server-zd: migrate the history of existing runs to segments

docs/reference/server/configuration.mdx

@@ -354,10 +354,22 @@ pruning then deletes local history only after its segment is uploaded. Each run
 records **history.segmentTargetBytes** when it starts, so a new value applies only
 to runs that start after the node restarts with it.
 
+Runs that already exist when offload turns on migrate, including runs that
+started while it was off. The first command that leaves such a run running seals
+the oldest part of its local history. Each later command of the run seals one
+more segment, until a segment reaches the command itself. The run then continues
+like a run that started with offload on. Events that the history window removed
+before the migration stay removed. A run that is already closed is not migrated;
+deleting it after its retention deletes its local history. A run that closes
+during its migration keeps the rest of its history on local disk until it is
+deleted.
+
 Turning offload off stops sealing. History of segments sealed earlier stays on
 local disk until each segment is uploaded; then pruning returns to the window
 rule. Commands applied while offload is off belong to no segment, so turning it
-on again starts a new segment after them.
+on again starts a new segment after them. Turning offload off also ends every
+migration; the rest of each such run's history follows the window rule like
+those commands.
 
 The lease holder uploads each sealed segment to the primary object store, under
 the **history/** prefix of the bucket that holds the commit log. It uploads a

server-zd/crates/dexzd-apply/README.md

@@ -52,7 +52,8 @@ superseded inside it, or when the command closed the run. Sealing writes a SEALE
 with a superseded seq inside the range, and the run's `HistoryUploadPending`
 index row, and emits `HistorySegmentSealed`. A value revived later keeps its row,
 so the segment's content never changes. Runs that started while offload was off
-keep no segment state. While the partition does not offload, apply seals nothing,
+keep no segment state until they migrate; see History migration. While the
+partition does not offload, apply seals nothing,
 and each command of a run with segment state moves the open segment past itself.
 The seqs and events of those commands belong to no segment, and their local rows
 follow window pruning, so a segment opened after offload is turned on again never
@@ -74,6 +75,35 @@ run state limit. Run purge deletes the pending index row with the run. Retention
 segment, as it does for an unacknowledged outbox message, so purging never
 abandons an upload.
 
+## History migration
+
+A run that started while its partition did not offload history has no segment
+state. The first command that leaves such a run running while its partition
+offloads starts the migration: it sets `history_migration_started_seq` to its
+seq and opens the segment at the run's first seq and event 1. While the run
+migrates, the open segment holds every unsealed seq up to the current command
+and tracks no bytes, and each command of the run seals one segment of it.
+
+A migration segment ends at a whole seq: the first one where it holds at least
+two history records and reaches the target bytes, 10,000 superseded values, or
+10,000 history rows. Otherwise it ends at the command's own seq, which ends the
+migration. Cutting it reads one bounded page of history metadata and of table G.
+Because segments end only at whole seqs, a migration step is bounded by the
+largest single command in the migrated range, as sealing is: one seq whose
+command superseded 30,000 values seals all of them in one step.
+Two records per segment let the migration gain on the run, since each command
+adds one record. The segment holds the rows still stored in its seqs, so seqs
+that window pruning deleted earlier are holes inside the reported pruned event
+range. Its event range comes from the metadata `last_assigned_event_id`; records
+written before that field are decoded instead. Pruning keeps every row above U
+while the partition offloads, so nothing is deleted before its segment uploads.
+
+A closed run never starts migrating, so retention purges its local history
+directly. A run that closes during its migration seals one more segment at its
+closing command and then keeps the rest local until purge; a time travel that
+reopens it continues the migration. Turning offload off ends a migration, and the
+unsealed rest joins the gap of commands applied while offload is off.
+
 ## Value and storage accounting
 
 Each run tracks encoded Flow and History row bytes, including keys. Live and superseded value bytes use the encoded value sizes in birth records; garbage rows identify the superseded portion. Apply updates these totals from each command's writes and deletes.

server-zd/crates/dexzd-apply/src/command_handlers/run_cursor.rs

@@ -17,7 +17,7 @@ use crate::command_handlers::{
     CommandScope, HandlerOutput, ReplayedCommandCounters, ReplayedCommandOutput,
 };
 use crate::history;
-use crate::history_offload::segment_seal::seal_history_segment_if_due;
+use crate::history_offload::segment_migration::seal_or_migrate_history_segment;
 use crate::projections::history_event_links::{
     HistoryEventIndex, PendingHistoryEvents, StepEventLink, resolve_step_event_links,
     with_recorded_event_ids,
@@ -283,7 +283,7 @@ impl RunCursor {
         );
         search_document::refresh_search_document(transaction, &self, scope)?;
         history::after_command(transaction, &mut self, window_events)?;
-        seal_history_segment_if_due(transaction, &mut self, scope.now_micros, has_closed_run)?;
+        seal_or_migrate_history_segment(transaction, &mut self, scope.now_micros, has_closed_run)?;
         self.rows.put_run_record(transaction, &self.record);
         if scope.is_replay {
             reconcile_run_storage(transaction, &mut self)?;

server-zd/crates/dexzd-apply/src/history_offload/mod.rs

@@ -9,5 +9,6 @@
 //! Offloading run history to object storage, gated by a committed partition mode.
 
 pub(crate) mod offload_mode;
+pub(crate) mod segment_migration;
 pub(crate) mod segment_seal;
 pub(crate) mod segment_upload;

server-zd/crates/dexzd-apply/src/history_offload/segment_migration.rs

@@ -0,0 +1,299 @@
+// Copyright (c) 2026 Super Durable, Inc.
+//
+// Licensed under the Sustainable Use License 1.0.
+// You may not use this file except in compliance with the License.
+// See the LICENSE file in the repository root.
+//
+// SPDX-License-Identifier: LicenseRef-Sustainable-Use-1.0
+
+//! Migrating history a run recorded before it had segment state (plan section 5.17).
+
+use std::collections::BTreeMap;
+
+use dexzd_keyspace::row_keys::history_record_key;
+use dexzd_keyspace::{ColumnFamily, RunTable, run_table_prefix};
+use dexzd_protocol::dex::FlowStatus;
+use dexzd_protocol::dexzd::v1::{HistoryRecord, HistoryRowMetadata, RunRecord, ValueGarbageRecord};
+
+use crate::ApplyError;
+use crate::command_handlers::run_cursor::RunCursor;
+use crate::history::genesis_seq;
+use crate::history_offload::offload_mode::is_history_offload_enabled;
+use crate::history_offload::segment_seal::{
+    MAX_SEGMENT_SUPERSEDED_VALUES, SealedSegmentBounds, garbage_identity,
+    open_segment_after_command, seal_history_segment_if_due, segment_target_bytes,
+    write_sealed_segment,
+};
+use crate::state_access::FlowTransaction;
+
+/// Bounds the history rows of one migration segment, so cutting it reads a bounded page.
+const MAX_MIGRATION_SEGMENT_HISTORY_ROWS: usize = 10_000;
+/// FlowStart records event 1 of every run.
+const FIRST_EVENT_ID: i64 = 1;
+
+/// Seals a due open segment, or starts or continues migrating older history, sealing at most one.
+///
+/// `has_closed_run` tells whether this command closed the run; `now_micros` is its proposal time.
+pub(crate) fn seal_or_migrate_history_segment(
+    transaction: &mut FlowTransaction<'_>,
+    cursor: &mut RunCursor,
+    now_micros: i64,
+    has_closed_run: bool,
+) -> Result<(), ApplyError> {
+    let record = &cursor.record;
+    if record.history_open_segment_first_seq == 0 {
+        // Closed runs never start: migrating them only uploads history that purge deletes next.
+        if is_running(record) && !record.is_purging && is_history_offload_enabled(transaction)? {
+            start_migration(transaction, cursor)?;
+            migrate_one_segment(transaction, cursor, now_micros)?;
+        }
+        return Ok(());
+    }
+    if record.history_migration_started_seq == 0 {
+        return seal_history_segment_if_due(transaction, cursor, now_micros, has_closed_run);
+    }
+    if !is_history_offload_enabled(transaction)? {
+        // The unsealed rest joins the gap of commands applied while offload is off.
+        cursor.record.history_migration_started_seq = 0;
+        return seal_history_segment_if_due(transaction, cursor, now_micros, has_closed_run);
+    }
+    if record.is_purging || !(has_closed_run || is_running(record)) {
+        return Ok(());
+    }
+    migrate_one_segment(transaction, cursor, now_micros)
+}
+
+/// Opens the segment state of a run without one at its first seq and event 1.
+fn start_migration(
+    transaction: &FlowTransaction<'_>,
+    cursor: &mut RunCursor,
+) -> Result<(), ApplyError> {
+    let first_seq = genesis_seq(transaction, cursor.rows)?;
+    let record = &mut cursor.record;
+    record.history_migration_started_seq = cursor.seq;
+    record.history_open_segment_first_seq = first_seq;
+    record.history_open_segment_first_event_id = FIRST_EVENT_ID;
+    record.history_open_segment_bytes = 0;
+    record.history_open_segment_superseded_value_count = 0;
+    Ok(())
+}
+
+/// Seals the next migration segment; the migration ends once a segment reaches this command.
+fn migrate_one_segment(
+    transaction: &mut FlowTransaction<'_>,
+    cursor: &mut RunCursor,
+    now_micros: i64,
+) -> Result<(), ApplyError> {
+    let bounds = cut_migration_segment(transaction, cursor)?;
+    write_sealed_segment(transaction, cursor, &bounds, now_micros)?;
+    if bounds.last_seq == cursor.seq {
+        cursor.record.history_migration_started_seq = 0;
+        open_segment_after_command(cursor);
+    } else {
+        cursor.record.history_open_segment_first_seq = bounds.last_seq + 1;
+        cursor.record.history_open_segment_first_event_id = bounds.last_event_id + 1;
+    }
+    Ok(())
+}
+
+/// Stored history of one seq inside the migration range.
+#[derive(Default)]
+struct SeqHistory {
+    bytes: u64,
+    history_rows: usize,
+    superseded_values: u64,
+    record: Option<HistoryRowMetadata>,
+}
+
+/// What a migration segment holds so far.
+#[derive(Default)]
+struct SegmentTotals {
+    bytes: u64,
+    history_rows: usize,
+    superseded_values: u64,
+    records: usize,
+}
+
+impl SegmentTotals {
+    fn add(&mut self, seq_history: &SeqHistory) {
+        self.bytes = self.bytes.saturating_add(seq_history.bytes);
+        self.history_rows += seq_history.history_rows;
+        self.superseded_values += seq_history.superseded_values;
+        self.records += usize::from(seq_history.record.is_some());
+    }
+
+    /// Two records at least, so each command seals more history than it adds.
+    fn is_complete(&self, target_bytes: u64) -> bool {
+        self.records >= 2
+            && (self.bytes >= target_bytes
+                || self.superseded_values >= MAX_SEGMENT_SUPERSEDED_VALUES
+                || self.history_rows >= MAX_MIGRATION_SEGMENT_HISTORY_ROWS)
+    }
+}
+
+/// Returns the bounds of the next migration segment, which starts at the open segment and ends
+/// at a whole seq: the first one where the segment holds two history records and reaches the
+/// target bytes, the superseded value limit, or the history row limit; otherwise this command's
+/// seq. When a bounded page ends first, the segment ends with that page or its second record.
+fn cut_migration_segment(
+    transaction: &FlowTransaction<'_>,
+    cursor: &RunCursor,
+) -> Result<SealedSegmentBounds, ApplyError> {
+    let first_seq = cursor.record.history_open_segment_first_seq;
+    let first_event_id = cursor.record.history_open_segment_first_event_id;
+    let current_seq = cursor.seq;
+    let mut seqs = BTreeMap::<u64, SeqHistory>::new();
+    let metadata_complete_through = scan_history_metadata(transaction, cursor, &mut seqs)?;
+    let garbage_complete_through = scan_superseded_values(transaction, cursor, &mut seqs)?;
+    let complete_through = metadata_complete_through.min(garbage_complete_through);
+    // The command's own record is stored, so the range holds at least one record.
+    let second_record_seq = seqs
+        .iter()
+        .filter(|(_, seq_history)| seq_history.record.is_some())
+        .map(|(seq, _)| *seq)
+        .nth(1)
+        .unwrap_or(current_seq);
+    let target_bytes = segment_target_bytes(&cursor.record);
+    let mut totals = SegmentTotals::default();
+    let cut_seq = seqs
+        .iter()
+        .take_while(|(seq, _)| **seq <= complete_through)
+        .find_map(|(seq, seq_history)| {
+            totals.add(seq_history);
+            totals.is_complete(target_bytes).then_some(*seq)
+        });
+    // Whole seqs can pass the limits; the command that recorded such a seq wrote as many rows.
+    let last_seq = cut_seq.unwrap_or_else(|| complete_through.max(second_record_seq));
+    let last_event_id = if last_seq == current_seq {
+        cursor.record.next_event_id.max(1) - 1
+    } else {
+        newest_event_id_through(transaction, cursor, &seqs, last_seq, first_event_id - 1)?
+    };
+    Ok(SealedSegmentBounds {
+        first_seq,
+        last_seq,
+        first_event_id,
+        last_event_id,
+    })
+}
+
+/// Adds the history records and checkpoints of the migration range to `seqs`, reading at most
+/// one row more than the row limit. Returns the last seq whose rows were all read.
+fn scan_history_metadata(
+    transaction: &FlowTransaction<'_>,
+    cursor: &RunCursor,
+    seqs: &mut BTreeMap<u64, SeqHistory>,
+) -> Result<u64, ApplyError> {
+    let rows = cursor.rows;
+    let prefix = run_table_prefix(rows.flow_hash, rows.run_id, RunTable::HistoryRowMetadata);
+    let limit = MAX_MIGRATION_SEGMENT_HISTORY_ROWS + 1;
+    let page = transaction.scan_messages::<HistoryRowMetadata>(
+        ColumnFamily::Flow,
+        &seq_bound(&prefix, cursor.record.history_open_segment_first_seq),
+        &seq_bound(&prefix, cursor.seq + 1),
+        limit,
+        "history metadata",
+    )?;
+    let is_truncated = page.len() == limit;
+    let mut last_seq = cursor.seq;
+    for (key, metadata) in page {
+        let suffix = key
+            .strip_prefix(prefix.as_slice())
+            .filter(|suffix| suffix.len() == 13)
+            .ok_or_else(|| ApplyError::InvariantViolation {
+                detail: "history metadata key must have a 13-byte suffix".to_owned(),
+            })?;
+        let seq = u64::from_be_bytes(suffix[..8].try_into().expect("eight seq bytes"));
+        last_seq = seq;
+        let seq_history = seqs.entry(seq).or_default();
+        seq_history.bytes = seq_history.bytes.saturating_add(metadata.row_size_bytes);
+        seq_history.history_rows += 1;
+        if suffix[8] == RunTable::HistoryRecord as u8 {
+            seq_history.record = Some(metadata);
+        }
+    }
+    Ok(if is_truncated {
+        last_seq - 1
+    } else {
+        cursor.seq
+    })
+}
+
+/// Adds the values superseded inside the migration range to `seqs`, reading at most one more
+/// than the superseded value limit. Returns the last seq whose values were all read.
+fn scan_superseded_values(
+    transaction: &FlowTransaction<'_>,
+    cursor: &RunCursor,
+    seqs: &mut BTreeMap<u64, SeqHistory>,
+) -> Result<u64, ApplyError> {
+    let rows = cursor.rows;
+    let prefix = run_table_prefix(rows.flow_hash, rows.run_id, RunTable::ValueGarbage);
+    let limit =
+        usize::try_from(MAX_SEGMENT_SUPERSEDED_VALUES).expect("the value limit fits in usize") + 1;
+    let page = transaction.scan_messages::<ValueGarbageRecord>(
+        ColumnFamily::Flow,
+        &seq_bound(&prefix, cursor.record.history_open_segment_first_seq),
+        &seq_bound(&prefix, cursor.seq + 1),
+        limit,
+        "value garbage",
+    )?;
+    let is_truncated = page.len() == limit;
+    let mut last_seq = cursor.seq;
+    for (key, garbage) in page {
+        let (seq, _) = garbage_identity(&key, &prefix)?;
+        last_seq = seq;
+        let seq_history = seqs.entry(seq).or_default();
+        seq_history.bytes = seq_history.bytes.saturating_add(garbage.size_bytes);
+        seq_history.superseded_values += 1;
+    }
+    Ok(if is_truncated {
+        last_seq - 1
+    } else {
+        cursor.seq
+    })
+}
+
+/// Returns the newest event ID that the records in `seqs` through `last_seq` assigned, or
+/// `before_range` when none did.
+///
+/// Records older than `last_assigned_event_id` are decoded; events of pruned records then widen
+/// the next range.
+fn newest_event_id_through(
+    transaction: &FlowTransaction<'_>,
+    cursor: &RunCursor,
+    seqs: &BTreeMap<u64, SeqHistory>,
+    last_seq: u64,
+    before_range: i64,
+) -> Result<i64, ApplyError> {
+    let mut newest = before_range;
+    for (seq, seq_history) in seqs.range(..=last_seq) {
+        let Some(metadata) = &seq_history.record else {
+            continue;
+        };
+        if metadata.last_assigned_event_id > 0 {
+            newest = newest.max(metadata.last_assigned_event_id);
+            continue;
+        }
+        let key = history_record_key(cursor.rows.flow_hash, cursor.rows.run_id, *seq);
+        let record = transaction
+            .get_message::<HistoryRecord>(ColumnFamily::History, &key, "history record")?
+            .ok_or_else(|| ApplyError::InvariantViolation {
+                detail: format!("history record {seq} has metadata but no row"),
+            })?;
+        if let Some(event_id) = record.events.iter().map(|event| event.event_id).max() {
+            newest = newest.max(event_id);
+        }
+    }
+    Ok(newest)
+}
+
+/// Returns the first key of `seq` in a run table whose keys start with an eight-byte seq.
+fn seq_bound(prefix: &[u8], seq: u64) -> Vec<u8> {
+    let mut key = prefix.to_vec();
+    key.extend_from_slice(&seq.to_be_bytes());
+    key
+}
+
+fn is_running(record: &RunRecord) -> bool {
+    record.status == FlowStatus::Running as i32
+}