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
//! Cross-node shuffle (E1/E4a/E4b/E5b), routed-surrogate-exchange (F1b),
//! and routed Calvin-submit (Cv1) RPC bodies.
use crate::error::Result;
use crate::forward::PlanExecutor;
use crate::rpc_codec::{
AssignSurrogateRequest, AssignSurrogateResponse, ReleaseReservationRequest,
ReleaseReservationResponse, ReserveReadRequest, ReserveReadResponse,
ShuffleAggregateConsumeRequest, ShuffleAggregateConsumeResponse, ShuffleConsumeRequest,
ShuffleConsumeResponse, ShuffleProduceRequest, ShuffleProduceResponse, ShufflePushRequest,
SubmitCalvinInboxRequest, SubmitCalvinInboxResponse, SubmitCalvinTxnRequest,
SubmitCalvinTxnResponse, TypedClusterError,
};
use super::super::loop_core::{CommitApplier, RaftLoop};
impl<A: CommitApplier, P: PlanExecutor> RaftLoop<A, P> {
// Cross-node streaming shuffle (E1) — delegate to the host-crate
// `ShuffleReceiver` (backed by `nodedb`'s `ShuffleReceiverRegistry`). When
// no receiver is installed (cluster-only tests / single-node), the request
// is a no-op and a chunk surfaces a typed "not configured" error.
pub(super) async fn on_shuffle_request_impl(&self, req: ShufflePushRequest) {
if let Some(recv) = &self.shuffle_receiver {
recv.on_shuffle_request(req.shuffle_id, req.part, req.side, req.producer_count)
.await;
}
}
pub(super) async fn on_shuffle_chunk_impl(
&self,
shuffle_id: u64,
part: u32,
side: u8,
payload: Vec<u8>,
) -> Result<()> {
match &self.shuffle_receiver {
Some(recv) => recv.on_shuffle_chunk(shuffle_id, part, side, payload).await,
None => Err(crate::error::ClusterError::Transport {
detail: "shuffle receiver not configured (no ShuffleReceiver installed)".into(),
}),
}
}
pub(super) async fn on_shuffle_end_impl(
&self,
shuffle_id: u64,
part: u32,
side: u8,
error: Option<TypedClusterError>,
) {
if let Some(recv) = &self.shuffle_receiver {
recv.on_shuffle_end(shuffle_id, part, side, error).await;
}
}
// Cross-node shuffle PRODUCER trigger (E4a) — delegate to the host-crate
// `ShuffleProducer`. When no producer is installed (cluster-only tests /
// single-node), return a typed "not configured" error so the coordinator
// learns the trigger could not run rather than silently succeeding.
pub(super) async fn on_shuffle_produce_impl(
&self,
req: ShuffleProduceRequest,
) -> ShuffleProduceResponse {
match &self.shuffle_producer {
Some(producer) => producer.on_shuffle_produce(req).await,
None => ShuffleProduceResponse {
error: Some(TypedClusterError::Internal {
code: 0,
message: "shuffle producer not configured (no ShuffleProducer installed)"
.into(),
}),
read_version_lsn: 0,
},
}
}
// Cross-node shuffle CONSUMER trigger (E4b) — delegate to the host-crate
// `ShuffleConsumer`. When no consumer is installed (cluster-only tests /
// single-node), return a typed "not configured" error (empty rows) so the
// coordinator learns the trigger could not run rather than silently
// receiving zero join rows.
pub(super) async fn on_shuffle_consume_impl(
&self,
req: ShuffleConsumeRequest,
) -> ShuffleConsumeResponse {
match &self.shuffle_consumer {
Some(consumer) => consumer.on_shuffle_consume(req).await,
None => ShuffleConsumeResponse {
rows: Vec::new(),
error: Some(TypedClusterError::Internal {
code: 0,
message: "shuffle consumer not configured (no ShuffleConsumer installed)"
.into(),
}),
},
}
}
// Cross-node distributed GROUP BY shuffle CONSUMER trigger (E5b) — delegate
// to the host-crate `ShuffleAggregator`. The single-sided aggregate sibling
// of `on_shuffle_consume`. When no aggregator is installed (cluster-only
// tests / single-node), return a typed "not configured" error (empty rows)
// so the coordinator learns the trigger could not run rather than silently
// receiving zero aggregate rows.
pub(super) async fn on_shuffle_aggregate_impl(
&self,
req: ShuffleAggregateConsumeRequest,
) -> ShuffleAggregateConsumeResponse {
match &self.shuffle_aggregator {
Some(aggregator) => aggregator.on_shuffle_aggregate(req).await,
None => ShuffleAggregateConsumeResponse {
rows: Vec::new(),
error: Some(TypedClusterError::Internal {
code: 0,
message: "shuffle aggregator not configured (no ShuffleAggregator installed)"
.into(),
}),
},
}
}
// Routed-surrogate-exchange (F1b) — delegate to the host-crate
// `AssignRemoteSurrogate`. This node is the home vShard's leader; a LOCAL
// assign through it yields the authoritative surrogate. When no assigner is
// installed (cluster-only tests / single-node), return a typed "not
// configured" error (surrogate 0) so the coordinator learns the request could
// not run rather than silently receiving a bogus zero surrogate.
pub(super) async fn on_assign_surrogate_impl(
&self,
req: AssignSurrogateRequest,
) -> AssignSurrogateResponse {
match &self.assign_remote_surrogate {
Some(assigner) => assigner.on_assign_surrogate(req).await,
None => AssignSurrogateResponse {
surrogate: 0,
error: Some(TypedClusterError::Internal {
code: 0,
message: "assign-remote-surrogate not configured \
(no AssignRemoteSurrogate installed)"
.into(),
}),
},
}
}
// Routed Calvin-submit (Cv1) — delegate to the host-crate `CalvinSubmit`.
// This node is the sequencer-group leader; submitting + awaiting through it
// is correct because the leader's sequencer service assigns and the leader's
// registry receives the replicated completion ack. When no Calvin-submit
// hook is installed (cluster-only tests / single-node), return a typed "not
// configured" error so the coordinator learns the request could not run
// rather than silently believing the cross-shard write committed.
pub(super) async fn on_submit_calvin_txn_impl(
&self,
req: SubmitCalvinTxnRequest,
) -> SubmitCalvinTxnResponse {
match &self.calvin_submit {
Some(submit) => submit.on_submit_calvin_txn(req).await,
None => SubmitCalvinTxnResponse {
error: Some(TypedClusterError::Internal {
code: 0,
message: "calvin-submit not configured (no CalvinSubmit installed)".into(),
}),
payload_bytes: None,
},
}
}
// Routed Calvin-INBOX submit (Cv1) — delegate to the host-crate
// `CalvinSubmitInbox`. The OLLP dependent sibling of `on_submit_calvin_txn`.
// This node is the sequencer-group leader; submitting + awaiting the
// assignment through it is correct because the leader's sequencer service
// assigns. When no Calvin-inbox hook is installed (cluster-only tests /
// single-node), return a typed "not configured" error so the coordinator
// learns the request could not run rather than silently believing the
// dependent transaction was assigned.
pub(super) async fn on_submit_calvin_inbox_impl(
&self,
req: SubmitCalvinInboxRequest,
) -> SubmitCalvinInboxResponse {
match &self.calvin_submit_inbox {
Some(submit) => submit.on_submit_calvin_inbox(req).await,
None => SubmitCalvinInboxResponse {
inbox_seq: 0,
epoch: 0,
position: 0,
participants: 0,
error: Some(TypedClusterError::Internal {
code: 0,
message: "calvin-inbox not configured (no CalvinSubmitInbox installed)".into(),
}),
},
}
}
// Routed reserve-read (Calvin OLLP) — delegate to the host-crate
// `ReserveRead`. This node is the sequencer-group leader; reserving
// through it is correct because the leader's scheduler holds the
// authoritative lock table for its local sequencer inbox. When no
// reserve-read hook is installed (cluster-only tests / single-node),
// return a typed "not configured" error so the coordinator learns the
// request could not run rather than silently believing the read was
// reserved.
pub(super) async fn on_reserve_read_impl(
&self,
req: ReserveReadRequest,
) -> ReserveReadResponse {
match &self.reserve_read {
Some(hook) => hook.on_reserve_read(req).await,
None => ReserveReadResponse {
owner_bytes: None,
error: Some(TypedClusterError::Internal {
code: 0,
message: "reserve-read not configured (no ReserveRead installed)".into(),
}),
},
}
}
// Routed release-reservation (Calvin OLLP) — delegate to the host-crate
// `ReleaseReservation`. The ack-only sibling of `on_reserve_read`. When no
// release-reservation hook is installed (cluster-only tests /
// single-node), return a typed "not configured" error so the coordinator
// learns the request could not run rather than silently believing the
// reservation was released.
pub(super) async fn on_release_reservation_impl(
&self,
req: ReleaseReservationRequest,
) -> ReleaseReservationResponse {
match &self.release_reservation {
Some(hook) => hook.on_release_reservation(req).await,
None => ReleaseReservationResponse {
error: Some(TypedClusterError::Internal {
code: 0,
message: "release-reservation not configured (no ReleaseReservation installed)"
.into(),
}),
},
}
}
}