server-zd: read history through a HistorySource and select time travel targets from metadata

server-zd/crates/dexzd-apply/src/history/diagnostics.rs

@@ -16,7 +16,7 @@ use dexzd_protocol::dexzd::v1::{
 use prost::Message;
 
 use super::attempt_scope::{AttemptRow, decode_timer_id, read_attempt_rows, run_status_of};
-use super::lineage::{Lineage, genesis_seq};
+use super::lineage::{Lineage, LocalHistorySource, genesis_seq};
 use super::reset_target::resolve_target;
 use super::state_rebuild::{HistoryPoint, RebuiltAttempt, rebuild_attempt_state};
 use crate::command_handlers::CommandScope;
@@ -161,12 +161,26 @@ fn rebuild_time_travel_target(
         _ => return Ok(None),
     };
     let lineage = Lineage::of_attempt(transaction, cursor.rows, selected_attempt)?;
-    let Ok(target) = resolve_target(transaction, &cursor, scope.now_micros, &request, &lineage)?
+    let history = LocalHistorySource::new(transaction, cursor.rows);
+    let Ok(target) = resolve_target(
+        transaction,
+        &history,
+        cursor.rows,
+        &cursor.record,
+        scope.now_micros,
+        &request,
+        &lineage,
+    )?
     else {
         return Ok(None);
     };
-    let Ok(rebuilt) =
-        rebuild_attempt_state(transaction, cursor.rows, &cursor.record, target.point)?
+    let Ok(rebuilt) = rebuild_attempt_state(
+        transaction,
+        &history,
+        cursor.rows,
+        &cursor.record,
+        target.point,
+    )?
     else {
         return Ok(None);
     };
@@ -238,3 +252,163 @@ fn append_bytes(encoded: &mut Vec<u8>, bytes: &[u8]) {
     encoded.extend_from_slice(&(bytes.len() as u64).to_be_bytes());
     encoded.extend_from_slice(bytes);
 }
+
+#[cfg(feature = "test-util")]
+pub use target_resolution::{
+    TargetResolutionForTest, TimeTravelTargetForTest, resolve_time_travel_target_for_test,
+};
+
+/// Test entry points that compare the ways of resolving a TimeTravel target.
+#[cfg(feature = "test-util")]
+mod target_resolution {
+    use std::ops::RangeInclusive;
+
+    use dexzd_keyspace::RunId;
+    use dexzd_protocol::dex::ResetFlowRequest;
+    use dexzd_protocol::dexzd::v1::{CheckpointRecord, CommandReply};
+
+    use crate::history::lineage::{HistoryRecords, HistorySource, Lineage, LocalHistorySource};
+    use crate::history::reset_target::{
+        CandidateSource, resolve_target, resolve_target_from_records_for_test,
+        resolve_target_from_source_for_test,
+    };
+    use crate::state_access::{FlowTransaction, RunRows};
+    use crate::{ApplyContext, ApplyError, FlowStateReader};
+
+    /// Which target resolution [`resolve_time_travel_target_for_test`] runs.
+    #[derive(Debug, Clone, Copy, PartialEq, Eq)]
+    pub enum TargetResolutionForTest {
+        /// TimeTravel's resolution: metadata, or records when the run's metadata is older.
+        Apply,
+        /// TimeTravel's resolution with a history source that fails every read.
+        ApplyWithoutHistoryReads,
+        /// Candidates from metadata rows, even when the run's metadata is older.
+        Metadata,
+        /// Candidates from History records read through the local history source.
+        Records,
+        /// The resolution from before metadata-based selection, reading every window record.
+        RecordOracle,
+    }
+
+    /// A resolved TimeTravel target.
+    #[derive(Debug, Clone, PartialEq, Eq)]
+    pub struct TimeTravelTargetForTest {
+        /// Attempt of the target point.
+        pub run_attempt: u32,
+        /// Run seq of the target point.
+        pub run_seq: u64,
+        /// Largest event ID of the selected lineage's semantic points up to the target.
+        pub last_event_id: i64,
+        /// Message of the last FlowClosed event of those points.
+        pub close_history_message: Option<String>,
+    }
+
+    /// Resolves the target of `request` on run `run_id` the way `resolution` names.
+    ///
+    /// Returns the target or the rejection reply. Returns an error when state cannot be read,
+    /// or when the request names an attempt the run does not have.
+    pub fn resolve_time_travel_target_for_test(
+        context: &ApplyContext,
+        state: &dyn FlowStateReader,
+        run_id: RunId,
+        now_micros: i64,
+        request: &ResetFlowRequest,
+        resolution: TargetResolutionForTest,
+    ) -> Result<Result<TimeTravelTargetForTest, CommandReply>, ApplyError> {
+        let transaction = FlowTransaction::new(state, *context);
+        let rows = RunRows::new(context.flow_hash, run_id);
+        let record =
+            rows.run_record(&transaction)?
+                .ok_or_else(|| ApplyError::InvariantViolation {
+                    detail: format!("run {run_id} has no run record"),
+                })?;
+        let selected_attempt = match u32::try_from(request.run_attempt) {
+            Ok(0) => record.current_attempt,
+            Ok(attempt) if attempt <= record.current_attempt => attempt,
+            _ => {
+                return Err(ApplyError::InvariantViolation {
+                    detail: format!("run attempt {} does not exist", request.run_attempt),
+                });
+            }
+        };
+        let lineage = Lineage::of_attempt(&transaction, rows, selected_attempt)?;
+        let local_history = LocalHistorySource::new(&transaction, rows);
+        let target = match resolution {
+            TargetResolutionForTest::Apply => resolve_target(
+                &transaction,
+                &local_history,
+                rows,
+                &record,
+                now_micros,
+                request,
+                &lineage,
+            )?,
+            TargetResolutionForTest::ApplyWithoutHistoryReads => resolve_target(
+                &transaction,
+                &UnreadableHistorySource,
+                rows,
+                &record,
+                now_micros,
+                request,
+                &lineage,
+            )?,
+            TargetResolutionForTest::Metadata | TargetResolutionForTest::Records => {
+                let source = if resolution == TargetResolutionForTest::Metadata {
+                    CandidateSource::Metadata
+                } else {
+                    CandidateSource::Records
+                };
+                resolve_target_from_source_for_test(
+                    &transaction,
+                    &local_history,
+                    rows,
+                    &record,
+                    now_micros,
+                    request,
+                    &lineage,
+                    source,
+                )?
+            }
+            TargetResolutionForTest::RecordOracle => resolve_target_from_records_for_test(
+                &transaction,
+                &local_history,
+                rows,
+                &record,
+                now_micros,
+                request,
+                &lineage,
+            )?,
+        };
+        Ok(target.map(|target| TimeTravelTargetForTest {
+            run_attempt: target.point.attempt,
+            run_seq: target.point.seq,
+            last_event_id: target.last_event_id,
+            close_history_message: target.close_history_message,
+        }))
+    }
+
+    /// A history source whose every read fails, proving a resolution read no history.
+    struct UnreadableHistorySource;
+
+    impl HistorySource for UnreadableHistorySource {
+        fn history_records(
+            &self,
+            seqs: RangeInclusive<u64>,
+            _batch_records: usize,
+        ) -> HistoryRecords<'_> {
+            Box::new(std::iter::once(Err(ApplyError::InvariantViolation {
+                detail: format!("History records {seqs:?} were read"),
+            })))
+        }
+
+        fn checkpoint(
+            &self,
+            attempt: u32,
+            seq: u64,
+        ) -> Result<Option<CheckpointRecord>, ApplyError> {
+            Err(ApplyError::InvariantViolation {
+                detail: format!("checkpoint ({attempt}, {seq}) was read"),
+            })
+        }
+    }
+}

server-zd/crates/dexzd-apply/src/history/lineage.rs

@@ -6,19 +6,23 @@
 //
 // SPDX-License-Identifier: LicenseRef-Sustainable-Use-1.0
 
-//! Attempt lineages and history record scans.
+//! Attempt lineages, and the history source that streams records and checkpoints.
 //!
 //! Only the current attempt receives commands, so attempts cover disjoint, contiguous seq ranges.
 
-use dexzd_keyspace::row_keys::{history_record_key, prefix_range};
+use std::collections::VecDeque;
+use std::ops::RangeInclusive;
+
+use dexzd_keyspace::row_keys::{checkpoint_key, history_record_key, prefix_range};
 use dexzd_keyspace::{ColumnFamily, RunTable, run_table_prefix};
-use dexzd_protocol::dexzd::v1::{AttemptRecord, HistoryRecord};
+use dexzd_protocol::dexzd::v1::{AttemptRecord, CheckpointRecord, HistoryRecord};
 
 use crate::ApplyError;
+use crate::history::attempt_scope::hex;
 use crate::state_access::{FlowTransaction, RunRows};
 
-/// History records read per scan page.
-const HISTORY_PAGE_RECORDS: usize = 64;
+/// History records read per batch by time travel's history scans.
+pub(crate) const HISTORY_PAGE_RECORDS: usize = 64;
 
 /// Attempts of a lineage back to attempt 1, each with its largest seq in the lineage.
 #[derive(Debug, Clone)]
@@ -79,50 +83,129 @@ pub(crate) fn genesis_seq(
     Ok(required_attempt_record(transaction, rows, 1)?.start_seq)
 }
 
-/// Returns up to one page of history records with seq >= `from_seq`, ascending.
-pub(crate) fn history_page(
-    transaction: &FlowTransaction<'_>,
+/// Synchronously streams one run's History records and checkpoints, which may live outside
+/// the local store.
+pub trait HistorySource {
+    /// Streams records with seq in `seqs` ascending, ending after an error. `batch_records`
+    /// hints the read size and never changes which records return.
+    fn history_records(
+        &self,
+        seqs: RangeInclusive<u64>,
+        batch_records: usize,
+    ) -> HistoryRecords<'_>;
+
+    /// Returns the checkpoint of `attempt` at `seq`, or `None` when the source holds none.
+    ///
+    /// Returns an error when the checkpoint cannot be read or decoded.
+    fn checkpoint(&self, attempt: u32, seq: u64) -> Result<Option<CheckpointRecord>, ApplyError>;
+}
+
+/// History records of a [`HistorySource`] in ascending seq order.
+pub type HistoryRecords<'source> =
+    Box<dyn Iterator<Item = Result<HistoryRecord, ApplyError>> + 'source>;
+
+/// The [`HistorySource`] over the History rows that a transaction reads.
+pub(crate) struct LocalHistorySource<'transaction, 'reader> {
+    transaction: &'transaction FlowTransaction<'reader>,
     rows: RunRows,
-    from_seq: u64,
-) -> Result<Vec<HistoryRecord>, ApplyError> {
-    let start = history_record_key(rows.flow_hash, rows.run_id, from_seq);
-    let prefix = run_table_prefix(rows.flow_hash, rows.run_id, RunTable::HistoryRecord);
-    let (_, end) = prefix_range(&prefix);
-    Ok(transaction
-        .scan_messages(
+}
+
+impl<'transaction, 'reader> LocalHistorySource<'transaction, 'reader> {
+    pub(crate) fn new(transaction: &'transaction FlowTransaction<'reader>, rows: RunRows) -> Self {
+        Self { transaction, rows }
+    }
+}
+
+impl HistorySource for LocalHistorySource<'_, '_> {
+    fn history_records(
+        &self,
+        seqs: RangeInclusive<u64>,
+        batch_records: usize,
+    ) -> HistoryRecords<'_> {
+        Box::new(LocalHistoryRecords {
+            source: self,
+            next_seq: (!seqs.is_empty()).then_some(*seqs.start()),
+            last_seq: *seqs.end(),
+            batch_records: batch_records.max(1),
+            batch: VecDeque::new(),
+        })
+    }
+
+    fn checkpoint(&self, attempt: u32, seq: u64) -> Result<Option<CheckpointRecord>, ApplyError> {
+        self.transaction.get_message(
+            ColumnFamily::History,
+            &checkpoint_key(self.rows.flow_hash, self.rows.run_id, attempt, seq),
+            "checkpoint",
+        )
+    }
+}
+
+/// Reads local History rows in bounded batches as the caller consumes them.
+struct LocalHistoryRecords<'source, 'transaction, 'reader> {
+    source: &'source LocalHistorySource<'transaction, 'reader>,
+    /// Seq where the next batch starts, or `None` when no rows remain.
+    next_seq: Option<u64>,
+    last_seq: u64,
+    batch_records: usize,
+    batch: VecDeque<HistoryRecord>,
+}
+
+impl Iterator for LocalHistoryRecords<'_, '_, '_> {
+    type Item = Result<HistoryRecord, ApplyError>;
+
+    fn next(&mut self) -> Option<Self::Item> {
+        if self.batch.is_empty() {
+            let next_seq = self.next_seq?;
+            if let Err(error) = self.read_batch(next_seq) {
+                self.next_seq = None;
+                self.batch.clear();
+                return Some(Err(error));
+            }
+        }
+        self.batch.pop_front().map(Ok)
+    }
+}
+
+impl LocalHistoryRecords<'_, '_, '_> {
+    /// Reads the batch starting at `next_seq`, checking each key against its record's seq.
+    fn read_batch(&mut self, next_seq: u64) -> Result<(), ApplyError> {
+        let rows = self.source.rows;
+        let prefix = run_table_prefix(rows.flow_hash, rows.run_id, RunTable::HistoryRecord);
+        let start = history_record_key(rows.flow_hash, rows.run_id, next_seq);
+        let end = match self.last_seq.checked_add(1) {
+            Some(end_seq) => history_record_key(rows.flow_hash, rows.run_id, end_seq),
+            None => prefix_range(&prefix).1,
+        };
+        let page: Vec<(Vec<u8>, HistoryRecord)> = self.source.transaction.scan_messages(
             ColumnFamily::History,
             &start,
             &end,
-            HISTORY_PAGE_RECORDS,
+            self.batch_records,
             "history record",
-        )?
-        .into_iter()
-        .map(|(_, record)| record)
-        .collect())
-}
-
-/// Reads every history record with seq in `[from_seq, to_seq]`, ascending.
-pub(crate) fn history_records(
-    transaction: &FlowTransaction<'_>,
-    rows: RunRows,
-    from_seq: u64,
-    to_seq: u64,
-) -> Result<Vec<HistoryRecord>, ApplyError> {
-    let mut records = Vec::new();
-    let mut next_seq = from_seq;
-    while next_seq <= to_seq {
-        let page = history_page(transaction, rows, next_seq)?;
-        let is_last_page = page.len() < HISTORY_PAGE_RECORDS;
-        for record in page {
-            if record.seq > to_seq {
-                return Ok(records);
+        )?;
+        let is_last_batch = page.len() < self.batch_records;
+        self.next_seq = None;
+        for (key, record) in page {
+            let key_seq = key
+                .strip_prefix(prefix.as_slice())
+                .and_then(|suffix| <[u8; 8]>::try_from(suffix).ok())
+                .map(u64::from_be_bytes)
+                .ok_or_else(|| ApplyError::InvariantViolation {
+                    detail: format!("history record key {key:02x?} has no seq suffix"),
+                })?;
+            if record.seq != key_seq {
+                return Err(ApplyError::CorruptState {
+                    table: "history record",
+                    key: hex(&key),
+                    detail: format!("the record has seq {} under key seq {key_seq}", record.seq),
+                });
             }
-            next_seq = record.seq.saturating_add(1);
-            records.push(record);
+            self.next_seq = key_seq.checked_add(1);
+            self.batch.push_back(record);
         }
-        if is_last_page {
-            break;
+        if is_last_batch {
+            self.next_seq = None;
         }
+        Ok(())
     }
-    Ok(records)
 }