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

Simplify on-demand run projection loading

Browse source 
Drop the unreachable active-run mismatch guard that was copied into
load_run_projection: the active-runs map is only ever inserted under the
handle's own run ID, so the check could never fire. Remove it from the two
pre-existing sites too and delete matches_run.

Trim install_in_memory_state to take only the committed projection, since
the event envelope duplicated last_seq and the inner scope only existed to
release the lock before the now-removed shared cache update. Add a From
impl so RunDatabase::build no longer hand-builds EventProjectionCache, and
rename projected_state_locked to match its projection_snapshot sibling.

In fabro-server, have reject_if_archived and ensure_run_exists read the
run summary row instead of replaying the full event history for inactive
runs; the summary is written in the same transaction as the event.

Fold the repeated store-reopen fixtures in fabro-store and fabro-server
tests into helpers, and fix a stale comment about the deleted shared
projection cache.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
This commit is contained in:
Scott Werner 2026-09-02 11:43:17 -04:00
parent c1e9364490
commit 13f3ff867b
6 changed files with 93 additions and 120 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

10
lib/apps/fabro-server/src/server.rs
View file

@ -4534,8 +4534,14 @@ async fn append_control_request(
/// run is currently archived. Returns `None` otherwise (including when the run
/// doesn't exist — the caller's own not-found handling will surface that).
async fn reject_if_archived(state: &AppState, run_id: &RunId) -> Option<Response> {
let projection = state.load_run_projection(run_id).await.ok()?;
projection.archived_at.is_some().then(|| {
let summary = state
.stores
.run_summaries
.get(run_id, Utc::now())
.await
.ok()
.flatten()?;
summary.lifecycle.archived_at.is_some().then(|| {
ApiError::new(
StatusCode::CONFLICT,
operations::archived_rejection_message(run_id),

15
lib/apps/fabro-server/src/server/handler/artifacts.rs
View file

@ -131,11 +131,18 @@ async fn read_run_blob(
}
async fn ensure_run_exists(state: &AppState, run_id: &RunId) -> Result<(), Response> {
state
.load_run_projection(run_id)
match state
.stores
.run_summaries
.get(run_id, chrono::Utc::now())
.await
.map(|_| ())
.map_err(IntoResponse::into_response)
{
Ok(Some(_)) => Ok(()),
Ok(None) => Err(ApiError::not_found("Run not found.").into_response()),
Err(err) => {
Err(ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string()).into_response())
}
}
}
async fn list_run_artifacts(

56
lib/apps/fabro-server/src/server/tests.rs
View file

@ -11629,27 +11629,37 @@ async fn get_run_state_exposes_pending_interviews() {
);
}
/// Builds an app state over shared object, blob, and summary stores so a test
/// can drop it and open a second state that sees the same durable data.
fn test_app_state_over_shared_stores(
object_store: &Arc<dyn object_store::ObjectStore>,
blobs: &Arc<fabro_store::BlobStore>,
summaries: &Arc<fabro_store::RunSummaryStore>,
) -> Arc<AppState> {
let store = Arc::new(fabro_store::test_support::test_database_with_stores(
Arc::clone(object_store),
"runs",
std::time::Duration::from_millis(1),
None,
Arc::clone(blobs),
Arc::clone(summaries),
));
test_app_state_with_store(
default_test_server_settings(),
RunLayer::default(),
5,
store,
ArtifactStore::new(Arc::clone(object_store), "artifacts"),
)
}
#[tokio::test]
async fn restarted_run_state_details_load_from_sql_and_preserve_error_statuses() {
let object_store: Arc<dyn object_store::ObjectStore> =
Arc::new(object_store::memory::InMemory::new());
let summaries = fabro_store::test_support::test_run_summary_store();
let blobs = fabro_store::test_support::test_blob_store();
let first_store = Arc::new(fabro_store::test_support::test_database_with_stores(
Arc::clone(&object_store),
"runs",
std::time::Duration::from_millis(1),
None,
Arc::clone(&blobs),
Arc::clone(&summaries),
));
let first_state = test_app_state_with_store(
default_test_server_settings(),
RunLayer::default(),
5,
first_store,
ArtifactStore::new(Arc::clone(&object_store), "artifacts"),
);
let first_state = test_app_state_over_shared_stores(&object_store, &blobs, &summaries);
let healthy_id = fixtures::RUN_1;
let broken_id = fixtures::RUN_2;
create_durable_run_with_events(&first_state, healthy_id, &[
@ -11667,21 +11677,7 @@ async fn restarted_run_state_details_load_from_sql_and_preserve_error_statuses()
.unwrap();
drop(first_state);
let reopened_store = Arc::new(fabro_store::test_support::test_database_with_stores(
Arc::clone(&object_store),
"runs",
std::time::Duration::from_millis(1),
None,
blobs,
summaries,
));
let reopened_state = test_app_state_with_store(
default_test_server_settings(),
RunLayer::default(),
5,
reopened_store,
ArtifactStore::new(object_store, "artifacts"),
);
let reopened_state = test_app_state_over_shared_stores(&object_store, &blobs, &summaries);
assert_eq!(
reconcile_incomplete_runs_on_startup(&reopened_state)
.await

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

@ -28,8 +28,8 @@ use crate::{Error, EventEnvelope, Result};
#[derive(Debug, Clone, Default)]
pub(crate) struct EventProjectionCache {
pub last_seq: u32,
// Arc-shared with the shared projection cache so opening a run does not
// deep-copy the projection; mutated copy-on-write via `Arc::make_mut`.
// Arc-shared with readers so a snapshot never deep-copies the projection;
// mutated copy-on-write via `Arc::make_mut`.
pub state: Option<Arc<RunProjection>>,
}

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

@ -129,7 +129,7 @@ impl Database {
) -> 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(&envelope, &cached);
run_store.install_in_memory_state(&cached);
Self::cache_active_run(&mut active_runs, &run_store);
run_store.publish(&envelope);
Ok(run_store)
@ -168,11 +168,6 @@ impl Database {
let mut active_runs = self.active_runs.lock().await;
if let Some(active) = active_run_from(&active_runs, run_id) {
if !active.matches_run(run_id) {
return Err(Error::Other(format!(
"active run cache mismatch for run_id {run_id:?}"
)));
}
return Ok(active);
}
if !self.run_summary_store.contains(run_id).await? {
@ -185,11 +180,6 @@ impl Database {
pub async fn open_run_reader(&self, run_id: &RunId) -> Result<RunDatabase> {
if let Some(active) = self.get_active_run(run_id).await {
if !active.matches_run(run_id) {
return Err(Error::Other(format!(
"active run cache mismatch for run_id {run_id:?}"
)));
}
return Ok(active.read_only_clone());
}
if !self.run_summary_store.contains(run_id).await? {
@ -257,11 +247,6 @@ impl Database {
pub async fn load_run_projection(&self, run_id: &RunId) -> Result<Option<Arc<RunProjection>>> {
if let Some(active) = self.get_active_run(run_id).await {
if !active.matches_run(run_id) {
return Err(Error::Other(format!(
"active run cache mismatch for run_id {run_id:?}"
)));
}
return active.projection_snapshot().await.map(Some);
}
Ok(
@ -416,6 +401,22 @@ mod tests {
(object_store, store)
}
/// Reopens a `Database` over an existing object store and SQLite summary
/// store, simulating a process restart with no active run handles.
fn reopen_store(
object_store: Arc<dyn ObjectStore>,
run_summaries: Arc<RunSummaryStore>,
) -> Database {
store_test_support::test_database_with_stores(
object_store,
"runs",
Duration::from_millis(1),
None,
store_test_support::test_blob_store(),
run_summaries,
)
}
#[tokio::test]
async fn retire_refresh_token_keyspace_clears_the_prefix_and_is_idempotent() {
let (_object_store, store) = make_store();
@ -1393,14 +1394,7 @@ mod tests {
let run = store.create_run(&test_run_id("run-1")).await.unwrap();
append_completed(&run, "run-1", dt("2026-03-27T12:00:00Z")).await;
let reopened = store_test_support::test_database_with_stores(
object_store,
"runs",
Duration::from_millis(1),
None,
store_test_support::test_blob_store(),
summaries,
);
let reopened = reopen_store(object_store, summaries);
let summary = reopened
.run_summary_store()
.list_all(Utc::now())
@ -1424,14 +1418,7 @@ mod tests {
append_completed(&run_2, "run-2", dt("2026-03-27T12:00:10Z")).await;
append_completed(&run_3, "run-3", dt("2026-03-27T12:00:20Z")).await;
let reopened = store_test_support::test_database_with_stores(
object_store,
"runs",
Duration::from_millis(1),
None,
store_test_support::test_blob_store(),
summaries,
);
let reopened = reopen_store(object_store, summaries);
assert!(reopened.active_runs.lock().await.is_empty());
let projection = reopened
@ -1505,14 +1492,7 @@ mod tests {
.unwrap();
summaries.test_delete_run_events(&broken_id).await.unwrap();
let reopened = store_test_support::test_database_with_stores(
object_store,
"runs",
Duration::from_millis(1),
None,
store_test_support::test_blob_store(),
summaries,
);
let reopened = reopen_store(object_store, summaries);
assert!(matches!(
reopened.load_run_projection(&broken_id).await,
Err(Error::RunHeadMismatch { .. })
@ -1534,14 +1514,7 @@ mod tests {
let run_id = test_run_id("run-1");
let writer = writer_store.create_run(&run_id).await.unwrap();
append_created(&writer, "run-1", dt("2026-03-27T12:00:00Z")).await;
let reader_store = store_test_support::test_database_with_stores(
object_store,
"runs",
Duration::from_millis(1),
None,
store_test_support::test_blob_store(),
summaries,
);
let reader_store = reopen_store(object_store, summaries);
let original_title = writer.state().await.unwrap().title;
let update = event_payload(
"run-1",
@ -1705,14 +1678,7 @@ mod tests {
let run = store.create_run(&run_id).await.unwrap();
append_completed(&run, "run-1", dt("2026-03-27T12:00:00Z")).await;
let reopened = store_test_support::test_database_with_stores(
object_store,
"runs",
Duration::from_millis(1),
None,
store_test_support::test_blob_store(),
summaries,
);
let reopened = reopen_store(object_store, summaries);
// Opening and projecting from canonical SQLite must not inspect an
// unreadable key in the retained legacy event history.
let mut unreadable_old_key = keys::run_event_seq_prefix(&run_id, 2).as_ref().to_vec();
@ -1804,13 +1770,13 @@ mod tests {
reason: FailureReason::WorkflowError,
});
let cached = reopened
let projection = reopened
.load_run_projection(&run_id)
.await
.unwrap()
.unwrap();
assert_eq!(cached.title, "Renamed failed run");
assert_eq!(cached.status, RunStatus::Failed {
assert_eq!(projection.title, "Renamed failed run");
assert_eq!(projection.status, RunStatus::Failed {
reason: FailureReason::WorkflowError,
});
}

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

@ -16,6 +16,9 @@ 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,
@ -33,6 +36,15 @@ impl CachedRunProjection {
}
}
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>,
@ -76,16 +88,12 @@ impl RunDatabase {
let cached = Self::build_projection(&run_summary_store, &run_id)
.await?
.ok_or_else(|| Error::RunNotFound(run_id.to_string()))?;
let projection_cache = EventProjectionCache {
last_seq: cached.last_seq,
state: Some(cached.projection),
};
Ok(Self::from_event_projection_cache(
run_id,
read_only,
blob_store,
run_summary_store,
projection_cache,
cached.into(),
))
}
@ -150,10 +158,6 @@ impl RunDatabase {
self.inner.event_tx.subscribe()
}
pub(crate) fn matches_run(&self, run_id: &RunId) -> bool {
self.inner.run_id == *run_id
}
pub(crate) async fn build_projection(
store: &RunSummaryStore,
run_id: &RunId,
@ -175,10 +179,10 @@ impl RunDatabase {
pub(super) async fn projection_snapshot(&self) -> Result<Arc<RunProjection>> {
let _state_guard = self.inner.state_lock.lock().await;
self.projected_state_locked()
self.projection_snapshot_locked()
}
fn projected_state_locked(&self) -> Result<Arc<RunProjection>> {
fn projection_snapshot_locked(&self) -> Result<Arc<RunProjection>> {
self.inner
.lock_projection_cache()
.state
@ -191,16 +195,10 @@ impl RunDatabase {
})
}
pub(crate) fn install_in_memory_state(
&self,
event: &EventEnvelope,
cached: &CachedRunProjection,
) {
{
let mut projection_cache = self.inner.lock_projection_cache();
projection_cache.state = Some(Arc::clone(&cached.projection));
projection_cache.last_seq = event.seq;
}
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 publish(&self, event: &EventEnvelope) {
@ -244,7 +242,7 @@ impl RunDatabase {
payload.validate(&self.inner.run_id)?;
let event = RunEvent::try_from(payload)?;
let _state_guard = self.inner.state_lock.lock().await;
let projection = self.projected_state_locked()?;
let projection = self.projection_snapshot_locked()?;
if !predicate(&projection) {
return Ok(None);
}
@ -274,7 +272,7 @@ impl RunDatabase {
let (envelope, cached) = 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(&envelope, &cached);
self.install_in_memory_state(&cached);
self.publish(&envelope);
Ok(envelope)
}