Reject double-spends of nullifiers in the finalized state

Applies a RocksDB merge operator to the nullifier column families,
so each nullifier key can only be written once.

As a defence-in-depth against state corruption,
we also prevent double-inserts of:
- block heights
- block hashes
- transaction hashes
- transparent `OutPoint`s

But we expect that earlier checks should catch these double-inserts,
before they are committed to RocksDB.

zebra-state/src/service/finalized_state.rs

@@ -1,6 +1,7 @@
 //! The primary implementation of the `zebra_state::Service` built upon rocksdb
 
 mod disk_format;
+mod merge_operator;
 
 #[cfg(test)]
 mod tests;
@@ -16,7 +17,10 @@ use zebra_chain::{
 
 use crate::{BoxError, Config, FinalizedBlock, HashOrHeight, Utxo};
 
-use self::disk_format::{DiskDeserialize, DiskSerialize, FromDisk, IntoDisk, TransactionLocation};
+use self::{
+    disk_format::{DiskDeserialize, DiskSerialize, FromDisk, IntoDisk, TransactionLocation},
+    merge_operator::reject_updates_to_existing_values,
+};
 
 use super::QueuedFinalized;
 
@@ -39,16 +43,36 @@ pub struct FinalizedState {
 impl FinalizedState {
     pub fn new(config: &Config, network: Network) -> Self {
         let (path, db_options) = config.db_config(network);
+
+        // Reject updates to existing values in column families
+        // with the following Merge operators:
+        // - Never: don't update existing values or delete keys, or
+        // - Delete: don't update existing values, but allow key deletion.
+        //
+        // For details see:
+        // https://github.com/ZcashFoundation/zebra/blob/main/book/src/dev/rfcs/0005-state-updates.md#rocksdb-data-structures
+        let mut reject_updates = db_options.clone();
+        // the merge operator name is saved in the database
+        // changing the name requires a Zebra database version increment
+        reject_updates.set_merge_operator_associative(
+            "reject_updates_to_existing_values",
+            reject_updates_to_existing_values,
+        );
+
         let column_families = vec![
-            rocksdb::ColumnFamilyDescriptor::new("hash_by_height", db_options.clone()),
-            rocksdb::ColumnFamilyDescriptor::new("height_by_hash", db_options.clone()),
-            rocksdb::ColumnFamilyDescriptor::new("block_by_height", db_options.clone()),
-            rocksdb::ColumnFamilyDescriptor::new("tx_by_hash", db_options.clone()),
-            rocksdb::ColumnFamilyDescriptor::new("utxo_by_outpoint", db_options.clone()),
-            rocksdb::ColumnFamilyDescriptor::new("sprout_nullifiers", db_options.clone()),
-            rocksdb::ColumnFamilyDescriptor::new("sapling_nullifiers", db_options.clone()),
-            rocksdb::ColumnFamilyDescriptor::new("orchard_nullifiers", db_options.clone()),
+            rocksdb::ColumnFamilyDescriptor::new("hash_by_height", reject_updates.clone()),
+            rocksdb::ColumnFamilyDescriptor::new("height_by_hash", reject_updates.clone()),
+            rocksdb::ColumnFamilyDescriptor::new("block_by_height", reject_updates.clone()),
+            rocksdb::ColumnFamilyDescriptor::new("tx_by_hash", reject_updates.clone()),
+            rocksdb::ColumnFamilyDescriptor::new("utxo_by_outpoint", reject_updates.clone()),
+            rocksdb::ColumnFamilyDescriptor::new("sprout_nullifiers", reject_updates.clone()),
+            rocksdb::ColumnFamilyDescriptor::new("sapling_nullifiers", reject_updates.clone()),
+            rocksdb::ColumnFamilyDescriptor::new("orchard_nullifiers", reject_updates),
+            // value_pools allows Update merges:
+            // chain value pool balances get updated for each block
+            //rocksdb::ColumnFamilyDescriptor::new("value_pools", db_options.clone()),
         ];
+
         let db_result = rocksdb::DB::open_cf_descriptors(&db_options, &path, column_families);
 
         let db = match db_result {