mcpmem_core/
events.rs

1//! Durable change log and delivery leases. No network work occurs here.
2use rusqlite::{Connection, OptionalExtension, params};
3use serde::{Deserialize, Serialize};
4use sha2::{Digest, Sha256};
5use uuid::Uuid;
6
7use crate::errors::{MCSError, Result};
8use crate::graph::TxGuard;
9use crate::mutation::{CommittedChangeSet, EntityChange, MutationContext};
10
11pub fn sql_error(error: rusqlite::Error) -> MCSError {
12    MCSError::IoError(std::io::Error::other(error))
13}
14
15pub fn now_us() -> i64 {
16    std::time::SystemTime::now()
17        .duration_since(std::time::UNIX_EPOCH)
18        .unwrap_or_default()
19        .as_micros()
20        .min(i64::MAX as u128) as i64
21}
22
23pub fn sha256(bytes: &[u8]) -> String {
24    format!("{:x}", Sha256::digest(bytes))
25}
26
27/// Method, normalized path and exact ingress bytes are length-delimited so
28/// neither separators inside a body nor JSON reserialization can collide.
29pub fn request_fingerprint(method: &str, normalized_path: &str, raw_body: &[u8]) -> String {
30    let mut hash = Sha256::new();
31    for part in [method.as_bytes(), normalized_path.as_bytes(), raw_body] {
32        hash.update((part.len() as u64).to_be_bytes());
33        hash.update(part);
34    }
35    format!("{hash:x}", hash = hash.finalize())
36}
37
38/// The ordered migration set, embedded at compile time.
39///
40/// Public because a test that builds a historical database needs the exact SQL
41/// and its checksum. It must not read the `.sql` file itself: the files belong
42/// to this crate, and `cargo package` copies only the files under one crate
43/// root, so an `include_str!` from another crate ships a crate that cannot
44/// compile. That is how the `v1.0.0-rc.1` release failed.
45pub const MIGRATIONS: [(i64, &str); 10] = [
46    (1, include_str!("../migrations/0001_change_events.sql")),
47    (
48        2,
49        include_str!("../migrations/0002_webhook_subscriptions.sql"),
50    ),
51    (
52        3,
53        include_str!("../migrations/0003_observation_metadata.sql"),
54    ),
55    (4, include_str!("../migrations/0004_oauth.sql")),
56    (5, include_str!("../migrations/0005_principals.sql")),
57    (6, include_str!("../migrations/0006_code_repos.sql")),
58    (7, include_str!("../migrations/0007_taxonomy_index.sql")),
59    (8, include_str!("../migrations/0008_type_descriptions.sql")),
60    (9, include_str!("../migrations/0009_chunked_embeddings.sql")),
61    (10, include_str!("../migrations/0010_embedding_cleanup.sql")),
62];
63
64#[cfg(test)]
65mod migration_inventory {
66    /// The one inventory anchor for the migration set. Every other test derives
67    /// its expectation from [`super::MIGRATIONS`], so a registry entry deleted
68    /// by a bad merge, misnumbered, or edited in place would otherwise pass the
69    /// whole suite. Checksums are what production verifies at every startup, so
70    /// pinning them here pins count, order and content together.
71    #[test]
72    fn every_migration_version_and_checksum_is_pinned() {
73        let inventory: Vec<(i64, String)> = super::MIGRATIONS
74            .iter()
75            .map(|(version, sql)| (*version, super::sha256(sql.as_bytes())))
76            .collect();
77        assert_eq!(
78            inventory,
79            vec![
80                (
81                    1,
82                    "a48def8b25e9ecf813d3fa27a785893ba5de346a8b82f012cc543af2fecd5af2".to_string()
83                ),
84                (
85                    2,
86                    "2c267315d89203d5223895a845b275d8add904310d2c0a5a7a5b0d1873b984ec".to_string()
87                ),
88                (
89                    3,
90                    "0b82809aca1b4e90edfea4796e33a9c32963e055b7b57b8524b86dcb580bd454".to_string()
91                ),
92                (
93                    4,
94                    "5d18e23d999661dda33bfd2358659530760afec0a079c3a1b96689b537af28a2".to_string()
95                ),
96                (
97                    5,
98                    "9e40e5c633bef1facda4361563d4c7952f4e1a8d86d11a4c20844bf1ff63b412".to_string()
99                ),
100                (
101                    6,
102                    "f57f6103c82269627ada7de2cea989a2494aabdb066eb6f9775e3ea6d32ed564".to_owned()
103                ),
104                (
105                    7,
106                    "4949f6c6f73c22bce5de31219d2d5d52739cde05967363c2fc89e88b35c044e8".to_string()
107                ),
108                (
109                    8,
110                    "9e0721309bf535e79ccf8561c7556f663dda3b6ffd8eb176973630ebf7d39a54".to_string()
111                ),
112                (
113                    9,
114                    "22810c6a00ba60b3360b4ada74c3e33519a654d1e5f94b890718dd057e373acb".to_string()
115                ),
116                (
117                    10,
118                    "fc42f71ef3edc7e6ccb7a695456d519e394c9ab92de8f2078bdc3b60c4f4bca2".to_string()
119                ),
120            ],
121            "a migration was added, removed, renumbered or edited"
122        );
123    }
124}
125
126/// Apply all pending ordered migrations in one transaction, including their
127/// ledger entries. Any failure rolls the entire pending set back; historical
128/// checksums are verified even when no migrations remain to apply.
129///
130/// Startup callers use [`crate::schema::initialize_database`] to establish the
131/// legacy graph tables and statistics before these migrations can reference them.
132pub fn migrate(conn: &Connection) -> Result<()> {
133    let tx = TxGuard::begin(conn)?;
134    conn.execute_batch("CREATE TABLE IF NOT EXISTS schema_migration(version INTEGER PRIMARY KEY, checksum TEXT NOT NULL, applied_at_us INTEGER NOT NULL) STRICT;").map_err(sql_error)?;
135    let migrations = MIGRATIONS;
136    let newest: i64 = conn
137        .query_row(
138            "SELECT coalesce(max(version),0) FROM schema_migration",
139            [],
140            |r| r.get(0),
141        )
142        .map_err(sql_error)?;
143    if newest > migrations.last().map_or(0, |(version, _)| *version) {
144        return Err(MCSError::MemoryError(
145            "database schema is newer than this binary".into(),
146        ));
147    }
148    for (version, sql) in migrations {
149        let checksum = sha256(sql.as_bytes());
150        let existing: Option<String> = conn
151            .query_row(
152                "SELECT checksum FROM schema_migration WHERE version=?1",
153                [version],
154                |r| r.get(0),
155            )
156            .optional()
157            .map_err(sql_error)?;
158        match existing {
159            Some(existing) if existing != checksum => {
160                return Err(MCSError::MemoryError(format!(
161                    "migration {version} checksum mismatch"
162                )));
163            }
164            Some(_) => {}
165            None => {
166                conn.execute_batch(sql).map_err(sql_error)?;
167                conn.execute(
168                    "INSERT INTO schema_migration VALUES(?1,?2,?3)",
169                    params![version, checksum, now_us()],
170                )
171                .map_err(sql_error)?;
172            }
173        }
174    }
175    tx.commit()
176}
177
178#[derive(Clone, Debug, Serialize, Deserialize)]
179#[serde(rename_all = "camelCase")]
180pub struct ChangeEvent {
181    pub event_id: Uuid,
182    pub transaction_id: Uuid,
183    pub entity_id: i64,
184    pub entity_revision: i64,
185    pub occurred_at_us: i64,
186    pub change: EntityChange,
187    pub provenance: MutationContext,
188}
189
190pub(crate) fn persist_changes(
191    conn: &Connection,
192    changes: &CommittedChangeSet,
193    context: &MutationContext,
194) -> Result<()> {
195    for change in &changes.changes {
196        let snapshot = change
197            .after
198            .as_ref()
199            .or(change.before.as_ref())
200            .ok_or_else(|| MCSError::MemoryError("empty entity change".into()))?;
201        let deleted = change.after.is_none();
202        let revision: i64 = conn.query_row("INSERT INTO entity_revision VALUES(?1,1,?2) ON CONFLICT(entity_id) DO UPDATE SET revision=revision+1, deleted=excluded.deleted RETURNING revision", params![snapshot.entity_id, deleted], |r| r.get(0)).map_err(sql_error)?;
203        let event = ChangeEvent {
204            event_id: Uuid::new_v4(),
205            transaction_id: changes.transaction_id,
206            entity_id: snapshot.entity_id,
207            entity_revision: revision,
208            occurred_at_us: now_us(),
209            change: change.clone(),
210            provenance: context.clone(),
211        };
212        conn.execute(
213            "INSERT INTO change_event VALUES(?1,?2,?3,?4,?5,?6)",
214            params![
215                event.event_id.to_string(),
216                event.transaction_id.to_string(),
217                event.entity_id,
218                revision,
219                event.occurred_at_us,
220                serde_json::to_string(&event)?
221            ],
222        )
223        .map_err(sql_error)?;
224        crate::jobs::enqueue_change(conn, snapshot.entity_id, revision, deleted)?;
225        crate::subscriptions::SubscriptionRepository::new(conn).enqueue_matching(&event)?;
226    }
227    Ok(())
228}
229
230#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
231pub struct Lease {
232    pub token: Uuid,
233    pub epoch: i64,
234    pub until_us: i64,
235}
236
237pub(crate) fn lease_until(now: i64, duration_us: i64) -> Result<i64> {
238    if duration_us <= 0 || duration_us > 3_600_000_000 {
239        return Err(MCSError::InvalidParams(
240            "lease duration must be 1us..1h".into(),
241        ));
242    }
243    now.checked_add(duration_us)
244        .ok_or_else(|| MCSError::InvalidParams("lease time overflow".into()))
245}
246
247#[derive(Clone, Debug, Serialize, Deserialize)]
248pub struct EventDelivery {
249    pub delivery_id: Uuid,
250    pub subscription_id: Uuid,
251    pub event: ChangeEvent,
252    pub lease: Lease,
253    pub attempts: i64,
254}
255
256pub struct EventRepository<'a> {
257    conn: &'a Connection,
258}
259
260impl<'a> EventRepository<'a> {
261    pub const fn new(conn: &'a Connection) -> Self {
262        Self { conn }
263    }
264
265    pub fn get(&self, event_id: Uuid) -> Result<Option<ChangeEvent>> {
266        let payload: Option<String> = self
267            .conn
268            .query_row(
269                "SELECT payload FROM change_event WHERE event_id=?1",
270                [event_id.to_string()],
271                |r| r.get(0),
272            )
273            .optional()
274            .map_err(sql_error)?;
275        payload
276            .map(|text| serde_json::from_str(&text).map_err(Into::into))
277            .transpose()
278    }
279
280    pub fn enqueue_delivery(&self, event_id: Uuid, subscription_id: Uuid) -> Result<()> {
281        let tx = TxGuard::begin(self.conn)?;
282        if self.get(event_id)?.is_none() {
283            return Err(MCSError::InvalidParams("unknown event".into()));
284        }
285        self.conn.execute("INSERT INTO event_outbox(delivery_id,event_id,subscription_id) VALUES(?1,?2,?3) ON CONFLICT(event_id,subscription_id) DO NOTHING", params![Uuid::new_v4().to_string(), event_id.to_string(), subscription_id.to_string()]).map_err(sql_error)?;
286        tx.commit()
287    }
288
289    /// At most one active delivery per subscription, including claims made by
290    /// other processes. Expired tokens are superseded by a strictly newer epoch.
291    pub fn claim_due(&self, now: i64, duration_us: i64) -> Result<Option<EventDelivery>> {
292        let until = lease_until(now, duration_us)?;
293        let tx = TxGuard::begin(self.conn)?;
294        let row: Option<(String,String,String,i64,i64)> = self.conn.query_row("SELECT delivery_id,subscription_id,event_id,lease_epoch,attempts FROM event_outbox e WHERE ((state='pending' AND next_attempt_us<=?1) OR (state='leased' AND lease_until_us<=?1)) AND NOT EXISTS(SELECT 1 FROM event_outbox busy WHERE busy.subscription_id=e.subscription_id AND busy.state='leased' AND busy.lease_until_us>?1) ORDER BY next_attempt_us,delivery_id LIMIT 1", [now], |r| Ok((r.get(0)?,r.get(1)?,r.get(2)?,r.get(3)?,r.get(4)?))).optional().map_err(sql_error)?;
295        let result = row.map(|(delivery, subscription, event, epoch, attempts)| -> Result<EventDelivery> {
296            let token = Uuid::new_v4();
297            self.conn.execute("UPDATE event_outbox SET state='leased',lease_token=?2,lease_epoch=lease_epoch+1,lease_until_us=?3,attempts=attempts+1 WHERE delivery_id=?1", params![delivery,token.to_string(),until]).map_err(sql_error)?;
298            Ok(EventDelivery { delivery_id: parse_uuid(&delivery)?, subscription_id: parse_uuid(&subscription)?, event: self.get(parse_uuid(&event)?)?.ok_or_else(|| MCSError::MemoryError("delivery event missing".into()))?, lease: Lease { token,epoch:epoch+1,until_us:until }, attempts:attempts+1 })
299        }).transpose()?;
300        tx.commit()?;
301        Ok(result)
302    }
303
304    pub fn complete(&self, delivery: &EventDelivery, now: i64) -> Result<bool> {
305        let tx = TxGuard::begin(self.conn)?;
306        let changed = self.conn.execute("UPDATE event_outbox SET state='done' WHERE delivery_id=?1 AND lease_token=?2 AND lease_epoch=?3 AND (state='done' OR (state='leased' AND lease_until_us>?4))", params![delivery.delivery_id.to_string(),delivery.lease.token.to_string(),delivery.lease.epoch,now]).map_err(sql_error)?;
307        tx.commit()?;
308        Ok(changed == 1)
309    }
310
311    pub fn retry(
312        &self,
313        delivery: &EventDelivery,
314        now: i64,
315        next_attempt_us: i64,
316        error: &str,
317        dead: bool,
318    ) -> Result<bool> {
319        let tx = TxGuard::begin(self.conn)?;
320        let error: String = error.chars().take(2048).collect();
321        let changed = self.conn.execute("UPDATE event_outbox SET state=?5,next_attempt_us=?6,last_error=?7 WHERE delivery_id=?1 AND lease_token=?2 AND lease_epoch=?3 AND state='leased' AND lease_until_us>?4", params![delivery.delivery_id.to_string(),delivery.lease.token.to_string(),delivery.lease.epoch,now,if dead {"dead"} else {"pending"},next_attempt_us,error]).map_err(sql_error)?;
322        tx.commit()?;
323        Ok(changed == 1)
324    }
325}
326
327pub(crate) fn parse_uuid(value: &str) -> Result<Uuid> {
328    Uuid::parse_str(value)
329        .map_err(|error| MCSError::MemoryError(format!("invalid persisted UUID: {error}")))
330}
mcpmem_core/events.rs

mcpmem_core/
events.rs
