From d787531fa7685ef6d439b1c09727f1cc7a022de0 Mon Sep 17 00:00:00 2001 From: Fabro Date: Wed, 8 Jul 2026 21:43:08 +0000 Subject: [PATCH] fabro(01KX1P0VV0DAQTT0N2NADX8J8J): simplify_fable (succeeded) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Fabro-Run: 01KX1P0VV0DAQTT0N2NADX8J8J Fabro-Completed: 6 ⚒️ Generated with [Fabro](https://fabro.sh) --- .../fabro-cli/src/commands/run/runner.rs | 9 ++-- .../fabro-redact/src/secret_registry.rs | 22 +++----- .../fabro-server/src/server/handler/events.rs | 31 +++++------ lib/crates/fabro-workflow/src/event.rs | 2 +- .../fabro-workflow/src/event/redaction.rs | 43 +++++++++++---- lib/crates/fabro-workflow/src/event/sink.rs | 54 +++++++++++++++---- .../fabro-workflow/src/handler/command.rs | 5 +- .../fabro-workflow/src/lifecycle/git.rs | 2 +- .../fabro-workflow/src/pipeline/finalize.rs | 2 +- .../fabro-workflow/src/runtime_store.rs | 27 +++++----- 10 files changed, 120 insertions(+), 77 deletions(-) diff --git a/lib/crates/fabro-cli/src/commands/run/runner.rs b/lib/crates/fabro-cli/src/commands/run/runner.rs index 2fca9a250..88547c52b 100644 --- a/lib/crates/fabro-cli/src/commands/run/runner.rs +++ b/lib/crates/fabro-cli/src/commands/run/runner.rs @@ -26,7 +26,7 @@ use fabro_types::{ }; use fabro_vault::Vault; use fabro_workflow::artifact_upload::{ArtifactSink, StageArtifactUploader}; -use fabro_workflow::event::{Emitter, RunEventSink}; +use fabro_workflow::event::{Emitter, RedactedRunEvent, RunEventSink}; use fabro_workflow::operations::{self, StartServices}; use fabro_workflow::run_control::RunControlState; use fabro_workflow::runtime_store::{RunStoreBackend, RunStoreHandle}; @@ -178,12 +178,12 @@ pub(crate) async fn execute( fabro_run_tools, }; - let execution = Box::pin(async { + let execution = async { match mode { RunWorkerMode::Start => operations::start(&run_dir, services).await, RunWorkerMode::Resume => operations::resume(&run_dir, services).await, } - }); + }; if let Some(mut control_manager) = control_manager { tokio::select! { @@ -1005,7 +1005,8 @@ impl RunStoreBackend for HttpRunStore { Ok(events) } - async fn append_run_event(&self, event: &RunEvent) -> Result<()> { + async fn append_run_event(&self, event: &RedactedRunEvent) -> Result<()> { + let event = event.event(); let seq = Box::pin(self.with_retries("append run event", || { let client = self.client.clone_for_reuse(); let run_id = self.run_id; diff --git a/lib/crates/fabro-redact/src/secret_registry.rs b/lib/crates/fabro-redact/src/secret_registry.rs index 764eb0fe5..414121787 100644 --- a/lib/crates/fabro-redact/src/secret_registry.rs +++ b/lib/crates/fabro-redact/src/secret_registry.rs @@ -40,9 +40,10 @@ impl SecretRedactor { /// Redact all registered secret values from `s`. pub fn redact_into(&self, s: &str) -> String { - let Some(values) = self.values_snapshot() else { + let values = self.read(); + if values.is_empty() { return s.to_string(); - }; + } redact_string_values(s, &values) } @@ -50,11 +51,10 @@ impl SecretRedactor { /// /// Object keys and non-string values are left unchanged. pub fn redact_json(&self, mut value: Value) -> Value { - let Some(values) = self.values_snapshot() else { - return value; - }; - - redact_json_leaves(&mut value, &values); + let values = self.read(); + if !values.is_empty() { + redact_json_leaves(&mut value, &values); + } value } @@ -65,14 +65,6 @@ impl SecretRedactor { fn write(&self) -> RwLockWriteGuard<'_, Vec> { self.values.write().unwrap_or_else(PoisonError::into_inner) } - - fn values_snapshot(&self) -> Option> { - let values = self.read(); - if values.is_empty() { - return None; - } - Some(values.clone()) - } } fn redact_json_leaves(value: &mut Value, values: &[String]) { diff --git a/lib/crates/fabro-server/src/server/handler/events.rs b/lib/crates/fabro-server/src/server/handler/events.rs index bfebaeebf..ce3bb3281 100644 --- a/lib/crates/fabro-server/src/server/handler/events.rs +++ b/lib/crates/fabro-server/src/server/handler/events.rs @@ -160,33 +160,26 @@ async fn append_run_event( if let Some(response) = reject_if_archived(state.as_ref(), &id).await { return response; } - let event = match RunEvent::from_value(value.clone()) { - Ok(event) => event, - Err(err) => { - return ApiError::bad_request(format!("Invalid run event: {err}")).into_response(); - } - }; - if event.run_id != id { - return ApiError::bad_request("Event run_id does not match path run ID.").into_response(); - } - if let Some(denied) = denied_lifecycle_event_name(&event.body) { - return ApiError::bad_request(format!( - "{denied} is a lifecycle event; clients must call the corresponding operation endpoint instead of injecting it via append_run_event" - )) - .into_response(); - } - let redacted_value = redact_json_value(value); - let payload = match EventPayload::new(redacted_value, &id) { + // Ingest trust boundary: content-redact before validation and storage so + // credential-shaped values from arbitrary clients never reach the store. + // `EventPayload::new` validates required fields and the run_id match, and + // the single parse below feeds both storage checks and the live-run update. + let payload = match EventPayload::new(redact_json_value(value), &id) { Ok(payload) => payload, Err(err) => return ApiError::bad_request(err.to_string()).into_response(), }; let event = match RunEvent::from_ref(payload.as_value()) { Ok(event) => event, Err(err) => { - return ApiError::bad_request(format!("Invalid redacted run event: {err}")) - .into_response(); + return ApiError::bad_request(format!("Invalid run event: {err}")).into_response(); } }; + if let Some(denied) = denied_lifecycle_event_name(&event.body) { + return ApiError::bad_request(format!( + "{denied} is a lifecycle event; clients must call the corresponding operation endpoint instead of injecting it via append_run_event" + )) + .into_response(); + } match state.stores.runs.open_run(&id).await { Ok(run_store) => match run_store.append_event(&payload).await { diff --git a/lib/crates/fabro-workflow/src/event.rs b/lib/crates/fabro-workflow/src/event.rs index e64b2ae30..9326f8471 100644 --- a/lib/crates/fabro-workflow/src/event.rs +++ b/lib/crates/fabro-workflow/src/event.rs @@ -15,7 +15,7 @@ pub use self::emitter::Emitter; pub use self::events::Event; pub use self::names::event_name; pub use self::redaction::{ - build_redacted_event_payload, event_payload_from_redacted_json, redacted_event_json, + RedactedRunEvent, build_redacted_event_payload, event_payload_from_redacted_json, }; pub use self::sink::{ RunEventLogger, RunEventSink, StoreProgressLogger, append_event, append_event_to_sink, diff --git a/lib/crates/fabro-workflow/src/event/redaction.rs b/lib/crates/fabro-workflow/src/event/redaction.rs index 1dca025d3..5e5cc97e4 100644 --- a/lib/crates/fabro-workflow/src/event/redaction.rs +++ b/lib/crates/fabro-workflow/src/event/redaction.rs @@ -14,13 +14,36 @@ pub fn build_redacted_event_payload( EventPayload::new(value, run_id).map_err(anyhow::Error::from) } -pub fn redacted_event_json(event: &RunEvent, redactor: &SecretRedactor) -> Result { - serde_json::to_string(&redacted_event_value(event, redactor)?).map_err(anyhow::Error::from) +/// A run event passed through the full redaction pipeline exactly once. +/// +/// The redacted JSON value feeds payload and JSON-line surfaces, and the typed +/// event is reparsed from that same value, so every downstream consumer +/// observes identical redacted data without re-running the redaction passes. +pub struct RedactedRunEvent { + event: RunEvent, + value: Value, } -pub(super) fn redacted_run_event(event: &RunEvent, redactor: &SecretRedactor) -> Result { - let value = redacted_event_value(event, redactor)?; - RunEvent::from_ref(&value).context("Failed to reparse redacted event payload") +impl RedactedRunEvent { + pub fn new(event: &RunEvent, redactor: &SecretRedactor) -> Result { + let value = redacted_event_value(event, redactor)?; + let event = + RunEvent::from_ref(&value).context("Failed to reparse redacted event payload")?; + Ok(Self { event, value }) + } + + #[must_use] + pub fn event(&self) -> &RunEvent { + &self.event + } + + pub fn payload(&self) -> Result { + EventPayload::new(self.value.clone(), &self.event.run_id).map_err(anyhow::Error::from) + } + + pub fn json_line(&self) -> Result { + serde_json::to_string(&self.value).map_err(anyhow::Error::from) + } } fn normalized_event_value(event: &RunEvent) -> Result { @@ -40,10 +63,7 @@ fn redact_registered_secrets_in_event_value(value: &mut Value, redactor: &Secret } if let Some(Value::String(node_label)) = value.get_mut("node_label") { - let redacted = redactor.redact_into(node_label); - if redacted != *node_label { - *node_label = redacted; - } + *node_label = redactor.redact_into(node_label); } if let Some(properties) = value.get_mut("properties") { @@ -271,7 +291,10 @@ mod tests { fabro_redact::redact_json_value(normalized_event_value(&stored).unwrap()); let content_only_json = serde_json::to_string(&content_only).unwrap(); - let with_empty_registry = redacted_event_json(&stored, &SecretRedactor::default()).unwrap(); + let with_empty_registry = RedactedRunEvent::new(&stored, &SecretRedactor::default()) + .unwrap() + .json_line() + .unwrap(); assert_eq!(with_empty_registry, content_only_json); } diff --git a/lib/crates/fabro-workflow/src/event/sink.rs b/lib/crates/fabro-workflow/src/event/sink.rs index bb9bc7c69..89f6d4020 100644 --- a/lib/crates/fabro-workflow/src/event/sink.rs +++ b/lib/crates/fabro-workflow/src/event/sink.rs @@ -10,7 +10,7 @@ use tokio::io::{AsyncWrite, AsyncWriteExt}; use tokio::sync::{Mutex as AsyncMutex, mpsc, oneshot}; use super::emitter::Emitter; -use super::redaction::{build_redacted_event_payload, redacted_event_json, redacted_run_event}; +use super::redaction::{RedactedRunEvent, build_redacted_event_payload}; use super::{Event, to_run_event}; use crate::runtime_store::RunStoreHandle; @@ -105,30 +105,36 @@ impl RunEventSink { } pub async fn write_run_event(&self, event: &RunEvent, redactor: &SecretRedactor) -> Result<()> { - let mut pending = vec![(self, event.clone())]; + let mut pending = vec![(self, PendingEvent::Raw(event.clone()))]; while let Some((sink, event)) = pending.pop() { match sink { Self::Store(run_store) => { - let event = redacted_run_event(&event, redactor)?; - run_store.append_run_event(&event).await?; + let redacted = event.into_redacted(redactor)?; + run_store.append_run_event(&redacted).await?; } Self::JsonLines(writer) => { - let line = redacted_event_json(&event, redactor)?; + let line = event.into_redacted(redactor)?.json_line()?; let mut writer = writer.lock().await; writer.write_all(line.as_bytes()).await?; writer.write_all(b"\n").await?; writer.flush().await?; } Self::Callback(callback) => { - let event = redacted_run_event(&event, redactor)?; - callback(event).await?; + callback(event.into_redacted(redactor)?.event().clone()).await?; } Self::Map { transform, inner } => { - pending.push((inner.as_ref(), transform(event))); + // A transform can add new fields, so its output re-enters + // the pipeline as a raw event and is redacted downstream. + pending.push(( + inner.as_ref(), + PendingEvent::Raw(transform(event.into_run_event())), + )); } Self::Composite(sinks) => { + // Redact once and share the result with every branch. + let redacted = event.into_redacted(redactor)?; for sink in sinks.iter().rev() { - pending.push((sink, event.clone())); + pending.push((sink, PendingEvent::Redacted(Arc::clone(&redacted)))); } } } @@ -137,6 +143,33 @@ impl RunEventSink { } } +/// Worklist entry for [`RunEventSink::write_run_event`]: events are redacted at +/// most once per fan-out and the shared result reused by every consumer. +#[allow( + clippy::large_enum_variant, + reason = "Worklist entries stay inline to avoid boxing hot-path payloads." +)] +enum PendingEvent { + Raw(RunEvent), + Redacted(Arc), +} + +impl PendingEvent { + fn into_redacted(self, redactor: &SecretRedactor) -> Result> { + match self { + Self::Raw(event) => Ok(Arc::new(RedactedRunEvent::new(&event, redactor)?)), + Self::Redacted(redacted) => Ok(redacted), + } + } + + fn into_run_event(self) -> RunEvent { + match self { + Self::Raw(event) => event, + Self::Redacted(redacted) => redacted.event().clone(), + } + } +} + #[allow( clippy::large_enum_variant, reason = "Logger queue messages stay inline to avoid boxing hot-path payloads." @@ -155,13 +188,12 @@ impl RunEventLogger { #[must_use] pub fn new(sink: RunEventSink, redactor: SecretRedactor) -> Self { let (tx, mut rx) = mpsc::unbounded_channel(); - let logger_redactor = redactor; tokio::spawn(async move { while let Some(command) = rx.recv().await { match command { RunEventCommand::Event(event) => { - if let Err(err) = sink.write_run_event(&event, &logger_redactor).await { + if let Err(err) = sink.write_run_event(&event, &redactor).await { tracing::warn!(error = %err, "Failed to write run event"); } } diff --git a/lib/crates/fabro-workflow/src/handler/command.rs b/lib/crates/fabro-workflow/src/handler/command.rs index 09186ffc3..a22b8f6bc 100644 --- a/lib/crates/fabro-workflow/src/handler/command.rs +++ b/lib/crates/fabro-workflow/src/handler/command.rs @@ -263,7 +263,10 @@ mod tests { Ok(Vec::new()) } - async fn append_run_event(&self, _event: &fabro_types::RunEvent) -> anyhow::Result<()> { + async fn append_run_event( + &self, + _event: &crate::event::RedactedRunEvent, + ) -> anyhow::Result<()> { Ok(()) } diff --git a/lib/crates/fabro-workflow/src/lifecycle/git.rs b/lib/crates/fabro-workflow/src/lifecycle/git.rs index a1aab9bc7..ca7d51c2e 100644 --- a/lib/crates/fabro-workflow/src/lifecycle/git.rs +++ b/lib/crates/fabro-workflow/src/lifecycle/git.rs @@ -1305,7 +1305,7 @@ mod tests { Ok(Vec::new()) } - async fn append_run_event(&self, _event: &RunEvent) -> Result<()> { + async fn append_run_event(&self, _event: &crate::event::RedactedRunEvent) -> Result<()> { Ok(()) } diff --git a/lib/crates/fabro-workflow/src/pipeline/finalize.rs b/lib/crates/fabro-workflow/src/pipeline/finalize.rs index 4d00dda40..c6a61ec6d 100644 --- a/lib/crates/fabro-workflow/src/pipeline/finalize.rs +++ b/lib/crates/fabro-workflow/src/pipeline/finalize.rs @@ -1416,7 +1416,7 @@ mod tests { Ok(Vec::new()) } - async fn append_run_event(&self, _event: &RunEvent) -> Result<()> { + async fn append_run_event(&self, _event: &crate::event::RedactedRunEvent) -> Result<()> { Ok(()) } diff --git a/lib/crates/fabro-workflow/src/runtime_store.rs b/lib/crates/fabro-workflow/src/runtime_store.rs index 82bb8dba3..899271c44 100644 --- a/lib/crates/fabro-workflow/src/runtime_store.rs +++ b/lib/crates/fabro-workflow/src/runtime_store.rs @@ -3,17 +3,19 @@ use std::sync::Arc; use anyhow::Result; use async_trait::async_trait; use bytes::Bytes; -use fabro_redact::SecretRedactor; use fabro_store::{EventEnvelope, RunDatabase, RunProjection}; -use fabro_types::{RunBlobId, RunEvent}; +use fabro_types::RunBlobId; -use crate::event::build_redacted_event_payload; +use crate::event::RedactedRunEvent; #[async_trait] pub trait RunStoreBackend: Send + Sync { async fn load_state(&self) -> Result; async fn list_events(&self) -> Result>; - async fn append_run_event(&self, event: &RunEvent) -> Result<()>; + /// Append an event that already passed the redaction pipeline. Taking + /// [`RedactedRunEvent`] keeps unredacted events out of every backend and + /// lets backends reuse the single redaction pass instead of re-running it. + async fn append_run_event(&self, event: &RedactedRunEvent) -> Result<()>; async fn write_blob(&self, data: &[u8]) -> Result; async fn read_blob(&self, id: &RunBlobId) -> Result>; async fn read_run_log(&self) -> Result>>; @@ -43,7 +45,7 @@ impl RunStoreHandle { self.backend.list_events().await } - pub async fn append_run_event(&self, event: &RunEvent) -> Result<()> { + pub async fn append_run_event(&self, event: &RedactedRunEvent) -> Result<()> { self.backend.append_run_event(event).await } @@ -83,14 +85,9 @@ impl RunStoreBackend for LocalRunStoreBackend { .map_err(anyhow::Error::from) } - async fn append_run_event(&self, event: &RunEvent) -> Result<()> { - // Local backends can be called from server/test paths that do not have a - // worker run registry. RunEventSink applies the per-run redactor before - // calling this backend during workflow execution. - let payload = - build_redacted_event_payload(event, &event.run_id, &SecretRedactor::default())?; + async fn append_run_event(&self, event: &RedactedRunEvent) -> Result<()> { self.run_store - .append_event(&payload) + .append_event(&event.payload()?) .await .map(|_| ()) .map_err(anyhow::Error::from) @@ -129,7 +126,7 @@ mod tests { use object_store::memory::InMemory; use super::RunStoreHandle; - use crate::event::{Event, append_event}; + use crate::event::{Event, RedactedRunEvent, append_event}; use crate::records::RunSpec; async fn test_run_store() -> fabro_store::RunDatabase { @@ -223,7 +220,9 @@ mod tests { definition_blob: None, }), }; - handle.append_run_event(&event).await.unwrap(); + let redacted = + RedactedRunEvent::new(&event, &fabro_redact::SecretRedactor::default()).unwrap(); + handle.append_run_event(&redacted).await.unwrap(); let blob_id = handle.write_blob(br#"{"ok":true}"#).await.unwrap(); let blob = handle.read_blob(&blob_id).await.unwrap().unwrap();