Repository navigation
feat #327, #328: cluster membership streaming (EmbeddedEngine + GrpcClient watch_membership) - #366
Merged
Merged
Conversation
…t API - New `MembershipSnapshot` struct (members, learners, committed_index) - `EmbeddedEngine::watch_membership()` returns a `watch::Receiver<MembershipSnapshot>` that fires on every Raft-consensus-backed conf change (AddNode, Promote, BatchRemove) - Notification reaches all nodes (leader, follower, learner) via the existing CommitHandler::apply_config_change path — no extra machinery - 7 integration tests covering: initial snapshot, join, zombie warn-only, learner promotion, all-nodes notification, multiple subscribers, monotonic index
…or peer removed The inner reconnect loop in process_batch called open_replication_stream in a tight loop without checking whether the owning LeaderState had been dropped. When a peer was removed or the leader stepped down, the worker kept retrying indefinitely. Fix: check task_rx.is_closed() at the top of the reconnect loop. The task channel sender lives inside LeaderState; once it drops, is_closed() returns true and the worker returns immediately instead of spinning forever. Also hide leader_id in ClusterConf responses until noop commits: before the noop entry commits, the leader has not confirmed quorum readiness, so exposing its id too early causes clients to route to it before it is ready to serve.
…via BatchRemove Previously handle_zombie_node called execute_request_immediately(BatchRemove) when a node exceeded the connection failure threshold. This was too aggressive: a node that is temporarily restarting would be permanently ejected from the cluster. Membership changes are high-risk Raft consensus operations. The framework should detect and report, not decide. Now handle_zombie_node only emits warn!(Zombie detected) with the node_id and status. Removal remains a deliberate operator or upper-layer decision. Peer failure/success telemetry channels (grpc_transport → health_monitor → zombie bridge) are preserved so the warning still fires reliably when the threshold is crossed.
|
Warning Rate limit exceeded
To keep reviews running without waiting, you can enable usage-based add-on for your organization. This allows additional reviews beyond the hourly cap. Account admins can enable it under billing. ⌛ How to resolve this issue?After the wait time has elapsed, a review can be triggered using the We recommend that you space out your commits to avoid hitting the rate limit. 🚦 How do rate limits work?CodeRabbit enforces hourly rate limits for each developer per organization. Our paid plans have higher rate limits than the trial, open-source and free plans. In all cases, we re-allow further reviews after a brief timeout. Please see our FAQ for further information. ℹ️ Review info⚙️ Run configurationConfiguration used: defaults Review profile: CHILL Plan: Pro Run ID: ⛔ Files ignored due to path filters (1)
📒 Files selected for processing (5)
📝 WalkthroughWalkthroughThis PR implements cluster membership streaming capabilities through a new ChangesCluster Membership Streaming and Zombie Detection Updates
Sequence Diagram(s)sequenceDiagram
participant Client as gRPC/Embedded Client
participant EmbeddedEngine as EmbeddedEngine /<br>GrpcService
participant Node as Node<br>(Raft)
participant RaftMembership as RaftMembership
participant MembershipNotifier as Membership<br>Watch Channel
Client->>EmbeddedEngine: watch_membership() /<br>GrpcClient::watch_membership()
activate EmbeddedEngine
EmbeddedEngine->>Node: membership_change_notifier()
activate Node
Node-->>EmbeddedEngine: watch::Receiver<MembershipSnapshot>
deactivate Node
EmbeddedEngine-->>Client: Receiver / Streaming<MembershipSnapshot>
deactivate EmbeddedEngine
activate Client
Note over Client: Await first snapshot
Client->>MembershipNotifier: borrow() / next()
activate MembershipNotifier
MembershipNotifier-->>Client: Initial MembershipSnapshot
deactivate MembershipNotifier
deactivate Client
Node->>RaftMembership: ConfChange committed
activate RaftMembership
RaftMembership->>MembershipNotifier: send_replace(snapshot)
activate MembershipNotifier
MembershipNotifier->>Client: Emit snapshot update
deactivate MembershipNotifier
deactivate RaftMembership
Client->>Client: changed() fires, read updated snapshot
Estimated code review effort🎯 4 (Complex) | ⏱️ ~45 minutes Possibly related issues
Possibly related PRs
Poem
🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing Touches🧪 Generate unit tests (beta)
Tip 💬 Introducing Slack Agent: The best way for teams to turn conversations into code.Slack Agent is built on CodeRabbit's deep understanding of your code, so your team can collaborate across the entire SDLC without losing context.
Built for teams:
One agent for your entire SDLC. Right inside Slack. 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: 5
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
d-engine-server/tests/watch_and_subscriptions/mod.rs (1)
6-9:⚠️ Potential issue | 🟡 Minor | ⚡ Quick winModule doc comment is stale — update to list the two new submodules.
watch_membership_embeddedandwatch_membership_standaloneare declared but omitted from the## Test Coveragesection.📝 Proposed fix
//! ## Test Coverage //! //! - `watch_events_embedded.rs` - In-process key change subscriptions (embedded mode) //! - `watch_events_grpc_standalone.rs` - gRPC streaming watch events (standalone mode) //! - `watch_performance_gate_embedded.rs` - Watch latency and throughput benchmarks (embedded mode) +//! - `watch_membership_embedded.rs` - Committed membership change notifications (embedded mode) +//! - `watch_membership_standalone.rs` - gRPC membership snapshot streaming (standalone mode)🤖 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 `@d-engine-server/tests/watch_and_subscriptions/mod.rs` around lines 6 - 9, Update the module doc comment at the top of tests/watch_and_subscriptions/mod.rs to reflect the current submodules: replace the stale entries (e.g., watch_events_embedded.rs, watch_events_grpc_standalone.rs, watch_performance_gate_embedded.rs) with the actual submodules now declared (include watch_membership_embedded and watch_membership_standalone) and add both names to the "## Test Coverage" section so those two new submodules are listed; ensure the submodule list and coverage section match the declared mod statements (watch_membership_embedded, watch_membership_standalone).
🧹 Nitpick comments (7)
d-engine-server/tests/common/mod.rs (2)
595-614: ⚡ Quick win
wait_for_stable_leadercan loop indefinitely — add an explicit deadline.
client.refresh(None).await.ok()silently discards all errors. If the cluster is permanently unreachable (or a test bug prevents recovery),refreshkeeps failing silently,get()keeps returningConnectionTimeout, and the function sleeps 100 ms then loops — forever. There is no escape hatch.Contrast this with the Phase 5 loop in
leader_failover_cas_standalone.rswhich usesrefresh(None).await?(propagates errors) and is therefore bounded by the 30 scluster_ready_timeout.wait_for_stable_leaderhas no equivalent bound.A CI job will eventually time it out, but without a clear test-failure message, making regressions hard to diagnose.
⏱️ Proposed fix: wrap in an explicit timeout
-pub async fn wait_for_stable_leader(client: &Client) -> Result<(), ClientApiError> { - loop { - client.refresh(None).await.ok(); - match client.get(b"__stability_probe__").await { - Ok(_) => return Ok(()), - Err(ClientApiError::Business { - code: ErrorCode::StaleOperation, - .. - }) => continue, - Err(ClientApiError::Network { - code: ErrorCode::ConnectionTimeout, - .. - }) => { - tokio::time::sleep(Duration::from_millis(100)).await; - continue; - } - Err(e) => return Err(e), - } - } -} +pub async fn wait_for_stable_leader(client: &Client) -> Result<(), ClientApiError> { + const STABILITY_TIMEOUT: Duration = Duration::from_secs(30); + tokio::time::timeout(STABILITY_TIMEOUT, async { + loop { + client.refresh(None).await.ok(); + match client.get(b"__stability_probe__").await { + Ok(_) => return Ok(()), + Err(ClientApiError::Business { + code: ErrorCode::StaleOperation, + .. + }) => continue, + Err(ClientApiError::Network { + code: ErrorCode::ConnectionTimeout, + .. + }) => { + tokio::time::sleep(Duration::from_millis(100)).await; + continue; + } + Err(e) => return Err(e), + } + } + }) + .await + .map_err(|_elapsed| { + std::io::Error::new( + std::io::ErrorKind::TimedOut, + "wait_for_stable_leader: no stable leader within 30s", + ) + .into() + })? +}🤖 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 `@d-engine-server/tests/common/mod.rs` around lines 595 - 614, The wait_for_stable_leader function currently swallows refresh errors via client.refresh(None).await.ok() and can loop forever on repeated ConnectionTimeouts; change it to enforce an explicit deadline (e.g., use tokio::time::timeout or track Instant::now() + duration) and propagate refresh errors instead of ignoring them (replace the .ok() with ?.await or check the Result), returning a clear timeout/err result if the deadline is exceeded; update references in wait_for_stable_leader and the loop that matches client.get(...) so the function returns an Err with a descriptive error when the overall timeout is reached.
600-611: ⚡ Quick winAdd handling for
NotLeaderandLeaderChangederrors to the retry loop.The retry set currently covers only
StaleOperation(4002) andConnectionTimeout(1001), but linearizable reads on non-leader nodes returnNotLeader(4001, Business layer) and leadership changes returnLeaderChanged(1003, Network layer). Both errors will escape viaErr(e) => return Err(e)and fail callers during failover settling, which contradicts the function's purpose.Add match arms to retry on:
ClientApiError::Business { code: ErrorCode::NotLeader, .. }ClientApiError::Network { code: ErrorCode::LeaderChanged, .. }with similar backoff (e.g., 100ms sleep) as the
ConnectionTimeoutcase.🤖 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 `@d-engine-server/tests/common/mod.rs` around lines 600 - 611, The retry loop currently matches only StaleOperation and ConnectionTimeout; add two new match arms to also handle ClientApiError::Business { code: ErrorCode::NotLeader, .. } and ClientApiError::Network { code: ErrorCode::LeaderChanged, .. } so they behave like the retry cases: await tokio::time::sleep(Duration::from_millis(100)).await and then continue the loop rather than returning Err(e). Update the match in the function containing the shown match on Err(ClientApiError::...) to include these two arms so linearizable reads and leader changes are retried with the same backoff.d-engine-server/src/membership/membership_snapshot.rs (2)
3-7: 💤 Low valueOptional: rustdoc intra-doc link for
EmbeddedEngine::watch_membership.The bracketed reference
[EmbeddedEngine::watch_membership]is unlikely to resolve from within themembershipmodule sinceEmbeddedEnginelives incrate::api. With#![warn(missing_docs)]enabled at the crate level you may want to also enablebroken_intra_doc_links(or use an explicit path) so this link doesn't silently render as plain text.Suggested fix
-/// Delivered via [`EmbeddedEngine::watch_membership`] whenever a `ConfChange` +/// Delivered via [`crate::api::EmbeddedEngine::watch_membership`] whenever a `ConfChange`🤖 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 `@d-engine-server/src/membership/membership_snapshot.rs` around lines 3 - 7, Doc link to EmbeddedEngine::watch_membership will not resolve from the membership module; update the intra-doc link in membership_snapshot.rs to use the full path (e.g. `crate::api::EmbeddedEngine::watch_membership`) or enable broken intra-doc links at the crate root (add `#![warn(broken_intra_doc_links)]` or `#![allow(broken_intra_doc_links)]` as appropriate) so the bracketed reference resolves correctly; locate the bracketed link in the membership snapshot doc comment and replace it with the explicit path or add the crate-level attribute to fix the link.
35-46: 💤 Low valueConsider documenting the
Defaultsemantics forcommitted_index.
MembershipSnapshot::default()yieldscommitted_index = 0, which is also the value used to seed the watch channel inmock_node_builder.rs(Lines 321, 370). The doc comment states the index is "strictly monotonically increasing", but a realConfChangecould in principle commit at index0(or, more realistically, the first published snapshot may share0with the seed). Subscribers usingcommitted_index <= last_appliedas an idempotency guard need to know whether0is a valid sentinel "no snapshot yet" value or a legitimate snapshot index. A short note in the rustdoc would prevent confusion.🤖 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 `@d-engine-server/src/membership/membership_snapshot.rs` around lines 35 - 46, Update the rustdoc for MembershipSnapshot::committed_index to state the Default semantics clearly: indicate that MembershipSnapshot::default() sets committed_index = 0 and that 0 is used as the sentinel "no snapshot yet" value (and matches the seed used by the watch channel), or alternatively document that 0 can be a valid index and recommend a specific idempotency check (e.g., use strict < or > comparisons rather than <=). Modify the comment on the committed_index field in the MembershipSnapshot struct to explicitly describe which convention is followed so subscribers using committed_index <= last_applied know how to interpret 0.d-engine-core/src/raft_role/leader_state.rs (2)
1966-1990: 💤 Low valueReconnect loop guard looks correct; consider also bailing on shutdown.
The new
task_rx.is_closed()check at the top of the reconnect loop correctly prevents the worker from spinning onopen_replication_streamafter the leader has stepped down (verified bytest_replication_worker_exits_when_handle_dropped). One small consideration: if the underlying transport keeps failing fast (Errimmediately) with a longmax_delay_ms, the worker now sleeps viatokio::time::sleepwithout a select on a shutdown signal —is_closed()is only re-checked once per backoff cycle. That's acceptable since dropping the handle still ensures eventual exit, but atokio::select!oversleepand a closed-channel future would shave off up tomax_delay_msof idle time on step-down. Fine to leave as-is given the existing test coverage.🤖 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 `@d-engine-core/src/raft_role/leader_state.rs` around lines 1966 - 1990, The reconnect loop currently sleeps with tokio::time::sleep between retries, which only re-checks task_rx.is_closed() once per backoff cycle; update the retry sleep to use tokio::select! so the worker can exit immediately on shutdown: replace the tokio::time::sleep(Duration::from_millis(backoff_ms)).await call with a tokio::select! that awaits either the sleep future or task_rx.closed() (or task_rx.is_closed()’s async closed notification) and break/return early if the channel is closed; keep the existing backoff_ms doubling logic and error logging around transport.open_replication_stream(peer_id, membership.clone(), response_compress_enabled).
3620-3646: ⚡ Quick winWarn-only zombie handling is intentional; consider emitting a metric for observability.
The shift from auto-
BatchRemoveto warn-only is well-justified by the comment (a node failing N attempts may simply be restarting). Since operators are now expected to act on these warnings, it would help to also emit a counter (similar tomembership.stale_learner_removedat Line 3607) so dashboards/alerts can surface persistent zombies without scraping logs:Suggested addition
warn!( node_id, ?status, "Zombie detected: node is persistently unreachable — manual intervention may be required" ); + metrics::counter!( + "membership.zombie_detected", + &[("node_id", node_id.to_string())] + ) + .increment(1); Ok(())Also note that
_role_txis now an unused parameter on thispub asyncmethod; since it's still required by theRaftRoleState::handle_zombie_detectedtrait method signature, keeping it (with the leading underscore) is the right call.🤖 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 `@d-engine-core/src/raft_role/leader_state.rs` around lines 3620 - 3646, Add an observability counter increment when a zombie is detected: inside pub async fn handle_zombie_node (the function shown) after confirming the node exists (i.e., where you currently call warn!), call the same membership metrics increment used for membership.stale_learner_removed to increment a new counter (e.g. "membership.zombie_detected" or similar) via the membership/metrics API (use ctx.membership().<metrics increment helper> or the same helper used by stale_learner_removed) and then emit the warn log; keep the unused _role_tx parameter as-is.d-engine-server/src/network/grpc/grpc_raft_service.rs (1)
528-558: 💤 Low valueRemove redundant
mark_changed()call and use the public accessor for consistency.The
tokio_stream::wrappers::WatchStream::new(in 0.1.16) yields the current value on its first poll automatically via an internalborrow_and_update()call. Therx.mark_changed()on line 539 is redundant; WatchStream handles the initial snapshot delivery without it. The comment incorrectly attributes this behavior tomark_changed().Additionally, per the pattern already used in
d-engine-server/src/api/embedded.rs, use the public accessorself.membership_change_notifier()instead of directly accessingself.membership_rx. This maintains consistency with the embedded API and provides a single chokepoint if the storage type ever changes.♻️ Suggested cleanup
- let mut rx = self.membership_rx.clone(); - // Deliver the current snapshot immediately; subsequent items arrive on each ConfChange. - rx.mark_changed(); + // WatchStream::new yields the current value on first poll, so subscribers always + // receive the snapshot at subscribe time, then one item per committed ConfChange. + let rx = self.membership_change_notifier();🤖 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 `@d-engine-server/src/network/grpc/grpc_raft_service.rs` around lines 528 - 558, Remove the redundant rx.mark_changed() call in watch_membership and switch to using the public accessor membership_change_notifier() instead of directly cloning self.membership_rx: locate the watch_membership method, remove the mark_changed() invocation (it’s unnecessary because tokio_stream::wrappers::WatchStream::new yields the current value on first poll), replace let mut rx = self.membership_rx.clone() with let mut rx = self.membership_change_notifier().clone() (or the exact accessor call used elsewhere), and keep the rest of the stream mapping and chain logic unchanged.
🤖 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 `@d-engine-core/src/raft_role/leader_state_test/worker_lifecycle_test.rs`:
- Around line 425-427: The timeout only unwraps the Result from
tokio::time::timeout but not the inner Option from first_attempt_rx.recv(), so
add a second expect (or assert_some) after the timeout to fail if recv()
returned None; specifically, change the await chain around
tokio::time::timeout(std::time::Duration::from_secs(5),
first_attempt_rx.recv()).await.expect("timed out waiting for first reconnect
attempt") to also unwrap the Option (e.g., .expect("first reconnect signal not
sent") on the value returned) so the test fails if the channel was closed
without sending.
In `@d-engine-proto/proto/client/client_api.proto`:
- Around line 179-181: The doc comment for the MembershipSnapshot/ConfChange
event is inconsistent: it mentions "Remove" but the code/tests use
"BatchRemove"; update the comment near MembershipSnapshot and ConfChange to use
the exact operation name "BatchRemove" (or list both as "Remove (BatchRemove)"
if you want backward clarity) so client-facing docs match the actual API
operation names such as AddNode, Promote, and BatchRemove.
In `@d-engine-server/src/membership/raft_membership.rs`:
- Around line 444-460: When AddLearner re-applies and finds an existing learner
(the if let Some(existing) guard.nodes branch), avoid silently returning Ok(())
if the incoming address or status differs; instead compare incoming
address/status with existing.address and existing.status and emit a warn!
(including self.node_id, node_id, existing.address, existing.status and the
incoming values) before the early return so operators see stale-address or
status mismatches; keep the early return to preserve idempotence but add that
warning in the AddLearner handling code path.
In `@d-engine-server/tests/watch_and_subscriptions/watch_membership_embedded.rs`:
- Line 471: The test doc comments incorrectly state Promotable is value 2 while
the code constant STATUS_PROMOTABLE: i32 = 1 (matching
d_engine_proto::common::NodeStatus::Promotable); update the doc comment in
watch_membership_embedded.rs (the block around the test that currently says
"Promotable (2)" and "status=2") to "Promotable (1)" and "status=1" (also apply
the same correction to the other comment block at the referenced second
occurrence around lines 535-537) so the comments match the STATUS_PROMOTABLE
constant and the proto enum.
- Around line 141-154: Update the comment above with_fast_zombie to remove the
stale reference to "tests 3 and 7" and the incorrect claim about node removal:
state that it is used only by test_watch_membership_zombie_warns_without_removal
(test 3), that test_watch_membership_committed_index_monotonically_increasing
(test 7) does not use this override and instead builds TOML via
make_toml/node_toml, and clarify that with_fast_zombie triggers immediate zombie
warnings only (no automatic BatchRemove or node removal) per current design.
---
Outside diff comments:
In `@d-engine-server/tests/watch_and_subscriptions/mod.rs`:
- Around line 6-9: Update the module doc comment at the top of
tests/watch_and_subscriptions/mod.rs to reflect the current submodules: replace
the stale entries (e.g., watch_events_embedded.rs,
watch_events_grpc_standalone.rs, watch_performance_gate_embedded.rs) with the
actual submodules now declared (include watch_membership_embedded and
watch_membership_standalone) and add both names to the "## Test Coverage"
section so those two new submodules are listed; ensure the submodule list and
coverage section match the declared mod statements (watch_membership_embedded,
watch_membership_standalone).
---
Nitpick comments:
In `@d-engine-core/src/raft_role/leader_state.rs`:
- Around line 1966-1990: The reconnect loop currently sleeps with
tokio::time::sleep between retries, which only re-checks task_rx.is_closed()
once per backoff cycle; update the retry sleep to use tokio::select! so the
worker can exit immediately on shutdown: replace the
tokio::time::sleep(Duration::from_millis(backoff_ms)).await call with a
tokio::select! that awaits either the sleep future or task_rx.closed() (or
task_rx.is_closed()’s async closed notification) and break/return early if the
channel is closed; keep the existing backoff_ms doubling logic and error logging
around transport.open_replication_stream(peer_id, membership.clone(),
response_compress_enabled).
- Around line 3620-3646: Add an observability counter increment when a zombie is
detected: inside pub async fn handle_zombie_node (the function shown) after
confirming the node exists (i.e., where you currently call warn!), call the same
membership metrics increment used for membership.stale_learner_removed to
increment a new counter (e.g. "membership.zombie_detected" or similar) via the
membership/metrics API (use ctx.membership().<metrics increment helper> or the
same helper used by stale_learner_removed) and then emit the warn log; keep the
unused _role_tx parameter as-is.
In `@d-engine-server/src/membership/membership_snapshot.rs`:
- Around line 3-7: Doc link to EmbeddedEngine::watch_membership will not resolve
from the membership module; update the intra-doc link in membership_snapshot.rs
to use the full path (e.g. `crate::api::EmbeddedEngine::watch_membership`) or
enable broken intra-doc links at the crate root (add
`#![warn(broken_intra_doc_links)]` or `#![allow(broken_intra_doc_links)]` as
appropriate) so the bracketed reference resolves correctly; locate the bracketed
link in the membership snapshot doc comment and replace it with the explicit
path or add the crate-level attribute to fix the link.
- Around line 35-46: Update the rustdoc for MembershipSnapshot::committed_index
to state the Default semantics clearly: indicate that
MembershipSnapshot::default() sets committed_index = 0 and that 0 is used as the
sentinel "no snapshot yet" value (and matches the seed used by the watch
channel), or alternatively document that 0 can be a valid index and recommend a
specific idempotency check (e.g., use strict < or > comparisons rather than <=).
Modify the comment on the committed_index field in the MembershipSnapshot struct
to explicitly describe which convention is followed so subscribers using
committed_index <= last_applied know how to interpret 0.
In `@d-engine-server/src/network/grpc/grpc_raft_service.rs`:
- Around line 528-558: Remove the redundant rx.mark_changed() call in
watch_membership and switch to using the public accessor
membership_change_notifier() instead of directly cloning self.membership_rx:
locate the watch_membership method, remove the mark_changed() invocation (it’s
unnecessary because tokio_stream::wrappers::WatchStream::new yields the current
value on first poll), replace let mut rx = self.membership_rx.clone() with let
mut rx = self.membership_change_notifier().clone() (or the exact accessor call
used elsewhere), and keep the rest of the stream mapping and chain logic
unchanged.
In `@d-engine-server/tests/common/mod.rs`:
- Around line 595-614: The wait_for_stable_leader function currently swallows
refresh errors via client.refresh(None).await.ok() and can loop forever on
repeated ConnectionTimeouts; change it to enforce an explicit deadline (e.g.,
use tokio::time::timeout or track Instant::now() + duration) and propagate
refresh errors instead of ignoring them (replace the .ok() with ?.await or check
the Result), returning a clear timeout/err result if the deadline is exceeded;
update references in wait_for_stable_leader and the loop that matches
client.get(...) so the function returns an Err with a descriptive error when the
overall timeout is reached.
- Around line 600-611: The retry loop currently matches only StaleOperation and
ConnectionTimeout; add two new match arms to also handle
ClientApiError::Business { code: ErrorCode::NotLeader, .. } and
ClientApiError::Network { code: ErrorCode::LeaderChanged, .. } so they behave
like the retry cases: await tokio::time::sleep(Duration::from_millis(100)).await
and then continue the loop rather than returning Err(e). Update the match in the
function containing the shown match on Err(ClientApiError::...) to include these
two arms so linearizable reads and leader changes are retried with the same
backoff.
🪄 Autofix (Beta)
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: defaults
Review profile: CHILL
Plan: Pro
Run ID: 08a4b1c7-8c9b-4550-bc20-30d991d1afe9
⛔ Files ignored due to path filters (3)
d-engine-proto/src/generated/d_engine.client.rsis excluded by!**/generated/**examples/single-node-expansion/Cargo.lockis excluded by!**/*.lockexamples/three-nodes-embedded/Cargo.lockis excluded by!**/*.lock
📒 Files selected for processing (33)
CHANGELOG.mdd-engine-client/src/grpc_client.rsd-engine-client/src/grpc_client_test.rsd-engine-client/src/mock_rpc.rsd-engine-client/src/mock_rpc_service.rsd-engine-core/src/raft_role/follower_state_test.rsd-engine-core/src/raft_role/leader_state.rsd-engine-core/src/raft_role/leader_state_test/event_handling_test.rsd-engine-core/src/raft_role/leader_state_test/membership_change_test.rsd-engine-core/src/raft_role/leader_state_test/worker_lifecycle_test.rsd-engine-core/src/test_utils/mock/mock_rpc.rsd-engine-proto/proto/client/client_api.protod-engine-server/src/api/embedded.rsd-engine-server/src/lib.rsd-engine-server/src/membership/membership_snapshot.rsd-engine-server/src/membership/mod.rsd-engine-server/src/membership/raft_membership.rsd-engine-server/src/membership/raft_membership_test.rsd-engine-server/src/network/grpc/grpc_raft_service.rsd-engine-server/src/network/grpc/grpc_raft_service_test.rsd-engine-server/src/network/grpc/grpc_transport.rsd-engine-server/src/node/builder.rsd-engine-server/src/node/mod.rsd-engine-server/src/test_utils/mock/mock_node_builder.rsd-engine-server/src/test_utils/mock/mock_rpc.rsd-engine-server/tests/cas_operations/leader_failover_cas_standalone.rsd-engine-server/tests/cas_operations/snapshot_recovery_standalone.rsd-engine-server/tests/cluster_lifecycle/scale_single_to_three_node_embedded.rsd-engine-server/tests/common/mod.rsd-engine-server/tests/failover_and_recovery/leader_failover_standalone.rsd-engine-server/tests/watch_and_subscriptions/mod.rsd-engine-server/tests/watch_and_subscriptions/watch_membership_embedded.rsd-engine-server/tests/watch_and_subscriptions/watch_membership_standalone.rs
Contributor
Author
|
@coderabbitai review |
✅ Actions performedReview triggered.
|
Codecov Report❌ Patch coverage is 📢 Thoughts on this report? Let us know! |
10 of 17 tasks
9 tasks done
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 Does This PR Do?
Adds real-time cluster membership change notifications for both embedded and standalone (gRPC)
modes. Callers subscribe once and receive a snapshot on every committed ConfChange — no polling.
Type:
Why Is This Needed?
For features: Issues #327 and #328
Operators and upper-layer tooling (service discovery, load balancers, health dashboards) need
to react to cluster topology changes (node join, promotion, removal) without polling
get_cluster_metadata. A push-based membership stream enables zero-latency reactions andeliminates unnecessary RPC traffic.
Checklist
Required:
make testpassesIf changing APIs:
Testing
How tested:
Unit tests (server):
test_watch_membership_returns_unavailable_when_node_not_ready,test_watch_membership_yields_current_snapshot_then_sentinel_on_sender_drop— verify the gRPC handler's not-ready guard and the
mark_changed()+ UNAVAILABLE sentinel behavior.Unit tests (client):
test_watch_membership_returns_err_when_server_rejects,test_watch_membership_receives_snapshots_in_order,test_watch_membership_empty_stream_closes_cleanly— verify
GrpcClient::watch_membership()against a mock gRPC server.Integration tests (embedded): 7 tests in
watch_membership_embedded.rscoveringinitial snapshot delivery, node join/promotion, zombie no-op, multi-subscriber fanout,
and
committed_indexmonotonicity.Integration test (standalone/gRPC):
watch_membership_standalone.rs— boots a real2-node cluster, opens a gRPC stream, joins a 3rd learner node, asserts the stream yields
a snapshot with
learners=[3]andcommitted_index > 0.All 13
watch_membershiptests pass:cargo nextest run --all-features -E 'test(watch_membership)'Does This Follow d-engine's Principles?
Reviewer Notes
4 commits, 3 concerns to focus on:
mark_changed()beforeWatchStream::new()(grpc_raft_service.rs): This forces thefirst stream item to be the current snapshot, matching
watch::Receiver::borrow()semantics.Requires tokio ≥ 1.37 (project uses 1.51.1). Alternative would be an extra initial
send()which adds complexity;
mark_changed()is idiomatic.UNAVAILABLE sentinel (
grpc_raft_service.rs): The stream is chained with afutures::stream::once(Err(Status::unavailable(...)))so clients get a clean signal insteadof a silent stream close. Clients should reconnect and resubscribe on receiving this error.
GrpcClient::watch_membership()is NOT inClientApitrait: Membership watch isstandalone-only (the embedded equivalent uses
EmbeddedEngine::watch_membership()whichreturns a
watch::Receiver). Putting both under one trait would force type-system gymnastics;keeping them separate preserves simplicity.
Estimated review complexity:
Summary by CodeRabbit
New Features
Important Fixes