add invalid commit floor corruption detection

Author: roberto-bayardo

storage/src/qmdb/any/db.rs

@@ -524,6 +524,11 @@ where
                 .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",
+                ));
+            }
             if let Some(known_inactivity_floor) = known_inactivity_floor {
                 (*known_inactivity_floor..*inactivity_floor_loc)
                     .for_each(|_| callback(false, None));

storage/src/qmdb/immutable/mod.rs

@@ -208,6 +208,9 @@ where
             let inactivity_floor_loc = last_op
                 .has_floor()
                 .expect("last operation should be a commit with floor");
+            if inactivity_floor_loc > last_commit_loc {
+                return Err(Error::DataCorrupted("inactivity floor exceeds last commit"));
+            }
 
             // Replay the log from the inactivity floor to build the snapshot.
             build_snapshot_from_log::<F, _, _, _>(

storage/src/qmdb/store/db.rs

@@ -370,6 +370,11 @@ where
             let reader = log.reader().await;
             let op = reader.read(*last_commit_loc).await?;
             let inactivity_floor_loc = op.has_floor().expect("last op should be a commit");
+            if inactivity_floor_loc > last_commit_loc {
+                return Err(crate::qmdb::Error::DataCorrupted(
+                    "inactivity floor exceeds last commit",
+                ));
+            }
             let active_keys =
                 build_snapshot_from_log(inactivity_floor_loc, &reader, &mut snapshot, |_, _| {})
                     .await?;