test: update tests for SystemTimeScope and add audit-log and bitemporal cluster coverage

Update all call sites that previously used `system_as_of_ms: None` or
`system_as_of: None` to use `system_time: SystemTimeScope::Current`.

New tests:
- `slice_all_versions_emits_audit_log_ascending` — verifies that three
  system-time versions of the same array cell are emitted in ascending
  system-time order when `AllVersions` is requested.
- `coord_agg_or_reduces_truncated_before_horizon` — coordinator-level test
  verifying the OR-reduction of the below-horizon flag across shards for
  aggregate fan-outs.
- `handle_slice_propagates_truncated_before_horizon` — handler test verifying
  the executor's flag is forwarded into the wire response.
- `versioned_scan_all` unit tests in `btree_versioned/tests.rs`.
- `cluster_array_agg_as_of_returns_versioned_sum_across_shards` — end-to-end
  cluster test exercising the `ARRAY_AGG ... AS OF SYSTEM TIME` path across
  three nodes.

Author: farhan-syah

nodedb-cluster-tests/tests/bitemporal_array_cluster.rs

@@ -80,6 +80,106 @@ fn parse_int_attr(row: &str) -> i64 {
         .unwrap_or_else(|| panic!("cannot extract attrs[0] as i64 from: {row}"))
 }
 
+/// Wall-clock milliseconds since the Unix epoch.
+fn now_ms() -> i64 {
+    std::time::SystemTime::now()
+        .duration_since(std::time::UNIX_EPOCH)
+        .expect("system time before epoch")
+        .as_millis() as i64
+}
+
+/// Extract the scalar aggregate value from an `ARRAY_AGG(..., 'sum')` result.
+///
+/// Robust to the exact projection shape: accepts a bare numeric `result`
+/// column (`"10"`) or a JSON object carrying a `result` key (`{"result":10}`),
+/// and skips any trailing `{"truncated_before_horizon": ...}` summary row that
+/// temporal aggregates append (it has no numeric `result`).
+async fn scalar_agg_sum(client: &tokio_postgres::Client, sql: &str) -> f64 {
+    let rows = query_col0(client, sql).await;
+    rows.iter()
+        .find_map(|r| {
+            r.parse::<f64>().ok().or_else(|| {
+                serde_json::from_str::<serde_json::Value>(r)
+                    .ok()
+                    .and_then(|v| v.get("result").and_then(|x| x.as_f64()))
+            })
+        })
+        .unwrap_or_else(|| panic!("no numeric aggregate result in rows: {rows:?}\n  sql: {sql}"))
+}
+
+/// Distributed bitemporal **aggregate** across shards.
+///
+/// Mirrors the slice test but exercises the `ARRAY_AGG ... AS OF SYSTEM TIME`
+/// path end-to-end: pgwire → coordinator `coord_agg` fan-out → shard
+/// `exec_agg` (which decodes the `(partials, truncated_before_horizon)` tuple)
+/// → partial merge → `finalize_agg_partials`. A wire-shape mismatch anywhere
+/// on that path (the class of bug that broke this silently when only mock
+/// executors were tested) surfaces here as a codec error or a wrong sum.
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn cluster_array_agg_as_of_returns_versioned_sum_across_shards() {
+    let cluster = TestCluster::spawn_three()
+        .await
+        .expect("3-node cluster spawn");
+
+    let leader_idx = cluster
+        .exec_ddl_on_any_leader(
+            "CREATE ARRAY bt3 \
+             DIMS (x INT64 [0..15]) \
+             ATTRS (v INT64) \
+             TILE_EXTENTS (16) \
+             CELL_ORDER ROW_MAJOR",
+        )
+        .await
+        .expect("CREATE ARRAY bt3");
+    let client = &cluster.nodes[leader_idx].client;
+
+    // v1 = 10 at x=0.
+    client
+        .simple_query("INSERT INTO ARRAY bt3 COORDS (0) VALUES (10)")
+        .await
+        .expect("insert v1");
+    client
+        .simple_query("SELECT ARRAY_FLUSH('bt3')")
+        .await
+        .expect("flush v1");
+
+    // Capture a cutoff strictly between v1 and v2.
+    tokio::time::sleep(std::time::Duration::from_millis(20)).await;
+    let cutoff_ms = now_ms();
+    tokio::time::sleep(std::time::Duration::from_millis(20)).await;
+
+    // v2 = 99 at x=0.
+    client
+        .simple_query("INSERT INTO ARRAY bt3 COORDS (0) VALUES (99)")
+        .await
+        .expect("insert v2");
+    client
+        .simple_query("SELECT ARRAY_FLUSH('bt3')")
+        .await
+        .expect("flush v2");
+
+    // Live aggregate sees the latest version: sum = v2 = 99.
+    let live = scalar_agg_sum(client, "SELECT * FROM ARRAY_AGG('bt3', 'v', 'sum')").await;
+    assert!(
+        (live - 99.0).abs() < 1e-4,
+        "live agg sum must be v2=99, got {live}"
+    );
+
+    // AS OF between versions: the distributed aggregate must resolve the cell
+    // ceiling to v1 on every shard and sum to 10.
+    let as_of = scalar_agg_sum(
+        client,
+        &format!("SELECT * FROM ARRAY_AGG('bt3', 'v', 'sum') AS OF SYSTEM TIME {cutoff_ms}"),
+    )
+    .await;
+    assert!(
+        (as_of - 10.0).abs() < 1e-4,
+        "AS OF SYSTEM TIME {cutoff_ms} agg sum must be v1=10, got {as_of}"
+    );
+
+    cluster.shutdown().await;
+}
+
 /// Bring up a 3-node cluster, write v1 then v2 of a cell at coord `x=0`,
 /// capturing the system-time cutoff between the two writes. Assert that
 /// every node returns v1 when queried `AS OF SYSTEM TIME <cutoff>`.

nodedb-cluster-tests/tests/cluster_procedures.rs

@@ -52,7 +52,7 @@ fn procedure_reads_not_replicated() {
         projection: vec![],
         computed_columns: vec![],
         window_functions: vec![],
-        system_as_of_ms: None,
+        system_time: nodedb_types::SystemTimeScope::Current,
         valid_at_ms: None,
         prefilter: None,
     });

nodedb-cluster-tests/tests/cluster_triggers.rs

@@ -204,7 +204,7 @@ fn read_ops_not_replicated() {
         collection: "orders".into(),
         document_id: "doc-1".into(),
         rls_filters: vec![],
-        system_as_of_ms: None,
+        system_time: nodedb_types::SystemTimeScope::Current,
         valid_at_ms: None,
         surrogate: nodedb_types::Surrogate::ZERO,
         pk_bytes: Vec::new(),

nodedb/src/data/executor/dispatch/array/tests_dispatch.rs

@@ -214,7 +214,11 @@ fn json_to_value(v: serde_json::Value) -> Value {
                 .cloned()
                 .map(json_to_value)
                 .collect();
-            Value::ArrayCell(ArrayCell { coords, attrs })
+            Value::ArrayCell(ArrayCell {
+                coords,
+                attrs,
+                system_time: None,
+            })
         }
         serde_json::Value::Number(n) => {
             if let Some(i) = n.as_i64() {
@@ -264,7 +268,7 @@ fn slice_returns_only_cells_in_range() {
         limit: 0,
         cell_filter: None,
         hilbert_range: None,
-        system_as_of: None,
+        system_time: nodedb_types::SystemTimeScope::Current,
         valid_at_ms: None,
     });
     assert_eq!(r.status, Status::Ok, "slice failed: {r:?}");
@@ -283,6 +287,79 @@ fn slice_returns_only_cells_in_range() {
     assert!((sums - 7.0).abs() < 1e-9);
 }
 
+#[test]
+fn slice_all_versions_emits_audit_log_ascending() {
+    let mut h = Harness::new();
+    let s = schema_2d_f64("t6_audit");
+    let aid = ArrayId::new(TenantId::new(1), "t6_audit");
+    h.open(&aid, &s, 0xA7);
+
+    // Three system-time versions of the same cell (0,0), each sealed into
+    // its own segment so they are distinct retained tile-versions.
+    let mk = |v: f64, sys: i64| ArrayPutCell {
+        coord: vec![CoordValue::Int64(0), CoordValue::Int64(0)],
+        attrs: vec![CellValue::Float64(v)],
+        surrogate: Surrogate::ZERO,
+        system_from_ms: sys,
+        valid_from_ms: 0,
+        valid_until_ms: i64::MAX,
+    };
+    h.put(&aid, vec![mk(1.0, 100)], 1);
+    h.flush(&aid);
+    h.put(&aid, vec![mk(2.0, 200)], 2);
+    h.flush(&aid);
+    h.put(&aid, vec![mk(3.0, 300)], 3);
+    h.flush(&aid);
+
+    let slice = ArraySlice::new(vec![None, None]);
+    let slice_bytes = zerompk::to_msgpack_vec(&slice).unwrap();
+    let r = h.send(ArrayOp::Slice {
+        array_id: aid.clone(),
+        slice_msgpack: slice_bytes,
+        attr_projection: vec![],
+        limit: 0,
+        cell_filter: None,
+        hilbert_range: None,
+        system_time: nodedb_types::SystemTimeScope::AllVersions,
+        valid_at_ms: None,
+    });
+    assert_eq!(r.status, Status::Ok, "all-versions slice failed: {r:?}");
+
+    // Read rows as raw JSON: `json_to_value` drops the injected `_ts_system`
+    // column, so inspect the envelope directly.
+    use crate::data::executor::response_codec::ArraySliceResponse;
+    let env: ArraySliceResponse =
+        zerompk::from_msgpack(r.payload.as_ref()).expect("ArraySliceResponse envelope");
+    assert!(
+        !env.truncated_before_horizon,
+        "AllVersions applies no system-time horizon"
+    );
+    let json = nodedb_types::msgpack_to_json_string(&env.rows_msgpack).expect("rows msgpack→json");
+    let rows: Vec<serde_json::Value> = serde_json::from_str(&json).expect("rows json parse");
+
+    let got: Vec<(i64, f64)> = rows
+        .iter()
+        .map(|row| {
+            let ts = row
+                .get("_ts_system")
+                .and_then(|v| v.as_i64())
+                .expect("_ts_system column present on audit-log rows");
+            let v = row
+                .get("attrs")
+                .and_then(|a| a.as_array())
+                .and_then(|a| a.first())
+                .and_then(|v| v.as_f64())
+                .expect("attr value");
+            (ts, v)
+        })
+        .collect();
+    assert_eq!(
+        got,
+        vec![(100, 1.0), (200, 2.0), (300, 3.0)],
+        "audit log must return every version ascending by system time"
+    );
+}
+
 #[test]
 fn aggregate_sum_scalar_across_multiple_tiles() {
     let mut h = Harness::new();
@@ -317,6 +394,127 @@ fn aggregate_sum_scalar_across_multiple_tiles() {
     assert!((f - 10.0).abs() < 1e-9, "sum got {f}");
 }
 
+#[test]
+fn aggregate_return_partial_emits_tuple_wire_shape() {
+    // Distributed shard aggregates set `return_partial = true`. The Data Plane
+    // MUST emit the `(partials, truncated_before_horizon)` tuple — the same
+    // shape the bitemporal path uses — so the cluster `exec_agg` (which always
+    // decodes a tuple) can read non-temporal aggregates too. A bare `Vec` here
+    // (the old `encode_partials` shape) would fail that decode.
+    let mut h = Harness::new();
+    let s = schema_2d_f64("t6_partial");
+    let aid = ArrayId::new(TenantId::new(1), "t6_partial");
+    h.open(&aid, &s, 0xA9);
+    h.put(&aid, vec![cell(0, 0, 1.0), cell(1, 1, 2.0)], 1);
+
+    let r = h.send(ArrayOp::Aggregate {
+        array_id: aid.clone(),
+        attr_idx: 0,
+        reducer: ArrayReducer::Sum,
+        group_by_dim: -1,
+        cell_filter: None,
+        return_partial: true,
+        hilbert_range: None,
+        system_as_of: None,
+        valid_at_ms: None,
+    });
+    assert_eq!(r.status, Status::Ok, "agg failed: {r:?}");
+
+    // Must decode as a 2-tuple (partials, flag), NOT a bare Vec.
+    let (partials, truncated): (
+        Vec<nodedb_cluster::distributed_array::merge::ArrayAggPartial>,
+        bool,
+    ) = zerompk::from_msgpack(r.payload.as_ref())
+        .expect("return_partial payload must be a (Vec<ArrayAggPartial>, bool) tuple");
+    assert_eq!(partials.len(), 1);
+    assert!(
+        (partials[0].sum - 3.0).abs() < 1e-9,
+        "sum got {}",
+        partials[0].sum
+    );
+    assert!(
+        !truncated,
+        "non-temporal current-state read is never below-horizon"
+    );
+}
+
+#[test]
+fn aggregate_temporal_appends_horizon_summary_row() {
+    // Temporal aggregates surface the below-horizon signal as a trailing
+    // {"truncated_before_horizon": bool} summary row — the same shape the
+    // cluster `finalize_agg_partials` produces. Non-temporal aggregates (tested
+    // above) carry no such row, so the two modes stay distinguishable.
+    let mut h = Harness::new();
+    let s = schema_2d_f64("t6_agg_ts");
+    let aid = ArrayId::new(TenantId::new(1), "t6_agg_ts");
+    h.open(&aid, &s, 0xAB);
+    let mk = |x: i64, y: i64, v: f64, sys: i64| ArrayPutCell {
+        coord: vec![CoordValue::Int64(x), CoordValue::Int64(y)],
+        attrs: vec![CellValue::Float64(v)],
+        surrogate: Surrogate::ZERO,
+        system_from_ms: sys,
+        valid_from_ms: 0,
+        valid_until_ms: i64::MAX,
+    };
+    h.put(&aid, vec![mk(0, 0, 10.0, 100), mk(1, 1, 20.0, 100)], 1);
+    h.flush(&aid);
+
+    // AS OF after the writes: both cells visible, not below horizon.
+    let r = h.send(ArrayOp::Aggregate {
+        array_id: aid.clone(),
+        attr_idx: 0,
+        reducer: ArrayReducer::Sum,
+        group_by_dim: -1,
+        cell_filter: None,
+        return_partial: false,
+        hilbert_range: None,
+        system_as_of: Some(150),
+        valid_at_ms: None,
+    });
+    assert_eq!(r.status, Status::Ok, "temporal agg failed: {r:?}");
+    let rows = decode_agg_rows(r.payload.as_ref());
+    assert_eq!(
+        rows.len(),
+        2,
+        "temporal scalar agg = result row + summary row"
+    );
+    let sum = rows[0]
+        .get("result")
+        .and_then(|v| v.as_f64())
+        .expect("result");
+    assert!((sum - 30.0).abs() < 1e-9, "sum got {sum}");
+    assert_eq!(
+        rows[1]
+            .get("truncated_before_horizon")
+            .and_then(|v| v.as_bool()),
+        Some(false),
+        "cutoff after all data is not below-horizon"
+    );
+
+    // AS OF before the writes: below horizon → trailing flag is true.
+    let r2 = h.send(ArrayOp::Aggregate {
+        array_id: aid.clone(),
+        attr_idx: 0,
+        reducer: ArrayReducer::Sum,
+        group_by_dim: -1,
+        cell_filter: None,
+        return_partial: false,
+        hilbert_range: None,
+        system_as_of: Some(50),
+        valid_at_ms: None,
+    });
+    assert_eq!(r2.status, Status::Ok, "below-horizon agg failed: {r2:?}");
+    let rows2 = decode_agg_rows(r2.payload.as_ref());
+    assert_eq!(
+        rows2
+            .last()
+            .and_then(|row| row.get("truncated_before_horizon"))
+            .and_then(|v| v.as_bool()),
+        Some(true),
+        "cutoff before all data must report below-horizon"
+    );
+}
+
 #[test]
 fn aggregate_group_by_dim_buckets_per_x() {
     let mut h = Harness::new();
@@ -428,7 +626,7 @@ fn slice_cell_filter_excludes_non_member_surrogates() {
         limit: 0,
         cell_filter: Some(bm),
         hilbert_range: None,
-        system_as_of: None,
+        system_time: nodedb_types::SystemTimeScope::Current,
         valid_at_ms: None,
     });
     assert_eq!(r.status, Status::Ok, "slice+filter failed: {r:?}");