re-apply pyramid-mmb work onto main bitmap API

Reconstructs the policy and sync changes that were dropped when the rebase
accepted main's version for the conflicting files. Adds split_root and
root_bagging to qmdb::any::Config and Db, threads policy through sync
(build_db, proof_spec, has_local_target_state), updates init_from_log to
take + cache the policy, recomputes the cached root after rewind, and
restores the floor-bound corruption check, all on top of main's
shared-bitmap reorg.

Author: roberto-bayardo

storage/src/qmdb/any/db.rs

@@ -13,7 +13,7 @@ use crate::{
         contiguous::{Contiguous, Mutable, Reader},
         Error as JournalError,
     },
-    merkle::{Family, Location, Proof},
+    merkle::{Bagging, Family, Location, Proof, RootSpec},
     qmdb::{
         bitmap::Shared, build_snapshot_from_log, delete_known_loc,
         operation::Operation as OperationTrait, update_known_loc, Error,
@@ -73,6 +73,15 @@ pub struct Db<
     /// - There is always at least one commit operation in the log.
     pub(crate) log: AuthenticatedLog<F, E, C, H>,
 
+    /// Cached operations root for this database.
+    pub(crate) root: H::Digest,
+
+    /// Whether the operations root uses inactive-prefix split semantics.
+    pub(crate) split_root: bool,
+
+    /// Bagging used by the operations root.
+    pub(crate) root_bagging: Bagging,
+
     /// A location before which all operations are "inactive" (that is, operations before this point
     /// are over keys that have been updated by some operation at or after this point).
     pub(crate) inactivity_floor_loc: Location<F>,
@@ -145,8 +154,15 @@ where
         }
     }
 
-    pub fn root(&self) -> H::Digest {
-        self.log.root()
+    /// Return the canonical QMDB operations root.
+    pub const fn root(&self) -> H::Digest {
+        self.root
+    }
+
+    /// Return the operations-root spec for the given leaf count and inactivity floor.
+    pub(crate) fn root_spec(&self, leaves: Location<F>, inactivity_floor: Location<F>) -> RootSpec {
+        let inactive_peaks = F::inactive_peaks(F::location_to_position(leaves), inactivity_floor);
+        RootSpec::from_split_policy(self.split_root, self.root_bagging, inactive_peaks)
     }
 
     /// Get the value of `key` in the db, or None if it has no value.
@@ -302,14 +318,46 @@ where
         Ok(())
     }
 
+    /// Returns a historical proof for `historical_size` operations, anchored at `start_loc`
+    /// and bounded by `max_ops`.
+    ///
+    /// # Contract
+    ///
+    /// When this database is configured with `split_root: true`, `historical_size` must be a
+    /// commit-boundary size: the operation at `historical_size - 1` must itself be a commit
+    /// op declaring the governing inactivity floor. With `split_root: false` the spec does
+    /// not depend on the floor and any retained `historical_size` works.
+    ///
+    /// # Errors
+    ///
+    /// Returns [`crate::qmdb::Error::HistoricalFloorPruned`] (split_root only) if
+    /// `historical_size - 1` is retained but is not a commit op, either because the caller
+    /// passed a non-commit-boundary size or because pruning removed the commit that would
+    /// have governed it.
     pub async fn historical_proof(
         &self,
         historical_size: Location<F>,
         start_loc: Location<F>,
         max_ops: NonZeroU64,
     ) -> Result<(Proof<F, H::Digest>, Vec<Operation<F, U>>), crate::qmdb::Error<F>> {
+        if historical_size > self.log.size().await {
+            return Err(crate::qmdb::Error::Merkle(
+                crate::merkle::Error::RangeOutOfBounds(historical_size),
+            ));
+        }
+
+        let inactivity_floor = if self.split_root {
+            let reader = self.log.reader().await;
+            crate::qmdb::find_inactivity_floor_at::<F, _>(&reader, historical_size, |op| {
+                op.has_floor()
+            })
+            .await?
+        } else {
+            Location::new(0)
+        };
+        let spec = self.root_spec(historical_size, inactivity_floor);
         self.log
-            .historical_proof(historical_size, start_loc, max_ops)
+            .historical_proof(historical_size, start_loc, max_ops, spec)
             .await
             .map_err(Into::into)
     }
@@ -484,6 +532,9 @@ where
             ))?;
         self.last_commit_loc = Location::new(rewind_size - 1);
         self.inactivity_floor_loc = rewind_floor;
+        self.root = self
+            .log
+            .root(self.root_spec(Location::new(rewind_size), rewind_floor))?;
 
         Ok(())
     }
@@ -512,13 +563,20 @@ where
         mut index: I,
         log: AuthenticatedLog<F, E, C, H>,
         shared_bitmap: Option<Arc<Shared<N>>>,
+        split_root: bool,
+        root_bagging: Bagging,
     ) -> Result<Self, crate::qmdb::Error<F>> {
         let (last_commit_loc, inactivity_floor_loc, active_keys, bitmap) = {
             let reader = log.reader().await;
             let bounds = reader.bounds();
             let last_commit_loc = bounds.end.checked_sub(1).expect("commit should exist");
             let last_commit = reader.read(last_commit_loc).await?;
             let inactivity_floor_loc = last_commit.has_floor().expect("should be a commit");
+            if *inactivity_floor_loc > last_commit_loc {
+                return Err(crate::qmdb::Error::DataCorrupted(
+                    "inactivity floor exceeds last commit",
+                ));
+            }
 
             // Seed the bitmap so its pruned prefix matches the retained log boundary. Bits in
             // [pruned_bits, bounds.start) correspond to pruned operations and remain 0; replay
@@ -585,8 +643,21 @@ where
             ));
         }
 
+        let inactive_peaks = F::inactive_peaks(
+            F::location_to_position(log.merkle.leaves()),
+            inactivity_floor_loc,
+        );
+        let root = log.root(RootSpec::from_split_policy(
+            split_root,
+            root_bagging,
+            inactive_peaks,
+        ))?;
+
         Ok(Self {
             log,
+            root,
+            split_root,
+            root_bagging,
             inactivity_floor_loc,
             snapshot: index,
             last_commit_loc,