Upload sealed history segments from the server-zd engine

While a partition offloads history, the lease holder now uploads every
committed SEALED history segment to the primary object store under
history/ and proposes the HistorySegmentUploaded acknowledgments in seq
order.

- One upload task per run (TrackedWork::HistoryUpload). The committed
  HistorySegmentSealed effect starts it, partition rebuild starts it
  from the HistoryUploadPending index after a restart, takeover, or
  snapshot restore, and a periodic rescan restarts stopped work.
- Each segment is read from committed state and encoded on a blocking
  thread, then written with CreateOnly. An existing object that fully
  decodes, on a blocking thread, with the same content checksum is
  acknowledged. Any other existing object is replaced with If-Match,
  logged as an error, and counted.
- Encoded bytes live only together with an upload slot, so no object
  bytes are held while an acknowledgment or a backoff waits, and memory
  stays bounded by history.uploadConcurrency (default 16).
- Every history object call ends when the node stops, so no request
  outlives its owner.
- Retryable object storage failures retry with backoff and never delay
  commands. A failure without a retry, such as a permanent rejection or
  corrupt state, is logged and counted, and every trigger then skips the
  run until the next rescan allows it again. Nothing is acknowledged
  meanwhile. Uploads stop without a retry once the run is gone or
  purging.
- The rescan logs, counts, and skips a run whose rows cannot be read,
  so the other runs of its partition are still rescanned.
- Uploads continue after the offload mode is turned off, so every
  sealed segment is eventually acknowledged.
- Retention postpones a closed run while it has a SEALED segment. The
  purge of a run that kept segment state waits one minute, so the
  client side of every earlier write has ended, then deletes the run's
  objects by catalog key, sweeps the run's history/ prefix, and retries
  a permanent rejection each minute before RunPurge. A write the store
  had already accepted can still leak an object of the purged run; a
  later orphan sweep is to remove it.
- New metrics: upload backlog segments, bytes, and oldest seal age;
  uploads by outcome; uploaded bytes; checksum mismatches; rescan
  errors; and the history object kind of object store requests.
- The format compatibility workload acknowledges the short-lived run's
  sealed segment before its retention fires, and testdata/v1 is
  regenerated with its generator.

Refs #1325

Author: longquanzheng

docs/reference/server/configuration.mdx

@@ -338,6 +338,7 @@ disk or fail Steps.
 | --- | --- | --- |
 | **history.offloadEnabled** | false | Experimental. Whether partitions offload run history to object storage. |
 | **history.segmentTargetBytes** | 4194304 (4 MiB) | Experimental. Since 0.1.21. Bytes of history records, checkpoints, and superseded values after which a run seals its open history segment for upload. From 1 byte to 268435456 (256 MiB). |
+| **history.uploadConcurrency** | 16 | Experimental. Since 0.1.21. Most sealed history segments the lease holder reads, encodes, and uploads at once. From 1 to 1024. Each upload in flight holds about one encoded segment in memory. |
 
 The setting applies to every partition, not to each run. When a node takes the
 lease, it records the value in every partition whose recorded value differs,
@@ -358,6 +359,27 @@ 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.
 
+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
+run's segments oldest first and records each upload before the next one. A
+segment's content depends only on committed state. A new lease holder therefore
+writes the same object again and accepts an existing object with the same
+content. An existing object with different content is replaced; the node logs an
+error and increments **dexzd_history_upload_checksum_mismatches_total**.
+
+Uploads never delay commands. While object storage fails them, they retry
+with backoff, sealed segments stay on local disk, and the upload backlog metrics
+grow; see [Metrics](/reference/server/metrics). A permanent rejection, such as
+denied access, is retried once a minute instead. Uploads of segments sealed
+earlier continue after **history.offloadEnabled** is turned off. Retention
+postpones the deletion of a closed run until its sealed segments are uploaded.
+Deleting a run also deletes every object under its **history/** prefix. That
+deletion starts a minute after the run's deletion begins, so that uploads an
+earlier lease holder abandoned have ended on its side. An upload that object
+storage accepted just before can still leave an object of the deleted run; no
+read returns it. Do not configure a bucket lifecycle rule that expires objects
+under **history/**.
+
 ## Limits {#limits}
 
 server-zd enforces these limits. A request that a limit rejects fails at once;

docs/reference/server/metrics.mdx

@@ -35,6 +35,7 @@ No series has a Flow, Flow type, partition, tenant, or key label.
 | Timers | **dexzd_timer_lag_seconds** |
 | Snapshots | **dexzd_snapshots_total** with **outcome**; **dexzd_snapshot_last_duration_seconds** |
 | Local disk | **dexzd_local_storage_bytes** |
+| History offload (experimental) | **dexzd_history_upload_backlog_segments**; **dexzd_history_upload_backlog_bytes**; **dexzd_history_upload_oldest_sealed_age_seconds**; **dexzd_history_uploads_total** with **outcome**; **dexzd_history_upload_bytes_total**; **dexzd_history_upload_checksum_mismatches_total**; **dexzd_history_upload_rescan_errors_total** |
 | Attribute Store sync | See [Monitoring](/reference/server/attribute-store-sync#attribute-store-monitoring) |
 | External search projection | See [Monitoring](/reference/server/search#search-projection-monitoring) |
 
@@ -78,6 +79,18 @@ Worker polls use one persistent stream, so there is no empty-poll count.
 **event="delivered"**, every task frame sent, including RPC task frames. A returned attempt
 that Dex hands out again counts again. Long-poll waits count toward **flow_read** latency.
 
+The history offload gauges come from a rescan of every partition each minute.
+**dexzd_history_upload_backlog_segments** counts sealed history segments that
+are not uploaded yet. **dexzd_history_upload_backlog_bytes** is their
+uncompressed size. **outcome** of **dexzd_history_uploads_total** is one of
+**created**, **already_present** (a repeated upload found the same content),
+**replaced** (an existing object had other content), each counted once the
+upload is recorded, and **failed**. A transient object storage failure is
+retried with backoff; any other failure is retried at the next rescan. Alert when
+**dexzd_history_upload_oldest_sealed_age_seconds** stays above 900, or when
+**dexzd_history_upload_checksum_mismatches_total** or
+**dexzd_history_upload_rescan_errors_total** rises.
+
 ~~~text
 # Failed API calls by class and code
 sum by (rpc_class, code) (rate(dexzd_api_requests_total{code!="ok"}[5m]))

server-zd/Cargo.lock

@@ -151,7 +151,7 @@ Every label value is from a closed set. No series has a Flow, Flow type, partiti
 | `dexzd_commit_records_total` | counter | — | Committed commands |
 | `dexzd_commit_bytes_total` | counter | — | Committed batch bytes |
 | `dexzd_commit_latency_seconds` | histogram, buckets 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2.5 | — | Batch seal to durable acknowledgement |
-| `dexzd_object_store_requests_total` | counter | `operation`: put, get, list, delete; `object_kind`: log, snapshot, manifest, lease, other | Object storage requests; multipart upload counts once as put |
+| `dexzd_object_store_requests_total` | counter | `operation`: put, get, list, delete; `object_kind`: log, snapshot, manifest, lease, history, other | Object storage requests; multipart upload counts once as put |
 | `dexzd_object_store_request_errors_total` | counter | `operation`: put, get, list, delete | Failed requests after SDK retries |
 | `dexzd_object_store_put_bytes_total` | counter | — | Uploaded bytes |
 | `dexzd_snapshots_total` | counter | `outcome`: ok, error | Snapshot attempts |
@@ -189,6 +189,13 @@ Every label value is from a closed set. No series has a Flow, Flow type, partiti
 | `dexzd_search_initial_loads_total` | counter | — | Partition loads started, including loads of progress that cannot continue; the log gives the reason |
 | `dexzd_search_projection_mode_skipped_partitions_total` | counter | — | Stopped partitions that kept their previous search projection mode at startup |
 | `dexzd_history_offload_mode_skipped_partitions_total` | counter | — | Stopped partitions that kept their previous history offload mode at startup |
+| `dexzd_history_upload_backlog_segments` | gauge | — | Sealed history segments not yet uploaded, summed over each partition's last rescan |
+| `dexzd_history_upload_backlog_bytes` | gauge | — | Uncompressed bytes of the history rows and superseded values of `dexzd_history_upload_backlog_segments` |
+| `dexzd_history_upload_oldest_sealed_age_seconds` | gauge | — | Age at scrape time of the oldest sealed history segment not yet uploaded, 0 when none |
+| `dexzd_history_uploads_total` | counter | `outcome`: created, already_present, replaced, failed | History segment uploads, counted once acknowledged; failed attempts retry with backoff when transient, otherwise at the next rescan |
+| `dexzd_history_upload_bytes_total` | counter | — | Bytes of the history segment objects this node created or replaced |
+| `dexzd_history_upload_checksum_mismatches_total` | counter | — | Existing history segment objects whose content differed from committed state and were replaced |
+| `dexzd_history_upload_rescan_errors_total` | counter | — | Runs or index rows that the history upload rescan could not read; the log names each and the rescan continues |
 | `dexzd_search_initial_load_documents_total` | counter | — | Documents written by initial SQL visibility loads |
 
 Worker polling uses a persistent stream, so there is no empty poll outcome. The stream event metric counts each task delivery and each stream close.

server-zd/README.md

@@ -84,6 +84,7 @@ history:
   # Experimental: keep false until the release notes announce history offload.
   offloadEnabled: false
   segmentTargetBytes: 4194304
+  uploadConcurrency: 16
 values:
   wireInlineThresholdBytes: 100
   maxValueBytes: 16777216

server-zd/config/dex-server-zd.example.yaml

@@ -70,7 +70,9 @@ at or below U while the partition offloads history and, whatever the mode, while
 the run has a SEALED segment. So turning offload off never deletes sealed content
 before its acknowledgment; once every SEALED segment is acknowledged, pruning
 returns to the window rule. Segment rows count as history bytes, never toward the
-run state limit. Run purge deletes the pending index row with the run.
+run state limit. Run purge deletes the pending index row with the run. Retention postpones a closed run while it has a SEALED
+segment, as it does for an unacknowledged outbox message, so purging never
+abandons an upload.
 
 ## Value and storage accounting
 

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

@@ -74,7 +74,10 @@ pub(crate) fn on_retention_fired(
     cursor: &mut RunCursor,
     scope: &CommandScope<'_>,
 ) -> Result<(), ApplyError> {
-    if has_unacknowledged_outbox(transaction, cursor.rows)? {
+    // Purging stops uploads, so a sealed segment would never reach object storage.
+    if has_unacknowledged_outbox(transaction, cursor.rows)?
+        || has_sealed_history_segment(transaction, cursor.rows)?
+    {
         postpone_retention(transaction, cursor, scope.now_micros);
         return Ok(());
     }
@@ -261,6 +264,22 @@ fn has_unacknowledged_outbox(
         .any(|(_, message)| message.acknowledged_at_micros == 0))
 }
 
+/// Returns whether the run has a SEALED history segment; its upload pending index row exists
+/// exactly while one remains.
+fn has_sealed_history_segment(
+    transaction: &FlowTransaction<'_>,
+    rows: RunRows,
+) -> Result<bool, ApplyError> {
+    let key = history_upload_pending_index_key(
+        transaction.context().partition,
+        rows.flow_hash,
+        rows.run_id,
+    );
+    Ok(transaction
+        .get(ColumnFamily::PartitionIndex, &key)?
+        .is_some())
+}
+
 fn remove_timer_indexes(
     transaction: &mut FlowTransaction<'_>,
     rows: RunRows,

server-zd/crates/dexzd-apply/src/cross_flow/run_purge.rs

@@ -1251,37 +1251,82 @@ impl Harness {
 }
 
 #[test]
-fn purging_a_run_deletes_its_upload_pending_index_row() {
+fn retention_waits_for_every_sealed_segment_then_the_purge_deletes_every_row() {
     let mut harness = harness_with_three_sealed_segments();
     harness.stop();
     assert!(harness.has_upload_pending_index_row());
-    let record = harness.run_record();
-    let retention = record.retention_deadline_micros;
+    let retention = harness.run_record().retention_deadline_micros;
     assert!(retention > 0);
     harness.now_micros = retention;
-    let fired = harness.apply(Command::TimersFired(
-        dexzd_protocol::dexzd::v1::TimersFiredCommand {
-            timers: vec![dexzd_protocol::dexzd::v1::TimerId {
-                run_attempt: 0,
-                kind: Some(dexzd_protocol::dexzd::v1::timer_id::Kind::Retention(
-                    dexzd_protocol::dexzd::v1::RetentionTimerId {},
-                )),
-            }],
-        },
+    let postponed = harness.apply(retention_fired());
+    assert!(!requests_purge(&postponed), "{:?}", postponed.effects);
+    let record = harness.run_record();
+    assert!(!record.is_purging, "purging would stop the uploads");
+    assert!(record.retention_deadline_micros > retention);
+
+    for segment in harness.segments() {
+        if segment.state == HistorySegmentState::Sealed as i32 {
+            let acknowledged = harness.ack(segment.first_seq, segment.last_seq);
+            assert!(is_ok(&acknowledged.reply), "{:?}", acknowledged.reply);
+        }
+    }
+    assert!(!harness.has_upload_pending_index_row());
+    harness.now_micros = record.retention_deadline_micros;
+    let fired = harness.apply(retention_fired());
+    assert!(requests_purge(&fired), "{:?}", fired.effects);
+    let purge = harness.apply(Command::RunPurge(
+        dexzd_protocol::dexzd::v1::RunPurgeCommand {},
     ));
-    assert!(
-        fired
-            .effects
-            .iter()
-            .any(|effect| matches!(effect, ApplyEffect::RunPurgeRequested { .. })),
-        "{:?}",
-        fired.effects
+    assert!(is_ok(&purge.reply), "{:?}", purge.reply);
+    assert_every_run_row_purged(&harness);
+}
+
+#[test]
+fn purging_a_run_deletes_its_upload_pending_index_row() {
+    let mut harness = harness_with_three_sealed_segments();
+    harness.stop();
+    assert!(harness.has_upload_pending_index_row());
+    // Retention never purges a run with a SEALED segment, so the test marks the purge itself.
+    let key = run_singleton_key(
+        harness.context.flow_hash,
+        harness.run(),
+        RunTable::RunRecord,
     );
+    let purging = RunRecord {
+        is_purging: true,
+        ..harness.run_record()
+    };
+    harness
+        .state
+        .rows
+        .insert((ColumnFamily::Flow, key), purging.encode_to_vec());
     let purge = harness.apply(Command::RunPurge(
         dexzd_protocol::dexzd::v1::RunPurgeCommand {},
     ));
     assert!(is_ok(&purge.reply), "{:?}", purge.reply);
     assert!(!harness.has_upload_pending_index_row());
+    assert_every_run_row_purged(&harness);
+}
+
+fn retention_fired() -> Command {
+    Command::TimersFired(dexzd_protocol::dexzd::v1::TimersFiredCommand {
+        timers: vec![dexzd_protocol::dexzd::v1::TimerId {
+            run_attempt: 0,
+            kind: Some(dexzd_protocol::dexzd::v1::timer_id::Kind::Retention(
+                dexzd_protocol::dexzd::v1::RetentionTimerId {},
+            )),
+        }],
+    })
+}
+
+fn requests_purge(outcome: &ApplyOutcome) -> bool {
+    outcome
+        .effects
+        .iter()
+        .any(|effect| matches!(effect, ApplyEffect::RunPurgeRequested { .. }))
+}
+
+fn assert_every_run_row_purged(harness: &Harness) {
     assert!(
         harness
             .state

server-zd/crates/dexzd-apply/tests/history_segments.rs

@@ -11,8 +11,11 @@ publish.workspace = true
 [dependencies]
 
 [dev-dependencies]
+dexzd-history-segment.workspace = true
+dexzd-keyspace.workspace = true
 dexzd-protocol.workspace = true
 dexzd-server = { workspace = true, features = ["test-util"] }
+dexzd-storage.workspace = true
 futures.workspace = true
 prost.workspace = true
 prost-types.workspace = true

server-zd/crates/dexzd-e2e-tests/Cargo.toml

@@ -104,6 +104,22 @@ impl DevServer {
         WorkerTaskServiceClient::new(self.channel.clone())
     }
 
+    /// Stops the server gracefully and returns its data directory, so a test can open the
+    /// stopped server's local store.
+    pub(crate) async fn shutdown_keeping_data(self) -> TempDir {
+        let Self {
+            server,
+            _data_directory: data_directory,
+            ..
+        } = self;
+        server
+            .expect("the server is shut down once")
+            .shutdown()
+            .await
+            .expect("the server stops cleanly");
+        data_directory
+    }
+
     /// Stops the server gracefully and returns how long the shutdown took.
     pub(crate) async fn shutdown(mut self) -> Duration {
         let server = self.server.take().expect("the server is shut down once");