diff --git a/docker-compose.yml b/docker-compose.yml index e0ab9aa7..6fd69c8d 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -362,8 +362,10 @@ services: # ── status.* (1) ── CTX_STATUS_CHANNEL_PROBE_INTERVAL: ${CTX_STATUS_CHANNEL_PROBE_INTERVAL:-} - # ── embed_backfill.* (4) ── + # ── embed_backfill.* (6) ── CTX_EMBED_BACKFILL_SYNC_CAP: ${CTX_EMBED_BACKFILL_SYNC_CAP:-} + CTX_EMBED_BACKFILL_SYNC_BUDGET: ${CTX_EMBED_BACKFILL_SYNC_BUDGET:-} + CTX_EMBED_BACKFILL_INTERVAL: ${CTX_EMBED_BACKFILL_INTERVAL:-} CTX_EMBED_BACKFILL_MAX_TOKENS: ${CTX_EMBED_BACKFILL_MAX_TOKENS:-} CTX_EMBED_BACKFILL_BACKOFF_BASE: ${CTX_EMBED_BACKFILL_BACKOFF_BASE:-} CTX_EMBED_BACKFILL_BACKOFF_CAP: ${CTX_EMBED_BACKFILL_BACKOFF_CAP:-} diff --git a/docs/operations.md b/docs/operations.md index 487c31bc..c78180bb 100644 --- a/docs/operations.md +++ b/docs/operations.md @@ -477,6 +477,8 @@ The `llama-embed` sidecar (`docker-compose.override.yml`, Qwen3-Embedding-8B, `- **Backfill failure memo (migration 113, Achse 04 W04-2).** Both backfill paths — the scheduler arm (`backfillOneEmbedding`) and the query-path pre-search loop (`backfillPending`) — used to retry a NULL-embedding block **indefinitely** if it exceeded the embed window: no truncate, no skip, nothing written to the DB on failure, so the next pick landed on the exact same block (observed 2026-07-10: a 13k-char block at 4704 tokens against a 4096 window looped every ~20 s and blocked ~265 pending blocks behind it). This is closed structurally, not operationally: every failed embed attempt now upserts a row into `context_embed_failures` (`block_id`, `attempts`, `last_error` — normalized, ≤500 chars, never raw wire/embed-text content —, `last_class`, `next_attempt_at`), and both pick queries exclude a block whose `next_attempt_at` is still in the future. A non-oversize failure (backend down, transient wire error) gets an exponential backoff (`embed_backfill.backoff_base` × 2^(attempts-1), capped at `embed_backfill.backoff_cap`; defaults 1min/24h; the base must be > 0 — V21 rejects 0, which would mean "retry immediately" and a tight loop against a failing backend, issue #38) — the next PEEK reaches the next-oldest block instead of re-picking the same one. An oversize block — classified either by a conservative pre-wire token estimate (`len(text)/4` against `embed_backfill.max_tokens`, default 24000, chosen to sit under the live kv-unified pool's 32k minus interactive headroom — see the tuning section above) or by an HTTP 400 whose body names `exceed_context_size` — gets `next_attempt_at = 'infinity'` instead: permanently parked, never a blind 24h-in-slow-motion retry, but never silently dropped either (the memo row stays visible in `context_embed_failures`). Known limitation: `ClearEmbedding` does NOT currently clear the block's failure memo, so a content edit that shrinks a previously oversize-parked block below the token threshold does **not** reactivate it on its own — the `infinity` exclusion persists until the memo row is cleaned up (a janitor for this is not yet built; see the Achse-04 backlog). The query-path loop additionally carries a synchronous cap (`embed_backfill.sync_cap`, default 4): before this wave a large rest-transient (e.g. after a bulk ingest that outran the scheduler arm) made the *first* interactive query pay the entire nachzug inline; now it embeds at most `sync_cap` blocks per request and leaves the rest to the scheduler arm. Keeping blocks compact (~1–1.5k chars per the store guidance) or sizing the embed window above your largest block is still the actual fix for an oversize block's content — the memo mechanism only stops it from starving everything behind it. +**The dedicated backfill arm and the query-path time budget (tracking issue: force-push.me/damienmoon/ctx/issues/1).** Two gaps in the above showed up on a CPU embed backend (Qwen3-Embedding-8B at the documented ~23 tok/s) with a ~10k-token block. First, the scheduler arm ran only at the top of the dream loop, so with `dream.enabled=false` — the default — nothing backfilled pending embeddings in the background at all; the query path was the only path. `runEmbedBackfillLoop` is now its own goroutine, started regardless of Dream: a successful pick loops at once, an empty pick waits `embed_backfill.interval` (default 15 s), `0` turns the arm off (polled every 15 s for a hot re-enable). The dream loop keeps its inline call; the tx-wrapped `FOR UPDATE SKIP LOCKED` pick makes two pickers safe. Second, `sync_cap` bounds how *many* blocks a query embeds inline, not how *long*: one block that needs 7 minutes consumed the caller's whole 120 s deadline, the question's own embed was never admitted, and the query answered 500 instead of degrading. `embed_backfill.sync_budget` (seconds, default 30, `0` = unbounded) now bounds the whole pre-search loop per request; each wire call runs under what is left of it. A block that overruns the budget — or dies with the request context — is memoized with the class `caller_timeout`: the query path never picks it again (`EmbedFailureQueryPathExcludedPredicate`), the background arm picks it at once (`next_attempt_at = now()`), and a later wire failure there reclassifies it as `wire` with the ordinary backoff. The block stays FTS-visible meanwhile. Previously such a failure was memoized as `wire` and re-burned the deadline of every query that landed after the backoff lapsed. + **Backfill skips retrieval-excluded types (2026-07-25).** Both backfill paths now also skip a block whose *type* carries retrieval policy `excluded` in the block-type registry (shipped builtins: `checkpoint`, `system-meta`). Such a vector has no consumer at all — every ranked path filters against the registry's visible-type allowlist, so the embedding was written once and then read by nobody. The trigger was a live incident on 2026-07-25 with two independent arms: 33 Hermes "Compaction source" parts (`type=checkpoint`, ~36.5 kB each) saturated the CPU embed backend for hours, and the digest rebuild rewrites `topic-map-private` (`type=system-meta`, 73.7 kB ≈ ~32k real tokens) on **every** boot, which clears its embedding and bought ~60 min of CPU embed for a dead vector — recurring, forever. The filter is policy-driven, not a hardcoded type list (same mechanic as `test.sh` T19), and deliberately conservative in the fail-safe direction: only `_global`-`excluded` skips; `damped` and `aggregate-to-parent` stay embeddable, a block whose type has no registry row at all stays embeddable, and a tenant scope that overrides an excluded type to a visible policy re-opens that type for the backfill (the pick has no reader identity, so the reader-aware overlay rule is mirrored as the union over all possible readers). Two operational consequences: blocks of an excluded type keep `embedding IS NULL` forever — that is the intended end state, not a stuck queue — and the `/status` `embed_backlog` metric still counts them, so a corpus with such blocks shows a permanent backlog floor (live at the time of writing: 24 `checkpoint` + 1 `system-meta`). Existing `infinity` failure memos for such blocks become moot: the predicate no longer even sees the rows. Flipping a type's policy from `excluded` to a visible one makes its blocks embeddable again on the next cycle without any manual step. **Re-embed migration worker (Achse 04 W04-4).** With an active migration in `running` or `verifying`, a dedicated scheduler arm (`migrateOneEmbedding`, BACKGROUND class, defers to interactive demand) walks all blocks that carry an old-space vector (`embedding IS NOT NULL AND embedding_next IS NULL AND NOT is_archived`) oldest-first from a persistent cursor, embeds them via the model-map key `embed_next` on the regular embed chain, and writes `embedding_next`/`embed_model_next` — the serving pair stays byte-identical until the cutover. Operator knobs: `embed_migration.batch_per_cycle` (default 8, `<=0` disables the arm), `embed_migration.max_tokens` (default 24000, pre-wire oversize gate; `0` disables the estimate — the wire classification of HTTP 400 `exceed_context_size` stays), `embed_migration.backoff_base`/`backoff_cap` (defaults 1min/24h; the base must be > 0 — V21, same tight-loop reasoning as the backfill sibling) — all hot, global-only. Failure memos are migration-scoped rows in `context_embed_failures` (independent from the regular backfill's memos): transient wire errors back off exponentially, `oversize` and `sensitivity_ineligible` (no backend trusted for the block's sensitivity) park at `infinity` — visible in the memo table, surfaced in status/state.sh with W04-7. **If a migration sits in `paused` with `last_error = "model guard: resolved , expected "`**: the target backend's model_map lost (or never had) the `embed_next` key resolving to the migration's `to_model` — fix the backend row, then resume; the guard re-checks every cycle and exists because a missing key silently falls back to the OLD model (relabeling old vectors as migrated). The arm maintains a runtime partial index `idx_embedding_next_pending` (created `CONCURRENTLY` when a migration is active, INVALID leftovers from a crashed build are dropped and rebuilt automatically, dropped again once the migration is terminal) — seeing it in the catalog is normal only while a migration exists. Measured backend behavior (2026-07-24, CPU llama.cpp, kv-unified): a parallel embed that overflows the shared pool WAITS (no error path), and even below the pool limit a small interactive embed queues behind a large prefill — at ~20 tok/s CPU prefill a 24k-token block holds a slot for ~19 min, which is what `max_tokens` actually bounds. diff --git a/go/internal/config/config.go b/go/internal/config/config.go index 55687c04..af0f5f96 100644 --- a/go/internal/config/config.go +++ b/go/internal/config/config.go @@ -1592,6 +1592,29 @@ type EmbedBackfillConfig struct { // unit test that never sets this field disables the cap rather than // silently blocking every backfillPending call. SyncCap int `key:"embed_backfill.sync_cap" env:"CTX_EMBED_BACKFILL_SYNC_CAP" default:"4" mut:"hot" tenancy:"global-only"` + // SyncBudget bounds Pfad A in SECONDS, next to SyncCap's count: the + // whole pre-search loop of one request runs inside this wall-clock + // budget, and each block's embed call runs under what is left of it. + // SyncCap alone could not stop one slow block from eating the caller's + // entire deadline (the tracking issue, force-push.me/damienmoon/ctx/issues/1: a ~10k-token block against a + // CPU embed backend at ~23 tok/s needs ~7 min, the CLI client's deadline + // is 120 s — the question's own embed was never admitted and the query + // answered 500 instead of degrading). A block that overruns the budget + // is memoized as caller_timeout (store.EmbedFailureCallerTimeout) and + // left to the background arm; the question embed keeps the rest of the + // deadline. Default 30 s (a quarter of the CLI deadline). 0 = unbounded, + // the pre-budget behavior (explicit opt-out, SyncCap convention). + SyncBudget time.Duration `key:"embed_backfill.sync_budget" env:"CTX_EMBED_BACKFILL_SYNC_BUDGET" default:"30" mut:"hot" tenancy:"global-only"` + // Interval is the idle wait in seconds of the dedicated background + // backfill arm (runEmbedBackfillLoop) between two empty picks. Until + // this key the only background backfill was the call at the top of the + // dream loop, so with dream.enabled=false (the default) NOTHING embedded + // pending blocks in the background and Pfad A was the only path (issue + // force-push.me/damienmoon/ctx/issues/1). The arm runs regardless of Dream; a successful + // pick loops at once, an empty pick waits this long. 0 = arm off + // (polled every 15 s for a hot re-enable, contract.recheck_interval + // convention). + Interval time.Duration `key:"embed_backfill.interval" env:"CTX_EMBED_BACKFILL_INTERVAL" default:"15" mut:"hot" tenancy:"global-only"` // MaxTokens is the pre-wire Oversize-Gate estimate threshold // (design/04 §4.4): len(embedText)/4 above this skips the block WITHOUT // a wire call (last_class='oversize', next_attempt_at='infinity' — diff --git a/go/internal/config/descriptions.go b/go/internal/config/descriptions.go index 3b653642..95ef754a 100644 --- a/go/internal/config/descriptions.go +++ b/go/internal/config/descriptions.go @@ -175,6 +175,8 @@ var keyDescriptions = map[string]string{ //nolint:gosec // G101-Fehlalarm: Schl "contract.recheck_interval": "Seconds between periodic schema-drift re-checks; 0 runs no checks while the ticker keeps polling so a later hot enable takes effect", "status.channel_probe_interval": "Seconds between latency probes of the four retrieval channels (semantic, fts_de, fts_en, trigram); 0 (default) disables the probe", "embed_backfill.sync_cap": "Maximum pending blocks a query embeds inline before its search runs; 0 = uncapped, the cap bounds first-query latency after a backlog", + "embed_backfill.sync_budget": "Seconds a query may spend embedding pending blocks before its own search; a block that overruns it is memoized as caller_timeout and left to the background arm, 0 = unbounded", + "embed_backfill.interval": "Seconds the background embed-backfill arm waits between two empty picks; it runs regardless of Dream, 0 = arm off (polled for a hot re-enable)", "embed_backfill.max_tokens": "Estimated-token threshold (length/4) above which a block is permanently parked as oversize without a wire call; 0 disables the pre-check", "embed_backfill.backoff_base": "Base delay in seconds of the exponential retry curve (base * 2^(attempts-1)) for transient embed failures. Must be > 0 (V21) — 0 would mean retry immediately, a tight loop against a failing embed backend", "embed_backfill.backoff_cap": "Upper bound in seconds on the embed-failure retry delay; a persistently failing block retries at most this often", diff --git a/go/internal/config/validate_test.go b/go/internal/config/validate_test.go index 44604181..d027bb74 100644 --- a/go/internal/config/validate_test.go +++ b/go/internal/config/validate_test.go @@ -393,7 +393,7 @@ func TestV17bIsAdvisoryOnly(t *testing.T) { // changes TYPE (seconds → int, or a struct field that stops being a // time.Duration) would leave the walk without anyone noticing. Raise this // number in the same commit that adds a duration key. -const wantDurationKeys = 45 +const wantDurationKeys = 47 // +2 on 2026-10-09: embed_backfill.sync_budget and embed_backfill.interval, both V17-covered // TestValidateRejectsNegativeOnEveryDurationKey is the registry-wide form of // V17 (issue #29): before it, exactly two of the seconds keys had a sign check diff --git a/go/internal/events/backfill_arm_integration_test.go b/go/internal/events/backfill_arm_integration_test.go new file mode 100644 index 00000000..55383e6c --- /dev/null +++ b/go/internal/events/backfill_arm_integration_test.go @@ -0,0 +1,124 @@ +//go:build integration + +// Rot-Gate for the tracking issue, force-push.me/damienmoon/ctx/issues/1, scheduler half: the embed-backfill +// arm (backfillOneEmbedding) was called ONLY at the top of runDreamLoop, so +// with dream.enabled=false — the registry default — no goroutine ever +// embedded a pending block in the background. Live: a block stayed pending +// for two hours while the scheduler logged guard and digest ticks only; the +// query path was the sole embedder, inside its caller's deadline. +// +// Rot is the premise, stated against the production wiring: a scheduler +// with DreamEnabled=false starts no dream worker, and nothing else in Run +// called backfillOneEmbedding before this wave. Grün is the dedicated arm +// (runEmbedBackfillLoop): started regardless of Dream, it embeds the pending +// block within a few intervals, and a caller_timeout memo (the query path's +// hand-off) does not hold it back. +// +// Run: `go test -tags=integration ./internal/events/ -run TestEmbedBackfillArm -count=1 -v`. +package events + +import ( + "context" + "testing" + "time" + + "github.com/GottZ/ctx/internal/backends" + "github.com/GottZ/ctx/internal/config" + "github.com/GottZ/ctx/internal/dispatch" + "github.com/GottZ/ctx/internal/store" + "github.com/GottZ/ctx/internal/testdb" +) + +func TestEmbedBackfillArm_RunsWithoutDream_Integration(t *testing.T) { + if testing.Short() { + t.Skip("integration test") + } + pool := testdb.SetupTestDB(t) + ctx := context.Background() + + srv := headOfLineEmbedServer(t) + bpool := backends.NewPool(nil, nil) + bpool.SeedSnapshotForTest([]backends.Backend{embedPoolRow("embed-a", srv.URL, 100)}) + d := dispatch.New(nil, dispatch.DefaultSettings()) // empty policy: pass-through + t.Cleanup(d.Close) + + cfg := &config.Config{EmbedBackfill: config.EmbedBackfillConfig{ + Interval: 100 * time.Millisecond, + MaxTokens: 0, + BackoffBase: time.Hour, + BackoffCap: 24 * time.Hour, + }} + // DreamEnabled=false: the RED premise — before this wave, Run started no + // goroutine that could reach backfillOneEmbedding in this configuration. + s := NewScheduler(pool, config.NewStore(cfg), bpool, StartupConfig{DreamEnabled: false}) + s.SetDispatcher(d) + + waitEmbedded := func(t *testing.T, within time.Duration) bool { + t.Helper() + deadline := time.Now().Add(within) + for time.Now().Before(deadline) { + if pendingCount(t, pool) == 0 { + return true + } + time.Sleep(25 * time.Millisecond) + } + return false + } + + t.Run("arm_embeds_a_pending_block_with_dream_off", func(t *testing.T) { + seedPendingBlock(t, pool, "arm-pending-normal", time.Hour) + defer clearBlocks(t, pool) + + armCtx, stop := context.WithCancel(ctx) + defer stop() + go s.runEmbedBackfillLoop(armCtx) + + if !waitEmbedded(t, 5*time.Second) { + t.Fatalf("block still pending after 5 s: the dedicated arm did not embed it (pending=%d)", pendingCount(t, pool)) + } + }) + + t.Run("arm_picks_a_caller_timeout_block_at_once", func(t *testing.T) { + seedPendingBlock(t, pool, "arm-handed-off", time.Hour) + defer clearBlocks(t, pool) + + // The query path's hand-off: a caller_timeout memo, due now. + var blockID string + if err := pool.QueryRow(ctx, `SELECT id FROM context_blocks WHERE title = 'arm-handed-off'`).Scan(&blockID); err != nil { + t.Fatalf("read block id: %v", err) + } + if err := store.RecordEmbedFailure(ctx, pool, blockID, store.EmbedFailureCallerTimeout, + store.NormalizeEmbedError(store.EmbedFailureCallerTimeout, "context deadline exceeded"), + cfg.EmbedBackfill.BackoffBase, cfg.EmbedBackfill.BackoffCap); err != nil { + t.Fatalf("record caller_timeout memo: %v", err) + } + t.Cleanup(func() { _, _ = pool.Exec(ctx, `DELETE FROM context_embed_failures WHERE block_id = $1`, blockID) }) + + armCtx, stop := context.WithCancel(ctx) + defer stop() + go s.runEmbedBackfillLoop(armCtx) + + if !waitEmbedded(t, 5*time.Second) { + t.Fatalf("caller_timeout block still pending after 5 s: the arm must pick it at once, not after a backoff") + } + }) + + t.Run("interval_zero_turns_the_arm_off", func(t *testing.T) { + seedPendingBlock(t, pool, "arm-off-pending", time.Hour) + defer clearBlocks(t, pool) + + off := *cfg + off.EmbedBackfill.Interval = 0 + sOff := NewScheduler(pool, config.NewStore(&off), bpool, StartupConfig{DreamEnabled: false}) + sOff.SetDispatcher(d) + + armCtx, stop := context.WithCancel(ctx) + defer stop() + go sOff.runEmbedBackfillLoop(armCtx) + + time.Sleep(600 * time.Millisecond) + if got := pendingCount(t, pool); got != 1 { + t.Fatalf("pending = %d, want 1 (Interval=0 is the explicit off switch)", got) + } + }) +} diff --git a/go/internal/events/scheduler.go b/go/internal/events/scheduler.go index ee3e2686..20f754e5 100644 --- a/go/internal/events/scheduler.go +++ b/go/internal/events/scheduler.go @@ -782,6 +782,14 @@ func (s *Scheduler) Run(ctx context.Context) { embedMigrateTicker := time.NewTicker(embedMigrateInterval) defer embedMigrateTicker.Stop() + // The dedicated embed-backfill arm (tracking issue: force-push.me/damienmoon/ctx/issues/1): pending + // embeddings used to be backfilled only at the top of the dream loop, so + // with dream.enabled=false (the default) or Dream switched off at runtime + // nothing embedded them in the background. The arm runs regardless of + // Dream; the dream loop keeps its inline call, the tx-wrapped + // FOR UPDATE SKIP LOCKED pick makes two pickers safe (Welle-49). + go s.runEmbedBackfillLoop(ctx) + // Dream runs in its own goroutine(s) as continuous loop(s). // DreamParallelism workers all share the same DB; PickBlock's FOR UPDATE // SKIP LOCKED ensures distinct blocks per worker. Backfill stays single- @@ -2043,6 +2051,78 @@ func (t *txTail) done() { } } +// backfillArmPoll is how often the dedicated embed-backfill arm re-reads its +// config while embed_backfill.interval is 0 (arm off), so a hot re-enable +// takes effect without a restart (contract.recheck_interval convention). +const backfillArmPoll = 15 * time.Second + +// runEmbedBackfillLoop is the dedicated background arm for pending +// embeddings (tracking issue: force-push.me/damienmoon/ctx/issues/1). It is the same work as the call at the +// top of runDreamLoop — backfillOneEmbedding under the iterated tenant's +// snapshot and router — on its own cadence: a successful pick loops at once, +// an empty pick waits embed_backfill.interval, an error waits 10 s. Under +// !enforcing() it yields to interactive demand exactly like the dream loop +// (P-Fallback); under Enforcing the wire call waits at its target in its +// background lease. Interval <= 0 turns the arm off; it then polls the +// config every backfillArmPoll so a hot enable needs no restart. +func (s *Scheduler) runEmbedBackfillLoop(ctx context.Context) { + slog.Info("scheduler: embed backfill arm started", "poll", backfillArmPoll) + var tenantCursor uint64 + wait := func(d time.Duration) bool { + select { + case <-ctx.Done(): + return false + case <-time.After(d): + return true + } + } + for ctx.Err() == nil { + tenants := s.backgroundTenantsFn(ctx) + if len(tenants) == 0 { + tenants = []backgroundTenant{{scope: store.GlobalScope}} + } + bt := tenants[tenantCursor%uint64(len(tenants))] + tenantCursor++ + cfg := s.cfg.SnapshotForTenant(ctx, bt.scope) + + interval := cfg.EmbedBackfill.Interval + if interval <= 0 { + if !wait(backfillArmPoll) { + return + } + continue + } + for !s.enforcing() && s.interactiveDemand() > 0 { + if !wait(dreamYieldWait) { + return + } + } + + backfilled, err := s.backfillArmOnce(ctx, cfg, bt.scope) + switch { + case err != nil: + slog.Error("scheduler: embed backfill arm error", "error", err) + if !wait(10 * time.Second) { + return + } + case backfilled: + continue // Loop at once: more may be pending. + default: + if !wait(interval) { + return + } + } + } +} + +// backfillArmOnce is one pick of the dedicated arm with its own panic guard, +// so a panic in one cycle is logged and the arm goes on (the dream loop's +// runDreamCycle has the same shape). +func (s *Scheduler) backfillArmOnce(ctx context.Context, cfg *config.Config, tenant string) (bool, error) { + defer guardPanic("embed backfill arm") + return s.backfillOneEmbedding(ctx, s.newRouter(cfg, tenant), cfg) +} + // backfillOneEmbedding finds one block with missing embedding, generates it, and stores it. // Returns true if a block was backfilled, false if none needed. // The embed chains over the pool (G28): role dream-embed when configured, diff --git a/go/internal/handler/backfill_syncbudget_integration_test.go b/go/internal/handler/backfill_syncbudget_integration_test.go new file mode 100644 index 00000000..bcfba1b2 --- /dev/null +++ b/go/internal/handler/backfill_syncbudget_integration_test.go @@ -0,0 +1,230 @@ +//go:build integration + +// Rot-Gate for the tracking issue, force-push.me/damienmoon/ctx/issues/1: the query-path pre-search backfill +// (backfillPending) had a COUNT cap (embed_backfill.sync_cap) but no TIME +// budget. One block whose embed could not finish inside the caller's +// deadline consumed the entire deadline (live: a ~10k-token block against a +// CPU embed backend at ~23 tok/s needs ~7 min, the CLI client's deadline is +// 120 s), the question's own embed was never admitted ("embedcache: +// admission: context canceled") and the query answered 500. The failure was +// memoized as class `wire` — backend-down semantics for a backend that was +// healthy, just slow for this caller — so every query that landed after the +// backoff lapsed repeated the burn. +// +// Rot and Grün run through the SAME production function. SyncBudget=0 is BY +// CONSTRUCTION the historical shape (no deadline on the wire call, no +// budget check at the loop head); SyncBudget>0 is the fix. +// +// SyncBudget=0 (Ist): the slow block holds the loop for its full embed time. +// SyncBudget=300ms (Soll): the loop returns inside the budget, the block +// is memoized caller_timeout with next_attempt_at <= now(), the query +// path never picks it again, the background arm's predicate still does. +// +// Run: `go test -tags=integration ./internal/handler/ -run TestBackfillPending_SyncBudget -count=1 -v`. +package handler + +import ( + "context" + "encoding/json" + "io" + "net/http" + "net/http/httptest" + "sync/atomic" + "testing" + "time" + + "github.com/GottZ/ctx/internal/blocktype" + "github.com/GottZ/ctx/internal/config" + "github.com/GottZ/ctx/internal/store" + "github.com/GottZ/ctx/internal/testdb" +) + +// slowEmbedServer answers like fakeEmbedServer, but only after `delay`, or +// at once with the request's cancellation if the client goes away first. +func slowEmbedServer(t *testing.T, delay time.Duration) (*httptest.Server, *atomic.Int32, *atomic.Int32) { + t.Helper() + var hits, canceled atomic.Int32 + vec := make([]float64, 1024) + for i := range vec { + vec[i] = float64((i % 2) * 2) + } + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + hits.Add(1) + // Drain the body first: net/http starts the background read that + // turns a client disconnect into r.Context() cancellation only once + // the request body has been consumed (registerOnHitEOF in + // readRequest). Without this the handler never sees the cancel. + _, _ = io.Copy(io.Discard, r.Body) + select { + case <-time.After(delay): + case <-r.Context().Done(): + canceled.Add(1) + return + } + w.Header().Set("Content-Type", "application/json") + _ = json.NewEncoder(w).Encode(map[string]any{"embeddings": [][]float64{vec}}) + })) + t.Cleanup(srv.Close) + return srv, &hits, &canceled +} + +func TestBackfillPending_SyncBudget_Integration(t *testing.T) { + if testing.Short() { + t.Skip("integration test") + } + pool := testdb.SetupTestDB(t) + ctx := context.Background() + + const title = "syncbudget-slow-block" + seed := func(t *testing.T) { + t.Helper() + if _, err := pool.Exec(ctx, + `INSERT INTO context_blocks (category, title, content, scope, created_at, updated_at) + VALUES ('learnings', $1, 'body of the sync-budget fixture block', 'private', now(), now())`, title); err != nil { + t.Fatalf("seed block: %v", err) + } + t.Cleanup(func() { + _, _ = pool.Exec(ctx, `DELETE FROM context_embed_failures WHERE block_id IN (SELECT id FROM context_blocks WHERE title = $1)`, title) + _, _ = pool.Exec(ctx, `DELETE FROM context_blocks WHERE title = $1`, title) + }) + } + readMemo := func(t *testing.T) (class string, dueNow bool, attempts int) { + t.Helper() + err := pool.QueryRow(ctx, + `SELECT f.last_class, f.next_attempt_at <= now(), f.attempts + FROM context_embed_failures f JOIN context_blocks cb ON cb.id = f.block_id + WHERE cb.title = $1 AND f.migration_id IS NULL`, title).Scan(&class, &dueNow, &attempts) + if err != nil { + t.Fatalf("read memo: %v", err) + } + return class, dueNow, attempts + } + newHandler := func(t *testing.T, delay time.Duration) (*QueryHandler, *atomic.Int32, *atomic.Int32) { + t.Helper() + srv, hits, canceled := slowEmbedServer(t, delay) + st := &countingStore{} + st.cfg.Store(snapshotTestConfig()) + h := NewQueryHandler(pool, st, embedPool(srv.URL), nil, blocktype.NewRegistry(), snapshotTestAdmitter(t)) + return h, hits, canceled + } + + // Rot: SyncBudget=0 is the historical shape — the loop waits for the + // slow backend however long it takes. Nothing bounds the request's time. + t.Run("red_unbudgeted_waits_for_the_slow_backend", func(t *testing.T) { + seed(t) + h, hits, _ := newHandler(t, 1500*time.Millisecond) + cfg := &config.Config{EmbedBackfill: config.EmbedBackfillConfig{ + SyncCap: 4, SyncBudget: 0, MaxTokens: 1_000_000, BackoffBase: time.Minute, BackoffCap: time.Hour, + }} + + started := time.Now() + got := h.backfillPending(ctx, nil, "private", h.embedAdmission(), cfg) + elapsed := time.Since(started) + if got != 1 { + t.Fatalf("RED premise: backfilled = %d, want 1 (unbudgeted loop embeds the slow block)", got) + } + if elapsed < 1400*time.Millisecond { + t.Fatalf("RED premise: returned after %v, want >= 1.4 s (the loop waited for the backend)", elapsed) + } + if hits.Load() != 1 { + t.Errorf("wire hits = %d, want 1", hits.Load()) + } + t.Logf("RED confirmed: SyncBudget=0 held the request for %v on one slow block", elapsed) + }) + + // Grün: SyncBudget bounds the loop. The slow block is cut at the budget, + // memoized caller_timeout and due for the background arm at once; the + // parent context is untouched, so the question embed keeps its time. + t.Run("budget_cuts_the_slow_block_and_hands_it_to_the_arm", func(t *testing.T) { + seed(t) + h, hits, canceled := newHandler(t, 5*time.Second) + cfg := &config.Config{EmbedBackfill: config.EmbedBackfillConfig{ + SyncCap: 4, SyncBudget: 300 * time.Millisecond, MaxTokens: 1_000_000, BackoffBase: time.Minute, BackoffCap: time.Hour, + }} + reqCtx, cancelReq := context.WithCancel(ctx) + defer cancelReq() + + started := time.Now() + got := h.backfillPending(reqCtx, nil, "private", h.embedAdmission(), cfg) + elapsed := time.Since(started) + if got != 0 { + t.Fatalf("backfilled = %d, want 0 (the slow block must not count)", got) + } + if elapsed > 2*time.Second { + t.Fatalf("returned after %v, want well inside the 5 s backend delay (budget 300 ms)", elapsed) + } + if reqCtx.Err() != nil { + t.Fatalf("request context is dead (%v); the budget must cut only the embed call", reqCtx.Err()) + } + if hits.Load() != 1 { + t.Errorf("wire hits = %d, want 1", hits.Load()) + } + // The backend saw the cancellation: no 7-minute embed keeps burning CPU + // for a caller that is gone (the live task was cancelled at 30 %). + deadline := time.Now().Add(2 * time.Second) + for canceled.Load() == 0 && time.Now().Before(deadline) { + time.Sleep(20 * time.Millisecond) + } + if canceled.Load() != 1 { + t.Errorf("backend saw %d cancellations, want 1", canceled.Load()) + } + + class, dueNow, attempts := readMemo(t) + if class != string(store.EmbedFailureCallerTimeout) { + t.Errorf("memo class = %q, want %q", class, store.EmbedFailureCallerTimeout) + } + if !dueNow { + t.Errorf("memo next_attempt_at is in the future; a caller_timeout must be due for the background arm at once") + } + if attempts != 1 { + t.Errorf("memo attempts = %d, want 1", attempts) + } + + // Pfad A never picks it again, however much budget it has. + cfg.EmbedBackfill.SyncBudget = 10 * time.Second + h2, hits2, _ := newHandler(t, 0) + if again := h2.backfillPending(ctx, nil, "private", h2.embedAdmission(), cfg); again != 0 { + t.Fatalf("second backfillPending embedded %d, want 0 (caller_timeout blocks belong to the background arm)", again) + } + if hits2.Load() != 0 { + t.Errorf("second handler made %d wire calls, want 0", hits2.Load()) + } + + // The background arm's predicate (the shared one) still sees it. + var visibleToArm int + if err := pool.QueryRow(ctx, + `SELECT count(*) FROM context_blocks + WHERE embedding IS NULL AND NOT is_archived AND title = $1`+store.EmbedFailureExcludedPredicate, title). + Scan(&visibleToArm); err != nil { + t.Fatalf("arm predicate: %v", err) + } + if visibleToArm != 1 { + t.Errorf("background predicate sees %d block(s), want 1 (caller_timeout is due at once for the arm)", visibleToArm) + } + }) + + // A request that is cancelled mid-embed (client gone) is the same case + // for the block: caller_timeout, not wire — the backend did nothing wrong. + t.Run("request_cancel_is_caller_timeout_not_wire", func(t *testing.T) { + seed(t) + h, _, _ := newHandler(t, 5*time.Second) + cfg := &config.Config{EmbedBackfill: config.EmbedBackfillConfig{ + SyncCap: 4, SyncBudget: 10 * time.Second, MaxTokens: 1_000_000, BackoffBase: time.Minute, BackoffCap: time.Hour, + }} + reqCtx, cancelReq := context.WithCancel(ctx) + go func() { + time.Sleep(200 * time.Millisecond) + cancelReq() + }() + if got := h.backfillPending(reqCtx, nil, "private", h.embedAdmission(), cfg); got != 0 { + t.Fatalf("backfilled = %d, want 0", got) + } + class, dueNow, _ := readMemo(t) + if class != string(store.EmbedFailureCallerTimeout) { + t.Errorf("memo class = %q, want %q", class, store.EmbedFailureCallerTimeout) + } + if !dueNow { + t.Errorf("memo must be due at once for the background arm") + } + }) +} diff --git a/go/internal/handler/query.go b/go/internal/handler/query.go index 3ab3d61d..e18abbb5 100644 --- a/go/internal/handler/query.go +++ b/go/internal/handler/query.go @@ -729,7 +729,10 @@ func (h *QueryHandler) HandleQuery(w http.ResponseWriter, r *http.Request) { "request_id", requestID, ) } else if llm.DetectGerman(query) { - slog.Info("german detected, translating", + // The heuristic fires on umlauts and German stopwords, so Swedish, + // Danish and Finnish questions land here too; the translation to + // English is right for them as well, only the old label was not. + slog.Info("non-english heuristic matched, translating", "request_id", requestID, ) translatedQuery, err := llm.TranslateQuery(ctx, h.pool, h.backendPool, querySens, query, ar.ApiKeyID, h.admission()) @@ -1710,12 +1713,34 @@ func (h *QueryHandler) logAccess(ar *auth.AuthResult, results []rrf.SearchResult func (h *QueryHandler) backfillPending(ctx context.Context, floor config.ScopeFloor, scope string, adm embedcache.Admission, cfg *config.Config) int { count := 0 syncCap := cfg.EmbedBackfill.SyncCap + // Time budget next to the count cap (embed_backfill.sync_budget, issue + // force-push.me/damienmoon/ctx/issues/1): SyncCap bounds how MANY blocks a request embeds + // inline, this bounds how LONG. One block that cannot embed inside the + // caller's deadline used to consume the whole deadline — the question's + // own embed was then never admitted and the query answered 500. Every + // block's wire call now runs under what is left of the budget; a block + // that overruns it is memoized caller_timeout and left to the + // background arm (runEmbedBackfillLoop), and the loop ends so the + // question embed keeps the rest of the request. Zero = unbounded, the + // pre-budget shape (explicit opt-out, same convention as SyncCap). + var budgetDeadline time.Time + if budget := cfg.EmbedBackfill.SyncBudget; budget > 0 { + budgetDeadline = time.Now().Add(budget) + } for attempts := 0; syncCap <= 0 || attempts < syncCap; attempts++ { + if !budgetDeadline.IsZero() && !time.Now().Before(budgetDeadline) { + slog.Info("query backfill: sync budget spent, rest left to the background arm", + "budget", cfg.EmbedBackfill.SyncBudget, "backfilled", count) + break + } var blockID, title, content, sens, scope string + // Pfad A's predicate: the shared backoff exclusion PLUS every block + // already memoized caller_timeout (store.EmbedFailureQueryPath + // ExcludedPredicate) — such a block is the background arm's job. err := h.pool.QueryRow(ctx, `SELECT id, title, content, sensitivity, scope FROM context_blocks WHERE embedding IS NULL AND NOT is_archived`+ - store.EmbedFailureExcludedPredicate+store.RetrievalExcludedTypePredicate+` + store.EmbedFailureQueryPathExcludedPredicate+store.RetrievalExcludedTypePredicate+` LIMIT 1`).Scan(&blockID, &title, &content, &sens, &scope) if err != nil { break // No more pending blocks (or error). @@ -1758,9 +1783,17 @@ func (h *QueryHandler) backfillPending(ctx context.Context, floor config.ScopeFl // pool=nil: document embeddings land in the block row, not the cache // (today's semantics — the cache is for repeated query/keyword text). + // The wire call runs under the remaining sync budget; callerCut is + // read BEFORE cancelEmbed, which would make Err() non-nil by itself. + embedCtx, cancelEmbed := ctx, func() {} + if !budgetDeadline.IsZero() { + embedCtx, cancelEmbed = context.WithDeadline(ctx, budgetDeadline) + } vec, served, wireAttempts, wired, err := embedcache.EmbedChain( - ctx, nil, chain, backends.RoleEmbed, embedText, embed.PrefixDocument, + embedCtx, nil, chain, backends.RoleEmbed, embedText, embed.PrefixDocument, embedcache.ReportFunc(llm.PoolReporter(h.backendPool)), adm) + callerCut := err != nil && embedCtx.Err() != nil + cancelEmbed() if wired { // TENANT-DECISION(backfill-attribution): "" → NULL. backfillPending is // query-triggered but maintenance in nature — it embeds whatever blocks @@ -1782,7 +1815,20 @@ func (h *QueryHandler) backfillPending(ctx context.Context, floor config.ScopeFl // failure most often means the backend itself is unavailable, // and hammering it for the rest of the sync-cap budget within // one request helps nobody. - class, normalized := store.ClassifyEmbedError(err) + // A budget or request deadline is a statement about THIS caller, + // not about the backend: class caller_timeout, which the query + // path never re-picks and the background arm picks at once. + // Everything else keeps the wire/oversize classification. + var ( + class store.EmbedFailureClass + normalized string + ) + if callerCut { + class = store.EmbedFailureCallerTimeout + normalized = store.NormalizeEmbedError(class, err.Error()) + } else { + class, normalized = store.ClassifyEmbedError(err) + } // The memo write must survive the request context: the wire // failure above IS frequently the request deadline itself // (canceled context), and a memo lost to that same cancellation diff --git a/go/internal/store/embed_failures.go b/go/internal/store/embed_failures.go index 4d54d09f..2c111372 100644 --- a/go/internal/store/embed_failures.go +++ b/go/internal/store/embed_failures.go @@ -44,6 +44,17 @@ const ( // memo-free "stays unembedded, retried every cycle" semantics for this // case — a pool outage there must not permanently park blocks). EmbedFailureSensitivityIneligible EmbedFailureClass = "sensitivity_ineligible" + // EmbedFailureCallerTimeout marks a block whose embed did not fit the + // query path's sync budget (embed_backfill.sync_budget) or died with + // the request context — the backend is not known to be down, it was + // slow relative to THIS caller's deadline (the tracking issue, force-push.me/damienmoon/ctx/issues/1: a + // 10k-token block on a CPU backend). The memo therefore says nothing + // about the backend and must not park the block on the wire curve: the + // query path (EmbedFailureQueryPathExcludedPredicate) never picks a + // caller_timeout block again, the background arm picks it at once + // (next_attempt_at = now()). A later wire failure in the background + // arm reclassifies it as wire with the ordinary backoff. + EmbedFailureCallerTimeout EmbedFailureClass = "caller_timeout" ) // maxLastErrorLen mirrors migration 113's documented last_error contract: @@ -153,6 +164,27 @@ func RecordEmbedFailure(ctx context.Context, q pgxdb.Execer, blockID string, cla return nil } + if class == EmbedFailureCallerTimeout { + // Eligible for the background arm at once: the backoff curve is + // for a backend that failed, this backend only took longer than + // the caller could wait. Pfad A's own exclusion of this class is + // in the predicate, not in the timestamp. + _, err := q.Exec(ctx, + `INSERT INTO context_embed_failures (block_id, migration_id, attempts, last_error, last_class, next_attempt_at) + VALUES ($1, NULL, 1, $2, $3, now()) + ON CONFLICT (block_id) WHERE migration_id IS NULL + DO UPDATE SET attempts = context_embed_failures.attempts + 1, + last_error = EXCLUDED.last_error, + last_class = EXCLUDED.last_class, + next_attempt_at = now()`, + blockID, normalizedErr, string(class), + ) + if err != nil { + return fmt.Errorf("store: record embed failure (caller_timeout): %w", err) + } + return nil + } + baseSecs := backoffBase.Seconds() capSecs := backoffCap.Seconds() _, err := q.Exec(ctx, @@ -265,3 +297,21 @@ const EmbedFailureExcludedPredicate = ` AND f.migration_id IS NULL AND f.next_attempt_at > now() )` + +// EmbedFailureQueryPathExcludedPredicate is Pfad A's (query.go +// backfillPending) variant of EmbedFailureExcludedPredicate: the same +// backoff exclusion PLUS every block memoized as caller_timeout, for good. +// A block that once overran the sync budget is the background arm's job +// from then on; re-picking it inline would burn the budget of every later +// query for a block the arm is about to embed anyway (and, while the arm +// holds its FOR UPDATE row, embed it a second time in parallel). The +// background arm keeps the plain predicate, so for it a caller_timeout memo +// (next_attempt_at = now()) is immediately eligible. Same positional +// contract as its sibling: bare context_blocks as the outer FROM. +const EmbedFailureQueryPathExcludedPredicate = ` + AND NOT EXISTS ( + SELECT 1 FROM context_embed_failures f + WHERE f.block_id = context_blocks.id + AND f.migration_id IS NULL + AND (f.next_attempt_at > now() OR f.last_class = 'caller_timeout') + )`