metrics: sift_batch_* series (jobs, targets, depth, in-flight, durations, bytes)

New internal/metrics/batch.go adds the batch series to metrics.Set, registered
in New: BatchJobsTotal and BatchTargetsTotal counters (label "outcome", any
value outside the known set folded to "other" so a raw string never becomes a
label), BatchQueueDepth and BatchTargetsInflight gauges, BatchJobSeconds and
BatchTargetSeconds histograms (the existing request buckets) and
BatchResultBytes (1K,4K,16K,64K,256K,1M). The Observe…/Set… methods are
nil-receiver safe like ObservePDF and flow through the existing /metrics
handler; no new endpoint or Snapshot change.

Co-Authored-By: ornith-1.5-35b-a3b <ornith-1.5-35b-a3b@llama.cpp.invalid>

docs/implementer-log.md

@@ -20,6 +20,7 @@ One row per task session that ends in a commit. Newest last.
 | p6/01-batchstore | 2026-09-26 | done | 1 | gate: ok | none | Storage-only batch store in `internal/dcache` (new `batch.go`; schema added to `Open`; `BatchH`/`BatchM` counters added to `DB`). `CreateBatchJob` stores as given. Reads never write the lease: `BatchJob`, `BatchTargets` and the counts are pure, so exact lease values hold after a read. `BatchJobByIDem` returns the newest `(caller,idemKey)` job with `lease+ttl>now` and status queued/running/done/partial; failed/cancelled never reused. `PutBatchTargets` inserts in order; `BatchTargets` returns idx order with NULLs as nil pointers. `UpdateBatchTarget` bumps lease to `now`, increments `targets_done` exactly once per target entering done/failed (previous-status check prevents re-write double-count), and adds `len(result)` to `nbytes` when a result is written — all in one tx. `SetBatchJobStatus`/`TouchBatchJob` set lease; `SweepBatch` deletes by `lease+ttl<=now` with its targets and returns the job count; `EvictBatchOldest` evicts whole non-live jobs oldest-lease-first (never a queued/running one) until `BatchNbytes()<=maxBytes`; `MarkInFlightFailed` fails queued/running on restart; `BatchNbytes` sums job nbytes, kept out of `Size()`. Accessors return `(v,bool)`/`error`, count misses in `BatchM`, reuse `ClassFleet`/`ClassRestricted`. A first pass failed `TestBatchIDemLookup` because I forgot `.Scan(&id)` on the `QueryRow` (the un-scanned `*sql.Row` was always non-nil, so the method always missed); fixed and verified. All dcache tests pass under `-race -count=3`; make gate ok; given file byte-identical.
 | p6/02-runner-core | 2026-09-26 | done | 1 | gate: ok | caller-cleanup hand-off to task 03 | Runner core in new internal/batch with no workers (task 03 adds them; jobs stay queued, so a queued job counts as live for the caps). Public surface in batch.go: Options/Target/Result/TargetError, FetchFunc, Job/TargetStatus, Hooks and Config with defaults (TTL 24h, 32 targets/job, 1 job/caller, 8 total, worker clamped to 2, 64 MiB, 2 s/10 m retry). Submit runs the given order — empty→ErrNoTargets, idempotency via hex SHA-256 of `name+NUL+json.Marshal(targets)` (a live hit returns the id reused, touching no caps), then TargetsPerJob/JobsPerCaller/JobsTotal caps, then stores the job queued (random 16-byte hex id, CreatedAt=Lease=now) and its targets (pending, Options=json of t.Options) before returning so a queued job is live; the submitting caller is held in memory by job id, never persisted. Status maps rows and decodes Result without writing; Touch moves the lease; Cancel sets a queued job cancelled at once and leaves its targets pending; Close is idempotent. Store errors are Warn-logged with the job id, never a URL/host. |
 | p6/03-runner-workers | 2026-09-26 | done | 1 | gate: ok | none | Workers in new internal/batch/worker.go: one rotation of live jobs in submit order, round-robin via an absolute cursor (advances in pick, wraps at access, so it survives a second job being added while the first is in flight — the mod-n cursor from the design stuck at 0 for one job and re-picked it), serial per job (target i+1 never starts before i is stored), one target per fetch under the Close-cancelled context (panic→ErrInternal, never echoed; ErrRetryLater retried after RetryWait, bounded by RetryFor then stored failed with the wrapped text). Terminal status once per job: all done→done, all failed→failed, mix→partial. Cancel of a running job sets cancelled, stores the in-flight target without overwriting cancelled, leaves later targets pending. Close sets closed, cancels the context, broadcasts, waits the workers, idempotent. batch.go: New starts cfg.Workers, Submit appends the liveJob and wakes, Cancel handles the in-flight case, Close stops. Absolute cursor is the one deviation from the design's "index in live" comment. All race tests pass -count=3. | |
+| p6/04-metrics | 2026-09-26 | done | 1 | gate: ok | none | Batch series in new internal/metrics/batch.go, registered in New: `BatchJobsTotal`/`BatchTargetsTotal` CounterVecs (label "outcome", any value outside the known set folded to "other" via `batchJobOutcomes`/`batchTargetOutcomes` maps so a raw string never becomes a label), `BatchQueueDepth`/`BatchTargetsInflight` gauges, and `BatchJobSeconds`/`BatchTargetSeconds` histograms (durationBuckets) plus `BatchResultBytes` (1K,4K,16K,64K,256K,1M). Nil-receiver-safe `Observe…`/`Set…` methods mirroring ObservePDF/ObserveDigest. Series served through the existing /metrics handler; no new endpoint, no Snapshot change. Given test byte-identical; metrics tests pass under -race -count=3; make gate ok. |
 
 ## P5a review (owner, 2026-09-26)
 

internal/metrics/batch.go

@@ -0,0 +1,80 @@
+package metrics
+
+import "time"
+
+var batchResultBytesBuckets = []float64{1024, 4096, 16384, 65536, 262144, 1048576}
+
+var batchJobOutcomes = map[string]struct{}{
+	"created": {}, "reused": {}, "done": {}, "partial": {},
+	"failed": {}, "cancelled": {}, "restart": {},
+}
+
+var batchTargetOutcomes = map[string]struct{}{
+	"ok": {}, "blocked": {}, "timeout": {}, "busy": {},
+	"upstream_4xx": {}, "upstream_5xx": {}, "unsupported": {},
+	"too_long": {}, "cancel": {},
+}
+
+// ObserveBatchJob records one batch outcome. Values outside the known set fold
+// to "other" so a raw string never becomes a label.
+func (s *Set) ObserveBatchJob(outcome string) {
+	if s == nil {
+		return
+	}
+	if _, ok := batchJobOutcomes[outcome]; !ok {
+		outcome = "other"
+	}
+	s.BatchJobsTotal.WithLabelValues(outcome).Inc()
+}
+
+// ObserveBatchTarget records one target outcome. Values outside the known set
+// fold to "other" so a raw string never becomes a label.
+func (s *Set) ObserveBatchTarget(outcome string) {
+	if s == nil {
+		return
+	}
+	if _, ok := batchTargetOutcomes[outcome]; !ok {
+		outcome = "other"
+	}
+	s.BatchTargetsTotal.WithLabelValues(outcome).Inc()
+}
+
+// SetBatchQueueDepth sets the live job count gauge (queued+running).
+func (s *Set) SetBatchQueueDepth(n int) {
+	if s == nil {
+		return
+	}
+	s.BatchQueueDepth.Set(float64(n))
+}
+
+// SetBatchTargetsInflight sets the in-flight targets gauge.
+func (s *Set) SetBatchTargetsInflight(n int) {
+	if s == nil {
+		return
+	}
+	s.BatchTargetsInflight.Set(float64(n))
+}
+
+// ObserveBatchJobSeconds records the wall time from submit to a batch's terminal status.
+func (s *Set) ObserveBatchJobSeconds(d time.Duration) {
+	if s == nil {
+		return
+	}
+	s.BatchJobSeconds.Observe(d.Seconds())
+}
+
+// ObserveBatchTargetSeconds records the wall time for one target fetch.
+func (s *Set) ObserveBatchTargetSeconds(d time.Duration) {
+	if s == nil {
+		return
+	}
+	s.BatchTargetSeconds.Observe(d.Seconds())
+}
+
+// ObserveBatchResultBytes records the stored byte count of one target result.
+func (s *Set) ObserveBatchResultBytes(n int) {
+	if s == nil {
+		return
+	}
+	s.BatchResultBytes.Observe(float64(n))
+}

internal/metrics/batch_test.go

@@ -0,0 +1,61 @@
+package metrics
+
+import (
+	"io"
+	"net/http/httptest"
+	"strings"
+	"testing"
+	"time"
+)
+
+type batchSrc struct{}
+
+func (batchSrc) Snapshot() Snapshot { return Snapshot{} }
+
+func TestBatchSeries(t *testing.T) {
+	s := New(batchSrc{})
+	s.ObserveBatchJob("created")
+	s.ObserveBatchJob("done")
+	s.ObserveBatchJob("done")
+	s.ObserveBatchJob("reused")
+	s.ObserveBatchJob("some raw string") // folded to "other", never a label
+	s.ObserveBatchTarget("ok")
+	s.ObserveBatchTarget("ok")
+	s.ObserveBatchTarget("blocked")
+	s.ObserveBatchTarget("upstream_5xx")
+	s.ObserveBatchTarget("leak me") // folded to "other"
+	s.SetBatchQueueDepth(3)
+	s.SetBatchTargetsInflight(1)
+	s.ObserveBatchJobSeconds(4 * time.Second)
+	s.ObserveBatchTargetSeconds(1500 * time.Millisecond)
+	s.ObserveBatchResultBytes(4096)
+
+	req := httptest.NewRequest("GET", "/metrics", nil)
+	req.RemoteAddr, req.Host = "127.0.0.1:9", "127.0.0.1:8890"
+	rec := httptest.NewRecorder()
+	s.Handler().ServeHTTP(rec, req)
+	body, _ := io.ReadAll(rec.Body)
+	out := string(body)
+	for _, want := range []string{
+		`sift_batch_jobs_total{outcome="created"} 1`,
+		`sift_batch_jobs_total{outcome="done"} 2`,
+		`sift_batch_jobs_total{outcome="reused"} 1`,
+		`sift_batch_jobs_total{outcome="other"} 1`,
+		`sift_batch_targets_total{outcome="ok"} 2`,
+		`sift_batch_targets_total{outcome="blocked"} 1`,
+		`sift_batch_targets_total{outcome="upstream_5xx"} 1`,
+		`sift_batch_targets_total{outcome="other"} 1`,
+		`sift_batch_queue_depth 3`,
+		`sift_batch_targets_inflight 1`,
+		`sift_batch_job_seconds_count 1`,
+		`sift_batch_target_seconds_count 1`,
+		`sift_batch_result_bytes_count 1`,
+	} {
+		if !strings.Contains(out, want) {
+			t.Errorf("metrics lack %q", want)
+		}
+	}
+	if strings.Contains(out, "raw") || strings.Contains(out, "leak") {
+		t.Error("a raw outcome string leaked into a label")
+	}
+}

internal/metrics/metrics.go

@@ -50,15 +50,22 @@ var pdfOutcomes = map[string]struct{}{
 
 // Set is a private registry of sift_* series.
 type Set struct {
-	Reg          *prometheus.Registry
-	Requests     *prometheus.CounterVec
-	Duration     *prometheus.HistogramVec
-	PDFTotal     *prometheus.CounterVec
-	PDFSeconds   prometheus.Histogram
-	PDFPages     prometheus.Counter
-	PDFWait      prometheus.Histogram
-	PDFCacheHits prometheus.Counter
-	handler      http.Handler
+	Reg                  *prometheus.Registry
+	Requests             *prometheus.CounterVec
+	Duration             *prometheus.HistogramVec
+	PDFTotal             *prometheus.CounterVec
+	PDFSeconds           prometheus.Histogram
+	PDFPages             prometheus.Counter
+	PDFWait              prometheus.Histogram
+	PDFCacheHits         prometheus.Counter
+	BatchJobsTotal       *prometheus.CounterVec
+	BatchTargetsTotal    *prometheus.CounterVec
+	BatchQueueDepth      prometheus.Gauge
+	BatchTargetsInflight prometheus.Gauge
+	BatchJobSeconds      prometheus.Histogram
+	BatchTargetSeconds   prometheus.Histogram
+	BatchResultBytes     prometheus.Histogram
+	handler              http.Handler
 }
 
 func New(src Source) *Set {
@@ -94,17 +101,57 @@ func New(src Source) *Set {
 		Name: "sift_pdf_cache_hits_total",
 		Help: "PDF answers served from the document cache.",
 	})
-	reg.MustRegister(requests, duration, pdfTotal, pdfSeconds, pdfPages, pdfWait, pdfCacheHits, &collector{src: src})
+	batchJobs := prometheus.NewCounterVec(prometheus.CounterOpts{
+		Name: "sift_batch_jobs_total",
+		Help: "Batch jobs by outcome.",
+	}, []string{"outcome"})
+	batchTargets := prometheus.NewCounterVec(prometheus.CounterOpts{
+		Name: "sift_batch_targets_total",
+		Help: "Batch targets by outcome.",
+	}, []string{"outcome"})
+	batchQueueDepth := prometheus.NewGauge(prometheus.GaugeOpts{
+		Name: "sift_batch_queue_depth",
+		Help: "Live batch jobs (queued+running).",
+	})
+	batchTargetsInflight := prometheus.NewGauge(prometheus.GaugeOpts{
+		Name: "sift_batch_targets_inflight",
+		Help: "Batch targets currently being fetched.",
+	})
+	batchJobSeconds := prometheus.NewHistogram(prometheus.HistogramOpts{
+		Name:    "sift_batch_job_seconds",
+		Help:    "Batch wall time from submit to terminal status.",
+		Buckets: durationBuckets,
+	})
+	batchTargetSeconds := prometheus.NewHistogram(prometheus.HistogramOpts{
+		Name:    "sift_batch_target_seconds",
+		Help:    "Batch target fetch wall time.",
+		Buckets: durationBuckets,
+	})
+	batchResultBytes := prometheus.NewHistogram(prometheus.HistogramOpts{
+		Name:    "sift_batch_result_bytes",
+		Help:    "Stored bytes per batch target result.",
+		Buckets: batchResultBytesBuckets,
+	})
+	reg.MustRegister(requests, duration, pdfTotal, pdfSeconds, pdfPages, pdfWait, pdfCacheHits,
+		batchJobs, batchTargets, batchQueueDepth, batchTargetsInflight,
+		batchJobSeconds, batchTargetSeconds, batchResultBytes, &collector{src: src})
 	return &Set{
-		Reg:          reg,
-		Requests:     requests,
-		Duration:     duration,
-		PDFTotal:     pdfTotal,
-		PDFSeconds:   pdfSeconds,
-		PDFPages:     pdfPages,
-		PDFWait:      pdfWait,
-		PDFCacheHits: pdfCacheHits,
-		handler:      promhttp.HandlerFor(reg, promhttp.HandlerOpts{EnableOpenMetrics: false}),
+		Reg:                  reg,
+		Requests:             requests,
+		Duration:             duration,
+		PDFTotal:             pdfTotal,
+		PDFSeconds:           pdfSeconds,
+		PDFPages:             pdfPages,
+		PDFWait:              pdfWait,
+		PDFCacheHits:         pdfCacheHits,
+		BatchJobsTotal:       batchJobs,
+		BatchTargetsTotal:    batchTargets,
+		BatchQueueDepth:      batchQueueDepth,
+		BatchTargetsInflight: batchTargetsInflight,
+		BatchJobSeconds:      batchJobSeconds,
+		BatchTargetSeconds:   batchTargetSeconds,
+		BatchResultBytes:     batchResultBytes,
+		handler:              promhttp.HandlerFor(reg, promhttp.HandlerOpts{EnableOpenMetrics: false}),
 	}
 }