From 642433a8b82980322c3845c4d3e2b354d9380d25 Mon Sep 17 00:00:00 2001 From: ScriptedAlchemy Date: Sun, 18 Oct 2026 16:14:11 +0000 Subject: [PATCH 01/10] feat(global-db): add transact_or and JobExit for job bodies --- .../src/configuration/store.rs | 2 +- .../src/observability_rollup.rs | 8 ++- .../src/observability_rollup/retention.rs | 4 +- .../src/observation_projection/apply.rs | 1 + crates/tracedecay-global-db/src/registered.rs | 71 +++++++++++++------ .../src/registry_maintenance.rs | 2 +- .../src/runtime/workflow/workflow_index.rs | 2 +- .../runtime/workflow/workflow_index/port.rs | 6 -- 8 files changed, 61 insertions(+), 35 deletions(-) diff --git a/crates/tracedecay-global-db/src/configuration/store.rs b/crates/tracedecay-global-db/src/configuration/store.rs index e8eed26376..b5401b5075 100644 --- a/crates/tracedecay-global-db/src/configuration/store.rs +++ b/crates/tracedecay-global-db/src/configuration/store.rs @@ -75,7 +75,7 @@ where R: Send + 'static, E: Send + 'static, { - db.transact(body).await.map_err(unavailable)? + db.transact_or(unavailable, body).await } /// Concrete control-plane adapter over one already-open owned session store. diff --git a/crates/tracedecay-global-db/src/observability_rollup.rs b/crates/tracedecay-global-db/src/observability_rollup.rs index 06b8f0c0cc..dce17ddbd5 100644 --- a/crates/tracedecay-global-db/src/observability_rollup.rs +++ b/crates/tracedecay-global-db/src/observability_rollup.rs @@ -41,9 +41,11 @@ where R: Send + 'static, E: From + Send + 'static, { - db.transact(body) - .await - .map_err(|error| E::from(format!("failed to {context}: {error}")))? + db.transact_or( + |error| E::from(format!("failed to {context}: {error}")), + body, + ) + .await } /// A rebuild job that ends without publishing. A replayed receipt rolls the diff --git a/crates/tracedecay-global-db/src/observability_rollup/retention.rs b/crates/tracedecay-global-db/src/observability_rollup/retention.rs index 5d8880efb0..23a22bd8c0 100644 --- a/crates/tracedecay-global-db/src/observability_rollup/retention.rs +++ b/crates/tracedecay-global-db/src/observability_rollup/retention.rs @@ -1,4 +1,4 @@ -use crate::RegisteredGlobalDbWriteTransaction; +use tracedecay_runtime_core::db::engine::Executor; use super::{ OBSERVABILITY_ROLLUP_RETENTION_DAYS_V1, ObservabilityRollupRetentionReceiptV1, SECONDS_PER_DAY, @@ -11,7 +11,7 @@ use super::{ /// expired day as known-empty from already-pruned events. A query reaching /// past the horizon reports partial coverage. pub(crate) async fn expire_observability_rollups( - transaction: &RegisteredGlobalDbWriteTransaction<'_>, + transaction: &impl Executor, now_seconds: i64, ) -> Result { let cutoff = diff --git a/crates/tracedecay-global-db/src/observation_projection/apply.rs b/crates/tracedecay-global-db/src/observation_projection/apply.rs index aefe8f89fa..533ef92e4b 100644 --- a/crates/tracedecay-global-db/src/observation_projection/apply.rs +++ b/crates/tracedecay-global-db/src/observation_projection/apply.rs @@ -679,6 +679,7 @@ impl_send_projection_derive!(DatabaseWriteTransaction<'_>); impl_send_projection_derive!(DatabaseEngineReadConnection); impl_send_projection_derive!(DatabaseEngineReadSnapshot); impl_send_projection_derive!(crate::registered::RegisteredGlobalDbWriteTransaction<'_>); +impl_send_projection_derive!(crate::registered::RegisteredGlobalDbWriteTx<'_, '_>); /// Objective used for goal-state dedupe: prefer native `/objective` (Codex), /// else the already-extracted `content_text`. diff --git a/crates/tracedecay-global-db/src/registered.rs b/crates/tracedecay-global-db/src/registered.rs index e3e52c44f6..23b9cbec73 100644 --- a/crates/tracedecay-global-db/src/registered.rs +++ b/crates/tracedecay-global-db/src/registered.rs @@ -634,6 +634,42 @@ impl RegisteredGlobalDb { .await } + /// [`Self::transact`] for a body whose error type is the caller's own: a + /// refused or failed job becomes `wrap(cause)`, and the body's error is + /// returned as is. + pub(crate) async fn transact_or( + &self, + wrap: impl FnOnce(TraceDecayError) -> E, + body: impl AsyncFn(RegisteredGlobalDbWriteTx<'_, '_>) -> Result + Send + Sync + 'static, + ) -> Result + where + R: Send + 'static, + E: Send + 'static, + { + self.transact(body).await.map_err(wrap)? + } + + /// [`Self::transact_or`] for a body that can also end early with a value + /// the caller wants as success. The job rolls back either way, so an + /// early exit leaves no commit behind. + pub(crate) async fn transact_or_rollback( + &self, + wrap: impl FnOnce(TraceDecayError) -> E, + body: impl AsyncFn(RegisteredGlobalDbWriteTx<'_, '_>) -> Result> + + Send + + Sync + + 'static, + ) -> Result + where + R: Send + 'static, + E: Send + 'static, + { + match self.transact(body).await.map_err(wrap)? { + Ok(value) | Err(JobExit::Rollback(value)) => Ok(value), + Err(JobExit::Fail(error)) => Err(error), + } + } + pub fn binding(&self) -> &tracedecay_store::StoreRuntimeBindingV1 { self.database.registered_binding() } @@ -866,27 +902,6 @@ impl tracedecay_sessions::runtime::git_correlation::GitCorrelationWriteTxn } } -impl tracedecay_sessions::runtime::workflow_index::WorkflowIngestWriteTxn - for RegisteredGlobalDbWriteTransaction<'_> -{ - #[allow(clippy::manual_async_fn)] - fn commit( - self, - ) -> impl Future< - Output = Result<(), tracedecay_sessions::runtime::workflow_index::WorkflowIndexError>, - > + Send { - async move { - RegisteredGlobalDbWriteTransaction::commit(self) - .await - .map_err(|error| { - tracedecay_sessions::runtime::workflow_index::WorkflowIndexError::Db( - error.to_string(), - ) - }) - } - } -} - impl RegisteredGlobalDbWriteTransaction<'_> { pub async fn execute

( &self, @@ -955,6 +970,20 @@ impl RegisteredGlobalDbWriteTransaction<'_> { } } +/// The error side of a job body that can end without committing. `Rollback` +/// carries a value for the caller, so a replay can be reported as a success +/// without leaving a commit behind; `Fail` is an ordinary failure. +pub(crate) enum JobExit { + Rollback(R), + Fail(E), +} + +impl From for JobExit { + fn from(error: E) -> Self { + Self::Fail(error) + } +} + /// The transaction a [`RegisteredGlobalDb::transact`] body runs in. It is a /// borrowed handle, so the body takes it by value and lends `&transaction` to /// helpers written against the executor traits. diff --git a/crates/tracedecay-global-db/src/registry_maintenance.rs b/crates/tracedecay-global-db/src/registry_maintenance.rs index fb474b5a1e..a8f82c7e9a 100644 --- a/crates/tracedecay-global-db/src/registry_maintenance.rs +++ b/crates/tracedecay-global-db/src/registry_maintenance.rs @@ -6,7 +6,7 @@ use serde::Serialize; use crate::{ CodeProjectRecord, GraphScopeUpsert, ProjectRegistryContext, RegisteredGlobalDb, - RegisteredGlobalDbWriteTransaction, StoreArtifactUpsert, StoreInstanceUpsert, + StoreArtifactUpsert, StoreInstanceUpsert, }; use tracedecay_runtime_core::branch_meta; use tracedecay_runtime_core::storage::{ diff --git a/crates/tracedecay-sessions/src/runtime/workflow/workflow_index.rs b/crates/tracedecay-sessions/src/runtime/workflow/workflow_index.rs index 17cfda8d2a..0ca3ddc53f 100644 --- a/crates/tracedecay-sessions/src/runtime/workflow/workflow_index.rs +++ b/crates/tracedecay-sessions/src/runtime/workflow/workflow_index.rs @@ -698,7 +698,7 @@ async fn runs_for_git_scope( } mod port; -pub use port::{WorkflowIngestSink, WorkflowIngestWriteTxn}; +pub use port::WorkflowIngestSink; #[cfg(test)] #[allow(clippy::unwrap_used, clippy::expect_used)] diff --git a/crates/tracedecay-sessions/src/runtime/workflow/workflow_index/port.rs b/crates/tracedecay-sessions/src/runtime/workflow/workflow_index/port.rs index 782d744a1e..8dd90aee1c 100644 --- a/crates/tracedecay-sessions/src/runtime/workflow/workflow_index/port.rs +++ b/crates/tracedecay-sessions/src/runtime/workflow/workflow_index/port.rs @@ -6,15 +6,9 @@ use std::future::Future; -use tracedecay_runtime_core::db::engine::{Executor, QueryExecutor}; - use super::WorkflowIndexError; use crate::{WorkflowAgent, WorkflowRun}; -pub trait WorkflowIngestWriteTxn: QueryExecutor + Executor + Sized + Send { - fn commit(self) -> impl Future> + Send; -} - /// Authoritative sink one workflow-ingest sweep upserts through. /// /// The sweep needs an authority check, a watermark it can advance, and a From 156af12380ec8be06ca8169884d1ad478c91a782 Mon Sep 17 00:00:00 2001 From: ScriptedAlchemy Date: Sun, 18 Oct 2026 16:14:12 +0000 Subject: [PATCH 02/10] refactor(global-db): run accounting and workflow writes as jobs --- .../src/registered_accounting.rs | 72 ++++++++++--------- .../src/workflow_adapter.rs | 60 ++++++++-------- 2 files changed, 68 insertions(+), 64 deletions(-) diff --git a/crates/tracedecay-global-db/src/registered_accounting.rs b/crates/tracedecay-global-db/src/registered_accounting.rs index 9d7895488a..4b056d7147 100644 --- a/crates/tracedecay-global-db/src/registered_accounting.rs +++ b/crates/tracedecay-global-db/src/registered_accounting.rs @@ -1,6 +1,6 @@ use std::path::Path; -use tracedecay_runtime_core::db::engine::Value; +use tracedecay_runtime_core::db::engine::{Executor as _, Value}; use super::{ RegisteredGlobalDb, SavingsDay, SavingsTotal, global_db_operation_error, @@ -19,20 +19,22 @@ impl RegisteredGlobalDb { tokens_saved: u64, ) -> tracedecay_domain::errors::Result<()> { let path = super::project_path_alias_key(project_path); - let transaction = self.begin_write_transaction().await?; - transaction - .execute( - "INSERT INTO projects (path, tokens_saved) VALUES (?1, ?2) - ON CONFLICT(path) DO UPDATE SET - tokens_saved = MAX(tokens_saved, excluded.tokens_saved)", - tracedecay_runtime_core::db::engine::params![path, tokens_saved as i64], - ) - .await - .map_err(|error| global_db_operation_error("update project token total", error))?; - transaction - .commit() - .await - .map_err(|error| global_db_operation_error("commit project token total", error)) + self.transact(async move |transaction| { + transaction + .execute( + "INSERT INTO projects (path, tokens_saved) VALUES (?1, ?2) + ON CONFLICT(path) DO UPDATE SET + tokens_saved = MAX(tokens_saved, excluded.tokens_saved)", + tracedecay_runtime_core::db::engine::params![ + path.as_str(), + tokens_saved as i64 + ], + ) + .await + .map_err(|error| global_db_operation_error("update project token total", error))?; + Ok(()) + }) + .await? } /// A project with no registry row has genuinely saved nothing, so an @@ -87,26 +89,26 @@ impl RegisteredGlobalDb { timestamp: i64, ) -> tracedecay_domain::errors::Result<()> { let project_path = RegisteredGlobalDb::canonical_project_key(Path::new(project_path)); - let transaction = self.begin_write_transaction().await?; - transaction - .execute( - "INSERT INTO savings_ledger - (ts, project_path, tool_name, before_tokens, after_tokens) - VALUES (?1, ?2, ?3, ?4, ?5)", - tracedecay_runtime_core::db::engine::params![ - timestamp, - project_path, - tool_name, - before_tokens as i64, - after_tokens as i64 - ], - ) - .await - .map_err(|error| global_db_operation_error("append savings ledger entry", error))?; - transaction - .commit() - .await - .map_err(|error| global_db_operation_error("commit savings ledger entry", error)) + let tool_name = tool_name.to_owned(); + self.transact(async move |transaction| { + transaction + .execute( + "INSERT INTO savings_ledger + (ts, project_path, tool_name, before_tokens, after_tokens) + VALUES (?1, ?2, ?3, ?4, ?5)", + tracedecay_runtime_core::db::engine::params![ + timestamp, + project_path.as_str(), + tool_name.as_str(), + before_tokens as i64, + after_tokens as i64 + ], + ) + .await + .map_err(|error| global_db_operation_error("append savings ledger entry", error))?; + Ok(()) + }) + .await? } /// Sums settled savings-ledger rows for an optional project path. diff --git a/crates/tracedecay-global-db/src/workflow_adapter.rs b/crates/tracedecay-global-db/src/workflow_adapter.rs index 0dc375dbc5..bd14d14813 100644 --- a/crates/tracedecay-global-db/src/workflow_adapter.rs +++ b/crates/tracedecay-global-db/src/workflow_adapter.rs @@ -7,14 +7,13 @@ use std::path::Path; use tracedecay_domain::ProjectId; use tracedecay_store::StoreShardScopeV1; -use crate::{RegisteredGlobalDb, RegisteredGlobalDbWriteTransaction}; -use tracedecay_runtime_core::db::engine::params; +use crate::RegisteredGlobalDb; +use tracedecay_runtime_core::db::engine::{Executor as _, params}; #[cfg(any(test, feature = "test-helpers"))] use tracedecay_sessions::runtime::source::TranscriptIngestResult; use tracedecay_sessions::runtime::workflow_index::{ INGEST_WATERMARK_KEY, RegisteredWorkflowIndexSnapshot, WorkflowAgent, WorkflowIndexError, - WorkflowIngestSink, WorkflowIngestWriteTxn, WorkflowRun, read_ingest_watermark, upsert_agent, - upsert_run, + WorkflowIngestSink, WorkflowRun, read_ingest_watermark, upsert_agent, upsert_run, }; #[cfg(any(test, feature = "test-helpers"))] use tracedecay_sessions::runtime::workflow_ingest::{ @@ -118,26 +117,27 @@ where #[tracing::instrument(name = "global_db.workflow.bump_watermark", level = "trace", skip_all)] async fn bump_ingest_watermark(&self, value: i64) { - let Ok(transaction) = self.db().begin_write_transaction().await else { - tracing::debug!("workflow ingest writer unavailable"); - return; - }; - let write = async { - transaction - .execute( - "INSERT INTO workflow_index_meta(key, value) + let write = self + .db() + .transact(async move |transaction| { + transaction + .execute( + "INSERT INTO workflow_index_meta(key, value) VALUES (?1, ?2) ON CONFLICT(key) DO UPDATE SET value = MAX(value, excluded.value), updated_at = unixepoch()", - params![INGEST_WATERMARK_KEY, value], - ) - .await?; - RegisteredGlobalDbWriteTransaction::commit(transaction).await - } - .await; - if let Err(err) = write { - tracing::debug!(error = %err, "workflow ingest watermark not advanced"); + params![INGEST_WATERMARK_KEY, value], + ) + .await + }) + .await; + match write { + Ok(Ok(_)) => {} + Ok(Err(err)) => { + tracing::debug!(error = %err, "workflow ingest watermark not advanced"); + } + Err(err) => tracing::debug!(error = %err, "workflow ingest writer unavailable"), } } @@ -147,15 +147,17 @@ where run: &WorkflowRun, agents: &[WorkflowAgent], ) -> Result<(), WorkflowIndexError> { - let transaction = self - .db() - .begin_write_transaction() + let run = run.clone(); + let agents = agents.to_vec(); + self.db() + .transact(async move |transaction| { + upsert_run(&transaction, &run).await?; + for agent in &agents { + upsert_agent(&transaction, agent).await?; + } + Ok(()) + }) .await - .map_err(|error| WorkflowIndexError::Db(error.to_string()))?; - upsert_run(&transaction, run).await?; - for agent in agents { - upsert_agent(&transaction, agent).await?; - } - WorkflowIngestWriteTxn::commit(transaction).await + .map_err(|error| WorkflowIndexError::Db(error.to_string()))? } } From 9111a8fb3f587ca2bf45d547a60d6203947fec96 Mon Sep 17 00:00:00 2001 From: ScriptedAlchemy Date: Sun, 18 Oct 2026 16:14:13 +0000 Subject: [PATCH 03/10] refactor(global-db): run topology anchor and discovery writes as jobs --- .../src/discovery_queue.rs | 143 +++++++++--------- .../src/git_topology_anchor.rs | 83 +++++----- 2 files changed, 109 insertions(+), 117 deletions(-) diff --git a/crates/tracedecay-global-db/src/discovery_queue.rs b/crates/tracedecay-global-db/src/discovery_queue.rs index abe0d86ace..c8bc387903 100644 --- a/crates/tracedecay-global-db/src/discovery_queue.rs +++ b/crates/tracedecay-global-db/src/discovery_queue.rs @@ -7,7 +7,7 @@ use std::path::{Path, PathBuf}; use base64::{Engine as _, engine::general_purpose::URL_SAFE_NO_PAD}; -use tracedecay_runtime_core::db::engine::params; +use tracedecay_runtime_core::db::engine::{Executor as _, QueryExecutor as _, params}; use crate::RegisteredGlobalDb; @@ -133,78 +133,77 @@ impl RegisteredGlobalDb { let Some(last_path) = paths.last().cloned() else { return Ok(None); }; - let transaction = self - .begin_write_transaction() - .await - .map_err(|error| format!("begin host discovery queue transaction: {error}"))?; let (range_start, range_end) = exclusive_prefix_range(&prefix)?; - let mut sequence_rows = transaction - .query( - "SELECT COALESCE(MAX(byte_offset), 0) - FROM parse_offsets - WHERE file_path >= ?1 AND file_path < ?2 AND file_id = 1", - params![range_start, range_end], - ) - .await - .map_err(|error| format!("read host discovery queue tail: {error}"))?; - let sequence_row = sequence_rows - .next() - .await - .map_err(|error| format!("read host discovery queue tail: {error}"))? - .ok_or_else(|| "host discovery queue tail is unavailable".to_owned())?; - let mut sequence = sequence_row - .get::(0) - .map_err(|error| format!("decode host discovery queue tail: {error}"))?; - if sequence < 0 { - return Err("host discovery queue tail is negative".to_owned()); - } - drop(sequence_rows); - for path in paths { - let key = queue_key(&prefix, &path); - let proposed = sequence - .checked_add(1) - .ok_or_else(|| "host discovery queue sequence is exhausted".to_owned())?; - let inserted = transaction - .execute( - "INSERT INTO parse_offsets (file_path, byte_offset, mtime, file_id) - VALUES (?1, ?2, 0, 1) - ON CONFLICT(file_path) DO NOTHING", - params![key, proposed], - ) - .await - .map_err(|error| format!("enqueue host discovery path: {error}"))?; - if inserted > 0 { - sequence = proposed; - } - } - let last_key = queue_key(&prefix, &last_path); - let mut rows = transaction - .query( - "SELECT byte_offset, file_path - FROM parse_offsets - WHERE file_path = ?1 AND file_id = 1", - params![last_key], - ) - .await - .map_err(|error| format!("resolve host discovery queue identity: {error}"))?; - let row = rows - .next() - .await - .map_err(|error| format!("read host discovery queue identity: {error}"))? - .ok_or_else(|| "enqueued host discovery path is unavailable".to_owned())?; - let entry = decode_entry( - &prefix, - row.get::(0) - .map_err(|error| format!("decode host discovery queue sequence: {error}"))?, - row.get::(1) - .map_err(|error| format!("decode host discovery queue key: {error}"))?, - )?; - drop(rows); - transaction - .commit() - .await - .map_err(|error| format!("commit host discovery queue: {error}"))?; - Ok(Some(entry)) + self.transact_or( + |error| format!("run host discovery queue transaction: {error}"), + async move |transaction| { + let mut sequence_rows = transaction + .query( + "SELECT COALESCE(MAX(byte_offset), 0) + FROM parse_offsets + WHERE file_path >= ?1 AND file_path < ?2 AND file_id = 1", + params![range_start.as_str(), range_end.as_str()], + ) + .await + .map_err(|error| format!("read host discovery queue tail: {error}"))?; + let sequence_row = sequence_rows + .next() + .await + .map_err(|error| format!("read host discovery queue tail: {error}"))? + .ok_or_else(|| "host discovery queue tail is unavailable".to_owned())?; + let mut sequence = sequence_row + .get::(0) + .map_err(|error| format!("decode host discovery queue tail: {error}"))?; + if sequence < 0 { + return Err("host discovery queue tail is negative".to_owned()); + } + drop(sequence_rows); + for path in &paths { + let key = queue_key(&prefix, path); + let proposed = sequence + .checked_add(1) + .ok_or_else(|| "host discovery queue sequence is exhausted".to_owned())?; + let inserted = transaction + .execute( + "INSERT INTO parse_offsets (file_path, byte_offset, mtime, file_id) + VALUES (?1, ?2, 0, 1) + ON CONFLICT(file_path) DO NOTHING", + params![key, proposed], + ) + .await + .map_err(|error| format!("enqueue host discovery path: {error}"))?; + if inserted > 0 { + sequence = proposed; + } + } + let last_key = queue_key(&prefix, &last_path); + let mut rows = transaction + .query( + "SELECT byte_offset, file_path + FROM parse_offsets + WHERE file_path = ?1 AND file_id = 1", + params![last_key], + ) + .await + .map_err(|error| format!("resolve host discovery queue identity: {error}"))?; + let row = rows + .next() + .await + .map_err(|error| format!("read host discovery queue identity: {error}"))? + .ok_or_else(|| "enqueued host discovery path is unavailable".to_owned())?; + let entry = decode_entry( + &prefix, + row.get::(0).map_err(|error| { + format!("decode host discovery queue sequence: {error}") + })?, + row.get::(1) + .map_err(|error| format!("decode host discovery queue key: {error}"))?, + )?; + drop(rows); + Ok(Some(entry)) + }, + ) + .await } #[tracing::instrument(name = "global_db.discovery_queue.query", level = "trace", skip_all)] diff --git a/crates/tracedecay-global-db/src/git_topology_anchor.rs b/crates/tracedecay-global-db/src/git_topology_anchor.rs index 9875b44b8c..88b5b79060 100644 --- a/crates/tracedecay-global-db/src/git_topology_anchor.rs +++ b/crates/tracedecay-global-db/src/git_topology_anchor.rs @@ -8,7 +8,7 @@ use tracedecay_contracts::retrieval::{ GitTopologyAnchorResolutionOutcome, }; use tracedecay_domain::{ObservationScopeV1, RetrievalAnchorRecord, RetrievalAnchorTarget}; -use tracedecay_runtime_core::db::engine::{params, params_from_iter}; +use tracedecay_runtime_core::db::engine::{Executor as _, params, params_from_iter}; use tracedecay_store::StoreShardScopeV1; use crate::{RegisteredGlobalDb, RegisteredGlobalDbLeaseV1}; @@ -30,58 +30,51 @@ impl RegisteredGitTopologyAnchorAuthority { if !binding_matches_owner(&self.database, publication.owner()) { return Err(GitTopologyAnchorAuthorityError::Unavailable); } - let transaction = self - .database - .begin_write_transaction() - .await - .map_err(map_database_error)?; let records = publication.into_records(); let anchor_ids = records .iter() .map(|record| record.anchor_id().as_str().to_owned()) .collect::>(); - let existing = read_records(&transaction, &anchor_ids).await?; - let mut published = false; - for candidate in records { - let existing = existing.get(candidate.anchor_id().as_str()); - match existing { - Some(existing) if existing.is_semantic_replay_of(&candidate) => continue, - Some(_) => { + self.database + .transact(async move |transaction| { + let existing = read_records(&transaction, &anchor_ids).await?; + let mut published = false; + for candidate in &records { + let existing = existing.get(candidate.anchor_id().as_str()); + match existing { + Some(existing) if existing.is_semantic_replay_of(candidate) => continue, + Some(_) => return Err(GitTopologyAnchorAuthorityError::Conflict), + None => {} + } + let anchor_json = serde_json::to_string(candidate) + .map_err(|_| GitTopologyAnchorAuthorityError::Conflict)?; + let owner_json = candidate + .owner_column_json() + .map_err(|_| GitTopologyAnchorAuthorityError::Conflict)?; transaction - .rollback() + .execute( + "INSERT INTO retrieval_anchors ( + anchor_id, anchor_json, owner_json, projection_generation + ) VALUES (?1, ?2, ?3, ?4)", + params![ + candidate.anchor_id().as_str(), + anchor_json, + owner_json, + candidate.projection_generation().as_str(), + ], + ) .await - .map_err(|_| GitTopologyAnchorAuthorityError::Unavailable)?; - return Err(GitTopologyAnchorAuthorityError::Conflict); + .map_err(map_engine_error)?; + published = true; } - None => {} - } - let anchor_json = serde_json::to_string(&candidate) - .map_err(|_| GitTopologyAnchorAuthorityError::Conflict)?; - let owner_json = candidate - .owner_column_json() - .map_err(|_| GitTopologyAnchorAuthorityError::Conflict)?; - transaction - .execute( - "INSERT INTO retrieval_anchors ( - anchor_id, anchor_json, owner_json, projection_generation - ) VALUES (?1, ?2, ?3, ?4)", - params![ - candidate.anchor_id().as_str(), - anchor_json, - owner_json, - candidate.projection_generation().as_str(), - ], - ) - .await - .map_err(map_engine_error)?; - published = true; - } - transaction.commit().await.map_err(map_engine_error)?; - Ok(if published { - GitTopologyAnchorPublicationOutcome::Published - } else { - GitTopologyAnchorPublicationOutcome::Replayed - }) + Ok(if published { + GitTopologyAnchorPublicationOutcome::Published + } else { + GitTopologyAnchorPublicationOutcome::Replayed + }) + }) + .await + .map_err(map_database_error)? } async fn resolve_record( From cca8a2f5bf4059c90c883a2f0eb20e40a9212df7 Mon Sep 17 00:00:00 2001 From: ScriptedAlchemy Date: Sun, 18 Oct 2026 16:14:14 +0000 Subject: [PATCH 04/10] refactor(global-db): run stack delivery writes as jobs --- .../src/registered/delivery_settlement.rs | 239 +++++----- .../src/stack_delivery.rs | 448 +++++++++--------- 2 files changed, 352 insertions(+), 335 deletions(-) diff --git a/crates/tracedecay-global-db/src/registered/delivery_settlement.rs b/crates/tracedecay-global-db/src/registered/delivery_settlement.rs index 3eec135196..1ac55c2bcd 100644 --- a/crates/tracedecay-global-db/src/registered/delivery_settlement.rs +++ b/crates/tracedecay-global-db/src/registered/delivery_settlement.rs @@ -1,9 +1,10 @@ -use super::{RegisteredGlobalDb, RegisteredGlobalDbWriteTransaction}; +use super::RegisteredGlobalDb; use tracedecay_domain::{ CoverageStateV1, DeliveryDropReasonV1, DeliveryEventClassV1, DeliverySettlementAttemptV1, DeliverySettlementCensusV1, DeliverySettlementOutcomeV1, DeliverySettlementV1, DeliverySurfaceFamilyV1, UtcMicros, WorkAttemptIdentityV1, canonical_sha256, }; +use tracedecay_runtime_core::db::engine::{Executor, QueryExecutor}; /// One Work attempt can have at most this many separately addressed delivery /// fan-outs. The recipient bound is enforced per fan-out by the domain; this @@ -151,59 +152,62 @@ impl RegisteredGlobalDb { source_receipt_ref: Option<&str>, ) -> Result { validate_project_and_attempt(project_id, attempt)?; - let transaction = self - .begin_write_transaction() - .await - .map_err(|error| format!("failed to begin delivery attempt transaction: {error}"))?; - ensure_fanout_identity(&transaction, project_id, attempt).await?; + let project_id = project_id.to_owned(); + let attempt = attempt.clone(); + let source_receipt_ref = source_receipt_ref.map(str::to_owned); + self.transact_or( + |error| format!("failed to run delivery attempt transaction: {error}"), + async move |transaction| { + let project_id = project_id.as_str(); + let attempt = &attempt; + ensure_fanout_identity(&transaction, project_id, attempt).await?; - if let Some(stored) = read_delivery_attempt(&transaction, project_id, attempt).await? { - if stored.attempt != *attempt { - return Err("delivery attempt identity conflict".to_owned()); - } - let outcome = if stored.outcome.is_some() { - DeliveryAttemptClaimV1::AlreadySettled(stored.into_settlement()?) - } else { - DeliveryAttemptClaimV1::ReplayedAttempt - }; - if let Some(source_receipt_ref) = source_receipt_ref { - bind_source_receipt(&transaction, project_id, attempt, source_receipt_ref).await?; - } - transaction - .commit() - .await - .map_err(|error| format!("failed to close delivery attempt replay: {error}"))?; - return Ok(outcome); - } + if let Some(stored) = + read_delivery_attempt(&transaction, project_id, attempt).await? + { + if stored.attempt != *attempt { + return Err("delivery attempt identity conflict".to_owned()); + } + let outcome = if stored.outcome.is_some() { + DeliveryAttemptClaimV1::AlreadySettled(stored.into_settlement()?) + } else { + DeliveryAttemptClaimV1::ReplayedAttempt + }; + if let Some(source_receipt_ref) = source_receipt_ref.as_deref() { + bind_source_receipt(&transaction, project_id, attempt, source_receipt_ref) + .await?; + } + return Ok(outcome); + } - let existing = count_attempts(&transaction, project_id, attempt).await?; - if existing >= u64::from(attempt.eligible) { - return Err("delivery attempt exceeds eligible recipients".to_owned()); - } - transaction - .execute( - "INSERT INTO delivery_settlements - (project_id, owner_event_id, surface, channel_ref, attempted_at_micros, - outcome, settled_at_micros, drop_reason, census_json) - VALUES (?1, ?2, ?3, ?4, ?5, NULL, NULL, NULL, NULL)", - tracedecay_runtime_core::db::engine::params![ - project_id, - attempt.owner_event_id.as_str(), - surface_name(attempt.channel.surface), - attempt.channel.channel_ref.as_str(), - attempt.attempted_at.0, - ], - ) - .await - .map_err(|error| format!("failed to record delivery attempt: {error}"))?; - if let Some(source_receipt_ref) = source_receipt_ref { - bind_source_receipt(&transaction, project_id, attempt, source_receipt_ref).await?; - } - transaction - .commit() - .await - .map_err(|error| format!("failed to commit delivery attempt: {error}"))?; - Ok(DeliveryAttemptClaimV1::Claimed) + let existing = count_attempts(&transaction, project_id, attempt).await?; + if existing >= u64::from(attempt.eligible) { + return Err("delivery attempt exceeds eligible recipients".to_owned()); + } + transaction + .execute( + "INSERT INTO delivery_settlements + (project_id, owner_event_id, surface, channel_ref, attempted_at_micros, + outcome, settled_at_micros, drop_reason, census_json) + VALUES (?1, ?2, ?3, ?4, ?5, NULL, NULL, NULL, NULL)", + tracedecay_runtime_core::db::engine::params![ + project_id, + attempt.owner_event_id.as_str(), + surface_name(attempt.channel.surface), + attempt.channel.channel_ref.as_str(), + attempt.attempted_at.0, + ], + ) + .await + .map_err(|error| format!("failed to record delivery attempt: {error}"))?; + if let Some(source_receipt_ref) = source_receipt_ref.as_deref() { + bind_source_receipt(&transaction, project_id, attempt, source_receipt_ref) + .await?; + } + Ok(DeliveryAttemptClaimV1::Claimed) + }, + ) + .await } /// Resolves a project-scoped source acknowledgement token to the exact @@ -359,67 +363,68 @@ impl RegisteredGlobalDb { settlement .validate() .map_err(|error| format!("invalid delivery settlement: {error}"))?; - let transaction = self - .begin_write_transaction() - .await - .map_err(|error| format!("failed to begin delivery settlement transaction: {error}"))?; - ensure_fanout_identity(&transaction, project_id, &settlement.attempt).await?; - let stored = read_delivery_attempt(&transaction, project_id, &settlement.attempt) - .await? - .ok_or_else(|| "delivery attempt is unavailable".to_owned())?; - if stored.attempt != settlement.attempt { - return Err("delivery attempt identity conflict".to_owned()); - } - let (replayed, census) = if stored.outcome.is_some() { - let census = stored - .census - .clone() - .ok_or_else(|| "delivery settlement census is missing".to_owned())?; - if stored.into_settlement()? != *settlement { - return Err("delivery settlement conflict".to_owned()); - } - (true, census) - } else { - let census = read_pending_delivery_census( - &transaction, - project_id, - &settlement.attempt, - settlement, - ) - .await?; - let census_json = serde_json::to_string(&census) - .map_err(|error| format!("failed to serialize delivery census: {error}"))?; - transaction - .execute( - "UPDATE delivery_settlements - SET outcome = ?5, settled_at_micros = ?6, - drop_reason = ?7, census_json = ?8 - WHERE project_id = ?1 AND owner_event_id = ?2 AND surface = ?3 - AND channel_ref = ?4 AND outcome IS NULL", - tracedecay_runtime_core::db::engine::params![ + let project_id = project_id.to_owned(); + let settlement = settlement.clone(); + self.transact_or( + |error| format!("failed to run delivery settlement transaction: {error}"), + async move |transaction| { + let project_id = project_id.as_str(); + ensure_fanout_identity(&transaction, project_id, &settlement.attempt).await?; + let stored = read_delivery_attempt(&transaction, project_id, &settlement.attempt) + .await? + .ok_or_else(|| "delivery attempt is unavailable".to_owned())?; + if stored.attempt != settlement.attempt { + return Err("delivery attempt identity conflict".to_owned()); + } + let (replayed, census) = if stored.outcome.is_some() { + let census = stored + .census + .clone() + .ok_or_else(|| "delivery settlement census is missing".to_owned())?; + if stored.into_settlement()? != settlement { + return Err("delivery settlement conflict".to_owned()); + } + (true, census) + } else { + let census = read_pending_delivery_census( + &transaction, project_id, - settlement.attempt.owner_event_id.as_str(), - surface_name(settlement.attempt.channel.surface), - settlement.attempt.channel.channel_ref.as_str(), - outcome_name(settlement.outcome), - settlement.settled_at.0, - settlement.drop_reason.map(drop_reason_name), - census_json, - ], - ) - .await - .map_err(|error| format!("failed to settle delivery attempt: {error}"))?; - (false, census) - }; - transaction - .commit() - .await - .map_err(|error| format!("failed to commit delivery settlement: {error}"))?; - Ok(DurableDeliverySettlementReceiptV1 { - settlement: settlement.clone(), - census, - replayed, - }) + &settlement.attempt, + &settlement, + ) + .await?; + let census_json = serde_json::to_string(&census) + .map_err(|error| format!("failed to serialize delivery census: {error}"))?; + transaction + .execute( + "UPDATE delivery_settlements + SET outcome = ?5, settled_at_micros = ?6, + drop_reason = ?7, census_json = ?8 + WHERE project_id = ?1 AND owner_event_id = ?2 AND surface = ?3 + AND channel_ref = ?4 AND outcome IS NULL", + tracedecay_runtime_core::db::engine::params![ + project_id, + settlement.attempt.owner_event_id.as_str(), + surface_name(settlement.attempt.channel.surface), + settlement.attempt.channel.channel_ref.as_str(), + outcome_name(settlement.outcome), + settlement.settled_at.0, + settlement.drop_reason.map(drop_reason_name), + census_json, + ], + ) + .await + .map_err(|error| format!("failed to settle delivery attempt: {error}"))?; + (false, census) + }; + Ok(DurableDeliverySettlementReceiptV1 { + settlement: settlement.clone(), + census, + replayed, + }) + }, + ) + .await } } @@ -497,7 +502,7 @@ fn validate_source_receipt_ref(source_receipt_ref: &str) -> Result<(), String> { } async fn bind_source_receipt( - transaction: &RegisteredGlobalDbWriteTransaction<'_>, + transaction: &impl Executor, project_id: &str, attempt: &DeliverySettlementAttemptV1, source_receipt_ref: &str, @@ -587,7 +592,7 @@ async fn bind_source_receipt( } async fn ensure_fanout_identity( - transaction: &RegisteredGlobalDbWriteTransaction<'_>, + transaction: &impl Executor, project_id: &str, attempt: &DeliverySettlementAttemptV1, ) -> Result<(), String> { @@ -697,7 +702,7 @@ impl StoredDeliveryAttempt { } async fn read_delivery_attempt( - transaction: &RegisteredGlobalDbWriteTransaction<'_>, + transaction: &impl QueryExecutor, project_id: &str, attempt: &DeliverySettlementAttemptV1, ) -> Result, String> { @@ -756,7 +761,7 @@ async fn read_delivery_attempt( } async fn count_attempts( - transaction: &RegisteredGlobalDbWriteTransaction<'_>, + transaction: &impl QueryExecutor, project_id: &str, attempt: &DeliverySettlementAttemptV1, ) -> Result { @@ -785,7 +790,7 @@ async fn count_attempts( } async fn read_pending_delivery_census( - transaction: &RegisteredGlobalDbWriteTransaction<'_>, + transaction: &impl QueryExecutor, project_id: &str, attempt: &DeliverySettlementAttemptV1, settlement: &DeliverySettlementV1, diff --git a/crates/tracedecay-global-db/src/stack_delivery.rs b/crates/tracedecay-global-db/src/stack_delivery.rs index 585fc67f37..0e2ff3e286 100644 --- a/crates/tracedecay-global-db/src/stack_delivery.rs +++ b/crates/tracedecay-global-db/src/stack_delivery.rs @@ -10,7 +10,7 @@ use std::collections::{BTreeMap, BTreeSet}; use serde::{Deserialize, Serialize}; use tracedecay_runtime_core::db::engine::{Executor, QueryExecutor, params, params_from_iter}; -use crate::RegisteredGlobalDb; +use crate::{RegisteredGlobalDb, registered::JobExit}; /// Maximum number of rows in the `pending` state for one project. pub const MAX_GITHUB_STACK_ACTIVE_PENDING_V1: usize = 256; @@ -540,93 +540,88 @@ impl RegisteredGlobalDb { MAX_GITHUB_STACK_DELIVERY_BATCH_V1 )); } - let transaction = self - .begin_write_transaction() - .await - .map_err(|error| format!("begin GitHub stack signal append: {error}"))?; - if let Some(existing) = - lookup_signal(&transaction, &record.project_id, &record.signal_id).await? - { - if existing != record { + self.transact_or_rollback( + |error| format!("GitHub stack signal append: {error}"), + async move |transaction| { + if let Some(existing) = + lookup_signal(&transaction, &record.project_id, &record.signal_id).await? + { + if existing != record { + return Err("GitHub stack signal identity conflict".to_owned().into()); + } + let (pending_count, deferred_count) = + counts(&transaction, &record.project_id).await?; + return Err(JobExit::Rollback( + GitHubStackSignalAppendOutcomeV1::Replayed { + pending_count, + deferred_count, + }, + )); + } + transaction - .rollback() + .execute( + "INSERT INTO github_stack_delivery_signals + (project_id, signal_id, scope_digest, repository_id, + watermark_id, observed_at_micros, signal_json) + VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7)", + params![ + record.project_id.as_str(), + record.signal_id.as_str(), + record.scope_digest.as_str(), + record.repository_id.as_str(), + record.watermark_id.as_str(), + record.observed_at_micros, + record.signal_json.as_str() + ], + ) .await - .map_err(|error| format!("close conflicting GitHub stack append: {error}"))?; - return Err("GitHub stack signal identity conflict".to_owned()); - } - let (pending_count, deferred_count) = counts(&transaction, &record.project_id).await?; - transaction - .rollback() - .await - .map_err(|error| format!("close replayed GitHub stack append: {error}"))?; - return Ok(GitHubStackSignalAppendOutcomeV1::Replayed { - pending_count, - deferred_count, - }); - } - - transaction - .execute( - "INSERT INTO github_stack_delivery_signals - (project_id, signal_id, scope_digest, repository_id, - watermark_id, observed_at_micros, signal_json) - VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7)", - params![ - record.project_id.as_str(), - record.signal_id.as_str(), - record.scope_digest.as_str(), - record.repository_id.as_str(), - record.watermark_id.as_str(), - record.observed_at_micros, - record.signal_json.as_str() - ], - ) - .await - .map_err(|error| format!("append GitHub stack signal: {error}"))?; - // Repair a deferred queue left by an interrupted older writer before - // assigning this append's capacity. All callers hold the immediate - // write transaction, so the quota decision is serialized. - promote_deferred(&transaction, &record.project_id).await?; - let (pending_before, _) = counts(&transaction, &record.project_id).await?; - let capacity = MAX_GITHUB_STACK_ACTIVE_PENDING_V1.saturating_sub(pending_before); - for (index, recipient) in recipient_set.into_iter().enumerate() { - let state = if index < capacity { - GitHubStackDeliveryStateV1::Pending - } else { - GitHubStackDeliveryStateV1::Deferred - }; - transaction - .execute( - "INSERT INTO github_stack_delivery_recipients - (project_id, signal_id, recipient, state) - VALUES (?1, ?2, ?3, ?4)", - params![ - record.project_id.as_str(), - record.signal_id.as_str(), - recipient, - state.as_str() - ], - ) - .await - .map_err(|error| format!("append GitHub stack delivery recipient: {error}"))?; - } - let (pending_count, deferred_count) = counts(&transaction, &record.project_id).await?; - let saturated = deferred_count > 0; - transaction - .commit() - .await - .map_err(|error| format!("commit GitHub stack signal append: {error}"))?; - Ok(if saturated { - GitHubStackSignalAppendOutcomeV1::Saturated { - pending_count, - deferred_count, - } - } else { - GitHubStackSignalAppendOutcomeV1::Appended { - pending_count, - deferred_count, - } - }) + .map_err(|error| format!("append GitHub stack signal: {error}"))?; + // Repair a deferred queue left by an interrupted older writer before + // assigning this append's capacity. The write job is serialized, so + // the quota decision is too. + promote_deferred(&transaction, &record.project_id).await?; + let (pending_before, _) = counts(&transaction, &record.project_id).await?; + let capacity = MAX_GITHUB_STACK_ACTIVE_PENDING_V1.saturating_sub(pending_before); + for (index, recipient) in recipient_set.iter().enumerate() { + let state = if index < capacity { + GitHubStackDeliveryStateV1::Pending + } else { + GitHubStackDeliveryStateV1::Deferred + }; + transaction + .execute( + "INSERT INTO github_stack_delivery_recipients + (project_id, signal_id, recipient, state) + VALUES (?1, ?2, ?3, ?4)", + params![ + record.project_id.as_str(), + record.signal_id.as_str(), + recipient.as_str(), + state.as_str() + ], + ) + .await + .map_err(|error| { + format!("append GitHub stack delivery recipient: {error}") + })?; + } + let (pending_count, deferred_count) = + counts(&transaction, &record.project_id).await?; + Ok(if deferred_count > 0 { + GitHubStackSignalAppendOutcomeV1::Saturated { + pending_count, + deferred_count, + } + } else { + GitHubStackSignalAppendOutcomeV1::Appended { + pending_count, + deferred_count, + } + }) + }, + ) + .await } /// Returns a deterministic page of coordinator-pending deliveries. @@ -742,46 +737,47 @@ impl RegisteredGlobalDb { validate_text(&key.signal_id, "signal id")?; validate_recipient(&key.recipient)?; } - let transaction = self - .begin_write_transaction() - .await - .map_err(|error| format!("begin GitHub stack host publication: {error}"))?; - let batch = read_delivery_batch(&transaction, project_id, &keys).await?; - let mut pending = Vec::with_capacity(keys.len()); - for key in &keys { - let row = batch - .get(key) - .ok_or_else(|| "GitHub stack delivery batch is incomplete".to_owned())?; - let signal_watermark = row - .watermark_id - .as_deref() - .ok_or_else(|| "GitHub stack delivery signal is unavailable".to_owned())?; - if signal_watermark != watermark_id { - return Err("GitHub stack delivery watermark mismatch".to_owned()); - } - let state = row - .state - .ok_or_else(|| "GitHub stack recipient binding is unavailable".to_owned())?; - match state { - GitHubStackDeliveryStateV1::Pending => { - pending.push(key.clone()); + let project_id = project_id.to_owned(); + let watermark_id = watermark_id.to_owned(); + self.transact_or( + |error| format!("GitHub stack host publication: {error}"), + async move |transaction| { + let batch = read_delivery_batch(&transaction, &project_id, &keys).await?; + let mut pending = Vec::with_capacity(keys.len()); + for key in &keys { + let row = batch + .get(key) + .ok_or_else(|| "GitHub stack delivery batch is incomplete".to_owned())?; + let signal_watermark = row + .watermark_id + .as_deref() + .ok_or_else(|| "GitHub stack delivery signal is unavailable".to_owned())?; + if signal_watermark != watermark_id { + return Err("GitHub stack delivery watermark mismatch".to_owned()); + } + let state = row.state.ok_or_else(|| { + "GitHub stack recipient binding is unavailable".to_owned() + })?; + match state { + GitHubStackDeliveryStateV1::Pending => { + pending.push(key.clone()); + } + GitHubStackDeliveryStateV1::HostPending => {} + GitHubStackDeliveryStateV1::Deferred => { + return Err("GitHub stack delivery is deferred".to_owned()); + } + GitHubStackDeliveryStateV1::Settled => {} + GitHubStackDeliveryStateV1::AuthorizationLost => { + return Err("GitHub stack delivery authorization was lost".to_owned()); + } + } } - GitHubStackDeliveryStateV1::HostPending => {} - GitHubStackDeliveryStateV1::Deferred => { - return Err("GitHub stack delivery is deferred".to_owned()); - } - GitHubStackDeliveryStateV1::Settled => {} - GitHubStackDeliveryStateV1::AuthorizationLost => { - return Err("GitHub stack delivery authorization was lost".to_owned()); - } - } - } - transition_pending_batch(&transaction, project_id, &pending).await?; - promote_deferred(&transaction, project_id).await?; - transaction - .commit() - .await - .map_err(|error| format!("commit GitHub stack host publication: {error}")) + transition_pending_batch(&transaction, &project_id, &pending).await?; + promote_deferred(&transaction, &project_id).await?; + Ok(()) + }, + ) + .await } /// Final host receipt. Replaying a receipt after settlement is harmless; @@ -800,58 +796,70 @@ impl RegisteredGlobalDb { validate_text(project_id, "project id")?; validate_text(signal_id, "signal id")?; validate_recipient(recipient)?; - let transaction = self - .begin_write_transaction() - .await - .map_err(|error| format!("begin GitHub stack host acknowledgement: {error}"))?; - let mut rows = transaction - .query( - "SELECT state FROM github_stack_delivery_recipients - WHERE project_id = ?1 AND signal_id = ?2 AND recipient = ?3", - params![project_id, signal_id, recipient], - ) - .await - .map_err(|error| format!("query GitHub stack host acknowledgement: {error}"))?; - let Some(row) = rows - .next() - .await - .map_err(|error| format!("read GitHub stack host acknowledgement: {error}"))? - else { - return Err("GitHub stack recipient binding is unavailable".to_owned()); - }; - let state = GitHubStackDeliveryStateV1::parse( - row.get::(0) - .map_err(|error| format!("decode GitHub stack host state: {error}"))?, - )?; - match state { - GitHubStackDeliveryStateV1::HostPending => { - transaction - .execute( - "UPDATE github_stack_delivery_recipients - SET state = 'settled' - WHERE project_id = ?1 AND signal_id = ?2 AND recipient = ?3 - AND state = 'host_pending'", - params![project_id, signal_id, recipient], + let (project_id, signal_id, recipient) = ( + project_id.to_owned(), + signal_id.to_owned(), + recipient.to_owned(), + ); + self.transact_or( + |error| format!("GitHub stack host acknowledgement: {error}"), + async move |transaction| { + let mut rows = transaction + .query( + "SELECT state FROM github_stack_delivery_recipients + WHERE project_id = ?1 AND signal_id = ?2 AND recipient = ?3", + params![project_id.as_str(), signal_id.as_str(), recipient.as_str()], ) .await - .map_err(|error| format!("settle GitHub stack host delivery: {error}"))?; - } - GitHubStackDeliveryStateV1::Settled => {} - GitHubStackDeliveryStateV1::Pending => { - return Err("GitHub stack delivery is still pending host publication".to_owned()); - } - GitHubStackDeliveryStateV1::Deferred => { - return Err("GitHub stack delivery is deferred".to_owned()); - } - GitHubStackDeliveryStateV1::AuthorizationLost => { - return Err("GitHub stack delivery authorization was lost".to_owned()); - } - } - promote_deferred(&transaction, project_id).await?; - transaction - .commit() - .await - .map_err(|error| format!("commit GitHub stack host acknowledgement: {error}")) + .map_err(|error| format!("query GitHub stack host acknowledgement: {error}"))?; + let Some(row) = rows + .next() + .await + .map_err(|error| format!("read GitHub stack host acknowledgement: {error}"))? + else { + return Err("GitHub stack recipient binding is unavailable".to_owned()); + }; + let state = GitHubStackDeliveryStateV1::parse( + row.get::(0) + .map_err(|error| format!("decode GitHub stack host state: {error}"))?, + )?; + match state { + GitHubStackDeliveryStateV1::HostPending => { + transaction + .execute( + "UPDATE github_stack_delivery_recipients + SET state = 'settled' + WHERE project_id = ?1 AND signal_id = ?2 AND recipient = ?3 + AND state = 'host_pending'", + params![ + project_id.as_str(), + signal_id.as_str(), + recipient.as_str() + ], + ) + .await + .map_err(|error| { + format!("settle GitHub stack host delivery: {error}") + })?; + } + GitHubStackDeliveryStateV1::Settled => {} + GitHubStackDeliveryStateV1::Pending => { + return Err( + "GitHub stack delivery is still pending host publication".to_owned() + ); + } + GitHubStackDeliveryStateV1::Deferred => { + return Err("GitHub stack delivery is deferred".to_owned()); + } + GitHubStackDeliveryStateV1::AuthorizationLost => { + return Err("GitHub stack delivery authorization was lost".to_owned()); + } + } + promote_deferred(&transaction, &project_id).await?; + Ok(()) + }, + ) + .await } /// Marks a pending/deferred recipient as permanently unauthorized. The @@ -866,49 +874,53 @@ impl RegisteredGlobalDb { validate_text(project_id, "project id")?; validate_text(signal_id, "signal id")?; validate_recipient(recipient)?; - let transaction = self - .begin_write_transaction() - .await - .map_err(|error| format!("begin GitHub stack authorization loss: {error}"))?; - let changed = transaction - .execute( - "UPDATE github_stack_delivery_recipients - SET state = 'authorization_lost' - WHERE project_id = ?1 AND signal_id = ?2 AND recipient = ?3 - AND state IN ('pending', 'deferred')", - params![project_id, signal_id, recipient], - ) - .await - .map_err(|error| format!("record GitHub stack authorization loss: {error}"))?; - if changed == 0 { - let mut rows = transaction - .query( - "SELECT state FROM github_stack_delivery_recipients - WHERE project_id = ?1 AND signal_id = ?2 AND recipient = ?3", - params![project_id, signal_id, recipient], - ) - .await - .map_err(|error| format!("query GitHub stack authorization state: {error}"))?; - let Some(row) = rows - .next() - .await - .map_err(|error| format!("read GitHub stack authorization state: {error}"))? - else { - return Err("GitHub stack recipient binding is unavailable".to_owned()); - }; - let state = - GitHubStackDeliveryStateV1::parse(row.get::(0).map_err(|error| { - format!("decode GitHub stack authorization state: {error}") - })?)?; - if state != GitHubStackDeliveryStateV1::AuthorizationLost { - return Err("GitHub stack delivery is no longer authorizable".to_owned()); - } - } - promote_deferred(&transaction, project_id).await?; - transaction - .commit() - .await - .map_err(|error| format!("commit GitHub stack authorization loss: {error}")) + let (project_id, signal_id, recipient) = ( + project_id.to_owned(), + signal_id.to_owned(), + recipient.to_owned(), + ); + self.transact_or( + |error| format!("GitHub stack authorization loss: {error}"), + async move |transaction| { + let changed = transaction + .execute( + "UPDATE github_stack_delivery_recipients + SET state = 'authorization_lost' + WHERE project_id = ?1 AND signal_id = ?2 AND recipient = ?3 + AND state IN ('pending', 'deferred')", + params![project_id.as_str(), signal_id.as_str(), recipient.as_str()], + ) + .await + .map_err(|error| format!("record GitHub stack authorization loss: {error}"))?; + if changed == 0 { + let mut rows = transaction + .query( + "SELECT state FROM github_stack_delivery_recipients + WHERE project_id = ?1 AND signal_id = ?2 AND recipient = ?3", + params![project_id.as_str(), signal_id.as_str(), recipient.as_str()], + ) + .await + .map_err(|error| { + format!("query GitHub stack authorization state: {error}") + })?; + let Some(row) = rows.next().await.map_err(|error| { + format!("read GitHub stack authorization state: {error}") + })? + else { + return Err("GitHub stack recipient binding is unavailable".to_owned()); + }; + let state = GitHubStackDeliveryStateV1::parse(row.get::(0).map_err( + |error| format!("decode GitHub stack authorization state: {error}"), + )?)?; + if state != GitHubStackDeliveryStateV1::AuthorizationLost { + return Err("GitHub stack delivery is no longer authorizable".to_owned()); + } + } + promote_deferred(&transaction, &project_id).await?; + Ok(()) + }, + ) + .await } /// Looks up the immutable signal payload by its exact registered-project From 45baa936c45fa216684c440099ee063e27c3fbe9 Mon Sep 17 00:00:00 2001 From: ScriptedAlchemy Date: Sun, 18 Oct 2026 16:14:15 +0000 Subject: [PATCH 05/10] refactor(global-db): run source rewrite writes as jobs --- .../src/registered_source_rewrite.rs | 297 +++++++++--------- 1 file changed, 152 insertions(+), 145 deletions(-) diff --git a/crates/tracedecay-global-db/src/registered_source_rewrite.rs b/crates/tracedecay-global-db/src/registered_source_rewrite.rs index 48cc673b79..e36a2eab5b 100644 --- a/crates/tracedecay-global-db/src/registered_source_rewrite.rs +++ b/crates/tracedecay-global-db/src/registered_source_rewrite.rs @@ -12,11 +12,12 @@ use tracedecay_domain::{ CanonicalObservationIdV1, ObservationScopeV1, ObservationSourceGenerationV1, ObservationSourceIdentityV1, }; -use tracedecay_runtime_core::db::engine::{QueryExecutor, params}; +use tracedecay_runtime_core::db::engine::{Executor, QueryExecutor, params}; use tracedecay_store::{ProjectionStoreError, ProjectionStoreResult}; use super::RegisteredGlobalDb; use super::observation_projection::retire_source_record; +use super::registered::JobExit; /// Unoffered-record reads are keyset-paginated at this page size. The exact- /// SQL engine refuses a query that materializes more rows than its hard cap, @@ -87,33 +88,42 @@ impl RegisteredGlobalDb { return Ok(()); } let key = source_key(source, scope)?; - let transaction = self - .begin_write_transaction() - .await - .map_err(|error| storage("begin source presence transaction", error))?; - for record in records { - transaction - .execute( - "INSERT INTO observation_source_presence ( - source_key, observation_id, generation, start_offset - ) VALUES (?1, ?2, ?3, ?4) - ON CONFLICT(source_key, observation_id) DO UPDATE SET - generation = excluded.generation, - start_offset = excluded.start_offset", - params![ - key.as_str(), - record.observation_id.as_str(), - record.generation.generation_id().to_string(), - offset(record.start_offset)?, - ], - ) - .await - .map_err(|error| storage("record source presence", error))?; - } - transaction - .commit() - .await - .map_err(|error| storage("commit source presence", error)) + let rows = records + .iter() + .map(|record| { + Ok(( + record.observation_id.as_str().to_owned(), + record.generation.generation_id().to_string(), + offset(record.start_offset)?, + )) + }) + .collect::>>()?; + self.transact_or( + |error| storage("run source presence transaction", error), + async move |transaction| { + for (observation_id, generation, start_offset) in &rows { + transaction + .execute( + "INSERT INTO observation_source_presence ( + source_key, observation_id, generation, start_offset + ) VALUES (?1, ?2, ?3, ?4) + ON CONFLICT(source_key, observation_id) DO UPDATE SET + generation = excluded.generation, + start_offset = excluded.start_offset", + params![ + key.as_str(), + observation_id.as_str(), + generation.as_str(), + *start_offset, + ], + ) + .await + .map_err(|error| storage("record source presence", error))?; + } + Ok(()) + }, + ) + .await } /// Starts a rewrite of the source from `previous` to `generation`. The @@ -138,44 +148,45 @@ impl RegisteredGlobalDb { } let key = source_key(source, scope)?; let generation = generation.generation_id().to_string(); - let transaction = self - .begin_write_transaction() - .await - .map_err(|error| storage("begin source rewrite transaction", error))?; - // A rewrite the previous generation never finished still names its - // own layout: its retained prefix was relabeled to it. - let unfinished = pending_rewrite(&transaction, &key).await?; - transaction - .execute( - "UPDATE observation_source_presence SET generation = ?2 - WHERE source_key = ?1 AND start_offset < ?3 - AND generation IN (?4, ?5)", - params![ - key.as_str(), - generation.as_str(), - offset(retained_through)?, - previous.generation_id().to_string(), - unfinished, - ], - ) - .await - .map_err(|error| storage("retain source prefix", error))?; - transaction - .execute( - "INSERT INTO observation_source_rewrites ( - source_key, generation, retained_through - ) VALUES (?1, ?2, ?3) - ON CONFLICT(source_key) DO UPDATE SET - generation = excluded.generation, - retained_through = excluded.retained_through", - params![key.as_str(), generation.as_str(), offset(retained_through)?], - ) - .await - .map_err(|error| storage("record source rewrite", error))?; - transaction - .commit() - .await - .map_err(|error| storage("commit source rewrite", error)) + let retained_through = offset(retained_through)?; + let previous = previous.generation_id().to_string(); + self.transact_or( + |error| storage("run source rewrite transaction", error), + async move |transaction| { + // A rewrite the previous generation never finished still names its + // own layout: its retained prefix was relabeled to it. + let unfinished = pending_rewrite(&transaction, &key).await?; + transaction + .execute( + "UPDATE observation_source_presence SET generation = ?2 + WHERE source_key = ?1 AND start_offset < ?3 + AND generation IN (?4, ?5)", + params![ + key.as_str(), + generation.as_str(), + retained_through, + previous.as_str(), + unfinished.as_deref(), + ], + ) + .await + .map_err(|error| storage("retain source prefix", error))?; + transaction + .execute( + "INSERT INTO observation_source_rewrites ( + source_key, generation, retained_through + ) VALUES (?1, ?2, ?3) + ON CONFLICT(source_key) DO UPDATE SET + generation = excluded.generation, + retained_through = excluded.retained_through", + params![key.as_str(), generation.as_str(), retained_through], + ) + .await + .map_err(|error| storage("record source rewrite", error))?; + Ok(()) + }, + ) + .await } /// Completes a pending rewrite once `generation` was read to the end of @@ -202,85 +213,81 @@ impl RegisteredGlobalDb { return Ok(0); } drop(snapshot); - let transaction = self - .begin_write_transaction() - .await - .map_err(|error| storage("begin source rewrite completion", error))?; - if pending_rewrite(&transaction, &key).await?.as_deref() != Some(generation.as_str()) { - transaction - .rollback() - .await - .map_err(|error| storage("release source rewrite completion", error))?; - return Ok(0); - } - let mut retired = 0_u64; - // Keyset-paginate the unoffered ids instead of materializing the whole - // set in one statement: a single rewritten source can hold more - // unoffered records than the engine's per-query materialization cap, - // and one oversized read wedged every completion attempt forever. - // `retire_source_record` never touches `observation_source_presence`, - // so the ordered id cursor stays stable across pages; the settle - // deletes below remain the authoritative cleanup. - let mut after_observation_id = String::new(); - loop { - let mut rows = transaction - .query( - "SELECT observation_id FROM observation_source_presence - WHERE source_key = ?1 AND generation <> ?2 - AND observation_id > ?3 - ORDER BY observation_id - LIMIT ?4", - params![ - key.as_str(), - generation.as_str(), - after_observation_id, - UNOFFERED_RECORD_PAGE_SIZE - ], - ) - .await - .map_err(|error| storage("read unoffered source records", error))?; - let mut page = Vec::new(); - while let Some(row) = rows - .next() - .await - .map_err(|error| storage("read unoffered source records", error))? - { - page.push( - row.get::(0) - .map_err(|error| storage("read unoffered source records", error))?, - ); - } - drop(rows); - if page.is_empty() { - break; - } - let page_len = page.len(); - after_observation_id = page[page_len - 1].clone(); - for observation_id in &page { - let retired_one: Pin< - Box> + Send + '_>, - > = Box::pin(retire_source_record(&transaction, observation_id)); - if retired_one.await? { - retired += 1; + self.transact_or_rollback( + |error| storage("run source rewrite completion", error), + async move |transaction| { + if pending_rewrite(&transaction, &key).await?.as_deref() + != Some(generation.as_str()) + { + return Err(JobExit::Rollback(0)); } - } - if (page_len as i64) < UNOFFERED_RECORD_PAGE_SIZE { - break; - } - } - for sql in [ - "DELETE FROM observation_source_presence WHERE source_key = ?1 AND generation <> ?2", - "DELETE FROM observation_source_rewrites WHERE source_key = ?1 AND generation = ?2", - ] { - transaction - .execute(sql, params![key.as_str(), generation.as_str()]) - .await - .map_err(|error| storage("settle source rewrite", error))?; - } - transaction - .commit() - .await - .map_err(|error| storage("commit source rewrite completion", error))?; - Ok(retired) + let mut retired = 0_u64; + // Keyset-paginate the unoffered ids instead of materializing the whole + // set in one statement: a single rewritten source can hold more + // unoffered records than the engine's per-query materialization cap, + // and one oversized read wedged every completion attempt forever. + // `retire_source_record` never touches `observation_source_presence`, + // so the ordered id cursor stays stable across pages; the settle + // deletes below remain the authoritative cleanup. + let mut after_observation_id = String::new(); + loop { + let mut rows = transaction + .query( + "SELECT observation_id FROM observation_source_presence + WHERE source_key = ?1 AND generation <> ?2 + AND observation_id > ?3 + ORDER BY observation_id + LIMIT ?4", + params![ + key.as_str(), + generation.as_str(), + after_observation_id.as_str(), + UNOFFERED_RECORD_PAGE_SIZE + ], + ) + .await + .map_err(|error| storage("read unoffered source records", error))?; + let mut page = Vec::new(); + while let Some(row) = rows + .next() + .await + .map_err(|error| storage("read unoffered source records", error))? + { + page.push( + row.get::(0) + .map_err(|error| storage("read unoffered source records", error))?, + ); + } + drop(rows); + if page.is_empty() { + break; + } + let page_len = page.len(); + after_observation_id = page[page_len - 1].clone(); + for observation_id in &page { + let retired_one: Pin< + Box> + Send + '_>, + > = Box::pin(retire_source_record(&transaction, observation_id)); + if retired_one.await? { + retired += 1; + } + } + if (page_len as i64) < UNOFFERED_RECORD_PAGE_SIZE { + break; + } + } + for sql in [ + "DELETE FROM observation_source_presence WHERE source_key = ?1 AND generation <> ?2", + "DELETE FROM observation_source_rewrites WHERE source_key = ?1 AND generation = ?2", + ] { + transaction + .execute(sql, params![key.as_str(), generation.as_str()]) + .await + .map_err(|error| storage("settle source rewrite", error))?; + } + Ok(retired) + }, + ) + .await } } From 6d7ab42a99e981c5389589d9ad16b0e9fcc2d5db Mon Sep 17 00:00:00 2001 From: ScriptedAlchemy Date: Sun, 18 Oct 2026 16:14:16 +0000 Subject: [PATCH 06/10] refactor(global-db): run registry and remote deletion writes as jobs --- .../src/registry_maintenance/lifecycle.rs | 41 ++-- .../src/remote_deletion.rs | 221 +++++++++--------- 2 files changed, 131 insertions(+), 131 deletions(-) diff --git a/crates/tracedecay-global-db/src/registry_maintenance/lifecycle.rs b/crates/tracedecay-global-db/src/registry_maintenance/lifecycle.rs index 09ca393e33..2c7ee003c4 100644 --- a/crates/tracedecay-global-db/src/registry_maintenance/lifecycle.rs +++ b/crates/tracedecay-global-db/src/registry_maintenance/lifecycle.rs @@ -209,7 +209,7 @@ pub async fn apply_registry_gc( } pub(super) async fn delete_registry_gc_candidates_in_transaction( - transaction: &RegisteredGlobalDbWriteTransaction<'_>, + transaction: &impl tracedecay_runtime_core::db::engine::Executor, project_ids: &[String], project_paths: &[PathBuf], ) -> tracedecay_domain::errors::Result<(usize, usize)> { @@ -269,19 +269,22 @@ pub async fn forget_registry_project( project_id: &str, project_paths: &[PathBuf], ) -> tracedecay_domain::errors::Result { - let transaction = db.begin_write_transaction().await?; - let (code_projects_deleted, path_ledger_rows_deleted) = - delete_registry_gc_candidates_in_transaction( - &transaction, - &[project_id.to_string()], - project_paths, - ) - .await?; - transaction.commit().await?; - Ok(ForgetRegistryProjectRows { - code_projects_deleted, - path_ledger_rows_deleted, + let project_ids = [project_id.to_string()]; + let project_paths = project_paths.to_vec(); + db.transact(async move |transaction| { + let (code_projects_deleted, path_ledger_rows_deleted) = + delete_registry_gc_candidates_in_transaction( + &transaction, + &project_ids, + &project_paths, + ) + .await?; + Ok::<_, tracedecay_domain::errors::TraceDecayError>(ForgetRegistryProjectRows { + code_projects_deleted, + path_ledger_rows_deleted, + }) }) + .await? } /// Retires exact registry rows for project roots already removed by an @@ -290,9 +293,11 @@ pub async fn retire_registry_project_paths( db: &RegisteredGlobalDb, project_paths: &[PathBuf], ) -> tracedecay_domain::errors::Result { - let transaction = db.begin_write_transaction().await?; - let (_, retired) = - delete_registry_gc_candidates_in_transaction(&transaction, &[], project_paths).await?; - transaction.commit().await?; - Ok(retired) + let project_paths = project_paths.to_vec(); + db.transact(async move |transaction| { + let (_, retired) = + delete_registry_gc_candidates_in_transaction(&transaction, &[], &project_paths).await?; + Ok::<_, tracedecay_domain::errors::TraceDecayError>(retired) + }) + .await? } diff --git a/crates/tracedecay-global-db/src/remote_deletion.rs b/crates/tracedecay-global-db/src/remote_deletion.rs index d85975e412..8a4fd115c4 100644 --- a/crates/tracedecay-global-db/src/remote_deletion.rs +++ b/crates/tracedecay-global-db/src/remote_deletion.rs @@ -2,7 +2,7 @@ use serde::{Deserialize, Serialize}; use tracedecay_domain::errors::TraceDecayError; -use tracedecay_runtime_core::db::engine::{QueryExecutor, params}; +use tracedecay_runtime_core::db::engine::{Executor, QueryExecutor, params}; use crate::RegisteredGlobalDb; @@ -323,45 +323,44 @@ impl RegisteredGlobalDb { tombstone: RemoteDeletionTombstone, ) -> Result { tombstone.validate_new()?; - let transaction = self.begin_write_transaction().await?; - if let Some(existing) = read_tombstone( - &transaction, - &tombstone.profile_id, - tombstone.target, - tombstone.project_key(), - ) - .await? - { - transaction.commit().await.map_err(|error| { - remote_deletion_error("commit remote deletion tombstone replay", error) - })?; - return if existing.tombstone_id == tombstone.tombstone_id { - Ok(RemoteDeletionTombstoneRecordOutcome::Replayed(existing)) - } else { - Ok(RemoteDeletionTombstoneRecordOutcome::Conflict { existing }) - }; - } - transaction - .execute( - "INSERT INTO remote_deletion_tombstones - (profile_id, target_kind, project_id, tombstone_id, recorded_at_micros, - cleanup_status, failure_code, failure_phase, retryable) - VALUES (?1, ?2, ?3, ?4, ?5, 'pending', NULL, NULL, NULL)", - params![ - tombstone.profile_id.as_str(), - tombstone.target.as_str(), - tombstone.project_key(), - tombstone.tombstone_id.as_str(), - tombstone.recorded_at_micros, - ], + self.transact(async move |transaction| { + if let Some(existing) = read_tombstone( + &transaction, + &tombstone.profile_id, + tombstone.target, + tombstone.project_key(), ) - .await - .map_err(|error| remote_deletion_error("record remote deletion tombstone", error))?; - transaction - .commit() - .await - .map_err(|error| remote_deletion_error("commit remote deletion tombstone", error))?; - Ok(RemoteDeletionTombstoneRecordOutcome::Recorded(tombstone)) + .await? + { + return Ok(if existing.tombstone_id == tombstone.tombstone_id { + RemoteDeletionTombstoneRecordOutcome::Replayed(existing) + } else { + RemoteDeletionTombstoneRecordOutcome::Conflict { existing } + }); + } + transaction + .execute( + "INSERT INTO remote_deletion_tombstones + (profile_id, target_kind, project_id, tombstone_id, recorded_at_micros, + cleanup_status, failure_code, failure_phase, retryable) + VALUES (?1, ?2, ?3, ?4, ?5, 'pending', NULL, NULL, NULL)", + params![ + tombstone.profile_id.as_str(), + tombstone.target.as_str(), + tombstone.project_key(), + tombstone.tombstone_id.as_str(), + tombstone.recorded_at_micros, + ], + ) + .await + .map_err(|error| { + remote_deletion_error("record remote deletion tombstone", error) + })?; + Ok(RemoteDeletionTombstoneRecordOutcome::Recorded( + tombstone.clone(), + )) + }) + .await? } #[tracing::instrument( @@ -376,70 +375,65 @@ impl RegisteredGlobalDb { next: RemoteDeletionCleanupState, ) -> Result { validate_transition(&next)?; - let transaction = self.begin_write_transaction().await?; - let existing = read_tombstone( - &transaction, - &tombstone.profile_id, - tombstone.target, - tombstone.project_key(), - ) - .await? - .ok_or_else(|| { - remote_deletion_error( - "transition remote deletion tombstone", - "tombstone is unavailable", - ) - })?; - if existing.tombstone_id != tombstone.tombstone_id { - transaction.commit().await.map_err(|error| { - remote_deletion_error("commit remote deletion conflict read", error) - })?; - return Ok(RemoteDeletionTombstoneTransitionOutcome::Conflict { existing }); - } - if existing.cleanup != expected { - transaction.commit().await.map_err(|error| { - remote_deletion_error("commit remote deletion state read", error) - })?; - return Ok(RemoteDeletionTombstoneTransitionOutcome::StateChanged { existing }); - } - let (failure_code, failure_phase, retryable) = - next.failure() - .map_or((None, None, None), |(code, phase, retryable)| { - ( - Some(code.as_str()), - Some(phase.as_str()), - Some(i64::from(retryable)), - ) - }); - transaction - .execute( - "UPDATE remote_deletion_tombstones - SET cleanup_status = ?1, failure_code = ?2, failure_phase = ?3, retryable = ?4 - WHERE profile_id = ?5 AND target_kind = ?6 AND project_id = ?7 - AND tombstone_id = ?8", - params![ - next.status(), - failure_code, - failure_phase, - retryable, - tombstone.profile_id.as_str(), - tombstone.target.as_str(), - tombstone.project_key(), - tombstone.tombstone_id.as_str(), - ], + let tombstone = tombstone.clone(); + let expected = expected.clone(); + self.transact(async move |transaction| { + let existing = read_tombstone( + &transaction, + &tombstone.profile_id, + tombstone.target, + tombstone.project_key(), ) - .await - .map_err(|error| { - remote_deletion_error("transition remote deletion tombstone", error) + .await? + .ok_or_else(|| { + remote_deletion_error( + "transition remote deletion tombstone", + "tombstone is unavailable", + ) })?; - transaction.commit().await.map_err(|error| { - remote_deletion_error("commit remote deletion tombstone transition", error) - })?; - let updated = RemoteDeletionTombstone { - cleanup: next, - ..existing - }; - Ok(RemoteDeletionTombstoneTransitionOutcome::Updated(updated)) + if existing.tombstone_id != tombstone.tombstone_id { + return Ok(RemoteDeletionTombstoneTransitionOutcome::Conflict { existing }); + } + if existing.cleanup != expected { + return Ok(RemoteDeletionTombstoneTransitionOutcome::StateChanged { existing }); + } + let (failure_code, failure_phase, retryable) = + next.failure() + .map_or((None, None, None), |(code, phase, retryable)| { + ( + Some(code.as_str()), + Some(phase.as_str()), + Some(i64::from(retryable)), + ) + }); + transaction + .execute( + "UPDATE remote_deletion_tombstones + SET cleanup_status = ?1, failure_code = ?2, failure_phase = ?3, retryable = ?4 + WHERE profile_id = ?5 AND target_kind = ?6 AND project_id = ?7 + AND tombstone_id = ?8", + params![ + next.status(), + failure_code, + failure_phase, + retryable, + tombstone.profile_id.as_str(), + tombstone.target.as_str(), + tombstone.project_key(), + tombstone.tombstone_id.as_str(), + ], + ) + .await + .map_err(|error| { + remote_deletion_error("transition remote deletion tombstone", error) + })?; + let updated = RemoteDeletionTombstone { + cleanup: next.clone(), + ..existing + }; + Ok(RemoteDeletionTombstoneTransitionOutcome::Updated(updated)) + }) + .await? } pub async fn remote_deletion_tombstone_for_project( @@ -475,19 +469,20 @@ impl RegisteredGlobalDb { pub async fn delete_remote_deleted_project_registry_row(&self, project_id: &str) -> Result<()> { validate_identifier("remote deletion project id", project_id)?; - let transaction = self.begin_write_transaction().await?; - transaction - .execute( - "DELETE FROM code_projects WHERE project_id = ?1", - params![project_id], - ) - .await - .map_err(|error| { - remote_deletion_error("remove remote-deleted project registry row", error) - })?; - transaction.commit().await.map_err(|error| { - remote_deletion_error("commit remote-deleted project registry row", error) + let project_id = project_id.to_owned(); + self.transact(async move |transaction| { + transaction + .execute( + "DELETE FROM code_projects WHERE project_id = ?1", + params![project_id.as_str()], + ) + .await + .map_err(|error| { + remote_deletion_error("remove remote-deleted project registry row", error) + })?; + Ok(()) }) + .await? } } From 68af603fdbd6e396ddad1fba3105e1d8b8836455 Mon Sep 17 00:00:00 2001 From: ScriptedAlchemy Date: Sun, 18 Oct 2026 16:14:18 +0000 Subject: [PATCH 07/10] refactor(global-db): run analytics writes as jobs --- .../src/registered_analytics.rs | 454 +++++++++--------- 1 file changed, 231 insertions(+), 223 deletions(-) diff --git a/crates/tracedecay-global-db/src/registered_analytics.rs b/crates/tracedecay-global-db/src/registered_analytics.rs index 609e51ecf1..468eef0bc4 100644 --- a/crates/tracedecay-global-db/src/registered_analytics.rs +++ b/crates/tracedecay-global-db/src/registered_analytics.rs @@ -3,7 +3,7 @@ use std::fmt::Write as _; use std::future::Future; use std::pin::Pin; -use tracedecay_runtime_core::db::engine::{Value, opt_text}; +use tracedecay_runtime_core::db::engine::{Executor, Value, opt_text}; use super::{ AnalyticsEventInsert, AnalyticsEventQuery, AnalyticsEventRecord, AnalyticsHintCounts, @@ -236,23 +236,19 @@ impl RegisteredGlobalDb { if events.is_empty() { return Ok(Vec::new()); } - let transaction = self - .begin_write_transaction() - .await - .map_err(|error| format!("failed to begin analytics append transaction: {error}"))?; - let ids = match kind { - AnalyticsAppendKind::General => { - append_analytics_events_in_existing_tx(&transaction, events).await? - } - AnalyticsAppendKind::Observability => { - append_observability_events_in_existing_tx(&transaction, events).await? - } - }; - transaction - .commit() - .await - .map_err(|error| format!("failed to commit analytics append transaction: {error}"))?; - Ok(ids) + let events = events.to_vec(); + self.transact_or( + |error| format!("failed to run analytics append transaction: {error}"), + async move |transaction| match kind { + AnalyticsAppendKind::General => { + append_analytics_events_in_existing_tx(&transaction, &events).await + } + AnalyticsAppendKind::Observability => { + append_observability_events_in_existing_tx(&transaction, &events).await + } + }, + ) + .await } #[tracing::instrument( @@ -323,52 +319,55 @@ impl RegisteredGlobalDb { owner_fact_json, delivery_envelope_json, )?; - let transaction = self - .begin_write_transaction() - .await - .map_err(|error| format!("failed to begin observability outbox claim: {error}"))?; - if let Some(stored) = read_outbox_record(&transaction, project_id, owner_event_id).await? { - if stored.owner_fact_json != owner_fact_json { - return Err("observability owner fact conflict".to_owned()); - } - let outcome = if let Some(analytics_event_id) = stored.analytics_event_id { - ObservabilityEmissionClaimV1::Settled { - delivery_envelope_json: stored.delivery_envelope_json, - analytics_event_id, - } - } else { - ObservabilityEmissionClaimV1::Pending { - delivery_envelope_json: stored.delivery_envelope_json, + let project_id = project_id.to_owned(); + let owner_event_id = owner_event_id.to_owned(); + let owner_fact_json = owner_fact_json.to_owned(); + let delivery_envelope_json = delivery_envelope_json.to_owned(); + self.transact_or( + |error| format!("failed to run observability outbox claim: {error}"), + async move |transaction| { + if let Some(stored) = + read_outbox_record(&transaction, &project_id, &owner_event_id).await? + { + if stored.owner_fact_json != owner_fact_json { + return Err("observability owner fact conflict".to_owned()); + } + return Ok( + if let Some(analytics_event_id) = stored.analytics_event_id { + ObservabilityEmissionClaimV1::Settled { + delivery_envelope_json: stored.delivery_envelope_json, + analytics_event_id, + } + } else { + ObservabilityEmissionClaimV1::Pending { + delivery_envelope_json: stored.delivery_envelope_json, + } + }, + ); } - }; - transaction - .commit() - .await - .map_err(|error| format!("failed to close observability outbox replay: {error}"))?; - return Ok(outcome); - } - transaction - .execute( - "INSERT INTO observability_emission_outbox - (project_id, owner_event_id, owner_fact_json, - delivery_envelope_json, state, analytics_event_id) - VALUES (?1, ?2, ?3, ?4, 'pending', NULL)", - tracedecay_runtime_core::db::engine::params![ - project_id, - owner_event_id, - owner_fact_json, - delivery_envelope_json - ], - ) - .await - .map_err(|error| format!("failed to claim observability outbox event: {error}"))?; - transaction - .commit() - .await - .map_err(|error| format!("failed to commit observability outbox claim: {error}"))?; - Ok(ObservabilityEmissionClaimV1::Claimed { - delivery_envelope_json: delivery_envelope_json.to_owned(), - }) + transaction + .execute( + "INSERT INTO observability_emission_outbox + (project_id, owner_event_id, owner_fact_json, + delivery_envelope_json, state, analytics_event_id) + VALUES (?1, ?2, ?3, ?4, 'pending', NULL)", + tracedecay_runtime_core::db::engine::params![ + project_id.as_str(), + owner_event_id.as_str(), + owner_fact_json.as_str(), + delivery_envelope_json.as_str() + ], + ) + .await + .map_err(|error| { + format!("failed to claim observability outbox event: {error}") + })?; + Ok(ObservabilityEmissionClaimV1::Claimed { + delivery_envelope_json: delivery_envelope_json.clone(), + }) + }, + ) + .await } /// CAS a pending delivery to its delayed-coverage representation. @@ -391,59 +390,56 @@ impl RegisteredGlobalDb { owner_fact_json, delayed_delivery_envelope_json, )?; - let transaction = self - .begin_write_transaction() - .await - .map_err(|error| format!("failed to begin observability outbox delay: {error}"))?; - let stored = read_outbox_record(&transaction, project_id, owner_event_id) - .await? - .ok_or_else(|| "observability outbox event is unavailable".to_owned())?; - if stored.owner_fact_json != owner_fact_json { - return Err("observability owner fact conflict".to_owned()); - } - if let Some(analytics_event_id) = stored.analytics_event_id { - transaction - .commit() - .await - .map_err(|error| format!("failed to close settled outbox delay: {error}"))?; - return Ok(ObservabilityEmissionClaimV1::Settled { - delivery_envelope_json: stored.delivery_envelope_json, - analytics_event_id, - }); - } - if stored.delivery_envelope_json != expected_delivery_envelope_json { - transaction - .commit() - .await - .map_err(|error| format!("failed to close replayed outbox delay: {error}"))?; - return Ok(ObservabilityEmissionClaimV1::Pending { - delivery_envelope_json: stored.delivery_envelope_json, - }); - } - transaction - .execute( - "UPDATE observability_emission_outbox - SET delivery_envelope_json = ?4 - WHERE project_id = ?1 AND owner_event_id = ?2 - AND owner_fact_json = ?3 AND state = 'pending' - AND delivery_envelope_json = ?5", - tracedecay_runtime_core::db::engine::params![ - project_id, - owner_event_id, - owner_fact_json, - delayed_delivery_envelope_json, - expected_delivery_envelope_json - ], - ) - .await - .map_err(|error| format!("failed to mark observability delivery delayed: {error}"))?; - transaction - .commit() - .await - .map_err(|error| format!("failed to commit observability outbox delay: {error}"))?; - Ok(ObservabilityEmissionClaimV1::Pending { - delivery_envelope_json: delayed_delivery_envelope_json.to_owned(), - }) + let project_id = project_id.to_owned(); + let owner_event_id = owner_event_id.to_owned(); + let owner_fact_json = owner_fact_json.to_owned(); + let expected_delivery_envelope_json = expected_delivery_envelope_json.to_owned(); + let delayed_delivery_envelope_json = delayed_delivery_envelope_json.to_owned(); + self.transact_or( + |error| format!("failed to run observability outbox delay: {error}"), + async move |transaction| { + let stored = read_outbox_record(&transaction, &project_id, &owner_event_id) + .await? + .ok_or_else(|| "observability outbox event is unavailable".to_owned())?; + if stored.owner_fact_json != owner_fact_json { + return Err("observability owner fact conflict".to_owned()); + } + if let Some(analytics_event_id) = stored.analytics_event_id { + return Ok(ObservabilityEmissionClaimV1::Settled { + delivery_envelope_json: stored.delivery_envelope_json, + analytics_event_id, + }); + } + if stored.delivery_envelope_json != expected_delivery_envelope_json { + return Ok(ObservabilityEmissionClaimV1::Pending { + delivery_envelope_json: stored.delivery_envelope_json, + }); + } + transaction + .execute( + "UPDATE observability_emission_outbox + SET delivery_envelope_json = ?4 + WHERE project_id = ?1 AND owner_event_id = ?2 + AND owner_fact_json = ?3 AND state = 'pending' + AND delivery_envelope_json = ?5", + tracedecay_runtime_core::db::engine::params![ + project_id.as_str(), + owner_event_id.as_str(), + owner_fact_json.as_str(), + delayed_delivery_envelope_json.as_str(), + expected_delivery_envelope_json.as_str() + ], + ) + .await + .map_err(|error| { + format!("failed to mark observability delivery delayed: {error}") + })?; + Ok(ObservabilityEmissionClaimV1::Pending { + delivery_envelope_json: delayed_delivery_envelope_json.clone(), + }) + }, + ) + .await } /// Append the exact delivery and settle its outbox row in one transaction. @@ -472,48 +468,54 @@ impl RegisteredGlobalDb { { return Err("observability outbox delivery binding conflict".to_owned()); } - let transaction = self - .begin_write_transaction() - .await - .map_err(|error| format!("failed to begin observability outbox settlement: {error}"))?; - let stored = read_outbox_record(&transaction, project_id, owner_event_id) - .await? - .ok_or_else(|| "observability outbox event is unavailable".to_owned())?; - if stored.owner_fact_json != owner_fact_json - || stored.delivery_envelope_json != delivery_envelope_json - { - return Err("observability outbox settlement conflict".to_owned()); - } - let id = append_observability_event_in_existing_tx(&transaction, event) - .await - .map_err(|error| error.to_string())?; - if let Some(settled_id) = stored.analytics_event_id { - if settled_id != id { - return Err("observability outbox receipt conflict".to_owned()); - } - } else { - transaction - .execute( - "UPDATE observability_emission_outbox - SET state = 'settled', analytics_event_id = ?5 - WHERE project_id = ?1 AND owner_event_id = ?2 - AND owner_fact_json = ?3 AND delivery_envelope_json = ?4 - AND state = 'pending'", - tracedecay_runtime_core::db::engine::params![ - project_id, - owner_event_id, - owner_fact_json, - delivery_envelope_json, - id - ], - ) - .await - .map_err(|error| format!("failed to settle observability outbox event: {error}"))?; - } - transaction.commit().await.map_err(|error| { - format!("failed to commit observability outbox settlement: {error}") - })?; - Ok(id) + let project_id = project_id.to_owned(); + let owner_event_id = owner_event_id.to_owned(); + let owner_fact_json = owner_fact_json.to_owned(); + let delivery_envelope_json = delivery_envelope_json.to_owned(); + let event = event.clone(); + self.transact_or( + |error| format!("failed to run observability outbox settlement: {error}"), + async move |transaction| { + let stored = read_outbox_record(&transaction, &project_id, &owner_event_id) + .await? + .ok_or_else(|| "observability outbox event is unavailable".to_owned())?; + if stored.owner_fact_json != owner_fact_json + || stored.delivery_envelope_json != delivery_envelope_json + { + return Err("observability outbox settlement conflict".to_owned()); + } + let id = append_observability_event_in_existing_tx(&transaction, &event) + .await + .map_err(|error| error.to_string())?; + if let Some(settled_id) = stored.analytics_event_id { + if settled_id != id { + return Err("observability outbox receipt conflict".to_owned()); + } + } else { + transaction + .execute( + "UPDATE observability_emission_outbox + SET state = 'settled', analytics_event_id = ?5 + WHERE project_id = ?1 AND owner_event_id = ?2 + AND owner_fact_json = ?3 AND delivery_envelope_json = ?4 + AND state = 'pending'", + tracedecay_runtime_core::db::engine::params![ + project_id.as_str(), + owner_event_id.as_str(), + owner_fact_json.as_str(), + delivery_envelope_json.as_str(), + id + ], + ) + .await + .map_err(|error| { + format!("failed to settle observability outbox event: {error}") + })?; + } + Ok(id) + }, + ) + .await } /// Claim and settle one run of worker-owned owner facts in a single write. @@ -630,43 +632,43 @@ impl RegisteredGlobalDb { if now_seconds < 0 { return Err("invalid observability retention time".to_owned()); } - let transaction = self - .begin_write_transaction() - .await - .map_err(|error| format!("failed to begin observability retention: {error}"))?; - // Expired dirty-day markers go first: they protect their source rows - // from the analytics pass below, and a day past the horizon has no - // generation left to rebuild. - let expired_rollups = expire_observability_rollups(&transaction, now_seconds).await?; - let (expired_detail_outbox, expired_detail) = prune_observability_retention_class( - &transaction, - now_seconds.saturating_sub(OBSERVABILITY_DETAIL_RETENTION_SECONDS), - "optional_local_detail30d", - ) - .await?; - let (expired_rollup_outbox, expired_rollup) = prune_observability_retention_class( - &transaction, - now_seconds.saturating_sub(OBSERVABILITY_ROLLUP_RETENTION_SECONDS), - "local_rollup395d", - ) - .await?; - let has_more = observability_retention_has_more( - &transaction, - now_seconds.saturating_sub(OBSERVABILITY_DETAIL_RETENTION_SECONDS), - now_seconds.saturating_sub(OBSERVABILITY_ROLLUP_RETENTION_SECONDS), + self.transact_or( + |error| format!("failed to run observability retention: {error}"), + async move |transaction| { + // Expired dirty-day markers go first: they protect their source rows + // from the analytics pass below, and a day past the horizon has no + // generation left to rebuild. + let expired_rollups = + expire_observability_rollups(&transaction, now_seconds).await?; + let (expired_detail_outbox, expired_detail) = prune_observability_retention_class( + &transaction, + now_seconds.saturating_sub(OBSERVABILITY_DETAIL_RETENTION_SECONDS), + "optional_local_detail30d", + ) + .await?; + let (expired_rollup_outbox, expired_rollup) = prune_observability_retention_class( + &transaction, + now_seconds.saturating_sub(OBSERVABILITY_ROLLUP_RETENTION_SECONDS), + "local_rollup395d", + ) + .await?; + let has_more = observability_retention_has_more( + &transaction, + now_seconds.saturating_sub(OBSERVABILITY_DETAIL_RETENTION_SECONDS), + now_seconds.saturating_sub(OBSERVABILITY_ROLLUP_RETENTION_SECONDS), + ) + .await?; + Ok(ObservabilityRetentionReceiptV1 { + expired_detail, + expired_rollup, + expired_settled_outbox: expired_detail_outbox + .saturating_add(expired_rollup_outbox), + expired_rollups, + has_more, + }) + }, ) - .await?; - transaction - .commit() - .await - .map_err(|error| format!("failed to commit observability retention: {error}"))?; - Ok(ObservabilityRetentionReceiptV1 { - expired_detail, - expired_rollup, - expired_settled_outbox: expired_detail_outbox.saturating_add(expired_rollup_outbox), - expired_rollups, - has_more, - }) + .await } #[tracing::instrument( @@ -701,22 +703,28 @@ impl RegisteredGlobalDb { expected_cursor: super::ParseOffset, cursor: super::ParseOffset, ) -> Result, String> { - let transaction = self - .begin_write_transaction() - .await - .map_err(|error| format!("failed to begin analytics import transaction: {error}"))?; - super::transcript::require_expected_offset(&transaction, cursor_path, expected_cursor) - .await - .map_err(|error| format!("failed to claim analytics import cursor: {error}"))?; - let ids = append_analytics_events_in_existing_tx(&transaction, events).await?; - super::transcript::set_parse_offset(&transaction, cursor_path, cursor) - .await - .map_err(|error| format!("failed to persist analytics import cursor: {error}"))?; - transaction - .commit() - .await - .map_err(|error| format!("failed to commit analytics import transaction: {error}"))?; - Ok(ids) + let events = events.to_vec(); + let cursor_path = cursor_path.to_owned(); + self.transact_or( + |error| format!("failed to run analytics import transaction: {error}"), + async move |transaction| { + super::transcript::require_expected_offset( + &transaction, + &cursor_path, + expected_cursor, + ) + .await + .map_err(|error| format!("failed to claim analytics import cursor: {error}"))?; + let ids = append_analytics_events_in_existing_tx(&transaction, &events).await?; + super::transcript::set_parse_offset(&transaction, &cursor_path, cursor) + .await + .map_err(|error| { + format!("failed to persist analytics import cursor: {error}") + })?; + Ok(ids) + }, + ) + .await } #[tracing::instrument( @@ -950,7 +958,7 @@ impl RegisteredGlobalDb { } async fn append_analytics_events_in_existing_tx( - transaction: &super::RegisteredGlobalDbWriteTransaction<'_>, + transaction: &impl Executor, events: &[AnalyticsEventInsert], ) -> Result, String> { let mut ids = Vec::with_capacity(events.len()); @@ -1016,7 +1024,7 @@ async fn append_analytics_events_in_existing_tx( } async fn append_analytics_event_in_existing_tx( - transaction: &super::RegisteredGlobalDbWriteTransaction<'_>, + transaction: &impl Executor, event: &AnalyticsEventInsert, ) -> Result { let mut rows = transaction @@ -1098,7 +1106,7 @@ fn validate_prepared_owner_emission( } async fn claim_and_settle_owner_facts_in_tx( - transaction: &RegisteredGlobalDbWriteTransaction<'_>, + transaction: &impl Executor, emissions: &[ObservabilityOwnerEmissionWriteV1], prepare_new: F, ) -> Result, String> @@ -1131,7 +1139,7 @@ where /// Resolves every still-open fact against the outbox and prepares a delivery /// only for a fact no row owns yet. async fn prepare_unclaimed_owner_facts( - transaction: &RegisteredGlobalDbWriteTransaction<'_>, + transaction: &impl Executor, emissions: &[ObservabilityOwnerEmissionWriteV1], outcomes: &mut [Option], mut prepare_new: F, @@ -1183,7 +1191,7 @@ where /// Adopts an exact stored event, appends the rest in one statement, and /// settles their outbox rows in one statement. async fn settle_prepared_owner_facts( - transaction: &RegisteredGlobalDbWriteTransaction<'_>, + transaction: &impl Executor, emissions: &[ObservabilityOwnerEmissionWriteV1], outcomes: &mut [Option], prepared: Vec<(usize, PreparedObservabilityEmissionV1)>, @@ -1263,7 +1271,7 @@ fn bind_key_pairs(mut sql: String, keys: Vec<(String, String)>) -> (String, Vec< /// Stored owner facts for `keys`, by `(project_id, owner_event_id)`. async fn stored_owner_facts( - transaction: &RegisteredGlobalDbWriteTransaction<'_>, + transaction: &impl Executor, keys: Vec<(String, String)>, ) -> Result, String> { let mut stored = BTreeMap::new(); @@ -1298,7 +1306,7 @@ async fn stored_owner_facts( /// Stored canonical observability events for `keys`, by /// `(project_id, hint_id)`. async fn stored_observability_events( - transaction: &RegisteredGlobalDbWriteTransaction<'_>, + transaction: &impl Executor, keys: Vec<(String, String)>, ) -> Result, String> { let mut stored = BTreeMap::new(); @@ -1336,7 +1344,7 @@ async fn stored_observability_events( } async fn insert_settled_outbox_rows( - transaction: &RegisteredGlobalDbWriteTransaction<'_>, + transaction: &impl Executor, rows: &[(&ObservabilityOwnerEmissionWriteV1, String, i64)], ) -> Result<(), String> { for chunk in rows.chunks(ANALYTICS_INSERT_ROWS_PER_STATEMENT) { @@ -1407,7 +1415,7 @@ fn validate_outbox_input( } async fn read_outbox_record( - transaction: &RegisteredGlobalDbWriteTransaction<'_>, + transaction: &impl Executor, project_id: &str, owner_event_id: &str, ) -> Result, String> { @@ -1454,7 +1462,7 @@ enum ObservabilityAppendError { /// All runs share the caller's transaction, so a later conflict rolls back /// earlier appends just as the single-event path does. async fn append_observability_events_in_existing_tx( - transaction: &RegisteredGlobalDbWriteTransaction<'_>, + transaction: &impl Executor, events: &[AnalyticsEventInsert], ) -> Result, String> { let mut ids = Vec::with_capacity(events.len()); @@ -1514,7 +1522,7 @@ async fn append_observability_events_in_existing_tx( } async fn append_observability_event_in_existing_tx( - transaction: &RegisteredGlobalDbWriteTransaction<'_>, + transaction: &impl Executor, event: &AnalyticsEventInsert, ) -> Result { validate_observability_event(event).map_err(ObservabilityAppendError::Failed)?; @@ -1583,7 +1591,7 @@ fn analytics_record_matches_insert( } async fn prune_observability_retention_class( - transaction: &RegisteredGlobalDbWriteTransaction<'_>, + transaction: &impl Executor, cutoff_seconds: i64, retention_class: &str, ) -> Result<(u64, u64), String> { @@ -1625,7 +1633,7 @@ async fn prune_observability_retention_class( } async fn observability_retention_has_more( - transaction: &RegisteredGlobalDbWriteTransaction<'_>, + transaction: &impl Executor, detail_cutoff_seconds: i64, rollup_cutoff_seconds: i64, ) -> Result { From f2f81a2598fe438307613a0fd53890f8c7125a2a Mon Sep 17 00:00:00 2001 From: ScriptedAlchemy Date: Sun, 18 Oct 2026 16:14:19 +0000 Subject: [PATCH 08/10] test(global-db): prove job exits and scope loss roll back --- crates/tracedecay-global-db/src/tests.rs | 86 ++++++++++++++++++++++++ 1 file changed, 86 insertions(+) diff --git a/crates/tracedecay-global-db/src/tests.rs b/crates/tracedecay-global-db/src/tests.rs index 9e471a6fdd..8629cda561 100644 --- a/crates/tracedecay-global-db/src/tests.rs +++ b/crates/tracedecay-global-db/src/tests.rs @@ -1483,6 +1483,92 @@ async fn transact_rolls_back_when_scope_ends_before_commit() { ); } +#[asupersync::test] +async fn transact_or_rollback_keeps_an_early_value_but_never_a_commit() { + use crate::registered::JobExit; + use tracedecay_runtime_core::db::engine::Executor as _; + let harness = RegisteredGlobalDbHarness::open("transact-or-rollback-exits").await; + let db = harness.registered.clone(); + + let replayed = db + .transact_or_rollback( + |error| error.to_string(), + async move |transaction| { + transaction + .execute("CREATE TABLE job_exit_rolled_back (value INTEGER)", ()) + .await + .map_err(|error| error.to_string())?; + Err::(JobExit::Rollback(7)) + }, + ) + .await; + assert_eq!(replayed, Ok(7)); + assert!( + !table_exists(&db, "job_exit_rolled_back").await, + "an early exit left its writes behind" + ); + + let failed = db + .transact_or_rollback( + |error| error.to_string(), + async move |transaction| { + transaction + .execute("CREATE TABLE job_exit_failed (value INTEGER)", ()) + .await + .map_err(|error| error.to_string())?; + Err::(JobExit::Fail("body failed".to_owned())) + }, + ) + .await; + assert_eq!(failed, Err("body failed".to_owned())); + assert!(!table_exists(&db, "job_exit_failed").await); +} + +#[asupersync::test] +async fn transact_or_wraps_a_scope_that_ended_mid_body_and_rolls_back() { + use tracedecay_runtime_core::db::engine::Executor as _; + let mut harness = RegisteredGlobalDbHarness::open("transact-or-scope-end").await; + let db = harness.registered.clone(); + let scope = std::sync::Arc::new(std::sync::Mutex::new(harness.take_scope())); + + let error = db + .transact_or( + |error| format!("wrapped: {error}"), + async move |transaction| { + transaction + .execute("CREATE TABLE transact_or_staged (value INTEGER)", ()) + .await + .map_err(|error| error.to_string())?; + drop(scope.lock().unwrap().take()); + Ok::<_, String>(()) + }, + ) + .await + .expect_err("transact_or committed after the scope ended"); + + assert!(error.starts_with("wrapped: "), "{error}"); + assert!(error.contains("active daemon"), "{error}"); + assert!( + !table_exists(&db, "transact_or_staged").await, + "scope loss did not roll the job back" + ); +} + +#[asupersync::test] +async fn converted_write_is_refused_without_a_write_scope() { + let mut harness = RegisteredGlobalDbHarness::open("converted-write-no-scope").await; + let db = harness.registered.clone(); + harness.revoke(); + + let error = db + .try_record_savings("/project", "tool", 10, 4, 1) + .await + .expect_err("a converted write ran without a write scope"); + + assert!(error.to_string().contains("active daemon"), "{error}"); + assert_eq!(row_count(&db, "savings_ledger").await, 0); +} + #[asupersync::test] async fn begun_registered_transaction_can_roll_back_after_scope_drop() { let mut harness = RegisteredGlobalDbHarness::open("registered-rollback").await; From 9afc4aa0a30a8c4960d40bba6f3c6471ab99025d Mon Sep 17 00:00:00 2001 From: ScriptedAlchemy Date: Sun, 18 Oct 2026 16:14:39 +0000 Subject: [PATCH 09/10] docs(storage): record step 5i-4 global-db job writes --- docs/plans/storage/README.md | 18 ++++++++++++++++++ 1 file changed, 18 insertions(+) diff --git a/docs/plans/storage/README.md b/docs/plans/storage/README.md index 021f5d860f..67e644e9a0 100644 --- a/docs/plans/storage/README.md +++ b/docs/plans/storage/README.md @@ -773,6 +773,24 @@ changes a file that holds memory. clones what it owns into the closure. A rollup rebuild that finds its stored receipt still ends by rolling back, so replay does not advance the transaction generation. +- Step 5i-4, global-db job writes (2026-10-18, branch + `refactor/storage-global-db-jobs-2`). The accounting, workflow-index, topology + anchor, discovery queue, GitHub stack delivery, delivery settlement, source + rewrite, registry forget and retire, remote deletion and analytics outbox writes + now run as `RegisteredGlobalDb::transact` jobs. `transact_or` maps a refused job + to the site's own error type. `transact_or_rollback` lets a body end early with + a value the caller wants as success, as a replayed append does, and the job + rolls back so a replay leaves no commit. Helpers under these sites take + `&impl Executor`, so one body serves the job and the remaining interactive + transactions. `WorkflowIngestWriteTxn` had no caller left and is deleted. Still + interactive, for later slices: `apply_registry_gc` (it reads outside the + transaction while holding it), `relink_orphan_store_instance` (a manifest file + effect sits between the SQL and the commit), `claim_and_settle_observability_emissions` + (its `FnMut` allocates producer sequences, so a re-run would allocate twice), + `append_analytics_event` (a writer connection, not an interactive transaction), + and the project registry, observation projection, LCM, provider usage, + dashboard, session, schema-stage and invariant, git-index and observation + retention and canonical-body groups. - #4138 (2026-10-16). Shutdown store close waits up to 3 s for leases to drain before deciding a store's WAL cannot be truncated. Under load, owners joined before store close could still be unwinding the tasks that held their leases, From 19dd56dcd3d9ba773da7c218552d77752dd6575e Mon Sep 17 00:00:00 2001 From: ScriptedAlchemy Date: Sun, 18 Oct 2026 20:59:25 +0000 Subject: [PATCH 10/10] test(observability): reserve space for every live carrier --- .../tests/application_suite/observability_runtime_contract.rs | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/crates/tracedecay-application/tests/application_suite/observability_runtime_contract.rs b/crates/tracedecay-application/tests/application_suite/observability_runtime_contract.rs index f76f1bec62..98425ae3e8 100644 --- a/crates/tracedecay-application/tests/application_suite/observability_runtime_contract.rs +++ b/crates/tracedecay-application/tests/application_suite/observability_runtime_contract.rs @@ -341,7 +341,8 @@ async fn live_queued_owner_claim_is_not_recovered_as_delayed_work() { configuration_revision: "configuration.owner.v1".into(), policy_revision: "policy.owner.v1".into(), }, - 2, + // Both ordinary carriers and the owner must fit before the native worker polls. + 3, ) .expect("producer"), ); @@ -351,7 +352,6 @@ async fn live_queued_owner_claim_is_not_recovered_as_delayed_work() { .expect("first queue carrier"), ObservabilityEmissionOutcomeV1::Enqueued ); - tokio::task::yield_now().await; assert_eq!( producer .try_emit(envelope(&scope, "caller", 73, 7_300_000))