feat: add state replicator crate on top of the engine - #25
Open
bmuddha wants to merge 7 commits into
Open
Conversation
This was referenced Jul 13, 2026
Collaborator
Author
|
Warning This pull request is not mergeable via GitHub because a downstack PR is open. Once all requirements are satisfied, merge this PR as a stack on Graphite.
This stack of pull requests is managed by Graphite. Learn more about stacking. |
bmuddha
force-pushed
the
replicator
branch
2 times, most recently
from
July 23, 2026 18:53
996c13e to
80d6710
Compare
bmuddha
force-pushed
the
replicator
branch
2 times, most recently
from
July 27, 2026 13:19
482523d to
a16851e
Compare
bmuddha
marked this pull request as ready for review
July 28, 2026 12:15
This was referenced Jul 30, 2026
|
No actionable comments were generated in the recent review. 🎉 ℹ️ Recent review info⚙️ Run configurationConfiguration used: Path: .coderabbit.yaml Review profile: CHILL Plan: Pro Plus Run ID: ⛔ Files ignored due to path filters (1)
📒 Files selected for processing (12)
🚧 Files skipped from review as they are similar to previous changes (10)
📝 WalkthroughWalkthroughAdded the ChangesDurable engine replication
Estimated code review effort: 5 (Critical) | ~120 minutes Sequence Diagram(s)sequenceDiagram
participant Follower
participant ReplicationClient
participant ReplicationDispatcher
participant LeaderStorage
participant FollowerEngine
Follower->>ReplicationClient: start replication
ReplicationClient->>ReplicationDispatcher: send signed cursor request
ReplicationDispatcher->>LeaderStorage: select retained stream or snapshot
LeaderStorage-->>ReplicationDispatcher: return replication data
ReplicationDispatcher-->>ReplicationClient: stream entries or snapshot
ReplicationClient->>FollowerEngine: apply entries and seals
ReplicationClient->>ReplicationDispatcher: reconnect from flushed cursor
Possibly related PRs
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✨ Finishing Touches📝 Generate docstrings
🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 15
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@replicator/Cargo.toml`:
- Line 24: Update the replicator tokio dependency feature list to include "rt"
alongside the existing features, ensuring server.rs and client.rs can use
tokio::spawn and runtime Builder APIs when the crate is built independently.
In `@replicator/src/client.rs`:
- Around line 92-110: Update consume so ReadError::Io only reconnects for
genuine transport failures; inspect the underlying io::ErrorKind and continue
the loop for WouldBlock or TimedOut. Ensure partial-frame reads are handled
safely: if blockstore::decode cannot resume after a timeout, preserve reconnect
behavior and increase the socket read timeout above the leader’s block interval.
- Around line 206-220: Update stage_snapshot to write the archive to a temporary
path, verify the copied length, sync it, and rename it to
ACCOUNTSDB_SNAPSHOT_FILE only after completion. Ensure any io::copy,
length-validation, or subsequent staging failure removes the temporary archive
and staged dir, including when the socket read timeout interrupts the copy;
retain the existing superblock append only after successful publication.
In `@replicator/src/error.rs`:
- Around line 11-55: Restrict the derive_more From generation in
ReplicationError by adding #[from] only to the intended external-error variants:
IO, State, Engine, Ledger, Serde, and Timeout. Leave RestartRequired, Handshake,
Snapshot, PositionNotFound, VersionMismatch, SnapshotUnavailable,
ReconnectExhausted, and StreamClosed without #[from] so they cannot be
constructed implicitly through into() or ?.
In `@replicator/src/protocol.rs`:
- Around line 88-103: Update the handshake flow around `verify` to prevent
captured `HandshakeRequest` messages from being replayed: have the leader issue
a fresh random nonce before accepting the request, carry it through the
handshake, and include it in the bytes passed to `message` for signature
verification. If challenge-response is not implemented in this change, document
the residual replay risk in `README.md` beside the transport-encryption
exclusion.
In `@replicator/src/server.rs`:
- Around line 128-137: Bound concurrency in the dispatch path before spawning
workers: update the owner of dispatch and ReplicationServer::spawn coordination
to track active workers, reject or defer sockets once the configured maximum is
reached, and release the slot when each worker exits. Ensure the limit applies
before the allowlist check and prevents unbounded threads and Tokio runtimes for
stalled peers.
- Around line 300-311: The send method in replicator/src/server.rs lines 300-311
must return early when end <= start before computing the byte length, preventing
send_range from receiving a wrapped value; in send_range at lines 314-328,
detect a zero-byte result from snedfile::send_exact and return an UnexpectedEof
error instead of continuing the loop.
- Around line 300-311: Update the send method to validate that end is greater
than or equal to self.position.current.offset before subtracting; explicitly
reject inverted ranges with the function’s existing Result error path, and only
call send_range and advance the cursor after validation succeeds.
In `@replicator/tests/integration.rs`:
- Around line 510-528: Update the integration test authority setup so the middle
node uses its own distinct Keypair instead of shared, while retaining the leader
identity as middle_authority.remote. Ensure middle_identity is derived from the
middle keypair, so the dispatcher allowlist validates the follower identity
rather than the leader’s or the middle node’s self-identity.
- Around line 322-327: Update replicator/tests/integration.rs at lines 322-327
to verify the rejected handshake produces a non-IO ReplicationError and returns
immediately; if it remains IO-based, increase TIMEOUT beyond the cumulative
reconnect backoff from ReplicationClient::reconnect. At lines 427-441, assert
the client is still running before starting the second dispatcher, or explicitly
derive the close-and-reopen window from MAX_RECONNECT_ATTEMPTS and RETRY_DELAY.
- Around line 265-271: Extract the repeated linked leader/follower authority
construction into a shared authorities helper, then replace the duplicated setup
in engines, the current integration test, and the cascade test with that helper.
Preserve each caller’s existing pacing configuration, including the leader’s
Internal pacing here.
- Around line 349-353: Rewrite the comment above the post-outage replication
flow to clarify that replaying one entry would incorrectly produce 6, while the
expected value of 5 demonstrates exactly-once application. Leave the assertion
and surrounding code unchanged.
- Around line 34-38: Update loopback_addr and the ReplicationDispatcher::spawn
setup to retain the bound TcpListener through dispatcher startup, converting it
to Tokio as needed instead of releasing and rebinding the port. Ensure the
dispatcher listener configuration supports rebinds after TIME_WAIT, particularly
for streams_and_resumes_without_duplicate_application and
resumes_after_leader_restart; alternatively retry the complete
address-allocation and spawn sequence on EADDRINUSE.
- Around line 120-131: Update await_replication’s positions.recv() handling to
explicitly continue on broadcast::error::RecvError::Lagged, while still failing
only when the stream is closed; remove the strict observed == expected assertion
from the receive loop so skipped positions do not cause false failures. Preserve
the helper’s existing durable-position validation, relaxing it to require
follower.superblocks().position() >= expected if lag can skip the exact target.
- Around line 91-96: Update restart_from_snapshot so the
follower.shutdown().wait() asynchronous wait is bounded by the file’s existing
TIMEOUT constant, preserving the current shutdown, close, and restart sequence
while allowing the test to fail with the standard timeout context instead of
hanging.
🪄 Autofix
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: ASSERTIVE
Plan: Pro Plus
Run ID: 3f83ee67-9c0a-4c13-b6f4-de7f9d3e5e24
📒 Files selected for processing (17)
accountsdb/src/store/index.rsaccountsdb/src/store/mmap.rskeeper/src/lib.rsledger/Cargo.tomlledger/src/index.rsledger/src/lib.rsledger/src/reader.rsreplicator/Cargo.tomlreplicator/README.mdreplicator/src/client.rsreplicator/src/error.rsreplicator/src/lib.rsreplicator/src/metrics.rsreplicator/src/protocol.rsreplicator/src/server.rsreplicator/tests/integration.rsrust-toolchain.toml
Comment on lines
+349
to
+353
| // Resume after an outage from the durable cursor; replaying one entry yields 6. | ||
| first_dispatcher.terminate().await; | ||
| let expected = commit_increment(&mut leader, state).await; | ||
| let mut second_dispatcher = dispatcher(upstream, &leader, &[follower_identity]).await; | ||
| await_replication(&mut positions, &follower, expected, state, 5).await; |
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick win
The comment states an outcome that contradicts the assertion.
The comment says "replaying one entry yields 6", but Line 353 expects 5. The intended meaning is that a duplicate application would produce 6, and 5 proves exactly-once application. Rewrite the comment so it does not read as the expected value.
📝 Proposed wording
- // Resume after an outage from the durable cursor; replaying one entry yields 6.
+ // Resume after an outage from the durable cursor. Exactly-once application
+ // yields 5; a duplicate application of the same entry would yield 6.📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
Suggested change
| // Resume after an outage from the durable cursor; replaying one entry yields 6. | |
| first_dispatcher.terminate().await; | |
| let expected = commit_increment(&mut leader, state).await; | |
| let mut second_dispatcher = dispatcher(upstream, &leader, &[follower_identity]).await; | |
| await_replication(&mut positions, &follower, expected, state, 5).await; | |
| // Resume after an outage from the durable cursor. Exactly-once application | |
| // yields 5; a duplicate application of the same entry would yield 6. | |
| first_dispatcher.terminate().await; | |
| let expected = commit_increment(&mut leader, state).await; | |
| let mut second_dispatcher = dispatcher(upstream, &leader, &[follower_identity]).await; | |
| await_replication(&mut positions, &follower, expected, state, 5).await; |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@replicator/tests/integration.rs` around lines 349 - 353, Rewrite the comment
above the post-outage replication flow to clarify that replaying one entry would
incorrectly produce 6, while the expected value of 5 demonstrates exactly-once
application. Leave the assertion and surrounding code unchanged.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
What changed
Added authenticated state replication over the engine, including a leader-side
TCP dispatcher, follower client, versioned protocol, snapshot fallback, and
full-stack replication coverage.
Why
Authorized followers need to resume from an exact durable cursor, catch up
without duplicate application, and recover from a snapshot when the requested
stream is no longer retained.
Closes #27.
Impact
allowlist and clients verify the canonical engine authority.
distinct-key followers are terminal.
accountsdb snapshot after validating the requested cursor.
boundaries flow through the external pacer to preserve ordering.
Reviewer notes
The security boundary is the canonical signer plus follower allowlist. The
durability boundary requires a follower to flush before every handshake and to
flush its cursor before dumping volatile state during shutdown.
Follow-up
Transport encryption, key rotation, and dynamic allowlist management remain
outside this change.