fix(asap-tools): send paced feeder ticks concurrently, ordered per series, in (t, t+1s]

asap-tools/data-sources/paced-feeder/README.md

@@ -53,12 +53,18 @@ credentials.
 
 ## Pacing and stats
 
-Each tick's rows are split into `--batch-rows` requests and sent with up to
-`--concurrency` in flight. For remote write, a tick sends its
-`--samples-per-sec` sub-second slots in turn, and the next slot starts only
-after every request of the current one returns, so each aggregation group's
-samples reach ASAPQuery in timestamp order. ClickHouse sorts on insert, so it
-gets the whole tick at once.
+Each tick's rows are split into requests of about `--batch-rows` rows and
+sent with up to `--concurrency` in flight. A tick's rows are series-major and
+requests are cut only between series, so each series' samples of a tick go
+in one request, in timestamp order; the next tick starts only after every
+request of the current one returns. Each series' samples therefore arrive in
+order (counters never seem to reset, and Prometheus needs no out-of-order
+window), while samples of different series of a group may arrive out of
+order within a tick. A tick starting at `t` holds samples spaced evenly over
+`(t, t + 1 s]`, the same left-open interval as ASAPQuery's one-second
+windows, so a tick never spans two windows, and that reordering is safe for
+ASAPQuery as long as its allowed lateness is at least 1 s (the default is
+5 s).
 `--samples-per-sec` is at most 1000, so timestamps (in ms) never repeat. The
 default batch size is 50,000 rows for remote write, which keeps a compressed
 request under axum's 2 MB body limit in ASAPQuery's ingest, and 100,000 for

asap-tools/data-sources/paced-feeder/src/encode.rs

@@ -157,7 +157,7 @@ mod tests {
     #[test]
     fn remote_write_decodes_to_one_series_per_row() {
         let w = workload();
-        let rows = w.slot_rows(2, 0, 1_700_000_000_000);
+        let rows = w.tick_rows(2, 1_700_000_000_000);
         let body = RemoteWriteEncoder::new(&w, "data").encode(&rows);
 
         let raw = snap::raw::Decoder::new().decompress_vec(&body).unwrap();
@@ -191,7 +191,7 @@ mod tests {
     #[test]
     fn row_binary_round_trips_every_column() {
         let w = workload();
-        let rows = w.slot_rows(0, 0, 42_000);
+        let rows = w.tick_rows(0, 42_000);
         let buf = RowBinaryEncoder::new(&w).encode(&rows);
 
         let mut cursor = buf.as_slice();

asap-tools/data-sources/paced-feeder/src/main.rs

@@ -16,7 +16,7 @@ use serde::Serialize;
 use std::sync::Arc;
 use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
 use tokio::sync::Semaphore;
-use workload::{Row, Workload};
+use workload::{series_batches, Workload};
 
 #[derive(Debug, Clone, Copy, ValueEnum, Serialize)]
 #[serde(rename_all = "kebab-case")]
@@ -71,9 +71,10 @@ struct Args {
     #[arg(long)]
     start_ms: Option<i64>,
 
-    /// Rows per HTTP request. Defaults to 50,000 for remote write, which keeps
-    /// a compressed request under axum's 2 MB body limit in ASAPQuery's ingest,
-    /// and 100,000 for ClickHouse, where each INSERT creates a part.
+    /// Rows per HTTP request, rounded down to whole series (at least one).
+    /// Defaults to 50,000 for remote write, which keeps a compressed request
+    /// under axum's 2 MB body limit in ASAPQuery's ingest, and 100,000 for
+    /// ClickHouse, where each INSERT creates a part.
     #[arg(long)]
     batch_rows: Option<usize>,
     /// Concurrent requests within one tick.
@@ -179,6 +180,15 @@ impl Sink {
     }
 }
 
+/// A tick's samples lie in one 1s window only if ticks start on whole seconds;
+/// a tick straddling a boundary can close a window before its other series arrive.
+fn validate_start_ms(start_ms: i64) -> Result<()> {
+    if start_ms % 1_000 != 0 {
+        bail!("--start-ms must be a whole second (multiple of 1000), got {start_ms}");
+    }
+    Ok(())
+}
+
 fn next_whole_second_ms() -> i64 {
     let now_ms = SystemTime::now()
         .duration_since(UNIX_EPOCH)
@@ -229,6 +239,9 @@ async fn main() -> Result<()> {
     if batch_rows == 0 || args.concurrency == 0 {
         bail!("--batch-rows and --concurrency must be positive");
     }
+    if let Some(start_ms) = args.start_ms {
+        validate_start_ms(start_ms)?;
+    }
 
     let workload = Workload {
         seed: args.seed,
@@ -263,43 +276,38 @@ async fn main() -> Result<()> {
         let started = Instant::now();
 
         let (mut rows, mut bytes) = (0, 0);
-        // ASAPQuery can drop samples behind a group's watermark, so remote write
-        // sends one slot at a time: a slot's requests all return before the
-        // next slot's are sent. ClickHouse sorts on insert, so it sends the
-        // whole tick at once.
-        let slots = (0..args.samples_per_sec).map(|slot| workload.slot_rows(tick, slot, start_ms));
-        let waves: Vec<Vec<Row>> = match args.sink {
-            SinkKind::RemoteWrite => slots.collect(),
-            SinkKind::Clickhouse => vec![slots.flatten().collect()],
-        };
-        for wave in waves {
-            let wave_rows = Arc::new(wave);
-            let mut tasks = tokio::task::JoinSet::new();
-            for begin in (0..wave_rows.len()).step_by(batch_rows) {
-                let end = (begin + batch_rows).min(wave_rows.len());
-                let (wave_rows, sink) = (Arc::clone(&wave_rows), Arc::clone(&sink));
-                let permit = Arc::clone(&permits).acquire_owned().await?;
-                tasks.spawn(async move {
-                    let _permit = permit;
-                    let body = sink.encoder.encode(&wave_rows[begin..end]);
-                    let bytes = body.len();
-                    sink.send(body).await.map(|()| (end - begin, bytes))
-                });
-            }
+        // A tick's requests go out concurrently, each holding whole series
+        // (see `series_batches`): a series' samples arrive in order, but a
+        // group's samples across series may not. That is safe for ASAPQuery:
+        // the tick lies in one window (see `tick_rows`) and is under one
+        // second behind the watermark, within the allowed lateness.
+        let tick_rows = Arc::new(workload.tick_rows(tick, start_ms));
+        let per_series = args.samples_per_sec as usize;
+        let mut tasks = tokio::task::JoinSet::new();
+        for range in series_batches(tick_rows.len(), per_series, batch_rows) {
+            let (tick_rows, sink) = (Arc::clone(&tick_rows), Arc::clone(&sink));
+            let permit = Arc::clone(&permits).acquire_owned().await?;
+            tasks.spawn(async move {
+                let _permit = permit;
+                let rows = range.len();
+                let body = sink.encoder.encode(&tick_rows[range]);
+                let bytes = body.len();
+                sink.send(body).await.map(|()| (rows, bytes))
+            });
+        }
 
-            while let Some(joined) = tasks.join_next().await {
-                // A panicked task counts as a failed request, so the stats of
-                // the ticks before it are still written.
-                match joined.map_err(anyhow::Error::from).and_then(|sent| sent) {
-                    Ok((n_rows, n_bytes)) => {
-                        rows += n_rows;
-                        bytes += n_bytes;
-                    }
-                    Err(e) => {
-                        failed_requests += 1;
-                        if errors.len() < 10 {
-                            errors.push(format!("tick {tick}: {e:#}"));
-                        }
+        while let Some(joined) = tasks.join_next().await {
+            // A panicked task counts as a failed request, so the stats of
+            // the ticks before it are still written.
+            match joined.map_err(anyhow::Error::from).and_then(|sent| sent) {
+                Ok((n_rows, n_bytes)) => {
+                    rows += n_rows;
+                    bytes += n_bytes;
+                }
+                Err(e) => {
+                    failed_requests += 1;
+                    if errors.len() < 10 {
+                        errors.push(format!("tick {tick}: {e:#}"));
                     }
                 }
             }
@@ -374,3 +382,14 @@ async fn main() -> Result<()> {
     }
     Ok(())
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn start_ms_must_be_whole_second() {
+        assert!(validate_start_ms(1_700_000_000_000).is_ok());
+        assert!(validate_start_ms(1_700_000_000_500).is_err());
+    }
+}

asap-tools/data-sources/paced-feeder/src/workload.rs

@@ -40,19 +40,21 @@ impl Workload {
         (format!("g{group}"), format!("i{instance}"))
     }
 
-    /// Samples for sub-second slot `slot` of one-second tick `tick`, one per
-    /// series, with event time starting at `start_ms`. Slots within the second
-    /// are spaced evenly.
-    pub fn slot_rows(&self, tick: u64, slot: u32, start_ms: i64) -> Vec<Row> {
+    /// The samples of one-second tick `tick`, series-major: each series'
+    /// `samples_per_sec` samples in timestamp order, then the next series'.
+    /// Slot `j` of `n` is at `t + (j + 1) · 1 s / n`, where `t` is the tick's
+    /// start, so the tick lies in `(t, t + 1 s]`: the same left-open interval as
+    /// ASAPQuery's one-second windows, so a tick never spans two of them.
+    pub fn tick_rows(&self, tick: u64, start_ms: i64) -> Vec<Row> {
         let per_sec = u64::from(self.samples_per_sec);
-        let slot = u64::from(slot);
-        let timestamp_ms = start_ms + tick as i64 * 1_000 + (slot * 1_000 / per_sec) as i64;
-        let sample_index = tick * per_sec + slot;
+        let tick_ms = start_ms + tick as i64 * 1_000;
         (0..self.num_series())
-            .map(|series| Row {
-                series,
-                timestamp_ms,
-                value: self.value(series, sample_index),
+            .flat_map(|series| {
+                (0..per_sec).map(move |slot| Row {
+                    series,
+                    timestamp_ms: tick_ms + ((slot + 1) * 1_000 / per_sec) as i64,
+                    value: self.value(series, tick * per_sec + slot),
+                })
             })
             .collect()
     }
@@ -65,6 +67,23 @@ impl Workload {
     }
 }
 
+/// Row ranges of the requests for `rows` series-major rows with
+/// `per_series` rows per series: about `batch_rows` each, cut only between
+/// series (at least one series per request), so every series' samples of a
+/// tick reach the sink in one request, in timestamp order. Concurrent requests
+/// then reorder samples only across series, never within one, which keeps
+/// counters monotonic for consumers that read them in arrival order.
+pub fn series_batches(
+    rows: usize,
+    per_series: usize,
+    batch_rows: usize,
+) -> impl Iterator<Item = std::ops::Range<usize>> {
+    let batch = (batch_rows / per_series).max(1) * per_series;
+    (0..rows)
+        .step_by(batch)
+        .map(move |begin| begin..(begin + batch).min(rows))
+}
+
 /// Counter-based uniform draw in (0, 1]: a stateless hash of the inputs, so
 /// any sample can be regenerated independently of the ones before it.
 fn unit_uniform(seed: u64, series: u64, sample_index: u64) -> f64 {
@@ -96,37 +115,59 @@ mod tests {
     }
 
     #[test]
-    fn each_slot_has_every_series_at_its_timestamp() {
+    fn a_tick_is_series_major_in_timestamp_order() {
         let w = workload();
-        for (slot, timestamp_ms) in [(0, 15_000), (1, 15_500)] {
-            let rows = w.slot_rows(5, slot, 10_000);
-            assert_eq!(rows.len(), 6);
-            assert!(rows.iter().all(|r| r.timestamp_ms == timestamp_ms));
-            assert!(rows.iter().map(|r| r.series).eq(0..6));
+        let rows = w.tick_rows(5, 10_000);
+        assert_eq!(rows.len(), 12);
+        for (series, samples) in rows.chunks(2).enumerate() {
+            assert!(samples.iter().all(|r| r.series == series));
+            let timestamps: Vec<i64> = samples.iter().map(|r| r.timestamp_ms).collect();
+            assert_eq!(timestamps, [15_500, 16_000]);
+        }
+    }
+
+    #[test]
+    fn batches_keep_each_series_in_one_request_in_order() {
+        let w = Workload {
+            samples_per_sec: 3,
+            ..workload()
+        };
+        let rows = w.tick_rows(2, 0);
+        // 7 rows round down to 2 series (6 rows); 1 row still takes a series.
+        for (batch_rows, sizes) in [(7, vec![6, 6, 6]), (1, vec![3; 6]), (100, vec![18])] {
+            let batches: Vec<_> = series_batches(rows.len(), 3, batch_rows).collect();
+            assert_eq!(batches.iter().map(|b| b.len()).collect::<Vec<_>>(), sizes);
+            for series in 0..w.num_series() {
+                let holding: Vec<_> = batches
+                    .iter()
+                    .filter(|b| rows[(*b).clone()].iter().any(|r| r.series == series))
+                    .collect();
+                assert_eq!(holding.len(), 1, "series {series}");
+                let timestamps: Vec<i64> = rows[holding[0].clone()]
+                    .iter()
+                    .filter(|r| r.series == series)
+                    .map(|r| r.timestamp_ms)
+                    .collect();
+                assert_eq!(timestamps.len(), 3);
+                assert!(timestamps.windows(2).all(|t| t[0] < t[1]));
+            }
         }
     }
 
     #[test]
     fn values_depend_only_on_seed_series_and_sample_index() {
         let w = workload();
-        // Same slot, different start time: same values.
-        let a: Vec<f64> = w.slot_rows(4, 1, 0).iter().map(|r| r.value).collect();
-        let b: Vec<f64> = w
-            .slot_rows(4, 1, 1_000_000)
-            .iter()
-            .map(|r| r.value)
-            .collect();
-        assert_eq!(a, b);
+        let values = |w: &Workload, start_ms| -> Vec<f64> {
+            w.tick_rows(4, start_ms).iter().map(|r| r.value).collect()
+        };
+        // Same tick, different start time: same values.
+        assert_eq!(values(&w, 0), values(&w, 1_000_000));
         // A different seed changes them.
-        let other = Workload { seed: 8, ..w };
-        assert_ne!(
-            a,
-            other
-                .slot_rows(4, 1, 0)
-                .iter()
-                .map(|r| r.value)
-                .collect::<Vec<_>>()
-        );
+        let other = Workload {
+            seed: 8,
+            ..w.clone()
+        };
+        assert_ne!(values(&w, 0), values(&other, 0));
     }
 
     #[test]