spc: dedup enrich backfill enqueue against active tasks

[miles0521]

Fixes the first half of #62 (queue flood). The attempts-cap bypass stays open in #62 pending a design call.

Problem

The enrich retry sweeper re-enqueues every due doc after its 600s lease expires, with no awareness of the backfill task it already enqueued. Once queue latency exceeds the lease, duplicates accumulate without bound — field-observed on the on-prem benchmark: 35,729 queued kTaskEnrichBackfill duplicates over ~18h under a flapping LLM provider. The 5s Enqueue debounce window cannot catch a re-submission period measured in minutes.

Fix (design first: F36-LR addendum §3.7.6, hub commit pushed)

• New TaskScheduler::HasActiveTaskFor(ns, doc, task_type): reports whether the most recent task for the triple is still queued/processing/cancelling.
• EnrichRetrySweeper::RunSweepNow: a due doc with an active backfill task is skipped without consuming its lease — it stays due and is re-checked next tick. Queue invariant: at most one active backfill task per doc.
• Terminal tasks do not block recovery: once the previous task completes or fails, a due doc is enqueued again.

Verification

• New SPCPipelineR7Test.EnrichSweeper_SkipsDocsWithActiveBackfillTask: lease-expiry while queued → no duplicate (task rows stay 1); after the task reaches a terminal state → re-enqueue works (rows 2). The test disables the watcher debounce to isolate the sweeper-side guard (with debounce on, the recovery enqueue merges into the just-completed row — the intended duplicate-submission dedup — and would mask the signal).
• Defect injection: reverting only enrich_retry_sweeper.cpp to main flips the new test red; restoring flips it green.
• Adjacent sweep: 46 tests across scheduler/SPC pipeline suites pass.

🤖 Generated with Claude Code

https://claude.ai/code/session_01SRNhu7QnvaQ3RjHQv7ViEC

[ScottSiu1983]

Blocking: HasActiveTaskFor cannot detect every state it promises. It delegates to FindRecentTaskByDocId, whose SQL excludes cancelling and applies a seven-day created_at cutoff. A cancelling or older queued/processing backfill is therefore reported inactive, allowing the sweeper to enqueue another duplicate and violating the stated at-most-one-active invariant. Please query active statuses directly without an age cutoff and add coverage for cancelling and old active tasks.

[miles0521]

Thanks — the review is correct on both counts (excluded cancelling and the created_at cutoff both punched holes in the invariant). Fixed in the new head:

• TaskManager::HasActiveTask(ns, doc, type): direct status IN ('queued','processing','cancelling') probe, no age cutoff; the scheduler guard now routes through it. FindRecentTaskByDocId is left untouched — its shape is right for debounce, just wrong for this guard.
• New SPCPipelineR7Test.EnrichSweeper_GuardSeesCancellingAndOldActiveTasks: (a) processing → cancelling via RequestCancel, (b) a queued task backdated to 2000 (asserted invisible to the recency lookup as a precondition) — both keep the sweeper from enqueuing again.
• Defect injection: reverting to the delegating guard flips the new test red; restoring flips it green. Scheduler/manager/sweeper neighborhood: 73 tests pass.

Field note for context: the earlier head has been running on the on-prem enriched-arm box since 08-14 with the backfill queue bounded at ≤181 rows (vs 35,729 before) — the dedup path itself is doing its job; this closes the status/age gaps in it.

[ScottSiu1983]

Blocking: the new dedup query error path fails open. In RunSweepNow, any HasActiveTaskFor error is treated like inactive because only a successful true result skips. A transient task-store read failure can therefore consume the lease and enqueue a duplicate if the following write succeeds, violating the at-most-one-active invariant. Please fail closed by skipping and logging or propagating the lookup error, and cover that path.

[miles0521]

Fixed in 741f014. You were right that the error path failed open, and it is reachable in production rather than only in tests: TaskManager::HasActiveTask returns kStorageFailed on prepare failure and on any step rc that is neither SQLITE_ROW nor SQLITE_DONE, so SQLITE_BUSY under load lands here.

The lookup now fails closed — an unanswerable check is treated as "an active task may exist", and the doc is skipped with its lease intact, so it stays due and the next tick re-checks. The continue sits above LeaseDocRetries, which is what makes the skip free: nothing consumed, nothing lost.

I took skip-and-log rather than propagating. Propagating would abort the whole namespace batch over one doc's transient read error, and skip-and-log matches the existing ListDueDocs handler a few lines up in the same function.

On the test, one thing worth flagging — the obvious assertion does not work. RunSweepNow returns 0 in both the fixed and the broken build, because once the tasks table is unavailable Enqueue fails too. A test that checks the return value passes against the defect. So the assertion is on the lease, which lives in the namespace store and is untouched by the injection: failing open pushes next_retry_at to now+600, failing closed leaves the doc due at 1. The table is renamed rather than dropped so the second half can restore it and prove the outage was transient-safe — no duplicate left behind, and the doc enqueues normally once the original task goes terminal.

Verification, red/green both observed rather than assumed:

• defect injection — reverting only the sweeper arm to the fail-open form turns the new test red at the lease assertion (test_spc_pipeline_r7.cpp:1205, "guard failed open"); restoring turns it green
• adjacent regression — 94 tests pass across the SPC pipeline, TaskScheduler and TaskManager suites
• the injected failure is real, not mocked around: the run logs CX_ERR_STORAGE_FAILED: HasActiveTask step - skipping with lease intact, next tick re-checks

Note this closes only the queue-flood half of #62. The attempts-cap bypass stays open there — I will follow the counting model you outlined in that issue (count when a backfill task actually begins a provider call, persist the claim before the call, idempotent per task/attempt token, so transport failures and timeouts count while pre-call queue/scheduler/storage failures do not).

Re-requesting review.

[ScottSiu1983]

Blocking: the new check still does not make the at-most-one-active invariant atomic. RunSweepNow is callable by both the background thread and the backfill route, while HasActiveTaskFor releases its locks before LeaseDocRetries and Enqueue. Two concurrent sweeps can both observe no active task; with f42.watcher_debounce_seconds=0, both can then create rows. Please serialize or atomically claim a sweep, or enforce uniqueness at task creation, and add a concurrent-sweep test with debounce disabled.

[miles0521]

Fixed in 924a09b by serializing the sweep — the first of the three directions you offered.

RunSweepNow now takes a dedicated std::mutex for the duration of its body. That covers all three interleavings, including one not named in the review: httplib serves the backfill route from a thread pool, so route-vs-route is a third pairing alongside timer-vs-route. RunSweepNow has exactly three call sites and all are in-process, so this closes the invariant for the single-instance deployment the product supports.

Why a lock rather than uniqueness at task creation. This repo already answers exactly this problem the same way everywhere else it occurs: GcManager::RunOnce/Purge take mu_, and CleanupScheduler carries run_mu_ documented as "advisory lock — one sweep at a time". EnrichRetrySweeper was the only sweeper reachable from both a background thread and an ops route that was missing one. So this is a gap being closed, not a new pattern.

I did evaluate the partial unique index and rejected it on measurement, which I'd rather show than assert. Against a table already holding two duplicate rows:

CREATE UNIQUE INDEX IF NOT EXISTS ux ON tasks(namespace_id, doc_id, task_type)
  WHERE task_type = 4 AND status IN ('queued','processing','cancelling');
-> Error: UNIQUE constraint failed: tasks.namespace_id, tasks.doc_id, tasks.task_type (19)

That DDL would live in CreateTasksTable's sqlite3_exec batch, which aborts at the failing statement and returns kStorageFailed, so TaskManager::Init fails and bootstrap.cpp:709-713 returns 1. The on-prem box carrying the 35,729 duplicates from this issue would not start, and tasks.db is not on the SchemaMigrator framework, so there is nowhere to express a pre-clean step. Two smaller problems on the same index: cancelling has no exit path (no sweep clears it — task_manager.cpp:889/915/941 all filter status='processing'), so one worker crash would pin a doc permanently; and doc_id is nullable and binds as NULL, which a SQLite unique index does not constrain at all, so it would be blind to exactly the submissions that carry no doc_id.

Blocking rather than try_lock, matching GcManager: a missed try_lock could only report zero enqueued, which the route cannot distinguish from "nothing was due" — and that route exists precisely to force a backfill. The mutex is separate from cv_mu_, which RunLoop releases before calling RunSweepNow, so Stop() cannot deadlock against an in-flight sweep. Worst-case shutdown latency is one sweep, bounded by kMaxDocsPerSweep per namespace with no network calls in the loop.

Test — EnrichSweeper_ConcurrentSweepsCreateOneTask. Two threads released from a spin barrier into RunSweepNow("test-ns") (the single-namespace form the route uses), f42.watcher_debounce_seconds=0 as you asked, 120 rounds on a fresh doc each round.

It asserts the row count in the tasks table, not the return value. That distinction matters here: the debounce merge branch also counts toward the returned total, so a returned 1 can still hide a second row — a test written against the return value would pass on the defect.

Verification, red and green both observed:

build	runs	result
lock removed	5	5 failed, every one tripping in round 0 or 1 with concurrent sweeps created 2 backfill tasks
lock restored	5	5 passed

All 5 EnrichSweeper tests pass; adjacent regression is 95 tests across the SPC pipeline, TaskScheduler and TaskManager suites.

One thing I want to flag rather than quietly rely on. The duplicate row is not reachable in a shipped binary: f42.watcher_debounce_seconds has no YAML key, is not among the four keys bootstrap.cpp:685-706 seeds, and no route can set it, so DebounceSeconds() is always 5 — and within that window Enqueue holds mutex_ across find→merge→create while the sweeper submits content_hash == doc_id, so the loser merges. The ordering hazard you identified is real and worth closing; I just don't want the PR to read as if it were fixing a live production duplicate.

Follow-up filed rather than done here: the invariant is still enforced by the sweeper, not the queue. Its long-term home is inside TaskScheduler::Enqueue's existing critical section — task_scheduler.cpp:91 is the only CreateTask call site in src/ and is already under the lock — which would make it hold for every producer, not just this one. That changes shared F42 queue semantics, so it doesn't belong in a PR scoped to the enrich sweeper.

Re-requesting review.

[ScottSiu1983]

Reviewed on 924a09b. The dedicated sweep mutex closes the concurrent RunSweepNow race. The active-task query now covers queued, processing, and cancelling states without an age cutoff, and lookup failures preserve the lease. The concurrent regression test addresses the earlier blocker, and all seven CI checks pass.

[miles0521]

Merged as b007ba91. Thanks for the three rounds — each one closed a hole the previous fix left open, and the last one was the only version that actually held.

The follow-up I mentioned is filed: #78 — moving the invariant into TaskScheduler::Enqueue's existing critical section so it holds for every producer rather than depending on each one serializing itself. It records why the partial unique index was rejected (measured: fails to build on any DB holding duplicates, and that failure aborts CreateTasksTable → Init → bootstrap.cpp:709-713 return 1), so that reasoning does not have to be rediscovered.

Issue #62's remaining half — the attempts-cap bypass — stays open there, and I'll follow the counting model you outlined in that issue.