diff --git a/crates/tracedecay-global-db/src/configuration/store.rs b/crates/tracedecay-global-db/src/configuration/store.rs index 52fd67d000..e8eed26376 100644 --- a/crates/tracedecay-global-db/src/configuration/store.rs +++ b/crates/tracedecay-global-db/src/configuration/store.rs @@ -1,8 +1,5 @@ //! `SQLite` persistence behind the final configuration control plane. -use std::future::Future; -use std::pin::Pin; - use super::contracts::{ ActivationDriftV1, AuthorizedActor, CONFIGURATION_AUDIT_PAGE_LIMIT, ComponentConfigurationState, ConfigurationAuditPage, ConfigurationAuditQuery, @@ -13,7 +10,7 @@ use super::contracts::{ }; use super::registry::ConfigurationRegistry; use super::resolver::{ConfigurationResolutionV1, registry_default_candidate}; -use crate::{RegisteredGlobalDb, RegisteredGlobalDbLeaseV1}; +use crate::{RegisteredGlobalDb, RegisteredGlobalDbLeaseV1, RegisteredGlobalDbWriteTx}; use tracedecay_domain::configuration::{ ACCESS_RULES_SETTING_KEY, CandidateDispositionV1, ChangePlanId, CodeIndexWorkerSelectionV1, ConfigurationAuditEvent, ConfigurationAuditEventId, ConfigurationAuditEventKindV1, @@ -67,6 +64,20 @@ pub use mutation::{ ConfigurationDirectCommitOutcomeV1, commit_direct_in_transaction, protected_change_snapshot_v1, }; +/// Runs `body` as one write job. A refused or failed job becomes the error +/// `unavailable` builds; the body's own error is returned as is. +pub(super) async fn transact_config( + db: &RegisteredGlobalDb, + unavailable: impl FnOnce(tracedecay_domain::errors::TraceDecayError) -> E, + body: impl AsyncFn(RegisteredGlobalDbWriteTx<'_, '_>) -> Result + Send + Sync + 'static, +) -> Result +where + R: Send + 'static, + E: Send + 'static, +{ + db.transact(body).await.map_err(unavailable)? +} + /// Concrete control-plane adapter over one already-open owned session store. /// It never accepts an arbitrary connection, opens a fallback database, or /// owns policy resolution; every write obtains the selected store's serialized @@ -184,56 +195,51 @@ impl<'db> GlobalDbConfigurationControlStore<'db> { .map_err(ConfigurationError::validation)?; validate_snapshot_registry_completeness_with_registry(&resolution.snapshot, registry) .map_err(map_store_error)?; - let transaction = self - .db - .begin_write_transaction() - .await - .map_err(|_| ConfigurationError::Unavailable)?; - let outcome = async { - let mut rows = transaction - .query("SELECT COUNT(*) FROM configuration_revisions", ()) - .await - .map_err(|_| ConfigurationError::Unavailable)?; - let existing = rows - .next() - .await - .map_err(|_| ConfigurationError::Unavailable)? - .ok_or_else(|| { - ConfigurationError::validation_message( - "configuration initialization count returned no row", - ) - })? - .get::(0) - .map_err(|_| { - ConfigurationError::validation_message( - "configuration initialization count is not an integer", - ) - })?; - if existing != 0 { - return Err(ConfigurationError::RevisionConflict); - } - let actor_id = ActorId::new("actor.configuration-initialization".to_owned()) - .map_err(ConfigurationError::validation)?; - let revision = ConfigurationRevisionRecordV1 { - revision_id: revision_id.clone(), - parent_revision_id: None, - snapshot: resolution.snapshot.clone(), - actor_id, - operation_kind: "canonical_initialization".to_owned(), - created_at: occurred_at, - }; - insert_revision_with_registry(&transaction, &revision, registry) - .await - .map_err(map_store_error) - } - .await; - match outcome { - Ok(()) => transaction - .commit() - .await - .map_err(|_| ConfigurationError::Unavailable), - Err(error) => Err(error), - } + let revision_id = revision_id.clone(); + let snapshot = resolution.snapshot.clone(); + let registry = registry.clone(); + transact_config( + self.db, + |_| ConfigurationError::Unavailable, + async move |transaction| { + let mut rows = transaction + .query("SELECT COUNT(*) FROM configuration_revisions", ()) + .await + .map_err(|_| ConfigurationError::Unavailable)?; + let existing = rows + .next() + .await + .map_err(|_| ConfigurationError::Unavailable)? + .ok_or_else(|| { + ConfigurationError::validation_message( + "configuration initialization count returned no row", + ) + })? + .get::(0) + .map_err(|_| { + ConfigurationError::validation_message( + "configuration initialization count is not an integer", + ) + })?; + if existing != 0 { + return Err(ConfigurationError::RevisionConflict); + } + let actor_id = ActorId::new("actor.configuration-initialization".to_owned()) + .map_err(ConfigurationError::validation)?; + let revision = ConfigurationRevisionRecordV1 { + revision_id: revision_id.clone(), + parent_revision_id: None, + snapshot: snapshot.clone(), + actor_id, + operation_kind: "canonical_initialization".to_owned(), + created_at: occurred_at, + }; + insert_revision_with_registry(&transaction, &revision, ®istry) + .await + .map_err(map_store_error) + }, + ) + .await } /// Publishes one daemon-owned source binding: the project-open binding @@ -271,64 +277,57 @@ impl<'db> GlobalDbConfigurationControlStore<'db> { } }; change.validate().map_err(ConfigurationError::validation)?; - let transaction = self - .db - .begin_write_transaction() - .await - .map_err(|_| ConfigurationError::Unavailable)?; - let outcome = async { - let current = current_state_from_transaction(&transaction).await?; - if ¤t.revision_id != expected_revision_id { - return Err(ConfigurationError::RevisionConflict); - } - let operation_digest = - canonical_sha256(&("tracedecay.configuration.daemon-source-binding.v1", change)) - .map_err(ConfigurationError::validation)?; - let next_revision_id: ConfigurationRevisionId = derived_identifier( - "configuration.revision.v1", - &canonical_sha256(&( - "tracedecay.configuration.result-revision.v1", - expected_revision_id, - &operation_digest, + let expected_revision_id = expected_revision_id.clone(); + let change = change.clone(); + transact_config( + self.db, + |_| ConfigurationError::Unavailable, + async move |transaction| { + let current = current_state_from_transaction(&transaction).await?; + if current.revision_id != expected_revision_id { + return Err(ConfigurationError::RevisionConflict); + } + let operation_digest = canonical_sha256(&( + "tracedecay.configuration.daemon-source-binding.v1", + &change, )) - .map_err(ConfigurationError::validation)?, - "configuration daemon source binding revision id", - )?; - let snapshot = - protected_change_snapshot_v1(¤t.snapshot, change, &next_revision_id)?; - validate_snapshot_registry_completeness(&snapshot).map_err(map_store_error)?; - let actor_id = ActorId::new("actor.tracedecay-daemon.source-binding".to_owned()) .map_err(ConfigurationError::validation)?; - let revision = ConfigurationRevisionRecordV1 { - revision_id: next_revision_id.clone(), - parent_revision_id: Some(expected_revision_id.clone()), - snapshot, - actor_id, - operation_kind: operation_kind.to_owned(), - created_at: occurred_at, - }; - insert_revision(&transaction, &revision) - .await - .map_err(map_store_error)?; - advance_component_desired_state(&transaction, &next_revision_id, occurred_at) - .await - .map_err(map_store_error)?; - Ok(ConfigurationCurrentStateV1 { - revision_id: revision.revision_id, - snapshot: revision.snapshot, - }) - } - .await; - match outcome { - Ok(state) => { - transaction - .commit() + let next_revision_id: ConfigurationRevisionId = derived_identifier( + "configuration.revision.v1", + &canonical_sha256(&( + "tracedecay.configuration.result-revision.v1", + &expected_revision_id, + &operation_digest, + )) + .map_err(ConfigurationError::validation)?, + "configuration daemon source binding revision id", + )?; + let snapshot = + protected_change_snapshot_v1(¤t.snapshot, &change, &next_revision_id)?; + validate_snapshot_registry_completeness(&snapshot).map_err(map_store_error)?; + let actor_id = ActorId::new("actor.tracedecay-daemon.source-binding".to_owned()) + .map_err(ConfigurationError::validation)?; + let revision = ConfigurationRevisionRecordV1 { + revision_id: next_revision_id.clone(), + parent_revision_id: Some(expected_revision_id.clone()), + snapshot, + actor_id, + operation_kind: operation_kind.to_owned(), + created_at: occurred_at, + }; + insert_revision(&transaction, &revision) .await - .map_err(|_| ConfigurationError::Unavailable)?; - Ok(state) - } - Err(error) => Err(error), - } + .map_err(map_store_error)?; + advance_component_desired_state(&transaction, &next_revision_id, occurred_at) + .await + .map_err(map_store_error)?; + Ok(ConfigurationCurrentStateV1 { + revision_id: revision.revision_id, + snapshot: revision.snapshot, + }) + }, + ) + .await } /// Converge a snapshot written by an earlier published registry onto the @@ -376,31 +375,30 @@ impl<'db> GlobalDbConfigurationControlStore<'db> { expected_revision_id .validate() .map_err(ConfigurationError::validation)?; - let transaction = self - .db - .begin_write_transaction() - .await - .map_err(|_| ConfigurationError::Unavailable)?; - let outcome: Pin< - Box< - dyn Future> - + Send - + '_, - >, - > = Box::pin(async { + let expected_revision_id = expected_revision_id.clone(); + let registry = registry.clone(); + let additive_keys = additive_keys + .iter() + .map(|key| (*key).to_owned()) + .collect::>(); + let retired_keys = retired_keys + .iter() + .map(|key| (*key).to_owned()) + .collect::>(); + transact_config(self.db, |_| ConfigurationError::Unavailable, async move |transaction| { let current = current_state_from_transaction(&transaction).await?; - if ¤t.revision_id != expected_revision_id { + if current.revision_id != expected_revision_id { return Err(ConfigurationError::RevisionConflict); } let additive_keys = additive_keys .iter() - .copied() + .cloned() .map(SettingKey::new) .collect::, _>>() .map_err(ConfigurationError::validation)?; let retired_keys = retired_keys .iter() - .copied() + .cloned() .map(SettingKey::new) .collect::, _>>() .map_err(ConfigurationError::validation)?; @@ -442,7 +440,7 @@ impl<'db> GlobalDbConfigurationControlStore<'db> { .map(|definition| definition.key.clone()) .collect::>(); if missing_keys.is_empty() && removals.is_empty() && stale_defaults.is_empty() { - validate_snapshot_registry_completeness_with_registry(¤t.snapshot, registry) + validate_snapshot_registry_completeness_with_registry(¤t.snapshot, ®istry) .map_err(map_store_error)?; return Ok(current); } @@ -470,7 +468,7 @@ impl<'db> GlobalDbConfigurationControlStore<'db> { .map_err(ConfigurationError::validation)?; let operation_digest = canonical_sha256(&( "tracedecay.configuration.registry-shape-convergence.v1", - expected_revision_id, + &expected_revision_id, &additions, &removals, )) @@ -479,7 +477,7 @@ impl<'db> GlobalDbConfigurationControlStore<'db> { "configuration.revision.v1", &canonical_sha256(&( "tracedecay.configuration.result-revision.v1", - expected_revision_id, + &expected_revision_id, &operation_digest, )) .map_err(ConfigurationError::validation)?, @@ -497,7 +495,7 @@ impl<'db> GlobalDbConfigurationControlStore<'db> { } let snapshot = ConfigurationSnapshotV1::new(effective_values, provenance) .map_err(ConfigurationError::validation)?; - validate_snapshot_registry_completeness_with_registry(&snapshot, registry) + validate_snapshot_registry_completeness_with_registry(&snapshot, ®istry) .map_err(map_store_error)?; let revision = ConfigurationRevisionRecordV1 { revision_id: next_revision_id.clone(), @@ -510,7 +508,7 @@ impl<'db> GlobalDbConfigurationControlStore<'db> { operation_kind: "registry_shape_convergence".to_owned(), created_at: occurred_at, }; - insert_revision_with_registry(&transaction, &revision, registry) + insert_revision_with_registry(&transaction, &revision, ®istry) .await .map_err(map_store_error)?; advance_component_desired_state(&transaction, &next_revision_id, occurred_at) @@ -520,18 +518,8 @@ impl<'db> GlobalDbConfigurationControlStore<'db> { revision_id: revision.revision_id, snapshot: revision.snapshot, }) - }); - let outcome = outcome.await; - match outcome { - Ok(state) => { - transaction - .commit() - .await - .map_err(|_| ConfigurationError::Unavailable)?; - Ok(state) - } - Err(error) => Err(error), - } + }) + .await } /// Records a daemon/component activation result. Failed activation keeps @@ -548,70 +536,65 @@ impl<'db> GlobalDbConfigurationControlStore<'db> { .map_err(|reason| map_store_error(invalid_store_data(reason)))?; validate_activation_error_code(activation_error_code.as_deref()) .map_err(|reason| map_store_error(invalid_store_data(reason)))?; - let transaction = self - .db - .begin_write_transaction() - .await - .map_err(|_| ConfigurationError::Unavailable)?; - let outcome = async { - let current = current_state_from_transaction(&transaction).await?; - if let Some(observed_revision_id) = &observed_revision_id - && read_revision_from_executor(&transaction, observed_revision_id) + transact_config( + self.db, + |_| ConfigurationError::Unavailable, + async move |transaction| { + let component = component.clone(); + let observed_revision_id = observed_revision_id.clone(); + let activation_error_code = activation_error_code.clone(); + let current = current_state_from_transaction(&transaction).await?; + if let Some(observed_revision_id) = &observed_revision_id + && read_revision_from_executor(&transaction, observed_revision_id) + .await + .map_err(map_store_error)? + .is_none() + { + return Err(ConfigurationError::PlanStale); + } + let prior = latest_component_activation_state(&transaction, &component) .await - .map_err(map_store_error)? - .is_none() - { - return Err(ConfigurationError::PlanStale); - } - let prior = latest_component_activation_state(&transaction, &component) - .await - .map_err(map_store_error)?; - let prior_last_working = prior - .as_ref() - .and_then(|state| state.last_working_revision_id.clone()) - .or_else(|| { - prior - .as_ref() - .and_then(|state| state.observed_revision_id.clone()) - }); - let failed = activation_error_code.is_some(); - let last_working_revision_id = if failed { - prior_last_working.clone() - } else { - observed_revision_id - .clone() - .or_else(|| prior_last_working.clone()) - }; - let observed_revision_id = if failed { - prior_last_working - } else { - observed_revision_id - }; - let restart_required = - observed_revision_id.as_ref() != Some(¤t.revision_id) || failed; - insert_component_activation_event( - &transaction, - &StoredComponentActivationState { - component, - desired_revision_id: current.revision_id, - observed_revision_id, - last_working_revision_id, - restart_required, - activation_error_code, - }, - occurred_at, - ) - .await - .map_err(map_store_error) - } - .await; - match outcome { - Ok(()) => transaction - .commit() + .map_err(map_store_error)?; + let prior_last_working = prior + .as_ref() + .and_then(|state| state.last_working_revision_id.clone()) + .or_else(|| { + prior + .as_ref() + .and_then(|state| state.observed_revision_id.clone()) + }); + let failed = activation_error_code.is_some(); + let last_working_revision_id = if failed { + prior_last_working.clone() + } else { + observed_revision_id + .clone() + .or_else(|| prior_last_working.clone()) + }; + let observed_revision_id = if failed { + prior_last_working + } else { + observed_revision_id + }; + let restart_required = + observed_revision_id.as_ref() != Some(¤t.revision_id) || failed; + insert_component_activation_event( + &transaction, + &StoredComponentActivationState { + component, + desired_revision_id: current.revision_id, + observed_revision_id, + last_working_revision_id, + restart_required, + activation_error_code, + }, + occurred_at, + ) .await - .map_err(|_| ConfigurationError::Unavailable), - Err(error) => Err(error), - } + .map_err(map_store_error) + }, + ) + .await }) } } @@ -748,7 +731,7 @@ impl<'db> ProfileConfigurationStore<'db> { .effective_values .contains_key(&definition.key) }) { - return self.project(current, ®istry); + return Self::project(&self.profile_id, current, ®istry); } let converged = match self .store @@ -765,7 +748,7 @@ impl<'db> ProfileConfigurationStore<'db> { Err(ConfigurationError::RevisionConflict) => self.current_state().await?, Err(error) => return Err(error), }; - self.project(converged, ®istry) + Self::project(&self.profile_id, converged, ®istry) } #[tracing::instrument(name = "global_db.configuration.query", level = "trace", skip_all)] @@ -808,34 +791,30 @@ impl<'db> ProfileConfigurationStore<'db> { expected_revision: &ConfigurationRevisionId, ) -> Result { let registry = ConfigurationRegistry::profile().map_err(ConfigurationError::validation)?; - let transaction = self - .store - .db - .begin_write_transaction() - .await - .map_err(|_| ConfigurationError::Unavailable)?; - let outcome = commit_direct_in_transaction_with_registry( - &transaction, - authority, - mutation, - expected_revision, - ®istry, - ) - .await; - match outcome { - Ok(outcome) => { - let current = self.project(outcome.current, ®istry)?; - transaction - .commit() - .await - .map_err(|_| ConfigurationError::Unavailable)?; + let authority = authority.clone(); + let mutation = mutation.clone(); + let expected_revision = expected_revision.clone(); + let profile_id = self.profile_id.clone(); + transact_config( + self.store.db, + |_| ConfigurationError::Unavailable, + async move |transaction| { + let outcome = commit_direct_in_transaction_with_registry( + &transaction, + &authority, + &mutation, + &expected_revision, + ®istry, + ) + .await?; + let current = Self::project(&profile_id, outcome.current, ®istry)?; Ok(ProfileConfigurationCommitV1 { receipt: outcome.receipt, current, }) - } - Err(error) => Err(error), - } + }, + ) + .await } #[tracing::instrument( @@ -862,7 +841,7 @@ impl<'db> ProfileConfigurationStore<'db> { /// Every value is registered and typed, and every winning candidate is /// this profile's own layer or the registry default. fn project( - &self, + expected_profile: &UserProfileId, current: ConfigurationCurrentStateV1, registry: &ConfigurationRegistry, ) -> Result { @@ -878,7 +857,7 @@ impl<'db> ProfileConfigurationStore<'db> { if provenance.iter().any(|candidate| match &candidate.layer { ConfigurationLayerIdV1::Default => false, ConfigurationLayerIdV1::UserProfile { profile_id } => { - profile_id != &self.profile_id + profile_id != expected_profile } ConfigurationLayerIdV1::Project { .. } | ConfigurationLayerIdV1::Collection { .. } => true, @@ -923,19 +902,10 @@ impl OwnedGlobalDbConfigurationControlStore { self.db.clone() } - /// Revalidate mutation access by acquiring the exact guarded writer - /// transaction. The capability carries the client-bound authority, so no - /// path-derived authority can be substituted here. - async fn require_active_mutation_scope( - db: &RegisteredGlobalDb, - ) -> Result<(), ConfigurationError> { - let transaction = db - .begin_write_transaction() - .await - .map_err(|_| ConfigurationError::Unavailable)?; - transaction - .rollback() - .await + /// Revalidate mutation access against the capability's client-bound write + /// authority, so no path-derived authority can be substituted here. + fn require_active_mutation_scope(db: &RegisteredGlobalDb) -> Result<(), ConfigurationError> { + db.require_active_write_scope("begin registered global database transaction") .map_err(|_| ConfigurationError::Unavailable) } @@ -948,7 +918,7 @@ impl OwnedGlobalDbConfigurationControlStore { ) -> ConfigurationOperationFuture<'_, ()> { let db = self.database(); Box::pin(async move { - Self::require_active_mutation_scope(db.as_ref()).await?; + Self::require_active_mutation_scope(db.as_ref())?; let store = GlobalDbConfigurationControlStore::new_registered(db.as_ref()); store .record_component_activation( @@ -1013,8 +983,7 @@ macro_rules! forward_to_registered { let db = $self.database(); $(let $owned = $owned.clone();)* Box::pin(async move { - OwnedGlobalDbConfigurationControlStore::require_active_mutation_scope(db.as_ref()) - .await?; + OwnedGlobalDbConfigurationControlStore::require_active_mutation_scope(db.as_ref())?; let $store = GlobalDbConfigurationControlStore::new_registered(db.as_ref()); $call.await }) diff --git a/crates/tracedecay-global-db/src/configuration/store/control.rs b/crates/tracedecay-global-db/src/configuration/store/control.rs index 16d16152b2..096293a722 100644 --- a/crates/tracedecay-global-db/src/configuration/store/control.rs +++ b/crates/tracedecay-global-db/src/configuration/store/control.rs @@ -9,6 +9,7 @@ use super::mutation::{ rollback_redacted_changes, validate_apply_request, validate_plan_evidence, }; use super::read::{read_change_plan_from_executor, read_revision_from_executor}; +use super::transact_config; use super::write::insert_change_plan; use super::{ ActivationDriftV1, AuthorizedActor, CONFIGURATION_AUDIT_PAGE_LIMIT, ChangePlanId, @@ -49,19 +50,19 @@ impl ConfigurationControlStore for GlobalDbConfigurationControlStore<'_> { operation: ConfigurationProtectedOperationV1::Change(Box::new(operation)), }; record.validate().map_err(ConfigurationError::validation)?; - let transaction = self - .db - .begin_write_transaction() - .await - .map_err(|_| ConfigurationError::Unavailable)?; - let outcome = match read_change_plan_from_executor(&transaction, &record.plan.plan_id) + transact_config( + self.db, + |_| ConfigurationError::Unavailable, + async move |transaction| match read_change_plan_from_executor( + &transaction, + &record.plan.plan_id, + ) .await .map_err(map_store_error)? - { - Some(existing) if existing == record => Ok(()), - Some(_) => Err(ConfigurationError::IdempotencyConflict), - None => { - async { + { + Some(existing) if existing == record => Ok(()), + Some(_) => Err(ConfigurationError::IdempotencyConflict), + None => { insert_change_plan(&transaction, &record) .await .map_err(map_store_error)?; @@ -69,16 +70,9 @@ impl ConfigurationControlStore for GlobalDbConfigurationControlStore<'_> { .await .map_err(map_store_error) } - .await - } - }; - match outcome { - Ok(()) => transaction - .commit() - .await - .map_err(|_| ConfigurationError::Unavailable), - Err(error) => Err(error), - } + }, + ) + .await }) } @@ -147,26 +141,21 @@ impl ConfigurationControlStore for GlobalDbConfigurationControlStore<'_> { let mutation = mutation.clone(); let expected_revision = expected_revision.clone(); Box::pin(async move { - let transaction = self - .db - .begin_write_transaction() - .await - .map_err(|_| ConfigurationError::Unavailable)?; - let outcome = commit_direct_in_transaction( - &transaction, - &authority, - &mutation, - &expected_revision, - ) - .await; - match outcome { - Ok(outcome) => transaction - .commit() + transact_config( + self.db, + |_| ConfigurationError::Unavailable, + async move |transaction| { + commit_direct_in_transaction( + &transaction, + &authority, + &mutation, + &expected_revision, + ) .await - .map(|()| outcome.receipt) - .map_err(|_| ConfigurationError::Unavailable), - Err(error) => Err(error), - } + .map(|outcome| outcome.receipt) + }, + ) + .await }) } @@ -196,108 +185,100 @@ impl ConfigurationControlStore for GlobalDbConfigurationControlStore<'_> { { return Err(ConfigurationError::PlanStale); } - let transaction = self - .db - .begin_write_transaction() - .await - .map_err(|_| ConfigurationError::Unavailable)?; - let outcome = async { - if let Some(receipt) = replay_control_receipt( - &transaction, - &authority.receipt.actor_id, - &request.idempotency_key, - &plan.base_revision_id, - &request.operation_digest, - Some(&plan.plan_id), - ) - .await? - { - return Ok(receipt); - } - let current = current_state_from_transaction(&transaction).await?; - if current.revision_id != plan.base_revision_id { - return Err(ConfigurationError::PlanStale); - } - let record = read_change_plan_from_executor(&transaction, &plan.plan_id) - .await - .map_err(map_store_error)? - .ok_or(ConfigurationError::PlanStale)?; - if record.plan != plan { - return Err(ConfigurationError::PlanStale); - } - let ConfigurationProtectedOperationV1::Change(change) = &record.operation else { - return Err(ConfigurationError::PlanStale); - }; - if record - .operation - .operation_digest() - .map_err(ConfigurationError::validation)? - != request.operation_digest - { - return Err(ConfigurationError::PlanStale); - } - let next_revision_id = result_revision_id( - &plan.base_revision_id, - &request.idempotency_key, - &request.operation_digest, - )?; - let snapshot = - protected_change_snapshot_v1(¤t.snapshot, change, &next_revision_id)?; - let sealed_target = - StoredConfigurationProtectedOperationV1::from(&record.operation); - let (commit, sealed_target_reference) = build_configuration_commit( - &transaction, - ConfigurationCommitDraft { - expected_base_revision_id: &plan.base_revision_id, - next_revision_id, - snapshot, - actor_id: &authority.receipt.actor_id, - operation_kind: "protected_apply", - operation_digest: request.operation_digest.clone(), - idempotency_key: request.idempotency_key.clone(), - authorization_policy_epoch: authority.receipt.policy_epoch, - authorization_policy_digest: authority.receipt.policy_digest.clone(), - authority_revalidated_at: authority.receipt.issued_at, - change_plan: Some(plan.clone()), - event_kind: ConfigurationAuditEventKindV1::Applied, - created_at: authority.receipt.issued_at, - effective_deadline_at: authority.receipt.expires_at, - target: &sealed_target, - }, - ) - .await?; - let receipt = commit_configuration_transaction( - &transaction, - &commit, - false, - Some(&sealed_target_reference), - ) - .await - .map_err(map_store_error)?; - Ok(ConfigurationMutationReceipt { - receipt_id: receipt.receipt_id, - base_revision_id: receipt.base_revision_id, - result_revision_id: receipt.result_revision_id, - snapshot_id: commit.next_revision.snapshot.snapshot_id, - operation_digest: receipt.operation_digest, - settlement_authority: ConfigurationSettlementAuthorityV1 { - policy_epoch: receipt.authorization_policy_epoch, - policy_digest: receipt.authorization_policy_digest, - revalidated_at: receipt.authority_revalidated_at, - }, - created_at: receipt.created_at, - effective_deadline_at: receipt.effective_deadline_at, - }) - } - .await; - match outcome { - Ok(receipt) => transaction - .commit() + transact_config( + self.db, + |_| ConfigurationError::Unavailable, + async move |transaction| { + if let Some(receipt) = replay_control_receipt( + &transaction, + &authority.receipt.actor_id, + &request.idempotency_key, + &plan.base_revision_id, + &request.operation_digest, + Some(&plan.plan_id), + ) + .await? + { + return Ok(receipt); + } + let current = current_state_from_transaction(&transaction).await?; + if current.revision_id != plan.base_revision_id { + return Err(ConfigurationError::PlanStale); + } + let record = read_change_plan_from_executor(&transaction, &plan.plan_id) + .await + .map_err(map_store_error)? + .ok_or(ConfigurationError::PlanStale)?; + if record.plan != plan { + return Err(ConfigurationError::PlanStale); + } + let ConfigurationProtectedOperationV1::Change(change) = &record.operation + else { + return Err(ConfigurationError::PlanStale); + }; + if record + .operation + .operation_digest() + .map_err(ConfigurationError::validation)? + != request.operation_digest + { + return Err(ConfigurationError::PlanStale); + } + let next_revision_id = result_revision_id( + &plan.base_revision_id, + &request.idempotency_key, + &request.operation_digest, + )?; + let snapshot = + protected_change_snapshot_v1(¤t.snapshot, change, &next_revision_id)?; + let sealed_target = + StoredConfigurationProtectedOperationV1::from(&record.operation); + let (commit, sealed_target_reference) = build_configuration_commit( + &transaction, + ConfigurationCommitDraft { + expected_base_revision_id: &plan.base_revision_id, + next_revision_id, + snapshot, + actor_id: &authority.receipt.actor_id, + operation_kind: "protected_apply", + operation_digest: request.operation_digest.clone(), + idempotency_key: request.idempotency_key.clone(), + authorization_policy_epoch: authority.receipt.policy_epoch, + authorization_policy_digest: authority.receipt.policy_digest.clone(), + authority_revalidated_at: authority.receipt.issued_at, + change_plan: Some(plan.clone()), + event_kind: ConfigurationAuditEventKindV1::Applied, + created_at: authority.receipt.issued_at, + effective_deadline_at: authority.receipt.expires_at, + target: &sealed_target, + }, + ) + .await?; + let receipt = commit_configuration_transaction( + &transaction, + &commit, + false, + Some(&sealed_target_reference), + ) .await - .map(|()| receipt) - .map_err(|_| ConfigurationError::Unavailable), - Err(error) => Err(error), - } + .map_err(map_store_error)?; + Ok(ConfigurationMutationReceipt { + receipt_id: receipt.receipt_id, + base_revision_id: receipt.base_revision_id, + result_revision_id: receipt.result_revision_id, + snapshot_id: commit.next_revision.snapshot.snapshot_id, + operation_digest: receipt.operation_digest, + settlement_authority: ConfigurationSettlementAuthorityV1 { + policy_epoch: receipt.authorization_policy_epoch, + policy_digest: receipt.authorization_policy_digest, + revalidated_at: receipt.authority_revalidated_at, + }, + created_at: receipt.created_at, + effective_deadline_at: receipt.effective_deadline_at, + }) + }, + ) + .await }) } @@ -318,91 +299,82 @@ impl ConfigurationControlStore for GlobalDbConfigurationControlStore<'_> { .target_revision_id .validate() .map_err(ConfigurationError::validation)?; - let transaction = self - .db - .begin_write_transaction() - .await - .map_err(|_| ConfigurationError::Unavailable)?; - let outcome = async { - let current = current_state_from_transaction(&transaction).await?; - if current.revision_id != authority.receipt.expected_configuration_revision { - return Err(ConfigurationError::RevisionConflict); - } - let target = - read_revision_from_executor(&transaction, &rollback.target_revision_id) + transact_config( + self.db, + |_| ConfigurationError::Unavailable, + async move |transaction| { + let current = current_state_from_transaction(&transaction).await?; + if current.revision_id != authority.receipt.expected_configuration_revision { + return Err(ConfigurationError::RevisionConflict); + } + let target = + read_revision_from_executor(&transaction, &rollback.target_revision_id) + .await + .map_err(map_store_error)? + .ok_or(ConfigurationError::PlanStale)?; + let operation = ConfigurationProtectedOperationV1::Rollback { + target_revision_id: rollback.target_revision_id.clone(), + mode: rollback.mode, + }; + let operation_digest = operation + .operation_digest() + .map_err(ConfigurationError::validation)?; + let plan_id = derived_identifier( + "configuration.plan.rollback.v1", + &canonical_sha256(&( + "tracedecay.configuration.rollback-plan.v1", + &authority.receipt.actor_id, + ¤t.revision_id, + &operation_digest, + &authority.receipt.scope_digest, + authority.receipt.policy_epoch, + &authority.receipt.policy_digest, + now, + )) + .map_err(ConfigurationError::validation)?, + "configuration rollback plan id", + )?; + let changes = rollback_redacted_changes(¤t.snapshot, &target.snapshot)?; + if changes.is_empty() { + return Err(ConfigurationError::PlanStale); + } + let plan = ProtectedChangePlan { + plan_id, + actor_id: authority.receipt.actor_id.clone(), + base_revision_id: current.revision_id, + operation_digest, + resolved_scope_digest: authority.receipt.scope_digest.clone(), + membership_digest: None, + authorization_policy_digest: authority.receipt.policy_digest.clone(), + policy_epoch: authority.receipt.policy_epoch, + expires_at: UtcMicros(now.0.saturating_add(300_000_000)), + created_at: now, + redacted_changes: changes, + }; + let record = ConfigurationProtectedPlanRecordV1 { + plan: plan.clone(), + operation, + }; + record.validate().map_err(ConfigurationError::validation)?; + match read_change_plan_from_executor(&transaction, &plan.plan_id) .await .map_err(map_store_error)? - .ok_or(ConfigurationError::PlanStale)?; - let operation = ConfigurationProtectedOperationV1::Rollback { - target_revision_id: rollback.target_revision_id.clone(), - mode: rollback.mode, - }; - let operation_digest = operation - .operation_digest() - .map_err(ConfigurationError::validation)?; - let plan_id = derived_identifier( - "configuration.plan.rollback.v1", - &canonical_sha256(&( - "tracedecay.configuration.rollback-plan.v1", - &authority.receipt.actor_id, - ¤t.revision_id, - &operation_digest, - &authority.receipt.scope_digest, - authority.receipt.policy_epoch, - &authority.receipt.policy_digest, - now, - )) - .map_err(ConfigurationError::validation)?, - "configuration rollback plan id", - )?; - let changes = rollback_redacted_changes(¤t.snapshot, &target.snapshot)?; - if changes.is_empty() { - return Err(ConfigurationError::PlanStale); - } - let plan = ProtectedChangePlan { - plan_id, - actor_id: authority.receipt.actor_id.clone(), - base_revision_id: current.revision_id, - operation_digest, - resolved_scope_digest: authority.receipt.scope_digest.clone(), - membership_digest: None, - authorization_policy_digest: authority.receipt.policy_digest.clone(), - policy_epoch: authority.receipt.policy_epoch, - expires_at: UtcMicros(now.0.saturating_add(300_000_000)), - created_at: now, - redacted_changes: changes, - }; - let record = ConfigurationProtectedPlanRecordV1 { - plan: plan.clone(), - operation, - }; - record.validate().map_err(ConfigurationError::validation)?; - match read_change_plan_from_executor(&transaction, &plan.plan_id) - .await - .map_err(map_store_error)? - { - Some(existing) if existing == record => Ok(plan), - Some(_) => Err(ConfigurationError::IdempotencyConflict), - None => { - insert_change_plan(&transaction, &record) - .await - .map_err(map_store_error)?; - insert_dry_run_audit_event(&transaction, &record) - .await - .map_err(map_store_error)?; - Ok(plan) + { + Some(existing) if existing == record => Ok(plan), + Some(_) => Err(ConfigurationError::IdempotencyConflict), + None => { + insert_change_plan(&transaction, &record) + .await + .map_err(map_store_error)?; + insert_dry_run_audit_event(&transaction, &record) + .await + .map_err(map_store_error)?; + Ok(plan) + } } - } - } - .await; - match outcome { - Ok(plan) => transaction - .commit() - .await - .map(|()| plan) - .map_err(|_| ConfigurationError::Unavailable), - Err(error) => Err(error), - } + }, + ) + .await }) } @@ -432,117 +404,108 @@ impl ConfigurationControlStore for GlobalDbConfigurationControlStore<'_> { { return Err(ConfigurationError::PlanStale); } - let transaction = self - .db - .begin_write_transaction() - .await - .map_err(|_| ConfigurationError::Unavailable)?; - let outcome = async { - if let Some(receipt) = replay_control_receipt( - &transaction, - &authority.receipt.actor_id, - &request.idempotency_key, - &plan.base_revision_id, - &request.operation_digest, - Some(&plan.plan_id), - ) - .await? - { - return Ok(receipt); - } - let current = current_state_from_transaction(&transaction).await?; - if current.revision_id != plan.base_revision_id { - return Err(ConfigurationError::PlanStale); - } - let record = read_change_plan_from_executor(&transaction, &plan.plan_id) - .await - .map_err(map_store_error)? - .ok_or(ConfigurationError::PlanStale)?; - if record.plan != plan { - return Err(ConfigurationError::PlanStale); - } - let ConfigurationProtectedOperationV1::Rollback { - target_revision_id, - mode, - } = &record.operation - else { - return Err(ConfigurationError::PlanStale); - }; - if *mode == RollbackModeV1::Partial { - return Err(ConfigurationError::Unavailable); - } - if record - .operation - .operation_digest() - .map_err(ConfigurationError::validation)? - != request.operation_digest - { - return Err(ConfigurationError::PlanStale); - } - let target = read_revision_from_executor(&transaction, target_revision_id) - .await - .map_err(map_store_error)? - .ok_or(ConfigurationError::PlanStale)?; - let next_revision_id = result_revision_id( - &plan.base_revision_id, - &request.idempotency_key, - &request.operation_digest, - )?; - let sealed_target = - StoredConfigurationProtectedOperationV1::from(&record.operation); - let (commit, sealed_target_reference) = build_configuration_commit( - &transaction, - ConfigurationCommitDraft { - expected_base_revision_id: &plan.base_revision_id, - next_revision_id, - snapshot: target.snapshot, - actor_id: &authority.receipt.actor_id, - operation_kind: "rollback_apply", - operation_digest: request.operation_digest.clone(), - idempotency_key: request.idempotency_key.clone(), - authorization_policy_epoch: authority.receipt.policy_epoch, - authorization_policy_digest: authority.receipt.policy_digest.clone(), - authority_revalidated_at: authority.receipt.issued_at, - change_plan: Some(plan.clone()), - event_kind: ConfigurationAuditEventKindV1::RollbackApplied, - created_at: authority.receipt.issued_at, - effective_deadline_at: authority.receipt.expires_at, - target: &sealed_target, - }, - ) - .await?; - let receipt = commit_configuration_transaction( - &transaction, - &commit, - false, - Some(&sealed_target_reference), - ) - .await - .map_err(map_store_error)?; - Ok(ConfigurationMutationReceipt { - receipt_id: receipt.receipt_id, - base_revision_id: receipt.base_revision_id, - result_revision_id: receipt.result_revision_id, - snapshot_id: commit.next_revision.snapshot.snapshot_id, - operation_digest: receipt.operation_digest, - settlement_authority: ConfigurationSettlementAuthorityV1 { - policy_epoch: receipt.authorization_policy_epoch, - policy_digest: receipt.authorization_policy_digest, - revalidated_at: receipt.authority_revalidated_at, - }, - created_at: receipt.created_at, - effective_deadline_at: receipt.effective_deadline_at, - }) - } - .await; - match outcome { - Ok(receipt) => transaction - .commit() + transact_config( + self.db, + |_| ConfigurationError::Unavailable, + async move |transaction| { + if let Some(receipt) = replay_control_receipt( + &transaction, + &authority.receipt.actor_id, + &request.idempotency_key, + &plan.base_revision_id, + &request.operation_digest, + Some(&plan.plan_id), + ) + .await? + { + return Ok(receipt); + } + let current = current_state_from_transaction(&transaction).await?; + if current.revision_id != plan.base_revision_id { + return Err(ConfigurationError::PlanStale); + } + let record = read_change_plan_from_executor(&transaction, &plan.plan_id) + .await + .map_err(map_store_error)? + .ok_or(ConfigurationError::PlanStale)?; + if record.plan != plan { + return Err(ConfigurationError::PlanStale); + } + let ConfigurationProtectedOperationV1::Rollback { + target_revision_id, + mode, + } = &record.operation + else { + return Err(ConfigurationError::PlanStale); + }; + if *mode == RollbackModeV1::Partial { + return Err(ConfigurationError::Unavailable); + } + if record + .operation + .operation_digest() + .map_err(ConfigurationError::validation)? + != request.operation_digest + { + return Err(ConfigurationError::PlanStale); + } + let target = read_revision_from_executor(&transaction, target_revision_id) + .await + .map_err(map_store_error)? + .ok_or(ConfigurationError::PlanStale)?; + let next_revision_id = result_revision_id( + &plan.base_revision_id, + &request.idempotency_key, + &request.operation_digest, + )?; + let sealed_target = + StoredConfigurationProtectedOperationV1::from(&record.operation); + let (commit, sealed_target_reference) = build_configuration_commit( + &transaction, + ConfigurationCommitDraft { + expected_base_revision_id: &plan.base_revision_id, + next_revision_id, + snapshot: target.snapshot, + actor_id: &authority.receipt.actor_id, + operation_kind: "rollback_apply", + operation_digest: request.operation_digest.clone(), + idempotency_key: request.idempotency_key.clone(), + authorization_policy_epoch: authority.receipt.policy_epoch, + authorization_policy_digest: authority.receipt.policy_digest.clone(), + authority_revalidated_at: authority.receipt.issued_at, + change_plan: Some(plan.clone()), + event_kind: ConfigurationAuditEventKindV1::RollbackApplied, + created_at: authority.receipt.issued_at, + effective_deadline_at: authority.receipt.expires_at, + target: &sealed_target, + }, + ) + .await?; + let receipt = commit_configuration_transaction( + &transaction, + &commit, + false, + Some(&sealed_target_reference), + ) .await - .map(|()| receipt) - .map_err(|_| ConfigurationError::Unavailable), - Err(error) => Err(error), - } + .map_err(map_store_error)?; + Ok(ConfigurationMutationReceipt { + receipt_id: receipt.receipt_id, + base_revision_id: receipt.base_revision_id, + result_revision_id: receipt.result_revision_id, + snapshot_id: commit.next_revision.snapshot.snapshot_id, + operation_digest: receipt.operation_digest, + settlement_authority: ConfigurationSettlementAuthorityV1 { + policy_epoch: receipt.authorization_policy_epoch, + policy_digest: receipt.authorization_policy_digest, + revalidated_at: receipt.authority_revalidated_at, + }, + created_at: receipt.created_at, + effective_deadline_at: receipt.effective_deadline_at, + }) + }, + ) + .await }) } diff --git a/crates/tracedecay-global-db/src/configuration/store/revision.rs b/crates/tracedecay-global-db/src/configuration/store/revision.rs index ee47867d2b..2aa472396d 100644 --- a/crates/tracedecay-global-db/src/configuration/store/revision.rs +++ b/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 { 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 } diff --git a/crates/tracedecay-global-db/src/lib.rs b/crates/tracedecay-global-db/src/lib.rs index c19618b03d..ad09a790d4 100644 --- a/crates/tracedecay-global-db/src/lib.rs +++ b/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}; diff --git a/crates/tracedecay-global-db/src/observability_rollup.rs b/crates/tracedecay-global-db/src/observability_rollup.rs index 7fd7e185a0..06b8f0c0cc 100644 --- a/crates/tracedecay-global-db/src/observability_rollup.rs +++ b/crates/tracedecay-global-db/src/observability_rollup.rs @@ -11,7 +11,7 @@ use tracedecay_domain::CoverageStateV1; use tracedecay_domain::canonical_text::sha256_hex; use tracedecay_runtime_core::db::engine::{Executor, QueryExecutor}; -use crate::RegisteredGlobalDb; +use crate::{RegisteredGlobalDb, RegisteredGlobalDbWriteTx}; mod compaction; mod dirty; mod frontier; @@ -30,6 +30,35 @@ use types::{ PublishedGeneration, SECONDS_PER_DAY, validate_day, validate_identifier, }; +/// Runs `body` as one write job. A refused or failed job becomes +/// `failed to {context}: {cause}`; the body's own error is returned as is. +async fn transact_rollup( + db: &RegisteredGlobalDb, + context: &'static str, + body: impl AsyncFn(RegisteredGlobalDbWriteTx<'_, '_>) -> Result + Send + Sync + 'static, +) -> Result +where + R: Send + 'static, + E: From + Send + 'static, +{ + db.transact(body) + .await + .map_err(|error| E::from(format!("failed to {context}: {error}")))? +} + +/// A rebuild job that ends without publishing. A replayed receipt rolls the +/// job back, like any other early exit, but is returned as success. +enum RebuildAbort { + Replayed(ObservabilityRollupRebuildReceiptV1), + Failed(String), +} + +impl From for RebuildAbort { + fn from(message: String) -> Self { + Self::Failed(message) + } +} + impl RegisteredGlobalDb { #[tracing::instrument( name = "global_db.observability_rollup.persist.rebuild", @@ -53,145 +82,146 @@ impl RegisteredGlobalDb { coverage_name(request.coverage), content_digest.as_str(), ))?; - let transaction = self - .begin_write_transaction() - .await - .map_err(|error| format!("failed to begin observability rollup rebuild: {error}"))?; + let outcome = transact_rollup( + self, + "publish observability rollup", + async move |transaction| { + if let Some((stored_digest, receipt)) = read_journal_receipt( + &transaction, + &request.authorized_scope_ref, + request.day_start_seconds, + &request.idempotency_key, + ) + .await? + { + if stored_digest == request_digest { + return Err(RebuildAbort::Replayed(receipt)); + } + return Err("observability rollup idempotency conflict" + .to_owned() + .into()); + } - if let Some((stored_digest, receipt)) = read_journal_receipt( - &transaction, - &request.authorized_scope_ref, - request.day_start_seconds, - &request.idempotency_key, - ) - .await? - { - transaction - .rollback() - .await - .map_err(|error| format!("failed to close rollup replay transaction: {error}"))?; - if stored_digest == request_digest { - return Ok(receipt); - } - return Err("observability rollup idempotency conflict".to_owned()); - } + if let Some(claim) = &request.dirty_claim + && !dirty_claim_is_current(&transaction, claim).await? + { + return Err("observability rollup dirty-day claim was superseded" + .to_owned() + .into()); + } + if let Some(claim) = &request.empty_day_claim + && !empty_day_claim_is_current(&transaction, claim).await? + { + return Err("observability rollup empty-day claim was superseded" + .to_owned() + .into()); + } - if let Some(claim) = &request.dirty_claim - && !dirty_claim_is_current(&transaction, claim).await? - { - transaction.rollback().await.map_err(|error| { - format!("failed to close superseded dirty-day rebuild: {error}") - })?; - return Err("observability rollup dirty-day claim was superseded".to_owned()); - } - if let Some(claim) = &request.empty_day_claim - && !empty_day_claim_is_current(&transaction, claim).await? - { - transaction.rollback().await.map_err(|error| { - format!("failed to close superseded empty-day rebuild: {error}") - })?; - return Err("observability rollup empty-day claim was superseded".to_owned()); - } - - let current = read_published_generation( - &transaction, - &request.authorized_scope_ref, - request.day_start_seconds, - ) - .await?; - if current - .as_ref() - .is_some_and(|value| request.source_watermark < value.source_watermark) - { - transaction - .rollback() - .await - .map_err(|error| format!("failed to close stale rollup transaction: {error}"))?; - return Err("observability rollup stale source watermark".to_owned()); - } - if let Some(current) = ¤t - && request.source_watermark == current.source_watermark - && request.projector_revision == current.projector_revision - && (request.coverage != current.coverage || content_digest != current.content_digest) - { - transaction.rollback().await.map_err(|error| { - format!("failed to close conflicting rollup transaction: {error}") - })?; - return Err("observability rollup changed at the same projector watermark".to_owned()); - } + let current = read_published_generation( + &transaction, + &request.authorized_scope_ref, + request.day_start_seconds, + ) + .await?; + if current + .as_ref() + .is_some_and(|value| request.source_watermark < value.source_watermark) + { + return Err("observability rollup stale source watermark" + .to_owned() + .into()); + } + if let Some(current) = ¤t + && request.source_watermark == current.source_watermark + && request.projector_revision == current.projector_revision + && (request.coverage != current.coverage + || content_digest != current.content_digest) + { + return Err( + "observability rollup changed at the same projector watermark" + .to_owned() + .into(), + ); + } - let unchanged = current.as_ref().is_some_and(|value| { - request.source_watermark == value.source_watermark - && request.projector_revision == value.projector_revision - && request.coverage == value.coverage - && content_digest == value.content_digest - }); - let generation = current.as_ref().map_or(1, |value| { - value.generation.saturating_add(u64::from(!unchanged)) - }); - let late_correction = current - .as_ref() - .is_some_and(|value| request.source_watermark > value.source_watermark); + let unchanged = current.as_ref().is_some_and(|value| { + request.source_watermark == value.source_watermark + && request.projector_revision == value.projector_revision + && request.coverage == value.coverage + && content_digest == value.content_digest + }); + let generation = current.as_ref().map_or(1, |value| { + value.generation.saturating_add(u64::from(!unchanged)) + }); + let late_correction = current + .as_ref() + .is_some_and(|value| request.source_watermark > value.source_watermark); - if !unchanged { - replace_published_fragment(&transaction, &request, generation, &content_digest).await?; - } - let receipt = ObservabilityRollupRebuildReceiptV1 { - authorized_scope_ref: request.authorized_scope_ref.clone(), - day_start_seconds: request.day_start_seconds, - generation, - projector_revision: request.projector_revision.clone(), - source_watermark: request.source_watermark, - coverage: request.coverage, - content_digest, - late_correction, - }; - insert_journal_receipt( - &transaction, - &request.idempotency_key, - &request_digest, - &receipt, - ) - .await?; - if let Some(claim) = &request.dirty_claim { - let changed = transaction - .execute( - "DELETE FROM observability_rollup_dirty_days - WHERE scope_ref = ?1 AND day_start_seconds = ?2 - AND source_watermark = ?3 AND claimant_id = ?4 - AND lease_until_seconds = ?5", - tracedecay_runtime_core::db::engine::params![ - claim.authorized_scope_ref.as_str(), - claim.day_start_seconds, - claim.source_watermark, - claim.claimant_id.as_str(), - claim.lease_until_seconds - ], + if !unchanged { + replace_published_fragment(&transaction, &request, generation, &content_digest) + .await?; + } + let receipt = ObservabilityRollupRebuildReceiptV1 { + authorized_scope_ref: request.authorized_scope_ref.clone(), + day_start_seconds: request.day_start_seconds, + generation, + projector_revision: request.projector_revision.clone(), + source_watermark: request.source_watermark, + coverage: request.coverage, + content_digest: content_digest.clone(), + late_correction, + }; + insert_journal_receipt( + &transaction, + &request.idempotency_key, + &request_digest, + &receipt, ) - .await - .map_err(|error| format!("failed to settle observability dirty day: {error}"))?; - if changed != 1 { - transaction.rollback().await.map_err(|error| { - format!("failed to close unsettled dirty-day rebuild: {error}") - })?; - return Err( - "observability rollup dirty-day claim changed during rebuild".to_owned(), - ); - } - } - if let Some(claim) = &request.empty_day_claim - && !settle_empty_day_claim(&transaction, claim).await? - { - transaction.rollback().await.map_err(|error| { - format!("failed to close unadvanced empty-day rebuild: {error}") - })?; - return Err("observability rollup empty-day claim changed during rebuild".to_owned()); + .await?; + if let Some(claim) = &request.dirty_claim { + let changed = transaction + .execute( + "DELETE FROM observability_rollup_dirty_days + WHERE scope_ref = ?1 AND day_start_seconds = ?2 + AND source_watermark = ?3 AND claimant_id = ?4 + AND lease_until_seconds = ?5", + tracedecay_runtime_core::db::engine::params![ + claim.authorized_scope_ref.as_str(), + claim.day_start_seconds, + claim.source_watermark, + claim.claimant_id.as_str(), + claim.lease_until_seconds + ], + ) + .await + .map_err(|error| { + format!("failed to settle observability dirty day: {error}") + })?; + if changed != 1 { + return Err( + "observability rollup dirty-day claim changed during rebuild" + .to_owned() + .into(), + ); + } + } + if let Some(claim) = &request.empty_day_claim + && !settle_empty_day_claim(&transaction, claim).await? + { + return Err( + "observability rollup empty-day claim changed during rebuild" + .to_owned() + .into(), + ); + } + Ok(receipt) + }, + ) + .await; + match outcome { + Ok(receipt) | Err(RebuildAbort::Replayed(receipt)) => Ok(receipt), + Err(RebuildAbort::Failed(message)) => Err(message), } - transaction - .commit() - .await - .map_err(|error| format!("failed to publish observability rollup: {error}"))?; - Ok(receipt) } /// Reads only bounded, projector-owned merge fragments. This internal diff --git a/crates/tracedecay-global-db/src/observability_rollup/compaction.rs b/crates/tracedecay-global-db/src/observability_rollup/compaction.rs index abb945bd9c..d4da55c8e5 100644 --- a/crates/tracedecay-global-db/src/observability_rollup/compaction.rs +++ b/crates/tracedecay-global-db/src/observability_rollup/compaction.rs @@ -1,9 +1,11 @@ +use tracedecay_runtime_core::db::engine::Executor; + use crate::RegisteredGlobalDb; use super::{ ObservabilityRollupCompactionCandidateV1, ObservabilityRollupCompactionReceiptV1, - ObservabilityRollupCompactionV1, SECONDS_PER_DAY, digest_json, encode_u64, validate_day, - validate_fragment_json, validate_identifier, + ObservabilityRollupCompactionV1, SECONDS_PER_DAY, digest_json, encode_u64, transact_rollup, + validate_day, validate_fragment_json, validate_identifier, }; const DETAIL_RETENTION_SECONDS: i64 = 30 * SECONDS_PER_DAY; @@ -105,13 +107,10 @@ impl RegisteredGlobalDb { } else { request.candidate.content_digest.clone() }; - let transaction = self - .begin_write_transaction() - .await - .map_err(|error| format!("failed to begin rollup fragment compaction: {error}"))?; - let changed = transaction - .execute( - "UPDATE observability_rollup_generations + transact_rollup(self, "compact rollup fragment", async move |transaction| { + let changed = transaction + .execute( + "UPDATE observability_rollup_generations SET generation = ?7, content_digest = ?8, fragment_json = ?9, retention_checked_at_seconds = unixepoch(), published_at_seconds = unixepoch() WHERE scope_ref = ?1 AND day_start_seconds = ?2 @@ -119,37 +118,32 @@ impl RegisteredGlobalDb { AND source_watermark = ?5 AND content_digest = ?6 AND retention_checked_at_seconds IS NULL AND day_start_seconds + 86400 <= unixepoch() - ?10", - tracedecay_runtime_core::db::engine::params![ - request.candidate.authorized_scope_ref.as_str(), - request.candidate.day_start_seconds, - encode_u64(request.candidate.generation, "compaction generation")?, - request.candidate.projector_revision.as_str(), - request.candidate.source_watermark, - request.candidate.content_digest.as_str(), - encode_u64(generation, "compacted generation")?, - content_digest, - request.fragment_json.as_str(), - DETAIL_RETENTION_SECONDS - ], - ) - .await - .map_err(|error| format!("failed to CAS compacted rollup fragment: {error}"))?; - if changed != 1 { - transaction.rollback().await.map_err(|error| { - format!("failed to close stale rollup compaction transaction: {error}") - })?; - return Err("observability rollup compaction candidate changed".to_owned()); - } - transaction - .commit() - .await - .map_err(|error| format!("failed to commit rollup fragment compaction: {error}"))?; - Ok(ObservabilityRollupCompactionReceiptV1 { - day_start_seconds: request.candidate.day_start_seconds, - previous_generation: request.candidate.generation, - generation, - changed: changed_fragment, + tracedecay_runtime_core::db::engine::params![ + request.candidate.authorized_scope_ref.as_str(), + request.candidate.day_start_seconds, + encode_u64(request.candidate.generation, "compaction generation")?, + request.candidate.projector_revision.as_str(), + request.candidate.source_watermark, + request.candidate.content_digest.as_str(), + encode_u64(generation, "compacted generation")?, + content_digest.as_str(), + request.fragment_json.as_str(), + DETAIL_RETENTION_SECONDS + ], + ) + .await + .map_err(|error| format!("failed to CAS compacted rollup fragment: {error}"))?; + if changed != 1 { + return Err("observability rollup compaction candidate changed".to_owned()); + } + Ok(ObservabilityRollupCompactionReceiptV1 { + day_start_seconds: request.candidate.day_start_seconds, + previous_generation: request.candidate.generation, + generation, + changed: changed_fragment, + }) }) + .await } } diff --git a/crates/tracedecay-global-db/src/observability_rollup/dirty.rs b/crates/tracedecay-global-db/src/observability_rollup/dirty.rs index 8c293e55f3..99eaac445d 100644 --- a/crates/tracedecay-global-db/src/observability_rollup/dirty.rs +++ b/crates/tracedecay-global-db/src/observability_rollup/dirty.rs @@ -1,8 +1,10 @@ -use tracedecay_runtime_core::db::engine::QueryExecutor; +use tracedecay_runtime_core::db::engine::{Executor, QueryExecutor}; use crate::RegisteredGlobalDb; -use super::{ObservabilityRollupDirtyDayClaimV1, validate_day, validate_identifier}; +use super::{ + ObservabilityRollupDirtyDayClaimV1, transact_rollup, validate_day, validate_identifier, +}; impl RegisteredGlobalDb { /// Leases the oldest dirty execution-topology day for one exact scope. @@ -24,13 +26,15 @@ impl RegisteredGlobalDb { if lease_seconds == 0 || lease_seconds > 300 { return Err("invalid observability rollup dirty-day lease".to_owned()); } - let transaction = self - .begin_write_transaction() - .await - .map_err(|error| format!("failed to begin observability dirty-day claim: {error}"))?; - let mut rows = transaction - .query( - "UPDATE observability_rollup_dirty_days + let scope = authorized_scope_ref.to_owned(); + let claimant = claimant_id.to_owned(); + transact_rollup( + self, + "claim observability dirty day", + async move |transaction| { + let mut rows = transaction + .query( + "UPDATE observability_rollup_dirty_days SET claimant_id = ?2, lease_until_seconds = unixepoch() + ?3 WHERE rowid = ( SELECT rowid FROM observability_rollup_dirty_days @@ -41,44 +45,41 @@ impl RegisteredGlobalDb { ) RETURNING scope_ref, day_start_seconds, source_watermark, claimant_id, lease_until_seconds", - tracedecay_runtime_core::db::engine::params![ - authorized_scope_ref, - claimant_id, - i64::from(lease_seconds) - ], - ) - .await - .map_err(|error| format!("failed to claim observability dirty day: {error}"))?; - let claim = match rows - .next() - .await - .map_err(|error| format!("failed to read observability dirty-day claim: {error}"))? - { - Some(row) => Some(ObservabilityRollupDirtyDayClaimV1 { - authorized_scope_ref: row - .get(0) - .map_err(|error| format!("failed to decode dirty-day scope: {error}"))?, - day_start_seconds: row - .get(1) - .map_err(|error| format!("failed to decode dirty-day bucket: {error}"))?, - source_watermark: row - .get(2) - .map_err(|error| format!("failed to decode dirty-day watermark: {error}"))?, - claimant_id: row - .get(3) - .map_err(|error| format!("failed to decode dirty-day claimant: {error}"))?, - lease_until_seconds: row - .get(4) - .map_err(|error| format!("failed to decode dirty-day lease: {error}"))?, - }), - None => None, - }; - drop(rows); - transaction - .commit() - .await - .map_err(|error| format!("failed to commit observability dirty-day claim: {error}"))?; - Ok(claim) + tracedecay_runtime_core::db::engine::params![ + scope.as_str(), + claimant.as_str(), + i64::from(lease_seconds) + ], + ) + .await + .map_err(|error| format!("failed to claim observability dirty day: {error}"))?; + let claim = match rows.next().await.map_err(|error| { + format!("failed to read observability dirty-day claim: {error}") + })? { + Some(row) => Some(ObservabilityRollupDirtyDayClaimV1 { + authorized_scope_ref: row.get(0).map_err(|error| { + format!("failed to decode dirty-day scope: {error}") + })?, + day_start_seconds: row.get(1).map_err(|error| { + format!("failed to decode dirty-day bucket: {error}") + })?, + source_watermark: row.get(2).map_err(|error| { + format!("failed to decode dirty-day watermark: {error}") + })?, + claimant_id: row.get(3).map_err(|error| { + format!("failed to decode dirty-day claimant: {error}") + })?, + lease_until_seconds: row.get(4).map_err(|error| { + format!("failed to decode dirty-day lease: {error}") + })?, + }), + None => None, + }; + drop(rows); + Ok(claim) + }, + ) + .await } /// Releases one exact lease after a bounded rebuild attempt could not @@ -93,31 +94,34 @@ impl RegisteredGlobalDb { claim: &ObservabilityRollupDirtyDayClaimV1, ) -> Result { validate_dirty_claim(claim)?; - let transaction = self - .begin_write_transaction() - .await - .map_err(|error| format!("failed to begin observability dirty-day release: {error}"))?; - let changed = transaction - .execute( - "UPDATE observability_rollup_dirty_days + let claim = claim.clone(); + transact_rollup( + self, + "release observability dirty day", + async move |transaction| { + let changed = transaction + .execute( + "UPDATE observability_rollup_dirty_days SET claimant_id = NULL, lease_until_seconds = NULL WHERE scope_ref = ?1 AND day_start_seconds = ?2 AND source_watermark = ?3 AND claimant_id = ?4 AND lease_until_seconds = ?5", - tracedecay_runtime_core::db::engine::params![ - claim.authorized_scope_ref.as_str(), - claim.day_start_seconds, - claim.source_watermark, - claim.claimant_id.as_str(), - claim.lease_until_seconds - ], - ) - .await - .map_err(|error| format!("failed to release observability dirty day: {error}"))?; - transaction.commit().await.map_err(|error| { - format!("failed to commit observability dirty-day release: {error}") - })?; - Ok(changed == 1) + tracedecay_runtime_core::db::engine::params![ + claim.authorized_scope_ref.as_str(), + claim.day_start_seconds, + claim.source_watermark, + claim.claimant_id.as_str(), + claim.lease_until_seconds + ], + ) + .await + .map_err(|error| { + format!("failed to release observability dirty day: {error}") + })?; + Ok(changed == 1) + }, + ) + .await } } diff --git a/crates/tracedecay-global-db/src/observability_rollup/frontier.rs b/crates/tracedecay-global-db/src/observability_rollup/frontier.rs index 2c53d15b3b..41a07f7e56 100644 --- a/crates/tracedecay-global-db/src/observability_rollup/frontier.rs +++ b/crates/tracedecay-global-db/src/observability_rollup/frontier.rs @@ -4,7 +4,8 @@ use crate::RegisteredGlobalDb; use super::{ ObservabilityRollupEmptyDayClaimOutcomeV1, ObservabilityRollupEmptyDayClaimV1, - ObservabilityRollupFrontierV1, SECONDS_PER_DAY, validate_day, validate_identifier, + ObservabilityRollupFrontierV1, SECONDS_PER_DAY, transact_rollup, validate_day, + validate_identifier, }; impl RegisteredGlobalDb { @@ -20,48 +21,55 @@ impl RegisteredGlobalDb { authorized_scope_ref: &str, ) -> Result { validate_identifier("scope", authorized_scope_ref)?; - let transaction = self.begin_write_transaction().await.map_err(|error| { - format!("failed to begin empty-day frontier initialization: {error}") - })?; - transaction - .execute( - "INSERT OR IGNORE INTO observability_rollup_frontiers + let authorized_scope_ref = authorized_scope_ref.to_owned(); + transact_rollup( + self, + "initialize empty-day frontier", + async move |transaction| { + let authorized_scope_ref = authorized_scope_ref.as_str(); + transaction + .execute( + "INSERT OR IGNORE INTO observability_rollup_frontiers (scope_ref, coverage_start_day_seconds, next_day_start_seconds) VALUES (?1, ((unixepoch() / 86400) + 1) * 86400, ((unixepoch() / 86400) + 1) * 86400)", - tracedecay_runtime_core::db::engine::params![authorized_scope_ref], - ) - .await - .map_err(|error| format!("failed to initialize empty-day frontier: {error}"))?; - let mut rows = transaction - .query( - "SELECT coverage_start_day_seconds, next_day_start_seconds + tracedecay_runtime_core::db::engine::params![authorized_scope_ref], + ) + .await + .map_err(|error| format!("failed to initialize empty-day frontier: {error}"))?; + let mut rows = transaction + .query( + "SELECT coverage_start_day_seconds, next_day_start_seconds FROM observability_rollup_frontiers WHERE scope_ref = ?1", - tracedecay_runtime_core::db::engine::params![authorized_scope_ref], - ) - .await - .map_err(|error| { - format!("failed to inspect initialized empty-day frontier: {error}") - })?; - let row = rows - .next() - .await - .map_err(|error| format!("failed to read initialized empty-day frontier: {error}"))? - .ok_or_else(|| "empty-day frontier initialization returned no row".to_owned())?; - let frontier = ObservabilityRollupFrontierV1 { - coverage_start_day_seconds: row - .get(0) - .map_err(|error| format!("failed to decode coverage start day: {error}"))?, - next_day_start_seconds: row - .get(1) - .map_err(|error| format!("failed to decode next empty day: {error}"))?, - }; - drop(rows); - transaction.commit().await.map_err(|error| { - format!("failed to commit empty-day frontier initialization: {error}") - })?; - Ok(frontier) + tracedecay_runtime_core::db::engine::params![authorized_scope_ref], + ) + .await + .map_err(|error| { + format!("failed to inspect initialized empty-day frontier: {error}") + })?; + let row = rows + .next() + .await + .map_err(|error| { + format!("failed to read initialized empty-day frontier: {error}") + })? + .ok_or_else(|| { + "empty-day frontier initialization returned no row".to_owned() + })?; + let frontier = ObservabilityRollupFrontierV1 { + coverage_start_day_seconds: row + .get(0) + .map_err(|error| format!("failed to decode coverage start day: {error}"))?, + next_day_start_seconds: row + .get(1) + .map_err(|error| format!("failed to decode next empty day: {error}"))?, + }; + drop(rows); + Ok(frontier) + }, + ) + .await } /// Claims at most one completed quiet UTC day at or after this scope's @@ -79,148 +87,136 @@ impl RegisteredGlobalDb { if lease_seconds == 0 || lease_seconds > 300 { return Err("invalid observability rollup empty-day lease".to_owned()); } - let transaction = self - .begin_write_transaction() - .await - .map_err(|error| format!("failed to begin empty-day frontier claim: {error}"))?; - transaction - .execute( - "INSERT OR IGNORE INTO observability_rollup_frontiers + let authorized_scope_ref = authorized_scope_ref.to_owned(); + let claimant_id = claimant_id.to_owned(); + transact_rollup(self, "claim empty-day frontier", async move |transaction| { + let authorized_scope_ref = authorized_scope_ref.as_str(); + let claimant_id = claimant_id.as_str(); + transaction + .execute( + "INSERT OR IGNORE INTO observability_rollup_frontiers (scope_ref, coverage_start_day_seconds, next_day_start_seconds) VALUES (?1, ((unixepoch() / 86400) + 1) * 86400, ((unixepoch() / 86400) + 1) * 86400)", - tracedecay_runtime_core::db::engine::params![authorized_scope_ref], - ) - .await - .map_err(|error| format!("failed to initialize empty-day frontier: {error}"))?; - let mut rows = transaction - .query( - "SELECT coverage_start_day_seconds, next_day_start_seconds, + tracedecay_runtime_core::db::engine::params![authorized_scope_ref], + ) + .await + .map_err(|error| format!("failed to initialize empty-day frontier: {error}"))?; + let mut rows = transaction + .query( + "SELECT coverage_start_day_seconds, next_day_start_seconds, claimant_id, lease_until_seconds, (unixepoch() / 86400) * 86400, unixepoch() FROM observability_rollup_frontiers WHERE scope_ref = ?1", - tracedecay_runtime_core::db::engine::params![authorized_scope_ref], - ) - .await - .map_err(|error| format!("failed to inspect empty-day frontier: {error}"))?; - let row = rows - .next() - .await - .map_err(|error| format!("failed to read empty-day frontier: {error}"))? - .ok_or_else(|| "empty-day frontier initialization returned no row".to_owned())?; - let coverage_start_day_seconds = row - .get::(0) - .map_err(|error| format!("failed to decode coverage start day: {error}"))?; - let next_day_start_seconds = row - .get::(1) - .map_err(|error| format!("failed to decode next empty day: {error}"))?; - let active_claimant = row - .get::>(2) - .map_err(|error| format!("failed to decode empty-day claimant: {error}"))?; - let lease_until_seconds = row - .get::>(3) - .map_err(|error| format!("failed to decode empty-day lease: {error}"))?; - let current_day_start_seconds = row - .get::(4) - .map_err(|error| format!("failed to decode current UTC day: {error}"))?; - let now_seconds = row - .get::(5) - .map_err(|error| format!("failed to decode current time: {error}"))?; - drop(rows); - - if next_day_start_seconds >= current_day_start_seconds { - transaction.commit().await.map_err(|error| { - format!("failed to close not-ready empty-day frontier: {error}") - })?; - return Ok(ObservabilityRollupEmptyDayClaimOutcomeV1::NotReady { - coverage_start_day_seconds, - next_day_start_seconds, - }); - } - if active_claimant.is_some() && lease_until_seconds.is_some_and(|lease| lease > now_seconds) - { - transaction - .commit() + tracedecay_runtime_core::db::engine::params![authorized_scope_ref], + ) .await - .map_err(|error| format!("failed to close leased empty-day frontier: {error}"))?; - return Ok(ObservabilityRollupEmptyDayClaimOutcomeV1::Leased { - day_start_seconds: next_day_start_seconds, - }); - } - if day_is_dirty(&transaction, authorized_scope_ref, next_day_start_seconds).await? { - transaction - .commit() + .map_err(|error| format!("failed to inspect empty-day frontier: {error}"))?; + let row = rows + .next() .await - .map_err(|error| format!("failed to close dirty empty-day frontier: {error}"))?; - return Ok(ObservabilityRollupEmptyDayClaimOutcomeV1::DirtyDay { - day_start_seconds: next_day_start_seconds, - }); - } - if day_has_generation(&transaction, authorized_scope_ref, next_day_start_seconds).await? { - let following_day = next_day_start_seconds - .checked_add(SECONDS_PER_DAY) - .ok_or_else(|| "empty-day frontier overflow".to_owned())?; - transaction - .execute( - "UPDATE observability_rollup_frontiers + .map_err(|error| format!("failed to read empty-day frontier: {error}"))? + .ok_or_else(|| "empty-day frontier initialization returned no row".to_owned())?; + let coverage_start_day_seconds = row + .get::(0) + .map_err(|error| format!("failed to decode coverage start day: {error}"))?; + let next_day_start_seconds = row + .get::(1) + .map_err(|error| format!("failed to decode next empty day: {error}"))?; + let active_claimant = row + .get::>(2) + .map_err(|error| format!("failed to decode empty-day claimant: {error}"))?; + let lease_until_seconds = row + .get::>(3) + .map_err(|error| format!("failed to decode empty-day lease: {error}"))?; + let current_day_start_seconds = row + .get::(4) + .map_err(|error| format!("failed to decode current UTC day: {error}"))?; + let now_seconds = row + .get::(5) + .map_err(|error| format!("failed to decode current time: {error}"))?; + drop(rows); + + if next_day_start_seconds >= current_day_start_seconds { + return Ok(ObservabilityRollupEmptyDayClaimOutcomeV1::NotReady { + coverage_start_day_seconds, + next_day_start_seconds, + }); + } + if active_claimant.is_some() + && lease_until_seconds.is_some_and(|lease| lease > now_seconds) + { + return Ok(ObservabilityRollupEmptyDayClaimOutcomeV1::Leased { + day_start_seconds: next_day_start_seconds, + }); + } + if day_is_dirty(&transaction, authorized_scope_ref, next_day_start_seconds).await? { + return Ok(ObservabilityRollupEmptyDayClaimOutcomeV1::DirtyDay { + day_start_seconds: next_day_start_seconds, + }); + } + if day_has_generation(&transaction, authorized_scope_ref, next_day_start_seconds) + .await? + { + let following_day = next_day_start_seconds + .checked_add(SECONDS_PER_DAY) + .ok_or_else(|| "empty-day frontier overflow".to_owned())?; + transaction + .execute( + "UPDATE observability_rollup_frontiers SET next_day_start_seconds = ?2, claimant_id = NULL, lease_until_seconds = NULL WHERE scope_ref = ?1 AND next_day_start_seconds = ?3", + tracedecay_runtime_core::db::engine::params![ + authorized_scope_ref, + following_day, + next_day_start_seconds + ], + ) + .await + .map_err(|error| format!("failed to advance existing-day frontier: {error}"))?; + return Ok( + ObservabilityRollupEmptyDayClaimOutcomeV1::AdvancedExisting { + day_start_seconds: next_day_start_seconds, + }, + ); + } + + let mut claimed = transaction + .query( + "UPDATE observability_rollup_frontiers + SET claimant_id = ?2, lease_until_seconds = unixepoch() + ?3 + WHERE scope_ref = ?1 AND next_day_start_seconds = ?4 + AND (claimant_id IS NULL OR lease_until_seconds <= unixepoch()) + RETURNING lease_until_seconds", tracedecay_runtime_core::db::engine::params![ authorized_scope_ref, - following_day, + claimant_id, + i64::from(lease_seconds), next_day_start_seconds ], ) .await - .map_err(|error| format!("failed to advance existing-day frontier: {error}"))?; - transaction.commit().await.map_err(|error| { - format!("failed to commit existing-day frontier advance: {error}") - })?; - return Ok( - ObservabilityRollupEmptyDayClaimOutcomeV1::AdvancedExisting { + .map_err(|error| format!("failed to claim empty-day frontier: {error}"))?; + let lease_until_seconds = claimed + .next() + .await + .map_err(|error| format!("failed to read empty-day claim: {error}"))? + .ok_or_else(|| "empty-day frontier claim changed concurrently".to_owned())? + .get::(0) + .map_err(|error| format!("failed to decode claimed empty-day lease: {error}"))?; + drop(claimed); + Ok(ObservabilityRollupEmptyDayClaimOutcomeV1::Claimed( + ObservabilityRollupEmptyDayClaimV1 { + authorized_scope_ref: authorized_scope_ref.to_owned(), day_start_seconds: next_day_start_seconds, + claimant_id: claimant_id.to_owned(), + lease_until_seconds, }, - ); - } - - let mut claimed = transaction - .query( - "UPDATE observability_rollup_frontiers - SET claimant_id = ?2, lease_until_seconds = unixepoch() + ?3 - WHERE scope_ref = ?1 AND next_day_start_seconds = ?4 - AND (claimant_id IS NULL OR lease_until_seconds <= unixepoch()) - RETURNING lease_until_seconds", - tracedecay_runtime_core::db::engine::params![ - authorized_scope_ref, - claimant_id, - i64::from(lease_seconds), - next_day_start_seconds - ], - ) - .await - .map_err(|error| format!("failed to claim empty-day frontier: {error}"))?; - let lease_until_seconds = claimed - .next() - .await - .map_err(|error| format!("failed to read empty-day claim: {error}"))? - .ok_or_else(|| "empty-day frontier claim changed concurrently".to_owned())? - .get::(0) - .map_err(|error| format!("failed to decode claimed empty-day lease: {error}"))?; - drop(claimed); - transaction - .commit() - .await - .map_err(|error| format!("failed to commit empty-day claim: {error}"))?; - Ok(ObservabilityRollupEmptyDayClaimOutcomeV1::Claimed( - ObservabilityRollupEmptyDayClaimV1 { - authorized_scope_ref: authorized_scope_ref.to_owned(), - day_start_seconds: next_day_start_seconds, - claimant_id: claimant_id.to_owned(), - lease_until_seconds, - }, - )) + )) + }) + .await } pub async fn release_observability_rollup_empty_day( @@ -228,30 +224,26 @@ impl RegisteredGlobalDb { claim: &ObservabilityRollupEmptyDayClaimV1, ) -> Result { validate_empty_day_claim(claim)?; - let transaction = self - .begin_write_transaction() - .await - .map_err(|error| format!("failed to begin empty-day release: {error}"))?; - let changed = transaction - .execute( - "UPDATE observability_rollup_frontiers + let claim = claim.clone(); + transact_rollup(self, "release empty-day claim", async move |transaction| { + let changed = transaction + .execute( + "UPDATE observability_rollup_frontiers SET claimant_id = NULL, lease_until_seconds = NULL WHERE scope_ref = ?1 AND next_day_start_seconds = ?2 AND claimant_id = ?3 AND lease_until_seconds = ?4", - tracedecay_runtime_core::db::engine::params![ - claim.authorized_scope_ref.as_str(), - claim.day_start_seconds, - claim.claimant_id.as_str(), - claim.lease_until_seconds - ], - ) - .await - .map_err(|error| format!("failed to release empty-day claim: {error}"))?; - transaction - .commit() - .await - .map_err(|error| format!("failed to commit empty-day release: {error}"))?; - Ok(changed == 1) + tracedecay_runtime_core::db::engine::params![ + claim.authorized_scope_ref.as_str(), + claim.day_start_seconds, + claim.claimant_id.as_str(), + claim.lease_until_seconds + ], + ) + .await + .map_err(|error| format!("failed to release empty-day claim: {error}"))?; + Ok(changed == 1) + }) + .await } } diff --git a/crates/tracedecay-global-db/src/registered.rs b/crates/tracedecay-global-db/src/registered.rs index 1dc8e8c6ee..e3e52c44f6 100644 --- a/crates/tracedecay-global-db/src/registered.rs +++ b/crates/tracedecay-global-db/src/registered.rs @@ -13,7 +13,7 @@ use tracedecay_runtime_core::{ DatabaseOwnerErrorV1, DatabaseOwnerRetirementReservationV1, DatabaseOwnerV1, DatabaseOwnerWeakLeaseIssuerErrorV1, DatabaseOwnerWeakLeaseIssuerV1, DatabaseRuntimeClientV1, DatabaseStorageTelemetryHandle, DatabaseWriteTransaction, - engine::{Executor, IntoParams, QueryExecutor, Result as EngineResult, Rows}, + engine::{Executor, IntoParams, QueryExecutor, Result as EngineResult, Rows, WriteTx}, }, shard_runtime::{VerifiedGraphRuntimePortV1, VerifiedGraphRuntimeWeakProxyV1}, }; @@ -599,6 +599,41 @@ impl RegisteredGlobalDb { }) } + pub(crate) fn require_active_write_scope( + &self, + operation: &str, + ) -> tracedecay_domain::errors::Result<()> { + self.database + .write_authority()? + .require_active_write_scope(operation) + } + + /// Runs `body` as one write job: commit when it returns `Ok`, rollback + /// when it returns `Err`. The daemon write scope is checked before the job + /// is queued. The runtime verifies the same authority again at COMMIT, so a + /// scope that ends while the body runs rolls the job back and surfaces as + /// the outer error. + #[tracing::instrument(name = "global_db.registered.txn.job", level = "trace", skip_all)] + pub async fn transact( + &self, + body: F, + ) -> tracedecay_domain::errors::Result> + where + F: AsyncFn(RegisteredGlobalDbWriteTx<'_, '_>) -> Result + Send + Sync + 'static, + R: Send + 'static, + E: Send + 'static, + { + self.require_active_write_scope("begin registered global database transaction")?; + self.database + .transact( + "begin registered global database transaction", + async move |transaction: &WriteTx<'_, '_>| { + body(RegisteredGlobalDbWriteTx { transaction }).await + }, + ) + .await + } + pub fn binding(&self) -> &tracedecay_store::StoreRuntimeBindingV1 { self.database.registered_binding() } @@ -920,6 +955,36 @@ impl RegisteredGlobalDbWriteTransaction<'_> { } } +/// 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. +#[derive(Clone, Copy)] +pub struct RegisteredGlobalDbWriteTx<'a, 'b> { + transaction: &'a WriteTx<'a, 'b>, +} + +impl QueryExecutor for RegisteredGlobalDbWriteTx<'_, '_> { + async fn query

(&self, sql: &str, params: P) -> EngineResult + where + P: IntoParams, + { + self.transaction.query(sql, params).await + } +} + +impl Executor for RegisteredGlobalDbWriteTx<'_, '_> { + async fn execute

(&self, sql: &str, params: P) -> EngineResult + where + P: IntoParams, + { + self.transaction.execute(sql, params).await + } + + async fn execute_batch(&self, sql: &str) -> EngineResult<()> { + self.transaction.execute_batch(sql).await + } +} + fn registered_error(operation: &str, error: impl std::fmt::Display) -> TraceDecayError { TraceDecayError::Database { operation: operation.to_owned(), diff --git a/crates/tracedecay-global-db/src/tests.rs b/crates/tracedecay-global-db/src/tests.rs index 37d68444b6..62ff88e522 100644 --- a/crates/tracedecay-global-db/src/tests.rs +++ b/crates/tracedecay-global-db/src/tests.rs @@ -1419,6 +1419,70 @@ async fn begun_registered_transaction_rolls_back_when_scope_drops_before_commit( ); } +#[asupersync::test] +async fn copied_transact_handle_cannot_write_after_scope_drop_inside_the_body() { + use tracedecay_runtime_core::db::engine::Executor as _; + let mut harness = RegisteredGlobalDbHarness::open("transact-copied-handle").await; + let db = harness.registered.clone(); + let scope = std::sync::Arc::new(std::sync::Mutex::new(harness.take_scope())); + + let outcome = db + .transact(async move |transaction| { + transaction + .execute( + "CREATE TABLE transact_written_before_revoke (value INTEGER)", + (), + ) + .await?; + let copy = transaction; + drop(scope.lock().unwrap().take()); + copy.execute( + "CREATE TABLE transact_written_after_revoke (value INTEGER)", + (), + ) + .await?; + Ok::<_, tracedecay_runtime_core::db::engine::Error>(()) + }) + .await + .expect("job transport"); + + let error = outcome.expect_err("copied handle wrote after the scope ended"); + assert!(error.to_string().contains("active daemon"), "{error}"); + assert!(!table_exists(&db, "transact_written_after_revoke").await); + assert!( + !table_exists(&db, "transact_written_before_revoke").await, + "refused body did not roll back" + ); +} + +#[asupersync::test] +async fn transact_rolls_back_when_scope_ends_before_commit() { + use tracedecay_runtime_core::db::engine::Executor as _; + let mut harness = RegisteredGlobalDbHarness::open("transact-commit-check").await; + let db = harness.registered.clone(); + let scope = std::sync::Arc::new(std::sync::Mutex::new(harness.take_scope())); + + let error = db + .transact(async move |transaction| { + transaction + .execute( + "CREATE TABLE transact_staged_before_commit (value INTEGER)", + (), + ) + .await?; + drop(scope.lock().unwrap().take()); + Ok::<_, tracedecay_runtime_core::db::engine::Error>(()) + }) + .await + .expect_err("transact committed after the scope ended"); + + assert!(error.to_string().contains("active daemon"), "{error}"); + assert!( + !table_exists(&db, "transact_staged_before_commit").await, + "failed commit did not roll back staged writes" + ); +} + #[asupersync::test] async fn begun_registered_transaction_can_roll_back_after_scope_drop() { let mut harness = RegisteredGlobalDbHarness::open("registered-rollback").await; diff --git a/crates/tracedecay-global-db/src/tests/harness.rs b/crates/tracedecay-global-db/src/tests/harness.rs index 1cf4e188c7..6b9b56c1ad 100644 --- a/crates/tracedecay-global-db/src/tests/harness.rs +++ b/crates/tracedecay-global-db/src/tests/harness.rs @@ -477,7 +477,12 @@ impl RegisteredGlobalDbHarness { #[cfg(test)] pub(crate) fn revoke(&mut self) { - drop(self._scope.take()); + drop(self.take_scope()); + } + + #[cfg(test)] + pub(crate) fn take_scope(&mut self) -> Option { + self._scope.take() } /// Remounts through daemon admission, which defers historical convergence diff --git a/docs/plans/storage/README.md b/docs/plans/storage/README.md index 2864a62da9..fff1f78f47 100644 --- a/docs/plans/storage/README.md +++ b/docs/plans/storage/README.md @@ -760,6 +760,19 @@ changes a file that holds memory. behind interactive writers instead of retrying `Busy`. `set_metadata`, the retrieval-anchor disposition and derivative writes, and `execute_write_engine` and `execute_write_batch` run as jobs. +- Step 5i-3, global-db configuration and rollup writes (2026-10-18, branch + `refactor/storage-global-db-config-jobs`). `RegisteredGlobalDb::transact` runs a + body as one writer job over `Database::transact`, handing it a + `RegisteredGlobalDbWriteTx` that implements the executor traits, so the existing + `&impl Executor` helpers run in it unchanged. It checks the daemon write scope + before queueing; the runtime verifies the same authority again at COMMIT, so a + scope that ends mid-body rolls the job back. The configuration store, control and + revision writes and the observability rollup rebuild, claim, release, frontier + and compaction writes now use it. The interactive transaction stays for the other + global-db sites, which move in the next slices. Bodies are `Fn`, so each site + 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. - #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, diff --git a/docs/plans/storage/decisions.tsv b/docs/plans/storage/decisions.tsv index 84be0ca024..e1a817f3aa 100644 --- a/docs/plans/storage/decisions.tsv +++ b/docs/plans/storage/decisions.tsv @@ -61,3 +61,8 @@ ts phase decision why evidence result 2026-10-17T05:10:00Z implement a synchronous store wait on a pool thread refuses with the existing unavailable errors in every build profile; work the pool refuses to spawn is adopted by the polling task a parked pool thread can hold the thread the awaited task needs; enough of them deadlock the pool storage_pool tests, a_pool_task_is_refused_snapshot_calls_instead_of_parked, a_refused_deferred_return_on_a_pool_thread_still_runs_to_completion (mutations killed) no debug-only guard 2026-10-18T01:30:00Z implement pool default is the host's available parallelism clamped to 6..=16 instead of a flat 6 at N=6 the tracedecay unit_test binary timed out cold opens in 3 of 5 runs and ran 24% slower than master (219 s vs 177 s avg); at N=16 3 of 3 passed at 164 s; at the new default 3 of 3 at 171 s /fast/tmp/storage-takeover/7a-unit3.log, 7a-unit3-16.log, 7a-unit3-def.log, master-unit3.log deviation from section 4 recorded in README 2026-10-18T01:30:00Z implement no foreground-class thread reservation in step 7a today's files mix classes (Work shares sessions.db with observation ingest), so a reservation by file has nothing to key on until step 8 README Implementation reconciliation step 7 deferred to the partition +2026-10-18T05:30:00Z implement RegisteredGlobalDb::transact checks the write scope once, before queueing, and leaves the commit-time check to the runtime the runtime verifies the write authority at COMMIT inside every job, so a second check in the body wrapper never decided an outcome mutation removing the wrapper's commit-time check left transact_rolls_back_when_scope_ends_before_commit and copied_transact_handle_cannot_write_after_scope_drop_inside_the_body green; command.rs verify_write_authority(Commit) check deleted; the test now proves the runtime check +2026-10-18T05:30:00Z implement RegisteredGlobalDbWriteTx is exported from tracedecay-global-db converted call sites and later slices name the job transaction in helper signatures Zack approved the export exported beside RegisteredGlobalDbWriteTransaction, which stays until the interactive API is deleted +2026-10-18T05:30:00Z implement the configuration mutation-scope probe is a direct require_active_write_scope check begin-and-rollback only proved the scope and made the caller wait for the writer OwnedGlobalDbConfigurationControlStore::require_active_mutation_scope; owned_global_control_adapter_rejects_writes_after_daemon_scope_ends probe no longer touches the writer +2026-10-18T05:30:00Z implement an idempotent observability rollup replay still ends its job by rolling back a committed job advances graph_transaction_generation and the old replay path rolled back without bumping RebuildAbort::Replayed carries the stored receipt out of the aborted job; rollup_rebuild_is_idempotent_version_monotone_and_late_correctable replay unchanged, no generation bump +2026-10-18T05:30:00Z implement a failed rollup or configuration job reports one error instead of separate begin, commit and rollback messages a job has a single failure point; the begin/commit/close split no longer exists transact_config and transact_rollup map the job error messages say failed to instead of failed to begin/commit