From 4983454b3ff0a2504c9924821da33f5d842a909a Mon Sep 17 00:00:00 2001
From: zz_y <zeyingz@umd.edu>
Date: Thu, 3 Sep 2026 13:53:14 -0600
Subject: [PATCH] feat: restore Remote Write compatibility profile

---
 Cargo.lock                                    |   7 +
 data_plane/Cargo.toml                         |   1 +
 .../examples/asapquery/streaming-config.yaml  |  35 +
 data_plane/src/drivers/ingest/mod.rs          |   4 +
 .../drivers/ingest/prometheus_remote_write.rs | 727 ++++++++++++++++++
 data_plane/src/drivers/query/servers/http.rs  | 232 +++++-
 data_plane/src/lib.rs                         |   3 +
 data_plane/src/main.rs                        | 162 +++-
 .../src/precompute_engine/series_router.rs    |  66 ++
 docs/README.md                                |   1 +
 docs/user_guide/asapquery-profile.md          |  59 ++
 11 files changed, 1288 insertions(+), 9 deletions(-)
 create mode 100644 data_plane/examples/asapquery/streaming-config.yaml
 create mode 100644 data_plane/src/drivers/ingest/prometheus_remote_write.rs
 create mode 100644 docs/user_guide/asapquery-profile.md

diff --git a/Cargo.lock b/Cargo.lock
index e84f81018..cb41bc184 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -1012,6 +1012,7 @@ dependencies = [
  "serde",
  "serde_json",
  "serde_yaml",
+ "snap",
  "structopt",
  "tempfile",
  "thiserror 1.0.69",
@@ -3250,6 +3251,12 @@ version = "1.15.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "67b1b7a3b5fe4f1376887184045fcf45c69e92af734b7aaddc05fb777b6fbd03"
 
+[[package]]
+name = "snap"
+version = "1.1.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "199905e6153d6405f9728fe44daace35f8f837bbf830bb6e85fbd5828709a886"
+
 [[package]]
 name = "socket2"
 version = "0.5.10"
diff --git a/data_plane/Cargo.toml b/data_plane/Cargo.toml
index e42164a84..89ec7817c 100644
--- a/data_plane/Cargo.toml
+++ b/data_plane/Cargo.toml
@@ -71,6 +71,7 @@ hex = "0.4"
 arrow = "53.4.1"
 futures = "0.3"
 prost = "0.13"
+snap = "1.1"
 # Vendored modified opentelemetry-proto with first-class sketch metric
 # variants from DataCollector (DDSketch / KLLSketch / CountSketch /
 # CountMinSketch / HLLSketch). See docs/pipeline-query-catalog.md §5.4 in the
diff --git a/data_plane/examples/asapquery/streaming-config.yaml b/data_plane/examples/asapquery/streaming-config.yaml
new file mode 100644
index 000000000..c2174a06c
--- /dev/null
+++ b/data_plane/examples/asapquery/streaming-config.yaml
@@ -0,0 +1,35 @@
+aggregations:
+  - aggregationType: Sum
+    aggregationSubType: ''
+    metric: compat_value
+    labels:
+      grouping: [job]
+      rollup: []
+      aggregated: []
+    parameters: {}
+    windowSize: 30
+    windowType: tumbling
+    spatialFilter: ''
+  - aggregationType: Increase
+    aggregationSubType: ''
+    metric: compat_counter_total
+    labels:
+      grouping: [job]
+      rollup: []
+      aggregated: []
+    parameters: {}
+    windowSize: 30
+    windowType: tumbling
+    spatialFilter: ''
+  - aggregationType: DDSketch
+    aggregationSubType: ''
+    metric: compat_latency_seconds
+    labels:
+      grouping: [job]
+      rollup: []
+      aggregated: []
+    parameters:
+      relative_accuracy: 0.01
+    windowSize: 30
+    windowType: tumbling
+    spatialFilter: ''
diff --git a/data_plane/src/drivers/ingest/mod.rs b/data_plane/src/drivers/ingest/mod.rs
index 21ca4d43f..39d0e8eff 100644
--- a/data_plane/src/drivers/ingest/mod.rs
+++ b/data_plane/src/drivers/ingest/mod.rs
@@ -1,5 +1,9 @@
 pub mod otel;
+pub mod prometheus_remote_write;
 pub mod series_resolver;
 
 pub use otel::{OtlpReceiver, OtlpReceiverConfig};
+pub use prometheus_remote_write::{
+    PrometheusRemoteWriteConfig, PrometheusRemoteWriteReceiver, RemoteWriteStats,
+};
 pub use series_resolver::{canonical_attrs_fingerprint, SeriesIdResolver};
diff --git a/data_plane/src/drivers/ingest/prometheus_remote_write.rs b/data_plane/src/drivers/ingest/prometheus_remote_write.rs
new file mode 100644
index 000000000..55d1a3767
--- /dev/null
+++ b/data_plane/src/drivers/ingest/prometheus_remote_write.rs
@@ -0,0 +1,727 @@
+//! Prometheus Remote Write v1 adapter for the backend-local precompute path.
+//!
+//! This module deliberately stops at wire validation, canonical series
+//! identity, retry deduplication, and config-driven routing. Aggregation-family
+//! selection remains owned by the installed precompute plan.
+
+use crate::precompute_engine::ingest_handler::IngestState;
+use crate::precompute_engine::series_router::{TryRouteError, WorkerMessage};
+use prost::Message;
+use std::collections::{HashMap, VecDeque};
+use std::sync::atomic::{AtomicU64, Ordering};
+use std::sync::{Arc, Mutex};
+use std::time::{Duration, Instant};
+
+pub const STALE_NAN_BITS: u64 = 0x7ff0_0000_0000_0002;
+
+#[derive(Clone, PartialEq, Message)]
+pub struct WriteRequest {
+    #[prost(message, repeated, tag = "1")]
+    pub timeseries: Vec<TimeSeries>,
+}
+
+#[derive(Clone, PartialEq, Message)]
+pub struct TimeSeries {
+    #[prost(message, repeated, tag = "1")]
+    pub labels: Vec<Label>,
+    #[prost(message, repeated, tag = "2")]
+    pub samples: Vec<Sample>,
+    #[prost(message, repeated, tag = "3")]
+    pub exemplars: Vec<OpaqueMessage>,
+    #[prost(message, repeated, tag = "4")]
+    pub histograms: Vec<OpaqueMessage>,
+}
+
+/// Presence-only decoder for unsupported non-scalar v1 payloads. Prost skips
+/// the nested fields while preserving whether such a payload was supplied.
+#[derive(Clone, PartialEq, Message)]
+pub struct OpaqueMessage {}
+
+#[derive(Clone, PartialEq, Message)]
+pub struct Label {
+    #[prost(string, tag = "1")]
+    pub name: String,
+    #[prost(string, tag = "2")]
+    pub value: String,
+}
+
+#[derive(Clone, PartialEq, Message)]
+pub struct Sample {
+    #[prost(double, tag = "1")]
+    pub value: f64,
+    #[prost(int64, tag = "2")]
+    pub timestamp: i64,
+}
+
+#[derive(Debug, Clone)]
+pub struct PrometheusRemoteWriteConfig {
+    pub max_compressed_bytes: usize,
+    pub max_decompressed_bytes: usize,
+    pub max_timeseries: usize,
+    pub max_samples: usize,
+    pub dedup_horizon: Duration,
+    pub max_dedup_entries: usize,
+}
+
+impl Default for PrometheusRemoteWriteConfig {
+    fn default() -> Self {
+        Self {
+            max_compressed_bytes: 32 * 1024 * 1024,
+            max_decompressed_bytes: 128 * 1024 * 1024,
+            max_timeseries: 100_000,
+            max_samples: 1_000_000,
+            dedup_horizon: Duration::from_secs(10 * 60),
+            max_dedup_entries: 2_000_000,
+        }
+    }
+}
+
+#[derive(Debug, Default)]
+pub struct RemoteWriteStats {
+    pub requests: AtomicU64,
+    pub samples: AtomicU64,
+    pub stale_markers: AtomicU64,
+    pub duplicates: AtomicU64,
+    pub rejected_requests: AtomicU64,
+    pub bytes: AtomicU64,
+}
+
+#[derive(Clone)]
+pub struct PrometheusRemoteWriteReceiver {
+    inner: Arc<ReceiverInner>,
+}
+
+struct ReceiverInner {
+    config: PrometheusRemoteWriteConfig,
+    ingest: Arc<IngestState>,
+    dedup: Mutex<DedupState>,
+    stats: Arc<RemoteWriteStats>,
+}
+
+#[derive(Default)]
+struct DedupState {
+    values: HashMap<(String, i64), DedupValue>,
+    expiry: VecDeque<(Instant, String, i64)>,
+}
+
+#[derive(Clone, Copy, PartialEq, Eq)]
+enum DedupValue {
+    Number(u64),
+    Stale,
+}
+
+#[derive(Debug, thiserror::Error)]
+pub enum RemoteWriteError {
+    #[error("compressed request exceeds {0} bytes")]
+    CompressedTooLarge(usize),
+    #[error("decompressed request exceeds {0} bytes")]
+    DecompressedTooLarge(usize),
+    #[error("snappy decompression failed: {0}")]
+    Snappy(String),
+    #[error("protobuf decoding failed: {0}")]
+    Protobuf(String),
+    #[error("request has {actual} time series; limit is {limit}")]
+    TooManySeries { actual: usize, limit: usize },
+    #[error("request has more than {0} samples")]
+    TooManySamples(usize),
+    #[error("invalid series: {0}")]
+    InvalidSeries(String),
+    #[error("invalid sample for {series} at {timestamp}: {reason}")]
+    InvalidSample {
+        series: String,
+        timestamp: i64,
+        reason: String,
+    },
+    #[error("conflicting sample for {series} at timestamp {timestamp}")]
+    Conflict { series: String, timestamp: i64 },
+    #[error("deduplication capacity ({0}) is exhausted")]
+    DedupCapacity(usize),
+    #[error(transparent)]
+    Backpressure(#[from] TryRouteError),
+}
+
+#[derive(Debug, Clone, PartialEq)]
+pub struct CanonicalSample {
+    pub metric: String,
+    pub labels: HashMap<String, String>,
+    pub series_key: String,
+    pub timestamp_ms: i64,
+    pub value: Option<f64>,
+}
+
+impl PrometheusRemoteWriteReceiver {
+    pub fn new(config: PrometheusRemoteWriteConfig, ingest: Arc<IngestState>) -> Self {
+        Self {
+            inner: Arc::new(ReceiverInner {
+                config,
+                ingest,
+                dedup: Mutex::new(DedupState::default()),
+                stats: Arc::new(RemoteWriteStats::default()),
+            }),
+        }
+    }
+
+    pub fn config(&self) -> &PrometheusRemoteWriteConfig {
+        &self.inner.config
+    }
+
+    pub fn stats(&self) -> Arc<RemoteWriteStats> {
+        self.inner.stats.clone()
+    }
+
+    /// Decode, validate, deduplicate, and enqueue one whole v1 request.
+    /// All validation and all queue reservations complete before any message
+    /// becomes visible to a worker.
+    pub fn accept(&self, body: &[u8]) -> Result<(), RemoteWriteError> {
+        self.inner.stats.requests.fetch_add(1, Ordering::Relaxed);
+        self.inner
+            .stats
+            .bytes
+            .fetch_add(body.len() as u64, Ordering::Relaxed);
+        let result = self.accept_inner(body);
+        if result.is_err() {
+            self.inner
+                .stats
+                .rejected_requests
+                .fetch_add(1, Ordering::Relaxed);
+        }
+        result
+    }
+
+    fn accept_inner(&self, body: &[u8]) -> Result<(), RemoteWriteError> {
+        let config = &self.inner.config;
+        if body.len() > config.max_compressed_bytes {
+            return Err(RemoteWriteError::CompressedTooLarge(
+                config.max_compressed_bytes,
+            ));
+        }
+        let decoded_len = snap::raw::decompress_len(body)
+            .map_err(|error| RemoteWriteError::Snappy(error.to_string()))?;
+        if decoded_len > config.max_decompressed_bytes {
+            return Err(RemoteWriteError::DecompressedTooLarge(
+                config.max_decompressed_bytes,
+            ));
+        }
+        let decoded = snap::raw::Decoder::new()
+            .decompress_vec(body)
+            .map_err(|error| RemoteWriteError::Snappy(error.to_string()))?;
+        let request = WriteRequest::decode(decoded.as_slice())
+            .map_err(|error| RemoteWriteError::Protobuf(error.to_string()))?;
+        let samples = canonicalize_request(&request, config)?;
+
+        let now = Instant::now();
+        let mut dedup = self
+            .inner
+            .dedup
+            .lock()
+            .expect("remote write dedup poisoned");
+        dedup.evict_before(now.checked_sub(config.dedup_horizon).unwrap_or(now));
+
+        // Validate conflicts both against committed history and inside this
+        // request before reserving any worker capacity.
+        let mut batch_values: HashMap<(String, i64), DedupValue> = HashMap::new();
+        let mut new_samples = Vec::with_capacity(samples.len());
+        let mut duplicates = 0u64;
+        for sample in samples {
+            let key = (sample.series_key.clone(), sample.timestamp_ms);
+            let value = sample
+                .value
+                .map(|v| DedupValue::Number(v.to_bits()))
+                .unwrap_or(DedupValue::Stale);
+            let prior = batch_values.get(&key).or_else(|| dedup.values.get(&key));
+            match prior {
+                Some(previous) if *previous == value => {
+                    duplicates += 1;
+                }
+                Some(_) => {
+                    return Err(RemoteWriteError::Conflict {
+                        series: sample.series_key,
+                        timestamp: sample.timestamp_ms,
+                    });
+                }
+                None => {
+                    batch_values.insert(key, value);
+                    new_samples.push(sample);
+                }
+            }
+        }
+        if dedup.values.len().saturating_add(batch_values.len()) > config.max_dedup_entries {
+            return Err(RemoteWriteError::DedupCapacity(config.max_dedup_entries));
+        }
+
+        let messages = route_messages(&new_samples, &self.inner.ingest);
+        self.inner
+            .ingest
+            .router
+            .try_route_group_batch_atomic(messages)?;
+
+        for ((series, timestamp), value) in batch_values {
+            dedup.values.insert((series.clone(), timestamp), value);
+            dedup.expiry.push_back((now, series, timestamp));
+        }
+        let stale_count = new_samples
+            .iter()
+            .filter(|sample| sample.value.is_none())
+            .count() as u64;
+        self.inner.stats.samples.fetch_add(
+            (new_samples.len() as u64).saturating_sub(stale_count),
+            Ordering::Relaxed,
+        );
+        self.inner
+            .stats
+            .stale_markers
+            .fetch_add(stale_count, Ordering::Relaxed);
+        self.inner
+            .stats
+            .duplicates
+            .fetch_add(duplicates, Ordering::Relaxed);
+        self.inner
+            .ingest
+            .samples_ingested
+            .fetch_add(new_samples.len() as u64, Ordering::Relaxed);
+        Ok(())
+    }
+}
+
+impl DedupState {
+    fn evict_before(&mut self, cutoff: Instant) {
+        while self
+            .expiry
+            .front()
+            .is_some_and(|(accepted_at, _, _)| *accepted_at < cutoff)
+        {
+            if let Some((_, series, timestamp)) = self.expiry.pop_front() {
+                self.values.remove(&(series, timestamp));
+            }
+        }
+    }
+}
+
+fn canonicalize_request(
+    request: &WriteRequest,
+    config: &PrometheusRemoteWriteConfig,
+) -> Result<Vec<CanonicalSample>, RemoteWriteError> {
+    if request.timeseries.len() > config.max_timeseries {
+        return Err(RemoteWriteError::TooManySeries {
+            actual: request.timeseries.len(),
+            limit: config.max_timeseries,
+        });
+    }
+    let mut out = Vec::new();
+    for timeseries in &request.timeseries {
+        if !timeseries.exemplars.is_empty() || !timeseries.histograms.is_empty() {
+            return Err(RemoteWriteError::InvalidSeries(
+                "exemplars and native histograms are outside compatibility level 1".into(),
+            ));
+        }
+        if out.len().saturating_add(timeseries.samples.len()) > config.max_samples {
+            return Err(RemoteWriteError::TooManySamples(config.max_samples));
+        }
+        let (metric, labels, series_key) = canonicalize_labels(&timeseries.labels)?;
+        for sample in &timeseries.samples {
+            let value = if sample.value.to_bits() == STALE_NAN_BITS {
+                None
+            } else if !sample.value.is_finite() {
+                return Err(RemoteWriteError::InvalidSample {
+                    series: series_key.clone(),
+                    timestamp: sample.timestamp,
+                    reason: "only finite values or the Prometheus stale marker are accepted".into(),
+                });
+            } else {
+                Some(sample.value)
+            };
+            out.push(CanonicalSample {
+                metric: metric.clone(),
+                labels: labels.clone(),
+                series_key: series_key.clone(),
+                timestamp_ms: sample.timestamp,
+                value,
+            });
+        }
+    }
+    Ok(out)
+}
+
+fn canonicalize_labels(
+    labels: &[Label],
+) -> Result<(String, HashMap<String, String>, String), RemoteWriteError> {
+    let mut metric = None;
+    let mut attrs = HashMap::new();
+    for label in labels {
+        if !valid_label_name(&label.name) {
+            return Err(RemoteWriteError::InvalidSeries(format!(
+                "invalid label name {:?}",
+                label.name
+            )));
+        }
+        if label.name == "__name__" {
+            if metric.replace(label.value.clone()).is_some() {
+                return Err(RemoteWriteError::InvalidSeries(
+                    "duplicate __name__ label".into(),
+                ));
+            }
+        } else if attrs
+            .insert(label.name.clone(), label.value.clone())
+            .is_some()
+        {
+            return Err(RemoteWriteError::InvalidSeries(format!(
+                "duplicate label {:?}",
+                label.name
+            )));
+        }
+    }
+    let metric =
+        metric.ok_or_else(|| RemoteWriteError::InvalidSeries("missing __name__".into()))?;
+    if !valid_metric_name(&metric) {
+        return Err(RemoteWriteError::InvalidSeries(format!(
+            "invalid metric name {metric:?}"
+        )));
+    }
+    let mut sorted: Vec<_> = attrs.iter().collect();
+    sorted.sort_by(|a, b| a.0.cmp(b.0));
+    let series_key = if sorted.is_empty() {
+        metric.clone()
+    } else {
+        let rendered = sorted
+            .into_iter()
+            .map(|(name, value)| format!("{name}=\"{}\"", escape_label_value(value)))
+            .collect::<Vec<_>>()
+            .join(",");
+        format!("{metric}{{{rendered}}}")
+    };
+    Ok((metric, attrs, series_key))
+}
+
+fn route_messages(samples: &[CanonicalSample], ingest: &Arc<IngestState>) -> Vec<WorkerMessage> {
+    type Bucket = (u64, asap_types::PolicyFingerprint, String);
+    type RoutedSample = (String, i64, f64);
+    let snapshot = ingest.config_snapshot();
+    let _ = crate::storage_engines::sketch_db::lifecycle::reconcile_if_config_changed(
+        ingest.sketch_index.as_ref(),
+        &snapshot,
+        crate::storage_engines::sketch_db::DEFAULT_RETIREMENT_RETENTION,
+    );
+    let mut buckets: HashMap<u64, (Bucket, Vec<RoutedSample>)> = HashMap::new();
+    for sample in samples {
+        let Some(value) = sample.value else {
+            // Staleness is a lifecycle signal, never an accumulator input.
+            continue;
+        };
+        for config in snapshot.get_all_aggregation_configs().values() {
+            if config.metric != sample.metric
+                && config.spatial_filter_normalized != sample.metric
+                && config.spatial_filter != sample.metric
+            {
+                continue;
+            }
+            let group_key = IngestState::extract_group_key_for(&sample.series_key, config);
+            let grouping_pairs: Vec<(&str, &str)> = config
+                .grouping_labels
+                .labels
+                .iter()
+                .map(|name| {
+                    (
+                        name.as_str(),
+                        sample.labels.get(name).map(String::as_str).unwrap_or(""),
+                    )
+                })
+                .collect();
+            let attrs_fp = super::canonical_attrs_fingerprint(&grouping_pairs);
+            let agg_kind = crate::storage_engines::sketch_db::data::AggKind::ExactAgg {
+                agg_type: config.aggregation_type,
+                parameters_canonical: crate::storage_engines::sketch_db::data::canonical_parameters(
+                    &config.parameters,
+                ),
+                spatial_filter_canonical: config.spatial_filter_normalized.clone(),
+            };
+            let sid = ingest.series_resolver.resolve(
+                &config.metric,
+                &attrs_fp,
+                &agg_kind.canonical_string(),
+            );
+            let policy_fp = asap_types::PolicyFingerprint(config.policy_fp_u64());
+            buckets
+                .entry(sid)
+                .or_insert_with(|| ((sid, policy_fp, group_key), Vec::new()))
+                .1
+                .push((sample.series_key.clone(), sample.timestamp_ms, value));
+        }
+    }
+    let received_at = Instant::now();
+    buckets
+        .into_values()
+        .map(
+            |((sid, policy_fp, group_key), samples)| WorkerMessage::GroupSamples {
+                sid,
+                policy_fp,
+                group_key,
+                samples,
+                ingest_received_at: received_at,
+            },
+        )
+        .collect()
+}
+
+fn valid_label_name(name: &str) -> bool {
+    let mut chars = name.chars();
+    matches!(chars.next(), Some('_' | 'a'..='z' | 'A'..='Z'))
+        && chars.all(|ch| matches!(ch, '_' | 'a'..='z' | 'A'..='Z' | '0'..='9'))
+}
+
+fn valid_metric_name(name: &str) -> bool {
+    let mut chars = name.chars();
+    matches!(chars.next(), Some('_' | ':' | 'a'..='z' | 'A'..='Z'))
+        && chars.all(|ch| matches!(ch, '_' | ':' | 'a'..='z' | 'A'..='Z' | '0'..='9'))
+}
+
+fn escape_label_value(value: &str) -> String {
+    let mut out = String::with_capacity(value.len());
+    for ch in value.chars() {
+        match ch {
+            '\\' => out.push_str("\\\\"),
+            '"' => out.push_str("\\\""),
+            '\n' => out.push_str("\\n"),
+            other => out.push(other),
+        }
+    }
+    out
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::precompute_engine::ingest_handler::IngestObservability;
+    use crate::precompute_engine::series_router::SeriesRouter;
+    use crate::storage_engines::types::{HotReloadStreamingConfig, StreamingConfig};
+    use tokio::sync::mpsc;
+
+    fn compressed(request: WriteRequest) -> Vec<u8> {
+        snap::raw::Encoder::new()
+            .compress_vec(&request.encode_to_vec())
+            .unwrap()
+    }
+
+    fn receiver(config: PrometheusRemoteWriteConfig) -> PrometheusRemoteWriteReceiver {
+        let (sender, _receiver) = mpsc::channel(8);
+        let ingest = Arc::new(IngestState {
+            router: SeriesRouter::new(vec![sender]),
+            samples_ingested: AtomicU64::new(0),
+            samples_blocked_by_schema_barrier: AtomicU64::new(0),
+            hot_reload_config: HotReloadStreamingConfig::new(StreamingConfig::default()),
+            pass_raw_samples: false,
+            sketch_snapshots: dashmap::DashMap::new(),
+            series_resolver: Arc::new(super::super::SeriesIdResolver::new()),
+            sketch_index: Arc::new(crate::storage_engines::sketch_db::index::SketchStore::new()),
+            observability: IngestObservability::default(),
+        });
+        PrometheusRemoteWriteReceiver::new(config, ingest)
+    }
+
+    fn configured_receiver() -> (PrometheusRemoteWriteReceiver, mpsc::Receiver<WorkerMessage>) {
+        use asap_types::enums::WindowKind;
+        use asap_types::{AggregationConfig, AggregationType, KeyByLabelNames};
+        let aggregation = AggregationConfig {
+            aggregation_type: AggregationType::Sum,
+            aggregation_sub_type: String::new(),
+            parameters: HashMap::new(),
+            grouping_labels: KeyByLabelNames::new(vec!["job".into()]),
+            aggregated_labels: KeyByLabelNames::empty(),
+            rollup_labels: KeyByLabelNames::empty(),
+            original_yaml: String::new(),
+            window_size: 60,
+            slide_interval: 60,
+            window_type: WindowKind::Tumbling,
+            spatial_filter: String::new(),
+            spatial_filter_normalized: String::new(),
+            metric: "requests_total".into(),
+            num_aggregates_to_retain: None,
+            table_name: None,
+            value_column: None,
+        };
+        let policy_fp = aggregation.policy_fp_u64();
+        let streaming = StreamingConfig::new(HashMap::from([(policy_fp, aggregation)]));
+        let (sender, receiver) = mpsc::channel(8);
+        let ingest = Arc::new(IngestState {
+            router: SeriesRouter::new(vec![sender]),
+            samples_ingested: AtomicU64::new(0),
+            samples_blocked_by_schema_barrier: AtomicU64::new(0),
+            hot_reload_config: HotReloadStreamingConfig::new(streaming),
+            pass_raw_samples: false,
+            sketch_snapshots: dashmap::DashMap::new(),
+            series_resolver: Arc::new(super::super::SeriesIdResolver::new()),
+            sketch_index: Arc::new(crate::storage_engines::sketch_db::index::SketchStore::new()),
+            observability: IngestObservability::default(),
+        });
+        (
+            PrometheusRemoteWriteReceiver::new(PrometheusRemoteWriteConfig::default(), ingest),
+            receiver,
+        )
+    }
+
+    fn one_sample(value: f64) -> Vec<u8> {
+        compressed(WriteRequest {
+            timeseries: vec![TimeSeries {
+                labels: vec![
+                    Label {
+                        name: "__name__".into(),
+                        value: "requests_total".into(),
+                    },
+                    Label {
+                        name: "job".into(),
+                        value: "api".into(),
+                    },
+                ],
+                samples: vec![Sample {
+                    value,
+                    timestamp: 100,
+                }],
+                exemplars: Vec::new(),
+                histograms: Vec::new(),
+            }],
+        })
+    }
+
+    #[test]
+    fn canonicalizes_labels_and_recognizes_staleness() {
+        let request = WriteRequest {
+            timeseries: vec![TimeSeries {
+                labels: vec![
+                    Label {
+                        name: "zone".into(),
+                        value: "a\n\"b".into(),
+                    },
+                    Label {
+                        name: "__name__".into(),
+                        value: "http_requests_total".into(),
+                    },
+                    Label {
+                        name: "job".into(),
+                        value: "api".into(),
+                    },
+                ],
+                samples: vec![
+                    Sample {
+                        value: 3.0,
+                        timestamp: 100,
+                    },
+                    Sample {
+                        value: f64::from_bits(STALE_NAN_BITS),
+                        timestamp: 200,
+                    },
+                ],
+                exemplars: Vec::new(),
+                histograms: Vec::new(),
+            }],
+        };
+        let decoded = snap::raw::Decoder::new()
+            .decompress_vec(&compressed(request))
+            .unwrap();
+        let request = WriteRequest::decode(decoded.as_slice()).unwrap();
+        let samples =
+            canonicalize_request(&request, &PrometheusRemoteWriteConfig::default()).unwrap();
+        assert_eq!(
+            samples[0].series_key,
+            "http_requests_total{job=\"api\",zone=\"a\\n\\\"b\"}"
+        );
+        assert_eq!(samples[0].value, Some(3.0));
+        assert_eq!(samples[1].value, None);
+    }
+
+    #[test]
+    fn rejects_non_stale_nan_and_conflicting_labels() {
+        let config = PrometheusRemoteWriteConfig::default();
+        let invalid_nan = WriteRequest {
+            timeseries: vec![TimeSeries {
+                labels: vec![Label {
+                    name: "__name__".into(),
+                    value: "m".into(),
+                }],
+                samples: vec![Sample {
+                    value: f64::NAN,
+                    timestamp: 1,
+                }],
+                exemplars: Vec::new(),
+                histograms: Vec::new(),
+            }],
+        };
+        assert!(matches!(
+            canonicalize_request(&invalid_nan, &config),
+            Err(RemoteWriteError::InvalidSample { .. })
+        ));
+        assert!(canonicalize_labels(&[
+            Label {
+                name: "__name__".into(),
+                value: "m".into()
+            },
+            Label {
+                name: "job".into(),
+                value: "a".into()
+            },
+            Label {
+                name: "job".into(),
+                value: "b".into()
+            },
+        ])
+        .is_err());
+    }
+
+    #[test]
+    fn resource_limits_apply_before_allocation() {
+        let bytes = compressed(WriteRequest {
+            timeseries: Vec::new(),
+        });
+        let mut config = PrometheusRemoteWriteConfig::default();
+        config.max_compressed_bytes = bytes.len() - 1;
+        assert!(matches!(
+            receiver(config).accept(&bytes),
+            Err(RemoteWriteError::CompressedTooLarge(_))
+        ));
+    }
+
+    #[test]
+    fn replay_is_a_noop_and_conflict_is_rejected() {
+        let receiver = receiver(PrometheusRemoteWriteConfig::default());
+        receiver.accept(&one_sample(4.0)).unwrap();
+        receiver.accept(&one_sample(4.0)).unwrap();
+        assert_eq!(receiver.stats().samples.load(Ordering::Relaxed), 1);
+        assert_eq!(receiver.stats().duplicates.load(Ordering::Relaxed), 1);
+        assert!(matches!(
+            receiver.accept(&one_sample(5.0)),
+            Err(RemoteWriteError::Conflict { .. })
+        ));
+        assert_eq!(receiver.stats().samples.load(Ordering::Relaxed), 1);
+        assert_eq!(
+            receiver.stats().rejected_requests.load(Ordering::Relaxed),
+            1
+        );
+    }
+
+    #[test]
+    fn stale_marker_is_deduplicated_but_never_counted_as_numeric() {
+        let receiver = receiver(PrometheusRemoteWriteConfig::default());
+        let stale = one_sample(f64::from_bits(STALE_NAN_BITS));
+        receiver.accept(&stale).unwrap();
+        receiver.accept(&stale).unwrap();
+        assert_eq!(receiver.stats().samples.load(Ordering::Relaxed), 0);
+        assert_eq!(receiver.stats().stale_markers.load(Ordering::Relaxed), 1);
+        assert_eq!(receiver.stats().duplicates.load(Ordering::Relaxed), 1);
+    }
+
+    #[tokio::test]
+    async fn valid_request_routes_canonical_sample_to_installed_plan() {
+        let (receiver, mut worker) = configured_receiver();
+        receiver.accept(&one_sample(4.0)).unwrap();
+        let message = worker.recv().await.expect("routed worker message");
+        let WorkerMessage::GroupSamples {
+            group_key, samples, ..
+        } = message
+        else {
+            panic!("expected GroupSamples");
+        };
+        assert_eq!(group_key, "api");
+        assert_eq!(
+            samples,
+            vec![("requests_total{job=\"api\"}".into(), 100, 4.0)]
+        );
+    }
+}
diff --git a/data_plane/src/drivers/query/servers/http.rs b/data_plane/src/drivers/query/servers/http.rs
index 95f41a8a7..ce33f9792 100644
--- a/data_plane/src/drivers/query/servers/http.rs
+++ b/data_plane/src/drivers/query/servers/http.rs
@@ -1,8 +1,8 @@
 use crate::drivers::query::adapters::{ParsedQueryRequest, ParsedRangeQueryRequest};
 use axum::{
     body::Bytes,
-    extract::{Form, Query, State},
-    http::StatusCode,
+    extract::{DefaultBodyLimit, Form, Query, State},
+    http::{HeaderMap, StatusCode},
     response::{IntoResponse, Json, Response},
     routing::{get, post},
     Router,
@@ -151,6 +151,7 @@ pub struct HttpServer {
     physical_plan_lock: Arc<tokio::sync::Mutex<()>>,
     active_physical_plan: Option<crate::storage_engines::types::HotReloadActivePhysicalPlan>,
     physical_plan_lifecycle: Option<crate::storage_engines::types::PhysicalPlanLifecycle>,
+    remote_write: Option<crate::drivers::ingest::PrometheusRemoteWriteReceiver>,
 }
 
 #[derive(Clone)]
@@ -179,6 +180,7 @@ struct AppState {
     physical_plan_lock: Arc<tokio::sync::Mutex<()>>,
     active_physical_plan: Option<crate::storage_engines::types::HotReloadActivePhysicalPlan>,
     physical_plan_lifecycle: Option<crate::storage_engines::types::PhysicalPlanLifecycle>,
+    remote_write: Option<crate::drivers::ingest::PrometheusRemoteWriteReceiver>,
 }
 
 impl HttpServer {
@@ -206,9 +208,19 @@ impl HttpServer {
             physical_plan_lock: Arc::new(tokio::sync::Mutex::new(())),
             active_physical_plan: None,
             physical_plan_lifecycle: None,
+            remote_write: None,
         }
     }
 
+    /// Enable Prometheus Remote Write v1 on the same public HTTP listener.
+    pub fn with_remote_write(
+        mut self,
+        receiver: crate::drivers::ingest::PrometheusRemoteWriteReceiver,
+    ) -> Self {
+        self.remote_write = Some(receiver);
+        self
+    }
+
     /// Plug an additional [`QueryEngine`] into the capability router.
     /// Used by the binary to register `GorillaQueryEngine` (cold
     /// archive) alongside the `ASAPQueryEngine` registered by `new`.
@@ -367,6 +379,11 @@ impl HttpServer {
         );
         info!("Runtime info endpoint: {}", runtime_info_path);
 
+        let request_body_limit = self
+            .remote_write
+            .as_ref()
+            .map(|receiver| receiver.config().max_compressed_bytes)
+            .unwrap_or(2 * 1024 * 1024);
         let app_state = AppState {
             config: self.config.clone(),
             query_engine: self.query_engine,
@@ -383,6 +400,7 @@ impl HttpServer {
             physical_plan_lock: self.physical_plan_lock.clone(),
             active_physical_plan: self.active_physical_plan.clone(),
             physical_plan_lifecycle: self.physical_plan_lifecycle.clone(),
+            remote_write: self.remote_write.clone(),
         };
 
         let range_query_endpoint = adapter.get_range_query_endpoint();
@@ -396,6 +414,11 @@ impl HttpServer {
             .route(runtime_info_path, post(handle_runtime_info))
             .route("/metrics", get(handle_metrics))
             .route("/api/v1/health", get(handle_health))
+            .route(
+                "/api/v1/write",
+                post(handle_prometheus_remote_write)
+                    .layer(DefaultBodyLimit::max(request_body_limit)),
+            )
             .route("/api/v1/store/metrics", get(handle_store_metrics))
             .route(
                 "/api/v1/streaming-config",
@@ -464,6 +487,11 @@ impl HttpServer {
         let query_endpoint = adapter.get_query_endpoint();
         let runtime_info_path = adapter.get_runtime_info_path();
 
+        let request_body_limit = self
+            .remote_write
+            .as_ref()
+            .map(|receiver| receiver.config().max_compressed_bytes)
+            .unwrap_or(2 * 1024 * 1024);
         let app_state = AppState {
             config: self.config.clone(),
             query_engine: self.query_engine.clone(),
@@ -480,6 +508,7 @@ impl HttpServer {
             physical_plan_lock: self.physical_plan_lock.clone(),
             active_physical_plan: self.active_physical_plan.clone(),
             physical_plan_lifecycle: self.physical_plan_lifecycle.clone(),
+            remote_write: self.remote_write.clone(),
         };
 
         let range_query_endpoint = adapter.get_range_query_endpoint();
@@ -490,6 +519,11 @@ impl HttpServer {
             .route(range_query_endpoint, get(handle_range_query))
             .route(range_query_endpoint, post(handle_range_query_post))
             .route(runtime_info_path, get(handle_runtime_info))
+            .route(
+                "/api/v1/write",
+                post(handle_prometheus_remote_write)
+                    .layer(DefaultBodyLimit::max(request_body_limit)),
+            )
             .route(
                 "/api/v1/streaming-config",
                 get(handle_get_streaming_config).post(handle_post_streaming_config),
@@ -1746,12 +1780,37 @@ fn query_status_label(response: &Response) -> &'static str {
 // Metrics Handler
 // ============================================================
 
-async fn handle_metrics() -> impl IntoResponse {
+async fn handle_metrics(State(state): State<AppState>) -> impl IntoResponse {
     let encoder = prometheus::TextEncoder::new();
     let metric_families = prometheus::gather();
     let mut buffer = Vec::new();
     prometheus::Encoder::encode(&encoder, &metric_families, &mut buffer)
         .unwrap_or_else(|e| tracing::error!("Failed to encode metrics: {}", e));
+    if let Some(receiver) = state.remote_write.as_ref() {
+        use std::sync::atomic::Ordering;
+        let stats = receiver.stats();
+        let extra = format!(
+            "# TYPE asap_remote_write_requests_total counter\n\
+             asap_remote_write_requests_total {}\n\
+             # TYPE asap_remote_write_samples_total counter\n\
+             asap_remote_write_samples_total {}\n\
+             # TYPE asap_remote_write_stale_markers_total counter\n\
+             asap_remote_write_stale_markers_total {}\n\
+             # TYPE asap_remote_write_duplicates_total counter\n\
+             asap_remote_write_duplicates_total {}\n\
+             # TYPE asap_remote_write_rejected_requests_total counter\n\
+             asap_remote_write_rejected_requests_total {}\n\
+             # TYPE asap_remote_write_bytes_total counter\n\
+             asap_remote_write_bytes_total {}\n",
+            stats.requests.load(Ordering::Relaxed),
+            stats.samples.load(Ordering::Relaxed),
+            stats.stale_markers.load(Ordering::Relaxed),
+            stats.duplicates.load(Ordering::Relaxed),
+            stats.rejected_requests.load(Ordering::Relaxed),
+            stats.bytes.load(Ordering::Relaxed),
+        );
+        buffer.extend_from_slice(extra.as_bytes());
+    }
     (
         [(
             axum::http::header::CONTENT_TYPE,
@@ -2077,10 +2136,19 @@ async fn handle_range_query_post(State(state): State<AppState>, body: Bytes) ->
 #[cfg(test)]
 mod tests {
     use super::*;
+    use crate::drivers::ingest::prometheus_remote_write::{
+        Label, PrometheusRemoteWriteConfig, PrometheusRemoteWriteReceiver, Sample, TimeSeries,
+        WriteRequest,
+    };
+    use crate::precompute_engine::ingest_handler::{IngestObservability, IngestState};
+    use crate::precompute_engine::series_router::SeriesRouter;
     use crate::query_engines::ASAPQueryEngine;
     use crate::storage_engines::types::{HotReloadStreamingConfig, StreamingConfig};
+    use prost::Message;
     use reqwest::Client;
+    use std::sync::atomic::AtomicU64;
     use std::sync::Arc;
+    use tokio::sync::mpsc;
 
     // ── warm-vs-archive range routing: tier classification ──────────────
     // The defect is that a recent (warm-resident, not-yet-archived) range
@@ -2154,6 +2222,90 @@ mod tests {
         setup_test_server_with_hot_reload(None).await
     }
 
+    async fn setup_remote_write_test_server() -> (u16, PrometheusRemoteWriteReceiver) {
+        let streaming_config = Arc::new(StreamingConfig::default());
+        let hot_reload = HotReloadStreamingConfig::new((*streaming_config).clone());
+        let (sender, _worker) = mpsc::channel(8);
+        let ingest = Arc::new(IngestState {
+            router: SeriesRouter::new(vec![sender]),
+            samples_ingested: AtomicU64::new(0),
+            samples_blocked_by_schema_barrier: AtomicU64::new(0),
+            hot_reload_config: hot_reload,
+            pass_raw_samples: false,
+            sketch_snapshots: dashmap::DashMap::new(),
+            series_resolver: Arc::new(crate::drivers::ingest::SeriesIdResolver::new()),
+            sketch_index: Arc::new(crate::storage_engines::sketch_db::index::SketchStore::new()),
+            observability: IngestObservability::default(),
+        });
+        let receiver =
+            PrometheusRemoteWriteReceiver::new(PrometheusRemoteWriteConfig::default(), ingest);
+        let adapter_config = AdapterConfig::prometheus_promql("http://127.0.0.1:9".into(), false);
+        let server = HttpServer::new(
+            HttpServerConfig {
+                port: 0,
+                handle_http_requests: true,
+                adapter_config,
+            },
+            Arc::new(ASAPQueryEngine::new(streaming_config, 15_000)),
+            Arc::new(crate::storage_engines::sketch_db::index::SketchStore::new()),
+        )
+        .with_remote_write(receiver.clone());
+        (server.start_test_server().await.unwrap(), receiver)
+    }
+
+    #[tokio::test]
+    async fn remote_write_http_contract_accepts_v1_and_rejects_wrong_encoding() {
+        let (port, receiver) = setup_remote_write_test_server().await;
+        let request = WriteRequest {
+            timeseries: vec![TimeSeries {
+                labels: vec![Label {
+                    name: "__name__".into(),
+                    value: "up".into(),
+                }],
+                samples: vec![Sample {
+                    value: 1.0,
+                    timestamp: 123,
+                }],
+                exemplars: Vec::new(),
+                histograms: Vec::new(),
+            }],
+        };
+        let body = snap::raw::Encoder::new()
+            .compress_vec(&request.encode_to_vec())
+            .unwrap();
+        let client = Client::new();
+        let accepted = client
+            .post(format!("http://127.0.0.1:{port}/api/v1/write"))
+            .header("content-encoding", "snappy")
+            .header("content-type", "application/x-protobuf")
+            .header("x-prometheus-remote-write-version", "0.1.0")
+            .body(body.clone())
+            .send()
+            .await
+            .unwrap();
+        assert_eq!(accepted.status().as_u16(), StatusCode::NO_CONTENT.as_u16());
+        assert_eq!(
+            receiver
+                .stats()
+                .samples
+                .load(std::sync::atomic::Ordering::Relaxed),
+            1
+        );
+
+        let rejected = client
+            .post(format!("http://127.0.0.1:{port}/api/v1/write"))
+            .header("content-encoding", "gzip")
+            .header("content-type", "application/x-protobuf")
+            .body(body)
+            .send()
+            .await
+            .unwrap();
+        assert_eq!(
+            rejected.status().as_u16(),
+            StatusCode::UNSUPPORTED_MEDIA_TYPE.as_u16()
+        );
+    }
+
     async fn setup_test_server_with_hot_reload(
         hot_reload: Option<HotReloadStreamingConfig>,
     ) -> u16 {
@@ -5083,6 +5235,80 @@ aggregations:
 }
 
 /// Health check endpoint for DataCollector controller to verify backend is alive.
+async fn handle_prometheus_remote_write(
+    State(state): State<AppState>,
+    headers: HeaderMap,
+    body: Bytes,
+) -> Response {
+    let Some(receiver) = state.remote_write.as_ref() else {
+        return (StatusCode::NOT_FOUND, "Remote Write is disabled").into_response();
+    };
+    let content_encoding = headers
+        .get(axum::http::header::CONTENT_ENCODING)
+        .and_then(|value| value.to_str().ok());
+    if !content_encoding.is_some_and(|value| value.eq_ignore_ascii_case("snappy")) {
+        return (
+            StatusCode::UNSUPPORTED_MEDIA_TYPE,
+            "Content-Encoding must be snappy",
+        )
+            .into_response();
+    }
+    if let Some(content_type) = headers
+        .get(axum::http::header::CONTENT_TYPE)
+        .and_then(|value| value.to_str().ok())
+    {
+        if !content_type
+            .split(';')
+            .next()
+            .is_some_and(|value| value.trim().eq_ignore_ascii_case("application/x-protobuf"))
+        {
+            return (
+                StatusCode::UNSUPPORTED_MEDIA_TYPE,
+                "Content-Type must be application/x-protobuf",
+            )
+                .into_response();
+        }
+    }
+    if let Some(version) = headers
+        .get("x-prometheus-remote-write-version")
+        .and_then(|value| value.to_str().ok())
+    {
+        if version != "0.1.0" {
+            return (
+                StatusCode::BAD_REQUEST,
+                "only Prometheus Remote Write v1 (0.1.0) is supported",
+            )
+                .into_response();
+        }
+    }
+
+    match receiver.accept(&body) {
+        Ok(()) => StatusCode::NO_CONTENT.into_response(),
+        Err(
+            crate::drivers::ingest::prometheus_remote_write::RemoteWriteError::CompressedTooLarge(
+                _,
+            )
+            | crate::drivers::ingest::prometheus_remote_write::RemoteWriteError::DecompressedTooLarge(
+                _,
+            )
+            | crate::drivers::ingest::prometheus_remote_write::RemoteWriteError::TooManySeries {
+                ..
+            }
+            | crate::drivers::ingest::prometheus_remote_write::RemoteWriteError::TooManySamples(
+                _,
+            ),
+        ) => (StatusCode::PAYLOAD_TOO_LARGE, "Remote Write request exceeds configured limits")
+            .into_response(),
+        Err(
+            error @ (crate::drivers::ingest::prometheus_remote_write::RemoteWriteError::Backpressure(
+                _,
+            )
+            | crate::drivers::ingest::prometheus_remote_write::RemoteWriteError::DedupCapacity(_)),
+        ) => (StatusCode::SERVICE_UNAVAILABLE, error.to_string()).into_response(),
+        Err(error) => (StatusCode::BAD_REQUEST, error.to_string()).into_response(),
+    }
+}
+
 async fn handle_health() -> &'static str {
     "ok"
 }
diff --git a/data_plane/src/lib.rs b/data_plane/src/lib.rs
index eba395d12..2c18f1b4a 100644
--- a/data_plane/src/lib.rs
+++ b/data_plane/src/lib.rs
@@ -24,6 +24,9 @@ pub use storage_engines::StoreResult;
 
 pub use query_engines::{ASAPQueryEngine, InstantVector, QueryResult};
 
+pub use drivers::ingest::{
+    PrometheusRemoteWriteConfig, PrometheusRemoteWriteReceiver, RemoteWriteStats,
+};
 pub use drivers::{HttpServer, HttpServerConfig, OtlpReceiver, OtlpReceiverConfig};
 
 pub use precompute_engine::config::{LateDataPolicy, PrecomputeEngineConfig};
diff --git a/data_plane/src/main.rs b/data_plane/src/main.rs
index cee6385c8..e534b7dc2 100644
--- a/data_plane/src/main.rs
+++ b/data_plane/src/main.rs
@@ -10,7 +10,7 @@
 // RuntimeSamples gRPC endpoint (port 4321) will be served from inside
 // this same backend process — there is no longer a separate
 // `asap-controller` container in `mvp-multinode/run_demo.sh`.
-use clap::Parser;
+use clap::{Parser, ValueEnum};
 use std::fs;
 use std::sync::Arc;
 use tokio::signal;
@@ -23,12 +23,25 @@ use data_plane::storage_engines::types::enums::{CleanupPolicy, LockStrategy};
 use data_plane::utils::file_io::read_streaming_config;
 use data_plane::{
     ASAPQueryEngine, HttpServer, HttpServerConfig, OtlpReceiver, OtlpReceiverConfig,
-    PrecomputeEngine, PrecomputeEngineConfig, Result, SketchStoreSink,
+    PrecomputeEngine, PrecomputeEngineConfig, PrometheusRemoteWriteConfig,
+    PrometheusRemoteWriteReceiver, Result, SketchStoreSink,
 };
 
+#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, ValueEnum)]
+enum RuntimeProfile {
+    #[default]
+    Distributed,
+    Asapquery,
+}
+
 #[derive(Parser, Debug)]
 #[command(author, version, about, long_about = None)]
 struct Args {
+    /// Runtime component profile. `asapquery` enables backend-local raw
+    /// Remote Write precompute and rejects Collector/OTLP-only components.
+    #[arg(long, value_enum, default_value = "distributed")]
+    profile: RuntimeProfile,
+
     /// File path for streaming_config
     #[arg(long)]
     streaming_config: String,
@@ -118,6 +131,39 @@ struct Args {
     #[arg(long)]
     enable_otel_ingest: bool,
 
+    /// Enable Prometheus Remote Write v1 at POST /api/v1/write. Automatically
+    /// enabled by `--profile asapquery`.
+    #[arg(long)]
+    enable_remote_write: bool,
+
+    /// Maximum compressed Remote Write request size.
+    #[arg(long, default_value = "33554432")]
+    remote_write_max_compressed_bytes: usize,
+
+    /// Maximum Snappy-decompressed Remote Write request size.
+    #[arg(long, default_value = "134217728")]
+    remote_write_max_decompressed_bytes: usize,
+
+    /// Maximum series in one Remote Write request.
+    #[arg(long, default_value = "100000")]
+    remote_write_max_timeseries: usize,
+
+    /// Maximum samples in one Remote Write request.
+    #[arg(long, default_value = "1000000")]
+    remote_write_max_samples: usize,
+
+    /// Bounded Remote Write idempotency horizon.
+    #[arg(long, default_value = "600000")]
+    remote_write_dedup_horizon_ms: u64,
+
+    /// Expected maximum interval over which Prometheus can retry a request.
+    #[arg(long, default_value = "60000")]
+    remote_write_expected_retry_interval_ms: u64,
+
+    /// Maximum retained (series,timestamp) idempotency keys.
+    #[arg(long, default_value = "2000000")]
+    remote_write_max_dedup_entries: usize,
+
     /// OTLP gRPC listen port
     #[arg(long, default_value = "4317")]
     otel_grpc_port: u16,
@@ -282,10 +328,92 @@ struct Args {
     backend_storage_routing: Option<std::path::PathBuf>,
 }
 
+fn validate_profile(args: &Args) -> Result<()> {
+    if args.profile != RuntimeProfile::Asapquery {
+        return Ok(());
+    }
+    let mut excluded = Vec::new();
+    if args.enable_otel_ingest {
+        excluded.push("--enable-otel-ingest");
+    }
+    if args.enable_monitor_coordinator {
+        excluded.push("--enable-monitor-coordinator");
+    }
+    if args.enable_backfill_worker {
+        excluded.push("--enable-backfill-worker");
+    }
+    if args.enable_schema_eviction {
+        excluded.push("--enable-schema-eviction");
+    }
+    if args.persistence_enabled {
+        excluded.push("--persistence-enabled");
+    }
+    if args.backend_storage_routing.is_some() {
+        excluded.push("--backend-storage-routing");
+    }
+    if args.control_plane_endpoint.is_some() {
+        excluded.push("--control-plane-endpoint");
+    }
+    if !excluded.is_empty() {
+        return Err(format!(
+            "--profile asapquery excludes distributed/durable components: {}",
+            excluded.join(", ")
+        )
+        .into());
+    }
+    if !args.forward_unsupported_queries {
+        return Err(
+            "--profile asapquery requires --forward-unsupported-queries for exact fallback".into(),
+        );
+    }
+    let required_horizon = (args.precompute_allowed_lateness_ms.max(0) as u64)
+        .saturating_add(args.remote_write_expected_retry_interval_ms);
+    if args.remote_write_dedup_horizon_ms < required_horizon {
+        return Err(format!(
+            "--remote-write-dedup-horizon-ms must cover allowed lateness plus the expected retry interval (at least {required_horizon}ms)"
+        )
+        .into());
+    }
+    if args.remote_write_max_compressed_bytes == 0
+        || args.remote_write_max_decompressed_bytes == 0
+        || args.remote_write_max_timeseries == 0
+        || args.remote_write_max_samples == 0
+        || args.remote_write_max_dedup_entries == 0
+    {
+        return Err("Remote Write resource limits must all be greater than zero".into());
+    }
+    if args.remote_write_max_decompressed_bytes < args.remote_write_max_compressed_bytes {
+        return Err(
+            "--remote-write-max-decompressed-bytes must be at least the compressed limit".into(),
+        );
+    }
+    Ok(())
+}
+
+async fn verify_prometheus_fallback(base_url: &str) -> Result<()> {
+    let url = format!("{}/-/healthy", base_url.trim_end_matches('/'));
+    let response = reqwest::Client::new()
+        .get(&url)
+        .timeout(std::time::Duration::from_secs(5))
+        .send()
+        .await
+        .map_err(|error| format!("Prometheus fallback health check {url} failed: {error}"))?;
+    if !response.status().is_success() {
+        return Err(format!(
+            "Prometheus fallback health check {url} returned {}",
+            response.status()
+        )
+        .into());
+    }
+    Ok(())
+}
+
 #[tokio::main]
 async fn main() -> Result<()> {
     let args = Args::parse();
 
+    validate_profile(&args)?;
+
     // Create output directory
     fs::create_dir_all(&args.output_dir)?;
 
@@ -296,11 +424,15 @@ async fn main() -> Result<()> {
     info!("Starting Query Engine Rust");
     info!("Output directory: {}", args.output_dir);
 
+    if args.profile == RuntimeProfile::Asapquery {
+        verify_prometheus_fallback(&args.prometheus_server).await?;
+        info!("ASAPQuery compatibility profile: Prometheus fallback is healthy");
+    }
+
     if let Some(ingest_port) = args.ingest_port {
         warn!(
-            "--ingest-port={ingest_port} is deprecated and ignored: the backend's only HTTP \
-             listener is the PromQL query surface (PRW ingest was removed in PR #100). \
-             Drop the flag from your compose `command:` block."
+            "--ingest-port={ingest_port} is deprecated and ignored: Remote Write and PromQL \
+             share --http-port. Drop the flag from your compose `command:` block."
         );
     }
 
@@ -564,7 +696,7 @@ async fn main() -> Result<()> {
         );
         let worker_diagnostics = engine.diagnostics();
         let ingest_state = engine.ingest_state();
-        info!("Starting precompute engine (OTLP-fed; no HTTP ingest port)");
+        info!("Starting precompute engine (ingest adapters share its bounded worker queues)");
 
         // Spawn periodic memory diagnostics logger — M2.3.6g routes
         // through SketchStore now that SketchStore no longer holds
@@ -787,6 +919,24 @@ async fn main() -> Result<()> {
         .with_active_physical_plan(active_physical_plan.clone())
         .with_probe_cache(probe_cache.clone());
 
+    if args.enable_remote_write || args.profile == RuntimeProfile::Asapquery {
+        let receiver = PrometheusRemoteWriteReceiver::new(
+            PrometheusRemoteWriteConfig {
+                max_compressed_bytes: args.remote_write_max_compressed_bytes,
+                max_decompressed_bytes: args.remote_write_max_decompressed_bytes,
+                max_timeseries: args.remote_write_max_timeseries,
+                max_samples: args.remote_write_max_samples,
+                dedup_horizon: std::time::Duration::from_millis(args.remote_write_dedup_horizon_ms),
+                max_dedup_entries: args.remote_write_max_dedup_entries,
+            },
+            precompute_ingest_state
+                .clone()
+                .expect("precompute ingest state is always constructed"),
+        );
+        info!("Prometheus Remote Write v1 enabled at POST /api/v1/write");
+        server = server.with_remote_write(receiver);
+    }
+
     // Per-metric storage-backend routing table (issue #46
     // criterion ⑤). Mirror the `precompute_engine` binary: load it
     // from `--backend-storage-routing` (or its env-var alias) so the
diff --git a/data_plane/src/precompute_engine/series_router.rs b/data_plane/src/precompute_engine/series_router.rs
index 85a63d0b2..f66b20738 100644
--- a/data_plane/src/precompute_engine/series_router.rs
+++ b/data_plane/src/precompute_engine/series_router.rs
@@ -190,6 +190,38 @@ impl SeriesRouter {
         Ok(())
     }
 
+    /// Atomically reserve queue capacity for an entire request and then
+    /// publish it. If any target worker is full or closed, every reservation
+    /// is dropped and no message is enqueued. Remote Write uses this to turn
+    /// bounded-queue pressure into a retryable HTTP response without leaving
+    /// an untracked partial request behind.
+    pub fn try_route_group_batch_atomic(
+        &self,
+        messages: Vec<WorkerMessage>,
+    ) -> Result<(), TryRouteError> {
+        let mut pending = Vec::with_capacity(messages.len());
+        for message in messages {
+            let worker_idx = match &message {
+                WorkerMessage::GroupSamples { sid, .. }
+                | WorkerMessage::AccumulatorInput { sid, .. } => self.worker_for_sid(*sid),
+                WorkerMessage::RawSamples { series_key, .. } => self.worker_for(series_key),
+                WorkerMessage::Flush | WorkerMessage::Shutdown => 0,
+            };
+            let permit = self.senders[worker_idx]
+                .clone()
+                .try_reserve_owned()
+                .map_err(|error| match error {
+                    mpsc::error::TrySendError::Full(_) => TryRouteError::Full,
+                    mpsc::error::TrySendError::Closed(_) => TryRouteError::Closed,
+                })?;
+            pending.push((permit, message));
+        }
+        for (permit, message) in pending {
+            permit.send(message);
+        }
+        Ok(())
+    }
+
     /// Broadcast a flush signal to all workers.
     pub async fn broadcast_flush(&self) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
         for (i, sender) in self.senders.iter().enumerate() {
@@ -229,6 +261,14 @@ impl SeriesRouter {
     }
 }
 
+#[derive(Debug, Clone, Copy, PartialEq, Eq, thiserror::Error)]
+pub enum TryRouteError {
+    #[error("precompute queue is full")]
+    Full,
+    #[error("precompute worker is unavailable")]
+    Closed,
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -264,4 +304,30 @@ mod tests {
         assert_eq!(w1, w2);
         assert!(router.worker_for("mem{host=\"a\"}") < 4);
     }
+
+    #[tokio::test]
+    async fn atomic_route_publishes_nothing_when_batch_exceeds_capacity() {
+        let (sender, mut receiver) = mpsc::channel::<WorkerMessage>(1);
+        let router = SeriesRouter::new(vec![sender]);
+        let messages = vec![
+            WorkerMessage::RawSamples {
+                series_key: "m{job=\"a\"}".into(),
+                samples: vec![(1, 1.0)],
+                ingest_received_at: Instant::now(),
+            },
+            WorkerMessage::RawSamples {
+                series_key: "m{job=\"b\"}".into(),
+                samples: vec![(1, 2.0)],
+                ingest_received_at: Instant::now(),
+            },
+        ];
+        assert_eq!(
+            router.try_route_group_batch_atomic(messages),
+            Err(TryRouteError::Full)
+        );
+        assert!(matches!(
+            receiver.try_recv(),
+            Err(mpsc::error::TryRecvError::Empty)
+        ));
+    }
 }
diff --git a/docs/README.md b/docs/README.md
index a026ff718..94c474108 100644
--- a/docs/README.md
+++ b/docs/README.md
@@ -23,6 +23,7 @@ backend operation material.
 ## Backend implementation
 
 - [ASAPQuery compatibility profile](design_docs/asapquery-compatibility-profile.md)
+  - [Run the compatibility profile](user_guide/asapquery-profile.md)
 - [Developer guide](developer_docs/README.md)
 - [Summary store engine](developer_docs/summary-store-engine/implementation.md)
 - [Query engine](developer_docs/query-engine/query-engine.md)
diff --git a/docs/user_guide/asapquery-profile.md b/docs/user_guide/asapquery-profile.md
new file mode 100644
index 000000000..e8adf5076
--- /dev/null
+++ b/docs/user_guide/asapquery-profile.md
@@ -0,0 +1,59 @@
+# Run the ASAPQuery compatibility profile
+
+The `asapquery` runtime profile runs without ASAPCollector. It accepts
+Prometheus Remote Write v1 and sends unsupported or not-yet-ready PromQL to the
+same Prometheus server as an exact fallback.
+
+Start Prometheus first, configure it to write to the backend, and keep its
+normal local storage enabled:
+
+```yaml
+remote_write:
+  - url: http://127.0.0.1:9091/api/v1/write
+```
+
+Then start the backend:
+
+```bash
+cargo run -p data_plane -- \
+  --profile asapquery \
+  --streaming-config data_plane/examples/asapquery/streaming-config.yaml \
+  --prometheus-server http://127.0.0.1:9090 \
+  --forward-unsupported-queries \
+  --http-port 9091 \
+  --output-dir /tmp/asapquery-backend
+```
+
+The profile checks Prometheus's `/-/healthy` endpoint before opening its public
+listener. It rejects OTLP ingest, monitor coordination, persistence, backfill,
+schema eviction, and archive routing flags. Remote Write and PromQL share the
+backend listener:
+
+- `POST /api/v1/write`
+- `GET` or `POST /api/v1/query`
+- `GET` or `POST /api/v1/query_range`
+
+Prometheus should use the standard Remote Write v1 headers
+`Content-Encoding: snappy`, `Content-Type: application/x-protobuf`, and
+`X-Prometheus-Remote-Write-Version: 0.1.0`. The receiver validates the entire
+batch before enqueueing it. Invalid batches return `400` or `413`; exhausted
+deduplication or worker-queue capacity returns retryable `503`; accepted batches
+return `204`.
+
+The deduplication horizon must cover allowed lateness plus the deployment's
+maximum expected Prometheus retry interval. Configure those assumptions with
+`--precompute-allowed-lateness-ms`,
+`--remote-write-expected-retry-interval-ms`, and
+`--remote-write-dedup-horizon-ms`. Startup fails when the horizon is too short.
+
+Receiver evidence is exported from `/metrics` as
+`asap_remote_write_requests_total`, `asap_remote_write_samples_total`,
+`asap_remote_write_stale_markers_total`,
+`asap_remote_write_duplicates_total`,
+`asap_remote_write_rejected_requests_total`, and
+`asap_remote_write_bytes_total`.
+
+The example streaming config is a checked-in precompute-plan fixture. Startup
+loading of canonical `QueryWorkload` and `DataWorkload` snapshots and automatic
+backend-only compilation remain required before the profile meets the complete
+compatibility contract in the design document.