STR-881 inscription changes

bin/strata-client/src/main.rs

@@ -25,9 +25,8 @@
 use strata_evmexec::{engine::RpcExecEngineCtl, EngineRpcClient};
 use strata_primitives::params::Params;
 use strata_rocksdb::{
-    broadcaster::db::BroadcastDb, init_broadcaster_database, init_core_dbs,
-    init_sequencer_database, open_rocksdb_database, sequencer::db::SequencerDB, CommonDb,
-    DbOpsConfig, RBSeqBlobDb, ROCKSDB_NAME,
+    broadcaster::db::BroadcastDb, init_broadcaster_database, init_core_dbs, init_writer_database,
+    open_rocksdb_database, CommonDb, DbOpsConfig, RBL1WriterDb, ROCKSDB_NAME,
 };
 use strata_rpc_api::{
     StrataAdminApiServer, StrataApiServer, StrataDebugApiServer, StrataSequencerApiServer,
@@ -143,14 +142,15 @@
                 ctx.bitcoin_client.clone(),
                 params.clone(),
             );
-            let seq_db = init_sequencer_database(rbdb.clone(), ops_config);
+            let writer_db = init_writer_database(rbdb.clone(), ops_config);
 
+            // TODO: split writer tasks from this
             start_sequencer_tasks(
                 ctx.clone(),
                 &config,
                 sequencer_config,
                 &executor,
-                seq_db,
+                writer_db,
                 checkpoint_handle.clone(),
                 broadcast_handle,
                 &mut methods,
@@ -373,7 +373,7 @@
     config: &Config,
     sequencer_config: &SequencerConfig,
     executor: &TaskExecutor,
-    seq_db: Arc<SequencerDB<RBSeqBlobDb>>,
+    writer_db: Arc<RBL1WriterDb>,
     checkpoint_handle: Arc<CheckpointHandle>,
     broadcast_handle: Arc<L1BroadcastHandle>,
     methods: &mut Methods,
@@ -417,7 +417,7 @@
         Arc::new(btcio_config.writer.clone()),
         params.clone(),
         sequencer_bitcoin_address,
-        seq_db,
+        writer_db,
         status_channel.clone(),
         pool.clone(),
         broadcast_handle.clone(),

crates/btcio/src/reader/query.rs

@@ -324,7 +324,8 @@
     let txs = block.txdata.len();
 
     let params = ctx.params.clone();
-    let filtered_txs = filter_protocol_op_tx_refs(&block, state.filter_config());
+    let filtered_txs =
+        filter_protocol_op_tx_refs(&block, ctx.params.rollup(), state.filter_config());
     let block_data = BlockData::new(height, block, filtered_txs);
     let l1blkid = block_data.block().block_hash();
     trace!(%height, %l1blkid, %txs, "fetched block from client");

crates/btcio/src/test_utils.rs

@@ -1,3 +1,5 @@
+use std::sync::Arc;
+
 use async_trait::async_trait;
 use bitcoin::{
     bip32::Xpriv,
@@ -7,7 +9,7 @@ use bitcoin::{
     Address, Amount, Block, BlockHash, Network, ScriptBuf, SignedAmount, Transaction, Txid, Work,
 };
 use strata_l1tx::envelope::builder::build_envelope_script;
-use strata_primitives::l1::payload::L1Payload;
+use strata_primitives::{l1::payload::L1Payload, params::Params};
 
 use crate::{
     rpc::{
@@ -213,11 +215,11 @@ impl SignerRpc for TestBitcoinClient {
 }
 
 pub fn generate_envelope_script_test(
-    envelope_data: L1Payload,
-    rollup_name: &str,
+    payloads: &[L1Payload],
+    params: Arc<Params>,
     version: u8,
 ) -> anyhow::Result<ScriptBuf> {
-    build_envelope_script(&envelope_data, rollup_name, version)
+    build_envelope_script(params.as_ref(), payloads, version)
 }
 
 pub fn build_reveal_transaction_test(

crates/btcio/src/writer/builder.rs

@@ -22,7 +22,7 @@ use bitcoin::{
 use rand::{rngs::OsRng, RngCore};
 use strata_config::btcio::FeePolicy;
 use strata_l1tx::envelope::builder::build_envelope_script;
-use strata_primitives::l1::payload::L1Payload;
+use strata_primitives::{l1::payload::L1Payload, params::Params};
 use thiserror::Error;
 
 use super::context::WriterContext;
@@ -52,7 +52,7 @@ pub enum EnvelopeError {
 // dependencies on `tx-parser`, we include {btcio, feature="strata_test_utils"} , so cyclic
 // dependency doesn't happen
 pub async fn build_envelope_txs<W: WriterRpc>(
-    payload: &L1Payload,
+    payloads: &[L1Payload],
     ctx: &WriterContext<W>,
 ) -> anyhow::Result<(Transaction, Transaction)> {
     let network = ctx.client.network().await?;
@@ -62,25 +62,25 @@ pub async fn build_envelope_txs<W: WriterRpc>(
         FeePolicy::Smart => ctx.client.estimate_smart_fee(1).await? * 2,
         FeePolicy::Fixed(val) => val,
     };
-    create_envelope_transactions(ctx, payload, utxos, fee_rate, network)
+    create_envelope_transactions(ctx, payloads, utxos, fee_rate, network)
         .map_err(|e| anyhow::anyhow!(e.to_string()))
 }
 
 #[allow(clippy::too_many_arguments)]
 pub fn create_envelope_transactions<W: WriterRpc>(
     ctx: &WriterContext<W>,
-    payload: &L1Payload,
+    payloads: &[L1Payload],
     utxos: Vec<ListUnspent>,
     fee_rate: u64,
     network: Network,
 ) -> Result<(Transaction, Transaction), EnvelopeError> {
     // Create commit key
     let key_pair = generate_key_pair()?;
     let public_key = XOnlyPublicKey::from_keypair(&key_pair).0;
-    let rollup_name = ctx.params.rollup().rollup_name.clone();
 
     // Start creating envelope content
-    let reveal_script = build_reveal_script(&rollup_name, &public_key, payload, ENVELOPE_VERSION)?;
+    let reveal_script =
+        build_reveal_script(ctx.params.as_ref(), &public_key, payloads, ENVELOPE_VERSION)?;
 
     // Create spend info for tapscript
     let taproot_spend_info = TaprootBuilder::new()
@@ -379,17 +379,17 @@ pub fn generate_key_pair() -> Result<UntweakedKeypair, anyhow::Error> {
 /// Builds reveal script such that it contains opcodes for verifying the internal key as well as the
 /// envelope block
 fn build_reveal_script(
-    rollup_name: &str,
+    params: &Params,
     taproot_public_key: &XOnlyPublicKey,
-    envelope_data: &L1Payload,
+    payloads: &[L1Payload],
     version: u8,
 ) -> Result<ScriptBuf, anyhow::Error> {
     let mut script_bytes = script::Builder::new()
         .push_x_only_key(taproot_public_key)
         .push_opcode(OP_CHECKSIG)
         .into_script()
         .into_bytes();
-    let script = build_envelope_script(envelope_data, rollup_name, version)?;
+    let script = build_envelope_script(params, payloads, version)?;
     script_bytes.extend(script.into_bytes());
     Ok(ScriptBuf::from(script_bytes))
 }
@@ -674,7 +674,7 @@ mod tests {
         let payload = L1Payload::new_da(vec![0u8; 100]);
         let (commit, reveal) = super::create_envelope_transactions(
             &ctx,
-            &payload,
+            &[payload],
             utxos.to_vec(),
             10,
             bitcoin::Network::Bitcoin,

crates/btcio/src/writer/bundler.rs

@@ -0,0 +1,83 @@
+use std::{sync::Arc, time::Duration};
+
+use strata_db::types::{BundledPayloadEntry, IntentEntry, IntentStatus};
+use strata_storage::ops::writer::EnvelopeDataOps;
+use tokio::time::sleep;
+use tracing::*;
+
+// TODO: get this from config
+const BUNDLE_INTERVAL: u64 = 200; // millis
+
+/// Periodically bundles unbundled intents into payload entries.
+pub(crate) async fn bundler_task(ops: Arc<EnvelopeDataOps>) -> anyhow::Result<()> {
+    let mut last_idx = 0;
+    loop {
+        let (unbundled, new_idx) = get_unbundled_intents_after(last_idx, ops.as_ref()).await?;
+        process_unbundled_entries(ops.as_ref(), unbundled).await?;
+        last_idx = new_idx;
+
+        let _ = sleep(Duration::from_millis(BUNDLE_INTERVAL)).await;
+    }
+}
+
+/// Processes and bundles a list of unbundled intents into payload entries.
+/// NOTE: The current logic is simply 1-1 mapping between intents and payloads, in future it can
+/// be sophisticated.
+async fn process_unbundled_entries(
+    ops: &EnvelopeDataOps,
+    unbundled: Vec<IntentEntry>,
+) -> anyhow::Result<()> {
+    for mut entry in unbundled {
+        // NOTE: In future, the logic to create payload will be different. We need to group
+        // intents and create payload entries accordingly
+        let payload_entry = BundledPayloadEntry::new_unsigned(vec![entry.payload().clone()]);
+
+        // TODO: the following block till "Atomic Ends" should be atomic.
+        let idx = ops.get_next_payload_idx_async().await?;
+        ops.put_payload_entry_async(idx, payload_entry).await?;
+
+        // Set the entry to be bundled so that it won't be processed next time.
+        entry.status = IntentStatus::Bundled(idx);
+        ops.put_intent_entry_async(*entry.intent.commitment(), entry)
+            .await?;
+        // Atomic Ends.
+    }
+    Ok(())
+}
+
+/// Retrieves unbundled intents after a given index in ascending order along with the latest
+/// unbundled entry idx.
+async fn get_unbundled_intents_after(
+    idx: u64,
+    ops: &EnvelopeDataOps,
+) -> anyhow::Result<(Vec<IntentEntry>, u64)> {
+    let latest_idx = ops.get_next_intent_idx_async().await?.saturating_sub(1);
+    let mut curr_idx = latest_idx;
+
+    let mut unbundled = Vec::new();
+
+    while curr_idx >= idx {
+        if let Some(intent) = ops.get_intent_by_idx_async(curr_idx).await? {
+            match intent.status {
+                IntentStatus::Unbundled => unbundled.push(intent),
+                IntentStatus::Bundled(_) => {
+                    // Bundled intent found, no more to scan
+                    break;
+                }
+            }
+        } else {
+            warn!(%curr_idx, "Could not find expected intent in db");
+            break;
+        }
+
+        if curr_idx == 0 {
+            break;
+        }
+        curr_idx -= 1;
+    }
+
+    // Reverse the items so that they are in ascending order of index
+    unbundled.reverse();
+
+    Ok((unbundled, latest_idx))
+}

crates/btcio/src/writer/mod.rs

@@ -1,4 +1,5 @@
 pub mod builder;
+mod bundler;
 pub(crate) mod context;
 mod signer;
 mod task;

crates/btcio/src/writer/signer.rs

@@ -1,7 +1,7 @@
 use std::sync::Arc;
 
 use bitcoin::{consensus, Transaction};
-use strata_db::types::{L1TxEntry, PayloadEntry};
+use strata_db::types::{BundledPayloadEntry, L1TxEntry};
 use strata_primitives::buf::Buf32;
 use tracing::*;
 
@@ -20,12 +20,12 @@ type BlobIdx = u64;
 /// 2. A signed intent needs to be resigned because somehow its inputs were spent/missing
 /// 3. A confirmed block that includes the tx gets reorged
 pub async fn create_and_sign_payload_envelopes<W: WriterRpc>(
-    payloadentry: &PayloadEntry,
+    payloadentry: &BundledPayloadEntry,
     broadcast_handle: &L1BroadcastHandle,
     ctx: Arc<WriterContext<W>>,
 ) -> Result<(Buf32, Buf32), EnvelopeError> {
     trace!("Creating and signing payload envelopes");
-    let (commit, reveal) = build_envelope_txs(&payloadentry.payload, ctx.as_ref()).await?;
+    let (commit, reveal) = build_envelope_txs(&payloadentry.payloads, ctx.as_ref()).await?;
 
     let ctxid = commit.compute_txid();
     debug!(commit_txid = ?ctxid, "Signing commit transaction");
@@ -59,8 +59,8 @@ pub async fn create_and_sign_payload_envelopes<W: WriterRpc>(
 
 #[cfg(test)]
 mod test {
-    use strata_db::types::{PayloadEntry, PayloadL1Status};
-    use strata_primitives::{hash, l1::payload::L1Payload};
+    use strata_db::types::{BundledPayloadEntry, L1BundleStatus};
+    use strata_primitives::l1::payload::L1Payload;
 
     use super::*;
     use crate::{
@@ -75,14 +75,14 @@ mod test {
         let ctx = get_writer_context();
 
         // First insert an unsigned blob
-        let entry = PayloadEntry::new_unsigned(L1Payload::new_da([1; 100].to_vec()));
+        let payload = L1Payload::new_da([1; 100].to_vec());
+        let entry = BundledPayloadEntry::new_unsigned(vec![payload]);
 
-        assert_eq!(entry.status, PayloadL1Status::Unsigned);
+        assert_eq!(entry.status, L1BundleStatus::Unsigned);
         assert_eq!(entry.commit_txid, Buf32::zero());
         assert_eq!(entry.reveal_txid, Buf32::zero());
 
-        let intent_hash = hash::raw(entry.payload.data());
-        iops.put_payload_entry_async(intent_hash, entry.clone())
+        iops.put_payload_entry_async(0, entry.clone())
             .await
             .unwrap();