feat(storage): store per-block events in a new column family

Cargo.lock

@@ -11752,6 +11752,7 @@ dependencies = [
  "borsh",
  "common",
  "lee",
+ "lee_core",
  "log",
  "programs",
  "rocksdb",

lez/indexer/core/src/block_store.rs

@@ -260,7 +260,7 @@ impl IndexerStore {
         let mut stored = block.clone();
         stored.bedrock_status = BedrockStatus::Finalized;
         self.dbio
-            .put_block(&stored, [0_u8; 32], l1_slot.into_inner(), &scratch)
+            .put_block(&stored, [0_u8; 32], l1_slot.into_inner(), &scratch, &[])
             .context("Failed to persist accepted block")?;
 
         // Commit in-memory state (infallible) only after the DB write succeeded.

lez/storage/Cargo.toml

@@ -19,5 +19,6 @@ tempfile.workspace = true
 zstd.workspace = true
 
 [dev-dependencies]
+lee_core.workspace = true
 programs.workspace = true
 system_accounts.workspace = true

lez/storage/src/indexer/indexer_cells.rs

@@ -1,4 +1,5 @@
 use borsh::{BorshDeserialize, BorshSerialize};
+use common::transaction::TxEvents;
 use lee::V03State;
 
 use crate::{
@@ -6,10 +7,11 @@ use crate::{
     cells::{SimpleReadableCell, SimpleStorableCell, SimpleWritableCell},
     error::DbError,
     indexer::{
-        ACC_NUM_CELL_NAME, BLOCK_HASH_CELL_NAME, BREAKPOINT_CELL_NAME, CF_ACC_META,
-        CF_BREAKPOINT_NAME, CF_HASH_TO_ID, CF_TX_TO_ID, DB_META_CROSS_ZONE_HALT_KEY,
-        DB_META_LAST_OBSERVED_L1_LIB_HEADER_ID_IN_DB_KEY, DB_META_STALL_REASON_KEY,
-        DB_META_TIP_SLOT_KEY, DB_META_ZONE_SDK_INDEXER_CURSOR_KEY, TX_HASH_CELL_NAME,
+        ACC_NUM_CELL_NAME, BLOCK_EVENTS_CELL_NAME, BLOCK_HASH_CELL_NAME, BREAKPOINT_CELL_NAME,
+        CF_ACC_META, CF_BREAKPOINT_NAME, CF_EVENTS, CF_HASH_TO_ID, CF_TX_TO_ID,
+        DB_META_CROSS_ZONE_HALT_KEY, DB_META_LAST_OBSERVED_L1_LIB_HEADER_ID_IN_DB_KEY,
+        DB_META_STALL_REASON_KEY, DB_META_TIP_SLOT_KEY, DB_META_ZONE_SDK_INDEXER_CURSOR_KEY,
+        TX_HASH_CELL_NAME,
     },
 };
 
@@ -122,6 +124,60 @@ impl SimpleWritableCell for BlockHashToBlockIdMapCell {
     }
 }
 
+#[derive(Debug, BorshSerialize, BorshDeserialize)]
+pub struct BlockEventsCellOwned(pub Vec<TxEvents>);
+
+impl SimpleStorableCell for BlockEventsCellOwned {
+    type KeyParams = u64;
+
+    const CELL_NAME: &'static str = BLOCK_EVENTS_CELL_NAME;
+    const CF_NAME: &'static str = CF_EVENTS;
+
+    fn key_constructor(params: Self::KeyParams) -> DbResult<Vec<u8>> {
+        borsh::to_vec(&params).map_err(|err| {
+            DbError::borsh_cast_message(
+                err,
+                Some(format!(
+                    "Failed to serialize {:?} key params",
+                    Self::CELL_NAME
+                )),
+            )
+        })
+    }
+}
+
+impl SimpleReadableCell for BlockEventsCellOwned {}
+
+#[derive(BorshSerialize)]
+pub struct BlockEventsCellRef<'events>(pub &'events [TxEvents]);
+
+impl SimpleStorableCell for BlockEventsCellRef<'_> {
+    type KeyParams = u64;
+
+    const CELL_NAME: &'static str = BLOCK_EVENTS_CELL_NAME;
+    const CF_NAME: &'static str = CF_EVENTS;
+
+    fn key_constructor(params: Self::KeyParams) -> DbResult<Vec<u8>> {
+        borsh::to_vec(&params).map_err(|err| {
+            DbError::borsh_cast_message(
+                err,
+                Some(format!(
+                    "Failed to serialize {:?} key params",
+                    Self::CELL_NAME
+                )),
+            )
+        })
+    }
+}
+
+impl SimpleWritableCell for BlockEventsCellRef<'_> {
+    fn value_constructor(&self) -> DbResult<Vec<u8>> {
+        borsh::to_vec(&self).map_err(|err| {
+            DbError::borsh_cast_message(err, Some("Failed to serialize block events".to_owned()))
+        })
+    }
+}
+
 #[derive(Debug, BorshSerialize, BorshDeserialize)]
 pub struct TxHashToBlockIdMapCell(pub u64);
 

lez/storage/src/indexer/mod.rs

@@ -38,6 +38,8 @@ pub const BLOCK_HASH_CELL_NAME: &str = "block hash";
 pub const TX_HASH_CELL_NAME: &str = "tx hash";
 /// Cell name for a account number of transactions.
 pub const ACC_NUM_CELL_NAME: &str = "acc id";
+/// Cell name for the events emitted by a block's transactions.
+pub const BLOCK_EVENTS_CELL_NAME: &str = "block events";
 
 /// Name of breakpoint column family.
 pub const CF_BREAKPOINT_NAME: &str = "cf_breakpoint";
@@ -49,6 +51,8 @@ pub const CF_TX_TO_ID: &str = "cf_tx_to_id";
 pub const CF_ACC_META: &str = "cf_acc_meta";
 /// Name of account id to tx hash map column family.
 pub const CF_ACC_TO_TX: &str = "cf_acc_to_tx";
+/// Name of per-block events column family.
+pub const CF_EVENTS: &str = "cf_events";
 
 pub struct RocksDBIO {
     pub db: DBWithThreadMode<MultiThreaded>,
@@ -73,6 +77,7 @@ impl RocksDBIO {
         let cftti = ColumnFamilyDescriptor::new(CF_TX_TO_ID, cf_opts.clone());
         let cfameta = ColumnFamilyDescriptor::new(CF_ACC_META, cf_opts.clone());
         let cfatt = ColumnFamilyDescriptor::new(CF_ACC_TO_TX, cf_opts.clone());
+        let cfevents = ColumnFamilyDescriptor::new(CF_EVENTS, cf_opts.clone());
 
         let mut db_opts = Options::default();
         db_opts.create_missing_column_families(true);
@@ -80,7 +85,16 @@ impl RocksDBIO {
         let db = DBWithThreadMode::<MultiThreaded>::open_cf_descriptors(
             &db_opts,
             path,
-            vec![cfb, cfmeta, cfbreakpoint, cfhti, cftti, cfameta, cfatt],
+            vec![
+                cfb,
+                cfmeta,
+                cfbreakpoint,
+                cfhti,
+                cftti,
+                cfameta,
+                cfatt,
+                cfevents,
+            ],
         )
         .map_err(|err| DbError::RocksDbError {
             error: err,
@@ -141,6 +155,12 @@ impl RocksDBIO {
             .expect("Account id to tx map column should exist")
     }
 
+    pub fn events_column(&self) -> Arc<BoundColumnFamily<'_>> {
+        self.db
+            .cf_handle(CF_EVENTS)
+            .expect("Events column should exist")
+    }
+
     pub fn account_meta_column(&self) -> Arc<BoundColumnFamily<'_>> {
         self.db
             .cf_handle(CF_ACC_META)

lez/storage/src/indexer/read_multiple.rs

@@ -1,4 +1,4 @@
-use common::transaction::LeeTransaction;
+use common::transaction::{LeeTransaction, TxEvents};
 
 use super::{Block, DbError, DbResult, RocksDBIO};
 
@@ -76,6 +76,50 @@ impl RocksDBIO {
         Ok(block_batch)
     }
 
+    // A block whose transactions emitted nothing has no row at all, so unlike
+    // `get_block_batch_seq` a missing key is expected here and must be skipped rather
+    // than terminate the scan.
+    //
+    // Callers bound the span: this materializes one key per block in `from..=to`, so an
+    // unbounded range allocates proportionally.
+    pub fn get_block_events_range(
+        &self,
+        from: u64,
+        to: u64,
+    ) -> DbResult<Vec<(u64, Vec<TxEvents>)>> {
+        let cf_events = self.events_column();
+
+        let block_ids: Vec<u64> = (from..=to).collect();
+        let mut keys = Vec::with_capacity(block_ids.len());
+        for block_id in &block_ids {
+            keys.push((
+                &cf_events,
+                borsh::to_vec(block_id).map_err(|err| {
+                    DbError::borsh_cast_message(
+                        err,
+                        Some("Failed to serialize block id".to_owned()),
+                    )
+                })?,
+            ));
+        }
+
+        let mut block_events = vec![];
+        for (block_id, res) in block_ids.iter().zip(self.db.multi_get_cf(keys)) {
+            let Some(data) = res.map_err(|rerr| DbError::rocksdb_cast_message(rerr, None))? else {
+                continue;
+            };
+            let events = borsh::from_slice::<Vec<TxEvents>>(&data).map_err(|serr| {
+                DbError::borsh_cast_message(
+                    serr,
+                    Some("Failed to deserialize block events".to_owned()),
+                )
+            })?;
+            block_events.push((*block_id, events));
+        }
+
+        Ok(block_events)
+    }
+
     /// Get block ids by txs.
     ///
     /// `ToDo`: There may be multiple transactions in one block

lez/storage/src/indexer/read_once.rs

@@ -1,11 +1,13 @@
+use common::transaction::TxEvents;
+
 use super::{Block, DbResult, RocksDBIO, V03State};
 use crate::{
     DBIO as _,
     cells::shared_cells::{BlockCell, FirstBlockCell, FirstBlockSetCell, LastBlockCell},
     indexer::indexer_cells::{
-        AccNumTxCell, BlockHashToBlockIdMapCell, BreakpointCellOwned, CrossZoneHaltCellOwned,
-        LastObservedL1LibHeaderCell, StallReasonCellOwned, TipSlotCell, TxHashToBlockIdMapCell,
-        ZoneSdkIndexerCursorCellOwned,
+        AccNumTxCell, BlockEventsCellOwned, BlockHashToBlockIdMapCell, BreakpointCellOwned,
+        CrossZoneHaltCellOwned, LastObservedL1LibHeaderCell, StallReasonCellOwned, TipSlotCell,
+        TxHashToBlockIdMapCell, ZoneSdkIndexerCursorCellOwned,
     },
 };
 
@@ -44,6 +46,11 @@ impl RocksDBIO {
             .map(|opt| opt.map(|val| val.0))
     }
 
+    pub fn get_block_events(&self, block_id: u64) -> DbResult<Option<Vec<TxEvents>>> {
+        self.get_opt::<BlockEventsCellOwned>(block_id)
+            .map(|opt| opt.map(|cell| cell.0))
+    }
+
     // State
 
     pub fn get_breakpoint(&self, br_id: u64) -> DbResult<V03State> {

lez/storage/src/indexer/tests.rs

@@ -1,4 +1,4 @@
-use common::test_utils::produce_dummy_block;
+use common::{test_utils::produce_dummy_block, transaction::TxEvents};
 use lee::{Account, AccountId, PublicKey};
 use tempfile::tempdir;
 
@@ -87,7 +87,7 @@ fn one_block_insertion() {
     let dbio = RocksDBIO::open_or_create(temdir_path, &initial_state).unwrap();
 
     let genesis_block = genesis_block();
-    dbio.put_block(&genesis_block, [0; 32], 0, &initial_state)
+    dbio.put_block(&genesis_block, [0; 32], 0, &initial_state, &[])
         .unwrap();
 
     let prev_hash = genesis_block.header.hash;
@@ -99,7 +99,8 @@ fn one_block_insertion() {
         common::test_utils::create_transaction_native_token_transfer(from, 0, to, 1, &sign_key);
     let block = produce_dummy_block(2, Some(prev_hash), vec![transfer_tx]);
 
-    dbio.put_block(&block, [1; 32], 0, &initial_state).unwrap();
+    dbio.put_block(&block, [1; 32], 0, &initial_state, &[])
+        .unwrap();
 
     let last_id = dbio.get_meta_last_block_id_in_db().unwrap().unwrap();
     let first_id = dbio.get_meta_first_block_id_in_db().unwrap();
@@ -138,17 +139,17 @@ fn put_block_records_tip_inscription_slot() {
     assert_eq!(dbio.get_meta_tip_slot_in_db().unwrap(), None);
 
     let genesis_block = genesis_block();
-    dbio.put_block(&genesis_block, [0; 32], 1_000, &initial_state)
+    dbio.put_block(&genesis_block, [0; 32], 1_000, &initial_state, &[])
         .unwrap();
     assert_eq!(dbio.get_meta_tip_slot_in_db().unwrap(), Some(1_000));
 
     let block = produce_dummy_block(2, Some(genesis_block.header.hash), vec![]);
-    dbio.put_block(&block, [1; 32], 1_005, &initial_state)
+    dbio.put_block(&block, [1; 32], 1_005, &initial_state, &[])
         .unwrap();
     assert_eq!(dbio.get_meta_tip_slot_in_db().unwrap(), Some(1_005));
 
     // Re-inserting a block at/below the tip must not move the tip slot.
-    dbio.put_block(&genesis_block, [0; 32], 1_010, &initial_state)
+    dbio.put_block(&genesis_block, [0; 32], 1_010, &initial_state, &[])
         .unwrap();
     assert_eq!(dbio.get_meta_tip_slot_in_db().unwrap(), Some(1_005));
 }
@@ -182,7 +183,8 @@ fn put_block_stores_breakpoint_in_same_batch() {
         );
         let block = produce_dummy_block(i.into(), prev_hash, vec![transfer_tx]);
 
-        dbio.put_block(&block, [i; 32], 0, &initial_state).unwrap();
+        dbio.put_block(&block, [i; 32], 0, &initial_state, &[])
+            .unwrap();
     }
 
     let bp1 = dbio.get_breakpoint(1).unwrap();
@@ -221,7 +223,8 @@ fn state_replay_falls_back_over_missing_breakpoints() {
             &sign_key,
         );
         let block = produce_dummy_block(i, prev_hash, vec![transfer_tx]);
-        dbio.put_block(&block, [0; 32], 0, &initial_state).unwrap();
+        dbio.put_block(&block, [0; 32], 0, &initial_state, &[])
+            .unwrap();
     }
 
     // Simulate a store whose boundary snapshot was lost (#605).
@@ -256,7 +259,8 @@ fn simple_maps() {
 
     let control_hash1 = block.header.hash;
 
-    dbio.put_block(&block, [1; 32], 0, &initial_state).unwrap();
+    dbio.put_block(&block, [1; 32], 0, &initial_state, &[])
+        .unwrap();
 
     let last_id = dbio.get_meta_last_block_id_in_db().unwrap().unwrap();
     let last_block = dbio.get_block(last_id).unwrap().unwrap();
@@ -268,7 +272,8 @@ fn simple_maps() {
 
     let control_hash2 = block.header.hash;
 
-    dbio.put_block(&block, [2; 32], 0, &initial_state).unwrap();
+    dbio.put_block(&block, [2; 32], 0, &initial_state, &[])
+        .unwrap();
 
     let last_id = dbio.get_meta_last_block_id_in_db().unwrap().unwrap();
     let last_block = dbio.get_block(last_id).unwrap().unwrap();
@@ -280,7 +285,8 @@ fn simple_maps() {
     let control_tx_hash1 = transfer_tx.hash();
 
     let block = produce_dummy_block(3, Some(prev_hash), vec![transfer_tx]);
-    dbio.put_block(&block, [3; 32], 0, &initial_state).unwrap();
+    dbio.put_block(&block, [3; 32], 0, &initial_state, &[])
+        .unwrap();
 
     let last_id = dbio.get_meta_last_block_id_in_db().unwrap().unwrap();
     let last_block = dbio.get_block(last_id).unwrap().unwrap();
@@ -292,7 +298,8 @@ fn simple_maps() {
     let control_tx_hash2 = transfer_tx.hash();
 
     let block = produce_dummy_block(4, Some(prev_hash), vec![transfer_tx]);
-    dbio.put_block(&block, [4; 32], 0, &initial_state).unwrap();
+    dbio.put_block(&block, [4; 32], 0, &initial_state, &[])
+        .unwrap();
 
     let control_block_id1 = dbio.get_block_id_by_hash(control_hash1.0).unwrap().unwrap();
     let control_block_id2 = dbio.get_block_id_by_hash(control_hash2.0).unwrap().unwrap();
@@ -330,7 +337,8 @@ fn block_batch() {
     let block = produce_dummy_block(1, None, vec![transfer_tx]);
 
     block_res.push(block.clone());
-    dbio.put_block(&block, [1; 32], 0, &initial_state).unwrap();
+    dbio.put_block(&block, [1; 32], 0, &initial_state, &[])
+        .unwrap();
 
     let last_id = dbio.get_meta_last_block_id_in_db().unwrap().unwrap();
     let last_block = dbio.get_block(last_id).unwrap().unwrap();
@@ -341,7 +349,8 @@ fn block_batch() {
     let block = produce_dummy_block(2, Some(prev_hash), vec![transfer_tx]);
 
     block_res.push(block.clone());
-    dbio.put_block(&block, [2; 32], 0, &initial_state).unwrap();
+    dbio.put_block(&block, [2; 32], 0, &initial_state, &[])
+        .unwrap();
 
     let last_id = dbio.get_meta_last_block_id_in_db().unwrap().unwrap();
     let last_block = dbio.get_block(last_id).unwrap().unwrap();
@@ -352,7 +361,8 @@ fn block_batch() {
 
     let block = produce_dummy_block(3, Some(prev_hash), vec![transfer_tx]);
     block_res.push(block.clone());
-    dbio.put_block(&block, [3; 32], 0, &initial_state).unwrap();
+    dbio.put_block(&block, [3; 32], 0, &initial_state, &[])
+        .unwrap();
 
     let last_id = dbio.get_meta_last_block_id_in_db().unwrap().unwrap();
     let last_block = dbio.get_block(last_id).unwrap().unwrap();
@@ -363,7 +373,8 @@ fn block_batch() {
 
     let block = produce_dummy_block(4, Some(prev_hash), vec![transfer_tx]);
     block_res.push(block.clone());
-    dbio.put_block(&block, [4; 32], 0, &initial_state).unwrap();
+    dbio.put_block(&block, [4; 32], 0, &initial_state, &[])
+        .unwrap();
 
     let block_hashes_mem: Vec<[u8; 32]> =
         block_res.into_iter().map(|bl| bl.header.hash.0).collect();
@@ -423,7 +434,8 @@ fn account_map() {
 
     let block = produce_dummy_block(1, None, vec![transfer_tx1, transfer_tx2]);
 
-    dbio.put_block(&block, [1; 32], 0, &initial_state).unwrap();
+    dbio.put_block(&block, [1; 32], 0, &initial_state, &[])
+        .unwrap();
 
     let last_id = dbio.get_meta_last_block_id_in_db().unwrap().unwrap();
     let last_block = dbio.get_block(last_id).unwrap().unwrap();
@@ -438,7 +450,8 @@ fn account_map() {
 
     let block = produce_dummy_block(2, Some(prev_hash), vec![transfer_tx1, transfer_tx2]);
 
-    dbio.put_block(&block, [2; 32], 0, &initial_state).unwrap();
+    dbio.put_block(&block, [2; 32], 0, &initial_state, &[])
+        .unwrap();
 
     let last_id = dbio.get_meta_last_block_id_in_db().unwrap().unwrap();
     let last_block = dbio.get_block(last_id).unwrap().unwrap();
@@ -453,7 +466,8 @@ fn account_map() {
 
     let block = produce_dummy_block(3, Some(prev_hash), vec![transfer_tx1, transfer_tx2]);
 
-    dbio.put_block(&block, [3; 32], 0, &initial_state).unwrap();
+    dbio.put_block(&block, [3; 32], 0, &initial_state, &[])
+        .unwrap();
 
     let last_id = dbio.get_meta_last_block_id_in_db().unwrap().unwrap();
     let last_block = dbio.get_block(last_id).unwrap().unwrap();
@@ -465,7 +479,8 @@ fn account_map() {
 
     let block = produce_dummy_block(4, Some(prev_hash), vec![transfer_tx]);
 
-    dbio.put_block(&block, [4; 32], 0, &initial_state).unwrap();
+    dbio.put_block(&block, [4; 32], 0, &initial_state, &[])
+        .unwrap();
 
     let acc1_tx = dbio.get_acc_transactions(*acc1().value(), 0, 7).unwrap();
     let acc1_tx_hashes: Vec<[u8; 32]> = acc1_tx.into_iter().map(|tx| tx.hash().0).collect();
@@ -490,3 +505,101 @@ fn reopen_preserves_seeded_breakpoint() {
     let dbio = RocksDBIO::open_or_create(temp_dir.path(), &initial_state).unwrap();
     assert!(dbio.get_breakpoint_opt(0).unwrap().is_some());
 }
+
+fn tx_events_fixture(tx_index: u32, tx_hash: [u8; 32]) -> TxEvents {
+    TxEvents {
+        tx_index,
+        tx_hash: tx_hash.into(),
+        events: vec![
+            lee_core::program::TransactionEvent {
+                program_id: [7; 8],
+                event: lee_core::program::ProgramEvent {
+                    selector: [1; 8],
+                    data: vec![1, 2, 3],
+                },
+            },
+            lee_core::program::TransactionEvent {
+                program_id: [9; 8],
+                event: lee_core::program::ProgramEvent {
+                    selector: [2; 8],
+                    data: vec![],
+                },
+            },
+        ],
+    }
+}
+
+#[test]
+fn put_block_stores_events_in_same_batch() {
+    let initial_state = initial_state();
+    let temp_dir = tempdir().unwrap();
+    let dbio = RocksDBIO::open_or_create(temp_dir.path(), &initial_state).unwrap();
+
+    let block = genesis_block();
+    let events = vec![
+        tx_events_fixture(0, [11; 32]),
+        tx_events_fixture(3, [12; 32]),
+    ];
+
+    dbio.put_block(&block, [0; 32], 0, &initial_state, &events)
+        .unwrap();
+
+    // One put_block call makes both the block and its events readable.
+    assert!(dbio.get_block(1).unwrap().is_some());
+    assert_eq!(dbio.get_block_events(1).unwrap(), Some(events));
+}
+
+#[test]
+fn block_without_events_writes_no_row() {
+    let initial_state = initial_state();
+    let temp_dir = tempdir().unwrap();
+    let dbio = RocksDBIO::open_or_create(temp_dir.path(), &initial_state).unwrap();
+
+    dbio.put_block(&genesis_block(), [0; 32], 0, &initial_state, &[])
+        .unwrap();
+
+    assert!(dbio.get_block(1).unwrap().is_some());
+    assert_eq!(dbio.get_block_events(1).unwrap(), None);
+}
+
+#[test]
+fn get_block_events_is_none_for_unknown_block() {
+    let initial_state = initial_state();
+    let temp_dir = tempdir().unwrap();
+    let dbio = RocksDBIO::open_or_create(temp_dir.path(), &initial_state).unwrap();
+
+    assert_eq!(dbio.get_block_events(999).unwrap(), None);
+}
+
+#[test]
+fn get_block_events_range_skips_blocks_without_events() {
+    let initial_state = initial_state();
+    let temp_dir = tempdir().unwrap();
+    let dbio = RocksDBIO::open_or_create(temp_dir.path(), &initial_state).unwrap();
+
+    let mut prev_hash = None;
+    let mut expected = vec![];
+    for block_id in 1..=4_u64 {
+        let block = produce_dummy_block(block_id, prev_hash, vec![]);
+        prev_hash = Some(block.header.hash);
+
+        // Only odd blocks emit.
+        let events = if block_id.is_multiple_of(2) {
+            vec![]
+        } else {
+            vec![tx_events_fixture(0, [u8::try_from(block_id).unwrap(); 32])]
+        };
+        if !events.is_empty() {
+            expected.push((block_id, events.clone()));
+        }
+        dbio.put_block(&block, [0; 32], 0, &initial_state, &events)
+            .unwrap();
+    }
+
+    assert_eq!(dbio.get_block_events_range(1, 4).unwrap(), expected);
+    assert_eq!(
+        dbio.get_block_events_range(2, 2).unwrap(),
+        Vec::<(u64, Vec<TxEvents>)>::new()
+    );
+    assert_eq!(dbio.get_block_events_range(3, 3).unwrap(), expected[1..]);
+}

lez/storage/src/indexer/write_atomic.rs

@@ -1,5 +1,6 @@
 use std::collections::HashMap;
 
+use common::transaction::TxEvents;
 use rocksdb::WriteBatch;
 
 use super::{BREAKPOINT_INTERVAL, Block, DbError, DbResult, RocksDBIO, V03State};
@@ -7,8 +8,8 @@ use crate::{
     DBIO as _,
     cells::shared_cells::{FirstBlockCell, FirstBlockSetCell, LastBlockCell},
     indexer::indexer_cells::{
-        AccNumTxCell, BlockHashToBlockIdMapCell, BreakpointCellRef, LastObservedL1LibHeaderCell,
-        TipSlotCell, TxHashToBlockIdMapCell,
+        AccNumTxCell, BlockEventsCellRef, BlockHashToBlockIdMapCell, BreakpointCellRef,
+        LastObservedL1LibHeaderCell, TipSlotCell, TxHashToBlockIdMapCell,
     },
 };
 
@@ -152,6 +153,7 @@ impl RocksDBIO {
         l1_lib_header: [u8; 32],
         l1_slot: u64,
         post_state: &V03State,
+        events: &[TxEvents],
     ) -> DbResult<()> {
         let cf_block = self.block_column();
         let last_curr_block = self.get_meta_last_block_id_in_db()?.unwrap_or(0);
@@ -229,6 +231,15 @@ impl RocksDBIO {
             self.put_batch(&BreakpointCellRef(post_state), br_id, &mut write_batch)?;
         }
 
+        // No row at all for a block whose transactions emitted nothing.
+        if !events.is_empty() {
+            self.put_batch(
+                &BlockEventsCellRef(events),
+                block.header.block_id,
+                &mut write_batch,
+            )?;
+        }
+
         self.db.write(write_batch).map_err(|rerr| {
             DbError::rocksdb_cast_message(rerr, Some("Failed to write batch".to_owned()))
         })?;