From a94233407b4a2705ab776a7c150f35cfb0fa0c23 Mon Sep 17 00:00:00 2001 From: Fabro Date: Thu, 30 Jul 2026 00:22:06 +0000 Subject: [PATCH] fabro(01KYQMV1VW6139EGNHEM1RGF2G): simplify_sol (succeeded) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Fabro-Run: 01KYQMV1VW6139EGNHEM1RGF2G Fabro-Completed: 7 Fabro-Checkpoint: edcf7467e90f0195f72367b5c69a1b257bc72d1d ⚒️ Generated with [Fabro](https://fabro.sh) --- lib/apps/fabro-server/src/server/tests.rs | 54 ++++----- lib/components/fabro-store/src/error.rs | 7 +- lib/components/fabro-store/src/lib.rs | 4 +- .../fabro-store/src/run_summary_store.rs | 4 +- lib/components/fabro-store/src/slate/mod.rs | 40 +++--- .../fabro-store/src/slate/run_store.rs | 114 ++++++++++++------ .../src/{test_util.rs => test_support/mod.rs} | 22 +++- .../fabro-types/src/run_event/mod.rs | 53 +++++++- 8 files changed, 209 insertions(+), 89 deletions(-) rename lib/components/fabro-store/src/{test_util.rs => test_support/mod.rs} (55%) diff --git a/lib/apps/fabro-server/src/server/tests.rs b/lib/apps/fabro-server/src/server/tests.rs index 7735c5e99..063b48f36 100644 --- a/lib/apps/fabro-server/src/server/tests.rs +++ b/lib/apps/fabro-server/src/server/tests.rs @@ -6366,34 +6366,32 @@ async fn create_unreadable_durable_run(state: &Arc, run_id: RunId) { workflow_event::append_event(&run_store, &run_id, &workflow_event::Event::RunRunning) .await .unwrap(); - // Appends now validate before write, so a corrupted log can only exist if - // it predates that check. Write the poison event directly to simulate one. - state - .stores - .runs - .test_put_unvalidated_run_event( - &run_id, - 5, - &json!({ - "id": "evt-unreadable-run-completed", - "ts": "2026-05-05T20:46:33Z", - "run_id": run_id, - "event": "run.completed", - "properties": { - "timing": { - "wall_time_ms": 1, - "inference_time_ms": 0, - "tool_time_ms": 0, - "active_time_ms": 0 - }, - "artifact_count": 0, - "status": "legacy-status", - "reason": "completed", - }, - }), - ) - .await - .unwrap(); + let seq = run_store.last_event_seq().await.unwrap().unwrap() + 1; + let completed = workflow_event::to_run_event_at( + &run_id, + &workflow_event::Event::WorkflowRunCompleted { + timing: fabro_types::RunTiming::wall_only(1), + artifact_count: 0, + status: "legacy-status".to_string(), + reason: SuccessReason::Completed, + total_usd_micros: None, + final_git_commit_sha: None, + final_patch: None, + diff_summary: None, + billing: None, + }, + "2026-05-05T20:46:33Z".parse().unwrap(), + None, + ); + let payload = workflow_event::build_redacted_event_payload(&completed, &run_id).unwrap(); + fabro_store::test_support::put_unvalidated_run_event( + &state.stores.runs, + &run_id, + seq, + payload.as_value(), + ) + .await + .unwrap(); let err = run_store .state() .await diff --git a/lib/components/fabro-store/src/error.rs b/lib/components/fabro-store/src/error.rs index ff78213d0..df12e5994 100644 --- a/lib/components/fabro-store/src/error.rs +++ b/lib/components/fabro-store/src/error.rs @@ -14,8 +14,11 @@ pub enum Error { Io(#[from] std::io::Error), #[error("Invalid event payload: {0}")] InvalidEvent(String), - #[error("event rejected by run projection: {reason}")] - EventRejected { reason: String }, + #[error("event rejected by run projection: {source}")] + EventRejected { + #[source] + source: Box, + }, #[error("Run not found: {0}")] RunNotFound(String), #[error("Run already exists: {0}")] diff --git a/lib/components/fabro-store/src/lib.rs b/lib/components/fabro-store/src/lib.rs index 0eb246d04..e99eadd0f 100644 --- a/lib/components/fabro-store/src/lib.rs +++ b/lib/components/fabro-store/src/lib.rs @@ -10,8 +10,8 @@ mod run_state; mod run_summary_store; mod serializable_projection; mod slate; -#[cfg(test)] -mod test_util; +#[cfg(any(test, feature = "test-support"))] +pub mod test_support; mod types; pub use artifact_store::{ diff --git a/lib/components/fabro-store/src/run_summary_store.rs b/lib/components/fabro-store/src/run_summary_store.rs index c1a339b99..5db1a49b1 100644 --- a/lib/components/fabro-store/src/run_summary_store.rs +++ b/lib/components/fabro-store/src/run_summary_store.rs @@ -576,7 +576,7 @@ mod tests { RunSummaryVisibility, }; use crate::slate::CachedRunProjection; - use crate::test_util; + use crate::test_support as store_test_support; fn dt(value: &str) -> DateTime { value.parse().unwrap() @@ -613,7 +613,7 @@ mod tests { } async fn store() -> (tempfile::TempDir, RunSummaryStore) { - test_util::sqlite_summary_store().await + store_test_support::sqlite_summary_store().await } fn sample_status(kind: RunStatusKind) -> RunStatus { diff --git a/lib/components/fabro-store/src/slate/mod.rs b/lib/components/fabro-store/src/slate/mod.rs index d9f542498..893b0d8f0 100644 --- a/lib/components/fabro-store/src/slate/mod.rs +++ b/lib/components/fabro-store/src/slate/mod.rs @@ -322,10 +322,8 @@ impl Database { Ok(unreadable) } - /// Writes a raw event record without append validation, simulating a - /// pre-existing poison event in the log for unreadable-run tests. #[cfg(any(test, feature = "test-support"))] - pub async fn test_put_unvalidated_run_event( + pub(crate) async fn put_unvalidated_run_event( &self, run_id: &RunId, seq: u32, @@ -541,7 +539,7 @@ mod tests { use object_store::path::Path; use super::*; - use crate::{EventPayload, keys, test_util}; + use crate::{EventPayload, keys, test_support as store_test_support}; fn dt(value: &str) -> DateTime { value.parse().unwrap() @@ -590,7 +588,7 @@ mod tests { } async fn make_summary_store() -> (tempfile::TempDir, Arc) { - let (directory, store) = test_util::sqlite_summary_store().await; + let (directory, store) = store_test_support::sqlite_summary_store().await; (directory, Arc::new(store)) } @@ -931,13 +929,18 @@ mod tests { .await .unwrap_err(); - let Error::EventRejected { reason } = err else { + let Error::EventRejected { source } = err else { panic!("expected event rejection"); }; - assert_eq!( - reason, - "invalid status transition: runnable -> failed(workflow_error)" - ); + assert!(matches!( + *source, + Error::InvalidTransition(fabro_types::InvalidTransition { + from: RunStatus::Runnable, + to: RunStatus::Failed { + reason: FailureReason::WorkflowError, + }, + }) + )); assert_eq!(run.list_events().await.unwrap(), events_before); assert_eq!(run.state().await.unwrap().status, RunStatus::Runnable); let cached = store.get_cached_run(&run_id).await.unwrap().unwrap(); @@ -971,7 +974,7 @@ mod tests { #[tokio::test] async fn committed_append_succeeds_when_summary_update_fails_and_is_repairable() { - let (_object_store, store) = make_store(); + let (object_store, store) = make_store(); let (directory, summaries) = make_summary_store().await; store.attach_run_summary_store(Arc::clone(&summaries)); let run_id = test_run_id("run-1"); @@ -996,7 +999,8 @@ mod tests { let stored = run.get_event(2).await.unwrap().unwrap(); assert_eq!(stored.event, result.unwrap().event); - let repaired_summaries = test_util::sqlite_summary_store_at(directory.path()).await; + let repaired_summaries = + Arc::new(store_test_support::sqlite_summary_store_at(directory.path()).await); let stale = repaired_summaries .get(&run_id, Utc::now()) .await @@ -1004,11 +1008,9 @@ mod tests { .unwrap(); assert_ne!(stale.title, "Committed title"); - let entries = store - .list_cached_runs(&ListRunsQuery::default(), Utc::now()) - .await - .unwrap(); - repaired_summaries.reconcile(&entries).await.unwrap(); + let reopened = Database::new(object_store, "runs/", Duration::from_millis(1), None); + reopened.attach_run_summary_store(Arc::clone(&repaired_summaries)); + reopened.warm_projection_cache().await.unwrap(); let repaired = repaired_summaries .get(&run_id, Utc::now()) .await @@ -1541,7 +1543,7 @@ mod tests { .await .unwrap(); store - .test_put_unvalidated_run_event( + .put_unvalidated_run_event( &bad_run_id, 1, &serde_json::json!({ "not": "a valid run event" }), @@ -1590,7 +1592,7 @@ mod tests { .unwrap(); run_settings.remove("integrations"); store - .test_put_unvalidated_run_event( + .put_unvalidated_run_event( &bad_run_id, 1, &serde_json::json!({ diff --git a/lib/components/fabro-store/src/slate/run_store.rs b/lib/components/fabro-store/src/slate/run_store.rs index 57f54a311..be9ca78bf 100644 --- a/lib/components/fabro-store/src/slate/run_store.rs +++ b/lib/components/fabro-store/src/slate/run_store.rs @@ -231,10 +231,10 @@ impl RunDatabase { self.projected_state_option_locked().await } - async fn install_derived_state_after_append( + async fn install_in_memory_state_after_append( &self, event: &EventEnvelope, - cached: CachedRunProjection, + cached: &CachedRunProjection, ) { { let mut projection_cache = self.inner.projection_cache.lock().await; @@ -253,14 +253,16 @@ impl RunDatabase { } drop(recent_events); let _ = self.inner.event_tx.send(event.clone()); + } + async fn update_summary_after_committed_append(&self, cached: &CachedRunProjection) { if let Some(store) = self.inner.run_summary_store.get() { - if let Err(err) = store.upsert_projection(&cached).await { + if let Err(err) = store.upsert_projection(cached).await { warn!( run_id = %self.inner.run_id, - source_last_seq = event.seq, - error = %err, - "Failed to update SQLite run summary after committed append" + source_last_seq = cached.last_seq, + error = ?err, + "failed to update SQLite run summary after committed append" ); } } @@ -312,12 +314,19 @@ impl RunDatabase { return Err(Error::ReadOnly); } payload.validate(&self.inner.run_id)?; - let _state_guard = self.inner.state_lock.lock().await; - let projection = self.projected_state_locked().await?; - if !predicate(&projection) { - return Ok(None); - } - Ok(Some(self.append_event_envelope_locked(payload).await?.seq)) + let (envelope, cached) = { + let _state_guard = self.inner.state_lock.lock().await; + let projection = self.projected_state_locked().await?; + if !predicate(&projection) { + return Ok(None); + } + let event = RunEvent::try_from(payload)?; + let event_bytes = serde_json::to_vec(payload)?; + self.append_event_envelope_locked(event, event_bytes) + .await? + }; + self.update_summary_after_committed_append(&cached).await; + Ok(Some(envelope.seq)) } /// Appends and returns the stored event envelope after pre-write reduction. @@ -331,23 +340,30 @@ impl RunDatabase { return Err(Error::ReadOnly); } payload.validate(&self.inner.run_id)?; - let _state_guard = self.inner.state_lock.lock().await; - self.append_event_envelope_locked(payload).await + let event = RunEvent::try_from(payload)?; + let event_bytes = serde_json::to_vec(payload)?; + let (envelope, cached) = { + let _state_guard = self.inner.state_lock.lock().await; + self.append_event_envelope_locked(event, event_bytes) + .await? + }; + self.update_summary_after_committed_append(&cached).await; + Ok(envelope) } - async fn append_event_envelope_locked(&self, payload: &EventPayload) -> Result { + async fn append_event_envelope_locked( + &self, + event: RunEvent, + event_bytes: Vec, + ) -> Result<(EventEnvelope, CachedRunProjection)> { let event_seq = self.inner.event_seq.as_ref().ok_or(Error::ReadOnly)?; - let seq = allocate_event_seq(event_seq)?; - let event = EventEnvelope { - seq, - event: RunEvent::try_from(payload)?, - }; + let seq = next_event_seq(event_seq)?; + let envelope = EventEnvelope { seq, event }; // Validation reduces through the exact code replay uses, so an event // is written iff replay can reduce it. `Arc::make_mut` copy-on-writes, // leaving the local projection cache untouched on rejection. let mut next_state = self.projected_state_for_append_locked(seq).await?; - apply_cached_projection_event(&mut next_state, &event) - .map_err(|err| event_rejected(&err))?; + apply_cached_projection_event(&mut next_state, &envelope).map_err(event_rejected)?; let next_projection = next_state.expect("apply_cached_projection_event sets the state on success"); let cached = CachedRunProjection::from_projection( @@ -355,7 +371,7 @@ impl RunDatabase { Arc::unwrap_or_clone(next_projection), seq, ); - let event_bytes = serde_json::to_vec(payload)?; + reserve_event_seq(event_seq, seq)?; self.inner .db .put( @@ -366,8 +382,8 @@ impl RunDatabase { // Box::pin keeps this future small enough for the // clippy::large_futures budget of append_event_envelope's many // callers. - Box::pin(self.install_derived_state_after_append(&event, cached)).await; - Ok(event) + Box::pin(self.install_in_memory_state_after_append(&envelope, &cached)).await; + Ok((envelope, cached)) } pub async fn list_events(&self) -> Result> { @@ -583,20 +599,27 @@ impl RunDatabase { } } -fn event_rejected(error: &Error) -> Error { +fn event_rejected(error: Error) -> Error { Error::EventRejected { - reason: error.to_string(), + source: Box::new(error), } } -fn allocate_event_seq(event_seq: &AtomicU32) -> Result { - event_seq - .fetch_update(Ordering::SeqCst, Ordering::SeqCst, |seq| { - (seq <= keys::MAX_EVENT_SEQ).then_some(seq + 1) - }) - .map_err(|_| Error::EventSequenceExhausted { +fn next_event_seq(event_seq: &AtomicU32) -> Result { + let seq = event_seq.load(Ordering::SeqCst); + if seq > keys::MAX_EVENT_SEQ { + return Err(Error::EventSequenceExhausted { max_seq: keys::MAX_EVENT_SEQ, - }) + }); + } + Ok(seq) +} + +fn reserve_event_seq(event_seq: &AtomicU32, seq: u32) -> Result<()> { + event_seq + .compare_exchange(seq, seq + 1, Ordering::SeqCst, Ordering::SeqCst) + .map(|_| ()) + .map_err(|_| Error::Other("event sequence changed while append lock was held".to_string())) } fn apply_cached_projection_event( @@ -1288,6 +1311,29 @@ mod tests { assert_eq!(seqs, vec![5, 4, 3]); } + #[tokio::test] + async fn rejected_event_does_not_consume_last_available_sequence() { + let run = fresh_run().await; + let run_id = run.run_id(); + run.inner + .event_seq + .as_ref() + .unwrap() + .store(keys::MAX_EVENT_SEQ, Ordering::SeqCst); + + let err = run + .append_event(&run_created_payload(&run_id)) + .await + .unwrap_err(); + assert!(matches!(err, Error::EventRejected { .. })); + + let seq = run + .append_event(&stage_prompt_payload(&run_id, 1, Some("alpha"))) + .await + .unwrap(); + assert_eq!(seq, keys::MAX_EVENT_SEQ); + } + #[tokio::test] async fn append_event_rejects_sequences_beyond_key_order_limit() { let run = fresh_run().await; diff --git a/lib/components/fabro-store/src/test_util.rs b/lib/components/fabro-store/src/test_support/mod.rs similarity index 55% rename from lib/components/fabro-store/src/test_util.rs rename to lib/components/fabro-store/src/test_support/mod.rs index a5b7e17f9..7ae613652 100644 --- a/lib/components/fabro-store/src/test_util.rs +++ b/lib/components/fabro-store/src/test_support/mod.rs @@ -1,13 +1,33 @@ +#[cfg(test)] use std::path::Path; -use crate::RunSummaryStore; +use fabro_types::RunId; +#[cfg(test)] +use crate::RunSummaryStore; +use crate::{Database, Result}; + +/// Writes an event without append validation to model a log corrupted by an +/// older Fabro version. +pub async fn put_unvalidated_run_event( + database: &Database, + run_id: &RunId, + seq: u32, + payload: &serde_json::Value, +) -> Result<()> { + database + .put_unvalidated_run_event(run_id, seq, payload) + .await +} + +#[cfg(test)] pub(crate) async fn sqlite_summary_store() -> (tempfile::TempDir, RunSummaryStore) { let directory = tempfile::tempdir().unwrap(); let store = sqlite_summary_store_at(directory.path()).await; (directory, store) } +#[cfg(test)] pub(crate) async fn sqlite_summary_store_at(directory: &Path) -> RunSummaryStore { let database = fabro_db::Database::connect(directory.join("fabro.sqlite3")) .await diff --git a/lib/foundation/fabro-types/src/run_event/mod.rs b/lib/foundation/fabro-types/src/run_event/mod.rs index 34cc88ed8..6680fff2e 100644 --- a/lib/foundation/fabro-types/src/run_event/mod.rs +++ b/lib/foundation/fabro-types/src/run_event/mod.rs @@ -12,7 +12,7 @@ pub use fabro_model::BilledTokenCounts; pub use infra::*; pub use misc::*; pub use run::*; -use serde::de::Error as _; +use serde::de::Error as DeError; use serde::ser::Error as SerError; use serde::{Deserialize, Deserializer, Serialize, Serializer}; use serde_json::{Map, Value, json}; @@ -764,6 +764,57 @@ impl RunEvent { }) } + pub fn from_ref(value: &Value) -> serde_json::Result { + fn opt_field Deserialize<'a>>( + obj: &Map, + key: &str, + ) -> serde_json::Result> { + match obj.get(key) { + Some(value) if !value.is_null() => Ok(Some(T::deserialize(value)?)), + _ => Ok(None), + } + } + + let obj = value.as_object().ok_or_else(|| { + ::custom("run event must be a JSON object") + })?; + let opt_str = |key: &str| obj.get(key).and_then(Value::as_str).map(str::to_string); + let id = obj.get("id").and_then(Value::as_str).ok_or_else(|| { + ::custom("missing or non-string field: id") + })?; + let ts = obj + .get("ts") + .ok_or_else(|| ::custom("missing field: ts")) + .and_then(DateTime::::deserialize)?; + let run_id = obj + .get("run_id") + .ok_or_else(|| ::custom("missing field: run_id")) + .and_then(RunId::deserialize)?; + let event = obj.get("event").and_then(Value::as_str).ok_or_else(|| { + ::custom("missing or non-string field: event") + })?; + let properties = obj + .get("properties") + .cloned() + .unwrap_or_else(default_properties); + Self::from_parts(RunEventParts { + id: id.to_string(), + ts, + run_id, + node_id: opt_str("node_id"), + node_label: opt_str("node_label"), + stage_id: opt_field(obj, "stage_id")?, + parallel_group_id: opt_field(obj, "parallel_group_id")?, + parallel_branch_id: opt_field(obj, "parallel_branch_id")?, + session_id: opt_str("session_id"), + parent_session_id: opt_str("parent_session_id"), + tool_call_id: opt_str("tool_call_id"), + actor: opt_field(obj, "actor")?, + event, + properties: &properties, + }) + } + fn from_parts(parts: RunEventParts<'_>) -> serde_json::Result { let body_payload = json!({ "event": parts.event,