Refactor refresh into arq-backed worker queue - #30
Merged
Conversation
The previous refresh script built every task up front and handed them to asyncio.gather, which pinned O(N) coroutine frames in memory and stalled for the duration of the longest package. On larger deployments this caused memory pressure and unbounded run time. The ExecStart was also broken after commit 7aa4caf moved the script into the package. Replace it with an arq worker that pops short-lived (name, version) jobs, so memory is bounded by max_jobs (=5) and each unit of work is independently retryable. The weekly refresh is handled by arq's cron_jobs, removing the need for a systemd timer. - Add pypi_tea.worker (resolve_pkg_version, check_latest, enqueue_refresh, weekly_refresh) and pypi-tea-worker console script - Rewrite pypi_tea.refresh as a thin CLI that enqueues one enqueue_refresh job and exits - Replace pypi-tea-refresh.{service,timer} with a long-running pypi-tea-worker.service (MemoryMax=1500M, Restart=always) - Bump to 0.6.0; pin redis<6 indirectly via arq>=0.26 Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
6 tasks
There was a problem hiding this comment.
Pull request overview
Refactors the package refresh process from a one-shot asyncio.gather run into an arq/Redis-backed worker queue, with refresh now triggered by enqueuing a fan-out job and executed by a bounded worker (including a weekly cron schedule).
Changes:
- Add an arq worker (
pypi_tea.worker) with jobs for per-package-version resolution, latest-version checks, and refresh fan-out scheduling. - Rewrite the refresh CLI (
pypi_tea.refresh) to enqueue a singleenqueue_refreshjob and exit (worker owns concurrency now). - Update deployment to run a persistent
pypi-tea-workersystemd service and remove the old weekly timer approach; bump deps/version accordingly.
Reviewed changes
Copilot reviewed 7 out of 8 changed files in this pull request and generated 3 comments.
Show a summary per file
| File | Description |
|---|---|
| uv.lock | Adds arq and downgrades redis to satisfy arq constraints; updates locked version to 0.6.0. |
| src/pypi_tea/worker.py | New arq worker entrypoint + job definitions + cron scheduling. |
| src/pypi_tea/refresh.py | Replaces in-process refresh logic with “enqueue and exit” CLI. |
| src/pypi_tea/cache.py | Adds get_tracked_packages() helper used by the worker. |
| pyproject.toml | Bumps version to 0.6.0, adds arq dependency and pypi-tea-worker script, adjusts mypy overrides. |
| deploy/setup.sh | Installs/enables the new worker unit in addition to the API service. |
| deploy/pypi-tea-worker.service | Defines the long-running arq worker systemd service with resource limits. |
| deploy/pypi-tea-refresh.timer | Removes the old weekly timer unit. |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
- Replace Cache.get_tracked_packages() (SMEMBERS) with scan_tracked_packages() async iterator (SSCAN), mirroring scan_packages_with_sbom(). Avoids a single large Redis reply on the fan-out path. - enqueue_refresh now parallelises enqueues in batches of 50 via asyncio.gather, so the fan-out stays well under job_timeout even on large tracked sets. Every enqueued child carries a deterministic _job_id (resolve:name@version, check_latest:name) so if the fan-out job times out and retries, children already in-queue or in the result window are not re-enqueued. - Restore --dry-run on pypi-tea-refresh: the flag propagates through to enqueue_refresh, which logs what it would enqueue without actually enqueueing. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
There was a problem hiding this comment.
Pull request overview
Refactors the package refresh workflow to use an arq/Redis-backed worker queue, moving the weekly refresh schedule into the worker (cron) and updating deployment units accordingly to reduce memory pressure from the previous asyncio.gather approach.
Changes:
- Added an arq worker (
pypi-tea-worker) with refresh-related jobs and a weekly cron schedule. - Rewrote
pypi-tea-refreshinto a thin CLI that enqueues a single fan-out job and exits. - Updated Redis cache access to stream tracked packages via
SSCAN, and updated deploy systemd units/scripts for the new worker.
Reviewed changes
Copilot reviewed 7 out of 8 changed files in this pull request and generated 4 comments.
Show a summary per file
| File | Description |
|---|---|
uv.lock |
Adds arq and downgrades redis to satisfy arq constraints; bumps pypi-tea to 0.6.0 in the lock. |
src/pypi_tea/worker.py |
Introduces the arq worker jobs, batching fan-out enqueues, and weekly cron scheduling. |
src/pypi_tea/refresh.py |
Replaces the old refresh implementation with a CLI that only enqueues enqueue_refresh. |
src/pypi_tea/cache.py |
Adds scan_tracked_packages() to stream unique:packages via SSCAN. |
pyproject.toml |
Bumps version, adds arq, adds pypi-tea-worker script, updates mypy overrides. |
deploy/setup.sh |
Installs/enables the new worker systemd service alongside the server. |
deploy/pypi-tea-worker.service |
Adds a long-running arq worker unit with memory/task limits and restart policy. |
deploy/pypi-tea-refresh.timer |
Removes the old weekly timer unit file (refresh now scheduled via arq cron). |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
- refresh.py: logger.exception in cli() to capture traceback; clarify "unexpected arq/Redis state" message (no deterministic _job_id is passed) - worker.py check_latest: only log "enqueued new version" when enqueue_job actually created a job, not when it was deduped - worker.py enqueue_refresh: stream enqueues via a flush buffer instead of materialising the full jobs list; memory now bounded by per-package version lists plus one batch Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
There was a problem hiding this comment.
Pull request overview
Refactors the package refresh flow from an asyncio.gather-based one-shot run into an arq/Redis-backed worker queue, with the refresh CLI acting as a trigger that enqueues a single fan-out job and exits.
Changes:
- Add an arq worker (
pypi-tea-worker) implementing refresh jobs and weekly cron scheduling. - Rewrite
pypi-tea-refreshinto a thin CLI that enqueuesenqueue_refreshvia arq. - Update cache + deployment artifacts to support streaming tracked packages (SSCAN) and run the new worker service.
Reviewed changes
Copilot reviewed 7 out of 8 changed files in this pull request and generated 2 comments.
Show a summary per file
| File | Description |
|---|---|
uv.lock |
Adds arq and updates locked dependencies (including redis downgrade). |
pyproject.toml |
Bumps version to 0.6.0; adds arq dependency and pypi-tea-worker entrypoint; mypy override tweak. |
src/pypi_tea/worker.py |
New arq worker with fan-out refresh job, per-package latest checks, and weekly cron. |
src/pypi_tea/refresh.py |
CLI rewritten to enqueue a refresh job onto arq and exit (supports --existing/--limit/--dry-run). |
src/pypi_tea/cache.py |
Adds scan_tracked_packages() SSCAN iterator to avoid loading the full tracked set at once. |
deploy/setup.sh |
Installs/enables the new worker systemd unit alongside the API service. |
deploy/pypi-tea-worker.service |
Adds long-running worker service unit with resource limits and restart policy. |
deploy/pypi-tea-refresh.timer |
Removes the old weekly systemd timer. |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
- enqueue_refresh: also check the buffer flush threshold inside the inner versions loop so a single package with many tracked versions (existing=True) can't grow the buffer past _ENQUEUE_CONCURRENCY between per-package checks - check_latest: cache `set(known)` once instead of rebuilding it inline on the membership check Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
There was a problem hiding this comment.
Pull request overview
Refactors the weekly refresh workflow from an in-process asyncio.gather run into an arq/Redis-backed job queue with a long-running worker, aiming to reduce memory pressure and improve operational stability in production.
Changes:
- Adds an arq worker (
pypi-tea-worker) with cron-driven weekly refresh and job functions for fan-out, latest-version checks, and per-version resolution. - Rewrites
pypi-tea-refreshinto a thin CLI that enqueues a single fan-out job and exits (preserving--existing,--limit,--dry-run). - Updates deployment artifacts (systemd units + setup script) and adds an SSCAN-based iterator for tracked packages to avoid large
SMEMBERSreplies.
Reviewed changes
Copilot reviewed 7 out of 8 changed files in this pull request and generated 3 comments.
Show a summary per file
| File | Description |
|---|---|
| uv.lock | Adds arq and downgrades redis to satisfy arq’s constraints; bumps project version in lock to 0.6.0. |
| src/pypi_tea/worker.py | Introduces arq worker, job definitions, batching, dedup job IDs, and weekly cron entry. |
| src/pypi_tea/refresh.py | Converts refresh into an enqueue-only CLI using arq.create_pool(). |
| src/pypi_tea/cache.py | Adds scan_tracked_packages() (SSCAN iterator) for large tracked sets. |
| pyproject.toml | Bumps version to 0.6.0, adds arq>=0.26, and adds pypi-tea-worker console script; mypy override tweak for older redis stubs. |
| deploy/setup.sh | Installs/enables the new worker unit alongside the API service. |
| deploy/pypi-tea-worker.service | Adds long-running worker systemd unit with resource limits and restart policy. |
| deploy/pypi-tea-refresh.timer | Removes the old weekly systemd timer. |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
- Parse PYPI_TEA_WORKER_MAX_JOBS defensively so a bad override (non-int, empty, <1) falls back to the default with a warning instead of crashing the worker at import time. - Drop the full per-package versions list from check_latest's job payload. Membership is now checked via a new Cache.is_package_tracked() helper backed by a single SISMEMBER, keeping the check_latest payload O(1) regardless of how many versions a package has tracked. - Skip materialising per-package version lists in enqueue_refresh() when existing=False, since they're only needed to emit resolve jobs. - Document exit code 130 (SIGINT) in refresh.py's module docstring.
There was a problem hiding this comment.
Pull request overview
Refactors the package refresh mechanism from an in-process asyncio.gather run into an arq/Redis-backed worker queue, moving weekly scheduling into arq cron and updating deployment artifacts accordingly.
Changes:
- Introduces an arq worker (
pypi-tea-worker) with jobs for resolving package versions, checking latest versions, and enqueueing refresh fan-out. - Rewrites
pypi-tea-refreshinto a thin CLI that enqueues a single fan-out job and exits (preserving--existing,--limit,--dry-run). - Adds SSCAN-based iteration over tracked packages and updates systemd setup to run the worker instead of a timer.
Reviewed changes
Copilot reviewed 7 out of 8 changed files in this pull request and generated 3 comments.
Show a summary per file
| File | Description |
|---|---|
| uv.lock | Adds arq and adjusts Redis dependency to support it; bumps package version metadata. |
| src/pypi_tea/worker.py | New arq worker implementation with cron-based weekly refresh and bounded fan-out enqueueing. |
| src/pypi_tea/refresh.py | CLI rewritten to enqueue enqueue_refresh via arq instead of doing refresh work inline. |
| src/pypi_tea/cache.py | Adds scan_tracked_packages() to stream unique:packages via SSCAN. |
| pyproject.toml | Bumps version to 0.6.0; adds arq dependency and pypi-tea-worker console script; mypy override tweak. |
| deploy/setup.sh | Installs/enables the new worker systemd unit alongside the API service. |
| deploy/pypi-tea-worker.service | Updates unit to run the long-lived arq worker with resource limits and restart policy. |
| deploy/pypi-tea-refresh.timer | Removed (weekly refresh now handled by arq cron jobs). |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
- Guard check_latest against ValueError from resp.json() so a malformed PyPI response logs a warning and skips the job instead of triggering arq retries over a non-transient failure. - Lower keep_result from 3600 → 1200. Because deterministic _job_id dedupe uses the same Redis result-key window, the previous hour-long setting silently blocked manual reruns for up to an hour after any prior refresh. 1200s still covers the worst-case fan-out retry span (max_tries * job_timeout = 900s) with slack for backoff. (The MAX_JOBS parsing concern Copilot re-flagged was already fixed in the previous commit.)
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
asyncio.gather-over-all-packages refresh with an arq Redis queue: short-lived(name, version)jobs popped by a bounded worker (max_jobs=5) instead of materialising every task up front.cron_jobsentry, so the systemd timer is gone.ExecStartinpypi-tea-refresh.service(left over after7aa4cafmoved the script into the package).Why
On the production deployment the oneshot refresh was causing memory pressure and stability issues:
asyncio.gatherpins O(N) coroutine frames for every tracked package before any work starts.--existing, a single package's versions are re-resolved sequentially while holding a concurrency slot, so popular packages stall progress.ExecStartpointed atscripts/refresh_packages.py, which no longer exists after it was moved into the package — so the service was effectively broken on master.What changed
src/pypi_tea/worker.pywith jobs:resolve_pkg_version(name, version)— one PURL resolution, the atomic pop-and-process unit.check_latest(name, known)— queries PyPI for the latest version and enqueues a resolve if it's new.enqueue_refresh(existing, limit)— fan-out overunique:packages.weekly_refresh— cron wrapper scheduledSun 03:00viacron_jobs.WorkerSettings:max_jobs=5,job_timeout=300s,max_tries=3, configurable viaPYPI_TEA_WORKER_MAX_JOBS.src/pypi_tea/refresh.py— now a thin CLI that enqueues oneenqueue_refreshjob viaarq.create_pool()and exits. Drops--concurrency/--delay(worker concerns now); keeps--existing,--limit, and--dry-run(the last of which now runs inside the fan-out job and logs what it would enqueue).deploy/pypi-tea-refresh.serviceanddeploy/pypi-tea-refresh.timer.deploy/pypi-tea-worker.service:Type=exec,Restart=always,MemoryMax=1500M,MemoryHigh=1200M,TasksMax=128.deploy/setup.shinstalls and enables both units.src/pypi_tea/cache.py— small publicget_tracked_packages()helper so the worker doesn't reach into_client.pyproject.toml: version 0.5.1 → 0.6.0, addsarq>=0.26, addspypi-tea-workerconsole script. arq pinsredis<6, soredisdowngraded 7.4.0 → 5.3.1 — covered the resulting stub regression by addingno-untyped-callto the existingpypi_tea.cachemypy override.Upgrade notes (non-breaking within the repo, breaking for deploys)
pypi-tea-refresh --concurrency/--delayflags are gone.--existing,--limit, and--dry-runare preserved. Any external scripts relying on the dropped flags need updating.pypi-tea-refresh.{service,timer}enabled in/etc/systemd/system/. After deploying 0.6.0 the timer will fire at Sun 03:00 and fail. Before rolling out, on each host:deploy/setup.shto installpypi-tea-worker.service.Known follow-ups (not in this PR)
setup.sh(disable old units automatically).enqueue_refresh.Test plan
uv run ruff check src/ tests/— cleanuv run ruff format src/ tests/— no changesuv run mypy src/— clean (strict)uv run pytest tests/ -v— 79 passed, 5 skipped (unchanged)pypi-tea-refresh --helploadspypi-tea-workerbinary loads and attempts Redis connectionpypi-tea-worker, runpypi-tea-refresh --limit 5, confirm jobs execute and cache is populated🤖 Generated with Claude Code