|
|
@@ -15,7 +15,7 @@
|
|
|
|
|
|
use std::sync::Arc;
|
|
|
|
|
|
-use kuatia_core::{Account, Envelope, ReservationId, envelope_id};
|
|
|
+use kuatia_core::{Account, Envelope, Receipt, ReservationId, envelope_id};
|
|
|
use kuatia_storage::error::StoreError;
|
|
|
use kuatia_storage::events::{LedgerEvent, LedgerEventKind};
|
|
|
|
|
|
@@ -38,7 +38,7 @@ pub(super) enum SagaPhase {
|
|
|
/// Write-ahead record for an in-flight commit (reserve → finalize). Persisted
|
|
|
/// before the saga mutates anything and removed once it reaches a terminal
|
|
|
/// state.
|
|
|
-#[derive(serde::Serialize, serde::Deserialize)]
|
|
|
+#[derive(Clone, serde::Serialize, serde::Deserialize)]
|
|
|
pub(super) struct PendingSaga {
|
|
|
pub(super) envelope: Envelope,
|
|
|
pub(super) reservation: ReservationId,
|
|
|
@@ -71,19 +71,6 @@ pub(super) enum PendingRecord {
|
|
|
}
|
|
|
|
|
|
impl PendingRecord {
|
|
|
- /// A commit write-ahead record at the given phase.
|
|
|
- pub(super) fn envelope(
|
|
|
- envelope: Envelope,
|
|
|
- reservation: ReservationId,
|
|
|
- phase: SagaPhase,
|
|
|
- ) -> Self {
|
|
|
- Self::Envelope(PendingSaga {
|
|
|
- envelope,
|
|
|
- reservation,
|
|
|
- phase,
|
|
|
- })
|
|
|
- }
|
|
|
-
|
|
|
/// An account-transition write-ahead record.
|
|
|
pub(super) fn transition(next: Account, event: LedgerEventKind) -> Self {
|
|
|
Self::Transition(PendingTransition { next, event })
|
|
|
@@ -129,11 +116,117 @@ impl PendingRecord {
|
|
|
Self::Transition(PendingTransition { next, event }) => {
|
|
|
complete_transition(ledger, saga_id, next, event).await
|
|
|
}
|
|
|
- Self::Envelope(PendingSaga {
|
|
|
- envelope,
|
|
|
- reservation,
|
|
|
- phase,
|
|
|
- }) => complete_envelope(ledger, saga_id, envelope, reservation, phase).await,
|
|
|
+ // The commit lifecycle (phase rules, delete-safety) lives on
|
|
|
+ // `PendingSaga`; recovery just hands the decoded record to it.
|
|
|
+ Self::Envelope(saga) => saga.complete(ledger).await,
|
|
|
+ }
|
|
|
+ }
|
|
|
+}
|
|
|
+
|
|
|
+impl PendingSaga {
|
|
|
+ /// A fresh write-ahead record for a new commit, at the pre-mutation phase.
|
|
|
+ pub(super) fn new(envelope: Envelope, reservation: ReservationId) -> Self {
|
|
|
+ Self {
|
|
|
+ envelope,
|
|
|
+ reservation,
|
|
|
+ phase: SagaPhase::Reserving,
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ /// The same record advanced to the point of no return, used for the finalize
|
|
|
+ /// bump just before the consumed postings are removed.
|
|
|
+ pub(super) fn finalizing(envelope: Envelope, reservation: ReservationId) -> Self {
|
|
|
+ Self {
|
|
|
+ envelope,
|
|
|
+ reservation,
|
|
|
+ phase: SagaPhase::Finalizing,
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ /// An envelope saga is always keyed by its reservation id, so the live commit
|
|
|
+ /// path and recovery agree on where the record lives.
|
|
|
+ fn saga_id(&self) -> i64 {
|
|
|
+ self.reservation.0
|
|
|
+ }
|
|
|
+
|
|
|
+ /// Persist this record at its current phase (upsert). The single writer of a
|
|
|
+ /// commit write-ahead record: the Reserving→Finalizing bump is just a persist
|
|
|
+ /// of the [`finalizing`](Self::finalizing) variant.
|
|
|
+ pub(super) async fn persist(&self, ledger: &Ledger) -> Result<(), LedgerError> {
|
|
|
+ PendingRecord::Envelope(self.clone())
|
|
|
+ .save(ledger, self.saga_id())
|
|
|
+ .await
|
|
|
+ }
|
|
|
+
|
|
|
+ /// Run a fresh commit end to end: write-ahead at Reserving, drive the saga,
|
|
|
+ /// then clear the record when it is safe. The single home of the commit
|
|
|
+ /// write-ahead lifecycle; [`commit_envelope`](Ledger::commit_envelope) calls
|
|
|
+ /// this and recovery mirrors it.
|
|
|
+ pub(super) async fn run(self, ledger: &Arc<Ledger>) -> Result<Receipt, LedgerError> {
|
|
|
+ self.persist(ledger).await?;
|
|
|
+ // Commit does not touch the balance projection (ADR-0019): cache points
|
|
|
+ // are appended lazily on read, once enough credits/debits have accrued.
|
|
|
+ let result = ledger
|
|
|
+ .drive_envelope_saga(self.envelope.clone(), self.reservation)
|
|
|
+ .await;
|
|
|
+ self.clear_if_safe(ledger, result.is_ok()).await?;
|
|
|
+ result
|
|
|
+ }
|
|
|
+
|
|
|
+ /// Delete the write-ahead record unless the saga crossed its point of no
|
|
|
+ /// return on a failing run. The ONE place the delete-safety rule lives: safe
|
|
|
+ /// on success, or on a failure that never reached `Finalizing` (compensation
|
|
|
+ /// released our reservation, nothing of ours applied). A failure that reached
|
|
|
+ /// `Finalizing` keeps the record so recovery rolls the commit forward.
|
|
|
+ async fn clear_if_safe(
|
|
|
+ &self,
|
|
|
+ ledger: &Arc<Ledger>,
|
|
|
+ succeeded: bool,
|
|
|
+ ) -> Result<(), LedgerError> {
|
|
|
+ let safe =
|
|
|
+ succeeded || ledger.saga_phase(self.saga_id()).await? != Some(SagaPhase::Finalizing);
|
|
|
+ if safe {
|
|
|
+ ledger.store.delete_saga(&self.saga_id()).await?;
|
|
|
+ }
|
|
|
+ Ok(())
|
|
|
+ }
|
|
|
+
|
|
|
+ /// Complete a crash-interrupted commit from its persisted phase, clearing the
|
|
|
+ /// record when safe. The recovery counterpart of [`run`](Self::run): same
|
|
|
+ /// lifecycle rules, entered from a decoded record instead of a fresh one.
|
|
|
+ async fn complete(self, ledger: &Arc<Ledger>) -> Result<(), LedgerError> {
|
|
|
+ // A full commit is the transfer row plus its committed event (appended
|
|
|
+ // after store_transfer). If the row is present the commit reached the far
|
|
|
+ // side; repair the possibly-missing event (idempotent) and clear.
|
|
|
+ let tid = envelope_id(&self.envelope);
|
|
|
+ if ledger.store.get_transfer(&tid).await?.is_some() {
|
|
|
+ ledger.append_committed_event(tid).await?;
|
|
|
+ ledger.store.delete_saga(&self.saga_id()).await?;
|
|
|
+ return Ok(());
|
|
|
+ }
|
|
|
+
|
|
|
+ match self.phase {
|
|
|
+ // Validation passed and the postings are ours; roll forward through the
|
|
|
+ // verified finalize. Keep the record if it fails so a later run retries.
|
|
|
+ SagaPhase::Finalizing => {
|
|
|
+ if ledger
|
|
|
+ .finalize_envelope(&self.envelope, self.reservation)
|
|
|
+ .await
|
|
|
+ .is_ok()
|
|
|
+ {
|
|
|
+ ledger.store.delete_saga(&self.saga_id()).await?;
|
|
|
+ }
|
|
|
+ Ok(())
|
|
|
+ }
|
|
|
+ // Not past the point of no return: re-run the validating saga and clear
|
|
|
+ // under the same rule as a live commit. The saga's own failure is
|
|
|
+ // absorbed (record kept for the next run); only infra errors propagate.
|
|
|
+ SagaPhase::Reserving => {
|
|
|
+ let result = ledger
|
|
|
+ .drive_envelope_saga(self.envelope.clone(), self.reservation)
|
|
|
+ .await;
|
|
|
+ self.clear_if_safe(ledger, result.is_ok()).await
|
|
|
+ }
|
|
|
}
|
|
|
}
|
|
|
}
|
|
|
@@ -170,48 +263,3 @@ async fn complete_transition(
|
|
|
ledger.store.delete_saga(&saga_id).await?;
|
|
|
Ok(())
|
|
|
}
|
|
|
-
|
|
|
-/// Complete a crash-interrupted commit and clear its record when safe.
|
|
|
-async fn complete_envelope(
|
|
|
- ledger: &Arc<Ledger>,
|
|
|
- saga_id: i64,
|
|
|
- envelope: Envelope,
|
|
|
- reservation: ReservationId,
|
|
|
- phase: SagaPhase,
|
|
|
-) -> Result<(), LedgerError> {
|
|
|
- // The transfer record is durable, but a full commit is more than the transfer
|
|
|
- // row: it also includes the committed event, appended *after* store_transfer.
|
|
|
- // A crash in that window leaves the record present yet the event missing, so
|
|
|
- // repair the whole end-state (idempotent) before clearing the record.
|
|
|
- let tid = envelope_id(&envelope);
|
|
|
- if ledger.store.get_transfer(&tid).await?.is_some() {
|
|
|
- ledger.append_committed_event(tid).await?;
|
|
|
- ledger.store.delete_saga(&saga_id).await?;
|
|
|
- return Ok(());
|
|
|
- }
|
|
|
-
|
|
|
- match phase {
|
|
|
- SagaPhase::Finalizing => {
|
|
|
- // Validation passed and the postings are ours; roll forward. Keep the
|
|
|
- // record if completion fails so a later run retries.
|
|
|
- if ledger
|
|
|
- .finalize_envelope(&envelope, reservation)
|
|
|
- .await
|
|
|
- .is_ok()
|
|
|
- {
|
|
|
- ledger.store.delete_saga(&saga_id).await?;
|
|
|
- }
|
|
|
- }
|
|
|
- SagaPhase::Reserving => {
|
|
|
- // Re-run the validating saga. On failure, delete only if it did not
|
|
|
- // reach finalize (clean abort); otherwise keep for the next run.
|
|
|
- let result = ledger.drive_envelope_saga(envelope, reservation).await;
|
|
|
- let safe_to_delete =
|
|
|
- result.is_ok() || ledger.saga_phase(saga_id).await? != Some(SagaPhase::Finalizing);
|
|
|
- if safe_to_delete {
|
|
|
- ledger.store.delete_saga(&saga_id).await?;
|
|
|
- }
|
|
|
- }
|
|
|
- }
|
|
|
- Ok(())
|
|
|
-}
|