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
// SPDX-License-Identifier: BUSL-1.1
//! Lite sync-producer registration/fencing and live routing-table
//! placement host-side effects.
use tracing::{debug, warn};
use super::types::MetadataCommitApplier;
impl MetadataCommitApplier {
pub(super) fn apply_sync_producer_register(
&self,
lite_id: &str,
producer_id: u64,
tenant_id: u64,
epoch: u64,
created_ms: i64,
raft_index: u64,
) -> Result<(), crate::Error> {
if let Some(weak) = self.shared.get()
&& let Some(shared) = weak.upgrade()
{
let Some(registry) = shared.producer_registry.as_deref() else {
return Ok(());
};
// The registration row is durable replicated state. A write
// failure must not advance the watermark — Raft re-delivers
// and `apply_register` is idempotent, so the retry is safe.
if let Err(e) =
registry.apply_register(lite_id, producer_id, tenant_id, epoch, created_ms)
{
warn!(
lite_id = %lite_id,
producer_id,
error = %e,
"sync_producer_register apply failed — halting watermark for retry"
);
return Err(crate::Error::Internal {
detail: format!("sync_producer_register apply failed: {e}"),
});
}
debug!(lite_id = %lite_id, producer_id, raft_index, "sync producer registered via raft");
}
Ok(())
}
pub(super) fn apply_sync_producer_fence(
&self,
lite_id: &str,
new_epoch: u64,
raft_index: u64,
) -> Result<(), crate::Error> {
if let Some(weak) = self.shared.get()
&& let Some(shared) = weak.upgrade()
{
let Some(registry) = shared.producer_registry.as_deref() else {
return Ok(());
};
// Durable epoch advance; `apply_fence` is idempotent
// (max-wins) so re-delivery on failure is safe.
if let Err(e) = registry.apply_fence(lite_id, new_epoch) {
warn!(
lite_id = %lite_id,
new_epoch,
error = %e,
"sync_producer_fence apply failed — halting watermark for retry"
);
return Err(crate::Error::Internal {
detail: format!("sync_producer_fence apply failed: {e}"),
});
}
debug!(lite_id = %lite_id, new_epoch, raft_index, "sync producer fenced via raft");
}
Ok(())
}
/// Group membership and leadership converge through the Raft
/// conf-change path (which mutates the shared routing table on
/// every node). `SetPlacement` carries the *intended* voter set
/// for a group and has no conf-change equivalent, so it must be
/// written through to the live shared routing table here — the
/// same `RwLock<RoutingTable>` the reconciler and the
/// learner-promotion gate read. Without this write the placement
/// never leaves the metadata log and N>RF voter-cap convergence
/// is inert. The other `RoutingChange` variants are intentionally
/// not handled here to avoid double-applying the conf-change path.
pub(super) fn apply_set_placement(
&self,
group_id: u64,
placement: &[u64],
raft_index: u64,
) -> Result<(), crate::Error> {
if let Some(weak) = self.shared.get()
&& let Some(shared) = weak.upgrade()
&& let Some(routing) = shared.cluster_routing.as_ref()
{
routing
.write()
.unwrap_or_else(|p| p.into_inner())
.set_placement(group_id, placement.to_vec());
debug!(
group_id,
raft_index, "set_placement applied to live routing table"
);
}
Ok(())
}
}