Merge pull request #316 from amazon-mq/gc-anchor

Write anchors to the DB in order to identify dangling objects

Author: the-mikedavis

docs/architecture.md

@@ -155,7 +155,7 @@ Each stream on the writer node has exactly one replica reader. Its lifecycle:
 2. **Resolve manifest.** Fetches the current manifest state to determine where to resume uploading. If S3 is unreachable, retries with backoff. The writer is unaffected.
 3. **Open data reader.** Calls `osiris_writer:init_data_reader/3` starting at the manifest's `next_offset`.
 4. **Drain loop.** On each `{osiris_offset, ...}` notification, reads committed chunk headers via `osiris_log:read_header/1`. Accumulates metadata in the fragment assembly.
-5. **Cut.** When the assembly reaches the size target (default 64 MiB), the fragment is sealed. The core assigns a reference and returns a `{submit_transfer, ...}` effect.
+5. **Cut.** When the assembly reaches the size target (default 64 MiB), the fragment is sealed. The core assigns a reference and returns a `{submit_transfer, ...}` effect. On a brand-new stream the very first cut instead returns a `{write_anchor, ...}` effect and holds the fragment's submit: the shell writes the per-stream anchor to Khepri, and the core releases the held submits only once the anchor is durable, so no S3 object can exist under a prefix whose anchor is absent (this is what makes deleted-stream GC safe; see [operations.md](./operations.md#safety-guarantee-the-anchor)). A reader resuming an established stream already has the anchor and skips this step.
 6. **Submit.** The shell submits the transfer to the governor. The drain loop continues immediately without waiting for the upload to complete.
 7. **Transfer complete.** The governor reports completion. The core applies the completion in cut order (buffering out-of-order arrivals). When a contiguous prefix of completions is ready, they are applied to the in-memory manifest.
 8. **Durable persist.** When the persist threshold is reached (N fragments applied, or timer fires), the shell spawns a task that PUTs the manifest root to S3 and writes to Khepri (optimistic lock).
@@ -176,13 +176,13 @@ When the rate is configured as `unlimited` (the default), the governor imposes n
 
 When a stream queue is deleted:
 1. RabbitMQ removes the queue from `rabbit_db_queue` in Khepri.
-2. The plugin's keep-while condition on its Khepri entry fires, removing the entry.
+2. The plugin's keep-while condition on the per-stream container node fires, removing the container and the subtree below it (the manifest pointer and the anchor).
 3. The stored procedure `handle_queue_deletion` runs, calling `rabbitmq_stream_s3_reaper:delete_stream(StreamId)`.
 4. The reaper spawns a short-lived task that pages through `rabbitmq_stream_s3_api:list/2` on the stream's prefix.
 5. Each page of keys is sent back to the reaper as a delete cast.
 6. The reaper batches keys (up to 1000) and calls `rabbitmq_stream_s3_api:delete/1`.
 
-If the task crashes or S3 is unavailable, the orphan detection mechanism eventually cleans up.
+If the task crashes or S3 is unavailable, the objects are left behind, but the anchor is already gone (step 2 removed the whole subtree), so a later GC run reaps the prefix via the no-anchor path. See [operations.md](./operations.md#garbage-collection).
 
 ## Read path (consumer side)
 

docs/failure-modes.md

@@ -203,7 +203,7 @@ Committed data (offsets 76-100, confirmed to publishers) is lost. This requires
 - Deposed replica reader uploaded a fragment that was never applied to the manifest.
 - Manifest truncation removed entries but object deletion failed or is pending.
 - Stream deletion cleanup was interrupted.
-- A node's data directory was wiped (e.g. during node replacement or disaster recovery) and RabbitMQ restarted. The Khepri database is recreated from scratch and any streams that existed before the wipe are gone from Khepri, but not via a delete operation, so the keep-while condition that triggers S3 cleanup never evaluates and the S3 objects remain.
+- A node's data directory was wiped (e.g. during node replacement or disaster recovery) and RabbitMQ restarted. The Khepri database is recreated from scratch and any streams that existed before the wipe are gone from Khepri, but not via a delete operation, so the keep-while condition that triggers S3 cleanup never evaluates and the S3 objects remain. The per-stream anchor is also gone with the wiped database, so a GC run reclaims these objects via the no-anchor path (see Resolution below).
 
 **Impact assessment.** Wasted S3 storage. No correctness impact. No impact on consumers or publishers.
 
@@ -216,7 +216,7 @@ Committed data (offsets 76-100, confirmed to publishers) is lost. This requires
 aws s3 ls "s3://${BUCKET}/rabbitmq/stream/${STREAM_ID}/data/" --recursive
 ```
 
-**Mitigation.** None needed urgently. Orphans do not affect operation. For unbounded accumulation (typically only after data-directory resets), operators can manually delete S3 prefixes for streams that no longer exist in Khepri.
+**Mitigation.** None needed urgently. Orphans do not affect operation.
 
 **Resolution.** The `rabbitmq_stream_s3_gc` module identifies and optionally deletes orphaned objects. Run it via the CLI:
 
@@ -226,4 +226,4 @@ rabbitmq-streams stream_s3_gc --formatter json
 rabbitmq-streams stream_s3_gc --mode delete
 ```
 
-See [operations.md](./operations.md#garbage-collection) for details on the GC mechanism, its safety guarantees, and current scope limitations.
+Objects whose stream is still live are classified against its first_offset and epoch. Objects whose stream is gone from Khepri (a deleted stream, or one whose metadata was wiped) are classified against the per-stream anchor: with the stream gone the anchor is absent, so a GC run reaps them via the no-anchor path rather than requiring manual deletion. See [operations.md](./operations.md#garbage-collection) for details on the GC mechanism and its safety guarantees.

docs/invariants.md

@@ -69,6 +69,16 @@ Append, rebalance, and retention keep the remote tier a contiguous extension bel
 |---|---|---|
 | Reset safety | When `f_local > n`, the writer discards `M` and installs the empty manifest at the floor: `f := n := f_local` and `Frag := ∅`. Coverage and Durability hold vacuously, since `Cov = ∅ = [f, n)`; the step advances `n` over `[n, f_local)` only while emptying coverage, so it never claims durability it lacks. This is the inverse of the #206 hole, which advanced `n` while claiming coverage. `f` and `n` are non-decreasing, and the seam is restored at `f_local`, so Tier overlap holds again with the remote tier empty and the local segments covering `[f_local, T)`. The discarded range `[f, f_local)` is lost by design: it is the un-uploaded tail together with the now-disconnected durable prefix, which cannot be kept without a hole in `M`. | [architecture.md](./architecture.md) (Durability versus the local retention bound) |
 
+## Garbage collection
+
+GC reaps an S3 object only when it can prove the object is not live. For an object whose stream is still in the committed lookup that proof is the monotonic offset/epoch barrier. For an object whose stream is absent from the lookup (deleted, never committed, or missing a local replica) the proof is the per-stream anchor.
+
+| Invariant | Statement | Where |
+|---|---|---|
+| Anchor ordering | A stream's anchor is committed to Khepri before its first S3 object is written, so no object can exist under a prefix whose anchor is absent. The anchor's presence therefore witnesses a live stream, and its absence witnesses a prefix that is safe to reap. | [operations.md](./operations.md#safety-guarantee-the-anchor) |
+| Anchor removal | The anchor is kept alive by a `keep_while` on the stream queue, so it is removed in the same transaction that deletes the queue. The "stream deleted" signal is permanent and cannot be lost to a crash. | [operations.md](./operations.md#safety-guarantee-the-anchor) |
+| Anchor read | An object whose stream is absent from the lookup is reaped only when a strongly-consistent read reports its anchor absent; a present anchor or a non-quorum read fails closed. A stale local read could report a live stream's anchor absent, so the consistent read is load-bearing. Modeled in [PR #311](https://github.com/amazon-mq/rabbitmq-stream-s3/pull/311). | [operations.md](./operations.md#safety-guarantee-the-anchor) |
+
 ## Read path
 
 | Invariant | Statement | Where |

docs/operations.md

@@ -636,28 +636,39 @@ curl http://node:15692/metrics/per-object | grep 'queue="my-stream"'
 
 ## Garbage collection
 
-Objects in S3 can become orphaned (not referenced by any manifest) in several scenarios: a deposed writer uploads a fragment but loses the Khepri race, a manifest persist fails or the process crashes before completion, or retention deletes entries from the manifest but the corresponding object deletion does not complete.
+Objects in S3 can become orphaned (not referenced by any manifest) in several scenarios: a deposed writer uploads a fragment but loses the Khepri race, a manifest persist fails or the process crashes before completion, retention deletes entries from the manifest but the corresponding object deletion does not complete, or a stream is deleted (or its Khepri metadata is wiped) and its objects outlive the metadata that referenced them.
 
 The GC mechanism identifies these orphans by listing S3 objects and comparing their keys against current authoritative state. It is invoked on demand via the CLI.
 
 ### Safety guarantee: monotonicity
 
-The correctness of GC depends on a single structural property: the values used as barriers (`first_offset`, `epoch`) only ever increase. This makes false positives (deleting a live object) structurally impossible, regardless of timing or consistency:
+For an object whose stream is still in the committed lookup, correctness rests on a structural property: the values used as barriers (`first_offset`, `epoch`) only ever increase. This makes false positives (deleting a live object) structurally impossible, regardless of timing or consistency:
 
 - `first_offset` advances monotonically as retention removes data from the front of the manifest. An object whose offset is below first_offset was already removed by retention. It can never become live again.
 - `epoch` advances monotonically with each new writer. A manifest root whose epoch is below the current epoch belongs to a deposed writer. The new writer's manifest is authoritative and the old root can never become active again.
 
 Eventually-consistent reads (from Khepri or the manifest replica ETS cache) can only return stale values that are lower than or equal to the true current value. A stale barrier is more conservative: it classifies fewer objects as garbage. The GC may miss orphans on a given run (false negatives) but can never delete live objects (false positives).
 
+### Safety guarantee: the anchor
+
+Monotonicity covers objects whose stream is still in the committed lookup. An object whose stream is absent from the lookup (a deleted stream, a stream that never committed a manifest, or one missing a local manifest replica) has no offset or epoch barrier to classify it against. These are resolved against the per-stream anchor instead.
+
+Before the first remote-tier fragment is uploaded, the replica reader writes an anchor node in Khepri, kept alive by a `keep_while` condition on the stream queue. Two properties make it a correct-by-construction signal:
+
+- Anchor-before-fragment ordering: no S3 object can exist under a stream's prefix unless that stream's anchor committed first. An object under a prefix with no anchor is therefore junk.
+- Atomic removal: the `keep_while` removes the anchor in the same transaction that deletes the queue, so the "stream deleted" signal is permanent and cannot be lost to a crash.
+
+An object whose stream is not in the lookup is reaped only when a strongly-consistent (quorum-requiring) read reports its anchor absent. A present anchor, or a read that cannot reach a quorum, fails closed and the object is skipped. The consistent read is load-bearing: a stale local read could report a live stream's anchor absent and reap it. This safety rests on ordering and consistency rather than on the monotonicity argument above, and it is what lets a deleted stream's objects be reclaimed automatically rather than by hand.
+
 ### Classifying garbage
 
 | Object type   | Key format                                                  | Condition for "garbage"
 |---            |---                                                          |---
-| Fragment      | `rabbitmq/stream/<id>/data/<offset>.<uid>.fragment`         | offset < manifest first_offset
-| Group         | `rabbitmq/stream/<id>/metadata/<offset>.<uid>.<kind>`       | offset < manifest first_offset
-| Manifest root | `rabbitmq/stream/<id>/metadata/root.<epoch>.<uid>.manifest` | epoch < current epoch according to Khepri
+| Fragment      | `rabbitmq/stream/<id>/data/<offset>.<uid>.fragment`         | offset < manifest first_offset, or the stream's anchor is absent
+| Group         | `rabbitmq/stream/<id>/metadata/<offset>.<uid>.<kind>`       | offset < manifest first_offset, or the stream's anchor is absent
+| Manifest root | `rabbitmq/stream/<id>/metadata/root.<epoch>.<uid>.manifest` | epoch < current epoch according to Khepri, or the stream's anchor is absent
 
-Note that objects belonging to an unknown stream ID are not currently considered garbage. Handling this case is planned.
+A well-formed key whose stream is not in the committed lookup is classified against the per-stream anchor, described under [Safety guarantee: the anchor](#safety-guarantee-the-anchor) above: it is reaped (reason `no_anchor`) only when a strongly-consistent read confirms the anchor absent, which reclaims the objects of a deleted stream. A key whose format is unrecognised is skipped outright.
 
 ### Modes
 
@@ -671,7 +682,8 @@ With `--formatter json`, the output is a JSON array of objects:
 ```json
 [
   {"stream_id": "...", "key": "rabbitmq/stream/.../00000000000000000042.abcd1234.fragment", "reason": "below_first_offset"},
-  {"stream_id": "...", "key": "rabbitmq/stream/.../metadata/root.1.deadbeef.manifest", "reason": "stale_epoch"}
+  {"stream_id": "...", "key": "rabbitmq/stream/.../metadata/root.1.deadbeef.manifest", "reason": "stale_epoch"},
+  {"stream_id": "...", "key": "rabbitmq/stream/.../00000000000000000042.abcd1234.fragment", "reason": "no_anchor"}
 ]
 ```