faucet_source_mongodb_cdc/
state.rs

1//! State key derivation and resume-token bookmark serialization.
2//!
3//! Bookmark shape (round-trips through `serde_json::Value`):
4//!
5//! ```json
6//! { "resume_token": { "_data": "8264..." } }
7//! ```
8
9use crate::config::Scope;
10use faucet_core::FaucetError;
11use mongodb::bson::{self, Bson};
12use mongodb::change_stream::event::ResumeToken;
13use serde::{Deserialize, Serialize};
14use serde_json::Value;
15
16/// State key for a given watch scope.
17///
18/// `cluster` → `mongodb-cdc:cluster`
19/// `database` → `mongodb-cdc:db:<database>`
20/// `collection` → `mongodb-cdc:coll:<database>.<collection>`
21pub fn state_key(scope: &Scope) -> String {
22    match scope {
23        Scope::Cluster => "mongodb-cdc:cluster".to_string(),
24        Scope::Database { database } => format!("mongodb-cdc:db:{database}"),
25        Scope::Collection {
26            database,
27            collection,
28        } => {
29            format!("mongodb-cdc:coll:{database}.{collection}")
30        }
31    }
32}
33
34/// Durable bookmark wrapping a resume token.
35#[derive(Clone, Debug, Default, Serialize, Deserialize, PartialEq)]
36pub struct Bookmark {
37    /// The opaque resume token as relaxed extended JSON (e.g. `{"_data": "..."}`).
38    pub resume_token: Value,
39    /// True when this token was captured from an **Invalidate** change event
40    /// (the collection was dropped/renamed, or a dropDatabase for db scope).
41    /// MongoDB forbids `resumeAfter` on an invalidate token — only `startAfter`
42    /// — so on resume the source must open the next stream with `start_after`,
43    /// or every subsequent watch open fails and the CDC pipeline wedges
44    /// permanently (audit #321 M3). Absent from older bookmarks (defaults
45    /// `false`) and omitted from the serialized state when `false`.
46    #[serde(default, skip_serializing_if = "is_false")]
47    pub invalidate: bool,
48}
49
50/// `skip_serializing_if` predicate keeping non-invalidate bookmarks byte-compatible
51/// with the pre-#321 state format.
52fn is_false(b: &bool) -> bool {
53    !*b
54}
55
56impl Bookmark {
57    /// Build a bookmark from a driver `ResumeToken`.
58    pub fn from_token(token: &ResumeToken) -> Result<Self, FaucetError> {
59        let bson = bson::to_bson(token)
60            .map_err(|e| FaucetError::State(format!("mongodb-cdc resume token serialize: {e}")))?;
61        Ok(Self {
62            resume_token: bson.into_relaxed_extjson(),
63            invalidate: false,
64        })
65    }
66
67    /// Parse a bookmark previously emitted by `to_value`.
68    pub fn from_value(v: Value) -> Result<Self, FaucetError> {
69        serde_json::from_value(v)
70            .map_err(|e| FaucetError::State(format!("mongodb-cdc bookmark parse: {e}")))
71    }
72
73    /// Serialize for the state store.
74    pub fn to_value(&self) -> Result<Value, FaucetError> {
75        serde_json::to_value(self)
76            .map_err(|e| FaucetError::State(format!("mongodb-cdc bookmark serialize: {e}")))
77    }
78
79    /// Decode the stored resume token back into a driver `ResumeToken`.
80    pub fn to_token(&self) -> Result<ResumeToken, FaucetError> {
81        let bson: Bson = bson::to_bson(&self.resume_token)
82            .map_err(|e| FaucetError::State(format!("mongodb-cdc resume token to bson: {e}")))?;
83        bson::from_bson(bson)
84            .map_err(|e| FaucetError::State(format!("mongodb-cdc resume token decode: {e}")))
85    }
86}
87
88#[cfg(test)]
89mod tests {
90    use super::*;
91    use serde_json::json;
92
93    #[test]
94    fn state_key_cluster() {
95        assert_eq!(state_key(&Scope::Cluster), "mongodb-cdc:cluster");
96    }
97
98    #[test]
99    fn state_key_database() {
100        assert_eq!(
101            state_key(&Scope::Database {
102                database: "app".into()
103            }),
104            "mongodb-cdc:db:app"
105        );
106    }
107
108    #[test]
109    fn state_key_collection() {
110        assert_eq!(
111            state_key(&Scope::Collection {
112                database: "app".into(),
113                collection: "users".into()
114            }),
115            "mongodb-cdc:coll:app.users"
116        );
117    }
118
119    #[test]
120    fn bookmark_value_round_trip() {
121        let b = Bookmark {
122            resume_token: json!({ "_data": "8264AB" }),
123            ..Default::default()
124        };
125        let v = b.to_value().unwrap();
126        let parsed = Bookmark::from_value(v).unwrap();
127        assert_eq!(parsed, b);
128    }
129
130    #[test]
131    fn bookmark_token_round_trip() {
132        let b = Bookmark {
133            resume_token: json!({ "_data": "8264AB00" }),
134            ..Default::default()
135        };
136        let token = b.to_token().unwrap();
137        let b2 = Bookmark::from_token(&token).unwrap();
138        assert_eq!(b2.resume_token["_data"], json!("8264AB00"));
139    }
140
141    #[test]
142    fn from_value_rejects_garbage() {
143        assert!(Bookmark::from_value(json!({})).is_err());
144        assert!(Bookmark::from_value(json!("bare")).is_err());
145    }
146
147    #[test]
148    fn invalidate_flag_serde_is_backward_compatible() {
149        // #321 M3: a legacy bookmark without the field parses as invalidate=false
150        // and a non-invalidate bookmark omits the field entirely (compact state).
151        let legacy = Bookmark::from_value(json!({ "resume_token": { "_data": "AA" } })).unwrap();
152        assert!(!legacy.invalidate);
153        let normal = Bookmark {
154            resume_token: json!({ "_data": "AA" }),
155            invalidate: false,
156        };
157        let v = normal.to_value().unwrap();
158        assert!(v.get("invalidate").is_none(), "false flag is omitted: {v}");
159        // An invalidate bookmark serializes the flag and round-trips.
160        let inv = Bookmark {
161            resume_token: json!({ "_data": "BB" }),
162            invalidate: true,
163        };
164        let back = Bookmark::from_value(inv.to_value().unwrap()).unwrap();
165        assert!(back.invalidate);
166    }
167}
faucet_source_mongodb_cdc/state.rs

faucet_source_mongodb_cdc/
state.rs
