These files were not touched by the recent performance and
correctness PRs but had coverage gaps. Adds tests for:
- CLI: serve, queue init/migrate, config loading, cli() entry point
including MigrationRequiredError exit path
- filter: _default_supported_extensions, __call__ watchfiles callback,
FileFilter with supported_extensions=None
- registry: resolve_adhoc_fetcher with bucket-less S3 URI
- migrations: pragma no-cover on unreachable schema upgrade path
(no diff migrations exist until SCHEMA_VERSION > 1)
Workers share the pollers' Source instances for fetch(), so the httpx
clients must be closed only after the pool has stopped. Guards both
run_batch() and serve() against reintroducing the shutdown-order bug.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Workers share the same Source instances as pollers and use them for
fetch(). PollerManager.stop() was closing httpx clients before the
worker pool drained, so in-flight fetches during the shutdown grace
hit a closed client.
- Move source closing out of stop() into a separate close_sources()
- Call close_sources() after _stop_pool() in both serve() and
run_batch()
- Promote aclose() to the Source protocol with no-op defaults for
FS and S3, removing the hasattr duck-typing
The condition only checked claimed jobs, but queued jobs with no
live workers also hang forever. Check live_workers == 0 regardless
of whether outstanding work is queued or claimed.
Rewrite the test to actually crash workers: patch _process to raise
a bare Exception (which _worker_loop doesn't catch), use
worker_count=1 so the single crash leaves live_workers == 0, and
assert the abort log message fires.
The jitter-before-first-sleep block was duplicated verbatim in
PeriodicPoller.run() and FSPoller._sweep_loop(). Move it to
BasePoller._stagger_start() with a named _STAGGER_FRACTION constant.
This also gives a testable seam outside the pragma-no-cover
event-loop glue methods.
The previous test deleted a file mid-iteration, but is_file()
caught it before stat() ran — so the new try/except never executed.
Monkeypatch Path.stat to raise FileNotFoundError on the third call
for the victim path (after is_symlink and is_file pass), simulating
the exact TOCTOU window between is_file() and stat().
has_pending() already uses the leading column of uq_jobs_live
(source_id) with the same WHERE clause. The extra index just adds
write amplification on every insert/claim/complete without improving
reads.
Workers parked on job_available.wait() were not woken by stop(),
causing them to sleep out the full poll_idle_interval_s before
noticing _stop. With the default 1.0s interval, stop() took ~0.8s
instead of ~0.007s.
Notify all waiters on the condition in stop() so idle workers exit
immediately. Add tests for fast job pickup via notification and
fast shutdown with idle workers.
PermissionError is a subclass of OSError, so it was caught by the
broad timeout/io handler and classified as transient. An unreadable
file would retry 5 times then DLQ — permissions don't fix themselves
without operator intervention.
Add an explicit PermissionError check before the OSError catch so
unreadable files go straight to the DLQ.
If sync.upsert() or sync.delete() raises after a job is already
marked succeeded (e.g. disk full, DB locked), the unhandled
exception crashes the worker. The job stays succeeded but sync_state
is stale, and the crashed worker stops processing other jobs.
Wrap the post-success sync_state writes in a try/except. On failure,
log the error and continue. The worst case is a redundant re-ingest
on the next sweep — better than killing the worker.
The drain loop in run_batch() polls counts_by_status() waiting for
queued and claimed counts to reach zero. If all worker tasks crash
(unhandled exception, OOM), claimed jobs stay claimed forever and
the loop never exits — the CLI command hangs.
Check live_workers during the drain loop. If claimed jobs exist but
no workers are alive to process them, log an error and break out.
The stranded jobs will be reaped on the next start.
The expression `str(path.stat().st_mtime_ns) if path.exists() else None`
has a TOCTOU race: the file can be deleted between exists() and stat().
The resulting FileNotFoundError propagates up to _watch_loop's except
handler, which records a breaker failure and terminates the loop — no
more push events are processed until restart.
Replace with a try/except around stat() and return early on
FileNotFoundError. The deletion event from watchfiles will handle
cleanup.
FileNotFoundError is a subclass of OSError, so it was caught by the
broad timeout/io handler and classified as transient. A file deleted
between discovery and fetch would retry 5 times on a file that's
permanently gone, then DLQ with a confusing error message.
Add an explicit FileNotFoundError check before the OSError catch so
deleted files go straight to the DLQ.
A file deleted between os.walk() and path.stat() raises
FileNotFoundError, which propagated uncaught and failed the entire
discover() sweep. With enough failures this trips the circuit
breaker, silencing the poller.
Catch FileNotFoundError around the stat() call and skip the file.
The next sweep (or watchfiles) will emit the DELETE event.
All pollers sharing the same poll_interval_s previously woke up and
swept at exactly the same moment after startup. With 10+ sources
this causes a coordinated spike in listing traffic (S3 LIST, WebDAV
PROPFIND, HTTP HEAD) every interval.
Add a random initial delay of 0-25% of the poll interval after the
first sweep, applied to both PeriodicPoller and FSPoller's sweep
loop. Subsequent sweeps run on the normal fixed interval, now
staggered across sources.
HTTPSource and WebDAVSource previously created a new AsyncClient for
every head(), fetch(), and discover() call — no connection reuse, TLS
renegotiation on every request, and connection pool churn at scale.
Create the client once in __init__ and reuse it for the lifetime of
the source. Add aclose() to both sources, called by PollerManager on
shutdown to cleanly close the connection pool.
has_pending() scans jobs WHERE source_id=? AND status IN
('queued','claimed') on every poller sweep — without an index this
is a full table scan as the jobs table grows. Similarly,
count_succeeded_since() scans by completed_at for the dashboard's
rolling-throughput display.
Add two partial indexes:
- idx_jobs_pending_by_source: covers has_pending() lookups
- idx_jobs_succeeded_completed: covers count_succeeded_since()
Both use CREATE INDEX IF NOT EXISTS so they're idempotent on
existing databases.
Workers previously polled the queue with a fixed 1s sleep between
claim attempts, adding ~500ms average latency to job pickup. Now
JobRepo.job_available (an asyncio.Condition) is notified on every
successful enqueue, waking idle workers immediately. The poll
interval remains as a timeout fallback for stop signals and breaker
state changes.
