diff --git a/Cargo.lock b/Cargo.lock index 311ab94..e92493f 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -969,6 +969,19 @@ dependencies = [ "tracing-subscriber", ] +[[package]] +name = "deadcat-rfq-provider" +version = "0.1.0-alpha" +dependencies = [ + "elements", + "postcard", + "redb", + "serde", + "sha2 0.10.9", + "tempfile", + "thiserror 2.0.18", +] + [[package]] name = "deadcat-rpc" version = "0.1.0-alpha" diff --git a/Cargo.toml b/Cargo.toml index cdeef25..9376c9e 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -8,6 +8,7 @@ members = [ "crates/deadcat-iroh", "crates/deadcat-node", "crates/deadcat-cli", + "crates/deadcat-rfq-provider", ] [workspace.package] @@ -41,6 +42,7 @@ deadcat-contracts = { path = "crates/deadcat-contracts" } deadcat-client = { path = "crates/deadcat-client" } deadcat-rpc = { path = "crates/deadcat-rpc" } deadcat-iroh = { path = "crates/deadcat-iroh" } +deadcat-rfq-provider = { path = "crates/deadcat-rfq-provider" } # The CLI and Rust libraries must remain on the exact same smplx release. # smplx 0.0.9 caps simplicityhl at 0.6.x; Cargo.lock pins the newest diff --git a/README.md b/README.md index 622a72b..216ceaa 100644 --- a/README.md +++ b/README.md @@ -28,9 +28,14 @@ packets remain in `docs/` as explicitly marked historical records. direction: the planned initial venue is a separate noncustodial liquidity service, with a client-side router responsible for quote validation and transaction construction. A future AMM or decentralized limit-order book can -implement the same venue boundary. The RFQ service remains separate from -`deadcat-node`; future AMM and DLOB protocols are not implemented by this -repository today. +implement the same venue boundary. [ADR 0007](docs/adr/0007-rfq-provider-state-machine.md) +defines the provider's durable reservation and commit-before-sign boundary. +The transport-free provider state core is implemented, while its wallet, +pricing, transaction validator, signer adapter, remote service, and relay +layers remain future work. Until the validator and signer adapter land, the +safety-critical commit and signed-result transitions are intentionally +crate-internal. The RFQ provider remains separate from `deadcat-node`; future +AMM and DLOB protocols are not implemented by this repository today. ## Assurance diff --git a/crates/deadcat-rfq-provider/Cargo.toml b/crates/deadcat-rfq-provider/Cargo.toml new file mode 100644 index 0000000..efabf1c --- /dev/null +++ b/crates/deadcat-rfq-provider/Cargo.toml @@ -0,0 +1,20 @@ +[package] +name = "deadcat-rfq-provider" +description = "Durable inventory reservation and signing state for a noncustodial Deadcat RFQ provider." +version.workspace = true +edition.workspace = true +publish.workspace = true + +[lints] +workspace = true + +[dependencies] +elements.workspace = true +postcard.workspace = true +redb.workspace = true +serde.workspace = true +sha2.workspace = true +thiserror.workspace = true + +[dev-dependencies] +tempfile.workspace = true diff --git a/crates/deadcat-rfq-provider/src/lib.rs b/crates/deadcat-rfq-provider/src/lib.rs new file mode 100644 index 0000000..5fe0633 --- /dev/null +++ b/crates/deadcat-rfq-provider/src/lib.rs @@ -0,0 +1,31 @@ +//! Transport-free durable state for a noncustodial RFQ provider. +//! +//! This crate owns neither networking nor wallet keys. Its job is narrower: +//! make inventory allocation and the provider's signing point of no return +//! durable and auditable. The required ordering is: +//! +//! `validate -> commit exact payload -> sign -> persist signed bytes -> release` +//! +//! Only an uncommitted reservation can expire or be cancelled. Once a signing +//! payload is committed, every reserved outpoint remains retired even across +//! expiry, process restart, signer ambiguity, mempool eviction, or reorg. +//! +//! This first state-only layer does not yet expose the commit or signed-result +//! transitions outside the crate. A later concrete transaction validator and +//! signer adapter must be the only producers of those transition inputs. + +mod model; +mod store; + +pub use model::{ + AuditEntry, AuditEvent, Clock, FeePolicy, FeePolicyViolation, FeeSizeMetric, IdempotencyKey, + InventoryItem, InventoryState, InventoryView, MAX_RESERVATION_INPUTS, MAX_SETTLEMENT_BYTES, + ModelError, OwnerId, ProviderId, ProviderIdentity, QuoteCommitment, RecoveryAction, + ReleaseReason, ReservationAccess, ReservationId, ReservationPlan, ReservationState, + ReservationView, SignedArtifact, SignedArtifactDigest, SigningCommitment, SigningJob, + TransactionFee, UnixMillis, +}; +pub use store::{ + CommitOutcome, MAX_EXPIRATION_BATCH, ProviderError, ReservationBook, ReserveOutcome, + SCHEMA_VERSION, SignedOutcome, +}; diff --git a/crates/deadcat-rfq-provider/src/model.rs b/crates/deadcat-rfq-provider/src/model.rs new file mode 100644 index 0000000..746a7cb --- /dev/null +++ b/crates/deadcat-rfq-provider/src/model.rs @@ -0,0 +1,703 @@ +use elements::{AssetId, BlockHash, OutPoint}; +use thiserror::Error; + +/// Maximum number of provider inventory inputs one reservation may claim. +pub const MAX_RESERVATION_INPUTS: usize = 64; +/// Maximum exact pre-sign or signed settlement retained for recovery. +pub const MAX_SETTLEMENT_BYTES: usize = 1_000_000; + +macro_rules! fixed_id { + ($(#[$meta:meta])* $name:ident) => { + $(#[$meta])* + #[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord, Hash)] + pub struct $name([u8; 32]); + + impl $name { + #[must_use] + pub const fn new(bytes: [u8; 32]) -> Self { + Self(bytes) + } + + #[must_use] + pub const fn to_bytes(self) -> [u8; 32] { + self.0 + } + } + }; +} + +fixed_id!( + /// Stable identity of one independently operated RFQ provider. + ProviderId +); +fixed_id!( + /// Authenticated reservation owner, normally derived from a transport principal. + OwnerId +); +fixed_id!( + /// Client-chosen retry key. Reusing it with different terms is rejected. + IdempotencyKey +); +fixed_id!( + /// Provider-issued identifier for one durable reservation. + ReservationId +); +fixed_id!( + /// Commitment to the exact authenticated quote and leg economics. + QuoteCommitment +); +fixed_id!( + /// Domain-separated commitment to the exact durable pre-sign transcript. + SigningCommitment +); +fixed_id!( + /// Domain-separated commitment to the exact persisted signed response. + SignedArtifactDigest +); + +/// Absolute Unix time in milliseconds. +#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord, Hash)] +pub struct UnixMillis(u64); + +impl UnixMillis { + #[must_use] + pub const fn new(value: u64) -> Self { + Self(value) + } + + #[must_use] + pub const fn value(self) -> u64 { + self.0 + } +} + +/// Time source sampled exactly once after the durable writer is acquired. +/// +/// Implementations used by the service should return wall-clock Unix time. +/// Tests may pass a [`UnixMillis`] directly as a fixed clock. +pub trait Clock { + fn now(&self) -> UnixMillis; +} + +impl Clock for UnixMillis { + fn now(&self) -> UnixMillis { + *self + } +} + +/// Immutable identity binding for one provider database. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub struct ProviderIdentity { + provider: ProviderId, + genesis_hash: BlockHash, + policy_asset: AssetId, +} + +impl ProviderIdentity { + #[must_use] + pub const fn new(provider: ProviderId, genesis_hash: BlockHash, policy_asset: AssetId) -> Self { + Self { + provider, + genesis_hash, + policy_asset, + } + } + + #[must_use] + pub const fn provider(self) -> ProviderId { + self.provider + } + + #[must_use] + pub const fn genesis_hash(self) -> BlockHash { + self.genesis_hash + } + + #[must_use] + pub const fn policy_asset(self) -> AssetId { + self.policy_asset + } +} + +/// Provider-owned spendable output metadata. Wallet secrets live elsewhere. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub struct InventoryItem { + outpoint: OutPoint, + asset: AssetId, + amount: u64, +} + +impl InventoryItem { + pub fn new(outpoint: OutPoint, asset: AssetId, amount: u64) -> Result { + if outpoint.is_null() || outpoint.vout & 0xc000_0000 != 0 { + return Err(ModelError::InvalidInventoryOutpoint(outpoint)); + } + if amount == 0 { + return Err(ModelError::ZeroInventoryAmount); + } + Ok(Self { + outpoint, + asset, + amount, + }) + } + + #[must_use] + pub const fn outpoint(self) -> OutPoint { + self.outpoint + } + + #[must_use] + pub const fn asset(self) -> AssetId { + self.asset + } + + #[must_use] + pub const fn amount(self) -> u64 { + self.amount + } +} + +/// Transaction size measure used by the provider's broadcasting node. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum FeeSizeMetric { + RegularVbytes, + DiscountVbytes, +} + +/// Immutable fee and resource floor attached to a firm reservation. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub struct FeePolicy { + policy_asset: AssetId, + minimum_sats_per_kvb: u64, + minimum_absolute_fee: u64, + maximum_transaction_weight: u64, + size_metric: FeeSizeMetric, +} + +impl FeePolicy { + pub fn new( + policy_asset: AssetId, + minimum_sats_per_kvb: u64, + minimum_absolute_fee: u64, + maximum_transaction_weight: u64, + size_metric: FeeSizeMetric, + ) -> Result { + if minimum_sats_per_kvb == 0 { + return Err(ModelError::ZeroMinimumFeeRate); + } + if maximum_transaction_weight == 0 { + return Err(ModelError::ZeroMaximumTransactionWeight); + } + Ok(Self { + policy_asset, + minimum_sats_per_kvb, + minimum_absolute_fee, + maximum_transaction_weight, + size_metric, + }) + } + + #[must_use] + pub const fn policy_asset(self) -> AssetId { + self.policy_asset + } + + #[must_use] + pub const fn minimum_sats_per_kvb(self) -> u64 { + self.minimum_sats_per_kvb + } + + #[must_use] + pub const fn minimum_absolute_fee(self) -> u64 { + self.minimum_absolute_fee + } + + #[must_use] + pub const fn maximum_transaction_weight(self) -> u64 { + self.maximum_transaction_weight + } + + #[must_use] + pub const fn size_metric(self) -> FeeSizeMetric { + self.size_metric + } + + pub fn required_fee(self, transaction: TransactionFee) -> Result { + if transaction.policy_asset != self.policy_asset { + return Err(FeePolicyViolation::WrongPolicyAsset { + expected: self.policy_asset, + actual: transaction.policy_asset, + }); + } + if transaction.weight > self.maximum_transaction_weight { + return Err(FeePolicyViolation::TransactionOverweight { + maximum: self.maximum_transaction_weight, + actual: transaction.weight, + }); + } + let policy_vsize = match self.size_metric { + FeeSizeMetric::RegularVbytes => transaction.regular_vsize, + FeeSizeMetric::DiscountVbytes => transaction.discount_vsize, + }; + let numerator = u128::from(self.minimum_sats_per_kvb) + .checked_mul(u128::from(policy_vsize)) + .ok_or(FeePolicyViolation::RequiredFeeOverflow)?; + let rate_fee = numerator + .checked_add(999) + .ok_or(FeePolicyViolation::RequiredFeeOverflow)? + / 1_000; + let rate_fee = + u64::try_from(rate_fee).map_err(|_| FeePolicyViolation::RequiredFeeOverflow)?; + Ok(self.minimum_absolute_fee.max(rate_fee)) + } + + pub fn validate(self, transaction: TransactionFee) -> Result<(), FeePolicyViolation> { + let required = self.required_fee(transaction)?; + if transaction.amount < required { + return Err(FeePolicyViolation::FeeBelowMinimum { + required, + actual: transaction.amount, + }); + } + Ok(()) + } +} + +/// Fee facts computed from the fully blinded transaction, including the +/// provider's projected final witness. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub struct TransactionFee { + policy_asset: AssetId, + amount: u64, + weight: u64, + regular_vsize: u64, + discount_vsize: u64, +} + +impl TransactionFee { + pub fn new( + policy_asset: AssetId, + amount: u64, + weight: u64, + regular_vsize: u64, + discount_vsize: u64, + ) -> Result { + if weight == 0 || regular_vsize == 0 || discount_vsize == 0 { + return Err(ModelError::ZeroTransactionSize); + } + let expected_regular_vsize = weight / 4 + u64::from(!weight.is_multiple_of(4)); + if regular_vsize != expected_regular_vsize || discount_vsize > regular_vsize { + return Err(ModelError::InconsistentTransactionSize { + weight, + regular_vsize, + discount_vsize, + }); + } + Ok(Self { + policy_asset, + amount, + weight, + regular_vsize, + discount_vsize, + }) + } + + #[must_use] + pub const fn policy_asset(self) -> AssetId { + self.policy_asset + } + + #[must_use] + pub const fn amount(self) -> u64 { + self.amount + } + + #[must_use] + pub const fn weight(self) -> u64 { + self.weight + } + + #[must_use] + pub const fn regular_vsize(self) -> u64 { + self.regular_vsize + } + + #[must_use] + pub const fn discount_vsize(self) -> u64 { + self.discount_vsize + } +} + +/// Exact inventory allocation requested by one authenticated client operation. +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct ReservationPlan { + owner: OwnerId, + idempotency_key: IdempotencyKey, + quote_commitment: QuoteCommitment, + outpoints: Vec, + accept_before: UnixMillis, + fee_policy: FeePolicy, +} + +impl ReservationPlan { + pub fn new( + owner: OwnerId, + idempotency_key: IdempotencyKey, + quote_commitment: QuoteCommitment, + mut outpoints: Vec, + accept_before: UnixMillis, + fee_policy: FeePolicy, + ) -> Result { + if outpoints.is_empty() { + return Err(ModelError::EmptyReservation); + } + if outpoints.len() > MAX_RESERVATION_INPUTS { + return Err(ModelError::TooManyReservationInputs { + maximum: MAX_RESERVATION_INPUTS, + actual: outpoints.len(), + }); + } + outpoints.sort_unstable(); + if let Some(duplicate) = outpoints + .windows(2) + .find_map(|pair| (pair[0] == pair[1]).then_some(pair[0])) + { + return Err(ModelError::DuplicateReservationOutpoint(duplicate)); + } + Ok(Self { + owner, + idempotency_key, + quote_commitment, + outpoints, + accept_before, + fee_policy, + }) + } + + #[must_use] + pub const fn owner(&self) -> OwnerId { + self.owner + } + + #[must_use] + pub const fn idempotency_key(&self) -> IdempotencyKey { + self.idempotency_key + } + + #[must_use] + pub const fn quote_commitment(&self) -> QuoteCommitment { + self.quote_commitment + } + + #[must_use] + pub fn outpoints(&self) -> &[OutPoint] { + &self.outpoints + } + + #[must_use] + pub const fn accept_before(&self) -> UnixMillis { + self.accept_before + } + + #[must_use] + pub const fn fee_policy(&self) -> FeePolicy { + self.fee_policy + } +} + +/// Authenticated access to an existing reservation. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub struct ReservationAccess { + reservation_id: ReservationId, + owner: OwnerId, +} + +impl ReservationAccess { + #[must_use] + pub const fn new(reservation_id: ReservationId, owner: OwnerId) -> Self { + Self { + reservation_id, + owner, + } + } + + #[must_use] + pub const fn reservation_id(self) -> ReservationId { + self.reservation_id + } + + #[must_use] + pub const fn owner(self) -> OwnerId { + self.owner + } +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum ReleaseReason { + Expired, + ClientCancelled, + ProviderRejected, +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum ReservationState { + Reserved, + Released { + reason: ReleaseReason, + at: UnixMillis, + }, + Committed { + commitment: SigningCommitment, + committed_at: UnixMillis, + }, + Signed { + commitment: SigningCommitment, + artifact: SignedArtifactDigest, + committed_at: UnixMillis, + signed_at: UnixMillis, + }, +} + +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct ReservationView { + pub(crate) id: ReservationId, + pub(crate) owner: OwnerId, + pub(crate) quote_commitment: QuoteCommitment, + pub(crate) outpoints: Vec, + pub(crate) created_at: UnixMillis, + pub(crate) accept_before: UnixMillis, + pub(crate) fee_policy: FeePolicy, + pub(crate) state: ReservationState, +} + +impl ReservationView { + #[must_use] + pub const fn id(&self) -> ReservationId { + self.id + } + + #[must_use] + pub const fn owner(&self) -> OwnerId { + self.owner + } + + #[must_use] + pub const fn quote_commitment(&self) -> QuoteCommitment { + self.quote_commitment + } + + #[must_use] + pub fn outpoints(&self) -> &[OutPoint] { + &self.outpoints + } + + #[must_use] + pub const fn created_at(&self) -> UnixMillis { + self.created_at + } + + #[must_use] + pub const fn accept_before(&self) -> UnixMillis { + self.accept_before + } + + #[must_use] + pub const fn fee_policy(&self) -> FeePolicy { + self.fee_policy + } + + #[must_use] + pub const fn state(&self) -> ReservationState { + self.state + } +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum InventoryState { + Available, + Reserved { + reservation_id: ReservationId, + }, + Committed { + reservation_id: ReservationId, + commitment: SigningCommitment, + }, +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub struct InventoryView { + item: InventoryItem, + state: InventoryState, +} + +impl InventoryView { + pub(crate) const fn new(item: InventoryItem, state: InventoryState) -> Self { + Self { item, state } + } + + #[must_use] + pub const fn item(self) -> InventoryItem { + self.item + } + + #[must_use] + pub const fn state(self) -> InventoryState { + self.state + } +} + +/// Exact durable work item that a signer may consume. +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct SigningJob { + pub(crate) reservation_id: ReservationId, + pub(crate) commitment: SigningCommitment, + pub(crate) pre_sign_payload: Vec, + pub(crate) fee: TransactionFee, +} + +impl SigningJob { + #[must_use] + pub const fn reservation_id(&self) -> ReservationId { + self.reservation_id + } + + #[must_use] + pub const fn commitment(&self) -> SigningCommitment { + self.commitment + } + + #[must_use] + pub fn pre_sign_payload(&self) -> &[u8] { + &self.pre_sign_payload + } + + #[must_use] + pub const fn fee(&self) -> TransactionFee { + self.fee + } +} + +/// Exact signed bytes persisted before any response or relay attempt. +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct SignedArtifact { + pub(crate) reservation_id: ReservationId, + pub(crate) commitment: SigningCommitment, + pub(crate) digest: SignedArtifactDigest, + pub(crate) bytes: Vec, +} + +impl SignedArtifact { + #[must_use] + pub const fn reservation_id(&self) -> ReservationId { + self.reservation_id + } + + #[must_use] + pub const fn commitment(&self) -> SigningCommitment { + self.commitment + } + + #[must_use] + pub const fn digest(&self) -> SignedArtifactDigest { + self.digest + } + + #[must_use] + pub fn bytes(&self) -> &[u8] { + &self.bytes + } +} + +#[derive(Clone, Debug, PartialEq, Eq)] +pub enum RecoveryAction { + SignCommittedExact(SigningJob), + ReplaySignedExact(SignedArtifact), +} + +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct AuditEntry { + pub(crate) sequence: u64, + pub(crate) at: UnixMillis, + pub(crate) event: AuditEvent, +} + +impl AuditEntry { + #[must_use] + pub const fn sequence(&self) -> u64 { + self.sequence + } + + #[must_use] + pub const fn at(&self) -> UnixMillis { + self.at + } + + #[must_use] + pub const fn event(&self) -> &AuditEvent { + &self.event + } +} + +#[derive(Clone, Debug, PartialEq, Eq)] +pub enum AuditEvent { + InventoryImported { + outpoint: OutPoint, + }, + ReservationCreated { + reservation_id: ReservationId, + outpoints: Vec, + }, + ReservationReleased { + reservation_id: ReservationId, + reason: ReleaseReason, + }, + SigningCommitted { + reservation_id: ReservationId, + commitment: SigningCommitment, + }, + SignedArtifactStored { + reservation_id: ReservationId, + artifact: SignedArtifactDigest, + }, +} + +#[derive(Debug, Error, PartialEq, Eq)] +pub enum ModelError { + #[error("inventory outpoint is null or contains issuance flags: {0:?}")] + InvalidInventoryOutpoint(OutPoint), + #[error("inventory amount must be nonzero")] + ZeroInventoryAmount, + #[error("minimum fee rate must be nonzero")] + ZeroMinimumFeeRate, + #[error("maximum transaction weight must be nonzero")] + ZeroMaximumTransactionWeight, + #[error("transaction weight and virtual sizes must be nonzero")] + ZeroTransactionSize, + #[error( + "transaction size metrics disagree: weight={weight}, regular_vsize={regular_vsize}, discount_vsize={discount_vsize}" + )] + InconsistentTransactionSize { + weight: u64, + regular_vsize: u64, + discount_vsize: u64, + }, + #[error("a reservation must contain at least one outpoint")] + EmptyReservation, + #[error("a reservation contains {actual} inputs; maximum is {maximum}")] + TooManyReservationInputs { maximum: usize, actual: usize }, + #[error("reservation contains duplicate outpoint {0:?}")] + DuplicateReservationOutpoint(OutPoint), +} + +#[derive(Clone, Debug, Error, PartialEq, Eq)] +pub enum FeePolicyViolation { + #[error("fee uses asset {actual}, expected policy asset {expected}")] + WrongPolicyAsset { expected: AssetId, actual: AssetId }, + #[error("transaction weight {actual} exceeds provider maximum {maximum}")] + TransactionOverweight { maximum: u64, actual: u64 }, + #[error("required fee calculation overflowed")] + RequiredFeeOverflow, + #[error("network fee {actual} is below required minimum {required}")] + FeeBelowMinimum { required: u64, actual: u64 }, +} diff --git a/crates/deadcat-rfq-provider/src/store.rs b/crates/deadcat-rfq-provider/src/store.rs new file mode 100644 index 0000000..f30724e --- /dev/null +++ b/crates/deadcat-rfq-provider/src/store.rs @@ -0,0 +1,2565 @@ +use std::collections::{BTreeMap, BTreeSet}; +use std::path::Path; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::{Mutex, MutexGuard}; + +use elements::hashes::Hash as _; +use elements::{AssetId, BlockHash, OutPoint}; +use redb::{ + Database, Durability, ReadableDatabase as _, ReadableTable as _, TableDefinition, + WriteTransaction, +}; +use serde::de::DeserializeOwned; +use serde::{Deserialize, Serialize}; +use sha2::{Digest as _, Sha256}; +use thiserror::Error; + +use crate::model::{ + AuditEntry, AuditEvent, Clock, FeePolicy, FeePolicyViolation, FeeSizeMetric, IdempotencyKey, + InventoryItem, InventoryState, InventoryView, MAX_RESERVATION_INPUTS, MAX_SETTLEMENT_BYTES, + OwnerId, ProviderId, ProviderIdentity, QuoteCommitment, RecoveryAction, ReleaseReason, + ReservationAccess, ReservationId, ReservationPlan, ReservationState, ReservationView, + SignedArtifact, SignedArtifactDigest, SigningCommitment, SigningJob, TransactionFee, + UnixMillis, +}; + +pub const SCHEMA_VERSION: u32 = 1; +/// Maximum number of unrelated expirations one explicit sweep may mutate in a +/// single immediate-durability transaction. +pub const MAX_EXPIRATION_BATCH: usize = 256; +const RECORD_VERSION: u8 = 1; + +const META: TableDefinition<&str, &[u8]> = TableDefinition::new("meta"); +const INVENTORY: TableDefinition<&[u8], &[u8]> = TableDefinition::new("inventory"); +const ALLOCATIONS: TableDefinition<&[u8], &[u8]> = TableDefinition::new("allocations"); +const RESERVATIONS: TableDefinition<&[u8], &[u8]> = TableDefinition::new("reservations"); +const REQUEST_KEYS: TableDefinition<&[u8], &[u8]> = TableDefinition::new("request_keys"); +const EXPIRATIONS: TableDefinition<&[u8], &[u8]> = TableDefinition::new("expirations"); +const AUDIT: TableDefinition = TableDefinition::new("audit"); + +const SCHEMA_VERSION_KEY: &str = "schema_version"; +const PROVIDER_IDENTITY_KEY: &str = "provider_identity"; +const LAST_OBSERVED_TIME_KEY: &str = "last_observed_unix_millis"; +const AUDIT_SEQUENCE_KEY: &str = "audit_sequence"; + +const RESERVATION_ID_DOMAIN: &[u8] = b"deadcat/rfq/reservation-id/v1"; +const REQUEST_DOMAIN: &[u8] = b"deadcat/rfq/reservation-request/v1"; +const SIGNING_DOMAIN: &[u8] = b"deadcat/rfq/signing-transcript/v1"; +const SIGNED_ARTIFACT_DOMAIN: &[u8] = b"deadcat/rfq/signed-artifact/v1"; + +/// Graceful mutation failures used to prove that every logical transition is +/// one redb commit. These are not a simulation of process death, torn writes, +/// or redb's own crash-recovery machinery. +#[cfg(test)] +mod mutation_failpoints { + use std::cell::RefCell; + + use super::ProviderError; + + pub(super) const RESERVE_AFTER_RECORD: &str = "reserve.after_record"; + pub(super) const RESERVE_AFTER_REQUEST_KEY: &str = "reserve.after_request_key"; + pub(super) const RESERVE_AFTER_ALLOCATION: &str = "reserve.after_allocation"; + pub(super) const RESERVE_AFTER_EXPIRATION: &str = "reserve.after_expiration"; + pub(super) const RESERVE_AFTER_AUDIT: &str = "reserve.after_audit"; + pub(super) const RELEASE_AFTER_ALLOCATION: &str = "release.after_allocation"; + pub(super) const RELEASE_AFTER_EXPIRATION: &str = "release.after_expiration"; + pub(super) const RELEASE_AFTER_RECORD: &str = "release.after_record"; + pub(super) const RELEASE_AFTER_AUDIT: &str = "release.after_audit"; + pub(super) const COMMIT_AFTER_ALLOCATION: &str = "commit.after_allocation"; + pub(super) const COMMIT_AFTER_EXPIRATION: &str = "commit.after_expiration"; + pub(super) const COMMIT_AFTER_RECORD: &str = "commit.after_record"; + pub(super) const COMMIT_AFTER_AUDIT: &str = "commit.after_audit"; + pub(super) const SIGNED_AFTER_RECORD: &str = "signed.after_record"; + pub(super) const SIGNED_AFTER_AUDIT: &str = "signed.after_audit"; + + #[derive(Clone, Copy)] + struct Active { + name: &'static str, + remaining_hits: usize, + } + + thread_local! { + static ACTIVE: RefCell> = const { RefCell::new(None) }; + } + + pub(super) struct Guard; + + impl Drop for Guard { + fn drop(&mut self) { + ACTIVE.with(|active| *active.borrow_mut() = None); + } + } + + pub(super) fn arm(name: &'static str, occurrence: usize) -> Guard { + ACTIVE.with(|active| { + let mut active = active.borrow_mut(); + assert!(active.is_none(), "a mutation failpoint is already armed"); + *active = Some(Active { + name, + remaining_hits: occurrence, + }); + }); + Guard + } + + pub(super) fn hit(name: &'static str) -> Result<(), ProviderError> { + ACTIVE.with(|active| { + let mut active = active.borrow_mut(); + let Some(specification) = active.as_mut() else { + return Ok(()); + }; + if specification.name != name { + return Ok(()); + } + if specification.remaining_hits != 0 { + specification.remaining_hits -= 1; + return Ok(()); + } + *active = None; + Err(ProviderError::InjectedMutationFailure(name)) + }) + } +} + +/// Synchronous durable reservation state. Async service code should call this +/// through its blocking boundary rather than sharing wallet or network state. +pub struct ReservationBook { + database: Database, + identity: ProviderIdentity, + poisoned: AtomicBool, + operation_lock: Mutex<()>, +} + +impl ReservationBook { + pub fn open(path: impl AsRef, identity: ProviderIdentity) -> Result { + let database = Database::create(path)?; + let book = Self { + database, + identity, + poisoned: AtomicBool::new(false), + operation_lock: Mutex::new(()), + }; + book.initialize_schema()?; + Ok(book) + } + + #[must_use] + pub const fn identity(&self) -> ProviderIdentity { + self.identity + } + + pub fn schema_version(&self) -> Result { + self.ensure_healthy()?; + let read = self.database.begin_read()?; + let table = read.open_table(META)?; + let value = table + .get(SCHEMA_VERSION_KEY)? + .ok_or(ProviderError::MissingMetadata(SCHEMA_VERSION_KEY))?; + decode_u32(value.value()).map_err(|()| ProviderError::CorruptSchemaVersion) + } + + /// Add one wallet-discovered output without changing an existing record. + /// Exact retries are idempotent; conflicting metadata is rejected. + pub fn import_inventory( + &self, + item: InventoryItem, + clock: &C, + ) -> Result { + let (_operation_guard, write, now) = self.begin_timed_write(clock)?; + let key = outpoint_key(item.outpoint()); + let stored = StoredInventoryItem::from(item); + if let Some(existing) = read_record_from_write(&write, INVENTORY, &key)? { + let existing: StoredInventoryItem = existing; + if existing != stored { + return Err(ProviderError::InventoryMetadataConflict { + outpoint: item.outpoint(), + }); + } + self.commit_write(write)?; + return Ok(false); + } + write_record(&write, INVENTORY, &key, &stored)?; + append_audit( + &write, + now, + StoredAuditEvent::InventoryImported { + outpoint: item.outpoint(), + }, + )?; + self.commit_write(write)?; + Ok(true) + } + + pub fn inventory(&self, outpoint: OutPoint) -> Result, ProviderError> { + self.ensure_healthy()?; + let read = self.database.begin_read()?; + let inventory = read.open_table(INVENTORY)?; + let key = outpoint_key(outpoint); + let Some(item) = inventory.get(key.as_slice())? else { + return Ok(None); + }; + let item: StoredInventoryItem = decode_record(item.value())?; + drop(inventory); + let allocations = read.open_table(ALLOCATIONS)?; + let state = allocations + .get(key.as_slice())? + .map(|allocation| decode_record::(allocation.value())) + .transpose()? + .map_or(InventoryState::Available, StoredAllocation::to_view); + Ok(Some(InventoryView::new(item.to_domain()?, state))) + } + + pub fn inventory_all(&self) -> Result, ProviderError> { + self.ensure_healthy()?; + let read = self.database.begin_read()?; + let inventory = read.open_table(INVENTORY)?; + let allocations = read.open_table(ALLOCATIONS)?; + let mut result = Vec::new(); + for entry in inventory.iter()? { + let (key, item) = entry?; + let item: StoredInventoryItem = decode_record(item.value())?; + let state = allocations + .get(key.value())? + .map(|allocation| decode_record::(allocation.value())) + .transpose()? + .map_or(InventoryState::Available, StoredAllocation::to_view); + result.push(InventoryView::new(item.to_domain()?, state)); + } + Ok(result) + } + + /// Atomically reserve every requested outpoint or none of them. + /// + /// The clock is sampled after acquiring redb's serial writer, so a request + /// queued behind another writer cannot commit using a stale pre-lock time. + pub fn reserve( + &self, + plan: &ReservationPlan, + clock: &C, + ) -> Result { + if plan.fee_policy().policy_asset() != self.identity.policy_asset() { + return Err(ProviderError::WrongPolicyAsset { + expected: self.identity.policy_asset(), + actual: plan.fee_policy().policy_asset(), + }); + } + let request_digest = request_digest(self.identity, plan)?; + let reservation_id = derive_reservation_id(plan.owner(), plan.idempotency_key()); + let (_operation_guard, write, now) = self.begin_timed_write(clock)?; + expire_requested_in_write(&write, now, plan.outpoints())?; + + if let Some(binding) = read_request_binding(&write, plan.owner(), plan.idempotency_key())? { + if binding.request_digest != request_digest { + return Err(ProviderError::IdempotencyConflict { + owner: plan.owner(), + key: plan.idempotency_key(), + }); + } + let record = + read_reservation_from_write(&write, ReservationId::new(binding.reservation_id))? + .ok_or(ProviderError::CorruptState( + "idempotency binding references a missing reservation".to_owned(), + ))?; + self.commit_write(write)?; + return Ok(ReserveOutcome { + reservation: record.to_view()?, + created: false, + }); + } + + if now >= plan.accept_before() { + self.commit_write(write)?; + return Err(ProviderError::ReservationDeadlineElapsed { + accept_before: plan.accept_before(), + now, + }); + } + if read_reservation_from_write(&write, reservation_id)?.is_some() { + return Err(ProviderError::ReservationIdCollision(reservation_id)); + } + + for outpoint in plan.outpoints() { + let key = outpoint_key(*outpoint); + if read_record_from_write::(&write, INVENTORY, &key)?.is_none() { + return Err(ProviderError::UnknownInventory(*outpoint)); + } + if let Some(allocation) = + read_record_from_write::(&write, ALLOCATIONS, &key)? + { + return Err(ProviderError::OutpointUnavailable { + outpoint: *outpoint, + state: allocation.to_view(), + }); + } + } + + let record = StoredReservation { + id: reservation_id.to_bytes(), + owner: plan.owner().to_bytes(), + idempotency_key: plan.idempotency_key().to_bytes(), + request_digest, + quote_commitment: plan.quote_commitment().to_bytes(), + outpoints: plan.outpoints().to_vec(), + created_at: now.value(), + accept_before: plan.accept_before().value(), + fee_policy: StoredFeePolicy::from(plan.fee_policy()), + state: StoredReservationState::Reserved, + }; + write_record(&write, RESERVATIONS, &reservation_id.to_bytes(), &record)?; + #[cfg(test)] + mutation_failpoints::hit(mutation_failpoints::RESERVE_AFTER_RECORD)?; + let binding = StoredRequestBinding { + reservation_id: reservation_id.to_bytes(), + request_digest, + }; + write_record( + &write, + REQUEST_KEYS, + &request_key(plan.owner(), plan.idempotency_key()), + &binding, + )?; + #[cfg(test)] + mutation_failpoints::hit(mutation_failpoints::RESERVE_AFTER_REQUEST_KEY)?; + for outpoint in plan.outpoints() { + write_record( + &write, + ALLOCATIONS, + &outpoint_key(*outpoint), + &StoredAllocation::Reserved { + reservation_id: reservation_id.to_bytes(), + }, + )?; + #[cfg(test)] + mutation_failpoints::hit(mutation_failpoints::RESERVE_AFTER_ALLOCATION)?; + } + let expiration_key = expiration_key(plan.accept_before(), reservation_id); + let empty: &[u8] = &[]; + write + .open_table(EXPIRATIONS)? + .insert(expiration_key.as_slice(), empty)?; + #[cfg(test)] + mutation_failpoints::hit(mutation_failpoints::RESERVE_AFTER_EXPIRATION)?; + append_audit( + &write, + now, + StoredAuditEvent::ReservationCreated { + reservation_id: reservation_id.to_bytes(), + outpoints: plan.outpoints().to_vec(), + }, + )?; + #[cfg(test)] + mutation_failpoints::hit(mutation_failpoints::RESERVE_AFTER_AUDIT)?; + self.commit_write(write)?; + Ok(ReserveOutcome { + reservation: record.to_view()?, + created: true, + }) + } + + pub fn reservation( + &self, + reservation_id: ReservationId, + ) -> Result, ProviderError> { + self.ensure_healthy()?; + let read = self.database.begin_read()?; + let table = read.open_table(RESERVATIONS)?; + let record = table + .get(reservation_id.to_bytes().as_slice())? + .map(|value| decode_record::(value.value())) + .transpose()?; + record + .map(|record| { + if record.id() != reservation_id { + return Err(ProviderError::CorruptState( + "reservation key and record ID disagree".to_owned(), + )); + } + record.validate()?; + record.to_view() + }) + .transpose() + } + + /// Cancel a reservation only while it remains before the signing point of + /// no return. Cancellation at or after the deadline is recorded as expiry. + pub fn cancel( + &self, + access: ReservationAccess, + clock: &C, + ) -> Result { + let (_operation_guard, write, now) = self.begin_timed_write(clock)?; + let mut record = require_authorized_reservation(&write, access)?; + match record.state { + StoredReservationState::Reserved => { + let reason = if now >= UnixMillis::new(record.accept_before) { + ReleaseReason::Expired + } else { + ReleaseReason::ClientCancelled + }; + release_reserved(&write, &mut record, reason, now)?; + self.commit_write(write)?; + Ok(true) + } + StoredReservationState::Released { + reason: StoredReleaseReason::ClientCancelled, + .. + } => { + self.commit_write(write)?; + Ok(false) + } + StoredReservationState::Released { .. } => { + Err(ProviderError::ReservationAlreadyReleased(record.id())) + } + StoredReservationState::Committed { .. } | StoredReservationState::Signed { .. } => { + Err(ProviderError::PointOfNoReturn(record.id())) + } + } + } + + /// Provider-side rejection before commitment. This is distinct from a + /// client cancellation in the durable audit trail. + pub fn reject_uncommitted( + &self, + reservation_id: ReservationId, + clock: &C, + ) -> Result { + let (_operation_guard, write, now) = self.begin_timed_write(clock)?; + let mut record = read_reservation_from_write(&write, reservation_id)? + .ok_or(ProviderError::ReservationNotFound(reservation_id))?; + match record.state { + StoredReservationState::Reserved => { + let reason = if now >= UnixMillis::new(record.accept_before) { + ReleaseReason::Expired + } else { + ReleaseReason::ProviderRejected + }; + release_reserved(&write, &mut record, reason, now)?; + self.commit_write(write)?; + Ok(true) + } + StoredReservationState::Released { .. } => { + self.commit_write(write)?; + Ok(false) + } + StoredReservationState::Committed { .. } | StoredReservationState::Signed { .. } => { + Err(ProviderError::PointOfNoReturn(reservation_id)) + } + } + } + + /// Release up to `limit` expired reservations, oldest deadline first. + pub fn expire_due( + &self, + clock: &C, + limit: usize, + ) -> Result, ProviderError> { + if limit == 0 { + return Ok(Vec::new()); + } + let (_operation_guard, write, now) = self.begin_timed_write(clock)?; + let expired = expire_due_in_write(&write, now, limit.min(MAX_EXPIRATION_BATCH))?; + self.commit_write(write)?; + Ok(expired) + } + + /// Atomically bind the reserved inputs to exact, already-validated bytes + /// before any wallet or HSM signer is invoked. + /// + /// `pre_sign_payload` must be the complete immutable provider signing + /// transcript produced by the later settlement validator, including the + /// finalized transaction body, proofs, authoritative prevouts, existing + /// user witnesses, approved sighash profile, and quote economics. + // The concrete validator added in the next provider layer will be this + // method's only production caller. Keeping the transition crate-private + // prevents detached fee assertions from crossing the trust boundary. + #[allow(dead_code)] + pub(crate) fn commit_before_sign( + &self, + access: ReservationAccess, + pre_sign_payload: Vec, + fee: TransactionFee, + clock: &C, + ) -> Result { + validate_settlement_bytes(&pre_sign_payload)?; + let (_operation_guard, write, now) = self.begin_timed_write(clock)?; + let mut record = require_authorized_reservation(&write, access)?; + + match &record.state { + StoredReservationState::Committed { intent } => { + let proposed = signing_commitment(&record, &pre_sign_payload, fee)?; + if intent.commitment != proposed.to_bytes() + || intent.pre_sign_payload != pre_sign_payload + || intent.fee != StoredTransactionFee::from(fee) + { + return Err(ProviderError::DifferentSigningIntent(record.id())); + } + let job = intent.to_job(record.id())?; + self.commit_write(write)?; + return Ok(CommitOutcome::AlreadyCommitted(job)); + } + StoredReservationState::Signed { intent, artifact } => { + let proposed = signing_commitment(&record, &pre_sign_payload, fee)?; + if intent.commitment != proposed.to_bytes() + || intent.pre_sign_payload != pre_sign_payload + || intent.fee != StoredTransactionFee::from(fee) + { + return Err(ProviderError::DifferentSigningIntent(record.id())); + } + let artifact = artifact.to_domain(record.id(), proposed)?; + self.commit_write(write)?; + return Ok(CommitOutcome::AlreadySigned(artifact)); + } + StoredReservationState::Released { .. } => { + return Err(ProviderError::ReservationAlreadyReleased(record.id())); + } + StoredReservationState::Reserved => {} + } + + if now >= UnixMillis::new(record.accept_before) { + let deadline = UnixMillis::new(record.accept_before); + release_reserved(&write, &mut record, ReleaseReason::Expired, now)?; + self.commit_write(write)?; + return Err(ProviderError::ReservationDeadlineElapsed { + accept_before: deadline, + now, + }); + } + + let policy = record.fee_policy.to_domain()?; + policy.validate(fee)?; + let commitment = signing_commitment(&record, &pre_sign_payload, fee)?; + for outpoint in &record.outpoints { + let key = outpoint_key(*outpoint); + let allocation = read_record_from_write::(&write, ALLOCATIONS, &key)? + .ok_or_else(|| { + ProviderError::CorruptState(format!( + "reserved outpoint {outpoint:?} has no allocation" + )) + })?; + if allocation + != (StoredAllocation::Reserved { + reservation_id: record.id, + }) + { + return Err(ProviderError::CorruptState(format!( + "reserved outpoint {outpoint:?} has a different allocation" + ))); + } + } + + let intent = StoredSigningIntent { + commitment: commitment.to_bytes(), + pre_sign_payload, + fee: StoredTransactionFee::from(fee), + committed_at: now.value(), + }; + for outpoint in &record.outpoints { + write_record( + &write, + ALLOCATIONS, + &outpoint_key(*outpoint), + &StoredAllocation::Committed { + reservation_id: record.id, + commitment: commitment.to_bytes(), + }, + )?; + #[cfg(test)] + mutation_failpoints::hit(mutation_failpoints::COMMIT_AFTER_ALLOCATION)?; + } + let expiration_key = expiration_key(UnixMillis::new(record.accept_before), record.id()); + let removed_expiration = { + let mut expirations = write.open_table(EXPIRATIONS)?; + expirations.remove(expiration_key.as_slice())?.is_some() + }; + if !removed_expiration { + return Err(ProviderError::CorruptState( + "reserved reservation has no expiration index entry".to_owned(), + )); + } + #[cfg(test)] + mutation_failpoints::hit(mutation_failpoints::COMMIT_AFTER_EXPIRATION)?; + record.state = StoredReservationState::Committed { + intent: intent.clone(), + }; + write_record(&write, RESERVATIONS, &record.id, &record)?; + #[cfg(test)] + mutation_failpoints::hit(mutation_failpoints::COMMIT_AFTER_RECORD)?; + append_audit( + &write, + now, + StoredAuditEvent::SigningCommitted { + reservation_id: record.id, + commitment: commitment.to_bytes(), + }, + )?; + #[cfg(test)] + mutation_failpoints::hit(mutation_failpoints::COMMIT_AFTER_AUDIT)?; + let job = intent.to_job(record.id())?; + self.commit_write(write)?; + Ok(CommitOutcome::NewlyCommitted(job)) + } + + /// Persist exact signed bytes before they can be returned or relayed. + // The signer adapter added in the next provider layer will verify and + // canonicalize its result before invoking this crate-private transition. + #[allow(dead_code)] + pub(crate) fn record_signed( + &self, + reservation_id: ReservationId, + expected_commitment: SigningCommitment, + signed_bytes: Vec, + clock: &C, + ) -> Result { + validate_settlement_bytes(&signed_bytes)?; + let (_operation_guard, write, now) = self.begin_timed_write(clock)?; + let mut record = read_reservation_from_write(&write, reservation_id)? + .ok_or(ProviderError::ReservationNotFound(reservation_id))?; + let intent = match &record.state { + StoredReservationState::Committed { intent } => intent.clone(), + StoredReservationState::Signed { intent, artifact } => { + if intent.commitment != expected_commitment.to_bytes() + || artifact.bytes != signed_bytes + { + return Err(ProviderError::DifferentSignedArtifact(reservation_id)); + } + let artifact = artifact.to_domain(reservation_id, expected_commitment)?; + self.commit_write(write)?; + return Ok(SignedOutcome { + artifact, + recorded: false, + }); + } + StoredReservationState::Reserved => { + return Err(ProviderError::SigningIntentNotCommitted(reservation_id)); + } + StoredReservationState::Released { .. } => { + return Err(ProviderError::ReservationAlreadyReleased(reservation_id)); + } + }; + if intent.commitment != expected_commitment.to_bytes() { + return Err(ProviderError::SigningCommitmentMismatch { + reservation_id, + expected: SigningCommitment::new(intent.commitment), + actual: expected_commitment, + }); + } + let digest = signed_artifact_digest(expected_commitment, &signed_bytes); + let stored_artifact = StoredSignedArtifact { + digest: digest.to_bytes(), + bytes: signed_bytes, + signed_at: now.value(), + }; + let artifact = stored_artifact.to_domain(reservation_id, expected_commitment)?; + record.state = StoredReservationState::Signed { + intent, + artifact: stored_artifact, + }; + write_record(&write, RESERVATIONS, &reservation_id.to_bytes(), &record)?; + #[cfg(test)] + mutation_failpoints::hit(mutation_failpoints::SIGNED_AFTER_RECORD)?; + append_audit( + &write, + now, + StoredAuditEvent::SignedArtifactStored { + reservation_id: reservation_id.to_bytes(), + artifact: digest.to_bytes(), + }, + )?; + #[cfg(test)] + mutation_failpoints::hit(mutation_failpoints::SIGNED_AFTER_AUDIT)?; + self.commit_write(write)?; + Ok(SignedOutcome { + artifact, + recorded: true, + }) + } + + /// Exact actions safe to resume after restart. The caller must sign or + /// replay only the returned durable bytes. + pub fn recovery_actions(&self) -> Result, ProviderError> { + self.ensure_healthy()?; + let read = self.database.begin_read()?; + let table = read.open_table(RESERVATIONS)?; + let mut actions = Vec::new(); + for entry in table.iter()? { + let (_, value) = entry?; + let record: StoredReservation = decode_record(value.value())?; + record.validate()?; + match &record.state { + StoredReservationState::Committed { intent } => actions.push( + RecoveryAction::SignCommittedExact(intent.to_job(record.id())?), + ), + StoredReservationState::Signed { intent, artifact } => { + let commitment = SigningCommitment::new(intent.commitment); + actions.push(RecoveryAction::ReplaySignedExact( + artifact.to_domain(record.id(), commitment)?, + )); + } + StoredReservationState::Reserved | StoredReservationState::Released { .. } => {} + } + } + Ok(actions) + } + + pub fn audit_log(&self) -> Result, ProviderError> { + self.ensure_healthy()?; + let read = self.database.begin_read()?; + let table = read.open_table(AUDIT)?; + let mut entries = Vec::new(); + for entry in table.iter()? { + let (sequence, value) = entry?; + let stored: StoredAuditEntry = decode_record(value.value())?; + if stored.sequence != sequence.value() { + return Err(ProviderError::CorruptState( + "audit key and record sequence disagree".to_owned(), + )); + } + entries.push(stored.to_domain()); + } + Ok(entries) + } + + pub fn last_observed_time(&self) -> Result, ProviderError> { + self.ensure_healthy()?; + let read = self.database.begin_read()?; + let table = read.open_table(META)?; + table + .get(LAST_OBSERVED_TIME_KEY)? + .map(|value| { + decode_u64(value.value()) + .map(UnixMillis::new) + .map_err(|()| ProviderError::CorruptTimeHighWatermark) + }) + .transpose() + } + + fn initialize_schema(&self) -> Result<(), ProviderError> { + let write = self.begin_immediate_write()?; + create_tables(&write)?; + let existing_schema = { + let meta = write.open_table(META)?; + meta.get(SCHEMA_VERSION_KEY)? + .map(|value| value.value().to_vec()) + }; + match existing_schema { + Some(value) => { + let actual = + decode_u32(&value).map_err(|()| ProviderError::CorruptSchemaVersion)?; + if actual != SCHEMA_VERSION { + return Err(ProviderError::SchemaMismatch { + expected: SCHEMA_VERSION, + actual, + }); + } + let meta = write.open_table(META)?; + let identity = meta + .get(PROVIDER_IDENTITY_KEY)? + .ok_or(ProviderError::MissingMetadata(PROVIDER_IDENTITY_KEY))?; + let actual: StoredProviderIdentity = decode_record(identity.value())?; + let actual = actual.to_domain(); + if actual != self.identity { + return Err(ProviderError::ProviderIdentityMismatch { + expected: Box::new(actual), + actual: Box::new(self.identity), + }); + } + if meta.get(AUDIT_SEQUENCE_KEY)?.is_none() { + return Err(ProviderError::MissingMetadata(AUDIT_SEQUENCE_KEY)); + } + } + None => { + if provider_tables_are_nonempty(&write)? { + return Err(ProviderError::CorruptState( + "schema version is missing from a nonempty provider database".to_owned(), + )); + } + let mut meta = write.open_table(META)?; + meta.insert(SCHEMA_VERSION_KEY, SCHEMA_VERSION.to_be_bytes().as_slice())?; + let identity = encode_record(&StoredProviderIdentity::from(self.identity))?; + meta.insert(PROVIDER_IDENTITY_KEY, identity.as_slice())?; + meta.insert(AUDIT_SEQUENCE_KEY, 0_u64.to_be_bytes().as_slice())?; + } + } + validate_store_integrity(&write, self.identity)?; + self.commit_write(write)?; + Ok(()) + } + + fn begin_immediate_write(&self) -> Result { + self.ensure_healthy()?; + let mut write = self.database.begin_write()?; + self.ensure_healthy()?; + // A returned reservation or signing commitment is safe to expose only + // after redb guarantees that its input locks survived a crash. + write.set_durability(Durability::Immediate)?; + Ok(write) + } + + /// Serialize a complete timed operation and durably advance the clock + /// high-water mark before starting its logical mutation. + /// + /// The separate immediate commit is intentional: authentication, + /// validation, or injected mutation failures must roll back the business + /// transaction without erasing the fact that the later time was observed. + /// Holding this process lock across both transactions prevents an older + /// operation from mutating state after a newer observation. redb's + /// exclusive database open prevents a second process from bypassing it. + fn begin_timed_write( + &self, + clock: &C, + ) -> Result<(MutexGuard<'_, ()>, WriteTransaction, UnixMillis), ProviderError> { + let operation_guard = self + .operation_lock + .lock() + .map_err(|_| ProviderError::OperationLockPoisoned)?; + let observation = self.begin_immediate_write()?; + let now = observe_time(&observation, clock)?; + self.commit_write(observation)?; + let write = self.begin_immediate_write()?; + Ok((operation_guard, write, now)) + } + + fn commit_write(&self, write: WriteTransaction) -> Result<(), ProviderError> { + match write.commit() { + Ok(()) => Ok(()), + Err(error) => { + // A commit error is an ambiguous durability boundary. Require + // the service to drop and reopen the database before it makes + // another availability or signing decision. + self.poisoned.store(true, Ordering::SeqCst); + Err(ProviderError::Commit(error)) + } + } + } + + fn ensure_healthy(&self) -> Result<(), ProviderError> { + if self.poisoned.load(Ordering::SeqCst) { + return Err(ProviderError::DatabaseRequiresReopen); + } + Ok(()) + } +} + +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct ReserveOutcome { + reservation: ReservationView, + created: bool, +} + +impl ReserveOutcome { + #[must_use] + pub const fn reservation(&self) -> &ReservationView { + &self.reservation + } + + #[must_use] + pub const fn created(&self) -> bool { + self.created + } +} + +#[derive(Clone, Debug, PartialEq, Eq)] +pub enum CommitOutcome { + NewlyCommitted(SigningJob), + AlreadyCommitted(SigningJob), + AlreadySigned(SignedArtifact), +} + +impl CommitOutcome { + #[must_use] + pub const fn signing_job(&self) -> Option<&SigningJob> { + match self { + Self::NewlyCommitted(job) | Self::AlreadyCommitted(job) => Some(job), + Self::AlreadySigned(_) => None, + } + } + + #[must_use] + pub const fn signed_artifact(&self) -> Option<&SignedArtifact> { + match self { + Self::AlreadySigned(artifact) => Some(artifact), + Self::NewlyCommitted(_) | Self::AlreadyCommitted(_) => None, + } + } + + #[must_use] + pub const fn newly_committed(&self) -> bool { + matches!(self, Self::NewlyCommitted(_)) + } +} + +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct SignedOutcome { + artifact: SignedArtifact, + recorded: bool, +} + +impl SignedOutcome { + #[must_use] + pub const fn artifact(&self) -> &SignedArtifact { + &self.artifact + } + + #[must_use] + pub const fn recorded(&self) -> bool { + self.recorded + } +} + +fn observe_time( + write: &WriteTransaction, + clock: &C, +) -> Result { + let now = clock.now(); + let mut meta = write.open_table(META)?; + if let Some(previous) = meta.get(LAST_OBSERVED_TIME_KEY)? { + let previous = + decode_u64(previous.value()).map_err(|()| ProviderError::CorruptTimeHighWatermark)?; + if now.value() < previous { + return Err(ProviderError::ClockRegression { + previous: UnixMillis::new(previous), + now, + }); + } + } + meta.insert(LAST_OBSERVED_TIME_KEY, now.value().to_be_bytes().as_slice())?; + Ok(now) +} + +fn expire_due_in_write( + write: &WriteTransaction, + now: UnixMillis, + limit: usize, +) -> Result, ProviderError> { + if limit == 0 { + return Ok(Vec::new()); + } + let due = { + let expirations = write.open_table(EXPIRATIONS)?; + let mut due = Vec::new(); + for entry in expirations.iter()? { + let (key, _) = entry?; + let (deadline, reservation_id) = decode_expiration_key(key.value())?; + if deadline > now { + break; + } + due.push((deadline, reservation_id)); + if due.len() == limit { + break; + } + } + due + }; + for (deadline, reservation_id) in &due { + let mut record = read_reservation_from_write(write, *reservation_id)?.ok_or_else(|| { + ProviderError::CorruptState( + "expiration index references a missing reservation".to_owned(), + ) + })?; + if record.accept_before != deadline.value() { + return Err(ProviderError::CorruptState( + "expiration index and reservation deadline disagree".to_owned(), + )); + } + match record.state { + StoredReservationState::Reserved => { + if UnixMillis::new(record.accept_before) > now { + return Err(ProviderError::CorruptState( + "expiration index precedes reservation deadline".to_owned(), + )); + } + release_reserved(write, &mut record, ReleaseReason::Expired, now)?; + } + _ => { + return Err(ProviderError::CorruptState( + "expiration index references a terminal reservation".to_owned(), + )); + } + } + } + Ok(due + .into_iter() + .map(|(_, reservation_id)| reservation_id) + .collect()) +} + +/// Lazily reclaim only expired reservations that block this request. This +/// keeps the hot path bounded by the request and reservation input limits; +/// the service is responsible for draining unrelated expirations through +/// [`ReservationBook::expire_due`] with an explicit batch size. +fn expire_requested_in_write( + write: &WriteTransaction, + now: UnixMillis, + requested: &[OutPoint], +) -> Result, ProviderError> { + let mut expired = Vec::new(); + for requested_outpoint in requested { + let Some(allocation) = read_record_from_write::( + write, + ALLOCATIONS, + &outpoint_key(*requested_outpoint), + )? + else { + continue; + }; + let StoredAllocation::Reserved { reservation_id } = allocation else { + continue; + }; + let reservation_id = ReservationId::new(reservation_id); + if expired.contains(&reservation_id) { + continue; + } + let mut record = read_reservation_from_write(write, reservation_id)?.ok_or_else(|| { + ProviderError::CorruptState("allocation references a missing reservation".to_owned()) + })?; + if !record.outpoints.contains(requested_outpoint) { + return Err(ProviderError::CorruptState( + "reservation does not contain its allocated outpoint".to_owned(), + )); + } + if !matches!(record.state, StoredReservationState::Reserved) { + return Err(ProviderError::CorruptState( + "reserved allocation references a terminal reservation".to_owned(), + )); + } + if now >= UnixMillis::new(record.accept_before) { + release_reserved(write, &mut record, ReleaseReason::Expired, now)?; + expired.push(reservation_id); + } + } + Ok(expired) +} + +fn release_reserved( + write: &WriteTransaction, + record: &mut StoredReservation, + reason: ReleaseReason, + at: UnixMillis, +) -> Result<(), ProviderError> { + if !matches!(record.state, StoredReservationState::Reserved) { + return Err(ProviderError::PointOfNoReturn(record.id())); + } + for outpoint in &record.outpoints { + let key = outpoint_key(*outpoint); + let allocation = read_record_from_write::(write, ALLOCATIONS, &key)? + .ok_or_else(|| { + ProviderError::CorruptState(format!( + "reserved outpoint {outpoint:?} has no allocation" + )) + })?; + if allocation + != (StoredAllocation::Reserved { + reservation_id: record.id, + }) + { + return Err(ProviderError::CorruptState(format!( + "reserved outpoint {outpoint:?} has a different allocation" + ))); + } + } + for outpoint in &record.outpoints { + write + .open_table(ALLOCATIONS)? + .remove(outpoint_key(*outpoint).as_slice())?; + #[cfg(test)] + mutation_failpoints::hit(mutation_failpoints::RELEASE_AFTER_ALLOCATION)?; + } + let expiration_key = expiration_key(UnixMillis::new(record.accept_before), record.id()); + let removed_expiration = { + let mut expirations = write.open_table(EXPIRATIONS)?; + expirations.remove(expiration_key.as_slice())?.is_some() + }; + if !removed_expiration { + return Err(ProviderError::CorruptState( + "reserved reservation has no expiration index entry".to_owned(), + )); + } + #[cfg(test)] + mutation_failpoints::hit(mutation_failpoints::RELEASE_AFTER_EXPIRATION)?; + record.state = StoredReservationState::Released { + reason: StoredReleaseReason::from(reason), + at: at.value(), + }; + write_record(write, RESERVATIONS, &record.id, record)?; + #[cfg(test)] + mutation_failpoints::hit(mutation_failpoints::RELEASE_AFTER_RECORD)?; + append_audit( + write, + at, + StoredAuditEvent::ReservationReleased { + reservation_id: record.id, + reason: reason.into(), + }, + )?; + #[cfg(test)] + mutation_failpoints::hit(mutation_failpoints::RELEASE_AFTER_AUDIT)?; + Ok(()) +} + +fn require_authorized_reservation( + write: &WriteTransaction, + access: ReservationAccess, +) -> Result { + let record = read_reservation_from_write(write, access.reservation_id())? + .ok_or(ProviderError::ReservationNotFound(access.reservation_id()))?; + if record.owner != access.owner().to_bytes() { + return Err(ProviderError::ReservationOwnerMismatch( + access.reservation_id(), + )); + } + Ok(record) +} + +fn read_reservation_from_write( + write: &WriteTransaction, + reservation_id: ReservationId, +) -> Result, ProviderError> { + let record = read_record_from_write::( + write, + RESERVATIONS, + &reservation_id.to_bytes(), + )?; + record + .map(|record| { + if record.id() != reservation_id { + return Err(ProviderError::CorruptState( + "reservation key and record ID disagree".to_owned(), + )); + } + record.validate()?; + Ok(record) + }) + .transpose() +} + +fn read_request_binding( + write: &WriteTransaction, + owner: OwnerId, + key: IdempotencyKey, +) -> Result, ProviderError> { + read_record_from_write(write, REQUEST_KEYS, &request_key(owner, key)) +} + +fn signing_commitment( + reservation: &StoredReservation, + pre_sign_payload: &[u8], + fee: TransactionFee, +) -> Result { + let transcript = StoredSigningTranscript { + request_digest: reservation.request_digest, + reservation_id: reservation.id, + outpoints: reservation.outpoints.clone(), + quote_commitment: reservation.quote_commitment, + fee_policy: reservation.fee_policy, + fee: StoredTransactionFee::from(fee), + pre_sign_payload, + }; + Ok(SigningCommitment::new(domain_digest( + SIGNING_DOMAIN, + &transcript, + )?)) +} + +fn signed_artifact_digest(commitment: SigningCommitment, bytes: &[u8]) -> SignedArtifactDigest { + let mut hasher = Sha256::new(); + hasher.update(SIGNED_ARTIFACT_DOMAIN); + hasher.update(commitment.to_bytes()); + hasher.update((bytes.len() as u64).to_be_bytes()); + hasher.update(bytes); + SignedArtifactDigest::new(hasher.finalize().into()) +} + +fn request_digest( + identity: ProviderIdentity, + plan: &ReservationPlan, +) -> Result<[u8; 32], ProviderError> { + let fingerprint = StoredRequestFingerprint { + identity: StoredProviderIdentity::from(identity), + owner: plan.owner().to_bytes(), + idempotency_key: plan.idempotency_key().to_bytes(), + quote_commitment: plan.quote_commitment().to_bytes(), + outpoints: plan.outpoints(), + accept_before: plan.accept_before().value(), + fee_policy: StoredFeePolicy::from(plan.fee_policy()), + }; + domain_digest(REQUEST_DOMAIN, &fingerprint) +} + +fn stored_request_digest( + identity: ProviderIdentity, + reservation: &StoredReservation, +) -> Result<[u8; 32], ProviderError> { + let fingerprint = StoredRequestFingerprint { + identity: StoredProviderIdentity::from(identity), + owner: reservation.owner, + idempotency_key: reservation.idempotency_key, + quote_commitment: reservation.quote_commitment, + outpoints: &reservation.outpoints, + accept_before: reservation.accept_before, + fee_policy: reservation.fee_policy, + }; + domain_digest(REQUEST_DOMAIN, &fingerprint) +} + +fn derive_reservation_id(owner: OwnerId, key: IdempotencyKey) -> ReservationId { + let mut hasher = Sha256::new(); + hasher.update(RESERVATION_ID_DOMAIN); + hasher.update(owner.to_bytes()); + hasher.update(key.to_bytes()); + ReservationId::new(hasher.finalize().into()) +} + +fn domain_digest(domain: &[u8], value: &T) -> Result<[u8; 32], ProviderError> { + let encoded = postcard::to_allocvec(value)?; + let mut hasher = Sha256::new(); + hasher.update(domain); + hasher.update((encoded.len() as u64).to_be_bytes()); + hasher.update(encoded); + Ok(hasher.finalize().into()) +} + +fn validate_settlement_bytes(bytes: &[u8]) -> Result<(), ProviderError> { + if bytes.is_empty() { + return Err(ProviderError::EmptySettlementPayload); + } + if bytes.len() > MAX_SETTLEMENT_BYTES { + return Err(ProviderError::SettlementPayloadTooLarge { + maximum: MAX_SETTLEMENT_BYTES, + actual: bytes.len(), + }); + } + Ok(()) +} + +fn append_audit( + write: &WriteTransaction, + at: UnixMillis, + event: StoredAuditEvent, +) -> Result<(), ProviderError> { + let sequence = { + let mut meta = write.open_table(META)?; + let current = { + let current = meta + .get(AUDIT_SEQUENCE_KEY)? + .ok_or(ProviderError::MissingMetadata(AUDIT_SEQUENCE_KEY))?; + current.value().to_vec() + }; + let current = decode_u64(¤t).map_err(|()| ProviderError::CorruptAuditSequence)?; + let next = current + .checked_add(1) + .ok_or(ProviderError::AuditSequenceOverflow)?; + meta.insert(AUDIT_SEQUENCE_KEY, next.to_be_bytes().as_slice())?; + next + }; + let entry = StoredAuditEntry { + sequence, + at: at.value(), + event, + }; + let encoded = encode_record(&entry)?; + write + .open_table(AUDIT)? + .insert(sequence, encoded.as_slice())?; + Ok(()) +} + +fn provider_tables_are_nonempty(write: &WriteTransaction) -> Result { + { + let table = write.open_table(META)?; + if table.iter()?.next().transpose()?.is_some() { + return Ok(true); + } + } + for definition in [ + INVENTORY, + ALLOCATIONS, + RESERVATIONS, + REQUEST_KEYS, + EXPIRATIONS, + ] { + let table = write.open_table(definition)?; + if table.iter()?.next().transpose()?.is_some() { + return Ok(true); + } + } + let audit = write.open_table(AUDIT)?; + Ok(audit.iter()?.next().transpose()?.is_some()) +} + +/// Validate every durable relationship before the database can answer an +/// availability question or return a recovery job. In particular, absence +/// from `ALLOCATIONS` means available only after this proves that no live or +/// committed reservation still owns the inventory outpoint. +fn validate_store_integrity( + write: &WriteTransaction, + identity: ProviderIdentity, +) -> Result<(), ProviderError> { + let (audit_sequence, last_observed_time) = { + let meta = write.open_table(META)?; + let audit_sequence = meta + .get(AUDIT_SEQUENCE_KEY)? + .ok_or(ProviderError::MissingMetadata(AUDIT_SEQUENCE_KEY))?; + let audit_sequence = + decode_u64(audit_sequence.value()).map_err(|()| ProviderError::CorruptAuditSequence)?; + let last_observed_time = meta + .get(LAST_OBSERVED_TIME_KEY)? + .map(|value| { + decode_u64(value.value()).map_err(|()| ProviderError::CorruptTimeHighWatermark) + }) + .transpose()?; + (audit_sequence, last_observed_time) + }; + + let inventory = { + let table = write.open_table(INVENTORY)?; + let mut records = BTreeMap::new(); + for entry in table.iter()? { + let (key, value) = entry?; + let key = decode_table_key::<36>("inventory", key.value())?; + let item: StoredInventoryItem = decode_record(value.value())?; + let domain = item.to_domain()?; + if key != outpoint_key(domain.outpoint()) { + return Err(ProviderError::CorruptState(format!( + "inventory key does not match record outpoint {:?}", + domain.outpoint() + ))); + } + records.insert(key, item); + } + records + }; + + let reservations = { + let table = write.open_table(RESERVATIONS)?; + let mut records = BTreeMap::new(); + for entry in table.iter()? { + let (key, value) = entry?; + let key = decode_table_key::<32>("reservation", key.value())?; + let record: StoredReservation = decode_record(value.value())?; + if key != record.id { + return Err(ProviderError::CorruptState( + "reservation key and record ID disagree".to_owned(), + )); + } + record.validate()?; + if record.fee_policy.policy_asset != identity.policy_asset() { + return Err(ProviderError::CorruptState(format!( + "reservation {:?} uses a fee policy for the wrong asset", + record.id() + ))); + } + let expected_request_digest = stored_request_digest(identity, &record)?; + if record.request_digest != expected_request_digest { + return Err(ProviderError::CorruptState(format!( + "reservation {:?} request digest does not match its immutable terms", + record.id() + ))); + } + validate_reservation_times(&record, last_observed_time)?; + records.insert(key, record); + } + records + }; + + let request_bindings = { + let table = write.open_table(REQUEST_KEYS)?; + let mut records = BTreeMap::new(); + for entry in table.iter()? { + let (key, value) = entry?; + let key = decode_table_key::<64>("request binding", key.value())?; + let binding: StoredRequestBinding = decode_record(value.value())?; + records.insert(key, binding); + } + records + }; + + let allocations = { + let table = write.open_table(ALLOCATIONS)?; + let mut records = BTreeMap::new(); + for entry in table.iter()? { + let (key, value) = entry?; + let key = decode_table_key::<36>("allocation", key.value())?; + let allocation: StoredAllocation = decode_record(value.value())?; + records.insert(key, allocation); + } + records + }; + + let expirations = { + let table = write.open_table(EXPIRATIONS)?; + let mut keys = BTreeSet::new(); + for entry in table.iter()? { + let (key, value) = entry?; + let key = decode_table_key::<40>("expiration", key.value())?; + if !value.value().is_empty() { + return Err(ProviderError::CorruptState( + "expiration index value is not empty".to_owned(), + )); + } + keys.insert(key); + } + keys + }; + + for record in reservations.values() { + let expected_request_key = request_key( + OwnerId::new(record.owner), + IdempotencyKey::new(record.idempotency_key), + ); + let binding = request_bindings.get(&expected_request_key).ok_or_else(|| { + ProviderError::CorruptState(format!( + "reservation {:?} has no request-key binding", + record.id() + )) + })?; + if binding.reservation_id != record.id || binding.request_digest != record.request_digest { + return Err(ProviderError::CorruptState(format!( + "reservation {:?} request-key binding disagrees with its record", + record.id() + ))); + } + + let expected_expiration = + expiration_key(UnixMillis::new(record.accept_before), record.id()); + let has_expiration = expirations.contains(&expected_expiration); + match &record.state { + StoredReservationState::Reserved => { + if !has_expiration { + return Err(ProviderError::CorruptState(format!( + "reserved reservation {:?} has no expiration index entry", + record.id() + ))); + } + } + StoredReservationState::Released { .. } + | StoredReservationState::Committed { .. } + | StoredReservationState::Signed { .. } => { + if has_expiration { + return Err(ProviderError::CorruptState(format!( + "terminal reservation {:?} still has an expiration index entry", + record.id() + ))); + } + } + } + + for outpoint in &record.outpoints { + let key = outpoint_key(*outpoint); + if !inventory.contains_key(&key) { + return Err(ProviderError::CorruptState(format!( + "reservation {:?} references missing inventory {outpoint:?}", + record.id() + ))); + } + let allocation = allocations.get(&key); + match &record.state { + StoredReservationState::Reserved => { + if allocation + != Some(&StoredAllocation::Reserved { + reservation_id: record.id, + }) + { + return Err(ProviderError::CorruptState(format!( + "reserved reservation {:?} does not own allocation {outpoint:?}", + record.id() + ))); + } + } + StoredReservationState::Committed { intent } + | StoredReservationState::Signed { intent, .. } => { + if allocation + != Some(&StoredAllocation::Committed { + reservation_id: record.id, + commitment: intent.commitment, + }) + { + return Err(ProviderError::CorruptState(format!( + "committed reservation {:?} does not own its permanent allocation {outpoint:?}", + record.id() + ))); + } + } + StoredReservationState::Released { .. } => { + let still_owned = allocation.is_some_and(|allocation| match allocation { + StoredAllocation::Reserved { reservation_id } + | StoredAllocation::Committed { reservation_id, .. } => { + *reservation_id == record.id + } + }); + if still_owned { + return Err(ProviderError::CorruptState(format!( + "released reservation {:?} still owns allocation {outpoint:?}", + record.id() + ))); + } + } + } + } + } + + for (key, binding) in &request_bindings { + let record = reservations.get(&binding.reservation_id).ok_or_else(|| { + ProviderError::CorruptState( + "request-key binding references a missing reservation".to_owned(), + ) + })?; + let expected_key = request_key( + OwnerId::new(record.owner), + IdempotencyKey::new(record.idempotency_key), + ); + if *key != expected_key || binding.request_digest != record.request_digest { + return Err(ProviderError::CorruptState(format!( + "request-key binding for reservation {:?} has inconsistent key or digest", + record.id() + ))); + } + } + + for (key, allocation) in &allocations { + let item = inventory.get(key).ok_or_else(|| { + ProviderError::CorruptState("allocation references missing inventory".to_owned()) + })?; + let (reservation_id, allocated_commitment) = match allocation { + StoredAllocation::Reserved { reservation_id } => (*reservation_id, None), + StoredAllocation::Committed { + reservation_id, + commitment, + } => (*reservation_id, Some(*commitment)), + }; + let record = reservations.get(&reservation_id).ok_or_else(|| { + ProviderError::CorruptState("allocation references a missing reservation".to_owned()) + })?; + if record.outpoints.binary_search(&item.outpoint).is_err() { + return Err(ProviderError::CorruptState(format!( + "allocation for {:?} is absent from reservation {:?}", + item.outpoint, + record.id() + ))); + } + match (&record.state, allocated_commitment) { + (StoredReservationState::Reserved, None) => {} + (StoredReservationState::Committed { intent }, Some(commitment)) + | (StoredReservationState::Signed { intent, .. }, Some(commitment)) + if intent.commitment == commitment => {} + _ => { + return Err(ProviderError::CorruptState(format!( + "allocation for {:?} disagrees with reservation {:?} state", + item.outpoint, + record.id() + ))); + } + } + } + + for key in &expirations { + let (deadline, reservation_id) = decode_expiration_key(key)?; + let record = reservations + .get(&reservation_id.to_bytes()) + .ok_or_else(|| { + ProviderError::CorruptState( + "expiration index references a missing reservation".to_owned(), + ) + })?; + if !matches!(record.state, StoredReservationState::Reserved) + || record.accept_before != deadline.value() + { + return Err(ProviderError::CorruptState(format!( + "expiration index disagrees with reservation {:?}", + record.id() + ))); + } + } + + validate_audit_integrity( + write, + audit_sequence, + last_observed_time, + &inventory, + &reservations, + ) +} + +fn validate_reservation_times( + record: &StoredReservation, + last_observed_time: Option, +) -> Result<(), ProviderError> { + let high_watermark = last_observed_time.ok_or_else(|| { + ProviderError::CorruptState(format!( + "reservation {:?} exists without a clock high-water mark", + record.id() + )) + })?; + if record.created_at > high_watermark { + return Err(ProviderError::CorruptState(format!( + "reservation {:?} was created after the clock high-water mark", + record.id() + ))); + } + match &record.state { + StoredReservationState::Reserved => {} + StoredReservationState::Released { reason, at } => { + if *at < record.created_at || *at > high_watermark { + return Err(ProviderError::CorruptState(format!( + "reservation {:?} has an invalid release time", + record.id() + ))); + } + let deadline_relation_is_valid = match reason { + StoredReleaseReason::Expired => *at >= record.accept_before, + StoredReleaseReason::ClientCancelled | StoredReleaseReason::ProviderRejected => { + *at < record.accept_before + } + }; + if !deadline_relation_is_valid { + return Err(ProviderError::CorruptState(format!( + "reservation {:?} release reason disagrees with its deadline", + record.id() + ))); + } + } + StoredReservationState::Committed { intent } => { + validate_commit_time(record, intent.committed_at, high_watermark)?; + } + StoredReservationState::Signed { intent, artifact } => { + validate_commit_time(record, intent.committed_at, high_watermark)?; + if artifact.signed_at < intent.committed_at || artifact.signed_at > high_watermark { + return Err(ProviderError::CorruptState(format!( + "reservation {:?} has an invalid signed-artifact time", + record.id() + ))); + } + } + } + Ok(()) +} + +fn validate_commit_time( + record: &StoredReservation, + committed_at: u64, + high_watermark: u64, +) -> Result<(), ProviderError> { + if committed_at < record.created_at + || committed_at >= record.accept_before + || committed_at > high_watermark + { + return Err(ProviderError::CorruptState(format!( + "reservation {:?} has an invalid commit time", + record.id() + ))); + } + Ok(()) +} + +fn validate_audit_integrity( + write: &WriteTransaction, + declared_sequence: u64, + last_observed_time: Option, + inventory: &BTreeMap<[u8; 36], StoredInventoryItem>, + reservations: &BTreeMap<[u8; 32], StoredReservation>, +) -> Result<(), ProviderError> { + let table = write.open_table(AUDIT)?; + let mut previous_sequence = 0_u64; + let mut previous_time = None; + let mut imported = BTreeSet::new(); + let mut created = BTreeSet::new(); + let mut released = BTreeSet::new(); + let mut committed = BTreeSet::new(); + let mut signed = BTreeSet::new(); + + for entry in table.iter()? { + let (sequence, value) = entry?; + let sequence = sequence.value(); + let expected = previous_sequence + .checked_add(1) + .ok_or(ProviderError::AuditSequenceOverflow)?; + if sequence != expected { + return Err(ProviderError::CorruptState(format!( + "audit sequence is not contiguous: expected {expected}, found {sequence}" + ))); + } + let entry: StoredAuditEntry = decode_record(value.value())?; + if entry.sequence != sequence { + return Err(ProviderError::CorruptState( + "audit key and record sequence disagree".to_owned(), + )); + } + if previous_time.is_some_and(|previous| entry.at < previous) { + return Err(ProviderError::CorruptState( + "audit timestamps moved backwards".to_owned(), + )); + } + let high_watermark = last_observed_time.ok_or_else(|| { + ProviderError::CorruptState( + "audit entries exist without a clock high-water mark".to_owned(), + ) + })?; + if entry.at > high_watermark { + return Err(ProviderError::CorruptState( + "audit entry is later than the clock high-water mark".to_owned(), + )); + } + match &entry.event { + StoredAuditEvent::InventoryImported { outpoint } => { + let key = outpoint_key(*outpoint); + if !inventory.contains_key(&key) || !imported.insert(key) { + return Err(ProviderError::CorruptState( + "inventory import audit entry is missing its record or duplicated" + .to_owned(), + )); + } + } + StoredAuditEvent::ReservationCreated { + reservation_id, + outpoints, + } => { + let record = reservations.get(reservation_id).ok_or_else(|| { + ProviderError::CorruptState( + "reservation-created audit entry references a missing reservation" + .to_owned(), + ) + })?; + if entry.at != record.created_at + || *outpoints != record.outpoints + || !created.insert(*reservation_id) + { + return Err(ProviderError::CorruptState(format!( + "reservation-created audit entry disagrees with reservation {:?}", + record.id() + ))); + } + } + StoredAuditEvent::ReservationReleased { + reservation_id, + reason, + } => { + let record = reservations.get(reservation_id).ok_or_else(|| { + ProviderError::CorruptState( + "reservation-released audit entry references a missing reservation" + .to_owned(), + ) + })?; + let matches_state = matches!( + record.state, + StoredReservationState::Released { + reason: stored_reason, + at, + } if stored_reason == *reason && at == entry.at + ); + if !matches_state || !released.insert(*reservation_id) { + return Err(ProviderError::CorruptState(format!( + "reservation-released audit entry disagrees with reservation {:?}", + record.id() + ))); + } + } + StoredAuditEvent::SigningCommitted { + reservation_id, + commitment, + } => { + let record = reservations.get(reservation_id).ok_or_else(|| { + ProviderError::CorruptState( + "signing-committed audit entry references a missing reservation".to_owned(), + ) + })?; + let matches_state = matches!( + &record.state, + StoredReservationState::Committed { intent } + | StoredReservationState::Signed { intent, .. } + if intent.commitment == *commitment && intent.committed_at == entry.at + ); + if !matches_state || !committed.insert(*reservation_id) { + return Err(ProviderError::CorruptState(format!( + "signing-committed audit entry disagrees with reservation {:?}", + record.id() + ))); + } + } + StoredAuditEvent::SignedArtifactStored { + reservation_id, + artifact, + } => { + let record = reservations.get(reservation_id).ok_or_else(|| { + ProviderError::CorruptState( + "signed-artifact audit entry references a missing reservation".to_owned(), + ) + })?; + let matches_state = matches!( + &record.state, + StoredReservationState::Signed { + artifact: stored_artifact, + .. + } if stored_artifact.digest == *artifact && stored_artifact.signed_at == entry.at + ); + if !matches_state || !signed.insert(*reservation_id) { + return Err(ProviderError::CorruptState(format!( + "signed-artifact audit entry disagrees with reservation {:?}", + record.id() + ))); + } + } + } + previous_sequence = sequence; + previous_time = Some(entry.at); + } + + if previous_sequence != declared_sequence { + return Err(ProviderError::CorruptState(format!( + "audit sequence metadata is {declared_sequence}, but the log ends at {previous_sequence}" + ))); + } + for key in inventory.keys() { + if !imported.contains(key) { + return Err(ProviderError::CorruptState( + "inventory record has no import audit entry".to_owned(), + )); + } + } + for (reservation_id, record) in reservations { + if !created.contains(reservation_id) { + return Err(ProviderError::CorruptState(format!( + "reservation {:?} has no creation audit entry", + record.id() + ))); + } + let audit_shape_is_valid = match record.state { + StoredReservationState::Reserved => { + !released.contains(reservation_id) + && !committed.contains(reservation_id) + && !signed.contains(reservation_id) + } + StoredReservationState::Released { .. } => { + released.contains(reservation_id) + && !committed.contains(reservation_id) + && !signed.contains(reservation_id) + } + StoredReservationState::Committed { .. } => { + !released.contains(reservation_id) + && committed.contains(reservation_id) + && !signed.contains(reservation_id) + } + StoredReservationState::Signed { .. } => { + !released.contains(reservation_id) + && committed.contains(reservation_id) + && signed.contains(reservation_id) + } + }; + if !audit_shape_is_valid { + return Err(ProviderError::CorruptState(format!( + "reservation {:?} state disagrees with its audit history", + record.id() + ))); + } + } + Ok(()) +} + +fn decode_table_key( + table: &'static str, + bytes: &[u8], +) -> Result<[u8; LENGTH], ProviderError> { + bytes.try_into().map_err(|_| { + ProviderError::CorruptState(format!( + "{table} key has length {}, expected {LENGTH}", + bytes.len() + )) + }) +} + +fn create_tables(write: &WriteTransaction) -> Result<(), ProviderError> { + write.open_table(INVENTORY)?; + write.open_table(ALLOCATIONS)?; + write.open_table(RESERVATIONS)?; + write.open_table(REQUEST_KEYS)?; + write.open_table(EXPIRATIONS)?; + write.open_table(AUDIT)?; + Ok(()) +} + +fn write_record( + write: &WriteTransaction, + definition: TableDefinition<&[u8], &[u8]>, + key: &[u8], + value: &T, +) -> Result<(), ProviderError> { + let encoded = encode_record(value)?; + write + .open_table(definition)? + .insert(key, encoded.as_slice())?; + Ok(()) +} + +fn read_record_from_write( + write: &WriteTransaction, + definition: TableDefinition<&[u8], &[u8]>, + key: &[u8], +) -> Result, ProviderError> { + let table = write.open_table(definition)?; + table + .get(key)? + .map(|value| decode_record(value.value())) + .transpose() +} + +fn encode_record(value: &T) -> Result, ProviderError> { + let mut encoded = Vec::with_capacity(64); + encoded.push(RECORD_VERSION); + encoded.extend(postcard::to_allocvec(value)?); + Ok(encoded) +} + +fn decode_record(bytes: &[u8]) -> Result { + let (&version, payload) = bytes.split_first().ok_or(ProviderError::EmptyRecord)?; + if version != RECORD_VERSION { + return Err(ProviderError::RecordVersionMismatch { + expected: RECORD_VERSION, + actual: version, + }); + } + let (value, trailing) = postcard::take_from_bytes(payload)?; + if !trailing.is_empty() { + return Err(ProviderError::TrailingRecordBytes(trailing.len())); + } + Ok(value) +} + +fn outpoint_key(outpoint: OutPoint) -> [u8; 36] { + let mut key = [0_u8; 36]; + key[..32].copy_from_slice(&outpoint.txid.to_byte_array()); + key[32..].copy_from_slice(&outpoint.vout.to_be_bytes()); + key +} + +fn request_key(owner: OwnerId, key: IdempotencyKey) -> [u8; 64] { + let mut encoded = [0_u8; 64]; + encoded[..32].copy_from_slice(&owner.to_bytes()); + encoded[32..].copy_from_slice(&key.to_bytes()); + encoded +} + +fn expiration_key(deadline: UnixMillis, reservation_id: ReservationId) -> [u8; 40] { + let mut key = [0_u8; 40]; + key[..8].copy_from_slice(&deadline.value().to_be_bytes()); + key[8..].copy_from_slice(&reservation_id.to_bytes()); + key +} + +fn decode_expiration_key(bytes: &[u8]) -> Result<(UnixMillis, ReservationId), ProviderError> { + let bytes: [u8; 40] = bytes + .try_into() + .map_err(|_| ProviderError::CorruptExpirationKey(bytes.len()))?; + let deadline = u64::from_be_bytes(bytes[..8].try_into().expect("fixed slice")); + let reservation_id = bytes[8..].try_into().expect("fixed slice"); + Ok(( + UnixMillis::new(deadline), + ReservationId::new(reservation_id), + )) +} + +fn decode_u32(bytes: &[u8]) -> Result { + Ok(u32::from_be_bytes(bytes.try_into().map_err(|_| ())?)) +} + +fn decode_u64(bytes: &[u8]) -> Result { + Ok(u64::from_be_bytes(bytes.try_into().map_err(|_| ())?)) +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)] +struct StoredProviderIdentity { + provider: [u8; 32], + genesis_hash: BlockHash, + policy_asset: AssetId, +} + +impl From for StoredProviderIdentity { + fn from(value: ProviderIdentity) -> Self { + Self { + provider: value.provider().to_bytes(), + genesis_hash: value.genesis_hash(), + policy_asset: value.policy_asset(), + } + } +} + +impl StoredProviderIdentity { + fn to_domain(self) -> ProviderIdentity { + ProviderIdentity::new( + ProviderId::new(self.provider), + self.genesis_hash, + self.policy_asset, + ) + } +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)] +struct StoredInventoryItem { + outpoint: OutPoint, + asset: AssetId, + amount: u64, +} + +impl From for StoredInventoryItem { + fn from(value: InventoryItem) -> Self { + Self { + outpoint: value.outpoint(), + asset: value.asset(), + amount: value.amount(), + } + } +} + +impl StoredInventoryItem { + fn to_domain(self) -> Result { + InventoryItem::new(self.outpoint, self.asset, self.amount).map_err(|error| { + ProviderError::CorruptState(format!("invalid persisted inventory: {error}")) + }) + } +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)] +enum StoredFeeSizeMetric { + RegularVbytes, + DiscountVbytes, +} + +impl From for StoredFeeSizeMetric { + fn from(value: FeeSizeMetric) -> Self { + match value { + FeeSizeMetric::RegularVbytes => Self::RegularVbytes, + FeeSizeMetric::DiscountVbytes => Self::DiscountVbytes, + } + } +} + +impl StoredFeeSizeMetric { + const fn to_domain(self) -> FeeSizeMetric { + match self { + Self::RegularVbytes => FeeSizeMetric::RegularVbytes, + Self::DiscountVbytes => FeeSizeMetric::DiscountVbytes, + } + } +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)] +struct StoredFeePolicy { + policy_asset: AssetId, + minimum_sats_per_kvb: u64, + minimum_absolute_fee: u64, + maximum_transaction_weight: u64, + size_metric: StoredFeeSizeMetric, +} + +impl From for StoredFeePolicy { + fn from(value: FeePolicy) -> Self { + Self { + policy_asset: value.policy_asset(), + minimum_sats_per_kvb: value.minimum_sats_per_kvb(), + minimum_absolute_fee: value.minimum_absolute_fee(), + maximum_transaction_weight: value.maximum_transaction_weight(), + size_metric: value.size_metric().into(), + } + } +} + +impl StoredFeePolicy { + fn to_domain(self) -> Result { + FeePolicy::new( + self.policy_asset, + self.minimum_sats_per_kvb, + self.minimum_absolute_fee, + self.maximum_transaction_weight, + self.size_metric.to_domain(), + ) + .map_err(|error| { + ProviderError::CorruptState(format!("invalid persisted fee policy: {error}")) + }) + } +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)] +struct StoredTransactionFee { + policy_asset: AssetId, + amount: u64, + weight: u64, + regular_vsize: u64, + discount_vsize: u64, +} + +impl From for StoredTransactionFee { + fn from(value: TransactionFee) -> Self { + Self { + policy_asset: value.policy_asset(), + amount: value.amount(), + weight: value.weight(), + regular_vsize: value.regular_vsize(), + discount_vsize: value.discount_vsize(), + } + } +} + +impl StoredTransactionFee { + fn to_domain(self) -> Result { + TransactionFee::new( + self.policy_asset, + self.amount, + self.weight, + self.regular_vsize, + self.discount_vsize, + ) + .map_err(|error| { + ProviderError::CorruptState(format!("invalid persisted transaction fee: {error}")) + }) + } +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)] +enum StoredAllocation { + Reserved { + reservation_id: [u8; 32], + }, + Committed { + reservation_id: [u8; 32], + commitment: [u8; 32], + }, +} + +impl StoredAllocation { + const fn to_view(self) -> InventoryState { + match self { + Self::Reserved { reservation_id } => InventoryState::Reserved { + reservation_id: ReservationId::new(reservation_id), + }, + Self::Committed { + reservation_id, + commitment, + } => InventoryState::Committed { + reservation_id: ReservationId::new(reservation_id), + commitment: SigningCommitment::new(commitment), + }, + } + } +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)] +enum StoredReleaseReason { + Expired, + ClientCancelled, + ProviderRejected, +} + +impl From for StoredReleaseReason { + fn from(value: ReleaseReason) -> Self { + match value { + ReleaseReason::Expired => Self::Expired, + ReleaseReason::ClientCancelled => Self::ClientCancelled, + ReleaseReason::ProviderRejected => Self::ProviderRejected, + } + } +} + +impl StoredReleaseReason { + const fn to_domain(self) -> ReleaseReason { + match self { + Self::Expired => ReleaseReason::Expired, + Self::ClientCancelled => ReleaseReason::ClientCancelled, + Self::ProviderRejected => ReleaseReason::ProviderRejected, + } + } +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +struct StoredSigningIntent { + commitment: [u8; 32], + pre_sign_payload: Vec, + fee: StoredTransactionFee, + committed_at: u64, +} + +impl StoredSigningIntent { + fn to_job(&self, reservation_id: ReservationId) -> Result { + validate_settlement_bytes(&self.pre_sign_payload).map_err(|error| { + ProviderError::CorruptState(format!("invalid persisted signing payload: {error}")) + })?; + Ok(SigningJob { + reservation_id, + commitment: SigningCommitment::new(self.commitment), + pre_sign_payload: self.pre_sign_payload.clone(), + fee: self.fee.to_domain()?, + }) + } +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +struct StoredSignedArtifact { + digest: [u8; 32], + bytes: Vec, + signed_at: u64, +} + +impl StoredSignedArtifact { + fn to_domain( + &self, + reservation_id: ReservationId, + commitment: SigningCommitment, + ) -> Result { + validate_settlement_bytes(&self.bytes).map_err(|error| { + ProviderError::CorruptState(format!("invalid persisted signed artifact: {error}")) + })?; + let expected = signed_artifact_digest(commitment, &self.bytes); + if expected.to_bytes() != self.digest { + return Err(ProviderError::CorruptState( + "persisted signed artifact digest does not match its bytes".to_owned(), + )); + } + Ok(SignedArtifact { + reservation_id, + commitment, + digest: SignedArtifactDigest::new(self.digest), + bytes: self.bytes.clone(), + }) + } +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +enum StoredReservationState { + Reserved, + Released { + reason: StoredReleaseReason, + at: u64, + }, + Committed { + intent: StoredSigningIntent, + }, + Signed { + intent: StoredSigningIntent, + artifact: StoredSignedArtifact, + }, +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +struct StoredReservation { + id: [u8; 32], + owner: [u8; 32], + idempotency_key: [u8; 32], + request_digest: [u8; 32], + quote_commitment: [u8; 32], + outpoints: Vec, + created_at: u64, + accept_before: u64, + fee_policy: StoredFeePolicy, + state: StoredReservationState, +} + +impl StoredReservation { + const fn id(&self) -> ReservationId { + ReservationId::new(self.id) + } + + fn validate(&self) -> Result<(), ProviderError> { + if self.id() + != derive_reservation_id( + OwnerId::new(self.owner), + IdempotencyKey::new(self.idempotency_key), + ) + { + return Err(ProviderError::CorruptState( + "reservation ID does not match its owner and idempotency key".to_owned(), + )); + } + if self.outpoints.is_empty() || self.outpoints.len() > MAX_RESERVATION_INPUTS { + return Err(ProviderError::CorruptState( + "reservation has an invalid outpoint count".to_owned(), + )); + } + if self.accept_before <= self.created_at { + return Err(ProviderError::CorruptState( + "reservation deadline is not after its creation time".to_owned(), + )); + } + for outpoint in &self.outpoints { + if outpoint.is_null() || outpoint.vout & 0xc000_0000 != 0 { + return Err(ProviderError::CorruptState(format!( + "reservation contains invalid outpoint {outpoint:?}" + ))); + } + } + if self.outpoints.windows(2).any(|pair| pair[0] >= pair[1]) { + return Err(ProviderError::CorruptState( + "reservation outpoints are not strictly sorted".to_owned(), + )); + } + let policy = self.fee_policy.to_domain()?; + match &self.state { + StoredReservationState::Committed { intent } + | StoredReservationState::Signed { intent, .. } => { + validate_settlement_bytes(&intent.pre_sign_payload).map_err(|error| { + ProviderError::CorruptState(format!( + "invalid persisted signing payload: {error}" + )) + })?; + let fee = intent.fee.to_domain()?; + policy.validate(fee).map_err(|error| { + ProviderError::CorruptState(format!( + "persisted signing fee violates its policy: {error}" + )) + })?; + let expected = signing_commitment(self, &intent.pre_sign_payload, fee)?; + if expected.to_bytes() != intent.commitment { + return Err(ProviderError::CorruptState( + "persisted signing commitment does not match its transcript".to_owned(), + )); + } + } + StoredReservationState::Reserved | StoredReservationState::Released { .. } => {} + } + if let StoredReservationState::Signed { intent, artifact } = &self.state { + artifact.to_domain(self.id(), SigningCommitment::new(intent.commitment))?; + } + Ok(()) + } + + fn to_view(&self) -> Result { + self.validate()?; + let fee_policy = self.fee_policy.to_domain()?; + let state = match &self.state { + StoredReservationState::Reserved => ReservationState::Reserved, + StoredReservationState::Released { reason, at } => ReservationState::Released { + reason: reason.to_domain(), + at: UnixMillis::new(*at), + }, + StoredReservationState::Committed { intent } => ReservationState::Committed { + commitment: SigningCommitment::new(intent.commitment), + committed_at: UnixMillis::new(intent.committed_at), + }, + StoredReservationState::Signed { intent, artifact } => ReservationState::Signed { + commitment: SigningCommitment::new(intent.commitment), + artifact: SignedArtifactDigest::new(artifact.digest), + committed_at: UnixMillis::new(intent.committed_at), + signed_at: UnixMillis::new(artifact.signed_at), + }, + }; + Ok(ReservationView { + id: self.id(), + owner: OwnerId::new(self.owner), + quote_commitment: QuoteCommitment::new(self.quote_commitment), + outpoints: self.outpoints.clone(), + created_at: UnixMillis::new(self.created_at), + accept_before: UnixMillis::new(self.accept_before), + fee_policy, + state, + }) + } +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)] +struct StoredRequestBinding { + reservation_id: [u8; 32], + request_digest: [u8; 32], +} + +#[derive(Serialize)] +struct StoredRequestFingerprint<'a> { + identity: StoredProviderIdentity, + owner: [u8; 32], + idempotency_key: [u8; 32], + quote_commitment: [u8; 32], + outpoints: &'a [OutPoint], + accept_before: u64, + fee_policy: StoredFeePolicy, +} + +#[derive(Serialize)] +struct StoredSigningTranscript<'a> { + request_digest: [u8; 32], + reservation_id: [u8; 32], + outpoints: Vec, + quote_commitment: [u8; 32], + fee_policy: StoredFeePolicy, + fee: StoredTransactionFee, + pre_sign_payload: &'a [u8], +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +struct StoredAuditEntry { + sequence: u64, + at: u64, + event: StoredAuditEvent, +} + +impl StoredAuditEntry { + fn to_domain(&self) -> AuditEntry { + AuditEntry { + sequence: self.sequence, + at: UnixMillis::new(self.at), + event: self.event.to_domain(), + } + } +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +enum StoredAuditEvent { + InventoryImported { + outpoint: OutPoint, + }, + ReservationCreated { + reservation_id: [u8; 32], + outpoints: Vec, + }, + ReservationReleased { + reservation_id: [u8; 32], + reason: StoredReleaseReason, + }, + SigningCommitted { + reservation_id: [u8; 32], + commitment: [u8; 32], + }, + SignedArtifactStored { + reservation_id: [u8; 32], + artifact: [u8; 32], + }, +} + +impl StoredAuditEvent { + fn to_domain(&self) -> AuditEvent { + match self { + Self::InventoryImported { outpoint } => AuditEvent::InventoryImported { + outpoint: *outpoint, + }, + Self::ReservationCreated { + reservation_id, + outpoints, + } => AuditEvent::ReservationCreated { + reservation_id: ReservationId::new(*reservation_id), + outpoints: outpoints.clone(), + }, + Self::ReservationReleased { + reservation_id, + reason, + } => AuditEvent::ReservationReleased { + reservation_id: ReservationId::new(*reservation_id), + reason: reason.to_domain(), + }, + Self::SigningCommitted { + reservation_id, + commitment, + } => AuditEvent::SigningCommitted { + reservation_id: ReservationId::new(*reservation_id), + commitment: SigningCommitment::new(*commitment), + }, + Self::SignedArtifactStored { + reservation_id, + artifact, + } => AuditEvent::SignedArtifactStored { + reservation_id: ReservationId::new(*reservation_id), + artifact: SignedArtifactDigest::new(*artifact), + }, + } + } +} + +#[derive(Debug, Error)] +pub enum ProviderError { + #[cfg(test)] + #[error("injected provider mutation failure at {0}")] + InjectedMutationFailure(&'static str), + #[error("redb database error: {0}")] + Database(#[from] redb::DatabaseError), + #[error("redb transaction error: {0}")] + Transaction(#[from] redb::TransactionError), + #[error("redb table error: {0}")] + Table(#[from] redb::TableError), + #[error("redb storage error: {0}")] + Storage(#[from] redb::StorageError), + #[error("redb commit error: {0}")] + Commit(#[from] redb::CommitError), + #[error("a prior commit was ambiguous; close and reopen the provider database")] + DatabaseRequiresReopen, + #[error("the provider operation lock was poisoned; close and reopen the provider database")] + OperationLockPoisoned, + #[error("redb durability configuration error: {0}")] + Durability(#[from] redb::SetDurabilityError), + #[error("record codec error: {0}")] + Codec(#[from] postcard::Error), + #[error("schema version has an invalid encoding")] + CorruptSchemaVersion, + #[error("schema mismatch: expected {expected}, found {actual}")] + SchemaMismatch { expected: u32, actual: u32 }, + #[error("required metadata is missing: {0}")] + MissingMetadata(&'static str), + #[error("provider database identity mismatch: database has {expected:?}, requested {actual:?}")] + ProviderIdentityMismatch { + expected: Box, + actual: Box, + }, + #[error("persisted record is empty")] + EmptyRecord, + #[error("record version mismatch: expected {expected}, found {actual}")] + RecordVersionMismatch { expected: u8, actual: u8 }, + #[error("persisted record has {0} trailing bytes from an incompatible shape")] + TrailingRecordBytes(usize), + #[error("persisted clock high-water mark is corrupt")] + CorruptTimeHighWatermark, + #[error("clock moved backwards from {previous:?} to {now:?}")] + ClockRegression { + previous: UnixMillis, + now: UnixMillis, + }, + #[error("persisted audit sequence is corrupt")] + CorruptAuditSequence, + #[error("audit sequence overflowed")] + AuditSequenceOverflow, + #[error("expiration index key has length {0}, expected 40")] + CorruptExpirationKey(usize), + #[error("provider state is internally inconsistent: {0}")] + CorruptState(String), + #[error("fee policy uses {actual}, expected provider policy asset {expected}")] + WrongPolicyAsset { expected: AssetId, actual: AssetId }, + #[error("inventory outpoint is unknown: {0:?}")] + UnknownInventory(OutPoint), + #[error("inventory metadata conflicts at {outpoint:?}")] + InventoryMetadataConflict { outpoint: OutPoint }, + #[error("outpoint {outpoint:?} is unavailable: {state:?}")] + OutpointUnavailable { + outpoint: OutPoint, + state: InventoryState, + }, + #[error("idempotency key {key:?} for owner {owner:?} was reused with different terms")] + IdempotencyConflict { owner: OwnerId, key: IdempotencyKey }, + #[error("derived reservation ID collided: {0:?}")] + ReservationIdCollision(ReservationId), + #[error("reservation deadline {accept_before:?} elapsed at {now:?}")] + ReservationDeadlineElapsed { + accept_before: UnixMillis, + now: UnixMillis, + }, + #[error("reservation not found: {0:?}")] + ReservationNotFound(ReservationId), + #[error("reservation owner authentication failed: {0:?}")] + ReservationOwnerMismatch(ReservationId), + #[error("reservation is already released: {0:?}")] + ReservationAlreadyReleased(ReservationId), + #[error("reservation crossed the irreversible signing point: {0:?}")] + PointOfNoReturn(ReservationId), + #[error("reservation is already committed to a different signing intent: {0:?}")] + DifferentSigningIntent(ReservationId), + #[error("settlement payload must not be empty")] + EmptySettlementPayload, + #[error("settlement payload has {actual} bytes; maximum is {maximum}")] + SettlementPayloadTooLarge { maximum: usize, actual: usize }, + #[error("reservation has not committed a signing intent: {0:?}")] + SigningIntentNotCommitted(ReservationId), + #[error( + "reservation {reservation_id:?} expects signing commitment {expected:?}, got {actual:?}" + )] + SigningCommitmentMismatch { + reservation_id: ReservationId, + expected: SigningCommitment, + actual: SigningCommitment, + }, + #[error("reservation already stored a different signed artifact: {0:?}")] + DifferentSignedArtifact(ReservationId), + #[error("fee policy rejected the final transaction: {0}")] + FeePolicy(#[from] FeePolicyViolation), +} + +#[cfg(test)] +mod tests; diff --git a/crates/deadcat-rfq-provider/src/store/tests.rs b/crates/deadcat-rfq-provider/src/store/tests.rs new file mode 100644 index 0000000..439e0bc --- /dev/null +++ b/crates/deadcat-rfq-provider/src/store/tests.rs @@ -0,0 +1,1599 @@ +use std::sync::{Arc, Barrier}; +use std::thread; + +use elements::hashes::Hash as _; +use elements::{AssetId, BlockHash, OutPoint, Txid}; +use tempfile::TempDir; + +use super::*; + +fn asset(marker: u8) -> AssetId { + AssetId::from_slice(&[marker; 32]).expect("asset") +} + +fn outpoint(marker: u8, vout: u32) -> OutPoint { + OutPoint::new(Txid::from_byte_array([marker; 32]), vout) +} + +fn identity(marker: u8) -> ProviderIdentity { + ProviderIdentity::new( + ProviderId::new([marker; 32]), + BlockHash::from_byte_array([marker.wrapping_add(1); 32]), + asset(1), + ) +} + +fn open_book(directory: &TempDir, identity: ProviderIdentity) -> ReservationBook { + ReservationBook::open(directory.path().join("provider.redb"), identity).expect("book") +} + +fn fee_policy(identity: ProviderIdentity) -> FeePolicy { + FeePolicy::new( + identity.policy_asset(), + 2_000, + 50, + 4_000, + FeeSizeMetric::DiscountVbytes, + ) + .expect("fee policy") +} + +fn transaction_fee(identity: ProviderIdentity, amount: u64) -> TransactionFee { + TransactionFee::new(identity.policy_asset(), amount, 800, 200, 100).expect("transaction fee") +} + +fn inventory(marker: u8) -> InventoryItem { + InventoryItem::new(outpoint(marker, 0), asset(2), 10_000).expect("inventory") +} + +fn owner(marker: u8) -> OwnerId { + OwnerId::new([marker; 32]) +} + +fn plan( + identity: ProviderIdentity, + owner: OwnerId, + request_marker: u8, + quote_marker: u8, + outpoints: Vec, + deadline: u64, +) -> ReservationPlan { + ReservationPlan::new( + owner, + IdempotencyKey::new([request_marker; 32]), + QuoteCommitment::new([quote_marker; 32]), + outpoints, + UnixMillis::new(deadline), + fee_policy(identity), + ) + .expect("plan") +} + +fn reserve_one( + book: &ReservationBook, + identity: ProviderIdentity, + item: InventoryItem, + owner: OwnerId, + request_marker: u8, +) -> ReservationView { + let now = UnixMillis::new(100); + book.import_inventory(item, &now).expect("inventory import"); + book.reserve( + &plan( + identity, + owner, + request_marker, + request_marker.wrapping_add(1), + vec![item.outpoint()], + 1_000, + ), + &now, + ) + .expect("reservation") + .reservation() + .clone() +} + +#[test] +fn fee_policy_uses_checked_ceiling_and_both_parties_bounds() { + let identity = identity(10); + let policy = fee_policy(identity); + let exact = transaction_fee(identity, 200); + assert_eq!(policy.required_fee(exact), Ok(200)); + assert_eq!(policy.validate(exact), Ok(())); + + let under = transaction_fee(identity, 199); + assert_eq!( + policy.validate(under), + Err(FeePolicyViolation::FeeBelowMinimum { + required: 200, + actual: 199, + }) + ); + let overweight = + TransactionFee::new(identity.policy_asset(), 10_000, 4_001, 1_001, 1_001).expect("fee"); + assert_eq!( + policy.validate(overweight), + Err(FeePolicyViolation::TransactionOverweight { + maximum: 4_000, + actual: 4_001, + }) + ); + let wrong_asset = TransactionFee::new(asset(9), 10_000, 800, 200, 100).expect("fee"); + assert!(matches!( + policy.validate(wrong_asset), + Err(FeePolicyViolation::WrongPolicyAsset { .. }) + )); +} + +#[test] +fn reservation_is_atomic_idempotent_and_owner_authenticated() { + let directory = TempDir::new().expect("tempdir"); + let identity = identity(11); + let book = open_book(&directory, identity); + let first = inventory(20); + let second = inventory(21); + let now = UnixMillis::new(100); + book.import_inventory(first, &now).expect("first inventory"); + book.import_inventory(second, &now) + .expect("second inventory"); + let request = plan( + identity, + owner(1), + 2, + 3, + vec![second.outpoint(), first.outpoint()], + 1_000, + ); + + let created = book.reserve(&request, &now).expect("reserve"); + assert!(created.created()); + assert_eq!( + created.reservation().outpoints(), + &[first.outpoint(), second.outpoint()] + ); + let retry = book.reserve(&request, &now).expect("idempotent retry"); + assert!(!retry.created()); + assert_eq!(retry.reservation(), created.reservation()); + assert_eq!(book.audit_log().expect("audit").len(), 3); + + let wrong_owner = ReservationAccess::new(created.reservation().id(), owner(9)); + assert!(matches!( + book.cancel(wrong_owner, &now), + Err(ProviderError::ReservationOwnerMismatch(_)) + )); + for item in [first, second] { + assert!(matches!( + book.inventory(item.outpoint()).expect("inventory").unwrap().state(), + InventoryState::Reserved { reservation_id } + if reservation_id == created.reservation().id() + )); + } +} + +#[test] +fn changed_request_cannot_reuse_an_idempotency_key() { + let directory = TempDir::new().expect("tempdir"); + let identity = identity(12); + let book = open_book(&directory, identity); + let item = inventory(22); + let now = UnixMillis::new(100); + book.import_inventory(item, &now).expect("inventory"); + let first = plan(identity, owner(1), 2, 3, vec![item.outpoint()], 1_000); + book.reserve(&first, &now).expect("reserve"); + let changed = plan(identity, owner(1), 2, 4, vec![item.outpoint()], 1_000); + assert!(matches!( + book.reserve(&changed, &now), + Err(ProviderError::IdempotencyConflict { .. }) + )); +} + +#[test] +fn overlapping_multi_input_failure_never_partially_locks_inventory() { + let directory = TempDir::new().expect("tempdir"); + let identity = identity(13); + let book = open_book(&directory, identity); + let first = inventory(23); + let second = inventory(24); + let third = inventory(25); + let now = UnixMillis::new(100); + for item in [first, second, third] { + book.import_inventory(item, &now).expect("inventory"); + } + let winner = book + .reserve( + &plan( + identity, + owner(1), + 1, + 1, + vec![first.outpoint(), second.outpoint()], + 1_000, + ), + &now, + ) + .expect("winner"); + assert!(matches!( + book.reserve( + &plan( + identity, + owner(2), + 2, + 2, + vec![second.outpoint(), third.outpoint()], + 1_000, + ), + &now, + ), + Err(ProviderError::OutpointUnavailable { outpoint, .. }) if outpoint == second.outpoint() + )); + assert_eq!( + book.inventory(third.outpoint()) + .expect("third") + .unwrap() + .state(), + InventoryState::Available + ); + assert!( + book.reservation(derive_reservation_id( + owner(2), + IdempotencyKey::new([2; 32]) + )) + .expect("loser lookup") + .is_none() + ); + assert!(winner.created()); +} + +#[test] +fn deadline_is_exclusive_and_expiry_releases_only_uncommitted_inputs() { + let directory = TempDir::new().expect("tempdir"); + let identity = identity(14); + let book = open_book(&directory, identity); + let item = inventory(26); + let reservation = reserve_one(&book, identity, item, owner(1), 1); + + let deadline = UnixMillis::new(1_000); + assert!(matches!( + book.commit_before_sign( + ReservationAccess::new(reservation.id(), reservation.owner()), + vec![1, 2, 3], + transaction_fee(identity, 200), + &deadline, + ), + Err(ProviderError::ReservationDeadlineElapsed { .. }) + )); + assert!(matches!( + book.reservation(reservation.id()) + .expect("reservation") + .unwrap() + .state(), + ReservationState::Released { + reason: ReleaseReason::Expired, + .. + } + )); + assert_eq!( + book.inventory(item.outpoint()) + .expect("inventory") + .unwrap() + .state(), + InventoryState::Available + ); + + let replacement = book + .reserve( + &plan(identity, owner(2), 2, 2, vec![item.outpoint()], 2_000), + &UnixMillis::new(1_001), + ) + .expect("replacement"); + assert!(replacement.created()); +} + +#[test] +fn expire_due_is_ordered_bounded_inclusive_and_restart_safe() { + let directory = TempDir::new().expect("tempdir"); + let identity = identity(74); + let later_item = inventory(107); + let earliest_item = inventory(108); + let middle_item = inventory(109); + let (later_id, earliest_id, middle_id) = { + let book = open_book(&directory, identity); + let now = UnixMillis::new(100); + for item in [later_item, earliest_item, middle_item] { + book.import_inventory(item, &now).expect("inventory"); + } + + // Insert out of deadline order so this test exercises the expiration + // index rather than reservation insertion order. + let later = book + .reserve( + &plan(identity, owner(1), 1, 1, vec![later_item.outpoint()], 700), + &now, + ) + .expect("later reservation") + .reservation() + .id(); + let earliest = book + .reserve( + &plan( + identity, + owner(2), + 2, + 2, + vec![earliest_item.outpoint()], + 500, + ), + &now, + ) + .expect("earliest reservation") + .reservation() + .id(); + let middle = book + .reserve( + &plan(identity, owner(3), 3, 3, vec![middle_item.outpoint()], 600), + &now, + ) + .expect("middle reservation") + .reservation() + .id(); + + assert!( + book.expire_due(&UnixMillis::new(499), usize::MAX) + .expect("nothing due before the first deadline") + .is_empty() + ); + assert_eq!( + book.expire_due(&UnixMillis::new(700), 2) + .expect("bounded expiration batch"), + vec![earliest, middle] + ); + assert_eq!( + book.inventory(later_item.outpoint()) + .expect("later inventory") + .unwrap() + .state(), + InventoryState::Reserved { + reservation_id: later, + } + ); + for (reservation_id, item) in [(earliest, earliest_item), (middle, middle_item)] { + assert!(matches!( + book.reservation(reservation_id) + .expect("expired reservation") + .unwrap() + .state(), + ReservationState::Released { + reason: ReleaseReason::Expired, + at, + } if at == UnixMillis::new(700) + )); + assert_eq!( + book.inventory(item.outpoint()) + .expect("released inventory") + .unwrap() + .state(), + InventoryState::Available + ); + } + (later, earliest, middle) + }; + + let book = open_book(&directory, identity); + assert_eq!( + book.expire_due(&UnixMillis::new(700), 2) + .expect("inclusive deadline after reopen"), + vec![later_id] + ); + assert!( + book.expire_due(&UnixMillis::new(700), 2) + .expect("expiration retry") + .is_empty() + ); + for (reservation_id, item) in [ + (earliest_id, earliest_item), + (middle_id, middle_item), + (later_id, later_item), + ] { + assert!(matches!( + book.reservation(reservation_id) + .expect("reservation") + .unwrap() + .state(), + ReservationState::Released { + reason: ReleaseReason::Expired, + at, + } if at == UnixMillis::new(700) + )); + assert_eq!( + book.inventory(item.outpoint()) + .expect("inventory") + .unwrap() + .state(), + InventoryState::Available + ); + } + let audit = book.audit_log().expect("audit"); + assert_eq!(audit.len(), 9); + let expired_ids: Vec<_> = audit[6..] + .iter() + .map(|entry| match entry.event() { + AuditEvent::ReservationReleased { + reservation_id, + reason: ReleaseReason::Expired, + } => *reservation_id, + other => panic!("unexpected expiration audit event: {other:?}"), + }) + .collect(); + assert_eq!(expired_ids, vec![earliest_id, middle_id, later_id]); + drop(book); + + let reopened = open_book(&directory, identity); + assert!( + reopened + .expire_due(&UnixMillis::new(700), usize::MAX) + .expect("reopened expiration retry") + .is_empty() + ); + assert_eq!(reopened.audit_log().expect("reopened audit").len(), 9); +} + +#[test] +fn reserve_lazily_reclaims_only_the_expired_reservation_blocking_its_outpoint() { + let directory = TempDir::new().expect("tempdir"); + let identity = identity(75); + let requested_item = inventory(110); + let unrelated_item = inventory(111); + let book = open_book(&directory, identity); + let now = UnixMillis::new(100); + for item in [requested_item, unrelated_item] { + book.import_inventory(item, &now).expect("inventory"); + } + let requested_old = book + .reserve( + &plan( + identity, + owner(1), + 1, + 1, + vec![requested_item.outpoint()], + 500, + ), + &now, + ) + .expect("requested old reservation") + .reservation() + .id(); + let unrelated = book + .reserve( + &plan( + identity, + owner(2), + 2, + 2, + vec![unrelated_item.outpoint()], + 400, + ), + &now, + ) + .expect("unrelated reservation") + .reservation() + .id(); + + let replacement = book + .reserve( + &plan( + identity, + owner(3), + 3, + 3, + vec![requested_item.outpoint()], + 1_000, + ), + &UnixMillis::new(500), + ) + .expect("lazy reclaim replacement"); + assert!(replacement.created()); + assert!(matches!( + book.reservation(requested_old) + .expect("old reservation") + .unwrap() + .state(), + ReservationState::Released { + reason: ReleaseReason::Expired, + at, + } if at == UnixMillis::new(500) + )); + assert!(matches!( + book.inventory(requested_item.outpoint()) + .expect("requested inventory") + .unwrap() + .state(), + InventoryState::Reserved { reservation_id } + if reservation_id == replacement.reservation().id() + )); + assert_eq!( + book.reservation(unrelated) + .expect("unrelated reservation") + .unwrap() + .state(), + ReservationState::Reserved + ); + assert!(matches!( + book.inventory(unrelated_item.outpoint()) + .expect("unrelated inventory") + .unwrap() + .state(), + InventoryState::Reserved { reservation_id } if reservation_id == unrelated + )); +} + +#[test] +fn fee_policy_is_rechecked_before_the_irreversible_transition() { + let directory = TempDir::new().expect("tempdir"); + let identity = identity(15); + let book = open_book(&directory, identity); + let item = inventory(27); + let reservation = reserve_one(&book, identity, item, owner(1), 1); + let access = ReservationAccess::new(reservation.id(), reservation.owner()); + let now = UnixMillis::new(200); + + assert!(matches!( + book.commit_before_sign(access, vec![1, 2, 3], transaction_fee(identity, 199), &now,), + Err(ProviderError::FeePolicy( + FeePolicyViolation::FeeBelowMinimum { .. } + )) + )); + assert_eq!( + book.reservation(reservation.id()) + .expect("reservation") + .unwrap() + .state(), + ReservationState::Reserved + ); + + let committed = book + .commit_before_sign(access, vec![1, 2, 3], transaction_fee(identity, 200), &now) + .expect("commit"); + assert!(committed.newly_committed()); + assert_eq!( + committed + .signing_job() + .expect("new signing job") + .pre_sign_payload(), + &[1, 2, 3] + ); +} + +#[test] +fn committed_outpoints_never_reopen_after_deadline_cancel_or_restart() { + let directory = TempDir::new().expect("tempdir"); + let identity = identity(16); + let item = inventory(28); + let (reservation_id, commitment) = { + let book = open_book(&directory, identity); + let reservation = reserve_one(&book, identity, item, owner(1), 1); + let access = ReservationAccess::new(reservation.id(), reservation.owner()); + let committed = book + .commit_before_sign( + access, + vec![9, 8, 7], + transaction_fee(identity, 200), + &UnixMillis::new(999), + ) + .expect("commit"); + assert!(matches!( + book.cancel(access, &UnixMillis::new(2_000)), + Err(ProviderError::PointOfNoReturn(_)) + )); + assert!( + book.expire_due(&UnixMillis::new(2_000), usize::MAX) + .expect("expire") + .is_empty() + ); + ( + reservation.id(), + committed + .signing_job() + .expect("new signing job") + .commitment(), + ) + }; + + let reopened = open_book(&directory, identity); + assert!(matches!( + reopened + .inventory(item.outpoint()) + .expect("inventory") + .unwrap() + .state(), + InventoryState::Committed { + reservation_id: actual, + commitment: actual_commitment, + } if actual == reservation_id && actual_commitment == commitment + )); + assert!(matches!( + reopened.recovery_actions().expect("recovery").as_slice(), + [RecoveryAction::SignCommittedExact(job)] + if job.reservation_id() == reservation_id + && job.commitment() == commitment + && job.pre_sign_payload() == [9, 8, 7] + )); + assert!(matches!( + reopened.reserve( + &plan(identity, owner(2), 2, 2, vec![item.outpoint()], 3_000,), + &UnixMillis::new(2_001), + ), + Err(ProviderError::OutpointUnavailable { + state: InventoryState::Committed { .. }, + .. + }) + )); +} + +#[test] +fn commitment_and_signed_response_retries_are_exact_and_restart_safe() { + let directory = TempDir::new().expect("tempdir"); + let identity = identity(17); + let item = inventory(29); + let book = open_book(&directory, identity); + let reservation = reserve_one(&book, identity, item, owner(1), 1); + let access = ReservationAccess::new(reservation.id(), reservation.owner()); + let now = UnixMillis::new(200); + let first = book + .commit_before_sign(access, vec![1, 2, 3], transaction_fee(identity, 200), &now) + .expect("commit"); + let retry = book + .commit_before_sign( + access, + vec![1, 2, 3], + transaction_fee(identity, 200), + &UnixMillis::new(201), + ) + .expect("commit retry"); + assert!(!retry.newly_committed()); + assert_eq!(retry.signing_job(), first.signing_job()); + let commitment = first.signing_job().expect("new signing job").commitment(); + assert!(matches!( + book.commit_before_sign( + access, + vec![1, 2, 4], + transaction_fee(identity, 200), + &UnixMillis::new(202), + ), + Err(ProviderError::DifferentSigningIntent(_)) + )); + + let signed = book + .record_signed( + reservation.id(), + commitment, + vec![5, 6, 7], + &UnixMillis::new(203), + ) + .expect("signed"); + assert!(signed.recorded()); + let retry = book + .record_signed( + reservation.id(), + commitment, + vec![5, 6, 7], + &UnixMillis::new(204), + ) + .expect("signed retry"); + assert!(!retry.recorded()); + assert_eq!(retry.artifact(), signed.artifact()); + let completed_retry = book + .commit_before_sign( + access, + vec![1, 2, 3], + transaction_fee(identity, 200), + &UnixMillis::new(205), + ) + .expect("completed commit retry"); + assert_eq!(completed_retry.signed_artifact(), Some(signed.artifact())); + assert!(matches!( + book.record_signed( + reservation.id(), + commitment, + vec![5, 6, 8], + &UnixMillis::new(206), + ), + Err(ProviderError::DifferentSignedArtifact(_)) + )); + drop(book); + + let reopened = open_book(&directory, identity); + assert!(matches!( + reopened.recovery_actions().expect("recovery").as_slice(), + [RecoveryAction::ReplaySignedExact(artifact)] + if artifact == signed.artifact() + )); +} + +#[test] +fn signed_allocation_stays_retired_and_recoverable_after_deadline_and_reopen() { + let directory = TempDir::new().expect("tempdir"); + let identity = identity(76); + let item = inventory(112); + let (reservation_id, access, commitment, artifact) = { + let book = open_book(&directory, identity); + let reservation = reserve_one(&book, identity, item, owner(1), 1); + let access = ReservationAccess::new(reservation.id(), reservation.owner()); + let committed = book + .commit_before_sign( + access, + vec![1, 2, 3], + transaction_fee(identity, 200), + &UnixMillis::new(999), + ) + .expect("commit before deadline"); + let commitment = committed.signing_job().expect("signing job").commitment(); + let signed = book + .record_signed( + reservation.id(), + commitment, + vec![4, 5, 6], + &UnixMillis::new(2_000), + ) + .expect("signing may finish after durable acceptance deadline"); + assert!(signed.recorded()); + assert!( + book.expire_due(&UnixMillis::new(3_000), usize::MAX) + .expect("committed reservation is not expirable") + .is_empty() + ); + assert!(matches!( + book.inventory(item.outpoint()) + .expect("inventory") + .unwrap() + .state(), + InventoryState::Committed { + reservation_id, + commitment: actual, + } if reservation_id == reservation.id() && actual == commitment + )); + ( + reservation.id(), + access, + commitment, + signed.artifact().clone(), + ) + }; + + let reopened = open_book(&directory, identity); + assert!(matches!( + reopened + .inventory(item.outpoint()) + .expect("inventory") + .unwrap() + .state(), + InventoryState::Committed { + reservation_id: actual_id, + commitment: actual_commitment, + } if actual_id == reservation_id && actual_commitment == commitment + )); + assert!(matches!( + reopened.recovery_actions().expect("recovery").as_slice(), + [RecoveryAction::ReplaySignedExact(recovered)] if recovered == &artifact + )); + let completed_retry = reopened + .commit_before_sign( + access, + vec![1, 2, 3], + transaction_fee(identity, 200), + &UnixMillis::new(3_001), + ) + .expect("exact post-deadline commitment retry"); + assert_eq!(completed_retry.signed_artifact(), Some(&artifact)); + assert!(matches!( + reopened.reserve( + &plan(identity, owner(2), 2, 2, vec![item.outpoint()], 4_000), + &UnixMillis::new(3_002), + ), + Err(ProviderError::OutpointUnavailable { + state: InventoryState::Committed { .. }, + .. + }) + )); +} + +#[test] +fn persisted_clock_high_watermark_fails_closed_on_rollback() { + let directory = TempDir::new().expect("tempdir"); + let identity = identity(18); + let item = inventory(30); + { + let book = open_book(&directory, identity); + book.import_inventory(item, &UnixMillis::new(500)) + .expect("inventory"); + assert_eq!( + book.last_observed_time().expect("time"), + Some(UnixMillis::new(500)) + ); + } + let reopened = open_book(&directory, identity); + assert!(matches!( + reopened.reserve( + &plan( + identity, + owner(1), + 1, + 1, + vec![item.outpoint()], + 1_000, + ), + &UnixMillis::new(499), + ), + Err(ProviderError::ClockRegression { + previous, + now, + }) if previous == UnixMillis::new(500) && now == UnixMillis::new(499) + )); + assert_eq!( + reopened + .inventory(item.outpoint()) + .expect("inventory") + .unwrap() + .state(), + InventoryState::Available + ); +} + +#[test] +fn failed_wrong_owner_operation_durably_advances_the_clock_high_watermark() { + let directory = TempDir::new().expect("tempdir"); + let identity = identity(77); + let item = inventory(113); + let book = open_book(&directory, identity); + let reservation = reserve_one(&book, identity, item, owner(1), 1); + let wrong_access = ReservationAccess::new(reservation.id(), owner(2)); + assert!(matches!( + book.commit_before_sign( + wrong_access, + vec![1, 2, 3], + transaction_fee(identity, 200), + &UnixMillis::new(1_100), + ), + Err(ProviderError::ReservationOwnerMismatch(actual)) if actual == reservation.id() + )); + assert_eq!( + book.last_observed_time().expect("time"), + Some(UnixMillis::new(1_100)) + ); + assert!(matches!( + book.commit_before_sign( + ReservationAccess::new(reservation.id(), reservation.owner()), + vec![1, 2, 3], + transaction_fee(identity, 200), + &UnixMillis::new(999), + ), + Err(ProviderError::ClockRegression { previous, now }) + if previous == UnixMillis::new(1_100) && now == UnixMillis::new(999) + )); + assert_eq!( + book.reservation(reservation.id()) + .expect("reservation") + .unwrap() + .state(), + ReservationState::Reserved + ); + drop(book); + + let reopened = open_book(&directory, identity); + assert_eq!( + reopened.last_observed_time().expect("reopened time"), + Some(UnixMillis::new(1_100)) + ); + assert!(matches!( + reopened + .inventory(item.outpoint()) + .expect("inventory") + .unwrap() + .state(), + InventoryState::Reserved { reservation_id } if reservation_id == reservation.id() + )); +} + +#[test] +fn concurrent_reservations_have_one_durable_winner() { + let directory = TempDir::new().expect("tempdir"); + let identity = identity(19); + let item = inventory(31); + let book = Arc::new(open_book(&directory, identity)); + book.import_inventory(item, &UnixMillis::new(100)) + .expect("inventory"); + let barrier = Arc::new(Barrier::new(8)); + let mut handles = Vec::new(); + for marker in 1..=8_u8 { + let book = Arc::clone(&book); + let barrier = Arc::clone(&barrier); + handles.push(thread::spawn(move || { + let request = plan( + identity, + owner(marker), + marker, + marker, + vec![item.outpoint()], + 1_000, + ); + barrier.wait(); + book.reserve(&request, &UnixMillis::new(200)) + })); + } + let results: Vec<_> = handles + .into_iter() + .map(|handle| handle.join().expect("thread")) + .collect(); + assert_eq!(results.iter().filter(|result| result.is_ok()).count(), 1); + assert_eq!( + results + .iter() + .filter(|result| matches!(result, Err(ProviderError::OutpointUnavailable { .. }))) + .count(), + 7 + ); + let winning_id = results + .iter() + .find_map(|result| result.as_ref().ok()) + .expect("winner") + .reservation() + .id(); + drop(results); + drop(book); + + let reopened = open_book(&directory, identity); + assert!(matches!( + reopened + .inventory(item.outpoint()) + .expect("inventory") + .unwrap() + .state(), + InventoryState::Reserved { reservation_id } if reservation_id == winning_id + )); +} + +#[test] +fn concurrent_overlapping_multi_input_reservations_remain_atomic_after_reopen() { + let directory = TempDir::new().expect("tempdir"); + let identity = identity(78); + let first = inventory(114); + let shared = inventory(115); + let third = inventory(116); + let book = Arc::new(open_book(&directory, identity)); + for item in [first, shared, third] { + book.import_inventory(item, &UnixMillis::new(100)) + .expect("inventory"); + } + let left_owner = owner(1); + let right_owner = owner(2); + let left_key = IdempotencyKey::new([1; 32]); + let right_key = IdempotencyKey::new([2; 32]); + let left_id = derive_reservation_id(left_owner, left_key); + let right_id = derive_reservation_id(right_owner, right_key); + let barrier = Arc::new(Barrier::new(2)); + + let left_book = Arc::clone(&book); + let left_barrier = Arc::clone(&barrier); + let left = thread::spawn(move || { + let request = plan( + identity, + left_owner, + 1, + 1, + vec![first.outpoint(), shared.outpoint()], + 1_000, + ); + left_barrier.wait(); + left_book.reserve(&request, &UnixMillis::new(200)) + }); + let right_book = Arc::clone(&book); + let right_barrier = Arc::clone(&barrier); + let right = thread::spawn(move || { + let request = plan( + identity, + right_owner, + 2, + 2, + vec![shared.outpoint(), third.outpoint()], + 1_000, + ); + right_barrier.wait(); + right_book.reserve(&request, &UnixMillis::new(200)) + }); + let left = left.join().expect("left thread"); + let right = right.join().expect("right thread"); + assert_ne!(left.is_ok(), right.is_ok()); + assert_eq!( + [&left, &right] + .into_iter() + .filter(|result| { + matches!( + result, + Err(ProviderError::OutpointUnavailable { outpoint, .. }) + if *outpoint == shared.outpoint() + ) + }) + .count(), + 1 + ); + + let (winning_id, winning_items, losing_id, losing_only_item) = if left.is_ok() { + (left_id, [first, shared], right_id, third) + } else { + (right_id, [shared, third], left_id, first) + }; + for item in winning_items { + assert!(matches!( + book.inventory(item.outpoint()) + .expect("winning inventory") + .unwrap() + .state(), + InventoryState::Reserved { reservation_id } if reservation_id == winning_id + )); + } + assert_eq!( + book.inventory(losing_only_item.outpoint()) + .expect("losing-only inventory") + .unwrap() + .state(), + InventoryState::Available + ); + assert!( + book.reservation(losing_id) + .expect("losing reservation") + .is_none() + ); + assert_eq!(book.audit_log().expect("audit").len(), 4); + drop(left); + drop(right); + drop(book); + + let reopened = open_book(&directory, identity); + assert_eq!( + reopened + .reservation(winning_id) + .expect("winning reservation") + .unwrap() + .outpoints(), + winning_items.map(InventoryItem::outpoint) + ); + assert!( + reopened + .reservation(losing_id) + .expect("losing reservation") + .is_none() + ); + for item in winning_items { + assert!(matches!( + reopened + .inventory(item.outpoint()) + .expect("reopened winning inventory") + .unwrap() + .state(), + InventoryState::Reserved { reservation_id } if reservation_id == winning_id + )); + } + assert_eq!( + reopened + .inventory(losing_only_item.outpoint()) + .expect("reopened losing-only inventory") + .unwrap() + .state(), + InventoryState::Available + ); + assert_eq!(reopened.audit_log().expect("reopened audit").len(), 4); +} + +#[test] +fn concurrent_cancel_and_commit_linearize_to_one_legal_state() { + for iteration in 0..16_u8 { + let directory = TempDir::new().expect("tempdir"); + let identity = identity(40_u8.wrapping_add(iteration)); + let item = inventory(80_u8.wrapping_add(iteration)); + let book = Arc::new(open_book(&directory, identity)); + let reservation = reserve_one(&book, identity, item, owner(1), 1); + let access = ReservationAccess::new(reservation.id(), reservation.owner()); + let barrier = Arc::new(Barrier::new(2)); + + let cancel_book = Arc::clone(&book); + let cancel_barrier = Arc::clone(&barrier); + let cancel = thread::spawn(move || { + cancel_barrier.wait(); + cancel_book.cancel(access, &UnixMillis::new(200)) + }); + let commit_book = Arc::clone(&book); + let commit_barrier = Arc::clone(&barrier); + let commit = thread::spawn(move || { + commit_barrier.wait(); + commit_book.commit_before_sign( + access, + vec![1, 2, 3], + transaction_fee(identity, 200), + &UnixMillis::new(200), + ) + }); + let cancel = cancel.join().expect("cancel thread"); + let commit = commit.join().expect("commit thread"); + assert_ne!(cancel.is_ok(), commit.is_ok()); + let state = book + .reservation(reservation.id()) + .expect("reservation") + .unwrap() + .state(); + match state { + ReservationState::Released { + reason: ReleaseReason::ClientCancelled, + .. + } => assert_eq!( + book.inventory(item.outpoint()) + .expect("inventory") + .unwrap() + .state(), + InventoryState::Available + ), + ReservationState::Committed { commitment, .. } => assert!(matches!( + book.inventory(item.outpoint()) + .expect("inventory") + .unwrap() + .state(), + InventoryState::Committed { + reservation_id, + commitment: actual, + } if reservation_id == reservation.id() && actual == commitment + )), + other => panic!("illegal race result: {other:?}"), + } + } +} + +#[test] +fn database_is_bound_to_one_provider_and_chain_identity() { + let directory = TempDir::new().expect("tempdir"); + let first = identity(60); + let book = open_book(&directory, first); + assert_eq!(book.identity(), first); + assert_eq!(book.schema_version().expect("schema"), SCHEMA_VERSION); + drop(book); + + let error = match ReservationBook::open(directory.path().join("provider.redb"), identity(61)) { + Ok(_) => panic!("identity mismatch must fail"), + Err(error) => error, + }; + assert!(matches!( + error, + ProviderError::ProviderIdentityMismatch { + expected, + actual, + } if *expected == first && *actual == identity(61) + )); +} + +#[test] +fn audit_log_is_ordered_and_records_the_safety_boundaries() { + let directory = TempDir::new().expect("tempdir"); + let identity = identity(62); + let book = open_book(&directory, identity); + let item = inventory(90); + let reservation = reserve_one(&book, identity, item, owner(1), 1); + let committed = book + .commit_before_sign( + ReservationAccess::new(reservation.id(), reservation.owner()), + vec![1], + transaction_fee(identity, 200), + &UnixMillis::new(200), + ) + .expect("commit"); + book.record_signed( + reservation.id(), + committed + .signing_job() + .expect("new signing job") + .commitment(), + vec![2], + &UnixMillis::new(201), + ) + .expect("signed"); + let audit = book.audit_log().expect("audit"); + assert_eq!( + audit.iter().map(AuditEntry::sequence).collect::>(), + vec![1, 2, 3, 4] + ); + assert!(matches!( + audit[0].event(), + AuditEvent::InventoryImported { .. } + )); + assert!(matches!( + audit[1].event(), + AuditEvent::ReservationCreated { .. } + )); + assert!(matches!( + audit[2].event(), + AuditEvent::SigningCommitted { .. } + )); + assert!(matches!( + audit[3].event(), + AuditEvent::SignedArtifactStored { .. } + )); +} + +#[test] +fn startup_integrity_rejects_a_missing_committed_allocation() { + let directory = TempDir::new().expect("tempdir"); + let path = directory.path().join("provider.redb"); + let identity = identity(79); + let item = inventory(117); + { + let book = ReservationBook::open(&path, identity).expect("book"); + let reservation = reserve_one(&book, identity, item, owner(1), 1); + book.commit_before_sign( + ReservationAccess::new(reservation.id(), reservation.owner()), + vec![1, 2, 3], + transaction_fee(identity, 200), + &UnixMillis::new(200), + ) + .expect("commit"); + } + + { + let database = Database::create(&path).expect("raw database"); + let mut write = database.begin_write().expect("raw write"); + write + .set_durability(Durability::Immediate) + .expect("durability"); + let removed = { + let mut allocations = write.open_table(ALLOCATIONS).expect("allocations"); + allocations + .remove(outpoint_key(item.outpoint()).as_slice()) + .expect("remove") + .is_some() + }; + assert!(removed); + write.commit().expect("commit corruption fixture"); + } + + let error = match ReservationBook::open(&path, identity) { + Ok(_) => panic!("missing committed allocation must fail closed"), + Err(error) => error, + }; + assert!(matches!(error, ProviderError::CorruptState(message) + if message.contains("permanent allocation"))); +} + +#[test] +fn missing_schema_metadata_cannot_reinitialize_a_nonempty_database() { + let directory = TempDir::new().expect("tempdir"); + let path = directory.path().join("provider.redb"); + let identity = identity(80); + { + let book = ReservationBook::open(&path, identity).expect("book"); + book.import_inventory(inventory(118), &UnixMillis::new(100)) + .expect("inventory"); + } + + { + let database = Database::create(&path).expect("raw database"); + let mut write = database.begin_write().expect("raw write"); + write + .set_durability(Durability::Immediate) + .expect("durability"); + let removed = { + let mut meta = write.open_table(META).expect("meta"); + meta.remove(SCHEMA_VERSION_KEY).expect("remove").is_some() + }; + assert!(removed); + write.commit().expect("commit corruption fixture"); + } + + let error = match ReservationBook::open(&path, identity) { + Ok(_) => panic!("nonempty database without schema metadata must fail closed"), + Err(error) => error, + }; + assert!(matches!(error, ProviderError::CorruptState(message) + if message.contains("schema version is missing"))); +} + +#[test] +fn strict_record_codec_rejects_wrong_versions_and_trailing_bytes() { + let encoded = encode_record(&StoredRequestBinding { + reservation_id: [1; 32], + request_digest: [2; 32], + }) + .expect("encode"); + let mut wrong_version = encoded.clone(); + wrong_version[0] = RECORD_VERSION.wrapping_add(1); + assert!(matches!( + decode_record::(&wrong_version), + Err(ProviderError::RecordVersionMismatch { .. }) + )); + let mut trailing = encoded; + trailing.push(0); + assert!(matches!( + decode_record::(&trailing), + Err(ProviderError::TrailingRecordBytes(1)) + )); +} + +#[test] +fn reservation_failpoints_rollback_every_logical_table() { + let failpoints = [ + (mutation_failpoints::RESERVE_AFTER_RECORD, 0), + (mutation_failpoints::RESERVE_AFTER_REQUEST_KEY, 0), + (mutation_failpoints::RESERVE_AFTER_ALLOCATION, 0), + (mutation_failpoints::RESERVE_AFTER_ALLOCATION, 1), + (mutation_failpoints::RESERVE_AFTER_EXPIRATION, 0), + (mutation_failpoints::RESERVE_AFTER_AUDIT, 0), + ]; + for (name, occurrence) in failpoints { + let directory = TempDir::new().expect("tempdir"); + let identity = identity(70); + let first = inventory(100); + let second = inventory(101); + let request = plan( + identity, + owner(1), + 1, + 1, + vec![first.outpoint(), second.outpoint()], + 1_000, + ); + let reservation_id = derive_reservation_id(owner(1), IdempotencyKey::new([1; 32])); + { + let book = open_book(&directory, identity); + for item in [first, second] { + book.import_inventory(item, &UnixMillis::new(100)) + .expect("inventory"); + } + let guard = mutation_failpoints::arm(name, occurrence); + assert!(matches!( + book.reserve(&request, &UnixMillis::new(200)), + Err(ProviderError::InjectedMutationFailure(actual)) if actual == name + )); + drop(guard); + } + let reopened = open_book(&directory, identity); + assert!( + reopened + .reservation(reservation_id) + .expect("reservation") + .is_none() + ); + assert_eq!(reopened.audit_log().expect("audit").len(), 2); + assert_eq!( + reopened.last_observed_time().expect("time"), + Some(UnixMillis::new(200)) + ); + for item in [first, second] { + assert_eq!( + reopened + .inventory(item.outpoint()) + .expect("inventory") + .unwrap() + .state(), + InventoryState::Available + ); + } + assert!( + reopened + .reserve(&request, &UnixMillis::new(200)) + .expect("retry") + .created() + ); + } +} + +#[test] +fn release_failpoints_never_partially_unlock_a_reservation() { + let failpoints = [ + (mutation_failpoints::RELEASE_AFTER_ALLOCATION, 0), + (mutation_failpoints::RELEASE_AFTER_ALLOCATION, 1), + (mutation_failpoints::RELEASE_AFTER_EXPIRATION, 0), + (mutation_failpoints::RELEASE_AFTER_RECORD, 0), + (mutation_failpoints::RELEASE_AFTER_AUDIT, 0), + ]; + for (name, occurrence) in failpoints { + let directory = TempDir::new().expect("tempdir"); + let identity = identity(71); + let first = inventory(102); + let second = inventory(103); + let access = { + let book = open_book(&directory, identity); + let now = UnixMillis::new(100); + for item in [first, second] { + book.import_inventory(item, &now).expect("inventory"); + } + let reservation = book + .reserve( + &plan( + identity, + owner(1), + 1, + 1, + vec![first.outpoint(), second.outpoint()], + 1_000, + ), + &now, + ) + .expect("reserve") + .reservation() + .clone(); + let access = ReservationAccess::new(reservation.id(), reservation.owner()); + let guard = mutation_failpoints::arm(name, occurrence); + assert!(matches!( + book.cancel(access, &UnixMillis::new(200)), + Err(ProviderError::InjectedMutationFailure(actual)) if actual == name + )); + drop(guard); + access + }; + let reopened = open_book(&directory, identity); + assert_eq!( + reopened + .reservation(access.reservation_id()) + .expect("reservation") + .unwrap() + .state(), + ReservationState::Reserved + ); + assert_eq!(reopened.audit_log().expect("audit").len(), 3); + for item in [first, second] { + assert!(matches!( + reopened + .inventory(item.outpoint()) + .expect("inventory") + .unwrap() + .state(), + InventoryState::Reserved { reservation_id } + if reservation_id == access.reservation_id() + )); + } + assert!( + reopened + .cancel(access, &UnixMillis::new(200)) + .expect("retry") + ); + } +} + +#[test] +fn signing_commitment_failpoints_never_cross_the_point_of_no_return() { + let failpoints = [ + (mutation_failpoints::COMMIT_AFTER_ALLOCATION, 0), + (mutation_failpoints::COMMIT_AFTER_ALLOCATION, 1), + (mutation_failpoints::COMMIT_AFTER_EXPIRATION, 0), + (mutation_failpoints::COMMIT_AFTER_RECORD, 0), + (mutation_failpoints::COMMIT_AFTER_AUDIT, 0), + ]; + for (name, occurrence) in failpoints { + let directory = TempDir::new().expect("tempdir"); + let identity = identity(72); + let first = inventory(104); + let second = inventory(105); + let access = { + let book = open_book(&directory, identity); + let now = UnixMillis::new(100); + for item in [first, second] { + book.import_inventory(item, &now).expect("inventory"); + } + let reservation = book + .reserve( + &plan( + identity, + owner(1), + 1, + 1, + vec![first.outpoint(), second.outpoint()], + 1_000, + ), + &now, + ) + .expect("reserve") + .reservation() + .clone(); + let access = ReservationAccess::new(reservation.id(), reservation.owner()); + let guard = mutation_failpoints::arm(name, occurrence); + assert!(matches!( + book.commit_before_sign( + access, + vec![1, 2, 3], + transaction_fee(identity, 200), + &UnixMillis::new(200), + ), + Err(ProviderError::InjectedMutationFailure(actual)) if actual == name + )); + drop(guard); + access + }; + let reopened = open_book(&directory, identity); + assert_eq!( + reopened + .reservation(access.reservation_id()) + .expect("reservation") + .unwrap() + .state(), + ReservationState::Reserved + ); + assert!(reopened.recovery_actions().expect("recovery").is_empty()); + assert_eq!(reopened.audit_log().expect("audit").len(), 3); + for item in [first, second] { + assert!(matches!( + reopened + .inventory(item.outpoint()) + .expect("inventory") + .unwrap() + .state(), + InventoryState::Reserved { reservation_id } + if reservation_id == access.reservation_id() + )); + } + assert!( + reopened + .commit_before_sign( + access, + vec![1, 2, 3], + transaction_fee(identity, 200), + &UnixMillis::new(200), + ) + .expect("retry") + .newly_committed() + ); + } +} + +#[test] +fn signed_artifact_failpoints_leave_an_exact_recoverable_signing_job() { + let failpoints = [ + (mutation_failpoints::SIGNED_AFTER_RECORD, 0), + (mutation_failpoints::SIGNED_AFTER_AUDIT, 0), + ]; + for (name, occurrence) in failpoints { + let directory = TempDir::new().expect("tempdir"); + let identity = identity(73); + let item = inventory(106); + let (reservation_id, commitment) = { + let book = open_book(&directory, identity); + let reservation = reserve_one(&book, identity, item, owner(1), 1); + let committed = book + .commit_before_sign( + ReservationAccess::new(reservation.id(), reservation.owner()), + vec![1, 2, 3], + transaction_fee(identity, 200), + &UnixMillis::new(200), + ) + .expect("commit"); + let commitment = committed.signing_job().expect("signing job").commitment(); + let guard = mutation_failpoints::arm(name, occurrence); + assert!(matches!( + book.record_signed( + reservation.id(), + commitment, + vec![4, 5, 6], + &UnixMillis::new(300), + ), + Err(ProviderError::InjectedMutationFailure(actual)) if actual == name + )); + drop(guard); + (reservation.id(), commitment) + }; + let reopened = open_book(&directory, identity); + assert!(matches!( + reopened.recovery_actions().expect("recovery").as_slice(), + [RecoveryAction::SignCommittedExact(job)] + if job.reservation_id() == reservation_id + && job.commitment() == commitment + && job.pre_sign_payload() == [1, 2, 3] + )); + assert_eq!(reopened.audit_log().expect("audit").len(), 3); + assert!( + reopened + .record_signed( + reservation_id, + commitment, + vec![4, 5, 6], + &UnixMillis::new(300), + ) + .expect("retry") + .recorded() + ); + } +} diff --git a/docs/adr/0006-rfq-first-liquidity-scope.md b/docs/adr/0006-rfq-first-liquidity-scope.md index 8593457..a09415a 100644 --- a/docs/adr/0006-rfq-first-liquidity-scope.md +++ b/docs/adr/0006-rfq-first-liquidity-scope.md @@ -5,7 +5,7 @@ - Supersedes: ADR 0002's release-scope decision - Retires as historical: ADR 0003 - Amends: ADR 0001's node-side advisory-routing responsibility -- Implementation status updated: 2026-08-10 +- Implementation status updated: 2026-08-11 ## Context @@ -120,9 +120,10 @@ links apply only to that revision. changing version constants. 3. **Completed in PR #25:** prove a two-wallet confidential RFQ settlement on liquidregtest before freezing a remote RFQ protocol. -4. **Implemented as a provisional client-local API:** add exact-in/exact-out +4. **Completed in PR #26 as a provisional client-local API:** add exact-in/exact-out aggregate intent, exact per-leg allocation, authenticated proposal binding, route-owned transaction composition, and no remote wire format. -5. Build the RFQ provider as a separate inventory-bearing service. +5. **Provider core implemented under ADR 0007:** complete the separate wallet, + quoting, transaction-validation, signer, relay, and remote-service layers. 6. Add production-shaped process, crash-recovery, mutation, reorg, and operational acceptance gates. diff --git a/docs/adr/0007-rfq-provider-state-machine.md b/docs/adr/0007-rfq-provider-state-machine.md new file mode 100644 index 0000000..8defebf --- /dev/null +++ b/docs/adr/0007-rfq-provider-state-machine.md @@ -0,0 +1,189 @@ +# ADR 0007: RFQ provider reservation and signing state machine + +- Status: Accepted +- Date: 2026-08-11 +- Extends: [ADR 0006](0006-rfq-first-liquidity-scope.md) + +## Context + +ADR 0006 selects a separate, noncustodial RFQ provider as the first liquidity +venue. The client constructs and authorizes the complete transaction; the +provider reserves only its own inventory, validates the final transaction, and +signs only its own inputs. + +Provider inventory is nevertheless a shared resource. A firm quote temporarily +removes exact outpoints from circulation, and a provider signature has no +service-level expiry while those outpoints remain spendable. Crash ambiguity, +response loss, a low-fee transaction, or a reorganization must never cause an +outpoint that may have a valid signature to be quoted again. + +The provider also needs an exact definition of quote expiry. Requiring a signer +or network response to finish before a wall-clock deadline cannot be made +atomic with durable storage. It would leave crash windows in which the service +could not know whether a valid signature exists. + +## Decision + +### Separate durable authority + +The RFQ provider owns a database and provider identity separate from +`deadcat-node`, the client, and every other provider. The database is bound to +one provider identity, Liquid genesis hash, and policy asset. It contains no +customer wallet secrets and gives no authority over customer funds. + +The initial provider core is transport-free. Its persistence types are private +versioned records, not wire DTOs. It does not extend the node RPC, reuse the +`deadcat/1` ALPN, or make a network compatibility promise. + +### Monotonic inventory states + +Each provider outpoint has one authoritative allocation: + +```text +Available + -> Reserved(reservation) + -> Available only by unused cancellation or expiry + -> CommittedToExactPayload + -> SignedBytesStored + -> relay and chain reconciliation +``` + +There is no transition from `CommittedToExactPayload` or any later state back +to `Available`. A confirmed settlement may create a new provider change output, +but that output has a new outpoint and enters inventory independently. + +Reservations, request-key bindings, input allocations, expiration indexes, and +audit entries change in one serializable redb write transaction. Terminal +reservation records and committed allocation tombstones remain durable for +retry and recovery. + +### Deadline and point of no return + +The quote deadline is an exclusive **durable accept-before deadline**: + +- a reservation is live only when `now < accept_before`; +- at `now >= accept_before`, an uncommitted reservation expires; and +- a commitment that durably succeeds before the deadline remains valid even + when signing, response delivery, relay, or restart recovery happens later. + +The exact point of no return is the durable `Reserved -> Committed` transition, +not quote creation and not signature delivery. The provider follows this +ordering: + +1. receive a complete blinded transaction with all required taker signatures; +2. validate its body, proofs, prevouts, economics, fee, and sighash policy; +3. atomically retire every reserved provider outpoint and persist the exact + pre-sign transcript plus a domain-separated commitment; +4. invoke the wallet or HSM signer using only those persisted bytes; +5. persist the exact signed response; and only then +6. return or relay those same signed bytes. + +A crash before step 3 leaves an ordinary reservation that may expire. A crash +after step 3 resumes only the persisted transcript. A crash after step 5 +replays only the persisted signed response. Signer failure, timeout, mempool +absence, fee-market movement, or reorganization never reopens committed +outpoints. + +This policy deliberately sacrifices provider inventory availability rather +than risk authorizing two transactions with the same outpoint. + +### Authentication and retry + +A public reservation ID is not authorization. Cancellation and commitment are +bound to an authenticated owner principal. Each owner supplies a high-entropy +idempotency key: + +- an exact retry returns the existing reservation or completed result; +- the same key with different terms is rejected; +- a terminal reservation is never resurrected; and +- a new quote requires a new key. + +The immutable reservation commits to the quote, exact outpoints, deadline, and +fee policy. The signing commitment additionally covers the exact pre-sign +payload and observed transaction fee facts. A transaction ID alone is +insufficient because Liquid proofs, witnesses, and PSET disclosures are not all +identified by the transaction ID. + +### Time safety + +The provider samples its clock once after acquiring the serial database writer. +Absolute Unix time is persisted because monotonic process time cannot survive a +restart. The database retains a last-observed time high-water mark; a backward +clock jump fails closed rather than extending a quote. Advancing that mark is a +separate immediate-durability commit performed while a process-wide operation +lock remains held. Consequently, a later time observation survives even when +authentication, policy validation, or the following logical mutation fails. +redb's exclusive database-open lock prevents another process from bypassing +that serialization. + +### Fee and resource admission + +Every firm reservation freezes: + +- a minimum effective fee rate in integer satoshis per 1,000 policy virtual + bytes; +- an optional minimum absolute fee; +- the regular or confidential-discounted size metric used by the provider's + broadcasting Elements node; and +- a maximum transaction weight. + +Before commitment, the provider recomputes policy size from the complete +blinded transaction, including the projected provider witness, and requires: + +```text +fee >= max(minimum_absolute_fee, + ceil(minimum_sats_per_kvb * policy_vsize / 1000)) +``` + +The calculation uses checked integer arithmetic. The client independently +retains its maximum absolute network-fee authorization. Thus the client caps +overpayment while the provider rejects transactions likely to strand shared +inventory. CPFP is a provider-operated recovery mechanism, not a substitute +for initial fee admission and not a cost silently imposed on later traders. + +The state-only crate in this change models and persists those validator-derived +facts, but deliberately does not expose its commit or signed-artifact recording +transitions as externally callable service APIs. They remain crate-internal +until the concrete PSET validator and signer adapter can construct their inputs; +detached caller assertions are not an admissible production trust boundary. + +## Consequences + +- The provider may strand inventory after an ambiguous signing failure, but it + cannot silently double-allocate that inventory. +- A client timeout after submitting its signature means status unknown, not + automatic cancellation. The later protocol must expose idempotent status and + replay. +- Immediate provider relay and optional provider-funded CPFP reduce the time + committed inventory remains unavailable; cooperative RBF is deferred. +- The state core stores no private keys and implements no pricing, inventory + discovery, transaction validation, signing, networking, relay, mempool, or + reorg policy. Those layers consume its transition-specific API. +- Multiple interactive RFQ signers remain deferred. Future AMM and DLOB legs + may coexist because a reservation covers only the provider's exact leg and + inputs, not the entire route. +- Reservation requests lazily expire only reservations blocking their requested + outpoints. A service worker drains unrelated expirations through explicitly + bounded batches (capped by the state core), so an accumulated expiry backlog + cannot make one request's write transaction unbounded. + +## Implementation and follow-up + +The first implementation is the `deadcat-rfq-provider` library. It provides +provider/chain database binding, durable inventory import, atomic multi-input +reservation, owner-scoped idempotency, bounded expiry and cancellation, +fee-policy evaluation over future validator-derived facts, commit-before-sign +recovery state, signed-response persistence state, clock rollback protection, +startup integrity validation, and an audit log. The safety-critical commit and +signed-artifact transitions remain crate-internal until their validator and +signer producers land. + +The next provider milestones are: + +1. choose the wallet/signer and inventory-discovery boundary; +2. add configurable inventory-aware quote construction; +3. validate a concrete final Liquid PSET and derive its exact fee metrics; +4. define a dedicated RFQ protocol, identity, and ALPN; +5. persist relay and chain-reconciliation observations without ever reopening + a committed outpoint; and +6. pass process-kill, signer ambiguity, mempool, confirmation, and reorg gates. diff --git a/docs/adr/README.md b/docs/adr/README.md index fc620a9..52bf251 100644 --- a/docs/adr/README.md +++ b/docs/adr/README.md @@ -11,3 +11,4 @@ change after covenant CMRs or public wire formats exist. | [0004](0004-chain-state-and-reorgs.md) | Chain transactions apply atomically; confirmed-tip state rolls back two blocks | | [0005](0005-rt-blinding-schedule.md) | **Proposed:** complementary A/B RT engineering evidence and protocol-owner approval are complete; focused external review remains | | [0006](0006-rfq-first-liquidity-scope.md) | **Accepted:** production is market-only with separate noncustodial RFQ liquidity and client-owned routing | +| [0007](0007-rfq-provider-state-machine.md) | **Accepted:** RFQ inputs commit durably before signing and never reopen after ambiguous authorization | diff --git a/docs/liquidity-roadmap.md b/docs/liquidity-roadmap.md index 5cf2bd9..5cca620 100644 --- a/docs/liquidity-roadmap.md +++ b/docs/liquidity-roadmap.md @@ -164,7 +164,8 @@ The RFQ service is a separate inventory-bearing security principal. It: - reserves exact inputs for a short-lived quote; - contributes its exact inputs and outputs to a final PSET; - validates the complete transaction; -- signs only while its reservation remains acceptable; and +- durably accepts a signing intent only while its reservation remains live; +- signs only the exact durably accepted transaction; and - returns its signature for the exact finalized transaction and may relay that same transaction immediately. @@ -488,16 +489,21 @@ An RFQ deadline is enforced by service behavior: 2. assemble and blind the final transaction; 3. have the user authenticate the final transaction body and proof set under an explicitly approved sighash and proof-authentication profile; -4. have the provider sign only while the reservation is live; and -5. return the provider signature for that exact finalized transaction and +4. while the reservation is live, have the provider durably commit its inputs + to the exact validated pre-sign transcript; +5. have the provider sign only that persisted transcript, then durably store + the signed response; and +6. return the provider signature for that exact finalized transaction and immediately relay it according to the quote policy. The provider commitment creates accountability and operational firmness, not a consensus guarantee that the provider cannot fail. Once created, a transaction signature has no service-level expiry while its inputs remain spendable. The -provider must therefore refuse to sign after the deadline, mark the reserved -inputs committed once it releases a signature, and never make them available -again merely because a local timer elapsed. The client retains final +provider must therefore refuse new durable acceptance after the deadline, mark +the reserved inputs committed before it invokes the signer, and never make +them available again merely because a local timer elapsed. Signing and relay +may finish after the deadline when durable acceptance won beforehand. The +client retains final verification and may relay the exact same transaction through any broadcaster. Ambiguous broadcast or deliberate conflict handling requires a documented state machine that checks the exact transaction and input outspends. An absolute @@ -845,10 +851,11 @@ The client must distinguish: - broadcast ambiguity; and - confirmation followed by reorganization. -Before any counterparty signature is released, failure can discard the -transaction, release or expire RFQ reservations, refresh state, and reroute. -After a provider releases a valid signature, the reservation is committed: -local timeout alone is not enough to recycle its inputs. +Before the provider durably accepts the exact signing transcript, failure can +discard the transaction, release or expire RFQ reservations, refresh state, +and reroute. After durable acceptance, the reservation is committed even if +the signer or response later becomes ambiguous: local timeout alone is not +enough to recycle its inputs. After ambiguous broadcast, the client first checks the exact transaction and its input outspends before constructing a conflicting replacement. After a @@ -865,9 +872,10 @@ that an orphaned venue transition remains live. - The user's signature cannot authorize a different input/output transaction body, and omitted witness commitments are covered by the approved proof-authentication protocol. -- The provider refuses to sign after the quote deadline, marks inputs committed - when it releases a signature, and immediately relays the exact transaction - according to policy. +- The provider refuses new signing commitments at or after the quote deadline, + durably commits exact inputs and transcript before invoking the signer, + stores the exact signed response before external release, and immediately + relays only that transaction according to policy. - Reserved provider inputs cannot be double-allocated, including across process crashes, restarts, or ambiguous broadcast. - Collaborative blinding outputs remain spendable by their intended recipients. @@ -947,8 +955,9 @@ three independently designed fragment layouts compose safely. trade privacy? - Which sighash profiles are supported for each input type, what fields does each profile commit, and how are omitted proofs or witnesses authenticated? -- What exact reservation, signature-release, relay, ambiguous-broadcast, and - input-retirement state machine does an RFQ provider implement? +- ADR 0007 resolves reservation, commit-before-sign, signature persistence, and + permanent input retirement. Exact relay, ambiguous-broadcast, and canonical + outspend reconciliation remain to be specified with the service layer. - Which chain and mempool evidence is required before a route is considered fresh enough to display or sign? - When should multiple RFQ signers be allowed in one transaction?