batch: runner core — types, Submit with idempotency and caps, Status, Touch, Cancel
New package internal/batch holding the runner core with no workers yet (task 03 adds them; until then every job stays queued, so a queued job counts as live for the caps). batch.go defines the public surface: Options/Target/Result/TargetError, FetchFunc, Job/TargetStatus, Hooks and Config with its defaults (TTL 24h, 32 targets/job, 1 job/caller, 8 total, worker clamped to 2, 64 MiB, 2 s / 10 m retry). Submit runs in the given order: empty → ErrNoTargets; idempotency via hex SHA-256 of `c.Name + NUL + json.Marshal(targets)` (a live BatchJobByIDem hit returns the id reused, touching no caps); then TargetsPerJob / JobsPerCaller / JobsTotal; then it stores the job queued (random 16-byte hex id, CreatedAt = Lease = now) and its targets (pending, Options = json of t.Options) before returning. The submitting caller is held in memory keyed by job id, never persisted (jobs never resume after a restart). Status maps the rows and decodes Result from the result column 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 or host. Co-Authored-By: ornith-1.5-35b-a3b <ornith-1.5-35b-a3b@llama.cpp.invalid>
This commit is contained in:
ornith-1.5-35b-a3b
@@ -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
|
||||
}
|
||||
@@ -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")
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user