This is a substantial architectural rewrite (~20k lines changed). The upload path is redesigned around a per-stream remote replica reader that behaves like an osiris replica but uploads to S3 instead of writing to disk. The old rabbitmq_stream_s3_server (1288 lines) and rabbitmq_stream_s3_machine (1848 lines) - a functional core / imperative shell pair that ran upload lifecycle inside the writer's hot path - are deleted and replaced by:
rabbitmq_stream_s3_replica_reader- per-streamgen_server(imperative shell)rabbitmq_stream_s3_replica_reader_core- pure functional corerabbitmq_stream_s3_governor- per-node token-bucket upload pacingrabbitmq_stream_s3_hooks- replacesosiris_log_manifestbehaviourrabbitmq_stream_s3_manifest_replica- per-node manifest cache and broadcast receiverrabbitmq_stream_s3_fragment_assembly- pure chunk-to-fragment accumulatorrabbitmq_stream_s3_fragment_iterator- tree traversal over manifest entriesrabbitmq_stream_s3_token_bucket- pure rate limiter
The stated goals (bounded time-to-S3, zero hot-path overhead, single upload owner) are well-served by the new design.
Strengths:
-
Functional core / imperative shell split is well-executed. The
_coremodule has no I/O and is deterministically testable. Thereplica_reader_SUITE+replica_reader_core_SUITEtogether give very solid coverage. The split is cleaner here than the old machine/server split because the core is truly pure - no ETS, noerlang:system_timeexcept ininit/2andtick/2where it is unavoidable. -
Out-of-order completion handling is correct. The
in_flightqueue enforces cut order.pending_completionsbuffers arrivals.drain_completions/1walks the front in order. This is a clean resolution to the problem that caused complexity in the old server. -
appended_since_persisttracking is subtle but correct. The PR description notes that fragments arriving during an in-flight persist must be held for the next broadcast. The binary slicing atpersist_complete(<<_:PersistedBytes/binary, Remaining/binary>>) is correct givenin_persist_countwas snapshotted atstart_persisttime. (A bug in this logic was found and fixed post-merge in1e62661.) -
Epoch fencing via Khepri optimistic concurrency. The
#if_payload_version{}condition indb.erlfences deposed writers correctly, and thereinitializeeffect on conflict is clean. -
Token bucket implementation is correct and the governor properly handles the
unlimitedfast path without branching inside every submission.
execute_effect(
{upload_group, StreamId, Kind, Entries, _Pos, _Len},
State0
) ->
Self = self(),
spawn_link(fun() ->
...
Self ! {group_upload_result, Result}
end),
State0#state{pending_group_kind = Kind};Three facts combine to create a permanent hang:
- The replica reader traps exits (
process_flag(trap_exit, true)at line 251), so linked process deaths arrive as{'EXIT', Pid, Reason}messages. - The catch-all
handle_info(_Info, State)at line 416 silently discards those messages. - The core state's
rebalance_in_flightflag is only cleared by receiving{group_upload_result, ...}— either viagroup_upload_complete(core line 306) orgroup_upload_failed(core line 349). No other code path resets it.
The do_upload_group call is already wrapped in a try/catch inside the spawned fun, covering Erlang-level exceptions. But if the process is killed externally (OOM killer, exit(Pid, kill) from a supervisor or memory alarm handler), no {group_upload_result, ...} message is sent.
Consequence: Once rebalance_in_flight is stuck true, maybe_start_persist (core line 467) returns {State, []} unconditionally — no further manifests are ever persisted to Khepri. The replica reader continues uploading fragments to S3, but:
- Replicas never learn about new fragments (broadcasts happen at persist time)
- On node restart, all fragments uploaded since the last persist are orphaned in S3
- Consumers reading from the remote tier see the stream stop growing
Fix: Replace spawn_link with spawn_monitor and add a {'DOWN', ...} handler that calls group_upload_failed, matching the pattern already used for the persist task (lines 679 + 406). Plain spawn (as used for delete_manifest_objects at line 499) removes the exit-signal hazard but does not solve the liveness problem — if the process dies without sending the result message, rebalance_in_flight is still stuck forever.
is_retriable({http, 500}) -> true;
is_retriable({http, 503}) -> true;
is_retriable(timeout) -> true;
is_retriable(_) -> false.S3 returns 429 when rate-limited. Under the current code a 429 is a fatal error that drops the fragment and accepts a gap (for individual transfers at line 203) or abandons the rebalance (for group uploads at line 350). With the governor pacing uploads this scenario is less likely, but it can still happen when a bucket policy threshold is lower than the configured rate. A 429 should be retriable.
drain_completions(State0) ->
case queue:peek(Q) of
{value, {Ref, _Meta}} ->
case maps:get(Ref, PC, undefined) of
{Uid, Meta} ->
...
{State3, Effects} = drain_completions(State2), %% recursive
...Each recursive call adds a stack frame. If a large batch of transfers completes concurrently (e.g. after a persist unblocks many that were queued), this recurses N times where N is the number of completed transfers. With the default persist_threshold = 5 this is bounded in normal operation, but if persist_threshold is set high or a burst of completions arrives post-reinitialize, the stack can grow large. Worth noting even if it does not cause problems at current thresholds.
case catch rabbitmq_stream_s3_db:get(StreamId) of
{ok, #{revision := Rev}} -> M#manifest{revision = Rev};
_ -> M
end;Both branches of resolve_manifest/1 (lines 817 and 822) use bare catch. catch returns {'EXIT', {Reason, Stack}} for errors/exits, which silently matches _. Using try ... catch _:_ -> M end (or #manifest{} for the second branch) is the correct pattern and makes the fallback intentional rather than accidental.
5. remove_for_max_bytes/4 never removes the last root entry - comment should explain why (manifest.erl:195)
case Kind of
?MANIFEST_KIND_FRAGMENT when Rest =/= <<>> ->
remove_for_max_bytes(Rest, TotalSize - Size, MaxBytes, N + 1);
_ ->
%% Hit a group entry or last entry. Stop here.
N
end.The Rest =/= <<>> guard (same in remove_for_max_age/3 at line 208) ensures the last fragment on the root is never deleted. The in-group variants (remove_for_max_bytes_in_group, remove_for_max_age_in_group) intentionally omit this guard (groups can be fully consumed). The current comment says "Hit a group entry or last entry. Stop here." - it should explain why we keep the last root entry (to avoid exposing an apparently empty stream while data still exists remotely but the manifest has been fully truncated).
get_meta_from_queue(Ref, [{Ref, Meta} | _]) -> Meta;
get_meta_from_queue(Ref, [_ | Rest]) -> get_meta_from_queue(Ref, Rest).If Ref is not in the queue, this crashes with a function_clause error. This cannot currently happen because transfer_complete/3 and transfer_failed/3 are the only callers and they use refs that were previously inserted into the queue. But it is a fragile ordering assumption with no guard. A fallback clause with a meaningful error (e.g. error({unknown_ref, Ref})) would make failures diagnosable.
- The module docs (using
-moduledoc) are excellent and detailed throughout. - Comments follow the project convention (above the line, not inline) correctly.
- The comment on
drain_completions/1explaining the timer fallback for low-throughput streams is a good example of a non-obvious invariant explained well. - The
?ENTRY/6binary macro with a matching?ENTRY/7pattern macro is clean and used consistently. - The
counter_fields/0export and the?NODE_COUNTERS/?STREAM_COUNTERSsplit for Prometheus monotonicity is well-documented inline. - Minor:
src/rabbitmq_stream_s3_log_reader.erl:507has a stale comment ("revise this comment as more moves into the server") that should be updated now that the server is gone.
Coverage is good at the unit level:
replica_reader_core_SUITE(836 lines) covers all state-machine interleavings including out-of-order completions, threshold triggers, tick behavior, persist failures, rebalancing, and recursive rebalancing - this is the most important test suite.fragment_iterator_SUITE,fragment_assembly_SUITE,prop_SUITEcover the pure modules well.broker_SUITEprovides end-to-end coverage.
Gaps:
- No test for the
{http, 429}retriability gap noted above. - No test for
drain_completionswith a large in-flight queue (stack depth concern). - No test exercising the group-upload
spawn_linkcrash path (it would hang rather than fail).
This is a high-quality rewrite that solves the stated problems cleanly. The core/shell split, the sequence-number broadcasting mechanism, and the functional test coverage are all well-implemented. Issues to address, in priority order:
- Must fix: Replace
spawn_linkwithspawn_monitor+'DOWN'handler for the group upload task (issue 1). - Should fix: Add
{http, 429}tois_retriable/1(issue 2). - Should fix: Replace
catchwithtry/catchinresolve_manifest/1(issue 4). - Minor: Add a base case or meaningful error to
get_meta_from_queue/2(issue 6). - Minor: Add a comment explaining why the last root entry is preserved (issue 5).
- Minor: Update the stale comment at
log_reader.erl:507.