feat: record accepted rows in every resource sample

README.md

@@ -129,12 +129,12 @@ 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 and background-work samples
+├── samples.jsonl   per-second Fanout ingest, 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.
+`samples.jsonl` includes cumulative `ingest_rows_total` by signal (`spans`, `logs`, `metrics`), `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. Derive per-second accepted-row rates from consecutive samples' `ingest_rows_total` deltas divided by their `at_seconds` difference. 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:
 
@@ -186,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, including rollup durations, write-gate hold/wait durations, compaction duration/count/backlog, publish wait duration, and ingest backpressure duration;
+- a raw time series in `samples.jsonl` for independent charting, including accepted rows by signal, 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

@@ -38,6 +38,7 @@ Each `samples.jsonl` row also records the following `/metrics` values. The same
 
 | Sample field | Source |
 |---|---|
+| `ingest_rows_total` | `fanout_ingest_rows_total`, keyed by `signal` (`spans`, `logs`, `metrics`) |
 | `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` |
@@ -47,7 +48,7 @@ Each `samples.jsonl` row also records the following `/metrics` values. The same
 | `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.
+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. For per-second accepted-row rates, divide each signal's `ingest_rows_total` delta by the difference in `at_seconds` between consecutive samples. `compaction_backlog_batches` is a current gauge rather than a cumulative counter. Series with additional labels are summed, retaining the indicated map key for ingest signals, 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
 

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, 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)
+	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 `ingest_rows_total` by signal (`spans`, `logs`, `metrics`), `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

@@ -405,6 +405,7 @@ func resourceSample(s *snapshot, at float64) ResourceSample {
 		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),
+		IngestRowsTotal:                sampleTotalsByLabel(s, "fanout_ingest_rows_total", "signal"),
 		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"),

internal/bench/runner_test.go

@@ -15,6 +15,61 @@ import (
 	"time"
 )
 
+func TestResourceSampleIngestRowsTotal(t *testing.T) {
+	for _, tt := range []struct {
+		name    string
+		fixture string
+		want    map[string]float64
+	}{
+		{
+			name: "present",
+			fixture: `fanout_ingest_rows_total{signal="spans",transport="grpc"} 123
+fanout_ingest_rows_total{signal="spans",transport="http"} 7
+fanout_ingest_rows_total{signal="logs"} 45
+fanout_ingest_rows_total{signal="metrics"} 0
+process_cpu_seconds_total 12
+`,
+			want: map[string]float64{"spans": 130, "logs": 45, "metrics": 0},
+		},
+		{
+			name:    "absent",
+			fixture: "process_cpu_seconds_total 12\n",
+		},
+	} {
+		t.Run(tt.name, func(t *testing.T) {
+			snap, err := parsePrometheus(strings.NewReader(tt.fixture))
+			if err != nil {
+				t.Fatal(err)
+			}
+			got := resourceSample(snap, 1)
+			if !reflect.DeepEqual(got.IngestRowsTotal, tt.want) {
+				t.Fatalf("ingest rows=%v; want %v", got.IngestRowsTotal, tt.want)
+			}
+			data, err := json.Marshal(got)
+			if err != nil {
+				t.Fatal(err)
+			}
+			var fields map[string]json.RawMessage
+			if err := json.Unmarshal(data, &fields); err != nil {
+				t.Fatal(err)
+			}
+			raw, present := fields["ingest_rows_total"]
+			if present != (tt.want != nil) {
+				t.Fatalf("JSON ingest_rows_total present=%t; want %t", present, tt.want != nil)
+			}
+			if present {
+				var decoded map[string]float64
+				if err := json.Unmarshal(raw, &decoded); err != nil {
+					t.Fatal(err)
+				}
+				if !reflect.DeepEqual(decoded, tt.want) {
+					t.Fatalf("JSON ingest rows=%v; want %v", decoded, tt.want)
+				}
+			}
+		})
+	}
+}
+
 func TestResourceSampleBackgroundMetrics(t *testing.T) {
 	const fixture = `process_cpu_seconds_total 12
 process_resident_memory_bytes 1024

internal/bench/types.go

@@ -144,6 +144,7 @@ type ResourceSample struct {
 	ParquetBytes                   float64            `json:"parquet_bytes"`
 	PublishWaiters                 float64            `json:"publish_waiters"`
 	PublishTimeouts                float64            `json:"publish_timeouts_total"`
+	IngestRowsTotal                map[string]float64 `json:"ingest_rows_total,omitempty"`
 	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"`