diff --git a/docs/implementer-log.md b/docs/implementer-log.md index 105bd40..e5460a3 100644 --- a/docs/implementer-log.md +++ b/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) diff --git a/internal/metrics/batch.go b/internal/metrics/batch.go new file mode 100644 index 0000000..f830456 --- /dev/null +++ b/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)) +} diff --git a/internal/metrics/batch_test.go b/internal/metrics/batch_test.go new file mode 100644 index 0000000..8e1cad3 --- /dev/null +++ b/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") + } +} diff --git a/internal/metrics/metrics.go b/internal/metrics/metrics.go index 525edb2..e08dc0e 100644 --- a/internal/metrics/metrics.go +++ b/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}), } }