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); 12] = [
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        11,
64        include_str!("../migrations/0011_relation_observations_and_attributes.sql"),
65    ),
66    (
67        12,
68        include_str!("../migrations/0012_rel_obs_fts_delete_fix.sql"),
69    ),
70];
71
72#[cfg(test)]
73mod migration_inventory {
74    /// The one inventory anchor for the migration set. Every other test derives
75    /// its expectation from [`super::MIGRATIONS`], so a registry entry deleted
76    /// by a bad merge, misnumbered, or edited in place would otherwise pass the
77    /// whole suite. Checksums are what production verifies at every startup, so
78    /// pinning them here pins count, order and content together.
79    #[test]
80    fn every_migration_version_and_checksum_is_pinned() {
81        let inventory: Vec<(i64, String)> = super::MIGRATIONS
82            .iter()
83            .map(|(version, sql)| (*version, super::sha256(sql.as_bytes())))
84            .collect();
85        assert_eq!(
86            inventory,
87            vec![
88                (
89                    1,
90                    "a48def8b25e9ecf813d3fa27a785893ba5de346a8b82f012cc543af2fecd5af2".to_string()
91                ),
92                (
93                    2,
94                    "2c267315d89203d5223895a845b275d8add904310d2c0a5a7a5b0d1873b984ec".to_string()
95                ),
96                (
97                    3,
98                    "0b82809aca1b4e90edfea4796e33a9c32963e055b7b57b8524b86dcb580bd454".to_string()
99                ),
100                (
101                    4,
102                    "5d18e23d999661dda33bfd2358659530760afec0a079c3a1b96689b537af28a2".to_string()
103                ),
104                (
105                    5,
106                    "9e40e5c633bef1facda4361563d4c7952f4e1a8d86d11a4c20844bf1ff63b412".to_string()
107                ),
108                (
109                    6,
110                    "f57f6103c82269627ada7de2cea989a2494aabdb066eb6f9775e3ea6d32ed564".to_owned()
111                ),
112                (
113                    7,
114                    "4949f6c6f73c22bce5de31219d2d5d52739cde05967363c2fc89e88b35c044e8".to_string()
115                ),
116                (
117                    8,
118                    "9e0721309bf535e79ccf8561c7556f663dda3b6ffd8eb176973630ebf7d39a54".to_string()
119                ),
120                (
121                    9,
122                    "22810c6a00ba60b3360b4ada74c3e33519a654d1e5f94b890718dd057e373acb".to_string()
123                ),
124                (
125                    10,
126                    "fc42f71ef3edc7e6ccb7a695456d519e394c9ab92de8f2078bdc3b60c4f4bca2".to_string()
127                ),
128                (
129                    11,
130                    "11456cb24b55e78a80c3e05ff37f4fd3411f9d16c5898132f7b783109f5474a8".to_string()
131                ),
132                (
133                    12,
134                    "2a713d3eb83bc75449063b94cf7089b2361b3b92612065407ee063f9468f8a2c".to_string()
135                ),
136            ],
137            "a migration was added, removed, renumbered or edited"
138        );
139    }
140}
141
142/// Apply all pending ordered migrations in one transaction, including their
143/// ledger entries. Any failure rolls the entire pending set back; historical
144/// checksums are verified even when no migrations remain to apply.
145///
146/// Startup callers use [`crate::schema::initialize_database`] to establish the
147/// legacy graph tables and statistics before these migrations can reference them.
148pub fn migrate(conn: &Connection) -> Result<()> {
149    let tx = TxGuard::begin(conn)?;
150    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)?;
151    let migrations = MIGRATIONS;
152    let newest: i64 = conn
153        .query_row(
154            "SELECT coalesce(max(version),0) FROM schema_migration",
155            [],
156            |r| r.get(0),
157        )
158        .map_err(sql_error)?;
159    if newest > migrations.last().map_or(0, |(version, _)| *version) {
160        return Err(MCSError::MemoryError(
161            "database schema is newer than this binary".into(),
162        ));
163    }
164    for (version, sql) in migrations {
165        let checksum = sha256(sql.as_bytes());
166        let existing: Option<String> = conn
167            .query_row(
168                "SELECT checksum FROM schema_migration WHERE version=?1",
169                [version],
170                |r| r.get(0),
171            )
172            .optional()
173            .map_err(sql_error)?;
174        match existing {
175            Some(existing) if existing != checksum => {
176                return Err(MCSError::MemoryError(format!(
177                    "migration {version} checksum mismatch"
178                )));
179            }
180            Some(_) => {}
181            None => {
182                conn.execute_batch(sql).map_err(sql_error)?;
183                conn.execute(
184                    "INSERT INTO schema_migration VALUES(?1,?2,?3)",
185                    params![version, checksum, now_us()],
186                )
187                .map_err(sql_error)?;
188            }
189        }
190    }
191    tx.commit()
192}
193
194#[derive(Clone, Debug, Serialize, Deserialize)]
195#[serde(rename_all = "camelCase")]
196pub struct ChangeEvent {
197    pub event_id: Uuid,
198    pub transaction_id: Uuid,
199    pub entity_id: i64,
200    pub entity_revision: i64,
201    pub occurred_at_us: i64,
202    pub change: EntityChange,
203    pub provenance: MutationContext,
204}
205
206pub(crate) fn persist_changes(
207    conn: &Connection,
208    changes: &CommittedChangeSet,
209    context: &MutationContext,
210) -> Result<()> {
211    for change in &changes.changes {
212        let snapshot = change
213            .after
214            .as_ref()
215            .or(change.before.as_ref())
216            .ok_or_else(|| MCSError::MemoryError("empty entity change".into()))?;
217        let deleted = change.after.is_none();
218        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)?;
219        let event = ChangeEvent {
220            event_id: Uuid::new_v4(),
221            transaction_id: changes.transaction_id,
222            entity_id: snapshot.entity_id,
223            entity_revision: revision,
224            occurred_at_us: now_us(),
225            change: change.clone(),
226            provenance: context.clone(),
227        };
228        conn.execute(
229            "INSERT INTO change_event VALUES(?1,?2,?3,?4,?5,?6)",
230            params![
231                event.event_id.to_string(),
232                event.transaction_id.to_string(),
233                event.entity_id,
234                revision,
235                event.occurred_at_us,
236                serde_json::to_string(&event)?
237            ],
238        )
239        .map_err(sql_error)?;
240        crate::jobs::enqueue_change(conn, snapshot.entity_id, revision, deleted)?;
241        crate::subscriptions::SubscriptionRepository::new(conn).enqueue_matching(&event)?;
242    }
243    Ok(())
244}
245
246#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
247pub struct Lease {
248    pub token: Uuid,
249    pub epoch: i64,
250    pub until_us: i64,
251}
252
253pub(crate) fn lease_until(now: i64, duration_us: i64) -> Result<i64> {
254    if duration_us <= 0 || duration_us > 3_600_000_000 {
255        return Err(MCSError::InvalidParams(
256            "lease duration must be 1us..1h".into(),
257        ));
258    }
259    now.checked_add(duration_us)
260        .ok_or_else(|| MCSError::InvalidParams("lease time overflow".into()))
261}
262
263#[derive(Clone, Debug, Serialize, Deserialize)]
264pub struct EventDelivery {
265    pub delivery_id: Uuid,
266    pub subscription_id: Uuid,
267    pub event: ChangeEvent,
268    pub lease: Lease,
269    pub attempts: i64,
270}
271
272pub struct EventRepository<'a> {
273    conn: &'a Connection,
274}
275
276impl<'a> EventRepository<'a> {
277    pub const fn new(conn: &'a Connection) -> Self {
278        Self { conn }
279    }
280
281    pub fn get(&self, event_id: Uuid) -> Result<Option<ChangeEvent>> {
282        let payload: Option<String> = self
283            .conn
284            .query_row(
285                "SELECT payload FROM change_event WHERE event_id=?1",
286                [event_id.to_string()],
287                |r| r.get(0),
288            )
289            .optional()
290            .map_err(sql_error)?;
291        payload
292            .map(|text| serde_json::from_str(&text).map_err(Into::into))
293            .transpose()
294    }
295
296    pub fn enqueue_delivery(&self, event_id: Uuid, subscription_id: Uuid) -> Result<()> {
297        let tx = TxGuard::begin(self.conn)?;
298        if self.get(event_id)?.is_none() {
299            return Err(MCSError::InvalidParams("unknown event".into()));
300        }
301        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)?;
302        tx.commit()
303    }
304
305    /// At most one active delivery per subscription, including claims made by
306    /// other processes. Expired tokens are superseded by a strictly newer epoch.
307    pub fn claim_due(&self, now: i64, duration_us: i64) -> Result<Option<EventDelivery>> {
308        let until = lease_until(now, duration_us)?;
309        let tx = TxGuard::begin(self.conn)?;
310        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)?;
311        let result = row.map(|(delivery, subscription, event, epoch, attempts)| -> Result<EventDelivery> {
312            let token = Uuid::new_v4();
313            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)?;
314            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 })
315        }).transpose()?;
316        tx.commit()?;
317        Ok(result)
318    }
319
320    pub fn complete(&self, delivery: &EventDelivery, now: i64) -> Result<bool> {
321        let tx = TxGuard::begin(self.conn)?;
322        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)?;
323        tx.commit()?;
324        Ok(changed == 1)
325    }
326
327    pub fn retry(
328        &self,
329        delivery: &EventDelivery,
330        now: i64,
331        next_attempt_us: i64,
332        error: &str,
333        dead: bool,
334    ) -> Result<bool> {
335        let tx = TxGuard::begin(self.conn)?;
336        let error: String = error.chars().take(2048).collect();
337        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)?;
338        tx.commit()?;
339        Ok(changed == 1)
340    }
341}
342
343pub(crate) fn parse_uuid(value: &str) -> Result<Uuid> {
344    Uuid::parse_str(value)
345        .map_err(|error| MCSError::MemoryError(format!("invalid persisted UUID: {error}")))
346}
mcpmem_core/events.rs

mcpmem_core/
events.rs
