diff --git a/cmd/sift/serve.go b/cmd/sift/serve.go index 6095982..1773304 100644 --- a/cmd/sift/serve.go +++ b/cmd/sift/serve.go @@ -10,9 +10,11 @@ import ( "os/signal" "strings" "syscall" + "time" "git.wntrmute.dev/kyle/sift/internal/api" "git.wntrmute.dev/kyle/sift/internal/auth" + "git.wntrmute.dev/kyle/sift/internal/batch" "git.wntrmute.dev/kyle/sift/internal/config" "git.wntrmute.dev/kyle/sift/internal/dcache" "git.wntrmute.dev/kyle/sift/internal/digest" @@ -119,7 +121,7 @@ func serveMain() { cachePath = "/var/lib/sift/cache.db" } var cache *dcache.DB - if cfg.Dixie.Chat != "" || cfg.Search.AnyEnabled() { + if cfg.Dixie.Chat != "" || cfg.Search.AnyEnabled() || !cfg.Batch.Disabled { c, err := dcache.Open(cachePath, 500<<20, log) if err != nil { log.Warn("cache", "err", err) @@ -127,6 +129,23 @@ func serveMain() { cache = c } } + if !cfg.Batch.Disabled && cache != nil { + now := time.Now().Unix() + if n := cache.MarkInFlightFailed("restart", now); n > 0 { + for i := 0; i < n; i++ { + srv.Prom.ObserveBatchJob("restart") + } + log.Info("batch reaped", "jobs", n) + } + srv.Batch = srv.NewBatchRunner(cache, batch.Config{ + TTL: config.ParseDuration(cfg.Batch.TTL, 0), + TargetsPerJob: cfg.Batch.TargetsPerJob, + JobsPerCaller: cfg.Batch.JobsPerCaller, + JobsTotal: cfg.Batch.JobsTotal, + Workers: cfg.Batch.Workers, + MaxBytes: config.ParseSize(cfg.Batch.MaxBytes, 0), + }) + } if cfg.Dixie.Chat != "" { srv.Digest = &digest.Engine{ ChatURL: cfg.Dixie.Chat, @@ -213,8 +232,14 @@ func serveMain() { case <-sig: stop() _ = hs.Close() + if srv.Batch != nil { + srv.Batch.Close() + } case err := <-errCh: stop() + if srv.Batch != nil { + srv.Batch.Close() + } if err != nil { log.Error("serve", "err", err) os.Exit(1) diff --git a/dist/sift.toml.example b/dist/sift.toml.example index c0b426c..6ff7801 100644 --- a/dist/sift.toml.example +++ b/dist/sift.toml.example @@ -54,3 +54,12 @@ listen = "127.0.0.1:8890" # url = "https://api.tavily.com/search" # daily_credit_cap = 500 # no_rerank = true + +# [batch] +# disabled = false +# targets_per_job = 32 +# jobs_per_caller = 1 +# jobs_total = 8 +# workers = 1 # at most 2: each worker fetches one target at a time +# ttl = "24h" # a job's results live this long after its last read or write +# max_bytes = "64MiB" # stored results; the oldest finished jobs go first diff --git a/docs/implementer-log.md b/docs/implementer-log.md index cd50411..a2823b7 100644 --- a/docs/implementer-log.md +++ b/docs/implementer-log.md @@ -24,6 +24,7 @@ One row per task session that ends in a commit. Newest last. | 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. | +| p6/08-serve-config | 2026-09-26 | done | 1 | gate: ok | none | `[batch]` config in `internal/config`: the `Batch` type (disabled/targets_per_job/jobs_per_caller/jobs_total/workers/ttl/max_bytes) and `File.Batch`, plus `ParseSize` (trims spaces, accepts only binary KiB/MiB/GiB suffixes or a plain byte count, zero/negative/other → def). serve.go opens the dcache store when batch is on (`!cfg.Batch.Disabled`) alongside dixie/search — same path, same call, Warn on failure — then, before building the runner, reaps in-flight jobs a restart killed via `MarkInFlightFailed("restart", now)` with one `ObserveBatchJob("restart")` per reaped job and a `batch reaped` info log when n>0; builds `srv.Batch` from the config (TTL via `ParseDuration`, MaxBytes via `ParseSize`, runner defaults fill the zeros) only when enabled and the store opened; closes `srv.Batch` in both shutdown branches when non-nil. Added a commented `[batch]` block to dist/sift.toml.example. Given test byte-identical; all config and cmd tests pass under -race -count=3. | ## P5a review (owner, 2026-09-26) diff --git a/internal/config/batch_test.go b/internal/config/batch_test.go new file mode 100644 index 0000000..b4ea9c1 --- /dev/null +++ b/internal/config/batch_test.go @@ -0,0 +1,56 @@ +package config + +// P6 task 08: the [batch] block and ParseSize. + +import ( + "os" + "path/filepath" + "testing" +) + +func TestBatchBlock(t *testing.T) { + p := filepath.Join(t.TempDir(), "sift.toml") + body := "[batch]\ndisabled = false\ntargets_per_job = 16\njobs_per_caller = 2\njobs_total = 4\nworkers = 2\nttl = \"12h\"\nmax_bytes = \"32MiB\"\n" + if err := os.WriteFile(p, []byte(body), 0o600); err != nil { + t.Fatal(err) + } + f, err := Load(p) + if err != nil { + t.Fatal(err) + } + b := f.Batch + if b.Disabled || b.TargetsPerJob != 16 || b.JobsPerCaller != 2 || b.JobsTotal != 4 || b.Workers != 2 || b.TTL != "12h" || b.MaxBytes != "32MiB" { + t.Fatalf("batch block %+v", b) + } + // No block: zero values, and batch is on (the runner applies its own defaults). + empty := filepath.Join(t.TempDir(), "empty.toml") + if err := os.WriteFile(empty, []byte("listen = \":8443\"\n"), 0o600); err != nil { + t.Fatal(err) + } + if f, err := Load(empty); err != nil || f.Batch != (Batch{}) { + t.Fatalf("no [batch] block: %+v %v", f.Batch, err) + } +} + +func TestParseSize(t *testing.T) { + const def = 7 + for _, c := range []struct { + in string + want int64 + }{ + {"64MiB", 64 << 20}, + {"512KiB", 512 << 10}, + {"1GiB", 1 << 30}, + {"1000", 1000}, + {" 2MiB ", 2 << 20}, + {"", def}, + {"junk", def}, + {"-1MiB", def}, + {"0", def}, + {"12MB", def}, // decimal units are not accepted: say MiB + } { + if got := ParseSize(c.in, def); got != c.want { + t.Errorf("ParseSize(%q) = %d, want %d", c.in, got, c.want) + } + } +} diff --git a/internal/config/config.go b/internal/config/config.go index e0205de..796007c 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -2,6 +2,8 @@ package config import ( "os" + "strconv" + "strings" "time" "github.com/pelletier/go-toml/v2" @@ -17,6 +19,19 @@ type File struct { Digest Digest `toml:"digest"` Context Context `toml:"context"` Search Search `toml:"search"` + Batch Batch `toml:"batch"` +} + +// Batch is the [batch] block. Batch is on by default: no block means the runner runs with its own +// defaults. disabled = true turns it off and the endpoints answer 503. +type Batch struct { + Disabled bool `toml:"disabled"` + TargetsPerJob int `toml:"targets_per_job"` + JobsPerCaller int `toml:"jobs_per_caller"` + JobsTotal int `toml:"jobs_total"` + Workers int `toml:"workers"` + TTL string `toml:"ttl"` + MaxBytes string `toml:"max_bytes"` } type Dixie struct { @@ -110,3 +125,28 @@ func ParseDuration(s string, def time.Duration) time.Duration { } return d } + +// ParseSize parses a positive byte count: a plain integer, or one with a binary suffix KiB, MiB or +// GiB. Anything else, zero or negative returns def. +func ParseSize(s string, def int64) int64 { + s = strings.TrimSpace(s) + if s == "" { + return def + } + var mult int64 + switch { + case strings.HasSuffix(s, "KiB"): + mult, s = 1<<10, s[:len(s)-3] + case strings.HasSuffix(s, "MiB"): + mult, s = 1<<20, s[:len(s)-3] + case strings.HasSuffix(s, "GiB"): + mult, s = 1<<30, s[:len(s)-3] + default: + mult = 1 + } + n, err := strconv.ParseInt(s, 10, 64) + if err != nil || n <= 0 { + return def + } + return n * mult +}