feat: record Fanout's background work in resource samples

README.md

@@ -129,11 +129,13 @@ Every run produces one self-contained directory:
 results/20261014T190000Z/
 ├── report.json     authoritative schema-versioned result
 ├── report.md       readable, publication-ready report
-├── samples.jsonl   per-second Fanout resource samples
+├── samples.jsonl   per-second Fanout resource and background-work samples
 ├── fanout-repair.txt  post-run authoritative-storage verification
 └── fanout.log      setup-token-scrubbed server log
 ```
 
+`samples.jsonl` includes cumulative `rollup_seconds` by rollup (`service`, `edge`, `version`, `read_cache`), `write_gate_hold_seconds` and `write_gate_wait_seconds` by operation, `compaction_seconds_total`, `compactions_total`, `publish_wait_seconds_total`, and `ingest_backpressure_seconds_total`, plus the current `compaction_backlog_batches` gauge. Missing metrics from older Fanout releases leave these fields absent or zero. See [sample definitions](docs/metrics.md#background-work-samples) for their sources and interpretation.
+
 The terminal output is optimized for decisions:
 
 ```text
@@ -184,7 +186,7 @@ For every phase the report includes:
 - generator process count/failures and Fanout dropped-row/restart checks;
 - average Fanout CPU cores, peak/end RSS, peak/end Go heap, allocations, GC pause;
 - peak ingest queue depth, Parquet file count/bytes, and optional local data-directory growth;
-- a raw time series in `samples.jsonl` for independent charting;
+- a raw time series in `samples.jsonl` for independent charting, including rollup durations, write-gate hold/wait durations, compaction duration/count/backlog, publish wait duration, and ingest backpressure duration;
 - attempted/completed read rate, latency, failures, scheduler shedding, and phase-end cancellations when query load is enabled.
 
 See [docs/metrics.md](docs/metrics.md) for definitions and [docs/publishing.md](docs/publishing.md) for the public-results checklist.

docs/metrics.md

@@ -32,6 +32,23 @@ CPU is the Fanout process only. On a same-host run it does not include telemetry
 
 The sampled series above, plus `fanout_rows_dropped_total`, `process_start_time_seconds`, `fanout_parquet_publish_waiters`, and `fanout_parquet_publish_timeouts_total`, come from Fanout v2026.10.1-alpha and its registered Go/process collectors.
 
+## Background work samples
+
+Each `samples.jsonl` row also records the following `/metrics` values. The same fields appear in `report.json` under `trials[].samples`.
+
+| Sample field | Source |
+|---|---|
+| `rollup_seconds` | `fanout_rollup_component_duration_seconds_sum`, keyed by `rollup` (`service`, `edge`, `version`, `read_cache`) |
+| `write_gate_hold_seconds` | `fanout_write_gate_hold_seconds_sum`, keyed by `operation` |
+| `write_gate_wait_seconds` | `fanout_write_gate_wait_seconds_sum`, keyed by `operation` |
+| `compaction_seconds_total` | `fanout_telemetry_operation_duration_seconds_sum` for `operation="compaction"` |
+| `compactions_total` | `fanout_telemetry_operation_duration_seconds_count` for `operation="compaction"` |
+| `publish_wait_seconds_total` | `fanout_parquet_publish_wait_seconds_sum` |
+| `ingest_backpressure_seconds_total` | `fanout_ingest_backpressure_seconds_total` |
+| `compaction_backlog_batches` | `fanout_compaction_backlog_batches` |
+
+Durations and counts are cumulative values at scrape time, not phase deltas. Subtract adjacent samples to attribute work during an interval, accounting for process restarts. `compaction_backlog_batches` is a current gauge rather than a cumulative counter. Series with additional labels are summed, retaining the indicated map key for rollups and write gates. Older Fanout releases may omit these metrics; absent maps and zero scalar values are omitted from JSON, and missing metrics do not fail a run. These samples provide attribution evidence without changing headline tables or quality gates.
+
 ## Generator resources
 
 Split trials read `/proc/stat` on the generator before and after each phase using the same SSH target, port, identity, known-hosts file, and connection options as telemetrygen. The interval includes startup and final exports. `report.json` records `trials[].generator_cpu.average_cpu_cores` and `logical_cpus`; `report.md` shows both in each phase row. Same-host and older reports omit this measurement and render it as `n/a`.

docs/publishing.md

@@ -10,7 +10,7 @@ A performance number is useful only when another engineer can explain and reprod
 - [ ] Fanout and generator Hetzner host descriptions, Fanout CPU model, architecture, logical CPU count, memory, OS, and generator placement;
 - [ ] transport, authentication path, batch size, topology, rate, duration, and repetitions;
 - [ ] average generator-host CPU cores and logical CPUs for each split trial, plus the selected sustained process count per signal;
-- [ ] raw `report.json`, `report.md`, and `samples.jsonl` from the same run;
+- [ ] raw `report.json`, `report.md`, and `samples.jsonl` from the same run, retaining the [background-work fields](metrics.md#background-work-samples) for rollups, write gates, compaction, publish waits, and ingest backpressure when available;
 - [ ] three publish-preset repetitions with Fanout alone on a fresh default `ccx33` VM and generators on a separate `ccx33`;
 - [ ] signal-specific peak results and a sustained mixed result;
 - [ ] a separate read-under-write result when claiming interactive performance;

docs/telemetrygen.md

@@ -46,4 +46,4 @@ Let `B` be batch size, `W` workers, `r` requested rate per worker, and `L` time
 - **Traces:** generation and export overlap through the queue. Long-run throughput is bounded approximately by `min(actual generation rate, B / L)`. Larger `W` can fill the queue faster without increasing export concurrency. Longer acknowledgements lower throughput and can cause queue drops. At 1,000 spans/batch, 13,000 spans/s would correspond to roughly 77 ms per serial export **if export were the active bottleneck**.
 - **Logs/metrics:** each worker alternates filling a batch and exporting it. Ignoring startup/final partial batches, per-process throughput is approximately `W × B / (actual batch-fill time + L)`. Even with ideal pacing, batch fill is about `B / r`, so the idealized rate is `W × B / (B / r + L)`, rather than a latency-independent `W × r`. Burst-one pacing prevents recovering most tokens missed during export.
 
-The supplied split report's single-process trials accepted approximately 13.5k spans/s, 12.5k records/s and 12.5k points/s. Similar rates across these different export paths cannot be attributed solely to the serialized trace processor. Neither that report nor `samples.jsonl` measures generator pacing or RPC acknowledgement latency, so the exact contribution of each mechanism is unresolved. The wider process sweep and generator CPU measurements provide the next attribution evidence without changing the workload shape.
+The supplied split report's single-process trials accepted approximately 13.5k spans/s, 12.5k records/s and 12.5k points/s. Similar rates across these different export paths cannot be attributed solely to the serialized trace processor. Neither that report nor `samples.jsonl` measures generator pacing or RPC acknowledgement latency, so the exact contribution of each mechanism is unresolved. New samples include Fanout's [rollup, write-gate, compaction, publish-wait, and ingest-backpressure metrics](metrics.md#background-work-samples) to help attribute server work. The wider process sweep and generator CPU measurements provide the next attribution evidence without changing the workload shape.

internal/bench/report.go

@@ -186,7 +186,7 @@ func Markdown(r Report) string {
 	for _, warning := range r.Summary.Warnings {
 		fmt.Fprintf(&b, "- %s\n", warning)
 	}
-	fmt.Fprintf(&b, "\n## Reproduce\n\n```sh\n%s\n```\n\nRaw evidence: `report.json` (authoritative structured report) and `samples.jsonl` (time-series resource samples).\n", r.ReproductionCommand)
+	fmt.Fprintf(&b, "\n## Reproduce\n\n```sh\n%s\n```\n\nRaw evidence: `report.json` (authoritative structured report) and `samples.jsonl` (time-series resource samples, including cumulative `rollup_seconds`, `write_gate_hold_seconds`, `write_gate_wait_seconds`, `compaction_seconds_total`, `compactions_total`, `publish_wait_seconds_total`, and `ingest_backpressure_seconds_total`, plus the current `compaction_backlog_batches` gauge when available).\n", r.ReproductionCommand)
 	return b.String()
 }
 

internal/bench/runner.go

@@ -404,10 +404,36 @@ func resourceSample(s *snapshot, at float64) ResourceSample {
 		RSSBytes: s.total("process_resident_memory_bytes", nil), HeapBytes: s.total("go_memstats_heap_alloc_bytes", nil),
 		QueueDepth: s.total("fanout_ingest_queue_depth", nil), ParquetFiles: s.total("fanout_parquet_files", nil),
 		ParquetBytes: s.total("fanout_parquet_size_bytes", nil), PublishWaiters: s.total("fanout_parquet_publish_waiters", nil),
-		PublishTimeouts: s.total("fanout_parquet_publish_timeouts_total", nil),
+		PublishTimeouts:                s.total("fanout_parquet_publish_timeouts_total", nil),
+		RollupSeconds:                  sampleTotalsByLabel(s, "fanout_rollup_component_duration_seconds_sum", "rollup"),
+		WriteGateHoldSeconds:           sampleTotalsByLabel(s, "fanout_write_gate_hold_seconds_sum", "operation"),
+		WriteGateWaitSeconds:           sampleTotalsByLabel(s, "fanout_write_gate_wait_seconds_sum", "operation"),
+		CompactionSecondsTotal:         s.total("fanout_telemetry_operation_duration_seconds_sum", map[string]string{"operation": "compaction"}),
+		CompactionsTotal:               s.total("fanout_telemetry_operation_duration_seconds_count", map[string]string{"operation": "compaction"}),
+		PublishWaitSecondsTotal:        s.total("fanout_parquet_publish_wait_seconds_sum", nil),
+		IngestBackpressureSecondsTotal: s.total("fanout_ingest_backpressure_seconds_total", nil),
+		CompactionBacklogBatches:       s.total("fanout_compaction_backlog_batches", nil),
 	}
 }
 
+func sampleTotalsByLabel(s *snapshot, name, label string) map[string]float64 {
+	var totals map[string]float64
+	if s == nil {
+		return nil
+	}
+	for _, sample := range s.samples {
+		key := sample.labels[label]
+		if sample.name != name || key == "" {
+			continue
+		}
+		if totals == nil {
+			totals = make(map[string]float64)
+		}
+		totals[key] += sample.value
+	}
+	return totals
+}
+
 func resourceDelta(before, after *snapshot, samples []ResourceSample, elapsed float64) Resources {
 	result := Resources{
 		EndRSSBytes: after.total("process_resident_memory_bytes", nil), EndHeapBytes: after.total("go_memstats_heap_alloc_bytes", nil),

internal/bench/runner_test.go

@@ -2,17 +2,129 @@ package bench
 
 import (
 	"context"
+	"encoding/json"
 	"errors"
 	"io"
 	"net/http"
 	"os"
+	"reflect"
 	"slices"
 	"strings"
 	"sync/atomic"
 	"testing"
 	"time"
 )
 
+func TestResourceSampleBackgroundMetrics(t *testing.T) {
+	const fixture = `process_cpu_seconds_total 12
+process_resident_memory_bytes 1024
+go_memstats_heap_alloc_bytes 512
+fanout_ingest_queue_depth 3
+fanout_parquet_files 4
+fanout_parquet_size_bytes 2048
+fanout_parquet_publish_waiters 2
+fanout_parquet_publish_timeouts_total 1
+fanout_rollup_component_duration_seconds_sum{rollup="service",signal="spans"} 1.25
+fanout_rollup_component_duration_seconds_sum{rollup="service",signal="logs"} 0.75
+fanout_rollup_component_duration_seconds_sum{rollup="edge"} 2.5
+fanout_rollup_component_duration_seconds_sum{rollup="version"} 3.5
+fanout_rollup_component_duration_seconds_sum{rollup="read_cache"} 4.5
+fanout_rollup_component_duration_seconds_count{rollup="service"} 100
+fanout_write_gate_hold_seconds_sum{operation="ingest",signal="spans"} 5
+fanout_write_gate_hold_seconds_sum{operation="ingest",signal="logs"} 0.5
+fanout_write_gate_hold_seconds_sum{operation="compaction"} 6.5
+fanout_write_gate_wait_seconds_sum{operation="ingest",signal="spans"} 7
+fanout_write_gate_wait_seconds_sum{operation="ingest",signal="logs"} 0.5
+fanout_write_gate_wait_seconds_sum{operation="compaction"} 8.5
+fanout_write_gate_wait_seconds_count{operation="ingest"} 100
+fanout_telemetry_operation_duration_seconds_sum{operation="compaction",signal="spans"} 9
+fanout_telemetry_operation_duration_seconds_sum{operation="compaction",signal="logs"} 0.5
+fanout_telemetry_operation_duration_seconds_sum{operation="ingest"} 100
+fanout_telemetry_operation_duration_seconds_count{operation="compaction",signal="spans"} 10
+fanout_telemetry_operation_duration_seconds_count{operation="compaction",signal="logs"} 2
+fanout_telemetry_operation_duration_seconds_count{operation="ingest"} 100
+fanout_parquet_publish_wait_seconds_sum{signal="spans"} 11
+fanout_parquet_publish_wait_seconds_sum{signal="logs"} 0.5
+fanout_ingest_backpressure_seconds_total 12.5
+fanout_compaction_backlog_batches 13
+`
+	snap, err := parsePrometheus(strings.NewReader(fixture))
+	if err != nil {
+		t.Fatal(err)
+	}
+	got := resourceSample(snap, 1.23456)
+	want := ResourceSample{
+		AtSeconds: 1.235, CPUSeconds: 12, RSSBytes: 1024, HeapBytes: 512,
+		QueueDepth: 3, ParquetFiles: 4, ParquetBytes: 2048, PublishWaiters: 2, PublishTimeouts: 1,
+		RollupSeconds:                  map[string]float64{"service": 2, "edge": 2.5, "version": 3.5, "read_cache": 4.5},
+		WriteGateHoldSeconds:           map[string]float64{"ingest": 5.5, "compaction": 6.5},
+		WriteGateWaitSeconds:           map[string]float64{"ingest": 7.5, "compaction": 8.5},
+		CompactionSecondsTotal:         9.5,
+		CompactionsTotal:               12,
+		PublishWaitSecondsTotal:        11.5,
+		IngestBackpressureSecondsTotal: 12.5,
+		CompactionBacklogBatches:       13,
+	}
+	if !reflect.DeepEqual(got, want) {
+		t.Fatalf("sample=%+v; want %+v", got, want)
+	}
+	// Check the public JSON names as well as the in-memory sample.
+	data, err := json.Marshal(got)
+	if err != nil {
+		t.Fatal(err)
+	}
+	var fields map[string]any
+	if err := json.Unmarshal(data, &fields); err != nil {
+		t.Fatal(err)
+	}
+	for key, want := range map[string]any{
+		"rollup_seconds":                    map[string]any{"service": float64(2), "edge": 2.5, "version": 3.5, "read_cache": 4.5},
+		"write_gate_hold_seconds":           map[string]any{"ingest": 5.5, "compaction": 6.5},
+		"write_gate_wait_seconds":           map[string]any{"ingest": 7.5, "compaction": 8.5},
+		"compaction_seconds_total":          9.5,
+		"compactions_total":                 float64(12),
+		"publish_wait_seconds_total":        11.5,
+		"ingest_backpressure_seconds_total": 12.5,
+		"compaction_backlog_batches":        float64(13),
+	} {
+		if !reflect.DeepEqual(fields[key], want) {
+			t.Errorf("JSON %s=%v; want %v", key, fields[key], want)
+		}
+	}
+}
+
+func TestResourceSampleWithoutBackgroundMetrics(t *testing.T) {
+	const fixture = `process_cpu_seconds_total 12
+process_resident_memory_bytes 1024
+go_memstats_heap_alloc_bytes 512
+fanout_ingest_queue_depth 3
+fanout_parquet_files 4
+fanout_parquet_size_bytes 2048
+fanout_parquet_publish_waiters 2
+fanout_parquet_publish_timeouts_total 1
+`
+	snap, err := parsePrometheus(strings.NewReader(fixture))
+	if err != nil {
+		t.Fatal(err)
+	}
+	got := resourceSample(snap, 1.23456)
+	want := ResourceSample{
+		AtSeconds: 1.235, CPUSeconds: 12, RSSBytes: 1024, HeapBytes: 512,
+		QueueDepth: 3, ParquetFiles: 4, ParquetBytes: 2048, PublishWaiters: 2, PublishTimeouts: 1,
+	}
+	if !reflect.DeepEqual(got, want) {
+		t.Fatalf("legacy sample=%+v; want %+v", got, want)
+	}
+	data, err := json.Marshal(got)
+	if err != nil {
+		t.Fatal(err)
+	}
+	const wantJSON = `{"at_seconds":1.235,"cpu_seconds":12,"rss_bytes":1024,"heap_bytes":512,"queue_depth":3,"parquet_files":4,"parquet_bytes":2048,"publish_waiters":2,"publish_timeouts_total":1}`
+	if string(data) != wantJSON {
+		t.Fatalf("legacy JSON=%s; want %s", data, wantJSON)
+	}
+}
+
 func TestRunPhaseStopsQueriesBeforeFinalGeneratorCPURead(t *testing.T) {
 	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
 	defer cancel()

internal/bench/types.go

@@ -135,15 +135,23 @@ type Counts struct {
 func (c Counts) Total() float64 { return c.Spans + c.LogRecords + c.MetricPoints }
 
 type ResourceSample struct {
-	AtSeconds       float64 `json:"at_seconds"`
-	CPUSeconds      float64 `json:"cpu_seconds"`
-	RSSBytes        float64 `json:"rss_bytes"`
-	HeapBytes       float64 `json:"heap_bytes"`
-	QueueDepth      float64 `json:"queue_depth"`
-	ParquetFiles    float64 `json:"parquet_files"`
-	ParquetBytes    float64 `json:"parquet_bytes"`
-	PublishWaiters  float64 `json:"publish_waiters"`
-	PublishTimeouts float64 `json:"publish_timeouts_total"`
+	AtSeconds                      float64            `json:"at_seconds"`
+	CPUSeconds                     float64            `json:"cpu_seconds"`
+	RSSBytes                       float64            `json:"rss_bytes"`
+	HeapBytes                      float64            `json:"heap_bytes"`
+	QueueDepth                     float64            `json:"queue_depth"`
+	ParquetFiles                   float64            `json:"parquet_files"`
+	ParquetBytes                   float64            `json:"parquet_bytes"`
+	PublishWaiters                 float64            `json:"publish_waiters"`
+	PublishTimeouts                float64            `json:"publish_timeouts_total"`
+	RollupSeconds                  map[string]float64 `json:"rollup_seconds,omitempty"`
+	WriteGateHoldSeconds           map[string]float64 `json:"write_gate_hold_seconds,omitempty"`
+	WriteGateWaitSeconds           map[string]float64 `json:"write_gate_wait_seconds,omitempty"`
+	CompactionSecondsTotal         float64            `json:"compaction_seconds_total,omitempty"`
+	CompactionsTotal               float64            `json:"compactions_total,omitempty"`
+	PublishWaitSecondsTotal        float64            `json:"publish_wait_seconds_total,omitempty"`
+	IngestBackpressureSecondsTotal float64            `json:"ingest_backpressure_seconds_total,omitempty"`
+	CompactionBacklogBatches       float64            `json:"compaction_backlog_batches,omitempty"`
 }
 
 type Latency struct {