feat(lsql): add local replication flow (#3322)

Cargo.lock

@@ -1567,7 +1567,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "ccc2776f0c61eca1ca32528f85548abd1a4be8fb53d1b21c013e4f18da1e7090"
 dependencies = [
  "data-encoding",
- "syn 1.0.109",
+ "syn 2.0.118",
 ]
 
 [[package]]
@@ -1752,7 +1752,7 @@ dependencies = [
  "libc",
  "option-ext",
  "redox_users",
- "windows-sys 0.59.0",
+ "windows-sys 0.61.2",
 ]
 
 [[package]]
@@ -2002,6 +2002,18 @@ dependencies = [
  "once_cell",
 ]
 
+[[package]]
+name = "fallible-iterator"
+version = "0.3.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "2acce4a10f12dc2fb14a218589d4f1f62ef011b2d0cc4b3cb1bba8e94da14649"
+
+[[package]]
+name = "fallible-streaming-iterator"
+version = "0.1.9"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "7360491ce676a36bf9bb3c56c1aa791658183a54d2744120f27285738d90465a"
+
 [[package]]
 name = "fastbloom"
 version = "0.14.1"
@@ -2774,7 +2786,7 @@ dependencies = [
  "libc",
  "percent-encoding",
  "pin-project-lite",
- "socket2 0.5.10",
+ "socket2 0.6.4",
  "tokio",
  "tower-service",
  "tracing",
@@ -3874,6 +3886,17 @@ dependencies = [
  "libz-sys",
 ]
 
+[[package]]
+name = "libsqlite3-sys"
+version = "0.33.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "947e6816f7825b2b45027c2c32e7085da9934defa535de4a6a46b10a4d5257fa"
+dependencies = [
+ "cc",
+ "pkg-config",
+ "vcpkg",
+]
+
 [[package]]
 name = "libz-sys"
 version = "1.1.29"
@@ -5384,6 +5407,27 @@ dependencies = [
  "tracing",
 ]
 
+[[package]]
+name = "logos-sql"
+version = "0.0.0"
+dependencies = [
+ "bincode",
+ "clap",
+ "hex",
+ "logos-blockchain-groth16",
+ "logos-blockchain-key-management-system-service",
+ "logos-blockchain-log-targets",
+ "logos-blockchain-zone-sdk",
+ "rand 0.8.6",
+ "reqwest 0.12.28",
+ "rusqlite",
+ "serde",
+ "tempfile",
+ "thiserror 2.0.18",
+ "tokio",
+ "tracing",
+]
+
 [[package]]
 name = "loki-api"
 version = "0.1.3"
@@ -5884,7 +5928,7 @@ version = "0.50.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "7957b9740744892f114936ab4a57b3f487491bbeafaf8083688b16841a4240e5"
 dependencies = [
- "windows-sys 0.59.0",
+ "windows-sys 0.61.2",
 ]
 
 [[package]]
@@ -6909,7 +6953,7 @@ dependencies = [
  "quinn-udp",
  "rustc-hash 2.1.2",
  "rustls",
- "socket2 0.5.10",
+ "socket2 0.6.4",
  "thiserror 2.0.18",
  "tokio",
  "tracing",
@@ -6947,7 +6991,7 @@ dependencies = [
  "cfg_aliases",
  "libc",
  "once_cell",
- "socket2 0.5.10",
+ "socket2 0.6.4",
  "tracing",
  "windows-sys 0.59.0",
 ]
@@ -7365,6 +7409,20 @@ dependencies = [
  "tokio",
 ]
 
+[[package]]
+name = "rusqlite"
+version = "0.35.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "a22715a5d6deef63c637207afbe68d0c72c3f8d0022d7cf9714c442d6157606b"
+dependencies = [
+ "bitflags 2.13.0",
+ "fallible-iterator",
+ "fallible-streaming-iterator",
+ "hashlink 0.10.0",
+ "libsqlite3-sys",
+ "smallvec",
+]
+
 [[package]]
 name = "rust-embed"
 version = "8.11.0"
@@ -8306,7 +8364,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "230a1b821ccbd75b185820a1f1ff7b14d21da1e442e22c0863ea5f08771a8874"
 dependencies = [
  "rustix 1.1.4",
- "windows-sys 0.59.0",
+ "windows-sys 0.61.2",
 ]
 
 [[package]]

Cargo.toml

@@ -50,6 +50,7 @@ members = [
   "kms/operators",
   "ledger",
   "libp2p",
+  "logos_sql",
   "merkle/blake2btree",
   "merkle/dynamic-merkle",
   "merkle/tree",
@@ -254,6 +255,7 @@ reqwest                            = { default-features = false, version = "0.12
 rocksdb                            = { default-features = false, version = "0.24" }
 rpds                               = { default-features = false, version = "1" }
 rs-merkle-tree                     = { default-features = false, version = "0.1" }
+rusqlite                           = { default-features = false, version = "0.35" }
 tar                                = { default-features = false, version = "0.4.46" }
 # Must stay in sync with the rust-rapidsnark flake input in flake.nix.
 rust-rapidsnark       = { default-features = false, git = "https://github.com/logos-blockchain/logos-blockchain-rust-rapidsnark.git", rev = "e91187f8ccb5bbfc7bb00dac88169112428da78f" } # feat/nixify

logos_sql/Cargo.toml

@@ -0,0 +1,33 @@
+[package]
+categories  = { workspace = true }
+description = "Logos SQL (λSQL) - decentralized SQLite over Logos Blockchain"
+edition     = { workspace = true }
+keywords    = { workspace = true }
+license     = { workspace = true }
+name        = "logos-sql"
+publish     = false
+readme      = { workspace = true }
+repository  = { workspace = true }
+version     = { workspace = true }
+
+[lints]
+workspace = true
+
+[dependencies]
+bincode                          = { workspace = true }
+lb-key-management-system-service = { workspace = true }
+lb-log-targets                   = { workspace = true }
+lb-zone-sdk                      = { workspace = true }
+rand                             = { features = ["getrandom"], workspace = true }
+reqwest                          = { workspace = true }
+rusqlite                         = { features = ["bundled", "functions", "hooks"], workspace = true }
+serde                            = { features = ["derive"], workspace = true }
+thiserror                        = { workspace = true }
+tokio                            = { features = ["macros", "rt", "sync", "time"], workspace = true }
+tracing                          = { workspace = true }
+
+[dev-dependencies]
+clap       = { features = ["derive", "env", "help", "std"], workspace = true }
+hex        = { workspace = true }
+lb-groth16 = { workspace = true }
+tempfile   = { workspace = true }

logos_sql/examples/password_manager/config.rs

@@ -0,0 +1,87 @@
+//! Startup configuration supplied through flags or environment variables.
+
+use std::path::PathBuf;
+
+use clap::Parser;
+use lb_groth16::fr_from_bytes;
+use lb_key_management_system_service::keys::{Ed25519Key, ZkPublicKey};
+use lb_zone_sdk::{node_types::ChannelId, sequencer::FundingConfig};
+use logos_sql::LogosSqlConfig;
+use reqwest::Url;
+
+#[derive(Parser)]
+#[command(name = "password-manager")]
+struct Options {
+    /// Channel carrying the shared password-manager writes.
+    #[arg(long, env = "LOGOS_SQL_CHANNEL_ID", value_parser = parse_channel_id)]
+    channel_id: ChannelId,
+
+    /// Ed25519 secret key used to sign inscriptions.
+    #[arg(long, env = "LOGOS_SQL_SIGNING_KEY", value_parser = parse_signing_key)]
+    signing_key: Ed25519Key,
+
+    /// Public key funding inscription fees.
+    #[arg(long, env = "LOGOS_SQL_FUNDING_KEY", value_parser = parse_funding_key)]
+    funding_key: ZkPublicKey,
+
+    /// Maximum fee accepted for one inscription.
+    #[arg(long, env = "LOGOS_SQL_MAX_TX_FEE")]
+    max_tx_fee: u64,
+
+    /// Percentage of the maximum fee offered as priority fee.
+    #[arg(
+        long,
+        env = "LOGOS_SQL_PRIORITY_FEE_PERCENT",
+        default_value_t = FundingConfig::DEFAULT_PRIORITY_FEE_PERCENT
+    )]
+    priority_fee_percent: u64,
+
+    /// Node HTTP API used by `ZoneSDK`.
+    #[arg(long, env = "LOGOS_SQL_NODE_URL")]
+    node_url: Url,
+
+    /// Participant-local directory containing the `SQLite` databases.
+    #[arg(long, env = "LOGOS_SQL_STATE_DIR")]
+    state_dir: PathBuf,
+}
+
+/// Parses startup options and builds the `λSQL` configuration.
+pub fn from_args() -> LogosSqlConfig {
+    let options = Options::parse();
+
+    LogosSqlConfig {
+        channel_id: options.channel_id,
+        signing_key: options.signing_key,
+        node_url: options.node_url,
+        funding: FundingConfig {
+            funding_pk: options.funding_key,
+            max_tx_fee: options.max_tx_fee.into(),
+            priority_fee_percent: options.priority_fee_percent,
+        },
+        state_dir: options.state_dir,
+    }
+}
+
+fn parse_channel_id(value: &str) -> Result<ChannelId, String> {
+    parse_hex(value).map(ChannelId::from)
+}
+
+fn parse_signing_key(value: &str) -> Result<Ed25519Key, String> {
+    parse_hex(value).map(|bytes| Ed25519Key::from_bytes(&bytes))
+}
+
+fn parse_funding_key(value: &str) -> Result<ZkPublicKey, String> {
+    let bytes = hex::decode(value).map_err(|error| error.to_string())?;
+
+    fr_from_bytes(&bytes)
+        .map(ZkPublicKey::new)
+        .map_err(|error| error.to_string())
+}
+
+fn parse_hex<const N: usize>(value: &str) -> Result<[u8; N], String> {
+    let bytes = hex::decode(value).map_err(|error| error.to_string())?;
+
+    bytes
+        .try_into()
+        .map_err(|bytes: Vec<u8>| format!("expected {N} bytes, received {}", bytes.len()))
+}

logos_sql/examples/password_manager/main.rs

@@ -0,0 +1,38 @@
+//! A password manager built on `λSQL`.
+//!
+//! This first version stores and replicates passwords in plaintext. It exists
+//! only to demonstrate the `λSQL` application API. Do not enter real
+//! credentials. Application-side encryption will be added in a follow-up.
+//!
+//! Run with `--help` to see startup configuration. Each option also accepts its
+//! corresponding `LOGOS_SQL_*` environment variable. After startup, enter
+//! `help` to see the available commands. A short session could look like this:
+//!
+//! ```text
+//! add email andrus@example.org not-a-real-password
+//! update email another-fake-password
+//! show email
+//! list
+//! exit
+//! ```
+
+mod config;
+mod passwords;
+mod repl;
+
+use std::error::Error;
+
+use passwords::PasswordManager;
+
+type AppResult<T> = Result<T, Box<dyn Error + Send + Sync>>;
+
+#[tokio::main(flavor = "current_thread")]
+async fn main() -> AppResult<()> {
+    let manager = PasswordManager::start(config::from_args()).await?;
+
+    let result = repl::run(&manager).await;
+
+    manager.shutdown().await?;
+
+    result
+}

logos_sql/examples/password_manager/passwords.rs

@@ -0,0 +1,180 @@
+//! Password-manager data and operations.
+//!
+//! `PasswordManager` owns `LogosSql` directly so the example shows where the
+//! application ends and the replication library begins. Domain operations are
+//! translated into parameterized SQL writes here.
+
+use logos_sql::{Error as LogosSqlError, LogosSql, LogosSqlConfig, TransactionBuilder, TxId};
+
+use crate::AppResult;
+
+const SELECT_CREDENTIALS: &str = "
+    SELECT label, account
+    FROM credentials
+    ORDER BY label
+";
+const SELECT_SCHEMA_EXISTS: &str = "
+    SELECT COUNT(*) = 1
+    FROM sqlite_schema
+    WHERE type = 'table' AND name = 'credentials'
+";
+
+fn prepare_schema() -> TransactionBuilder {
+    TransactionBuilder::new(
+        "CREATE TABLE IF NOT EXISTS credentials (
+            label TEXT PRIMARY KEY,
+            account TEXT NOT NULL,
+            password TEXT NOT NULL
+        )",
+    )
+}
+
+fn prepare_credential_insert(label: &str, account: &str, password: &str) -> TransactionBuilder {
+    TransactionBuilder::new(
+        "INSERT INTO credentials (label, account, password)
+         VALUES (?1, ?2, ?3)",
+    )
+    .bind(label)
+    .bind(account)
+    .bind(password)
+}
+
+fn prepare_password_update(label: &str, password: &str) -> TransactionBuilder {
+    TransactionBuilder::new(
+        "UPDATE credentials
+         SET password = ?2
+         WHERE label = ?1",
+    )
+    .bind(label)
+    .bind(password)
+}
+
+fn prepare_credential_delete(label: &str) -> TransactionBuilder {
+    TransactionBuilder::new("DELETE FROM credentials WHERE label = ?1").bind(label)
+}
+
+/// One credential in the current local database view.
+///
+/// Passwords are deliberately plaintext in this first example slice. Do not
+/// use this example with real credentials.
+#[derive(Debug)]
+pub struct Credential {
+    pub label: String,
+    pub account: String,
+    pub password: String,
+}
+
+/// Non-secret credential information shown when listing the database.
+#[derive(Debug)]
+pub struct CredentialSummary {
+    pub label: String,
+    pub account: String,
+}
+
+/// Application-facing password-manager operations over one `λSQL` database.
+pub struct PasswordManager {
+    logos_sql: LogosSql,
+}
+
+impl PasswordManager {
+    /// Starts `λSQL`, catches up with the channel, and prepares the database.
+    pub async fn start(config: LogosSqlConfig) -> AppResult<Self> {
+        let manager = Self {
+            logos_sql: LogosSql::start(config).await?,
+        };
+
+        manager.initialize().await?;
+
+        Ok(manager)
+    }
+
+    async fn initialize(&self) -> AppResult<()> {
+        if self.schema_exists()? {
+            return Ok(());
+        }
+
+        self.logos_sql.execute(prepare_schema()).await?;
+
+        Ok(())
+    }
+
+    /// Existing participants install the schema through channel replay before
+    /// startup completes. Only an empty database needs to publish the DDL.
+    fn schema_exists(&self) -> AppResult<bool> {
+        let connection = self.logos_sql.read_connection()?;
+
+        Ok(connection.query_row(SELECT_SCHEMA_EXISTS, [], |row| row.get(0))?)
+    }
+
+    /// Adds one credential.
+    ///
+    /// Concurrent attempts to use the same label produce a primary-key
+    /// conflict.
+    pub async fn add(&self, label: String, account: String, password: String) -> AppResult<TxId> {
+        // TODO(security): Encrypt the password before it enters `λSQL`. The
+        // resulting ciphertext, salt, and nonce should be bound instead.
+        Ok(self
+            .logos_sql
+            .execute(prepare_credential_insert(&label, &account, &password))
+            .await?)
+    }
+
+    /// Replaces the password stored under one label.
+    pub async fn update_password(&self, label: String, password: String) -> AppResult<TxId> {
+        Ok(self
+            .logos_sql
+            .execute(prepare_password_update(&label, &password))
+            .await?)
+    }
+
+    /// Removes one credential.
+    pub async fn remove(&self, label: String) -> AppResult<TxId> {
+        Ok(self
+            .logos_sql
+            .execute(prepare_credential_delete(&label))
+            .await?)
+    }
+
+    /// Reads one credential from the local `SQLite` database.
+    pub fn credential(&self, label: &str) -> AppResult<Option<Credential>> {
+        let connection = self.logos_sql.read_connection()?;
+        let mut statement = connection.prepare(
+            "SELECT label, account, password
+             FROM credentials
+             WHERE label = ?1",
+        )?;
+        let mut rows = statement.query([label])?;
+
+        let Some(row) = rows.next()? else {
+            return Ok(None);
+        };
+
+        Ok(Some(Credential {
+            label: row.get("label")?,
+            account: row.get("account")?,
+            password: row.get("password")?,
+        }))
+    }
+
+    /// Lists credential labels and accounts from the local database.
+    pub fn credentials(&self) -> AppResult<Vec<CredentialSummary>> {
+        let connection = self.logos_sql.read_connection()?;
+        let mut statement = connection.prepare(SELECT_CREDENTIALS)?;
+
+        let credentials = statement
+            .query_map([], |row| {
+                Ok(CredentialSummary {
+                    label: row.get("label")?,
+                    account: row.get("account")?,
+                })
+            })?
+            .collect::<Result<Vec<_>, _>>()?;
+
+        Ok(credentials)
+    }
+
+    /// Gracefully stops the owned `λSQL` runtime.
+    pub async fn shutdown(self) -> Result<(), LogosSqlError> {
+        self.logos_sql.shutdown().await
+    }
+}

logos_sql/examples/password_manager/repl.rs

@@ -0,0 +1,149 @@
+//! Interactive terminal adapter for the password manager.
+//!
+//! Clap owns command parsing and help text. This module maps parsed commands to
+//! [`PasswordManager`] operations while leaving SQL and `λSQL` concerns in the
+//! domain module.
+
+use std::io::{self, Write as _};
+
+use clap::{Parser, Subcommand};
+use logos_sql::TxId;
+
+use crate::{
+    AppResult,
+    passwords::{Credential, CredentialSummary, PasswordManager},
+};
+
+#[derive(Debug, Parser)]
+#[command(name = "password-manager")]
+struct Input {
+    #[command(subcommand)]
+    command: Command,
+}
+
+/// A command accepted by the running password manager.
+#[derive(Debug, Subcommand)]
+enum Command {
+    /// Adds a credential.
+    Add {
+        label: String,
+        account: String,
+        #[arg(required = true, num_args = 1..)]
+        password: Vec<String>,
+    },
+    /// Replaces the password stored under a label.
+    Update {
+        label: String,
+        #[arg(required = true, num_args = 1..)]
+        password: Vec<String>,
+    },
+    /// Shows one credential from the local database.
+    Show { label: String },
+    /// Removes one credential.
+    Remove { label: String },
+    /// Lists credential labels and accounts.
+    List,
+    /// Stops the application.
+    #[command(alias = "quit")]
+    Exit,
+}
+
+impl Command {
+    /// Parses one line entered at the password-manager prompt.
+    fn parse(input: &str) -> Result<Self, clap::Error> {
+        let args = std::iter::once("password-manager").chain(input.split_whitespace());
+        Input::try_parse_from(args).map(|input| input.command)
+    }
+}
+
+/// Reads commands from the terminal until the user exits or input closes.
+pub async fn run(manager: &PasswordManager) -> AppResult<()> {
+    println!("Password manager is running. Enter `help` to list commands.");
+    println!("WARNING: passwords are replicated in plaintext; do not enter real credentials.");
+
+    while let Some(input) = read_input().await? {
+        if input.trim().is_empty() {
+            continue;
+        }
+
+        let command = match Command::parse(&input) {
+            Ok(command) => command,
+            Err(error) => {
+                error.print()?;
+                continue;
+            }
+        };
+
+        if matches!(command, Command::Exit) {
+            break;
+        }
+
+        match handle_command(manager, command).await {
+            Ok(Some(tx_id)) => {
+                println!("committed locally as {tx_id}");
+            }
+            Ok(None) => {}
+            Err(error) => eprintln!("error: {error}"),
+        }
+    }
+
+    Ok(())
+}
+
+async fn handle_command(manager: &PasswordManager, command: Command) -> AppResult<Option<TxId>> {
+    let tx_id = match command {
+        Command::Add {
+            label,
+            account,
+            password,
+        } => manager.add(label, account, password.join(" ")).await?,
+        Command::Update { label, password } => {
+            manager.update_password(label, password.join(" ")).await?
+        }
+        Command::Show { label } => {
+            print_credential(manager.credential(&label)?);
+            return Ok(None);
+        }
+        Command::Remove { label } => manager.remove(label).await?,
+        Command::List => {
+            print_credentials(manager.credentials()?);
+            return Ok(None);
+        }
+        Command::Exit => return Ok(None),
+    };
+
+    Ok(Some(tx_id))
+}
+
+/// Reads terminal input without blocking the runtime that drives `λSQL`.
+async fn read_input() -> AppResult<Option<String>> {
+    let input = tokio::task::spawn_blocking(|| -> io::Result<Option<String>> {
+        print!("password-manager> ");
+        io::stdout().flush()?;
+
+        let mut input = String::new();
+        let bytes_read = io::stdin().read_line(&mut input)?;
+
+        Ok((bytes_read > 0).then_some(input))
+    })
+    .await??;
+
+    Ok(input)
+}
+
+fn print_credential(credential: Option<Credential>) {
+    let Some(credential) = credential else {
+        println!("credential not found");
+        return;
+    };
+
+    println!("{}", credential.label);
+    println!("  account: {}", credential.account);
+    println!("  password: {}", credential.password);
+}
+
+fn print_credentials(credentials: Vec<CredentialSummary>) {
+    for credential in credentials {
+        println!("{} ({})", credential.label, credential.account);
+    }
+}

logos_sql/src/applier.rs

@@ -0,0 +1,235 @@
+//! Inbound path for applying channel history to participant-local state.
+
+use lb_zone_sdk::{
+    node_types::ChannelId,
+    sequencer::{
+        ChannelUpdate, ChannelUpdateTx, Event, FinalizedOp, FinalizedTx, channel_inscriptions,
+    },
+};
+
+use crate::{db::Databases, error::Error, protocol};
+
+const TARGET: &str = lb_log_targets::logos_sql::APPLIER;
+
+/// Handles one sequencer event.
+///
+/// Checkpoints advance over traffic that does not change `λSQL` state.
+/// Encountering an adopted, orphaned, or finalized `λSQL` transaction
+/// currently stops at the unfinished replay path.
+///
+/// # Errors
+///
+/// Returns an error if the checkpoint cannot be persisted or a channel payload
+/// cannot be decoded.
+///
+/// # Panics
+///
+/// Panics when a `λSQL` transaction reaches an unfinished apply path.
+pub fn on_event(db: &mut Databases, event: &Event, channel_id: ChannelId) -> Result<(), Error> {
+    match event {
+        Event::BlocksProcessed {
+            checkpoint,
+            channel_update,
+            finalized,
+        } => {
+            tracing::debug!(
+                target: TARGET,
+                adopted_txs = channel_update.adopted.len(),
+                orphaned_txs = channel_update.orphaned.len(),
+                finalized_txs = finalized.len(),
+                "blocks processed"
+            );
+
+            // TODO: Apply finalized transactions to LIB and reconcile LIVE with
+            // adopted and orphaned channel transactions before persisting the
+            // checkpoint.
+            if update_contains_logos_sql(channel_update, channel_id) {
+                todo!("reconcile adopted and orphaned \u{3bb}SQL transactions");
+            }
+
+            if let Some(payload) = find_logos_sql_payload(finalized) {
+                let decoded = protocol::ChannelInscription::decode(payload)?;
+
+                tracing::debug!(
+                    target: TARGET,
+                    tx_id = ?decoded.tx_id,
+                    statements = decoded.transaction.statements().len(),
+                    "\u{3bb}SQL transaction requires replay"
+                );
+
+                todo!("apply finalized \u{3bb}SQL transactions");
+            }
+
+            db.persist_checkpoint(checkpoint)?;
+        }
+        Event::Ready => {
+            tracing::info!(target: TARGET, "sequencer ready");
+        }
+        Event::MempoolPending(_) | Event::TurnNotification { .. } => {}
+    }
+
+    Ok(())
+}
+
+fn update_contains_logos_sql(channel_update: &ChannelUpdate, channel_id: ChannelId) -> bool {
+    channel_update
+        .orphaned
+        .iter()
+        .chain(&channel_update.adopted)
+        .any(|transaction| transaction_contains_logos_sql(transaction, channel_id))
+}
+
+fn transaction_contains_logos_sql(transaction: &ChannelUpdateTx, channel_id: ChannelId) -> bool {
+    if let Some(inscription) = transaction.inscription() {
+        return protocol::is_logos_sql_payload(inscription.payload.as_ref());
+    }
+
+    let ChannelUpdateTx::Custom(transaction) = transaction else {
+        return false;
+    };
+
+    channel_inscriptions(transaction, channel_id)
+        .iter()
+        .any(|inscription| protocol::is_logos_sql_payload(inscription.payload.as_ref()))
+}
+
+fn find_logos_sql_payload(finalized: &[FinalizedTx]) -> Option<&[u8]> {
+    finalized
+        .iter()
+        .flat_map(|transaction| &transaction.ops)
+        .find_map(|operation| {
+            let FinalizedOp::Inscription(info) = operation else {
+                return None;
+            };
+
+            let payload = info.payload.as_ref();
+
+            protocol::is_logos_sql_payload(payload).then_some(payload)
+        })
+}
+
+#[cfg(test)]
+mod tests {
+    use lb_zone_sdk::{
+        node_types::{ChannelId, HeaderId, MsgId, Slot, TxHash},
+        sequencer::{
+            ChannelUpdate, ChannelUpdateTx, Event, FinalizedOp, FinalizedTx, InscriptionInfo,
+            SequencerCheckpoint,
+        },
+    };
+    use tempfile::TempDir;
+
+    use super::{on_event, update_contains_logos_sql};
+    use crate::{
+        db::Databases,
+        protocol::{EncodedWrite, Statement, Transaction, Value},
+    };
+
+    fn checkpoint(byte: u8, slot: u64) -> SequencerCheckpoint {
+        SequencerCheckpoint {
+            last_msg_id: MsgId::root(),
+            pending_txs: Vec::new(),
+            lib: HeaderId::from([byte; 32]),
+            lib_slot: Slot::from(slot),
+            channel_notes: Vec::new(),
+        }
+    }
+
+    fn inscription(payload: &[u8]) -> InscriptionInfo {
+        InscriptionInfo {
+            tx_hash: TxHash::from([1; 32]),
+            parent_msg: MsgId::root(),
+            this_msg: MsgId::root(),
+            payload: payload
+                .to_vec()
+                .try_into()
+                .expect("test payload should fit an inscription"),
+        }
+    }
+
+    fn blocks_processed(checkpoint: SequencerCheckpoint, payload: &[u8]) -> Event {
+        Event::BlocksProcessed {
+            checkpoint,
+            channel_update: ChannelUpdate {
+                orphaned: Vec::new(),
+                adopted: Vec::new(),
+            },
+            finalized: vec![FinalizedTx {
+                tx_hash: TxHash::from([1; 32]),
+                l1_slot: Slot::from(2),
+                ops: vec![FinalizedOp::Inscription(inscription(payload))],
+            }],
+        }
+    }
+
+    fn orphaned(checkpoint: SequencerCheckpoint, payload: &[u8]) -> Event {
+        Event::BlocksProcessed {
+            checkpoint,
+            channel_update: ChannelUpdate {
+                orphaned: vec![ChannelUpdateTx::Inscription(inscription(payload))],
+                adopted: Vec::new(),
+            },
+            finalized: Vec::new(),
+        }
+    }
+
+    fn encoded_write() -> Vec<u8> {
+        let transaction = Transaction::new(vec![
+            Statement::new(
+                "INSERT INTO items(value) VALUES (?1)".to_owned(),
+                vec![Value::Integer(1)],
+            )
+            .expect("statement should be valid"),
+        ])
+        .expect("transaction should be valid");
+        EncodedWrite::new(&transaction)
+            .expect("payload should encode")
+            .payload
+    }
+
+    #[test]
+    #[should_panic(expected = "apply finalized \u{3bb}SQL transactions")]
+    fn finalized_logos_sql_payload_reaches_replay_placeholder() {
+        let dir = TempDir::new().expect("temporary directory should be created");
+        let path = dir.path().join("zone");
+        let first = checkpoint(1, 1);
+        let second = checkpoint(2, 2);
+        let mut db = Databases::open(&path).expect("databases should open");
+
+        db.persist_checkpoint(&first)
+            .expect("initial checkpoint should commit");
+
+        let payload = encoded_write();
+
+        let event = blocks_processed(second, &payload);
+        on_event(&mut db, &event, ChannelId::from([9; 32]))
+            .expect("event should reach replay placeholder");
+    }
+
+    #[test]
+    #[should_panic(expected = "reconcile adopted and orphaned \u{3bb}SQL transactions")]
+    fn orphaned_logos_sql_payload_reaches_reconciliation_placeholder() {
+        let dir = TempDir::new().expect("temporary directory should be created");
+        let path = dir.path().join("zone");
+        let mut db = Databases::open(&path).expect("databases should open");
+        let payload = encoded_write();
+        let event = orphaned(checkpoint(2, 2), &payload);
+
+        on_event(&mut db, &event, ChannelId::from([9; 32]))
+            .expect("event should reach reconciliation placeholder");
+    }
+
+    #[test]
+    fn adopted_logos_sql_payload_is_detected() {
+        let payload = encoded_write();
+        let channel_update = ChannelUpdate {
+            orphaned: Vec::new(),
+            adopted: vec![ChannelUpdateTx::Inscription(inscription(&payload))],
+        };
+
+        assert!(update_contains_logos_sql(
+            &channel_update,
+            ChannelId::from([9; 32])
+        ));
+    }
+}

logos_sql/src/db.rs

@@ -0,0 +1,519 @@
+//! Participant-local `SQLite` databases owned by the `λSQL` runtime.
+
+use std::{
+    fs,
+    path::{Path, PathBuf},
+    time::Duration,
+};
+
+use bincode::Options as _;
+use lb_zone_sdk::sequencer::SequencerCheckpoint;
+use rusqlite::{
+    Connection, OpenFlags, OptionalExtension as _, Row,
+    hooks::{AuthAction, AuthContext, Authorization},
+    params, params_from_iter,
+};
+
+use crate::{
+    error::Error,
+    protocol::{EncodedWrite, Transaction, TxId},
+};
+
+const DATABASE_BUSY_TIMEOUT: Duration = Duration::from_secs(5);
+const LIVE_DATABASE_FILE: &str = "LIVE.db";
+const CONTROL_DATABASE_FILE: &str = "control.db";
+
+// Stores locally committed writes and their exact channel payload. At most one
+// write may be waiting for ZoneSDK publication.
+const LIVE_SCHEMA: &str = "
+    CREATE TABLE IF NOT EXISTS __logos_sql_pending_write (
+        singleton INTEGER PRIMARY KEY CHECK (singleton = 1),
+        tx_id BLOB NOT NULL UNIQUE CHECK (length(tx_id) = 32),
+        payload BLOB NOT NULL
+    ) STRICT;
+";
+
+// Stores the participant-local ZoneSDK checkpoint independently of live state.
+const CONTROL_SCHEMA: &str = "
+    CREATE TABLE IF NOT EXISTS __logos_sql_state (
+        singleton INTEGER PRIMARY KEY CHECK (singleton = 1),
+        checkpoint BLOB
+    ) STRICT;
+";
+
+const INITIALIZE_CONTROL_STATE: &str = "
+    INSERT OR IGNORE INTO __logos_sql_state (singleton, checkpoint)
+    VALUES (1, NULL)
+";
+
+const SELECT_CHECKPOINT: &str = "
+    SELECT checkpoint
+    FROM __logos_sql_state
+    WHERE singleton = 1
+";
+
+const UPDATE_CHECKPOINT: &str = "
+    UPDATE __logos_sql_state
+    SET checkpoint = ?1
+    WHERE singleton = 1
+";
+
+const INSERT_PENDING_WRITE: &str = "
+    INSERT INTO __logos_sql_pending_write (singleton, tx_id, payload)
+    VALUES (1, ?1, ?2)
+";
+
+const SELECT_PENDING_PUBLISH: &str = "
+    SELECT tx_id, payload
+    FROM __logos_sql_pending_write
+    WHERE singleton = 1
+";
+
+const MARK_PUBLISH_COMPLETE: &str = "
+    DELETE FROM __logos_sql_pending_write
+    WHERE singleton = 1 AND tx_id = ?1
+";
+
+const WRITER_PRAGMAS: &str = "
+    PRAGMA journal_mode = WAL;
+    PRAGMA synchronous = FULL;
+";
+
+const FOREIGN_KEYS_PRAGMA: &str = "PRAGMA foreign_keys = ON;";
+
+/// Raw database representation of a write waiting for publication.
+struct StoredPendingPublish {
+    tx_id: Vec<u8>,
+    payload: Vec<u8>,
+}
+
+impl StoredPendingPublish {
+    fn from_row(row: &Row<'_>) -> rusqlite::Result<Self> {
+        Ok(Self {
+            tx_id: row.get(0)?,
+            payload: row.get(1)?,
+        })
+    }
+}
+
+/// A write committed to `LIVE.db` but not yet present in the persisted
+/// `ZoneSDK` checkpoint.
+pub struct PendingPublish {
+    pub tx_id: TxId,
+    pub payload: Vec<u8>,
+}
+
+/// Owns the participant-local database connections.
+pub struct Databases {
+    live: Connection,
+    control: Connection,
+    live_path: PathBuf,
+}
+
+impl Databases {
+    /// Opens or creates the participant state under `directory`.
+    pub(crate) fn open(directory: &Path) -> Result<Self, Error> {
+        fs::create_dir_all(directory)?;
+
+        let live_path = directory.join(LIVE_DATABASE_FILE);
+        let control_path = directory.join(CONTROL_DATABASE_FILE);
+
+        let live = open_writer(&live_path)?;
+        let control = open_writer(&control_path)?;
+
+        live.execute_batch(LIVE_SCHEMA)?;
+        control.execute_batch(CONTROL_SCHEMA)?;
+        control.execute(INITIALIZE_CONTROL_STATE, [])?;
+
+        Ok(Self {
+            live,
+            control,
+            live_path,
+        })
+    }
+
+    pub(crate) fn live_path(&self) -> &Path {
+        &self.live_path
+    }
+
+    pub(crate) fn load_checkpoint(&self) -> Result<Option<SequencerCheckpoint>, Error> {
+        let bytes = self
+            .control
+            .query_row(SELECT_CHECKPOINT, [], |row| {
+                row.get::<_, Option<Vec<u8>>>(0)
+            })
+            .optional()?
+            .flatten();
+
+        bytes
+            .map(|bytes| checkpoint_options().deserialize(&bytes))
+            .transpose()
+            .map_err(Error::from)
+    }
+
+    pub(crate) fn persist_checkpoint(
+        &mut self,
+        checkpoint: &SequencerCheckpoint,
+    ) -> Result<(), Error> {
+        let encoded = checkpoint_options().serialize(checkpoint)?;
+        let transaction = self.control.transaction()?;
+
+        transaction.execute(UPDATE_CHECKPOINT, [encoded])?;
+        transaction.commit()?;
+
+        Ok(())
+    }
+
+    /// 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<TxId, Error> {
+        if self.pending_publish()?.is_some() {
+            return Err(Error::PublishPending);
+        }
+
+        let db_transaction = self.live.transaction()?;
+
+        // Application SQL runs inside the transaction that also records its
+        // pending publication. Keep it from escaping that transaction.
+        db_transaction.authorizer(Some(authorize_application_sql));
+
+        let apply_result = transaction.statements().iter().try_for_each(|statement| {
+            // TODO: Capture nondeterministic function results and include them
+            // in the transaction published to other participants.
+            db_transaction.execute(statement.sql(), params_from_iter(statement.params()))?;
+
+            Ok::<_, Error>(())
+        });
+
+        db_transaction.authorizer(None::<fn(AuthContext<'_>) -> Authorization>);
+        apply_result?;
+
+        db_transaction.execute(
+            INSERT_PENDING_WRITE,
+            params![encoded.tx_id.as_ref(), encoded.payload],
+        )?;
+        db_transaction.commit()?;
+
+        Ok(encoded.tx_id)
+    }
+
+    pub(crate) fn pending_publish(&self) -> Result<Option<PendingPublish>, Error> {
+        let record = self
+            .live
+            .query_row(SELECT_PENDING_PUBLISH, [], StoredPendingPublish::from_row)
+            .optional()?;
+
+        let Some(record) = record else {
+            return Ok(None);
+        };
+
+        let tx_id = decode_tx_id(record.tx_id)?;
+
+        Ok(Some(PendingPublish {
+            tx_id,
+            payload: record.payload,
+        }))
+    }
+
+    pub(crate) fn mark_publish_complete(&self, tx_id: TxId) -> Result<(), Error> {
+        let changed = self.live.execute(MARK_PUBLISH_COMPLETE, [tx_id.as_ref()])?;
+
+        if changed != 1 {
+            return Err(Error::InvalidLocalState(
+                "pending publish record is missing",
+            ));
+        }
+
+        Ok(())
+    }
+
+    pub(crate) fn open_reader(path: &Path) -> Result<Connection, Error> {
+        let conn = Connection::open_with_flags(path, OpenFlags::SQLITE_OPEN_READ_ONLY)?;
+
+        configure_connection(&conn)?;
+
+        Ok(conn)
+    }
+}
+
+fn open_writer(path: &Path) -> Result<Connection, Error> {
+    let conn = Connection::open(path)?;
+
+    configure_connection(&conn)?;
+    conn.execute_batch(WRITER_PRAGMAS)?;
+
+    Ok(conn)
+}
+
+fn configure_connection(conn: &Connection) -> Result<(), Error> {
+    conn.busy_timeout(DATABASE_BUSY_TIMEOUT)?;
+    conn.execute_batch(FOREIGN_KEYS_PRAGMA)?;
+
+    Ok(())
+}
+
+// Application statements share a transaction with the pending publication
+// record, so they cannot take over transaction or connection management.
+const fn authorize_application_sql(context: AuthContext<'_>) -> Authorization {
+    let denied = matches!(
+        context.action,
+        AuthAction::Unknown { .. }
+            | AuthAction::Transaction { .. }
+            | AuthAction::Savepoint { .. }
+            | AuthAction::Attach { .. }
+            | AuthAction::Detach { .. }
+            | AuthAction::Pragma { .. }
+    );
+
+    if denied {
+        Authorization::Deny
+    } else {
+        Authorization::Allow
+    }
+}
+
+fn decode_tx_id(bytes: Vec<u8>) -> Result<TxId, Error> {
+    let bytes: [u8; 32] = bytes
+        .try_into()
+        .map_err(|_| Error::InvalidLocalState("stored transaction id is malformed"))?;
+
+    Ok(bytes.into())
+}
+
+fn checkpoint_options() -> impl bincode::Options {
+    bincode::DefaultOptions::new()
+        .with_little_endian()
+        .with_fixint_encoding()
+        .reject_trailing_bytes()
+}
+
+#[cfg(test)]
+mod tests {
+    use lb_zone_sdk::{
+        node_types::{HeaderId, MsgId, Slot},
+        sequencer::SequencerCheckpoint,
+    };
+    use tempfile::TempDir;
+
+    use super::Databases;
+    use crate::{
+        error::Error,
+        local_write,
+        protocol::{EncodedWrite, Statement, Transaction, Value},
+    };
+
+    fn checkpoint(byte: u8, slot: u64) -> SequencerCheckpoint {
+        SequencerCheckpoint {
+            last_msg_id: MsgId::root(),
+            pending_txs: Vec::new(),
+            lib: HeaderId::from([byte; 32]),
+            lib_slot: Slot::from(slot),
+            channel_notes: Vec::new(),
+        }
+    }
+
+    fn insert(value: &str) -> Transaction {
+        Transaction::new(vec![
+            Statement::new(
+                "INSERT INTO items(value) VALUES (?1)".to_owned(),
+                vec![Value::Text(value.to_owned())],
+            )
+            .expect("statement should be valid"),
+        ])
+        .expect("transaction should be valid")
+    }
+
+    fn transaction(sql: &str) -> Transaction {
+        Transaction::new(vec![
+            Statement::new(sql.to_owned(), Vec::new()).expect("statement should be valid"),
+        ])
+        .expect("transaction should be valid")
+    }
+
+    fn encoded_write(transaction: &Transaction) -> EncodedWrite {
+        EncodedWrite::new(transaction).expect("write should encode")
+    }
+
+    fn assert_application_sql_rejected(sql: &str) {
+        let dir = TempDir::new().expect("temporary directory should be created");
+        let mut db = Databases::open(dir.path()).expect("databases should open");
+        let transaction = transaction(sql);
+        let encoded = encoded_write(&transaction);
+
+        let error = db
+            .commit_local_write(&transaction, &encoded)
+            .expect_err("application SQL should be rejected");
+
+        assert!(matches!(error, Error::Database(_)));
+        assert!(
+            db.pending_publish()
+                .expect("pending publication should load")
+                .is_none()
+        );
+    }
+
+    #[test]
+    fn checkpoint_survives_reopen() {
+        let dir = TempDir::new().expect("temporary directory should be created");
+        let expected = checkpoint(7, 42);
+        let mut db = Databases::open(dir.path()).expect("databases should open");
+
+        db.persist_checkpoint(&expected)
+            .expect("checkpoint should persist");
+        drop(db);
+
+        let db = Databases::open(dir.path()).expect("databases should reopen");
+        let actual = db
+            .load_checkpoint()
+            .expect("checkpoint should load")
+            .expect("checkpoint should exist");
+
+        assert_eq!(actual.lib, expected.lib);
+        assert_eq!(actual.lib_slot, expected.lib_slot);
+    }
+
+    #[test]
+    fn application_write_and_pending_publish_commit_together() {
+        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 TEXT NOT NULL)", [])
+            .expect("application table should be created");
+
+        let transaction = insert("hello");
+        let encoded = encoded_write(&transaction);
+
+        db.commit_local_write(&transaction, &encoded)
+            .expect("write should commit");
+
+        let count: i64 = db
+            .live
+            .query_row("SELECT count(*) FROM items", [], |row| row.get(0))
+            .expect("row should be readable");
+
+        assert_eq!(count, 1);
+        assert_eq!(
+            db.pending_publish()
+                .expect("pending publish should load")
+                .expect("pending publish should exist")
+                .tx_id,
+            encoded.tx_id
+        );
+    }
+
+    #[test]
+    fn repeated_write_is_a_new_transaction() {
+        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 TEXT NOT NULL)", [])
+            .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");
+        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");
+
+        assert_ne!(second_tx_id, first_tx_id);
+
+        let count: i64 = db
+            .live
+            .query_row("SELECT count(*) FROM items", [], |row| row.get(0))
+            .expect("row should be readable");
+
+        assert_eq!(count, 2);
+    }
+
+    #[test]
+    fn application_write_can_create_persistent_schema() {
+        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)
+            .expect("schema write should commit");
+
+        db.live
+            .execute("INSERT INTO items(value) VALUES ('hello')", [])
+            .expect("created table should be writable");
+    }
+
+    #[test]
+    fn application_sql_cannot_control_wrapper_transaction() {
+        for control in ["COMMIT", "ROLLBACK", "SAVEPOINT application"] {
+            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 TEXT NOT NULL)", [])
+                .expect("application table should be created");
+
+            let transaction = Transaction::new(vec![
+                Statement::new(
+                    "INSERT INTO items(value) VALUES ('hello')".to_owned(),
+                    Vec::new(),
+                )
+                .expect("statement should be valid"),
+                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)
+                .expect_err("transaction control should be rejected");
+
+            let count: i64 = db
+                .live
+                .query_row("SELECT count(*) FROM items", [], |row| row.get(0))
+                .expect("row count should be readable");
+
+            assert_eq!(count, 0);
+            assert!(
+                db.pending_publish()
+                    .expect("pending publication should load")
+                    .is_none()
+            );
+        }
+    }
+
+    #[test]
+    fn application_sql_cannot_change_runtime_state() {
+        for sql in [
+            "PRAGMA synchronous = OFF",
+            "ATTACH DATABASE ':memory:' AS other",
+        ] {
+            assert_application_sql_rejected(sql);
+        }
+    }
+
+    #[test]
+    fn application_read_connection_is_read_only() {
+        let dir = TempDir::new().expect("temporary directory should be created");
+        let db = Databases::open(dir.path()).expect("databases should open");
+
+        db.live
+            .execute("CREATE TABLE items(value TEXT NOT NULL)", [])
+            .expect("application table should be created");
+
+        let read = Databases::open_reader(db.live_path()).expect("read connection should open");
+
+        read.query_row("SELECT count(*) FROM items", [], |row| row.get::<_, i64>(0))
+            .expect("application table should be readable");
+
+        assert!(
+            read.execute("INSERT INTO items(value) VALUES ('hello')", [])
+                .is_err()
+        );
+    }
+}

logos_sql/src/error.rs

@@ -0,0 +1,74 @@
+//! Crate error type.
+
+/// Errors returned by the `λSQL` library.
+#[derive(Debug, thiserror::Error)]
+pub enum Error {
+    /// A participant state file or directory could not be accessed.
+    #[error("I/O error: {0}")]
+    Io(#[from] std::io::Error),
+
+    /// The local `SQLite` database could not be opened or written.
+    #[error("database error: {0}")]
+    Database(#[from] rusqlite::Error),
+
+    /// Local draft serialization failed.
+    #[error("encoding error: {0}")]
+    Encoding(#[from] bincode::Error),
+
+    /// The zone sequencer reported an error.
+    #[error("sequencer error: {0}")]
+    Sequencer(#[from] lb_zone_sdk::sequencer::Error),
+
+    /// A submitted transaction is outside the currently supported shape.
+    #[error("invalid transaction: {0}")]
+    InvalidTransaction(&'static str),
+
+    /// A bound parameter is malformed.
+    #[error("invalid SQL parameter: {0}")]
+    InvalidParameter(&'static str),
+
+    /// A bound parameter uses a `SQLite` representation that cannot be
+    /// replayed.
+    #[error("SQL parameter representation is not supported by \u{3bb}SQL")]
+    UnsupportedParameter,
+
+    /// Participant-local bookkeeping is missing or malformed.
+    #[error("invalid local state: {0}")]
+    InvalidLocalState(&'static str),
+
+    /// A local write is committed but has not yet been accepted by `ZoneSDK`.
+    #[error("a committed write is still waiting for ZoneSDK to accept it")]
+    PublishPending,
+
+    /// The encoded transaction does not conform to the `λSQL` protocol.
+    #[error("invalid \u{3bb}SQL payload: {0}")]
+    InvalidPayload(&'static str),
+
+    /// The inscription uses a protocol version this library cannot execute.
+    #[error("unsupported \u{3bb}SQL protocol version {0}")]
+    UnsupportedProtocolVersion(u16),
+
+    /// The encoded payload exceeds the inscription limit.
+    #[error("transaction is too large for one inscription")]
+    InscriptionTooLarge,
+
+    /// No Tokio runtime is active on the calling thread.
+    #[error("LogosSql::start must be called from within a Tokio runtime")]
+    RuntimeUnavailable,
+
+    /// The sequencer has not completed backfill and cannot accept writes yet.
+    #[error("the zone sequencer is not ready to accept writes")]
+    SequencerNotReady,
+
+    /// Chain application has halted and new local writes are unsafe.
+    #[error("\u{3bb}SQL runtime is halted while applying channel history")]
+    RuntimeHalted,
+
+    /// The participant runtime stopped before it could process a command.
+    #[error("\u{3bb}SQL runtime is not running")]
+    RuntimeStopped,
+
+    /// The runtime task could not be joined.
+    #[error("runtime task failed: {0}")]
+    RuntimeJoin(#[from] tokio::task::JoinError),
+}

logos_sql/src/lib.rs

@@ -0,0 +1,21 @@
+//! `λSQL`: replicated `SQLite` state over Logos Blockchain.
+//!
+//! Applications read through a normal `SQLite` connection and submit replicated
+//! writes through [`LogosSql::execute`]. One runtime task owns the zone
+//! sequencer and database writer, so the SQL effects and pending publication
+//! record commit together before the payload is given to `ZoneSDK`.
+
+mod applier;
+mod db;
+mod error;
+mod local_write;
+mod logos_sql;
+mod protocol;
+mod runtime;
+mod sql;
+
+pub use error::Error;
+pub use logos_sql::{LogosSql, LogosSqlConfig};
+pub use protocol::TxId;
+pub use rusqlite::types::ToSql;
+pub use sql::TransactionBuilder;

logos_sql/src/local_write.rs

@@ -0,0 +1,21 @@
+//! Outbound write path for transactions initiated by the local application.
+//!
+//! Transactions arriving through channel history are handled by the applier.
+
+use lb_zone_sdk::node_types::Inscription;
+
+use crate::{
+    db::Databases,
+    error::Error,
+    protocol::{EncodedWrite, Transaction, TxId},
+};
+
+pub fn commit(db: &mut Databases, transaction: &Transaction) -> Result<TxId, Error> {
+    let encoded = EncodedWrite::new(transaction)?;
+
+    if encoded.payload.len() > Inscription::MAX {
+        return Err(Error::InscriptionTooLarge);
+    }
+
+    db.commit_local_write(transaction, &encoded)
+}

logos_sql/src/logos_sql.rs

@@ -0,0 +1,148 @@
+//! Application entry point: a running `λSQL` participant.
+
+use std::path::PathBuf;
+
+use lb_key_management_system_service::keys::Ed25519Key;
+use lb_zone_sdk::{
+    CommonHttpClient,
+    adapter::NodeHttpClient,
+    node_types::ChannelId,
+    sequencer::{FundingConfig, ZoneSequencer},
+};
+use reqwest::Url;
+use rusqlite::Connection;
+
+use crate::{db::Databases, error::Error, protocol::TxId, runtime, sql::TransactionBuilder};
+
+/// Configuration for one `λSQL` database.
+pub struct LogosSqlConfig {
+    /// Channel carrying the write log.
+    pub channel_id: ChannelId,
+    /// Key used to sign published inscriptions.
+    pub signing_key: Ed25519Key,
+    /// Base URL of the node HTTP API.
+    pub node_url: Url,
+    /// Fee funding for published transactions.
+    pub funding: FundingConfig,
+    /// Directory containing this participant's local databases.
+    pub state_dir: PathBuf,
+}
+
+/// A running `λSQL` database.
+///
+/// `LogosSql` owns one background task. That task is the only owner of both the
+/// `ZoneSDK` sequencer and the database writer. Dropping `LogosSql` aborts the
+/// task; call [`Self::shutdown`] to stop it gracefully and observe errors.
+pub struct LogosSql {
+    live_path: PathBuf,
+    runtime: Option<runtime::RuntimeHandle>,
+}
+
+impl LogosSql {
+    /// Opens local state, starts replication, and waits for the initial channel
+    /// history to be processed.
+    ///
+    /// Must be called from within a tokio runtime.
+    ///
+    /// # Errors
+    ///
+    /// Returns an error if no Tokio runtime is active, the local state cannot
+    /// be opened, or the replication task stops before becoming ready.
+    pub async fn start(config: LogosSqlConfig) -> Result<Self, Error> {
+        tokio::runtime::Handle::try_current().map_err(|_| Error::RuntimeUnavailable)?;
+
+        let db = Databases::open(&config.state_dir)?;
+        let live_path = db.live_path().to_owned();
+        let checkpoint = db.load_checkpoint()?;
+        let node = NodeHttpClient::new(CommonHttpClient::new(None), config.node_url);
+
+        let sequencer = ZoneSequencer::init(
+            config.channel_id,
+            config.signing_key,
+            node,
+            config.funding,
+            checkpoint.clone(),
+        );
+
+        let runtime = runtime::spawn(sequencer, db, config.channel_id, checkpoint);
+
+        let mut logos_sql = Self {
+            live_path,
+            runtime: Some(runtime),
+        };
+
+        logos_sql
+            .runtime
+            .as_mut()
+            .ok_or(Error::RuntimeStopped)?
+            .wait_until_ready()
+            .await?;
+
+        Ok(logos_sql)
+    }
+
+    /// Executes a prepared SQL transaction locally and submits it for
+    /// publication.
+    ///
+    /// ```no_run
+    /// # use logos_sql::{Error, LogosSql, TransactionBuilder, TxId};
+    /// # async fn create_task(logos_sql: &LogosSql) -> Result<TxId, Error> {
+    /// let transaction = TransactionBuilder::new(
+    ///     "INSERT INTO tasks (id, title) VALUES (?1, ?2)",
+    /// )
+    ///     .bind(42i64)
+    ///     .bind("Write documentation");
+    ///
+    /// logos_sql.execute(transaction).await
+    /// # }
+    /// ```
+    ///
+    /// A successful return means the SQL effects and recovery record are
+    /// committed locally. Publication and finality remain asynchronous.
+    ///
+    /// # Errors
+    ///
+    /// Returns an error when a parameter cannot be represented by the `λSQL`
+    /// protocol, validation or the local commit fails, the sequencer is not
+    /// ready, or the runtime has halted.
+    pub async fn execute(&self, transaction: TransactionBuilder) -> Result<TxId, Error> {
+        let transaction = transaction.finish()?;
+
+        self.runtime
+            .as_ref()
+            .ok_or(Error::RuntimeStopped)?
+            .execute(transaction)
+            .await
+    }
+
+    /// Opens a read-only connection to the replicated database.
+    ///
+    /// # Errors
+    ///
+    /// Returns an error when the database file cannot be opened.
+    pub fn read_connection(&self) -> Result<Connection, Error> {
+        Databases::open_reader(&self.live_path)
+    }
+
+    /// Stops the runtime after its current atomic operation and waits for it.
+    ///
+    /// # Errors
+    ///
+    /// Returns the runtime error if the task had already failed, or a join
+    /// error if the task was cancelled or panicked.
+    pub async fn shutdown(mut self) -> Result<(), Error> {
+        if let Some(runtime) = self.runtime.take() {
+            runtime.shutdown().await?;
+        }
+
+        Ok(())
+    }
+}
+
+impl Drop for LogosSql {
+    fn drop(&mut self) {
+        if let Some(runtime) = &self.runtime {
+            runtime.abort();
+        }
+    }
+}

logos_sql/src/protocol/mod.rs

@@ -0,0 +1,281 @@
+//! SQL transactions exchanged by `λSQL` instances.
+
+use std::fmt::{self, Display, Formatter};
+
+use bincode::Options as _;
+use rand::RngCore as _;
+pub use rusqlite::types::Value;
+use serde::{Deserialize, Serialize};
+
+use crate::error::Error;
+
+const PAYLOAD_MARKER: [u8; 9] = *b"LOGOS_SQL";
+const PAYLOAD_VERSION: u16 = 1;
+const PAYLOAD_HEADER_LEN: usize = PAYLOAD_MARKER.len() + size_of::<u16>();
+
+/// Stable identity of one application write.
+#[derive(Clone, Copy, Debug, Eq, Hash, PartialEq, Serialize, Deserialize)]
+pub struct TxId([u8; 32]);
+
+impl Display for TxId {
+    fn fmt(&self, formatter: &mut Formatter<'_>) -> fmt::Result {
+        for byte in self.0 {
+            write!(formatter, "{byte:02x}")?;
+        }
+
+        Ok(())
+    }
+}
+
+impl TxId {
+    fn generate() -> Self {
+        let mut bytes = [0; 32];
+        rand::rngs::OsRng.fill_bytes(&mut bytes);
+
+        Self(bytes)
+    }
+}
+
+impl From<[u8; 32]> for TxId {
+    fn from(bytes: [u8; 32]) -> Self {
+        Self(bytes)
+    }
+}
+
+impl AsRef<[u8; 32]> for TxId {
+    fn as_ref(&self) -> &[u8; 32] {
+        &self.0
+    }
+}
+
+impl From<TxId> for [u8; 32] {
+    fn from(tx_id: TxId) -> Self {
+        tx_id.0
+    }
+}
+
+/// One parameterized SQL statement.
+#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
+pub struct Statement {
+    sql: String,
+    #[serde(with = "serialized_values")]
+    params: Vec<Value>,
+}
+
+impl Statement {
+    /// Creates one non-empty statement.
+    ///
+    /// # Errors
+    ///
+    /// Returns an error if `sql` is empty.
+    pub fn new(sql: String, params: Vec<Value>) -> Result<Self, Error> {
+        if sql.trim().is_empty() {
+            return Err(Error::InvalidTransaction("statement SQL must not be empty"));
+        }
+
+        Ok(Self { sql, params })
+    }
+
+    /// Returns the SQL text.
+    #[must_use]
+    pub fn sql(&self) -> &str {
+        &self.sql
+    }
+
+    /// Returns parameters in `SQLite` binding order.
+    #[must_use]
+    pub fn params(&self) -> &[Value] {
+        &self.params
+    }
+}
+
+mod serialized_values {
+    use serde::{Deserialize, Deserializer, Serialize, Serializer};
+
+    use super::Value;
+
+    /// Serializable representation of a `rusqlite` parameter value.
+    #[derive(Deserialize, Serialize)]
+    enum SqlValue {
+        Null,
+        Integer(i64),
+        Real(f64),
+        Text(String),
+        Blob(Vec<u8>),
+    }
+
+    impl From<&Value> for SqlValue {
+        fn from(value: &Value) -> Self {
+            match value {
+                Value::Null => Self::Null,
+                Value::Integer(value) => Self::Integer(*value),
+                Value::Real(value) => Self::Real(*value),
+                Value::Text(value) => Self::Text(value.clone()),
+                Value::Blob(value) => Self::Blob(value.clone()),
+            }
+        }
+    }
+
+    impl From<SqlValue> for Value {
+        fn from(value: SqlValue) -> Self {
+            match value {
+                SqlValue::Null => Self::Null,
+                SqlValue::Integer(value) => Self::Integer(value),
+                SqlValue::Real(value) => Self::Real(value),
+                SqlValue::Text(value) => Self::Text(value),
+                SqlValue::Blob(value) => Self::Blob(value),
+            }
+        }
+    }
+
+    pub fn serialize<S>(values: &[Value], serializer: S) -> Result<S::Ok, S::Error>
+    where
+        S: Serializer,
+    {
+        values
+            .iter()
+            .map(SqlValue::from)
+            .collect::<Vec<_>>()
+            .serialize(serializer)
+    }
+
+    pub fn deserialize<'de, D>(deserializer: D) -> Result<Vec<Value>, D::Error>
+    where
+        D: Deserializer<'de>,
+    {
+        Vec::<SqlValue>::deserialize(deserializer)
+            .map(|values| values.into_iter().map(Value::from).collect())
+    }
+}
+
+/// Statements applied atomically at one channel position.
+#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
+pub struct Transaction {
+    statements: Vec<Statement>,
+}
+
+impl Transaction {
+    /// Creates a non-empty transaction.
+    ///
+    /// # Errors
+    ///
+    /// Returns an error if no statements are provided.
+    pub fn new(statements: Vec<Statement>) -> Result<Self, Error> {
+        if statements.is_empty() {
+            return Err(Error::InvalidTransaction(
+                "transaction must contain a statement",
+            ));
+        }
+
+        Ok(Self { statements })
+    }
+
+    /// Returns statements in execution order.
+    #[must_use]
+    pub fn statements(&self) -> &[Statement] {
+        &self.statements
+    }
+}
+
+/// Transaction payload carried by a `λSQL` channel inscription.
+#[derive(Serialize, Deserialize)]
+pub struct ChannelInscription {
+    pub tx_id: TxId,
+    pub transaction: Transaction,
+}
+
+impl ChannelInscription {
+    fn encode(&self) -> Result<Vec<u8>, Error> {
+        let mut payload = Vec::from(PAYLOAD_MARKER);
+        payload.extend_from_slice(&PAYLOAD_VERSION.to_le_bytes());
+        payload.extend(codec().serialize(self)?);
+
+        Ok(payload)
+    }
+
+    pub fn decode(payload: &[u8]) -> Result<Self, Error> {
+        let (header, body) = payload
+            .split_at_checked(PAYLOAD_HEADER_LEN)
+            .ok_or(Error::InvalidPayload("header is missing"))?;
+
+        if header[..PAYLOAD_MARKER.len()] != PAYLOAD_MARKER {
+            return Err(Error::InvalidPayload("protocol marker does not match"));
+        }
+
+        let version_offset = PAYLOAD_MARKER.len();
+        let version = u16::from_le_bytes([header[version_offset], header[version_offset + 1]]);
+
+        if version != PAYLOAD_VERSION {
+            return Err(Error::UnsupportedProtocolVersion(version));
+        }
+
+        codec()
+            .deserialize(body)
+            .map_err(|_| Error::InvalidPayload("body cannot be decoded"))
+    }
+}
+
+/// A local write after its identity and channel payload have been encoded.
+pub struct EncodedWrite {
+    pub tx_id: TxId,
+    pub payload: Vec<u8>,
+}
+
+impl EncodedWrite {
+    pub fn new(transaction: &Transaction) -> Result<Self, Error> {
+        let tx_id = TxId::generate();
+
+        let channel_inscription = ChannelInscription {
+            tx_id,
+            transaction: transaction.clone(),
+        };
+
+        let payload = channel_inscription.encode()?;
+
+        Ok(Self { tx_id, payload })
+    }
+}
+
+/// Returns whether an inscription belongs to `λSQL`.
+#[must_use]
+pub fn is_logos_sql_payload(payload: &[u8]) -> bool {
+    payload.starts_with(&PAYLOAD_MARKER)
+}
+
+fn codec() -> impl bincode::Options {
+    bincode::DefaultOptions::new()
+        .with_little_endian()
+        .with_fixint_encoding()
+        .reject_trailing_bytes()
+}
+
+#[cfg(test)]
+mod tests {
+    use super::{ChannelInscription, EncodedWrite, Statement, Transaction, TxId, Value};
+
+    #[test]
+    fn transaction_id_is_displayed_as_hex() {
+        let tx_id = TxId::from([0xab; 32]);
+
+        assert_eq!(tx_id.to_string(), "ab".repeat(32));
+    }
+
+    #[test]
+    fn channel_inscription_round_trips() {
+        let transaction = Transaction::new(vec![
+            Statement::new(
+                "INSERT INTO messages VALUES (?1)".to_owned(),
+                vec![Value::Text("hello".to_owned())],
+            )
+            .expect("statement should be valid"),
+        ])
+        .expect("transaction should be valid");
+
+        let encoded = EncodedWrite::new(&transaction).expect("submission should encode");
+        let decoded = ChannelInscription::decode(&encoded.payload)
+            .expect("channel inscription should decode");
+
+        assert_eq!(decoded.tx_id, encoded.tx_id);
+        assert_eq!(decoded.transaction, transaction);
+    }
+}

logos_sql/src/runtime.rs

@@ -0,0 +1,352 @@
+//! Single-owner runtime for SQL writes and channel events.
+
+use std::time::Duration;
+
+use lb_zone_sdk::{
+    adapter::NodeHttpClient,
+    node_types::{ChannelId, Inscription},
+    sequencer::{Event, SequencerCheckpoint, ZoneSequencer, channel_inscriptions},
+};
+use tokio::{
+    sync::{mpsc, oneshot},
+    task::JoinHandle,
+};
+
+use crate::{
+    applier,
+    db::Databases,
+    error::Error,
+    local_write,
+    protocol::{Transaction, TxId},
+};
+
+const COMMAND_CHANNEL_CAPACITY: usize = 16;
+const PUBLISH_RETRY_INTERVAL: Duration = Duration::from_secs(5);
+const TARGET: &str = lb_log_targets::logos_sql::RUNTIME;
+
+/// Requests processed by the task that owns the sequencer and database writer.
+enum Command {
+    Execute {
+        transaction: Transaction,
+        response_tx: oneshot::Sender<Result<TxId, Error>>,
+    },
+    Shutdown,
+}
+
+/// Control surface for the owning runtime task.
+pub struct RuntimeHandle {
+    command_tx: mpsc::Sender<Command>,
+    ready_rx: oneshot::Receiver<()>,
+    task: JoinHandle<Result<(), Error>>,
+}
+
+/// Starts the task that owns the sequencer and writable database connections.
+pub fn spawn(
+    sequencer: ZoneSequencer<NodeHttpClient>,
+    db: Databases,
+    channel_id: ChannelId,
+    restored_checkpoint: Option<SequencerCheckpoint>,
+) -> RuntimeHandle {
+    let (command_tx, command_rx) = mpsc::channel(COMMAND_CHANNEL_CAPACITY);
+    let (ready_tx, ready_rx) = oneshot::channel();
+
+    let runtime = Runtime {
+        sequencer,
+        db,
+        channel_id,
+        command_rx,
+        sequencer_ready: false,
+        ready_tx: Some(ready_tx),
+        event_pending_retry: None,
+        publish_state: PublishState::Idle,
+    };
+    let task = tokio::spawn(runtime.run(restored_checkpoint));
+
+    RuntimeHandle {
+        command_tx,
+        ready_rx,
+        task,
+    }
+}
+
+impl RuntimeHandle {
+    pub(crate) async fn wait_until_ready(&mut self) -> Result<(), Error> {
+        tokio::select! {
+            biased;
+
+            result = &mut self.task => {
+                match result? {
+                    Ok(()) => Err(Error::RuntimeStopped),
+                    Err(error) => Err(error),
+                }
+            }
+            result = &mut self.ready_rx => {
+                result.map_err(|_| Error::RuntimeStopped)
+            }
+        }
+    }
+
+    pub(crate) async fn execute(&self, transaction: Transaction) -> Result<TxId, Error> {
+        let (response_tx, response_rx) = oneshot::channel();
+        self.command_tx
+            .send(Command::Execute {
+                transaction,
+                response_tx,
+            })
+            .await
+            .map_err(|_| Error::RuntimeStopped)?;
+
+        response_rx.await.map_err(|_| Error::RuntimeStopped)?
+    }
+
+    pub(crate) async fn shutdown(self) -> Result<(), Error> {
+        drop(self.command_tx.send(Command::Shutdown).await);
+
+        self.task.await?
+    }
+
+    pub(crate) fn abort(&self) {
+        self.task.abort();
+    }
+}
+
+/// A `ZoneSDK` publish whose returned checkpoint may still need to be
+/// persisted.
+enum PublishState {
+    Idle,
+    CheckpointPending {
+        tx_id: TxId,
+        checkpoint: SequencerCheckpoint,
+    },
+}
+
+/// State owned exclusively by the participant's background task.
+struct Runtime {
+    sequencer: ZoneSequencer<NodeHttpClient>,
+    db: Databases,
+    channel_id: ChannelId,
+    command_rx: mpsc::Receiver<Command>,
+    sequencer_ready: bool,
+    ready_tx: Option<oneshot::Sender<()>>,
+    event_pending_retry: Option<Event>,
+    publish_state: PublishState,
+}
+
+impl Runtime {
+    async fn run(mut self, restored_checkpoint: Option<SequencerCheckpoint>) -> Result<(), Error> {
+        self.reconcile_restored_publish(restored_checkpoint.as_ref())?;
+
+        let mut retry = tokio::time::interval(PUBLISH_RETRY_INTERVAL);
+        retry.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
+
+        loop {
+            let retry_needed = self.has_pending_work()?;
+
+            tokio::select! {
+                command = self.command_rx.recv() => {
+                    let Some(command) = command else {
+                        return Ok(());
+                    };
+
+                    if !self.handle_command(command).await {
+                        return Ok(());
+                    }
+                },
+                event = self.sequencer.next_event(), if self.event_pending_retry.is_none() && !matches!(self.publish_state, PublishState::CheckpointPending { .. }) => {
+                    self.handle_event(event).await;
+                },
+                _ = retry.tick(), if retry_needed => {
+                    self.retry_pending_work().await?;
+                }
+            }
+        }
+    }
+
+    async fn handle_command(&mut self, command: Command) -> bool {
+        let Command::Execute {
+            transaction,
+            response_tx,
+        } = command
+        else {
+            return false;
+        };
+
+        let result = if self.event_pending_retry.is_some() {
+            Err(Error::RuntimeHalted)
+        } else if !self.sequencer_ready {
+            Err(Error::SequencerNotReady)
+        } else {
+            let committed = local_write::commit(&mut self.db, &transaction);
+
+            if let Ok(tx_id) = committed {
+                tracing::trace!(
+                    target: TARGET,
+                    ?tx_id,
+                    statements = transaction.statements().len(),
+                    "local write committed"
+                );
+
+                if let Err(error) = self.advance_publish().await {
+                    tracing::warn!(
+                        target: TARGET,
+                        %error,
+                        ?tx_id,
+                        "write committed; ZoneSDK publish remains pending"
+                    );
+                }
+            }
+
+            committed
+        };
+
+        drop(response_tx.send(result));
+
+        true
+    }
+
+    async fn handle_event(&mut self, event: Event) {
+        match applier::on_event(&mut self.db, &event, self.channel_id) {
+            Ok(()) => {
+                self.mark_ready(&event);
+
+                if self.sequencer_ready
+                    && let Err(error) = self.advance_publish().await
+                {
+                    tracing::warn!(
+                        target: TARGET,
+                        %error,
+                        "ZoneSDK publish remains pending"
+                    );
+                }
+            }
+            Err(error) => {
+                tracing::error!(target: TARGET, %error, "applier halted");
+                self.event_pending_retry = Some(event);
+            }
+        }
+    }
+
+    async fn retry_pending_work(&mut self) -> Result<(), Error> {
+        if let Some(event) = self.event_pending_retry.take() {
+            if let Err(error) = applier::on_event(&mut self.db, &event, self.channel_id) {
+                if !is_retryable_apply_error(&error) {
+                    return Err(error);
+                }
+
+                tracing::debug!(target: TARGET, %error, "applier retry failed");
+                self.event_pending_retry = Some(event);
+            } else {
+                self.mark_ready(&event);
+            }
+        } else if self.sequencer_ready
+            && let Err(error) = self.advance_publish().await
+        {
+            tracing::debug!(target: TARGET, %error, "ZoneSDK publish retry failed");
+        }
+
+        Ok(())
+    }
+
+    fn mark_ready(&mut self, event: &Event) {
+        if self.sequencer_ready || !matches!(event, Event::Ready) {
+            return;
+        }
+
+        self.sequencer_ready = true;
+
+        if let Some(ready_tx) = self.ready_tx.take() {
+            let _ = ready_tx.send(());
+        }
+    }
+
+    async fn advance_publish(&mut self) -> Result<(), Error> {
+        self.persist_publish_checkpoint()?;
+
+        let Some(pending) = self.db.pending_publish()? else {
+            return Ok(());
+        };
+
+        let inscription: Inscription = pending
+            .payload
+            .try_into()
+            .map_err(|_| Error::InscriptionTooLarge)?;
+
+        let (_, checkpoint) = self.sequencer.handle().publish(inscription).await?;
+
+        tracing::trace!(
+            target: TARGET,
+            tx_id = ?pending.tx_id,
+            "write accepted by ZoneSDK"
+        );
+
+        self.publish_state = PublishState::CheckpointPending {
+            tx_id: pending.tx_id,
+            checkpoint,
+        };
+
+        self.persist_publish_checkpoint()
+    }
+
+    fn persist_publish_checkpoint(&mut self) -> Result<(), Error> {
+        let PublishState::CheckpointPending { tx_id, checkpoint } = &self.publish_state else {
+            return Ok(());
+        };
+
+        self.db.persist_checkpoint(checkpoint)?;
+        self.db.mark_publish_complete(*tx_id)?;
+
+        tracing::trace!(
+            target: TARGET,
+            ?tx_id,
+            "write publication recorded"
+        );
+
+        self.publish_state = PublishState::Idle;
+
+        Ok(())
+    }
+
+    fn reconcile_restored_publish(
+        &self,
+        checkpoint: Option<&SequencerCheckpoint>,
+    ) -> Result<(), Error> {
+        let Some(pending) = self.db.pending_publish()? else {
+            return Ok(());
+        };
+
+        let Some(checkpoint) = checkpoint else {
+            return Ok(());
+        };
+
+        let already_submitted = checkpoint.pending_txs.iter().any(|(_, transaction)| {
+            channel_inscriptions(transaction, self.channel_id)
+                .iter()
+                .any(|inscription| inscription.payload.as_ref() == pending.payload)
+        });
+
+        if already_submitted {
+            self.db.mark_publish_complete(pending.tx_id)?;
+
+            tracing::debug!(
+                target: TARGET,
+                tx_id = ?pending.tx_id,
+                "restored ZoneSDK checkpoint matched pending write"
+            );
+        }
+
+        Ok(())
+    }
+
+    fn has_pending_work(&self) -> Result<bool, Error> {
+        Ok(self.event_pending_retry.is_some()
+            || matches!(self.publish_state, PublishState::CheckpointPending { .. })
+            || self.db.pending_publish()?.is_some())
+    }
+}
+
+const fn is_retryable_apply_error(error: &Error) -> bool {
+    !matches!(
+        error,
+        Error::InvalidPayload(_) | Error::UnsupportedProtocolVersion(_)
+    )
+}

logos_sql/src/sql.rs

@@ -0,0 +1,192 @@
+//! Construction of replicated SQL transactions.
+
+use rusqlite::types::{ToSql, ToSqlOutput, Value, ValueRef};
+
+use crate::{
+    error::Error,
+    protocol::{Statement, Transaction},
+};
+
+/// A replicated SQL transaction with a current query.
+///
+/// Parameter values are converted immediately into owned `SQLite` values, so
+/// borrowed application data does not need to outlive the call to
+/// [`Self::bind`].
+#[must_use = "the transaction must be passed to LogosSql::execute"]
+pub struct TransactionBuilder {
+    transaction: TransactionDraft,
+}
+
+struct TransactionDraft {
+    statements: Vec<Statement>,
+    current: PendingStatement,
+    error: Option<Error>,
+}
+
+impl TransactionBuilder {
+    /// Starts a transaction with its first SQL query.
+    pub fn new(sql: impl Into<String>) -> Self {
+        Self {
+            transaction: TransactionDraft::new(sql),
+        }
+    }
+
+    /// Adds the next SQL query to this transaction.
+    pub fn query(mut self, sql: impl Into<String>) -> Self {
+        self.transaction.query(sql);
+
+        self
+    }
+
+    /// Binds one parameter to the current statement.
+    ///
+    /// Values use [`rusqlite::types::ToSql`], the same conversion interface as
+    /// ordinary `SQLite` writes. Any conversion error is returned by
+    /// [`crate::LogosSql::execute`].
+    pub fn bind<T>(mut self, value: T) -> Self
+    where
+        T: ToSql,
+    {
+        self.transaction.bind(&value);
+
+        self
+    }
+
+    pub(crate) fn finish(self) -> Result<Transaction, Error> {
+        self.transaction.finish()
+    }
+}
+
+impl TransactionDraft {
+    fn new(sql: impl Into<String>) -> Self {
+        Self {
+            statements: Vec::new(),
+            current: PendingStatement::new(sql),
+            error: None,
+        }
+    }
+
+    fn query(&mut self, sql: impl Into<String>) {
+        if self.error.is_some() {
+            return;
+        }
+
+        let current = std::mem::replace(&mut self.current, PendingStatement::new(sql));
+
+        match current.finish() {
+            Ok(statement) => self.statements.push(statement),
+            Err(error) => self.error = Some(error),
+        }
+    }
+
+    fn bind(&mut self, value: &impl ToSql) {
+        if self.error.is_some() {
+            return;
+        }
+
+        match to_owned_value(value) {
+            Ok(value) => self.current.params.push(value),
+            Err(error) => self.error = Some(error),
+        }
+    }
+
+    fn finish(self) -> Result<Transaction, Error> {
+        if let Some(error) = self.error {
+            return Err(error);
+        }
+
+        let mut statements = self.statements;
+        statements.push(self.current.finish()?);
+
+        Transaction::new(statements)
+    }
+}
+
+struct PendingStatement {
+    sql: String,
+    params: Vec<Value>,
+}
+
+impl PendingStatement {
+    fn new(sql: impl Into<String>) -> Self {
+        Self {
+            sql: sql.into(),
+            params: Vec::new(),
+        }
+    }
+
+    fn finish(self) -> Result<Statement, Error> {
+        Statement::new(self.sql, self.params)
+    }
+}
+
+fn to_owned_value(value: &impl ToSql) -> Result<Value, Error> {
+    match value.to_sql()? {
+        ToSqlOutput::Borrowed(value) => owned_value_ref(value),
+        ToSqlOutput::Owned(value) => Ok(value),
+        _ => Err(Error::UnsupportedParameter),
+    }
+}
+
+fn owned_value_ref(value: ValueRef<'_>) -> Result<Value, Error> {
+    match value {
+        ValueRef::Null => Ok(Value::Null),
+        ValueRef::Integer(value) => Ok(Value::Integer(value)),
+        ValueRef::Real(value) => Ok(Value::Real(value)),
+        ValueRef::Text(value) => String::from_utf8(value.to_vec())
+            .map(Value::Text)
+            .map_err(|_| Error::InvalidParameter("text parameter is not valid UTF-8")),
+        ValueRef::Blob(value) => Ok(Value::Blob(value.to_vec())),
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use rusqlite::types::Value;
+
+    use super::{Statement, Transaction, TransactionBuilder, to_owned_value};
+
+    #[test]
+    fn assembles_statements_and_parameters_in_order() {
+        let transaction =
+            TransactionBuilder::new("INSERT INTO credentials (label, account) VALUES (?1, ?2)")
+                .bind("email")
+                .bind("andrus@example.org")
+                .query("UPDATE credentials SET password = ?2 WHERE label = ?1")
+                .bind("email")
+                .bind(42i64)
+                .finish()
+                .unwrap();
+        let expected = Transaction::new(vec![
+            Statement::new(
+                "INSERT INTO credentials (label, account) VALUES (?1, ?2)".to_owned(),
+                vec![
+                    Value::Text("email".to_owned()),
+                    Value::Text("andrus@example.org".to_owned()),
+                ],
+            )
+            .unwrap(),
+            Statement::new(
+                "UPDATE credentials SET password = ?2 WHERE label = ?1".to_owned(),
+                vec![Value::Text("email".to_owned()), Value::Integer(42)],
+            )
+            .unwrap(),
+        ])
+        .unwrap();
+
+        assert_eq!(transaction, expected);
+    }
+
+    #[test]
+    fn converts_sql_parameters_to_owned_values() {
+        let text = String::from("hello");
+
+        assert_eq!(to_owned_value(&42i64).unwrap(), Value::Integer(42));
+        assert_eq!(to_owned_value(&text.as_str()).unwrap(), Value::Text(text));
+    }
+
+    #[test]
+    fn converts_none_to_sql_null() {
+        assert_eq!(to_owned_value(&Option::<i64>::None).unwrap(), Value::Null);
+    }
+}

tracing/targets/src/lib.rs

@@ -7,6 +7,7 @@ log_target_modules! {
     blend,
     http_client,
     kms,
+    logos_sql,
     libp2p,
     mantle,
     mempool,

tracing/targets/src/logos_sql.rs

@@ -0,0 +1,8 @@
+use lb_log_targets_macros::log_targets;
+
+log_targets! {
+    root = logos_sql;
+
+    APPLIER,
+    RUNTIME,
+}