From 7db0c733389d5773b030cf33d0e62dd2a4b8fec7 Mon Sep 17 00:00:00 2001 From: Umang Date: Thu, 30 Jul 2026 18:33:42 -0500 Subject: [PATCH] =?UTF-8?q?PR=208=20=E2=80=94=20Background=20processing=20?= =?UTF-8?q?and=20recovery?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Makes the system run by itself, and proves it recovers (IMPLEMENTATION_PLAN.md §13's PR 8). Solid Queue runs in its own `queue` database inside the same PostgreSQL container, a `worker` container runs its supervisor, and two recurring tasks fire every 60 seconds. A tick is not a poll. PollEventSourceJob pre-filters with EventSource.poll_due — mode-scoped, enabled, idle, projection arrived — and Github::IngestionRunner still re-reads each source inside its advisory lock and applies §9's five components. At the 300-second cadence four ticks in five cost one indexed SELECT and open no run row. A busy source is INFO and a completed tick, never a failed execution: §2A's poller attempts once, and the next tick is 60 seconds away. Enrichment is scheduled twice over, deliberately. IngestionRunner#call dispatches once per run that created events, after every row is committed and the source lock is released; ReconcilePendingEnrichmentsJob sweeps the committed entity rows every 60 seconds. The enqueue is a hint and the entity rows are the durable record, so work committed before a crash but never enqueued is rediscovered within a minute with no cleanup step. Both go through Github::Enrichment::Dispatch, which schedules at most one cycle per class — the runner picks the entity by fairness under a lease, so N enqueues carry no more information than one, and queue depth is set by §10's hourly allowance rather than by arrival rate. No retry_on anywhere: PollState and EntityState already own durable, ledger-aware retry ladders, and a second uncoordinated one would spend the allowance twice on the same failure. ApplicationJob adds §11's job fields — job_id, job_class, queue, attempt, duration_ms — so a reviewer's trace is job_id → run_id → every ingestion.* line. Solid Queue's schema lands as a migration in db/queue_migrate *and* the regenerated dump, because db:prepare loads a schema file only for an uninitialized database and PR 2's version-0 placeholder already created schema_migrations in every development queue database (verified against one). Recovery tests under spec/recovery/ cover §12's list: work committed but never enqueued, a job delivered twice, a lease left by a crashed worker, contention between pollers, and advisory-lock release on session death — proven with pg_terminate_backend rather than a cooperative close, because a container kill runs no ensure block. spec/queue/ holds the only examples that touch the queue test database; spec/job_boundary_spec.rb keeps it that way. Processing semantics are unchanged and stated exactly: at-least-once execution plus idempotent writes plus unique constraints gives effectively-once persisted outcomes. Not exactly-once execution. Closes #18 Co-Authored-By: Claude Opus 5 (1M context) --- .env.example | 6 +- .github/workflows/ci.yml | 26 ++- Gemfile | 8 + Gemfile.lock | 14 ++ README.md | 136 +++++++++--- app/jobs/application_job.rb | 62 +++++- app/jobs/enrich_actor_job.rb | 24 +++ app/jobs/enrich_repository_job.rb | 15 ++ app/jobs/poll_event_source_job.rb | 94 +++++++++ app/jobs/reconcile_pending_enrichments_job.rb | 23 ++ app/models/event_source.rb | 28 +++ app/services/github/enrichment/dispatch.rb | 103 +++++++++ app/services/github/enrichment_runner.rb | 7 +- app/services/github/ingestion_runner.rb | 43 +++- app/services/github/source_lock.rb | 5 +- bin/enrich | 5 +- bin/jobs | 17 ++ config/application.rb | 40 ++++ config/ci.rb | 11 + config/database.yml | 6 +- config/environments/development.rb | 6 +- config/environments/production.rb | 6 +- config/environments/test.rb | 9 + config/queue.yml | 31 +++ config/recurring.yml | 43 ++++ ...000_add_poll_due_index_to_event_sources.rb | 25 +++ ...0260731100100_create_solid_queue_tables.rb | 165 +++++++++++++++ db/queue_schema.rb | 144 ++++++++++++- db/schema.rb | 3 +- docker-compose.yml | 40 +++- ...nqueue-and-entity-scoped-reconciliation.md | 117 ++++++++++ spec/db/schema_spec.rb | 24 ++- spec/docker_compose_spec.rb | 49 ++++- spec/job_boundary_spec.rb | 40 ++++ spec/jobs/application_job_spec.rb | 113 ++++++++++ spec/jobs/enrich_actor_job_spec.rb | 35 +++ spec/jobs/enrich_repository_job_spec.rb | 35 +++ spec/jobs/poll_event_source_job_spec.rb | 199 ++++++++++++++++++ .../reconcile_pending_enrichments_job_spec.rb | 56 +++++ spec/models/event_source_spec.rb | 65 ++++++ spec/queue/configuration_spec.rb | 103 +++++++++ spec/queue/solid_queue_integration_spec.rb | 84 ++++++++ .../advisory_lock_session_death_spec.rb | 102 +++++++++ spec/recovery/duplicate_job_execution_spec.rb | 88 ++++++++ .../pending_enrichment_recovery_spec.rb | 106 ++++++++++ spec/recovery/source_contention_spec.rb | 80 +++++++ spec/recovery/worker_crash_lease_spec.rb | 93 ++++++++ .../github/enrichment/dispatch_spec.rb | 139 ++++++++++++ spec/services/github/ingestion_runner_spec.rb | 56 ++++- spec/support/advisory_lock_helpers.rb | 72 ++++++- spec/support/queue_helpers.rb | 56 +++++ .../support/shared_examples/enrichment_job.rb | 66 ++++++ spec/support/webmock.rb | 4 +- 53 files changed, 2853 insertions(+), 74 deletions(-) create mode 100644 app/jobs/enrich_actor_job.rb create mode 100644 app/jobs/enrich_repository_job.rb create mode 100644 app/jobs/poll_event_source_job.rb create mode 100644 app/jobs/reconcile_pending_enrichments_job.rb create mode 100644 app/services/github/enrichment/dispatch.rb create mode 100755 bin/jobs create mode 100644 config/queue.yml create mode 100644 config/recurring.yml create mode 100644 db/migrate/20260731100000_add_poll_due_index_to_event_sources.rb create mode 100644 db/queue_migrate/20260731100100_create_solid_queue_tables.rb create mode 100644 docs/adr/0008-post-commit-enqueue-and-entity-scoped-reconciliation.md create mode 100644 spec/job_boundary_spec.rb create mode 100644 spec/jobs/application_job_spec.rb create mode 100644 spec/jobs/enrich_actor_job_spec.rb create mode 100644 spec/jobs/enrich_repository_job_spec.rb create mode 100644 spec/jobs/poll_event_source_job_spec.rb create mode 100644 spec/jobs/reconcile_pending_enrichments_job_spec.rb create mode 100644 spec/queue/configuration_spec.rb create mode 100644 spec/queue/solid_queue_integration_spec.rb create mode 100644 spec/recovery/advisory_lock_session_death_spec.rb create mode 100644 spec/recovery/duplicate_job_execution_spec.rb create mode 100644 spec/recovery/pending_enrichment_recovery_spec.rb create mode 100644 spec/recovery/source_contention_spec.rb create mode 100644 spec/recovery/worker_crash_lease_spec.rb create mode 100644 spec/services/github/enrichment/dispatch_spec.rb create mode 100644 spec/support/queue_helpers.rb create mode 100644 spec/support/shared_examples/enrichment_job.rb diff --git a/.env.example b/.env.example index 53c208c..f084a0d 100644 --- a/.env.example +++ b/.env.example @@ -7,10 +7,12 @@ # DEBUG adds per-request / per-page lines; INFO carries the run summaries. LOG_LEVEL=info -# Rails environment used by the compose app services (web, setup). +# Rails environment used by the compose app services (web, worker, setup). # RAILS_ENV=development -# Connection pool / Puma thread sizing. +# Connection pool / Puma thread sizing. The worker holds a pool against each of the two +# databases config/database.yml declares, so config/queue.yml keeps its thread count +# inside this number. # RAILS_MAX_THREADS=5 # Database connection. Inside compose these are managed by the topology diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 6ba193a..9e79024 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -57,8 +57,9 @@ jobs: with: bundler-cache: true - # Prepares both isolated test databases declared in config/database.yml. The - # queue database is schema-empty until PR 8; a version-0 schema loads cleanly. + # Prepares both isolated test databases declared in config/database.yml, including + # Solid Queue's tables in the queue database (db/queue_migrate, dumped to + # db/queue_schema.rb). - name: Prepare test databases run: bin/rails db:test:prepare @@ -86,3 +87,24 @@ jobs: env: GITHUB_MODE: fixture run: bin/enrich --limit 6 + + # Solid Queue's own validator over config/queue.yml and config/recurring.yml: a + # schedule Fugit cannot parse, or a task naming a class that does not exist, fails + # here rather than in a worker container at 3am. Starts no process. + - name: Validate the queue configuration + run: bin/rails solid_queue:check + + # The supervisor's own boot path, for the reason the two smokes above exist: the suite + # never runs bin/jobs, so nothing else would catch a queue database without its schema, + # a scheduler that cannot register its recurring tasks, or a TERM it does not honour. + # Deliberately not long enough to guarantee a tick — the 60-second schedule would make + # that a 70-second CI step — so what it proves is boot, registration and clean + # shutdown; the tick's own behaviour is spec/jobs' business. + # + # GITHUB_MODE=fixture is not optional: this boots a real scheduler that can run a real + # poll, in a separate process WebMock cannot see, and a live-mode worker would spend a + # runner's unauthenticated quota on every push. + - name: Smoke test the worker supervisor + env: + GITHUB_MODE: fixture + run: timeout --preserve-status --signal=TERM 20 bin/jobs diff --git a/Gemfile b/Gemfile index 6d64521..b0068c3 100644 --- a/Gemfile +++ b/Gemfile @@ -7,6 +7,14 @@ gem "pg", "~> 1.1" # Use the Puma web server [https://github.com/puma/puma] gem "puma", ">= 5.0" +# The Active Job backend pinned by IMPLEMENTATION_PLAN.md §2A: PostgreSQL-backed, no +# Redis, recurring tasks drive polling, and it runs in its own `queue` database inside the +# same Postgres container (config/database.yml declares it; db/queue_migrate carries its +# schema). That separation is what makes §8 step 10's post-commit enqueue necessary — an +# enqueue cannot join the business transaction, so the committed entity rows are the +# durable record of pending work and Github::Enrichment::Dispatch is only a hint. +gem "solid_queue", "~> 1.5" + # HTTP client for every live GitHub request, pinned by IMPLEMENTATION_PLAN.md §2A. # Reached only through Github::Transports::Faraday, and configured with no middleware: # retries and redirects belong to Github::RequestExecutor because every attempt diff --git a/Gemfile.lock b/Gemfile.lock index 3b9f828..d70254c 100644 --- a/Gemfile.lock +++ b/Gemfile.lock @@ -102,12 +102,17 @@ GEM drb (2.2.3) erb (6.0.6) erubi (1.13.1) + et-orbi (1.4.0) + tzinfo faraday (2.14.3) faraday-net_http (>= 2.0, < 3.5) json logger faraday-net_http (3.4.4) net-http (~> 0.5) + fugit (1.13.0) + et-orbi (~> 1.4) + raabro (~> 1.4) globalid (1.4.0) activesupport (>= 6.1) hashdiff (1.2.1) @@ -184,6 +189,7 @@ GEM public_suffix (7.0.5) puma (8.0.2) nio4r (~> 2.0) + raabro (1.5.0) racc (1.8.1) rack (3.2.6) rack-session (2.1.2) @@ -285,6 +291,13 @@ GEM rubocop-rails (>= 2.30) ruby-progressbar (1.13.0) securerandom (0.4.1) + solid_queue (1.5.1) + activejob (>= 7.1) + activerecord (>= 7.1) + concurrent-ruby (>= 1.3.1) + fugit (~> 1.11) + railties (>= 7.1) + thor (>= 1.3.1) thor (1.5.0) timeout (0.6.1) tsort (0.2.0) @@ -327,6 +340,7 @@ DEPENDENCIES rails (~> 8.1.3) rspec-rails rubocop-rails-omakase + solid_queue (~> 1.5) tzinfo-data webmock diff --git a/README.md b/README.md index 8690d45..8ca7d31 100644 --- a/README.md +++ b/README.md @@ -10,7 +10,7 @@ and worker crashes. ## Status -Enrichment budget and fairness stage (PR 7). The Rails 8.1 API, PostgreSQL, Docker +Background processing and recovery stage (PR 8). The Rails 8.1 API, PostgreSQL, Docker Compose topology, health endpoints, and structured JSON logging landed in PR 2; the seven core tables and their idempotent write paths in PR 3; the chain every GitHub request flows through — request gate, class-aware budget ledger, SSRF URL policy, @@ -18,17 +18,18 @@ live and offline transports, fixture corpus, per-source advisory lock, event-sou adapter — in PR 4; the processor registry, the tolerant `PushEvent` parser, the quarantine taxonomy, the ingest transaction, and the one-shot command in PR 5; `Link`-header pagination, the page-one ETag, the five independent components behind -`effective_poll_time`, and global-versus-class blocking in PR 6. - -This stage fills the stubs in. Actors and repositories are now fetched from the -URLs their own push events carried, under an hourly allowance split fairly between -the two classes — because one observed live page referenced ~89 distinct actors -against ~92 distinct repositories, and a queue ordered purely by recency would -starve actor enrichment to zero. It adds the entity state machine and its write -matrix, the per-class fairness guarantees with eligibility-aware borrowing, -newest-first eligibility, the freshness cache and its refresh TTLs, `skipped_budget` -with distinct-event reactivation, error classification that tells an entity failure -from a source failure, and `effective_enrichment_time`. +`effective_poll_time`, and global-versus-class blocking in PR 6; the entity state +machine, per-class fairness with eligibility-aware borrowing, the freshness cache and +its refresh TTLs, `skipped_budget` with distinct-event reactivation, and +`effective_enrichment_time` in PR 7. + +**This stage makes the system run by itself, and proves it recovers.** Solid Queue +now runs in its own `queue` database inside the same PostgreSQL container, a `worker` +container runs its supervisor, and two recurring tasks fire every 60 seconds: +`PollEventSourceJob`, which polls each source that §9's schedule says is due, and +`ReconcilePendingEnrichmentsJob`, which sweeps the committed entity rows for +enrichment work and schedules a cycle per class. A completed poll that created events +schedules enrichment immediately, after commit. **Enrichment is bounded best-effort sampling, and says so.** Against ~2,172 cold entity requests an hour of demand and 40 available, partial coverage is the design @@ -36,10 +37,15 @@ rather than a shortfall — `skipped_budget` is a normal documented outcome, and `bin/enrich` prints the per-class usage so the sampling rate is visible instead of a mysteriously growing queue (plan §10). -**Nothing fires either schedule yet.** `docker compose run --rm ingest` and -`docker compose run --rm enrich` are still the only things that poll and enrich; the -always-on `worker` container, its recurring task, and the entity-scoped reconciler -are PR 8. +**With the default `GITHUB_MODE=live`, `docker compose up` starts spending real +unauthenticated quota** — twelve poll requests an hour at the default cadence, plus +enrichment inside its allowance. That is the intended runtime behaviour (plan §2A); +`GITHUB_MODE=fixture docker compose up --build` runs the same flow entirely offline. + +The processing guarantee is unchanged and stated exactly: at-least-once execution +plus idempotent writes plus unique constraints gives **effectively-once persisted +outcomes**. This system does not claim exactly-once execution — a job may run twice, +and the second run changes nothing. Ingestion capabilities land PR by PR; each README section below is completed by the pull request that ships the capability it documents. The authoritative @@ -66,6 +72,9 @@ This starts, in dependency order (plan §2A): queue databases 3. `web` — the Rails API on http://localhost:3000, started only after `setup` completes successfully +4. `worker` — the Solid Queue supervisor, started on the same condition. + **Continuous polling begins here**: its scheduler fires the 60-second tick, so + the first poll happens within a minute and every 300 seconds after that. Verify it is healthy: @@ -83,12 +92,12 @@ development databases): docker compose run --rm test ``` -Rails and application logs (requests, and jobs from PR 8 onward) are one -structured JSON stream; PostgreSQL and Puma startup output remain their own -plain-text formats: +Rails, Active Job and application logs are one structured JSON stream; +PostgreSQL and Puma startup output remain their own plain-text formats: ```bash docker compose logs -f +docker compose logs -f worker # the poll tick, enrichment cycles, reconciliation ``` Stop everything (add `-v` to also drop the database volume): @@ -392,6 +401,23 @@ system — plus a line per persisted, duplicate and ignored event. Every line carries the run's `run_id`, except `ingestion.not_due`: a poll the schedule turned away opens no run, so it reports `event_source_id` instead. +Background work adds the job vocabulary: `job.completed` for every job the worker +runs — carrying `job_id`, `job_class`, `queue`, `attempt`, `duration_ms` and the +identifiers that job produced — and `job.failed` with the error class and message +when one raises. `ingestion.source_busy` reports a tick that found the source owned +by another poller, `ingestion.cycle_failed` one source that failed without stopping +the tick, and `enrichment.dispatched` is the reconciliation summary: what it +scheduled, what blocked it, and the per-class state counts and share usage. A tick +that scheduled nothing keeps that line at debug, so an exhausted window does not +emit a line a minute for the rest of the hour. + +The trace is one hop: a `job_id` on `job.completed` gives the `run_id`s that job +opened, and every `ingestion.*` line carries the `run_id`. + +```bash +docker compose logs worker | grep -E 'job\.(completed|failed)|enrichment\.dispatched' +``` + ## Environment variables Compose runs with working defaults — no `.env` file is required. The template @@ -409,7 +435,7 @@ is [`.env.example`](.env.example). | `MAX_HTTP_RETRIES` | `2` | Retries after a 5xx or network timeout. Each retry is a fresh budget reservation (plan §10) | | `MAX_REDIRECTS` | `2` | Redirect hops followed per request, each re-validated and separately reserved | | `SOURCE_LOCK_WAIT_SECONDS` | `30` | How long the one-shot waits for a busy source lock; the poller attempts once (plan §9) | -| `POLL_INTERVAL_SECONDS` | `300` | The poll cadence, and an allowance-formula input. A source polled at T is due again at T + this; an unforced run before then is deferred rather than made. Nothing fires the cadence automatically until PR 8 (plan §9, §10) | +| `POLL_INTERVAL_SECONDS` | `300` | The poll cadence, and an allowance-formula input. A source polled at T is due again at T + this; an unforced run before then is deferred rather than made. The worker's 60-second tick checks the schedule; it does not replace it (plan §9, §10) | | `MAX_PAGES_PER_POLL` | `1` | How many `Link`-followed pages one poll may fetch, and an allowance-formula input. Raising it trades enrichment allowance for capture depth: at `3` the poll allowance becomes 36 attempts an hour and enrichment drops to 16 (plan §9, §10) | | `ENABLED_LIVE_SOURCE_COUNT` | `1` | Allowance-formula input: live sources sharing one per-IP budget | | `RATE_LIMIT_RESERVE` | `8` | Requests per hour left deliberately unspent (plan §10) | @@ -558,15 +584,50 @@ EnrichmentRunner ────────────────┘ │ Decisions behind this are recorded in [`docs/adr/`](docs/adr/): advisory locks and the gate (0002), the source and transport seams (0003), the class-aware ledger (0004), at-least-once processing with idempotent -writes (0005), decomposed poll deferral state (0006), and enrichment fairness shares -and borrowing (0007). +writes (0005), decomposed poll deferral state (0006), enrichment fairness shares +and borrowing (0007), and post-commit enqueue with entity-scoped reconciliation (0008). + +## Continuous ingestion + +The `worker` container runs one Solid Queue supervisor: a dispatcher, the worker +threads from [`config/queue.yml`](config/queue.yml), and a scheduler running +[`config/recurring.yml`](config/recurring.yml). Two tasks fire every 60 seconds. + +**A tick is not a poll.** `PollEventSourceJob` selects the sources whose cached +`next_poll_at` has arrived, and `Github::IngestionRunner` then re-reads each one +inside its advisory lock and applies §9's five components before spending anything. +At the default 300-second cadence roughly four ticks in five cost one indexed +`SELECT` and nothing else. The tick exists so a source that becomes due at T is +polled within a minute of T — not so that polls happen every minute, which the +allowance formula (twelve poll requests an hour, no headroom) could not pay for. +A source another process is polling is reported at INFO and left alone; the tick +never retries it, because the next tick is 60 seconds away. + +**Enrichment is scheduled twice over, deliberately.** A run that created events +schedules one cycle per class as soon as its rows are committed and its advisory +lock is released. That enqueue is a *hint*: the durable record of pending work is the +entity rows themselves, so `ReconcilePendingEnrichmentsJob` sweeps them every 60 +seconds and schedules a cycle for any class that has claimable work and is not +blocked by the ledger. Work committed before a crash but never enqueued is +rediscovered on the next tick — no operator step, no cleanup job, and no queue +inspection (plan §8, [ADR 0008](docs/adr/0008-post-commit-enqueue-and-entity-scoped-reconciliation.md)). + +Each cycle enriches at most one entity, chosen by §10's fairness policy under a +lease, so a backlog of ninety pending actors is one queued job rather than ninety. +Steady state at the defaults: twelve polls an hour, and at most forty enrichment +requests an hour split between the two classes. + +**Crashes need no cleanup.** A killed worker's PostgreSQL session dies with it, and +the source advisory lock goes with the session; its claim lease is a timestamp on +`next_retry_at` that expires by arithmetic; and any job it was running may run again, +because every write on the path is idempotent. That is at-least-once execution with +effectively-once persisted outcomes — never exactly-once execution. ## Planned contents | Section | Lands with | |---|---| | Data model reference | PR 12 (design brief) | -| Continuous ingestion behavior — what the always-on poller does on a schedule | PR 8 | | Full fixture scenario matrix (retries, rate limits, redirects) | PR 11 | | API and database inspection examples | PR 10, 12 | | Crash-recovery verification (container kills) | PR 11, 12 | @@ -575,15 +636,42 @@ and borrowing (0007). ## Reviewer commands ```bash -docker compose up --build # available now +docker compose up --build # available now — starts continuous polling docker compose run --rm test # available now (real suite; runs in CI too) +docker compose logs -f worker # available now — the tick, cycles, reconciliation docker compose logs -f # available now docker compose run --rm ingest # available now — one ingestion cycle docker compose run --rm enrich --limit 6 # available now — up to six enrichment cycles +GITHUB_MODE=fixture docker compose up --build # available now — the whole system, no network GITHUB_MODE=fixture docker compose run --rm ingest # available now — deterministic, no network GITHUB_MODE=fixture docker compose run --rm enrich --limit 6 ``` +The queue is a database, so it is inspectable with the same tool as everything else: + +```bash +docker compose exec db psql -U postgres -d github_push_ingestor_queue_development \ + -c "SELECT key, class_name, schedule FROM solid_queue_recurring_tasks;" \ + -c "SELECT class_name, count(*) FROM solid_queue_jobs GROUP BY 1;" \ + -c "SELECT kind, name, last_heartbeat_at FROM solid_queue_processes;" +``` + +Recovery is watchable in under a minute — stop the worker, put the entities back into +`pending`, empty the queue (the crash), and start it again: + +```bash +docker compose stop worker +docker compose exec db psql -U postgres -d github_push_ingestor_development \ + -c "UPDATE github_actors SET enrichment_status = 'pending', fetched_at = NULL, next_retry_at = NULL;" +docker compose exec db psql -U postgres -d github_push_ingestor_queue_development \ + -c "TRUNCATE solid_queue_jobs CASCADE;" +docker compose start worker +docker compose logs -f worker # enrichment.dispatched, then enrichment.completed +``` + +Container-kill verification against the restart policies is `IMPLEMENTATION_PLAN.md` +§15's reviewer step and lands with PR 11; nothing above claims it. + ## Development AI-assisted development guidance for this repository lives in diff --git a/app/jobs/application_job.rb b/app/jobs/application_job.rb index d394c3d..559dbde 100644 --- a/app/jobs/application_job.rb +++ b/app/jobs/application_job.rb @@ -1,7 +1,61 @@ +# The base every job in this application inherits (IMPLEMENTATION_PLAN.md §5, §11). +# +# Two things live here and nothing else: the enqueue-after-commit contract, and the job half +# of §11's common log fields — "timestamp, level, service, environment, event name, run_id, +# job ID, ...". Story 4 asks for job IDs in the logs, and this is where they enter the +# stream, once, for every job. +# +# **No retry_on, deliberately, anywhere in this application.** Every job here can spend +# GitHub request budget, and both retry ladders already exist and are durable: +# Github::Ingestion::PollState writes consecutive_failures + retry_not_before_at for a +# source, and Github::Enrichment::EntityState writes next_retry_at for an entity. A second, +# uncoordinated Active Job ladder would re-poll a source whose backoff was just written and +# spend the hourly allowance twice on the same failure. The 60-second recurring tick is the +# retry — an escaped exception is a defect, so it fails the execution, job.failed says why, +# and the next tick starts from committed state. +# +# No discard_on ActiveJob::DeserializationError either: no job here takes a record argument. class ApplicationJob < ActiveJob::Base - # Automatically retry jobs that encountered a deadlock - # retry_on ActiveRecord::Deadlocked + # §2A's enqueue semantics, stated on the class a reader of a job actually opens. Solid + # Queue runs in its own database, so an enqueue can never join the business transaction; + # this makes the boundary explicit rather than incidental, and it holds even for a future + # caller that enqueues from inside a transaction the way this application's call sites + # deliberately do not. + self.enqueue_after_transaction_commit = true - # Most jobs are safe to ignore if the underlying records are no longer available - # discard_on ActiveJob::DeserializationError + around_perform do |job, block| + started = job.monotonic_now + Rails.logger.debug(event: "job.started", **job.log_context) + + block.call + + Rails.logger.info(event: "job.completed", **job.log_context, + duration_ms: job.elapsed_ms(started), **job.outcome) + rescue StandardError => error + # Logged and re-raised: Solid Queue has to see the failure to record it, and §11 wants + # the reason in the same stream as everything else rather than only in + # solid_queue_failed_executions. + Rails.logger.error(event: "job.failed", **job.log_context, duration_ms: job.elapsed_ms(started), + error_class: error.class.name, error_message: error.message) + raise + end + + # §11's common fields for a job. `attempt` is Active Job's own execution counter, which it + # increments in perform_now *before* these callbacks run — so it already reads 1 on a first + # delivery and 2 on a redelivery after a crash, which is the at-least-once behaviour §8 + # describes, made visible. + def log_context + { job_id: job_id, job_class: self.class.name, queue: queue_name, attempt: executions } + end + + # Identifiers a job wants on its completion line, so a reviewer's trace is one hop: + # job_id → run_id → every ingestion.* line, all of which already carry run_id. A job sets + # @outcome; one that does not simply reports nothing extra. + def outcome + @outcome.to_h + end + + def monotonic_now = Process.clock_gettime(Process::CLOCK_MONOTONIC) + + def elapsed_ms(started) = ((monotonic_now - started) * 1000).round(1) end diff --git a/app/jobs/enrich_actor_job.rb b/app/jobs/enrich_actor_job.rb new file mode 100644 index 0000000..fc873b9 --- /dev/null +++ b/app/jobs/enrich_actor_job.rb @@ -0,0 +1,24 @@ +# One actor enrichment cycle (IMPLEMENTATION_PLAN.md §5, §8 step 10). +# +# It takes no actor id, and that is the design rather than an omission: +# Github::EnrichmentRunner enriches at most one entity per call and *chooses* it through +# §10's fairness policy and a FOR UPDATE SKIP LOCKED lease, newest-first. An id-addressed job +# would have to bypass that ordering to honour its argument, which is how a repository flood +# starves actors. So the job says "do one actor's worth of work" and the runner decides +# whose — which is also why a duplicate delivery is harmless: it is one more cycle, and it +# finds either different work or none. +# +# It never takes a source lock (§8 step 1: "enrichment jobs skip this step — they take only +# the request gate"), and Github::LockOrder enforces that structurally. +class EnrichActorJob < ApplicationJob + def perform + result = Github::EnrichmentRunner.new.call(entity_class: GithubActor) + + # idle and deferred are ordinary outcomes, not errors: nothing was eligible, or the + # ledger refused. Github::EnrichmentRunner already logged the cycle; this joins it to the + # job. Errors::FixtureMiss and anything unexpected propagate — the runner released the + # lease before re-raising, and ApplicationJob turns it into job.failed. + @outcome = { entity_type: result.entity_type, github_actor_id: result.github_id, + enrichment_outcome: result.status }.compact + end +end diff --git a/app/jobs/enrich_repository_job.rb b/app/jobs/enrich_repository_job.rb new file mode 100644 index 0000000..3f784b7 --- /dev/null +++ b/app/jobs/enrich_repository_job.rb @@ -0,0 +1,15 @@ +# One repository enrichment cycle (IMPLEMENTATION_PLAN.md §5, §8 step 10). EnrichActorJob's +# twin, and separate rather than one parameterised class because §5 names both and a class +# per queue-visible unit of work is what makes `SELECT class_name, count(*) FROM +# solid_queue_jobs GROUP BY 1` answer a reviewer's question. +# +# It takes no repository id, for the reason EnrichActorJob's comment gives: the entity is +# chosen by §10's fairness policy under a lease, not by the caller. +class EnrichRepositoryJob < ApplicationJob + def perform + result = Github::EnrichmentRunner.new.call(entity_class: GithubRepository) + + @outcome = { entity_type: result.entity_type, github_repository_id: result.github_id, + enrichment_outcome: result.status }.compact + end +end diff --git a/app/jobs/poll_event_source_job.rb b/app/jobs/poll_event_source_job.rb new file mode 100644 index 0000000..b1a838d --- /dev/null +++ b/app/jobs/poll_event_source_job.rb @@ -0,0 +1,94 @@ +# The recurring poll tick (IMPLEMENTATION_PLAN.md §2A): "Solid Queue recurring task fires +# every 60s → PollEventSourceJob computes effective_poll_time and no-ops unless a poll is +# due". config/recurring.yml is what fires it. +# +# **A tick is not a poll.** Against the default POLL_INTERVAL_SECONDS (300) roughly four +# ticks in five find nothing due, and those cost one indexed SELECT: EventSource.poll_due is +# a pre-filter, and a row it skips writes no ingestion_runs row, takes no lock, and spends no +# budget. The 60-second cadence exists so a source that becomes due at T is polled within a +# minute of T, not so that polls happen every minute — §10's allowance formula grants twelve +# poll requests an hour with zero headroom. +# +# The authority is still Github::IngestionRunner, which reloads the source *inside* its +# advisory lock before deciding. This job's scope only decides what is worth asking about. +# +# It takes no argument. §9's multi-source case is one row per source_type today, the request +# gate makes outbound concurrency exactly one application-wide, and Solid Queue concurrency +# limits keyed by source id (§9's third bullet) are deliberately not used: a semaphore has a +# fixed duration, so a container killed mid-poll would suppress that source until it expired, +# where the session advisory lock is released by PostgreSQL the moment the backend dies. PR 8 +# is the PR about surviving container kills; a weaker duplicate of a lock we already hold +# would be a regression. Revisit in PR 11, where multi-poller tests could justify one. +class PollEventSourceJob < ApplicationJob + # Facts about the process rather than about one source: continuing to the next source + # would only repeat them, and a tick that "completed" after boot-level breakage would be a + # lie. The same line Github::Ingestion::PageWriter::FATAL_ERRORS and + # Github::Ingestion::OneShot::REFUSING_ERRORS draw. + FATAL_ERRORS = [ + Github::Errors::ConfigurationError, Github::Errors::FixtureCorpusError, + Github::Errors::LockOrderViolation, Github::Errors::ReentrantLock, + Github::Errors::LockSessionChanged, + ActiveRecord::ConnectionNotEstablished, ActiveRecord::ConnectionFailed + ].freeze + + def perform + now = Time.current + # Nothing seeds event_sources — Github::Ingestion::SourceProvisioner's comment explains + # why lazily-at-the-point-of-use is the only correct answer — so on a clean checkout this + # is what makes the worker able to poll at all. Every call after the first is one SELECT. + Github::Ingestion::SourceProvisioner.ensure!(now: now) + + sources = due_sources(now: now) + results = sources.filter_map { |event_source| poll(event_source) } + + # run_ids rather than a count of results, because §7's rule is that a run row exists iff + # the process tried to reach GitHub: a source the runner found not-due after all returns a + # Result with no run_id, and counting it as a poll would say a request happened. + # sources_skipped is the contention-and-failure half, which has no run row either but for + # a reason an operator acts on. + @outcome = { sources_due: sources.size, sources_skipped: sources.size - results.size, + run_ids: results.filter_map(&:run_id) } + end + + private + + # Scoped to the current mode's source_type. A development database routinely holds both + # rows — the README's reviewer path creates a github_fixture_events source — and a live + # worker polling the fixture row would raise Errors::FixtureMiss once a minute forever. + def due_sources(now:) + source_type = Github::EventSources::Base.for_mode(Github.configuration.mode).source_type + + EventSource.poll_due(source_type: source_type, now: now).to_a + end + + # @return [Github::IngestionRunner::Result, nil] nil when this source contributed no + # attempt — a busy source or one that failed in a way the tick can survive. + def poll(event_source) + Github::IngestionRunner.new.call(event_source: event_source) + rescue *FATAL_ERRORS + raise + rescue Github::Errors::SourceBusy + # §2A pins the poller's contract: "attempts once and exits if unavailable". §11 lists + # "source lock acquired/busy" at INFO, and this stays INFO rather than becoming a failed + # execution because the system's own mutual exclusion working is not a defect — a raise + # here would put a row in solid_queue_failed_executions every minute a one-shot ran long. + Rails.logger.info(event: "ingestion.source_busy", event_source_id: event_source.id, **log_context) + nil + rescue Github::Errors::FixtureMiss => error + # A corpus gap is an authoring bug (§6 requires it raised rather than laundered), but it + # is a fact about *this source's* scenario. Fixture mode is offline, so the tick reports + # it and keeps going rather than failing every remaining source with it. + cycle_failed(event_source, error) + rescue StandardError => error + # Already durable by the time this is reached: IngestionRunner finalizes the run row and + # PollState writes the source's backoff before re-raising. One bad source must not + # abandon the others in this tick. + cycle_failed(event_source, error) + end + + def cycle_failed(event_source, error) + Rails.logger.error(event: "ingestion.cycle_failed", event_source_id: event_source.id, + error_class: error.class.name, error_message: error.message, **log_context) + nil + end +end diff --git a/app/jobs/reconcile_pending_enrichments_job.rb b/app/jobs/reconcile_pending_enrichments_job.rb new file mode 100644 index 0000000..4f02b27 --- /dev/null +++ b/app/jobs/reconcile_pending_enrichments_job.rb @@ -0,0 +1,23 @@ +# §8 step 11 — "Reconcile entities whose enrichment was not scheduled or completed" — as the +# recurring sweep behind §2A's outbox-style recovery. config/recurring.yml fires it every 60 +# seconds. +# +# **This is the crash-recovery mechanism, and its input is committed entity state.** The +# post-commit enqueue in Github::IngestionRunner is a hint; a process killed between the +# COMMIT and the enqueue, a worker that was down for an hour, a job that failed permanently +# — all of them lose the hint, and none of them lose the work, because the entity rows still +# say `pending` and the partial index that answers that predicate has existed since PR 3. +# +# **Entity-scoped, structurally.** It reads github_actors, github_repositories and one +# github_api_budget row through Github::Enrichment::Dispatch, and never push_events. §8's +# words are "a small, entity-scoped set, not N event rows per entity": fifty events +# referencing one actor are one candidate here, not fifty. +# +# It enqueues nothing when nothing is claimable, and nothing while §9's global block or the +# derived class block is in force — so an exhausted window costs one indexed EXISTS per class +# per minute rather than a queue full of cycles the ledger would refuse. +class ReconcilePendingEnrichmentsJob < ApplicationJob + def perform + @outcome = Github::Enrichment::Dispatch.call(reason: "reconcile") + end +end diff --git a/app/models/event_source.rb b/app/models/event_source.rb index d832174..8487eff 100644 --- a/app/models/event_source.rb +++ b/app/models/event_source.rb @@ -28,6 +28,34 @@ class EventSource < ApplicationRecord enum :status, STATUSES.index_by(&:itself), validate: true + # PR 8's recurring tick asks this before it asks anything else, and it is a **pre-filter, + # never the decision**. Github::PollSchedule reading the four components under the source + # lock stays the authority — Github::IngestionRunner reloads the row inside the lock + # precisely so a decision is made against committed state. + # + # Filtering on next_poll_at is safe because that column can only be conservative. It is a + # projection written at the end of each run from values that had already been committed, + # and §9's components only ever move later or are cleared by the run that clears them — + # so a row this scope skips was genuinely not due, and no source can be stranded by it. + # The cost of being wrong in the other direction is nil: the runner re-checks and returns + # `deferred` without opening a run row. + # + # source_type is not a nicety. A development database routinely holds two rows — the + # README's reviewer path creates a github_fixture_events source with + # `GITHUB_MODE=fixture docker compose run --rm ingest` — and a live worker polling the + # fixture row (or the reverse) would either be refused by Github::UrlPolicy or raise + # Errors::FixtureMiss, once a minute, forever. + # + # enabled/status are excluded here rather than left to the runner because + # IngestionRunner#out_of_service warns on every attempt: a failed source is + # operator-recoverable only, so the tick would emit a warning a minute until someone + # looked, burying the lines §11 asks reviewers to read. + scope :poll_due, ->(source_type:, now:) { + where(source_type: source_type, enabled: true, status: "idle") + .where("next_poll_at IS NULL OR next_poll_at <= :now", now: now) + .order(:id) + } + validates :source_type, :status, presence: true validates :consecutive_failures, numericality: { greater_than_or_equal_to: 0 } end diff --git a/app/services/github/enrichment/dispatch.rb b/app/services/github/enrichment/dispatch.rb new file mode 100644 index 0000000..b743f1e --- /dev/null +++ b/app/services/github/enrichment/dispatch.rb @@ -0,0 +1,103 @@ +module Github + module Enrichment + # §8 step 10's enqueue and step 11's reconciliation, as one rule with two callers. + # + # Github::IngestionRunner calls it after a run whose events committed; PR 8's recurring + # ReconcilePendingEnrichmentsJob calls it every 60 seconds. Both ask the same question — + # "is there durable enrichment work this class could do right now?" — and the answer is + # read from the committed entity rows and the ledger, never from the queue. That is what + # makes the enqueue a *hint*: §2A's outbox-style recovery says "the committed entity + # state is the durable record of pending work", so a process killed between the COMMIT + # and the enqueue loses the hint and never the work. + # + # **At most one job per class per call**, however deep the backlog. §5 gives each class + # one job and Github::EnrichmentRunner enriches at most one entity per call, so the + # queue depth that matters is set by §10's hourly allowance (40 at the defaults), not by + # how fast jobs can be created. One live page carries ~90 distinct actors and ~90 + # distinct repositories; enqueuing per created event would put ~2,400 argument-identical + # cycles an hour on a queue that can spend 40 requests, and every surplus one would run + # the age-out sweep and the fairness reads to be told no. The reconciler's 60-second + # cadence is what refills the pipeline instead — it is faster than the budget can be + # spent, and it self-limits when the budget is gone. + # + # It takes no lock, opens no transaction, and makes no request. + class Dispatch + # Job classes by name, constantized at the call, for EntityType's reason: a constant + # holding the class object would pin it across a development reload. + JOBS = { actor: "EnrichActorJob", repository: "EnrichRepositoryJob" }.freeze + + def self.call(reason:, **options) + new(**options).call(reason: reason) + end + + def initialize(configuration: Github.configuration, clock: -> { Time.current }, selector: nil) + @configuration = configuration + @clock = clock + @selector = selector || CandidateSelector.new(configuration: configuration) + end + + # @param reason [String] what asked — "ingestion" or "reconcile". It is on every line + # because the two have different meanings when they disagree: an ingestion dispatch + # that enqueues nothing means the events created no new work, while a reconcile one + # that enqueues means something was committed and never scheduled. + # @return [Hash] the payload it logged, so a caller can assert on it. + def call(reason:) + now = @clock.call + schedule = class_schedule(now: now) + blocked = !schedule.due?(now: now) + + payload = EntityType.all.each_with_object({}) do |entity_type, counts| + enqueue = !blocked && @selector.claimable?(entity_type, now: now) + JOBS.fetch(entity_type.key).constantize.perform_later if enqueue + + counts[:"#{entity_type.key}_enqueued"] = enqueue ? 1 : 0 + end + + log(payload.merge(reason: reason, blocked_by: (schedule.binding_component if blocked)).compact, now: now) + end + + private + + # §9's effective_enrichment_time with the entity component omitted, because this object + # is not choosing an entity — Github::Enrichment::Claim does that, under a lease, after + # Github::Enrichment::Fairness has chosen a class. What it can answer cheaply is + # whether *any* enrichment is legal right now, and both of the remaining components are + # single reads of one row. + # + # The per-class share is deliberately absent, for the reason + # Github::EnrichmentSchedule's own comment gives: a share exhaustion is a denial + # relieved by borrowing, not a deferral, so refusing to enqueue on it would withhold + # work the ledger would have granted. + # + # find_by, never bootstrap!: a read path must not create the ledger row. A clean + # checkout has no row, every component is nil, and the schedule is due — which is + # right, because the first poll is what initializes the window. + def class_schedule(now:) + budget = GithubApiBudget.find_by(id: GithubApiBudget::SINGLETON_ID) + + EnrichmentSchedule.new( + next_retry_at: nil, + global_blocked_until: budget&.global_blocked_until, + enrichment_class_blocked_until: budget&.enrichment_class_blocked_until(now: now) + ) + end + + # §11 lists "reconciliation summaries" among the INFO events, and this is that line — + # but only when it scheduled something. A tick that enqueued nothing is the ordinary + # steady state of an exhausted window, and at 60-second cadence it would emit a line a + # minute for the rest of the hour: the volume argument + # Github::BudgetLedger#log_class_exhausted and Github::EnrichmentRunner#log already make. + # + # The summary is PR 7's, unchanged: per-status counts per class, per-class share usage, + # the window state, and when enrichment is next due. + def log(payload, now:) + enqueued = payload.fetch(:actor_enqueued) + payload.fetch(:repository_enqueued) + entry = { event: "enrichment.dispatched", **payload, + **Summary.capture(now: now, configuration: @configuration, selector: @selector).to_log } + + enqueued.positive? ? Rails.logger.info(entry) : Rails.logger.debug(entry) + payload + end + end + end +end diff --git a/app/services/github/enrichment_runner.rb b/app/services/github/enrichment_runner.rb index 018c067..55d964d 100644 --- a/app/services/github/enrichment_runner.rb +++ b/app/services/github/enrichment_runner.rb @@ -15,9 +15,10 @@ module Github # budget is exhausted — precisely when boundedness matters. Behind the fairness decision # it would stop exactly then, and the backlog would grow without limit. # - # **At most one entity per call.** §5 names EnrichActorJob and EnrichRepositoryJob, so - # one entity is the unit PR 8 wraps; batching is the caller's loop. - # Github::Enrichment::OneShot is that loop today. + # **At most one entity per call.** §5 names EnrichActorJob and EnrichRepositoryJob, and one + # entity is what each of them performs; batching is the caller's loop, which + # Github::Enrichment::OneShot is for the operator and the 60-second reconciler tick is for + # the worker. # # **No source lock, ever.** §8 step 1: "Enrichment jobs skip this step — they take only # the request gate." This class is never handed an EventSource and never reaches for diff --git a/app/services/github/ingestion_runner.rb b/app/services/github/ingestion_runner.rb index 01c44d5..ae89dca 100644 --- a/app/services/github/ingestion_runner.rb +++ b/app/services/github/ingestion_runner.rb @@ -36,10 +36,20 @@ module Github # refuses a reservation inside an open application transaction, because an outer # rollback would refund a request GitHub has already counted. # - # It still enqueues nothing. §8 step 10 is PR 8, and §8 already says why no list of - # created ids is needed: "the committed entity state is the durable record of pending - # work (outbox-style recovery)" — and the enrichment_candidates partial index for - # exactly that predicate already exists. + # §8 step 10 — "enqueue enrichment after commit" — is #call's last act, once the lock is + # released, and it needs no list of created ids because §8 says why: "the committed entity + # state is the durable record of pending work (outbox-style recovery)", and the + # enrichment_candidates partial index for exactly that predicate already exists. + # Github::Enrichment::Dispatch reads that state; this class only tells it that a run + # created something worth looking at. + # + # One dispatch per run rather than per created event, and after the run row is finalized + # rather than inside PageWriter's per-envelope transaction. Both follow from enrichment + # jobs being class-scoped: the runner chooses the entity by fairness, so N enqueues carry + # no more information than one, and §8's property that matters — the enqueue happens after + # the rows are durable — is stronger here, where every commit of the run is behind us. + # Losing the dispatch to a crash loses nothing: ReconcilePendingEnrichmentsJob sweeps the + # same committed state every 60 seconds. class IngestionRunner MONOTONIC = -> { Process.clock_gettime(Process::CLOCK_MONOTONIC) } @@ -70,10 +80,11 @@ def to_log def initialize(executor: Github.executor, writer: Ingestion::PageWriter.new, configuration: Github.configuration, clock: -> { Time.current }, monotonic: MONOTONIC, rate_limit_policy: RateLimitPolicy.new, - page_loop: nil, poll_state: nil) + page_loop: nil, poll_state: nil, dispatch: nil) @configuration = configuration @clock = clock @monotonic = monotonic + @dispatch = dispatch || Enrichment::Dispatch.new(configuration: configuration, clock: clock) @page_loop = page_loop || Ingestion::PageLoop.new( executor: executor, writer: writer, configuration: configuration, rate_limit_policy: rate_limit_policy, clock: clock @@ -96,13 +107,33 @@ def initialize(executor: Github.executor, writer: Ingestion::PageWriter.new, def call(event_source:, wait_seconds: SourceLock::POLLER_WAIT_SECONDS, force: false) requested_at = @monotonic.call - SourceLock.acquire(event_source.id, wait_seconds: wait_seconds) do + result = SourceLock.acquire(event_source.id, wait_seconds: wait_seconds) do run(event_source.reload, force: force, lock_wait_ms: elapsed_ms(requested_at)) end + + dispatch_enrichment(result) + result end private + # §8 step 10, outside the source lock on purpose: every row of this run is committed, the + # run row is finalized, and there is no reason to hold a source's mutual exclusion across + # a write to a different database. + # + # Only when the run created events. A page of duplicates, a 304, a deferral and a not-due + # tick changed no entity's state, and everything that was pending before this run is + # already ReconcilePendingEnrichmentsJob's business. + # + # A crash before this line — or a queue database that refuses the insert — loses the + # dispatch and no work: that is what §2A's outbox-style recovery means, and the reconciler + # closes the gap within its 60-second cadence. + def dispatch_enrichment(result) + return unless result.tally.events_created.positive? + + @dispatch.call(reason: "ingestion") + end + def run(event_source, force:, lock_wait_ms:) return out_of_service(event_source) if event_source.failed? diff --git a/app/services/github/source_lock.rb b/app/services/github/source_lock.rb index b159668..6945559 100644 --- a/app/services/github/source_lock.rb +++ b/app/services/github/source_lock.rb @@ -6,8 +6,9 @@ module Github # A `FOR UPDATE` row claim cannot own an HTTP operation, because a row lock ends at # transaction end and a transaction must not span network I/O. A session advisory # lock gives operation-wide ownership that PostgreSQL releases automatically when - # the session dies — which is what makes a hard container kill safe. Verifying that - # release-on-death behaviour is PR 8's half of B7; PR 4 delivers acquisition. + # the session dies — which is what makes a hard container kill safe. That + # release-on-death behaviour is verified rather than assumed, with a terminated backend + # rather than a cooperative close: spec/recovery/advisory_lock_session_death_spec.rb. # # PR 5's IngestionRunner is the first caller; the one-shot ingestion command is the # second. diff --git a/bin/enrich b/bin/enrich index 6fc6ac4..530174e 100755 --- a/bin/enrich +++ b/bin/enrich @@ -10,8 +10,9 @@ # # Separate from bin/ingest rather than a flag on it, because §5 gives enrichment its own # path: it belongs to no event source, never takes a source lock, and spends a different -# class of the budget. PR 8 wraps the same Github::EnrichmentRunner in EnrichActorJob and -# EnrichRepositoryJob; this is the operator's handle on it until then, and after. +# class of the budget. EnrichActorJob and EnrichRepositoryJob wrap the same +# Github::EnrichmentRunner one cycle at a time; this stays the operator's handle on it, +# for a burst of cycles on demand rather than the worker's budgeted trickle. # # config/environment rather than config/boot: this command uses models, so Rails has to be # initialized rather than merely loadable. diff --git a/bin/jobs b/bin/jobs new file mode 100755 index 0000000..2c7c42f --- /dev/null +++ b/bin/jobs @@ -0,0 +1,17 @@ +#!/usr/bin/env ruby +# The Solid Queue supervisor (IMPLEMENTATION_PLAN.md §2A), the entry point behind the +# `worker` compose service: a dispatcher, a scheduler running config/recurring.yml, and the +# worker threads defined in config/queue.yml. +# +# Generated by `bin/rails generate solid_queue:install` and left as generated. It loads +# config/environment, so config/initializers/github.rb's budget validation runs here too — a +# configuration whose polling requirement leaves no enrichment capacity stops the worker at +# boot rather than letting it poll into an over-commitment (§10). +# +# `bin/jobs check` validates both YAML files without starting a process; CI runs that, and +# then runs this supervisor for real in fixture mode. + +require_relative "../config/environment" +require "solid_queue/cli" + +SolidQueue::Cli.start(ARGV) diff --git a/config/application.rb b/config/application.rb index 6612df0..10acc84 100644 --- a/config/application.rb +++ b/config/application.rb @@ -40,5 +40,45 @@ class Application < Rails::Application logger.formatter = JsonLogFormatter.new config.logger = logger config.log_level = ENV.fetch("LOG_LEVEL", "info") + + # Solid Queue is the job backend (§2A), and it lives in its own `queue` database inside + # the same PostgreSQL container — config/database.yml has declared that database since + # PR 2, and db/queue_migrate carries its schema. + # + # Configured here rather than in config/environments/production.rb, where the installer + # put it, for the same reason logging is configured here: the `worker` container runs + # RAILS_ENV=development by default, so development is the environment reviewers actually + # exercise. A job backend that differed by environment would make PR 8's recovery + # behaviour untestable exactly where it runs. config/environments/test.rb overrides the + # adapter with :test — §2A's "ordinary specs use Active Job's test adapter; only + # dedicated queue integration tests touch the queue test database". + config.active_job.queue_adapter = :solid_queue + config.solid_queue.connects_to = { database: { writing: :queue } } + + # Its polling SELECTs would otherwise be the loudest thing in the stream: two processes + # asking for work every second against ~12 ingestion runs an hour (§11 requires the + # events reviewers trace not to be buried). + config.solid_queue.silence_polling = true + + # Active Job's own subscriber, at warn. Its "Performing X (Job ID: ...)" pair is an + # unstructured `message` string in a JSON stream, it says nothing ApplicationJob's + # job.started/job.completed lines do not say with real fields, and at ~120 jobs an hour + # it would be two thirds of the INFO stream. Warn rather than silence: the framework's + # own error and retry lines still land, in the same format, on the same stream. + active_job_logger = ActiveSupport::Logger.new($stdout) + active_job_logger.formatter = JsonLogFormatter.new + active_job_logger.level = :warn + config.active_job.logger = active_job_logger + + # Derived from §2A's pinned defaults, the way Github::RequestGate::WAIT_SECONDS is: + # HTTP_OPEN_TIMEOUT_SECONDS (5) + HTTP_READ_TIMEOUT_SECONDS (15) is the longest an + # attempt that already holds the request gate can still be running, so a TERM waits that + # long for in-flight work and no longer. docker-compose.yml pairs it with + # stop_grace_period: 30s, leaving margin for the supervisor to reap its children. + # + # A job still *waiting* for the gate has reserved nothing and is safe to kill: §8's + # at-least-once execution with idempotent writes is what makes that true, and both + # advisory locks die with the session. + config.solid_queue.shutdown_timeout = 20.seconds end end diff --git a/config/ci.rb b/config/ci.rb index 53b49f8..c855158 100644 --- a/config/ci.rb +++ b/config/ci.rb @@ -19,6 +19,17 @@ # enrich — across two real processes, offline and deterministically. step "Tests: One-shot enrichment smoke", "env RAILS_ENV=test GITHUB_MODE=fixture bin/enrich" + # Solid Queue's own validator over config/queue.yml and config/recurring.yml. Starts no + # process; catches an unparseable schedule or a task naming a class that does not exist. + step "Tests: Queue configuration", "env RAILS_ENV=test bin/rails solid_queue:check" + + # The supervisor's own boot path, for the same reason the two smokes above exist — nothing + # else runs bin/jobs. It proves boot, recurring-task registration and a clean TERM, not that + # a tick fired: the schedule is 60 seconds and waiting for one would treble the step. Fixture + # mode is mandatory — this boots a real scheduler in a process WebMock cannot see. + step "Tests: Worker supervisor smoke", + "env RAILS_ENV=test GITHUB_MODE=fixture timeout --preserve-status --signal=TERM 20 bin/jobs" + step "Style: Ruby", "bin/rubocop" step "Security: Gem audit", "bin/bundler-audit" diff --git a/config/database.yml b/config/database.yml index b9c77a2..2ff8181 100644 --- a/config/database.yml +++ b/config/database.yml @@ -1,8 +1,8 @@ # Two databases in the same PostgreSQL 16 instance (IMPLEMENTATION_PLAN.md §2A): # primary — the business tables, the system of record -# queue — Solid Queue's database (jobs run there from PR 8 onward; declared -# here already so the compose `setup` service prepares both and -# `web`/`worker` never race concurrent db:prepare runs) +# queue — Solid Queue's database (db/queue_migrate carries its schema; the compose +# `setup` service prepares both, so `web` and `worker` never race +# concurrent db:prepare runs) # # Test databases are fully isolated from development (plan §2A, §12): # github_push_ingestor_test / github_push_ingestor_queue_test diff --git a/config/environments/development.rb b/config/environments/development.rb index 389a31b..c0b4f8c 100644 --- a/config/environments/development.rb +++ b/config/environments/development.rb @@ -41,8 +41,10 @@ # Append comments with runtime information tags to SQL queries in logs. config.active_record.query_log_tags_enabled = true - # Highlight code that enqueued background job in logs. - config.active_job.verbose_enqueue_logs = true + # Off, now that jobs run continuously: the enqueue backtrace line is an unstructured + # string in a JSON stream, and every enqueue this application makes comes from one place + # (Github::Enrichment::Dispatch) that logs it structurally with the job's own identifiers. + config.active_job.verbose_enqueue_logs = false # Highlight code that triggered redirect in logs. config.action_dispatch.verbose_redirect_logs = true diff --git a/config/environments/production.rb b/config/environments/production.rb index 905ed3c..492c8a0 100644 --- a/config/environments/production.rb +++ b/config/environments/production.rb @@ -39,8 +39,10 @@ # Replace the default in-process memory cache store with a durable alternative. # config.cache_store = :mem_cache_store - # Replace the default in-process and non-durable queuing backend for Active Job. - # config.active_job.queue_adapter = :resque + # Active Job's backend and its queue-database routing are configured once in + # config/application.rb, identically in every environment — the `worker` container runs + # RAILS_ENV=development by default, so a production-only backend would leave the + # environment reviewers exercise on the inline adapter (plan §2A). # Enable locale fallbacks for I18n (makes lookups for any locale fall back to # the I18n.default_locale when a translation cannot be found). diff --git a/config/environments/test.rb b/config/environments/test.rb index 14bc29e..6a3e6b5 100644 --- a/config/environments/test.rb +++ b/config/environments/test.rb @@ -31,6 +31,15 @@ # Print deprecation notices to the stderr. config.active_support.deprecation = :stderr + # §2A: "Ordinary specs use Active Job's test adapter; only dedicated queue integration + # tests touch the queue test database." The suite would otherwise write solid_queue_jobs + # rows from every ingestion spec — rows no example transaction on the *primary* + # connection would necessarily clean up, and enqueue volume nobody asserted. + # + # spec/support/queue_helpers.rb swaps this per example for the handful tagged :queue, and + # spec/job_boundary_spec.rb fails loudly if this override ever stops applying. + config.active_job.queue_adapter = :test + # Raises error for missing translations. # config.i18n.raise_on_missing_translations = true diff --git a/config/queue.yml b/config/queue.yml new file mode 100644 index 0000000..ee30fdd --- /dev/null +++ b/config/queue.yml @@ -0,0 +1,31 @@ +# Solid Queue's process topology (IMPLEMENTATION_PLAN.md §2A). One supervisor per `worker` +# container: a dispatcher, a scheduler for config/recurring.yml, and one worker process. +# +# Two threads, not the generator's three, and no JOB_CONCURRENCY knob. §5's global request +# gate makes outbound concurrency exactly one application-wide, so a third thread could only +# queue behind the gate — and while it waited it would hold a primary-database connection +# for up to Github::RequestGate::WAIT_SECONDS (45) out of the RAILS_MAX_THREADS (5) that +# config/database.yml grants per database. Two keeps a reconciler tick from sitting behind a +# poll that is mid-fetch, which is the only concurrency this system actually needs. +# +# polling_interval stays at 1s. Nothing here is latency-sensitive: the poll cadence is +# POLL_INTERVAL_SECONDS (300), enrichment is capped by the hourly allowance (40 at the +# defaults), and both recurring tasks fire on a 60-second schedule. +default: &default + dispatchers: + - polling_interval: 1 + batch_size: 500 + workers: + - queues: "*" + threads: 2 + processes: 1 + polling_interval: 1 + +development: + <<: *default + +test: + <<: *default + +production: + <<: *default diff --git a/config/recurring.yml b/config/recurring.yml new file mode 100644 index 0000000..bf8374d --- /dev/null +++ b/config/recurring.yml @@ -0,0 +1,43 @@ +# The recurring tasks that make this system run by itself (IMPLEMENTATION_PLAN.md §2A, §8). +# +# poll_event_sources — §2A's "Solid Queue recurring task fires every 60s → +# PollEventSourceJob computes effective_poll_time and no-ops unless a poll is due". A tick +# is not a poll: against the default POLL_INTERVAL_SECONDS (300) roughly four ticks in five +# find nothing due and cost one indexed SELECT. That ratio is the point — it is what makes a +# source that becomes due at T get polled within a minute of T, without a cadence that the +# allowance formula (12 poll requests an hour, zero headroom) cannot pay for. +# +# reconcile_pending_enrichments — §8 step 11, the sweep behind the outbox-style recovery. +# The post-commit enqueue in Github::IngestionRunner is only a hint; a SIGKILL between the +# COMMIT and the enqueue loses the hint and never the work, because the committed entity +# rows are the durable record. This task rediscovers that work, entity-scoped, and enqueues +# nothing when there is none or when the ledger says the class is spent. +# +# The cadences live here rather than in an environment variable because §9's cadence is +# POLL_INTERVAL_SECONDS and it is enforced in the scheduling components, not in the tick. +# Two tunables for one behaviour is the drift trap the plan avoids everywhere else. +# +# Solid Queue enforces uniqueness per occurrence with a unique index on +# (task_key, run_at) in solid_queue_recurring_executions, so a second worker container +# cannot double-enqueue a tick. That guarantee needs preserve_finished_jobs (its default, +# true), which is why the installer's hourly cleanup below is kept rather than replaced by +# turning retention off. +default: &default + poll_event_sources: + class: PollEventSourceJob + schedule: every 60 seconds + reconcile_pending_enrichments: + class: ReconcilePendingEnrichmentsJob + schedule: every 60 seconds + clear_solid_queue_finished_jobs: + command: "SolidQueue::Job.clear_finished_in_batches(sleep_between_batches: 0.3)" + schedule: every hour at minute 12 + +development: + <<: *default + +test: + <<: *default + +production: + <<: *default diff --git a/db/migrate/20260731100000_add_poll_due_index_to_event_sources.rb b/db/migrate/20260731100000_add_poll_due_index_to_event_sources.rb new file mode 100644 index 0000000..7903c3a --- /dev/null +++ b/db/migrate/20260731100000_add_poll_due_index_to_event_sources.rb @@ -0,0 +1,25 @@ +# PR 8's recurring tick asks a question nothing asked before: "which sources might be due +# right now?" (EventSource.poll_due). Until this PR, every path reached exactly one row +# through Github::Ingestion::SourceProvisioner.ensure!, so event_sources carried no index +# at all — and a PR 3 spec asserted that absence deliberately, so that adding the query and +# adding its index would land in the same reviewable change. +# +# Column order is (source_type, next_poll_at): source_type is the equality predicate and +# next_poll_at carries the range. The partial predicate is the scope's remaining WHERE, so +# the index covers the whole query and the two cannot drift without the schema spec +# noticing. +# +# The table is tiny — one row per mode today — so this index buys no measurable time now. +# It is here because the query is written to survive §9's "multiple event sources", and an +# index added with its query is checkable, while one added later is archaeology. +# +# Same deliberate omission as AddEnrichmentRefreshIndexes: not algorithm: :concurrently, +# because disable_ddl_transaction! complicates the compose `setup` service's db:prepare and +# the lock here is sub-second at any plausible size. +class AddPollDueIndexToEventSources < ActiveRecord::Migration[8.1] + def change + add_index :event_sources, %i[source_type next_poll_at], + where: "enabled AND status = 'idle'", + name: "index_event_sources_on_poll_due" + end +end diff --git a/db/queue_migrate/20260731100100_create_solid_queue_tables.rb b/db/queue_migrate/20260731100100_create_solid_queue_tables.rb new file mode 100644 index 0000000..062781e --- /dev/null +++ b/db/queue_migrate/20260731100100_create_solid_queue_tables.rb @@ -0,0 +1,165 @@ +# Solid Queue's tables, in the `queue` database (IMPLEMENTATION_PLAN.md §2A, §13's PR 8). +# +# The bodies below are `bin/rails generate solid_queue:install`'s output for solid_queue +# 1.5.1, transcribed rather than hand-written: `force: :cascade` dropped because a +# migration must never drop an existing table, and everything else — every index, every +# `on_delete: :cascade` foreign key — left exactly as the gem defines it. +# +# **Why a migration at all, when the gem ships a schema file.** `db:prepare` loads +# `db/queue_schema.rb` only for a database it finds *uninitialized*; anything already +# carrying `schema_migrations` is migrated instead. PR 2 committed a version-0 +# `db/queue_schema.rb` placeholder so the `setup` service could prepare both databases +# from the start, and `ActiveRecord::Schema.define` created `schema_migrations` when it +# ran — so every development queue database created since then already looks initialized +# (verified: `SELECT * FROM schema_migrations` returns the single row `0`). Shipping only +# a new schema file would silently skip the load there, and a reviewer who has run this +# project before would get a worker crash-looping against an empty queue database. +# +# With both, the two paths converge: a fresh database loads the regenerated dump, an +# existing one migrates, and `db:test:prepare` — which purges and loads the dump — is +# unaffected either way. +class CreateSolidQueueTables < ActiveRecord::Migration[8.1] + def change + create_table :solid_queue_jobs do |t| + t.string :queue_name, null: false + t.string :class_name, null: false + t.text :arguments + t.integer :priority, default: 0, null: false + t.string :active_job_id + t.datetime :scheduled_at + t.datetime :finished_at + t.string :concurrency_key + t.datetime :created_at, null: false + t.datetime :updated_at, null: false + + t.index [ :active_job_id ], name: "index_solid_queue_jobs_on_active_job_id" + t.index [ :class_name ], name: "index_solid_queue_jobs_on_class_name" + t.index [ :finished_at ], name: "index_solid_queue_jobs_on_finished_at" + t.index [ :queue_name, :finished_at ], name: "index_solid_queue_jobs_for_filtering" + t.index [ :scheduled_at, :finished_at ], name: "index_solid_queue_jobs_for_alerting" + end + + create_table :solid_queue_blocked_executions do |t| + t.bigint :job_id, null: false + t.string :queue_name, null: false + t.integer :priority, default: 0, null: false + t.string :concurrency_key, null: false + t.datetime :expires_at, null: false + t.datetime :created_at, null: false + + t.index [ :concurrency_key, :priority, :job_id ], name: "index_solid_queue_blocked_executions_for_release" + t.index [ :expires_at, :concurrency_key ], name: "index_solid_queue_blocked_executions_for_maintenance" + t.index [ :job_id ], name: "index_solid_queue_blocked_executions_on_job_id", unique: true + end + + create_table :solid_queue_claimed_executions do |t| + t.bigint :job_id, null: false + t.bigint :process_id + t.datetime :created_at, null: false + + t.index [ :job_id ], name: "index_solid_queue_claimed_executions_on_job_id", unique: true + t.index [ :process_id, :job_id ], name: "index_solid_queue_claimed_executions_on_process_id_and_job_id" + end + + create_table :solid_queue_failed_executions do |t| + t.bigint :job_id, null: false + t.text :error + t.datetime :created_at, null: false + + t.index [ :job_id ], name: "index_solid_queue_failed_executions_on_job_id", unique: true + end + + create_table :solid_queue_pauses do |t| + t.string :queue_name, null: false + t.datetime :created_at, null: false + + t.index [ :queue_name ], name: "index_solid_queue_pauses_on_queue_name", unique: true + end + + create_table :solid_queue_processes do |t| + t.string :kind, null: false + t.datetime :last_heartbeat_at, null: false + t.bigint :supervisor_id + t.integer :pid, null: false + t.string :hostname + t.text :metadata + t.datetime :created_at, null: false + t.string :name, null: false + + t.index [ :last_heartbeat_at ], name: "index_solid_queue_processes_on_last_heartbeat_at" + t.index [ :name, :supervisor_id ], name: "index_solid_queue_processes_on_name_and_supervisor_id", unique: true + t.index [ :supervisor_id ], name: "index_solid_queue_processes_on_supervisor_id" + end + + create_table :solid_queue_ready_executions do |t| + t.bigint :job_id, null: false + t.string :queue_name, null: false + t.integer :priority, default: 0, null: false + t.datetime :created_at, null: false + + t.index [ :job_id ], name: "index_solid_queue_ready_executions_on_job_id", unique: true + t.index [ :priority, :job_id ], name: "index_solid_queue_poll_all" + t.index [ :queue_name, :priority, :job_id ], name: "index_solid_queue_poll_by_queue" + end + + # The unique index on (task_key, run_at) is the whole of §2A's "a second worker + # container cannot double-enqueue a tick": two schedulers racing on the same + # occurrence collide here, in PostgreSQL, rather than in application logic. + create_table :solid_queue_recurring_executions do |t| + t.bigint :job_id, null: false + t.string :task_key, null: false + t.datetime :run_at, null: false + t.datetime :created_at, null: false + + t.index [ :job_id ], name: "index_solid_queue_recurring_executions_on_job_id", unique: true + t.index [ :task_key, :run_at ], name: "index_solid_queue_recurring_executions_on_task_key_and_run_at", unique: true + end + + create_table :solid_queue_recurring_tasks do |t| + t.string :key, null: false + t.string :schedule, null: false + t.string :command, limit: 2048 + t.string :class_name + t.text :arguments + t.string :queue_name + t.integer :priority, default: 0 + t.boolean :static, default: true, null: false + t.text :description + t.datetime :created_at, null: false + t.datetime :updated_at, null: false + + t.index [ :key ], name: "index_solid_queue_recurring_tasks_on_key", unique: true + t.index [ :static ], name: "index_solid_queue_recurring_tasks_on_static" + end + + create_table :solid_queue_scheduled_executions do |t| + t.bigint :job_id, null: false + t.string :queue_name, null: false + t.integer :priority, default: 0, null: false + t.datetime :scheduled_at, null: false + t.datetime :created_at, null: false + + t.index [ :job_id ], name: "index_solid_queue_scheduled_executions_on_job_id", unique: true + t.index [ :scheduled_at, :priority, :job_id ], name: "index_solid_queue_dispatch_all" + end + + create_table :solid_queue_semaphores do |t| + t.string :key, null: false + t.integer :value, default: 1, null: false + t.datetime :expires_at, null: false + t.datetime :created_at, null: false + t.datetime :updated_at, null: false + + t.index [ :expires_at ], name: "index_solid_queue_semaphores_on_expires_at" + t.index [ :key, :value ], name: "index_solid_queue_semaphores_on_key_and_value" + t.index [ :key ], name: "index_solid_queue_semaphores_on_key", unique: true + end + + add_foreign_key :solid_queue_blocked_executions, :solid_queue_jobs, column: :job_id, on_delete: :cascade + add_foreign_key :solid_queue_claimed_executions, :solid_queue_jobs, column: :job_id, on_delete: :cascade + add_foreign_key :solid_queue_failed_executions, :solid_queue_jobs, column: :job_id, on_delete: :cascade + add_foreign_key :solid_queue_ready_executions, :solid_queue_jobs, column: :job_id, on_delete: :cascade + add_foreign_key :solid_queue_recurring_executions, :solid_queue_jobs, column: :job_id, on_delete: :cascade + add_foreign_key :solid_queue_scheduled_executions, :solid_queue_jobs, column: :job_id, on_delete: :cascade + end +end diff --git a/db/queue_schema.rb b/db/queue_schema.rb index d79dfb5..c764866 100644 --- a/db/queue_schema.rb +++ b/db/queue_schema.rb @@ -1,8 +1,144 @@ -# Solid Queue's database schema lands with PR 8 (IMPLEMENTATION_PLAN.md §13). -# The queue database is declared and prepared from PR 2 onward so the compose -# `setup` service owns preparation of both databases (plan §2A). +# This file is auto-generated from the current state of the database. Instead +# of editing this file, please use the migrations feature of Active Record to +# incrementally modify your database, and then regenerate this schema definition. +# +# This file is the source Rails uses to define your schema when running `bin/rails +# db:schema:load`. When creating a new database, `bin/rails db:schema:load` tends to +# be faster and is potentially less error prone than running all of your +# migrations from scratch. Old migrations may fail to apply correctly if those +# migrations use external dependencies or application code. +# +# It's strongly recommended that you check this file into your version control system. -ActiveRecord::Schema[8.1].define(version: 0) do +ActiveRecord::Schema[8.1].define(version: 2026_07_31_100100) do # These are extensions that must be enabled in order to support this database enable_extension "pg_catalog.plpgsql" + + create_table "solid_queue_blocked_executions", force: :cascade do |t| + t.string "concurrency_key", null: false + t.datetime "created_at", null: false + t.datetime "expires_at", null: false + t.bigint "job_id", null: false + t.integer "priority", default: 0, null: false + t.string "queue_name", null: false + t.index ["concurrency_key", "priority", "job_id"], name: "index_solid_queue_blocked_executions_for_release" + t.index ["expires_at", "concurrency_key"], name: "index_solid_queue_blocked_executions_for_maintenance" + t.index ["job_id"], name: "index_solid_queue_blocked_executions_on_job_id", unique: true + end + + create_table "solid_queue_claimed_executions", force: :cascade do |t| + t.datetime "created_at", null: false + t.bigint "job_id", null: false + t.bigint "process_id" + t.index ["job_id"], name: "index_solid_queue_claimed_executions_on_job_id", unique: true + t.index ["process_id", "job_id"], name: "index_solid_queue_claimed_executions_on_process_id_and_job_id" + end + + create_table "solid_queue_failed_executions", force: :cascade do |t| + t.datetime "created_at", null: false + t.text "error" + t.bigint "job_id", null: false + t.index ["job_id"], name: "index_solid_queue_failed_executions_on_job_id", unique: true + end + + create_table "solid_queue_jobs", force: :cascade do |t| + t.string "active_job_id" + t.text "arguments" + t.string "class_name", null: false + t.string "concurrency_key" + t.datetime "created_at", null: false + t.datetime "finished_at" + t.integer "priority", default: 0, null: false + t.string "queue_name", null: false + t.datetime "scheduled_at" + t.datetime "updated_at", null: false + t.index ["active_job_id"], name: "index_solid_queue_jobs_on_active_job_id" + t.index ["class_name"], name: "index_solid_queue_jobs_on_class_name" + t.index ["finished_at"], name: "index_solid_queue_jobs_on_finished_at" + t.index ["queue_name", "finished_at"], name: "index_solid_queue_jobs_for_filtering" + t.index ["scheduled_at", "finished_at"], name: "index_solid_queue_jobs_for_alerting" + end + + create_table "solid_queue_pauses", force: :cascade do |t| + t.datetime "created_at", null: false + t.string "queue_name", null: false + t.index ["queue_name"], name: "index_solid_queue_pauses_on_queue_name", unique: true + end + + create_table "solid_queue_processes", force: :cascade do |t| + t.datetime "created_at", null: false + t.string "hostname" + t.string "kind", null: false + t.datetime "last_heartbeat_at", null: false + t.text "metadata" + t.string "name", null: false + t.integer "pid", null: false + t.bigint "supervisor_id" + t.index ["last_heartbeat_at"], name: "index_solid_queue_processes_on_last_heartbeat_at" + t.index ["name", "supervisor_id"], name: "index_solid_queue_processes_on_name_and_supervisor_id", unique: true + t.index ["supervisor_id"], name: "index_solid_queue_processes_on_supervisor_id" + end + + create_table "solid_queue_ready_executions", force: :cascade do |t| + t.datetime "created_at", null: false + t.bigint "job_id", null: false + t.integer "priority", default: 0, null: false + t.string "queue_name", null: false + t.index ["job_id"], name: "index_solid_queue_ready_executions_on_job_id", unique: true + t.index ["priority", "job_id"], name: "index_solid_queue_poll_all" + t.index ["queue_name", "priority", "job_id"], name: "index_solid_queue_poll_by_queue" + end + + create_table "solid_queue_recurring_executions", force: :cascade do |t| + t.datetime "created_at", null: false + t.bigint "job_id", null: false + t.datetime "run_at", null: false + t.string "task_key", null: false + t.index ["job_id"], name: "index_solid_queue_recurring_executions_on_job_id", unique: true + t.index ["task_key", "run_at"], name: "index_solid_queue_recurring_executions_on_task_key_and_run_at", unique: true + end + + create_table "solid_queue_recurring_tasks", force: :cascade do |t| + t.text "arguments" + t.string "class_name" + t.string "command", limit: 2048 + t.datetime "created_at", null: false + t.text "description" + t.string "key", null: false + t.integer "priority", default: 0 + t.string "queue_name" + t.string "schedule", null: false + t.boolean "static", default: true, null: false + t.datetime "updated_at", null: false + t.index ["key"], name: "index_solid_queue_recurring_tasks_on_key", unique: true + t.index ["static"], name: "index_solid_queue_recurring_tasks_on_static" + end + + create_table "solid_queue_scheduled_executions", force: :cascade do |t| + t.datetime "created_at", null: false + t.bigint "job_id", null: false + t.integer "priority", default: 0, null: false + t.string "queue_name", null: false + t.datetime "scheduled_at", null: false + t.index ["job_id"], name: "index_solid_queue_scheduled_executions_on_job_id", unique: true + t.index ["scheduled_at", "priority", "job_id"], name: "index_solid_queue_dispatch_all" + end + + create_table "solid_queue_semaphores", force: :cascade do |t| + t.datetime "created_at", null: false + t.datetime "expires_at", null: false + t.string "key", null: false + t.datetime "updated_at", null: false + t.integer "value", default: 1, null: false + t.index ["expires_at"], name: "index_solid_queue_semaphores_on_expires_at" + t.index ["key", "value"], name: "index_solid_queue_semaphores_on_key_and_value" + t.index ["key"], name: "index_solid_queue_semaphores_on_key", unique: true + end + + add_foreign_key "solid_queue_blocked_executions", "solid_queue_jobs", column: "job_id", on_delete: :cascade + add_foreign_key "solid_queue_claimed_executions", "solid_queue_jobs", column: "job_id", on_delete: :cascade + add_foreign_key "solid_queue_failed_executions", "solid_queue_jobs", column: "job_id", on_delete: :cascade + add_foreign_key "solid_queue_ready_executions", "solid_queue_jobs", column: "job_id", on_delete: :cascade + add_foreign_key "solid_queue_recurring_executions", "solid_queue_jobs", column: "job_id", on_delete: :cascade + add_foreign_key "solid_queue_scheduled_executions", "solid_queue_jobs", column: "job_id", on_delete: :cascade end diff --git a/db/schema.rb b/db/schema.rb index 650934e..15037c8 100644 --- a/db/schema.rb +++ b/db/schema.rb @@ -10,7 +10,7 @@ # # It's strongly recommended that you check this file into your version control system. -ActiveRecord::Schema[8.1].define(version: 2026_07_31_090000) do +ActiveRecord::Schema[8.1].define(version: 2026_07_31_100000) do # These are extensions that must be enabled in order to support this database enable_extension "pg_catalog.plpgsql" @@ -30,6 +30,7 @@ t.text "source_type", null: false t.text "status", null: false t.datetime "updated_at", null: false + t.index ["source_type", "next_poll_at"], name: "index_event_sources_on_poll_due", where: "(enabled AND (status = 'idle'::text))" t.check_constraint "consecutive_failures >= 0", name: "event_sources_consecutive_failures_nonnegative" t.check_constraint "status = ANY (ARRAY['idle'::text, 'failed'::text])", name: "event_sources_status_known" end diff --git a/docker-compose.yml b/docker-compose.yml index 67150f2..00101b7 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -1,7 +1,8 @@ # Compose topology per IMPLEMENTATION_PLAN.md §2A. # -# `docker compose up --build` starts db + setup + web (worker arrives with -# PR 8; continuous polling starts then). The `tools`-profiled services run +# `docker compose up --build` starts db + setup + web + worker, and continuous +# polling begins immediately: the worker's Solid Queue scheduler fires +# config/recurring.yml's 60-second tick. The `tools`-profiled services run # only when explicitly targeted: # # docker compose run --rm ingest # one ingestion cycle, then the state @@ -68,8 +69,8 @@ services: start_period: 10s # One-shot: prepares BOTH the primary and queue databases declared in - # config/database.yml, so web (and later worker) never race concurrent - # db:prepare runs. + # config/database.yml, so web and worker never race concurrent db:prepare + # runs. setup: build: . image: github-push-ingestor-app @@ -102,6 +103,37 @@ services: retries: 3 start_period: 20s + # The always-on background worker (plan §2A's topology table): one Solid Queue + # supervisor running the 60-second poll tick, the enrichment jobs, and the + # entity-scoped reconciler. Same image and the same shared environment as web, + # because one ledger row serves every process and two policies against it + # would be worse than either. + # + # stop_grace_period pairs with SolidQueue.shutdown_timeout (20s, set in + # config/application.rb from §2A's pinned HTTP timeouts): 20s for an in-flight + # attempt that already holds the request gate, and 10s of margin for the + # supervisor to reap its children. A job still *waiting* for the gate has + # reserved nothing and is safe to kill — §8's at-least-once execution with + # idempotent writes is what makes that true, and both advisory locks die with + # the session. + # + # No healthcheck. Nothing depends on this service, `unless-stopped` already + # covers process death, and the only probe that could tell a hung worker from + # a busy one would have to boot Rails on every interval; Solid Queue's own + # heartbeats in solid_queue_processes are the durable evidence instead. + worker: + build: . + image: github-push-ingestor-app + restart: unless-stopped + command: ["bin/jobs"] + stop_grace_period: 30s + environment: *app_env + depends_on: + db: + condition: service_healthy + setup: + condition: service_completed_successfully + # One-shot ingestion (plan §2A's topology table, §9's contract). Profiled so # a plain `up` never starts it, and `restart: "no"` because a one-shot that # restarts is a poller. diff --git a/docs/adr/0008-post-commit-enqueue-and-entity-scoped-reconciliation.md b/docs/adr/0008-post-commit-enqueue-and-entity-scoped-reconciliation.md new file mode 100644 index 0000000..7139fbf --- /dev/null +++ b/docs/adr/0008-post-commit-enqueue-and-entity-scoped-reconciliation.md @@ -0,0 +1,117 @@ +# 8. Enrichment is enqueued after commit and reconciled from the entity rows + +Date: 2026-07-31 + +Status: Accepted + +## Context + +§2A puts Solid Queue in its own `queue` database inside the same PostgreSQL container — +the separate-database configuration new Rails 8 applications generate. That choice is what +creates this decision: an enqueue cannot join the business transaction, because the two +writes are to different databases. Rails names the boundary +(`enqueue_after_transaction_commit`) but naming it does not answer what happens to work +whose enqueue never ran. + +The crash window is real and small: a `push_events` row commits, the process is killed, +and the enrichment the run was about to schedule is gone. §8 says what must be true +anyway — "the event and its stub entities remain durable in PostgreSQL. Pending enrichment +is rediscovered after restart by scanning entity rows — a small, entity-scoped set, not N +event rows per entity." + +A second fact shapes the design as much as the first: **enrichment jobs carry no entity +id.** `Github::EnrichmentRunner` enriches one entity per call and chooses it itself, +through §10's fairness policy and a `FOR UPDATE SKIP LOCKED` lease, newest-first. That is +not an accident to route around — an id-addressed job would have to bypass that ordering to +honour its argument, which is precisely how a repository flood starves actors (ADR 0007). +So an enqueue cannot mean "enrich this actor". It can only mean "there may be actor work". + +## Decision + +1. **The committed entity rows are the durable record of pending work; the enqueue is a + hint.** `github_actors.enrichment_status` and its partial index + `index_*_on_enrichment_candidates` already express the predicate exactly. Nothing is + written to represent pending work a second time. + +2. **One dispatch per run, after the lock is released.** `Github::IngestionRunner#call` + calls `Github::Enrichment::Dispatch` once, only when the run created events, after every + row of the run has committed, the run row is finalized, and the source advisory lock is + gone. + +3. **`ReconcilePendingEnrichmentsJob` runs every 60 seconds and is the recovery + mechanism.** It asks the same object the same question, from committed state alone. Work + whose enqueue was lost — to a kill, to a queue-database failure, to a discarded job — + is rediscovered on the next tick, within a minute, with no operator step and no cleanup + job. + +4. **At most one job per class per dispatch**, whatever the backlog depth. + +5. **No `retry_on` on any job.** The durable retry ladders already exist and are + coordinated with the ledger: `Github::Ingestion::PollState` for a source, + `Github::Enrichment::EntityState` for an entity. The 60-second tick is the retry. + +## Consequences + +Positive: + +- The crash window has no special case. The system's recovery path and its steady-state + path are the same code, which means the recovery path is exercised on every tick rather + than only during an incident. +- Queue depth is bounded by the hourly allowance rather than by arrival rate. One live page + references ~181 distinct entities; enqueuing per created event would produce ~2,400 + argument-identical cycles an hour against 40 spendable requests, and each surplus cycle + would run the age-out sweep and the fairness reads to be told no. +- The queue is never read to answer a question about business state, which keeps + CLAUDE.md's source-of-truth rule intact: PostgreSQL business tables are the durable + record, not the queue. + +Negative, and accepted: + +- Enrichment latency after a lost enqueue is up to 60 seconds. That is the cadence's + cost, and it is far inside the 30s–6h latency of the feed being sampled. +- A dispatch that finds work still only *schedules* one cycle per class, so a large backlog + drains at the reconciler's cadence rather than in a burst. `bin/enrich --limit N` remains + the operator's handle when a burst is wanted. +- The reconciler runs whether or not anything is pending. Its cost when idle is one indexed + `EXISTS` per class plus one ledger read, and it logs at debug so an exhausted window does + not emit a line a minute. + +## Alternatives rejected + +**Enqueue per created push event.** The natural reading of §8 step 10, and wrong here for a +reason specific to this design: the job carries no entity id, so N enqueues carry exactly +the information one does. The arithmetic above (~2,400 cycles/hour against 40 requests) +makes it a queue full of no-ops, and the dedupe it would then need — Solid Queue's +`limits_concurrency … on_conflict: :discard` — is a mechanism bolted on to undo a decision +rather than to make one. + +**Enqueue inside `Github::Ingestion::PageWriter`'s per-envelope transaction.** Closest to +§8's step ordering, and it breaks the property the step exists for: the enqueue would +precede the commit it is supposed to follow. + +**Solid Queue concurrency limits keyed by source id** (§9's third multi-poller bullet). +Rejected for PR 8 with a reason, not deferred silently: a `limits_concurrency` semaphore has +a fixed `duration`, so a container killed mid-poll would suppress that source until the +semaphore expired. The session advisory lock this system already holds is released by +PostgreSQL the instant the backend dies. Adopting a weaker, crash-unsafe duplicate of an +existing lock — in the PR whose subject is surviving container kills — would be a +regression. The source lock, the global request gate and the unique event constraint are +the protections in force; revisit when PR 11's multi-poller tests can measure a gap. + +**A dedicated outbox table.** A row per pending enrichment would be a second representation +of a fact the entity row already carries, with its own drift and its own cleanup. §2A calls +this design outbox-*style* precisely because there is no such record. + +**Per-source fan-out from the tick.** One job per due source, rather than one job that +iterates them. The global request gate makes outbound concurrency exactly one +application-wide, so the fanned-out jobs would serialize on the same advisory lock while +each held a database connection. Deferred to PR 9, which owns multi-source allocation. + +## Related + +- ADR 0002 — advisory locks and the request gate (the crash-safety property this decision + leans on) +- ADR 0004 — the class-aware budget ledger (what bounds enrichment throughput) +- ADR 0005 — at-least-once execution with idempotent writes (why a duplicate delivery is + safe) +- ADR 0007 — enrichment fairness shares and borrowing (why a job cannot carry an entity id) diff --git a/spec/db/schema_spec.rb b/spec/db/schema_spec.rb index 4da616c..23f2bbe 100644 --- a/spec/db/schema_spec.rb +++ b/spec/db/schema_spec.rb @@ -128,11 +128,25 @@ expect(names).to include("event_sources_status_known") end - # PR 8 adds the "which sources are due" query and should add the index that serves it - # in the same change, where its plan is checkable. Asserting the absence keeps that a - # visible decision rather than an oversight. - it "carries no index yet, because the query that would use one does not exist" do - expect(connection.indexes("event_sources")).to be_empty + # PR 8's recurring tick is the query that needed one, and it arrived in the same change. + # The predicate here and EventSource.poll_due's WHERE are the same sentence written + # twice, so this example is what keeps them from drifting apart. + it "indexes the recurring tick's due-source query" do + index = connection.indexes("event_sources").find { |i| i.name == "index_event_sources_on_poll_due" } + + expect(index).not_to be_nil, "expected index_event_sources_on_poll_due" + expect(index.columns).to eq(%w[source_type next_poll_at]) + expect(index.where).to include("enabled").and include("idle") + end + end + + # Solid Queue lives in its own database (§2A), and the outbox-style recovery argument + # depends on that being true rather than intended: if these tables were here, an enqueue + # could join the business transaction and "the committed entity state is the durable record + # of pending work" would stop being the reason the reconciler exists. + describe "the queue database boundary" do + it "keeps Solid Queue's tables out of the primary database" do + expect(connection.tables.grep(/solid_queue/)).to be_empty end end end diff --git a/spec/docker_compose_spec.rb b/spec/docker_compose_spec.rb index 712c2e2..4666ff2 100644 --- a/spec/docker_compose_spec.rb +++ b/spec/docker_compose_spec.rb @@ -12,9 +12,8 @@ def unprofiled services.reject { |_name, service| service.key?("profiles") }.keys end - # worker arrives with PR 8, when continuous polling starts. - it "starts exactly db, setup and web on a plain up" do - expect(unprofiled).to match_array(%w[db setup web]) + it "starts exactly db, setup, web and worker on a plain up" do + expect(unprofiled).to match_array(%w[db setup web worker]) end it "keeps the one-shots behind the tools profile" do @@ -98,6 +97,50 @@ def unprofiled end end + # §2A's topology table, and the service that makes this system run by itself: one Solid + # Queue supervisor per container, running the poll tick, the enrichment jobs and the + # reconciler. + describe "the worker service" do + let(:worker) { services.fetch("worker") } + + it "runs the Solid Queue supervisor" do + expect(worker.fetch("command")).to eq([ "bin/jobs" ]) + expect(worker).not_to have_key("entrypoint") + end + + # Docker's default policy is `no`, so §2A's crash recovery has to be declared. This is + # the service whose death would silently stop all ingestion. + it "restarts after a crash, like the other long-running services" do + expect(worker.fetch("restart")).to eq("unless-stopped") + end + + # 30s pairs with SolidQueue.shutdown_timeout (20s in config/application.rb, itself + # HTTP_OPEN_TIMEOUT_SECONDS + HTTP_READ_TIMEOUT_SECONDS), leaving margin for the + # supervisor to reap its children. + it "gives in-flight GitHub work time to finish on the way down" do + expect(worker.fetch("stop_grace_period")).to eq("30s") + end + + it "waits for the schema both databases need" do + expect(worker.dig("depends_on", "setup", "condition")).to eq("service_completed_successfully") + expect(worker.dig("depends_on", "db", "condition")).to eq("service_healthy") + end + + # One ledger row serves every process, so the worker has to read the same §10 policy the + # one-shots do. A worker on a different ACTOR_ENRICHMENT_SHARE would enforce a second + # policy against the same row. + it "inherits the same shared environment the one-shots do" do + expect(worker.fetch("environment")).to eq(services.fetch("ingest").fetch("environment")) + end + + # Nothing depends on this service, `unless-stopped` already covers process death, and the + # only probe that could tell a hung worker from a busy one would have to boot Rails on + # every interval; solid_queue_processes heartbeats are the durable evidence instead. + it "declares no healthcheck, unlike web" do + expect(worker).not_to have_key("healthcheck") + end + end + # §5's two request paths, and §13's PR 7. A separate service rather than a flag on # `ingest`, because enrichment belongs to no event source, takes no source lock, and # spends a different class of the budget. diff --git a/spec/job_boundary_spec.rb b/spec/job_boundary_spec.rb new file mode 100644 index 0000000..a210e2b --- /dev/null +++ b/spec/job_boundary_spec.rb @@ -0,0 +1,40 @@ +require "rails_helper" + +# The queue counterpart of spec/network_boundary_spec.rb. That file proves no spec can reach +# live GitHub; this one proves no ordinary spec can reach the queue database — §2A's other +# containment rule, and the one that silently stops holding if config/environments/test.rb's +# adapter override is ever dropped. +RSpec.describe "the job boundary" do + it "runs the suite on Active Job's test adapter" do + expect(ActiveJob::Base.queue_adapter).to be_a(ActiveJob::QueueAdapters::TestAdapter) + end + + it "keeps that adapter in the test environment's own configuration" do + expect(Rails.application.config.active_job.queue_adapter).to eq(:test) + end + + # §2A's enqueue semantics. A default that flipped would move every enqueue in this + # application inside the caller's transaction — the one boundary the outbox-style recovery + # argument depends on. + it "defers enqueues until after the enclosing transaction commits" do + expect(ApplicationJob.enqueue_after_transaction_commit).to be(true) + end + + # Solid Queue is what the worker container runs, so the routing has to be configured for + # every environment rather than only for production, where nothing in this project runs. + it "routes Solid Queue at the queue database in every environment" do + expect(Rails.application.config.solid_queue.connects_to).to eq(database: { writing: :queue }) + end + + # Grep-based, like the live-probe example in spec/network_boundary_spec.rb: the :queue tag + # is what swaps the adapter, so a file outside spec/queue/ carrying it would quietly start + # writing queue rows from the middle of the ordinary suite. + it "confines the :queue tag to spec/queue" do + tagged = Dir[Rails.root.join("spec/**/*_spec.rb")].select do |path| + File.read(path).match?(/^RSpec\.describe.*, :queue\b/) + end + + expect(tagged.map { |path| Pathname(path).relative_path_from(Rails.root).to_s }) + .to all(start_with("spec/queue/")) + end +end diff --git a/spec/jobs/application_job_spec.rb b/spec/jobs/application_job_spec.rb new file mode 100644 index 0000000..6a9b930 --- /dev/null +++ b/spec/jobs/application_job_spec.rb @@ -0,0 +1,113 @@ +require "rails_helper" + +# The job half of §11's common fields, and §2A's enqueue-after-commit contract. Both are +# properties every job in this application inherits, so they are asserted once, here, against +# a job defined for the purpose rather than against whichever real job happens to be handy. +RSpec.describe ApplicationJob do + # Named, not anonymous: Active Job serializes the class name, so an anonymous class cannot + # be enqueued and half of these examples would be untestable. + before do + stub_const("SpecProbeJob", Class.new(ApplicationJob) do + def self.name = "SpecProbeJob" + + cattr_accessor :behaviour, default: -> { } + + def perform + @outcome = { probe: "ran" } + self.class.behaviour.call + end + end) + end + + describe "the lines every job emits" do + before { allow(Rails.logger).to receive(:info).and_call_original } + + it "reports the job id, class, queue and attempt on completion" do + job = SpecProbeJob.new + job.perform_now + + expect(Rails.logger).to have_received(:info).with( + hash_including(event: "job.completed", job_id: job.job_id, job_class: "SpecProbeJob", + queue: "default", attempt: 1, probe: "ran") + ) + end + + it "measures how long the job took" do + allow(Rails.logger).to receive(:info) + + SpecProbeJob.new.perform_now + + expect(Rails.logger).to have_received(:info) + .with(hash_including(event: "job.completed", duration_ms: an_instance_of(Float))) + end + + # §8's at-least-once execution, made visible: a redelivery after a crash is the same job + # id at a higher attempt, which is how an operator tells one from a fresh tick. + it "counts a redelivery as a later attempt of the same job" do + allow(Rails.logger).to receive(:info) + job = SpecProbeJob.new + + job.perform_now + job.perform_now + + expect(Rails.logger).to have_received(:info) + .with(hash_including(event: "job.completed", job_id: job.job_id, attempt: 2)) + end + + # Logged *and* re-raised: Solid Queue has to see the failure to record it, and §11 wants + # the reason in the same stream as everything else rather than only in + # solid_queue_failed_executions. + it "reports a failure with its cause and lets it out" do + allow(Rails.logger).to receive(:error) + SpecProbeJob.behaviour = -> { raise Github::Errors::MalformedResponse, "not an array" } + + job = SpecProbeJob.new + expect { job.perform_now }.to raise_error(Github::Errors::MalformedResponse) + + expect(Rails.logger).to have_received(:error).with( + hash_including(event: "job.failed", job_id: job.job_id, + error_class: "Github::Errors::MalformedResponse", error_message: "not an array") + ) + end + end + + describe "retries" do + # Every job here can spend GitHub budget, and both retry ladders are already durable and + # coordinated with the ledger (Github::Ingestion::PollState and + # Github::Enrichment::EntityState). A second, uncoordinated Active Job ladder would + # re-poll a source whose backoff was just written. The 60-second recurring tick is the + # retry. + it "declares none, in any job in this application" do + jobs = [ ApplicationJob, PollEventSourceJob, EnrichActorJob, EnrichRepositoryJob, + ReconcilePendingEnrichmentsJob ] + + expect(jobs.map { |job| job.rescue_handlers.map(&:first) }.flatten).to be_empty + end + end + + describe "enqueue semantics (§2A)" do + it "defers an enqueue until the enclosing transaction commits" do + expect(described_class.enqueue_after_transaction_commit).to be(true) + end + + # The property that setting buys, rather than the setting alone. The example's fixture + # transaction is non-joinable, so this is a genuine inner transaction — the same + # distinction Github::IngestionRunner's "no application transaction across a fetch" + # example relies on. + it "enqueues nothing until the transaction has committed" do + enqueued_at_the_time = nil + + ActiveRecord::Base.transaction do + SpecProbeJob.perform_later + enqueued_at_the_time = ActiveJob::Base.queue_adapter.enqueued_jobs.size + end + + expect(enqueued_at_the_time).to eq(0) + expect(ActiveJob::Base.queue_adapter.enqueued_jobs.size).to eq(1) + end + + it "still enqueues when no transaction is open" do + expect { SpecProbeJob.perform_later }.to have_enqueued_job(SpecProbeJob) + end + end +end diff --git a/spec/jobs/enrich_actor_job_spec.rb b/spec/jobs/enrich_actor_job_spec.rb new file mode 100644 index 0000000..f05464c --- /dev/null +++ b/spec/jobs/enrich_actor_job_spec.rb @@ -0,0 +1,35 @@ +require "rails_helper" + +RSpec.describe EnrichActorJob do + it_behaves_like "an enrichment job", + entity_class: GithubActor, entity_type: :actor, log_key: :github_actor_id + + # The whole job over the real runner and the offline corpus, so "one cycle" is known to mean + # one entity and one request rather than only to be stubbed that way. + describe "with the real runner in fixture mode", type: :integration do + before do + allow(Github).to receive(:configuration).and_return(configuration_with("GITHUB_MODE" => "fixture")) + allow(Github::EnrichmentRunner).to receive(:new) + .and_return(fixture_enrichment_runner(transport: fixture_transport, now: frozen_time)) + + active_budget_window(now: frozen_time) + create_actor(github_id: 583_231, last_seen_at: frozen_time, + api_url: "https://api.github.com/users/octocat") + end + + it "enriches one actor and spends one request" do + described_class.new.perform_now + + expect(GithubActor.sole).to have_attributes(enrichment_status: "complete", name: "The Octocat") + expect(current_budget).to have_attributes(enrichment_used: 1, actor_share_used: 1) + end + + it "leaves repositories alone, whatever the fairness policy would have preferred" do + create_repository(github_id: 1_296_269, last_seen_at: frozen_time) + + described_class.new.perform_now + + expect(GithubRepository.sole.enrichment_status).to eq("pending") + end + end +end diff --git a/spec/jobs/enrich_repository_job_spec.rb b/spec/jobs/enrich_repository_job_spec.rb new file mode 100644 index 0000000..9e42371 --- /dev/null +++ b/spec/jobs/enrich_repository_job_spec.rb @@ -0,0 +1,35 @@ +require "rails_helper" + +RSpec.describe EnrichRepositoryJob do + it_behaves_like "an enrichment job", + entity_class: GithubRepository, entity_type: :repository, log_key: :github_repository_id + + describe "with the real runner in fixture mode", type: :integration do + before do + allow(Github).to receive(:configuration).and_return(configuration_with("GITHUB_MODE" => "fixture")) + allow(Github::EnrichmentRunner).to receive(:new) + .and_return(fixture_enrichment_runner(transport: fixture_transport, now: frozen_time)) + + active_budget_window(now: frozen_time) + create_repository(github_id: 1_296_269, last_seen_at: frozen_time, + api_url: "https://api.github.com/repos/octocat/Hello-World") + end + + it "enriches one repository and spends one request" do + described_class.new.perform_now + + expect(GithubRepository.sole).to have_attributes( + enrichment_status: "complete", description: "My first repository on GitHub!", language: "Ruby" + ) + expect(current_budget).to have_attributes(enrichment_used: 1, repository_share_used: 1) + end + + it "leaves actors alone, whatever the fairness policy would have preferred" do + create_actor(github_id: 583_231, last_seen_at: frozen_time) + + described_class.new.perform_now + + expect(GithubActor.sole.enrichment_status).to eq("pending") + end + end +end diff --git a/spec/jobs/poll_event_source_job_spec.rb b/spec/jobs/poll_event_source_job_spec.rb new file mode 100644 index 0000000..b29a1d8 --- /dev/null +++ b/spec/jobs/poll_event_source_job_spec.rb @@ -0,0 +1,199 @@ +require "rails_helper" + +# §2A's recurring tick. The runner is a double in most of these because the contract under +# test is the *tick's*: which sources it asks about, what it does with each answer, and what +# it refuses to turn into a failed execution. Github::IngestionRunner's own behaviour has its +# own spec, and the last group here drives the real one over the fixture corpus so the two +# contracts are known to fit. +# +# Due-ness is expressed relative to real time rather than to frozen_time, because the job +# reads Time.current: it is the one object in this flow that cannot be handed a clock, since +# Solid Queue constructs it. +RSpec.describe PollEventSourceJob do + let(:runner) { instance_double(Github::IngestionRunner) } + let(:result) { Github::IngestionRunner::Result.new(run_id: "run-1", status: "completed") } + + before { allow(Github::IngestionRunner).to receive(:new).and_return(runner) } + + def poll!(job = described_class.new) + job.perform_now + job + end + + # Nothing seeds event_sources, so without this a tick would run forever against an empty + # table (Github::Ingestion::SourceProvisioner's comment explains why lazily, at the point of + # use, is the only correct answer here). + describe "on a clean database" do + it "provisions the source it is about to poll" do + allow(runner).to receive(:call).and_return(result) + + expect { poll! }.to change(EventSource, :count).by(1) + expect(EventSource.sole.source_type).to eq("github_public_events") + end + end + + describe "choosing what to poll" do + it "polls a due source once, attempting the lock only once (§2A's poller contract)" do + source = create_event_source(next_poll_at: nil) + expect(runner).to receive(:call).with(event_source: source).once.and_return(result) + + poll! + end + + it "asks the runner about nothing when no source is due" do + create_event_source(next_poll_at: 1.hour.from_now) + expect(runner).not_to receive(:call) + + poll! + end + + # Filtered rather than left to the runner, which logs ingestion.source_unavailable at + # warn — once a minute, forever, for a source only an operator can restore. + it "asks the runner about nothing when the only source is out of service" do + create_event_source(status: "failed", next_poll_at: nil) + expect(runner).not_to receive(:call) + + poll! + end + + # A development database routinely holds both rows: the README's reviewer path creates a + # fixture source with `GITHUB_MODE=fixture docker compose run --rm ingest`. + it "never polls a source belonging to another mode" do + create_event_source(source_type: "github_fixture_events", next_poll_at: nil) + allow(runner).to receive(:call).and_return(result) + + poll! + + expect(runner).to have_received(:call) + .with(event_source: having_attributes(source_type: "github_public_events")).once + end + end + + describe "what it reports" do + before { create_event_source(next_poll_at: nil) } + + it "puts the job id and every run it opened on one line" do + allow(runner).to receive(:call).and_return(result) + allow(Rails.logger).to receive(:info) + + job = poll! + + expect(Rails.logger).to have_received(:info).with( + hash_including(event: "job.completed", job_id: job.job_id, job_class: "PollEventSourceJob", + attempt: 1, sources_due: 1, sources_skipped: 0, run_ids: [ "run-1" ]) + ) + end + + # §7's rule is that a run row exists iff the process tried to reach GitHub. A source the + # runner reloaded and found not-due after all opened no run, so it contributes no run id — + # and it was not skipped either, which is a different fact. + it "reports no run id for a source the runner found not due" do + allow(runner).to receive(:call) + .and_return(Github::IngestionRunner::Result.new(run_id: nil, status: "deferred", + deferral_reason: "cadence_due_at")) + + expect(poll!.outcome).to include(sources_due: 1, sources_skipped: 0, run_ids: []) + end + end + + # §2A: "The poller attempts once and exits if unavailable." A raise here would put a row in + # solid_queue_failed_executions every minute a one-shot ran long — the system's own mutual + # exclusion working is not a defect. + describe "a source another process is already polling" do + before do + create_event_source(next_poll_at: nil) + allow(runner).to receive(:call).and_raise(Github::Errors::SourceBusy) + end + + it "reports the contention at INFO and completes the tick" do + allow(Rails.logger).to receive(:info) + + job = described_class.new + expect { job.perform_now }.not_to raise_error + + expect(Rails.logger).to have_received(:info) + .with(hash_including(event: "ingestion.source_busy", job_id: job.job_id)) + expect(job.outcome).to include(sources_skipped: 1, run_ids: []) + end + + # The guard against someone later adding retry_on and building a retry storm on top of a + # 60-second recurring task. + it "does not re-enqueue itself, because the next tick is 60 seconds away" do + expect { poll! }.not_to have_enqueued_job + end + end + + describe "a source that fails" do + let!(:failing) { create_event_source(next_poll_at: nil) } + + # By the time an error escapes the runner, the run row is finalized and the source's + # backoff is written — so the remaining sources in this tick are still worth polling. + it "logs the cycle and keeps going" do + other = create_event_source(next_poll_at: nil) + allow(Rails.logger).to receive(:error) + allow(runner).to receive(:call).with(event_source: failing) + .and_raise(Github::Errors::ConnectionFailed, "boom") + allow(runner).to receive(:call).with(event_source: other).and_return(result) + + job = poll! + + expect(Rails.logger).to have_received(:error).with( + hash_including(event: "ingestion.cycle_failed", event_source_id: failing.id, + error_class: "Github::Errors::ConnectionFailed", job_id: job.job_id) + ) + expect(job.outcome).to include(sources_due: 2, sources_skipped: 1, run_ids: [ "run-1" ]) + end + + # A misconfiguration, a broken lock invariant or a dead connection is a fact about the + # process. Continuing to the next source would only repeat it, and a tick that "completed" + # after boot-level breakage would be a lie. + it "lets a process-level failure fail the job" do + allow(runner).to receive(:call).and_raise(Github::Errors::LockOrderViolation, "gate first") + + expect { poll! }.to raise_error(Github::Errors::LockOrderViolation) + end + + it "logs the failure with the job id before letting it out" do + allow(Rails.logger).to receive(:error) + allow(runner).to receive(:call).and_raise(Github::Errors::LockOrderViolation, "gate first") + + job = described_class.new + expect { job.perform_now }.to raise_error(Github::Errors::LockOrderViolation) + + expect(Rails.logger).to have_received(:error).with( + hash_including(event: "job.failed", job_id: job.job_id, + error_class: "Github::Errors::LockOrderViolation") + ) + end + end + + # The tick and the real runner, over the offline corpus, so the two contracts are known to + # fit: the scope hands over a source the runner accepts, and a real poll's work reaches the + # queue. + describe "with the real runner in fixture mode", type: :integration do + before do + allow(Github).to receive(:configuration).and_return(configuration_with("GITHUB_MODE" => "fixture")) + + # The outer stub has to come off first: #fixture_runner builds its runner through + # Github::IngestionRunner.new, so with the double still in place it would hand back the + # double and this group would assert nothing. + allow(Github::IngestionRunner).to receive(:new).and_call_original + real_runner = fixture_runner + allow(Github::IngestionRunner).to receive(:new).and_return(real_runner) + + active_budget_window(now: frozen_time) + end + + it "provisions, polls and persists the corpus page" do + poll! + + expect(EventSource.sole.source_type).to eq("github_fixture_events") + expect(PushEvent.count).to eq(4) + end + + it "hands the run's enrichment work to the queue" do + expect { poll! }.to have_enqueued_job(EnrichActorJob).exactly(:once) + .and have_enqueued_job(EnrichRepositoryJob).exactly(:once) + end + end +end diff --git a/spec/jobs/reconcile_pending_enrichments_job_spec.rb b/spec/jobs/reconcile_pending_enrichments_job_spec.rb new file mode 100644 index 0000000..3a2803b --- /dev/null +++ b/spec/jobs/reconcile_pending_enrichments_job_spec.rb @@ -0,0 +1,56 @@ +require "rails_helper" + +# §8 step 11's sweep. Its own behaviour is Github::Enrichment::Dispatch's, specified there; +# what this file pins is that the job is that sweep and nothing else — no source lock, no +# request, no reading of push_events, and a summary on its own line. +# +# The recovery property it exists for — work committed before a crash but never enqueued — +# is spec/recovery/pending_enrichment_recovery_spec.rb. +RSpec.describe ReconcilePendingEnrichmentsJob do + before { active_budget_window(now: frozen_time) } + + it "reconciles, and reports what it scheduled on the job's line" do + create_actor(github_id: 583_231, last_seen_at: Time.current) + allow(Rails.logger).to receive(:info) + + job = described_class.new + expect { job.perform_now }.to have_enqueued_job(EnrichActorJob).exactly(:once) + + expect(Rails.logger).to have_received(:info).with( + hash_including(event: "job.completed", job_id: job.job_id, reason: "reconcile", actor_enqueued: 1) + ) + end + + it "enqueues nothing when there is nothing durable to do" do + expect { described_class.new.perform_now }.not_to have_enqueued_job + end + + # §8: "a small, entity-scoped set, not N event rows per entity". Held structurally — the + # sweep never reads push_events at all — so this asserts the structure rather than a count. + it "never reads the event table it is recovering work for" do + create_actor(github_id: 583_231, last_seen_at: Time.current) + tables = [] + subscriber = ActiveSupport::Notifications.subscribe("sql.active_record") do |*, payload| + tables << payload[:sql] + end + + begin + described_class.new.perform_now + ensure + ActiveSupport::Notifications.unsubscribe(subscriber) + end + + expect(tables.grep(/push_events/)).to be_empty + end + + it "takes no lock and makes no request" do + create_actor(github_id: 583_231, last_seen_at: Time.current) + expect(Github::SourceLock).not_to receive(:acquire) + expect(Github::RequestGate).not_to receive(:hold) + + described_class.new.perform_now + + expect(WebMock).not_to have_requested(:any, //) + expect(Github::LockOrder.held_keys).to be_empty + end +end diff --git a/spec/models/event_source_spec.rb b/spec/models/event_source_spec.rb index b50c7a2..5aed7de 100644 --- a/spec/models/event_source_spec.rb +++ b/spec/models/event_source_spec.rb @@ -134,6 +134,71 @@ end end + # PR 8's recurring tick asks this once a minute. It is a pre-filter over the cached + # projection, never the decision — Github::IngestionRunner reloads the row inside the + # source lock and Github::PollSchedule decides there, from §9's four components. + describe ".poll_due" do + def due(now: frozen_time, source_type: "github_public_events") + described_class.poll_due(source_type: source_type, now: now) + end + + # A freshly provisioned source has never been polled, so its projection is NULL. If that + # did not count as due, a clean checkout would never poll at all. + it "includes a source that has never been polled" do + source = create_event_source(next_poll_at: nil) + + expect(due).to contain_exactly(source) + end + + it "includes a source whose projection has arrived, and excludes one whose has not" do + overdue = create_event_source(next_poll_at: frozen_time - 1) + create_event_source(next_poll_at: frozen_time + 1) + + expect(due).to contain_exactly(overdue) + end + + # <= rather than <, matching PollSchedule#due? and for its reason: PostgreSQL truncates a + # timestamp to microseconds on the way in and Time.current does not. + it "includes a source due at exactly this instant" do + source = create_event_source(next_poll_at: frozen_time) + + expect(due).to contain_exactly(source) + end + + # A live worker polling a fixture source raises Errors::FixtureMiss once a minute + # forever, and the reverse is refused by Github::UrlPolicy — and a development database + # routinely holds both rows, because the README's reviewer path creates one. + it "excludes a source belonging to another mode" do + create_event_source(source_type: "github_fixture_events", next_poll_at: nil) + + expect(due).to be_empty + end + + # An operator turned it off; the tick has nothing to say about that. + it "excludes a disabled source" do + create_event_source(enabled: false, next_poll_at: nil) + + expect(due).to be_empty + end + + # §10's "/events returns permanent 4xx → source failed": operator-recoverable only. The + # runner would refuse it anyway, with a warning — once a minute, until someone looked. + it "excludes a source that is out of service" do + create_event_source(status: "failed", next_poll_at: nil) + + expect(due).to be_empty + end + + # Ordered, so a tick with several due sources spends the budget in the same order every + # time rather than in whatever order PostgreSQL finds convenient. + it "returns due sources in a stable order" do + first = create_event_source(next_poll_at: nil) + second = create_event_source(next_poll_at: frozen_time - 60) + + expect(due.map(&:id)).to eq([ first.id, second.id ].sort) + end + end + describe "associations" do it "will not be destroyed while runs reference it" do source = create_event_source diff --git a/spec/queue/configuration_spec.rb b/spec/queue/configuration_spec.rb new file mode 100644 index 0000000..43a2ce3 --- /dev/null +++ b/spec/queue/configuration_spec.rb @@ -0,0 +1,103 @@ +require "rails_helper" + +# config/queue.yml and config/recurring.yml decide whether this system runs at all, and a +# mistyped key in either is silent: the supervisor starts, the tick never fires, and nothing +# in the suite would notice. The same argument spec/docker_compose_spec.rb makes about a +# profile key, applied to the two files PR 8 adds. +# +# No database and no adapter swap here — this is the YAML and the constants. The real +# validator runs in spec/queue/solid_queue_integration_spec.rb, where a queue connection is +# already open. +RSpec.describe "Solid Queue configuration" do + let(:queue_config) { YAML.safe_load(Rails.root.join("config/queue.yml").read, aliases: true) } + let(:recurring) { YAML.safe_load(Rails.root.join("config/recurring.yml").read, aliases: true) } + + ENVIRONMENTS = %w[development test production].freeze + + describe "config/recurring.yml" do + # Development is what `docker compose up` runs and what a reviewer watches; test is what + # CI's supervisor smoke boots. A task defined only in production would mean neither ever + # proved the tick fires. + it "defines the same tasks in every environment" do + task_sets = ENVIRONMENTS.map { |environment| recurring.fetch(environment).keys.sort } + + expect(task_sets.uniq.size).to eq(1) + expect(task_sets.first) + .to contain_exactly("clear_solid_queue_finished_jobs", "poll_event_sources", + "reconcile_pending_enrichments") + end + + it "schedules a job class this application actually defines" do + classes = recurring.fetch("production").values.filter_map { |task| task["class"] } + + expect(classes).to contain_exactly("PollEventSourceJob", "ReconcilePendingEnrichmentsJob") + expect(classes.map(&:constantize)).to all(be < ApplicationJob) + end + + # §2A: "Solid Queue recurring task fires every 60s". A 300-second tick would be the + # cadence twice over, and a source that became due at T would wait up to five minutes past + # it; a 1-second tick would spend the poll allowance on SELECTs of an unchanged table. + it "ticks every 60 seconds, for the poll and the reconciler alike" do + %w[poll_event_sources reconcile_pending_enrichments].each do |task| + expect(recurring.dig("production", task, "schedule")).to eq("every 60 seconds") + end + end + + # Solid Queue's recurring uniqueness guarantee — the unique index on (task_key, run_at) — + # holds "as long as you keep the jobs around", so preserve_finished_jobs stays at its + # default and the installer's hourly cleanup is what bounds the table instead. + it "keeps the installer's finished-job cleanup, which is what makes retention safe" do + expect(recurring.dig("production", "clear_solid_queue_finished_jobs", "command")) + .to include("clear_finished_in_batches") + end + end + + describe "config/queue.yml" do + it "configures a dispatcher and a worker in every environment" do + ENVIRONMENTS.each do |environment| + expect(queue_config.dig(environment, "workers")).to be_present + expect(queue_config.dig(environment, "dispatchers")).to be_present + end + end + + # A job enqueued into a queue no worker polls is a silent, total failure, and nothing else + # in the suite would catch it. Every job here uses the default queue, so one worker on "*" + # is the whole guarantee. + it "works every queue this application enqueues into" do + # Through an instance: Active Job's default queue name is a lambda until a job resolves + # it. + queues = [ PollEventSourceJob, EnrichActorJob, EnrichRepositoryJob, + ReconcilePendingEnrichmentsJob ].map { _1.new.queue_name }.uniq + + expect(queues).to eq([ "default" ]) + expect(queue_config.dig("production", "workers").map { _1["queues"] }).to all(eq("*")) + end + + # §5's request gate makes outbound concurrency exactly one application-wide, so extra + # threads could only queue behind it while holding a primary-database connection for up to + # Github::RequestGate::WAIT_SECONDS. config/database.yml grants RAILS_MAX_THREADS (5) per + # database, and the worker holds both. + it "keeps the thread pool inside the connection pool the worker is granted" do + threads = queue_config.dig("production", "workers").sum { _1.fetch("threads") } + + expect(threads).to be <= Integer(ENV.fetch("RAILS_MAX_THREADS", 5)) + expect(threads).to be >= 2 + end + end + + describe "bin/jobs" do + let(:path) { Rails.root.join("bin/jobs") } + + it "exists and is executable, because the worker service runs it directly" do + expect(path).to be_file + expect(path).to be_executable + end + + # config/environment rather than config/boot: config/initializers/github.rb validates the + # budget configuration in to_prepare, so a worker that would over-commit the hourly + # allowance stops at boot instead of polling into it. + it "boots the full application, so startup validation reaches the worker" do + expect(path.read).to include('require_relative "../config/environment"') + end + end +end diff --git a/spec/queue/solid_queue_integration_spec.rb b/spec/queue/solid_queue_integration_spec.rb new file mode 100644 index 0000000..6d2a30f --- /dev/null +++ b/spec/queue/solid_queue_integration_spec.rb @@ -0,0 +1,84 @@ +require "rails_helper" + +# The only file in the suite that writes to github_push_ingestor_queue_test (§2A: "Ordinary +# specs use Active Job's test adapter; only dedicated queue integration tests touch the queue +# test database"). Everything here is what no double can prove: that an enqueue reaches a +# different database, that connects_to routes it there, and that the schema this PR ships is +# the schema Solid Queue expects. +RSpec.describe "Solid Queue", :queue do + it "is the adapter these examples actually run against" do + expect(ActiveJob::Base.queue_adapter).to be_a(ActiveJob::QueueAdapters::SolidQueueAdapter) + end + + describe "enqueueing" do + it "writes a job row and a ready execution" do + expect { EnrichActorJob.perform_later }.to change(SolidQueue::Job, :count).by(1) + + job = SolidQueue::Job.last + expect(job.class_name).to eq("EnrichActorJob") + expect(job.queue_name).to eq("default") + expect(SolidQueue::ReadyExecution.where(job_id: job.id)).to exist + end + + it "round-trips a job's arguments" do + ReconcilePendingEnrichmentsJob.perform_later + + expect(SolidQueue::Job.last.arguments.fetch("arguments")).to eq([]) + end + + # Ordering is random, so this example and the two above are an empirical check that + # transactional fixtures really do roll back the *second* database's connection. If they + # did not, whichever of these ran last would find the others' rows. + it "leaves no rows behind for the next example" do + expect(SolidQueue::Job.count).to eq(0) + end + end + + describe "the queue database" do + it "routes Solid Queue's models to the queue connection, not the primary one" do + expect(SolidQueue::Job.connection_db_config.name).to eq("queue") + expect(SolidQueue::Job.connection_db_config.database).to eq("github_push_ingestor_queue_test") + end + + it "is a different database from the one the business tables live in" do + expect(SolidQueue::Job.connection_db_config.database) + .not_to eq(ActiveRecord::Base.connection_db_config.database) + end + + it "carries every table db/queue_migrate creates" do + expect(SolidQueue::Job.connection.tables).to include(*QueueHelpers::SOLID_QUEUE_TABLES) + end + + # The whole basis of §2A's "a second worker container cannot double-enqueue a tick", so it + # is asserted rather than trusted to the gem. + it "makes a recurring task's occurrence unique, which is what stops two schedulers racing" do + index = SolidQueue::Job.connection.indexes("solid_queue_recurring_executions") + .find { |i| i.columns == %w[task_key run_at] } + + expect(index).not_to be_nil + expect(index.unique).to be(true) + end + end + + # Solid Queue's own validator, over the real config/queue.yml and config/recurring.yml — + # the same object `bin/jobs check` runs, so a schedule Fugit cannot parse or a task naming a + # class that does not exist fails here rather than at 3am in a worker container. + describe "the shipped configuration" do + subject(:configuration) { SolidQueue::Configuration.new } + + it "is valid" do + expect(configuration).to be_valid, -> { configuration.errors.full_messages.join("; ") } + end + + it "configures the three processes the worker container runs" do + expect(configuration.configured_processes.map(&:kind)).to contain_exactly(:dispatcher, :worker, :scheduler) + end + + it "hands the scheduler this application's two ticks" do + scheduler = configuration.configured_processes.find { |process| process.kind == :scheduler } + + expect(scheduler.attributes.fetch(:recurring_tasks).map(&:class_name)) + .to include("PollEventSourceJob", "ReconcilePendingEnrichmentsJob") + end + end +end diff --git a/spec/recovery/advisory_lock_session_death_spec.rb b/spec/recovery/advisory_lock_session_death_spec.rb new file mode 100644 index 0000000..0a14278 --- /dev/null +++ b/spec/recovery/advisory_lock_session_death_spec.rb @@ -0,0 +1,102 @@ +require "rails_helper" + +# §12's "Advisory locks released on session death (simulated connection kill)" — the +# verification half of Extension B's crash-safe source ownership, and the property the whole +# lock design rests on: §2A chose session advisory locks over a FOR UPDATE row claim precisely +# because "hard process/container death closes the session and releases the lock +# automatically". +# +# The kill is pg_terminate_backend, not a client-side close. A close is the client +# cooperating — it proves locks are session-scoped and no application ensure block was +# involved, which the suite's own teardown demonstrates incidentally. A terminate ends the +# backend with no client cooperation and no Ruby running anywhere in the "dead process", which +# is what a `docker kill` does to a worker mid-poll. +# +# The residual gap, stated rather than papered over: the dying session cannot be the RSpec +# process's own pooled connection, because killing that backend would take the example's +# fixture transaction with it. §15's container kill closes that last gap, and it is a reviewer +# step (PR 11), not a unit test. +RSpec.describe "advisory locks after session death", type: :integration do + let(:source_namespace) { Github::AdvisoryLock::SOURCE_LOCK_NAMESPACE } + let(:gate_namespace) { Github::AdvisoryLock::REQUEST_GATE_NAMESPACE } + let(:event_source) { create_event_source } + let(:source_key) { Github::AdvisoryLock.key_for(event_source.id) } + + describe "a source lock held by a session that dies" do + before { acquire_in_other_session(source_namespace, source_key) } + + it "blocks the application while that session is alive" do + expect { Github::SourceLock.acquire(event_source.id) { :polled } } + .to raise_error(Github::Errors::SourceBusy) + end + + it "is released by PostgreSQL, with nothing in this application running to release it" do + pid = terminate_second_session! + wait_for_advisory_lock_release(source_namespace, source_key) + + expect(advisory_lock_holders(source_namespace, source_key)).to be_empty + expect(pid).to be_positive + end + + # Through the production wait rather than a hand-rolled sleep: pg_terminate_backend + # returns before the backend has finished exiting, so what is asserted is the contract a + # real poller relies on — SourceLock retries for its wait_seconds and gets the lock. + it "lets the next poller take the lock" do + terminate_second_session! + + expect(Github::SourceLock.acquire(event_source.id, wait_seconds: 5) { :polled }).to eq(:polled) + end + + it "leaves the lock free again once that poller is done" do + terminate_second_session! + Github::SourceLock.acquire(event_source.id, wait_seconds: 5) { :polled } + + expect(advisory_lock_holders(source_namespace, source_key)).to be_empty + end + end + + # The gate is the lock whose leak would stop every request in the system, polling and + # enrichment alike — so it gets the same proof rather than an argument by analogy. + describe "the request gate held by a session that dies" do + before { acquire_in_other_session(gate_namespace, Github::AdvisoryLock::REQUEST_GATE_KEY) } + + it "blocks the application while that session is alive" do + expect { Github::RequestGate.hold(wait_seconds: 0.1) { :requested } } + .to raise_error(Github::Errors::GateUnavailable) + end + + it "lets the next request through once the holder's session is gone" do + terminate_second_session! + wait_for_advisory_lock_release(gate_namespace, Github::AdvisoryLock::REQUEST_GATE_KEY) + + expect(Github::RequestGate.hold(wait_seconds: 5) { :requested }).to eq(:requested) + end + end + + # The headline: a whole polling operation, through production code on both sides of the + # kill. Before, the poller finds the source owned and leaves no trace; after, the same + # runner polls it and persists the page. + describe "a poll blocked by a dead worker's lock", type: :integration do + let(:transport) { fixture_transport } + let(:event_source) { fixture_event_source } + + before { active_budget_window(now: frozen_time) } + + it "recovers the source without an operator or a cleanup job" do + acquire_in_other_session(source_namespace, source_key) + + expect { fixture_runner(transport: transport).call(event_source: event_source) } + .to raise_error(Github::Errors::SourceBusy) + expect(IngestionRun.count).to eq(0) + expect(transport.requests).to be_empty + + terminate_second_session! + + result = fixture_runner(transport: transport) + .call(event_source: event_source, wait_seconds: 5) + + expect(result).to be_completed + expect(PushEvent.count).to eq(4) + end + end +end diff --git a/spec/recovery/duplicate_job_execution_spec.rb b/spec/recovery/duplicate_job_execution_spec.rb new file mode 100644 index 0000000..9695049 --- /dev/null +++ b/spec/recovery/duplicate_job_execution_spec.rb @@ -0,0 +1,88 @@ +require "rails_helper" + +# §12's "Enrichment job executed twice", which is §8's processing semantics stated as a test: +# at-least-once execution + idempotent writes + unique constraints = effectively-once +# persisted outcomes. Never exactly-once execution — the second execution really happens here, +# and what is asserted is that it changes nothing. +# +# The redelivery is modelled as the *same job instance* performed twice: one job id, two +# executions, which is what Solid Queue produces when a worker dies after the job ran and +# before its claim was released. +RSpec.describe "an enrichment job delivered twice", type: :integration do + let(:transport) { fixture_transport } + let(:job) { EnrichActorJob.new } + + before do + active_budget_window(now: frozen_time) + create_actor(github_id: 583_231, last_seen_at: frozen_time, + api_url: "https://api.github.com/users/octocat") + + allow(Github).to receive(:configuration).and_return(configuration_with("GITHUB_MODE" => "fixture")) + allow(Github::EnrichmentRunner).to receive(:new) + .and_return(fixture_enrichment_runner(transport: transport, now: frozen_time)) + + job.perform_now + end + + it "enriched the actor on the first delivery" do + expect(GithubActor.sole) + .to have_attributes(enrichment_status: "complete", name: "The Octocat", fetched_at: frozen_time) + expect(current_budget.enrichment_used).to eq(1) + end + + it "leaves the durable state byte-identical after the second" do + before_attributes = GithubActor.sole.attributes + + job.perform_now + + expect(GithubActor.sole.attributes).to eq(before_attributes) + end + + # The freshness cache is what makes this true — not a dedup table, and not the queue. A + # fresh record is not a candidate, so the second delivery has nothing to claim. + it "spends no second request and no second reservation" do + expect { job.perform_now }.not_to change { current_budget.enrichment_used }.from(1) + expect(transport.requests.size).to eq(1) + end + + it "reports idle rather than pretending it did the work again" do + job.perform_now + + expect(job.outcome).to include(enrichment_outcome: "idle") + end + + it "creates no duplicate entity row" do + expect { job.perform_now }.not_to change(GithubActor, :count).from(1) + end + + # A redelivery that lands while the first execution is still in flight: the lease on + # next_retry_at excludes the row from all four selector queries, so the second finds nothing + # rather than fetching the same entity twice. + describe "arriving while the first execution still holds the lease" do + it "finds nothing to do and spends nothing" do + GithubActor.update_all(enrichment_status: "pending", fetched_at: nil, + next_retry_at: frozen_time + 600) + + expect { job.perform_now }.not_to change { current_budget.enrichment_used }.from(1) + expect(job.outcome).to include(enrichment_outcome: "idle") + end + end + + # §10: "actor or repo URL returns 404/410 → entity permanent_failure". A redelivery must not + # re-attempt a decided entity, and must not reset the attempt counter that decided it. + describe "after a permanent failure" do + let(:ghost) { GithubActor.find_by(github_id: 7_700_421) } + + it "does not re-attempt the entity" do + create_actor(github_id: 7_700_421, login: "ghostuser", last_seen_at: frozen_time, + api_url: "https://api.github.com/users/ghostuser") + job.perform_now + before_attributes = ghost.attributes + + job.perform_now + + expect(ghost.reload).to have_attributes(enrichment_status: "permanent_failure") + expect(ghost.attributes).to eq(before_attributes) + end + end +end diff --git a/spec/recovery/pending_enrichment_recovery_spec.rb b/spec/recovery/pending_enrichment_recovery_spec.rb new file mode 100644 index 0000000..8f5deb9 --- /dev/null +++ b/spec/recovery/pending_enrichment_recovery_spec.rb @@ -0,0 +1,106 @@ +require "rails_helper" + +# §12's recovery tests: "Event committed but job not scheduled (reconciler sweep)" and +# "Pending enrichment rediscovered (entity-scoped)". +# +# The crash is expressed the way a crash actually presents itself: rows committed, queue +# empty. Nothing is stubbed to raise, because a SIGKILL runs no ensure block and no rescue — +# what it leaves behind is exactly this state, and §2A's claim is that this state is +# recoverable *because* the entity rows are the durable record of pending work. +# +# The page is ingested at Time.current rather than at frozen_time, because the reconciler +# reads the clock the worker will actually be holding — Solid Queue constructs the job, so +# there is nothing to inject — and §10's eligibility window is measured against it. +RSpec.describe "recovering enrichment work that was never enqueued", type: :integration do + let(:transport) { fixture_transport } + let(:ingested_at) { Time.current } + + before do + active_budget_window(now: frozen_time) + fixture_runner(transport: transport, now: ingested_at).call(event_source: fixture_event_source) + + # This line is the crash: the four push events and their six stub entities are committed, + # and the enqueue the run had just made is gone. + clear_enqueued_jobs + end + + it "leaves the work durable in the business tables, where nothing could lose it" do + expect(PushEvent.count).to eq(4) + expect(GithubActor.enrichment_candidates.count).to eq(3) + expect(GithubRepository.enrichment_candidates.count).to eq(3) + expect(enqueued_jobs).to be_empty + end + + it "rediscovers it on the next reconciler tick" do + expect { ReconcilePendingEnrichmentsJob.perform_now } + .to have_enqueued_job(EnrichActorJob).exactly(:once) + .and have_enqueued_job(EnrichRepositoryJob).exactly(:once) + end + + # §8: "a small, entity-scoped set, not N event rows per entity." Three actors behind four + # events are one cycle, not three and not four — the queue is not where the backlog lives. + it "schedules one cycle per class, not one per pending entity or per event" do + ReconcilePendingEnrichmentsJob.perform_now + + expect(enqueued_jobs.map { _1[:job] }).to contain_exactly(EnrichActorJob, EnrichRepositoryJob) + end + + # The sweep reads state and schedules; it never writes entity rows. If it did, a worker that + # came back after an hour would silently reset the backoff of everything it found. + it "changes no entity row while rediscovering the work" do + before_rows = GithubActor.order(:id).pluck(:id, :enrichment_status, :enrichment_attempts, :updated_at) + + ReconcilePendingEnrichmentsJob.perform_now + + expect(GithubActor.order(:id).pluck(:id, :enrichment_status, :enrichment_attempts, :updated_at)) + .to eq(before_rows) + end + + it "keeps scheduling on every tick until the work is actually done" do + 2.times { ReconcilePendingEnrichmentsJob.perform_now } + + expect(enqueued_jobs.count { _1[:job] == EnrichActorJob }).to eq(2) + end + + # The end of the story rather than the middle: the rediscovered work runs, and the entities + # reach the same durable state the un-crashed run would have produced. + it "completes the recovered work when the scheduled cycles run" do + allow(Github).to receive(:configuration).and_return(configuration_with("GITHUB_MODE" => "fixture")) + allow(Github::EnrichmentRunner).to receive(:new) + .and_return(fixture_enrichment_runner(transport: transport, now: frozen_time)) + + 6.times do + ReconcilePendingEnrichmentsJob.perform_now + perform_enqueued_jobs + end + + expect(GithubActor.find_by(github_id: 583_231)) + .to have_attributes(enrichment_status: "complete", name: "The Octocat") + expect(GithubRepository.find_by(github_id: 1_296_269).enrichment_status).to eq("complete") + end + + describe "when there is nothing left to recover" do + before do + GithubActor.update_all(enrichment_status: "complete", fetched_at: ingested_at) + GithubRepository.update_all(enrichment_status: "complete", fetched_at: ingested_at) + end + + it "schedules nothing" do + expect { ReconcilePendingEnrichmentsJob.perform_now }.not_to have_enqueued_job + end + end + + # A live worker's in-flight rows are not pending work: the claim lease is written onto + # next_retry_at, and every one of the selector's queries excludes it. Without this the + # reconciler would pile cycles onto entities another thread already holds. + describe "while a live worker holds every candidate" do + before do + GithubActor.update_all(next_retry_at: ingested_at + 600) + GithubRepository.update_all(next_retry_at: ingested_at + 600) + end + + it "schedules nothing" do + expect { ReconcilePendingEnrichmentsJob.perform_now }.not_to have_enqueued_job + end + end +end diff --git a/spec/recovery/source_contention_spec.rb b/spec/recovery/source_contention_spec.rb new file mode 100644 index 0000000..76915b0 --- /dev/null +++ b/spec/recovery/source_contention_spec.rb @@ -0,0 +1,80 @@ +require "rails_helper" + +# §12's "Multiple pollers attempt the same source", at the level PR 8 introduces it: the +# recurring tick, running in a worker container that may not be the only one. +# +# The contended lock is taken from a genuinely separate PostgreSQL session, never a thread — +# session advisory locks are re-entrant within a session, so a thread-based version of this +# spec would pass even if SourceLock did nothing at all (spec/support/advisory_lock_helpers.rb +# spells the trap out). +RSpec.describe "a tick against a source another poller owns", type: :integration do + let(:transport) { fixture_transport } + let!(:event_source) { fixture_event_source } + let(:key) { Github::AdvisoryLock.key_for(event_source.id) } + + before do + active_budget_window(now: frozen_time) + allow(Github).to receive(:configuration).and_return(configuration_with("GITHUB_MODE" => "fixture")) + + allow(Github::IngestionRunner).to receive(:new).and_call_original + runner = fixture_runner(transport: transport) + allow(Github::IngestionRunner).to receive(:new).and_return(runner) + end + + def tick(&block) + other_session_holding(Github::AdvisoryLock::SOURCE_LOCK_NAMESPACE, key) do + job = PollEventSourceJob.new + job.perform_now + block&.call(job) + job + end + end + + # §2A: "The poller attempts once and exits if unavailable." A raise would record a failed + # execution once a minute for as long as a one-shot ran, which is the system's own mutual + # exclusion working — not a defect. + it "completes the tick instead of failing the job" do + expect { tick }.not_to raise_error + end + + it "reports the contention at INFO, with the job id" do + allow(Rails.logger).to receive(:info) + + job = tick + + expect(Rails.logger).to have_received(:info).with( + hash_including(event: "ingestion.source_busy", event_source_id: event_source.id, job_id: job.job_id) + ) + end + + # The run row is opened inside the lock, which is what makes this provable rather than + # merely likely. + it "leaves no trace at all: no run, no request, no budget spent, no schedule moved" do + tick + + expect(IngestionRun.count).to eq(0) + expect(transport.requests).to be_empty + expect(current_budget.poll_used).to eq(0) + expect(event_source.reload) + .to have_attributes(consecutive_failures: 0, last_polled_at: nil, cadence_due_at: nil) + end + + it "schedules no enrichment, because no event was created" do + expect { tick }.not_to have_enqueued_job + end + + # Contention on one source must not cost the others their turn in this tick. + it "still polls the sources the other poller does not hold" do + other = create_event_source(source_type: "github_fixture_events", next_poll_at: nil) + + tick + + expect(IngestionRun.pluck(:event_source_id)).to eq([ other.id ]) + end + + it "leaves the lock-order tracking clean, on the busy path as much as the polled one" do + tick + + expect(Github::LockOrder.held_keys).to be_empty + end +end diff --git a/spec/recovery/worker_crash_lease_spec.rb b/spec/recovery/worker_crash_lease_spec.rb new file mode 100644 index 0000000..858c726 --- /dev/null +++ b/spec/recovery/worker_crash_lease_spec.rb @@ -0,0 +1,93 @@ +require "rails_helper" + +# §12's "Worker failure before completion". A real crash runs no ensure block, so it is +# modelled as what it leaves behind — a claim lease with no worker behind it — rather than by +# stubbing an exception, which would exercise the release path the crash skipped. +# +# The recovery has no cleanup code anywhere, and that is the design: the lease is written onto +# next_retry_at, so it expires by arithmetic. Nothing has to notice the worker died. +RSpec.describe "a lease left behind by a crashed worker", type: :integration do + let(:transport) { fixture_transport } + let(:configuration) { Github.configuration } + let(:claim) { Github::Enrichment::Claim.new(configuration: configuration) } + let(:actor_type) { Github::Enrichment::EntityType.fetch(:actor) } + let!(:actor) do + create_actor(github_id: 583_231, last_seen_at: frozen_time, + api_url: "https://api.github.com/users/octocat") + end + + # The crash: a lease taken, and then nothing. + let!(:lease) { claim.acquire(actor_type, pool: :pending, now: frozen_time) } + + before do + active_budget_window(now: frozen_time) + allow(Github).to receive(:configuration).and_return(configuration_with("GITHUB_MODE" => "fixture")) + end + + # The stub has to come off before the next runner is built: #fixture_enrichment_runner goes + # through Github::EnrichmentRunner.new itself, so a second call with the previous stub still + # in place would hand back the previous runner — and its clock, which is the one thing every + # example here is varying. + def enrich_at(instant) + allow(Github::EnrichmentRunner).to receive(:new).and_call_original + runner = fixture_enrichment_runner(transport: transport, now: instant) + allow(Github::EnrichmentRunner).to receive(:new).and_return(runner) + + job = EnrichActorJob.new + job.perform_now + job + end + + # Derived from §2A's pinned defaults — attempts × redirect hops × (gate wait + open + read) + # plus the backoff — so a configuration change moves this expectation with the code instead + # of leaving a stale literal behind. + it "holds the entity for exactly the derived lease window" do + expect(lease.leased_until - frozen_time).to eq(claim.lease_seconds) + end + + describe "while the lease is still live" do + it "finds nothing to do, and says so" do + expect(enrich_at(frozen_time).outcome).to include(enrichment_outcome: "idle") + end + + it "leaves the entity exactly as the dead worker left it" do + before_attributes = actor.reload.attributes + + enrich_at(frozen_time) + + expect(actor.reload.attributes).to eq(before_attributes) + expect(actor.enrichment_attempts).to eq(0) + expect(actor.last_error).to be_nil + end + + it "spends no budget on an entity it cannot claim" do + expect { enrich_at(frozen_time) }.not_to change { current_budget.enrichment_used }.from(0) + expect(transport.requests).to be_empty + end + end + + describe "once the lease expires" do + it "enriches the entity, with no sweeper and no cleanup step in between" do + enrich_at(lease.leased_until) + + expect(actor.reload) + .to have_attributes(enrichment_status: "complete", name: "The Octocat") + end + + it "spends exactly one request for the whole crash-and-recovery sequence" do + enrich_at(frozen_time) + enrich_at(lease.leased_until) + + expect(current_budget.enrichment_used).to eq(1) + expect(transport.requests.size).to eq(1) + end + + # The crash cost the entity nothing: attempts count attempts *since the last success*, and + # a claim that was never used is not one. + it "charges the entity no attempt for the crash" do + enrich_at(lease.leased_until) + + expect(actor.reload.enrichment_attempts).to eq(0) + end + end +end diff --git a/spec/services/github/enrichment/dispatch_spec.rb b/spec/services/github/enrichment/dispatch_spec.rb new file mode 100644 index 0000000..6b75d84 --- /dev/null +++ b/spec/services/github/enrichment/dispatch_spec.rb @@ -0,0 +1,139 @@ +require "rails_helper" + +# The one rule both enqueue paths share (§8 steps 10 and 11): "is there durable enrichment +# work this class could do right now?", answered from the committed entity rows and the +# ledger, never from the queue. +RSpec.describe Github::Enrichment::Dispatch do + subject(:dispatch) { described_class.new(clock: -> { frozen_time }) } + + def actor(**overrides) + create_actor(github_id: 583_231, last_seen_at: frozen_time, **overrides) + end + + def repository(**overrides) + create_repository(github_id: 1_296_269, last_seen_at: frozen_time, **overrides) + end + + before { active_budget_window(now: frozen_time) } + + describe "when a class has work" do + it "enqueues one cycle for each class that does, and none for the class that does not" do + actor + + expect { dispatch.call(reason: "reconcile") } + .to have_enqueued_job(EnrichActorJob).exactly(:once) + expect(ActiveJob::Base.queue_adapter.enqueued_jobs.map { _1[:job] }).to eq([ EnrichActorJob ]) + end + + # However deep the backlog. Github::EnrichmentRunner enriches at most one entity per call, + # so queue depth is set by §10's hourly allowance and not by how many rows are waiting — + # 90 pending actors would otherwise become 90 cycles the ledger refuses 40 requests in. + it "enqueues one cycle whether one entity is pending or fifty" do + 50.times { |index| create_actor(github_id: 1_000 + index, last_seen_at: frozen_time) } + + expect { dispatch.call(reason: "reconcile") }.to have_enqueued_job(EnrichActorJob).exactly(:once) + end + + it "reports what it scheduled" do + actor + repository + + expect(dispatch.call(reason: "ingestion")) + .to eq(actor_enqueued: 1, repository_enqueued: 1, reason: "ingestion") + end + + # §11 lists "reconciliation summaries" among the INFO events, and PR 7's Summary is what + # fills it — per-status counts, per-class share usage, the window state. + it "logs the summary at INFO, because a tick that scheduled work is worth reading" do + actor + allow(Rails.logger).to receive(:info) + + dispatch.call(reason: "reconcile") + + expect(Rails.logger).to have_received(:info).with( + hash_including(event: "enrichment.dispatched", reason: "reconcile", actor_enqueued: 1, + enrichment_used: 0, enrichment_allowance: 40, window_status: "active") + ) + end + end + + describe "when there is nothing to do" do + it "enqueues nothing when every entity is decided" do + actor(enrichment_status: "complete", fetched_at: frozen_time) + repository(enrichment_status: "permanent_failure") + + expect { dispatch.call(reason: "reconcile") }.not_to have_enqueued_job + end + + # An entity another worker is mid-way through is not pending work: the claim lease lives + # on next_retry_at, and every one of the selector's queries excludes it. + it "enqueues nothing while the only candidate is leased by a live worker" do + actor + GithubActor.update_all(next_retry_at: frozen_time + 600) + + expect { dispatch.call(reason: "reconcile") }.not_to have_enqueued_job + end + + # The steady state of an exhausted window, once a minute for the rest of the hour — the + # volume argument Github::BudgetLedger#log_class_exhausted already makes. + it "keeps the line at DEBUG so an idle tick is not INFO noise" do + allow(Rails.logger).to receive(:info) + allow(Rails.logger).to receive(:debug) + + dispatch.call(reason: "reconcile") + + expect(Rails.logger).to have_received(:debug).with(hash_including(event: "enrichment.dispatched")) + expect(Rails.logger).not_to have_received(:info).with(hash_including(event: "enrichment.dispatched")) + end + end + + # §9's effective_enrichment_time, minus the entity component this object is not choosing. + describe "when the ledger says enrichment cannot happen" do + it "enqueues nothing while a global block is in force, and names it" do + actor + active_budget_window(now: frozen_time, global_blocked_until: frozen_time + 300) + + expect { dispatch.call(reason: "reconcile") }.not_to have_enqueued_job + expect(dispatch.call(reason: "reconcile")).to include(blocked_by: :global_blocked_until) + end + + it "enqueues nothing once the enrichment class has spent its allowance" do + actor + active_budget_window(now: frozen_time, enrichment_used: 40, reset_at: frozen_time + 600) + + expect { dispatch.call(reason: "reconcile") }.not_to have_enqueued_job + expect(dispatch.call(reason: "reconcile")).to include(blocked_by: :enrichment_class_blocked_until) + end + + # §10: a share exhaustion is a *denial* relieved by borrowing, not a deferral. Refusing to + # enqueue on it would withhold work the ledger would have granted — the reason + # Github::EnrichmentSchedule leaves the share out too. + it "still enqueues when only one class's fairness share is spent" do + actor + active_budget_window(now: frozen_time, actor_share_used: 20, enrichment_used: 20) + + expect { dispatch.call(reason: "reconcile") }.to have_enqueued_job(EnrichActorJob) + end + end + + # A clean checkout has no ledger row: nothing seeds it, and only a reservation inside the + # request gate creates one. Reading a schedule must not. + describe "before the first window exists" do + it "enqueues without creating the ledger row" do + GithubApiBudget.delete_all + actor + + expect { dispatch.call(reason: "ingestion") }.to have_enqueued_job(EnrichActorJob) + expect(GithubApiBudget.count).to eq(0) + end + end + + it "makes no GitHub request and takes no lock" do + actor + expect(Github::RequestGate).not_to receive(:hold) + + dispatch.call(reason: "reconcile") + + expect(WebMock).not_to have_requested(:any, //) + end +end diff --git a/spec/services/github/ingestion_runner_spec.rb b/spec/services/github/ingestion_runner_spec.rb index 201b5f7..18d0338 100644 --- a/spec/services/github/ingestion_runner_spec.rb +++ b/spec/services/github/ingestion_runner_spec.rb @@ -429,8 +429,60 @@ def ingest(runner = fixture_runner, **options) expect(observed).to all(be(false)) end - # §9's poll cadence. Nothing fires it on a schedule until PR 8 — this is the gate every - # caller passes through, whoever calls it. + # §8 step 10. The enqueue is a hint and the committed entity rows are the durable record, + # so what these examples pin is *when* the hint is emitted — after every row is durable, + # outside the source lock, and only when the run created something. + describe "enqueueing enrichment after commit" do + it "enqueues one cycle per class when the run created events" do + expect { ingest }.to have_enqueued_job(EnrichActorJob).exactly(:once) + .and have_enqueued_job(EnrichRepositoryJob).exactly(:once) + end + + # §7 merge rule 4's boundary, at the queue: a replay refreshes identity and reactivates + # nothing, so there is no new work to schedule. Anything still pending from the first run + # is ReconcilePendingEnrichmentsJob's business, not this run's. + it "enqueues nothing for a replay that created no events" do + ingest + + replay = fixture_runner(transport: fixture_transport, now: frozen_time + 301) + + expect { ingest(replay) }.not_to have_enqueued_job + end + + it "enqueues nothing for a tick that was not due" do + ingest + + expect { ingest(fixture_runner(now: frozen_time + 60)) }.not_to have_enqueued_job + end + + # The three properties in one place, because each of them alone would let a plausible + # refactor through: enqueue inside PageWriter's transaction (rows not yet durable), + # enqueue in #finish before the run row closes (a "pending work" hint that outlives an + # unfinished run), or enqueue inside SourceLock (a source's mutual exclusion held across + # a write to another database). + it "enqueues only once the rows are durable, the run row is closed, and the lock is released" do + observed = [] + allow(EnrichActorJob).to receive(:perform_later) do + transaction = ActiveRecord::Base.lease_connection.current_transaction + + observed << { + open_transaction: transaction.open? && transaction.joinable?, + run_status: IngestionRun.sole.status, + persisted: PushEvent.count, + source_lock_free: lock_available_to_other_session?( + Github::AdvisoryLock::SOURCE_LOCK_NAMESPACE, Github::AdvisoryLock.key_for(event_source.id) + ) + } + end + + ingest + + expect(observed).to eq([ { open_transaction: false, run_status: "completed", persisted: 4, + source_lock_free: true } ]) + end + end + + # §9's poll cadence — the gate every caller passes through, whoever calls it. describe "the poll cadence" do let(:transport) { fixture_transport } diff --git a/spec/support/advisory_lock_helpers.rb b/spec/support/advisory_lock_helpers.rb index c70ba6c..0bdfd16 100644 --- a/spec/support/advisory_lock_helpers.rb +++ b/spec/support/advisory_lock_helpers.rb @@ -50,6 +50,65 @@ def close_second_session @second_session = nil end + # Takes the lock and returns, with no ensure to undo it — the point of a session-death + # example is that nothing runs on the way out. #other_session_holding cannot be used for + # that: its ensure would exec on a connection that no longer exists. + def acquire_in_other_session(namespace, key) + second_session.exec_params("SELECT pg_advisory_lock($1::int, $2::int)", [ namespace, key ]) + end + + def second_session_pid + second_session.exec("SELECT pg_backend_pid()").getvalue(0, 0).to_i + end + + # Session death the way a container kill produces it: PostgreSQL ends the backend, the + # client never says goodbye, and no Ruby ensure block runs anywhere. A plain #close is a + # *disconnect* — the client cooperating — and proves something weaker. + # + # Issued through the pooled connection rather than a third socket, so this adds no cleanup + # surface; the local socket is closed afterwards because its backend is already gone. + # + # pg_terminate_backend returns before the backend has finished exiting, so a caller must + # never assert on the very next line — use the production wait (SourceLock's wait_seconds) + # or #wait_for_advisory_lock_release. + # @return [Integer] the pid that was terminated + def terminate_second_session! + pid = second_session_pid + ActiveRecord::Base.connection.exec_query( + ActiveRecord::Base.sanitize_sql_array([ "SELECT pg_terminate_backend(?)", pid ]), + "AdvisoryLockHelpers Terminate" + ) + close_second_session + + pid + end + + # Which backends hold this lock right now, straight from the server. Uncached, because the + # query cache would happily answer a second question with the first answer. + def advisory_lock_holders(namespace, key) + ActiveRecord::Base.uncached do + ActiveRecord::Base.connection.select_values( + ActiveRecord::Base.sanitize_sql_array([ <<~SQL.squish, namespace, key ]) + SELECT pid FROM pg_locks + WHERE locktype = 'advisory' AND classid::bigint = ? AND objid::bigint = ? + SQL + ) + end + end + + # Bounded, and deliberately not a bare sleep: the release lands within a millisecond or two + # of the terminate, and if it ever stops landing the example must fail rather than hang. + def wait_for_advisory_lock_release(namespace, key, timeout: 5) + deadline = Process.clock_gettime(Process::CLOCK_MONOTONIC) + timeout + + until advisory_lock_holders(namespace, key).empty? + raise "advisory lock #{namespace}:#{key} still held #{timeout}s after session death" if + Process.clock_gettime(Process::CLOCK_MONOTONIC) > deadline + + sleep 0.01 + end + end + # A scripted clock and a no-op sleeper, so a wait contract is asserted in zero # wall-clock time. Readings are consumed in order: the first is the deadline # baseline, then one per loop turn. @@ -69,6 +128,14 @@ def scripted_clock(*readings) config.after do close_second_session + # Github::LockOrder tracks held locks per execution context, not per connection, and PR 8 + # runs code inside job threads that Solid Queue reuses. A marker left behind by one job + # would make the *next* job on that thread raise Errors::ReentrantLock for a lock it never + # took — a non-local failure that is nearly impossible to read backwards. Green today, and + # the only thing that would catch it. + tracked = Github::LockOrder.held_keys.dup + Github::LockOrder.held_keys.clear + namespaces = [ Github::AdvisoryLock::SOURCE_LOCK_NAMESPACE, Github::AdvisoryLock::REQUEST_GATE_NAMESPACE ] # An example that made the connection unavailable — spec/requests/health_spec.rb @@ -87,9 +154,12 @@ def scripted_clock(*readings) [] end + raise "Github::LockOrder tracking leaked out of this example: #{tracked.inspect}" if + leaked.empty? && tracked.any? + next if leaked.empty? ActiveRecord::Base.connection_pool.with_connection { |c| c.execute("SELECT pg_advisory_unlock_all()") } - raise "advisory lock(s) leaked out of this example: #{leaked.join(", ")}" + raise "advisory lock(s) leaked out of this example: #{leaked.join(", ")}#{" (tracked: #{tracked.inspect})" if tracked.any?}" end end diff --git a/spec/support/queue_helpers.rb b/spec/support/queue_helpers.rb new file mode 100644 index 0000000..7a83ad3 --- /dev/null +++ b/spec/support/queue_helpers.rb @@ -0,0 +1,56 @@ +# Active Job's adapter policy for the suite (IMPLEMENTATION_PLAN.md §2A: "Ordinary specs use +# Active Job's test adapter; only dedicated queue integration tests touch the queue test +# database"). +# +# config/environments/test.rb sets the test adapter, so the default needs no help here. What +# this file adds is the exception — the handful of examples tagged :queue, which swap in the +# real Solid Queue adapter to prove the parts no double can: that an enqueue reaches the +# queue *database*, that connects_to routes it there, and that the schema is loaded. +module QueueHelpers + # Every table db/queue_migrate creates, jobs last so the foreign keys unwind in order. + SOLID_QUEUE_TABLES = %w[ + solid_queue_blocked_executions solid_queue_claimed_executions solid_queue_failed_executions + solid_queue_ready_executions solid_queue_recurring_executions solid_queue_scheduled_executions + solid_queue_pauses solid_queue_processes solid_queue_recurring_tasks solid_queue_semaphores + solid_queue_jobs + ].freeze +end + +RSpec.configure do |config| + # ActiveJob::TestHelper forces the test adapter in before_setup — a prepend_before, so it + # runs *inside* the around hook below and would silently undo the swap. Deriving the + # include from metadata is what keeps the two policies from fighting; spec/job_boundary_spec.rb + # asserts the split holds. + config.define_derived_metadata do |metadata| + metadata[:active_job_test_adapter] = true unless metadata[:queue] + end + + config.include ActiveJob::TestHelper, :active_job_test_adapter + + config.around(:each, :queue) do |example| + previous = ActiveJob::Base.queue_adapter + ActiveJob::Base.queue_adapter = :solid_queue + + begin + example.run + ensure + ActiveJob::Base.queue_adapter = previous + end + end + + # Two jobs at once: it clears rows a crashed run — or CI's `bin/jobs` supervisor smoke, + # which runs against RAILS_ENV=test — committed where no example transaction will ever roll + # them back, and it turns a queue test database that was never prepared into one actionable + # sentence instead of a PG::UndefinedTable inside whichever :queue example ran first. + # + # before(:suite), because this is the only point at which no fixture transaction is open. + config.before(:suite) do + SolidQueue::Job.connection.truncate_tables(*QueueHelpers::SOLID_QUEUE_TABLES) + rescue ActiveRecord::StatementInvalid => error + abort <<~MESSAGE + The queue test database is not prepared: #{error.message.lines.first&.strip} + + Run `bin/rails db:test:prepare` (the `test` compose service and CI both do). + MESSAGE + end +end diff --git a/spec/support/shared_examples/enrichment_job.rb b/spec/support/shared_examples/enrichment_job.rb new file mode 100644 index 0000000..8000073 --- /dev/null +++ b/spec/support/shared_examples/enrichment_job.rb @@ -0,0 +1,66 @@ +# EnrichActorJob and EnrichRepositoryJob are the same job over §7's identical state machine, +# so their contract is written once — the precedent spec/support/shared_examples/ +# enrichable_entity.rb set for the models themselves. +# +# What is actually being asserted is the job's *boundary*: which class it asks for, that one +# call is one entity, that it never reaches for a source lock, and that an outcome nobody has +# to act on is not an error. +RSpec.shared_examples "an enrichment job" do |entity_class:, entity_type:, log_key:| + let(:runner) { instance_double(Github::EnrichmentRunner) } + let(:enriched) do + Github::EnrichmentRunner::Result.new(status: "enriched", entity_type: entity_type, github_id: 4_242) + end + + before { allow(Github::EnrichmentRunner).to receive(:new).and_return(runner) } + + it "runs exactly one cycle, narrowed to its own class" do + expect(runner).to receive(:call).with(entity_class: entity_class).once.and_return(enriched) + + described_class.new.perform_now + end + + it "joins the cycle to the job on one line" do + allow(runner).to receive(:call).and_return(enriched) + allow(Rails.logger).to receive(:info) + + job = described_class.new + job.perform_now + + expect(Rails.logger).to have_received(:info).with( + hash_including(event: "job.completed", job_id: job.job_id, job_class: described_class.name, + log_key => 4_242, enrichment_outcome: "enriched") + ) + end + + # §8 step 1: "Enrichment jobs skip this step — they take only the request gate." Asserted at + # the job boundary as well as inside the runner, because this is where a future "just lock + # the source while we enrich it" would be written. + it "never takes a source lock" do + allow(runner).to receive(:call).and_return(enriched) + expect(Github::SourceLock).not_to receive(:acquire) + + described_class.new.perform_now + + expect(Github::LockOrder.held_keys).to be_empty + end + + # Nothing eligible, or a ledger that refused: ordinary outcomes of a system whose budget is + # 40 requests an hour. Failing the job would fill solid_queue_failed_executions with the + # steady state. + %w[idle deferred].each do |status| + it "treats a #{status} cycle as a completed job" do + allow(runner).to receive(:call) + .and_return(Github::EnrichmentRunner::Result.new(status: status, deferral_reason: "no_candidate")) + + expect { described_class.new.perform_now }.not_to raise_error + end + end + + # §6 requires a corpus gap to be raised rather than laundered into a failed fetch, and the + # runner has already put the lease back by the time it arrives here. + it "lets a fixture corpus gap fail the job" do + allow(runner).to receive(:call).and_raise(Github::Errors::FixtureMiss, "no such body") + + expect { described_class.new.perform_now }.to raise_error(Github::Errors::FixtureMiss) + end +end diff --git a/spec/support/webmock.rb b/spec/support/webmock.rb index ff0098b..902373b 100644 --- a/spec/support/webmock.rb +++ b/spec/support/webmock.rb @@ -8,6 +8,6 @@ # a socket, so this application has no local HTTP dependency to exempt. # # PostgreSQL is unaffected. WebMock hooks Ruby HTTP client libraries; the pg gem talks -# to the server through libpq's own socket, which WebMock never sees. Solid Queue -# (PR 8) is PostgreSQL-backed for the same reason. +# to the server through libpq's own socket, which WebMock never sees. Solid Queue is +# PostgreSQL-backed and reaches its own database the same way, unhooked. WebMock.disable_net_connect!(allow_localhost: false)