Skip to content

Commit 9600257

Browse files
farhan-syahfarhan-syah
committed
refactor(nodedb): replace ClusterVersionState with topology-derived ClusterVersionView
Split rolling_upgrade.rs into a rolling_upgrade/ submodule (mod.rs, versions.rs, view.rs). Replace the mutable ClusterVersionState — which maintained a shadow map of node versions under a Mutex — with ClusterVersionView, computed on demand from the live topology snapshot. Stamp wire_version on NodeInfo in nodedb-cluster so every node carries its version in the topology. Add CLUSTER_WIRE_FORMAT_VERSION constant to topology.rs as the single source of truth. Propagate wire_version through JoinRequest and JoinNodeInfo wire types so peers learn each other's version in one RPC round-trip. Remove cluster_version_state field from SharedState and its Mutex initialisation from both single-node and cluster init paths. Update all call sites (metadata_proposer, drain_propose, health check, metadata applier, start_raft, handle_rpc, bootstrap) to call shared.cluster_version_view() instead of locking the old field.
1 parent 9a426d4 commit 9600257

18 files changed

Lines changed: 455 additions & 351 deletions

File tree

‎nodedb-cluster/src/bootstrap/handle_join.rs‎

Lines changed: 15 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -68,8 +68,14 @@ pub fn handle_join_request(
6868
return build_response(topology, routing, cluster_id);
6969
}
7070

71-
// Brand new node — admit as Active.
72-
topology.add_node(NodeInfo::new(req.node_id, addr, NodeState::Active));
71+
// Brand new node — admit as Active. Stamp the joiner's own
72+
// wire version onto its NodeInfo so every peer that replays
73+
// this topology sees the correct version in
74+
// `cluster_version_view` without round-tripping through the
75+
// old in-memory shadow map.
76+
topology.add_node(
77+
NodeInfo::new(req.node_id, addr, NodeState::Active).with_wire_version(req.wire_version),
78+
);
7379
build_response(topology, routing, cluster_id)
7480
}
7581

@@ -86,6 +92,7 @@ fn build_response(
8692
addr: n.addr.clone(),
8793
state: n.state.as_u8(),
8894
raft_groups: n.raft_groups.clone(),
95+
wire_version: n.wire_version,
8996
})
9097
.collect();
9198

@@ -144,6 +151,7 @@ mod tests {
144151
let req = JoinRequest {
145152
node_id: 2,
146153
listen_addr: "10.0.0.2:9400".into(),
154+
wire_version: crate::topology::CLUSTER_WIRE_FORMAT_VERSION,
147155
};
148156

149157
let resp = handle_join_request(&req, &mut topology, &routing, 42);
@@ -165,6 +173,7 @@ mod tests {
165173
let req = JoinRequest {
166174
node_id: 2,
167175
listen_addr: "10.0.0.2:9400".into(),
176+
wire_version: crate::topology::CLUSTER_WIRE_FORMAT_VERSION,
168177
};
169178

170179
let _ = handle_join_request(&req, &mut topology, &routing, 42);
@@ -186,6 +195,7 @@ mod tests {
186195
let req = JoinRequest {
187196
node_id: 2,
188197
listen_addr: "10.0.0.2:9400".into(),
198+
wire_version: crate::topology::CLUSTER_WIRE_FORMAT_VERSION,
189199
};
190200

191201
let resp1 = handle_join_request(&req, &mut topology, &routing, 7);
@@ -216,6 +226,7 @@ mod tests {
216226
let req1 = JoinRequest {
217227
node_id: 2,
218228
listen_addr: "10.0.0.2:9400".into(),
229+
wire_version: crate::topology::CLUSTER_WIRE_FORMAT_VERSION,
219230
};
220231
let resp1 = handle_join_request(&req1, &mut topology, &routing, 11);
221232
assert!(resp1.success);
@@ -224,6 +235,7 @@ mod tests {
224235
let req2 = JoinRequest {
225236
node_id: 2,
226237
listen_addr: "10.0.0.99:9400".into(),
238+
wire_version: crate::topology::CLUSTER_WIRE_FORMAT_VERSION,
227239
};
228240
let resp2 = handle_join_request(&req2, &mut topology, &routing, 11);
229241

@@ -247,6 +259,7 @@ mod tests {
247259
let req = JoinRequest {
248260
node_id: 2,
249261
listen_addr: "not-a-valid-address".into(),
262+
wire_version: crate::topology::CLUSTER_WIRE_FORMAT_VERSION,
250263
};
251264

252265
let resp = handle_join_request(&req, &mut topology, &routing, 42);

‎nodedb-cluster/src/bootstrap/join.rs‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -120,6 +120,7 @@ pub(super) async fn join(
120120
let req_template = JoinRequest {
121121
node_id: config.node_id,
122122
listen_addr: config.listen_addr.to_string(),
123+
wire_version: crate::topology::CLUSTER_WIRE_FORMAT_VERSION,
123124
};
124125

125126
let mut last_err: Option<ClusterError> = None;
@@ -260,6 +261,7 @@ fn apply_join_response(
260261
addr: node.addr.clone(),
261262
state,
262263
raft_groups: node.raft_groups.clone(),
264+
wire_version: node.wire_version,
263265
};
264266
if node.node_id == config.node_id {
265267
info.state = NodeState::Active;

‎nodedb-cluster/src/health.rs‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -240,6 +240,7 @@ pub fn broadcast_topology(
240240
addr: n.addr.clone(),
241241
state: n.state.as_u8(),
242242
raft_groups: n.raft_groups.clone(),
243+
wire_version: n.wire_version,
243244
})
244245
.collect(),
245246
});
@@ -292,6 +293,7 @@ pub fn handle_topology_update(
292293
addr: node.addr.clone(),
293294
state,
294295
raft_groups: node.raft_groups.clone(),
296+
wire_version: node.wire_version,
295297
};
296298
new_topo.add_node(info);
297299
}
@@ -342,12 +344,14 @@ mod tests {
342344
addr: "10.0.0.1:9400".into(),
343345
state: 1,
344346
raft_groups: vec![],
347+
wire_version: crate::topology::CLUSTER_WIRE_FORMAT_VERSION,
345348
},
346349
JoinNodeInfo {
347350
node_id: 2,
348351
addr: "10.0.0.2:9400".into(),
349352
state: 1,
350353
raft_groups: vec![],
354+
wire_version: crate::topology::CLUSTER_WIRE_FORMAT_VERSION,
351355
},
352356
],
353357
};

‎nodedb-cluster/src/raft_loop/handle_rpc.rs‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -297,6 +297,7 @@ mod tests {
297297
let req = RaftRpc::JoinRequest(crate::rpc_codec::JoinRequest {
298298
node_id: 2,
299299
listen_addr: "127.0.0.1:9401".into(),
300+
wire_version: crate::topology::CLUSTER_WIRE_FORMAT_VERSION,
300301
});
301302

302303
let resp = raft_loop.handle_rpc(req).await.unwrap();

‎nodedb-cluster/src/rpc_codec.rs‎

Lines changed: 13 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -168,6 +168,12 @@ pub struct JoinRequest {
168168
pub node_id: u64,
169169
/// Listen address for Raft RPCs (e.g. "10.0.0.5:9400").
170170
pub listen_addr: String,
171+
/// Wire format version the joiner is running. The leader
172+
/// stamps this onto the joiner's `NodeInfo` so every peer
173+
/// sees the correct version in the topology snapshot they
174+
/// receive back. See
175+
/// `topology::CLUSTER_WIRE_FORMAT_VERSION`.
176+
pub wire_version: u16,
171177
}
172178

173179
/// Wire-level redirect contract between the join-flow producer
@@ -209,6 +215,10 @@ pub struct JoinNodeInfo {
209215
/// NodeState as u8 (0=Joining, 1=Active, 2=Draining, 3=Decommissioned).
210216
pub state: u8,
211217
pub raft_groups: Vec<u64>,
218+
/// Mirror of `NodeInfo::wire_version` so joiners learn the
219+
/// version of every peer in one RPC round-trip and never
220+
/// silently fall back to the minimum-supported default.
221+
pub wire_version: u16,
212222
}
213223

214224
/// Raft group membership in the join response wire format.
@@ -880,6 +890,7 @@ mod tests {
880890
let req = JoinRequest {
881891
node_id: 42,
882892
listen_addr: "10.0.0.5:9400".into(),
893+
wire_version: crate::topology::CLUSTER_WIRE_FORMAT_VERSION,
883894
};
884895

885896
let rpc = RaftRpc::JoinRequest(req);
@@ -907,12 +918,14 @@ mod tests {
907918
addr: "10.0.0.1:9400".into(),
908919
state: 1,
909920
raft_groups: vec![0, 1],
921+
wire_version: crate::topology::CLUSTER_WIRE_FORMAT_VERSION,
910922
},
911923
JoinNodeInfo {
912924
node_id: 2,
913925
addr: "10.0.0.2:9400".into(),
914926
state: 1,
915927
raft_groups: vec![0, 1],
928+
wire_version: crate::topology::CLUSTER_WIRE_FORMAT_VERSION,
916929
},
917930
],
918931
vshard_to_group: (0..1024u64).map(|i| i % 4).collect(),

‎nodedb-cluster/src/topology.rs‎

Lines changed: 40 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -3,6 +3,26 @@
33
use std::collections::HashMap;
44
use std::net::SocketAddr;
55

6+
/// Wire format version carried on every `NodeInfo`. This is the
7+
/// single source of truth for "what version is this node running"
8+
/// once the node has registered itself with the cluster — the
9+
/// in-memory shadow map used by the old `ClusterVersionState` is
10+
/// gone. See `control::rolling_upgrade::view::ClusterVersionView`
11+
/// for the consumer.
12+
///
13+
/// Must match `nodedb::version::WIRE_FORMAT_VERSION`. The cluster
14+
/// crate defines the constant so the wire format lives next to
15+
/// the types it stamps.
16+
pub const CLUSTER_WIRE_FORMAT_VERSION: u16 = 4;
17+
18+
fn default_wire_version() -> u16 {
19+
// Records persisted by an older build that did not carry a
20+
// `wire_version` field default to `1` — the minimum supported
21+
// wire version — so rolling-upgrade feature gates treat them
22+
// as N-1 until the node re-registers with its real version.
23+
1
24+
}
25+
626
/// Node lifecycle states.
727
#[derive(
828
Debug,
@@ -82,18 +102,38 @@ pub struct NodeInfo {
82102
pub state: NodeState,
83103
/// Raft groups hosted on this node.
84104
pub raft_groups: Vec<u64>,
105+
/// Wire format version this node is running. Stamped by the
106+
/// node itself on self-registration (bootstrap / join) and by
107+
/// `handle_join` when learning about a remote joiner. Read by
108+
/// `control::rolling_upgrade::view::compute` to derive the
109+
/// cluster-wide min/max/mixed view on demand from the live
110+
/// topology.
111+
#[serde(default = "default_wire_version")]
112+
pub wire_version: u16,
85113
}
86114

87115
impl NodeInfo {
116+
/// Construct a NodeInfo stamped with this build's wire version.
117+
/// Use [`NodeInfo::with_wire_version`] when stamping a remote
118+
/// node whose version arrived over the wire.
88119
pub fn new(node_id: u64, addr: SocketAddr, state: NodeState) -> Self {
89120
Self {
90121
node_id,
91122
addr: addr.to_string(),
92123
state,
93124
raft_groups: Vec::new(),
125+
wire_version: CLUSTER_WIRE_FORMAT_VERSION,
94126
}
95127
}
96128

129+
/// Override the wire version on an otherwise-fresh NodeInfo.
130+
/// Builder-style so call sites read as
131+
/// `NodeInfo::new(id, addr, state).with_wire_version(remote_v)`.
132+
pub fn with_wire_version(mut self, wire_version: u16) -> Self {
133+
self.wire_version = wire_version;
134+
self
135+
}
136+
97137
pub fn socket_addr(&self) -> Option<SocketAddr> {
98138
self.addr.parse().ok()
99139
}

‎nodedb/src/control/cluster/metadata_applier.rs‎

Lines changed: 6 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -145,20 +145,15 @@ impl MetadataCommitApplier {
145145
// treat it as "unknown, always re-fetch". Only the proposing
146146
// node's apply path observes the gate; followers run the
147147
// same applier and will reach the same conclusion because
148-
// every node observes the same `cluster_version_state`
149-
// (replicated via the gossip path).
148+
// every node observes the same live topology (same
149+
// `wire_version` on every NodeInfo, replicated via the
150+
// gossip path).
150151
let stamped = if let Some(weak) = self.shared.get()
151152
&& let Some(shared) = weak.upgrade()
152153
{
153-
let compat = {
154-
let vs = shared
155-
.cluster_version_state
156-
.lock()
157-
.unwrap_or_else(|p| p.into_inner());
158-
!vs.can_activate_feature(
159-
crate::control::rolling_upgrade::DESCRIPTOR_VERSIONING_VERSION,
160-
)
161-
};
154+
let compat = !shared.cluster_version_view().can_activate_feature(
155+
crate::control::rolling_upgrade::DESCRIPTOR_VERSIONING_VERSION,
156+
);
162157
if compat {
163158
catalog_entry
164159
} else {

‎nodedb/src/control/cluster/start_raft.rs‎

Lines changed: 12 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -120,28 +120,22 @@ pub fn start_raft(
120120
}
121121
});
122122

123-
// Register this node's wire version in the rolling upgrade tracker.
123+
// Wire version of every node is now carried on the live
124+
// `NodeInfo` in `cluster_topology`, stamped by the joiner on
125+
// join_request and the self-register path on bootstrap — no
126+
// shadow map to populate here. Log the derived view for
127+
// observability.
124128
{
125-
let mut vs = shared
126-
.cluster_version_state
127-
.lock()
128-
.unwrap_or_else(|p| p.into_inner());
129-
vs.report_version(handle.node_id, crate::version::WIRE_FORMAT_VERSION);
130-
131-
if let Ok(topo) = handle.topology.read() {
132-
for node in topo.active_nodes() {
133-
if node.node_id != handle.node_id {
134-
vs.report_version(node.node_id, crate::version::WIRE_FORMAT_VERSION);
135-
}
136-
}
137-
}
138-
let compat = crate::control::rolling_upgrade::should_compat_mode(&vs);
129+
let view = shared.cluster_version_view();
130+
let compat = crate::control::rolling_upgrade::should_compat_mode(&view);
139131
info!(
140132
node_id = handle.node_id,
141-
nodes = vs.node_count,
142-
mixed = vs.is_mixed_version(),
133+
nodes = view.node_count,
134+
min_version = view.min_version,
135+
max_version = view.max_version,
136+
mixed = view.is_mixed_version(),
143137
compat_mode = compat,
144-
"cluster version state initialized"
138+
"cluster version view derived from topology"
145139
);
146140
}
147141

‎nodedb/src/control/lease/drain_propose.rs‎

Lines changed: 2 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -26,7 +26,7 @@
2626
//! The `MetadataEntry::DescriptorDrainStart` / `End` variants are
2727
//! wire-format v4. Mixed clusters running v3 binaries can't decode
2828
//! them, so the proposer gates on
29-
//! `cluster_version_state.can_activate_feature(DESCRIPTOR_DRAIN_VERSION)`
29+
//! `cluster_version_view().can_activate_feature(DESCRIPTOR_DRAIN_VERSION)`
3030
//! and returns `Ok(())` immediately in compat mode — the same
3131
//! "degrade to no drain" fallback catalog DDL uses. Mixed clusters
3232
//! behave without drain safety until all nodes are upgraded.
@@ -67,10 +67,7 @@ pub fn drain_for_ddl(
6767
) -> Result<(), Error> {
6868
// Rolling upgrade gate: no drain in mixed-version clusters.
6969
{
70-
let vs = shared
71-
.cluster_version_state
72-
.lock()
73-
.unwrap_or_else(|p| p.into_inner());
70+
let vs = shared.cluster_version_view();
7471
if !vs.can_activate_feature(DESCRIPTOR_DRAIN_VERSION) {
7572
tracing::warn!(
7673
min_version = vs.min_version,

‎nodedb/src/control/metadata_proposer.rs‎

Lines changed: 1 addition & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -142,10 +142,7 @@ pub fn propose_catalog_entry_with_timeout(
142142
// diverge catalog state across nodes — see
143143
// `control/rolling_upgrade.rs`.
144144
{
145-
let vs = shared
146-
.cluster_version_state
147-
.lock()
148-
.unwrap_or_else(|p| p.into_inner());
145+
let vs = shared.cluster_version_view();
149146
if !vs.can_activate_feature(crate::control::rolling_upgrade::DISTRIBUTED_CATALOG_VERSION) {
150147
tracing::warn!(
151148
min_version = vs.min_version,

0 commit comments

Comments
 (0)