diff --git a/logos_sql/src/applier.rs b/logos_sql/src/applier.rs index 4851b68d3..420c0e281 100644 --- a/logos_sql/src/applier.rs +++ b/logos_sql/src/applier.rs @@ -221,7 +221,10 @@ mod tests { use super::on_event; use crate::{ db::Databases, - protocol::{ChannelInscription, EncodedWrite, PAYLOAD_MARKER, Statement, Transaction}, + protocol::{ + CapturedFunctionCalls, ChannelInscription, EncodedWrite, PAYLOAD_MARKER, Statement, + Transaction, + }, }; const CHANNEL_ID: [u8; 32] = [9; 32]; @@ -278,7 +281,8 @@ mod tests { } fn encoded_write(transaction: &Transaction) -> EncodedWrite { - EncodedWrite::new(transaction).expect("payload should encode") + EncodedWrite::new(transaction, CapturedFunctionCalls::empty()) + .expect("payload should encode") } fn item_count(path: &std::path::Path) -> i64 { @@ -431,26 +435,28 @@ mod tests { let live_path = db.live_path().to_owned(); let setup = transaction("CREATE TABLE items(value INTEGER NOT NULL)", Vec::new()); - let setup_encoded = EncodedWrite::new(&setup).expect("setup write should encode"); - - db.commit_local_write(&setup, &setup_encoded) + let setup_tx_id = db + .commit_local_write(&setup) .expect("setup write should commit"); - db.mark_publish_complete(setup_encoded.tx_id) + db.mark_publish_complete(setup_tx_id) .expect("setup publish should be complete"); let insert = transaction( "INSERT INTO items(value) VALUES (?1)", vec![Value::Integer(1)], ); - let insert_encoded = EncodedWrite::new(&insert).expect("insert write should encode"); - - db.commit_local_write(&insert, &insert_encoded) + db.commit_local_write(&insert) .expect("insert write should commit"); + let insert_payload = db + .pending_publish() + .expect("pending write should load") + .expect("pending write should exist") + .payload; let event = blocks_processed( checkpoint(2, 2), vec![ChannelUpdateTx::Inscription(inscription( - &insert_encoded.payload, + &insert_payload, 2, ))], Vec::new(), @@ -567,6 +573,7 @@ mod tests { let conflicting = ChannelInscription { tx_id: first.tx_id, transaction: transaction("CREATE TABLE conflicting_write(value INTEGER)", Vec::new()), + captured_function_calls: CapturedFunctionCalls::empty(), } .encode() .expect("conflicting write should encode"); @@ -618,7 +625,7 @@ mod tests { .payload; let version_offset = PAYLOAD_MARKER.len(); unsupported[version_offset..version_offset + size_of::()] - .copy_from_slice(&2u16.to_le_bytes()); + .copy_from_slice(&3u16.to_le_bytes()); let following = encoded_write(&transaction( "CREATE TABLE following_write(value INTEGER)", @@ -659,11 +666,15 @@ mod tests { let live_path = db.live_path().to_owned(); let local = transaction("CREATE TABLE local_write(value INTEGER)", Vec::new()); - let local_encoded = EncodedWrite::new(&local).expect("local write should encode"); - - db.commit_local_write(&local, &local_encoded) + let local_tx_id = db + .commit_local_write(&local) .expect("local write should commit"); - db.mark_publish_complete(local_encoded.tx_id) + let local_payload = db + .pending_publish() + .expect("pending write should load") + .expect("pending write should exist") + .payload; + db.mark_publish_complete(local_tx_id) .expect("local publish should be complete"); let adopted = encoded_write(&transaction( @@ -676,10 +687,7 @@ mod tests { &adopted.payload, 2, ))], - vec![ChannelUpdateTx::Inscription(inscription( - &local_encoded.payload, - 1, - ))], + vec![ChannelUpdateTx::Inscription(inscription(&local_payload, 1))], Vec::new(), ); diff --git a/logos_sql/src/db.rs b/logos_sql/src/db.rs index 44764cc88..cf9a44928 100644 --- a/logos_sql/src/db.rs +++ b/logos_sql/src/db.rs @@ -2,6 +2,7 @@ use std::{ fs, + ops::Deref, path::{Path, PathBuf}, time::Duration, }; @@ -16,6 +17,7 @@ use rusqlite::{ use crate::{ error::Error, + functions::FunctionOverrides, protocol::{ChannelInscription, EncodedWrite, Transaction, TxId}, }; @@ -40,7 +42,7 @@ const PENDING_WRITE_SCHEMA: &str = " const APPLIED_WRITE_SCHEMA: &str = " CREATE TABLE IF NOT EXISTS __logos_sql_applied_writes ( tx_id BLOB PRIMARY KEY CHECK (length(tx_id) = 32), - transaction_digest BLOB NOT NULL CHECK (length(transaction_digest) = 32) + content_digest BLOB NOT NULL CHECK (length(content_digest) = 32) ) STRICT; "; @@ -81,6 +83,7 @@ const INSERT_REJECTED_WRITE: &str = " VALUES (?1, ?2, ?3) ON CONFLICT (this_msg) DO NOTHING "; + const INSERT_PENDING_WRITE: &str = " INSERT INTO __logos_sql_pending_write (singleton, tx_id, payload) VALUES (1, ?1, ?2) @@ -98,13 +101,13 @@ const MARK_PUBLISH_COMPLETE: &str = " "; const SELECT_APPLIED_WRITE: &str = " - SELECT transaction_digest + SELECT content_digest FROM __logos_sql_applied_writes WHERE tx_id = ?1 "; const INSERT_APPLIED_WRITE: &str = " - INSERT INTO __logos_sql_applied_writes (tx_id, transaction_digest) + INSERT INTO __logos_sql_applied_writes (tx_id, content_digest) VALUES (?1, ?2) "; @@ -115,6 +118,19 @@ const WRITER_PRAGMAS: &str = " const FOREIGN_KEYS_PRAGMA: &str = "PRAGMA foreign_keys = ON;"; +// These functions depend on one connection, database file, or SQLite build. +// Their results cannot be reproduced from the ordered channel history. +const UNSUPPORTED_FUNCTIONS: [&str; 9] = [ + "changes", + "last_insert_rowid", + "load_extension", + "sqlite_compileoption_get", + "sqlite_compileoption_used", + "sqlite_offset", + "sqlite_source_id", + "sqlite_version", + "total_changes", +]; /// Raw database representation of a write waiting for publication. struct StoredPendingPublish { tx_id: Vec, @@ -137,10 +153,24 @@ pub struct PendingPublish { pub payload: Vec, } +/// A replicated database connection and the function state attached to it. +struct ReplicatedDatabase { + connection: Connection, + functions: FunctionOverrides, +} + +impl Deref for ReplicatedDatabase { + type Target = Connection; + + fn deref(&self) -> &Self::Target { + &self.connection + } +} + /// Owns the participant-local database connections. pub struct Databases { - lib: Connection, - live: Connection, + lib: ReplicatedDatabase, + live: ReplicatedDatabase, control: Connection, lib_path: PathBuf, live_path: PathBuf, @@ -157,7 +187,7 @@ impl Databases { let lib = open_writer(&lib_path)?; let live = open_writer(&live_path)?; - let control = open_writer(&control_path)?; + let control = open_connection(&control_path)?; for connection in [&lib, &live] { connection.execute_batch(PENDING_WRITE_SCHEMA)?; @@ -236,26 +266,21 @@ impl Databases { } /// Commits application effects and their pending publish record together in /// `LIVE.db`. - pub(crate) fn commit_local_write( - &mut self, - transaction: &Transaction, - encoded: &EncodedWrite, - ) -> Result { + pub(crate) fn commit_local_write(&mut self, transaction: &Transaction) -> Result { if self.pending_publish()?.is_some() { return Err(Error::PublishPending); } - let db_transaction = self.live.transaction()?; + let capture = self.live.functions.capture(); + let db_transaction = self.live.connection.transaction()?; - // TODO: Capture nondeterministic function results and include them in - // the transaction published to other participants. apply_statements(&db_transaction, transaction)?; - - let transaction_digest = transaction.digest(); + let captured_function_calls = capture.finish()?; + let encoded = EncodedWrite::new(transaction, captured_function_calls)?; db_transaction.execute( INSERT_APPLIED_WRITE, - params![encoded.tx_id.as_ref(), transaction_digest], + params![encoded.tx_id.as_ref(), encoded.content_digest], )?; db_transaction.execute( @@ -281,10 +306,10 @@ impl Databases { &mut self, write: &ChannelInscription, ) -> Result<(), Error> { - let transaction_digest = write.transaction.digest(); + let content_digest = write.content_digest(); - is_write_applied(&self.lib, write.tx_id, &transaction_digest)?; - is_write_applied(&self.live, write.tx_id, &transaction_digest)?; + is_write_applied(&self.lib.connection, write.tx_id, &content_digest)?; + is_write_applied(&self.live.connection, write.tx_id, &content_digest)?; // LIB and LIVE are separate SQLite files. If LIVE fails after LIB // commits, the checkpoint remains behind. Redelivery then skips LIB @@ -307,6 +332,7 @@ impl Databases { pub(crate) fn pending_publish(&self) -> Result, Error> { let record = self .live + .connection .query_row(SELECT_PENDING_PUBLISH, [], StoredPendingPublish::from_row) .optional()?; @@ -323,7 +349,10 @@ impl Databases { } pub(crate) fn mark_publish_complete(&self, tx_id: TxId) -> Result<(), Error> { - let changed = self.live.execute(MARK_PUBLISH_COMPLETE, [tx_id.as_ref()])?; + let changed = self + .live + .connection + .execute(MARK_PUBLISH_COMPLETE, [tx_id.as_ref()])?; if changed != 1 { return Err(Error::InvalidLocalState( @@ -343,7 +372,17 @@ impl Databases { } } -fn open_writer(path: &Path) -> Result { +fn open_writer(path: &Path) -> Result { + let connection = open_connection(path)?; + let functions = FunctionOverrides::install(&connection)?; + + Ok(ReplicatedDatabase { + connection, + functions, + }) +} + +fn open_connection(path: &Path) -> Result { let conn = Connection::open(path)?; configure_connection(&conn)?; @@ -360,17 +399,24 @@ fn configure_connection(conn: &Connection) -> Result<(), Error> { } fn apply_channel_write( - connection: &mut Connection, + database: &mut ReplicatedDatabase, write: &ChannelInscription, ) -> Result<(), Error> { - let transaction_digest = write.transaction.digest(); - let db_transaction = connection.transaction()?; + let content_digest = write.content_digest(); + let replay = database.functions.replay(&write.captured_function_calls); + let db_transaction = database.connection.transaction()?; - if is_write_applied(&db_transaction, write.tx_id, &transaction_digest)? { + if is_write_applied(&db_transaction, write.tx_id, &content_digest)? { return Ok(()); } if let Err(error) = apply_statements(&db_transaction, &write.transaction) { + if replay.failed() { + return Err(Error::InvalidPayload( + "captured SQLite function call does not match replay", + )); + } + return match error { Error::Database(error) if is_deterministic_sql_error(&error) => { Err(Error::RejectedSql(error)) @@ -379,9 +425,11 @@ fn apply_channel_write( }; } + replay.finish()?; + db_transaction.execute( INSERT_APPLIED_WRITE, - params![write.tx_id.as_ref(), transaction_digest], + params![write.tx_id.as_ref(), content_digest], )?; db_transaction.commit()?; @@ -391,7 +439,7 @@ fn apply_channel_write( fn is_write_applied( connection: &Connection, tx_id: TxId, - transaction_digest: &[u8; 32], + content_digest: &[u8; 32], ) -> Result { let stored_digest = connection .query_row(SELECT_APPLIED_WRITE, [tx_id.as_ref()], |row| { @@ -400,7 +448,7 @@ fn is_write_applied( .optional()?; if let Some(stored_digest) = stored_digest { - if stored_digest.as_slice() != transaction_digest { + if stored_digest.as_slice() != content_digest { return Err(Error::InvalidPayload( "transaction id was reused for different content", )); @@ -482,7 +530,14 @@ fn authorize_application_sql(context: AuthContext<'_>) -> Authorization { | AuthAction::Detach { .. } | AuthAction::Pragma { .. } ) || is_temporary_object_action(context.action) - || action_uses_reserved_name(context.action); + || action_uses_reserved_name(context.action) + || matches!( + context.action, + AuthAction::Function { function_name } + if UNSUPPORTED_FUNCTIONS + .iter() + .any(|name| function_name.eq_ignore_ascii_case(name)) + ); if denied { Authorization::Deny @@ -567,14 +622,16 @@ mod tests { node_types::{HeaderId, MsgId, Slot}, sequencer::SequencerCheckpoint, }; - use rusqlite::types::Value; + use rusqlite::{Connection, types::Value}; use tempfile::TempDir; use super::Databases; use crate::{ error::Error, - local_write, - protocol::{ChannelInscription, EncodedWrite, Statement, Transaction, TxId}, + protocol::{ + CapturedFunction, CapturedFunctionCall, CapturedFunctionCalls, ChannelInscription, + Statement, Transaction, TxId, + }, }; fn checkpoint(byte: u8, slot: u64) -> SequencerCheckpoint { @@ -606,11 +663,17 @@ mod tests { .expect("transaction should be valid") } - fn encoded_write(transaction: &Transaction) -> EncodedWrite { - EncodedWrite::new(transaction).expect("write should encode") + fn row_values(connection: &Connection, table: &str) -> Vec { + connection + .query_row(&format!("SELECT * FROM {table}"), [], |row| { + (0..row.as_ref().column_count()) + .map(|column| row.get(column)) + .collect() + }) + .expect("captured row should be readable") } - fn assert_application_sql_rejected(sql: &str) { + fn rejected_application_sql(sql: &str) -> Error { let dir = TempDir::new().expect("temporary directory should be created"); let mut db = Databases::open(dir.path()).expect("databases should open"); @@ -619,28 +682,35 @@ mod tests { .expect("application table should be created"); let transaction = transaction(sql); - let encoded = encoded_write(&transaction); - let error = db - .commit_local_write(&transaction, &encoded) + .commit_local_write(&transaction) .expect_err("application SQL should be rejected"); + assert!( + db.pending_publish() + .expect("pending publication should load") + .is_none() + ); + + error + } + + fn assert_application_sql_rejected(sql: &str) { + assert!(matches!(rejected_application_sql(sql), Error::Database(_))); + } + + fn assert_application_sql_denied(sql: &str) { + let error = rejected_application_sql(sql); + assert!(matches!( error, Error::Database(ref error) if error.sqlite_error_code() == Some(rusqlite::ErrorCode::AuthorizationForStatementDenied) )); - assert!( - db.pending_publish() - .expect("pending publication should load") - .is_none() - ); } - fn internal_schema( - connection: &rusqlite::Connection, - ) -> Vec<(String, String, String, Option)> { + fn internal_schema(connection: &Connection) -> Vec<(String, String, String, Option)> { let mut statement = connection .prepare( "SELECT type, name, tbl_name, sql @@ -690,9 +760,8 @@ mod tests { .expect("application table should be created"); let transaction = insert("hello"); - let encoded = encoded_write(&transaction); - - db.commit_local_write(&transaction, &encoded) + let tx_id = db + .commit_local_write(&transaction) .expect("write should commit"); let count: i64 = db @@ -706,7 +775,7 @@ mod tests { .expect("pending publish should load") .expect("pending publish should exist") .tx_id, - encoded.tx_id + tx_id ); } @@ -728,13 +797,15 @@ mod tests { .expect("application table should be created"); let transaction = insert("hello"); - let first_tx_id = - local_write::commit(&mut db, &transaction).expect("first write should commit"); + let first_tx_id = db + .commit_local_write(&transaction) + .expect("first write should commit"); db.mark_publish_complete(first_tx_id) .expect("first publication should complete"); - let second_tx_id = - local_write::commit(&mut db, &transaction).expect("second write should commit"); + let second_tx_id = db + .commit_local_write(&transaction) + .expect("second write should commit"); assert_ne!(second_tx_id, first_tx_id); @@ -746,6 +817,193 @@ mod tests { assert_eq!(count, 2); } + #[test] + fn function_results_are_replayed_exactly() { + let dir = TempDir::new().expect("temporary directory should be created"); + let mut db = Databases::open(dir.path()).expect("databases should open"); + let schema = "CREATE TABLE captured( + random_value, + random_blob_value, + date_value, + time_value, + datetime_value, + julian_day_value, + unix_epoch_value, + strftime_value, + time_diff_value, + current_date_value, + current_time_value, + current_timestamp_value + )"; + + for connection in [&db.lib, &db.live] { + connection + .execute(schema, []) + .expect("application table should be created"); + } + + let write = transaction( + "INSERT INTO captured VALUES ( + random(), + randomblob(16), + date('now'), + time('now'), + datetime('now'), + julianday('now'), + unixepoch('now'), + strftime('%s', 'now'), + timediff('now', 'now'), + CURRENT_DATE, + CURRENT_TIME, + CURRENT_TIMESTAMP + )", + ); + db.commit_local_write(&write) + .expect("local write should commit"); + + let pending = db + .pending_publish() + .expect("pending publish should load") + .expect("pending publish should exist"); + let channel_inscription = + ChannelInscription::decode(&pending.payload).expect("payload should decode"); + let functions = channel_inscription + .captured_function_calls + .as_slice() + .iter() + .map(|call| call.function) + .collect::>(); + + assert_eq!( + functions, + vec![ + CapturedFunction::Random, + CapturedFunction::RandomBlob, + CapturedFunction::Date, + CapturedFunction::Time, + CapturedFunction::DateTime, + CapturedFunction::JulianDay, + CapturedFunction::UnixEpoch, + CapturedFunction::Strftime, + CapturedFunction::TimeDiff, + CapturedFunction::CurrentDate, + CapturedFunction::CurrentTime, + CapturedFunction::CurrentTimestamp, + ] + ); + + db.apply_finalized_write(&channel_inscription) + .expect("captured write should replay"); + + assert_eq!( + row_values(&db.live, "captured"), + row_values(&db.lib, "captured") + ); + } + + #[test] + fn function_calls_inside_defaults_and_triggers_are_captured() { + let dir = TempDir::new().expect("temporary directory should be created"); + let mut db = Databases::open(dir.path()).expect("databases should open"); + let schema = " + CREATE TABLE items( + value TEXT, + created_at TEXT DEFAULT CURRENT_TIMESTAMP + ); + CREATE TABLE audit(random_value INTEGER); + CREATE TRIGGER audit_insert AFTER INSERT ON items BEGIN + INSERT INTO audit VALUES (random()); + END; + "; + + for connection in [&db.lib, &db.live] { + connection + .execute_batch(schema) + .expect("application schema should be created"); + } + + let write = transaction("INSERT INTO items(value) VALUES ('hello')"); + db.commit_local_write(&write) + .expect("local write should commit"); + + let pending = db + .pending_publish() + .expect("pending publish should load") + .expect("pending publish should exist"); + let channel_inscription = + ChannelInscription::decode(&pending.payload).expect("payload should decode"); + + assert_eq!( + channel_inscription.captured_function_calls.as_slice().len(), + 2 + ); + + db.apply_finalized_write(&channel_inscription) + .expect("trigger write should replay"); + + assert_eq!(row_values(&db.live, "items"), row_values(&db.lib, "items")); + assert_eq!(row_values(&db.live, "audit"), row_values(&db.lib, "audit")); + } + + #[test] + fn missing_function_result_rejects_channel_inscription() { + let dir = TempDir::new().expect("temporary directory should be created"); + let mut db = Databases::open(dir.path()).expect("databases should open"); + + db.live + .execute("CREATE TABLE items(value INTEGER)", []) + .expect("application table should be created"); + + let write = ChannelInscription { + tx_id: TxId::from([7; 32]), + transaction: transaction("INSERT INTO items VALUES (random())"), + captured_function_calls: CapturedFunctionCalls::empty(), + }; + + let error = db + .apply_adopted_write(&write) + .expect_err("missing result should reject the write"); + + assert!(matches!(error, Error::InvalidPayload(_))); + assert_eq!( + db.live + .query_row("SELECT count(*) FROM items", [], |row| row.get::<_, i64>(0)) + .expect("row count should be readable"), + 0 + ); + } + + #[test] + fn unused_function_result_rejects_channel_inscription() { + let dir = TempDir::new().expect("temporary directory should be created"); + let mut db = Databases::open(dir.path()).expect("databases should open"); + + db.live + .execute("CREATE TABLE items(value INTEGER)", []) + .expect("application table should be created"); + + let captured = CapturedFunctionCall::new(CapturedFunction::Random, Value::Integer(7)) + .expect("captured result should be valid"); + let write = ChannelInscription { + tx_id: TxId::from([7; 32]), + transaction: transaction("INSERT INTO items VALUES (1)"), + captured_function_calls: CapturedFunctionCalls::new(vec![captured]) + .expect("captured calls should be valid"), + }; + + let error = db + .apply_adopted_write(&write) + .expect_err("unused result should reject the write"); + + assert!(matches!(error, Error::InvalidPayload(_))); + assert_eq!( + db.live + .query_row("SELECT count(*) FROM items", [], |row| row.get::<_, i64>(0)) + .expect("row count should be readable"), + 0 + ); + } + #[test] fn reused_transaction_id_with_different_content_is_rejected() { let dir = TempDir::new().expect("temporary directory should be created"); @@ -759,10 +1017,12 @@ mod tests { let first = ChannelInscription { tx_id, transaction: insert("first"), + captured_function_calls: CapturedFunctionCalls::empty(), }; let conflicting = ChannelInscription { tx_id, transaction: insert("conflicting"), + captured_function_calls: CapturedFunctionCalls::empty(), }; db.apply_adopted_write(&first) @@ -797,10 +1057,12 @@ mod tests { let first = ChannelInscription { tx_id, transaction: insert("first"), + captured_function_calls: CapturedFunctionCalls::empty(), }; let conflicting = ChannelInscription { tx_id, transaction: insert("conflicting"), + captured_function_calls: CapturedFunctionCalls::empty(), }; db.apply_adopted_write(&first) @@ -825,9 +1087,7 @@ mod tests { let dir = TempDir::new().expect("temporary directory should be created"); let mut db = Databases::open(dir.path()).expect("databases should open"); let transaction = transaction("CREATE TABLE items(value TEXT NOT NULL)"); - let encoded = encoded_write(&transaction); - - db.commit_local_write(&transaction, &encoded) + db.commit_local_write(&transaction) .expect("schema write should commit"); db.live @@ -854,9 +1114,7 @@ mod tests { Statement::new(control.to_owned(), Vec::new()).expect("statement should be valid"), ]) .expect("transaction should be valid"); - let encoded = encoded_write(&transaction); - - db.commit_local_write(&transaction, &encoded) + db.commit_local_write(&transaction) .expect_err("transaction control should be rejected"); let count: i64 = db @@ -879,7 +1137,7 @@ mod tests { "PRAGMA synchronous = OFF", "ATTACH DATABASE ':memory:' AS other", ] { - assert_application_sql_rejected(sql); + assert_application_sql_denied(sql); } } @@ -894,7 +1152,19 @@ mod tests { "CREATE TRIGGER __logos_sql_trigger AFTER INSERT ON items BEGIN SELECT 1; END", "CREATE TABLE __LOGOS_SQL_mixed_case(value INTEGER)", ] { - assert_application_sql_rejected(sql); + assert_application_sql_denied(sql); + } + } + + #[test] + fn connection_dependent_functions_are_rejected() { + for function in [ + "changes()", + "last_insert_rowid()", + "sqlite_version()", + "total_changes()", + ] { + assert_application_sql_rejected(&format!("SELECT {function}")); } } @@ -904,7 +1174,7 @@ mod tests { "CREATE TEMP TABLE temporary_items(value INTEGER)", "CREATE TEMP VIEW temporary_items AS SELECT 1", ] { - assert_application_sql_rejected(sql); + assert_application_sql_denied(sql); } } @@ -913,9 +1183,8 @@ mod tests { let dir = TempDir::new().expect("temporary directory should be created"); let mut db = Databases::open(dir.path()).expect("databases should open"); let transaction = transaction("CREATE TABLE aaaaaaaaaa\u{65e5}(value INTEGER)"); - let encoded = encoded_write(&transaction); - db.commit_local_write(&transaction, &encoded) + db.commit_local_write(&transaction) .expect("non-reserved Unicode name should be accepted"); } diff --git a/logos_sql/src/functions.rs b/logos_sql/src/functions.rs new file mode 100644 index 000000000..45cbd9e50 --- /dev/null +++ b/logos_sql/src/functions.rs @@ -0,0 +1,298 @@ +//! Capture and replay of `SQLite` functions whose results vary between runs. +//! +//! Selected built-ins are overridden on each replicated writer connection. +//! During local execution, the override calls the original built-in on a +//! separate in-memory connection and records its result. During replay, it +//! returns the recorded result without evaluating the function again. + +use std::{ + collections::VecDeque, + sync::{Arc, Mutex, MutexGuard}, +}; + +use rusqlite::{ + Connection, + functions::{Context, FunctionFlags}, + params_from_iter, + types::Value, +}; + +use crate::{ + error::Error, + protocol::{CapturedFunction, CapturedFunctionCall, CapturedFunctionCalls}, +}; + +/// Describes one `SQLite` built-in intercepted by the replicated connection. +#[derive(Clone, Copy)] +struct FunctionDefinition { + function: CapturedFunction, + name: &'static str, + argument_count: i32, + keyword: bool, +} + +// SQLite functions whose results are captured during local execution and +// supplied in the same order during replay on other replicas. +const FUNCTIONS: [FunctionDefinition; 12] = [ + FunctionDefinition::new(CapturedFunction::Random, "random", 0), + FunctionDefinition::new(CapturedFunction::RandomBlob, "randomblob", 1), + FunctionDefinition::variadic(CapturedFunction::Date, "date"), + FunctionDefinition::variadic(CapturedFunction::Time, "time"), + FunctionDefinition::variadic(CapturedFunction::DateTime, "datetime"), + FunctionDefinition::variadic(CapturedFunction::JulianDay, "julianday"), + FunctionDefinition::variadic(CapturedFunction::UnixEpoch, "unixepoch"), + FunctionDefinition::variadic(CapturedFunction::Strftime, "strftime"), + FunctionDefinition::new(CapturedFunction::TimeDiff, "timediff", 2), + FunctionDefinition::keyword(CapturedFunction::CurrentDate, "current_date"), + FunctionDefinition::keyword(CapturedFunction::CurrentTime, "current_time"), + FunctionDefinition::keyword(CapturedFunction::CurrentTimestamp, "current_timestamp"), +]; + +impl FunctionDefinition { + const fn new(function: CapturedFunction, name: &'static str, argument_count: i32) -> Self { + Self { + function, + name, + argument_count, + keyword: false, + } + } + + const fn variadic(function: CapturedFunction, name: &'static str) -> Self { + Self::new(function, name, -1) + } + + const fn keyword(function: CapturedFunction, name: &'static str) -> Self { + Self { + function, + name, + argument_count: 0, + keyword: true, + } + } + + fn query(self, argument_count: usize) -> String { + if self.keyword { + return format!("SELECT {}", self.name); + } + + let parameters = std::iter::repeat_n("?", argument_count) + .collect::>() + .join(", "); + + format!("SELECT {}({parameters})", self.name) + } +} + +/// Behavior of the installed callbacks for the current writer operation. +/// Only one capture or replay session can be active on a connection. +enum Mode { + Passthrough, + Capture(Vec), + Replay { + calls: VecDeque, + failed: bool, + }, +} + +/// Controls the function implementations installed on one replicated writer. +pub struct FunctionOverrides { + state: Arc>, +} + +impl FunctionOverrides { + /// Installs the overrides and creates the untouched `SQLite` connection + /// used to evaluate the original built-ins during capture. + pub fn install(connection: &Connection) -> Result { + let state = Arc::new(Mutex::new(Mode::Passthrough)); + let evaluator = Arc::new(Mutex::new(Connection::open_in_memory()?)); + + for definition in FUNCTIONS { + let state = Arc::clone(&state); + let evaluator = Arc::clone(&evaluator); + + connection.create_scalar_function( + definition.name, + definition.argument_count, + FunctionFlags::SQLITE_UTF8, + move |context| invoke(definition, context, &state, &evaluator), + )?; + } + + Ok(Self { state }) + } + + /// Starts recording function calls made by one local transaction. + pub fn capture(&mut self) -> CaptureSession<'_> { + *lock(&self.state) = Mode::Capture(Vec::new()); + + CaptureSession::new(self) + } + + /// Starts replaying the recorded calls for one received transaction. + pub fn replay(&mut self, calls: &CapturedFunctionCalls) -> ReplaySession<'_> { + *lock(&self.state) = Mode::Replay { + calls: calls.as_slice().iter().cloned().collect(), + failed: false, + }; + + ReplaySession::new(self) + } +} + +/// Restores passthrough mode if capture exits before [`Self::finish`]. +pub struct CaptureSession<'a> { + overrides: &'a mut FunctionOverrides, + active: bool, +} + +impl<'a> CaptureSession<'a> { + const fn new(overrides: &'a mut FunctionOverrides) -> Self { + Self { + overrides, + active: true, + } + } + + pub fn finish(mut self) -> Result { + let mode = take_mode(&self.overrides.state); + self.active = false; + + let Mode::Capture(calls) = mode else { + unreachable!("capture guard must own an active capture session"); + }; + + CapturedFunctionCalls::new(calls) + } +} + +impl Drop for CaptureSession<'_> { + fn drop(&mut self) { + if self.active { + reset(&self.overrides.state); + } + } +} + +/// Restores passthrough mode if replay exits before [`Self::finish`]. +pub struct ReplaySession<'a> { + overrides: &'a mut FunctionOverrides, + active: bool, +} + +impl<'a> ReplaySession<'a> { + const fn new(overrides: &'a mut FunctionOverrides) -> Self { + Self { + overrides, + active: true, + } + } + + pub fn failed(&self) -> bool { + matches!( + &*lock(&self.overrides.state), + Mode::Replay { failed: true, .. } + ) + } + + pub fn finish(mut self) -> Result<(), Error> { + let mode = take_mode(&self.overrides.state); + self.active = false; + + let Mode::Replay { calls, failed } = mode else { + unreachable!("replay guard must own an active replay session"); + }; + + if failed { + return Err(Error::InvalidPayload( + "captured SQLite function call does not match replay", + )); + } + + if !calls.is_empty() { + return Err(Error::InvalidPayload( + "captured SQLite function results were not consumed", + )); + } + + Ok(()) + } +} + +impl Drop for ReplaySession<'_> { + fn drop(&mut self) { + if self.active { + reset(&self.overrides.state); + } + } +} + +fn invoke( + definition: FunctionDefinition, + context: &Context<'_>, + state: &Mutex, + evaluator: &Mutex, +) -> rusqlite::Result { + let mut mode = lock(state); + + match &mut *mode { + Mode::Passthrough => evaluate(definition, context, evaluator), + Mode::Capture(calls) => { + let result = evaluate(definition, context, evaluator)?; + let call = CapturedFunctionCall::new(definition.function, result.clone()) + .map_err(|error| rusqlite::Error::UserFunctionError(Box::new(error)))?; + + calls.push(call); + + Ok(result) + } + Mode::Replay { calls, failed } => { + let Some(call) = calls.pop_front() else { + *failed = true; + return Err(replay_error("captured SQLite function result is missing")); + }; + + if call.function != definition.function { + *failed = true; + return Err(replay_error( + "captured SQLite function order does not match", + )); + } + + Ok(call.result.into_value()) + } + } +} + +fn evaluate( + definition: FunctionDefinition, + context: &Context<'_>, + evaluator: &Mutex, +) -> rusqlite::Result { + // Calling the function on the replicated connection would recurse into + // this override. The evaluator connection still has SQLite's built-in. + let arguments = (0..context.len()) + .map(|index| context.get_raw(index).into()) + .collect::>(); + let query = definition.query(arguments.len()); + + lock(evaluator).query_row(&query, params_from_iter(arguments), |row| row.get(0)) +} + +fn replay_error(message: &'static str) -> rusqlite::Error { + rusqlite::Error::UserFunctionError(message.into()) +} + +fn take_mode(state: &Mutex) -> Mode { + std::mem::replace(&mut *lock(state), Mode::Passthrough) +} + +fn reset(state: &Mutex) { + *lock(state) = Mode::Passthrough; +} + +fn lock(mutex: &Mutex) -> MutexGuard<'_, T> { + mutex + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) +} diff --git a/logos_sql/src/lib.rs b/logos_sql/src/lib.rs index 7229a5993..081f4a4ec 100644 --- a/logos_sql/src/lib.rs +++ b/logos_sql/src/lib.rs @@ -8,7 +8,7 @@ mod applier; mod db; mod error; -mod local_write; +mod functions; mod logos_sql; mod protocol; mod runtime; diff --git a/logos_sql/src/local_write.rs b/logos_sql/src/local_write.rs deleted file mode 100644 index c0ba191c9..000000000 --- a/logos_sql/src/local_write.rs +++ /dev/null @@ -1,15 +0,0 @@ -//! Outbound write path for transactions initiated by the local application. -//! -//! Transactions arriving through channel history are handled by the applier. - -use crate::{ - db::Databases, - error::Error, - protocol::{EncodedWrite, Transaction, TxId}, -}; - -pub fn commit(db: &mut Databases, transaction: &Transaction) -> Result { - let encoded = EncodedWrite::new(transaction)?; - - db.commit_local_write(transaction, &encoded) -} diff --git a/logos_sql/src/protocol/codec.rs b/logos_sql/src/protocol/codec.rs index dc289f80f..9804baed9 100644 --- a/logos_sql/src/protocol/codec.rs +++ b/logos_sql/src/protocol/codec.rs @@ -4,14 +4,31 @@ use lb_codec::{BinaryDecode, BinaryEncode, DecodeError}; use lb_utils::bounded::UpperBoundedVec; use rusqlite::types::Value; -use super::{MAX_PAYLOAD_BYTES, SqlParameter, SqlText}; +use super::{CapturedFunction, MAX_PAYLOAD_BYTES, SqlParameter, SqlText}; +// Stable wire tags for SQLite value variants. Existing values must never be +// renumbered because inscriptions remain part of the channel history. const NULL: u8 = 0; const INTEGER: u8 = 1; const REAL: u8 = 2; const TEXT: u8 = 3; const BLOB: u8 = 4; +// Stable wire tags for captured SQLite functions. New functions may be +// appended, but existing values must keep their encoding. +const RANDOM: u8 = 0; +const RANDOM_BLOB: u8 = 1; +const DATE: u8 = 2; +const TIME: u8 = 3; +const DATE_TIME: u8 = 4; +const JULIAN_DAY: u8 = 5; +const UNIX_EPOCH: u8 = 6; +const STRFTIME: u8 = 7; +const TIME_DIFF: u8 = 8; +const CURRENT_DATE: u8 = 9; +const CURRENT_TIME: u8 = 10; +const CURRENT_TIMESTAMP: u8 = 11; + type BoundedBytes = UpperBoundedVec; // Every variable-length field uses the same fixed-width prefix. Keep the two @@ -19,6 +36,64 @@ type BoundedBytes = UpperBoundedVec; const _: () = assert!(MAX_PAYLOAD_BYTES > u16::MAX as usize); const _: () = assert!(MAX_PAYLOAD_BYTES <= u32::MAX as usize); +impl BinaryEncode for CapturedFunction { + fn encoded_length(&self) -> usize { + size_of::() + } + + fn encode_into(&self, out: &mut Vec) { + let discriminant = match self { + Self::Random => RANDOM, + Self::RandomBlob => RANDOM_BLOB, + Self::Date => DATE, + Self::Time => TIME, + Self::DateTime => DATE_TIME, + Self::JulianDay => JULIAN_DAY, + Self::UnixEpoch => UNIX_EPOCH, + Self::Strftime => STRFTIME, + Self::TimeDiff => TIME_DIFF, + Self::CurrentDate => CURRENT_DATE, + Self::CurrentTime => CURRENT_TIME, + Self::CurrentTimestamp => CURRENT_TIMESTAMP, + }; + + discriminant.encode_into(out); + } +} + +impl BinaryDecode for CapturedFunction { + type Context = (); + + fn decode<'input>( + input: &'input [u8], + (): &Self::Context, + ) -> Result<(&'input [u8], Self), DecodeError> { + let (input, discriminant) = ::decode(input, &())?; + + let function = match discriminant { + RANDOM => Self::Random, + RANDOM_BLOB => Self::RandomBlob, + DATE => Self::Date, + TIME => Self::Time, + DATE_TIME => Self::DateTime, + JULIAN_DAY => Self::JulianDay, + UNIX_EPOCH => Self::UnixEpoch, + STRFTIME => Self::Strftime, + TIME_DIFF => Self::TimeDiff, + CURRENT_DATE => Self::CurrentDate, + CURRENT_TIME => Self::CurrentTime, + CURRENT_TIMESTAMP => Self::CurrentTimestamp, + _ => { + return Err(DecodeError::unknown_discriminant::(u64::from( + discriminant, + ))); + } + }; + + Ok((input, function)) + } +} + impl BinaryEncode for SqlText { fn encoded_length(&self) -> usize { size_of::() + self.as_str().len() diff --git a/logos_sql/src/protocol/fixtures.rs b/logos_sql/src/protocol/fixtures.rs index 711e06805..609934b0a 100644 --- a/logos_sql/src/protocol/fixtures.rs +++ b/logos_sql/src/protocol/fixtures.rs @@ -3,7 +3,26 @@ use lb_codec::codec_fixtures; use rusqlite::types::Value; -use super::{ChannelInscription, SqlParameter, SqlText, Statement, Transaction, TxId}; +use super::{ + CapturedFunction, CapturedFunctionCall, CapturedFunctionCalls, ChannelInscription, + SqlParameter, SqlText, Statement, Transaction, TxId, +}; + +codec_fixtures!( + CapturedFunction, + CapturedFunction::Random => "00", + CapturedFunction::RandomBlob => "01", + CapturedFunction::Date => "02", + CapturedFunction::Time => "03", + CapturedFunction::DateTime => "04", + CapturedFunction::JulianDay => "05", + CapturedFunction::UnixEpoch => "06", + CapturedFunction::Strftime => "07", + CapturedFunction::TimeDiff => "08", + CapturedFunction::CurrentDate => "09", + CapturedFunction::CurrentTime => "0a", + CapturedFunction::CurrentTimestamp => "0b" +); codec_fixtures!( TxId, @@ -30,6 +49,26 @@ codec_fixtures!( "040200000000ff" ); +fn captured_function_call_fixture() -> CapturedFunctionCall { + CapturedFunctionCall::new(CapturedFunction::Random, Value::Integer(42)) + .expect("fixture should be valid") +} + +codec_fixtures!( + CapturedFunctionCall, + captured_function_call_fixture() => "00012a00000000000000" +); + +fn captured_function_calls_fixture() -> CapturedFunctionCalls { + CapturedFunctionCalls::new(vec![captured_function_call_fixture()]) + .expect("fixture should be valid") +} + +codec_fixtures!( + CapturedFunctionCalls, + captured_function_calls_fixture() => "0100000000012a00000000000000" +); + fn statement_fixture() -> Statement { Statement::new("SELECT 1".to_owned(), Vec::new()).expect("fixture should be valid") } @@ -74,6 +113,7 @@ fn channel_inscription_fixture() -> ChannelInscription { ChannelInscription { tx_id: TxId::from([3; 32]), transaction: transaction_fixture(), + captured_function_calls: CapturedFunctionCalls::empty(), } } @@ -81,6 +121,7 @@ codec_fixtures!( ChannelInscription, channel_inscription_fixture() => concat!( "0303030303030303030303030303030303030303030303030303030303030303", - "010000000800000053454c454354203100000000" + "010000000800000053454c454354203100000000", + "00000000" ) ); diff --git a/logos_sql/src/protocol/mod.rs b/logos_sql/src/protocol/mod.rs index 0f7dec08e..6ada39458 100644 --- a/logos_sql/src/protocol/mod.rs +++ b/logos_sql/src/protocol/mod.rs @@ -14,9 +14,12 @@ use crate::error::Error; mod codec; mod fixtures; +// Every payload starts with this marker and version before the encoded body. pub const PAYLOAD_MARKER: [u8; 9] = *b"LOGOS_SQL"; -const PAYLOAD_VERSION: u16 = 1; +const PAYLOAD_VERSION: u16 = 2; const PAYLOAD_HEADER_LEN: usize = PAYLOAD_MARKER.len() + size_of::(); + +// A complete λSQL transaction must fit into one channel inscription. const MAX_PAYLOAD_BYTES: usize = Inscription::MAX; /// Stable identity of one application write. @@ -113,6 +116,12 @@ impl ToSql for SqlParameter { } } +impl SqlParameter { + pub fn into_value(self) -> Value { + self.0 + } +} + /// One parameterized SQL statement. #[derive(Clone, Debug, PartialEq, BinaryCodec)] pub struct Statement { @@ -177,9 +186,62 @@ impl Transaction { pub fn statements(&self) -> &[Statement] { self.statements.as_slice() } +} - pub(crate) fn digest(&self) -> [u8; 32] { - Blake2b::::digest(self.encode_to_vec()).into() +/// `SQLite` function whose result must be reproduced during channel replay. +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub enum CapturedFunction { + Random, + RandomBlob, + Date, + Time, + DateTime, + JulianDay, + UnixEpoch, + Strftime, + TimeDiff, + CurrentDate, + CurrentTime, + CurrentTimestamp, +} + +/// One captured `SQLite` function call and the value returned locally. +#[derive(Clone, Debug, PartialEq, BinaryCodec)] +pub struct CapturedFunctionCall { + pub function: CapturedFunction, + pub result: SqlParameter, +} + +impl CapturedFunctionCall { + pub fn new(function: CapturedFunction, result: Value) -> Result { + Ok(Self { + function, + result: SqlParameter::try_from(result)?, + }) + } +} + +/// Function results captured while executing one replicated transaction. +#[derive(Clone, Debug, PartialEq, BinaryCodec)] +pub struct CapturedFunctionCalls { + calls: UpperBoundedVec, +} + +impl CapturedFunctionCalls { + pub fn new(calls: Vec) -> Result { + let calls = UpperBoundedVec::try_from(calls).map_err(|_| Error::InscriptionTooLarge)?; + + Ok(Self { calls }) + } + + pub const fn empty() -> Self { + Self { + calls: UpperBoundedVec::new_unchecked(Vec::new()), + } + } + + pub fn as_slice(&self) -> &[CapturedFunctionCall] { + self.calls.as_slice() } } @@ -188,6 +250,7 @@ impl Transaction { pub struct ChannelInscription { pub tx_id: TxId, pub transaction: Transaction, + pub captured_function_calls: CapturedFunctionCalls, } impl ChannelInscription { @@ -225,26 +288,40 @@ impl ChannelInscription { ::decode_all(body, &()) .map_err(|_| Error::InvalidPayload("body cannot be decoded")) } + + pub fn content_digest(&self) -> [u8; 32] { + Blake2b::::digest(self.encode_to_vec()).into() + } } /// A local write after its identity and channel payload have been encoded. pub struct EncodedWrite { pub tx_id: TxId, + pub content_digest: [u8; 32], pub payload: Vec, } impl EncodedWrite { - pub fn new(transaction: &Transaction) -> Result { + pub fn new( + transaction: &Transaction, + captured_function_calls: CapturedFunctionCalls, + ) -> Result { let tx_id = TxId::generate(); let channel_inscription = ChannelInscription { tx_id, transaction: transaction.clone(), + captured_function_calls, }; + let content_digest = channel_inscription.content_digest(); let payload = channel_inscription.encode()?; - Ok(Self { tx_id, payload }) + Ok(Self { + tx_id, + content_digest, + payload, + }) } } @@ -271,7 +348,8 @@ mod tests { use rusqlite::types::Value; use super::{ - ChannelInscription, EncodedWrite, MAX_PAYLOAD_BYTES, Statement, Transaction, TxId, + CapturedFunctionCalls, ChannelInscription, EncodedWrite, MAX_PAYLOAD_BYTES, Statement, + Transaction, TxId, }; #[test] @@ -292,7 +370,8 @@ mod tests { ]) .expect("transaction should be valid"); - let encoded = EncodedWrite::new(&transaction).expect("submission should encode"); + let encoded = EncodedWrite::new(&transaction, CapturedFunctionCalls::empty()) + .expect("payload should encode"); let decoded = ChannelInscription::decode(&encoded.payload) .expect("channel inscription should decode"); @@ -309,6 +388,7 @@ mod tests { let write = ChannelInscription { tx_id: TxId::from([3; 32]), transaction, + captured_function_calls: CapturedFunctionCalls::empty(), }; let mut payload = write.encode().expect("payload should encode"); payload.push(0); @@ -325,18 +405,37 @@ mod tests { let write = ChannelInscription { tx_id: TxId::from([3; 32]), transaction, + captured_function_calls: CapturedFunctionCalls::empty(), }; let expected = hex::decode(concat!( - "4c4f474f535f53514c0100", + "4c4f474f535f53514c0200", "0303030303030303030303030303030303030303030303030303030303030303", - "010000000800000053454c454354203100000000" + "010000000800000053454c45435420310000000000000000" )) .expect("fixture should be valid hex"); assert_eq!(write.encode().expect("payload should encode"), expected); } + #[test] + fn content_digest_is_pinned() { + let transaction = Transaction::new(vec![ + Statement::new("SELECT 1".to_owned(), Vec::new()).expect("statement should be valid"), + ]) + .expect("transaction should be valid"); + let write = ChannelInscription { + tx_id: TxId::from([3; 32]), + transaction, + captured_function_calls: CapturedFunctionCalls::empty(), + }; + + assert_eq!( + hex::encode(write.content_digest()), + "b119823633ba6fbe90618b226b0d68eae1805876a86c1057237aca6205874b30" + ); + } + #[test] fn complete_payload_must_fit_one_inscription() { let transaction = Transaction::new(vec![ @@ -347,7 +446,7 @@ mod tests { .expect("statement should be valid"), ]) .expect("transaction should be valid"); - let result = EncodedWrite::new(&transaction); + let result = EncodedWrite::new(&transaction, CapturedFunctionCalls::empty()); assert!(matches!(result, Err(crate::Error::InscriptionTooLarge))); } diff --git a/logos_sql/src/runtime.rs b/logos_sql/src/runtime.rs index 392daa59e..c06a0cd5b 100644 --- a/logos_sql/src/runtime.rs +++ b/logos_sql/src/runtime.rs @@ -16,7 +16,6 @@ use crate::{ applier, db::Databases, error::Error, - local_write, protocol::{Transaction, TxId}, }; @@ -176,7 +175,7 @@ impl Runtime { } else if !self.sequencer_ready { Err(Error::SequencerNotReady) } else { - let committed = local_write::commit(&mut self.db, &transaction); + let committed = self.db.commit_local_write(&transaction); if let Ok(tx_id) = committed { tracing::trace!(