diff --git a/lib/crates/fabro-store/src/keys.rs b/lib/crates/fabro-store/src/keys.rs index 2451df632..db5786c79 100644 --- a/lib/crates/fabro-store/src/keys.rs +++ b/lib/crates/fabro-store/src/keys.rs @@ -66,6 +66,16 @@ pub(crate) fn run_event_seq_prefix(run_id: &RunId, seq: u32) -> SlateKey { .with(format!("{seq:06}-")) } +pub(crate) fn main_events_prefix() -> SlateKey { + SlateKey::new("events").with("main").into_prefix() +} + +pub(crate) fn main_event_key(seq: u32, epoch_ms: i64) -> SlateKey { + SlateKey::new("events") + .with("main") + .with(format!("{seq:06}-{epoch_ms}")) +} + pub(crate) fn blobs_prefix() -> SlateKey { SlateKey::new("blobs").with("sha256").into_prefix() } @@ -90,6 +100,17 @@ pub(crate) fn parse_event_seq(key: &str) -> Option { segments.next()?.split_once('-')?.0.parse().ok() } +pub(crate) fn parse_main_event_seq(key: &str) -> Option { + let mut segments = SlateKey::segments(key); + if segments.next()? != "events" { + return None; + } + if segments.next()? != "main" { + return None; + } + segments.next()?.split_once('-')?.0.parse().ok() +} + pub(crate) fn parse_blob_id(key: &str) -> Option { let mut segments = SlateKey::segments(key); if segments.next()? != "blobs" { @@ -136,6 +157,13 @@ mod tests { ]); } + #[test] + fn main_event_key_segments() { + let key = main_event_key(7, 123); + let segments: Vec<&str> = SlateKey::segments(key.as_str()).collect(); + assert_eq!(segments, ["events", "main", "000007-123"]); + } + #[test] fn blob_key_segments() { let blob_id = RunBlobId::new(b"summary"); @@ -165,6 +193,20 @@ mod tests { assert_eq!(parse_blob_id(key.as_str()), Some(blob_id)); } + #[test] + fn parse_main_event_seq_roundtrip() { + assert_eq!(parse_main_event_seq(main_event_key(7, 123).as_str()), Some(7)); + assert_eq!( + parse_main_event_seq( + SlateKey::new("events") + .with("other") + .with("000007-123") + .as_str() + ), + None + ); + } + #[test] fn parse_helpers_reject_invalid_keys() { assert_eq!( diff --git a/lib/crates/fabro-store/src/lib.rs b/lib/crates/fabro-store/src/lib.rs index cfc8e1c0c..cf6809878 100644 --- a/lib/crates/fabro-store/src/lib.rs +++ b/lib/crates/fabro-store/src/lib.rs @@ -28,7 +28,8 @@ pub use run_state::RunProjectionReducer; pub use serializable_projection::SerializableProjection; pub use slate::{ AuthCode, AuthCodeStore, Blob, BlobStore, CachedRunProjection, ConsumeOutcome, Database, - RefreshToken, RefreshTokenStore, RunCatalogIndex, RunDatabase, Runs, UnreadableRun, + MainEventStore, RefreshToken, RefreshTokenStore, RunCatalogIndex, RunDatabase, Runs, + UnreadableRun, }; pub use types::EventPayload; diff --git a/lib/crates/fabro-store/src/slate/main_event_store.rs b/lib/crates/fabro-store/src/slate/main_event_store.rs new file mode 100644 index 000000000..3d85b8e8f --- /dev/null +++ b/lib/crates/fabro-store/src/slate/main_event_store.rs @@ -0,0 +1,125 @@ +use std::sync::Arc; +use std::sync::atomic::{AtomicU32, Ordering}; + +use bytes::Bytes; +use chrono::Utc; +use fabro_types::{MainEvent, MainEventBody, MainEventEnvelope}; +use slatedb::{Db, DbRead}; +use tokio::sync::{Mutex, broadcast}; +use uuid::Uuid; + +use crate::{Error, Result, keys}; + +const DEFAULT_MAIN_EVENT_TAIL_LIMIT: usize = 1024; + +#[derive(Clone)] +pub struct MainEventStore { + inner: Arc, +} + +impl std::fmt::Debug for MainEventStore { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("MainEventStore") + .finish_non_exhaustive() + } +} + +struct MainEventStoreInner { + db: Db, + event_seq: AtomicU32, + append_lock: Mutex<()>, + event_tx: broadcast::Sender, +} + +impl MainEventStore { + pub(crate) async fn open(db: Db) -> Result { + let event_seq = recover_next_seq(&db).await?; + let (event_tx, _) = broadcast::channel(DEFAULT_MAIN_EVENT_TAIL_LIMIT.max(16)); + Ok(Self { + inner: Arc::new(MainEventStoreInner { + db, + event_seq: AtomicU32::new(event_seq), + append_lock: Mutex::new(()), + event_tx, + }), + }) + } + + pub async fn append(&self, body: MainEventBody) -> Result { + let _guard = self.inner.append_lock.lock().await; + let seq = self.inner.event_seq.fetch_add(1, Ordering::SeqCst); + let now = Utc::now(); + let event = MainEvent { + id: Uuid::now_v7().to_string(), + ts: now, + body, + }; + self.inner + .db + .put( + keys::main_event_key(seq, now.timestamp_millis()), + serde_json::to_vec(&event)?, + ) + .await?; + let envelope = MainEventEnvelope { seq, event }; + let _ = self.inner.event_tx.send(envelope.clone()); + Ok(envelope) + } + + pub fn subscribe(&self) -> broadcast::Receiver { + self.inner.event_tx.subscribe() + } + + pub async fn list_from_with_limit( + &self, + start_seq: u32, + limit: usize, + ) -> Result> { + let mut events = list_events_from(&self.inner.db, start_seq).await?; + events.truncate(limit.saturating_add(1)); + Ok(events) + } +} + +async fn recover_next_seq(db: &R) -> Result +where + R: DbRead + Sync, +{ + let mut iter = db.scan_prefix(keys::main_events_prefix()).await?; + let mut max_seq = 0; + while let Some(entry) = iter.next().await? { + let key = key_to_string(&entry.key)?; + if let Some(seq) = keys::parse_main_event_seq(&key) { + max_seq = max_seq.max(seq); + } + } + Ok(max_seq.saturating_add(1).max(1)) +} + +async fn list_events_from(db: &R, start_seq: u32) -> Result> +where + R: DbRead + Sync, +{ + let mut iter = db.scan_prefix(keys::main_events_prefix()).await?; + let mut events = Vec::new(); + while let Some(entry) = iter.next().await? { + let key = key_to_string(&entry.key)?; + let Some(seq) = keys::parse_main_event_seq(&key) else { + continue; + }; + if seq < start_seq { + continue; + } + events.push(MainEventEnvelope { + seq, + event: serde_json::from_slice(&entry.value)?, + }); + } + events.sort_by_key(|event| event.seq); + Ok(events) +} + +fn key_to_string(key: &Bytes) -> Result { + String::from_utf8(key.to_vec()) + .map_err(|err| Error::Other(format!("stored key is not valid UTF-8: {err}"))) +} diff --git a/lib/crates/fabro-store/src/slate/mod.rs b/lib/crates/fabro-store/src/slate/mod.rs index 8cc1bd41d..09fbd82cc 100644 --- a/lib/crates/fabro-store/src/slate/mod.rs +++ b/lib/crates/fabro-store/src/slate/mod.rs @@ -1,6 +1,7 @@ mod auth_codes; mod auth_tokens; mod blob_store; +mod main_event_store; mod projection_cache; mod run_catalog_index; mod run_store; @@ -15,6 +16,7 @@ pub use auth_tokens::{ConsumeOutcome, RefreshToken, RefreshTokenStore}; pub use blob_store::{Blob, BlobStore}; use chrono::{DateTime, Utc}; use fabro_types::{Run, RunId, SessionId}; +pub use main_event_store::MainEventStore; use object_store::ObjectStore; pub use projection_cache::CachedRunProjection; use projection_cache::RunProjectionCache; @@ -49,6 +51,7 @@ pub struct Database { active_runs: Arc>>>, blobs: Arc>>, catalog_index: Arc>>, + main_events: Arc>>, auth_codes: Arc>>, refresh_tokens: Arc>>, projection_cache: Arc, @@ -81,6 +84,7 @@ impl Database { active_runs: Arc::new(Mutex::new(HashMap::new())), blobs: Arc::new(OnceCell::new()), catalog_index: Arc::new(OnceCell::new()), + main_events: Arc::new(OnceCell::new()), auth_codes: Arc::new(OnceCell::new()), refresh_tokens: Arc::new(OnceCell::new()), projection_cache: Arc::new(RunProjectionCache::default()), @@ -410,6 +414,15 @@ impl Database { Ok(Arc::clone(store)) } + pub async fn main_events(&self) -> Result> { + let db = self.open_db().await?; + let store = self + .main_events + .get_or_try_init(|| async { Ok::<_, Error>(Arc::new(MainEventStore::open(db).await?)) }) + .await?; + Ok(Arc::clone(store)) + } + pub async fn refresh_tokens(&self) -> Result> { let store = self .refresh_tokens @@ -471,8 +484,9 @@ fn active_run_from( mod tests { use chrono::{DateTime, Utc}; use fabro_types::{ - AttrValue, FailureReason, Graph, RunControlAction, RunSpec, RunStatus, StageId, - SuccessReason, WorkflowSettings, test_support, + AttrValue, FailureReason, Graph, MainEventBody, MainEventLifecycleProps, MainEventSource, + MainEventStartedProps, RunControlAction, RunSpec, RunStatus, StageId, SuccessReason, + WorkflowSettings, test_support, }; use futures::TryStreamExt; use object_store::memory::InMemory; @@ -608,6 +622,30 @@ mod tests { .unwrap(); } + fn main_event_body(run_id: RunId, source_event_name: &str) -> MainEventBody { + let source = MainEventSource { + run_id, + source_event_id: format!("evt-{source_event_name}"), + source_event_ts: dt("2026-06-21T12:00:00Z"), + source_event_name: source_event_name.to_string(), + actor: None, + }; + match source_event_name { + "run.started" => MainEventBody::RunStarted(MainEventStartedProps { + source, + name: "test".to_string(), + base_branch: None, + base_sha: None, + run_branch: None, + worktree_dir: None, + goal: None, + }), + "run.running" => MainEventBody::RunRunning(MainEventLifecycleProps { source }), + "run.paused" => MainEventBody::RunPaused(MainEventLifecycleProps { source }), + other => panic!("unsupported test main event source {other}"), + } + } + async fn append_created_with_parent( run: &RunDatabase, label: &str, @@ -729,6 +767,95 @@ mod tests { assert!(!list_paths(object_store, "runs/").await.is_empty()); } + #[tokio::test] + async fn main_events_append_list_and_subscribe_in_order() { + let (_object_store, store) = make_store(); + let main_events = store.main_events().await.unwrap(); + let run_id = test_run_id("run-1"); + let mut rx = main_events.subscribe(); + + let first = main_events + .append(main_event_body(run_id, "run.started")) + .await + .unwrap(); + let second = main_events + .append(main_event_body(run_id, "run.running")) + .await + .unwrap(); + + assert_eq!(first.seq, 1); + assert_eq!(second.seq, 2); + assert_eq!(first.event.event_name(), "fabro.run.started"); + + assert_eq!(rx.recv().await.unwrap().seq, 1); + assert_eq!(rx.recv().await.unwrap().seq, 2); + + let listed = main_events.list_from_with_limit(1, 10).await.unwrap(); + assert_eq!( + listed.iter().map(|event| event.seq).collect::>(), + vec![1, 2] + ); + } + + #[tokio::test] + async fn main_events_list_returns_limit_plus_one_from_start_seq() { + let (_object_store, store) = make_store(); + let main_events = store.main_events().await.unwrap(); + let run_id = test_run_id("run-1"); + + for source in ["run.started", "run.running", "run.paused"] { + main_events + .append(main_event_body(run_id, source)) + .await + .unwrap(); + } + + let listed = main_events.list_from_with_limit(2, 1).await.unwrap(); + assert_eq!( + listed.iter().map(|event| event.seq).collect::>(), + vec![2, 3] + ); + } + + #[tokio::test] + async fn main_events_recover_next_sequence_after_reopen() { + let (object_store, store) = make_store(); + let run_id = test_run_id("run-1"); + let main_events = store.main_events().await.unwrap(); + main_events + .append(main_event_body(run_id, "run.started")) + .await + .unwrap(); + + let reopened = Database::new(object_store, "runs", Duration::from_millis(1), None); + let reopened_main_events = reopened.main_events().await.unwrap(); + let appended = reopened_main_events + .append(main_event_body(run_id, "run.running")) + .await + .unwrap(); + + assert_eq!(appended.seq, 2); + } + + #[tokio::test] + async fn delete_run_keeps_main_events() { + let (_object_store, store) = make_store(); + let run = store.create_run(&test_run_id("run-1")).await.unwrap(); + append_created(&run, "run-1", dt("2026-03-27T12:00:00Z")).await; + + let main_events = store.main_events().await.unwrap(); + main_events + .append(main_event_body(test_run_id("run-1"), "run.started")) + .await + .unwrap(); + + store.delete_run(&test_run_id("run-1")).await.unwrap(); + + let listed = main_events.list_from_with_limit(1, 10).await.unwrap(); + assert_eq!(listed.len(), 1); + assert_eq!(listed[0].event.event_name(), "fabro.run.started"); + } + #[tokio::test] async fn delete_run_keeps_global_cas_blobs() { let (_object_store, store) = make_store();