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

Move run projection replay into the summary store

Browse source
This commit is contained in:
Scott Werner 2026-09-02 13:32:43 -04:00
parent 7c23e08868
commit 2360e8046b
4 changed files with 174 additions and 82 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

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

@ -16,7 +16,7 @@ use tokio::sync::Barrier;
use tracing::debug;
use crate::keys::SlateKey;
use crate::slate::CachedRunProjection;
use crate::slate::ProjectedRun;
use crate::{Database, EventPayload, RunProjectionReducer, RunSummaryStore, keys};
/// Count-only observations about the legacy catalog and session indexes.
@ -458,7 +458,7 @@ struct ValidatedLegacyRunEvent {
struct ValidatedLegacyRunHistory {
run_id: RunId,
events: Vec<ValidatedLegacyRunEvent>,
current: CachedRunProjection,
current: ProjectedRun,
}
struct LegacyRunHistorySource {
@ -534,7 +534,7 @@ impl LegacyRunHistorySource {
.expect("a validated history contains at least one event")
.envelope
.seq;
let current = CachedRunProjection::from_projection(run_id, projection, last_seq);
let current = ProjectedRun::new(run_id, projection, last_seq);
Ok(Some(ValidatedLegacyRunHistory {
run_id,
events,
@ -1159,7 +1159,7 @@ fn require_exact_prefix(
fn replay_destination(
run_id: &RunId,
events: &[(EventEnvelope, String)],
) -> crate::Result<CachedRunProjection> {
) -> crate::Result<ProjectedRun> {
let Some((first, _event_json)) = events.first() else {
return Err(crate::Error::InvalidEvent(
"run projection requires an event".to_owned(),
@ -1181,9 +1181,7 @@ fn replay_destination(
.last()
.expect("a destination history validated as nonempty")
.seq;
Ok(CachedRunProjection::from_projection(
*run_id, projection, last_seq,
))
Ok(ProjectedRun::new(*run_id, projection, last_seq))
}
fn usize_to_import_count(value: usize) -> Result<u64, LegacyRunHistoryImportFailure> {
@ -1273,7 +1271,7 @@ mod tests {
parse_source_event,
};
use crate::keys::SlateKey;
use crate::slate::CachedRunProjection;
use crate::slate::ProjectedRun;
use crate::{
Database, EventEnvelope, EventPayload, RunProjectionReducer, RunSummaryStore, keys,
test_support as store_test_support,
@ -1479,8 +1477,7 @@ mod tests {
.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::new(*run_id, projection, events.last().unwrap().0);
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) {
@ -1514,7 +1511,7 @@ mod tests {
.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::new(*run_id, projection, next.0);
let mut transaction = pool.begin().await?;
RunSummaryStore::append_event_on_connection(
&mut transaction,

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

@ -3,8 +3,8 @@ use std::sync::LazyLock;
use chrono::{DateTime, Utc};
use fabro_types::{
BilledTokenCounts, EventEnvelope, Run, RunEvent, RunId, RunSize, RunStatusKind, RunTiming,
SessionId, StageId, timing,
BilledTokenCounts, EventEnvelope, Run, RunEvent, RunId, RunProjection, RunSize, RunStatusKind,
RunTiming, SessionId, StageId, timing,
};
use sqlx::pool::PoolConnection;
use sqlx::query::Query;
@ -12,8 +12,8 @@ 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::{RunProjectionReducer, build_summary, projected_billing};
use crate::slate::ProjectedRun;
use crate::{Error, EventPayload, Result, keys};
const INSERT_RUN_SQL: &str = r"
@ -288,14 +288,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 +351,22 @@ ON CONFLICT(singleton) DO NOTHING
select_run_head(&mut connection, run_id).await
}
/// Replays one run's canonical history from one validated SQLite snapshot,
/// returning `None` when the run does not exist.
pub(crate) async fn load_projection(&self, run_id: &RunId) -> Result<Option<ProjectedRun>> {
let events = match self.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 projection = RunProjection::apply_events(&events)?;
Ok(Some(ProjectedRun::new(*run_id, projection, last_seq)))
}
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 +654,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 +673,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 +750,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 +784,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 +1020,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 +1052,7 @@ impl PreparedRunSummary {
}
fn ensure_entry_identity(
entry: &CachedRunProjection,
entry: &ProjectedRun,
record: &PreparedRunSummary,
seq: u32,
) -> Result<()> {
@ -1544,8 +1560,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::slate::ProjectedRun;
use crate::{Error, EventPayload, RunProjectionReducer, test_support as store_test_support};
fn dt(value: &str) -> DateTime<Utc> {
value.parse().unwrap()
@ -1580,8 +1596,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, projection, last_seq)
}
async fn store() -> (tempfile::TempDir, RunSummaryStore) {
@ -1920,6 +1936,94 @@ 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().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!(store.load_projection(&missing).await.unwrap().is_none());
}
#[tokio::test]
async fn load_projection_reports_removed_events_without_poisoning_following_reads() {
let (_directory, store) = store().await;
let created_at = dt("2026-08-27T12:00:00Z");
let broken_id = run_id(created_at.timestamp_millis().cast_unsigned(), 33);
let healthy_id = run_id(created_at.timestamp_millis().cast_unsigned() + 1, 34);
for id in [broken_id, healthy_id] {
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(&broken_id).await.unwrap();
assert!(matches!(
store.load_projection(&broken_id).await,
Err(Error::RunHeadMismatch {
expected_last_seq: 1,
actual_last_seq: None,
..
})
));
let loaded = store.load_projection(&healthy_id).await.unwrap().unwrap();
assert_eq!(loaded.run_id, healthy_id);
assert_eq!(loaded.last_seq, 1);
assert_eq!(loaded.projection.title, "created");
}
#[tokio::test]
async fn session_owner_lookup_resolves_only_typed_creation_events() {
let (_directory, store) = store().await;

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

@ -8,7 +8,7 @@ use std::time::Duration;
use chrono::{DateTime, Utc};
use fabro_types::{RunId, SessionId};
use object_store::ObjectStore;
pub(crate) use run_store::CachedRunProjection;
pub(crate) use run_store::ProjectedRun;
pub use run_store::RunDatabase;
use run_store::RunDatabaseInner;
use slatedb::config::{CompressionCodec, Settings};
@ -128,8 +128,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 +170,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 +179,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,7 +186,7 @@ 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 {
match self.run_summary_store.load_projection(&run_id).await {
Ok(Some(_)) => {}
Ok(None) => unreadable.push(UnreadableRun {
run_id,
@ -249,11 +243,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),
)
Ok(self
.run_summary_store
.load_projection(run_id)
.await?
.map(|projected| projected.projection))
}
/// Resolves the run that owns `session_id` from the canonical typed
@ -902,6 +896,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,7 +1025,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(
let entries = [ProjectedRun::new(
run_id,
Arc::unwrap_or_clone(projection),
last_seq,

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

@ -20,14 +20,14 @@ const EVENT_BROADCAST_CAPACITY: usize = 1024;
/// 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) struct ProjectedRun {
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 {
impl ProjectedRun {
pub(crate) fn new(run_id: RunId, projection: RunProjection, last_seq: u32) -> Self {
Self {
run_id,
projection: Arc::new(projection),
@ -36,11 +36,11 @@ impl CachedRunProjection {
}
}
impl From<CachedRunProjection> for EventProjectionCache {
fn from(cached: CachedRunProjection) -> Self {
impl From<ProjectedRun> for EventProjectionCache {
fn from(projected: ProjectedRun) -> Self {
Self {
last_seq: cached.last_seq,
state: Some(cached.projection),
last_seq: projected.last_seq,
state: Some(projected.projection),
}
}
}
@ -85,7 +85,8 @@ impl RunDatabase {
blob_store: Arc<BlobStore>,
run_summary_store: Arc<RunSummaryStore>,
) -> Result<Self> {
let cached = Self::build_projection(&run_summary_store, &run_id)
let projected = run_summary_store
.load_projection(&run_id)
.await?
.ok_or_else(|| Error::RunNotFound(run_id.to_string()))?;
Ok(Self::from_event_projection_cache(
@ -93,7 +94,7 @@ impl RunDatabase {
read_only,
blob_store,
run_summary_store,
cached.into(),
projected.into(),
))
}
@ -158,25 +159,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 +177,10 @@ impl RunDatabase {
})
}
pub(crate) fn install_in_memory_state(&self, cached: &CachedRunProjection) {
pub(crate) fn install_in_memory_state(&self, projected: &ProjectedRun) {
let mut projection_cache = self.inner.lock_projection_cache();
projection_cache.state = Some(Arc::clone(&cached.projection));
projection_cache.last_seq = cached.last_seq;
projection_cache.state = Some(Arc::clone(&projected.projection));
projection_cache.last_seq = projected.last_seq;
}
pub(crate) fn publish(&self, event: &EventEnvelope) {
@ -208,7 +190,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 +251,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 +263,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,7 +273,7 @@ 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(
let projected = ProjectedRun::new(
self.inner.run_id,
Arc::unwrap_or_clone(next_projection),
seq,
@ -299,19 +281,19 @@ impl RunDatabase {
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>> {