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

Load inactive run projections on demand

Browse source
This commit is contained in:
Scott Werner 2026-09-01 16:06:54 -04:00
parent d260ae89b5
commit c1e9364490
15 changed files with 343 additions and 267 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
lib/apps/fabro-server/src/run_files.rs
View file

@ -1195,7 +1195,7 @@ async fn load_projection(
state: &Arc<AppState>,
run_id: &RunId,
) -> std::result::Result<Arc<fabro_store::RunProjection>, ApiError> {
state.cached_run_projection(run_id).await
state.load_run_projection(run_id).await
}
async fn reconnect_run_sandbox(

6
lib/apps/fabro-server/src/serve.rs
View file

@ -833,12 +833,6 @@ where
#[cfg(any(test, feature = "test-support"))]
automation_materializer_override: None,
})?;
state
.stores
.runs
.warm_projection_cache()
.await
.context("warming run projection cache from SQLite")?;
let reconciled = reconcile_incomplete_runs_on_startup(&state).await?;
if reconciled > 0 {
info!(

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

@ -794,7 +794,7 @@ impl SlackService {
return;
};
let event_name = event.body.event_name();
let projection = match state.stores.runs.get_cached_projection(&event.run_id).await {
let projection = match state.stores.runs.load_run_projection(&event.run_id).await {
Ok(Some(projection)) => projection,
Ok(None) => {
warn!(
@ -1508,16 +1508,16 @@ impl AppState {
&self.stores.runs
}
/// Current cached projection for `run_id`, with the standard HTTP error
/// Loads the current projection for `run_id`, with the standard HTTP error
/// mapping: storage failures become 500s and a missing run becomes the
/// canonical 404.
pub(crate) async fn cached_run_projection(
pub(crate) async fn load_run_projection(
&self,
run_id: &RunId,
) -> Result<Arc<fabro_store::RunProjection>, ApiError> {
self.stores
.runs
.get_cached_projection(run_id)
.load_run_projection(run_id)
.await
.map_err(|err| ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string()))?
.ok_or_else(|| ApiError::not_found("Run not found."))
@ -3744,7 +3744,7 @@ async fn load_pending_interview(
qid: &str,
) -> Result<LoadedPendingInterview, Response> {
let projection = state
.cached_run_projection(&run_id)
.load_run_projection(&run_id)
.await
.map_err(IntoResponse::into_response)?;
let Some(record) = projection.pending_interviews.get(qid) else {
@ -4534,7 +4534,7 @@ 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.cached_run_projection(run_id).await.ok()?;
let projection = state.load_run_projection(run_id).await.ok()?;
projection.archived_at.is_some().then(|| {
ApiError::new(
StatusCode::CONFLICT,

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

@ -86,7 +86,7 @@ async fn get_checkpoint(
Ok(id) => id,
Err(response) => return response,
};
match state.cached_run_projection(&id).await {
match state.load_run_projection(&id).await {
Ok(projection) => match projection.current_checkpoint() {
Some(cp) => (StatusCode::OK, Json(cp.clone())).into_response(),
None => (StatusCode::OK, Json(serde_json::json!(null))).into_response(),
@ -132,7 +132,7 @@ async fn read_run_blob(
async fn ensure_run_exists(state: &AppState, run_id: &RunId) -> Result<(), Response> {
state
.cached_run_projection(run_id)
.load_run_projection(run_id)
.await
.map(|_| ())
.map_err(IntoResponse::into_response)
@ -330,7 +330,7 @@ async fn download_run_artifacts(
Ok(id) => id,
Err(response) => return response,
};
let projection = match state.cached_run_projection(&id).await {
let projection = match state.load_run_projection(&id).await {
Ok(projection) => projection,
Err(error) => return error.into_response(),
};

4
lib/apps/fabro-server/src/server/handler/billing.rs
View file

@ -69,7 +69,7 @@ async fn list_run_stages(
Err(response) => return response,
};
let projection = match state.cached_run_projection(&id).await {
let projection = match state.load_run_projection(&id).await {
Ok(projection) => projection,
Err(err) => return err.into_response(),
};
@ -90,7 +90,7 @@ async fn get_run_billing(
State(state): State<Arc<AppState>>,
Path(id): Path<RunId>,
) -> Response {
let projection = match state.cached_run_projection(&id).await {
let projection = match state.load_run_projection(&id).await {
Ok(projection) => projection,
Err(err) => return err.into_response(),
};

2
lib/apps/fabro-server/src/server/handler/graph.rs
View file

@ -243,7 +243,7 @@ async fn load_run_dot_source(state: &AppState, id: &RunId) -> Result<String, Res
Some(dot)
} else {
state
.cached_run_projection(id)
.load_run_projection(id)
.await
.map_err(IntoResponse::into_response)?
.spec

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

@ -151,7 +151,7 @@ async fn load_pull_request_record(
state: &Arc<AppState>,
id: &RunId,
) -> Result<PullRequestLink, ApiError> {
let projection = state.cached_run_projection(id).await?;
let projection = state.load_run_projection(id).await?;
projection.pull_request.clone().ok_or_else(|| {
ApiError::with_code(
StatusCode::NOT_FOUND,
@ -319,7 +319,7 @@ async fn create_run_pull_request(
let Ok(run_store) = state.stores.runs.open_run(&id).await else {
return ApiError::not_found("Run not found.").into_response();
};
let run_state = match state.cached_run_projection(&id).await {
let run_state = match state.load_run_projection(&id).await {
Ok(run_state) => run_state,
Err(err) => return err.into_response(),
};
@ -372,7 +372,7 @@ async fn create_run_pull_request(
}
};
let run_state = match state.cached_run_projection(&id).await {
let run_state = match state.load_run_projection(&id).await {
Ok(run_state) => run_state,
Err(err) => return err.into_response(),
};
@ -427,7 +427,7 @@ async fn get_run_pull_request_creation(
RequireRunScoped(id): RequireRunScoped,
State(state): State<Arc<AppState>>,
) -> Response {
let run_state = match state.cached_run_projection(&id).await {
let run_state = match state.load_run_projection(&id).await {
Ok(run_state) => run_state,
Err(err) => return err.into_response(),
};
@ -473,7 +473,7 @@ async fn unlink_run_pull_request(
let Ok(run_store) = state.stores.runs.open_run(&id).await else {
return ApiError::not_found("Run not found.").into_response();
};
let run_state = match state.cached_run_projection(&id).await {
let run_state = match state.load_run_projection(&id).await {
Ok(run_state) => run_state,
Err(err) => return err.into_response(),
};

12
lib/apps/fabro-server/src/server/handler/runs.rs
View file

@ -330,7 +330,7 @@ async fn run_summary_at(
return Ok(None);
};
if summary.timestamps.completed_at.is_none() {
let projection = state.stores.runs.get_cached_projection(run_id).await?;
let projection = state.stores.runs.load_run_projection(run_id).await?;
if let Some(timing) = projection.and_then(|projection| projection.live_run_timing(now)) {
summary.timing = Some(timing);
}
@ -1608,7 +1608,7 @@ async fn get_run_settings(
Ok(id) => id,
Err(response) => return response,
};
let projection = match state.cached_run_projection(&id).await {
let projection = match state.load_run_projection(&id).await {
Ok(projection) => projection,
Err(err) => return err.into_response(),
};
@ -1619,7 +1619,7 @@ async fn get_questions(
RequireRunManagementTarget(id, _actor): RequireRunManagementTarget,
State(state): State<Arc<AppState>>,
) -> Response {
match state.cached_run_projection(&id).await {
match state.load_run_projection(&id).await {
Ok(projection) => {
let questions = projection
.pending_interviews
@ -1660,7 +1660,7 @@ async fn get_run_state(
RequireRunManagementTarget(id, _actor): RequireRunManagementTarget,
State(state): State<Arc<AppState>>,
) -> Response {
match state.cached_run_projection(&id).await {
match state.load_run_projection(&id).await {
Ok(projection) => Json(&*projection).into_response(),
Err(err) => err.into_response(),
}
@ -1697,7 +1697,7 @@ async fn get_run_stage_context_window(
Ok(stage_id) => stage_id,
Err(response) => return response,
};
let projection = match state.cached_run_projection(&id).await {
let projection = match state.load_run_projection(&id).await {
Ok(projection) => projection,
Err(err) => return err.into_response(),
};
@ -1753,7 +1753,7 @@ async fn get_run_stage_command_log(
return ApiError::bad_request("limit must be greater than 0").into_response();
}
let limit = query.limit.min(MAX_COMMAND_LOG_LIMIT);
let projection = match state.cached_run_projection(&id).await {
let projection = match state.load_run_projection(&id).await {
Ok(projection) => projection,
Err(err) => return err.into_response(),
};

2
lib/apps/fabro-server/src/server/handler/sandbox.rs
View file

@ -952,7 +952,7 @@ async fn load_run_sandbox_instance(
run_id: &RunId,
) -> Result<fabro_types::RunSandboxInstance, Response> {
let projection = state
.cached_run_projection(run_id)
.load_run_projection(run_id)
.await
.map_err(IntoResponse::into_response)?;
projection

2
lib/apps/fabro-server/src/server/handler/worker_control.rs
View file

@ -35,7 +35,7 @@ async fn worker_control_stream(
Query(query): Query<WorkerControlStreamQuery>,
ws: WebSocketUpgrade,
) -> Response {
let projection = match state.cached_run_projection(&id).await {
let projection = match state.load_run_projection(&id).await {
Ok(projection) => projection,
Err(err) => return err.into_response(),
};

4
lib/apps/fabro-server/src/server/pull_request_supervisor.rs
View file

@ -139,7 +139,7 @@ pub(in crate::server) async fn process_pull_request_creation(
) -> anyhow::Result<()> {
let _create_guard = state.pull_request_create_locks.lock(run_id).await;
let run_store = state.stores.runs.open_run(&run_id).await?;
let Some(run_state) = state.stores.runs.get_cached_projection(&run_id).await? else {
let Some(run_state) = state.stores.runs.load_run_projection(&run_id).await? else {
return Ok(());
};
let Some(creation) = run_state
@ -257,7 +257,7 @@ pub(super) async fn recover_pending_pull_request_creations(
if !can_dispatch(&run_id, active, failures) {
continue;
}
let projection = match state.stores.runs.get_cached_projection(&run_id).await {
let projection = match state.stores.runs.load_run_projection(&run_id).await {
Ok(Some(projection)) => projection,
Ok(None) => continue,
Err(error) => {

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

@ -11630,7 +11630,109 @@ async fn get_run_state_exposes_pending_interviews() {
}
#[tokio::test]
async fn cache_backed_run_endpoints_reflect_events_appended_after_warmup() {
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 healthy_id = fixtures::RUN_1;
let broken_id = fixtures::RUN_2;
create_durable_run_with_events(&first_state, healthy_id, &[
workflow_event::Event::RunSubmitted {
definition_blob: None,
},
])
.await;
create_succeeded_run(&first_state, broken_id).await;
first_state
.stores
.run_summaries
.test_delete_run_events(&broken_id)
.await
.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"),
);
assert_eq!(
reconcile_incomplete_runs_on_startup(&reopened_state)
.await
.unwrap(),
0,
"startup reconciliation must not replay terminal histories"
);
let app = crate::test_support::build_test_router(reopened_state);
let healthy = app
.clone()
.oneshot(
Request::builder()
.method("GET")
.uri(api(&format!("/runs/{healthy_id}/state")))
.body(Body::empty())
.unwrap(),
)
.await
.unwrap();
let healthy_body = response_json!(healthy, StatusCode::OK).await;
assert_eq!(healthy_body["spec"]["run_id"], healthy_id.to_string());
let missing = app
.clone()
.oneshot(
Request::builder()
.method("GET")
.uri(api(&format!("/runs/{}/state", fixtures::RUN_3)))
.body(Body::empty())
.unwrap(),
)
.await
.unwrap();
assert_status!(missing, StatusCode::NOT_FOUND).await;
let broken = app
.oneshot(
Request::builder()
.method("GET")
.uri(api(&format!("/runs/{broken_id}/state")))
.body(Body::empty())
.unwrap(),
)
.await
.unwrap();
assert_status!(broken, StatusCode::INTERNAL_SERVER_ERROR).await;
}
#[tokio::test]
async fn run_projection_endpoints_reflect_events_appended_to_an_open_run() {
let state = test_app_state();
let app = crate::test_support::build_test_router(Arc::clone(&state));
let run_id = create_run(&app, MINIMAL_DOT)
@ -11638,8 +11740,6 @@ async fn cache_backed_run_endpoints_reflect_events_appended_after_warmup() {
.parse::<RunId>()
.unwrap();
state.stores.runs.warm_projection_cache().await.unwrap();
let run_store = state.stores.runs.open_run(&run_id).await.unwrap();
workflow_event::append_event(&run_store, &run_id, &workflow_event::Event::RunRunnable {
source: fabro_types::RunRunnableSource::StartRequested,
@ -12976,14 +13076,14 @@ async fn run_tool_worker_token_can_use_client_backend_routes_across_runs() {
assert_ne!(response.status(), StatusCode::FORBIDDEN);
let created_child = create_run_with_bearer(&app, &run_tool_worker_token).await;
let cached = state
let projection = state
.stores
.runs
.get_cached_projection(&created_child)
.load_run_projection(&created_child)
.await
.unwrap()
.expect("created run should be cached");
assert_eq!(cached.spec.provenance.subject, Principal::Worker {
.expect("created run should have a projection");
assert_eq!(projection.spec.provenance.subject, Principal::Worker {
run_id: parent_run_id,
},);

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

@ -1,4 +1,3 @@
mod projection_cache;
mod run_store;
use std::collections::HashMap;
@ -9,13 +8,11 @@ use std::time::Duration;
use chrono::{DateTime, Utc};
use fabro_types::{RunId, SessionId};
use object_store::ObjectStore;
pub(crate) use projection_cache::CachedRunProjection;
use projection_cache::RunProjectionCache;
pub(crate) use run_store::CachedRunProjection;
pub use run_store::RunDatabase;
use run_store::RunDatabaseInner;
use slatedb::config::{CompressionCodec, Settings};
use tokio::sync::{Mutex, MutexGuard, OnceCell};
use tracing::warn;
use crate::{BlobStore, Error, EventPayload, Result, RunProjection, RunSummaryStore, keys};
@ -28,15 +25,13 @@ pub struct UnreadableRun {
#[derive(Clone)]
pub struct Database {
object_store: Arc<dyn ObjectStore>,
base_prefix: String,
flush_interval: Duration,
cache_path: Option<PathBuf>,
db: Arc<OnceCell<slatedb::Db>>,
active_runs: Arc<Mutex<HashMap<RunId, Arc<RunDatabaseInner>>>>,
blobs: Arc<BlobStore>,
projection_cache: Arc<RunProjectionCache>,
projection_cache_warmed: Arc<OnceCell<()>>,
object_store: Arc<dyn ObjectStore>,
base_prefix: String,
flush_interval: Duration,
cache_path: Option<PathBuf>,
db: Arc<OnceCell<slatedb::Db>>,
active_runs: Arc<Mutex<HashMap<RunId, Arc<RunDatabaseInner>>>>,
blobs: Arc<BlobStore>,
run_summary_store: Arc<RunSummaryStore>,
}
@ -67,8 +62,6 @@ impl Database {
db: Arc::new(OnceCell::new()),
active_runs: Arc::new(Mutex::new(HashMap::new())),
blobs,
projection_cache: Arc::new(RunProjectionCache::default()),
projection_cache_warmed: Arc::new(OnceCell::new()),
run_summary_store,
}
}
@ -126,14 +119,7 @@ impl Database {
/// Builds a run handle wired to the Database-owned shared stores.
async fn open_run_database(&self, run_id: &RunId, read_only: bool) -> Result<RunDatabase> {
RunDatabase::build(
*run_id,
read_only,
self.blobs(),
Arc::clone(&self.projection_cache),
self.run_summary_store(),
)
.await
RunDatabase::build(*run_id, read_only, self.blobs(), self.run_summary_store()).await
}
pub async fn create_run_with_first_event(
@ -168,22 +154,15 @@ impl Database {
MutexGuard<'_, HashMap<RunId, Arc<RunDatabaseInner>>>,
RunDatabase,
)> {
self.warm_projection_cache().await?;
let active_runs = self.active_runs.lock().await;
if active_runs.contains_key(run_id) || self.run_summary_store.contains(run_id).await? {
return Err(Error::RunAlreadyExists(run_id.to_string()));
}
let run_store = RunDatabase::build_empty(
*run_id,
self.blobs(),
Arc::clone(&self.projection_cache),
self.run_summary_store(),
);
let run_store = RunDatabase::build_empty(*run_id, self.blobs(), self.run_summary_store());
Ok((active_runs, run_store))
}
pub async fn open_run(&self, run_id: &RunId) -> Result<RunDatabase> {
self.warm_projection_cache().await?;
// Keep the active-writer miss and insert atomic. Otherwise concurrent
// callers can create independent writers with the same recovered seq.
let mut active_runs = self.active_runs.lock().await;
@ -219,38 +198,11 @@ impl Database {
self.open_run_database(run_id, true).await
}
pub async fn warm_projection_cache(&self) -> Result<()> {
self.projection_cache_warmed
.get_or_try_init(|| async {
let run_ids = self.run_summary_store.list_run_ids().await?;
let mut entries = Vec::new();
for run_id in run_ids {
match RunDatabase::build_cached_projection(&self.run_summary_store, &run_id)
.await
{
Ok(Some(entry)) => entries.push(entry),
Ok(None) => {}
Err(err) => {
warn!(
run_id = %run_id,
error = %err,
"Skipping run during projection cache warmup"
);
}
}
}
self.projection_cache.replace_all(entries);
Ok::<_, Error>(())
})
.await?;
Ok(())
}
pub async fn list_unreadable_runs(&self) -> Result<Vec<UnreadableRun>> {
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_cached_projection(&self.run_summary_store, &run_id).await {
match RunDatabase::build_projection(&self.run_summary_store, &run_id).await {
Ok(Some(_)) => {}
Ok(None) => unreadable.push(UnreadableRun {
run_id,
@ -284,7 +236,6 @@ impl Database {
.test_insert_unvalidated_event(run_id, seq, payload)
.await?;
self.active_runs.lock().await.remove(run_id);
self.projection_cache.remove(run_id);
Ok(())
}
@ -304,15 +255,20 @@ impl Database {
Ok(())
}
pub async fn get_cached_projection(
&self,
run_id: &RunId,
) -> Result<Option<Arc<RunProjection>>> {
self.warm_projection_cache().await?;
Ok(self
.projection_cache
.projection_snapshot(run_id)
.map(|(projection, _)| projection))
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(
RunDatabase::build_projection(&self.run_summary_store, run_id)
.await?
.map(|entry| entry.projection),
)
}
/// Resolves the run that owns `session_id` from the canonical typed
@ -321,10 +277,6 @@ impl Database {
self.run_summary_store.find_session_owner(session_id).await
}
pub(crate) fn remove_cached_run(&self, run_id: &RunId) {
self.projection_cache.remove(run_id);
}
pub async fn delete_run(&self, run_id: &RunId) -> Result<()> {
let mut active_runs = self.active_runs.lock().await;
let active = active_runs.get(run_id).cloned();
@ -336,7 +288,6 @@ impl Database {
.delete_canonical(run_id, Utc::now().timestamp_millis())
.await?;
active_runs.remove(run_id);
self.remove_cached_run(run_id);
Ok(())
}
@ -1017,7 +968,7 @@ mod tests {
}
#[tokio::test]
async fn rejected_transition_writes_nothing_and_preserves_projection_cache() {
async fn rejected_transition_writes_nothing_and_preserves_projection_state() {
let (_object_store, store) = make_store();
let run_id = test_run_id("run-1");
let run = store.create_run(&run_id).await.unwrap();
@ -1043,8 +994,8 @@ mod tests {
));
assert_eq!(run.list_events().await.unwrap(), events_before);
assert_eq!(run.state().await.unwrap().status, RunStatus::Runnable);
let (projection, last_seq) = store.projection_cache.projection_snapshot(&run_id).unwrap();
assert_eq!(last_seq, 4);
let projection = store.load_run_projection(&run_id).await.unwrap().unwrap();
assert_eq!(run.last_event_seq().await.unwrap(), Some(4));
assert_eq!(projection.status, RunStatus::Runnable);
}
@ -1062,7 +1013,8 @@ mod tests {
.unwrap_err();
assert!(matches!(err, Error::EventRejected { .. }));
let (projection, last_seq) = store.projection_cache.projection_snapshot(&run_id).unwrap();
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),
@ -1080,7 +1032,7 @@ mod tests {
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;
let cached_before = store.projection_cache.projection_snapshot(&run_id).unwrap();
let projection_before = store.load_run_projection(&run_id).await.unwrap().unwrap();
summaries.close_pool().await;
let result = run
@ -1096,9 +1048,8 @@ mod tests {
result,
Err(Error::Sqlite(sqlx::Error::PoolClosed))
));
let cached = store.projection_cache.projection_snapshot(&run_id).unwrap();
assert_eq!(cached.1, cached_before.1);
assert_eq!(cached.0.title, cached_before.0.title);
let projection_after = store.load_run_projection(&run_id).await.unwrap().unwrap();
assert!(Arc::ptr_eq(&projection_before, &projection_after));
}
#[tokio::test]
@ -1463,13 +1414,15 @@ mod tests {
}
#[tokio::test]
async fn projection_cache_warmup_rebuilds_full_projections() {
async fn inactive_projection_loads_on_demand_without_registering_run() {
let (_directory, summaries) = make_run_summary_store().await;
let (object_store, store) = make_store_with_run_summaries(Arc::clone(&summaries));
let run_1 = store.create_run(&test_run_id("run-1")).await.unwrap();
let run_2 = store.create_run(&test_run_id("run-2")).await.unwrap();
let run_3 = store.create_run(&test_run_id("run-3")).await.unwrap();
append_completed(&run_1, "run-1", dt("2026-03-27T12:00:00Z")).await;
append_running(&run_2, "run-2", dt("2026-03-27T12:00:10Z")).await;
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,
@ -1479,32 +1432,141 @@ mod tests {
store_test_support::test_blob_store(),
summaries,
);
reopened.warm_projection_cache().await.unwrap();
assert!(reopened.active_runs.lock().await.is_empty());
let (run_1_projection, run_1_last_seq) = reopened
.projection_cache
.projection_snapshot(&test_run_id("run-1"))
let projection = reopened
.load_run_projection(&test_run_id("run-2"))
.await
.unwrap()
.unwrap();
assert_eq!(run_1_projection.status, RunStatus::Succeeded {
assert_eq!(projection.spec.run_id, test_run_id("run-2"));
assert_eq!(projection.status, RunStatus::Succeeded {
reason: SuccessReason::Completed,
});
assert_eq!(run_1_last_seq, 5);
let (run_2_projection, run_2_last_seq) = reopened
.projection_cache
.projection_snapshot(&test_run_id("run-2"))
.unwrap();
assert_eq!(run_2_projection.status, RunStatus::Running);
assert_eq!(run_2_last_seq, 4);
assert!(
reopened.active_runs.lock().await.is_empty(),
"inactive detail reads must stay request-local"
);
}
#[tokio::test]
async fn required_run_summary_append_refreshes_cache_and_delete_removes_rows() {
async fn active_projection_reads_are_coherent_immutable_snapshots() {
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;
let first = store.load_run_projection(&run_id).await.unwrap().unwrap();
let second = store.load_run_projection(&run_id).await.unwrap().unwrap();
assert!(Arc::ptr_eq(&first, &second));
let original_title = first.title.clone();
run.append_event(&event_payload(
"run-1",
"2026-03-27T12:00:01Z",
"run.title.updated",
&serde_json::json!({ "title": "New title" }),
))
.await
.unwrap();
let updated = store.load_run_projection(&run_id).await.unwrap().unwrap();
assert!(!Arc::ptr_eq(&first, &updated));
assert_eq!(first.title, original_title);
assert_eq!(updated.title, "New title");
}
#[tokio::test]
async fn inactive_projection_replay_accepts_sequence_gaps_and_failures_do_not_poison_reads() {
let (_directory, summaries) = make_run_summary_store().await;
let (object_store, store) = make_store_with_run_summaries(Arc::clone(&summaries));
let healthy_id = test_run_id("run-1");
let broken_id = test_run_id("run-2");
let healthy = store.create_run(&healthy_id).await.unwrap();
let broken = store.create_run(&broken_id).await.unwrap();
append_created(&healthy, "run-1", dt("2026-03-27T12:00:00Z")).await;
append_created(&broken, "run-2", dt("2026-03-27T12:00:10Z")).await;
store.remove_active_run(&healthy_id).await;
store.remove_active_run(&broken_id).await;
store
.put_unvalidated_run_event(
&healthy_id,
3,
event_payload(
"run-1",
"2026-03-27T12:00:02Z",
"run.title.updated",
&serde_json::json!({ "title": "Imported gap" }),
)
.as_value(),
)
.await
.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,
);
assert!(matches!(
reopened.load_run_projection(&broken_id).await,
Err(Error::RunHeadMismatch { .. })
));
let projection = reopened
.load_run_projection(&healthy_id)
.await
.unwrap()
.unwrap();
assert_eq!(projection.title, "Imported gap");
assert!(reopened.active_runs.lock().await.is_empty());
}
#[tokio::test]
async fn inactive_projection_read_concurrent_with_append_is_one_coherent_snapshot() {
let (_directory, summaries) = make_run_summary_store().await;
let (object_store, writer_store) = make_store_with_run_summaries(Arc::clone(&summaries));
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 original_title = writer.state().await.unwrap().title;
let update = event_payload(
"run-1",
"2026-03-27T12:00:01Z",
"run.title.updated",
&serde_json::json!({ "title": "Concurrent title" }),
);
let (loaded, appended) = tokio::join!(
reader_store.load_run_projection(&run_id),
writer.append_event(&update)
);
appended.unwrap();
let loaded = loaded.unwrap().unwrap();
assert!(loaded.title == original_title || loaded.title == "Concurrent title");
assert!(reader_store.active_runs.lock().await.is_empty());
}
#[tokio::test]
async fn required_run_summary_append_refreshes_active_projection_and_delete_removes_rows() {
let (_directory, summaries) = make_run_summary_store().await;
let (_object_store, store) = make_store_with_run_summaries(Arc::clone(&summaries));
let run = store.create_run(&test_run_id("run-1")).await.unwrap();
append_created(&run, "run-1", dt("2026-03-27T12:00:00Z")).await;
store.warm_projection_cache().await.unwrap();
run.append_event(&event_payload(
"run-1",
"2026-03-27T12:00:01Z",
@ -1583,12 +1645,13 @@ mod tests {
.await
.unwrap();
let (projection, last_seq) = store
.projection_cache
.projection_snapshot(&test_run_id("run-1"))
let projection = store
.load_run_projection(&test_run_id("run-1"))
.await
.unwrap()
.unwrap();
assert_eq!(projection.status, RunStatus::Running);
assert_eq!(last_seq, 7);
assert_eq!(run.last_event_seq().await.unwrap(), Some(7));
assert_eq!(
projection
.stage(&StageId::new("review", 1))
@ -1620,8 +1683,9 @@ mod tests {
store.delete_run(&test_run_id("run-1")).await.unwrap();
assert!(
store
.projection_cache
.projection_snapshot(&test_run_id("run-1"))
.load_run_projection(&test_run_id("run-1"))
.await
.unwrap()
.is_none()
);
assert!(
@ -1649,11 +1713,8 @@ mod tests {
store_test_support::test_blob_store(),
summaries,
);
reopened.warm_projection_cache().await.unwrap();
// If opening or projecting the run starts at the beginning, this
// unreadable old key makes the operation fail. A hydrated run starts
// after the shared projection's last sequence instead.
// 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();
unreadable_old_key.push(0xff);
reopened
@ -1744,7 +1805,7 @@ mod tests {
});
let cached = reopened
.get_cached_projection(&run_id)
.load_run_projection(&run_id)
.await
.unwrap()
.unwrap();

78
lib/components/fabro-store/src/slate/projection_cache.rs
View file

@ -1,78 +0,0 @@
use std::collections::HashMap;
use std::sync::{Arc, Mutex, MutexGuard};
use fabro_types::{RunId, RunProjection};
#[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,
}
}
}
#[derive(Debug, Default)]
pub(crate) struct RunProjectionCache {
// Cache operations are bounded in-memory work and never await. Keeping
// this lock synchronous lets a committed event update both projection
// caches without introducing a cancellation point.
state: Mutex<RunProjectionCacheState>,
}
#[derive(Debug, Default)]
struct RunProjectionCacheState {
entries: HashMap<RunId, CachedRunProjection>,
}
impl RunProjectionCacheState {
fn replace_all(&mut self, entries: Vec<CachedRunProjection>) {
self.entries.clear();
for entry in entries {
self.insert(entry);
}
}
fn insert(&mut self, entry: CachedRunProjection) {
self.entries.insert(entry.run_id, entry);
}
fn remove(&mut self, run_id: &RunId) {
self.entries.remove(run_id);
}
}
impl RunProjectionCache {
fn lock(&self) -> MutexGuard<'_, RunProjectionCacheState> {
self.state.lock().expect(
"run projection cache mutex is never poisoned: no code panics while holding this lock",
)
}
pub(crate) fn replace_all(&self, entries: Vec<CachedRunProjection>) {
self.lock().replace_all(entries);
}
pub(crate) fn replace(&self, entry: CachedRunProjection) {
self.lock().insert(entry);
}
pub(crate) fn projection_snapshot(&self, run_id: &RunId) -> Option<(Arc<RunProjection>, u32)> {
self.lock()
.entries
.get(run_id)
.map(|entry| (Arc::clone(&entry.projection), entry.last_seq))
}
pub(crate) fn remove(&self, run_id: &RunId) {
self.lock().remove(run_id);
}
}

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

@ -6,7 +6,6 @@ use futures::Stream;
use tokio::sync::{Mutex as AsyncMutex, broadcast, mpsc};
use tokio_stream::wrappers::UnboundedReceiverStream;
use super::projection_cache::{CachedRunProjection, RunProjectionCache};
use crate::run_state::{EventProjectionCache, RunProjectionReducer};
use crate::{
BlobStore, Error, EventEnvelope, EventPayload, Result, RunProjection, RunSummaryStore, StageId,
@ -17,6 +16,23 @@ use crate::{
/// from SQLite.
const EVENT_BROADCAST_CAPACITY: usize = 1024;
#[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,
}
}
}
#[derive(Clone)]
pub struct RunDatabase {
inner: Arc<RunDatabaseInner>,
@ -34,13 +50,12 @@ impl std::fmt::Debug for RunDatabase {
}
pub(crate) struct RunDatabaseInner {
pub(crate) run_id: RunId,
blob_store: Arc<BlobStore>,
pub(crate) state_lock: AsyncMutex<()>,
projection_cache: StdMutex<EventProjectionCache>,
shared_projection_cache: Arc<RunProjectionCache>,
run_summary_store: Arc<RunSummaryStore>,
event_tx: broadcast::Sender<EventEnvelope>,
pub(crate) run_id: RunId,
blob_store: Arc<BlobStore>,
pub(crate) state_lock: AsyncMutex<()>,
projection_cache: StdMutex<EventProjectionCache>,
run_summary_store: Arc<RunSummaryStore>,
event_tx: broadcast::Sender<EventEnvelope>,
}
impl RunDatabaseInner {
@ -56,30 +71,19 @@ impl RunDatabase {
run_id: RunId,
read_only: bool,
blob_store: Arc<BlobStore>,
shared_projection_cache: Arc<RunProjectionCache>,
run_summary_store: Arc<RunSummaryStore>,
) -> Result<Self> {
let projection_cache = if let Some((projection, last_seq)) =
shared_projection_cache.projection_snapshot(&run_id)
{
EventProjectionCache {
last_seq,
state: Some(projection),
}
} else {
let cached = Self::build_cached_projection(&run_summary_store, &run_id)
.await?
.ok_or_else(|| Error::RunNotFound(run_id.to_string()))?;
EventProjectionCache {
last_seq: cached.last_seq,
state: Some(cached.projection),
}
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_projection_cache(
Ok(Self::from_event_projection_cache(
run_id,
read_only,
blob_store,
shared_projection_cache,
run_summary_store,
projection_cache,
))
@ -88,24 +92,21 @@ impl RunDatabase {
pub(crate) fn build_empty(
run_id: RunId,
blob_store: Arc<BlobStore>,
shared_projection_cache: Arc<RunProjectionCache>,
run_summary_store: Arc<RunSummaryStore>,
) -> Self {
Self::from_projection_cache(
Self::from_event_projection_cache(
run_id,
false,
blob_store,
shared_projection_cache,
run_summary_store,
EventProjectionCache::default(),
)
}
fn from_projection_cache(
fn from_event_projection_cache(
run_id: RunId,
read_only: bool,
blob_store: Arc<BlobStore>,
shared_projection_cache: Arc<RunProjectionCache>,
run_summary_store: Arc<RunSummaryStore>,
projection_cache: EventProjectionCache,
) -> Self {
@ -116,7 +117,6 @@ impl RunDatabase {
blob_store,
state_lock: AsyncMutex::new(()),
projection_cache: StdMutex::new(projection_cache),
shared_projection_cache,
run_summary_store,
event_tx,
}),
@ -154,7 +154,7 @@ impl RunDatabase {
self.inner.run_id == *run_id
}
pub(crate) async fn build_cached_projection(
pub(crate) async fn build_projection(
store: &RunSummaryStore,
run_id: &RunId,
) -> Result<Option<CachedRunProjection>> {
@ -173,7 +173,7 @@ impl RunDatabase {
)))
}
async fn projected_state(&self) -> Result<Arc<RunProjection>> {
pub(super) async fn projection_snapshot(&self) -> Result<Arc<RunProjection>> {
let _state_guard = self.inner.state_lock.lock().await;
self.projected_state_locked()
}
@ -201,7 +201,6 @@ impl RunDatabase {
projection_cache.state = Some(Arc::clone(&cached.projection));
projection_cache.last_seq = event.seq;
}
self.inner.shared_projection_cache.replace(cached.clone());
}
pub(crate) fn publish(&self, event: &EventEnvelope) {
@ -274,7 +273,7 @@ impl RunDatabase {
) -> Result<EventEnvelope> {
let (envelope, cached) = self.commit_event_locked(payload, event).await?;
// Keep post-commit propagation await-free: cancellation after SQLite
// commits must not leave either cache stale or omit the broadcast.
// commits must not leave in-memory state stale or omit the broadcast.
self.install_in_memory_state(&envelope, &cached);
self.publish(&envelope);
Ok(envelope)
@ -434,7 +433,7 @@ impl RunDatabase {
}
pub async fn state(&self) -> Result<RunProjection> {
Ok(Arc::unwrap_or_clone(self.projected_state().await?))
Ok(Arc::unwrap_or_clone(self.projection_snapshot().await?))
}
}