a3s_code_core/store/
watch.rs

1//! Commit watch notifications for session stores (KRN-6 / STORE-WATCH1).
2//!
3//! Backends that advertise [`super::SessionStoreCapabilities::watch`] publish
4//! one event after a complete snapshot generation becomes durable. Events
5//! carry session identity and digest only — never session plaintext.
6
7use anyhow::{bail, Result};
8use serde::{Deserialize, Serialize};
9use tokio::sync::broadcast;
10
11pub const SESSION_STORE_COMMIT_EVENT_SCHEMA_V1: &str = "a3s.code.session-store-commit-event.v1";
12
13/// Notification that one complete session generation was committed.
14#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
15#[serde(rename_all = "camelCase", deny_unknown_fields)]
16pub struct SessionStoreCommitEventV1 {
17    pub schema: String,
18    pub session_id: String,
19    pub snapshot_digest: String,
20    pub committed_at_ms: u64,
21}
22
23impl SessionStoreCommitEventV1 {
24    pub fn new(
25        session_id: impl Into<String>,
26        snapshot_digest: impl Into<String>,
27        committed_at_ms: u64,
28    ) -> Result<Self> {
29        let event = Self {
30            schema: SESSION_STORE_COMMIT_EVENT_SCHEMA_V1.to_owned(),
31            session_id: session_id.into(),
32            snapshot_digest: snapshot_digest.into(),
33            committed_at_ms,
34        };
35        event.validate()?;
36        Ok(event)
37    }
38
39    pub fn validate(&self) -> Result<()> {
40        if self.schema != SESSION_STORE_COMMIT_EVENT_SCHEMA_V1 {
41            bail!("session store commit event schema is unsupported");
42        }
43        if self.session_id.trim().is_empty() {
44            bail!("session store commit event session_id must be non-empty");
45        }
46        if self.snapshot_digest.trim().is_empty() {
47            bail!("session store commit event snapshot_digest must be non-empty");
48        }
49        Ok(())
50    }
51}
52
53/// Subscription to durable snapshot commit notifications.
54pub struct SessionStoreCommitWatch {
55    receiver: broadcast::Receiver<SessionStoreCommitEventV1>,
56}
57
58impl SessionStoreCommitWatch {
59    pub(super) fn new(receiver: broadcast::Receiver<SessionStoreCommitEventV1>) -> Self {
60        Self { receiver }
61    }
62
63    /// Wait for the next committed generation. Lagged receivers skip to the
64    /// newest available event rather than replaying missed ones.
65    pub async fn recv(&mut self) -> Result<SessionStoreCommitEventV1> {
66        loop {
67            match self.receiver.recv().await {
68                Ok(event) => {
69                    event.validate()?;
70                    return Ok(event);
71                }
72                Err(broadcast::error::RecvError::Lagged(_)) => continue,
73                Err(broadcast::error::RecvError::Closed) => {
74                    bail!("session store commit watch closed")
75                }
76            }
77        }
78    }
79}
80
81pub(super) fn commit_watch_channel() -> (
82    broadcast::Sender<SessionStoreCommitEventV1>,
83    broadcast::Receiver<SessionStoreCommitEventV1>,
84) {
85    broadcast::channel(64)
86}
87
88#[cfg(test)]
89mod tests {
90    use super::*;
91
92    #[test]
93    fn commit_event_rejects_empty_identity() {
94        assert!(SessionStoreCommitEventV1::new("", "sha256:aa", 1).is_err());
95        assert!(SessionStoreCommitEventV1::new("s", "", 1).is_err());
96        let ok = SessionStoreCommitEventV1::new("s", "sha256:aa", 1).unwrap();
97        assert_eq!(ok.session_id, "s");
98    }
99}
a3s_code_core/store/watch.rs

a3s_code_core/store/
watch.rs
