P6 plan: asynchronous batch fetch endpoint — phase spec, task plan, given tests
Co-Authored-By: DeepSeek V4 Pro <noreply@deepseek.com>
This commit is contained in:
DeepSeek V4 Pro
@@ -0,0 +1,254 @@
|
||||
package api
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"io"
|
||||
"log/slog"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"net/netip"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"git.wntrmute.dev/kyle/sift/internal/auth"
|
||||
"git.wntrmute.dev/kyle/sift/internal/batch"
|
||||
"git.wntrmute.dev/kyle/sift/internal/dcache"
|
||||
"git.wntrmute.dev/kyle/sift/internal/fetch"
|
||||
"git.wntrmute.dev/kyle/sift/internal/guard"
|
||||
)
|
||||
|
||||
// newBatchServer builds a Server with the batch runner wired over an in-memory
|
||||
// dcache, exactly as serve does in production. callers are the caller names to
|
||||
// provision (each with a bearer key).
|
||||
func newBatchServer(t *testing.T, g *guard.Guard, callers ...string) (*Server, map[string]string) {
|
||||
t.Helper()
|
||||
dir := t.TempDir()
|
||||
keysPath := filepath.Join(dir, "keys.toml")
|
||||
keys := map[string]string{}
|
||||
for _, name := range callers {
|
||||
out := filepath.Join(dir, name)
|
||||
if err := auth.Add(keysPath, name, "hermes-fleet", out); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
b, err := os.ReadFile(out)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
keys[name] = strings.TrimSpace(string(b))
|
||||
}
|
||||
store, err := auth.Load(keysPath)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if g == nil {
|
||||
g = &guard.Guard{}
|
||||
}
|
||||
f := &fetch.Fetcher{Guard: g, UA: "sift-test"}
|
||||
s := New("test", g, f, store, slog.New(slog.NewTextHandler(io.Discard, nil)))
|
||||
db, err := dcache.Open("", 1<<20, slog.New(slog.NewTextHandler(io.Discard, nil)))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
t.Cleanup(func() { _ = db.Close() })
|
||||
s.Batch = batch.New(db, s.fetchTarget, batch.Config{}, slog.New(slog.NewTextHandler(io.Discard, nil)))
|
||||
return s, keys
|
||||
}
|
||||
|
||||
func submitBatch(t *testing.T, s *Server, key string, targets []map[string]string) *httptest.ResponseRecorder {
|
||||
t.Helper()
|
||||
return postJSON(t, s, "/v1/batch", key, map[string]any{"targets": targets})
|
||||
}
|
||||
|
||||
func getBatch(t *testing.T, s *Server, key, id string) *httptest.ResponseRecorder {
|
||||
t.Helper()
|
||||
req := httptest.NewRequest(http.MethodGet, "/v1/batch/"+id, nil)
|
||||
req.Header.Set("Authorization", "Bearer "+key)
|
||||
rec := httptest.NewRecorder()
|
||||
s.ServeHTTP(rec, req)
|
||||
return rec
|
||||
}
|
||||
|
||||
func batchStatus(t *testing.T, s *Server, key, id string) map[string]any {
|
||||
t.Helper()
|
||||
rec := getBatch(t, s, key, id)
|
||||
var env map[string]any
|
||||
if err := json.Unmarshal(rec.Body.Bytes(), &env); err != nil {
|
||||
t.Fatalf("status decode: %v (%s)", err, rec.Body.String())
|
||||
}
|
||||
return env
|
||||
}
|
||||
|
||||
// waitBatchTerminal polls until the job reaches a terminal status.
|
||||
func waitBatchTerminal(t *testing.T, s *Server, key, id string) map[string]any {
|
||||
t.Helper()
|
||||
deadline := time.Now().Add(3 * time.Second)
|
||||
for time.Now().Before(deadline) {
|
||||
env := batchStatus(t, s, key, id)
|
||||
if st, _ := env["status"].(string); st == "done" || st == "partial" || st == "failed" || st == "cancelled" {
|
||||
return env
|
||||
}
|
||||
time.Sleep(time.Millisecond)
|
||||
}
|
||||
t.Fatal("batch never reached a terminal status")
|
||||
return nil
|
||||
}
|
||||
|
||||
func TestBatchSubmitAndCollect(t *testing.T) {
|
||||
g := &guard.Guard{AllowPrefixes: []netip.Prefix{netip.MustParsePrefix("127.0.0.0/8")}}
|
||||
ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
|
||||
w.Header().Set("Content-Type", "text/html; charset=utf-8")
|
||||
_, _ = w.Write([]byte(`<html><head><title>T</title></head><body><article><p>hello batch</p></article></body></html>`))
|
||||
}))
|
||||
defer ts.Close()
|
||||
s, keys := newBatchServer(t, g, "test")
|
||||
rec := submitBatch(t, s, keys["test"], []map[string]string{{"url": ts.URL}, {"url": ts.URL + "/2"}})
|
||||
if rec.Code != 202 {
|
||||
t.Fatalf("submit %d %s", rec.Code, rec.Body.String())
|
||||
}
|
||||
var sub map[string]any
|
||||
if err := json.Unmarshal(rec.Body.Bytes(), &sub); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if sub["success"] != true {
|
||||
t.Fatalf("success %v", sub["success"])
|
||||
}
|
||||
id, _ := sub["job_id"].(string)
|
||||
if id == "" {
|
||||
t.Fatal("missing job_id")
|
||||
}
|
||||
if reused, _ := sub["reused"].(bool); reused {
|
||||
t.Fatal("first submit must not be reused")
|
||||
}
|
||||
env := waitBatchTerminal(t, s, keys["test"], id)
|
||||
if env["status"] != "done" {
|
||||
t.Fatalf("status %v", env["status"])
|
||||
}
|
||||
targets, _ := env["targets"].([]any)
|
||||
if len(targets) != 2 {
|
||||
t.Fatalf("targets %d", len(targets))
|
||||
}
|
||||
first := targets[0].(map[string]any)
|
||||
if first["status"] != "done" {
|
||||
t.Fatalf("target0 %v", first["status"])
|
||||
}
|
||||
res, _ := first["result"].(map[string]any)
|
||||
if res == nil || !strings.Contains(res["content"].(string), "hello batch") {
|
||||
t.Fatalf("target0 result %v", res)
|
||||
}
|
||||
}
|
||||
|
||||
func TestBatchIdempotentResubmit(t *testing.T) {
|
||||
g := &guard.Guard{AllowPrefixes: []netip.Prefix{netip.MustParsePrefix("127.0.0.0/8")}}
|
||||
ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
|
||||
w.Header().Set("Content-Type", "text/plain")
|
||||
_, _ = w.Write([]byte("ok"))
|
||||
}))
|
||||
defer ts.Close()
|
||||
s, keys := newBatchServer(t, g, "test")
|
||||
targets := []map[string]string{{"url": ts.URL}}
|
||||
first := submitBatch(t, s, keys["test"], targets)
|
||||
second := submitBatch(t, s, keys["test"], targets)
|
||||
var a, b map[string]any
|
||||
_ = json.Unmarshal(first.Body.Bytes(), &a)
|
||||
_ = json.Unmarshal(second.Body.Bytes(), &b)
|
||||
if a["job_id"] != b["job_id"] {
|
||||
t.Fatalf("ids differ: %v vs %v", a["job_id"], b["job_id"])
|
||||
}
|
||||
if reused, _ := b["reused"].(bool); !reused {
|
||||
t.Fatal("resubmit must be reused=true")
|
||||
}
|
||||
}
|
||||
|
||||
func TestBatchGuardRefusedUpFront(t *testing.T) {
|
||||
s, keys := newBatchServer(t, nil, "test")
|
||||
rec := submitBatch(t, s, keys["test"], []map[string]string{{"url": "http://example.com:8080/"}})
|
||||
if rec.Code != 422 {
|
||||
t.Fatalf("want 422, got %d %s", rec.Code, rec.Body.String())
|
||||
}
|
||||
if !strings.Contains(rec.Body.String(), "blocked") {
|
||||
t.Fatalf("body %s", rec.Body.String())
|
||||
}
|
||||
}
|
||||
|
||||
func TestBatchTooManyTargets(t *testing.T) {
|
||||
s, keys := newBatchServer(t, nil, "test")
|
||||
var targets []map[string]string
|
||||
for i := 0; i < 33; i++ {
|
||||
targets = append(targets, map[string]string{"url": "https://example.com/"})
|
||||
}
|
||||
rec := submitBatch(t, s, keys["test"], targets)
|
||||
if rec.Code != 422 {
|
||||
t.Fatalf("want 422, got %d %s", rec.Code, rec.Body.String())
|
||||
}
|
||||
}
|
||||
|
||||
func TestBatchPartialFailure(t *testing.T) {
|
||||
g := &guard.Guard{AllowPrefixes: []netip.Prefix{netip.MustParsePrefix("127.0.0.0/8")}}
|
||||
ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
|
||||
w.Header().Set("Content-Type", "text/plain")
|
||||
_, _ = w.Write([]byte("ok"))
|
||||
}))
|
||||
defer ts.Close()
|
||||
s, keys := newBatchServer(t, g, "test")
|
||||
// target 1 is a private address: passes the submit pre-check (valid URL) but
|
||||
// is blocked by the connect-time guard, surfacing as a per-target failure.
|
||||
rec := submitBatch(t, s, keys["test"], []map[string]string{{"url": ts.URL}, {"url": "http://10.0.0.5/"}})
|
||||
if rec.Code != 202 {
|
||||
t.Fatalf("submit %d %s", rec.Code, rec.Body.String())
|
||||
}
|
||||
var sub map[string]any
|
||||
_ = json.Unmarshal(rec.Body.Bytes(), &sub)
|
||||
id := sub["job_id"].(string)
|
||||
env := waitBatchTerminal(t, s, keys["test"], id)
|
||||
if env["status"] != "partial" {
|
||||
t.Fatalf("status %v", env["status"])
|
||||
}
|
||||
targets := env["targets"].([]any)
|
||||
second := targets[1].(map[string]any)
|
||||
if second["status"] != "failed" {
|
||||
t.Fatalf("target1 status %v", second["status"])
|
||||
}
|
||||
if err, _ := second["error"].(string); !strings.Contains(err, "blocked") {
|
||||
t.Fatalf("target1 error %q", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestBatchOwnership404(t *testing.T) {
|
||||
s, keys := newBatchServer(t, nil, "test", "other")
|
||||
rec := submitBatch(t, s, keys["test"], []map[string]string{{"url": "https://example.com/"}})
|
||||
if rec.Code != 202 {
|
||||
t.Fatalf("submit %d", rec.Code)
|
||||
}
|
||||
var sub map[string]any
|
||||
_ = json.Unmarshal(rec.Body.Bytes(), &sub)
|
||||
id := sub["job_id"].(string)
|
||||
if got := getBatch(t, s, keys["other"], id); got.Code != 404 {
|
||||
t.Fatalf("other caller GET: %d (must not reveal the job)", got.Code)
|
||||
}
|
||||
if got := getBatch(t, s, keys["test"], id); got.Code != 200 {
|
||||
t.Fatalf("owner GET: %d", got.Code)
|
||||
}
|
||||
}
|
||||
|
||||
func TestBatchCancel(t *testing.T) {
|
||||
g := &guard.Guard{AllowPrefixes: []netip.Prefix{netip.MustParsePrefix("127.0.0.0/8")}}
|
||||
s, keys := newBatchServer(t, g, "test")
|
||||
rec := submitBatch(t, s, keys["test"], []map[string]string{{"url": "https://example.com/"}})
|
||||
var sub map[string]any
|
||||
_ = json.Unmarshal(rec.Body.Bytes(), &sub)
|
||||
id := sub["job_id"].(string)
|
||||
req := httptest.NewRequest(http.MethodDelete, "/v1/batch/"+id, nil)
|
||||
req.Header.Set("Authorization", "Bearer "+keys["test"])
|
||||
drec := httptest.NewRecorder()
|
||||
s.ServeHTTP(drec, req)
|
||||
if drec.Code != 200 {
|
||||
t.Fatalf("cancel %d %s", drec.Code, drec.Body.String())
|
||||
}
|
||||
env := waitBatchTerminal(t, s, keys["test"], id)
|
||||
if env["status"] != "cancelled" {
|
||||
t.Fatalf("status %v", env["status"])
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,234 @@
|
||||
package batch
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"io"
|
||||
"log/slog"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"git.wntrmute.dev/kyle/sift/internal/dcache"
|
||||
)
|
||||
|
||||
func newRunner(t *testing.T, fetch FetchFunc, cfg Config) *Runner {
|
||||
t.Helper()
|
||||
db, err := dcache.Open("", 1<<20, slog.New(slog.NewTextHandler(io.Discard, nil)))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
t.Cleanup(func() { _ = db.Close() })
|
||||
return New(db, fetch, cfg, slog.New(slog.NewTextHandler(io.Discard, nil)))
|
||||
}
|
||||
|
||||
func waitStatus(t *testing.T, r *Runner, id, want string) {
|
||||
t.Helper()
|
||||
deadline := time.Now().Add(3 * time.Second)
|
||||
for time.Now().Before(deadline) {
|
||||
j, _, ok := r.Status(id)
|
||||
if ok && j.Status == want {
|
||||
return
|
||||
}
|
||||
time.Sleep(time.Millisecond)
|
||||
}
|
||||
j, _, _ := r.Status(id)
|
||||
t.Fatalf("job %s never reached %q (now %q)", id, want, j.Status)
|
||||
}
|
||||
|
||||
func TestSerialOrder(t *testing.T) {
|
||||
var mu sync.Mutex
|
||||
var calls []string
|
||||
inflight, peak := 0, 0
|
||||
fetch := func(_ context.Context, tgt Target) (Result, error) {
|
||||
mu.Lock()
|
||||
inflight++
|
||||
if inflight > peak {
|
||||
peak = inflight
|
||||
}
|
||||
calls = append(calls, tgt.URL)
|
||||
mu.Unlock()
|
||||
time.Sleep(time.Millisecond)
|
||||
mu.Lock()
|
||||
inflight--
|
||||
mu.Unlock()
|
||||
return Result{Content: "ok " + tgt.URL}, nil
|
||||
}
|
||||
r := newRunner(t, fetch, Config{})
|
||||
id, reused, err := r.Submit(context.Background(), "alice", "fleet", []Target{
|
||||
{URL: "https://a.example/"}, {URL: "https://b.example/"}, {URL: "https://c.example/"},
|
||||
})
|
||||
if err != nil || reused {
|
||||
t.Fatalf("submit err=%v reused=%v", err, reused)
|
||||
}
|
||||
waitStatus(t, r, id, "done")
|
||||
mu.Lock()
|
||||
order := append([]string(nil), calls...)
|
||||
p := peak
|
||||
mu.Unlock()
|
||||
if len(order) != 3 || order[0] != "https://a.example/" || order[1] != "https://b.example/" || order[2] != "https://c.example/" {
|
||||
t.Fatalf("order %v", order)
|
||||
}
|
||||
if p != 1 {
|
||||
t.Fatalf("peak inflight %d, want 1 (serial)", p)
|
||||
}
|
||||
}
|
||||
|
||||
func TestFailureIsolation(t *testing.T) {
|
||||
fetch := func(_ context.Context, tgt Target) (Result, error) {
|
||||
if tgt.URL == "https://b.example/" {
|
||||
return Result{}, errors.New("upstream 500")
|
||||
}
|
||||
return Result{Content: "ok " + tgt.URL}, nil
|
||||
}
|
||||
r := newRunner(t, fetch, Config{})
|
||||
id, _, err := r.Submit(context.Background(), "alice", "fleet", []Target{
|
||||
{URL: "https://a.example/"}, {URL: "https://b.example/"}, {URL: "https://c.example/"},
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
waitStatus(t, r, id, "partial")
|
||||
j, ts, ok := r.Status(id)
|
||||
if !ok {
|
||||
t.Fatal("no job")
|
||||
}
|
||||
if j.Status != "partial" || j.TargetsDone != 3 {
|
||||
t.Fatalf("job %+v", j)
|
||||
}
|
||||
if ts[0].Status != "done" || ts[1].Status != "failed" || ts[2].Status != "done" {
|
||||
t.Fatalf("statuses %+v", ts)
|
||||
}
|
||||
if ts[1].Error == "" {
|
||||
t.Fatal("failed target missing error")
|
||||
}
|
||||
if ts[0].Result == nil || ts[0].Result.Content != "ok https://a.example/" {
|
||||
t.Fatalf("result %+v", ts[0].Result)
|
||||
}
|
||||
}
|
||||
|
||||
func TestIdempotentResubmit(t *testing.T) {
|
||||
var mu sync.Mutex
|
||||
n := 0
|
||||
fetch := func(_ context.Context, tgt Target) (Result, error) {
|
||||
mu.Lock()
|
||||
n++
|
||||
mu.Unlock()
|
||||
return Result{Content: "ok"}, nil
|
||||
}
|
||||
r := newRunner(t, fetch, Config{})
|
||||
targets := []Target{{URL: "https://a.example/"}, {URL: "https://b.example/"}}
|
||||
id1, reused1, err := r.Submit(context.Background(), "alice", "fleet", targets)
|
||||
if err != nil || reused1 {
|
||||
t.Fatalf("first submit err=%v reused=%v", err, reused1)
|
||||
}
|
||||
id2, reused2, err := r.Submit(context.Background(), "alice", "fleet", targets)
|
||||
if err != nil || !reused2 || id2 != id1 {
|
||||
t.Fatalf("resubmit id=%q reused=%v err=%v", id2, reused2, err)
|
||||
}
|
||||
waitStatus(t, r, id1, "done")
|
||||
mu.Lock()
|
||||
calls := n
|
||||
mu.Unlock()
|
||||
if calls != 2 {
|
||||
t.Fatalf("fetched %d times, want 2 (idempotent)", calls)
|
||||
}
|
||||
}
|
||||
|
||||
func TestTooManyTargets(t *testing.T) {
|
||||
fetch := func(_ context.Context, tgt Target) (Result, error) { return Result{Content: "ok"}, nil }
|
||||
r := newRunner(t, fetch, Config{TargetsPerJob: 2})
|
||||
three := []Target{{URL: "https://a.example/"}, {URL: "https://b.example/"}, {URL: "https://c.example/"}}
|
||||
if _, _, err := r.Submit(context.Background(), "alice", "fleet", three); !errors.Is(err, ErrTooManyTargets) {
|
||||
t.Fatalf("want ErrTooManyTargets, got %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestCallerBusy(t *testing.T) {
|
||||
release := make(chan struct{})
|
||||
fetch := func(_ context.Context, tgt Target) (Result, error) {
|
||||
<-release // hold the job in the running state
|
||||
return Result{Content: "ok"}, nil
|
||||
}
|
||||
r := newRunner(t, fetch, Config{})
|
||||
id, _, err := r.Submit(context.Background(), "alice", "fleet", []Target{{URL: "https://a.example/"}})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
waitStatus(t, r, id, "running")
|
||||
if _, _, err := r.Submit(context.Background(), "alice", "fleet", []Target{{URL: "https://b.example/"}}); !errors.Is(err, ErrCallerBusy) {
|
||||
t.Fatalf("want ErrCallerBusy, got %v", err)
|
||||
}
|
||||
close(release)
|
||||
}
|
||||
|
||||
func TestCancel(t *testing.T) {
|
||||
started := make(chan struct{}, 1)
|
||||
release := make(chan struct{})
|
||||
var mu sync.Mutex
|
||||
var calls []string
|
||||
fetch := func(_ context.Context, tgt Target) (Result, error) {
|
||||
mu.Lock()
|
||||
calls = append(calls, tgt.URL)
|
||||
mu.Unlock()
|
||||
select {
|
||||
case started <- struct{}{}:
|
||||
default:
|
||||
}
|
||||
<-release
|
||||
return Result{Content: "ok"}, nil
|
||||
}
|
||||
r := newRunner(t, fetch, Config{})
|
||||
id, _, err := r.Submit(context.Background(), "alice", "fleet", []Target{
|
||||
{URL: "https://a.example/"}, {URL: "https://b.example/"}, {URL: "https://c.example/"},
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
<-started // first target is in flight
|
||||
if !r.Cancel(id) {
|
||||
t.Fatal("cancel returned false")
|
||||
}
|
||||
close(release) // let the in-flight target finish
|
||||
waitStatus(t, r, id, "cancelled")
|
||||
mu.Lock()
|
||||
n := len(calls)
|
||||
mu.Unlock()
|
||||
if n != 1 {
|
||||
t.Fatalf("cancelled after %d fetches, want 1", n)
|
||||
}
|
||||
_, ts, ok := r.Status(id)
|
||||
if !ok {
|
||||
t.Fatal("no job")
|
||||
}
|
||||
if ts[0].Status != "done" {
|
||||
t.Fatalf("in-flight target should be stored done: %+v", ts[0])
|
||||
}
|
||||
if ts[1].Status != "pending" || ts[2].Status != "pending" {
|
||||
t.Fatalf("later targets must stay pending: %+v", ts[1:])
|
||||
}
|
||||
}
|
||||
|
||||
func TestPanicRecovered(t *testing.T) {
|
||||
fetch := func(_ context.Context, tgt Target) (Result, error) {
|
||||
if tgt.URL == "https://b.example/" {
|
||||
panic("boom")
|
||||
}
|
||||
return Result{Content: "ok"}, nil
|
||||
}
|
||||
r := newRunner(t, fetch, Config{})
|
||||
id, _, err := r.Submit(context.Background(), "alice", "fleet", []Target{
|
||||
{URL: "https://a.example/"}, {URL: "https://b.example/"}, {URL: "https://c.example/"},
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
waitStatus(t, r, id, "partial")
|
||||
_, ts, _ := r.Status(id)
|
||||
if ts[0].Status != "done" || ts[1].Status != "failed" || ts[2].Status != "done" {
|
||||
t.Fatalf("statuses %+v", ts)
|
||||
}
|
||||
if ts[1].Error == "" {
|
||||
t.Fatal("panic target missing error")
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,34 @@
|
||||
package main
|
||||
|
||||
import "testing"
|
||||
|
||||
func TestParseBatchArgs(t *testing.T) {
|
||||
cases := []struct {
|
||||
args []string
|
||||
verb string
|
||||
arg string
|
||||
ok bool
|
||||
}{
|
||||
{[]string{"submit", "targets.json"}, "submit", "targets.json", true},
|
||||
{[]string{"submit", "-"}, "submit", "-", true},
|
||||
{[]string{"status", "job-123"}, "status", "job-123", true},
|
||||
{[]string{"cancel", "job-123"}, "cancel", "job-123", true},
|
||||
{[]string{}, "", "", false},
|
||||
{[]string{"status"}, "", "", false},
|
||||
{[]string{"status", "a", "b"}, "", "", false},
|
||||
{[]string{"bogus", "x"}, "", "", false},
|
||||
}
|
||||
for _, c := range cases {
|
||||
verb, arg, err := parseBatchArgs(c.args)
|
||||
if c.ok {
|
||||
if err != nil {
|
||||
t.Fatalf("%v: unexpected error %v", c.args, err)
|
||||
}
|
||||
if verb != c.verb || arg != c.arg {
|
||||
t.Fatalf("%v: got verb=%q arg=%q", c.args, verb, arg)
|
||||
}
|
||||
} else if err == nil {
|
||||
t.Fatalf("%v: want error", c.args)
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,240 @@
|
||||
package dcache
|
||||
|
||||
import (
|
||||
"io"
|
||||
"log/slog"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
func newBatchDB(t *testing.T) *DB {
|
||||
t.Helper()
|
||||
db, err := Open("", 1<<20, slog.New(slog.NewTextHandler(io.Discard, nil)))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
t.Cleanup(func() { _ = db.Close() })
|
||||
return db
|
||||
}
|
||||
|
||||
func strp(s string) *string { return &s }
|
||||
|
||||
func TestBatchJobRoundTrip(t *testing.T) {
|
||||
db := newBatchDB(t)
|
||||
j := BatchJob{
|
||||
ID: "j1", Caller: "alice", CallerClass: ClassFleet, IDemKey: "k1",
|
||||
Status: "queued", TargetsTotal: 3, CreatedAt: 100, Lease: 100,
|
||||
}
|
||||
if err := db.CreateBatchJob(j); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
got, ok := db.BatchJob("j1")
|
||||
if !ok {
|
||||
t.Fatal("missing job")
|
||||
}
|
||||
if got.ID != j.ID || got.Caller != "alice" || got.CallerClass != ClassFleet ||
|
||||
got.Status != "queued" || got.TargetsTotal != 3 || got.Lease != 100 || got.CreatedAt != 100 {
|
||||
t.Fatalf("round trip mismatch: %+v", got)
|
||||
}
|
||||
if _, ok := db.BatchJob("nope"); ok {
|
||||
t.Fatal("unknown id must be ok=false")
|
||||
}
|
||||
}
|
||||
|
||||
func TestBatchIDemLookup(t *testing.T) {
|
||||
db := newBatchDB(t)
|
||||
if err := db.CreateBatchJob(BatchJob{ID: "j1", Caller: "a", CallerClass: ClassFleet, IDemKey: "k", Status: "running", TargetsTotal: 1, CreatedAt: 100, Lease: 1000}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if id, ok := db.BatchJobByIDem("a", "k", 2000, time.Hour); !ok || id != "j1" {
|
||||
t.Fatalf("live idem: %q %v", id, ok)
|
||||
}
|
||||
if _, ok := db.BatchJobByIDem("a", "k", 5000, time.Hour); ok {
|
||||
t.Fatal("lease-expired job must not match")
|
||||
}
|
||||
if _, ok := db.BatchJobByIDem("b", "k", 2000, time.Hour); ok {
|
||||
t.Fatal("other caller must not match")
|
||||
}
|
||||
}
|
||||
|
||||
func TestBatchLiveCounts(t *testing.T) {
|
||||
db := newBatchDB(t)
|
||||
mk := func(id, caller, status string) {
|
||||
if err := db.CreateBatchJob(BatchJob{ID: id, Caller: caller, CallerClass: ClassFleet, IDemKey: id, Status: status, TargetsTotal: 1, CreatedAt: 1, Lease: 1}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
mk("j1", "a", "queued")
|
||||
mk("j2", "a", "running")
|
||||
mk("j3", "a", "done")
|
||||
mk("j4", "b", "running")
|
||||
if n := db.LiveBatchJobs("a"); n != 2 {
|
||||
t.Fatalf("caller a live=%d want 2", n)
|
||||
}
|
||||
if n := db.LiveBatchJobsTotal(); n != 3 {
|
||||
t.Fatalf("total live=%d want 3", n)
|
||||
}
|
||||
}
|
||||
|
||||
func TestBatchTargetsOrderedAndUpdate(t *testing.T) {
|
||||
db := newBatchDB(t)
|
||||
must := func(err error) {
|
||||
t.Helper()
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
must(db.CreateBatchJob(BatchJob{ID: "j1", Caller: "a", CallerClass: ClassFleet, IDemKey: "k", Status: "running", TargetsTotal: 3, CreatedAt: 100, Lease: 100}))
|
||||
must(db.PutBatchTargets("j1", []BatchTarget{
|
||||
{JobID: "j1", Idx: 0, URL: "https://a.example/", Options: `{"mode":"clean"}`, Status: "pending"},
|
||||
{JobID: "j1", Idx: 1, URL: "https://b.example/", Status: "pending"},
|
||||
{JobID: "j1", Idx: 2, URL: "https://c.example/", Status: "pending"},
|
||||
}))
|
||||
got, ok := db.BatchTargets("j1")
|
||||
if !ok || len(got) != 3 {
|
||||
t.Fatalf("targets len=%d ok=%v", len(got), ok)
|
||||
}
|
||||
for i := range got {
|
||||
if got[i].Idx != i {
|
||||
t.Fatalf("order broken: got[%d].Idx=%d", i, got[i].Idx)
|
||||
}
|
||||
}
|
||||
|
||||
// A done write bumps the lease to now and counts the target once.
|
||||
res := strp(`{"content":"ok"}`)
|
||||
must(db.UpdateBatchTarget("j1", 0, 200, "done", res, nil, nil))
|
||||
j, _ := db.BatchJob("j1")
|
||||
if j.Lease != 200 || j.TargetsDone != 1 {
|
||||
t.Fatalf("after first write lease=%d done=%d", j.Lease, j.TargetsDone)
|
||||
}
|
||||
// Re-writing the same target (still a done transition) must not double-count.
|
||||
must(db.UpdateBatchTarget("j1", 0, 201, "done", res, nil, nil))
|
||||
j, _ = db.BatchJob("j1")
|
||||
if j.TargetsDone != 1 {
|
||||
t.Fatalf("double-counted: done=%d", j.TargetsDone)
|
||||
}
|
||||
// A failed write counts too.
|
||||
must(db.UpdateBatchTarget("j1", 1, 202, "failed", nil, strp("upstream 500"), strp("browser")))
|
||||
j, _ = db.BatchJob("j1")
|
||||
if j.TargetsDone != 2 || j.Lease != 202 {
|
||||
t.Fatalf("after failed write lease=%d done=%d", j.Lease, j.TargetsDone)
|
||||
}
|
||||
}
|
||||
|
||||
func TestBatchTouchAndStatus(t *testing.T) {
|
||||
db := newBatchDB(t)
|
||||
if err := db.CreateBatchJob(BatchJob{ID: "j1", Caller: "a", CallerClass: ClassFleet, IDemKey: "k", Status: "running", TargetsTotal: 1, CreatedAt: 100, Lease: 100}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := db.TouchBatchJob("j1", 500); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if j, _ := db.BatchJob("j1"); j.Lease != 500 {
|
||||
t.Fatalf("touch lease=%d want 500", j.Lease)
|
||||
}
|
||||
if err := db.SetBatchJobStatus("j1", "cancelled", "", 600); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if j, _ := db.BatchJob("j1"); j.Status != "cancelled" || j.Lease != 600 {
|
||||
t.Fatalf("status=%q lease=%d", j.Status, j.Lease)
|
||||
}
|
||||
}
|
||||
|
||||
func TestSweepBatch(t *testing.T) {
|
||||
db := newBatchDB(t)
|
||||
mk := func(id string, lease int64) {
|
||||
if err := db.CreateBatchJob(BatchJob{ID: id, Caller: "a", CallerClass: ClassFleet, IDemKey: id, Status: "done", TargetsTotal: 1, CreatedAt: 1, Lease: lease}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := db.PutBatchTargets(id, []BatchTarget{{JobID: id, Idx: 0, URL: "https://x.example/", Status: "done", Result: strp("x")}}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
mk("old", 100)
|
||||
mk("new", 1000)
|
||||
n := db.SweepBatch(300, 100*time.Second) // old: 100+100=200<=300 expired; new: 1100>300 kept
|
||||
if n != 1 {
|
||||
t.Fatalf("swept %d want 1", n)
|
||||
}
|
||||
if _, ok := db.BatchJob("old"); ok {
|
||||
t.Fatal("old job must be gone")
|
||||
}
|
||||
if _, ok := db.BatchJob("new"); !ok {
|
||||
t.Fatal("new job must remain")
|
||||
}
|
||||
if _, ok := db.BatchTargets("old"); ok {
|
||||
t.Fatal("old targets must be gone")
|
||||
}
|
||||
}
|
||||
|
||||
func TestEvictBatchOldest(t *testing.T) {
|
||||
db := newBatchDB(t)
|
||||
done := func(id string, lease int64, size int) {
|
||||
if err := db.CreateBatchJob(BatchJob{ID: id, Caller: "a", CallerClass: ClassFleet, IDemKey: id, Status: "done", TargetsTotal: 1, CreatedAt: 1, Lease: lease}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := db.PutBatchTargets(id, []BatchTarget{{JobID: id, Idx: 0, URL: "https://x.example/", Status: "pending"}}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := db.UpdateBatchTarget(id, 0, lease, "done", strp(strings.Repeat("x", size)), nil, nil); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
done("old1", 100, 100)
|
||||
done("old2", 200, 200)
|
||||
done("new", 300, 300)
|
||||
|
||||
// A running job with one completed target must never be byte-evicted.
|
||||
if err := db.CreateBatchJob(BatchJob{ID: "run", Caller: "a", CallerClass: ClassFleet, IDemKey: "run", Status: "running", TargetsTotal: 2, TargetsDone: 1, CreatedAt: 1, Lease: 400}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := db.PutBatchTargets("run", []BatchTarget{
|
||||
{JobID: "run", Idx: 0, URL: "https://x.example/", Status: "pending"},
|
||||
{JobID: "run", Idx: 1, URL: "https://y.example/", Status: "pending"},
|
||||
}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := db.UpdateBatchTarget("run", 0, 400, "done", strp(strings.Repeat("x", 500)), nil, nil); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
// total nbytes = 100+200+300+500 = 1100. Evict oldest done until <= 250:
|
||||
// old1(100), old2(200), new(300) all evicted (oldest first); run stays (live).
|
||||
db.EvictBatchOldest(250)
|
||||
for _, id := range []string{"old1", "old2", "new"} {
|
||||
if _, ok := db.BatchJob(id); ok {
|
||||
t.Fatalf("%s should be byte-evicted", id)
|
||||
}
|
||||
}
|
||||
if _, ok := db.BatchJob("run"); !ok {
|
||||
t.Fatal("running job must not be byte-evicted")
|
||||
}
|
||||
if n := db.BatchNbytes(); n != 500 {
|
||||
t.Fatalf("BatchNbytes=%d want 500", n)
|
||||
}
|
||||
}
|
||||
|
||||
func TestMarkInFlightFailed(t *testing.T) {
|
||||
db := newBatchDB(t)
|
||||
mk := func(id, status string) {
|
||||
if err := db.CreateBatchJob(BatchJob{ID: id, Caller: "a", CallerClass: ClassFleet, IDemKey: id, Status: status, TargetsTotal: 1, CreatedAt: 1, Lease: 1}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
mk("q", "queued")
|
||||
mk("r", "running")
|
||||
mk("d", "done")
|
||||
if n := db.MarkInFlightFailed("restart", 999); n != 2 {
|
||||
t.Fatalf("marked %d want 2", n)
|
||||
}
|
||||
if j, _ := db.BatchJob("q"); j.Status != "failed" || j.Error != "restart" || j.Lease != 999 {
|
||||
t.Fatalf("q %+v", j)
|
||||
}
|
||||
if j, _ := db.BatchJob("r"); j.Status != "failed" || j.Error != "restart" {
|
||||
t.Fatalf("r %+v", j)
|
||||
}
|
||||
if j, _ := db.BatchJob("d"); j.Status != "done" {
|
||||
t.Fatalf("done job changed: %+v", j)
|
||||
}
|
||||
}
|
||||
@@ -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")
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user