Swap saga (#1183)

# Implement Saga Pattern for Swap Operations with Recovery Mechanism

## Overview

This PR refactors the swap operation implementation to use the saga pattern - a distributed transaction pattern that provides reliable transaction management through explicit state tracking and compensation-based error handling. The implementation includes a robust recovery mechanism that automatically handles swap operations interrupted by crashes, power loss, or network failures.

## What Changed

**Saga Pattern Implementation:**
- Introduced a strict linear state machine for swaps: `Initial` → `SetupComplete` → `Signed` → `Completed`
- New modular `swap_saga` module with state validation, compensation logic, and saga orchestration
- Automatic rollback of database changes on failure, ensuring atomic swap operations
- Replaced previous swap implementation (`swap.rs`, `blinded_message_writer.rs`) with saga-based approach

**Recovery Mechanism:**
- Added `operation_id` and `operation_kind` columns to database schema for tracking which operation proofs belong to
- New `recover_from_bad_swaps()` method that runs on mint startup to handle incomplete swaps
- For proofs left in `PENDING` state from swap operations:
  - If blind signatures exist: marks proofs as `SPENT` (swap completed but not finalized)
  - If no blind signatures exist: removes proofs from database (swap failed partway through)
- Database migrations included for both PostgreSQL and SQLite

crates/cdk-sql-common/Cargo.toml

@@ -29,3 +29,4 @@ serde.workspace = true
 serde_json.workspace = true
 lightning-invoice.workspace = true
 once_cell.workspace = true
+uuid.workspace = true

crates/cdk-sql-common/src/mint/migrations/postgres/20251010144317_add_saga_support.sql

@@ -0,0 +1,24 @@
+-- Add operation and operation_id columns to proof table
+ALTER TABLE proof ADD COLUMN operation_kind TEXT;
+ALTER TABLE proof ADD COLUMN operation_id TEXT;
+
+-- Add operation and operation_id columns to blind_signature table
+ALTER TABLE blind_signature ADD COLUMN operation_kind TEXT;
+ALTER TABLE blind_signature ADD COLUMN operation_id TEXT;
+
+CREATE INDEX idx_proof_state_operation ON proof(state, operation_kind);
+CREATE INDEX idx_proof_operation_id ON proof(operation_kind, operation_id);
+CREATE INDEX idx_blind_sig_operation_id ON blind_signature(operation_kind, operation_id);
+
+-- Add saga_state table for persisting saga state
+CREATE TABLE IF NOT EXISTS saga_state (
+    operation_id TEXT PRIMARY KEY,
+    operation_kind TEXT NOT NULL,
+    state TEXT NOT NULL,
+    blinded_secrets TEXT NOT NULL,
+    input_ys TEXT NOT NULL,
+    created_at BIGINT NOT NULL,
+    updated_at BIGINT NOT NULL
+);
+
+CREATE INDEX IF NOT EXISTS idx_saga_state_operation_kind ON saga_state(operation_kind);

crates/cdk-sql-common/src/mint/migrations/sqlite/20251010144317_add_saga_support.sql

@@ -0,0 +1,24 @@
+-- Add operation and operation_id columns to proof table
+ALTER TABLE proof ADD COLUMN operation_kind TEXT;
+ALTER TABLE proof ADD COLUMN operation_id TEXT;
+
+-- Add operation and operation_id columns to blind_signature table
+ALTER TABLE blind_signature ADD COLUMN operation_kind TEXT;
+ALTER TABLE blind_signature ADD COLUMN operation_id TEXT;
+
+CREATE INDEX idx_proof_state_operation ON proof(state, operation_kind);
+CREATE INDEX idx_proof_operation_id ON proof(operation_kind, operation_id);
+CREATE INDEX idx_blind_sig_operation_id ON blind_signature(operation_kind, operation_id);
+
+-- Add saga_state table for persisting saga state
+CREATE TABLE IF NOT EXISTS saga_state (
+    operation_id TEXT PRIMARY KEY,
+    operation_kind TEXT NOT NULL,
+    state TEXT NOT NULL,
+    blinded_secrets TEXT NOT NULL,
+    input_ys TEXT NOT NULL,
+    created_at INTEGER NOT NULL,
+    updated_at INTEGER NOT NULL
+);
+
+CREATE INDEX IF NOT EXISTS idx_saga_state_operation_kind ON saga_state(operation_kind);

crates/cdk-sql-common/src/mint/mod.rs

@@ -15,7 +15,7 @@ use std::sync::Arc;
 
 use async_trait::async_trait;
 use bitcoin::bip32::DerivationPath;
-use cdk_common::database::mint::validate_kvstore_params;
+use cdk_common::database::mint::{validate_kvstore_params, SagaDatabase, SagaTransaction};
 use cdk_common::database::{
     self, ConversionError, Error, MintDatabase, MintDbWriterFinalizer, MintKeyDatabaseTransaction,
     MintKeysDatabase, MintProofsDatabase, MintQuotesDatabase, MintQuotesTransaction,
@@ -23,6 +23,7 @@ use cdk_common::database::{
 };
 use cdk_common::mint::{
     self, IncomingPayment, Issuance, MeltPaymentRequest, MeltQuote, MintKeySetInfo, MintQuote,
+    Operation,
 };
 use cdk_common::nut00::ProofsMethods;
 use cdk_common::payment::PaymentIdentifier;
@@ -138,6 +139,7 @@ where
         &mut self,
         proofs: Proofs,
         quote_id: Option<QuoteId>,
+        operation: &Operation,
     ) -> Result<(), Self::Err> {
         let current_time = unix_time();
 
@@ -165,9 +167,9 @@ where
             query(
                 r#"
                   INSERT INTO proof
-                  (y, amount, keyset_id, secret, c, witness, state, quote_id, created_time)
+                  (y, amount, keyset_id, secret, c, witness, state, quote_id, created_time, operation_kind, operation_id)
                   VALUES
-                  (:y, :amount, :keyset_id, :secret, :c, :witness, :state, :quote_id, :created_time)
+                  (:y, :amount, :keyset_id, :secret, :c, :witness, :state, :quote_id, :created_time, :operation_kind, :operation_id)
                   "#,
             )?
             .bind("y", proof.y()?.to_bytes().to_vec())
@@ -182,6 +184,8 @@ where
             .bind("state", "UNSPENT".to_string())
             .bind("quote_id", quote_id.clone().map(|q| q.to_string()))
             .bind("created_time", current_time as i64)
+            .bind("operation_kind", operation.kind())
+            .bind("operation_id", operation.id().to_string())
             .execute(&self.inner)
             .await?;
         }
@@ -574,6 +578,7 @@ where
         &mut self,
         quote_id: Option<&QuoteId>,
         blinded_messages: &[BlindedMessage],
+        operation: &Operation,
     ) -> Result<(), Self::Err> {
         let current_time = unix_time();
 
@@ -583,9 +588,9 @@ where
             match query(
                 r#"
                 INSERT INTO blind_signature
-                (blinded_message, amount, keyset_id, c, quote_id, created_time)
+                (blinded_message, amount, keyset_id, c, quote_id, created_time, operation_kind, operation_id)
                 VALUES
-                (:blinded_message, :amount, :keyset_id, NULL, :quote_id, :created_time)
+                (:blinded_message, :amount, :keyset_id, NULL, :quote_id, :created_time, :operation_kind, :operation_id)
                 "#,
             )?
             .bind(
@@ -596,6 +601,8 @@ where
             .bind("keyset_id", message.keyset_id.to_string())
             .bind("quote_id", quote_id.map(|q| q.to_string()))
             .bind("created_time", current_time as i64)
+            .bind("operation_kind", operation.kind())
+            .bind("operation_id", operation.id().to_string())
             .execute(&self.inner)
             .await
             {
@@ -2120,6 +2127,147 @@ where
     }
 }
 
+#[async_trait]
+impl<RM> SagaTransaction<'_> for SQLTransaction<RM>
+where
+    RM: DatabasePool + 'static,
+{
+    type Err = Error;
+
+    async fn get_saga(
+        &mut self,
+        operation_id: &uuid::Uuid,
+    ) -> Result<Option<mint::Saga>, Self::Err> {
+        Ok(query(
+            r#"
+            SELECT
+                operation_id,
+                operation_kind,
+                state,
+                blinded_secrets,
+                input_ys,
+                created_at,
+                updated_at
+            FROM
+                saga_state
+            WHERE
+                operation_id = :operation_id
+            FOR UPDATE
+            "#,
+        )?
+        .bind("operation_id", operation_id.to_string())
+        .fetch_one(&self.inner)
+        .await?
+        .map(sql_row_to_saga)
+        .transpose()?)
+    }
+
+    async fn add_saga(&mut self, saga: &mint::Saga) -> Result<(), Self::Err> {
+        let current_time = unix_time();
+
+        let blinded_secrets_json = serde_json::to_string(&saga.blinded_secrets)
+            .map_err(|e| Error::Internal(format!("Failed to serialize blinded_secrets: {}", e)))?;
+
+        let input_ys_json = serde_json::to_string(&saga.input_ys)
+            .map_err(|e| Error::Internal(format!("Failed to serialize input_ys: {}", e)))?;
+
+        query(
+            r#"
+            INSERT INTO saga_state
+            (operation_id, operation_kind, state, blinded_secrets, input_ys, created_at, updated_at)
+            VALUES
+            (:operation_id, :operation_kind, :state, :blinded_secrets, :input_ys, :created_at, :updated_at)
+            "#,
+        )?
+        .bind("operation_id", saga.operation_id.to_string())
+        .bind("operation_kind", saga.operation_kind.to_string())
+        .bind("state", saga.state.state())
+        .bind("blinded_secrets", blinded_secrets_json)
+        .bind("input_ys", input_ys_json)
+        .bind("created_at", saga.created_at as i64)
+        .bind("updated_at", current_time as i64)
+        .execute(&self.inner)
+        .await?;
+
+        Ok(())
+    }
+
+    async fn update_saga(
+        &mut self,
+        operation_id: &uuid::Uuid,
+        new_state: mint::SagaStateEnum,
+    ) -> Result<(), Self::Err> {
+        let current_time = unix_time();
+
+        query(
+            r#"
+            UPDATE saga_state
+            SET state = :state, updated_at = :updated_at
+            WHERE operation_id = :operation_id
+            "#,
+        )?
+        .bind("state", new_state.state())
+        .bind("updated_at", current_time as i64)
+        .bind("operation_id", operation_id.to_string())
+        .execute(&self.inner)
+        .await?;
+
+        Ok(())
+    }
+
+    async fn delete_saga(&mut self, operation_id: &uuid::Uuid) -> Result<(), Self::Err> {
+        query(
+            r#"
+            DELETE FROM saga_state
+            WHERE operation_id = :operation_id
+            "#,
+        )?
+        .bind("operation_id", operation_id.to_string())
+        .execute(&self.inner)
+        .await?;
+
+        Ok(())
+    }
+}
+
+#[async_trait]
+impl<RM> SagaDatabase for SQLMintDatabase<RM>
+where
+    RM: DatabasePool + 'static,
+{
+    type Err = Error;
+
+    async fn get_incomplete_sagas(
+        &self,
+        operation_kind: mint::OperationKind,
+    ) -> Result<Vec<mint::Saga>, Self::Err> {
+        let conn = self.pool.get().map_err(|e| Error::Database(Box::new(e)))?;
+        Ok(query(
+            r#"
+            SELECT
+                operation_id,
+                operation_kind,
+                state,
+                blinded_secrets,
+                input_ys,
+                created_at,
+                updated_at
+            FROM
+                saga_state
+            WHERE
+                operation_kind = :operation_kind
+            ORDER BY created_at ASC
+            "#,
+        )?
+        .bind("operation_kind", operation_kind.to_string())
+        .fetch_all(&*conn)
+        .await?
+        .into_iter()
+        .map(sql_row_to_saga)
+        .collect::<Result<Vec<_>, _>>()?)
+    }
+}
+
 #[async_trait]
 impl<RM> MintDatabase<Error> for SQLMintDatabase<RM>
 where
@@ -2383,6 +2531,53 @@ fn sql_row_to_blind_signature(row: Vec<Column>) -> Result<BlindSignature, Error>
     })
 }
 
+fn sql_row_to_saga(row: Vec<Column>) -> Result<mint::Saga, Error> {
+    unpack_into!(
+        let (
+            operation_id,
+            operation_kind,
+            state,
+            blinded_secrets,
+            input_ys,
+            created_at,
+            updated_at
+        ) = row
+    );
+
+    let operation_id_str = column_as_string!(&operation_id);
+    let operation_id = uuid::Uuid::parse_str(&operation_id_str)
+        .map_err(|e| Error::Internal(format!("Invalid operation_id UUID: {}", e)))?;
+
+    let operation_kind_str = column_as_string!(&operation_kind);
+    let operation_kind = mint::OperationKind::from_str(&operation_kind_str)
+        .map_err(|e| Error::Internal(format!("Invalid operation kind: {}", e)))?;
+
+    let state_str = column_as_string!(&state);
+    let state = mint::SagaStateEnum::new(operation_kind, &state_str)
+        .map_err(|e| Error::Internal(format!("Invalid saga state: {}", e)))?;
+
+    let blinded_secrets_str = column_as_string!(&blinded_secrets);
+    let blinded_secrets: Vec<PublicKey> = serde_json::from_str(&blinded_secrets_str)
+        .map_err(|e| Error::Internal(format!("Failed to deserialize blinded_secrets: {}", e)))?;
+
+    let input_ys_str = column_as_string!(&input_ys);
+    let input_ys: Vec<PublicKey> = serde_json::from_str(&input_ys_str)
+        .map_err(|e| Error::Internal(format!("Failed to deserialize input_ys: {}", e)))?;
+
+    let created_at: u64 = column_as_number!(created_at);
+    let updated_at: u64 = column_as_number!(updated_at);
+
+    Ok(mint::Saga {
+        operation_id,
+        operation_kind,
+        state,
+        blinded_secrets,
+        input_ys,
+        created_at,
+        updated_at,
+    })
+}
+
 #[cfg(test)]
 mod test {
     use super::*;