refactor(global-db): run more registered writes as job transactions

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))

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.

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::<i64>(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::<i64>(0)
-                .map_err(|error| format!("decode host discovery queue sequence: {error}"))?,
-            row.get::<String>(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::<i64>(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::<i64>(0).map_err(|error| {
+                        format!("decode host discovery queue sequence: {error}")
+                    })?,
+                    row.get::<String>(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)]

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::<Vec<_>>();
-        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(

crates/tracedecay-global-db/src/observability_rollup.rs

@@ -41,9 +41,11 @@ where
     R: Send + 'static,
     E: From<String> + 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

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<ObservabilityRollupRetentionReceiptV1, String> {
     let cutoff =

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`.