faucet_source_pubsub/
state.rs

1//! Bookmark shape + state key for the Pub/Sub source.
2//!
3//! Pub/Sub has **no client-side resume offset** — durability is the server
4//! tracking acked messages on the subscription. So the emitted bookmark is
5//! purely informational (a cumulative count + the last message id): it exists
6//! so each durable page triggers a `flush` + `StateStore::put`, which is the
7//! signal the streaming loop uses to ack the previous page. On resume the
8//! subscription redelivers whatever was never acked, so the persisted
9//! bookmark is not consulted to seek.
10
11use serde::{Deserialize, Serialize};
12use serde_json::Value;
13
14/// The persisted (informational) bookmark value.
15#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
16pub struct PubsubBookmark {
17    /// Cumulative messages emitted this run.
18    #[serde(default)]
19    pub delivered: u64,
20    /// The most recently emitted message id (for observability).
21    #[serde(default, skip_serializing_if = "Option::is_none")]
22    pub last_message_id: Option<String>,
23}
24
25impl PubsubBookmark {
26    /// Record one more delivered message.
27    pub fn advance(&mut self, message_id: &str) {
28        self.delivered += 1;
29        self.last_message_id = Some(message_id.to_string());
30    }
31
32    pub fn to_value(&self) -> Value {
33        serde_json::to_value(self).unwrap_or(Value::Null)
34    }
35
36    /// Parse a bookmark `Value`; a malformed value is treated as fresh (never
37    /// fails a run — Pub/Sub redelivers unacked messages regardless).
38    pub fn from_value(v: &Value) -> Self {
39        serde_json::from_value(v.clone()).unwrap_or_default()
40    }
41}
42
43/// The source's stable state key.
44pub fn state_key(subscription: &str) -> String {
45    format!("pubsub:{subscription}")
46}
47
48#[cfg(test)]
49mod tests {
50    use super::*;
51    use serde_json::json;
52
53    #[test]
54    fn bookmark_advances_and_round_trips() {
55        let mut b = PubsubBookmark::default();
56        b.advance("m1");
57        b.advance("m2");
58        assert_eq!(b.delivered, 2);
59        assert_eq!(b.last_message_id.as_deref(), Some("m2"));
60        let back = PubsubBookmark::from_value(&b.to_value());
61        assert_eq!(back, b);
62    }
63
64    #[test]
65    fn malformed_bookmark_is_fresh() {
66        assert_eq!(
67            PubsubBookmark::from_value(&json!("nope")),
68            PubsubBookmark::default()
69        );
70        assert_eq!(
71            PubsubBookmark::from_value(&json!(null)),
72            PubsubBookmark::default()
73        );
74    }
75
76    #[test]
77    fn state_key_shape_is_valid() {
78        assert_eq!(state_key("orders-sub"), "pubsub:orders-sub");
79        faucet_core::state::validate_state_key(&state_key("orders-sub")).unwrap();
80    }
81}
faucet_source_pubsub/state.rs

faucet_source_pubsub/
state.rs
