think
16px
820px
← all files

Task 7 Review — Job API + Manager + server wiring (KEYSTONE)

Reviewed at HEAD d4ec67d. go build ./..., go vet ./..., go test -race -count=1 ./... all green (8 packages, 0 failures).

Spec Compliance — ⚠️

The single-job path (submit → claim → admit → adapter → terminal → audit) is implemented correctly and the seven carried obligations are structurally wired. But there are real defects: an audit/metrics attribution bug (3 of 4 early-failure codes lose their upstream label), a broken truncation-hash invariant on OCR result bodies, a missing panic-recovery boundary around the only new "arbitrary code runs in a bare background goroutine" surface in the codebase, and a real test-coverage gap at exactly the integration seam (server.EnableJobs) the plan called out as required ("drain returns 503 for new submits" is an explicit Task-7 test requirement that was never written).

Obligations A–G

  • A. On-prem in job path — satisfied. internal/jobs/api.go:75-78 (Submit) and internal/jobs/manager.go:246-249 (process) both apply up.Class == "external_dev" && !cfg.AllowExternalUpstreams independently of proxy/handler.go. TestAPISubmitExternalUpstreamForbidden and TestManagerOnPremForbiddenInJobPath both pass.
  • B. Idempotent dedup — satisfied. api.go:80-84 400s on missing key; LocalStore.Create (store.go:78-107) and RedisStore.Create (SETNX-based, redisstore.go:181-) dedup atomically under their own lock/atomicity, so a second Create never inserts a second queued record — ClaimNext therefore can only ever pick up one run. TestAPISubmitIdempotentReturnsSameJob, TestRedisCreateDedupAtomic pass. Verified structurally, not just by the (weak) job_id-equality test.
  • C. Drain 503 — satisfied for the guard wrapper itself (server.go:179-191 reuses s.enter()/leave(), identical pattern to the /v1/ guard), but untested: no test anywhere calls EnableJobs (grep confirms zero references outside server.go/main.go), so the actual POST /jobs → 503+Retry-After-while-draining behavior has never been exercised end-to-end. See Issues.
  • D. ClaimNext priority loop — satisfied. claimOrder = [interactive, system, batch] (manager.go:25), swept in claim() (manager.go:203-210), matching pool.rank and reconciling LocalStore's advisory class arg with RedisStore's genuine per-class lists (per Task 2/3 carry note).
  • E. ClaimNext no-error backoff — satisfied. worker() (manager.go:171-199) backs off IdlePoll (default 500ms) on any nil claim without distinguishing "empty" from "store down," per the Task 3 carry note; no busy-spin.
  • F. Slot-lease release on every exit path — mostly satisfied, with one real gap. All paths in process() after tk := p.Enqueue(...) are covered by defer rel() (releaseOnce) registered immediately after admission (manager.go:282-283), including payload-missing, adapter success/failure/retry-exhaustion, and ctx-cancel/drain (which returns early but still runs the defer). The slot-wait-abandon path (SlotWaitTimeout / ctx-done / stopClaim while waiting) uses abandon(tk) (Cancel + drain-race Release) without a slot ever having been "acquired," so it's a different, also-correct code path. Not covered: an unrecovered panic inside adapter.Run — see Issues (Critical). Not tested: retryable-exhausted as a distinct scenario, and the slot-wait-abandon/drain scenario the last fix commit (d4ec67d) specifically targets — see Issues (Important).
  • G. report/store mutex non-reentrancy — satisfied. report (manager.go:295-300) only sets j.Stage/j.Progress inside the UpdateStatus mutation closure, never re-entering the store. TestManagerReportNoDeadlock exercises 3 sequential report calls and passes under -race.

Strengths

  • The on-prem check, idempotency, and slot-admission/release patterns are faithfully reused from P0 rather than reinvented (abandon/releaseOnce mirror proxy.cancelOrRelease almost verbatim, with a code comment cross-referencing it).
  • Manager.Stop grace-then-hard-cancel semantics are correct and race-clean: workers stop claiming (stopClaim) but finish in-flight work; only a grace-timeout triggers m.cancel(), which is what unblocks a truly-stuck adapter and correctly leaves the job in processing for ReclaimStale rather than marking it failed (manager.go:328-330).
  • The d4ec67d fix (adding <-m.stopClaim to the slot-wait select) is the right fix for the right bug — without it, Stop() would hang until SlotWaitTimeout (default 30m) for any worker parked on a saturated pool.
  • readCapped/bounded-read discipline is applied consistently across both adapters and carried through to the manager's own audit-body compress step.
  • Redis UpdateStatus's WATCH/MULTI optimistic retry avoids the store-mutex-reentrancy hazard by construction (Go closure runs outside any Redis-side lock), independently of the LocalStore mutex argument in obligation G.

Issues

Critical

  1. No panic recovery around adapter execution — a single malformed/buggy adapter response can crash the entire gateway process, not just fail one job. internal/jobs/manager.go:171-199 (worker) and manager.go:238-352 (process, which calls adapter.Run at manager.go:321) run in bare goroutines spawned by Start() (manager.go:118-127); there is no recover() anywhere in internal/jobs/ (confirmed by repo-wide grep — zero hits) or in internal/server/. This differs materially from the pre-existing P0 request path, where Go's net/http server recovers a panicking handler per-connection and keeps serving; the job manager's worker/reclaim/heartbeat goroutines get no such safety net; an unrecovered panic anywhere in the call chain terminates the whole OS process (all LLM traffic + all other jobs + all replicas' worth of local capacity, not just the offending job), which is strictly worse than every other terminal state task 7 was built to guard against (the "historically worst bug" this task explicitly asks about). Current adapter code (adapter.go, adapter_azuredi.go, adapter_classify.go) doesn't have an obvious panic today, but this is exactly the kind of regression Task 8's fan-out work (arbitrary per-child JSON/merge logic) is likely to introduce, and OCR upstreams are the least-trusted response shape in the whole system (attacker- or bug-controlled JSON/multipart, unlike the vLLM proxy's narrower response shapes).
    - Fix: wrap the body of worker()'s per-job dispatch (or process() itself) in a defer func() { if r := recover(); r != nil { ... mark job failed with ADAPTER_PANIC, release slot via the already-deferred rel(), emit audit, log ... } }(). Since rel()/hbCancel() are already defer-registered inside process(), adding a top-level recover in process() (or in worker() wrapping the m.process(job) call) is sufficient — the existing defers will still fire during the panic unwind before the recover stops propagation.

Important

  1. Audit/metrics upstream attribution lost for 3 of 4 early-failure codes. manager.go:241,247,252,287 all call m.finish(job, StatusFailed, nil, CODE, ...), and finish (manager.go:386-388) unconditionally forwards up=nil to finishWithPayload. For UNKNOWN_MODEL (line 241) up is genuinely unresolved so nil is correct — but for EXTERNAL_UPSTREAM_FORBIDDEN (line 247), NO_POOL (line 252), and PAYLOAD_MISSING (line 287), a valid non-nil up is already in scope at the call site and is simply not threaded through. The result: the audit event's Upstream field is empty and record() (the gateway_jobs_total metric) buckets these failures under upstream="" instead of the actual upstream id — exactly the observability the plan's "one audit event per job... operation from upstream" bullet exists to guarantee, and precisely the failure mode (a misconfigured/forbidden OCR upstream) an operator would most want attributed correctly. Not caught by TestManagerOnPremForbiddenInJobPath, which only asserts job.ErrorCode, never the audit event's Upstream field.
    - Fix: change these three call sites to call m.finishWithPayload(job, up, StatusFailed, nil, CODE, queueMs, upstreamMs, nil) directly instead of going through m.finish, leaving m.finish's nil-up behavior only for the true UNKNOWN_MODEL case.
  2. Truncated OCR/DI result bodies lose their verification hash. manager.go:426-430:
    go if len(result) > 0 { z, tr, _ := m.compress(result) ev.ResponseBodyZst = z ev.BodyTruncated = ev.BodyTruncated || tr }
    discards the response's fullHash outright (_). Contrast with proxy/handler.go:256-258 which does if fh != "" { ev.FullBodySHA256 = fh } — i.e., the existing P0 code does propagate the response-side hash when truncated. For OCR/Azure-DI this is the more likely truncation direction (an analyzeResult with per-page word/line boxes can dwarf body_cap_bytes even when the submitted document itself is well under the cap), so this is not a corner case — it's the common one for large multi-page documents. Net effect: body_truncated=true with full_body_sha256="", breaking CONVENTIONS' "truncated + full_body_sha256" pairing for exactly the direction (response) most likely to trigger truncation on an OCR job.
    - Fix: mirror proxy exactly: if tr && ev.FullBodySHA256 == "" { ev.FullBodySHA256 = fh } (or a dedicated field if request and response can both be truncated and both hashes matter — this same single-field ambiguity exists in proxy/handler.go today, so at minimum match its partial fix rather than dropping it entirely).
  3. Zero test coverage of the actual server.EnableJobs integration, despite the plan's Task 7 line explicitly requiring "drain returns 503 for new submits" and "submit→poll→completed against a stub adapter" as tests. grep -rln EnableJobs internal/ cmd/ returns only server.go and main.go — no test file calls it. internal/jobs/manager_test.go and internal/jobs/api_test.go each construct their own isolated Manager/API over independent LocalStores and never exercise the guard wrapper at server.go:179-191, the shared-store wiring between API.Store and the Manager (server.go:166-172), or Shutdown's jobMgr.Stop(ctx) call (server.go:229-231). The one thing this leaves genuinely unverified is exactly obligation C end-to-end: does a real HTTP POST /jobs actually get 503 + Retry-After while draining through the real mux? Code inspection says yes, but the specific test the plan asked for doesn't exist.
    - Fix: add a server-package test mirroring TestDrainRejectsNewFinishesInflight but calling s.EnableJobs(jobs.NewLocal(), jobs.ManagerConfig{}) first and posting to /jobs; and one submit→poll→completed HTTP-level test using a stub upstream, matching what the isolated jobs-package tests already do at the unit level.
  4. No regression test for the exact scenario the last fix commit (d4ec67d) addresses. The slot-wait-abandon-on-drain path (manager.go:270-274, the <-m.stopClaim case added in that commit) has no test that saturates a 1-slot pool, enqueues a second job, calls Stop() while it's parked in the tk.Ready() select, and asserts (a) Stop() returns promptly rather than blocking ~SlotWaitTimeout, and (b) the pool's Active()/Depth() return to a clean state and the job is left queued/processing-for-reclaim rather than incorrectly failed. Given this is described in the ledger as "the historically worst bug" area and this exact code was touched by a fix committed minutes before this review, it's the single highest-value missing test in the diff.
  5. TestManagerReleasesSlotOnEveryTerminalState doesn't cover "retryable-exhausted" as its own case, only success, non-retryable-error, and processing-timeout. Structurally the retryable-exhausted path re-joins the same if runErr != nil branch as non-retryable, so this is lower risk than #5, but it's one of the nine terminal kinds this review was asked to hand-trace and it's silently assumed rather than asserted.

Minor

  1. queue_ms is inflated after a reclaim bounce. ClaimNext (store.go:127-146) sets StartedAt = now on every claim, including a re-claim after ReclaimStale requeues a dead worker's job. manager.go:256-259 computes queueMs = StartedAt.Sub(CreatedAt), so for a job that died mid-processing and was later reclaimed, queue_ms ends up spanning the original queued time plus the entire dead-processing/visibility-timeout window, not the job's actual "time spent queued" — a metrics-accuracy nit, not a functional bug.
  2. Heartbeat goroutine isn't tracked by Manager.wg. manager.go:303-305 does go m.heartbeat(hbCtx, job.ID) outside of m.wg, so Stop()'s m.wg.Wait() (manager.go:136-137) can return before every in-flight heartbeat goroutine has observed hbCancel() and returned. In practice this window is a single scheduler tick (the goroutine's select sees the closed ctx.Done() essentially immediately), and the process exits shortly after Stop() returns in main.go anyway, so this is benign today — but it means Manager.Stop() doesn't give a hard "zero goroutines" guarantee, which could bite a future goleak-style test (the module already vendors go.uber.org/goleak transitively per go.sum, though nothing currently imports it).
  3. cmd/gateway/main.go:92-93 calls srv.Shutdown(dctx) then hs.Shutdown(dctx) sequentially against the same deadline context. If job-manager draining (jobMgr.Stop inside srv.Shutdown) consumes the whole 30s grace budget, hs.Shutdown is invoked with an already-expired context; http.Server.Shutdown still closes listeners immediately in that case, so this is not a hang, but it means the final "wait for idle HTTP conns to close" step gets effectively zero budget instead of its own share. Low practical impact, worth a one-line comment or splitting the budget if it ever becomes a real drain-time constraint.

Assessment

Task quality: Needs fixes before merge, not because the design is wrong — the architecture (Manager over Store over Pool, single shared audit/emit closure, guard-wrapper reuse) is sound and matches P0's patterns exactly as instructed — but because:

  • Issue #1 (no panic recovery in the job worker goroutines) is a genuine Critical: it converts what should be a per-job failure into a whole-gateway outage, which is a strictly worse failure mode than anything this task's obligations were written to prevent, and it's a cheap, mechanical fix (one recover() wrapping process()).
  • Issues #2 and #3 are real, demonstrable defects in the audit trail this whole task exists to produce correctly (attribution loss + broken truncation-hash invariant on the response side), not stylistic nits.
  • Issue #4 is a direct, named gap against the plan's own Task 7 test list ("drain returns 503 for new submits") — the behavior is very likely correct by inspection, but "very likely correct by inspection" is exactly the standard this plan explicitly rejected for the slot-lease area, and the same untested seam is where the Store/Manager/API three-way wiring actually gets exercised together for the first time.

None of these require a redesign — recommend: add the recover() boundary, thread up through the three finish-with-up-in-scope call sites, fix the response-hash drop to match proxy.compress's pattern, and add the two missing tests (server-level EnableJobs drain/submit-poll test, and a slot-wait-abandon-on-drain regression test) — then re-review should be quick since nothing here touches the core admission/claim/store logic that Tasks 1-6 already validated.