scriptoriumadmin/fabro
1
0
Fork 0
mirror of https://github.com/fabro-sh/fabro.git synced 2026-09-12 23:02:41 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 5

fix(store): hydrate projection cache before appending later events

Browse source
This commit is contained in:
Bryan Helmkamp 2026-05-10 13:38:06 -04:00
parent 5209d05623
commit da368366f0
No known key found for this signature in database
2 changed files with 103 additions and 2 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

73
lib/crates/fabro-store/src/slate/mod.rs
View file

@ -1083,4 +1083,77 @@ mod tests {
.is_empty()
);
}
#[tokio::test]
async fn append_event_hydrates_local_projection_cache_for_fresh_writer() {
let (object_store, store) = make_store();
let run_id = test_run_id("run-1");
let run = store.create_run(&run_id).await.unwrap();
append_created(&run, "run-1", dt("2026-03-27T12:00:00Z")).await;
run.append_event(&event_payload(
"run-1",
"2026-03-27T12:00:01Z",
"run.queued",
&serde_json::json!({}),
))
.await
.unwrap();
run.append_event(&event_payload(
"run-1",
"2026-03-27T12:00:02Z",
"run.starting",
&serde_json::json!({}),
))
.await
.unwrap();
run.append_event(&event_payload(
"run-1",
"2026-03-27T12:00:03Z",
"run.running",
&serde_json::json!({}),
))
.await
.unwrap();
run.append_event(&event_payload(
"run-1",
"2026-03-27T12:00:04Z",
"run.failed",
&serde_json::json!({
"error": "workflow failed",
"duration_ms": 1,
"reason": "workflow_error",
}),
))
.await
.unwrap();
let reopened = Database::new(
Arc::clone(&object_store),
"runs/",
Duration::from_millis(1),
None,
);
let fresh_writer = reopened.open_run(&run_id).await.unwrap();
fresh_writer
.append_event(&event_payload(
"run-1",
"2026-03-27T12:00:05Z",
"run.title.updated",
&serde_json::json!({ "title": "Renamed failed run" }),
))
.await
.unwrap();
let state = fresh_writer.state().await.unwrap();
assert_eq!(state.title, "Renamed failed run");
assert_eq!(state.status, RunStatus::Failed {
reason: FailureReason::WorkflowError,
});
let cached = reopened.get_cached_run(&run_id).await.unwrap().unwrap();
assert_eq!(cached.summary.title, "Renamed failed run");
assert_eq!(cached.summary.status, RunStatus::Failed {
reason: FailureReason::WorkflowError,
});
}
}

32
lib/crates/fabro-store/src/slate/run_store.rs
View file

@ -174,8 +174,14 @@ impl RunDatabase {
async fn cache_event(&self, event: &EventEnvelope) -> Result<()> {
{
let mut projection_cache = self.inner.projection_cache.lock().await;
apply_cached_projection_event(&mut projection_cache.state, event)?;
projection_cache.last_seq = event.seq;
if projection_cache.state.is_none() && event.seq > 1 {
drop(projection_cache);
self.rebuild_local_projection_cache_through(event.seq)
.await?;
} else {
apply_cached_projection_event(&mut projection_cache.state, event)?;
projection_cache.last_seq = event.seq;
}
}
let mut recent_events = self.inner.recent_events.lock().await;
recent_events.push_back(event.clone());
@ -186,6 +192,28 @@ impl RunDatabase {
Ok(())
}
async fn rebuild_local_projection_cache_through(&self, seq: u32) -> Result<()> {
let events = list_events_from(&self.inner.db, &self.inner.run_id, 1).await?;
let Some(last_seq) = events.last().map(|event| event.seq) else {
return Err(Error::InvalidEvent(format!(
"run {} has no events while rebuilding projection cache",
self.inner.run_id
)));
};
if last_seq < seq {
return Err(Error::InvalidEvent(format!(
"run {} projection cache rebuild stopped at seq {last_seq}, before appended seq {seq}",
self.inner.run_id
)));
}
let state = RunProjection::apply_events(&events)?;
let mut projection_cache = self.inner.projection_cache.lock().await;
projection_cache.state = Some(state);
projection_cache.last_seq = last_seq;
Ok(())
}
async fn cached_events_from(&self, start_seq: u32, limit: usize) -> Option<Vec<EventEnvelope>> {
let recent_events = self.inner.recent_events.lock().await;
let oldest_seq = recent_events.front().map(|event| event.seq)?;