feat(lsql): apply channel writes (#3333)

Cargo.lock

@@ -5477,6 +5477,7 @@ name = "logos-sql"
 version = "0.0.0"
 dependencies = [
  "bincode",
+ "blake2",
  "clap",
  "hex",
  "logos-blockchain-groth16",

logos_sql/Cargo.toml

@@ -15,6 +15,7 @@ workspace = true
 
 [dependencies]
 bincode                          = { workspace = true }
+blake2                           = { workspace = true }
 lb-key-management-system-service = { workspace = true }
 lb-log-targets                   = { workspace = true }
 lb-zone-sdk                      = { workspace = true }

logos_sql/src/applier.rs

@@ -3,28 +3,51 @@
 use lb_zone_sdk::{
     node_types::ChannelId,
     sequencer::{
-        ChannelUpdate, ChannelUpdateTx, Event, FinalizedOp, FinalizedTx, channel_inscriptions,
+        ChannelUpdate, ChannelUpdateTx, Event, FinalizedOp, FinalizedTx, InscriptionInfo,
+        channel_inscriptions,
     },
 };
 
-use crate::{db::Databases, error::Error, protocol};
+use crate::{
+    db::Databases,
+    error::Error,
+    protocol::{self, ChannelInscription, TxId},
+};
 
 const TARGET: &str = lb_log_targets::logos_sql::APPLIER;
 
+#[derive(Clone, Copy, Debug)]
+enum WriteState {
+    Adopted,
+    Finalized,
+}
+
+impl WriteState {
+    fn apply(self, db: &mut Databases, write: &ChannelInscription) -> Result<(), Error> {
+        match self {
+            Self::Adopted => db.apply_adopted_write(write),
+            Self::Finalized => db.apply_finalized_write(write),
+        }
+    }
+}
+
 /// 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.
+/// Every `BlocksProcessed` checkpoint is persisted, including events that
+/// contain no `λSQL` writes, so restart resumes from the latest processed
+/// block.
+/// Finalized writes apply to both finalized and live state, while newly
+/// adopted writes apply only to live state. Orphan recovery is deferred.
 ///
 /// # Errors
 ///
-/// Returns an error if the checkpoint cannot be persisted or a channel payload
-/// cannot be decoded.
+/// Returns an error if SQL cannot be applied, a channel payload cannot be
+/// decoded, or the checkpoint cannot be persisted.
 ///
 /// # Panics
 ///
-/// Panics when a `λSQL` transaction reaches an unfinished apply path.
+/// Panics when an orphaned `λSQL` transaction reaches the unfinished rebuild
+/// path.
 pub fn on_event(db: &mut Databases, event: &Event, channel_id: ChannelId) -> Result<(), Error> {
     match event {
         Event::BlocksProcessed {
@@ -40,25 +63,12 @@ pub fn on_event(db: &mut Databases, event: &Event, channel_id: ChannelId) -> Res
                 "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 orphaned_contains_logos_sql(channel_update, channel_id) {
+                todo!("reconcile 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");
-            }
+            apply_finalized(db, finalized)?;
+            apply_adopted(db, &channel_update.adopted, channel_id)?;
 
             db.persist_checkpoint(checkpoint)?;
         }
@@ -71,14 +81,117 @@ pub fn on_event(db: &mut Databases, event: &Event, channel_id: ChannelId) -> Res
     Ok(())
 }
 
-fn update_contains_logos_sql(channel_update: &ChannelUpdate, channel_id: ChannelId) -> bool {
+fn orphaned_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 apply_finalized(db: &mut Databases, finalized: &[FinalizedTx]) -> Result<(), Error> {
+    for transaction in finalized {
+        for operation in &transaction.ops {
+            let FinalizedOp::Inscription(inscription) = operation else {
+                continue;
+            };
+
+            apply_inscription(db, inscription, WriteState::Finalized)?;
+        }
+    }
+
+    Ok(())
+}
+
+fn apply_adopted(
+    db: &mut Databases,
+    adopted: &[ChannelUpdateTx],
+    channel_id: ChannelId,
+) -> Result<(), Error> {
+    for transaction in adopted {
+        if let Some(inscription) = transaction.inscription() {
+            apply_inscription(db, inscription, WriteState::Adopted)?;
+            continue;
+        }
+
+        let ChannelUpdateTx::Custom(transaction) = transaction else {
+            continue;
+        };
+
+        for inscription in channel_inscriptions(transaction, channel_id) {
+            apply_inscription(db, &inscription, WriteState::Adopted)?;
+        }
+    }
+
+    Ok(())
+}
+
+fn apply_inscription(
+    db: &mut Databases,
+    inscription: &InscriptionInfo,
+    state: WriteState,
+) -> Result<(), Error> {
+    let payload = inscription.payload.as_ref();
+
+    if !protocol::is_logos_sql_payload(payload) {
+        return Ok(());
+    }
+
+    let write = match ChannelInscription::decode(payload) {
+        Ok(write) => write,
+        Err(error) => return handle_write_error(db, inscription, None, error),
+    };
+
+    if let Err(error) = state.apply(db, &write) {
+        return handle_write_error(db, inscription, Some(write.tx_id), error);
+    }
+
+    tracing::debug!(
+        target: TARGET,
+        tx_id = ?write.tx_id,
+        statements = write.transaction.statements().len(),
+        ?state,
+        "channel write processed"
+    );
+
+    Ok(())
+}
+
+fn handle_write_error(
+    db: &Databases,
+    inscription: &InscriptionInfo,
+    tx_id: Option<TxId>,
+    error: Error,
+) -> Result<(), Error> {
+    if is_rejected_write(&error) {
+        record_rejection(db, inscription, tx_id, &error)
+    } else {
+        Err(error)
+    }
+}
+
+fn record_rejection(
+    db: &Databases,
+    inscription: &InscriptionInfo,
+    tx_id: Option<TxId>,
+    error: &Error,
+) -> Result<(), Error> {
+    db.record_rejected_write(inscription.this_msg, tx_id, &error.to_string())?;
+
+    tracing::warn!(
+        target: TARGET,
+        this_msg = %inscription.this_msg,
+        ?tx_id,
+        %error,
+        "channel write rejected"
+    );
+
+    Ok(())
+}
+
+const fn is_rejected_write(error: &Error) -> bool {
+    matches!(error, Error::InvalidPayload(_) | Error::RejectedSql(_))
+}
+
 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());
@@ -93,21 +206,6 @@ fn transaction_contains_logos_sql(transaction: &ChannelUpdateTx, channel_id: Cha
         .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::{
@@ -119,12 +217,14 @@ mod tests {
     };
     use tempfile::TempDir;
 
-    use super::{on_event, update_contains_logos_sql};
+    use super::on_event;
     use crate::{
         db::Databases,
-        protocol::{EncodedWrite, Statement, Transaction, Value},
+        protocol::{ChannelInscription, EncodedWrite, Statement, Transaction, Value},
     };
 
+    const CHANNEL_ID: [u8; 32] = [9; 32];
+
     fn checkpoint(byte: u8, slot: u64) -> SequencerCheckpoint {
         SequencerCheckpoint {
             last_msg_id: MsgId::root(),
@@ -135,11 +235,11 @@ mod tests {
         }
     }
 
-    fn inscription(payload: &[u8]) -> InscriptionInfo {
+    fn inscription(payload: &[u8], byte: u8) -> InscriptionInfo {
         InscriptionInfo {
-            tx_hash: TxHash::from([1; 32]),
+            tx_hash: TxHash::from([byte; 32]),
             parent_msg: MsgId::root(),
-            this_msg: MsgId::root(),
+            this_msg: MsgId::from([byte; 32]),
             payload: payload
                 .to_vec()
                 .try_into()
@@ -147,34 +247,139 @@ mod tests {
         }
     }
 
-    fn blocks_processed(checkpoint: SequencerCheckpoint, payload: &[u8]) -> Event {
+    fn blocks_processed(
+        checkpoint: SequencerCheckpoint,
+        adopted: Vec<ChannelUpdateTx>,
+        orphaned: Vec<ChannelUpdateTx>,
+        finalized: Vec<FinalizedTx>,
+    ) -> 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))],
-            }],
+            channel_update: ChannelUpdate { orphaned, adopted },
+            finalized,
         }
     }
 
-    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 finalized(payload: &[u8], byte: u8) -> FinalizedTx {
+        FinalizedTx {
+            tx_hash: TxHash::from([byte; 32]),
+            l1_slot: Slot::from(2),
+            ops: vec![FinalizedOp::Inscription(inscription(payload, byte))],
         }
     }
 
-    fn encoded_write() -> Vec<u8> {
+    fn transaction(sql: &str, params: Vec<Value>) -> Transaction {
+        Transaction::new(vec![
+            Statement::new(sql.to_owned(), params).expect("statement should be valid"),
+        ])
+        .expect("transaction should be valid")
+    }
+
+    fn encoded_write(transaction: &Transaction) -> EncodedWrite {
+        EncodedWrite::new(transaction).expect("payload should encode")
+    }
+
+    fn item_count(path: &std::path::Path) -> i64 {
+        let connection = Databases::open_reader(path).expect("database should open for reading");
+
+        connection
+            .query_row("SELECT count(*) FROM items", [], |row| row.get(0))
+            .expect("item count should be readable")
+    }
+
+    fn table_exists(path: &std::path::Path, table: &str) -> bool {
+        let connection = Databases::open_reader(path).expect("database should open for reading");
+
+        connection
+            .query_row(
+                "SELECT EXISTS(
+                    SELECT 1 FROM sqlite_schema WHERE type = 'table' AND name = ?1
+                )",
+                [table],
+                |row| row.get(0),
+            )
+            .expect("schema should be readable")
+    }
+
+    #[test]
+    fn adopted_writes_apply_to_live_in_channel_order() {
+        let dir = TempDir::new().expect("temporary directory should be created");
+        let mut db = Databases::open(dir.path()).expect("databases should open");
+        let live_path = db.live_path().to_owned();
+        let lib_path = db.lib_path().to_owned();
+
+        let create = encoded_write(&transaction(
+            "CREATE TABLE items(value INTEGER NOT NULL)",
+            Vec::new(),
+        ));
+        let insert = encoded_write(&transaction(
+            "INSERT INTO items(value) VALUES (?1)",
+            vec![Value::Integer(1)],
+        ));
+
+        let event = blocks_processed(
+            checkpoint(2, 2),
+            vec![
+                ChannelUpdateTx::Inscription(inscription(&create.payload, 1)),
+                ChannelUpdateTx::Inscription(inscription(&insert.payload, 2)),
+            ],
+            Vec::new(),
+            Vec::new(),
+        );
+
+        on_event(&mut db, &event, ChannelId::from(CHANNEL_ID))
+            .expect("adopted writes should apply");
+
+        on_event(&mut db, &event, ChannelId::from(CHANNEL_ID))
+            .expect("replayed event should be skipped");
+
+        assert_eq!(item_count(&live_path), 1);
+        assert!(!table_exists(&lib_path, "items"));
+    }
+
+    #[test]
+    fn finalized_backfill_applies_to_lib_and_live_once() {
+        let dir = TempDir::new().expect("temporary directory should be created");
+        let mut db = Databases::open(dir.path()).expect("databases should open");
+        let live_path = db.live_path().to_owned();
+        let lib_path = db.lib_path().to_owned();
+
+        let create = encoded_write(&transaction(
+            "CREATE TABLE items(value INTEGER NOT NULL)",
+            Vec::new(),
+        ));
+        let insert = encoded_write(&transaction(
+            "INSERT INTO items(value) VALUES (?1)",
+            vec![Value::Integer(1)],
+        ));
+
+        let event = blocks_processed(
+            checkpoint(2, 2),
+            Vec::new(),
+            Vec::new(),
+            vec![finalized(&create.payload, 1), finalized(&insert.payload, 2)],
+        );
+
+        on_event(&mut db, &event, ChannelId::from(CHANNEL_ID))
+            .expect("finalized writes should apply");
+        on_event(&mut db, &event, ChannelId::from(CHANNEL_ID))
+            .expect("replayed event should be skipped");
+
+        assert_eq!(item_count(&lib_path), 1);
+        assert_eq!(item_count(&live_path), 1);
+    }
+
+    #[test]
+    fn replay_after_apply_completes_checkpoint_without_duplicate_effects() {
+        let dir = TempDir::new().expect("temporary directory should be created");
+        let mut db = Databases::open(dir.path()).expect("databases should open");
+        let live_path = db.live_path().to_owned();
         let transaction = Transaction::new(vec![
+            Statement::new(
+                "CREATE TABLE items(value INTEGER NOT NULL)".to_owned(),
+                Vec::new(),
+            )
+            .expect("statement should be valid"),
             Statement::new(
                 "INSERT INTO items(value) VALUES (?1)".to_owned(),
                 vec![Value::Integer(1)],
@@ -182,54 +387,307 @@ mod tests {
             .expect("statement should be valid"),
         ])
         .expect("transaction should be valid");
-        EncodedWrite::new(&transaction)
-            .expect("payload should encode")
-            .payload
+        let encoded = encoded_write(&transaction);
+        let write = ChannelInscription::decode(&encoded.payload).expect("payload should decode");
+
+        // Simulate a crash after SQL and its applied marker commit but before
+        // the ZoneSDK checkpoint is persisted.
+        db.apply_adopted_write(&write)
+            .expect("write should apply before the simulated crash");
+        drop(db);
+
+        let mut db = Databases::open(dir.path()).expect("databases should reopen");
+
+        let expected_checkpoint = checkpoint(2, 2);
+        let event = blocks_processed(
+            expected_checkpoint.clone(),
+            vec![ChannelUpdateTx::Inscription(inscription(
+                &encoded.payload,
+                1,
+            ))],
+            Vec::new(),
+            Vec::new(),
+        );
+
+        on_event(&mut db, &event, ChannelId::from(CHANNEL_ID))
+            .expect("replayed event should complete");
+
+        let restored = db
+            .load_checkpoint()
+            .expect("checkpoint should load")
+            .expect("checkpoint should exist");
+
+        assert_eq!(item_count(&live_path), 1);
+        assert_eq!(restored.lib, expected_checkpoint.lib);
+        assert_eq!(restored.lib_slot, expected_checkpoint.lib_slot);
     }
 
     #[test]
-    #[should_panic(expected = "apply finalized \u{3bb}SQL transactions")]
-    fn finalized_logos_sql_payload_reaches_replay_placeholder() {
+    fn locally_applied_write_is_not_executed_when_adopted() {
         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");
+        let mut db = Databases::open(dir.path()).expect("databases should open");
+        let live_path = db.live_path().to_owned();
 
-        db.persist_checkpoint(&first)
-            .expect("initial checkpoint should commit");
+        let setup = transaction("CREATE TABLE items(value INTEGER NOT NULL)", Vec::new());
+        let setup_encoded = EncodedWrite::new(&setup).expect("setup write should encode");
 
-        let payload = encoded_write();
+        db.commit_local_write(&setup, &setup_encoded)
+            .expect("setup write should commit");
+        db.mark_publish_complete(setup_encoded.tx_id)
+            .expect("setup publish should be complete");
 
-        let event = blocks_processed(second, &payload);
-        on_event(&mut db, &event, ChannelId::from([9; 32]))
-            .expect("event should reach replay placeholder");
+        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)
+            .expect("insert write should commit");
+
+        let event = blocks_processed(
+            checkpoint(2, 2),
+            vec![ChannelUpdateTx::Inscription(inscription(
+                &insert_encoded.payload,
+                2,
+            ))],
+            Vec::new(),
+            Vec::new(),
+        );
+
+        on_event(&mut db, &event, ChannelId::from(CHANNEL_ID))
+            .expect("local adoption should be recognized");
+
+        assert_eq!(item_count(&live_path), 1);
     }
 
     #[test]
-    #[should_panic(expected = "reconcile adopted and orphaned \u{3bb}SQL transactions")]
-    fn orphaned_logos_sql_payload_reaches_reconciliation_placeholder() {
+    fn rejected_sql_does_not_block_following_writes() {
         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);
+        let mut db = Databases::open(dir.path()).expect("databases should open");
+        let live_path = db.live_path().to_owned();
 
-        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])
+        let insert = encoded_write(&transaction(
+            "INSERT INTO missing(value) VALUES (?1)",
+            vec![Value::Integer(1)],
         ));
+        let create = encoded_write(&transaction(
+            "CREATE TABLE items(value INTEGER NOT NULL)",
+            Vec::new(),
+        ));
+        let expected_checkpoint = checkpoint(2, 2);
+        let event = blocks_processed(
+            expected_checkpoint.clone(),
+            vec![
+                ChannelUpdateTx::Inscription(inscription(&insert.payload, 1)),
+                ChannelUpdateTx::Inscription(inscription(&create.payload, 2)),
+            ],
+            Vec::new(),
+            Vec::new(),
+        );
+
+        on_event(&mut db, &event, ChannelId::from(CHANNEL_ID))
+            .expect("rejected SQL should not halt channel replay");
+
+        let restored = db
+            .load_checkpoint()
+            .expect("checkpoint should load")
+            .expect("checkpoint should exist");
+
+        assert_eq!(restored.lib, expected_checkpoint.lib);
+        assert_eq!(restored.lib_slot, expected_checkpoint.lib_slot);
+        assert!(table_exists(&live_path, "items"));
+
+        drop(db);
+        let db = Databases::open(dir.path()).expect("databases should reopen");
+
+        assert_eq!(
+            db.rejected_write_count().expect("rejections should load"),
+            1
+        );
+    }
+
+    #[test]
+    fn malformed_payload_does_not_block_following_writes() {
+        let dir = TempDir::new().expect("temporary directory should be created");
+        let mut db = Databases::open(dir.path()).expect("databases should open");
+        let live_path = db.live_path().to_owned();
+
+        let mut malformed = encoded_write(&transaction(
+            "CREATE TABLE discarded(value INTEGER)",
+            Vec::new(),
+        ))
+        .payload;
+        malformed.truncate(b"LOGOS_SQL".len());
+
+        let create = encoded_write(&transaction(
+            "CREATE TABLE items(value INTEGER NOT NULL)",
+            Vec::new(),
+        ));
+        let expected_checkpoint = checkpoint(2, 2);
+        let event = blocks_processed(
+            expected_checkpoint.clone(),
+            vec![
+                ChannelUpdateTx::Inscription(inscription(&malformed, 1)),
+                ChannelUpdateTx::Inscription(inscription(&create.payload, 2)),
+            ],
+            Vec::new(),
+            Vec::new(),
+        );
+
+        on_event(&mut db, &event, ChannelId::from(CHANNEL_ID))
+            .expect("malformed payload should not halt channel replay");
+
+        let restored = db
+            .load_checkpoint()
+            .expect("checkpoint should load")
+            .expect("checkpoint should exist");
+
+        assert_eq!(restored.lib, expected_checkpoint.lib);
+        assert_eq!(restored.lib_slot, expected_checkpoint.lib_slot);
+        assert!(table_exists(&live_path, "items"));
+        assert_eq!(
+            db.rejected_write_count().expect("rejections should load"),
+            1
+        );
+    }
+
+    #[test]
+    fn reused_transaction_id_does_not_block_following_writes() {
+        let dir = TempDir::new().expect("temporary directory should be created");
+        let mut db = Databases::open(dir.path()).expect("databases should open");
+        let live_path = db.live_path().to_owned();
+
+        let first = encoded_write(&transaction(
+            "CREATE TABLE first_write(value INTEGER)",
+            Vec::new(),
+        ));
+        let conflicting = ChannelInscription {
+            tx_id: first.tx_id,
+            transaction: transaction("CREATE TABLE conflicting_write(value INTEGER)", Vec::new()),
+        }
+        .encode()
+        .expect("conflicting write should encode");
+        let following = encoded_write(&transaction(
+            "CREATE TABLE following_write(value INTEGER)",
+            Vec::new(),
+        ));
+        let expected_checkpoint = checkpoint(2, 2);
+        let event = blocks_processed(
+            expected_checkpoint.clone(),
+            vec![
+                ChannelUpdateTx::Inscription(inscription(&first.payload, 1)),
+                ChannelUpdateTx::Inscription(inscription(&conflicting, 2)),
+                ChannelUpdateTx::Inscription(inscription(&following.payload, 3)),
+            ],
+            Vec::new(),
+            Vec::new(),
+        );
+
+        on_event(&mut db, &event, ChannelId::from(CHANNEL_ID))
+            .expect("reused transaction ID should not halt channel replay");
+
+        let restored = db
+            .load_checkpoint()
+            .expect("checkpoint should load")
+            .expect("checkpoint should exist");
+
+        assert_eq!(restored.lib, expected_checkpoint.lib);
+        assert_eq!(restored.lib_slot, expected_checkpoint.lib_slot);
+        assert!(table_exists(&live_path, "first_write"));
+        assert!(!table_exists(&live_path, "conflicting_write"));
+        assert!(table_exists(&live_path, "following_write"));
+        assert_eq!(
+            db.rejected_write_count().expect("rejections should load"),
+            1
+        );
+    }
+
+    #[test]
+    fn unsupported_protocol_does_not_block_following_writes() {
+        let dir = TempDir::new().expect("temporary directory should be created");
+        let mut db = Databases::open(dir.path()).expect("databases should open");
+        let live_path = db.live_path().to_owned();
+
+        let mut unsupported = encoded_write(&transaction(
+            "CREATE TABLE discarded(value INTEGER)",
+            Vec::new(),
+        ))
+        .payload;
+        let version_offset = b"LOGOS_SQL".len();
+        unsupported[version_offset..version_offset + 2].copy_from_slice(&2u16.to_le_bytes());
+
+        let following = encoded_write(&transaction(
+            "CREATE TABLE following_write(value INTEGER)",
+            Vec::new(),
+        ));
+        let expected_checkpoint = checkpoint(2, 2);
+        let event = blocks_processed(
+            expected_checkpoint.clone(),
+            vec![
+                ChannelUpdateTx::Inscription(inscription(&unsupported, 1)),
+                ChannelUpdateTx::Inscription(inscription(&following.payload, 2)),
+            ],
+            Vec::new(),
+            Vec::new(),
+        );
+
+        on_event(&mut db, &event, ChannelId::from(CHANNEL_ID))
+            .expect("unsupported protocol should not halt channel replay");
+
+        let restored = db
+            .load_checkpoint()
+            .expect("checkpoint should load")
+            .expect("checkpoint should exist");
+
+        assert_eq!(restored.lib, expected_checkpoint.lib);
+        assert_eq!(restored.lib_slot, expected_checkpoint.lib_slot);
+        assert!(table_exists(&live_path, "following_write"));
+        assert_eq!(
+            db.rejected_write_count().expect("rejections should load"),
+            1
+        );
+    }
+
+    #[test]
+    fn orphaned_local_write_prevents_adopted_apply_until_rebuild() {
+        let dir = TempDir::new().expect("temporary directory should be created");
+        let mut db = Databases::open(dir.path()).expect("databases should open");
+        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)
+            .expect("local write should commit");
+        db.mark_publish_complete(local_encoded.tx_id)
+            .expect("local publish should be complete");
+
+        let adopted = encoded_write(&transaction(
+            "CREATE TABLE adopted_too_early(value INTEGER)",
+            Vec::new(),
+        ));
+        let event = blocks_processed(
+            checkpoint(2, 2),
+            vec![ChannelUpdateTx::Inscription(inscription(
+                &adopted.payload,
+                2,
+            ))],
+            vec![ChannelUpdateTx::Inscription(inscription(
+                &local_encoded.payload,
+                1,
+            ))],
+            Vec::new(),
+        );
+
+        let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
+            on_event(&mut db, &event, ChannelId::from(CHANNEL_ID))
+        }));
+
+        assert!(
+            result.is_err(),
+            "orphan recovery should reach its placeholder"
+        );
+        assert!(!table_exists(&live_path, "adopted_too_early"));
     }
 }

logos_sql/src/db.rs

@@ -7,25 +7,27 @@ use std::{
 };
 
 use bincode::Options as _;
-use lb_zone_sdk::sequencer::SequencerCheckpoint;
+use lb_zone_sdk::{node_types::MsgId, sequencer::SequencerCheckpoint};
 use rusqlite::{
-    Connection, OpenFlags, OptionalExtension as _, Row,
+    Connection, ErrorCode as SqliteErrorCode, OpenFlags, OptionalExtension as _, Row,
     hooks::{AuthAction, AuthContext, Authorization},
     params, params_from_iter,
 };
 
 use crate::{
     error::Error,
-    protocol::{EncodedWrite, Transaction, TxId},
+    protocol::{ChannelInscription, EncodedWrite, Transaction, TxId},
 };
 
 const DATABASE_BUSY_TIMEOUT: Duration = Duration::from_secs(5);
+const LIB_DATABASE_FILE: &str = "LIB.db";
 const LIVE_DATABASE_FILE: &str = "LIVE.db";
 const CONTROL_DATABASE_FILE: &str = "control.db";
+const RESERVED_OBJECT_PREFIX: &str = "__logos_sql_";
 
-// Stores locally committed writes and their exact channel payload. At most one
-// write may be waiting for ZoneSDK publication.
-const LIVE_SCHEMA: &str = "
+// Present in both state databases so replicated SQL observes the same schema.
+// Only LIVE.db stores a row, committed atomically with the local write.
+const PENDING_WRITE_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),
@@ -33,12 +35,28 @@ const LIVE_SCHEMA: &str = "
     ) STRICT;
 ";
 
-// Stores the participant-local ZoneSDK checkpoint independently of live state.
+// Records the SQL transactions whose effects exist in each database. The
+// marker commits with the effects, making channel-event replay idempotent.
+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)
+    ) STRICT;
+";
+
+// Stores participant-local progress and rejected channel writes independently
+// of replicated database state.
 const CONTROL_SCHEMA: &str = "
     CREATE TABLE IF NOT EXISTS __logos_sql_state (
         singleton INTEGER PRIMARY KEY CHECK (singleton = 1),
         checkpoint BLOB
     ) STRICT;
+
+    CREATE TABLE IF NOT EXISTS __logos_sql_rejected_writes (
+        this_msg BLOB PRIMARY KEY CHECK (length(this_msg) = 32),
+        tx_id BLOB CHECK (tx_id IS NULL OR length(tx_id) = 32),
+        reason TEXT NOT NULL
+    ) STRICT;
 ";
 
 const INITIALIZE_CONTROL_STATE: &str = "
@@ -58,6 +76,11 @@ const UPDATE_CHECKPOINT: &str = "
     WHERE singleton = 1
 ";
 
+const INSERT_REJECTED_WRITE: &str = "
+    INSERT INTO __logos_sql_rejected_writes (this_msg, tx_id, reason)
+    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)
@@ -74,6 +97,17 @@ const MARK_PUBLISH_COMPLETE: &str = "
     WHERE singleton = 1 AND tx_id = ?1
 ";
 
+const SELECT_APPLIED_WRITE: &str = "
+    SELECT transaction_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)
+    VALUES (?1, ?2)
+";
+
 const WRITER_PRAGMAS: &str = "
     PRAGMA journal_mode = WAL;
     PRAGMA synchronous = FULL;
@@ -105,8 +139,10 @@ pub struct PendingPublish {
 
 /// Owns the participant-local database connections.
 pub struct Databases {
+    lib: Connection,
     live: Connection,
     control: Connection,
+    lib_path: PathBuf,
     live_path: PathBuf,
 }
 
@@ -115,23 +151,35 @@ impl Databases {
     pub(crate) fn open(directory: &Path) -> Result<Self, Error> {
         fs::create_dir_all(directory)?;
 
+        let lib_path = directory.join(LIB_DATABASE_FILE);
         let live_path = directory.join(LIVE_DATABASE_FILE);
         let control_path = directory.join(CONTROL_DATABASE_FILE);
 
+        let lib = open_writer(&lib_path)?;
         let live = open_writer(&live_path)?;
         let control = open_writer(&control_path)?;
 
-        live.execute_batch(LIVE_SCHEMA)?;
+        for connection in [&lib, &live] {
+            connection.execute_batch(PENDING_WRITE_SCHEMA)?;
+            connection.execute_batch(APPLIED_WRITE_SCHEMA)?;
+        }
+
         control.execute_batch(CONTROL_SCHEMA)?;
         control.execute(INITIALIZE_CONTROL_STATE, [])?;
 
         Ok(Self {
+            lib,
             live,
             control,
+            lib_path,
             live_path,
         })
     }
 
+    pub(crate) fn lib_path(&self) -> &Path {
+        &self.lib_path
+    }
+
     pub(crate) fn live_path(&self) -> &Path {
         &self.live_path
     }
@@ -164,6 +212,28 @@ impl Databases {
         Ok(())
     }
 
+    /// Records a channel write that every replica must skip.
+    ///
+    /// Keeping the rejection in participant-local state allows replay to
+    /// advance past invalid input and preserves the outcome for later
+    /// application reporting. `tx_id` is absent when the payload could not be
+    /// decoded.
+    pub(crate) fn record_rejected_write(
+        &self,
+        this_msg: MsgId,
+        tx_id: Option<TxId>,
+        reason: &str,
+    ) -> Result<(), Error> {
+        let tx_id = tx_id.map(<[u8; 32]>::from);
+        let tx_id = tx_id.as_ref().map(<[u8; 32]>::as_slice);
+
+        self.control.execute(
+            INSERT_REJECTED_WRITE,
+            params![this_msg.as_ref(), tx_id, reason],
+        )?;
+
+        Ok(())
+    }
     /// Commits application effects and their pending publish record together in
     /// `LIVE.db`.
     pub(crate) fn commit_local_write(
@@ -177,20 +247,16 @@ impl Databases {
 
         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));
+        // TODO: Capture nondeterministic function results and include them in
+        // the transaction published to other participants.
+        apply_statements(&db_transaction, transaction)?;
 
-        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()))?;
+        let transaction_digest = transaction.digest()?;
 
-            Ok::<_, Error>(())
-        });
-
-        db_transaction.authorizer(None::<fn(AuthContext<'_>) -> Authorization>);
-        apply_result?;
+        db_transaction.execute(
+            INSERT_APPLIED_WRITE,
+            params![encoded.tx_id.as_ref(), transaction_digest],
+        )?;
 
         db_transaction.execute(
             INSERT_PENDING_WRITE,
@@ -201,6 +267,43 @@ impl Databases {
         Ok(encoded.tx_id)
     }
 
+    /// Applies a newly adopted channel write to the live database.
+    pub(crate) fn apply_adopted_write(&mut self, write: &ChannelInscription) -> Result<(), Error> {
+        apply_channel_write(&mut self.live, write)
+    }
+
+    /// Applies a finalized channel write to finalized and live state.
+    ///
+    /// Applying to `LIVE.db` as well covers writes first discovered through
+    /// finalized backfill. Writes already applied locally or while adopted
+    /// are skipped using their `TxId`.
+    pub(crate) fn apply_finalized_write(
+        &mut self,
+        write: &ChannelInscription,
+    ) -> Result<(), Error> {
+        let transaction_digest = write.transaction.digest()?;
+
+        is_write_applied(&self.lib, write.tx_id, &transaction_digest)?;
+        is_write_applied(&self.live, write.tx_id, &transaction_digest)?;
+
+        // LIB and LIVE are separate SQLite files. If LIVE fails after LIB
+        // commits, the checkpoint remains behind. Redelivery then skips LIB
+        // using its applied marker and completes LIVE.
+        apply_channel_write(&mut self.lib, write)?;
+        apply_channel_write(&mut self.live, write)
+    }
+
+    #[cfg(test)]
+    pub(crate) fn rejected_write_count(&self) -> Result<i64, Error> {
+        self.control
+            .query_row(
+                "SELECT count(*) FROM __logos_sql_rejected_writes",
+                [],
+                |row| row.get(0),
+            )
+            .map_err(Error::from)
+    }
+
     pub(crate) fn pending_publish(&self) -> Result<Option<PendingPublish>, Error> {
         let record = self
             .live
@@ -256,9 +359,120 @@ fn configure_connection(conn: &Connection) -> Result<(), Error> {
     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 {
+fn apply_channel_write(
+    connection: &mut Connection,
+    write: &ChannelInscription,
+) -> Result<(), Error> {
+    let transaction_digest = write.transaction.digest()?;
+    let db_transaction = connection.transaction()?;
+
+    if is_write_applied(&db_transaction, write.tx_id, &transaction_digest)? {
+        return Ok(());
+    }
+
+    if let Err(error) = apply_statements(&db_transaction, &write.transaction) {
+        return match error {
+            Error::Database(error) if is_deterministic_sql_error(&error) => {
+                Err(Error::RejectedSql(error))
+            }
+            error => Err(error),
+        };
+    }
+
+    db_transaction.execute(
+        INSERT_APPLIED_WRITE,
+        params![write.tx_id.as_ref(), transaction_digest],
+    )?;
+    db_transaction.commit()?;
+
+    Ok(())
+}
+
+fn is_write_applied(
+    connection: &Connection,
+    tx_id: TxId,
+    transaction_digest: &[u8; 32],
+) -> Result<bool, Error> {
+    let stored_digest = connection
+        .query_row(SELECT_APPLIED_WRITE, [tx_id.as_ref()], |row| {
+            row.get::<_, Vec<u8>>(0)
+        })
+        .optional()?;
+
+    if let Some(stored_digest) = stored_digest {
+        if stored_digest.as_slice() != transaction_digest {
+            return Err(Error::InvalidPayload(
+                "transaction id was reused for different content",
+            ));
+        }
+
+        return Ok(true);
+    }
+
+    Ok(false)
+}
+
+// Only failures determined by the received statement and parameters are safe
+// to skip. Storage, locking, and other local failures must halt replay so the
+// same channel position can be retried without diverging from other replicas.
+fn is_deterministic_sql_error(error: &rusqlite::Error) -> bool {
+    if error.sqlite_error().is_some_and(|error| {
+        matches!(
+            error.extended_code,
+            rusqlite::ffi::SQLITE_ERROR_RETRY | rusqlite::ffi::SQLITE_ERROR_SNAPSHOT
+        )
+    }) {
+        return false;
+    }
+
+    match error.sqlite_error_code() {
+        Some(
+            SqliteErrorCode::Unknown
+            | SqliteErrorCode::TooBig
+            | SqliteErrorCode::ConstraintViolation
+            | SqliteErrorCode::TypeMismatch
+            | SqliteErrorCode::AuthorizationForStatementDenied
+            | SqliteErrorCode::ParameterOutOfRange,
+        ) => true,
+        Some(_) => false,
+        None => matches!(
+            error,
+            rusqlite::Error::NulError(_)
+                | rusqlite::Error::InvalidParameterName(_)
+                | rusqlite::Error::ExecuteReturnedResults
+                | rusqlite::Error::InvalidFunctionParameterType(_, _)
+                | rusqlite::Error::UserFunctionError(_)
+                | rusqlite::Error::ToSqlConversionFailure(_)
+                | rusqlite::Error::InvalidQuery
+                | rusqlite::Error::UnwindingPanic
+                | rusqlite::Error::GetAuxWrongType
+                | rusqlite::Error::MultipleStatement
+                | rusqlite::Error::InvalidParameterCount(_, _)
+        ),
+    }
+}
+
+fn apply_statements(
+    db_transaction: &rusqlite::Transaction<'_>,
+    transaction: &Transaction,
+) -> Result<(), Error> {
+    db_transaction.authorizer(Some(authorize_application_sql));
+
+    let result = transaction.statements().iter().try_for_each(|statement| {
+        db_transaction.execute(statement.sql(), params_from_iter(statement.params()))?;
+
+        Ok::<_, Error>(())
+    });
+
+    db_transaction.authorizer(None::<fn(AuthContext<'_>) -> Authorization>);
+
+    result
+}
+
+// λSQL owns the surrounding transaction and its bookkeeping tables, so
+// application statements cannot modify either. Temporary objects are also
+// denied because they exist only for one connection and cannot be replicated.
+fn authorize_application_sql(context: AuthContext<'_>) -> Authorization {
     let denied = matches!(
         context.action,
         AuthAction::Unknown { .. }
@@ -267,7 +481,8 @@ const fn authorize_application_sql(context: AuthContext<'_>) -> Authorization {
             | AuthAction::Attach { .. }
             | AuthAction::Detach { .. }
             | AuthAction::Pragma { .. }
-    );
+    ) || is_temporary_object_action(context.action)
+        || action_uses_reserved_name(context.action);
 
     if denied {
         Authorization::Deny
@@ -276,6 +491,61 @@ const fn authorize_application_sql(context: AuthContext<'_>) -> Authorization {
     }
 }
 
+const fn is_temporary_object_action(action: AuthAction<'_>) -> bool {
+    matches!(
+        action,
+        AuthAction::CreateTempIndex { .. }
+            | AuthAction::CreateTempTable { .. }
+            | AuthAction::CreateTempTrigger { .. }
+            | AuthAction::CreateTempView { .. }
+            | AuthAction::DropTempIndex { .. }
+            | AuthAction::DropTempTable { .. }
+            | AuthAction::DropTempTrigger { .. }
+            | AuthAction::DropTempView { .. }
+    )
+}
+
+fn action_uses_reserved_name(action: AuthAction<'_>) -> bool {
+    match action {
+        AuthAction::CreateIndex {
+            index_name,
+            table_name,
+        }
+        | AuthAction::DropIndex {
+            index_name,
+            table_name,
+        } => is_reserved_name(index_name) || is_reserved_name(table_name),
+        AuthAction::CreateTrigger {
+            trigger_name,
+            table_name,
+        }
+        | AuthAction::DropTrigger {
+            trigger_name,
+            table_name,
+        } => is_reserved_name(trigger_name) || is_reserved_name(table_name),
+        AuthAction::CreateTable { table_name }
+        | AuthAction::Delete { table_name }
+        | AuthAction::DropTable { table_name }
+        | AuthAction::Insert { table_name }
+        | AuthAction::Read { table_name, .. }
+        | AuthAction::Update { table_name, .. }
+        | AuthAction::AlterTable { table_name, .. }
+        | AuthAction::Analyze { table_name }
+        | AuthAction::CreateVtable { table_name, .. }
+        | AuthAction::DropVtable { table_name, .. } => is_reserved_name(table_name),
+        AuthAction::CreateView { view_name } | AuthAction::DropView { view_name } => {
+            is_reserved_name(view_name)
+        }
+        AuthAction::Reindex { index_name } => is_reserved_name(index_name),
+        _ => false,
+    }
+}
+
+fn is_reserved_name(name: &str) -> bool {
+    name.get(..RESERVED_OBJECT_PREFIX.len())
+        .is_some_and(|prefix| prefix.eq_ignore_ascii_case(RESERVED_OBJECT_PREFIX))
+}
+
 fn decode_tx_id(bytes: Vec<u8>) -> Result<TxId, Error> {
     let bytes: [u8; 32] = bytes
         .try_into()
@@ -303,7 +573,7 @@ mod tests {
     use crate::{
         error::Error,
         local_write,
-        protocol::{EncodedWrite, Statement, Transaction, Value},
+        protocol::{ChannelInscription, EncodedWrite, Statement, Transaction, TxId, Value},
     };
 
     fn checkpoint(byte: u8, slot: u64) -> SequencerCheckpoint {
@@ -341,6 +611,11 @@ mod tests {
     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");
+
+        db.live
+            .execute("CREATE TABLE items(value INTEGER)", [])
+            .expect("application table should be created");
+
         let transaction = transaction(sql);
         let encoded = encoded_write(&transaction);
 
@@ -348,7 +623,12 @@ mod tests {
             .commit_local_write(&transaction, &encoded)
             .expect_err("application SQL should be rejected");
 
-        assert!(matches!(error, Error::Database(_)));
+        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")
@@ -356,6 +636,28 @@ mod tests {
         );
     }
 
+    fn internal_schema(
+        connection: &rusqlite::Connection,
+    ) -> Vec<(String, String, String, Option<String>)> {
+        let mut statement = connection
+            .prepare(
+                "SELECT type, name, tbl_name, sql
+                 FROM sqlite_schema
+                 WHERE name GLOB '__logos_sql_*'
+                    OR tbl_name GLOB '__logos_sql_*'
+                 ORDER BY type, name",
+            )
+            .expect("schema query should prepare");
+
+        statement
+            .query_map([], |row| {
+                Ok((row.get(0)?, row.get(1)?, row.get(2)?, row.get(3)?))
+            })
+            .expect("schema query should run")
+            .collect::<Result<_, _>>()
+            .expect("schema rows should decode")
+    }
+
     #[test]
     fn checkpoint_survives_reopen() {
         let dir = TempDir::new().expect("temporary directory should be created");
@@ -406,6 +708,14 @@ mod tests {
         );
     }
 
+    #[test]
+    fn state_databases_have_the_same_internal_schema() {
+        let dir = TempDir::new().expect("temporary directory should be created");
+        let db = Databases::open(dir.path()).expect("databases should open");
+
+        assert_eq!(internal_schema(&db.lib), internal_schema(&db.live));
+    }
+
     #[test]
     fn repeated_write_is_a_new_transaction() {
         let dir = TempDir::new().expect("temporary directory should be created");
@@ -434,6 +744,80 @@ mod tests {
         assert_eq!(count, 2);
     }
 
+    #[test]
+    fn reused_transaction_id_with_different_content_is_rejected() {
+        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 tx_id = TxId::from([7; 32]);
+        let first = ChannelInscription {
+            tx_id,
+            transaction: insert("first"),
+        };
+        let conflicting = ChannelInscription {
+            tx_id,
+            transaction: insert("conflicting"),
+        };
+
+        db.apply_adopted_write(&first)
+            .expect("first write should apply");
+
+        let error = db
+            .apply_adopted_write(&conflicting)
+            .expect_err("conflicting write should be rejected");
+
+        assert!(matches!(error, Error::InvalidPayload(_)));
+
+        let value: String = db
+            .live
+            .query_row("SELECT value FROM items", [], |row| row.get(0))
+            .expect("stored value should be readable");
+
+        assert_eq!(value, "first");
+    }
+
+    #[test]
+    fn conflicting_finalized_write_does_not_modify_lib() {
+        let dir = TempDir::new().expect("temporary directory should be created");
+        let mut db = Databases::open(dir.path()).expect("databases should open");
+
+        for connection in [&db.lib, &db.live] {
+            connection
+                .execute("CREATE TABLE items(value TEXT NOT NULL)", [])
+                .expect("application table should be created");
+        }
+
+        let tx_id = TxId::from([7; 32]);
+        let first = ChannelInscription {
+            tx_id,
+            transaction: insert("first"),
+        };
+        let conflicting = ChannelInscription {
+            tx_id,
+            transaction: insert("conflicting"),
+        };
+
+        db.apply_adopted_write(&first)
+            .expect("first write should apply to LIVE");
+
+        let error = db
+            .apply_finalized_write(&conflicting)
+            .expect_err("conflicting finalized write should be rejected");
+
+        assert!(matches!(error, Error::InvalidPayload(_)));
+
+        let lib_count: i64 = db
+            .lib
+            .query_row("SELECT count(*) FROM items", [], |row| row.get(0))
+            .expect("LIB row count should be readable");
+
+        assert_eq!(lib_count, 0);
+    }
+
     #[test]
     fn application_write_can_create_persistent_schema() {
         let dir = TempDir::new().expect("temporary directory should be created");
@@ -497,6 +881,42 @@ mod tests {
         }
     }
 
+    #[test]
+    fn application_sql_cannot_access_internal_objects() {
+        for sql in [
+            "DELETE FROM __logos_sql_applied_writes",
+            "DROP TABLE __logos_sql_applied_writes",
+            "ALTER TABLE __logos_sql_applied_writes RENAME TO application_table",
+            "CREATE INDEX __logos_sql_index ON items(value)",
+            "CREATE VIEW __logos_sql_view AS SELECT 1",
+            "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);
+        }
+    }
+
+    #[test]
+    fn application_sql_cannot_create_temporary_objects() {
+        for sql in [
+            "CREATE TEMP TABLE temporary_items(value INTEGER)",
+            "CREATE TEMP VIEW temporary_items AS SELECT 1",
+        ] {
+            assert_application_sql_rejected(sql);
+        }
+    }
+
+    #[test]
+    fn reserved_name_check_accepts_non_ascii_identifiers() {
+        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)
+            .expect("non-reserved Unicode name should be accepted");
+    }
+
     #[test]
     fn application_read_connection_is_read_only() {
         let dir = TempDir::new().expect("temporary directory should be created");

logos_sql/src/error.rs

@@ -44,9 +44,10 @@ pub enum Error {
     #[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),
+    /// SQL received from the channel was rejected deterministically by
+    /// `SQLite`.
+    #[error("channel SQL was rejected: {0}")]
+    RejectedSql(#[source] rusqlite::Error),
 
     /// The encoded payload exceeds the inscription limit.
     #[error("transaction is too large for one inscription")]

logos_sql/src/logos_sql.rs

@@ -34,6 +34,7 @@ pub struct LogosSqlConfig {
 /// `ZoneSDK` sequencer and the database writer. Dropping `LogosSql` aborts the
 /// task; call [`Self::shutdown`] to stop it gracefully and observe errors.
 pub struct LogosSql {
+    lib_path: PathBuf,
     live_path: PathBuf,
     runtime: Option<runtime::RuntimeHandle>,
 }
@@ -52,6 +53,7 @@ impl LogosSql {
         tokio::runtime::Handle::try_current().map_err(|_| Error::RuntimeUnavailable)?;
 
         let db = Databases::open(&config.state_dir)?;
+        let lib_path = db.lib_path().to_owned();
         let live_path = db.live_path().to_owned();
         let checkpoint = db.load_checkpoint()?;
         let node = NodeHttpClient::new(CommonHttpClient::new(None), config.node_url);
@@ -67,6 +69,7 @@ impl LogosSql {
         let runtime = runtime::spawn(sequencer, db, config.channel_id, checkpoint);
 
         let mut logos_sql = Self {
+            lib_path,
             live_path,
             runtime: Some(runtime),
         };
@@ -124,6 +127,18 @@ impl LogosSql {
         Databases::open_reader(&self.live_path)
     }
 
+    /// Opens a read-only connection to finalized state.
+    ///
+    /// Unlike [`Self::read_connection`], this state cannot be displaced by a
+    /// channel reorganization.
+    ///
+    /// # Errors
+    ///
+    /// Returns an error when the database file cannot be opened.
+    pub fn finalized_read_connection(&self) -> Result<Connection, Error> {
+        Databases::open_reader(&self.lib_path)
+    }
+
     /// Stops the runtime after its current atomic operation and waits for it.
     ///
     /// # Errors

logos_sql/src/protocol/mod.rs

@@ -3,6 +3,7 @@
 use std::fmt::{self, Display, Formatter};
 
 use bincode::Options as _;
+use blake2::{Blake2b, Digest as _, digest::consts::U32};
 use rand::RngCore as _;
 pub use rusqlite::types::Value;
 use serde::{Deserialize, Serialize};
@@ -175,6 +176,12 @@ impl Transaction {
     pub fn statements(&self) -> &[Statement] {
         &self.statements
     }
+
+    pub(crate) fn digest(&self) -> Result<[u8; 32], Error> {
+        let encoded = codec().serialize(self)?;
+
+        Ok(Blake2b::<U32>::digest(encoded).into())
+    }
 }
 
 /// Transaction payload carried by a `λSQL` channel inscription.
@@ -185,7 +192,7 @@ pub struct ChannelInscription {
 }
 
 impl ChannelInscription {
-    fn encode(&self) -> Result<Vec<u8>, Error> {
+    pub(crate) 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)?);
@@ -206,7 +213,7 @@ impl ChannelInscription {
         let version = u16::from_le_bytes([header[version_offset], header[version_offset + 1]]);
 
         if version != PAYLOAD_VERSION {
-            return Err(Error::UnsupportedProtocolVersion(version));
+            return Err(Error::InvalidPayload("protocol version is not supported"));
         }
 
         codec()

logos_sql/src/runtime.rs

@@ -229,10 +229,6 @@ impl Runtime {
     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 {
@@ -343,10 +339,3 @@ impl Runtime {
             || self.db.pending_publish()?.is_some())
     }
 }
-
-const fn is_retryable_apply_error(error: &Error) -> bool {
-    !matches!(
-        error,
-        Error::InvalidPayload(_) | Error::UnsupportedProtocolVersion(_)
-    )
-}