diff --git a/docs/implementer-log.md b/docs/implementer-log.md index 1854bed..cd50411 100644 --- a/docs/implementer-log.md +++ b/docs/implementer-log.md @@ -23,6 +23,7 @@ One row per task session that ends in a commit. Newest last. | 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. | | p6/05-fetchtarget | 2026-09-26 | done | 1 | gate: ok | none | Extracted v1fetch's per-target pipeline into new `internal/api/target.go`: `fetchTarget(ctx, c, t batch.Target) (batch.Result, error)`. Body moved unchanged in order — question truncation to 500 runes, mode default/check, wantCtx, doFetch, recover path, PDF branch (render, pages/grep errors, question-chunk limit), clean/raw branch, digest, context search, the fetch log line, and the envelope. `content` is returned as `Result.Content`; every other envelope key (`mode`,`untrusted`,`source`,`warnings`,`sizes`,`timings`, and `pdf`/`digest`/`context` when present) goes in `Result.Meta`, never content. `v1fetch` keeps reqN/caller/decode/400, builds a `batch.Target` (`URL` + `Mode`/`Question`/`Pages`/`Grep`/`JQ`/`Context`/`Recover` into `Options`), calls `fetchTarget`, `writeFetchErr` on error and `writeJSON(w,200,env)` on success where `env` is a fresh copy of `Meta` plus `"content"` (a copy, so `len(env)==len(Meta)+1` holds and Meta is not mutated). The two `writeErr` failures become `*fetch.Error` with the same status/msg/hint: 400 "invalid url" (bad mode), 400 "raw mode does not apply to PDFs". `doFetch` takes `context.Context` instead of `*http.Request` (its three `r.Context()` → `ctx`); `scrape` passes `r.Context()`. Removed now-unused `fmt`/`unicode/utf8`/`recover` imports, added `batch`. All api tests pass under -race -count=3; make gate ok; given test byte-identical. | | p6/06-batch-wiring | 2026-09-26 | done | 1 | gate: ok | none | New `internal/api/batch.go`: `NewBatchRunner` overwrites `cfg.Hooks` so Job→ObserveBatchJob, Target→ObserveBatchTarget(batchOutcome)+ObserveBatchTargetSeconds (+ObserveBatchResultBytes only when err==nil), and JobSeconds/Depth/Inflight→the matching `s.Prom` methods; the runner's FetchFunc is `batchFetch`, which tags ctx with an unexported `batchCtxKey`, runs under `requestDeadline`, and maps fetchTarget's errors (a *fetch.Error 429/503 → `fmt.Errorf("%s: %w", fe.Msg, batch.ErrRetryLater)` so the runner retries; any other *fetch.Error → TargetError with 502 rewritten to 500 "internal error" and `s.blockN.Add(1)` for 403; anything else → TargetError 500 logged Error with `s.logErr`). `batchOutcome` classifies nil→ok, ErrRetryLater-wrapped→busy, then by TargetError.Status (403 blocked, 429 busy, 504 timeout, 415 unsupported, 422 whose Msg starts "document too long"→too_long, other 4xx→upstream_4xx, 5xx→upstream_5xx, any other error→upstream_5xx). Added `Batch *batch.Runner` Server field (nil → the /v1/batch endpoints answer 503 "batch disabled"). `pdf.go` extractPDF charges the PDF pool to `c.Name+"/batch"` when the ctx carries the batch mark, single-flight key unchanged so a batch and an interactive extract of the same PDF still share one child; the only new log line carries caller and host="-" (never a URL/host). All api tests pass under -race -count=3; make gate ok; given file byte-identical. | +| p6/07-api-endpoints | 2026-09-26 | done | 1 | gate: ok | none | The four /v1/batch handlers in new `internal/api/batch.go`: `batchSubmit` (POST) decodes `{"targets":[{url,mode,question,pages,grep,jq,context,recover}]}` flat like `/v1/fetch`, 400 "invalid batch" on bad json/empty list/empty url/bad mode (only empty,clean,raw,digest), then `PreCheck` per target → 422 "blocked: target : " with no job, then `Submit` mapping ErrTooManyTargets→422 "too many targets (max 32)", ErrCallerBusy→429+Retry-After 30, ErrBusy→429+Retry-After 60, ErrNoTargets→400, and 202 {success,job_id,status,targets,reused} (status "queued" new, current status reused). `batchStatus` (GET) and `batchResult` (GET {id}/{idx}) and `batchCancel` (DELETE) share `ownership` (unknown or not `c.Name` → 404, never 403; nil batch → 503); status is a summary with no results unless `?results=1` inlines the envelope (Meta+content) on done targets, idx validated 0..targets_total-1 else 404, cancel of a terminal job → 409 "job already finished" else 200 {success,status:cancelled}. `api.go` scoped to the rule-2 changes: `batchMaxBody` (256 KiB) const, the four routes in `New`, and `ServeHTTP` applying the per-path cap once (`/v1/batch`→256 KiB, `/v1/extract`→pdfMaxUpload, else 64 KiB) and collapsing every batch path to a fixed metrics label via `batchObservePath` (`/v1/batch` or `/v1/batch/{id}`, never a job id). Each handler logs one `batch` line with caller/job_id/targets/status-or-reused and never a URL/hostname/idem_key. Given test byte-identical; all api tests pass under -race -count=3; make gate ok. | ## P5a review (owner, 2026-09-26) diff --git a/internal/api/api.go b/internal/api/api.go index 3818f9b..61a0a2b 100644 --- a/internal/api/api.go +++ b/internal/api/api.go @@ -37,6 +37,10 @@ const ( writeTimeout = 240 * time.Second // P5a: a PDF may take 120 s to download + 20 s queued + 60 s to extract requestDeadline = 210 * time.Second // P5a: the PDF path's worst case (see writeTimeout); other paths keep their own shorter timeouts + // batchMaxBody is the largest body POST /v1/batch accepts: 32 targets, each with a + // large question, exceed the 64 KiB cap of the other endpoints. + batchMaxBody = 256 << 10 + // DefaultPDFMaxUpload is the largest body POST /v1/extract accepts: an uploaded // PDF, addressed by its hash and extracted once per content. DefaultPDFMaxUpload = 100 << 20 @@ -108,6 +112,10 @@ func New(version string, g *guard.Guard, f *fetch.Fetcher, keys *auth.Store, log s.Mux.HandleFunc("POST /v1/extract", s.v1extract) s.Mux.HandleFunc("POST /v1/search", s.v1search) s.Mux.HandleFunc("POST /v2/search", s.v2search) + s.Mux.HandleFunc("POST /v1/batch", s.batchSubmit) + s.Mux.HandleFunc("GET /v1/batch/{id}", s.batchStatus) + s.Mux.HandleFunc("GET /v1/batch/{id}/{idx}", s.batchResult) + s.Mux.HandleFunc("DELETE /v1/batch/{id}", s.batchCancel) return s } @@ -125,8 +133,11 @@ func (s *Server) HTTPServer(addr string) *http.Server { func (s *Server) ServeHTTP(w http.ResponseWriter, r *http.Request) { var limit int64 = maxReqBody - if r.URL.Path == "/v1/extract" { + switch r.URL.Path { + case "/v1/extract": limit = int64(s.pdfMaxUpload()) + case "/v1/batch": + limit = batchMaxBody } r.Body = http.MaxBytesReader(w, r.Body, limit) switch r.URL.Path { @@ -144,9 +155,29 @@ func (s *Server) ServeHTTP(w http.ResponseWriter, r *http.Request) { s.Prom.Observe(r.URL.Path, time.Since(start), code) return } + if strings.HasPrefix(r.URL.Path, "/v1/batch") { + cw := &codeWriter{ResponseWriter: w} + start := time.Now() + s.Mux.ServeHTTP(cw, r) + code := cw.code + if code == 0 { + code = http.StatusOK + } + s.Prom.Observe(batchObservePath(r.URL.Path), time.Since(start), code) + return + } s.Mux.ServeHTTP(w, r) } +// batchObservePath collapses a batch path to a fixed label so a job id never becomes a +// metric label value: the submit path stays, every {id}/{idx} variant becomes {id}. +func batchObservePath(path string) string { + if path == "/v1/batch" { + return "/v1/batch" + } + return "/v1/batch/{id}" +} + type codeWriter struct { http.ResponseWriter code int diff --git a/internal/api/batch.go b/internal/api/batch.go index 7dc7c9e..1945972 100644 --- a/internal/api/batch.go +++ b/internal/api/batch.go @@ -2,8 +2,11 @@ package api import ( "context" + "encoding/json" "errors" "fmt" + "net/http" + "strconv" "strings" "time" @@ -111,3 +114,253 @@ func batchOutcome(err error) string { } return "upstream_5xx" } + +// The four /v1/batch handlers (task 07). A job that is unknown or owned by another caller is a +// 404 for every GET and DELETE, never a 403, so a caller cannot probe for another caller's jobs. + +// ownership answers the caller for id: a disabled batch is 503, an unknown id or one not owned by +// c is a 404. The empty job returned on either path signals "already answered" to the caller. +func (s *Server) ownership(w http.ResponseWriter, c *auth.Caller, id string) (batch.Job, []batch.TargetStatus) { + if s.Batch == nil { + writeErr(w, 503, "batch disabled", "") + return batch.Job{}, nil + } + j, ts, ok := s.Batch.Status(id) + if !ok || j.Caller != c.Name { + writeErr(w, 404, "not found", "") + return batch.Job{}, nil + } + return j, ts +} + +// logBatch writes the one batch line per handler: caller, job_id, targets, then either status or +// reused. It never carries a target URL, hostname or the idem_key. +func (s *Server) logBatch(c *auth.Caller, jobID string, targets int, extra ...any) { + s.Log.Info("batch", + append([]any{ + "caller", c.Name, + "job_id", jobID, + "targets", targets, + }, extra...)..., + ) +} + +func validBatchMode(m string) bool { + switch m { + case "", "clean", "raw", "digest": + return true + } + return false +} + +func (s *Server) batchSubmit(w http.ResponseWriter, r *http.Request) { + c := s.caller(w, r) + if c == nil { + return + } + var req struct { + Targets []v1Req `json:"targets"` + } + if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + writeErr(w, 400, "invalid batch: bad json", "") + return + } + if len(req.Targets) == 0 { + writeErr(w, 400, "invalid batch: no targets", "") + return + } + targets := make([]batch.Target, 0, len(req.Targets)) + for _, t := range req.Targets { + if t.URL == "" { + writeErr(w, 400, "invalid batch: empty url", "") + return + } + if !validBatchMode(t.Mode) { + writeErr(w, 400, "invalid batch: bad mode", "") + return + } + targets = append(targets, batch.Target{ + URL: t.URL, + Options: batch.Options{ + Mode: t.Mode, + Question: t.Question, + Pages: t.Pages, + Grep: t.Grep, + JQ: t.JQ, + Context: t.Context, + Recover: t.Recover, + }, + }) + } + for i, t := range targets { + if err := s.Guard.PreCheck(t.URL); err != nil { + s.logBatch(c, "", len(targets), "status", "blocked") + writeErr(w, 422, fmt.Sprintf("blocked: target %d: %v", i, err), "") + return + } + } + id, reused, err := s.Batch.Submit(r.Context(), c, s.callerClass(c), targets) + if err != nil { + s.logBatch(c, "", len(targets), "status", "failed") + switch { + case errors.Is(err, batch.ErrTooManyTargets): + writeErr(w, 422, "too many targets (max 32)", "") + case errors.Is(err, batch.ErrCallerBusy): + w.Header().Set("Retry-After", "30") + writeErr(w, 429, "busy: a batch of yours is still running", "") + case errors.Is(err, batch.ErrBusy): + w.Header().Set("Retry-After", "60") + writeErr(w, 429, "batch jobs busy", "") + case errors.Is(err, batch.ErrNoTargets): + writeErr(w, 400, "invalid batch: no targets", "") + default: + writeErr(w, 500, "internal error", "") + } + return + } + status := "queued" + if reused { + if j, _, ok := s.Batch.Status(id); ok { + status = j.Status + } + } + if reused { + s.logBatch(c, id, len(targets), "reused", true) + } else { + s.logBatch(c, id, len(targets), "status", status) + } + writeJSON(w, 202, map[string]any{ + "success": true, + "job_id": id, + "status": status, + "targets": len(targets), + "reused": reused, + }) +} + +func (s *Server) batchStatus(w http.ResponseWriter, r *http.Request) { + c := s.caller(w, r) + if c == nil { + return + } + id := r.PathValue("id") + j, ts := s.ownership(w, c, id) + if j.ID == "" { + return + } + s.Batch.Touch(id) + s.logBatch(c, id, j.TargetsTotal, "status", j.Status) + writeJSON(w, 200, statusEnvelope(j, ts, r)) +} + +func (s *Server) batchResult(w http.ResponseWriter, r *http.Request) { + c := s.caller(w, r) + if c == nil { + return + } + id := r.PathValue("id") + j, ts := s.ownership(w, c, id) + if j.ID == "" { + return + } + idx, err := strconv.Atoi(r.PathValue("idx")) + if err != nil || idx < 0 || idx >= j.TargetsTotal { + writeErr(w, 404, "not found", "") + return + } + s.Batch.Touch(id) + t := ts[idx] + out := map[string]any{ + "job_id": j.ID, + "idx": t.Idx, + "url": t.URL, + "status": t.Status, + } + if t.Result != nil { + out["result"] = resultEnvelope(t.Result) + } + if t.Error != "" { + out["error"] = t.Error + } + if t.Hint != "" { + out["hint"] = t.Hint + } + s.logBatch(c, id, j.TargetsTotal, "status", t.Status) + writeJSON(w, 200, out) +} + +func (s *Server) batchCancel(w http.ResponseWriter, r *http.Request) { + c := s.caller(w, r) + if c == nil { + return + } + id := r.PathValue("id") + j, _ := s.ownership(w, c, id) + if j.ID == "" { + return + } + switch j.Status { + case "done", "partial", "failed", "cancelled": + s.logBatch(c, id, j.TargetsTotal, "status", "failed") + writeErr(w, 409, "job already finished", "") + return + } + if s.Batch.Cancel(id) { + s.logBatch(c, id, j.TargetsTotal, "status", "cancelled") + writeJSON(w, 200, map[string]any{"success": true, "status": "cancelled"}) + return + } + s.logBatch(c, id, j.TargetsTotal, "status", "failed") + writeErr(w, 409, "job already finished", "") +} + +// statusEnvelope is the summary GET /v1/batch/{id} returns: statuses, never inline results, +// because 32 results can be several megabytes. With ?results=1 each done target carries the +// /v1/fetch envelope (Meta plus content). +func statusEnvelope(j batch.Job, ts []batch.TargetStatus, r *http.Request) map[string]any { + out := map[string]any{ + "job_id": j.ID, + "status": j.Status, + "targets_total": j.TargetsTotal, + "targets_done": j.TargetsDone, + "created_at": time.Unix(j.CreatedAt, 0).UTC().Format(time.RFC3339), + "targets": targetSummaries(ts, r), + } + if j.Error != "" { + out["error"] = j.Error + } + return out +} + +func targetSummaries(ts []batch.TargetStatus, r *http.Request) []map[string]any { + results := r.URL.Query().Get("results") == "1" + out := make([]map[string]any, 0, len(ts)) + for _, t := range ts { + m := map[string]any{ + "idx": t.Idx, + "url": t.URL, + "status": t.Status, + } + if t.Error != "" { + m["error"] = t.Error + } + if t.Hint != "" { + m["hint"] = t.Hint + } + if results && t.Status == "done" && t.Result != nil { + m["result"] = resultEnvelope(t.Result) + } + out = append(out, m) + } + return out +} + +// resultEnvelope is the /v1/fetch shape: every Meta field plus content. +func resultEnvelope(res *batch.Result) map[string]any { + env := make(map[string]any, len(res.Meta)+1) + for k, v := range res.Meta { + env[k] = v + } + env["content"] = res.Content + return env +} diff --git a/internal/api/batch_test.go b/internal/api/batch_test.go new file mode 100644 index 0000000..6b81ad6 --- /dev/null +++ b/internal/api/batch_test.go @@ -0,0 +1,280 @@ +package api + +import ( + "encoding/json" + "fmt" + "io" + "net/http" + "net/http/httptest" + "strings" + "testing" + "time" +) + +// Every test fetches only from local httptest servers or internal.test (fake DNS): no test +// touches the network. newBatchServer, pageServer and fakeDNS come from batchrun_test.go. + +func batchReq(t *testing.T, s *Server, method, path, key string, body any) *httptest.ResponseRecorder { + t.Helper() + var rd io.Reader + if body != nil { + b, err := json.Marshal(body) + if err != nil { + t.Fatal(err) + } + rd = strings.NewReader(string(b)) + } + req := httptest.NewRequest(method, path, rd) + req.Header.Set("Authorization", "Bearer "+key) + if body != nil { + req.Header.Set("Content-Type", "application/json") + } + rec := httptest.NewRecorder() + s.ServeHTTP(rec, req) + return rec +} + +func batchDecode(t *testing.T, rec *httptest.ResponseRecorder) map[string]any { + t.Helper() + var m map[string]any + if err := json.Unmarshal(rec.Body.Bytes(), &m); err != nil { + t.Fatalf("decode: %v (%d %s)", err, rec.Code, rec.Body.String()) + } + return m +} + +func submit(t *testing.T, s *Server, key string, targets ...map[string]any) map[string]any { + t.Helper() + rec := batchReq(t, s, http.MethodPost, "/v1/batch", key, map[string]any{"targets": targets}) + if rec.Code != 202 { + t.Fatalf("submit: %d %s", rec.Code, rec.Body.String()) + } + return batchDecode(t, rec) +} + +func waitTerminal(t *testing.T, s *Server, key, id string) map[string]any { + t.Helper() + deadline := time.Now().Add(10 * time.Second) + for time.Now().Before(deadline) { + env := batchDecode(t, batchReq(t, s, http.MethodGet, "/v1/batch/"+id, key, nil)) + switch env["status"] { + case "done", "partial", "failed", "cancelled": + return env + } + time.Sleep(5 * time.Millisecond) + } + t.Fatal("batch never reached a terminal status") + return nil +} + +func TestBatchSubmitAndCollect(t *testing.T) { + ts := pageServer(t, nil) + s, keys := newBatchServer(t, "test") + sub := submit(t, s, keys["test"], map[string]any{"url": ts.URL + "/one"}, map[string]any{"url": ts.URL + "/two"}) + id, _ := sub["job_id"].(string) + if sub["success"] != true || id == "" || sub["reused"] != false || sub["targets"] != float64(2) || sub["status"] != "queued" { + t.Fatalf("submit body %v", sub) + } + + // The status body is a summary: statuses, no results. + env := waitTerminal(t, s, keys["test"], id) + if env["status"] != "done" || env["targets_total"] != float64(2) || env["targets_done"] != float64(2) { + t.Fatalf("status %v", env) + } + if _, err := time.Parse(time.RFC3339, fmt.Sprint(env["created_at"])); err != nil { + t.Fatalf("created_at %v is not RFC 3339", env["created_at"]) + } + targets := env["targets"].([]any) + if len(targets) != 2 { + t.Fatalf("targets %d", len(targets)) + } + for i, x := range targets { + tg := x.(map[string]any) + if tg["idx"] != float64(i) || tg["status"] != "done" { + t.Fatalf("target %d: %v", i, tg) + } + if _, has := tg["result"]; has { + t.Fatalf("summary must not carry results: target %d has one", i) + } + } + + // One target's result: the /v1/fetch envelope. + rec := batchReq(t, s, http.MethodGet, "/v1/batch/"+id+"/1", keys["test"], nil) + if rec.Code != 200 { + t.Fatalf("GET idx: %d %s", rec.Code, rec.Body.String()) + } + one := batchDecode(t, rec) + res, _ := one["result"].(map[string]any) + if one["idx"] != float64(1) || one["status"] != "done" || res == nil { + t.Fatalf("GET idx body %v", one) + } + if c, _ := res["content"].(string); !strings.Contains(c, "hello batch /two") { + t.Fatalf("result content %q", c) + } + if res["untrusted"] != true || res["mode"] != "clean" { + t.Fatalf("result is not the /v1/fetch envelope: %v", res) + } + if src, _ := res["source"].(map[string]any); src == nil || src["final_url"] != ts.URL+"/two" { + t.Fatalf("result source %v", res["source"]) + } + + // ?results=1 inlines every result. + full := batchDecode(t, batchReq(t, s, http.MethodGet, "/v1/batch/"+id+"?results=1", keys["test"], nil)) + for i, x := range full["targets"].([]any) { + if _, has := x.(map[string]any)["result"]; !has { + t.Fatalf("results=1: target %d has no result", i) + } + } + + if got := batchReq(t, s, http.MethodGet, "/v1/batch/"+id+"/2", keys["test"], nil).Code; got != 404 { + t.Fatalf("GET idx out of range: %d, want 404", got) + } + if got := batchReq(t, s, http.MethodGet, "/v1/batch/"+id+"/x", keys["test"], nil).Code; got != 404 { + t.Fatalf("GET idx not a number: %d, want 404", got) + } +} + +func TestBatchIdempotentResubmit(t *testing.T) { + ts := pageServer(t, nil) + s, keys := newBatchServer(t, "test") + tg := map[string]any{"url": ts.URL, "mode": "clean"} + a := submit(t, s, keys["test"], tg) + b := submit(t, s, keys["test"], tg) + if a["job_id"] != b["job_id"] || b["reused"] != true { + t.Fatalf("resubmit: %v then %v", a, b) + } +} + +func TestBatchRefusedUpFront(t *testing.T) { + s, keys := newBatchServer(t, "test") + cases := []struct { + name string + body any + code int + want string + }{ + {"guard pre-check (port)", map[string]any{"targets": []any{map[string]any{"url": "http://127.0.0.1/"}, map[string]any{"url": "http://example.com:8080/"}}}, 422, "blocked"}, + {"guard pre-check (private literal)", map[string]any{"targets": []any{map[string]any{"url": "http://10.0.0.5/"}}}, 422, "blocked"}, + {"no targets", map[string]any{"targets": []any{}}, 400, ""}, + {"not json", "nope", 400, ""}, + {"empty url", map[string]any{"targets": []any{map[string]any{"url": ""}}}, 400, ""}, + {"bad mode", map[string]any{"targets": []any{map[string]any{"url": "http://127.0.0.1/", "mode": "shout"}}}, 400, ""}, + } + for _, c := range cases { + rec := batchReq(t, s, http.MethodPost, "/v1/batch", keys["test"], c.body) + if rec.Code != c.code || !strings.Contains(rec.Body.String(), c.want) { + t.Errorf("%s: %d %s, want %d containing %q", c.name, rec.Code, rec.Body.String(), c.code, c.want) + } + } + // The guard refusal names the offending index. + rec := batchReq(t, s, http.MethodPost, "/v1/batch", keys["test"], cases[0].body) + if !strings.Contains(rec.Body.String(), "1") { + t.Errorf("guard refusal does not name target 1: %s", rec.Body.String()) + } + + var many []any + for i := 0; i < 33; i++ { + many = append(many, map[string]any{"url": fmt.Sprintf("http://127.0.0.1/%d", i)}) + } + if rec := batchReq(t, s, http.MethodPost, "/v1/batch", keys["test"], map[string]any{"targets": many}); rec.Code != 422 { + t.Errorf("33 targets: %d, want 422", rec.Code) + } +} + +// 32 targets, each with a 500-rune question, fit the batch body cap even though they are well +// over the 64 KiB cap of the other endpoints. +func TestBatchBodyCap(t *testing.T) { + hold := make(chan struct{}) + defer close(hold) + ts := pageServer(t, hold) + s, keys := newBatchServer(t, "test") + q := strings.Repeat("q", 500) + var many []any + for i := 0; i < 32; i++ { + many = append(many, map[string]any{"url": fmt.Sprintf("%s/%d?pad=%s", ts.URL, i, strings.Repeat("p", 1500)), "question": q}) + } + if rec := batchReq(t, s, http.MethodPost, "/v1/batch", keys["test"], map[string]any{"targets": many}); rec.Code != 202 { + t.Fatalf("32 targets with questions (~%d KiB): %d %s", 32*2100/1024, rec.Code, rec.Body.String()) + } +} + +func TestBatchPartialFailure(t *testing.T) { + ts := pageServer(t, nil) + s, keys := newBatchServer(t, "test") + // Target 1 is a name that resolves to a private address. Names are not resolved at submit, + // so it passes the pre-check; the fetch-time guard blocks it: a per-target failure. + sub := submit(t, s, keys["test"], map[string]any{"url": ts.URL}, map[string]any{"url": "http://internal.test/"}) + env := waitTerminal(t, s, keys["test"], sub["job_id"].(string)) + if env["status"] != "partial" { + t.Fatalf("status %v", env["status"]) + } + second := env["targets"].([]any)[1].(map[string]any) + if second["status"] != "failed" { + t.Fatalf("target 1: %v", second) + } + if e, _ := second["error"].(string); !strings.Contains(e, "blocked") { + t.Fatalf("target 1 error %q", e) + } +} + +func TestBatchOwnership404(t *testing.T) { + hold := make(chan struct{}) + defer close(hold) + ts := pageServer(t, hold) + s, keys := newBatchServer(t, "test", "other") + id := submit(t, s, keys["test"], map[string]any{"url": ts.URL})["job_id"].(string) + for _, p := range []string{"/v1/batch/" + id, "/v1/batch/" + id + "/0"} { + if got := batchReq(t, s, http.MethodGet, p, keys["other"], nil).Code; got != 404 { + t.Fatalf("other caller GET %s: %d (must not reveal the job)", p, got) + } + } + if got := batchReq(t, s, http.MethodDelete, "/v1/batch/"+id, keys["other"], nil).Code; got != 404 { + t.Fatalf("other caller DELETE: %d", got) + } + if got := batchReq(t, s, http.MethodGet, "/v1/batch/"+id, keys["test"], nil).Code; got != 200 { + t.Fatalf("owner GET: %d", got) + } + if got := batchReq(t, s, http.MethodGet, "/v1/batch/no-such-job", keys["test"], nil).Code; got != 404 { + t.Fatalf("unknown job: %d", got) + } + if got := batchReq(t, s, http.MethodGet, "/v1/batch/"+id, "not-a-key", nil).Code; got != 401 { + t.Fatalf("bad key: %d", got) + } +} + +func TestBatchCancel(t *testing.T) { + hold := make(chan struct{}) + ts := pageServer(t, hold) + s, keys := newBatchServer(t, "test") + id := submit(t, s, keys["test"], map[string]any{"url": ts.URL + "/a"}, map[string]any{"url": ts.URL + "/b"})["job_id"].(string) + rec := batchReq(t, s, http.MethodDelete, "/v1/batch/"+id, keys["test"], nil) + if rec.Code != 200 { + t.Fatalf("cancel: %d %s", rec.Code, rec.Body.String()) + } + if b := batchDecode(t, rec); b["success"] != true || b["status"] != "cancelled" { + t.Fatalf("cancel body %v", b) + } + close(hold) + env := waitTerminal(t, s, keys["test"], id) + if env["status"] != "cancelled" { + t.Fatalf("status %v", env["status"]) + } + if tg := env["targets"].([]any)[1].(map[string]any); tg["status"] != "pending" { + t.Fatalf("target 1 after cancel: %v", tg) + } + if got := batchReq(t, s, http.MethodDelete, "/v1/batch/"+id, keys["test"], nil).Code; got != 409 { + t.Fatalf("cancelling a finished job: %d, want 409", got) + } +} + +// The request metrics never get a job id as a label value. +func TestBatchMetricsPathLabel(t *testing.T) { + ts := pageServer(t, nil) + s, keys := newBatchServer(t, "test") + id := submit(t, s, keys["test"], map[string]any{"url": ts.URL})["job_id"].(string) + waitTerminal(t, s, keys["test"], id) + batchReq(t, s, http.MethodGet, "/v1/batch/"+id+"/0", keys["test"], nil) + if strings.Contains(promText(t, s), id) { + t.Fatal("a job id appears in /metrics") + } +}