feat: add state replicator crate on top of the engine - #25
Conversation
|
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. |
55d60f6 to
1be4057
Compare
996c13e to
80d6710
Compare
482523d to
a16851e
Compare
|
Important Review skippedAuto reviews are disabled on base/target branches other than the default branch. Please check the settings in the CodeRabbit UI or the ⚙️ Run configurationConfiguration used: Organization UI Review profile: ASSERTIVE Plan: Pro Plus Run ID: You can disable this status message by setting the Use the checkbox below for a quick retry:
Warning
|
![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
| fn stage_snapshot(&self, connection: &mut TcpStream, meta: SnapshotMetadata) -> Result<()> { | ||
| let _timer = metrics::time(Operation::ClientStageSnapshot); | ||
| // Stage in the successor before seal rotation so restart can find it. | ||
| let dir = Superblock::init_dir(self.superblocks().directory(), meta.id + 1)?; | ||
| let archive = dir.join(ACCOUNTSDB_SNAPSHOT_FILE); | ||
| let mut file = File::options().write(true).create(true).truncate(true).open(&archive)?; | ||
| let written = io::copy(&mut connection.take(meta.len), &mut file)?; | ||
| file.sync_all()?; | ||
| if written != meta.len { | ||
| return Err(ReplicationError::Snapshot(meta.len, written)); | ||
| } | ||
| self.superblocks().append(meta.superblock)?; | ||
| info!(?meta, "replication snapshot staged"); | ||
| Ok(()) | ||
| } |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick win
A failed snapshot transfer leaves a partial archive that this node can serve.
stage_snapshot creates the successor superblock directory and writes ACCOUNTSDB_SNAPSHOT_FILE in place. If the transfer is short, or if the 4-second socket read timeout fires during a large archive copy, the function returns an error and leaves the truncated file on disk. The seal is not appended, so the follower does not install it, but the file remains.
ReplicationServer::snapshot in replicator/src/server.rs selects a snapshot with path.exists() alone. If this node later relays, it can advertise and send that truncated archive to a downstream follower. The leader-side writer in keeper/src/lib.rs avoids this by writing to a .tmp name and renaming only after sync_data.
Write to a temporary name, verify the length, then rename. Remove the staged directory on failure.
🛠️ Publish the archive only after it is complete
let dir = Superblock::init_dir(self.superblocks().directory(), meta.id + 1)?;
let archive = dir.join(ACCOUNTSDB_SNAPSHOT_FILE);
- let mut file = File::options().write(true).create(true).truncate(true).open(&archive)?;
- let written = io::copy(&mut connection.take(meta.len), &mut file)?;
- file.sync_all()?;
- if written != meta.len {
- return Err(ReplicationError::Snapshot(meta.len, written));
- }
+ let tmp = dir.join(format!("{ACCOUNTSDB_SNAPSHOT_FILE}.tmp"));
+ let mut file = File::options().write(true).create(true).truncate(true).open(&tmp)?;
+ let written = io::copy(&mut connection.take(meta.len), &mut file)?;
+ if written != meta.len {
+ let _ = fs::remove_file(&tmp);
+ return Err(ReplicationError::Snapshot(meta.len, written));
+ }
+ file.sync_all()?;
+ fs::rename(&tmp, &archive)?;
self.superblocks().append(meta.superblock)?;Also remove dir when io::copy itself fails, so no empty successor directory survives.
🤖 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/src/client.rs` around lines 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.
| #[derive(From, thiserror::Error, Debug)] | ||
| pub enum ReplicationError { | ||
| /// Socket or replicated-file access failed. | ||
| #[error("replication I/O failed: {0}")] | ||
| IO(#[source] io::Error), | ||
| /// Applying replicated state through the keeper failed. | ||
| #[error("failed to apply replicated state: {0}")] | ||
| State(#[source] KeeperError), | ||
| /// Applying a replicated entry through the execution engine failed. | ||
| #[error("replication engine operation failed: {0}")] | ||
| Engine(#[source] EngineError), | ||
| /// Reading or advancing replicated ledger storage failed. | ||
| #[error("replication ledger operation failed: {0}")] | ||
| Ledger(#[source] LedgerError), | ||
| /// A control message could not be encoded or decoded. | ||
| #[error("invalid replication control message: {0}")] | ||
| Serde(#[source] wincode::Error), | ||
| /// The peer uses a protocol version this crate cannot read. | ||
| #[error("replication protocol version mismatch; expected version {0}")] | ||
| VersionMismatch(u32), | ||
| /// The requested or published blockstore cursor is unavailable locally. | ||
| #[error("replication position is unavailable: {0:?}")] | ||
| PositionNotFound(BlockstorePosition), | ||
| /// The leader rejected the client's handshake. | ||
| #[error("replication handshake rejected: {0}")] | ||
| Handshake(String), | ||
| /// The snapshot connection ended before the advertised byte count arrived. | ||
| #[error("incomplete replication snapshot: expected {0} bytes, received {1}")] | ||
| Snapshot(u64, u64), | ||
| /// No complete retained snapshot can satisfy an unavailable cursor. | ||
| #[error("no complete replication snapshot is available")] | ||
| SnapshotUnavailable, | ||
| /// All bounded attempts to reconnect to the leader failed. | ||
| #[error("replication reconnect attempts exhausted")] | ||
| ReconnectExhausted, | ||
| /// A staged snapshot must be installed by restarting the engine. | ||
| #[error("replication snapshot for superblock {0} is staged; restart required")] | ||
| RestartRequired(u64), | ||
| /// A replication event stream closed before the transfer completed. | ||
| #[error("replication event stream closed")] | ||
| StreamClosed, | ||
| /// Waiting for a locally committed block boundary timed out. | ||
| #[error("timed out waiting for a replicated block boundary: {0}")] | ||
| Timeout(#[source] Elapsed), | ||
| } |
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick win
🧩 Analysis chain
🌐 Web query:
derive_more From derive on enum: does it generate From impls for all variants when no variant is annotated with #[from]?
💡 Result:
Yes, when you derive From for an enum using the derive_more crate, it generates From implementations for all variants by default if no variants are annotated with #[from] or #[from(ignore)] [1][2]. For enums, the derive_more crate treats each variant as if it were a struct and attempts to generate a From implementation for each one [1][2]. If your enum contains multiple variants that would result in conflicting From implementations—for example, if two variants contain the same type—you must use the #[from(ignore)] attribute to explicitly skip the variants you do not want to generate implementations for [2]. In summary: - By default, an implementation is generated for every variant [1][2]. - You can opt-out of specific variants using #[from(ignore)] [1][2]. - You can explicitly limit generation to specific variants by using #[from] only on those you want to include [1][2].
Citations:
- 1: https://docs.rs/derive_more/latest/derive_more/derive.From.html
- 2: https://jeltef.github.io/derive_more/derive_more/from.html
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "== locate error.rs and dependency files =="
fd -a 'error.rs|Cargo.toml|Cargo.lock' . | sed 's#^\./##' | head -100
echo
echo "== replicator/src/error.rs outline and relevant lines =="
if [ -f replicator/src/error.rs ]; then
wc -l replicator/src/error.rs
sed -n '1,90p' replicator/src/error.rs | cat -n
fi
echo
echo "== derive_more version references =="
rg -n 'derive_more|derive_more' Cargo.toml Cargo.lock replicator/Cargo.toml 2>/dev/null || true
echo
echo "== usages of ReplicationError construction / ? / into =="
rg -n 'ReplicationError|Timeout\(|RestartRequired\(|Handshake\(|Snapshot\(|PositionNotFound\(|\b\?\b|\binto\(\)' replicator/src -g '*.rs' | head -200Repository: magicblock-labs/magicblock-engine
Length of output: 8454
🌐 Web query:
derive_more 2.1.1 From derive enum all variants if no #[from] explicit
💡 Result:
In the derive_more crate version 2.1.1, the #[derive(From)] macro automatically generates From trait implementations for all enum variants by default [1][2]. You do not need to add explicit #[from] attributes to individual variants to enable this behavior; it is the standard behavior for the derive macro [1][2]. If you wish to opt-out of this behavior for specific variants, you can use the #[from(skip)] or #[from(ignore)] attributes on the variant you want to exclude [1][2]. Conversely, while the default is to include all variants, explicit #[from] annotations can also be used if you prefer to be explicit or if you need to configure specific field conversions [2]. Summary of behavior: - Automatic: By default, From implementations are generated for every enum variant [1][2]. - Opt-out: Use #[from(skip)] or #[from(ignore)] on specific variants to prevent From generation for them [2]. - Conflict avoidance: If multiple variants would result in overlapping From implementations (e.g., if two variants have the same field types), the compiler will generate an error unless you explicitly ignore the conflicting variants [1].
Citations:
- 1: https://jeltef.github.io/derive_more/derive_more/from.html
- 2: https://docs.rs/derive_more/latest/derive_more/derive.From.html
Restrict derive_more::From to the intended variants.
#[derive(From)] generates From implementations for every ReplicationError variant by default, so non-source variants such as RestartRequired(u64), Handshake(String), Snapshot(u64, u64), PositionNotFound(BlockstorePosition), VersionMismatch(u32), and unit variants can be created accidentally through .into() or ?. Add #[from] only to the variants that should accept external source errors, and leave the data-carrying and unit variants without #[from].
🤖 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/src/error.rs` around lines 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 ?.
| // 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.
| // 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.
1 participant
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.
Summary by CodeRabbit