scriptoriumadmin/fabro
1
0
Fork 0
mirror of https://github.com/fabro-sh/fabro.git synced 2026-09-08 22:21:45 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 1

Merge pull request #835 from fabro-sh/codex/projected-run-cleanup

Browse source 
Move run projection replay into the summary store
This commit is contained in:
Scott Werner 2026-09-03 11:40:45 -04:00 • committed by GitHub
commit a2002175e1
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
5 changed files with 202 additions and 152 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

61
lib/components/fabro-store/src/legacy_run_history_import.rs
View file

@ -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<ValidatedLegacyRunEvent>,
current: CachedRunProjection,
current: ProjectedRun,
}
struct LegacyRunHistorySource {
@ -527,14 +527,8 @@ impl LegacyRunHistorySource {
.iter()
.map(|event| event.envelope.clone())
.collect::<Vec<_>>();
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<CachedRunProjection> {
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<ProjectedRun> {
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::<Vec<_>>();
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<u64, LegacyRunHistoryImportFailure> {
@ -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::<Vec<_>>();
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, &current).await?;
for ((_, event_json), (payload, envelope)) in events.iter().zip(&decoded) {
@ -1513,8 +1493,7 @@ mod tests {
.iter()
.map(|(_payload, envelope)| envelope.clone())
.collect::<Vec<_>>();
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,

39
lib/components/fabro-store/src/run_state.rs
View file

@ -33,6 +33,45 @@ pub(crate) struct EventProjectionCache {
pub state: Option<Arc<RunProjection>>,
}
/// 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<RunProjection>,
pub(crate) last_seq: u32,
}
impl ProjectedRun {
pub(crate) fn new(run_id: RunId, projection: Arc<RunProjection>, 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<Self> {
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<ProjectedRun> 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<Self>
where

117
lib/components/fabro-store/src/run_summary_store.rs
View file

@ -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<ProjectedRun> {
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<Vec<EventEnvelope>> {
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<EventEnvelope> {
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<EventEnvelope> {
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<Utc>) {
#[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<Utc> {
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,
&current,
&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;

55
lib/components/fabro-store/src/slate/mod.rs
View file

@ -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<RunDatabase> {
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<Utc> {
@ -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);

82
lib/components/fabro-store/src/slate/run_store.rs
View file

@ -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<RunProjection>,
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<CachedRunProjection> for EventProjectionCache {
fn from(cached: CachedRunProjection) -> Self {
Self {
last_seq: cached.last_seq,
state: Some(cached.projection),
}
}
}
#[derive(Clone)]
pub struct RunDatabase {
inner: Arc<RunDatabaseInner>,
@ -85,15 +56,13 @@ impl RunDatabase {
blob_store: Arc<BlobStore>,
run_summary_store: Arc<RunSummaryStore>,
) -> Result<Self> {
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<Option<CachedRunProjection>> {
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<Arc<RunProjection>> {
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<EventEnvelope> {
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<Vec<EventEnvelope>> {