diff --git a/.gitignore b/.gitignore index 50f59a1f..6c4aba32 100644 --- a/.gitignore +++ b/.gitignore @@ -13,4 +13,5 @@ logs **/*/CLAUDE.md -/.claude \ No newline at end of file +/.claude +backups/ diff --git a/README.md b/README.md index 910588b6..6dec80e2 100644 --- a/README.md +++ b/README.md @@ -61,7 +61,7 @@ async fn main() -> Result<(), ddk::error::Error> { DDK is designed with a pluggable architecture, allowing you to choose or implement your own components: - **Transport**: Communication layer for DLC messages between peers. Implementations include Lightning Network gossip and Nostr protocol messaging. -- **Storage**: Persistence backend for contracts and wallet data. Implementations include Sled (embedded) and PostgreSQL. +- **Storage**: Persistence backend for contracts and wallet data. Implementations include Sled (embedded) and PostgreSQL. Databases written by releases up to 2.0 are moved to the columnar contract layout on upgrade; see [docs/postgres-contract-migration.md](./docs/postgres-contract-migration.md). - **Oracle**: External data source for contract attestations. Implementations include HTTP and Nostr-based oracle clients. You can create a custom DDK instance by implementing the required traits defined in [`ddk/src/lib.rs`](./ddk/src/lib.rs). diff --git a/ddk-manager/src/contract/offered_contract.rs b/ddk-manager/src/contract/offered_contract.rs index 0fc82615..78d43b46 100644 --- a/ddk-manager/src/contract/offered_contract.rs +++ b/ddk-manager/src/contract/offered_contract.rs @@ -197,6 +197,11 @@ impl OfferedContract { }) } + /// The id of the keys this contract signs with. + pub fn keys_id(&self) -> KeysId { + self.keys_id + } + /// The chain hash to put on offer messages for this contract. /// /// Contracts stored before ddk tracked the chain hash have none to diff --git a/ddk-node/src/bin/node.rs b/ddk-node/src/bin/node.rs index 90565961..8c489a8f 100644 --- a/ddk-node/src/bin/node.rs +++ b/ddk-node/src/bin/node.rs @@ -1,5 +1,5 @@ use clap::Parser; -use ddk_node::opts::NodeOpts; +use ddk_node::opts::{NodeCommand, NodeOpts}; use ddk_node::DdkNode; use std::str::FromStr; use tracing::level_filters::LevelFilter; @@ -34,7 +34,10 @@ async fn main() -> anyhow::Result<()> { tracing::subscriber::set_global_default(subscriber).unwrap(); - DdkNode::serve(opts).await?; + match opts.command { + Some(NodeCommand::Migrate) => DdkNode::migrate(opts).await?, + None => DdkNode::serve(opts).await?, + } Ok(()) } diff --git a/ddk-node/src/lib.rs b/ddk-node/src/lib.rs index d3f34571..77d2e4c3 100644 --- a/ddk-node/src/lib.rs +++ b/ddk-node/src/lib.rs @@ -58,6 +58,38 @@ impl DdkNode { } } + /// Applies the schema migrations and moves every contract still stored + /// in the legacy blob layout to the columnar layout, then returns. + /// + /// `PostgresStore::new` with migrations on does the same work, so a node + /// that starts normally migrates as well. This is for operators who want + /// to run the migration ahead of an upgrade, or against a restored + /// backup first. + pub async fn migrate(opts: NodeOpts) -> anyhow::Result<()> { + let logger = Arc::new(Logger::console( + "console_logger".to_string(), + LogLevel::from(opts.log), + )); + let storage = + PostgresStore::new(&opts.postgres_url, false, logger.clone(), opts.name).await?; + let report = storage.run_migrations().await?; + println!( + "Migrated {} contract(s) to the columnar layout. {} could not be moved.", + report.migrated, + report.failed.len() + ); + for (id, error) in &report.failed { + println!(" {id}: {error}"); + } + if !report.is_complete() { + anyhow::bail!( + "{} contract(s) are still in the legacy blob layout", + report.failed.len() + ); + } + Ok(()) + } + pub async fn serve(opts: NodeOpts) -> anyhow::Result<()> { let logger = Arc::new(Logger::console( "console_logger".to_string(), diff --git a/ddk-node/src/opts.rs b/ddk-node/src/opts.rs index 6a3a336d..3dab3c4f 100644 --- a/ddk-node/src/opts.rs +++ b/ddk-node/src/opts.rs @@ -1,4 +1,4 @@ -use clap::Parser; +use clap::{Parser, Subcommand}; use std::path::PathBuf; #[derive(Parser, Clone, Debug)] @@ -60,4 +60,15 @@ pub struct NodeOpts { #[arg(long)] #[arg(help = "Endpoint for bitcoind ZeroMQ blockhash notifications")] pub zmq_blockhash_endpoint: Option, + #[command(subcommand)] + pub command: Option, +} + +/// A one-shot task to run instead of serving the node. +#[derive(Subcommand, Clone, Debug)] +pub enum NodeCommand { + /// Apply the schema migrations and move every contract still stored in + /// the legacy blob layout to the columnar layout, then exit. Safe to run + /// again. Takes a backup of the database first. + Migrate, } diff --git a/ddk/src/storage/postgres/contract_row.rs b/ddk/src/storage/postgres/contract_row.rs new file mode 100644 index 00000000..71142905 --- /dev/null +++ b/ddk/src/storage/postgres/contract_row.rs @@ -0,0 +1,552 @@ +//! The columnar layout of a stored contract: one row in `dlc_contracts`. +//! +//! The DLC wire messages are the truth. The offer, accept, and sign messages +//! are stored byte for byte, so they keep their TLV streams and their own +//! protocol versioning. The state the manager derives and that the messages +//! do not carry, such as the adaptor info and the DLC transactions, lives in +//! its own column, one type per column. [`ContractRow::from_contract`] is the +//! one writer and [`ContractRow::into_contract`] is the one reader. Every +//! read path of the Postgres store goes through them. + +use bitcoin::consensus; +use bitcoin::secp256k1::PublicKey; +use bitcoin::{SignedAmount, Transaction, Txid}; +use ddk_manager::contract::accepted_contract::AcceptedContract; +use ddk_manager::contract::offered_contract::OfferedContract; +use ddk_manager::contract::ser::{dlc_transactions, ContractPrefix}; +use ddk_manager::contract::signed_contract::SignedContract; +use ddk_manager::contract::{ + AdaptorInfo, ClosedContract, Contract, FailedAcceptContract, FailedSignContract, + PreClosedContract, +}; +use ddk_manager::error::Error; +use ddk_messages::oracle_msgs::OracleAttestation; +use ddk_messages::ser_impls::{party_params, read_vec, write_vec}; +use ddk_messages::{AcceptDlc, OfferDlc, SignDlc}; +use lightning::util::ser::{Readable, Writeable}; +use sqlx::FromRow; +use std::str::FromStr; + +/// The layout version of the rows this module writes. +/// +/// Version 1 is the legacy blob layout in the `contract_data` table. It never +/// appears in `dlc_contracts`; rows are moved out of it by +/// `PostgresStore::migrate_legacy_contracts`. +pub const CONTRACT_ROW_FORMAT_VERSION: i16 = 2; + +/// The layout version of the legacy blob rows in `contract_data`. +pub const LEGACY_BLOB_FORMAT_VERSION: i16 = 1; + +/// The value of `announcement_id` and `oracle_pubkey` when a contract has no +/// announcement to report. Kept from the metadata table so consumers keep +/// their filter. +pub const NO_ANNOUNCEMENT: &str = "legacy_data"; + +/// One row of `dlc_contracts`. The column list is the struct, so adding a +/// field is a migration plus a change to the two functions below. +#[derive(Debug, Clone, FromRow)] +pub struct ContractRow { + /// Hex contract id, or the temporary id before the contract is accepted. + pub id: String, + pub format_version: i16, + /// The [`ContractPrefix`] of the stored state. + pub state: i16, + /// Hex temporary contract id. + pub temporary_id: String, + pub is_offer_party: bool, + /// Hex compressed public key of the counter party. + pub counter_party: String, + pub keys_id: Vec, + pub contract_flags: i16, + pub chain_hash: Option>, + pub offer_collateral: i64, + pub accept_collateral: i64, + pub total_collateral: i64, + pub fee_rate_per_vb: i64, + pub cet_locktime: i32, + pub refund_locktime: i32, + pub announcement_id: String, + pub oracle_pubkey: String, + pub funding_txid: Option, + pub cet_txid: Option, + pub pnl: Option, + /// The offer message, wire encoded. + pub offer_message: Vec, + /// The accept message, wire encoded. Set from the accepted state on, and + /// on a failed accept, where it is the message that failed. + pub accept_message: Option>, + /// The sign message, wire encoded. Set from the signed state on, and on + /// a failed sign, where it is the message that failed. + pub sign_message: Option>, + /// The offer party's params, wire encoded. + pub offer_params: Vec, + /// The accept party's params, wire encoded. Set from the accepted state on. + pub accept_params: Option>, + /// The adaptor infos, wire encoded. Set from the accepted state on. + pub adaptor_infos: Option>, + /// The DLC transactions, wire encoded. Set from the accepted state on. + pub dlc_transactions: Option>, + pub channel_id: Option>, + /// The oracle attestations, wire encoded. Set on pre-closed and closed + /// contracts that closed with an attestation. + pub attestations: Option>, + /// The signed CET, consensus encoded. Set on pre-closed contracts and on + /// closed contracts that closed with a CET. + pub signed_cet: Option>, + /// The error of a failed accept or failed sign. + pub error_message: Option, +} + +fn storage_error(column: &str, error: impl std::fmt::Debug) -> Error { + Error::StorageError(format!("dlc_contracts.{column}: {error:?}")) +} + +fn decode(column: &str, bytes: &[u8]) -> Result { + let mut cursor = lightning::io::Cursor::new(bytes); + T::read(&mut cursor).map_err(|e| storage_error(column, e)) +} + +fn decode_with<'a, T, R>(column: &str, bytes: &'a [u8], read: R) -> Result +where + R: FnOnce(&mut lightning::io::Cursor<&'a [u8]>) -> Result, +{ + let mut cursor = lightning::io::Cursor::new(bytes); + read(&mut cursor).map_err(|e| storage_error(column, e)) +} + +fn encode_with(column: &str, value: &T, write: W) -> Result, Error> +where + W: FnOnce(&T, &mut Vec) -> Result<(), lightning::io::Error>, +{ + let mut buffer = Vec::new(); + write(value, &mut buffer).map_err(|e| storage_error(column, e))?; + Ok(buffer) +} + +fn required(column: &str, value: Option) -> Result { + value.ok_or_else(|| storage_error(column, "missing for this state")) +} + +fn array_32(column: &str, bytes: &[u8]) -> Result<[u8; 32], Error> { + bytes + .try_into() + .map_err(|_| storage_error(column, format!("expected 32 bytes, got {}", bytes.len()))) +} + +fn hex_32(column: &str, hex: &str) -> Result<[u8; 32], Error> { + let bytes = hex::decode(hex).map_err(|e| storage_error(column, e))?; + array_32(column, &bytes) +} + +fn adaptor_signatures( + signatures: &ddk_messages::CetAdaptorSignatures, +) -> Vec { + signatures + .ecdsa_adaptor_signatures + .iter() + .map(|s| s.signature) + .collect() +} + +/// The offered, accepted, and signed contract a state wraps, whichever exist. +fn parts( + contract: &Contract, +) -> ( + &OfferedContract, + Option<&AcceptedContract>, + Option<&SignedContract>, +) { + match contract { + Contract::Offered(o) | Contract::Rejected(o) => (o, None, None), + Contract::Accepted(a) => (&a.offered_contract, Some(a), None), + Contract::Signed(s) | Contract::Confirmed(s) | Contract::Refunded(s) => ( + &s.accepted_contract.offered_contract, + Some(&s.accepted_contract), + Some(s), + ), + Contract::PreClosed(p) => parts_of_signed(&p.signed_contract), + Contract::Closed(c) => parts_of_signed(&c.signed_contract), + Contract::FailedAccept(f) => (&f.offered_contract, None, None), + Contract::FailedSign(f) => ( + &f.accepted_contract.offered_contract, + Some(&f.accepted_contract), + None, + ), + } +} + +fn parts_of_signed( + signed: &SignedContract, +) -> ( + &OfferedContract, + Option<&AcceptedContract>, + Option<&SignedContract>, +) { + ( + &signed.accepted_contract.offered_contract, + Some(&signed.accepted_contract), + Some(signed), + ) +} + +impl ContractRow { + /// The one writer: the row for a contract in any state. + pub fn from_contract(contract: &Contract) -> Result { + let (offered, accepted, signed) = parts(contract); + // The `Contract` accessors return zeros for a closed contract and + // unwrap its CET, so the metadata comes from the offered contract + // every state wraps. + let offer_collateral = offered.offer_params.collateral; + let total_collateral = offered.total_collateral; + let accept_collateral = total_collateral + .checked_sub(offer_collateral) + .unwrap_or(bitcoin::Amount::ZERO); + + let announcement = offered + .contract_info + .first() + .and_then(|info| info.oracle_announcements.first()); + let announcement_id = announcement + .map(|a| a.oracle_event.event_id.clone()) + .unwrap_or_else(|| NO_ANNOUNCEMENT.to_string()); + let oracle_pubkey = announcement + .map(|a| a.oracle_public_key.to_string()) + .unwrap_or_else(|| NO_ANNOUNCEMENT.to_string()); + + let offer_message = OfferDlc::from(offered).encode(); + let accept_message = match contract { + Contract::FailedAccept(f) => Some(f.accept_message.encode()), + _ => accepted.map(|a| a.get_accept_contract_msg(&a.adaptor_signatures).encode()), + }; + let sign_message = match contract { + Contract::FailedSign(f) => Some(f.sign_message.encode()), + _ => signed.map(|s| s.get_sign_dlc(s.adaptor_signatures.clone()).encode()), + }; + + let offer_params = encode_with("offer_params", &offered.offer_params, party_params::write)?; + let accept_params = accepted + .map(|a| encode_with("accept_params", &a.accept_params, party_params::write)) + .transpose()?; + let adaptor_infos = accepted + .map(|a| encode_with("adaptor_infos", &a.adaptor_infos, write_vec)) + .transpose()?; + let dlc_transactions = accepted + .map(|a| { + encode_with( + "dlc_transactions", + &a.dlc_transactions, + dlc_transactions::write, + ) + }) + .transpose()?; + + let (attestations, signed_cet) = match contract { + Contract::PreClosed(p) => (p.attestations.as_ref(), Some(&p.signed_cet)), + Contract::Closed(c) => (c.attestations.as_ref(), c.signed_cet.as_ref()), + _ => (None, None), + }; + let attestations = attestations + .map(|a| encode_with("attestations", a, write_vec)) + .transpose()?; + let signed_cet = signed_cet.map(consensus::serialize); + + let error_message = match contract { + Contract::FailedAccept(f) => Some(f.error_message.clone()), + Contract::FailedSign(f) => Some(f.error_message.clone()), + _ => None, + }; + + Ok(ContractRow { + id: hex::encode(contract.get_id()), + format_version: CONTRACT_ROW_FORMAT_VERSION, + state: ContractPrefix::get_prefix(contract) as i16, + temporary_id: hex::encode(contract.get_temporary_id()), + is_offer_party: offered.is_offer_party, + counter_party: hex::encode(contract.get_counter_party_id().serialize()), + keys_id: offered.keys_id().to_vec(), + contract_flags: offered.contract_flags as i16, + chain_hash: offered.chain_hash.map(|h| h.to_vec()), + offer_collateral: offer_collateral.to_sat() as i64, + accept_collateral: accept_collateral.to_sat() as i64, + total_collateral: total_collateral.to_sat() as i64, + fee_rate_per_vb: offered.fee_rate_per_vb as i64, + cet_locktime: offered.cet_locktime as i32, + refund_locktime: offered.refund_locktime as i32, + announcement_id, + oracle_pubkey, + funding_txid: contract.get_funding_txid().map(|txid| txid.to_string()), + cet_txid: contract.get_cet_txid().map(|txid| txid.to_string()), + pnl: Some(contract.get_pnl().to_sat()), + offer_message, + accept_message, + sign_message, + offer_params, + accept_params, + adaptor_infos, + dlc_transactions, + channel_id: signed.and_then(|s| s.channel_id).map(|id| id.to_vec()), + attestations, + signed_cet, + error_message, + }) + } + + /// The one reader: the contract a row stands for. + pub fn into_contract(self) -> Result { + if self.format_version != CONTRACT_ROW_FORMAT_VERSION { + return Err(storage_error( + "format_version", + format!("unknown row format {}", self.format_version), + )); + } + let prefix = ContractPrefix::try_from(self.state as u8)?; + Ok(match prefix { + ContractPrefix::Offered => Contract::Offered(self.offered()?), + ContractPrefix::Rejected => Contract::Rejected(self.offered()?), + ContractPrefix::Accepted => Contract::Accepted(self.accepted()?), + ContractPrefix::Signed => Contract::Signed(self.signed()?), + ContractPrefix::Confirmed => Contract::Confirmed(self.signed()?), + ContractPrefix::Refunded => Contract::Refunded(self.signed()?), + ContractPrefix::PreClosed => Contract::PreClosed(PreClosedContract { + signed_contract: self.signed()?, + attestations: self.attestations()?, + signed_cet: required("signed_cet", self.signed_cet()?)?, + }), + ContractPrefix::Closed => Contract::Closed(ClosedContract { + attestations: self.attestations()?, + signed_cet: self.signed_cet()?, + contract_id: hex_32("id", &self.id)?, + temporary_contract_id: hex_32("temporary_id", &self.temporary_id)?, + counter_party_id: self.counter_party()?, + funding_txid: Txid::from_str(required( + "funding_txid", + self.funding_txid.as_deref(), + )?) + .map_err(|e| storage_error("funding_txid", e))?, + pnl: SignedAmount::from_sat(required("pnl", self.pnl)?), + signed_contract: self.signed()?, + }), + ContractPrefix::FailedAccept => Contract::FailedAccept(FailedAcceptContract { + offered_contract: self.offered()?, + accept_message: self.accept_message()?, + error_message: required("error_message", self.error_message.clone())?, + }), + ContractPrefix::FailedSign => Contract::FailedSign(FailedSignContract { + accepted_contract: self.accepted()?, + sign_message: self.sign_message()?, + error_message: required("error_message", self.error_message.clone())?, + }), + }) + } + + fn counter_party(&self) -> Result { + PublicKey::from_str(&self.counter_party).map_err(|e| storage_error("counter_party", e)) + } + + fn offered(&self) -> Result { + let offer: OfferDlc = decode("offer_message", &self.offer_message)?; + let keys_id = array_32("keys_id", &self.keys_id)?; + let mut offered = + OfferedContract::try_from_offer_dlc(&offer, self.counter_party()?, keys_id) + .map_err(|e| storage_error("offer_message", e))?; + offered.is_offer_party = self.is_offer_party; + offered.chain_hash = self + .chain_hash + .as_deref() + .map(|h| array_32("chain_hash", h)) + .transpose()?; + offered.offer_params = decode_with("offer_params", &self.offer_params, party_params::read)?; + Ok(offered) + } + + fn accept_message(&self) -> Result { + decode( + "accept_message", + required("accept_message", self.accept_message.as_deref())?, + ) + } + + fn accepted(&self) -> Result { + let accept = self.accept_message()?; + Ok(AcceptedContract { + offered_contract: self.offered()?, + accept_params: decode_with( + "accept_params", + required("accept_params", self.accept_params.as_deref())?, + party_params::read, + )?, + funding_inputs: accept.funding_inputs, + adaptor_infos: decode_with::, _>( + "adaptor_infos", + required("adaptor_infos", self.adaptor_infos.as_deref())?, + read_vec, + )?, + adaptor_signatures: adaptor_signatures(&accept.cet_adaptor_signatures), + accept_refund_signature: accept.refund_signature, + dlc_transactions: decode_with( + "dlc_transactions", + required("dlc_transactions", self.dlc_transactions.as_deref())?, + dlc_transactions::read, + )?, + tlvs: accept.tlvs, + }) + } + + fn sign_message(&self) -> Result { + decode( + "sign_message", + required("sign_message", self.sign_message.as_deref())?, + ) + } + + fn signed(&self) -> Result { + let sign = self.sign_message()?; + Ok(SignedContract { + accepted_contract: self.accepted()?, + adaptor_signatures: adaptor_signatures(&sign.cet_adaptor_signatures), + offer_refund_signature: sign.refund_signature, + funding_signatures: sign.funding_signatures, + channel_id: self + .channel_id + .as_deref() + .map(|id| array_32("channel_id", id)) + .transpose()?, + tlvs: sign.tlvs, + }) + } + + fn attestations(&self) -> Result>, Error> { + self.attestations + .as_deref() + .map(|bytes| decode_with("attestations", bytes, read_vec)) + .transpose() + } + + fn signed_cet(&self) -> Result, Error> { + self.signed_cet + .as_deref() + .map(|bytes| consensus::deserialize(bytes).map_err(|e| storage_error("signed_cet", e))) + .transpose() + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::util::ser::{deserialize_contract, serialize_contract}; + use ddk_messages::tlv_stream::TlvStream; + + fn fixture(name: &str) -> Contract { + let path = format!( + "{}/../testconfig/contract_binaries/{name}", + env!("CARGO_MANIFEST_DIR") + ); + let bytes = std::fs::read(path).unwrap(); + deserialize_contract(&bytes).unwrap_or_else(|e| panic!("fixture {name}: {e}")) + } + + /// The row keeps everything the legacy blob kept: a contract written as + /// a row and read back serializes to the same legacy bytes. + fn assert_round_trips(contract: &Contract) { + let expected = serialize_contract(contract).unwrap(); + let row = ContractRow::from_contract(contract).unwrap(); + assert_eq!(row.format_version, CONTRACT_ROW_FORMAT_VERSION); + let read = row.into_contract().unwrap(); + assert_eq!(serialize_contract(&read).unwrap(), expected); + } + + #[test] + fn stored_contracts_round_trip_byte_for_byte() { + for name in [ + "Offered", + "Accepted", + "Signed", + "Confirmed", + "PreClosed", + "Closed", + "old/Offered", + ] { + assert_round_trips(&fixture(name)); + } + } + + /// The states that wrap another state's struct share its columns. + #[test] + fn wrapped_states_round_trip() { + let Contract::Signed(signed) = fixture("Signed") else { + panic!("fixture is not signed") + }; + let Contract::Offered(offered) = fixture("Offered") else { + panic!("fixture is not offered") + }; + let Contract::Accepted(accepted) = fixture("Accepted") else { + panic!("fixture is not accepted") + }; + let accept_message = accepted.get_accept_contract_msg(&accepted.adaptor_signatures); + let sign_message = signed.get_sign_dlc(signed.adaptor_signatures.clone()); + + assert_round_trips(&Contract::Rejected(offered.clone())); + assert_round_trips(&Contract::Refunded(signed.clone())); + assert_round_trips(&Contract::FailedAccept(FailedAcceptContract { + offered_contract: offered, + accept_message, + error_message: "bad accept".to_string(), + })); + assert_round_trips(&Contract::FailedSign(FailedSignContract { + accepted_contract: accepted, + sign_message, + error_message: "bad sign".to_string(), + })); + } + + /// A stream holding one record of type 65007 with `body` as its one-byte body. + fn stream_with_record(body: u8) -> TlvStream { + let bytes = [0xfd, 0xfd, 0xef, 0x01, body]; + TlvStream::read_to_end(&mut lightning::io::Cursor::new(bytes)).unwrap() + } + + /// The TLV streams travel inside the stored messages, on every state. + #[test] + fn tlv_streams_survive_on_every_state() { + let Contract::Closed(mut closed) = fixture("Closed") else { + panic!("fixture is not closed") + }; + closed + .signed_contract + .accepted_contract + .offered_contract + .tlvs = stream_with_record(1); + closed.signed_contract.accepted_contract.tlvs = stream_with_record(2); + closed.signed_contract.tlvs = stream_with_record(3); + + let row = ContractRow::from_contract(&Contract::Closed(closed)).unwrap(); + let Contract::Closed(read) = row.into_contract().unwrap() else { + panic!("state changed in storage") + }; + let signed = read.signed_contract; + assert_eq!( + signed.accepted_contract.offered_contract.tlvs, + stream_with_record(1) + ); + assert_eq!(signed.accepted_contract.tlvs, stream_with_record(2)); + assert_eq!(signed.tlvs, stream_with_record(3)); + } + + /// A row of an unknown layout is an error, not a misread. + #[test] + fn unknown_format_version_is_rejected() { + let mut row = ContractRow::from_contract(&fixture("Offered")).unwrap(); + row.format_version = 99; + assert!(row.into_contract().is_err()); + } + + /// A state that needs a column the row does not hold is an error that + /// names the column. + #[test] + fn missing_column_names_the_column() { + let mut row = ContractRow::from_contract(&fixture("Signed")).unwrap(); + row.sign_message = None; + let error = row.into_contract().unwrap_err().to_string(); + assert!(error.contains("sign_message"), "{error}"); + } +} diff --git a/ddk/src/storage/postgres/legacy.rs b/ddk/src/storage/postgres/legacy.rs new file mode 100644 index 00000000..ff11a4ef --- /dev/null +++ b/ddk/src/storage/postgres/legacy.rs @@ -0,0 +1,203 @@ +//! The legacy blob layout: one opaque blob per contract in `contract_data`, +//! with a copy of a few fields in `contract_metadata`. +//! +//! Releases up to 2.0 wrote this layout. It is the version one row format. +//! This module is the version one reader and the migration that moves each +//! blob into a `dlc_contracts` row. Both go away in the release that drops +//! the two legacy tables. + +use super::contract_row::ContractRow; +use super::PostgresStore; +use crate::error::to_storage_error; +use crate::logger::{log_error, log_info, log_warn, WriteLog}; +use crate::storage::sqlx::ContractData; +use crate::util::ser::deserialize_contract; +use ddk_manager::contract::Contract; +use ddk_manager::error::Error; +use sqlx::Postgres; + +/// A blob row that has no `dlc_contracts` row yet. +const LEGACY_ROWS: &str = "SELECT cd.id, cd.state, cd.contract_data, cd.is_compressed + FROM contract_data cd + WHERE NOT EXISTS (SELECT 1 FROM dlc_contracts d WHERE d.id = cd.id)"; + +/// What [`PostgresStore::migrate_legacy_contracts`] did. +#[derive(Debug, Default, Clone, PartialEq, Eq)] +pub struct LegacyMigrationReport { + /// Contracts moved into `dlc_contracts`. + pub migrated: usize, + /// Contracts left in the legacy tables, with the error that stopped each. + pub failed: Vec<(String, String)>, +} + +impl LegacyMigrationReport { + /// True when no contract was left behind. + pub fn is_complete(&self) -> bool { + self.failed.is_empty() + } +} + +impl PostgresStore { + /// Contracts still stored in the legacy blob layout. + pub async fn count_legacy_contracts(&self) -> Result { + let (count,): (i64,) = sqlx::query_as(&format!("SELECT COUNT(*) FROM ({LEGACY_ROWS}) l")) + .fetch_one(&self.pool) + .await + .map_err(to_storage_error)?; + Ok(count as u64) + } + + /// Moves every contract still in the legacy blob layout into a + /// `dlc_contracts` row. + /// + /// One transaction per contract: the row is inserted and the legacy rows + /// are deleted together, so a crash leaves each contract in exactly one + /// place. A contract that fails to convert stays in the legacy tables, + /// is reported, and does not stop the others. Safe to run again: it only + /// selects what is left. + pub async fn migrate_legacy_contracts(&self) -> Result { + let rows = sqlx::query_as::(LEGACY_ROWS) + .fetch_all(&self.pool) + .await + .map_err(to_storage_error)?; + + let mut report = LegacyMigrationReport::default(); + if rows.is_empty() { + return Ok(report); + } + log_info!( + self.logger, + "Migrating contracts from the legacy blob layout. count={}", + rows.len() + ); + + for legacy in rows { + match self.migrate_legacy_row(&legacy).await { + Ok(()) => report.migrated += 1, + Err(e) => { + log_error!( + self.logger, + "Could not migrate contract from the legacy blob layout. id={} error={}", + legacy.id, + e + ); + report.failed.push((legacy.id, e.to_string())); + } + } + } + + log_info!( + self.logger, + "Finished migrating contracts from the legacy blob layout. migrated={} failed={}", + report.migrated, + report.failed.len() + ); + Ok(report) + } + + async fn migrate_legacy_row(&self, legacy: &ContractData) -> Result<(), Error> { + let contract = deserialize_contract(&legacy.contract_data)?; + let row = ContractRow::from_contract(&contract)?; + if row.id != legacy.id { + return Err(Error::StorageError(format!( + "legacy row id {} does not match the contract id {}", + legacy.id, row.id + ))); + } + + let mut tx = self.pool.begin().await.map_err(to_storage_error)?; + super::upsert_contract_row(&mut tx, &row).await?; + delete_legacy_rows(&mut tx, &legacy.id).await?; + tx.commit().await.map_err(to_storage_error)?; + Ok(()) + } + + /// Logs a warning when contracts are still in the legacy blob layout. + pub async fn warn_if_legacy_contracts_remain(&self) -> Result { + let remaining = self.count_legacy_contracts().await?; + if remaining > 0 { + log_warn!( + self.logger, + "{} contract(s) are still stored in the legacy blob layout (contract_data). \ + They still load, but the legacy reader will be removed in a later release. \ + Run `ddk-node migrate --postgres-url ` or \ + `PostgresStore::migrate_legacy_contracts` to move them to dlc_contracts.", + remaining + ); + } + Ok(remaining) + } + + /// One legacy contract by id, when it has no `dlc_contracts` row. + pub(super) async fn legacy_contract(&self, id: &str) -> Result, Error> { + let row = + sqlx::query_as::(&format!("{LEGACY_ROWS} AND cd.id = $1")) + .bind(id) + .fetch_optional(&self.pool) + .await + .map_err(to_storage_error)?; + match row { + Some(row) => { + self.warn_legacy_read(1); + Ok(Some(deserialize_contract(&row.contract_data)?)) + } + None => Ok(None), + } + } + + /// The legacy contracts with no `dlc_contracts` row, in `state` when given. + pub(super) async fn legacy_contracts( + &self, + state: Option, + ) -> Result, Error> { + let rows = match state { + Some(state) => { + sqlx::query_as::(&format!( + "{LEGACY_ROWS} AND cd.state = $1" + )) + .bind(state) + .fetch_all(&self.pool) + .await + } + None => { + sqlx::query_as::(LEGACY_ROWS) + .fetch_all(&self.pool) + .await + } + } + .map_err(to_storage_error)?; + if rows.is_empty() { + return Ok(vec![]); + } + self.warn_legacy_read(rows.len()); + rows.iter() + .map(|row| deserialize_contract(&row.contract_data)) + .collect() + } + + fn warn_legacy_read(&self, count: usize) { + log_warn!( + self.logger, + "Read {} contract(s) from the legacy blob layout. Run `ddk-node migrate` to move them.", + count + ); + } +} + +/// Deletes the legacy rows of a contract, if any. +pub(super) async fn delete_legacy_rows( + tx: &mut sqlx::Transaction<'_, Postgres>, + id: &str, +) -> Result<(), Error> { + sqlx::query("DELETE FROM contract_data WHERE id = $1") + .bind(id) + .execute(&mut **tx) + .await + .map_err(to_storage_error)?; + sqlx::query("DELETE FROM contract_metadata WHERE id = $1") + .bind(id) + .execute(&mut **tx) + .await + .map_err(to_storage_error)?; + Ok(()) +} diff --git a/ddk/src/storage/postgres/migrations/0010_dlc_contracts.down.sql b/ddk/src/storage/postgres/migrations/0010_dlc_contracts.down.sql new file mode 100644 index 00000000..8fee70aa --- /dev/null +++ b/ddk/src/storage/postgres/migrations/0010_dlc_contracts.down.sql @@ -0,0 +1,10 @@ +-- The legacy blob cannot be rebuilt in SQL, so a table that still holds +-- contracts must not be dropped. Delete or export the rows first. +DO $$ +BEGIN + IF EXISTS (SELECT 1 FROM dlc_contracts) THEN + RAISE EXCEPTION 'dlc_contracts still holds contracts; reverting would strand them'; + END IF; +END $$; + +DROP TABLE dlc_contracts; diff --git a/ddk/src/storage/postgres/migrations/0010_dlc_contracts.up.sql b/ddk/src/storage/postgres/migrations/0010_dlc_contracts.up.sql new file mode 100644 index 00000000..230959e9 --- /dev/null +++ b/ddk/src/storage/postgres/migrations/0010_dlc_contracts.up.sql @@ -0,0 +1,50 @@ +-- One row per contract, with the wire messages and the manager-only state in +-- their own columns. Replaces the opaque blob in contract_data and the +-- duplicated columns in contract_metadata. Rows are moved here by +-- `PostgresStore::migrate_legacy_contracts`, which needs the Rust decoder for +-- the old blob, so this migration only creates the table. +-- +-- format_version: the layout of the row. 1 is the legacy blob layout that +-- lives in contract_data and never appears in this table. 2 is this layout. +CREATE TABLE dlc_contracts ( + id TEXT PRIMARY KEY, + format_version SMALLINT NOT NULL, + state SMALLINT NOT NULL CHECK (state >= 0), + temporary_id TEXT NOT NULL, + is_offer_party BOOLEAN NOT NULL, + counter_party TEXT NOT NULL, + keys_id BYTEA NOT NULL, + contract_flags SMALLINT NOT NULL CHECK (contract_flags >= 0), + chain_hash BYTEA, + offer_collateral BIGINT NOT NULL CHECK (offer_collateral >= 0), + accept_collateral BIGINT NOT NULL CHECK (accept_collateral >= 0), + total_collateral BIGINT NOT NULL CHECK (total_collateral >= 0), + fee_rate_per_vb BIGINT NOT NULL CHECK (fee_rate_per_vb >= 0), + cet_locktime INTEGER NOT NULL CHECK (cet_locktime >= 0), + refund_locktime INTEGER NOT NULL CHECK (refund_locktime >= 0), + announcement_id TEXT NOT NULL, + oracle_pubkey TEXT NOT NULL, + funding_txid TEXT, + cet_txid TEXT, + pnl BIGINT, + -- The DLC wire messages, byte for byte, including their TLV streams. + offer_message BYTEA NOT NULL, + accept_message BYTEA, + sign_message BYTEA, + -- Manager-only state that is not carried by the messages. Each column + -- holds exactly one type with its own wire encoding. + offer_params BYTEA NOT NULL, + accept_params BYTEA, + adaptor_infos BYTEA, + dlc_transactions BYTEA, + channel_id BYTEA, + attestations BYTEA, + signed_cet BYTEA, + error_message TEXT, + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + updated_at TIMESTAMPTZ NOT NULL DEFAULT now() +); + +CREATE INDEX idx_dlc_contracts_state ON dlc_contracts (state); +CREATE INDEX idx_dlc_contracts_counter_party ON dlc_contracts (counter_party); +CREATE INDEX idx_dlc_contracts_temporary_id ON dlc_contracts (temporary_id); diff --git a/ddk/src/storage/postgres/mod.rs b/ddk/src/storage/postgres/mod.rs index 0c0f2cc3..20713e39 100644 --- a/ddk/src/storage/postgres/mod.rs +++ b/ddk/src/storage/postgres/mod.rs @@ -1,12 +1,12 @@ -use super::sqlx::{ContractData, ContractMetadata, SqlxError}; +pub mod contract_row; +pub mod legacy; + +use super::sqlx::{ContractMetadata, SqlxError}; use crate::error::{StorageError, WalletError}; use crate::logger::Logger; -use crate::logger::{log_info, WriteLog}; +use crate::logger::{log_debug, log_info, WriteLog}; use crate::Storage; -use crate::{ - error::to_storage_error, - util::ser::{deserialize_contract, serialize_contract, ContractPrefix}, -}; +use crate::{error::to_storage_error, util::ser::ContractPrefix}; use bdk_chain::{ local_chain, tx_graph, Anchor, ConfirmationBlockTime, DescriptorExt, DescriptorId, Merge, }; @@ -22,6 +22,7 @@ use bdk_wallet::keys::DescriptorPublicKey; use bdk_wallet::ChangeSet; use bdk_wallet::KeychainKind; use bdk_wallet::KeychainKind::{External, Internal}; +use contract_row::ContractRow; use ddk_manager::{ contract::{ offered_contract::OfferedContract, signed_contract::SignedContract, Contract, @@ -29,6 +30,7 @@ use ddk_manager::{ }, Storage as ManagerStorage, }; +pub use legacy::LegacyMigrationReport; use serde_json::json; use sqlx::pool::PoolOptions; use sqlx::postgres::PgRow; @@ -99,19 +101,40 @@ impl PostgresStore { .connect(url) .await .map_err(|e| StorageError::Sqlx(e.into()))?; - if migrations { - log_info!(logger, "Migrating postgres"); - MIGRATOR - .run(&pool) - .await - .map_err(|e| StorageError::Sqlx(e.into()))?; - } - - Ok(Self { + let store = Self { pool, logger, wallet_name, - }) + }; + + if migrations { + store.run_migrations().await?; + } + // Without migrations the new table may not exist yet, and every read + // will say so; the count is not the place to fail. + if let Err(e) = store.warn_if_legacy_contracts_remain().await { + log_debug!( + store.logger, + "Could not count contracts in the legacy blob layout. error={}", + e + ); + } + + Ok(store) + } + + /// Applies the schema migrations, then moves every contract still in the + /// legacy blob layout to the columnar layout. `new` runs this when + /// `migrations` is on. + pub async fn run_migrations(&self) -> Result { + log_info!(self.logger, "Migrating postgres"); + MIGRATOR + .run(&self.pool) + .await + .map_err(|e| StorageError::Sqlx(e.into()))?; + self.migrate_legacy_contracts() + .await + .map_err(|e| StorageError::Init(e.to_string())) } pub async fn get_contract_metadata( @@ -124,7 +147,8 @@ impl PostgresStore { .collect::>() .join(", "); - let query = format!("SELECT * FROM contract_metadata WHERE state IN ({placeholders})"); + let query = + format!("SELECT * FROM ({CONTRACT_METADATA}) c WHERE c.state IN ({placeholders})"); let mut query = sqlx::query_as::<_, ContractMetadata>(&query); @@ -137,7 +161,7 @@ impl PostgresStore { .await .map_err(|e| StorageError::Sqlx(e.into()))? } else { - sqlx::query_as::("SELECT * FROM contract_metadata") + sqlx::query_as::(CONTRACT_METADATA) .fetch_all(&self.pool) .await .map_err(|e| StorageError::Sqlx(e.into()))? @@ -149,9 +173,9 @@ impl PostgresStore { &self, id: &str, ) -> Result { - let row = sqlx::query_as::( - "SELECT * FROM contract_metadata WHERE id = $1", - ) + let row = sqlx::query_as::(&format!( + "SELECT * FROM ({CONTRACT_METADATA}) c WHERE c.id = $1" + )) .bind(id) .fetch_one(&self.pool) .await @@ -160,15 +184,36 @@ impl PostgresStore { } pub async fn get_offer_metadata(&self) -> Result, StorageError> { - let rows = sqlx::query_as::( - "SELECT * FROM contract_metadata WHERE state = 1", - ) + let rows = sqlx::query_as::(&format!( + "SELECT * FROM ({CONTRACT_METADATA}) c WHERE c.state = 1" + )) .fetch_all(&self.pool) .await .map_err(|e| StorageError::Sqlx(e.into()))?; Ok(rows) } + /// The contracts in `state`, read through the one decoder, plus any + /// still in the legacy blob layout. + async fn contracts_in_state( + &self, + state: ContractPrefix, + ) -> Result, ddk_manager::error::Error> { + let state = state as i16; + let rows = + sqlx::query_as::("SELECT * FROM dlc_contracts WHERE state = $1") + .bind(state) + .fetch_all(&self.pool) + .await + .map_err(to_storage_error)?; + let mut contracts = rows + .into_iter() + .map(ContractRow::into_contract) + .collect::, _>>()?; + contracts.extend(self.legacy_contracts(Some(state)).await?); + Ok(contracts) + } + #[tracing::instrument(skip(self))] pub(crate) async fn read(&self) -> Result { log_info!( @@ -449,32 +494,32 @@ impl ManagerStorage for PostgresStore { &self, id: &ddk_manager::ContractId, ) -> Result, ddk_manager::error::Error> { - let contract = - sqlx::query_as::("SELECT * FROM contract_data WHERE id = $1") - .bind(hex::encode(id)) + let id = hex::encode(id); + let row = + sqlx::query_as::("SELECT * FROM dlc_contracts WHERE id = $1") + .bind(&id) .fetch_optional(&self.pool) .await .map_err(to_storage_error)?; - if let Some(contract) = contract { - Ok(Some(deserialize_contract(&contract.contract_data)?)) - } else { - Ok(None) + match row { + Some(row) => Ok(Some(row.into_contract()?)), + None => self.legacy_contract(&id).await, } } #[tracing::instrument(skip(self))] async fn get_contracts(&self) -> Result, ddk_manager::error::Error> { - let contracts = sqlx::query_as::("SELECT * FROM contract_data") + let rows = sqlx::query_as::("SELECT * FROM dlc_contracts") .fetch_all(&self.pool) .await .map_err(to_storage_error)?; - let contracts = contracts + let mut contracts = rows .into_iter() - .map(|c| deserialize_contract(&c.contract_data)) + .map(ContractRow::into_contract) .collect::, _>>()?; - + contracts.extend(self.legacy_contracts(None).await?); Ok(contracts) } @@ -482,53 +527,9 @@ impl ManagerStorage for PostgresStore { &self, contract: &OfferedContract, ) -> Result<(), ddk_manager::error::Error> { + let row = ContractRow::from_contract(&Contract::Offered(contract.clone()))?; let mut tx = self.pool.begin().await.map_err(to_storage_error)?; - let oracle_pubkey = contract.contract_info[0].oracle_announcements[0].oracle_public_key; - let announcement_id = contract.contract_info[0].oracle_announcements[0] - .oracle_event - .event_id - .clone(); - - sqlx::query( - r#" - INSERT INTO contract_metadata ( - id, state, is_offer_party, counter_party, - offer_collateral, accept_collateral, total_collateral, fee_rate_per_vb, - cet_locktime, refund_locktime, pnl, funding_txid, cet_txid, announcement_id, oracle_pubkey - ) - VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15) - "#, - ) - .bind(hex::encode(contract.id)) - .bind(1_i16) - .bind(contract.is_offer_party) - .bind(hex::encode(contract.counter_party.serialize())) - .bind(contract.offer_params.collateral.to_sat() as i64) - .bind((contract.total_collateral - contract.offer_params.collateral).to_sat() as i64) - .bind(contract.total_collateral.to_sat() as i64) - .bind(contract.fee_rate_per_vb as i64) - .bind(contract.cet_locktime as i32) - .bind(contract.refund_locktime as i32) - .bind(None as Option) - .bind(None as Option) - .bind(None as Option) - .bind(announcement_id) - .bind(oracle_pubkey.to_string()) - .execute(&mut *tx) - .await - .map_err(to_storage_error)?; - - sqlx::query( - "INSERT INTO contract_data (id, state, contract_data, is_compressed) VALUES ($1, $2, $3, $4)" - ) - .bind(hex::encode(contract.id)) - .bind(1_i16) - .bind(serialize_contract(&Contract::Offered(contract.clone()))?) - .bind(false) - .execute(&mut *tx) - .await - .map_err(to_storage_error)?; - + upsert_contract_row(&mut tx, &row).await?; tx.commit().await.map_err(to_storage_error)?; log_info!( @@ -547,18 +548,7 @@ impl ManagerStorage for PostgresStore { ) -> Result<(), ddk_manager::error::Error> { let mut tx = self.pool.begin().await.map_err(to_storage_error)?; let id = hex::encode(id); - sqlx::query("DELETE FROM contract_data WHERE id = $1") - .bind(id.clone()) - .execute(&mut *tx) - .await - .map_err(to_storage_error)?; - - sqlx::query("DELETE FROM contract_metadata WHERE id = $1") - .bind(id) - .execute(&mut *tx) - .await - .map_err(to_storage_error)?; - + delete_contract_rows(&mut tx, &id).await?; tx.commit().await.map_err(to_storage_error)?; Ok(()) @@ -570,99 +560,28 @@ impl ManagerStorage for PostgresStore { "Updating contract. id={}", hex::encode(contract.get_id()) ); - let prefix = ContractPrefix::get_prefix(contract); - let contract_id = hex::encode(contract.get_id()); - let (offer_collateral, accept_collateral, total_collateral) = contract.get_collateral(); + let row = ContractRow::from_contract(contract)?; - // Start a transaction let mut tx = self.pool.begin().await.map_err(to_storage_error)?; - // Step 1: Remove by temp_id if Accepted or Signed + // The offered row is keyed by the temporary id. Once the contract has + // its real id, that row goes. match contract { - a @ Contract::Accepted(_) | a @ Contract::Signed(_) => { + Contract::Accepted(_) | Contract::Signed(_) => { log_info!( self.logger, "Deleting contract by temp_id. tmp_id={}", - hex::encode(a.get_temporary_id()) + row.temporary_id ); - let temp_id = hex::encode(a.get_temporary_id()); - sqlx::query("DELETE FROM contract_data WHERE id = $1") - .bind(temp_id.clone()) - .execute(&mut *tx) - .await - .map_err(to_storage_error)?; - sqlx::query("DELETE FROM contract_metadata WHERE id = $1") - .bind(temp_id) - .execute(&mut *tx) - .await - .map_err(to_storage_error)?; + delete_contract_rows(&mut tx, &row.temporary_id).await?; } _ => {} } - let funding_txid = contract.get_funding_txid().map(|txid| txid.to_string()); - let cet_txid = contract.get_cet_txid().map(|txid| txid.to_string()); - let oracle_pubkey = contract - .get_oracle_announcement() - .map(|ann| ann.oracle_public_key.to_string()); - let announcement_id = contract - .get_oracle_announcement() - .map(|ann| ann.oracle_event.event_id.clone()); - - // A single atomic upsert: the read-modify-write it replaces raced under - // concurrent updates, and its insert arm hardcoded is_offer_party and - // fee_rate_per_vb. The update arm deliberately leaves the columns set - // at insert time untouched and only advances the mutable ones. - sqlx::query( - r#" - INSERT INTO contract_metadata ( - id, state, is_offer_party, counter_party, - offer_collateral, accept_collateral, total_collateral, fee_rate_per_vb, - cet_locktime, refund_locktime, pnl, funding_txid, cet_txid, announcement_id, oracle_pubkey - ) - VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15) - ON CONFLICT (id) DO UPDATE SET - state = EXCLUDED.state, - pnl = EXCLUDED.pnl, - funding_txid = COALESCE(EXCLUDED.funding_txid, contract_metadata.funding_txid), - cet_txid = COALESCE(EXCLUDED.cet_txid, contract_metadata.cet_txid) - "#, - ) - .bind(&contract_id) - .bind(prefix as i16) - .bind(contract.is_offer_party()) - .bind(hex::encode(contract.get_counter_party_id().serialize())) - .bind(offer_collateral.to_sat() as i64) - .bind(accept_collateral.to_sat() as i64) - .bind(total_collateral.to_sat() as i64) - .bind(contract.get_fee_rate_per_vb() as i64) - .bind(contract.get_cet_locktime() as i32) - .bind(contract.get_refund_locktime() as i32) - .bind(Some(contract.get_pnl().to_sat())) - .bind(&funding_txid) - .bind(&cet_txid) - .bind(announcement_id.unwrap_or_else(|| "legacy_data".to_string())) - .bind(oracle_pubkey.unwrap_or_else(|| "legacy_data".to_string())) - .execute(&mut *tx) - .await - .map_err(to_storage_error)?; - - let serialized_contract = serialize_contract(contract)?; - - sqlx::query( - "INSERT INTO contract_data (id, state, contract_data, is_compressed) - VALUES ($1, $2, $3, $4) - ON CONFLICT (id) DO UPDATE SET - state = EXCLUDED.state, - contract_data = EXCLUDED.contract_data", - ) - .bind(&contract_id) - .bind(prefix as i16) - .bind(&serialized_contract) - .bind(false) - .execute(&mut *tx) - .await - .map_err(to_storage_error)?; + upsert_contract_row(&mut tx, &row).await?; + // A contract read from the legacy blob layout moves over on its + // first update. + legacy::delete_legacy_rows(&mut tx, &row.id).await?; tx.commit().await.map_err(to_storage_error)?; @@ -671,89 +590,182 @@ impl ManagerStorage for PostgresStore { #[tracing::instrument(skip(self))] async fn get_signed_contracts(&self) -> Result, ddk_manager::error::Error> { - let contracts = - sqlx::query_as::("SELECT * FROM contract_data WHERE state = 3") - .fetch_all(&self.pool) - .await - .map_err(to_storage_error)?; - - let signed = contracts + self.contracts_in_state(ContractPrefix::Signed) + .await? .into_iter() - .map(|c| match deserialize_contract(&c.contract_data)? { + .map(|c| match c { Contract::Signed(s) => Ok(s), _ => Err(wrong_state_error("signed")), }) - .collect::, ddk_manager::error::Error>>()?; - - Ok(signed) + .collect() } #[tracing::instrument(skip(self))] async fn get_contract_offers(&self) -> Result, ddk_manager::error::Error> { - let contracts = sqlx::query_as::( - "SELECT cd.id, cd.state, cd.contract_data, cd.is_compressed - FROM contract_data cd - INNER JOIN contract_metadata cm ON cd.id = cm.id - WHERE cm.state = 1 AND cm.is_offer_party = false", - ) - .fetch_all(&self.pool) - .await - .map_err(to_storage_error)?; - - let offers = contracts + self.contracts_in_state(ContractPrefix::Offered) + .await? .into_iter() - .map(|c| match deserialize_contract(&c.contract_data)? { + .filter(|c| !c.is_offer_party()) + .map(|c| match c { Contract::Offered(o) => Ok(o), _ => Err(wrong_state_error("offered")), }) - .collect::, ddk_manager::error::Error>>()?; - - Ok(offers) + .collect() } #[tracing::instrument(skip(self))] async fn get_confirmed_contracts( &self, ) -> Result, ddk_manager::error::Error> { - let contracts = - sqlx::query_as::("SELECT * FROM contract_data WHERE state = 4") - .fetch_all(&self.pool) - .await - .map_err(to_storage_error)?; - - let signed = contracts + self.contracts_in_state(ContractPrefix::Confirmed) + .await? .into_iter() - .map(|c| match deserialize_contract(&c.contract_data)? { + .map(|c| match c { Contract::Confirmed(s) => Ok(s), _ => Err(wrong_state_error("confirmed")), }) - .collect::, ddk_manager::error::Error>>()?; - - Ok(signed) + .collect() } #[tracing::instrument(skip(self))] async fn get_preclosed_contracts( &self, ) -> Result, ddk_manager::error::Error> { - let contracts = - sqlx::query_as::("SELECT * FROM contract_data WHERE state = 5") - .fetch_all(&self.pool) - .await - .map_err(to_storage_error)?; - - let preclosed = contracts + self.contracts_in_state(ContractPrefix::PreClosed) + .await? .into_iter() - .map(|c| match deserialize_contract(&c.contract_data)? { + .map(|c| match c { Contract::PreClosed(p) => Ok(p), _ => Err(wrong_state_error("pre-closed")), }) - .collect::, ddk_manager::error::Error>>()?; - - Ok(preclosed) + .collect() } } +/// The metadata columns, from `dlc_contracts` and from the legacy +/// `contract_metadata` rows that have not been migrated yet. +const CONTRACT_METADATA: &str = "SELECT id, state, is_offer_party, counter_party, + offer_collateral, accept_collateral, total_collateral, fee_rate_per_vb, + cet_locktime, refund_locktime, pnl, funding_txid, cet_txid, announcement_id, oracle_pubkey + FROM dlc_contracts + UNION ALL + SELECT id, state, is_offer_party, counter_party, + offer_collateral, accept_collateral, total_collateral, fee_rate_per_vb, + cet_locktime, refund_locktime, pnl, funding_txid, cet_txid, announcement_id, oracle_pubkey + FROM contract_metadata m + WHERE NOT EXISTS (SELECT 1 FROM dlc_contracts d WHERE d.id = m.id)"; + +/// Writes a contract row, replacing the row with the same id. +pub(super) async fn upsert_contract_row( + tx: &mut Transaction<'_, Postgres>, + row: &ContractRow, +) -> Result<(), ddk_manager::error::Error> { + sqlx::query( + r#" + INSERT INTO dlc_contracts ( + id, format_version, state, temporary_id, is_offer_party, counter_party, + keys_id, contract_flags, chain_hash, + offer_collateral, accept_collateral, total_collateral, fee_rate_per_vb, + cet_locktime, refund_locktime, announcement_id, oracle_pubkey, + funding_txid, cet_txid, pnl, + offer_message, accept_message, sign_message, + offer_params, accept_params, adaptor_infos, dlc_transactions, + channel_id, attestations, signed_cet, error_message + ) + VALUES ( + $1, $2, $3, $4, $5, $6, + $7, $8, $9, + $10, $11, $12, $13, + $14, $15, $16, $17, + $18, $19, $20, + $21, $22, $23, + $24, $25, $26, $27, + $28, $29, $30, $31 + ) + ON CONFLICT (id) DO UPDATE SET + format_version = EXCLUDED.format_version, + state = EXCLUDED.state, + temporary_id = EXCLUDED.temporary_id, + is_offer_party = EXCLUDED.is_offer_party, + counter_party = EXCLUDED.counter_party, + keys_id = EXCLUDED.keys_id, + contract_flags = EXCLUDED.contract_flags, + chain_hash = EXCLUDED.chain_hash, + offer_collateral = EXCLUDED.offer_collateral, + accept_collateral = EXCLUDED.accept_collateral, + total_collateral = EXCLUDED.total_collateral, + fee_rate_per_vb = EXCLUDED.fee_rate_per_vb, + cet_locktime = EXCLUDED.cet_locktime, + refund_locktime = EXCLUDED.refund_locktime, + announcement_id = EXCLUDED.announcement_id, + oracle_pubkey = EXCLUDED.oracle_pubkey, + funding_txid = EXCLUDED.funding_txid, + cet_txid = EXCLUDED.cet_txid, + pnl = EXCLUDED.pnl, + offer_message = EXCLUDED.offer_message, + accept_message = EXCLUDED.accept_message, + sign_message = EXCLUDED.sign_message, + offer_params = EXCLUDED.offer_params, + accept_params = EXCLUDED.accept_params, + adaptor_infos = EXCLUDED.adaptor_infos, + dlc_transactions = EXCLUDED.dlc_transactions, + channel_id = EXCLUDED.channel_id, + attestations = EXCLUDED.attestations, + signed_cet = EXCLUDED.signed_cet, + error_message = EXCLUDED.error_message, + updated_at = now() + "#, + ) + .bind(&row.id) + .bind(row.format_version) + .bind(row.state) + .bind(&row.temporary_id) + .bind(row.is_offer_party) + .bind(&row.counter_party) + .bind(&row.keys_id) + .bind(row.contract_flags) + .bind(&row.chain_hash) + .bind(row.offer_collateral) + .bind(row.accept_collateral) + .bind(row.total_collateral) + .bind(row.fee_rate_per_vb) + .bind(row.cet_locktime) + .bind(row.refund_locktime) + .bind(&row.announcement_id) + .bind(&row.oracle_pubkey) + .bind(&row.funding_txid) + .bind(&row.cet_txid) + .bind(row.pnl) + .bind(&row.offer_message) + .bind(&row.accept_message) + .bind(&row.sign_message) + .bind(&row.offer_params) + .bind(&row.accept_params) + .bind(&row.adaptor_infos) + .bind(&row.dlc_transactions) + .bind(&row.channel_id) + .bind(&row.attestations) + .bind(&row.signed_cet) + .bind(&row.error_message) + .execute(&mut **tx) + .await + .map_err(to_storage_error)?; + Ok(()) +} + +/// Deletes a contract from `dlc_contracts` and from the legacy tables. +async fn delete_contract_rows( + tx: &mut Transaction<'_, Postgres>, + id: &str, +) -> Result<(), ddk_manager::error::Error> { + sqlx::query("DELETE FROM dlc_contracts WHERE id = $1") + .bind(id) + .execute(&mut **tx) + .await + .map_err(to_storage_error)?; + legacy::delete_legacy_rows(tx, id).await +} + /// Insert keychain descriptors. #[tracing::instrument(skip_all)] async fn insert_descriptor( @@ -1229,7 +1241,7 @@ mod tests { .unwrap(); let offered = include_bytes!("../../../../testconfig/contract_binaries/Offered"); - let offered_contract = deserialize_contract(&offered.to_vec()).unwrap(); + let offered_contract = deserialize_contract(offered).unwrap(); match offered_contract { Contract::Offered(offered_contract) => { store @@ -1240,32 +1252,32 @@ mod tests { _ => panic!("Offered contract is not an OfferedContract"), } let accept = include_bytes!("../../../../testconfig/contract_binaries/Accepted"); - let accepted_contract = deserialize_contract(&accept.to_vec()).unwrap(); + let accepted_contract = deserialize_contract(accept).unwrap(); store .update_contract(&accepted_contract) .await .expect("Failed to update accepted contract"); let signed = include_bytes!("../../../../testconfig/contract_binaries/Signed"); - let signed_contract = deserialize_contract(&signed.to_vec()).unwrap(); + let signed_contract = deserialize_contract(signed).unwrap(); store .update_contract(&signed_contract) .await .expect("Failed to update signed contract"); let confirmed = include_bytes!("../../../../testconfig/contract_binaries/Confirmed"); - let confirmed_contract = deserialize_contract(&confirmed.to_vec()).unwrap(); + let confirmed_contract = deserialize_contract(confirmed).unwrap(); store .update_contract(&confirmed_contract) .await .expect("Failed to update confirmed contract"); let preclosed = include_bytes!("../../../../testconfig/contract_binaries/PreClosed"); - let preclosed_contract = deserialize_contract(&preclosed.to_vec()).unwrap(); + let preclosed_contract = deserialize_contract(preclosed).unwrap(); store .update_contract(&preclosed_contract) .await .expect("Failed to update preclosed contract"); let closed = include_bytes!("../../../../testconfig/contract_binaries/Closed"); - let closed_contract = deserialize_contract(&closed.to_vec()).unwrap(); + let closed_contract = deserialize_contract(closed).unwrap(); store .update_contract(&closed_contract) .await @@ -1291,7 +1303,7 @@ mod tests { assert_eq!(confirmed_rows.len(), 1); assert_eq!(confirmed_rows[0].state, ContractPrefix::Closed as i16); let contracts = db.get_contracts().await.unwrap(); - assert!(contracts.len() > 0); + assert!(!contracts.is_empty()); } #[tokio::test] @@ -1303,9 +1315,11 @@ mod tests { .unwrap(); let did = descriptor.descriptor_id(); - let mut changeset = ChangeSet::default(); - changeset.network = Some(Network::Regtest); - changeset.descriptor = Some(descriptor); + let mut changeset = ChangeSet { + network: Some(Network::Regtest), + descriptor: Some(descriptor), + ..Default::default() + }; changeset.indexer.last_revealed.insert(did, 7); db.write(&changeset).await.unwrap(); @@ -1407,8 +1421,10 @@ mod tests { vout: 0, }; - let mut lock = ChangeSet::default(); - lock.network = Some(Network::Regtest); + let mut lock = ChangeSet { + network: Some(Network::Regtest), + ..Default::default() + }; lock.locked_outpoints.outpoints.insert(outpoint, true); db.write(&lock).await.unwrap(); let read = db.read().await.unwrap(); @@ -1431,9 +1447,11 @@ mod tests { .unwrap(); let did = descriptor.descriptor_id(); - let mut changeset = ChangeSet::default(); - changeset.network = Some(Network::Regtest); - changeset.descriptor = Some(descriptor.clone()); + let mut changeset = ChangeSet { + network: Some(Network::Regtest), + descriptor: Some(descriptor.clone()), + ..Default::default() + }; changeset.indexer.last_revealed.insert(did, 4); db.write(&changeset).await.unwrap(); @@ -1443,16 +1461,18 @@ mod tests { // A keychain row written without a revealed index must read back as // "nothing revealed", not index 0. - let mut fresh = ChangeSet::default(); - fresh.change_descriptor = Some( - "wpkh([73c5da0a/84'/1'/0']tpubDC8msFGeGuwnKG9Upg7DM2b4DaRqg3CUZa5g8v2SRQ6K4NSkxUgd7HsL2XVWbVm39yBA4LAxysQAm397zwQSQoQgewGiYZqrA9DsP4zbQ1M/1/*)" - .parse() - .unwrap(), - ); + let fresh = ChangeSet { + change_descriptor: Some( + "wpkh([73c5da0a/84'/1'/0']tpubDC8msFGeGuwnKG9Upg7DM2b4DaRqg3CUZa5g8v2SRQ6K4NSkxUgd7HsL2XVWbVm39yBA4LAxysQAm397zwQSQoQgewGiYZqrA9DsP4zbQ1M/1/*)" + .parse() + .unwrap(), + ), + ..Default::default() + }; db.write(&fresh).await.unwrap(); let read = db.read().await.unwrap(); let fresh_did = fresh.change_descriptor.as_ref().unwrap().descriptor_id(); - assert!(read.indexer.last_revealed.get(&fresh_did).is_none()); + assert!(!read.indexer.last_revealed.contains_key(&fresh_did)); let read = db.read().await.unwrap(); assert_eq!(read.network, Some(Network::Regtest)); @@ -1467,8 +1487,10 @@ mod tests { let hash_a = BlockHash::from_byte_array([0xAA; 32]); let hash_b = BlockHash::from_byte_array([0xBB; 32]); - let mut changeset = ChangeSet::default(); - changeset.network = Some(Network::Regtest); + let mut changeset = ChangeSet { + network: Some(Network::Regtest), + ..Default::default() + }; changeset.local_chain.blocks.insert(100, Some(hash_a)); db.write(&changeset).await.unwrap(); @@ -1502,7 +1524,7 @@ mod tests { db.write(&remove).await.unwrap(); let read = db.read().await.unwrap(); - assert!(read.local_chain.blocks.get(&100).is_none()); + assert!(!read.local_chain.blocks.contains_key(&100)); } #[tokio::test] @@ -1513,7 +1535,7 @@ mod tests { // delete (the Accepted transition); it must carry the contract's real // values instead of hardcoded ones. let accept = include_bytes!("../../../../testconfig/contract_binaries/Accepted"); - let accepted_contract = deserialize_contract(&accept.to_vec()).unwrap(); + let accepted_contract = deserialize_contract(accept).unwrap(); let metadata = db.get_contract_metadata(None).await.unwrap(); assert_eq!(metadata.len(), 1); @@ -1534,8 +1556,10 @@ mod tests { let txid = dummy_tx().compute_txid(); // No tx row exists yet for this txid; the value must not be dropped. - let mut changeset = ChangeSet::default(); - changeset.network = Some(Network::Regtest); + let mut changeset = ChangeSet { + network: Some(Network::Regtest), + ..Default::default() + }; changeset.tx_graph.last_seen.insert(txid, 100); db.write(&changeset).await.unwrap(); @@ -1559,8 +1583,10 @@ mod tests { let did = DescriptorId(sha256::Hash::from_byte_array([0x11; 32])); let script = ScriptBuf::from(vec![0x00, 0x14]); - let mut changeset = ChangeSet::default(); - changeset.network = Some(Network::Regtest); + let mut changeset = ChangeSet { + network: Some(Network::Regtest), + ..Default::default() + }; changeset.tx_graph.txs.insert(Arc::new(tx)); changeset.tx_graph.first_seen.insert(txid, 100); changeset.tx_graph.last_evicted.insert(txid, 200); @@ -1598,6 +1624,306 @@ mod tests { assert_eq!(read.tx_graph.last_evicted.get(&txid), Some(&250)); } + /// Writes a contract the way releases up to 2.0 did: a blob in + /// contract_data and a metadata row in contract_metadata. + async fn seed_legacy(pool: &Pool, contract: &Contract) { + let id = hex::encode(contract.get_id()); + let state = ContractPrefix::get_prefix(contract) as i16; + let (offer, accept, total) = contract.get_collateral(); + sqlx::query( + "INSERT INTO contract_metadata ( + id, state, is_offer_party, counter_party, offer_collateral, accept_collateral, + total_collateral, fee_rate_per_vb, cet_locktime, refund_locktime, pnl + ) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11)", + ) + .bind(&id) + .bind(state) + .bind(contract.is_offer_party()) + .bind(hex::encode(contract.get_counter_party_id().serialize())) + .bind(offer.to_sat() as i64) + .bind(accept.to_sat() as i64) + .bind(total.to_sat() as i64) + .bind(contract.get_fee_rate_per_vb() as i64) + .bind(contract.get_cet_locktime() as i32) + .bind(contract.get_refund_locktime() as i32) + .bind(Some(contract.get_pnl().to_sat())) + .execute(pool) + .await + .unwrap(); + sqlx::query( + "INSERT INTO contract_data (id, state, contract_data, is_compressed) + VALUES ($1, $2, $3, false)", + ) + .bind(&id) + .bind(state) + .bind(crate::util::ser::serialize_contract(contract).unwrap()) + .execute(pool) + .await + .unwrap(); + } + + fn fixture(name: &str) -> Contract { + let path = format!( + "{}/../testconfig/contract_binaries/{name}", + env!("CARGO_MANIFEST_DIR") + ); + deserialize_contract(&std::fs::read(path).unwrap()).unwrap() + } + + async fn open(server: &TestPostgres, migrations: bool) -> PostgresStore { + PostgresStore::new( + server.url(), + migrations, + Arc::new(Logger::console( + "console_logger".to_string(), + LogLevel::Info, + )), + "test".to_string(), + ) + .await + .unwrap() + } + + fn bytes(contract: &Contract) -> Vec { + crate::util::ser::serialize_contract(contract).unwrap() + } + + /// A database written by 2.0 is moved to the columnar layout when the + /// store opens with migrations on, and every contract survives. + #[tokio::test] + async fn legacy_contracts_migrate_at_startup() { + let server = TestPostgres::start("ddk").await; + let schema = open(&server, true).await; + let offered = fixture("Offered"); + let closed = fixture("Closed"); + seed_legacy(&schema.pool, &offered).await; + seed_legacy(&schema.pool, &closed).await; + assert_eq!(schema.count_legacy_contracts().await.unwrap(), 2); + drop(schema); + + let db = open(&server, true).await; + assert_eq!(db.count_legacy_contracts().await.unwrap(), 0); + + let (rows,): (i64,) = + sqlx::query_as("SELECT COUNT(*) FROM dlc_contracts WHERE format_version = 2") + .fetch_one(&db.pool) + .await + .unwrap(); + assert_eq!(rows, 2); + let (legacy,): (i64,) = sqlx::query_as("SELECT COUNT(*) FROM contract_data") + .fetch_one(&db.pool) + .await + .unwrap(); + assert_eq!(legacy, 0); + + let read = db.get_contract(&offered.get_id()).await.unwrap().unwrap(); + assert_eq!(bytes(&read), bytes(&offered)); + let read = db.get_contract(&closed.get_id()).await.unwrap().unwrap(); + assert_eq!(bytes(&read), bytes(&closed)); + assert_eq!(db.get_contract_metadata(None).await.unwrap().len(), 2); + } + + /// With migrations off, a legacy contract still loads through every read + /// path and moves to the columnar layout on its first update. + #[tokio::test] + async fn legacy_contract_loads_and_moves_on_update() { + let server = TestPostgres::start("ddk").await; + let schema = open(&server, true).await; + let signed = fixture("Signed"); + seed_legacy(&schema.pool, &signed).await; + drop(schema); + + let db = open(&server, false).await; + assert_eq!(db.count_legacy_contracts().await.unwrap(), 1); + let read = db.get_contract(&signed.get_id()).await.unwrap().unwrap(); + assert_eq!(bytes(&read), bytes(&signed)); + assert_eq!(db.get_contracts().await.unwrap().len(), 1); + assert_eq!(db.get_signed_contracts().await.unwrap().len(), 1); + assert_eq!(db.get_contract_metadata(None).await.unwrap().len(), 1); + + db.update_contract(&signed).await.unwrap(); + assert_eq!(db.count_legacy_contracts().await.unwrap(), 0); + assert_eq!(db.get_signed_contracts().await.unwrap().len(), 1); + assert_eq!(db.get_contract_metadata(None).await.unwrap().len(), 1); + let read = db.get_contract(&signed.get_id()).await.unwrap().unwrap(); + assert_eq!(bytes(&read), bytes(&signed)); + + let report = db.migrate_legacy_contracts().await.unwrap(); + assert_eq!(report, LegacyMigrationReport::default()); + } + + /// A blob that cannot be decoded stays where it is, is named in the + /// report, and does not stop the store from opening or the other + /// contracts from moving. + #[tokio::test] + async fn migration_reports_the_rows_it_cannot_move() { + let server = TestPostgres::start("ddk").await; + let schema = open(&server, true).await; + seed_legacy(&schema.pool, &fixture("Confirmed")).await; + sqlx::query( + "INSERT INTO contract_metadata ( + id, state, is_offer_party, counter_party, offer_collateral, accept_collateral, + total_collateral, fee_rate_per_vb, cet_locktime, refund_locktime + ) VALUES ('bad', 3, true, 'aa', 0, 0, 0, 0, 0, 0)", + ) + .execute(&schema.pool) + .await + .unwrap(); + sqlx::query( + "INSERT INTO contract_data (id, state, contract_data, is_compressed) + VALUES ('bad', 3, $1, false)", + ) + .bind(vec![3u8, 1, 2, 3]) + .execute(&schema.pool) + .await + .unwrap(); + drop(schema); + + let db = open(&server, true).await; + assert_eq!(db.count_legacy_contracts().await.unwrap(), 1); + assert_eq!(db.get_confirmed_contracts().await.unwrap().len(), 1); + + let report = db.migrate_legacy_contracts().await.unwrap(); + assert_eq!(report.migrated, 0); + assert_eq!(report.failed.len(), 1); + assert_eq!(report.failed[0].0, "bad"); + assert!(!report.is_complete()); + } + + /// Every blob in a live legacy database survives the row layout byte for + /// byte. Read only: it never writes to the database it is pointed at. + /// + /// ```sh + /// DDK_MIGRATION_CHECK_URL=postgres://... cargo test -p ddk --features postgres \ + /// legacy_blobs_round_trip_against_a_live_database -- --ignored --nocapture + /// ``` + #[tokio::test] + #[ignore = "needs DDK_MIGRATION_CHECK_URL"] + async fn legacy_blobs_round_trip_against_a_live_database() { + let url = std::env::var("DDK_MIGRATION_CHECK_URL").expect("DDK_MIGRATION_CHECK_URL"); + let pool = PoolOptions::::new() + .max_connections(1) + .connect(&url) + .await + .unwrap(); + let rows = sqlx::query_as::( + "SELECT id, state, contract_data, is_compressed FROM contract_data ORDER BY id", + ) + .fetch_all(&pool) + .await + .unwrap(); + assert!(!rows.is_empty(), "no legacy rows to check"); + for legacy in &rows { + let contract = deserialize_contract(&legacy.contract_data) + .unwrap_or_else(|e| panic!("{}: legacy decode: {e}", legacy.id)); + let row = ContractRow::from_contract(&contract) + .unwrap_or_else(|e| panic!("{}: to row: {e}", legacy.id)); + assert_eq!(row.id, legacy.id, "row id differs from the legacy id"); + let read = row + .into_contract() + .unwrap_or_else(|e| panic!("{}: from row: {e}", legacy.id)); + assert_eq!( + bytes(&read), + bytes(&contract), + "{}: bytes differ", + legacy.id + ); + println!("ok {} state={}", legacy.id, legacy.state); + } + println!("{} contracts round-trip byte for byte", rows.len()); + } + + /// A stream holding one record of type 65007 with `body` as its one-byte body. + fn stream_with_record(body: u8) -> ddk_messages::tlv_stream::TlvStream { + let bytes = [0xfd, 0xfd, 0xef, 0x01, body]; + ddk_messages::tlv_stream::TlvStream::read_to_end(&mut lightning::io::Cursor::new(bytes)) + .unwrap() + } + + /// TLV records on a legacy blob come through the migration and back out + /// of the row, on every layer of a signed contract. + #[tokio::test] + async fn tlv_records_survive_the_legacy_migration() { + let server = TestPostgres::start("ddk").await; + let schema = open(&server, true).await; + let Contract::Signed(mut signed) = fixture("Signed") else { + panic!("fixture is not signed") + }; + signed.accepted_contract.offered_contract.tlvs = stream_with_record(1); + signed.accepted_contract.tlvs = stream_with_record(2); + signed.tlvs = stream_with_record(3); + let contract = Contract::Signed(signed); + seed_legacy(&schema.pool, &contract).await; + drop(schema); + + let db = open(&server, true).await; + assert_eq!(db.count_legacy_contracts().await.unwrap(), 0); + let Some(Contract::Signed(read)) = db.get_contract(&contract.get_id()).await.unwrap() + else { + panic!("contract not found or not signed") + }; + assert_eq!( + read.accepted_contract.offered_contract.tlvs, + stream_with_record(1) + ); + assert_eq!(read.accepted_contract.tlvs, stream_with_record(2)); + assert_eq!(read.tlvs, stream_with_record(3)); + assert_eq!(bytes(&Contract::Signed(read)), bytes(&contract)); + } + + /// A contract closed by refund has no CET and no attestations. Both + /// columns are null and the contract still round-trips. + #[tokio::test] + async fn closed_by_refund_round_trips() { + let (_server, db) = seed_db().await; + let Contract::Closed(mut closed) = fixture("Closed") else { + panic!("fixture is not closed") + }; + closed.signed_cet = None; + closed.attestations = None; + let contract = Contract::Closed(closed); + + db.update_contract(&contract).await.unwrap(); + + let (cet, attestations): (Option>, Option>) = + sqlx::query_as("SELECT signed_cet, attestations FROM dlc_contracts WHERE id = $1") + .bind(hex::encode(contract.get_id())) + .fetch_one(&db.pool) + .await + .unwrap(); + assert!(cet.is_none()); + assert!(attestations.is_none()); + + let read = db.get_contract(&contract.get_id()).await.unwrap().unwrap(); + assert_eq!(bytes(&read), bytes(&contract)); + let metadata = db.get_contract_metadata(None).await.unwrap(); + assert_eq!(metadata.len(), 1); + assert!(metadata[0].cet_txid.is_none()); + } + + /// Offers we received are offers to act on; offers we made are not. + #[tokio::test] + async fn contract_offers_are_the_ones_we_received() { + let server = TestPostgres::start("ddk").await; + let db = open(&server, true).await; + let Contract::Offered(mut received) = fixture("Offered") else { + panic!("fixture is not offered") + }; + received.is_offer_party = false; + let mut made = received.clone(); + made.is_offer_party = true; + made.id = [9u8; 32]; + + db.create_contract(&received).await.unwrap(); + db.create_contract(&made).await.unwrap(); + + assert_eq!(db.get_contracts().await.unwrap().len(), 2); + let offers = db.get_contract_offers().await.unwrap(); + assert_eq!(offers.len(), 1); + assert_eq!(offers[0].id, received.id); + assert!(!offers[0].is_offer_party); + } + #[tokio::test] async fn delete_contract_removes_rows() { let (_server, db) = seed_db().await; diff --git a/docs/postgres-contract-migration.md b/docs/postgres-contract-migration.md new file mode 100644 index 00000000..63af0cda --- /dev/null +++ b/docs/postgres-contract-migration.md @@ -0,0 +1,70 @@ +# Postgres contract migration + +Releases up to 2.0 stored each contract as one opaque blob in `contract_data`, +with a copy of a few fields in `contract_metadata`. From this release the +Postgres store keeps one row per contract in `dlc_contracts`, with the offer, +accept, and sign messages in their own columns and the manager-only state in +typed columns. See issue #190 for why. + +## What happens on upgrade + +1. The schema migration `0010_dlc_contracts` creates the new table. It does + not touch the old tables. +2. When the store opens with migrations on, which is how `ddk-node` and + `Builder` open it, it moves every contract from the old tables to the new + table. One transaction per contract, so a contract is always in exactly one + place. +3. The store logs a warning at startup, and on every read that hits the old + tables, while any contract is still in the old layout. +4. Old rows still load. A contract that is read from the old layout moves to + the new one on its next update. + +Nothing is deleted from the old tables except the rows that were moved. The +two old tables are dropped in a later release, after the legacy reader goes. + +## Running the migration by hand + +Take a backup first: + +```sh +pg_dump "$DATABASE_URL" > before-migration.sql +``` + +Then run the migration and exit: + +```sh +ddk-node --postgres-url "$DATABASE_URL" migrate +``` + +The command prints how many contracts it moved and names every contract it +could not move, with the error. It exits non-zero when any contract is left +behind. It is safe to run again; it only selects what is left. + +An application that embeds the store calls the same function: + +```rust +let report = store.migrate_legacy_contracts().await?; +if !report.is_complete() { + for (id, error) in &report.failed { + eprintln!("{id}: {error}"); + } +} +``` + +`PostgresStore::count_legacy_contracts` tells how many contracts are left +without moving any. + +## The row layout + +`format_version` on each row says which layout the row uses. Version 1 is the +blob layout in `contract_data` and never appears in `dlc_contracts`. Version 2 +is the columnar layout. A reader that meets a version it does not know +returns an error instead of misreading the row. + +The one writer is `ContractRow::from_contract` and the one reader is +`ContractRow::into_contract`, both in `ddk/src/storage/postgres/contract_row.rs`. +Every read path of the store goes through them. Adding a field to a stored +contract is a schema migration plus a change to those two functions. + +The TLV streams of the offer, accept, and sign messages travel inside the +stored messages, so they survive on every state without a suffix.