manifest cache: Make branching on the cache state structurally total

Replace per-caller case expressions over get_manifest/1 with
with_manifest/2, a fold whose handler map has mandatory keys for all
three cache states (resolved, pending, absent). A call site that fails
to consider a state no longer compiles a quiet fallback into place; it
fails to match at the call site, and when a state is ever added every
caller crashes loudly instead of inheriting a neighbor's behavior.

This is the structural successor to the no-catch-alls convention: the
boot-window read bug existed because each consumer branched on the bare
accessor with its own clauses and a catch-all arm absorbed the state it
had not considered. get_manifest/1 remains for non-branching uses
(diagnostics, tests). get_range/1 and get_manifest_and_epoch/1 keep
their documented single-value projections.

Migrated: the log reader's four resolution clauses and resolve_first,
the retention-counter hook, GC's still_dangling re-check, the remote
reader's iterator refresh, and the replica reader's manifest resolution
and reconciler re-seed. No behavior change; the full suite set from the
pending-state change stays green, including the single-node boot-window
regression.

Author: the-mikedavis

docs/conventions.md

@@ -28,6 +28,8 @@ The chunk header is not a fixed 48 bytes. It includes a bloom filter whose size
 
 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

src/rabbitmq_stream_s3_gc.erl

@@ -286,8 +286,8 @@ still_dangling(#{reason := no_anchor}) ->
     %% the absence is permanent, so there is nothing to re-validate.
     true;
 still_dangling(#{reason := below_first_offset, stream_id := StreamId, key := Key}) ->
-    case rabbitmq_stream_s3_manifest_replica:get_manifest(StreamId) of
-        #manifest{first_offset = FirstOffset} = Manifest ->
+    rabbitmq_stream_s3_manifest_replica:with_manifest(StreamId, #{
+        resolved => fun(#manifest{first_offset = FirstOffset} = Manifest) ->
             case parse_key(Key) of
                 {data, _StreamId, Offset} ->
                     Offset < FirstOffset;
@@ -296,14 +296,14 @@ still_dangling(#{reason := below_first_offset, stream_id := StreamId, key := Key
                         not live_leading_group(StreamId, Key, Manifest);
                 _ ->
                     false
-            end;
-        Unresolved when Unresolved =:= undefined; Unresolved =:= pending ->
-            %% No live manifest to compare against (missing row, or a pending
-            %% marker whose manifest has not resolved yet): do not delete (a
-            %% later sweep reclaims a genuine orphan once a floor is known
-            %% again).
-            false
-    end.
+            end
+        end,
+        %% No live manifest to compare against (a pending marker whose manifest
+        %% has not resolved yet, or a missing row): do not delete (a later
+        %% sweep reclaims a genuine orphan once a floor is known again).
+        pending => fun() -> false end,
+        absent => fun() -> false end
+    }).
 
 %% Whether the group object Key is protected by the LIVE manifest's leading-group
 %% carve-out: it is the live referenced leading group, or the live manifest is in

src/rabbitmq_stream_s3_hooks.erl

@@ -151,20 +151,22 @@ on_retention_evaluated(Cnt, #{name := Name}) ->
     %% of -1, which must never reach the counter. get_range/1 returns `empty`
     %% in that case, so the override only runs when the remote tier actually
     %% holds data and its first_timestamp is a real timestamp.
-    case rabbitmq_stream_s3_manifest_replica:get_manifest(StreamId) of
-        #manifest{entries = <<>>} ->
-            ok;
-        #manifest{first_offset = RemoteFirst, first_timestamp = RemoteFirstTs} ->
-            LocalFirst = counters:get(Cnt, ?C_OSIRIS_LOG_FIRST_OFFSET),
-            counters:put(Cnt, ?C_OSIRIS_LOG_FIRST_OFFSET, min(LocalFirst, RemoteFirst)),
-            LocalFirstTs = counters:get(Cnt, ?C_OSIRIS_LOG_FIRST_TIMESTAMP),
-            counters:put(Cnt, ?C_OSIRIS_LOG_FIRST_TIMESTAMP, min(LocalFirstTs, RemoteFirstTs));
-        pending ->
-            %% Not yet resolved: nothing known to override the counters with.
-            ok;
-        undefined ->
-            ok
-    end.
+    rabbitmq_stream_s3_manifest_replica:with_manifest(StreamId, #{
+        resolved => fun
+            (#manifest{entries = <<>>}) ->
+                ok;
+            (#manifest{first_offset = RemoteFirst, first_timestamp = RemoteFirstTs}) ->
+                LocalFirst = counters:get(Cnt, ?C_OSIRIS_LOG_FIRST_OFFSET),
+                counters:put(Cnt, ?C_OSIRIS_LOG_FIRST_OFFSET, min(LocalFirst, RemoteFirst)),
+                LocalFirstTs = counters:get(Cnt, ?C_OSIRIS_LOG_FIRST_TIMESTAMP),
+                counters:put(
+                    Cnt, ?C_OSIRIS_LOG_FIRST_TIMESTAMP, min(LocalFirstTs, RemoteFirstTs)
+                )
+        end,
+        %% Not yet resolved: nothing known to override the counters with.
+        pending => fun() -> ok end,
+        absent => fun() -> ok end
+    }).
 
 -doc """
 Discover existing osiris writers and replicas on this node and attach