diff --git a/docs/implementer-log.md b/docs/implementer-log.md index 16658ac..12b81fd 100644 --- a/docs/implementer-log.md +++ b/docs/implementer-log.md @@ -17,7 +17,8 @@ One row per task session that ends in a commit. Newest last. | p5b/03-api | 2026-09-26 | done | 1 | gate: ok | none | The digest now answers long documents itself (task 02), so the API's refusal limit is only a safety cap: `DefaultPDFMaxQuestionChunks` 200→5000 in `internal/api/pdf.go` (the 422 message text unchanged, still "document too long for question (N chunks); use grep or pages"). In `v1fetch` (`internal/api/api.go`) the digest-meta map gains `"retrieval": dr.Retrieval` and `"embedded": dr.Embedded` next to `chunks_used`/`chunks_total`, only when the digest ran without a fallback. Copied the two given tests byte-identical from the plan — the replacement `pdf_test.go` is now the protected one and asserts `DefaultPDFMaxQuestionChunks == 5000`; `pdf_retrieval_test.go` asserts the first long-doc question returns `retrieval=bm25`, `embedded=0`. README PDF section: a long-doc question is answered first from BM25 (`retrieval: "bm25"`) while its chunks embed in the background and hybrid on later questions, plus the 5000-chunk cap. One sentence added to `skill/SKILL.md`. All api tests pass under `-race -count=3`; make gate ok; given files byte-identical. | | p5c/01-extract-api | 2026-09-26 | done | 1 | gate: ok | none | `POST /v1/extract` answers a PDF posted in the body like `/v1/fetch` answers the same PDF by URL, nothing fetched. `internal/api/extract.go`: auth via `caller`; `io.ReadAll` of the body (a `*http.MaxBytesError`→413 "too large (>N MB)", empty→400); `clean.Kind` must be "pdf" else 415; query `pages`/`grep`/`question`/`mode` (digest when a question is set, else clean; other→400). Content-addressed cache: `hash="sha256:"+HashBody(body)`, `key=PDFKey("upload:"+hash, callerClass)`; a `GetPDF` hit rebuilds the doc via `pdfFromEntry` with `cached:true` + `PDFCacheHit` and no child (bypasses the TTL/revalidation that `pdfFromCache` does, so identical content is extracted once); a miss builds a synthetic `fetch.Resource{URL:"upload",...}` and runs `extractPDF`+`storePDF`. Bad pages/grep→400 via `writeFetchErr`, the chunk limit→the same 422 as `v1fetch`, extraction errors→`pdfError` via `writeFetchErr`. Envelope mirrors the PDF fetch path (`content`/`mode`/`untrusted`/`warnings`/`sizes`/`timings`/`pdf`/`digest`); `source` is `{"kind":"upload","hash","bytes","cached","fetched_at"}` with no `url` key; no Honcho context for uploads; one log line "extract" with caller/bytes/pdf_pages/mode/cached. `internal/api/api.go` scoped to the rule-1 changes: `DefaultPDFMaxUpload` const (100 MiB), `PDFMaxUpload int64` Server field, the `POST /v1/extract` route in `New`, and `ServeHTTP` (per-path upload cap instead of the 64 KiB JSON limit, and `/v1/extract` added to the `requestDeadline` switch). The `pdfMaxUpload()` field-or-default resolver lives in extract.go. Given test byte-identical; all api tests pass under `-race -count=3`; make gate ok. | | p5c/02-extract-cli | 2026-09-26 | done | 1 | gate: ok | none | `sift extract ` uploads a local PDF to `/v1/extract` and prints the answer exactly as `sift fetch` does. `cmd/sift/extractcmd.go`: `extractOpts` + `parseExtractOpts` (reuses `splitAroundPositional` with `extractValueFlags` mode/pages/grep/question, one positional file, `--json`), `extractURL` builds `base+"/v1/extract"` with `url.Values` set only for the options that are set and `mode=digest` when a question is given and no mode — values are URL-encoded so a grep like `a b&c` survives round-trip; `extractMain` reuses `loadClient`, `stat`s the file and refuses >100 MiB before uploading (exit 3, matching the server's 413→exitFor), then `http.NewRequest` with the `*os.File` as the streaming body and `ContentLength` set (no second in-memory copy), `Content-Type: application/pdf` + bearer, and the same error/exitFor path as fetch; success prints `content` (or the `--json` envelope). `main.go`: `case "extract"` and one help line. README: extract line in the CLI block plus a sentence; SKILL.md: a PDF-on-disk bullet in the PDFs section. Given test byte-identical; all cmd/sift tests pass under `-race -count=3`; make gate ok; `bin/sift extract` prints the usage line. | -| 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/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. | | ## P5a review (owner, 2026-09-26) diff --git a/internal/batch/batch.go b/internal/batch/batch.go new file mode 100644 index 0000000..51ed5d5 --- /dev/null +++ b/internal/batch/batch.go @@ -0,0 +1,330 @@ +// Package batch is the batch runner core: it owns job lifecycle, idempotency and +// the per-caller / global caps. It knows nothing about HTTP, the guard or the real +// fetch pipeline — per-target work is an injected FetchFunc, and persistence is the +// dcache store. Jobs never resume after a restart, so the caller is held in memory, +// never written to the store. +package batch + +import ( + "context" + "crypto/rand" + "crypto/sha256" + "encoding/hex" + "encoding/json" + "errors" + "fmt" + "log/slog" + "sync" + "time" + + "git.wntrmute.dev/kyle/sift/internal/auth" + "git.wntrmute.dev/kyle/sift/internal/dcache" +) + +type Options struct { + Mode string `json:"mode,omitempty"` + Question string `json:"question,omitempty"` + Pages string `json:"pages,omitempty"` + Grep string `json:"grep,omitempty"` + JQ string `json:"jq,omitempty"` + Context *bool `json:"context,omitempty"` // nil = /v1/fetch's default (on when there is a question) + Recover bool `json:"recover,omitempty"` +} + +type Target struct { + URL string `json:"url"` + Options Options `json:"options"` +} + +// Result is a finished target: the /v1/fetch envelope's content, and every other envelope field +// in Meta. It is stored JSON-encoded in the target's result column. +type Result struct { + Content string `json:"content"` + Meta map[string]any `json:"meta,omitempty"` +} + +// TargetError is a failure with an agent-facing hint. Status is the HTTP status the same +// failure gets from /v1/fetch; the runner ignores it (the metrics hook reads it). Any other +// error is stored with its Error() text and no hint. +type TargetError struct { + Status int + Msg, Hint string +} + +func (e *TargetError) Error() string { return e.Msg } + +// FetchFunc does one target for the caller that submitted the job. +type FetchFunc func(ctx context.Context, c *auth.Caller, t Target) (Result, error) + +type Job struct { + ID string + Caller string + Status string // queued | running | done | partial | failed | cancelled + Error string + TargetsTotal int + TargetsDone int + CreatedAt int64 // unix seconds +} + +type TargetStatus struct { + Idx int + URL string + Status string // pending | running | done | failed + Result *Result + Error string + Hint string +} + +// Hooks observe the runner for metrics. Any nil field is skipped. +type Hooks struct { + Job func(outcome string) // created, reused, done, partial, failed, cancelled + Target func(err error, d time.Duration, resultBytes int) // once per stored target; err nil on success + JobSeconds func(d time.Duration) // submit → terminal, once per job + Depth func(live int) // live jobs, after every change + Inflight func(n int) // targets being fetched, after every change +} + +type Config struct { + TTL time.Duration // 0 → 24h + TargetsPerJob int // 0 → 32 + JobsPerCaller int // 0 → 1 + JobsTotal int // 0 → 8 + Workers int // 0 → 1; values above 2 are clamped to 2 + MaxBytes int64 // 0 → 64 MiB + RetryWait time.Duration // 0 → 2s: the pause before a busy target is tried again + RetryFor time.Duration // 0 → 10m: how long one target may keep coming back busy + Hooks Hooks +} + +var ( + ErrNoTargets = errors.New("no targets") + ErrTooManyTargets = errors.New("too many targets") + ErrCallerBusy = errors.New("busy") + ErrBusy = errors.New("jobs busy") + ErrRetryLater = errors.New("retry later") // a FetchFunc wraps this when the target was only busy +) + +// terminal statuses mark a job as finished; only a queued job can be cancelled. +func isJobTerminal(status string) bool { + switch status { + case "done", "partial", "failed", "cancelled": + return true + } + return false +} + +type Runner struct { + db *dcache.DB + fetch FetchFunc + log *slog.Logger + cfg Config + callers map[string]*auth.Caller // job id → submitting caller, held in memory only + callMu sync.Mutex + closeOnce sync.Once +} + +func New(db *dcache.DB, fetch FetchFunc, cfg Config, log *slog.Logger) *Runner { + if cfg.TTL == 0 { + cfg.TTL = 24 * time.Hour + } + if cfg.TargetsPerJob == 0 { + cfg.TargetsPerJob = 32 + } + if cfg.JobsPerCaller == 0 { + cfg.JobsPerCaller = 1 + } + if cfg.JobsTotal == 0 { + cfg.JobsTotal = 8 + } + if cfg.Workers == 0 { + cfg.Workers = 1 + } + if cfg.Workers > 2 { + cfg.Workers = 2 + } + if cfg.MaxBytes == 0 { + cfg.MaxBytes = 64 << 20 + } + if cfg.RetryWait == 0 { + cfg.RetryWait = 2 * time.Second + } + if cfg.RetryFor == 0 { + cfg.RetryFor = 10 * time.Minute + } + return &Runner{ + db: db, + fetch: fetch, + log: log, + cfg: cfg, + callers: make(map[string]*auth.Caller), + } +} + +// Submit stores the job and its targets before it returns. The order matters: an empty +// request, then idempotency (which never touches the caps), then the caps, then the write. +func (r *Runner) Submit(ctx context.Context, c *auth.Caller, callerClass string, targets []Target) (id string, reused bool, err error) { + if len(targets) == 0 { + return "", false, ErrNoTargets + } + + raw, err := json.Marshal(targets) + if err != nil { + return "", false, fmt.Errorf("batch: marshal targets: %w", err) + } + sum := sha256.Sum256([]byte(c.Name + "\x00" + string(raw))) + idemKey := hex.EncodeToString(sum[:]) + + now := time.Now().Unix() + if existing, ok := r.db.BatchJobByIDem(c.Name, idemKey, now, r.cfg.TTL); ok { + r.callJob("reused") + return existing, true, nil + } + + if len(targets) > r.cfg.TargetsPerJob { + return "", false, ErrTooManyTargets + } + if r.db.LiveBatchJobs(c.Name) >= r.cfg.JobsPerCaller { + return "", false, ErrCallerBusy + } + if r.db.LiveBatchJobsTotal() >= r.cfg.JobsTotal { + return "", false, ErrBusy + } + + id, err = randomID() + if err != nil { + return "", false, fmt.Errorf("batch: job id: %w", err) + } + + if err := r.db.CreateBatchJob(dcache.BatchJob{ + ID: id, + Caller: c.Name, + CallerClass: callerClass, + IDemKey: idemKey, + Status: "queued", + TargetsTotal: len(targets), + CreatedAt: now, + Lease: now, + }); err != nil { + r.log.Warn("batch: create job", "job", id, "err", err) + return "", false, fmt.Errorf("batch: create job: %w", err) + } + + btargets := make([]dcache.BatchTarget, len(targets)) + for i := range targets { + opts, _ := json.Marshal(targets[i].Options) + btargets[i] = dcache.BatchTarget{ + JobID: id, + Idx: i, + URL: targets[i].URL, + Options: string(opts), + Status: "pending", + } + } + if err := r.db.PutBatchTargets(id, btargets); err != nil { + r.log.Warn("batch: store targets", "job", id, "err", err) + return "", false, fmt.Errorf("batch: store targets: %w", err) + } + + r.callMu.Lock() + r.callers[id] = c + r.callMu.Unlock() + + r.callJob("created") + r.db.SweepBatch(now, r.cfg.TTL) + return id, false, nil +} + +// Cancel sets a queued job cancelled at once; its pending targets stay pending. It is +// false for an unknown id or a job already terminal. +func (r *Runner) Cancel(id string) bool { + jb, ok := r.db.BatchJob(id) + if !ok || isJobTerminal(jb.Status) { + return false + } + if err := r.db.SetBatchJobStatus(id, "cancelled", "", time.Now().Unix()); err != nil { + r.log.Warn("batch: cancel job", "job", id, "err", err) + return false + } + r.callMu.Lock() + delete(r.callers, id) + r.callMu.Unlock() + r.callJob("cancelled") + return true +} + +// Status maps the store rows for a job and its targets; Result is decoded from the result +// column. It writes nothing. +func (r *Runner) Status(id string) (Job, []TargetStatus, bool) { + jb, ok := r.db.BatchJob(id) + if !ok { + return Job{}, nil, false + } + j := Job{ + ID: jb.ID, + Caller: jb.Caller, + Status: jb.Status, + Error: jb.Error, + TargetsTotal: jb.TargetsTotal, + TargetsDone: jb.TargetsDone, + CreatedAt: jb.CreatedAt, + } + tbs, _ := r.db.BatchTargets(id) + ts := make([]TargetStatus, 0, len(tbs)) + for _, bt := range tbs { + var res *Result + if bt.Result != nil { + res = new(Result) + if err := json.Unmarshal([]byte(*bt.Result), res); err != nil { + r.log.Warn("batch: decode result", "job", id, "err", err) + } + } + ts = append(ts, TargetStatus{ + Idx: bt.Idx, + URL: bt.URL, + Status: bt.Status, + Result: res, + Error: strOrEmpty(bt.Error), + Hint: strOrEmpty(bt.Hint), + }) + } + return j, ts, true +} + +// Touch bumps the job's lease to now so a slow owner does not lose it. +func (r *Runner) Touch(id string) { + now := time.Now().Unix() + if err := r.db.TouchBatchJob(id, now); err != nil { + r.log.Warn("batch: touch job", "job", id, "err", err) + } +} + +// Close stops the workers and waits for an in-flight fetch to return. Safe to call more than +// once. With no workers (task 02) there is nothing to stop; the in-memory callers are dropped. +func (r *Runner) Close() { + r.closeOnce.Do(func() { + r.callMu.Lock() + r.callers = nil + r.callMu.Unlock() + }) +} + +func (r *Runner) callJob(outcome string) { + if r.cfg.Hooks.Job != nil { + r.cfg.Hooks.Job(outcome) + } +} + +func randomID() (string, error) { + var b [16]byte + if _, err := rand.Read(b[:]); err != nil { + return "", err + } + return hex.EncodeToString(b[:]), nil +} + +func strOrEmpty(s *string) string { + if s == nil { + return "" + } + return *s +} diff --git a/internal/batch/runner_core_test.go b/internal/batch/runner_core_test.go new file mode 100644 index 0000000..6576aae --- /dev/null +++ b/internal/batch/runner_core_test.go @@ -0,0 +1,182 @@ +package batch + +// P6 task 02: the runner's core — types, Submit (idempotency, caps, storage), Status, Touch and +// Cancel — with no workers yet: every job stays queued. The helpers here are shared with task +// 03's runner_test.go. + +import ( + "context" + "errors" + "io" + "log/slog" + "testing" + "time" + + "git.wntrmute.dev/kyle/sift/internal/auth" + "git.wntrmute.dev/kyle/sift/internal/dcache" +) + +var ( + alice = &auth.Caller{Name: "alice", Workspace: "hermes-fleet"} + bob = &auth.Caller{Name: "bob", Workspace: "hermes-fleet"} + carol = &auth.Caller{Name: "carol", Workspace: "hermes-fleet"} +) + +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() }) + r := New(db, fetch, cfg, slog.New(slog.NewTextHandler(io.Discard, nil))) + t.Cleanup(r.Close) // runs before db.Close (cleanups are LIFO) + return r +} + +func waitStatus(t *testing.T, r *Runner, id, want string) { + t.Helper() + deadline := time.Now().Add(5 * time.Second) + for time.Now().Before(deadline) { + j, _, ok := r.Status(id) + if ok && j.Status == want { + return + } + time.Sleep(2 * time.Millisecond) + } + j, _, _ := r.Status(id) + t.Fatalf("job %s never reached %q (now %q)", id, want, j.Status) +} + +func urls(ss ...string) []Target { + var out []Target + for _, s := range ss { + out = append(out, Target{URL: "https://" + s + ".example/"}) + } + return out +} + +// Submit stores the job queued with its targets pending, before it returns. +func TestSubmitStoresQueued(t *testing.T) { + r := newRunner(t, func(context.Context, *auth.Caller, Target) (Result, error) { + return Result{Content: "ok"}, nil + }, Config{Workers: 1}) + id, reused, err := r.Submit(context.Background(), alice, dcache.ClassFleet, urls("a", "b")) + if err != nil || reused || len(id) != 32 { + t.Fatalf("submit id=%q reused=%v err=%v, want a 32-hex-char id", id, reused, err) + } + j, ts, ok := r.Status(id) + if !ok || j.ID != id || j.Caller != "alice" || j.TargetsTotal != 2 || j.CreatedAt == 0 { + t.Fatalf("job %+v ok=%v", j, ok) + } + if j.Status != "queued" && j.Status != "running" && j.Status != "done" { + t.Fatalf("status %q", j.Status) + } + if len(ts) != 2 || ts[0].Idx != 0 || ts[0].URL != "https://a.example/" || ts[1].URL != "https://b.example/" { + t.Fatalf("targets %+v", ts) + } +} + +// Cancelling a job no worker has reached: it is cancelled at once, its targets stay pending, +// and the same list then starts a new job. +func TestCancelQueued(t *testing.T) { + release := make(chan struct{}) + defer close(release) + fetch := func(context.Context, *auth.Caller, Target) (Result, error) { + <-release + return Result{Content: "ok"}, nil + } + r := newRunner(t, fetch, Config{Workers: 1}) + busy, _, err := r.Submit(context.Background(), bob, dcache.ClassFleet, urls("x")) // occupies the only worker once there is one + if err != nil { + t.Fatal(err) + } + _ = busy + id, _, err := r.Submit(context.Background(), alice, dcache.ClassFleet, urls("a", "b")) + if err != nil { + t.Fatal(err) + } + if !r.Cancel(id) { + t.Fatal("cancel returned false") + } + j, ts, _ := r.Status(id) + if j.Status != "cancelled" || ts[0].Status != "pending" || ts[1].Status != "pending" { + t.Fatalf("after cancel: %+v %+v", j, ts) + } + if r.Cancel(id) || r.Cancel("no-such-job") { + t.Fatal("cancel of a finished or unknown job must return false") + } + id2, reused, err := r.Submit(context.Background(), alice, dcache.ClassFleet, urls("a", "b")) + if err != nil || reused || id2 == id { + t.Fatalf("resubmit after cancel id=%q reused=%v err=%v", id2, reused, err) + } +} + +// Touch keeps a job alive; Status never does. +func TestTouchAndStatusLease(t *testing.T) { + db, err := dcache.Open("", 1<<20, slog.New(slog.NewTextHandler(io.Discard, nil))) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = db.Close() }) + release := make(chan struct{}) + fetch := func(context.Context, *auth.Caller, Target) (Result, error) { + <-release + return Result{Content: "ok"}, nil + } + r := New(db, fetch, Config{Workers: 1}, slog.New(slog.NewTextHandler(io.Discard, nil))) + t.Cleanup(r.Close) + defer close(release) // runs before the cleanups + id, _, err := r.Submit(context.Background(), alice, dcache.ClassFleet, urls("a")) + if err != nil { + t.Fatal(err) + } + time.Sleep(50 * time.Millisecond) // let a worker (task 03) take the target; it then blocks + before, _ := db.BatchJob(id) + time.Sleep(1100 * time.Millisecond) // the lease is in whole seconds + r.Status(id) + if j, _ := db.BatchJob(id); j.Lease != before.Lease { + t.Fatalf("Status moved the lease %d → %d; a read must not write", before.Lease, j.Lease) + } + r.Touch(id) + if j, _ := db.BatchJob(id); j.Lease <= before.Lease { + t.Fatalf("Touch did not move the lease: %d → %d", before.Lease, j.Lease) + } +} + +func TestCaps(t *testing.T) { + release := make(chan struct{}) + fetch := func(context.Context, *auth.Caller, Target) (Result, error) { + <-release + return Result{Content: "ok"}, nil + } + r := newRunner(t, fetch, Config{TargetsPerJob: 2, JobsTotal: 2}) + defer close(release) + if _, _, err := r.Submit(context.Background(), alice, dcache.ClassFleet, urls("a", "b", "c")); !errors.Is(err, ErrTooManyTargets) { + t.Fatalf("3 targets over a cap of 2: want ErrTooManyTargets, got %v", err) + } + if _, _, err := r.Submit(context.Background(), alice, dcache.ClassFleet, nil); !errors.Is(err, ErrNoTargets) { + t.Fatalf("no targets: want ErrNoTargets, got %v", err) + } + if _, _, err := r.Submit(context.Background(), alice, dcache.ClassFleet, urls("a")); err != nil { + t.Fatal(err) + } + // alice's job is live (queued or running): a second, different list is refused. + if _, _, err := r.Submit(context.Background(), alice, dcache.ClassFleet, urls("z")); !errors.Is(err, ErrCallerBusy) { + t.Fatalf("want ErrCallerBusy, got %v", err) + } + if _, _, err := r.Submit(context.Background(), bob, dcache.ClassFleet, urls("b")); err != nil { + t.Fatal(err) + } + // Two live jobs in total: a third caller is refused. + if _, _, err := r.Submit(context.Background(), carol, dcache.ClassFleet, urls("c")); !errors.Is(err, ErrBusy) { + t.Fatalf("want ErrBusy, got %v", err) + } +} + +func TestStatusUnknown(t *testing.T) { + r := newRunner(t, func(context.Context, *auth.Caller, Target) (Result, error) { return Result{}, nil }, Config{}) + if _, _, ok := r.Status("no-such-job"); ok { + t.Fatal("unknown id must be ok=false") + } +}