refactor(global-db): run configuration and rollup writes as jobs

crates/tracedecay-global-db/src/configuration/store/revision.rs

@@ -9,6 +9,7 @@ use super::read::{
     current_revision_id_from_executor, read_change_plan_from_executor, read_revision_from_executor,
     validate_snapshot_registry_completeness_with_registry,
 };
+use super::transact_config;
 use super::write::{insert_change_plan, insert_snapshot_entries};
 use super::{
     ChangePlanId, ConfigurationAuditEvent, ConfigurationAuditEventId, ConfigurationCommitV1,
@@ -100,29 +101,17 @@ impl ConfigurationRevisionStore for GlobalDbConfigurationControlStore<'_> {
         let plan = plan.clone();
         async move {
             tracing::Instrument::instrument(
-                async {
+                async move {
                     plan.validate().map_err(ConfigurationStoreError::from)?;
-                    let transaction = self
-                        .db
-                        .begin_write_transaction()
-                        .await
-                        .map_err(unavailable_store)?;
-                    let outcome = match read_change_plan_from_executor(
-                        &transaction,
-                        &plan.plan.plan_id,
-                    )
-                    .await
-                    {
-                        Ok(Some(existing)) if existing == plan => Ok(()),
-                        Ok(Some(_)) => Err(ConfigurationStoreError::IdempotencyConflict),
-                        Ok(None) => insert_change_plan(&transaction, &plan).await,
-                        Err(error) => Err(error),
-                    };
-                    crate::sqlite_persist::commit_outcome(
-                        transaction,
-                        outcome,
-                        ConfigurationStoreError::Unavailable,
-                    )
+                    transact_config(self.db, unavailable_store, async move |transaction| {
+                        match read_change_plan_from_executor(&transaction, &plan.plan.plan_id).await
+                        {
+                            Ok(Some(existing)) if existing == plan => Ok(()),
+                            Ok(Some(_)) => Err(ConfigurationStoreError::IdempotencyConflict),
+                            Ok(None) => insert_change_plan(&transaction, &plan).await,
+                            Err(error) => Err(error),
+                        }
+                    })
                     .await
                 },
                 tracing::trace_span!("global_db.configuration.persist.plan"),
@@ -160,17 +149,9 @@ impl ConfigurationRevisionStore for GlobalDbConfigurationControlStore<'_> {
         commit: ConfigurationCommitV1,
     ) -> ConfigurationStoreResult<ConfigurationMutationReceiptV1> {
         commit.validate().map_err(ConfigurationStoreError::from)?;
-        let transaction = self
-            .db
-            .begin_write_transaction()
-            .await
-            .map_err(unavailable_store)?;
-        let outcome = commit_configuration_transaction(&transaction, &commit, false, None).await;
-        crate::sqlite_persist::commit_outcome(
-            transaction,
-            outcome,
-            ConfigurationStoreError::Unavailable,
-        )
+        transact_config(self.db, unavailable_store, async move |transaction| {
+            commit_configuration_transaction(&transaction, &commit, false, None).await
+        })
         .await
     }
 

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

@@ -134,8 +134,8 @@ pub use registered::{
     MAX_PENDING_RECEIPTED_DELIVERIES_V1, MAX_WORK_ATTEMPT_DELIVERY_FANOUTS_V1,
     PendingDeliverySourceReceiptV1, RegisteredGlobalDb, RegisteredGlobalDbLeaseV1,
     RegisteredGlobalDbOwnerV1, RegisteredGlobalDbWeakLeaseIssuerV1,
-    RegisteredGlobalDbWriteTransaction, RegisteredGlobalDbWriterConnection,
-    WorkAttemptDeliveryCensusReadV1,
+    RegisteredGlobalDbWriteTransaction, RegisteredGlobalDbWriteTx,
+    RegisteredGlobalDbWriterConnection, WorkAttemptDeliveryCensusReadV1,
 };
 pub use registered_analytics::ObservabilityRetentionReceiptV1;
 pub use registered_lcm_privacy::{LcmPrivacyRescanOutcomeV1, LcmPrivacyRescanReceiptV1};