Recover incoming settlement and persist immutable purchase journals

core/archipelago/src/content_purchase.rs

@@ -0,0 +1,909 @@
+//! Durable purchase intent and seller receipt primitives. No transport or wallet
+//! mutations happen here. Callers must authenticate both peers, negotiate this
+//! protocol and obtain a stable content offer before creating the contract.
+//!
+//! A caller must retain the same purchase UUID across retries. A saved token is
+//! not settlement evidence: only a successful, correlated recoverable wallet
+//! result may advance seller settlement. Received receipts must come from the
+//! authenticated seller; this local journal is not a wire-signature verifier.
+use anyhow::{Context, Result};
+use rand::RngCore;
+use serde::{Deserialize, Serialize};
+use sha2::{Digest, Sha256};
+use std::path::{Path, PathBuf};
+use tokio::{
+    fs,
+    io::{AsyncReadExt, AsyncWriteExt},
+};
+
+use crate::wallet::{cashu::CashuToken, ecash::EcashNetwork};
+const VERSION: u8 = 1;
+const MAX_RECORD_BYTES: u64 = 2 * 1024 * 1024;
+const MAX_TOKEN_BYTES: usize = 512 * 1024;
+
+fn hash(bytes: &[u8]) -> String {
+    hex::encode(Sha256::digest(bytes))
+}
+fn valid_hash(value: &str) -> bool {
+    value.len() == 64
+        && value
+            .bytes()
+            .all(|c| c.is_ascii_digit() || (b'a'..=b'f').contains(&c))
+}
+fn validate_id(id: &str) -> Result<()> {
+    anyhow::ensure!(
+        uuid::Uuid::parse_str(id)
+            .ok()
+            .is_some_and(|v| v.to_string() == id),
+        "Invalid purchase identifier"
+    );
+    Ok(())
+}
+fn canonical_mint(value: &str) -> Result<String> {
+    let url = reqwest::Url::parse(value).context("Invalid purchase mint")?;
+    anyhow::ensure!(
+        matches!(url.scheme(), "http" | "https")
+            && url.host_str().is_some()
+            && url.username().is_empty()
+            && url.password().is_none()
+            && url.query().is_none()
+            && url.fragment().is_none(),
+        "Invalid purchase mint"
+    );
+    Ok(url.to_string().trim_end_matches('/').to_owned())
+}
+
+/// Verified identities are supplied by the authenticated offer/request layer.
+/// Parsing a DID here checks its form; it does not authenticate its presenter.
+#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(deny_unknown_fields)]
+pub(crate) struct Contract {
+    pub version: u8,
+    pub id: String,
+    pub buyer_did: String,
+    pub seller_did: String,
+    pub content_id: String,
+    pub content_sha256: String,
+    pub content_size: u64,
+    pub terms_sha256: String,
+    pub network: EcashNetwork,
+    pub mint_url: String,
+    pub gross_token_sats: u64,
+    pub minimum_net_sats: u64,
+    pub offered_at: i64,
+    pub expires_at: i64,
+}
+impl Contract {
+    pub fn validate(&self) -> Result<()> {
+        validate_id(&self.id)?;
+        anyhow::ensure!(self.version == VERSION, "Unsupported purchase protocol");
+        crate::identity::pubkey_bytes_from_did_key(&self.buyer_did)?;
+        crate::identity::pubkey_bytes_from_did_key(&self.seller_did)?;
+        anyhow::ensure!(
+            self.buyer_did != self.seller_did,
+            "Purchase peers must be distinct"
+        );
+        anyhow::ensure!(
+            !self.content_id.is_empty()
+                && self.content_id.len() <= 256
+                && self
+                    .content_id
+                    .bytes()
+                    .all(|c| c.is_ascii_alphanumeric() || b"_-".contains(&c)),
+            "Invalid purchase content identifier"
+        );
+        anyhow::ensure!(
+            valid_hash(&self.content_sha256)
+                && valid_hash(&self.terms_sha256)
+                && self.content_size > 0,
+            "Invalid purchase content or terms"
+        );
+        anyhow::ensure!(
+            self.minimum_net_sats > 0 && self.gross_token_sats >= self.minimum_net_sats,
+            "Invalid gross/net purchase amounts"
+        );
+        anyhow::ensure!(
+            canonical_mint(&self.mint_url)? == self.mint_url,
+            "Purchase mint is not canonical"
+        );
+        anyhow::ensure!(
+            self.offered_at > 0 && self.expires_at > self.offered_at,
+            "Invalid purchase offer lifetime"
+        );
+        Ok(())
+    }
+    /// Stable context passed to both recoverable wallet operations.
+    pub fn context_hash(&self) -> Result<String> {
+        self.validate()?;
+        Ok(hash(&serde_json::to_vec(&(
+            "archipelago-content-purchase-v1",
+            self,
+        ))?))
+    }
+    fn validate_new_at(&self, now: i64) -> Result<()> {
+        self.validate()?;
+        anyhow::ensure!(
+            now >= self.offered_at && now < self.expires_at,
+            "Purchase offer is not current"
+        );
+        Ok(())
+    }
+}
+
+/// Private delivery capability; do not log or expose it to another buyer.
+/// The wire layer must authenticate this receipt before a buyer stores it.
+#[derive(Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(deny_unknown_fields)]
+pub(crate) struct Receipt {
+    pub contract_hash: String,
+    pub amount_received: u64,
+    pub capability: String,
+}
+impl Receipt {
+    fn validate(&self, contract: &Contract) -> Result<()> {
+        anyhow::ensure!(
+            self.contract_hash == contract.context_hash()?
+                && self.amount_received >= contract.minimum_net_sats
+                && self.amount_received <= contract.gross_token_sats
+                && valid_hash(&self.capability),
+            "Receipt does not match the purchase"
+        );
+        Ok(())
+    }
+}
+#[derive(Clone, Serialize, Deserialize)]
+#[serde(deny_unknown_fields)]
+struct PreparedToken {
+    encoded: String,
+    sha256: String,
+}
+impl PreparedToken {
+    fn new(contract: &Contract, encoded: String) -> Result<Self> {
+        let value = Self {
+            sha256: hash(encoded.as_bytes()),
+            encoded,
+        };
+        value.validate(contract)?;
+        Ok(value)
+    }
+    fn validate(&self, contract: &Contract) -> Result<()> {
+        anyhow::ensure!(
+            self.encoded.len() <= MAX_TOKEN_BYTES && self.sha256 == hash(self.encoded.as_bytes()),
+            "Prepared purchase token is damaged"
+        );
+        let token =
+            CashuToken::deserialize(&self.encoded).context("Invalid prepared purchase token")?;
+        anyhow::ensure!(
+            token.unit.as_deref().unwrap_or("sat") == "sat" && token.token.len() == 1,
+            "Purchase requires one sat-denominated mint"
+        );
+        let entry = &token.token[0];
+        anyhow::ensure!(
+            canonical_mint(&entry.mint)? == contract.mint_url,
+            "Purchase token mint changed"
+        );
+        let mut secrets = std::collections::HashSet::new();
+        let mut total = 0u64;
+        for proof in &entry.proofs {
+            anyhow::ensure!(
+                proof.amount.is_power_of_two()
+                    && !proof.secret.is_empty()
+                    && secrets.insert(&proof.secret),
+                "Invalid or duplicate purchase proof"
+            );
+            proof.c_as_pubkey()?;
+            total = total
+                .checked_add(proof.amount)
+                .context("Purchase amount overflow")?;
+        }
+        anyhow::ensure!(
+            total == contract.gross_token_sats,
+            "Purchase token amount changed"
+        );
+        Ok(())
+    }
+}
+#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
+pub(crate) enum BuyerPhase {
+    Intent,
+    TokenPrepared,
+    ReceiptSaved,
+    Delivered,
+}
+#[derive(Clone, Serialize, Deserialize)]
+#[serde(deny_unknown_fields)]
+pub(crate) struct BuyerRecord {
+    pub contract: Contract,
+    pub phase: BuyerPhase,
+    token: Option<PreparedToken>,
+    receipt: Option<Receipt>,
+}
+impl BuyerRecord {
+    pub fn token(&self) -> Option<&str> {
+        self.token.as_ref().map(|token| token.encoded.as_str())
+    }
+    pub fn receipt(&self) -> Option<&Receipt> {
+        self.receipt.as_ref()
+    }
+    fn validate(&self) -> Result<()> {
+        self.contract.validate()?;
+        if let Some(token) = &self.token {
+            token.validate(&self.contract)?;
+        }
+        if let Some(receipt) = &self.receipt {
+            receipt.validate(&self.contract)?;
+        }
+        anyhow::ensure!(
+            match self.phase {
+                BuyerPhase::Intent => self.token.is_none() && self.receipt.is_none(),
+                BuyerPhase::TokenPrepared => self.token.is_some() && self.receipt.is_none(),
+                BuyerPhase::ReceiptSaved | BuyerPhase::Delivered =>
+                    self.token.is_some() && self.receipt.is_some(),
+            },
+            "Invalid buyer purchase transition"
+        );
+        Ok(())
+    }
+}
+#[derive(Clone, Serialize, Deserialize)]
+#[serde(deny_unknown_fields)]
+pub(crate) enum SellerPhase {
+    Intent,
+    Settled { amount_received: u64 },
+    ReceiptSaved(Receipt),
+}
+#[derive(Clone, Serialize, Deserialize)]
+#[serde(deny_unknown_fields)]
+pub(crate) struct SellerRecord {
+    pub contract: Contract,
+    pub phase: SellerPhase,
+}
+impl SellerRecord {
+    fn validate(&self) -> Result<()> {
+        self.contract.validate()?;
+        match &self.phase {
+            SellerPhase::Intent => (),
+            SellerPhase::Settled { amount_received } => {
+                anyhow::ensure!(
+                    *amount_received >= self.contract.minimum_net_sats
+                        && *amount_received <= self.contract.gross_token_sats,
+                    "Invalid seller settlement amount"
+                );
+            }
+            SellerPhase::ReceiptSaved(receipt) => receipt.validate(&self.contract)?,
+        }
+        Ok(())
+    }
+}
+#[derive(Serialize, Deserialize)]
+#[serde(deny_unknown_fields)]
+struct Envelope {
+    version: u8,
+    payload: String,
+    checksum: String,
+}
+
+/// Exclusive journal access across tasks and processes. Hold this only while
+/// changing local purchase state; release it before transport/wallet calls.
+/// A later caller reopens and revalidates the immutable contract before advancing.
+pub(crate) struct Journal {
+    directory: PathBuf,
+    _lock: std::fs::File,
+    #[cfg(test)]
+    before_commit: Option<(
+        std::sync::Arc<tokio::sync::Notify>,
+        std::sync::Arc<tokio::sync::Notify>,
+    )>,
+}
+impl Journal {
+    pub async fn open(data_dir: &Path) -> Result<Self> {
+        fs::create_dir_all(data_dir).await?;
+        let data_dir = fs::canonicalize(data_dir).await?;
+        let directory = data_dir.join("content-purchases");
+        fs::create_dir_all(&directory).await?;
+        anyhow::ensure!(
+            fs::symlink_metadata(&directory).await?.is_dir(),
+            "Purchase journal directory is not regular"
+        );
+        #[cfg(unix)]
+        {
+            use std::os::unix::fs::PermissionsExt;
+            fs::set_permissions(&directory, std::fs::Permissions::from_mode(0o700)).await?;
+        }
+        let path = directory.join(".lock");
+        let lock = tokio::task::spawn_blocking(move || -> Result<std::fs::File> {
+            let mut options = std::fs::OpenOptions::new();
+            options.read(true).write(true).create(true);
+            #[cfg(unix)]
+            {
+                use std::os::unix::fs::OpenOptionsExt;
+                options
+                    .mode(0o600)
+                    .custom_flags(libc::O_NOFOLLOW | libc::O_NONBLOCK);
+            }
+            let file = options.open(path)?;
+            anyhow::ensure!(
+                file.metadata()?.is_file(),
+                "Purchase lock is not a regular file"
+            );
+            #[cfg(unix)]
+            {
+                use std::os::fd::AsRawFd;
+                loop {
+                    if unsafe { libc::flock(file.as_raw_fd(), libc::LOCK_EX) } == 0 {
+                        break;
+                    }
+                    let error = std::io::Error::last_os_error();
+                    if error.kind() != std::io::ErrorKind::Interrupted {
+                        return Err(error.into());
+                    }
+                }
+            }
+            #[cfg(not(unix))]
+            anyhow::bail!("Purchase journal locking requires Unix");
+            Ok(file)
+        })
+        .await??;
+        Ok(Self {
+            directory,
+            _lock: lock,
+            #[cfg(test)]
+            before_commit: None,
+        })
+    }
+    fn path(&self, role: &str, id: &str) -> Result<PathBuf> {
+        validate_id(id)?;
+        anyhow::ensure!(
+            matches!(role, "buyer" | "seller"),
+            "Invalid purchase journal role"
+        );
+        Ok(self.directory.join(format!("{role}-{id}.json")))
+    }
+    async fn read<T: serde::de::DeserializeOwned>(
+        &self,
+        role: &str,
+        id: &str,
+    ) -> Result<Option<T>> {
+        let mut options = fs::OpenOptions::new();
+        options.read(true);
+        #[cfg(unix)]
+        options.custom_flags(libc::O_NOFOLLOW | libc::O_NONBLOCK);
+        let file = match options.open(self.path(role, id)?).await {
+            Ok(file) => file,
+            Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(None),
+            Err(e) => return Err(e).context("Cannot read purchase recovery"),
+        };
+        anyhow::ensure!(
+            file.metadata().await?.is_file(),
+            "Purchase record is not a regular file"
+        );
+        let mut bytes = Vec::new();
+        file.take(MAX_RECORD_BYTES + 1)
+            .read_to_end(&mut bytes)
+            .await?;
+        anyhow::ensure!(
+            bytes.len() as u64 <= MAX_RECORD_BYTES,
+            "Purchase recovery exceeds size limit"
+        );
+        let envelope: Envelope = serde_json::from_slice(&bytes)
+            .context("Purchase recovery is damaged; do not pay again")?;
+        anyhow::ensure!(
+            envelope.version == VERSION && envelope.checksum == hash(envelope.payload.as_bytes()),
+            "Purchase recovery checksum/version failed; do not pay again"
+        );
+        Ok(Some(
+            serde_json::from_str(&envelope.payload)
+                .context("Purchase recovery contents are damaged")?,
+        ))
+    }
+    async fn write<T: Serialize>(&self, role: &str, id: &str, value: &T) -> Result<()> {
+        let payload = serde_json::to_string(value)?;
+        let bytes = serde_json::to_vec(&Envelope {
+            version: VERSION,
+            checksum: hash(payload.as_bytes()),
+            payload,
+        })?;
+        anyhow::ensure!(
+            bytes.len() as u64 <= MAX_RECORD_BYTES,
+            "Purchase recovery exceeds size limit"
+        );
+        struct Temporary(PathBuf);
+        impl Drop for Temporary {
+            fn drop(&mut self) {
+                let _ = std::fs::remove_file(&self.0);
+            }
+        }
+        let temporary = Temporary(
+            self.directory
+                .join(format!(".{}.tmp", uuid::Uuid::new_v4())),
+        );
+        let mut options = fs::OpenOptions::new();
+        options.write(true).create_new(true);
+        #[cfg(unix)]
+        options.mode(0o600);
+        let mut file = options.open(&temporary.0).await?;
+        file.write_all(&bytes).await?;
+        file.sync_all().await?;
+        drop(file);
+        #[cfg(test)]
+        if let Some((reached, resume)) = &self.before_commit {
+            reached.notify_one();
+            resume.notified().await;
+        }
+        // No await after the commit begins: a cancelled future must not release
+        // the journal lock while an async rename can still overwrite new state.
+        std::fs::rename(&temporary.0, self.path(role, id)?)?;
+        std::fs::File::open(&self.directory)?.sync_all()?;
+        std::fs::File::open(
+            self.directory
+                .parent()
+                .context("Purchase journal has no parent")?,
+        )?
+        .sync_all()?;
+        Ok(())
+    }
+    pub async fn buyer(&self, id: &str) -> Result<Option<BuyerRecord>> {
+        let result: Option<BuyerRecord> = self.read("buyer", id).await?;
+        if let Some(record) = &result {
+            record.validate()?;
+            anyhow::ensure!(record.contract.id == id, "Buyer journal identity changed");
+        }
+        Ok(result)
+    }
+    pub async fn seller(&self, id: &str) -> Result<Option<SellerRecord>> {
+        let result: Option<SellerRecord> = self.read("seller", id).await?;
+        if let Some(record) = &result {
+            record.validate()?;
+            anyhow::ensure!(record.contract.id == id, "Seller journal identity changed");
+        }
+        Ok(result)
+    }
+    pub async fn prepare_buyer(&self, contract: &Contract, now: i64) -> Result<BuyerRecord> {
+        contract.validate()?;
+        if let Some(record) = self.buyer(&contract.id).await? {
+            anyhow::ensure!(&record.contract == contract, "Buyer purchase terms changed");
+            return Ok(record);
+        }
+        contract.validate_new_at(now)?;
+        let record = BuyerRecord {
+            contract: contract.clone(),
+            phase: BuyerPhase::Intent,
+            token: None,
+            receipt: None,
+        };
+        self.write("buyer", &contract.id, &record).await?;
+        Ok(record)
+    }
+    pub async fn prepare_seller(&self, contract: &Contract, now: i64) -> Result<SellerRecord> {
+        contract.validate()?;
+        if let Some(record) = self.seller(&contract.id).await? {
+            anyhow::ensure!(
+                &record.contract == contract,
+                "Seller purchase terms changed"
+            );
+            return Ok(record);
+        }
+        contract.validate_new_at(now)?;
+        let record = SellerRecord {
+            contract: contract.clone(),
+            phase: SellerPhase::Intent,
+        };
+        self.write("seller", &contract.id, &record).await?;
+        Ok(record)
+    }
+    async fn bound_buyer(&self, contract: &Contract) -> Result<BuyerRecord> {
+        let record = self
+            .buyer(&contract.id)
+            .await?
+            .context("Buyer intent is not durable")?;
+        anyhow::ensure!(&record.contract == contract, "Buyer purchase terms changed");
+        Ok(record)
+    }
+    async fn bound_seller(&self, contract: &Contract) -> Result<SellerRecord> {
+        let record = self
+            .seller(&contract.id)
+            .await?
+            .context("Seller intent is not durable")?;
+        anyhow::ensure!(
+            &record.contract == contract,
+            "Seller purchase terms changed"
+        );
+        Ok(record)
+    }
+    /// Call only with the original correlated recoverable-send result.
+    pub async fn record_token(&self, contract: &Contract, encoded: &str) -> Result<BuyerRecord> {
+        let mut record = self.bound_buyer(contract).await?;
+        let token = PreparedToken::new(contract, encoded.into())?;
+        if let Some(previous) = &record.token {
+            anyhow::ensure!(
+                previous.encoded == token.encoded,
+                "Buyer already has a different prepared token"
+            );
+            return Ok(record);
+        }
+        anyhow::ensure!(
+            record.phase == BuyerPhase::Intent,
+            "Cannot prepare token in this phase"
+        );
+        record.token = Some(token);
+        record.phase = BuyerPhase::TokenPrepared;
+        record.validate()?;
+        self.write("buyer", &contract.id, &record).await?;
+        Ok(record)
+    }
+    /// Call only after receive_token_recoverable succeeds for this UUID/context.
+    /// A missing response, client's claim or balance delta is not settlement.
+    pub async fn record_settlement(
+        &self,
+        contract: &Contract,
+        amount_received: u64,
+    ) -> Result<SellerRecord> {
+        let mut record = self.bound_seller(contract).await?;
+        match &record.phase {
+            SellerPhase::Intent => record.phase = SellerPhase::Settled { amount_received },
+            SellerPhase::Settled {
+                amount_received: saved,
+            } => {
+                anyhow::ensure!(
+                    *saved == amount_received,
+                    "Seller settlement result changed"
+                );
+                return Ok(record);
+            }
+            SellerPhase::ReceiptSaved(receipt) => {
+                anyhow::ensure!(
+                    receipt.amount_received == amount_received,
+                    "Seller settlement result changed"
+                );
+                return Ok(record);
+            }
+        }
+        record.validate()?;
+        self.write("seller", &contract.id, &record).await?;
+        Ok(record)
+    }
+    /// Generate once and save before returning the capability to the wire layer.
+    pub async fn issue_receipt(&self, contract: &Contract) -> Result<Receipt> {
+        let mut record = self.bound_seller(contract).await?;
+        let amount_received = match &record.phase {
+            SellerPhase::Intent => anyhow::bail!("Seller settlement is not durable"),
+            SellerPhase::Settled { amount_received } => *amount_received,
+            SellerPhase::ReceiptSaved(receipt) => return Ok(receipt.clone()),
+        };
+        let mut capability = [0u8; 32];
+        rand::rngs::OsRng.fill_bytes(&mut capability);
+        let receipt = Receipt {
+            contract_hash: contract.context_hash()?,
+            amount_received,
+            capability: hex::encode(capability),
+        };
+        receipt.validate(contract)?;
+        record.phase = SellerPhase::ReceiptSaved(receipt.clone());
+        self.write("seller", &contract.id, &record).await?;
+        Ok(receipt)
+    }
+    /// The caller must first authenticate the seller's response and contract.
+    pub async fn record_receipt(
+        &self,
+        contract: &Contract,
+        receipt: &Receipt,
+    ) -> Result<BuyerRecord> {
+        let mut record = self.bound_buyer(contract).await?;
+        receipt.validate(contract)?;
+        if let Some(previous) = &record.receipt {
+            anyhow::ensure!(previous == receipt, "Buyer already has a different receipt");
+            return Ok(record);
+        }
+        anyhow::ensure!(
+            record.phase == BuyerPhase::TokenPrepared,
+            "Buyer token is not durable"
+        );
+        record.receipt = Some(receipt.clone());
+        record.phase = BuyerPhase::ReceiptSaved;
+        record.validate()?;
+        self.write("buyer", &contract.id, &record).await?;
+        Ok(record)
+    }
+    /// Call only after complete downloaded bytes and metadata have been flushed.
+    pub async fn record_delivery(
+        &self,
+        contract: &Contract,
+        sha256: &str,
+        size: u64,
+    ) -> Result<BuyerRecord> {
+        let mut record = self.bound_buyer(contract).await?;
+        anyhow::ensure!(
+            matches!(
+                record.phase,
+                BuyerPhase::ReceiptSaved | BuyerPhase::Delivered
+            ),
+            "Buyer receipt is not durable"
+        );
+        anyhow::ensure!(
+            sha256 == contract.content_sha256 && size == contract.content_size,
+            "Delivered content changed"
+        );
+        if record.phase != BuyerPhase::Delivered {
+            record.phase = BuyerPhase::Delivered;
+            self.write("buyer", &contract.id, &record).await?;
+        }
+        Ok(record)
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    fn contract() -> Contract {
+        Contract {
+            version: VERSION,
+            id: uuid::Uuid::new_v4().to_string(),
+            buyer_did: crate::identity::did_key_from_pubkey_hex(&hex::encode([1; 32])).unwrap(),
+            seller_did: crate::identity::did_key_from_pubkey_hex(&hex::encode([2; 32])).unwrap(),
+            content_id: "film-1".into(),
+            content_sha256: "ab".repeat(32),
+            content_size: 1024,
+            terms_sha256: "cd".repeat(32),
+            network: EcashNetwork::Mainnet,
+            mint_url: "https://mint.invalid".into(),
+            gross_token_sats: 8,
+            minimum_net_sats: 7,
+            offered_at: 1000,
+            expires_at: 2000,
+        }
+    }
+    fn token(contract: &Contract, secret: &str) -> String {
+        let key = bitcoin::secp256k1::SecretKey::from_slice(&[7; 32]).unwrap();
+        let c = bitcoin::secp256k1::PublicKey::from_secret_key(
+            &bitcoin::secp256k1::Secp256k1::new(),
+            &key,
+        )
+        .to_string();
+        CashuToken::new(
+            &contract.mint_url,
+            vec![crate::wallet::cashu::Proof {
+                id: "0011223344556677".into(),
+                amount: contract.gross_token_sats,
+                secret: secret.into(),
+                c,
+            }],
+        )
+        .serialize()
+        .unwrap()
+    }
+    #[test]
+    fn strict_contract_binds_identity_content_terms_network_and_fee_amounts() {
+        let original = contract();
+        original.validate().unwrap();
+        let context = original.context_hash().unwrap();
+        let mut changed = original.clone();
+        changed.version = 2;
+        assert!(changed.validate().is_err());
+        let mut value = serde_json::to_value(&original).unwrap();
+        value["unreviewed_field"] = serde_json::json!(true);
+        assert!(serde_json::from_value::<Contract>(value).is_err());
+        let mut changed = original.clone();
+        changed.minimum_net_sats = 9;
+        assert!(changed.validate().is_err());
+        changed = original.clone();
+        changed.buyer_did = "claimed".into();
+        assert!(changed.validate().is_err());
+        changed = original.clone();
+        changed.mint_url.push('/');
+        assert!(changed.validate().is_err());
+        changed = original.clone();
+        changed.content_sha256 = "ef".repeat(32);
+        assert_ne!(changed.context_hash().unwrap(), context);
+        changed = original.clone();
+        changed.network = EcashNetwork::Testnet;
+        assert_ne!(changed.context_hash().unwrap(), context);
+        changed = original.clone();
+        changed.minimum_net_sats = 8;
+        assert_ne!(changed.context_hash().unwrap(), context);
+    }
+    #[tokio::test]
+    async fn buyer_seller_resume_original_token_receipt_and_delivery_after_reopen() {
+        let root = tempfile::tempdir().unwrap();
+        let contract = contract();
+        let encoded = token(&contract, "original");
+        let journal = Journal::open(root.path()).await.unwrap();
+        assert!(journal.record_token(&contract, &encoded).await.is_err());
+        assert!(journal.record_settlement(&contract, 7).await.is_err());
+        journal.prepare_buyer(&contract, 1500).await.unwrap();
+        journal.prepare_seller(&contract, 1500).await.unwrap();
+        assert!(journal.issue_receipt(&contract).await.is_err());
+        journal.record_token(&contract, &encoded).await.unwrap();
+        journal.record_settlement(&contract, 7).await.unwrap();
+        drop(journal);
+        let journal = Journal::open(root.path()).await.unwrap();
+        // Expiry prevents a new sale, not recovery of the original durable one.
+        journal.prepare_buyer(&contract, 3000).await.unwrap();
+        journal.prepare_seller(&contract, 3000).await.unwrap();
+        assert_eq!(
+            journal.buyer(&contract.id).await.unwrap().unwrap().token(),
+            Some(encoded.as_str())
+        );
+        let receipt = journal.issue_receipt(&contract).await.unwrap();
+        journal.record_receipt(&contract, &receipt).await.unwrap();
+        let before = fs::read(journal.path("buyer", &contract.id).unwrap())
+            .await
+            .unwrap();
+        journal.record_token(&contract, &encoded).await.unwrap();
+        journal.record_receipt(&contract, &receipt).await.unwrap();
+        assert_eq!(
+            fs::read(journal.path("buyer", &contract.id).unwrap())
+                .await
+                .unwrap(),
+            before
+        );
+        assert!(journal
+            .record_delivery(&contract, &"ef".repeat(32), contract.content_size)
+            .await
+            .is_err());
+        journal
+            .record_delivery(&contract, &contract.content_sha256, contract.content_size)
+            .await
+            .unwrap();
+        drop(journal);
+        let journal = Journal::open(root.path()).await.unwrap();
+        assert!(journal.issue_receipt(&contract).await.unwrap() == receipt);
+        assert_eq!(
+            journal.buyer(&contract.id).await.unwrap().unwrap().phase,
+            BuyerPhase::Delivered
+        );
+        assert!(
+            journal
+                .buyer(&contract.id)
+                .await
+                .unwrap()
+                .unwrap()
+                .receipt()
+                .unwrap()
+                == &receipt
+        );
+        assert!(journal.record_settlement(&contract, 8).await.is_err());
+    }
+    #[tokio::test]
+    async fn changed_terms_tokens_and_foreign_receipts_preserve_original_record() {
+        let root = tempfile::tempdir().unwrap();
+        let contract = contract();
+        let journal = Journal::open(root.path()).await.unwrap();
+        journal.prepare_buyer(&contract, 1500).await.unwrap();
+        journal.prepare_seller(&contract, 1500).await.unwrap();
+        journal
+            .record_token(&contract, &token(&contract, "first"))
+            .await
+            .unwrap();
+        let before = fs::read(journal.path("buyer", &contract.id).unwrap())
+            .await
+            .unwrap();
+        let mut changed = contract.clone();
+        changed.terms_sha256 = "ef".repeat(32);
+        assert!(journal.prepare_buyer(&changed, 1500).await.is_err());
+        assert!(journal.prepare_seller(&changed, 1500).await.is_err());
+        assert!(journal
+            .record_token(&contract, &token(&contract, "other"))
+            .await
+            .is_err());
+        assert!(journal.record_settlement(&contract, 6).await.is_err());
+        journal.record_settlement(&contract, 7).await.unwrap();
+        let mut receipt = journal.issue_receipt(&contract).await.unwrap();
+        receipt.contract_hash = changed.context_hash().unwrap();
+        assert!(journal.record_receipt(&contract, &receipt).await.is_err());
+        assert_eq!(
+            fs::read(journal.path("buyer", &contract.id).unwrap())
+                .await
+                .unwrap(),
+            before
+        );
+        let mut expired = contract.clone();
+        expired.id = uuid::Uuid::new_v4().to_string();
+        assert!(journal.prepare_buyer(&expired, 2000).await.is_err());
+        assert!(journal.prepare_seller(&expired, 999).await.is_err());
+        assert!(journal.buyer(&expired.id).await.unwrap().is_none());
+    }
+    #[tokio::test]
+    async fn damaged_records_and_nonregular_targets_are_preserved() {
+        let root = tempfile::tempdir().unwrap();
+        let contract = contract();
+        let journal = Journal::open(root.path()).await.unwrap();
+        let path = journal.path("buyer", &contract.id).unwrap();
+        fs::write(&path, b"damaged").await.unwrap();
+        assert!(journal.prepare_buyer(&contract, 1500).await.is_err());
+        assert_eq!(fs::read(&path).await.unwrap(), b"damaged");
+        fs::remove_file(&path).await.unwrap();
+        fs::create_dir(&path).await.unwrap();
+        assert!(journal.prepare_buyer(&contract, 1500).await.is_err());
+        assert!(path.is_dir());
+        #[cfg(unix)]
+        {
+            fs::remove_dir(&path).await.unwrap();
+            let target = root.path().join("untouched");
+            fs::write(&target, b"preserve").await.unwrap();
+            std::os::unix::fs::symlink(&target, &path).unwrap();
+            assert!(journal.prepare_buyer(&contract, 1500).await.is_err());
+            assert_eq!(fs::read(&target).await.unwrap(), b"preserve");
+        }
+    }
+    #[tokio::test]
+    async fn private_records_reject_checksum_damage_and_skipped_phases() {
+        let root = tempfile::tempdir().unwrap();
+        let contract = contract();
+        let journal = Journal::open(root.path()).await.unwrap();
+        journal.prepare_buyer(&contract, 1500).await.unwrap();
+        let path = journal.path("buyer", &contract.id).unwrap();
+        #[cfg(unix)]
+        {
+            use std::os::unix::fs::PermissionsExt;
+            assert_eq!(
+                std::fs::metadata(&path).unwrap().permissions().mode() & 0o777,
+                0o600
+            );
+            assert_eq!(
+                std::fs::metadata(&journal.directory)
+                    .unwrap()
+                    .permissions()
+                    .mode()
+                    & 0o777,
+                0o700
+            );
+        }
+        let mut envelope: Envelope =
+            serde_json::from_slice(&fs::read(&path).await.unwrap()).unwrap();
+        envelope.checksum = "00".repeat(32);
+        fs::write(&path, serde_json::to_vec(&envelope).unwrap())
+            .await
+            .unwrap();
+        assert!(journal.buyer(&contract.id).await.is_err());
+        let mut record: BuyerRecord = serde_json::from_str(&envelope.payload).unwrap();
+        record.phase = BuyerPhase::Delivered;
+        envelope.payload = serde_json::to_string(&record).unwrap();
+        envelope.checksum = hash(envelope.payload.as_bytes());
+        fs::write(&path, serde_json::to_vec(&envelope).unwrap())
+            .await
+            .unwrap();
+        assert!(journal.buyer(&contract.id).await.is_err());
+    }
+    #[tokio::test]
+    async fn cancellation_before_commit_cannot_overwrite_the_next_writer() {
+        let root = tempfile::tempdir().unwrap();
+        let contract = contract();
+        let mut journal = Journal::open(root.path()).await.unwrap();
+        journal.prepare_buyer(&contract, 1500).await.unwrap();
+        let reached = std::sync::Arc::new(tokio::sync::Notify::new());
+        let resume = std::sync::Arc::new(tokio::sync::Notify::new());
+        journal.before_commit = Some((reached.clone(), resume.clone()));
+        let original_contract = contract.clone();
+        let stale_token = token(&contract, "cancelled");
+        let task =
+            tokio::spawn(
+                async move { journal.record_token(&original_contract, &stale_token).await },
+            );
+        reached.notified().await;
+        task.abort();
+        let result = task.await;
+        assert!(matches!(result, Err(error) if error.is_cancelled()));
+        let journal = Journal::open(root.path()).await.unwrap();
+        assert_eq!(
+            journal.buyer(&contract.id).await.unwrap().unwrap().phase,
+            BuyerPhase::Intent
+        );
+        let current_token = token(&contract, "current");
+        journal
+            .record_token(&contract, &current_token)
+            .await
+            .unwrap();
+        // Resuming the old hook cannot enqueue a stale rename: its future and
+        // private temporary file were already dropped before the lock released.
+        resume.notify_one();
+        tokio::task::yield_now().await;
+        assert_eq!(
+            journal.buyer(&contract.id).await.unwrap().unwrap().token(),
+            Some(current_token.as_str())
+        );
+        let mut directory = fs::read_dir(&journal.directory).await.unwrap();
+        while let Some(entry) = directory.next_entry().await.unwrap() {
+            assert!(!entry.file_name().to_string_lossy().ends_with(".tmp"));
+        }
+    }
+}

core/archipelago/src/wallet/ecash.rs

@@ -88,6 +88,10 @@ pub struct WalletState {
     #[serde(default)]
     pub mint_url: String,
 
+    /// Durable receive commit ownership; never prune with transaction history.
+    #[serde(default, skip_serializing_if = "std::collections::BTreeMap::is_empty")]
+    pub(super) receive_commits: std::collections::BTreeMap<String, String>,
+
     // ── Legacy compatibility ──
     // Old wallet format had a `tokens` field. If present during deserialization,
     // we migrate to proofs. This field is never written.
@@ -239,7 +243,7 @@ pub enum EcashNetwork {
 }
 
 impl EcashNetwork {
-    fn wallet_file(&self) -> &'static str {
+    pub(super) fn wallet_file(&self) -> &'static str {
         match self {
             Self::Mainnet => WALLET_FILE,
             Self::Testnet => "wallet/ecash.testnet.json",
@@ -367,13 +371,11 @@ async fn write_file_atomically(path: &Path, content: &str) -> Result<()> {
         .await
         .context("Failed to flush wallet file")?;
     drop(file);
-    fs::rename(&tmp.0, path)
-        .await
-        .context("Failed to replace wallet file")?;
-    fs::File::open(parent)
-        .await?
+    // Complete the commit without yielding: async filesystem work must not
+    // rename an old purse after cancellation releases the mutation guard.
+    std::fs::rename(&tmp.0, path).context("Failed to replace wallet file")?;
+    std::fs::File::open(parent)?
         .sync_all()
-        .await
         .context("Failed to flush wallet directory")?;
     Ok(())
 }
@@ -856,7 +858,9 @@ pub async fn send_token_recoverable(
             // Establish recovery support before reserving or spending inputs.
             // Newly derived outputs must not already exist at the mint.
             let existing = client.restore_prepared_swap(&prepared).await.map_err(|_| {
-                anyhow::anyhow!("The mint could not verify payment recovery support; no funds spent")
+                anyhow::anyhow!(
+                    "The mint could not verify payment recovery support; no funds spent"
+                )
             })?;
             anyhow::ensure!(
                 existing.is_none(),
@@ -1366,6 +1370,150 @@ fn target_liquidity_score(liq: &SwapLiquidity, to_mint: &str) -> i64 {
         .sum()
 }
 
+/// Settle one caller-owned incoming token without losing an ambiguous mint
+/// response. Persist and reuse operation_id/context_hash for the same purchase.
+/// This is not a delivery receipt, refund authorization, or purchase protocol.
+pub async fn receive_token_recoverable(
+    data_dir: &Path,
+    operation_id: &str,
+    network: EcashNetwork,
+    mint_url: &str,
+    token_str: &str,
+    minimum_sats: u64,
+    context_hash: &str,
+) -> Result<u64> {
+    use super::receive_journal::{canonical_mint, Binding, Journal, Phase};
+    use sha2::{Digest, Sha256};
+    let held = super::mutation::guard(data_dir).await?;
+    anyhow::ensure!(
+        load_network(data_dir).await? == network,
+        "Switch back to the settlement's original network"
+    );
+    anyhow::ensure!(
+        token_str.len() <= 512 * 1024,
+        "Incoming token exceeds settlement size limit"
+    );
+    let binding = Binding {
+        id: operation_id.into(),
+        network,
+        mint_url: canonical_mint(mint_url)?,
+        token_hash: hex::encode(Sha256::digest(token_str.as_bytes())),
+        context_hash: context_hash.into(),
+        minimum_sats,
+    };
+    binding.validate()?;
+    let journal = Journal::new(&held);
+    let previous = journal.load(operation_id).await?;
+    let recovering = previous.is_some();
+    let record = if let Some(record) = previous {
+        anyhow::ensure!(
+            record.binding == binding,
+            "Settlement operation terms changed; do not redeem again"
+        );
+        record
+    } else {
+        let token = CashuToken::deserialize(token_str)
+            .map_err(|_| anyhow::anyhow!("Invalid incoming settlement token"))?;
+        anyhow::ensure!(
+            token.unit.as_deref().unwrap_or("sat") == "sat" && token.token.len() == 1,
+            "Settlement requires one sat-denominated mint"
+        );
+        let entry = &token.token[0];
+        anyhow::ensure!(
+            canonical_mint(&entry.mint)? == binding.mint_url,
+            "Incoming token mint does not match settlement terms"
+        );
+        let amount = entry
+            .proofs
+            .iter()
+            .try_fold(0u64, |sum, proof| sum.checked_add(proof.amount))
+            .context("Incoming amount overflow")?;
+        anyhow::ensure!(
+            amount >= minimum_sats
+                && entry
+                    .proofs
+                    .iter()
+                    .all(|proof| proof.amount.is_power_of_two()
+                        && !proof.secret.is_empty()
+                        && proof.c_as_pubkey().is_ok()),
+            "Invalid incoming settlement proofs or amount"
+        );
+        journal
+            .ensure_unclaimed(&binding.mint_url, &entry.proofs, Some(operation_id))
+            .await?;
+        let accepted = load_accepted_mints(data_dir).await?;
+        anyhow::ensure!(accepted.mints.iter().any(|mint| canonical_mint(mint).ok().as_deref() == Some(binding.mint_url.as_str())), "Settlement mint is not accepted");
+        let wallet = load_wallet(data_dir).await?;
+        anyhow::ensure!(
+            !entry
+                .proofs
+                .iter()
+                .any(|proof| wallet.proofs.iter().any(|stored| !stored.spent
+                    && canonical_mint(&stored.mint_url).ok().as_deref()
+                        == Some(binding.mint_url.as_str())
+                    && stored.proof.secret == proof.secret)),
+            "Incoming settlement overlaps existing wallet funds"
+        );
+        anyhow::ensure!(
+            !wallet
+                .receive_commits
+                .contains_key(&format!("received:{operation_id}")),
+            "Settlement marker exists without its recovery record; manual recovery required"
+        );
+        let client = mint_client(data_dir, &binding.mint_url).await?;
+        let prepared = client.prepare_swap_at_least(&entry.proofs, &amount_to_denominations(amount), minimum_sats).await
+            .map_err(|_| anyhow::anyhow!("Could not prepare a recoverable settlement covering the agreed net price; no proofs redeemed"))?;
+        let existing = client.restore_prepared_swap(&prepared).await.map_err(|_| {
+            anyhow::anyhow!(
+                "The mint could not verify settlement recovery support; no proofs redeemed"
+            )
+        })?;
+        anyhow::ensure!(
+            existing.is_none(),
+            "New settlement outputs already exist; no proofs redeemed"
+        );
+        journal.prepare(binding.clone(), prepared).await?
+    };
+    if !matches!(record.phase, Phase::Prepared) {
+        return journal.commit_wallet(&binding).await;
+    }
+    let client = MintClient::new(&binding.mint_url)?;
+    let restored = if recovering {
+        client
+            .restore_prepared_swap(&record.request)
+            .await
+            .map_err(|_| {
+                anyhow::anyhow!(
+                    "Could not recover this settlement yet; retain the original operation"
+                )
+            })?
+    } else {
+        None
+    };
+    let result = if let Some(result) = restored {
+        result
+    } else {
+        if recovering {
+            let states = client.check_state(record.request.inputs()).await
+                .map_err(|_| anyhow::anyhow!("Could not verify incoming settlement inputs; do not redeem or refund again"))?;
+            anyhow::ensure!(
+                states.iter().all(|state| state.state == "UNSPENT"),
+                "Incoming settlement remains pending at the mint; do not redeem or refund again"
+            );
+        }
+        client
+            .execute_prepared_swap(&record.request)
+            .await
+            .map_err(|_| {
+                anyhow::anyhow!(
+                    "The mint did not confirm settlement; retry this same operation to recover it"
+                )
+            })?
+    };
+    journal.record_result(&binding, result.new_proofs).await?;
+    journal.commit_wallet(&binding).await
+}
+
 /// Receive a Cashu token from a peer — swaps proofs at the mint for fresh ones.
 pub async fn receive_token(data_dir: &Path, token_str: &str) -> Result<u64> {
     let _mutation = super::mutation::guard(data_dir).await?;
@@ -1375,6 +1523,11 @@ pub async fn receive_token(data_dir: &Path, token_str: &str) -> Result<u64> {
     }
 
     let token = CashuToken::deserialize(token_str)?;
+    for entry in &token.token {
+        super::receive_journal::Journal::new(&_mutation)
+            .ensure_unclaimed(&entry.mint, &entry.proofs, None)
+            .await?;
+    }
     let total_amount = token.total_amount();
 
     if total_amount == 0 {
@@ -1530,6 +1683,9 @@ pub async fn verify_and_receive_payment(
         [entry] => entry,
         _ => anyhow::bail!("Use a single-mint token for this payment"),
     };
+    super::receive_journal::Journal::new(&_mutation)
+        .ensure_unclaimed(&entry.mint, &entry.proofs, None)
+        .await?;
     let total = entry
         .proofs
         .iter()
@@ -1616,6 +1772,9 @@ pub struct RestoreOutcome {
 /// time to press this button is when something already looks wrong.
 pub async fn restore_from_seed(data_dir: &Path, mint_url: &str) -> Result<RestoreOutcome> {
     let _mutation = super::mutation::guard(data_dir).await?;
+    super::receive_journal::Journal::new(&_mutation)
+        .ensure_restore_allowed(load_network(data_dir).await?, mint_url)
+        .await?;
     let outgoing = super::send_journal::Journal::new(&_mutation)
         .restore_exclusions(load_network(data_dir).await?, mint_url)
         .await?;

core/archipelago/src/wallet/mod.rs

@@ -12,3 +12,4 @@ mod mutation;
 pub mod nut13;
 pub mod profits;
 mod send_journal;
+mod receive_journal;

core/archipelago/src/wallet/payment_tests.rs

@@ -729,16 +729,30 @@ async fn recoverable_send_rejects_broken_recovery_before_reserving_or_spending()
     let id = uuid::Uuid::new_v4().to_string();
     let context = "ab".repeat(32);
     let error = send_token_recoverable(
-        root.path(), &id, EcashNetwork::Mainnet, &mint.url, 4, &context,
-    ).await.unwrap_err();
+        root.path(),
+        &id,
+        EcashNetwork::Mainnet,
+        &mint.url,
+        4,
+        &context,
+    )
+    .await
+    .unwrap_err();
     assert!(error.to_string().contains("recovery support"));
     assert_eq!(load_wallet(root.path()).await.unwrap().balance(), 8);
     assert!(mint.requests.lock().unwrap().is_empty());
     // Once the mint responds correctly the same unspent operation can proceed.
     *mint.restore_reply.lock().unwrap() = None;
     assert!(send_token_recoverable(
-        root.path(), &id, EcashNetwork::Mainnet, &mint.url, 4, &context,
-    ).await.is_ok());
+        root.path(),
+        &id,
+        EcashNetwork::Mainnet,
+        &mint.url,
+        4,
+        &context,
+    )
+    .await
+    .is_ok());
     assert_eq!(mint.requests.lock().unwrap().len(), 1);
     assert_eq!(load_wallet(root.path()).await.unwrap().balance(), 4);
 }
@@ -1016,3 +1030,547 @@ async fn incomplete_duplicate_or_unknown_restoration_never_completes_a_swap() {
     }
     assert!(mint.requests.lock().unwrap().is_empty());
 }
+
+#[tokio::test]
+async fn recoverable_receive_lost_reply_claims_inputs_and_recovers_once() {
+    let mint = Mint::start(0, None).await;
+    let root = mint.wallet().await;
+    let incoming = CashuToken::new(&mint.url, vec![proof(V2, 8), proof(ACTIVE, 4)]);
+    let token = incoming.serialize_v4().unwrap();
+    let id = uuid::Uuid::new_v4().to_string();
+    let context = "ab".repeat(32);
+    mint.lose_swap_reply
+        .store(true, std::sync::atomic::Ordering::SeqCst);
+    assert!(receive_token_recoverable(
+        root.path(),
+        &id,
+        EcashNetwork::Mainnet,
+        &mint.url,
+        &token,
+        12,
+        &context
+    )
+    .await
+    .is_err());
+    assert_eq!(mint.requests.lock().unwrap().len(), 1);
+    assert_eq!(load_wallet(root.path()).await.unwrap().balance(), 0);
+    assert!(restore_from_seed(root.path(), &mint.url)
+        .await
+        .unwrap_err()
+        .to_string()
+        .contains("pending receipts"));
+    for duplicate in [
+        token.clone(),
+        incoming.serialize().unwrap(),
+        CashuToken::new(&mint.url, vec![proof(ACTIVE, 4), proof(ACTIVE, 2)])
+            .serialize()
+            .unwrap(),
+    ] {
+        let other = uuid::Uuid::new_v4().to_string();
+        assert!(receive_token_recoverable(
+            root.path(),
+            &other,
+            EcashNetwork::Mainnet,
+            &mint.url,
+            &duplicate,
+            1,
+            &context
+        )
+        .await
+        .unwrap_err()
+        .to_string()
+        .contains("another settlement"));
+    }
+    assert!(receive_token(root.path(), &token)
+        .await
+        .unwrap_err()
+        .to_string()
+        .contains("another settlement"));
+    *mint.restore_reply.lock().unwrap() = Some(json!({"outputs":[],"signatures":[]}));
+    assert!(receive_token_recoverable(
+        root.path(),
+        &id,
+        EcashNetwork::Mainnet,
+        &mint.url,
+        &token,
+        12,
+        &context
+    )
+    .await
+    .unwrap_err()
+    .to_string()
+    .contains("pending at the mint"));
+    assert_eq!(mint.requests.lock().unwrap().len(), 1);
+    *mint.restore_reply.lock().unwrap() = None;
+    assert_eq!(
+        receive_token_recoverable(
+            root.path(),
+            &id,
+            EcashNetwork::Mainnet,
+            &mint.url,
+            &token,
+            12,
+            &context
+        )
+        .await
+        .unwrap(),
+        12
+    );
+    let wallet = load_wallet(root.path()).await.unwrap();
+    assert_eq!(wallet.balance(), 12);
+    assert_eq!(wallet.transactions.len(), 1);
+    assert_eq!(wallet.transactions[0].id, format!("received:{id}"));
+    assert_eq!(wallet.receive_commits.len(), 1);
+    for output in &wallet.proofs {
+        assert_eq!(
+            output.proof.c,
+            signed_point(bdhke::hash_to_curve(output.proof.secret.as_bytes()).unwrap())
+        );
+        assert!(!incoming.token[0]
+            .proofs
+            .iter()
+            .any(|input| input.secret == output.proof.secret));
+    }
+    let before = std::fs::read(root.path().join("wallet/ecash.json")).unwrap();
+    assert_eq!(
+        receive_token_recoverable(
+            root.path(),
+            &id,
+            EcashNetwork::Mainnet,
+            &mint.url,
+            &token,
+            12,
+            &context
+        )
+        .await
+        .unwrap(),
+        12
+    );
+    assert_eq!(
+        std::fs::read(root.path().join("wallet/ecash.json")).unwrap(),
+        before
+    );
+    assert_eq!(mint.requests.lock().unwrap().len(), 1);
+    for (network, price, terms) in [
+        (EcashNetwork::Testnet, 12, context.clone()),
+        (EcashNetwork::Mainnet, 11, context.clone()),
+        (EcashNetwork::Mainnet, 12, "cd".repeat(32)),
+    ] {
+        assert!(receive_token_recoverable(
+            root.path(),
+            &id,
+            network,
+            &mint.url,
+            &token,
+            price,
+            &terms
+        )
+        .await
+        .is_err());
+    }
+    // Completed receipts remain valid without re-crediting a pruned wallet.
+    std::fs::remove_file(root.path().join("wallet/ecash.json")).unwrap();
+    assert_eq!(
+        receive_token_recoverable(
+            root.path(),
+            &id,
+            EcashNetwork::Mainnet,
+            &mint.url,
+            &token,
+            12,
+            &context
+        )
+        .await
+        .unwrap(),
+        12
+    );
+    assert!(!root.path().join("wallet/ecash.json").exists());
+    assert!(receive_token_recoverable(
+        root.path(),
+        &uuid::Uuid::new_v4().to_string(),
+        EcashNetwork::Mainnet,
+        &mint.url,
+        &incoming.serialize().unwrap(),
+        12,
+        &context
+    )
+    .await
+    .is_err());
+}
+
+#[tokio::test]
+async fn recoverable_receive_recovery_support_fees_and_terms_fail_before_spend() {
+    let mint = Mint::start(1000, None).await;
+    let root = mint.wallet().await;
+    let token = CashuToken::new(&mint.url, vec![proof(ACTIVE, 8)])
+        .serialize()
+        .unwrap();
+    let id = uuid::Uuid::new_v4().to_string();
+    let context = "ab".repeat(32);
+    assert!(receive_token_recoverable(
+        root.path(),
+        &id,
+        EcashNetwork::Mainnet,
+        &mint.url,
+        &token,
+        8,
+        &context
+    )
+    .await
+    .is_err());
+    *mint.restore_reply.lock().unwrap() = Some(json!({}));
+    assert!(receive_token_recoverable(
+        root.path(),
+        &id,
+        EcashNetwork::Mainnet,
+        &mint.url,
+        &token,
+        7,
+        &context
+    )
+    .await
+    .unwrap_err()
+    .to_string()
+    .contains("recovery support"));
+    assert!(mint.requests.lock().unwrap().is_empty());
+    assert!(!root.path().join("wallet/receive-operations").exists());
+    assert_eq!(load_wallet(root.path()).await.unwrap().balance(), 0);
+    let mut foreign = CashuToken::new(&mint.url, vec![proof(ACTIVE, 8)]);
+    foreign.unit = Some("usd".into());
+    assert!(receive_token_recoverable(
+        root.path(),
+        &id,
+        EcashNetwork::Mainnet,
+        &mint.url,
+        &foreign.serialize().unwrap(),
+        7,
+        &context
+    )
+    .await
+    .is_err());
+    foreign.unit = Some("sat".into());
+    foreign.token.push(foreign.token[0].clone());
+    assert!(receive_token_recoverable(
+        root.path(),
+        &id,
+        EcashNetwork::Mainnet,
+        &mint.url,
+        &foreign.serialize().unwrap(),
+        7,
+        &context
+    )
+    .await
+    .is_err());
+    *mint.restore_reply.lock().unwrap() = None;
+    assert_eq!(
+        receive_token_recoverable(
+            root.path(),
+            &id,
+            EcashNetwork::Mainnet,
+            &format!("{}/", mint.url),
+            &token,
+            7,
+            &context
+        )
+        .await
+        .unwrap(),
+        7
+    );
+    assert_eq!(load_wallet(root.path()).await.unwrap().balance(), 7);
+    assert_eq!(mint.requests.lock().unwrap().len(), 1);
+}
+
+fn rewrite_receive_phase(root: &std::path::Path, id: &str, phase: Value) {
+    use sha2::{Digest, Sha256};
+    let path = root.join(format!("wallet/receive-operations/{id}.json"));
+    let mut envelope: Value = serde_json::from_slice(&std::fs::read(&path).unwrap()).unwrap();
+    let mut record: Value = serde_json::from_str(envelope["payload"].as_str().unwrap()).unwrap();
+    record["phase"] = phase;
+    let payload = serde_json::to_string(&record).unwrap();
+    envelope["checksum"] = json!(hex::encode(Sha256::digest(payload.as_bytes())));
+    envelope["payload"] = json!(payload);
+    std::fs::write(path, serde_json::to_vec(&envelope).unwrap()).unwrap();
+}
+
+#[tokio::test]
+async fn recoverable_receive_commit_boundaries_survive_history_pruning_without_recredit() {
+    use sha2::{Digest, Sha256};
+    let mint = Mint::start(0, None).await;
+    let root = mint.wallet().await;
+    let token = CashuToken::new(&mint.url, vec![proof(ACTIVE, 8)])
+        .serialize()
+        .unwrap();
+    let id = uuid::Uuid::new_v4().to_string();
+    let context = "ab".repeat(32);
+    assert_eq!(
+        receive_token_recoverable(
+            root.path(),
+            &id,
+            EcashNetwork::Mainnet,
+            &mint.url,
+            &token,
+            8,
+            &context
+        )
+        .await
+        .unwrap(),
+        8
+    );
+    let mut wallet = load_wallet(root.path()).await.unwrap();
+    let proofs: Vec<_> = wallet.proofs.iter().map(|p| p.proof.clone()).collect();
+    let committing =
+        json!({"Committing":{"proofs":proofs,"before":hex::encode(Sha256::digest(b"missing"))}});
+    let journal_path = root
+        .path()
+        .join(format!("wallet/receive-operations/{id}.json"));
+    let committed_record = std::fs::read(&journal_path).unwrap();
+    // Crash after purse save but before phase save; unrelated history pruning
+    // preserves the durable marker and must not restore already-spent outputs.
+    rewrite_receive_phase(root.path(), &id, committing.clone());
+    wallet.transactions.clear();
+    wallet.proofs.clear();
+    save_wallet(root.path(), &wallet).await.unwrap();
+    let purse = std::fs::read(root.path().join("wallet/ecash.json")).unwrap();
+    assert_eq!(
+        receive_token_recoverable(
+            root.path(),
+            &id,
+            EcashNetwork::Mainnet,
+            &mint.url,
+            &token,
+            8,
+            &context
+        )
+        .await
+        .unwrap(),
+        8
+    );
+    assert_eq!(
+        std::fs::read(root.path().join("wallet/ecash.json")).unwrap(),
+        purse
+    );
+    // Old writers dropping the marker cause a recovery hold, never a guessed credit.
+    rewrite_receive_phase(root.path(), &id, committing.clone());
+    wallet.receive_commits.clear();
+    save_wallet(root.path(), &wallet).await.unwrap();
+    assert!(receive_token_recoverable(
+        root.path(),
+        &id,
+        EcashNetwork::Mainnet,
+        &mint.url,
+        &token,
+        8,
+        &context
+    )
+    .await
+    .unwrap_err()
+    .to_string()
+    .contains("manual recovery"));
+    assert_eq!(load_wallet(root.path()).await.unwrap().balance(), 0);
+    // Crash before the purse save: exact absent pre-image authorizes first commit.
+    std::fs::remove_file(root.path().join("wallet/ecash.json")).unwrap();
+    assert_eq!(
+        receive_token_recoverable(
+            root.path(),
+            &id,
+            EcashNetwork::Mainnet,
+            &mint.url,
+            &token,
+            8,
+            &context
+        )
+        .await
+        .unwrap(),
+        8
+    );
+    assert_eq!(load_wallet(root.path()).await.unwrap().balance(), 8);
+    assert_eq!(mint.requests.lock().unwrap().len(), 1);
+    // Completed receipt survives a missing purse without rebuilding its proofs.
+    std::fs::write(journal_path, committed_record).unwrap();
+    std::fs::remove_file(root.path().join("wallet/ecash.json")).unwrap();
+    assert_eq!(
+        receive_token_recoverable(
+            root.path(),
+            &id,
+            EcashNetwork::Mainnet,
+            &mint.url,
+            &token,
+            8,
+            &context
+        )
+        .await
+        .unwrap(),
+        8
+    );
+    assert!(!root.path().join("wallet/ecash.json").exists());
+}
+
+#[tokio::test]
+async fn recoverable_receive_corrupt_claims_and_write_failures_preserve_funds() {
+    let mint = Mint::start(0, None).await;
+    let root = mint.wallet().await;
+    let token = CashuToken::new(&mint.url, vec![proof(ACTIVE, 8)])
+        .serialize()
+        .unwrap();
+    let id = uuid::Uuid::new_v4().to_string();
+    let context = "ab".repeat(32);
+    // A directory at the target blocks the private journal replacement before POST.
+    let path = root
+        .path()
+        .join(format!("wallet/receive-operations/{id}.json"));
+    std::fs::create_dir_all(&path).unwrap();
+    assert!(receive_token_recoverable(
+        root.path(),
+        &id,
+        EcashNetwork::Mainnet,
+        &mint.url,
+        &token,
+        8,
+        &context
+    )
+    .await
+    .is_err());
+    assert!(mint.requests.lock().unwrap().is_empty());
+    std::fs::remove_dir(&path).unwrap();
+    mint.lose_swap_reply
+        .store(true, std::sync::atomic::Ordering::SeqCst);
+    assert!(receive_token_recoverable(
+        root.path(),
+        &id,
+        EcashNetwork::Mainnet,
+        &mint.url,
+        &token,
+        8,
+        &context
+    )
+    .await
+    .is_err());
+    let saved = std::fs::read(&path).unwrap();
+    #[cfg(unix)]
+    {
+        use std::os::unix::fs::PermissionsExt;
+        assert_eq!(
+            std::fs::metadata(&path).unwrap().permissions().mode() & 0o777,
+            0o600
+        );
+        assert_eq!(
+            std::fs::metadata(path.parent().unwrap())
+                .unwrap()
+                .permissions()
+                .mode()
+                & 0o777,
+            0o700
+        );
+    }
+    std::fs::write(&path, b"damaged").unwrap();
+    assert!(receive_token_recoverable(
+        root.path(),
+        &id,
+        EcashNetwork::Mainnet,
+        &mint.url,
+        &token,
+        8,
+        &context
+    )
+    .await
+    .is_err());
+    assert!(receive_token_recoverable(
+        root.path(),
+        &uuid::Uuid::new_v4().to_string(),
+        EcashNetwork::Mainnet,
+        &mint.url,
+        &token,
+        8,
+        &context
+    )
+    .await
+    .is_err());
+    assert!(receive_token(root.path(), &token).await.is_err());
+    assert_eq!(mint.requests.lock().unwrap().len(), 1);
+    std::fs::write(&path, saved).unwrap();
+    assert_eq!(
+        receive_token_recoverable(
+            root.path(),
+            &id,
+            EcashNetwork::Mainnet,
+            &mint.url,
+            &token,
+            8,
+            &context
+        )
+        .await
+        .unwrap(),
+        8
+    );
+}
+
+#[tokio::test]
+async fn recoverable_receive_unspent_retry_reuses_exact_request_and_waits_for_verified_state() {
+    let mint = Mint::start(0, Some(503)).await;
+    let root = mint.wallet().await;
+    let token = CashuToken::new(&mint.url, vec![proof(ACTIVE, 8)])
+        .serialize()
+        .unwrap();
+    let id = uuid::Uuid::new_v4().to_string();
+    let context = "ab".repeat(32);
+    assert!(receive_token_recoverable(
+        root.path(),
+        &id,
+        EcashNetwork::Mainnet,
+        &mint.url,
+        &token,
+        8,
+        &context
+    )
+    .await
+    .is_err());
+    assert_eq!(mint.requests.lock().unwrap().len(), 1);
+    // Legacy paid-content redemption must respect claims even while mint inputs
+    // remain unspent; otherwise it could steal the pending operation's proofs.
+    assert!(verify_and_receive_payment(root.path(), &token, 8)
+        .await
+        .unwrap_err()
+        .to_string()
+        .contains("another settlement"));
+    assert_eq!(mint.requests.lock().unwrap().len(), 1);
+    let original = mint.requests.lock().unwrap()[0].clone();
+    assert_eq!(load_wallet(root.path()).await.unwrap().balance(), 0);
+    // An empty state response must not authorize a second POST.
+    *mint.state_reply.lock().unwrap() = Some(json!({"states":[]}));
+    mint.failure.store(0, std::sync::atomic::Ordering::SeqCst);
+    assert!(receive_token_recoverable(
+        root.path(),
+        &id,
+        EcashNetwork::Mainnet,
+        &mint.url,
+        &token,
+        8,
+        &context
+    )
+    .await
+    .is_err());
+    assert_eq!(mint.requests.lock().unwrap().len(), 1);
+    *mint.state_reply.lock().unwrap() = None;
+    assert_eq!(
+        receive_token_recoverable(
+            root.path(),
+            &id,
+            EcashNetwork::Mainnet,
+            &mint.url,
+            &token,
+            8,
+            &context
+        )
+        .await
+        .unwrap(),
+        8
+    );
+    let requests = mint.requests.lock().unwrap();
+    assert_eq!(requests.len(), 2);
+    assert_eq!(requests[1], original);
+    drop(requests);
+    let wallet = load_wallet(root.path()).await.unwrap();
+    assert_eq!(wallet.balance(), 8);
+    assert_eq!(wallet.transactions.len(), 1);
+    assert_eq!(wallet.receive_commits.len(), 1);
+}

core/archipelago/src/wallet/receive_journal.rs

@@ -0,0 +1,456 @@
+//! Private incoming-proof claims and recoverable settlement. Incoming bearer
+//! proofs never become spendable locally: only fresh, saved swap outputs do.
+use super::{
+    cashu::Proof, ecash::EcashNetwork, mint_client::PreparedSwap, mutation::WalletMutation,
+};
+use anyhow::{Context, Result};
+use serde::{Deserialize, Serialize};
+use sha2::{Digest, Sha256};
+use std::{collections::HashSet, path::PathBuf};
+use tokio::{
+    fs,
+    io::{AsyncReadExt, AsyncWriteExt},
+};
+const MAX_BYTES: u64 = 1024 * 1024;
+
+pub(super) fn canonical_mint(value: &str) -> Result<String> {
+    let url = reqwest::Url::parse(value).context("Invalid settlement mint")?;
+    anyhow::ensure!(
+        matches!(url.scheme(), "http" | "https")
+            && url.host_str().is_some()
+            && url.username().is_empty()
+            && url.password().is_none()
+            && url.query().is_none()
+            && url.fragment().is_none(),
+        "Invalid settlement mint URL"
+    );
+    Ok(url.to_string().trim_end_matches('/').to_owned())
+}
+fn digest(bytes: &[u8]) -> String {
+    hex::encode(Sha256::digest(bytes))
+}
+fn is_hash(value: &str) -> bool {
+    value.len() == 64
+        && value
+            .bytes()
+            .all(|c| c.is_ascii_digit() || (b'a'..=b'f').contains(&c))
+}
+#[derive(Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(deny_unknown_fields)]
+pub(super) struct Binding {
+    pub id: String,
+    pub network: EcashNetwork,
+    pub mint_url: String,
+    pub token_hash: String,
+    pub context_hash: String,
+    pub minimum_sats: u64,
+}
+impl Binding {
+    pub fn validate(&self) -> Result<()> {
+        anyhow::ensure!(
+            uuid::Uuid::parse_str(&self.id)
+                .ok()
+                .is_some_and(|id| id.to_string() == self.id),
+            "Invalid settlement identifier"
+        );
+        anyhow::ensure!(
+            self.minimum_sats > 0 && is_hash(&self.token_hash) && is_hash(&self.context_hash),
+            "Invalid settlement terms"
+        );
+        anyhow::ensure!(
+            canonical_mint(&self.mint_url)? == self.mint_url,
+            "Settlement mint is not canonical"
+        );
+        Ok(())
+    }
+    fn history_id(&self) -> String {
+        format!("received:{}", self.id)
+    }
+}
+#[derive(Clone, Serialize, Deserialize)]
+pub(super) enum Phase {
+    Prepared,
+    Result(Vec<Proof>),
+    Committing {
+        proofs: Vec<Proof>,
+        before: String,
+    },
+    Committed {
+        amount_sats: u64,
+        commitment: String,
+    },
+}
+#[derive(Clone, Serialize, Deserialize)]
+#[serde(deny_unknown_fields)]
+pub(super) struct Record {
+    pub binding: Binding,
+    pub request: PreparedSwap,
+    pub phase: Phase,
+}
+impl std::fmt::Debug for Record {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("ReceiveJournalRecord")
+            .field("id", &self.binding.id)
+            .finish_non_exhaustive()
+    }
+}
+#[derive(Serialize, Deserialize)]
+#[serde(deny_unknown_fields)]
+struct Envelope {
+    version: u8,
+    payload: String,
+    checksum: String,
+}
+pub(super) struct Journal<'a> {
+    guard: &'a WalletMutation,
+}
+impl<'a> Journal<'a> {
+    pub fn new(guard: &'a WalletMutation) -> Self {
+        Self { guard }
+    }
+    fn directory(&self) -> PathBuf {
+        self.guard.data_dir.join("wallet/receive-operations")
+    }
+    fn path(&self, id: &str) -> Result<PathBuf> {
+        let uuid = uuid::Uuid::parse_str(id).context("Invalid settlement identifier")?;
+        anyhow::ensure!(
+            uuid.to_string() == id,
+            "Settlement identifier is not canonical"
+        );
+        Ok(self.directory().join(format!("{uuid}.json")))
+    }
+    fn validate(record: &Record) -> Result<()> {
+        record.binding.validate()?;
+        record.request.validate_for_mint(&record.binding.mint_url)?;
+        anyhow::ensure!(
+            record.request.covers_payment(record.binding.minimum_sats),
+            "Settlement does not cover the agreed price"
+        );
+        match &record.phase {
+            Phase::Prepared => (),
+            Phase::Result(proofs) => {
+                Self::validate_result(record, proofs)?;
+            }
+            Phase::Committing { proofs, before } => {
+                Self::validate_result(record, proofs)?;
+                anyhow::ensure!(is_hash(before), "Invalid settlement purse boundary");
+            }
+            Phase::Committed {
+                amount_sats,
+                commitment,
+            } => {
+                anyhow::ensure!(
+                    *amount_sats >= record.binding.minimum_sats
+                        && record.request.covers_payment(*amount_sats)
+                        && (amount_sats
+                            .checked_add(1)
+                            .is_none_or(|next| !record.request.covers_payment(next)))
+                        && is_hash(commitment),
+                    "Invalid committed settlement"
+                );
+            }
+        }
+        Ok(())
+    }
+    fn validate_result(record: &Record, proofs: &[Proof]) -> Result<u64> {
+        record.request.validate_result_proofs(proofs)?;
+        let amount = proofs
+            .iter()
+            .try_fold(0u64, |sum, proof| sum.checked_add(proof.amount))
+            .context("Settlement amount overflow")?;
+        anyhow::ensure!(
+            amount >= record.binding.minimum_sats,
+            "Settlement does not cover the agreed price"
+        );
+        Ok(amount)
+    }
+    pub async fn load(&self, id: &str) -> Result<Option<Record>> {
+        let mut options = fs::OpenOptions::new();
+        options.read(true);
+        #[cfg(unix)]
+        options.custom_flags(libc::O_NOFOLLOW | libc::O_NONBLOCK);
+        let file = match options.open(self.path(id)?).await {
+            Ok(file) => file,
+            Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(None),
+            Err(e) => return Err(e).context("Could not read settlement recovery"),
+        };
+        anyhow::ensure!(
+            file.metadata().await?.is_file(),
+            "Settlement record is not a regular file"
+        );
+        let mut bytes = Vec::new();
+        file.take(MAX_BYTES + 1).read_to_end(&mut bytes).await?;
+        anyhow::ensure!(
+            bytes.len() as u64 <= MAX_BYTES,
+            "Settlement recovery exceeds its size limit"
+        );
+        let envelope: Envelope = serde_json::from_slice(&bytes)
+            .map_err(|_| anyhow::anyhow!("Settlement recovery is damaged; do not redeem again"))?;
+        anyhow::ensure!(
+            envelope.version == 1 && envelope.checksum == digest(envelope.payload.as_bytes()),
+            "Settlement recovery checksum/version failed"
+        );
+        let record: Record = serde_json::from_str(&envelope.payload)
+            .map_err(|_| anyhow::anyhow!("Settlement recovery contents are damaged"))?;
+        anyhow::ensure!(record.binding.id == id, "Settlement identity mismatch");
+        Self::validate(&record)?;
+        Ok(Some(record))
+    }
+    async fn records(&self) -> Result<Vec<Record>> {
+        let mut directory = match fs::read_dir(self.directory()).await {
+            Ok(directory) => directory,
+            Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(vec![]),
+            Err(e) => return Err(e).context("Cannot inspect incoming settlement claims"),
+        };
+        let mut records = Vec::new();
+        while let Some(entry) = directory.next_entry().await? {
+            let name = entry.file_name();
+            let name = name.to_str().context("Invalid settlement filename")?;
+            if name
+                .strip_prefix('.')
+                .and_then(|name| name.strip_suffix(".tmp"))
+                .is_some_and(|id| uuid::Uuid::parse_str(id).is_ok())
+            {
+                continue;
+            }
+            let id = name
+                .strip_suffix(".json")
+                .context("Unexpected settlement recovery entry")?;
+            records.push(
+                self.load(id)
+                    .await?
+                    .context("Settlement recovery disappeared")?,
+            );
+        }
+        Ok(records)
+    }
+    /// Claims are based on proof secrets, not token serialization/keyset aliases.
+    /// A partial overlap must not be treated as a new payment or a refund.
+    pub async fn ensure_unclaimed(
+        &self,
+        mint_url: &str,
+        inputs: &[Proof],
+        owner: Option<&str>,
+    ) -> Result<()> {
+        let mint = canonical_mint(mint_url)?;
+        let secrets: HashSet<_> = inputs
+            .iter()
+            .map(|proof| digest(proof.secret.as_bytes()))
+            .collect();
+        anyhow::ensure!(
+            secrets.len() == inputs.len() && !inputs.is_empty(),
+            "Duplicate or missing settlement inputs"
+        );
+        for record in self.records().await? {
+            if record.binding.mint_url != mint || owner == Some(record.binding.id.as_str()) {
+                continue;
+            }
+            anyhow::ensure!(!record.request.inputs().iter().any(|proof| secrets.contains(&digest(proof.secret.as_bytes()))), "These incoming proofs already belong to another settlement; resume its original operation");
+        }
+        Ok(())
+    }
+    pub async fn ensure_restore_allowed(
+        &self,
+        network: EcashNetwork,
+        mint_url: &str,
+    ) -> Result<()> {
+        let mint = canonical_mint(mint_url)?;
+        for record in self.records().await? {
+            if record.binding.network == network && record.binding.mint_url == mint {
+                anyhow::ensure!(
+                    matches!(record.phase, Phase::Committed { .. }),
+                    "Recover pending receipts before restoring this mint from the backup phrase"
+                );
+            }
+        }
+        Ok(())
+    }
+    pub async fn prepare(&self, binding: Binding, request: PreparedSwap) -> Result<Record> {
+        binding.validate()?;
+        if let Some(previous) = self.load(&binding.id).await? {
+            anyhow::ensure!(
+                previous.binding == binding,
+                "Settlement operation terms changed"
+            );
+            return Ok(previous);
+        }
+        self.ensure_unclaimed(&binding.mint_url, request.inputs(), Some(&binding.id))
+            .await?;
+        let record = Record {
+            binding,
+            request,
+            phase: Phase::Prepared,
+        };
+        Self::validate(&record)?;
+        self.write(&record).await?;
+        Ok(record)
+    }
+    async fn bound(&self, binding: &Binding) -> Result<Record> {
+        let record = self
+            .load(&binding.id)
+            .await?
+            .context("Settlement recovery is missing")?;
+        anyhow::ensure!(
+            &record.binding == binding,
+            "Settlement operation terms changed"
+        );
+        anyhow::ensure!(
+            super::ecash::load_network(&self.guard.data_dir).await? == binding.network,
+            "Switch back to the settlement's original network"
+        );
+        Ok(record)
+    }
+    pub async fn record_result(&self, binding: &Binding, proofs: Vec<Proof>) -> Result<()> {
+        let mut record = self.bound(binding).await?;
+        Self::validate_result(&record, &proofs)?;
+        match &record.phase {
+            Phase::Prepared => record.phase = Phase::Result(proofs),
+            Phase::Result(saved) | Phase::Committing { proofs: saved, .. } => {
+                anyhow::ensure!(
+                    serde_json::to_vec(saved)? == serde_json::to_vec(&proofs)?,
+                    "Settlement already has a different result"
+                );
+                return Ok(());
+            }
+            Phase::Committed { .. } => anyhow::bail!("Settlement already committed"),
+        }
+        self.write(&record).await
+    }
+    async fn purse_snapshot(&self, network: EcashNetwork) -> Result<String> {
+        // Hash exact on-disk bytes, not a reserialized WalletState. Absence and
+        // empty file are deliberately different (empty fails wallet loading).
+        match fs::read(self.guard.data_dir.join(network.wallet_file())).await {
+            Ok(bytes) => {
+                let mut tagged = b"existing:".to_vec();
+                tagged.extend(bytes);
+                Ok(digest(&tagged))
+            }
+            Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(digest(b"missing")),
+            Err(e) => Err(e).context("Cannot establish settlement purse boundary"),
+        }
+    }
+    pub async fn commit_wallet(&self, binding: &Binding) -> Result<u64> {
+        use super::ecash::{load_wallet, save_wallet, TransactionType};
+        let mut record = self.bound(binding).await?;
+        if let Phase::Committed { amount_sats, .. } = record.phase {
+            return Ok(amount_sats);
+        }
+        let mut wallet = load_wallet(&self.guard.data_dir).await?;
+        let (proofs, before) = match record.phase.clone() {
+            Phase::Prepared => anyhow::bail!("Settlement result is not durable yet"),
+            Phase::Result(proofs) => {
+                let before = self.purse_snapshot(binding.network).await?;
+                record.phase = Phase::Committing {
+                    proofs: proofs.clone(),
+                    before: before.clone(),
+                };
+                self.write(&record).await?;
+                (proofs, before)
+            }
+            Phase::Committing { proofs, before } => (proofs, before),
+            Phase::Committed { .. } => unreachable!(),
+        };
+        let amount = Self::validate_result(&record, &proofs)?;
+        let commitment = digest(&serde_json::to_vec(&(binding, amount, &proofs))?);
+        let history_id = binding.history_id();
+        if let Some(marker) = wallet.receive_commits.get(&history_id) {
+            anyhow::ensure!(
+                is_hash(marker) && *marker == commitment,
+                "Settlement purse marker does not match; manual recovery required"
+            );
+        } else {
+            anyhow::ensure!(
+                self.purse_snapshot(binding.network).await? == before,
+                "Settlement purse changed without its commit marker; manual recovery required"
+            );
+            anyhow::ensure!(
+                !wallet.transactions.iter().any(|tx| tx.id == history_id),
+                "Settlement history exists without its commit marker; manual recovery required"
+            );
+            anyhow::ensure!(
+                !proofs
+                    .iter()
+                    .any(|proof| wallet.proofs.iter().any(|stored| canonical_mint(
+                        &stored.mint_url
+                    )
+                    .ok()
+                    .as_deref()
+                        == Some(binding.mint_url.as_str())
+                        && stored.proof.secret == proof.secret)),
+                "Settlement output exists without its commit marker; manual recovery required"
+            );
+            wallet.add_proofs(&binding.mint_url, proofs);
+            wallet.record_tx(
+                TransactionType::Receive,
+                amount,
+                "Received ecash",
+                &binding.mint_url,
+                "",
+            );
+            wallet
+                .transactions
+                .last_mut()
+                .context("Could not record settlement history")?
+                .id = history_id.clone();
+            wallet
+                .receive_commits
+                .insert(history_id, commitment.clone());
+            save_wallet(&self.guard.data_dir, &wallet).await?;
+        }
+        record.phase = Phase::Committed {
+            amount_sats: amount,
+            commitment,
+        };
+        self.write(&record).await?;
+        Ok(amount)
+    }
+    async fn write(&self, record: &Record) -> Result<()> {
+        Self::validate(record)?;
+        let payload = serde_json::to_string(record)?;
+        let bytes = serde_json::to_vec(&Envelope {
+            version: 1,
+            checksum: digest(payload.as_bytes()),
+            payload,
+        })?;
+        anyhow::ensure!(
+            bytes.len() as u64 <= MAX_BYTES,
+            "Settlement recovery exceeds its size limit"
+        );
+        let parent = self.directory();
+        fs::create_dir_all(&parent).await?;
+        anyhow::ensure!(
+            fs::symlink_metadata(&parent).await?.is_dir(),
+            "Settlement directory is not a regular directory"
+        );
+        #[cfg(unix)]
+        {
+            use std::os::unix::fs::PermissionsExt;
+            fs::set_permissions(&parent, std::fs::Permissions::from_mode(0o700)).await?;
+        }
+        struct Temporary(PathBuf);
+        impl Drop for Temporary {
+            fn drop(&mut self) {
+                let _ = std::fs::remove_file(&self.0);
+            }
+        }
+        let temporary = Temporary(parent.join(format!(".{}.tmp", uuid::Uuid::new_v4())));
+        let mut options = fs::OpenOptions::new();
+        options.write(true).create_new(true);
+        #[cfg(unix)]
+        options.mode(0o600);
+        let mut file = options.open(&temporary.0).await?;
+        file.write_all(&bytes).await?;
+        file.sync_all().await?;
+        drop(file);
+        // No asynchronous commit may outlive the wallet mutation guard.
+        std::fs::rename(&temporary.0, self.path(&record.binding.id)?)?;
+        for directory in [
+            parent,
+            self.guard.data_dir.join("wallet"),
+            self.guard.data_dir.clone(),
+        ] {
+            std::fs::File::open(directory)?.sync_all()?;
+        }
+        Ok(())
+    }
+}

core/archipelago/src/wallet/send_journal.rs

@@ -979,14 +979,16 @@ impl<'a> Journal<'a> {
         file.write_all(&bytes).await?;
         file.sync_all().await?;
         drop(file);
-        fs::rename(&temporary.0, &path).await?;
+        // Keep the commit synchronous under the mutation guard: cancellation
+        // cannot leave a rename queued after a newer wallet mutation starts.
+        std::fs::rename(&temporary.0, &path)?;
         // Persist every new directory entry down from the existing node root.
         for directory in [
             parent.to_path_buf(),
             self.guard.data_dir.join("wallet"),
             self.guard.data_dir.clone(),
         ] {
-            fs::File::open(directory).await?.sync_all().await?;
+            std::fs::File::open(directory)?.sync_all()?;
         }
         Ok(())
     }