@@ -92,6 +92,9 @@ jobs:
     - name: Type check with mypy
       working-directory: asap-tools/queriers/prometheus-client
       run: mypy .
+    - name: Run tests
+      working-directory: asap-tools/queriers/prometheus-client
+      run: python -m unittest discover -s tests -v
 
   test-utilities:
     needs: detect-changes
 
@@ -160,8 +160,7 @@ http: 2000000 records/s
 ```
 
 Event time starts at the next whole wall-clock second (or `--start-ms`), so
-`NOW()`-relative queries see the data as it arrives. `--send-at-ms` sends the
-first tick at a later wall-clock time, backdating the data by the difference.
+`NOW()`-relative queries see the data as it arrives.
 
 ## Usage
 
 
@@ -93,11 +93,6 @@ struct Args {
     /// next whole wall-clock second, so `NOW()`-relative queries see live data.
     #[arg(long)]
     start_ms: Option<i64>,
-    /// Wall-clock time to send the first tick, in ms since the epoch. Defaults
-    /// to `--start-ms`; a later time backdates the data by the difference, so
-    /// feeders launched at different moments share one event-time origin.
-    #[arg(long, requires = "start_ms")]
-    send_at_ms: Option<i64>,
 
     /// Rows per HTTP request, for gauges and counters rounded down to whole
     /// series (at least one). Defaults to 50,000 for remote write, which keeps
@@ -139,7 +134,6 @@ struct RunStats {
     samples_per_sec: Option<u32>,
     seed: u64,
     start_ms: i64,
-    send_at_ms: i64,
     duration_secs: u64,
     target_rows_per_sec: usize,
     rows_sent: usize,
@@ -279,14 +273,6 @@ fn next_whole_second_ms() -> i64 {
     (now_ms / 1_000 + 1) * 1_000
 }
 
-/// Data sent before its own timestamp would be in the future.
-fn check_send_at(start_ms: i64, send_at_ms: i64) -> Result<()> {
-    if send_at_ms < start_ms {
-        bail!("--send-at-ms ({send_at_ms}) must not be before --start-ms ({start_ms})");
-    }
-    Ok(())
-}
-
 /// Sleep until wall-clock `epoch_ms`, then return the monotonic instant it maps to.
 async fn wait_for_wall_clock(epoch_ms: i64) -> Instant {
     let now_ms = SystemTime::now()
@@ -452,8 +438,6 @@ async fn main() -> Result<()> {
     let sink = Arc::new(Sink::new(&args, sink_kind, url, &workload, batch_rows)?);
     let permits = Arc::new(Semaphore::new(args.concurrency));
     let start_ms = args.start_ms.unwrap_or_else(next_whole_second_ms);
-    let send_at_ms = args.send_at_ms.unwrap_or(start_ms);
-    check_send_at(start_ms, send_at_ms)?;
     let metric_names = workload.names();
 
     let series = if workload.metrics.is_empty() {
@@ -466,12 +450,12 @@ async fn main() -> Result<()> {
         )
     };
     eprintln!(
-        "paced_feeder: {sink_kind:?} -> {url}, {}{series} = {} rows/s for {duration_secs}s, start_ms={start_ms}, send_at_ms={send_at_ms}",
+        "paced_feeder: {sink_kind:?} -> {url}, {}{series} = {} rows/s for {duration_secs}s, start_ms={start_ms}",
         metric_names.join(", "),
         workload.rows_per_tick(),
     );
 
-    let base = wait_for_wall_clock(send_at_ms).await;
+    let base = wait_for_wall_clock(start_ms).await;
     let mut ticks = Vec::with_capacity(duration_secs as usize);
     let mut errors: Vec<String> = Vec::new();
     let mut failed_requests = 0;
@@ -564,7 +548,6 @@ async fn main() -> Result<()> {
         samples_per_sec: (!workload.metrics.is_empty()).then_some(workload.samples_per_sec),
         seed: args.seed,
         start_ms,
-        send_at_ms,
         duration_secs,
         target_rows_per_sec: workload.rows_per_tick(),
         rows_sent,
@@ -609,11 +592,4 @@ mod tests {
         assert!(validate_start_ms(1_700_000_000_000).is_ok());
         assert!(validate_start_ms(1_700_000_000_500).is_err());
     }
-
-    #[test]
-    fn send_at_may_backdate_but_not_postdate_the_data() {
-        assert!(check_send_at(1_000, 1_000).is_ok());
-        assert!(check_send_at(1_000, 301_000).is_ok());
-        assert!(check_send_at(1_000, 999).is_err());
-    }
 }
@@ -457,6 +457,48 @@ These parameters come from the `experiment_type` config group and are prefixed w
 - **Example**: `"gauge"`
 - **Note**: Counter metrics have `_total` suffix, gauge metrics don't
 
+#### Paced Feeder Parameters
+
+`paced_feeder` runs `asap-tools/data-sources/paced-feeder` (built by
+`installation/common/setup_dependencies.sh`) bare-metal on the worker nodes,
+one process per (sink, metric). It pushes over remote write instead of being
+scraped: in `sketchdb` mode to the query engine's ingest
+(`streaming.remote_write`, which must have `parallelism: 1`), plus Prometheus
+when the mode sets `query_prometheus_too`; in every other mode to Prometheus,
+which then starts with `--web.enable-remote-write-receiver`. All processes
+share the seed and table, so every system gets the same data. Its metrics are
+listed under `metrics` with `exporter: paced_feeder`; Prometheus does not
+forward them to the query engine. Generate a whole config from a sketch-bench
+table with `sketch_bench_e2e/sketch_bench_e2e_configs.py`.
+
+#### `experiment_params.exporters.exporter_list.paced_feeder.workload_table` (string, required)
+- **Description**: Local path of the sketch-bench workload table; copied to each feeder node
+- **Example**: `/path/to/synthetic-templatesclassic-shared1.json`
+
+#### `experiment_params.exporters.exporter_list.paced_feeder.metrics` (list, required)
+- **Description**: Metrics of the table's `workloads[0].e2e.metrics` to send. Gauge and counter metrics go to every sink; `records` metrics (one sample per value field, `<metric>.<field>`) only to ASAPQuery, so they are refused when any mode feeds Prometheus (baseline, or `query_prometheus_too`)
+- **Example**: `[data, data_total]`
+
+#### `experiment_params.exporters.exporter_list.paced_feeder.seed` (int, required)
+- **Description**: `--seed` of every feeder process
+- **Example**: `42`
+
+#### `experiment_params.exporters.exporter_list.paced_feeder.duration_secs` (int, required)
+- **Description**: Seconds of data to send. Must be at least the feeder's startup check (7 s) plus `flow.steady_state_wait` plus the query groups' running time
+- **Example**: `4000`
+
+#### `experiment_params.exporters.exporter_list.paced_feeder.samples_per_sec` (int, optional)
+- **Description**: Overrides the table's `samples_per_sec_per_series` (gauge and counter metrics)
+- **Example**: `20`
+
+#### `experiment_params.exporters.exporter_list.paced_feeder.records_per_sec` (int, optional)
+- **Description**: Overrides the table's `records_per_sec` (records metrics)
+- **Example**: `20000`
+
+#### `experiment_params.exporters.exporter_list.paced_feeder.extra_flags` (string, optional)
+- **Description**: Appended to every feeder command line
+- **Example**: `"--concurrency 32"`
+
 ### Query Group Configuration
 
 #### `experiment_params.query_groups` (list, required)
@@ -500,6 +542,15 @@ These parameters come from the `experiment_type` config group and are prefixed w
 - **Example**: `15`
 - **Usage**: Ensures data is available for lookback queries
 
+#### `experiment_params.query_groups[].client_options.align_to_interval` (boolean, optional)
+- **Description**: Evaluate repetition k at `t_k = data_start_time + (max(1, ceil(starting_delay / T)) + k) * T`, with `T = repetition_delay_ms` (whole seconds), sending it at `t_k + query_time_offset`. `starting_delay` counts from `data_start_time`, not as a sleep from client start
+- **Default**: `false`
+- **Usage**: Every evaluation time is a multiple of `T`, and repetition k covers the same samples in every mode however late the mode started; `query_time_offset` (at least 5 s, ASAPQuery's allowed lateness) is how long the data of the window ending at `t_k` gets to arrive. Overrides the query engine's per-aggregation alignment in sketchdb mode. Requires `data_start_time`; the LCM of the aligned intervals must be at most 600 s
+
+#### `experiment_params.query_groups[].client_options.data_start_time` (int)
+- **Description**: Time (epoch seconds) the paced feeder's data starts at
+- **Usage**: Not set by hand: `experiment_run_e2e.py` writes it into each mode's client config after starting the paced feeder, which waits for the next multiple of the aligned groups' intervals' LCM
+
 #### `experiment_params.query_groups[].client_options.starting_delay` (int, required)
 - **Description**: Initial delay in seconds before starting query execution
 - **Example**: `610`
 
@@ -48,6 +48,8 @@
 # Peak query memory is roughly 16 B per sample (~8 GB at this cap).
 PROMETHEUS_QUERY_MAX_SAMPLES = 500_000_000
 PROMETHEUS_REMOTE_WRITE_ENDPOINT = "/api/v1/write"
+# Lets Prometheus accept samples pushed to PROMETHEUS_REMOTE_WRITE_ENDPOINT.
+PROMETHEUS_REMOTE_WRITE_RECEIVER_FLAG = "--web.enable-remote-write-receiver"
 HTTP_READINESS_CURL_TIMEOUT_SECONDS = 5
 QUERY_ENGINE_RUNTIME_INFO_ENDPOINT = "/api/v1/status/runtimeinfo"
 # Query engine spill files live in this subdirectory of its output dir, which
 
@@ -1,8 +1,10 @@
 import os
+import atexit
 import json
 import time
 
 import hydra
+import yaml
 from omegaconf import DictConfig, OmegaConf
 
 import constants
@@ -25,6 +27,13 @@
     DockerPrometheusService,
     DockerVictoriaMetricsService,
     SystemExportersService,
+    PACED_FEEDER_EXPORTER,
+    PacedFeederService,
+    aligned_interval_lcm,
+    check_paced_feeder_config,
+    paced_feeder_min_duration_secs,
+    paced_feeder_sinks,
+    set_data_start,
 )
 from experiment_utils.services.misc import DiscoveryBackend
 
@@ -188,6 +197,39 @@ def main(cfg: DictConfig):
         cfg.experiment_params
     )
 
+    modes_by_name = {
+        m["mode"]: m
+        for m in OmegaConf.to_container(cfg.experiment_params.experiment, resolve=True)
+    }
+    paced_feeder_service = None
+    paced_feeder_config = exporter_config["exporter_list"].get(PACED_FEEDER_EXPORTER)
+    if paced_feeder_config is not None:
+        paced_feeder_config = OmegaConf.to_container(paced_feeder_config, resolve=True)
+        # The feeder runs from its own directory.
+        paced_feeder_config["workload_table"] = os.path.abspath(
+            paced_feeder_config["workload_table"]
+        )
+        check_paced_feeder_config(
+            paced_feeder_config,
+            modes=list(modes_by_name.values()),
+            parallelism=args.parallelism,
+            min_feed_secs=paced_feeder_min_duration_secs(
+                0 if skip_querying else args.steady_state_wait,
+                minimum_experiment_running_time,
+            ),
+            forward_unsupported_queries=args.forward_unsupported_queries,
+            auto_discover_labels=args.controller_auto_discover_labels,
+        )
+        # Also checks the aligned query groups before any service starts.
+        paced_feeder_align_secs = aligned_interval_lcm(
+            OmegaConf.to_container(cfg.experiment_params, resolve=True)
+        )
+        paced_feeder_service = PacedFeederService(provider, args)
+        if not args.no_teardown:
+            # Otherwise a failed run leaves feeders pushing into the next
+            # run's sinks.
+            atexit.register(paced_feeder_service.stop)
+
     for experiment_mode in experiment_modes:
         print(f"Running experiment mode: {experiment_mode}")
         experiment_output_dir = os.path.join(
@@ -228,6 +270,8 @@ def main(cfg: DictConfig):
         exporter_service.stop()
         deathstar_service.stop()
         prometheus_service.reset()
+        if paced_feeder_service is not None:
+            paced_feeder_service.stop()
 
         # Also stop avalanche exporters if they were started
         if config.check_exporter_and_queries_exist("avalanche", cfg.experiment_params):
@@ -481,6 +525,45 @@ def main(cfg: DictConfig):
             ), f"Expected PrometheusService but got {type(prometheus_service).__name__}"
             prometheus_service.start(experiment_output_dir)
 
+        # Pushes to its sinks, so it starts once they accept writes.
+        if paced_feeder_service is not None:
+            prometheus_url = (
+                f"http://{args.remote_write_ip}:"
+                f"{prometheus_service.get_query_endpoint_port()}"
+                f"{constants.PROMETHEUS_REMOTE_WRITE_ENDPOINT}"
+            )
+            sinks = paced_feeder_sinks(
+                modes_by_name[experiment_mode],
+                query_engine_url=f"http://{args.remote_write_ip}:"
+                f"{args.remote_write_base_port}{args.remote_write_path}",
+                prometheus_url=prometheus_url,
+            )
+            if prometheus_url in sinks.values():
+                prometheus_service.wait_until_ready()
+            local_client_config = os.path.join(
+                local_experiment_root_dir,
+                "controller_client_configs",
+                f"{experiment_mode}.yaml",
+            )
+            with open(local_client_config) as f:
+                client_config = yaml.safe_load(f)
+            start_ms = paced_feeder_service.start(
+                config=paced_feeder_config,
+                sinks=sinks,
+                experiment_output_dir=experiment_output_dir,
+                local_experiment_dir=local_experiment_dir,
+                align_secs=paced_feeder_align_secs,
+            )
+            set_data_start(client_config, start_ms)
+            with open(local_client_config, "w") as f:
+                yaml.safe_dump(client_config, f, sort_keys=False)
+            sync.rsync_controller_client_configs(
+                provider,
+                experiment_root_output_dir,
+                local_experiment_root_dir,
+                node_offset=args.node_offset,
+            )
+
         # Start Prometheus throughput monitoring if enabled
         if args.throughput_prometheus:
             prometheus_throughput_monitor = PrometheusThroughputMonitor(
@@ -562,6 +645,9 @@ def main(cfg: DictConfig):
 
         # Skip teardown if the no_teardown flag is set
         if not args.no_teardown:
+            # Before its sinks go away, so its log shows no spurious errors.
+            if paced_feeder_service is not None:
+                paced_feeder_service.stop()
             if experiment_mode == constants.SKETCHDB_EXPERIMENT_NAME:
                 query_engine_service.stop()
 
 
@@ -134,6 +134,17 @@ def read_exporter_config(experiment_params: DictConfig) -> Tuple[Optional[Dict],
                     "cluster_data_exporter with parts_mode='part-index' requires 'part_index'",
                 )
 
+    from experiment_utils.services.paced_feeder import (
+        PACED_FEEDER_EXPORTER,
+        PACED_FEEDER_REQUIRED_KEYS,
+    )
+
+    if PACED_FEEDER_EXPORTER in exporters_config.exporter_list:
+        feeder_config = exporters_config.exporter_list[PACED_FEEDER_EXPORTER]
+        missing_keys = [k for k in PACED_FEEDER_REQUIRED_KEYS if k not in feeder_config]
+        if missing_keys:
+            return None, f"Missing keys in paced_feeder section: {missing_keys}"
+
     return exporters_config, ""
 
 
@@ -186,9 +197,36 @@ def call_generate_prometheus_config(
     import generate_prometheus_config
     import generate_victoriametrics_config
 
+    from experiment_utils.services.paced_feeder import (
+        PACED_FEEDER_EXPORTER,
+        paced_feeder_metric_names,
+    )
+
+    # The paced feeder writes its metrics to the query engine itself, so
+    # Prometheus forwarding them too would send them twice.
+    fed_metrics = paced_feeder_metric_names(cfg.experiment_params)
+    metrics_to_remote_write = [
+        m
+        for m in get_metrics_to_remote_write(cfg.experiment_params)
+        if m not in fed_metrics
+    ]
+    # With only fed metrics there is nothing for Prometheus to forward. (An
+    # empty list without the feeder keeps meaning "forward everything".)
+    forward_to_query_engine = bool(metrics_to_remote_write) or not fed_metrics
+    exporter_list = cfg.experiment_params.get("exporters", {}).get("exporter_list", {})
+    other_exporters = sorted(set(exporter_list) - {PACED_FEEDER_EXPORTER})
+    if not forward_to_query_engine and other_exporters:
+        raise ValueError(
+            f"exporters {other_exporters} run next to the paced feeder but none "
+            "of their metrics are listed under experiment_params.metrics, so "
+            "Prometheus would forward nothing of theirs to the query engine"
+        )
     # Set up remote write if this is a SketchDB experiment
-    if experiment_mode == sketchdb_experiment_name and sketchdb_experiment_name:
-        metrics_to_remote_write = get_metrics_to_remote_write(cfg.experiment_params)
+    if (
+        experiment_mode == sketchdb_experiment_name
+        and sketchdb_experiment_name
+        and forward_to_query_engine
+    ):
         args.remote_write_metric_names = metrics_to_remote_write
 
         # Build remote_write_url from experiment params