khive_runtime/secret_gate_finalizer/
entity_transaction.rs

1//! Entity finalization inside the caller's existing atomic unit.
2
3use std::sync::{Arc, Mutex};
4
5use khive_db::stores::{
6    entity::{entity_insert_if_absent_statement, entity_replace_if_unchanged_statement},
7    event::event_insert_statements,
8};
9use khive_storage::{Entity, Event, SqlAccess, SqlStatement, SqlValue, SqlWriter, StorageError};
10use khive_types::{EventKind, EventOutcome, SubstrateKind};
11
12use super::entity_admission::{observe, PreparedEntityExemption};
13use super::log_sink::{FinalizerAuditGap, LogSink, TracingLogSink};
14use super::outcome::{
15    ExemptionCommit, FailureClass, FailureDiagnostic, FinalizerOutcome, Substrate,
16};
17use crate::atomic_plan::{AffectedRowGuard, PlanStatement, PostCommitEffect};
18use crate::atomic_runner::AtomicOpFailure;
19use crate::entity_write::EntityWriteGuard;
20use crate::{RuntimeError, RuntimeResult};
21
22/// The existing conditional-write semantics retained by manifest admission.
23#[derive(Debug, Clone)]
24#[allow(clippy::large_enum_variant)] // Own the captured snapshot without an additional caller allocation.
25pub enum EntityCandidateMutation {
26    CreateIfAbsent,
27    ReplaceIfUnchanged {
28        expected: Entity,
29        expected_version: Option<i64>,
30    },
31}
32
33#[derive(Debug)]
34enum RecordedFailure {
35    Contention,
36    Finalization {
37        diagnostic: FailureDiagnostic,
38        source: Option<Box<StorageError>>,
39    },
40}
41
42/// Opaque, runtime-constructed record/stamp/audit plan. There is no public
43/// constructor accepting SQL or caller-supplied exemption state.
44#[derive(Debug)]
45pub struct EntityFinalizationPlan {
46    prepared: PreparedEntityExemption,
47    row: SqlStatement,
48    guard: Option<EntityWriteGuard>,
49    required: Vec<PlanStatement>,
50    success_audit: Vec<SqlStatement>,
51    post_commit: PostCommitEffect,
52    replacing: bool,
53    committed_version: i64,
54    failure: Arc<Mutex<Option<RecordedFailure>>>,
55}
56
57impl Clone for EntityFinalizationPlan {
58    fn clone(&self) -> Self {
59        let mut plan = self.for_settlement();
60        // A replay/concurrent invocation owns a separate failure record.
61        plan.failure = Arc::new(Mutex::new(None));
62        plan
63    }
64}
65
66impl PreparedEntityExemption {
67    pub(crate) fn into_plan(
68        self,
69        mutation: EntityCandidateMutation,
70        required: Vec<PlanStatement>,
71        post_commit: PostCommitEffect,
72    ) -> RuntimeResult<EntityFinalizationPlan> {
73        let (row, guard, replacing, committed_version) = match mutation {
74            EntityCandidateMutation::CreateIfAbsent => (
75                entity_insert_if_absent_statement(&self.entity),
76                None,
77                false,
78                1,
79            ),
80            EntityCandidateMutation::ReplaceIfUnchanged {
81                expected,
82                expected_version,
83            } => {
84                crate::entity_write::validate_expected_version(expected_version)?;
85                if expected.id != self.entity.id
86                    || expected.version != self.entity.version
87                    || expected.namespace != self.entity.namespace
88                    || expected.created_at != self.entity.created_at
89                {
90                    return Err(RuntimeError::InvalidInput(
91                        "entity finalization CAS snapshot mismatch".into(),
92                    ));
93                }
94                let committed_version = expected
95                    .version
96                    .checked_add(1)
97                    .ok_or_else(|| RuntimeError::InvalidInput("entity version exhausted".into()))?;
98                let mut row = entity_replace_if_unchanged_statement(
99                    &self.entity,
100                    expected.updated_at,
101                    expected.deleted_at,
102                );
103                // Reuse the canonical CAS builder, binding its captured target
104                // namespace without depending on its current parameter count.
105                row.sql
106                    .push_str(&format!(" AND namespace = ?{}", row.params.len() + 1));
107                row.params.push(SqlValue::Text(expected.namespace));
108                row.sql
109                    .push_str(&format!(" AND created_at = ?{}", row.params.len() + 1));
110                row.params.push(SqlValue::Integer(expected.created_at));
111                let guard = expected_version.map(|expected_version| EntityWriteGuard {
112                    id: expected.id,
113                    expected_version,
114                });
115                (row, guard, true, committed_version)
116            }
117        };
118        let event = self.attribution.stamp(
119            Event::new(
120                &self.entity.namespace,
121                self.entry.verb(),
122                EventKind::Audit,
123                SubstrateKind::Entity,
124                "",
125            )
126            .with_target(self.entity.id)
127            .with_payload(serde_json::json!({
128                "mechanism": "content-sha256-manifest-v1",
129                "digest_sha256": self.digest_sha256,
130                "field_scope": self.field_scope,
131                "manifest_id": self.snapshot.manifest_id(),
132                "canonical_verb": self.entry.verb(),
133                "namespace": self.entity.namespace,
134                "overridden_detector": self.overridden_detector,
135                "outcome": "exempted",
136                "record_id": self.entity.id,
137                "entry_point": self.entry.family(replacing),
138            })),
139        );
140        let mut event = event;
141        event.payload["actor"] = event.actor.clone().into();
142        let success_audit = event_insert_statements(&event).map_err(|error| {
143            RuntimeError::Internal(format!("finalizer event preparation: {error}"))
144        })?;
145        Ok(EntityFinalizationPlan {
146            prepared: self,
147            row,
148            guard,
149            required,
150            success_audit,
151            post_commit,
152            replacing,
153            committed_version,
154            failure: Arc::new(Mutex::new(None)),
155        })
156    }
157}
158
159impl EntityFinalizationPlan {
160    pub(crate) fn for_settlement(&self) -> Self {
161        Self {
162            prepared: self.prepared.clone(),
163            row: self.row.clone(),
164            guard: self.guard.clone(),
165            required: self.required.clone(),
166            success_audit: self.success_audit.clone(),
167            post_commit: self.post_commit.clone(),
168            replacing: self.replacing,
169            committed_version: self.committed_version,
170            failure: Arc::clone(&self.failure),
171        }
172    }
173
174    pub(crate) fn committed_entity(&self) -> Entity {
175        let mut entity = self.prepared.entity.clone();
176        entity.version = self.committed_version;
177        entity
178    }
179
180    pub(crate) async fn apply(
181        &self,
182        writer: &mut dyn SqlWriter,
183    ) -> Result<Option<PostCommitEffect>, AtomicOpFailure> {
184        if let Some(guard) = &self.guard {
185            if let Some(conflict) = guard
186                .check(writer)
187                .await
188                .map_err(|source| self.fail(FailureClass::RecordWrite, Some(source)))?
189            {
190                self.contention();
191                return Err(AtomicOpFailure::EntityConflict(conflict));
192            }
193        }
194        if super::faults::consume_record_write_fail(&self.prepared.entity.namespace) {
195            return Err(self.fail(FailureClass::RecordWrite, None));
196        }
197        let affected = writer
198            .execute(self.row.clone())
199            .await
200            .map_err(|source| self.fail(FailureClass::RecordWrite, Some(source)))?;
201        if affected != 1 {
202            self.contention();
203            return Err(AtomicOpFailure::GuardFailed {
204                statement_label: self.row.label.clone(),
205                expected: AffectedRowGuard::exactly(1),
206                observed: affected,
207            });
208        }
209        // The stamp is already in the single row DML. This logical checkpoint
210        // proves rollback without a second UPDATE or revision increment.
211        if super::faults::consume_stamp_fail(&self.prepared.entity.namespace) {
212            return Err(self.fail(FailureClass::Stamp, None));
213        }
214        for statement in &self.required {
215            let affected = writer
216                .execute(statement.statement.clone())
217                .await
218                .map_err(|source| self.fail(FailureClass::RecordWrite, Some(source)))?;
219            if statement
220                .guard
221                .is_some_and(|guard| !guard.holds_for(affected))
222            {
223                return Err(self.fail(FailureClass::RecordWrite, None));
224            }
225        }
226        if super::faults::consume_success_audit_fail(&self.prepared.entity.namespace) {
227            return Err(self.fail(FailureClass::SuccessAudit, None));
228        }
229        for statement in &self.success_audit {
230            let affected = writer
231                .execute(statement.clone())
232                .await
233                .map_err(|source| self.fail(FailureClass::SuccessAudit, Some(source)))?;
234            if affected != 1 {
235                return Err(self.fail(FailureClass::SuccessAudit, None));
236            }
237        }
238        Ok((self.post_commit != PostCommitEffect::None).then(|| self.post_commit.clone()))
239    }
240
241    fn contention(&self) {
242        *self
243            .failure
244            .lock()
245            .expect("entity finalizer failure lock poisoned") = Some(RecordedFailure::Contention);
246    }
247
248    fn fail(&self, class: FailureClass, source: Option<StorageError>) -> AtomicOpFailure {
249        let diagnostic = self.diagnostic(class);
250        *self
251            .failure
252            .lock()
253            .expect("entity finalizer failure lock poisoned") =
254            Some(RecordedFailure::Finalization {
255                diagnostic,
256                source: source.map(Box::new),
257            });
258        AtomicOpFailure::SqlError {
259            statement_label: Some("entity-manifest-finalization".into()),
260            message: format!("entity manifest finalization failed at {class:?}"),
261        }
262    }
263
264    fn diagnostic(&self, class: FailureClass) -> FailureDiagnostic {
265        FailureDiagnostic::new(
266            self.prepared.entity.id,
267            Substrate::Entity,
268            &self.prepared.entity.namespace,
269            self.prepared.entry.family(self.replacing),
270            class,
271        )
272    }
273
274    pub(crate) fn finish_commit(&self) {
275        observe(
276            &self.prepared.entity.namespace,
277            FinalizerOutcome::Exempted(ExemptionCommit {
278                record_id: self.prepared.entity.id,
279                substrate: Substrate::Entity,
280                entry_point: self.prepared.entry.family(self.replacing),
281                digest_sha256: self.prepared.digest_sha256.clone(),
282                field_scope: self.prepared.field_scope,
283                manifest_id: self.prepared.snapshot.manifest_id().into(),
284            }),
285        );
286    }
287
288    /// Called only after the outer owner confirms rollback, never from the
289    /// failed writer callback. Native causes survive the best-effort audit.
290    pub(crate) async fn finish_rollback(&self, access: &dyn SqlAccess) -> Option<StorageError> {
291        let recorded = self
292            .failure
293            .lock()
294            .expect("entity finalizer failure lock poisoned")
295            .take();
296        let (diagnostic, source) = match recorded {
297            Some(RecordedFailure::Contention) => return None,
298            Some(RecordedFailure::Finalization { diagnostic, source }) => (diagnostic, source),
299            None => (self.diagnostic(FailureClass::RecordWrite), None),
300        };
301        let outcome = match diagnostic.failure_class {
302            FailureClass::RecordWrite => FinalizerOutcome::RecordWriteFailed(diagnostic.clone()),
303            FailureClass::Stamp => FinalizerOutcome::StampFailed(diagnostic.clone()),
304            FailureClass::SuccessAudit => FinalizerOutcome::AuditFailed(diagnostic.clone()),
305        };
306        observe(&diagnostic.namespace, outcome);
307        let event = self.prepared.attribution.stamp(Event::new(
308            &diagnostic.namespace, self.prepared.entry.verb(), EventKind::Audit, SubstrateKind::Entity, "",
309        ).with_target(diagnostic.record_id).with_outcome(EventOutcome::Error).with_payload(serde_json::json!({
310            "mechanism": "content-sha256-manifest-v1", "outcome": match diagnostic.failure_class {
311                FailureClass::RecordWrite => "record-write-failed",
312                FailureClass::Stamp => "stamp-failed",
313                FailureClass::SuccessAudit => "audit-failed",
314            },
315            "diagnostic_id": diagnostic.diagnostic_id, "record_id": diagnostic.record_id,
316            "entry_point": diagnostic.entry_point, "failure_class": format!("{:?}", diagnostic.failure_class),
317        })));
318        let failed = if super::faults::consume_failure_audit_fail(&diagnostic.namespace) {
319            true
320        } else {
321            match event_insert_statements(&event) {
322                Err(_) => true,
323                Ok(statements) => access
324                    .atomic_unit(Box::new(move |writer| {
325                        Box::pin(async move {
326                            for statement in statements {
327                                if writer.execute(statement).await? != 1 {
328                                    return Err(StorageError::Internal(
329                                        "entity finalizer failure audit was not persisted".into(),
330                                    ));
331                                }
332                            }
333                            Ok(Box::new(()) as Box<dyn std::any::Any + Send>)
334                        })
335                    }))
336                    .await
337                    .is_err(),
338            }
339        };
340        if failed {
341            let gap = FinalizerAuditGap {
342                failure_class: diagnostic.failure_class,
343                diagnostic_id: diagnostic.diagnostic_id,
344                record_id: diagnostic.record_id,
345                substrate: diagnostic.substrate,
346                namespace: diagnostic.namespace,
347                entry_point: diagnostic.entry_point,
348            };
349            TracingLogSink.record_audit_gap(&gap);
350            #[cfg(test)]
351            super::entity_admission::fixture::record_gap(gap);
352        }
353        source.map(|source| *source)
354    }
355}
khive_runtime/secret_gate_finalizer/entity_transaction.rs

khive_runtime/secret_gate_finalizer/
entity_transaction.rs
