1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
// SPDX-License-Identifier: BUSL-1.1
//! Phase 4 of `start_raft`: install the sync/async Raft proposer, compactor,
//! and durable applied-index closures onto `SharedState`, and spawn the
//! background apply loop that drains `DistributedApplier::apply_committed`
//! into the Data Plane and notifies propose waiters.
use std::collections::BTreeMap;
use std::sync::{Arc, Mutex};
use tokio::sync::mpsc::{self, Sender};
use crate::control::cluster::calvin::ReadResultEvent;
use crate::control::distributed_applier::{ApplyBatch, ProposeTracker, run_apply_loop};
use crate::control::state::SharedState;
use super::loop_build::RaftLoopType;
/// Install the sync `raft_proposer` / `raft_compactor` /
/// `raft_applied_index_sink`, the async `async_raft_proposer`, and spawn the
/// apply loop.
pub(super) fn wire_proposers(
shared: &Arc<SharedState>,
raft_loop: &Arc<RaftLoopType>,
tracker: Arc<ProposeTracker>,
apply_rx: mpsc::Receiver<ApplyBatch>,
calvin_read_result_senders: Arc<Mutex<BTreeMap<u32, Sender<ReadResultEvent>>>>,
sequencer_state_machine: Arc<Mutex<nodedb_cluster::calvin::SequencerStateMachine>>,
) {
// Wire the Raft proposer into SharedState so CP dispatch paths
// (pgwire, HTTP, array inbound) can route writes through Raft.
// Hold `raft_loop` weakly: `SharedState` owns this closure, and the
// closure must NOT keep `raft_loop` alive or the two form a strong
// reference cycle that pins `SharedState` forever. During normal
// operation the loop's spawned tasks keep it alive so `upgrade`
// always succeeds; `None` only occurs once those tasks have stopped
// on shutdown, where a clean "cluster not running" error is correct.
let raft_loop_for_propose = Arc::downgrade(raft_loop);
let proposer: Arc<crate::control::wal_replication::RaftProposer> =
Arc::new(move |vshard_id, data| {
let rl = raft_loop_for_propose
.upgrade()
.ok_or_else(|| crate::Error::Internal {
detail: "raft propose: cluster not running".into(),
})?;
rl.propose(vshard_id, data)
.map_err(|e| crate::Error::Internal {
detail: format!("raft propose: {e}"),
})
});
if shared.raft_proposer.set(proposer).is_err() {
tracing::warn!("raft_proposer already set — start_raft appears to have run twice");
}
// Wire the Raft log-compaction trigger. `run_apply_loop` invokes this
// after a committed entry has been durably applied to the Data Plane,
// so compaction is gated on the data-plane applied watermark — never
// raft's commit index. A no-op for groups whose
// `log_compaction_threshold` is `None`.
// Weak for the same cycle-breaking reason as `raft_proposer` above.
let raft_loop_for_compact = Arc::downgrade(raft_loop);
let sm_for_compact = Arc::clone(&sequencer_state_machine);
let compactor: Arc<crate::control::wal_replication::RaftCompactor> =
Arc::new(move |group_id, applied_index| {
let rl = raft_loop_for_compact
.upgrade()
.ok_or_else(|| crate::Error::Internal {
detail: "raft log compaction: cluster not running".into(),
})?;
// Sequencer-group hold-down. Unlike a data group — whose entries are
// replayable from each replica's own durable state — a cross-shard
// Calvin txn is re-derived on every replica ONLY from the sequencer
// log. A scheduler that missed a fan-out (channel full/closed, or it
// had not subscribed yet) recovers by replaying that log from its
// armed catch-up index. Compacting past an armed index destroys the
// only copy, permanently losing the txn on that replica — which for a
// cross-shard graph edge means the edge silently vanishes from that
// node's index. Floor the compaction boundary strictly below the
// lowest armed catch-up so the replay range always survives.
let effective_index = if group_id == nodedb_cluster::calvin::SEQUENCER_GROUP_ID {
match sm_for_compact
.lock()
.unwrap_or_else(|p| p.into_inner())
.min_catch_up_from()
{
// Keep index `m` itself: compaction discards entries at and
// below its boundary, and the replay range starts AT `m`.
Some(m) => applied_index.min(m.saturating_sub(1)),
None => applied_index,
}
} else {
applied_index
};
if effective_index == 0 {
// Nothing compactable once held down.
return Ok(false);
}
rl.maybe_compact_group(group_id, effective_index)
.map_err(|e| crate::Error::Internal {
detail: format!("raft log compaction: {e}"),
})
});
if shared.raft_compactor.set(compactor).is_err() {
tracing::warn!("raft_compactor already set — start_raft appears to have run twice");
}
// Wire the durable applied-index sink. `run_apply_loop` invokes this for
// each committed entry once the write funnel's durable-at-ack barrier has
// fsynced that entry's redo record, so the next boot resumes Raft delivery
// above it — without this floor the whole retained log is re-delivered on
// every boot and WAL replay applies the same entries a second time.
// Weak for the same cycle-breaking reason as `raft_proposer` above.
let raft_loop_for_applied = Arc::downgrade(raft_loop);
let applied_index_sink: Arc<crate::control::wal_replication::RaftAppliedIndexSink> =
Arc::new(move |group_id, applied_index| {
let rl = raft_loop_for_applied
.upgrade()
.ok_or_else(|| crate::Error::Internal {
detail: "raft applied index: cluster not running".into(),
})?;
rl.save_applied_index(group_id, applied_index)
.map_err(|e| crate::Error::Internal {
detail: format!("raft applied index: {e}"),
})
});
if shared
.raft_applied_index_sink
.set(applied_index_sink)
.is_err()
{
tracing::warn!(
"raft_applied_index_sink already set — start_raft appears to have run twice"
);
}
// Install the async proposer with transparent leader forwarding.
//
// Proposes via the data group leader (forwarding to a remote leader if
// needed), then registers a ProposeTracker waiter and awaits apply.
//
// The ProposeTracker is race-safe: if `run_apply_loop` calls complete()
// before register() is called (possible on fast clusters where the entry
// commits and applies on this node before the proposer returns), the
// result is stored and register() picks it up immediately with no timeout.
// Weak for the same cycle-breaking reason as `raft_proposer` above.
let raft_loop_async = Arc::downgrade(raft_loop);
let tracker_for_proposer = tracker.clone();
let deadline_secs = shared.tuning.network.default_deadline_secs;
let async_proposer: Arc<crate::control::wal_replication::AsyncRaftProposer> =
Arc::new(move |vshard_id, idempotency_key, data| {
let rl_weak = raft_loop_async.clone();
let tk = tracker_for_proposer.clone();
Box::pin(async move {
let rl = rl_weak.upgrade().ok_or_else(|| crate::Error::Internal {
detail: "raft propose (async): cluster not running".into(),
})?;
let (group_id, log_index) = rl
.propose_via_data_leader(vshard_id, data)
.await
.map_err(|e| crate::Error::Internal {
detail: format!("raft propose (async): {e}"),
})?;
// Register the waiter with the proposer's idempotency
// key. The apply path compares against the committed
// entry's key so a leader-change overwrite at the same
// (group_id, log_index) — by either an empty no-op or a
// different proposer's real entry — surfaces as
// `RetryableLeaderChange` instead of leaking a
// not-our-payload back to the caller.
let rx = tk.register(group_id, log_index, idempotency_key);
tokio::time::timeout(std::time::Duration::from_secs(deadline_secs), rx)
.await
.map_err(|_| crate::Error::Dispatch {
detail: format!(
"raft commit timeout for group {group_id} index {log_index}"
),
})?
.map_err(|_| crate::Error::Dispatch {
detail: "propose waiter channel closed".into(),
})?
// Preserve `RetryableLeaderChange` so the gateway
// retry loop can re-propose against the new leader
// — wrapping it in `Dispatch` would hide the
// retryable signal and surface as silent INSERT
// success. Other errors stay wrapped for
// diagnostics.
.map_err(|e| match e {
crate::Error::RetryableLeaderChange { .. } => e,
other => crate::Error::Dispatch {
detail: format!("apply error: {other}"),
},
})
// Carry out the write-version the APPLY side stamped, not
// `log_index`. The tracker resolves on the node that applied
// the entry locally, so `write_version` is this replica's own
// post-write `coll_write_lsn` — a WAL LSN, the same domain
// every other feed of that map records in, and the only
// domain the shard-local OCC read validator compares in. The
// raft log index is a per-group counter on a different scale
// entirely; publishing it here made reads validate a WAL LSN
// against a log index.
.map(|applied| (applied.payload, applied.write_version))
})
});
if shared.async_raft_proposer.set(async_proposer).is_err() {
tracing::warn!("async_raft_proposer already set — start_raft appears to have run twice");
}
// Spawn the background apply loop. It reads from the mpsc channel
// pushed by `DistributedApplier::apply_committed`, dispatches to the
// Data Plane, and notifies propose waiters. Registered via
// `spawn_loop_no_abort` so `LoopRegistry::shutdown_all` waits for it to
// exit (dropping its captured `Arc<SharedState>` deterministically) but
// NEVER force-aborts it — an abort mid-apply would strand
// committed-but-unapplied entries.
let apply_state = shared.clone();
let apply_tracker = tracker.clone();
let apply_calvin_read_result_senders = Arc::clone(&calvin_read_result_senders);
crate::control::shutdown::spawn_loop_no_abort(
&shared.loop_registry,
&shared.shutdown,
"raft_apply_loop",
move |mut shutdown| async move {
// `biased` polls `run_apply_loop` FIRST on every iteration: any
// committed batch already queued in `apply_rx` is drained to the
// Data Plane before the shutdown arm can win, so shutdown never
// cuts the loop off mid-apply.
tokio::select! {
biased;
_ = run_apply_loop(
apply_rx,
apply_state,
apply_tracker,
apply_calvin_read_result_senders,
) => {}
_ = shutdown.wait_cancelled() => {}
}
},
);
}