reddb_wire/
moved.rs

1//! MOVED-style stale-ownership redirect payload.
2//!
3//! The shape is shared by transports: a non-owner returns it when routing
4//! metadata says another member currently owns the target range. Clients update
5//! their routing cache from this payload and retry once against `owner_addr`.
6
7use serde_json::{Map, Value};
8
9use crate::redwire::{Frame, MessageKind};
10
11pub const MOVED_CODE: &str = "MOVED";
12
13#[derive(Debug, Clone, PartialEq, Eq)]
14pub struct MovedRedirect {
15    pub slot: Option<u64>,
16    pub collection: String,
17    pub range_id: u64,
18    pub owner_addr: String,
19    pub ownership_epoch: u64,
20    pub catalog_version: u64,
21    pub reason: String,
22}
23
24#[derive(Debug, Clone, PartialEq, Eq)]
25pub enum MovedRedirectError {
26    InvalidJson,
27    ExpectedObject,
28    MissingField(&'static str),
29    InvalidField(&'static str),
30}
31
32impl std::fmt::Display for MovedRedirectError {
33    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
34        match self {
35            Self::InvalidJson => write!(f, "MOVED payload is not valid JSON"),
36            Self::ExpectedObject => write!(f, "MOVED payload must be a JSON object"),
37            Self::MissingField(field) => write!(f, "MOVED payload missing field {field}"),
38            Self::InvalidField(field) => write!(f, "MOVED payload field {field} is invalid"),
39        }
40    }
41}
42
43impl std::error::Error for MovedRedirectError {}
44
45pub fn encode_moved_redirect(payload: &MovedRedirect) -> Vec<u8> {
46    let mut obj = Map::new();
47    obj.insert("code".to_string(), Value::String(MOVED_CODE.to_string()));
48    if let Some(slot) = payload.slot {
49        obj.insert("slot".to_string(), Value::Number(slot.into()));
50    }
51    obj.insert(
52        "collection".to_string(),
53        Value::String(payload.collection.clone()),
54    );
55    obj.insert(
56        "range_id".to_string(),
57        Value::Number(payload.range_id.into()),
58    );
59    obj.insert(
60        "owner_addr".to_string(),
61        Value::String(payload.owner_addr.clone()),
62    );
63    obj.insert(
64        "ownership_epoch".to_string(),
65        Value::Number(payload.ownership_epoch.into()),
66    );
67    obj.insert(
68        "catalog_version".to_string(),
69        Value::Number(payload.catalog_version.into()),
70    );
71    obj.insert("reason".to_string(), Value::String(payload.reason.clone()));
72    serde_json::to_vec(&Value::Object(obj)).expect("MOVED payload JSON encoding cannot fail")
73}
74
75pub fn decode_moved_redirect(bytes: &[u8]) -> Result<MovedRedirect, MovedRedirectError> {
76    let value: Value =
77        serde_json::from_slice(bytes).map_err(|_| MovedRedirectError::InvalidJson)?;
78    let obj = value
79        .as_object()
80        .ok_or(MovedRedirectError::ExpectedObject)?;
81    let code = required_str(obj, "code")?;
82    if code != MOVED_CODE {
83        return Err(MovedRedirectError::InvalidField("code"));
84    }
85    Ok(MovedRedirect {
86        slot: optional_u64(obj, "slot")?,
87        collection: required_str(obj, "collection")?.to_string(),
88        range_id: required_u64(obj, "range_id")?,
89        owner_addr: required_str(obj, "owner_addr")?.to_string(),
90        ownership_epoch: required_u64(obj, "ownership_epoch")?,
91        catalog_version: required_u64(obj, "catalog_version")?,
92        reason: required_str(obj, "reason")?.to_string(),
93    })
94}
95
96pub fn build_moved_redirect_frame(correlation_id: u64, payload: &MovedRedirect) -> Frame {
97    Frame::new(
98        MessageKind::MovedRedirect,
99        correlation_id,
100        encode_moved_redirect(payload),
101    )
102}
103
104fn required_str<'a>(
105    obj: &'a Map<String, Value>,
106    field: &'static str,
107) -> Result<&'a str, MovedRedirectError> {
108    obj.get(field)
109        .ok_or(MovedRedirectError::MissingField(field))?
110        .as_str()
111        .filter(|value| !value.is_empty())
112        .ok_or(MovedRedirectError::InvalidField(field))
113}
114
115fn required_u64(obj: &Map<String, Value>, field: &'static str) -> Result<u64, MovedRedirectError> {
116    obj.get(field)
117        .ok_or(MovedRedirectError::MissingField(field))?
118        .as_u64()
119        .ok_or(MovedRedirectError::InvalidField(field))
120}
121
122fn optional_u64(
123    obj: &Map<String, Value>,
124    field: &'static str,
125) -> Result<Option<u64>, MovedRedirectError> {
126    match obj.get(field) {
127        Some(value) => value
128            .as_u64()
129            .map(Some)
130            .ok_or(MovedRedirectError::InvalidField(field)),
131        None => Ok(None),
132    }
133}
134
135#[cfg(test)]
136mod tests {
137    use super::*;
138
139    fn payload() -> MovedRedirect {
140        MovedRedirect {
141            slot: Some(42),
142            collection: "orders".to_string(),
143            range_id: 7,
144            owner_addr: "node-b:5050".to_string(),
145            ownership_epoch: 9,
146            catalog_version: 11,
147            reason: "transaction".to_string(),
148        }
149    }
150
151    #[test]
152    fn moved_payload_round_trips_owner_epoch_version_and_slot() {
153        let encoded = encode_moved_redirect(&payload());
154        let decoded = decode_moved_redirect(&encoded).expect("decode MOVED");
155
156        assert_eq!(decoded, payload());
157    }
158
159    #[test]
160    fn moved_frame_uses_server_to_client_kind() {
161        let frame = build_moved_redirect_frame(99, &payload());
162
163        assert_eq!(frame.kind, MessageKind::MovedRedirect);
164        assert_eq!(frame.correlation_id, 99);
165        assert_eq!(
166            decode_moved_redirect(&frame.payload).expect("decode frame payload"),
167            payload()
168        );
169    }
170}
reddb_wire/moved.rs

reddb_wire/
moved.rs
