nik/liana
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

poller: save transactions in database

Browse Source 
This includes changes from darosior's commits:
3e7d968508a71ffa740576d6bf9432611e78f4a5
e2225a5110ca981c340eef329182e702b375b55f
This commit is contained in:
jp1ac4 2024-07-08 10:05:51 +01:00
parent ba4c819918
commit afa6a51601
No known key found for this signature in database
GPG Key ID: C61FA2110D7DC407
7 changed files with 389 additions and 217 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

36
src/bitcoin/poller/looper.rs
View File

@ -156,6 +156,40 @@ fn update_coins(
}
}
// Add new deposit and spend transactions to the database.
fn add_txs_to_db(
bit: &impl BitcoinInterface,
db_conn: &mut Box<dyn DatabaseConnection>,
updated_coins: &UpdatedCoins,
) {
let curr_txids: HashSet<_> = db_conn.list_saved_txids().into_iter().collect();
let mut new_txids = HashSet::new();
// First get all newly received coins that have not expired.
new_txids.extend(updated_coins.received.iter().filter_map(|c| {
if !updated_coins.expired.contains(&c.outpoint) {
Some(c.outpoint.txid)
} else {
None
}
}));
// Add spend txid for new & existing coins.
new_txids.extend(updated_coins.spending.iter().map(|(_, txid)| txid));
// Remove those txids we already have.
let missing_txids = new_txids.difference(&curr_txids);
log::debug!("Missing txids: {:?}", missing_txids);
// Now retrieve txs.
let txs: Vec<_> = missing_txids
.map(|txid| bit.wallet_transaction(txid).map(|(tx, _)| tx))
.collect::<Option<Vec<_>>>()
.expect("we must retrieve all txs");
if !txs.is_empty() {
db_conn.new_txs(&txs);
}
}
#[derive(Debug, Clone, Copy)]
enum TipUpdate {
// The best block is still the same as in the previous poll.
@ -233,6 +267,8 @@ fn updates(
return updates(db_conn, bit, descs, secp);
}
// Transactions must be added to the DB before coins due to foreign key constraints.
add_txs_to_db(bit, db_conn, &updated_coins);
// The chain tip did not change since we started our updates. Record them and the latest tip.
// Having the tip in database means that, as far as the chain is concerned, we've got all
// updates up to this block. But not more.

14
src/database/mod.rs
View File

@ -146,6 +146,12 @@ pub trait DatabaseConnection {
/// Retrieve a limited list of txids that where deposited or spent between the start and end timestamps (inclusive bounds)
fn list_txids(&mut self, start: u32, end: u32, limit: u64) -> Vec<bitcoin::Txid>;
/// Retrieves all txids from the transactions table whether or not they are referenced by a coin.
fn list_saved_txids(&mut self) -> Vec<bitcoin::Txid>;
/// Store transactions in database, ignoring any that already exist.
fn new_txs(&mut self, txs: &[bitcoin::Transaction]);
}
impl DatabaseConnection for SqliteConn {
@ -310,6 +316,14 @@ impl DatabaseConnection for SqliteConn {
fn list_txids(&mut self, start: u32, end: u32, limit: u64) -> Vec<bitcoin::Txid> {
self.db_list_txids(start, end, limit)
}
fn list_saved_txids(&mut self) -> Vec<bitcoin::Txid> {
self.db_list_saved_txids()
}
fn new_txs<'a>(&mut self, txs: &[bitcoin::Transaction]) {
self.new_txs(txs)
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]

406
src/database/sqlite/mod.rs
View File

File diff suppressed because one or more lines are too long

13
src/database/sqlite/schema.rs
View File

@ -58,6 +58,12 @@ CREATE TABLE coins (
is_immature BOOLEAN NOT NULL CHECK (is_immature IN (0,1)),
UNIQUE (txid, vout),
FOREIGN KEY (wallet_id) REFERENCES wallets (id)
ON UPDATE RESTRICT
ON DELETE RESTRICT,
FOREIGN KEY (txid) REFERENCES transactions (txid)
ON UPDATE RESTRICT
ON DELETE RESTRICT,
FOREIGN KEY (spend_txid) REFERENCES transactions (txid)
ON UPDATE RESTRICT
ON DELETE RESTRICT
);
@ -71,6 +77,13 @@ CREATE TABLE addresses (
derivation_index INTEGER NOT NULL UNIQUE
);
/* Transactions for all wallets. */
CREATE TABLE transactions (
id INTEGER PRIMARY KEY NOT NULL,
txid BLOB UNIQUE NOT NULL,
tx BLOB UNIQUE NOT NULL
);
/* Transactions we created that spend some of our coins. */
CREATE TABLE spend_transactions (
id INTEGER PRIMARY KEY NOT NULL,

86
src/database/sqlite/utils.rs
View File

@ -2,7 +2,7 @@ use crate::database::sqlite::{FreshDbOptions, SqliteDbError, DB_VERSION};
use std::{convert::TryInto, fs, path, time};
use miniscript::bitcoin::secp256k1;
use miniscript::bitcoin::{self, secp256k1};
pub const LOOK_AHEAD_LIMIT: u32 = 200;
@ -229,9 +229,80 @@ fn migrate_v3_to_v4(conn: &mut rusqlite::Connection) -> Result<(), SqliteDbError
Ok(())
}
fn migrate_v4_to_v5(
conn: &mut rusqlite::Connection,
bitcoin_txs: &[bitcoin::Transaction],
) -> Result<(), SqliteDbError> {
db_exec(conn, |db_tx| {
db_tx.execute(
"
CREATE TABLE transactions (
id INTEGER PRIMARY KEY NOT NULL,
txid BLOB UNIQUE NOT NULL,
tx BLOB UNIQUE NOT NULL
);",
rusqlite::params![],
)?;
for bitcoin_tx in bitcoin_txs {
let txid = &bitcoin_tx.txid()[..].to_vec();
let bitcoin_tx_ser = bitcoin::consensus::serialize(bitcoin_tx);
db_tx.execute(
"INSERT INTO transactions (txid, tx) VALUES (?1, ?2);",
rusqlite::params![txid, bitcoin_tx_ser,],
)?;
}
// Create new coins table with foreign key constraints on transactions table.
db_tx.execute_batch(
"
CREATE TABLE coins_new (
id INTEGER PRIMARY KEY NOT NULL,
wallet_id INTEGER NOT NULL,
blockheight INTEGER,
blocktime INTEGER,
txid BLOB NOT NULL,
vout INTEGER NOT NULL,
amount_sat INTEGER NOT NULL,
derivation_index INTEGER NOT NULL,
is_change BOOLEAN NOT NULL CHECK (is_change IN (0,1)),
spend_txid BLOB,
spend_block_height INTEGER,
spend_block_time INTEGER,
is_immature BOOLEAN NOT NULL CHECK (is_immature IN (0,1)),
UNIQUE (txid, vout),
FOREIGN KEY (wallet_id) REFERENCES wallets (id)
ON UPDATE RESTRICT
ON DELETE RESTRICT,
FOREIGN KEY (txid) REFERENCES transactions (txid)
ON UPDATE RESTRICT
ON DELETE RESTRICT,
FOREIGN KEY (spend_txid) REFERENCES transactions (txid)
ON UPDATE RESTRICT
ON DELETE RESTRICT
);
INSERT INTO coins_new SELECT * FROM coins;
DROP TABLE coins;
ALTER TABLE coins_new RENAME TO coins;
UPDATE version SET version = 5;",
)
})?;
Ok(())
}
/// Check the database version and if necessary apply the migrations to upgrade it to the current
/// one.
pub fn maybe_apply_migration(db_path: &path::Path) -> Result<(), SqliteDbError> {
/// one. The `bitcoin_txs` parameter is here for the migration from versions 4 and earlier, which
/// did not store the Bitcoin transactions in database, to versions 5 and later, which do. For a
/// migration from v4 or earlier to v5 or later it is assumed the caller passes *all* necessary
/// transactions, otherwise the migration will fail.
pub fn maybe_apply_migration(
db_path: &path::Path,
bitcoin_txs: &[bitcoin::Transaction],
) -> Result<(), SqliteDbError> {
let mut conn = rusqlite::Connection::open(db_path)?;
// Iteratively apply the database migrations necessary.
@ -262,6 +333,15 @@ pub fn maybe_apply_migration(db_path: &path::Path) -> Result<(), SqliteDbError>
migrate_v3_to_v4(&mut conn)?;
log::warn!("Migration from database version 3 to version 4 successful.");
}
4 => {
log::warn!("Upgrading database from version 4 to version 5.");
log::warn!(
"Number of bitcoin transactions to be inserted: {}.",
bitcoin_txs.len()
);
migrate_v4_to_v5(&mut conn, bitcoin_txs)?;
log::warn!("Migration from database version 4 to version 5 successful.");
}
_ => return Err(SqliteDbError::UnsupportedVersion(version)),
}
}

39
src/lib.rs
View File

@ -23,13 +23,13 @@ use crate::{
bitcoin::{poller, BitcoinInterface},
config::Config,
database::{
sqlite::{FreshDbOptions, SqliteDb, SqliteDbError},
sqlite::{FreshDbOptions, SqliteDb, SqliteDbError, MAX_DB_VERSION_NO_TX_DB},
DatabaseInterface,
},
};
use std::{
error, fmt, fs, io, path,
collections, error, fmt, fs, io, path,
sync::{self, mpsc},
thread,
};
@ -92,6 +92,7 @@ pub enum StartupError {
DefaultDataDirNotFound,
DatadirCreation(path::PathBuf, io::Error),
MissingBitcoindConfig,
DbMigrateBitcoinTxs(&'static str),
Database(SqliteDbError),
Bitcoind(BitcoindError),
#[cfg(unix)]
@ -116,6 +117,10 @@ impl fmt::Display for StartupError {
f,
"Our Bitcoin interface is bitcoind but we have no 'bitcoind_config' entry in the configuration."
),
Self::DbMigrateBitcoinTxs(msg) => write!(
f,
"Error when migrating Bitcoin transaction from Bitcoin backend to database: {}.", msg
),
Self::Database(e) => write!(f, "Error initializing database: '{}'.", e),
Self::Bitcoind(e) => write!(f, "Error setting up bitcoind interface: '{}'.", e),
#[cfg(unix)]
@ -184,6 +189,7 @@ fn setup_sqlite(
data_dir: &path::Path,
fresh_data_dir: bool,
secp: &secp256k1::Secp256k1<secp256k1::VerifyOnly>,
bitcoind: &Option<BitcoinD>,
) -> Result<SqliteDb, StartupError> {
let db_path: path::PathBuf = [data_dir, path::Path::new("lianad.sqlite3")]
.iter()
@ -197,8 +203,34 @@ fn setup_sqlite(
None
};
// If opening an existing wallet whose database does not yet store the wallet transactions,
// query them from the Bitcoin backend before proceeding to the migration.
let sqlite = SqliteDb::new(db_path, options, secp)?;
sqlite.maybe_apply_migrations()?;
if !fresh_data_dir {
let mut conn = sqlite.connection()?;
let wallet_txs = if conn.db_version() <= MAX_DB_VERSION_NO_TX_DB {
let bit = bitcoind.as_ref().ok_or(StartupError::DbMigrateBitcoinTxs(
"a connection to a Bitcoin backend is required",
))?;
let coins = conn.db_coins(&[]);
let coins_txids = coins
.iter()
.map(|c| c.outpoint.txid)
.chain(coins.iter().filter_map(|c| c.spend_txid))
.collect::<collections::HashSet<_>>();
coins_txids
.into_iter()
.map(|txid| bit.get_transaction(&txid).map(|res| res.tx))
.collect::<Option<Vec<_>>>()
.ok_or(StartupError::DbMigrateBitcoinTxs(
"missing transaction in Bitcoin backend",
))?
} else {
Vec::new()
};
sqlite.maybe_apply_migrations(&wallet_txs)?;
}
sqlite.sanity_check(config.bitcoin_config.network, &config.main_descriptor)?;
log::info!("Database initialized and checked.");
@ -355,6 +387,7 @@ impl DaemonHandle {
&data_dir,
fresh_data_dir,
&secp,
&bitcoind,
)?)) as sync::Arc<sync::Mutex<dyn DatabaseInterface>>,
};

12
src/testutils.rs
View File

@ -138,6 +138,7 @@ struct DummyDbState {
change_index: bip32::ChildNumber,
curr_tip: Option<BlockChainTip>,
coins: HashMap<bitcoin::OutPoint, Coin>,
txs: HashMap<bitcoin::Txid, bitcoin::Transaction>,
spend_txs: HashMap<bitcoin::Txid, (Psbt, Option<u32>)>,
timestamp: u32,
}
@ -169,6 +170,7 @@ impl DummyDatabase {
change_index: 0.into(),
curr_tip: None,
coins: HashMap::new(),
txs: HashMap::new(),
spend_txs: HashMap::new(),
timestamp: now,
})),
@ -437,6 +439,16 @@ impl DatabaseConnection for DummyDatabase {
txids_and_time.truncate(limit as usize);
txids_and_time.into_iter().map(|(txid, _)| txid).collect()
}
fn list_saved_txids(&mut self) -> Vec<bitcoin::Txid> {
self.db.read().unwrap().txs.keys().cloned().collect()
}
fn new_txs(&mut self, txs: &[bitcoin::Transaction]) {
for tx in txs {
self.db.write().unwrap().txs.insert(tx.txid(), tx.clone());
}
}
}
pub struct DummyLiana {