Merge pull request #339 from amazon-mq/fix/cold-cache-remote-skip

Mark manifests as pending during resolution

Author: the-mikedavis

docs/architecture.md

@@ -2,6 +2,16 @@
 
 This document describes how the plugin's processes fit together: what starts them, what they own, and how data flows between them.
 
+## Design stance: durability versus availability
+
+The plugin resolves the durability-versus-availability trade differently on its two paths, and every mechanism below follows from this asymmetry.
+
+On the write path, availability wins and durability is a bounded, explicitly scoped promise. Producers never wait on S3: publishing, replication, and local consumption are unaffected by an object-store outage of any length. Durability in S3 is promised only for the uploaded range `[f, n)`, and the user's retention bound is authoritative over upload progress, so an outage longer than the local retention window can buffer costs the un-uploaded tail. The window is an operator-sized knob, not a guarantee (see [Durability versus the local retention bound](#durability-versus-the-local-retention-bound)).
+
+On the read path, correctness wins and availability degrades only in two sanctioned, visible ways. A read whose answer is momentarily unknowable (a transient fetch failure, an unresolved manifest cache) fails closed and the consumer retries. A read whose data is permanently gone (retention, a gap from an outage) is repositioned at the oldest available offset, the same observable semantics as vanilla stream retention. What is never sanctioned is silently serving an incomplete answer: a stalled read is recoverable, a wrong one is not (see the Cached state and Read path sections of [invariants.md](./invariants.md)).
+
+In one sentence: the write path never lets an S3 outage become a local availability problem, at the cost of a bounded, un-uploaded tail; the read path never lets an unknown answer look like a correct one, at the cost of availability.
+
 ## Plugin lifecycle
 
 The plugin is enabled via `rabbitmq-plugins enable rabbitmq_stream_s3`. RabbitMQ starts it after the broker is fully up (after `core_started`). On start, the plugin's supervisor initializes all infrastructure and sets the osiris hooks. Streams that already exist at this point are discovered and attached to (see [Stream discovery](#stream-discovery)). Streams created after plugin start are handled by the hooks.

docs/conventions.md

@@ -24,6 +24,20 @@ The `log_hooks` key follows this pattern. Passing `log_hooks => undefined` in th
 
 The chunk header is not a fixed 48 bytes. It includes a bloom filter whose size depends on `filter_size` configuration. Do not hardcode chunk header sizes. Use `data_size` and `position`/`next_position` from the header map to reason about byte offsets.
 
+## Derived state has no catch-alls
+
+Any state derived from the manifest (the per-node manifest cache, the osiris first-offset counters, iterator snapshots) can be missing or not yet established, and "missing" is never the same value as "empty". When branching on such state, write one clause per state and no catch-all: a `_ ->` arm over cache states is how a cache miss gets silently collapsed into "no remote tier", which is the boot-window read bug (see the Cached state section of [invariants.md](./invariants.md)). If a new state is added, every consumer should fail to compile or crash loudly, not inherit an arbitrary neighbor's behavior.
+
+For the manifest cache this is enforced structurally, not by convention: branching goes through `rabbitmq_stream_s3_manifest_replica:with_manifest/2`, whose handler map has mandatory keys for all three states (`resolved`, `pending`, `absent`). A call site that fails to consider a state does not compile a quiet fallback into place; it fails to match. `get_manifest/1` remains only for non-branching uses (diagnostics, tests). New derived-state APIs should follow the same shape: expose a total fold, not a bare accessor that invites per-caller `case` expressions.
+
+The direction of each explicit miss clause matters: consumers must fail in the direction that cannot lose data. Readers fail closed (error, the consumer retries), retention deletes nothing, GC skips, resolution goes to the authoritative store. A fallback that degrades silently reads as robustness in review and is exactly where correctness leaks out.
+
+## Cold-state test dimension
+
+Every stateful fixture a test warms up implicitly is a state the suite must also construct cold. An end-to-end test that publishes and then reads has, as a side effect, seeded the manifest cache, attached the hooks, and resolved the manifest: the act of arranging the fixture destroys the cold-start state, so the warm path is the only path such suites can ever exercise.
+
+When a subsystem keeps per-node volatile state, its suite needs at least one case that restarts the owning process or node and exercises the consumer before anything re-warms the state (for streams: subscribe before any publish). Note the topology: with replicas present, the writer's sync re-seeds an acceptor's cache within milliseconds, so cold-cache cases need the single-node shape where no peer can heal the state (see `broker_SUITE:restarted_node_serves_first_from_remote_tier/1`).
+
 ## Pure functional cores
 
 Complex state machines use a functional-core / imperative-shell split inspired by Ra's `ra_server` / `ra_server_proc` separation.

docs/failure-modes.md

@@ -85,6 +85,24 @@ Discarding remote manifest and restarting from the local log.
 
 ---
 
+## Node restart with an idle stream (cold manifest cache)
+
+**Trigger.** A node restarts, emptying its in-memory manifest cache, and no publish follows. This is the most ordinary event in the catalog: an upgrade, a reboot, a crash-and-recover, followed by consumers reconnecting before producers do.
+
+**Impact assessment.** None. The member-init hooks mark the stream's cache row `pending` before the member finishes starting, and the replica reader's manifest resolution (writer node) or the writer's sync (replica nodes) replaces the marker moments later. A consumer that attaches inside that window has its subscription fail closed with `{error, {manifest_not_resolved, _}}` and retries; it cannot be silently attached at the local floor.
+
+**Detection.**
+
+- `rabbitmq_stream_s3_log_reader` counter `resolve_failed_closed` counts attaches that failed closed during the window; a burst right after a restart is expected, a sustained rate is not.
+- `rabbitmq_stream_s3_replica_reader` counter `manifest_resolution_failures` increments when resolution itself cannot complete (S3 or metadata store unavailable), which is what keeps rows pending.
+- Logs: `Failing reader setup closed for stream <stream_id>` (INFO, expected during the window); `Reconciliation: re-seeding the local manifest cache` (the reconciler healing a row that resolution could not seed, for example after a manifest_replica crash).
+
+**Mitigation.** None needed in the normal case; the window is milliseconds. If consumers fail closed for longer, manifest resolution is stuck: check S3 reachability and the metadata store, exactly as for an S3 outage.
+
+**Resolution.** Automatic. Resolution or sync replaces the pending marker; the reconciler heals leaked markers within one period.
+
+---
+
 ## Segment deleted before upload
 
 **Trigger.** Local retention deletes a segment the replica reader has not yet uploaded. Most likely during an S3 outage or when local retention is aggressive relative to throughput.

docs/invariants.md

@@ -79,9 +79,19 @@ GC reaps an S3 object only when it can prove the object is not live. For an obje
 | 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) |
 
+## Cached state
+
+The invariants above are stated over the manifest `M`, but no reader consults `M` directly: every consumer reads a node-local cache of it (`rabbitmq_stream_s3_manifest_replica`'s ETS table). The cache is volatile state with its own lifecycle, and the read-path invariants only follow from the durable-state invariants if the cache's relationship to `M` is itself pinned down. `Cache(node, S)` denotes the cache row for stream `S` on a node, with three states: a resolved manifest (possibly empty), `pending` (attached but not yet resolved or synced), and absent (no row).
+
+| Invariant | Statement | Where |
+|---|---|---|
+| Cache availability | On any node hosting a member of `S`, `Cache(node, S)` exists (pending or resolved) from before the member finishes init, and therefore from before any reader can attach. The member-init hooks write the pending marker synchronously; resolution or the writer's sync replaces it. An absent row is therefore a positive statement: the plugin never attached to `S` on this node (an un-tiered stream), whose local log is the whole stream. | [read-path.md](./read-path.md) (Offset spec resolution) |
+| Cache coherence | A resolved row is the writer's manifest at some point in time, and only moves forward: a replica's row is a prefix of the writer's edit history (Replica prefix), the writer's own row is written only by its resolution, persist, and reset paths, and `mark_pending` is insert-if-absent so a resolved row is never downgraded to pending. | [manifest.md](./manifest.md) (Manifest edit replication) |
+| Miss semantics | `pending` means the remote tier's extent is unknown, and every consumer must fail in the direction that cannot lose data: the read path fails closed (a retryable error, never a local fallback that would skip `[f, f_local)`), local retention deletes nothing, GC treats the stream as unsweepable, counters are left uncorrected, and the writer's own resolution ignores the row and reads the store. Absent means un-tiered: the read path serves locally, which is exact, not a fallback. No consumer may branch on a catch-all over cache states. | [read-path.md](./read-path.md) (Offset spec resolution) |
+
 ## Read path
 
 | Invariant | Statement | Where |
 |---|---|---|
-| Exactly-once | A read started at spec `s` delivers each offset of `[max(s, f), T)` exactly once, in strictly increasing order, across both tiers. A seek into a group resolves to the unique fragment containing the target (Ordering), not the group's first child. | [read-path.md](./read-path.md) |
+| Exactly-once | A read started at spec `s` delivers each offset of `[max(s, f), T)` exactly once, in strictly increasing order, across both tiers. A seek into a group resolves to the unique fragment containing the target (Ordering), not the group's first child. A fail-closed attach (Miss semantics) is not a violation: it delivers nothing and the consumer retries; only a delivery that skips an offset of `[max(s, f), T)` violates this invariant. | [read-path.md](./read-path.md) |
 | Tier overlap | The local segments cover `[n, T)` while the remote tier covers `[f, n)`; the ranges overlap and the reader hands off at a shared offset, so no offset is skipped or repeated at the seam. | [read-path.md](./read-path.md) |

docs/read-path.md

@@ -23,13 +23,18 @@ A consumer subscribes to a stream with an offset spec (`first`, `last`, `next`,
 
 Resolution determines which tier holds the requested offset and where within that tier to start reading.
 
-The log reader checks the manifest cache (`rabbitmq_stream_s3_manifest_replica:get_range/1`) to get the remote tier's `{first_offset, next_offset}`. It also checks the local log's first chunk ID via shared atomics.
+The log reader checks the manifest cache (`rabbitmq_stream_s3_manifest_replica:get_manifest/1`) for the remote tier's extent. It also checks the local log's first chunk ID via shared atomics. The cache row is in one of three states, and the decision depends on the state, not just the extent (see the Cached state section of [invariants.md](./invariants.md)):
 
-Decision logic:
+Decision logic with a resolved manifest:
 
+- If the offset is at or above the local first chunk ID: resolve to local (delegate to `osiris_log`). The cache is not consulted; `last` and `next` also short-circuit to local.
 - If the offset is below the local first chunk ID and within the manifest's range: resolve to remote.
-- If the offset is at or above the local first chunk ID: resolve to local (delegate to `osiris_log`).
-- If the offset is below both the local range and the manifest range: the data has been retained away. Return `{error, {offset_out_of_range, ...}}`.
+- If the offset is below both the local range and the manifest range: the data has been retained away. Attach at the oldest available data, matching vanilla stream retention behavior.
+
+Decision logic without one:
+
+- If the row is `pending` (the plugin is attached on this node but the manifest has not been resolved or synced yet, for example in the first moments after a member starts): fail closed with `{error, {manifest_not_resolved, StreamId}}` for any spec the remote tier may hold (`first`, a below-floor offset, a timestamp, `{abs, _}`). The subscription fails and the client retries; falling back to the local tier here would silently skip the remote range below the local floor. The `resolve_failed_closed` counter records these.
+- If there is no row at all: the plugin never attached to this stream on this node (an un-tiered stream), so the local log is the whole stream and local resolution is exact.
 
 For timestamp-based specs, the log reader binary-searches the manifest's entries (each entry has `first_timestamp` and `last_timestamp`) to find the fragment containing the target timestamp.
 

p/README.md

@@ -53,7 +53,7 @@ The read and tier-resolution seam: resolving the `first` offset spec when the le
 
 ### `tier-routing`
 
-The offset-to-tier routing branch of `resolve_remote_location`: when the local log is empty (`first_chunk_id == -1`), every offset must fall through to the remote tier. Verifies that dropping the `=/= -1` guard routes a remote-only offset to the local tail (a silent remote skip). A second INV#4 guard, distinct from the group-fetch catch-all. See [`tier-routing/README.md`](tier-routing/README.md).
+The offset-to-tier routing branch of `resolve_remote_location`, including the manifest cache's lifecycle (`Cold -> Pending -> Serving`). Verifies two INV#4 guards: the `=/= -1` empty-local floor guard, and the fail-closed handling of an unresolved cache (the boot-window bug: a node restart used to make the remote tier invisible to readers until a persist or the reconciler re-seeded the cache, so `first` attached at the local floor and silently skipped the remote range). The spec's oracle is the ground-truth extent announced by the driver, independent of the cache the reader consults, and the gates include a weakened-environment run (no pending marker) alongside the weakened-code runs. See [`tier-routing/README.md`](tier-routing/README.md).
 
 ### `trimmed-segment`
 

p/read-resolution/README.md

@@ -36,3 +36,7 @@ p check -tc tcReadResolveGuarded -i 5000   # 0 bugs
 p check -tc tcReadResolveBuggy   -i 2000   # 1 bug: INV#4 silent remote skip
 p check -tc tcReadResolveExplore -i 5000   # 0 bugs
 ```
+
+## Scope note
+
+This model covers the group-fetch classification inside `resolve_first` (a transient fetch failure must retry, never fall back local). The manifest cache's availability axis - `pending` versus resolved versus absent rows, and the boot window they create - lives in the sibling [`tier-routing`](../tier-routing/README.md) model; the `pending` clause in `resolve_first` itself is exercised directly by `read_resolution_SUITE`, which drives the real module.

p/tier-routing/PSpec/Monitors.p

@@ -1,13 +1,37 @@
 /* INV#4 (tier resolution total / correct): an offset held only by the remote
-   tier (covered by the remote extent, not by the local log) must route to the
-   remote tier. Routing it locally silently skips the remote data. An offset the
-   local tier also holds may route locally (osiris prefers the local reader), so
-   the obligation is only on offsets the local tier does NOT cover. */
-spec TierRoutingCorrect observes eResolution {
+   tier (covered by the remote extent, not by the local log) must not route to
+   the local tier. Routing it locally silently skips the remote data. RETRY is
+   acceptable there (fail closed, the consumer retries); LOCAL is not. An
+   offset the local tier also holds may route locally (osiris prefers the local
+   reader), so the obligation is only on offsets the local tier does NOT cover.
+
+   Independent oracle: coverage is re-derived here from the ground-truth extent
+   the DRIVER announced, never from anything the Reader or the cache reported.
+   The previous version of this spec computed remoteCovers from the
+   ManifestStore's reply, which made the cache simultaneously the component's
+   input and the monitor's oracle: a reader that believed an empty or missing
+   cache could not be flagged, and the boot-window bug was unrepresentable. */
+spec TierRoutingCorrect observes eGroundTruth, eResolution {
+  var truthKnown: bool;
+  var nonEmpty: bool;
+  var remoteFirst: int;
+  var remoteNext: int;
+
   start state Watching {
-    on eResolution do (r: (tier: Tier, remoteCovers: bool, localCovers: bool)) {
-      if (r.remoteCovers && !r.localCovers) {
-        assert r.tier == REMOTE,
+    on eGroundTruth do (g: (nonEmpty: bool, remoteFirst: int, remoteNext: int)) {
+      truthKnown = true;
+      nonEmpty = g.nonEmpty;
+      remoteFirst = g.remoteFirst;
+      remoteNext = g.remoteNext;
+    }
+    on eResolution do (r: (outcome: Outcome, offset: int, firstChunkId: int)) {
+      var remoteCovers: bool;
+      var localCovers: bool;
+      assert truthKnown, "eResolution observed before eGroundTruth: driver must announce the true extent first";
+      remoteCovers = nonEmpty && r.offset >= remoteFirst && r.offset < remoteNext;
+      localCovers = (r.firstChunkId != -1) && (r.offset >= r.firstChunkId);
+      if (remoteCovers && !localCovers) {
+        assert r.outcome != LOCAL,
           "INV#4 violated: an offset held only by the remote tier routed to the local tier (silent remote skip)";
       }
     }