arrow: add Prometheus series staleness detection and StalenessTracker

benches/decode.rs

@@ -27,7 +27,7 @@ use std::pin::{Pin, pin};
 use std::task::{Context, Poll, Waker};
 
 use benchfmt::{CountingAlloc, Suite};
-use expfmt::arrow::{ArrowDecoder, DecodeBudget, DecodeOptions, DecodeStep};
+use expfmt::arrow::{ArrowDecoder, DecodeBudget, DecodeOptions, DecodeStep, StalenessTracker};
 use expfmt::format::Format;
 
 #[global_allocator]
@@ -216,6 +216,32 @@ fn main() {
                         }
                     }
                 },
+            )
+            .bench(
+                format!("Decode/case={case}/mode=decode_staleness"),
+                move |b| {
+                    let mut decoder = ArrowDecoder::new();
+                    let mut tracker = StalenessTracker::new();
+                    black_box(
+                        decoder
+                            .decode_with_staleness(payload, &Format::TEXT, &opts, &mut tracker)
+                            .unwrap(),
+                    );
+
+                    b.set_bytes(bytes);
+                    b.report_allocs();
+                    while b.keep_running() {
+                        let records = decoder
+                            .decode_with_staleness(
+                                black_box(payload),
+                                &Format::TEXT,
+                                &opts,
+                                &mut tracker,
+                            )
+                            .unwrap();
+                        black_box(records);
+                    }
+                },
             );
     }
 

specs/SPEC.md

@@ -128,7 +128,8 @@ scrape bytes
   1. **Stage 1 (`__name__` series filter pushdown)**: Leading `keep`, `drop`, `keepequal`, and `dropequal` rules that inspect only `__name__` will run once per contiguous wire series-name transition (e.g. `foo_bucket`, `foo_sum`, `foo_count`, matching Prometheus `scrape/scrape.go`) so dropped series runs skip float parsing, label unescaping, string interning, and family assembly.
   2. **Stage 2 (dictionary-vectorized relabeling and `honor_labels`)**: Once kept series have been assembled into metric families (collapsing classic histogram `_bucket`, `_sum`, and `_count` lines into a single `HistogramDataPoints` row) and their label keys and values have been deduplicated in `key_interner` and `val_interner`, attribute-level `honor_labels` collision resolution, `labeldrop` / `labelkeep`, and single-label `keep` / `drop` / value transformations (`replace`, `lowercase`, `uppercase`, `hashmod`) will execute across unique dictionary entries (memoized per `source_labels[0]` value ID) instead of $O(N_{\text{samples}})$.
 - The metric family assembler groups classic histogram `_bucket`, `_sum`, and `_count` lines and summary `quantile`, `_sum`, and `_count` lines (plus OpenMetrics `_created` start timestamps) into complete metric families.
-- Configurable translation policies control counter `_total` and unit suffix handling and untyped series handling.
+- Configurable translation policies control counter `_total` and unit suffix handling, untyped series handling, and explicit-timestamp staleness tracking (`track_timestamps_staleness`, `false` by default to match Prometheus `scrape/scrape.go`).
+- Optional per-target series staleness tracking via `StalenessTracker` (`ArrowDecoder::decode_with_staleness`, `decode_stream_with_staleness`, `step_decode_with_staleness`, `finish_decode_with_staleness`, and `flush_target_stale`) detects disappeared time series across scrapes as well as incoming `STALE_NAN_BITS` (`0x7ff0_0000_0000_0002`) markers. Stale points set `DATA_POINT_FLAG_NO_RECORDED_VALUE` (`1`) in a lazily materialized `flags` (`UInt32`) column and omit `double_value` and `histogram_sum` (`NULL` in Arrow, `0.0` for `summary_sum`), leaving non-stale batch schemas unchanged. `StalenessTracker` interns active series labels in contiguous arenas with zero steady-state heap allocations, rolls back uncommitted changes on parse errors or dropped streams, and compacts deactivated series under label churn so per-target live heap stays proportional to active series. Callers that do not pass a `StalenessTracker` (`decode`, `decode_stream`, `step_decode`) pay zero cross-scrape memory overhead.
 - `ArrowDecoder` owns its scratch buffers and Arrow builders without internal locks or global state, so a per-core `!Send` receiver can reuse allocations across scrapes with bounded memory.
 
 ### Target and metric relabeling architecture