Read sourced history pages without local History rows in the B4 simulation

The page comparison passed the store that still holds every History row
to get_history_events_from_source, so a direct local read or an
accidental fallback would also pass. It now reads through a reader that
fails every History column access, and checks that a local read through
that reader fails.

Author: longquanzheng

server-zd/crates/dexzd-simulation/tests/time_travel_targets.rs

@@ -19,7 +19,10 @@ use dexzd_apply::diagnostics::{
     TargetResolutionForTest, TimeTravelTargetForTest, resolve_time_travel_target_for_test,
 };
 use dexzd_apply::read_model::{get_history_events, get_history_events_from_source};
-use dexzd_apply::{ApplyContext, ApplyError, HistoryRecords, HistorySource, StateWrite};
+use dexzd_apply::{
+    ApplyContext, ApplyError, BoundedRowSeek, FlowStateReader, HistoryRecords, HistorySource,
+    StateReadError, StateRow, StateWrite,
+};
 use dexzd_keyspace::{ColumnFamily, RunId, RunTable, run_table_prefix};
 use dexzd_protocol::dex::{FlowResetStepMethod, FlowResetType, ResetFlowRequest};
 use dexzd_protocol::dexzd::v1::{
@@ -233,6 +236,21 @@ impl CheckedFlow<'_> {
                 continue;
             }
             let history = MemoryHistorySource::of_run(state, &context, run_id, run);
+            // The sourced read must take every History row from `history`, never from the store.
+            let state_without_history = WithoutHistoryColumn { state };
+            let local_without_history = get_history_events(
+                &state_without_history,
+                context.flow_hash,
+                run_id,
+                0,
+                0,
+                &[],
+                0,
+            );
+            assert!(
+                local_without_history.is_err(),
+                "{site}: a local read of run {run_id} passed without History rows"
+            );
             let attempts = [
                 0,
                 1,
@@ -255,7 +273,7 @@ impl CheckedFlow<'_> {
                             panic!("{site}: local history page failed: {error}")
                         });
                         let from_source = get_history_events_from_source(
-                            state,
+                            &state_without_history,
                             &history,
                             context.flow_hash,
                             run_id,
@@ -515,6 +533,57 @@ fn strip_target_fields(
     stripped
 }
 
+/// A reader of `state` that fails every read of the History column family.
+struct WithoutHistoryColumn<'state> {
+    state: &'state MemoryFlowState,
+}
+
+impl WithoutHistoryColumn<'_> {
+    fn reject_history(column_family: ColumnFamily) -> Result<(), StateReadError> {
+        if column_family == ColumnFamily::History {
+            return Err(StateReadError {
+                column_family,
+                detail: "the sourced read accessed local History rows".to_owned(),
+            });
+        }
+        Ok(())
+    }
+}
+
+impl FlowStateReader for WithoutHistoryColumn<'_> {
+    fn get(
+        &self,
+        column_family: ColumnFamily,
+        key: &[u8],
+    ) -> Result<Option<Vec<u8>>, StateReadError> {
+        Self::reject_history(column_family)?;
+        self.state.get(column_family, key)
+    }
+
+    fn scan(
+        &self,
+        column_family: ColumnFamily,
+        start: &[u8],
+        end: &[u8],
+        limit: usize,
+    ) -> Result<Vec<StateRow>, StateReadError> {
+        Self::reject_history(column_family)?;
+        self.state.scan(column_family, start, end, limit)
+    }
+
+    fn seek_first_bounded(
+        &self,
+        column_family: ColumnFamily,
+        start: &[u8],
+        end: &[u8],
+        max_skipped_deletions: u64,
+    ) -> Result<BoundedRowSeek, StateReadError> {
+        Self::reject_history(column_family)?;
+        self.state
+            .seek_first_bounded(column_family, start, end, max_skipped_deletions)
+    }
+}
+
 /// A history source holding a run's records and checkpoints in memory, as a source built from
 /// downloaded history would.
 struct MemoryHistorySource {