nodedb_cluster/rpc_codec/
surrogate.rs

1// SPDX-License-Identifier: BUSL-1.1
2
3//! Routed-surrogate-exchange one-shot RPC (F1b).
4//!
5//! A coordinator planning a cross-shard graph edge needs the AUTHORITATIVE
6//! global surrogate (the stable `u32` cross-engine identity) for an endpoint key
7//! whose home vShard is owned by ANOTHER node. It sends an
8//! [`AssignSurrogateRequest`] to that vShard's LEADER; the leader runs a LOCAL
9//! `SurrogateAssigner::assign` for `(collection, pk)` — which, because the leader
10//! IS the home node, yields the authoritative (first-wins, idempotent) value —
11//! and replies with exactly one [`AssignSurrogateResponse`]. One-shot
12//! request/response; no streaming.
13//!
14//! Discriminants 34/35 are permanently assigned to these variants.
15
16use super::discriminants::*;
17use super::execute::TypedClusterError;
18use super::header::write_frame;
19use super::raft_rpc::RaftRpc;
20use crate::error::{ClusterError, Result};
21
22// ── Wire types ──────────────────────────────────────────────────────────────
23
24/// Coordinator → leader routed-surrogate-exchange request (F1b).
25///
26/// Carries the endpoint key `(collection, pk)` plus the identity scope
27/// (`database_id`, `tenant_id`) and routing key (`vshard_id`) the home vShard was
28/// resolved from. The `deadline_remaining_ms` / `trace_id` fields mirror
29/// [`ExecuteRequest`](super::execute::ExecuteRequest) so the leader-side handler
30/// shares the same deadline / tracing prologue as the other one-shot RPCs.
31///
32/// Cross-version safety: new optional fields should be added as `Option<T>`.
33#[derive(Debug, Clone, rkyv::Archive, rkyv::Serialize, rkyv::Deserialize)]
34pub struct AssignSurrogateRequest {
35    /// Home vShard the `(collection, pk)` endpoint key routes to.
36    pub vshard_id: u32,
37    pub database_id: u64,
38    pub tenant_id: u64,
39    /// Collection the surrogate is scoped to.
40    pub collection: String,
41    /// Primary-key bytes of the endpoint whose surrogate is being resolved.
42    pub pk: Vec<u8>,
43    pub deadline_remaining_ms: u64,
44    pub trace_id: [u8; 16],
45}
46
47/// Terminal reply to an [`AssignSurrogateRequest`].
48///
49/// `error: None` carries the authoritative `surrogate` (a non-zero `u32` when the
50/// leader has a wired catalog; `0` is only ever returned with an `error`).
51/// `error: Some(e)` means the leader-side assign failed (lock poisoned, backend
52/// error, or no assign-remote-surrogate hook configured); `surrogate` is `0` in
53/// that case.
54#[derive(Debug, Clone, rkyv::Archive, rkyv::Serialize, rkyv::Deserialize)]
55pub struct AssignSurrogateResponse {
56    /// The authoritative surrogate. `0` only on error.
57    pub surrogate: u32,
58    pub error: Option<TypedClusterError>,
59}
60
61// ── Codec ────────────────────────────────────────────────────────────────────
62
63macro_rules! to_bytes {
64    ($msg:expr) => {
65        rkyv::to_bytes::<rkyv::rancor::Error>($msg)
66            .map(|b| b.to_vec())
67            .map_err(|e| ClusterError::Codec {
68                detail: format!("rkyv serialize: {e}"),
69            })
70    };
71}
72
73macro_rules! from_bytes {
74    ($payload:expr, $T:ty, $name:expr) => {{
75        let mut aligned = rkyv::util::AlignedVec::<16>::with_capacity($payload.len());
76        aligned.extend_from_slice($payload);
77        rkyv::from_bytes::<$T, rkyv::rancor::Error>(&aligned).map_err(|e| ClusterError::Codec {
78            detail: format!("rkyv deserialize {}: {e}", $name),
79        })
80    }};
81}
82
83pub(super) fn encode_assign_surrogate_req(
84    msg: &AssignSurrogateRequest,
85    out: &mut Vec<u8>,
86) -> Result<()> {
87    write_frame(RPC_ASSIGN_SURROGATE_REQ, &to_bytes!(msg)?, out)
88}
89pub(super) fn encode_assign_surrogate_resp(
90    msg: &AssignSurrogateResponse,
91    out: &mut Vec<u8>,
92) -> Result<()> {
93    write_frame(RPC_ASSIGN_SURROGATE_RESP, &to_bytes!(msg)?, out)
94}
95
96pub(super) fn decode_assign_surrogate_req(payload: &[u8]) -> Result<RaftRpc> {
97    Ok(RaftRpc::AssignSurrogateRequest(from_bytes!(
98        payload,
99        AssignSurrogateRequest,
100        "AssignSurrogateRequest"
101    )?))
102}
103pub(super) fn decode_assign_surrogate_resp(payload: &[u8]) -> Result<RaftRpc> {
104    Ok(RaftRpc::AssignSurrogateResponse(from_bytes!(
105        payload,
106        AssignSurrogateResponse,
107        "AssignSurrogateResponse"
108    )?))
109}
110
111#[cfg(test)]
112mod tests {
113    use super::*;
114
115    fn roundtrip_req(req: AssignSurrogateRequest) -> AssignSurrogateRequest {
116        let rpc = RaftRpc::AssignSurrogateRequest(req);
117        let encoded = super::super::encode(&rpc).unwrap();
118        match super::super::decode(&encoded).unwrap() {
119            RaftRpc::AssignSurrogateRequest(r) => r,
120            other => panic!("expected AssignSurrogateRequest, got {other:?}"),
121        }
122    }
123
124    fn roundtrip_resp(resp: AssignSurrogateResponse) -> AssignSurrogateResponse {
125        let rpc = RaftRpc::AssignSurrogateResponse(resp);
126        let encoded = super::super::encode(&rpc).unwrap();
127        match super::super::decode(&encoded).unwrap() {
128            RaftRpc::AssignSurrogateResponse(r) => r,
129            other => panic!("expected AssignSurrogateResponse, got {other:?}"),
130        }
131    }
132
133    #[test]
134    fn roundtrip_assign_surrogate_request() {
135        let req = AssignSurrogateRequest {
136            vshard_id: 512,
137            database_id: 7,
138            tenant_id: 42,
139            collection: "people".into(),
140            pk: vec![0x61, 0x6C, 0x69, 0x63, 0x65],
141            deadline_remaining_ms: 5000,
142            trace_id: [9u8; 16],
143        };
144        let decoded = roundtrip_req(req.clone());
145        assert_eq!(decoded.vshard_id, 512);
146        assert_eq!(decoded.database_id, 7);
147        assert_eq!(decoded.tenant_id, 42);
148        assert_eq!(decoded.collection, "people");
149        assert_eq!(decoded.pk, vec![0x61, 0x6C, 0x69, 0x63, 0x65]);
150        assert_eq!(decoded.deadline_remaining_ms, 5000);
151        assert_eq!(decoded.trace_id, [9u8; 16]);
152    }
153
154    #[test]
155    fn roundtrip_assign_surrogate_request_empty_pk() {
156        let req = AssignSurrogateRequest {
157            vshard_id: 0,
158            database_id: 0,
159            tenant_id: 0,
160            collection: String::new(),
161            pk: vec![],
162            deadline_remaining_ms: 1000,
163            trace_id: [0u8; 16],
164        };
165        let decoded = roundtrip_req(req);
166        assert!(decoded.pk.is_empty());
167        assert!(decoded.collection.is_empty());
168        assert_eq!(decoded.deadline_remaining_ms, 1000);
169    }
170
171    #[test]
172    fn roundtrip_assign_surrogate_response_ok() {
173        let decoded = roundtrip_resp(AssignSurrogateResponse {
174            surrogate: 12345,
175            error: None,
176        });
177        assert_eq!(decoded.surrogate, 12345);
178        assert!(decoded.error.is_none());
179    }
180
181    #[test]
182    fn roundtrip_assign_surrogate_response_error() {
183        let decoded = roundtrip_resp(AssignSurrogateResponse {
184            surrogate: 0,
185            error: Some(TypedClusterError::Internal {
186                code: 0x7,
187                message: "assign-remote-surrogate not configured".into(),
188            }),
189        });
190        assert_eq!(decoded.surrogate, 0);
191        match decoded.error {
192            Some(TypedClusterError::Internal { code, message }) => {
193                assert_eq!(code, 0x7);
194                assert!(message.contains("assign-remote-surrogate"));
195            }
196            other => panic!("expected Internal, got {other:?}"),
197        }
198    }
199}
nodedb_cluster/rpc_codec/surrogate.rs

nodedb_cluster/rpc_codec/
surrogate.rs
