diff --git a/lib/components/fabro-store/src/legacy_run_history_import.rs b/lib/components/fabro-store/src/legacy_run_history_import.rs index dfad82b55..b253248e5 100644 --- a/lib/components/fabro-store/src/legacy_run_history_import.rs +++ b/lib/components/fabro-store/src/legacy_run_history_import.rs @@ -8,7 +8,7 @@ use std::collections::HashSet; use std::error::Error as StdError; use std::fmt; -use fabro_types::{EventEnvelope, RunEvent, RunId, RunProjection}; +use fabro_types::{EventEnvelope, RunEvent, RunId}; use sha2::{Digest as _, Sha256}; use sqlx::SqlitePool; #[cfg(test)] @@ -16,8 +16,8 @@ use tokio::sync::Barrier; use tracing::debug; use crate::keys::SlateKey; -use crate::slate::CachedRunProjection; -use crate::{Database, EventPayload, RunProjectionReducer, RunSummaryStore, keys}; +use crate::run_state::ProjectedRun; +use crate::{Database, EventPayload, RunSummaryStore, keys}; /// Count-only observations about the legacy catalog and session indexes. #[derive(Clone, Copy, Debug, Default, Eq, PartialEq)] @@ -458,7 +458,7 @@ struct ValidatedLegacyRunEvent { struct ValidatedLegacyRunHistory { run_id: RunId, events: Vec, - current: CachedRunProjection, + current: ProjectedRun, } struct LegacyRunHistorySource { @@ -527,14 +527,8 @@ impl LegacyRunHistorySource { .iter() .map(|event| event.envelope.clone()) .collect::>(); - let projection = RunProjection::apply_events(&envelopes) + let current = ProjectedRun::replay(run_id, &envelopes) .map_err(LegacyRunHistorySourceFailure::Replay)?; - let last_seq = events - .last() - .expect("a validated history contains at least one event") - .envelope - .seq; - let current = CachedRunProjection::from_projection(run_id, projection, last_seq); Ok(Some(ValidatedLegacyRunHistory { run_id, events, @@ -1159,31 +1153,21 @@ fn require_exact_prefix( fn replay_destination( run_id: &RunId, events: &[(EventEnvelope, String)], -) -> crate::Result { - let Some((first, _event_json)) = events.first() else { - return Err(crate::Error::InvalidEvent( - "run projection requires an event".to_owned(), - )); - }; - if first.seq != 1 { - return Err(crate::Error::RunEventMismatch { - run_id: run_id.to_string(), - seq: first.seq, - field: "seq", - }); +) -> crate::Result { + if let Some((first, _event_json)) = events.first() { + if first.seq != 1 { + return Err(crate::Error::RunEventMismatch { + run_id: run_id.to_string(), + seq: first.seq, + field: "seq", + }); + } } let envelopes = events .iter() .map(|(envelope, _event_json)| envelope.clone()) .collect::>(); - let projection = RunProjection::apply_events(&envelopes)?; - let last_seq = envelopes - .last() - .expect("a destination history validated as nonempty") - .seq; - Ok(CachedRunProjection::from_projection( - *run_id, projection, last_seq, - )) + ProjectedRun::replay(*run_id, &envelopes) } fn usize_to_import_count(value: usize) -> Result { @@ -1258,9 +1242,7 @@ mod tests { use std::time::Duration; use chrono::{TimeZone as _, Utc}; - use fabro_types::{ - Graph, RunEvent, RunId, RunProjection, SessionId, WorkflowSettings, test_support, - }; + use fabro_types::{Graph, RunEvent, RunId, SessionId, WorkflowSettings, test_support}; use fabro_util::error; use object_store::memory::InMemory; use tokio::sync::Barrier; @@ -1273,9 +1255,9 @@ mod tests { parse_source_event, }; use crate::keys::SlateKey; - use crate::slate::CachedRunProjection; + use crate::run_state::ProjectedRun; use crate::{ - Database, EventEnvelope, EventPayload, RunProjectionReducer, RunSummaryStore, keys, + Database, EventEnvelope, EventPayload, RunSummaryStore, keys, test_support as store_test_support, }; @@ -1478,9 +1460,7 @@ mod tests { .iter() .map(|(_payload, envelope)| envelope.clone()) .collect::>(); - let projection = RunProjection::apply_events(&envelopes)?; - let current = - CachedRunProjection::from_projection(*run_id, projection, events.last().unwrap().0); + let current = ProjectedRun::replay(*run_id, &envelopes)?; let mut transaction = pool.begin().await?; RunSummaryStore::insert_imported_run_on_connection(&mut transaction, ¤t).await?; for ((_, event_json), (payload, envelope)) in events.iter().zip(&decoded) { @@ -1513,8 +1493,7 @@ mod tests { .iter() .map(|(_payload, envelope)| envelope.clone()) .collect::>(); - let projection = RunProjection::apply_events(&envelopes)?; - let current = CachedRunProjection::from_projection(*run_id, projection, next.0); + let current = ProjectedRun::replay(*run_id, &envelopes)?; let mut transaction = pool.begin().await?; RunSummaryStore::append_event_on_connection( &mut transaction, diff --git a/lib/components/fabro-store/src/run_state.rs b/lib/components/fabro-store/src/run_state.rs index 28780c1a5..a6a4a12a7 100644 --- a/lib/components/fabro-store/src/run_state.rs +++ b/lib/components/fabro-store/src/run_state.rs @@ -33,6 +33,45 @@ pub(crate) struct EventProjectionCache { pub state: Option>, } +/// A run's projection at its committed head. `RunSummaryStore` replays it +/// from SQLite history and derives summary rows from it; `RunDatabase` builds +/// it from a newly committed event and seeds its in-memory cache with it. +#[derive(Debug, Clone)] +pub(crate) struct ProjectedRun { + pub(crate) run_id: RunId, + pub(crate) projection: Arc, + pub(crate) last_seq: u32, +} + +impl ProjectedRun { + pub(crate) fn new(run_id: RunId, projection: Arc, last_seq: u32) -> Self { + Self { + run_id, + projection, + last_seq, + } + } + + /// Replays a run's full history; the last event's `seq` becomes the head. + pub(crate) fn replay(run_id: RunId, events: &[EventEnvelope]) -> Result { + let projection = RunProjection::apply_events(events)?; + let last_seq = events + .last() + .expect("a successfully replayed history contains at least one event") + .seq; + Ok(Self::new(run_id, Arc::new(projection), last_seq)) + } +} + +impl From for EventProjectionCache { + fn from(projected: ProjectedRun) -> Self { + Self { + last_seq: projected.last_seq, + state: Some(projected.projection), + } + } +} + pub trait RunProjectionReducer { fn apply_events(events: &[EventEnvelope]) -> Result where diff --git a/lib/components/fabro-store/src/run_summary_store.rs b/lib/components/fabro-store/src/run_summary_store.rs index 97e07eee0..5181d9be5 100644 --- a/lib/components/fabro-store/src/run_summary_store.rs +++ b/lib/components/fabro-store/src/run_summary_store.rs @@ -12,8 +12,7 @@ use sqlx::sqlite::{SqliteArguments, SqliteConnection, SqliteRow}; use sqlx::{Connection as _, QueryBuilder, Row as _, Sqlite, SqlitePool, Transaction}; use strum::VariantArray as _; -use crate::run_state::{build_summary, projected_billing}; -use crate::slate::CachedRunProjection; +use crate::run_state::{ProjectedRun, build_summary, projected_billing}; use crate::{Error, EventPayload, Result, keys}; const INSERT_RUN_SQL: &str = r" @@ -288,14 +287,14 @@ ON CONFLICT(singleton) DO NOTHING } #[cfg(test)] - pub(crate) async fn upsert_projection(&self, entry: &CachedRunProjection) -> Result<()> { + pub(crate) async fn upsert_projection(&self, entry: &ProjectedRun) -> Result<()> { let record = PreparedRunSummary::from_entry(entry); let mut connection = self.pool.acquire().await?; upsert_run_on_connection(&mut connection, &record).await } #[cfg(test)] - pub(crate) async fn reconcile(&self, entries: &[CachedRunProjection]) -> Result<()> { + pub(crate) async fn reconcile(&self, entries: &[ProjectedRun]) -> Result<()> { use std::collections::{HashMap, HashSet}; let mut transaction = self.pool.begin().await?; @@ -351,6 +350,13 @@ ON CONFLICT(singleton) DO NOTHING select_run_head(&mut connection, run_id).await } + /// Replays one run's canonical history from one validated SQLite snapshot. + /// Fails with `RunNotFound` when the run does not exist. + pub(crate) async fn load_projection(&self, run_id: &RunId) -> Result { + let events = self.list_events_for_run(run_id).await?; + ProjectedRun::replay(*run_id, &events) + } + pub(crate) async fn list_events_for_run(&self, run_id: &RunId) -> Result> { let mut connection = self.acquire().await?; Self::list_events_on_connection(&mut connection, run_id).await @@ -638,7 +644,7 @@ FROM runs", impl RunSummaryStore { pub(crate) async fn insert_first_event_on_connection( connection: &mut SqliteConnection, - entry: &CachedRunProjection, + entry: &ProjectedRun, payload: &EventPayload, ) -> Result { let record = PreparedRunSummary::from_entry(entry); @@ -657,7 +663,7 @@ impl RunSummaryStore { pub(crate) async fn append_event_on_connection( connection: &mut SqliteConnection, expected_last_seq: u32, - entry: &CachedRunProjection, + entry: &ProjectedRun, payload: &EventPayload, ) -> Result { let next_seq = next_event_seq_after(expected_last_seq)?; @@ -734,7 +740,7 @@ impl RunSummaryStore { pub(crate) async fn insert_imported_run_on_connection( connection: &mut SqliteConnection, - entry: &CachedRunProjection, + entry: &ProjectedRun, ) -> Result<()> { let record = PreparedRunSummary::from_entry(entry); ensure_entry_identity(entry, &record, entry.last_seq)?; @@ -768,7 +774,7 @@ impl RunSummaryStore { pub(crate) async fn verify_current_run_on_connection( connection: &mut SqliteConnection, - entry: &CachedRunProjection, + entry: &ProjectedRun, ) -> Result<()> { let record = PreparedRunSummary::from_entry(entry); ensure_entry_identity(entry, &record, entry.last_seq)?; @@ -1004,7 +1010,7 @@ struct PreparedRunSummary { } impl PreparedRunSummary { - fn from_entry(entry: &CachedRunProjection) -> Self { + fn from_entry(entry: &ProjectedRun) -> Self { let mut run = build_summary(&entry.projection, &entry.run_id); if run.timing.is_none() { let at = run @@ -1036,7 +1042,7 @@ impl PreparedRunSummary { } fn ensure_entry_identity( - entry: &CachedRunProjection, + entry: &ProjectedRun, record: &PreparedRunSummary, seq: u32, ) -> Result<()> { @@ -1527,6 +1533,7 @@ fn overlay_live_wall_time(run: &mut Run, now: DateTime) { #[cfg(test)] mod tests { use std::collections::HashMap; + use std::sync::Arc; use std::time::Duration; use chrono::{DateTime, Utc}; @@ -1544,8 +1551,8 @@ mod tests { INSERT_EVENT_SQL, RunSummaryListQuery, RunSummarySort, RunSummarySortDirection, RunSummaryStore, RunSummaryVisibility, decode_event_row, }; - use crate::slate::CachedRunProjection; - use crate::{Error, EventPayload, test_support as store_test_support}; + use crate::run_state::ProjectedRun; + use crate::{Error, EventPayload, RunProjectionReducer, test_support as store_test_support}; fn dt(value: &str) -> DateTime { value.parse().unwrap() @@ -1580,8 +1587,8 @@ mod tests { ) } - fn entry(projection: RunProjection, last_seq: u32) -> CachedRunProjection { - CachedRunProjection::from_projection(projection.spec.run_id, projection, last_seq) + fn entry(projection: RunProjection, last_seq: u32) -> ProjectedRun { + ProjectedRun::new(projection.spec.run_id, Arc::new(projection), last_seq) } async fn store() -> (tempfile::TempDir, RunSummaryStore) { @@ -1920,6 +1927,88 @@ mod tests { assert_eq!(sequences, vec![1, 2, 3]); } + #[tokio::test] + async fn load_projection_replays_committed_events_and_reports_missing_run() { + let (_directory, store) = store().await; + let created_at = dt("2026-08-27T12:00:00Z"); + let id = run_id(created_at.timestamp_millis().cast_unsigned(), 31); + let first = entry(projection(id, "created", created_at), 1); + let first_payload = created_payload(&id); + + let mut transaction = store.pool.begin().await.unwrap(); + let first_envelope = RunSummaryStore::insert_first_event_on_connection( + &mut transaction, + &first, + &first_payload, + ) + .await + .unwrap(); + transaction.commit().await.unwrap(); + + let second = entry(projection(id, "updated", created_at), 2); + let second_payload = sql_event_payload( + &id, + "run.title.updated", + None, + None, + None, + serde_json::json!({ "title": "updated" }), + ); + let mut transaction = store.pool.begin().await.unwrap(); + let second_envelope = RunSummaryStore::append_event_on_connection( + &mut transaction, + 1, + &second, + &second_payload, + ) + .await + .unwrap(); + transaction.commit().await.unwrap(); + + let expected = RunProjection::apply_events(&[first_envelope, second_envelope]).unwrap(); + + let loaded = store.load_projection(&id).await.unwrap(); + assert_eq!(loaded.run_id, id); + assert_eq!(loaded.last_seq, 2); + assert_eq!( + serde_json::to_value(loaded.projection.as_ref()).unwrap(), + serde_json::to_value(expected).unwrap() + ); + + let missing = run_id(created_at.timestamp_millis().cast_unsigned() + 1, 32); + assert!(matches!( + store.load_projection(&missing).await, + Err(Error::RunNotFound(text)) if text == missing.to_string() + )); + } + + #[tokio::test] + async fn load_projection_reports_removed_events() { + let (_directory, store) = store().await; + let created_at = dt("2026-08-27T12:00:00Z"); + let id = run_id(created_at.timestamp_millis().cast_unsigned(), 33); + let current = entry(projection(id, "created", created_at), 1); + let mut transaction = store.pool.begin().await.unwrap(); + RunSummaryStore::insert_first_event_on_connection( + &mut transaction, + ¤t, + &created_payload(&id), + ) + .await + .unwrap(); + transaction.commit().await.unwrap(); + store.test_delete_run_events(&id).await.unwrap(); + + assert!(matches!( + store.load_projection(&id).await, + Err(Error::RunHeadMismatch { + expected_last_seq: 1, + actual_last_seq: None, + .. + }) + )); + } + #[tokio::test] async fn session_owner_lookup_resolves_only_typed_creation_events() { let (_directory, store) = store().await; diff --git a/lib/components/fabro-store/src/slate/mod.rs b/lib/components/fabro-store/src/slate/mod.rs index 5678ee954..1c1857666 100644 --- a/lib/components/fabro-store/src/slate/mod.rs +++ b/lib/components/fabro-store/src/slate/mod.rs @@ -8,7 +8,6 @@ use std::time::Duration; use chrono::{DateTime, Utc}; use fabro_types::{RunId, SessionId}; use object_store::ObjectStore; -pub(crate) use run_store::CachedRunProjection; pub use run_store::RunDatabase; use run_store::RunDatabaseInner; use slatedb::config::{CompressionCodec, Settings}; @@ -128,8 +127,8 @@ impl Database { payload: &EventPayload, ) -> Result { let (mut active_runs, run_store) = self.reserve_new_run(run_id).await?; - let (envelope, cached) = run_store.commit_first_event(payload).await?; - run_store.install_in_memory_state(&cached); + let (envelope, projected) = run_store.commit_first_event(payload).await?; + run_store.install_in_memory_state(projected); Self::cache_active_run(&mut active_runs, &run_store); run_store.publish(&envelope); Ok(run_store) @@ -170,9 +169,6 @@ impl Database { if let Some(active) = active_run_from(&active_runs, run_id) { return Ok(active); } - if !self.run_summary_store.contains(run_id).await? { - return Err(Error::RunNotFound(run_id.to_string())); - } let run_store = self.open_run_database(run_id, false).await?; Self::cache_active_run(&mut active_runs, &run_store); Ok(run_store) @@ -182,9 +178,6 @@ impl Database { if let Some(active) = self.get_active_run(run_id).await { return Ok(active.read_only_clone()); } - if !self.run_summary_store.contains(run_id).await? { - return Err(Error::RunNotFound(run_id.to_string())); - } self.open_run_database(run_id, true).await } @@ -192,18 +185,12 @@ impl Database { let run_ids = self.run_summary_store.list_run_ids().await?; let mut unreadable = Vec::new(); for run_id in run_ids { - match RunDatabase::build_projection(&self.run_summary_store, &run_id).await { - Ok(Some(_)) => {} - Ok(None) => unreadable.push(UnreadableRun { - run_id, - created_at: run_id.created_at(), - error: "run has no events".to_string(), - }), - Err(err) => unreadable.push(UnreadableRun { + if let Err(err) = self.run_summary_store.load_projection(&run_id).await { + unreadable.push(UnreadableRun { run_id, created_at: run_id.created_at(), error: err.to_string(), - }), + }); } } unreadable.sort_by(|left, right| { @@ -249,11 +236,11 @@ impl Database { if let Some(active) = self.get_active_run(run_id).await { return active.projection_snapshot().await.map(Some); } - Ok( - RunDatabase::build_projection(&self.run_summary_store, run_id) - .await? - .map(|entry| entry.projection), - ) + match self.run_summary_store.load_projection(run_id).await { + Ok(projected) => Ok(Some(projected.projection)), + Err(Error::RunNotFound(_)) => Ok(None), + Err(error) => Err(error), + } } /// Resolves the run that owns `session_id` from the canonical typed @@ -338,6 +325,7 @@ mod tests { use object_store::path::Path; use super::*; + use crate::run_state::ProjectedRun; use crate::{EventPayload, keys, test_support as store_test_support}; fn dt(value: &str) -> DateTime { @@ -902,6 +890,21 @@ mod tests { ); } + #[tokio::test] + async fn missing_run_open_paths_return_run_not_found() { + let (_object_store, store) = make_store(); + let run_id = test_run_id("run-4"); + + assert!(matches!( + store.open_run(&run_id).await, + Err(Error::RunNotFound(id)) if id == run_id.to_string() + )); + assert!(matches!( + store.open_run_reader(&run_id).await, + Err(Error::RunNotFound(id)) if id == run_id.to_string() + )); + } + #[tokio::test] async fn open_run_reader_is_read_only() { let (_object_store, store) = make_store(); @@ -1016,11 +1019,7 @@ mod tests { let projection = store.load_run_projection(&run_id).await.unwrap().unwrap(); let last_seq = run.last_event_seq().await.unwrap().unwrap(); - let entries = [CachedRunProjection::from_projection( - run_id, - Arc::unwrap_or_clone(projection), - last_seq, - )]; + let entries = [ProjectedRun::new(run_id, projection, last_seq)]; summaries.reconcile(&entries).await.unwrap(); let summary = summaries.get(&run_id, Utc::now()).await.unwrap().unwrap(); assert_eq!(summary.lifecycle.status, RunStatus::Runnable); diff --git a/lib/components/fabro-store/src/slate/run_store.rs b/lib/components/fabro-store/src/slate/run_store.rs index e64fe6fc1..e1a23fc26 100644 --- a/lib/components/fabro-store/src/slate/run_store.rs +++ b/lib/components/fabro-store/src/slate/run_store.rs @@ -6,7 +6,7 @@ use futures::Stream; use tokio::sync::{Mutex as AsyncMutex, broadcast, mpsc}; use tokio_stream::wrappers::UnboundedReceiverStream; -use crate::run_state::{EventProjectionCache, RunProjectionReducer}; +use crate::run_state::{EventProjectionCache, ProjectedRun, RunProjectionReducer}; use crate::{ BlobStore, Error, EventEnvelope, EventPayload, Result, RunProjection, RunSummaryStore, StageId, run_summary_store, @@ -16,35 +16,6 @@ use crate::{ /// from SQLite. const EVENT_BROADCAST_CAPACITY: usize = 1024; -/// A run's projection as of its last committed event. Produced by replaying -/// SQLite history or by applying a newly committed event, and consumed by -/// `RunSummaryStore` writes that must stay in step with the event log. -#[derive(Debug, Clone)] -pub(crate) struct CachedRunProjection { - pub(crate) run_id: RunId, - pub(crate) projection: Arc, - pub(crate) last_seq: u32, -} - -impl CachedRunProjection { - pub(crate) fn from_projection(run_id: RunId, projection: RunProjection, last_seq: u32) -> Self { - Self { - run_id, - projection: Arc::new(projection), - last_seq, - } - } -} - -impl From for EventProjectionCache { - fn from(cached: CachedRunProjection) -> Self { - Self { - last_seq: cached.last_seq, - state: Some(cached.projection), - } - } -} - #[derive(Clone)] pub struct RunDatabase { inner: Arc, @@ -85,15 +56,13 @@ impl RunDatabase { blob_store: Arc, run_summary_store: Arc, ) -> Result { - let cached = Self::build_projection(&run_summary_store, &run_id) - .await? - .ok_or_else(|| Error::RunNotFound(run_id.to_string()))?; + let projected = run_summary_store.load_projection(&run_id).await?; Ok(Self::from_event_projection_cache( run_id, read_only, blob_store, run_summary_store, - cached.into(), + projected.into(), )) } @@ -158,25 +127,6 @@ impl RunDatabase { self.inner.event_tx.subscribe() } - pub(crate) async fn build_projection( - store: &RunSummaryStore, - run_id: &RunId, - ) -> Result> { - let events = match store.list_events_for_run(run_id).await { - Ok(events) => events, - Err(Error::RunNotFound(_)) => return Ok(None), - Err(error) => return Err(error), - }; - let last_seq = events - .last() - .map(|event| event.seq) - .ok_or_else(|| Error::InvalidEvent(format!("run {run_id} has no run.created event")))?; - let state = RunProjection::apply_events(&events)?; - Ok(Some(CachedRunProjection::from_projection( - *run_id, state, last_seq, - ))) - } - pub(super) async fn projection_snapshot(&self) -> Result> { let _state_guard = self.inner.state_lock.lock().await; self.projection_snapshot_locked() @@ -195,10 +145,8 @@ impl RunDatabase { }) } - pub(crate) fn install_in_memory_state(&self, cached: &CachedRunProjection) { - let mut projection_cache = self.inner.lock_projection_cache(); - projection_cache.state = Some(Arc::clone(&cached.projection)); - projection_cache.last_seq = cached.last_seq; + pub(crate) fn install_in_memory_state(&self, projected: ProjectedRun) { + *self.inner.lock_projection_cache() = projected.into(); } pub(crate) fn publish(&self, event: &EventEnvelope) { @@ -208,7 +156,7 @@ impl RunDatabase { pub(crate) async fn commit_first_event( &self, payload: &EventPayload, - ) -> Result<(EventEnvelope, CachedRunProjection)> { + ) -> Result<(EventEnvelope, ProjectedRun)> { payload.validate(&self.inner.run_id)?; let event = RunEvent::try_from(payload)?; let _state_guard = self.inner.state_lock.lock().await; @@ -269,10 +217,10 @@ impl RunDatabase { payload: &EventPayload, event: RunEvent, ) -> Result { - let (envelope, cached) = self.commit_event_locked(payload, event).await?; + let (envelope, projected) = self.commit_event_locked(payload, event).await?; // Keep post-commit propagation await-free: cancellation after SQLite // commits must not leave in-memory state stale or omit the broadcast. - self.install_in_memory_state(&cached); + self.install_in_memory_state(projected); self.publish(&envelope); Ok(envelope) } @@ -281,7 +229,7 @@ impl RunDatabase { &self, payload: &EventPayload, event: RunEvent, - ) -> Result<(EventEnvelope, CachedRunProjection)> { + ) -> Result<(EventEnvelope, ProjectedRun)> { let (expected_last_seq, mut next_state) = { let cache = self.inner.lock_projection_cache(); (cache.last_seq, cache.state.clone()) @@ -291,27 +239,23 @@ impl RunDatabase { apply_cached_projection_event(&mut next_state, &prospective).map_err(event_rejected)?; let next_projection = next_state.expect("applying a valid event should always produce a projection"); - let cached = CachedRunProjection::from_projection( - self.inner.run_id, - Arc::unwrap_or_clone(next_projection), - seq, - ); + let projected = ProjectedRun::new(self.inner.run_id, next_projection, seq); let mut transaction = self.inner.run_summary_store.begin().await?; let envelope = if expected_last_seq == 0 { - RunSummaryStore::insert_first_event_on_connection(&mut transaction, &cached, payload) + RunSummaryStore::insert_first_event_on_connection(&mut transaction, &projected, payload) .await? } else { RunSummaryStore::append_event_on_connection( &mut transaction, expected_last_seq, - &cached, + &projected, payload, ) .await? }; transaction.commit().await?; - Ok((envelope, cached)) + Ok((envelope, projected)) } pub async fn list_events(&self) -> Result> {