scriptoriumadmin/fabro
1
0
Fork 0
mirror of https://github.com/fabro-sh/fabro.git synced 2026-10-09 03:20:56 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 6

fix(store): reuse shared run projections

Browse source
This commit is contained in:
Bryan Helmkamp 2026-07-24 07:49:58 -04:00
parent e3011481a1
commit 8f0ecfb170
No known key found for this signature in database
12 changed files with 262 additions and 78 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

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

@ -1195,15 +1195,13 @@ async fn load_projection(
state: &Arc<AppState>,
run_id: &RunId,
) -> std::result::Result<fabro_store::RunProjection, ApiError> {
let reader = state
let cached = state
.store_ref()
.open_run_reader(run_id)
.get_cached_run(run_id)
.await
.map_err(|_| ApiError::not_found("Run not found."))?;
reader
.state()
.await
.map_err(|err| ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string()))
.map_err(|err| ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string()))?
.ok_or_else(|| ApiError::not_found("Run not found."))?;
Ok((*cached.projection).clone())
}
async fn reconnect_run_sandbox(

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

@ -2686,9 +2686,8 @@ async fn delete_run_internal(
}
async fn load_durable_run_status(state: &AppState, id: &RunId) -> Option<RunStatus> {
let run_store = state.stores.runs.open_run(id).await.ok()?;
let projection = run_store.state().await.ok()?;
Some(projection.status)
let cached = state.stores.runs.get_cached_run(id).await.ok()??;
Some(cached.projection.status)
}
async fn delete_run_sandbox_resource(
@ -4521,9 +4520,8 @@ 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 run_store = state.stores.runs.open_run_reader(run_id).await.ok()?;
let projection = run_store.state().await.ok()?;
projection.archived_at.is_some().then(|| {
let cached = state.stores.runs.get_cached_run(run_id).await.ok()??;
cached.projection.archived_at.is_some().then(|| {
ApiError::new(
StatusCode::CONFLICT,
operations::archived_rejection_message(run_id),

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

@ -119,16 +119,16 @@ async fn read_run_blob(
}
async fn load_run_spec(state: &AppState, run_id: &RunId) -> Result<fabro_types::RunSpec, Response> {
let run_store = state
let cached = state
.stores
.runs
.open_run_reader(run_id)
.get_cached_run(run_id)
.await
.map_err(|_| ApiError::not_found("Run not found.").into_response())?;
let run_state = run_store.state().await.map_err(|err| {
ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string()).into_response()
})?;
Ok(run_state.spec)
.map_err(|err| {
ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string()).into_response()
})?
.ok_or_else(|| ApiError::not_found("Run not found.").into_response())?;
Ok(cached.projection.spec.clone())
}
async fn list_run_artifacts(

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

@ -390,8 +390,8 @@ async fn attach_run_events(
let start_seq = match params.since_seq {
Some(seq) if seq >= 1 => seq,
Some(_) => 1,
None => match run_store.list_events().await {
Ok(events) => events.last().map_or(1, |event| event.seq.saturating_add(1)),
None => match run_store.last_event_seq().await {
Ok(last_seq) => last_seq.map_or(1, |seq| seq.saturating_add(1)),
Err(err) => {
return ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string())
.into_response();

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

@ -242,17 +242,17 @@ async fn load_run_dot_source(state: &AppState, id: &RunId) -> Result<String, Res
let dot_source = if let Some(dot) = live_dot_source.filter(|d| !d.is_empty()) {
Some(dot)
} else {
match state.stores.runs.open_run_reader(id).await {
Ok(run_store) => match run_store.state().await {
Ok(run_state) => run_state.spec.graph_source,
Err(err) => {
return Err(
ApiError::new(StatusCode::BAD_GATEWAY, err.to_string()).into_response()
);
}
},
Err(_) => return Err(ApiError::not_found("Run not found.").into_response()),
}
state
.stores
.runs
.get_cached_run(id)
.await
.map_err(|err| ApiError::new(StatusCode::BAD_GATEWAY, err.to_string()).into_response())?
.ok_or_else(|| ApiError::not_found("Run not found.").into_response())?
.projection
.spec
.graph_source
.clone()
};
dot_source

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

@ -131,17 +131,14 @@ async fn load_pull_request_record(
state: &Arc<AppState>,
id: &RunId,
) -> Result<PullRequestLink, ApiError> {
let run_store = state
let cached = state
.stores
.runs
.open_run_reader(id)
.get_cached_run(id)
.await
.map_err(|_| ApiError::not_found("Run not found."))?;
let run_state = run_store
.state()
.await
.map_err(|err| ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string()))?;
run_state.pull_request.ok_or_else(|| {
.map_err(|err| ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string()))?
.ok_or_else(|| ApiError::not_found("Run not found."))?;
cached.projection.pull_request.clone().ok_or_else(|| {
ApiError::with_code(
StatusCode::NOT_FOUND,
format!("No pull request found in store. Create one first with: fabro pr create {id}"),
@ -297,14 +294,22 @@ 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 run_store.state().await {
Ok(run_state) => run_state,
let cached = match state.stores.runs.get_cached_run(&id).await {
Ok(Some(cached)) => cached,
Ok(None) => {
return ApiError::new(
StatusCode::INTERNAL_SERVER_ERROR,
"Run projection unavailable.",
)
.into_response();
}
Err(err) => {
return ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string())
.into_response();
}
};
let inputs = match RunPrInputs::extract(&run_state, body.force) {
let run_state = cached.projection.as_ref();
let inputs = match RunPrInputs::extract(run_state, body.force) {
Ok(inputs) => inputs,
Err(err) => return err.into_response(),
};
@ -343,7 +348,7 @@ async fn create_run_pull_request(
llm_source: state.llm_source.as_ref(),
catalog,
conclusion: Some(inputs.conclusion),
run_state: Some(&run_state),
run_state: Some(run_state),
};
let created_pull_request = match pull_request::maybe_open_pull_request(request).await {
Ok(Some(created)) => created,
@ -401,14 +406,21 @@ 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 run_store.state().await {
Ok(run_state) => run_state,
let cached = match state.stores.runs.get_cached_run(&id).await {
Ok(Some(cached)) => cached,
Ok(None) => {
return ApiError::new(
StatusCode::INTERNAL_SERVER_ERROR,
"Run projection unavailable.",
)
.into_response();
}
Err(err) => {
return ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string())
.into_response();
}
};
let Some(pull_request) = run_state.pull_request else {
let Some(pull_request) = cached.projection.pull_request.clone() else {
return ApiError::with_code(
StatusCode::NOT_FOUND,
format!("No pull request found in store. Create one first with: fabro pr create {id}"),

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

@ -1109,14 +1109,15 @@ async fn get_run_stage_command_log(
let Ok(run_store) = state.stores.runs.open_run_reader(&id).await else {
return ApiError::not_found("Run not found.").into_response();
};
let run_state = match run_store.state().await {
Ok(run_state) => run_state,
let cached = match state.stores.runs.get_cached_run(&id).await {
Ok(Some(cached)) => cached,
Ok(None) => return ApiError::not_found("Run not found.").into_response(),
Err(err) => {
return ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string())
.into_response();
}
};
let Some(node) = run_state.stage(&stage_id) else {
let Some(node) = cached.projection.stage(&stage_id) else {
return ApiError::not_found("Stage not found.").into_response();
};

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

@ -951,18 +951,21 @@ async fn load_run_sandbox_instance(
state: &Arc<AppState>,
run_id: &RunId,
) -> Result<fabro_types::RunSandboxInstance, Response> {
match state.stores.runs.open_run_reader(run_id).await {
Ok(run_store) => match run_store.state().await {
Ok(run_state) => run_state
.sandbox
.and_then(fabro_types::RunSandbox::into_instance)
.ok_or_else(|| ApiError::not_found("Run sandbox was not created.").into_response()),
Err(err) => Err(
ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string()).into_response(),
),
},
Err(_) => Err(ApiError::not_found("Run not found.").into_response()),
}
let cached = state
.stores
.runs
.get_cached_run(run_id)
.await
.map_err(|err| {
ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string()).into_response()
})?
.ok_or_else(|| ApiError::not_found("Run not found.").into_response())?;
cached
.projection
.sandbox
.clone()
.and_then(fabro_types::RunSandbox::into_instance)
.ok_or_else(|| ApiError::not_found("Run sandbox was not created.").into_response())
}
#[cfg(test)]

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

@ -305,8 +305,8 @@ async fn attach_session_events(
};
let start_seq = match params.since_seq {
Some(seq) => seq.max(1),
None => match run_store.list_events().await {
Ok(events) => events.last().map_or(1, |event| event.seq.saturating_add(1)),
None => match run_store.last_event_seq().await {
Ok(last_seq) => last_seq.map_or(1, |seq| seq.saturating_add(1)),
Err(err) => return store_error(&err).into_response(),
},
};

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

@ -1591,6 +1591,48 @@ mod tests {
});
}
#[tokio::test]
async fn opening_cached_run_does_not_read_older_event_history() {
let (object_store, store) = make_store();
let run_id = test_run_id("run-1");
let run = store.create_run(&run_id).await.unwrap();
append_completed(&run, "run-1", dt("2026-03-27T12:00:00Z")).await;
let reopened = Database::new(object_store, "runs", Duration::from_millis(1), None);
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.
let mut unreadable_old_key = keys::run_event_seq_prefix(&run_id, 2).as_ref().to_vec();
unreadable_old_key.push(0xff);
reopened
.open_db()
.await
.unwrap()
.put(unreadable_old_key, b"invalid json")
.await
.unwrap();
let fresh_writer = reopened.open_run(&run_id).await.unwrap();
assert_eq!(fresh_writer.last_event_seq().await.unwrap(), Some(5));
let state = fresh_writer.state().await.unwrap();
assert_eq!(state.status, RunStatus::Succeeded {
reason: SuccessReason::Completed,
});
let seq = fresh_writer
.append_event(&event_payload(
"run-1",
"2026-03-27T12:00:05Z",
"run.title.updated",
&serde_json::json!({ "title": "Renamed completed run" }),
))
.await
.unwrap();
assert_eq!(seq, 6);
}
#[tokio::test]
async fn append_event_hydrates_local_projection_cache_for_fresh_writer() {
let (object_store, store) = make_store();

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

@ -171,6 +171,15 @@ impl RunProjectionCache {
.map(|entry| state.with_children_count(entry))
}
pub(crate) async fn last_seq(&self, run_id: &RunId) -> Option<u32> {
self.state
.lock()
.await
.entries
.get(run_id)
.map(|entry| entry.last_seq)
}
pub(crate) async fn get_summary(&self, run_id: &RunId, now: DateTime<Utc>) -> Option<Run> {
let mut entry = {
let state = self.state.lock().await;

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

@ -84,12 +84,27 @@ impl RunDatabase {
shared_projection_cache: Arc<RunProjectionCache>,
run_summary_store: Arc<OnceLock<Arc<RunSummaryStore>>>,
) -> Result<Self> {
let event_seq = if read_only {
// Readers never append, so they do not need to scan the full event
// history to recover the next write sequence.
1
} else {
recover_next_seq(&db, keys::run_events_prefix(&run_id), keys::parse_event_seq).await?
let cached_projection = shared_projection_cache.get(&run_id).await;
let projection_cache =
cached_projection
.as_ref()
.map_or_else(EventProjectionCache::default, |cached| {
EventProjectionCache {
last_seq: cached.last_seq,
state: Some((*cached.projection).clone()),
}
});
let event_seq = match (&cached_projection, read_only) {
(Some(cached), _) => cached.last_seq.saturating_add(1),
(None, true) => {
// Readers never append, so they do not need to scan the full event
// history to recover the next write sequence.
1
}
(None, false) => {
recover_next_seq(&db, keys::run_events_prefix(&run_id), keys::parse_event_seq)
.await?
}
};
let (event_tx, _) = broadcast::channel(DEFAULT_EVENT_TAIL_LIMIT.max(16));
let blob_store = BlobStore::new(Arc::new(db.clone()));
@ -101,7 +116,7 @@ impl RunDatabase {
event_seq: AtomicU32::new(event_seq),
close_lock: Mutex::new(()),
state_lock: Mutex::new(()),
projection_cache: Mutex::new(EventProjectionCache::default()),
projection_cache: Mutex::new(projection_cache),
shared_projection_cache,
run_summary_store,
recent_events: Mutex::new(VecDeque::with_capacity(DEFAULT_EVENT_TAIL_LIMIT)),
@ -368,6 +383,31 @@ impl RunDatabase {
self.list_events_from_with_limit(1, usize::MAX / 2).await
}
/// Returns the newest stored event sequence without reading event bodies
/// when a current local or shared projection is available.
pub async fn last_event_seq(&self) -> Result<Option<u32>> {
let local_last_seq = self.inner.projection_cache.lock().await.last_seq;
if local_last_seq > 0 {
return Ok(Some(local_last_seq));
}
if let Some(last_seq) = self
.inner
.shared_projection_cache
.last_seq(&self.inner.run_id)
.await
{
return Ok(Some(last_seq));
}
let next_seq = recover_next_seq(
&self.inner.db,
keys::run_events_prefix(&self.inner.run_id),
keys::parse_event_seq,
)
.await?;
Ok(next_seq.checked_sub(1).filter(|seq| *seq > 0))
}
pub async fn list_events_from_with_limit(
&self,
start_seq: u32,
@ -540,9 +580,15 @@ async fn list_events_from<R>(db: &R, run_id: &RunId, start_seq: u32) -> Result<V
where
R: DbRead + Sync,
{
let mut iter = db.scan_prefix(keys::run_events_prefix(run_id)).await?;
let event_prefix = keys::run_events_prefix(run_id);
let mut iter = db
.scan(keys::run_event_seq_prefix(run_id, start_seq)..)
.await?;
let mut events = Vec::new();
while let Some(entry) = iter.next().await? {
if !entry.key.starts_with(event_prefix.as_ref()) {
break;
}
let key = key_to_string(&entry.key)?;
let Some(seq) = keys::parse_event_seq(&key) else {
continue;
@ -624,9 +670,10 @@ async fn list_events_for_stage_from_with_limit<R>(
where
R: DbRead + Sync,
{
// Unbounded scan first: filtering by stage identity with a generic
// limit-bounded scan would silently drop matches whenever the stage's
// events are sparse late in the event log.
// Scan without a storage-level item limit from the requested cursor:
// filtering by stage identity with a generic limit-bounded scan would
// silently drop matches whenever the stage's events are sparse late in
// the event log.
//
// We probe just the stage identity fields with a small partial deserialize and
// only run the full `RunEvent` parse on matches. Most events in a run
@ -642,9 +689,15 @@ where
let stage_id_string = stage_id.to_string();
let max_events = limit.saturating_add(1);
let mut iter = db.scan_prefix(keys::run_events_prefix(run_id)).await?;
let event_prefix = keys::run_events_prefix(run_id);
let mut iter = db
.scan(keys::run_event_seq_prefix(run_id, start_seq)..)
.await?;
let mut events: Vec<EventEnvelope> = Vec::new();
while let Some(entry) = iter.next().await? {
if !entry.key.starts_with(event_prefix.as_ref()) {
break;
}
let key = key_to_string(&entry.key)?;
let Some(seq) = keys::parse_event_seq(&key) else {
continue;
@ -702,9 +755,15 @@ where
let session_id_string = session_id.to_string();
let max_events = limit.saturating_add(1);
let mut iter = db.scan_prefix(keys::run_events_prefix(run_id)).await?;
let event_prefix = keys::run_events_prefix(run_id);
let mut iter = db
.scan(keys::run_event_seq_prefix(run_id, start_seq)..)
.await?;
let mut events = Vec::new();
while let Some(entry) = iter.next().await? {
if !entry.key.starts_with(event_prefix.as_ref()) {
break;
}
let key = key_to_string(&entry.key)?;
let Some(seq) = keys::parse_event_seq(&key) else {
continue;
@ -980,6 +1039,36 @@ mod tests {
assert_eq!(seqs, vec![4, 6]);
}
#[tokio::test]
async fn list_events_for_stage_seeks_to_start_sequence() {
let run = fresh_run().await;
let run_id = run.run_id();
run.append_event(&stage_prompt_payload(&run_id, 1, Some("alpha")))
.await
.unwrap();
run.append_event(&stage_prompt_payload(&run_id, 2, Some("beta")))
.await
.unwrap();
run.append_event(&stage_prompt_payload(&run_id, 3, Some("alpha")))
.await
.unwrap();
let mut unreadable_earlier_key = keys::run_event_seq_prefix(&run_id, 2).as_ref().to_vec();
unreadable_earlier_key.push(0xff);
run.inner
.db
.put(unreadable_earlier_key, b"invalid json")
.await
.unwrap();
let events = run
.list_events_for_stage_from_with_limit(&StageId::new("alpha", 1), 3, 100)
.await
.unwrap();
let seqs: Vec<u32> = events.iter().map(|event| event.seq).collect();
assert_eq!(seqs, vec![4]);
}
#[tokio::test]
async fn list_events_for_stage_walks_past_unrelated_events_for_sparse_matches() {
let run = fresh_run().await;
@ -1088,6 +1177,38 @@ mod tests {
assert_eq!(seqs, vec![3, 5]);
}
#[tokio::test]
async fn list_events_for_session_seeks_to_start_sequence() {
let run = fresh_run().await;
let run_id = run.run_id();
let session_id = SessionId::new();
let other_session_id = SessionId::new();
run.append_event(&session_message_payload(&run_id, 1, session_id))
.await
.unwrap();
run.append_event(&session_message_payload(&run_id, 2, other_session_id))
.await
.unwrap();
run.append_event(&session_message_payload(&run_id, 3, session_id))
.await
.unwrap();
let mut unreadable_earlier_key = keys::run_event_seq_prefix(&run_id, 2).as_ref().to_vec();
unreadable_earlier_key.push(0xff);
run.inner
.db
.put(unreadable_earlier_key, b"invalid json")
.await
.unwrap();
let events = run
.list_events_for_session_from_with_limit(session_id, 3, 100)
.await
.unwrap();
let seqs: Vec<u32> = events.iter().map(|event| event.seq).collect();
assert_eq!(seqs, vec![4]);
}
#[tokio::test]
async fn list_events_for_session_returns_limit_plus_one_for_has_more_signal() {
let run = fresh_run().await;