diff --git a/CHANGELOG.md b/CHANGELOG.md index 0c119cc..e69e9bd 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,112 @@ # Changelog +## Unreleased + +### Dashboard SSE hardening + +- Make SSE the default dashboard transport. `:polling` remains an explicit compatibility fallback. +- Replace one `PRAGMA data_version` loop **per browser connection** with one shared `EventHub` watcher per Rack app process while at least one dashboard tab is connected. It fans complete overview snapshots to every SSE body through latest-value mailboxes, so slow tabs do not accumulate unbounded event queues. +- Use complete snapshots on connect and after a change rather than a replay log: reconnecting EventSource clients cannot miss the current queue state. +- Coalesce client-side list refreshes after a burst of changes, cancel stale list fetches on tab switches, and retain previous pages when `Load more` is used. +- Add SSE retry control, 25-second heartbeat frames, `no-cache, no-transform`, and `X-Accel-Buffering: no`. Remove the hop-by-hop `Connection` response header. +- Fingerprint asset URLs and cache immutable assets by digest, so a dashboard deploy cannot leave a browser on incompatible HTML/JS/CSS. +- Add coverage for event fan-out, latest-value coalescing, clean stream shutdown, forced overview reads, and the SSE configuration constraints. + +## 1.1.0 + +Server-Sent Events transport for the dashboard. Replaces HTTP polling as the recommended transport. + +### Added + +- **SSE transport for the dashboard.** Set `c.transport = :sse` in `Async::Background::Web.configure` and the dashboard now uses a single long-lived `text/event-stream` connection per browser tab instead of polling `/api/overview` every 2 seconds. The browser opens `EventSource(mount_path + '/api/stream')` once; the server pushes an `overview` event whenever `PRAGMA data_version` changes, and a `:keepalive` comment frame every 30 seconds. Result: 1 HTTP connection per dashboard tab regardless of how long it stays open, instead of 30 req/min per tab. + + - New module `Async::Background::Web::Stream` implements the event loop as a Rack streaming body (responds to `#each`, yields SSE frames). Holds no state across requests. + - New route `GET /api/stream` returns `200 text/event-stream` when `transport == :sse`, `404` otherwise. Subject to the same auth gate as every other endpoint. + - New `Response.sse(body)` helper sets the correct headers including `x-accel-buffering: no` (disables nginx buffering for the streaming response). + - JS client (`assets.rb`) detects `state.config.transport === 'sse'` at boot and chooses `EventSource` over `setInterval(tick, ...)`. Both transports share the same `applyOverview()` and `refreshActiveList()` handlers, so the UI behaves identically. + +- **`Configuration#transport`** with default `:polling` (backward compatible) and accepted values `:polling | :sse`. Validation rejects anything else with `ConfigurationError`. The chosen transport is exposed at `/api/config` so the client knows which path to take. + +### Migration + +Existing deployments keep working unchanged. To opt into SSE: + +```ruby +Async::Background::Web.configure do |c| + c.queue_path = ... + c.auth = ->(env) { ... } + c.transport = :sse +end +``` + +### Server compatibility note + +SSE holds the request thread/fiber open for the lifetime of the dashboard tab. **Recommended for Falcon**, which handles long-lived connections natively via fibers. **Puma works** but each open dashboard tab holds one worker thread for its lifetime — fine for an admin dashboard with a handful of operators, problematic if many concurrent operators would starve the worker pool. **Unicorn does not work** for SSE since its blocking worker model can't hold long-lived connections without timeouts; stay on `:polling` there. + +### Backend-side polling + +The server still polls `PRAGMA data_version` every 500ms inside the snapshot connection to detect changes. This is a connection-local PRAGMA call, microseconds, never hits a rate limiter. Client-facing transport is push. + +### Tests + +- New `spec/async/background/web/stream_spec.rb` — covers overview event on data_version change, heartbeat after idle, graceful exit on `EPIPE`/`IOError`, error frame on `ClosedError`/`UnavailableError`. +- Extended `spec/async/background/web/app_spec.rb` — `/api/stream` returns 404 on polling default, 200 text/event-stream on `:sse`, 401 without auth. +- Extended `spec/async/background/web/configuration_spec.rb` — accepts `:sse`, rejects unknown transports. + +## 1.0.0 + +First stable release. The queue execution contract from 0.7.2 (claim-token CAS, lifecycle columns, barrier-based shutdown drain, per-status partial indexes, versioned migrations) is now considered the public API. + +### Features + +- **Web dashboard.** Rack-mountable read-only UI under `require 'async/background/web'`. Vanilla HTML/CSS/JS, no JS framework, no npm. + - Endpoints: `GET /`, `GET /assets/app.css`, `GET /assets/app.js`, `GET /api/overview`, `GET /api/executing`, `GET /api/claimed`, `GET /api/pending`, `GET /api/done`, `GET /api/failed`, `GET /api/metrics`, `GET /api/config`. + - Default transport is JSON polling (`poll_interval_ms`, default 2000). SSE adapter for Falcon is intentionally deferred to a later release; the dashboard already coalesces work via a shared overview cache, so adding SSE later is a backward-compatible change. + - Read path runs through `Async::Background::Web::Snapshot`, which opens SQLite with `file:?mode=ro`, wraps a `Mutex` around a single shared connection, and uses one read transaction per endpoint and caches each overview as one consistent snapshot. + - Distinguishes `Executing` (`status='running' AND started_at IS NOT NULL`) from `Claimed` (`status='running' AND started_at IS NULL`). + - Overview snapshot cache for `counts_cache_ttl` seconds (default 3.0) so a busy queue does not turn the dashboard into a hot reader. + - Cursor pagination for `done`/`failed`/`pending` using `(finished_at, id)` / `(run_at, id)` tuples. Stable on ties. + - Args hidden by default (`expose_args: false`); when enabled, content runs through `redact_args`. All user content rendered through `textContent`, never `innerHTML`. + - Auth hook is **mandatory**. `Configuration#validate!` rejects an unconfigured `auth`. There is no permissive default. + + - Add the optional Rack dashboard for the SQLite queue. + - Make sqlite3 an explicit runtime dependency for queue/dashboard installs. + +### Configuration + +```ruby +require 'async/background/web' + +Async::Background::Queue::Store.prepare_dashboard!(path: '/var/lib/app/queue.db') + +Async::Background::Web.configure do |c| + c.queue_path = '/var/lib/app/queue.db' + c.auth = ->(env) { env['warden'].user&.admin? } + c.expose_args = false + c.metrics_path = '/run/app/async-background.shm' + c.total_workers = 4 + c.counts_cache_ttl = 3.0 + c.poll_interval_ms = 2000 + c.list_limit = 50 + c.mount_path = '/admin/background' + c.title = 'My App background jobs' +end + +run Async::Background::Web.app +``` + +### Dependencies + +- `rack` is an optional dependency. Required only when `require 'async/background/web'` is loaded. Core gem and worker processes do not require it. + +### Breaking changes from 0.7.x + +None beyond what 0.7.2 already shipped. The 1.0 line locks the existing contract: + +- `Queue::Store#fetch` returns `claim_token` in the result hash. +- All terminal `Queue::Store` methods (`complete`, `fail`, `retry_or_fail`) require the `claim_token:` kwarg and return CAS success boolean / `:retried` / `:failed` / `nil`. +- Schema is versioned via `PRAGMA user_version`. Use `Queue::Store.migrate!(path:)` to upgrade. Use `Queue::Store.prepare_dashboard!(path:)` from the dashboard process to lazily create dashboard-only indexes (per-status partial indexes for `done` / `failed`, plus separate `executing` and `claimed` indexes). + ## 0.7.2 - Harden queue execution, retries, shutdown, and metrics. @@ -102,7 +209,7 @@ - Proper job distribution validation across worker pool - **Test fixtures** — dedicated `ci/fixtures/jobs.rb` and `ci/fixtures/schedule.yml` for scenario testing -### Bug Fixes +### Bug Fixes - **SQLite busy timeout** — added `PRAGMA busy_timeout = 5000` to `Queue::Store` to prevent `SQLITE_BUSY` errors under concurrent multi-process database access - **Enhanced Queue::Notifier error handling** — restructured IO error handling with clearer categorization: - `WRITE_DROPPED` for write failures (`IO::WaitWritable`, `Errno::EAGAIN`, `IOError`, `Errno::EPIPE`) — all non-fatal as job is already in store diff --git a/README.md b/README.md index 53c2fb2..b4f6607 100644 --- a/README.md +++ b/README.md @@ -12,7 +12,7 @@ A lightweight cron, interval, and job-queue scheduler for Ruby's [Async](https:/ - Ruby >= 3.3 - `async ~> 2.0`, `fugit ~> 1.0` -- `sqlite3 ~> 2.0` (optional, for the job queue) +- `sqlite3 ~> 2.0` (optional, storage) - `async-utilization >= 0.3, < 0.5` (optional, for metrics) ## Install @@ -20,7 +20,7 @@ A lightweight cron, interval, and job-queue scheduler for Ruby's [Async](https:/ ```ruby # Gemfile gem "async-background" -gem "sqlite3", "~> 2.0" # optional +gem "sqlite3", "~> 2.0" # optional gem "async-utilization", ">= 0.3", "< 0.5" # optional ``` @@ -126,6 +126,12 @@ The dynamic queue runs alongside it: Jobs are persisted in SQLite, so a missed wake-up is never a lost job — workers also poll every 5 seconds as a safety net. +### Queue-only workers + +Recurring schedules are optional. A worker that serves only dynamic jobs starts with +`config_path: nil` and a `queue_socket_dir`; it does not need a placeholder schedule file. +A supplied schedule path stays strict and raises when the file is missing or empty. + ### Schema migration during deploy Run queue migrations once in the release/pre-deploy step, before starting new web or worker @@ -140,18 +146,18 @@ A fresh database still self-initializes on first use for local development, but migration is the production path. For an existing queue, finish or stop 0.7.1 producers/workers, run the migration once, then start 0.7.2 processes. -### Future dashboard indexes +### Dashboard indexes The queue does **not** install dashboard indexes by default. They slow every enqueue even though -pending rows never enter terminal or in-flight read-model indexes. When the 1.0 dashboard module -is enabled, its installer will run this once in the same release step: +pending rows never enter terminal or in-flight read-model indexes. Enable them once before +mounting the dashboard: ```ruby Async::Background::Queue.prepare_dashboard!(path: ENV.fetch("QUEUE_DB_PATH")) ``` -It adds three compact indexes: one each for cursor-sorted done and failed jobs, plus one for -the bounded in-flight list. It does not change queue behavior or rerun the core migration. +It adds four compact indexes: cursor-sorted `done` / `failed` history plus separate +`executing` / `claimed` in-flight lists. It does not change queue behavior or rerun the core migration. ## Metrics diff --git a/async-background.gemspec b/async-background.gemspec index 1137948..a0fabb0 100644 --- a/async-background.gemspec +++ b/async-background.gemspec @@ -33,12 +33,14 @@ Gem::Specification.new do |spec| spec.add_dependency 'async', '~> 2.0' spec.add_dependency 'console', '~> 1.0' spec.add_dependency 'fugit', '~> 1.0' + spec.add_dependency 'base64', '~> 0.2' # Optional: add to your own Gemfile if you need these features - # gem 'sqlite3', '~> 2.0' # dynamic job queue + # gem 'sqlite3', '~> 2.0' # gem 'async-utilization', '>= 0.3', '< 0.5' # shared-memory worker metrics spec.add_development_dependency 'rake', '~> 13.0' spec.add_development_dependency 'rspec', '~> 3.12' + spec.add_development_dependency 'rack', '~> 3.0' spec.add_development_dependency 'async-utilization', '>= 0.3', '< 0.5' end diff --git a/docs/GET_STARTED.md b/docs/GET_STARTED.md index a13ddf1..4cb3dc2 100644 --- a/docs/GET_STARTED.md +++ b/docs/GET_STARTED.md @@ -172,8 +172,8 @@ Async::Background::Queue.migrate!(path: ENV.fetch("QUEUE_DB_PATH")) ``` For an existing queue, stop or drain 0.7.1 processes first, run the migration, then start 0.7.2. -The base queue keeps only its pending-job index. A future dashboard installer can later call -`Async::Background::Queue.prepare_dashboard!(path: DB_PATH)` once to add its three read-model +The base queue keeps only its pending-job index. Before mounting the dashboard, call +`Async::Background::Queue.prepare_dashboard!(path: DB_PATH)` once to add its four read-model indexes without slowing normal enqueue-only deployments. That's the full config — both web and scheduler share the same SQLite file and notify each other through Unix domain sockets. Web controllers can now enqueue: @@ -190,6 +190,25 @@ end > **How wake-up works.** When any process (web or scheduler) enqueues a job, `SocketNotifier` sends one byte to a Unix domain socket. The chosen background worker wakes in ~30–80 µs and reads from SQLite — no polling delay. +### Queue-only worker + +A recurring schedule is optional. For applications that use only `perform_async`, +`perform_in`, and `perform_at`, pass `config_path: nil`. The worker keeps listening +to the queue until it receives `SIGTERM` / `SIGINT`; no placeholder YAML file is needed. + +```ruby +Async::Background::Runner.new( + config_path: nil, + worker_index: 1, + total_workers: 1, + queue_db_path: Rails.root.join("storage/async-background.sqlite3").to_s, + queue_socket_dir: "/tmp" +).run +``` + +A non-`nil` `config_path` remains strict: a missing or empty schedule file raises +`Async::Background::ConfigError` rather than silently disabling recurring jobs. + ### Environment variables | Variable | Default | Description | @@ -217,6 +236,128 @@ is `false` and `Async::Background::Metrics.read_all(...)` returns `[]`. When web background run in separate containers, point `ASYNC_BACKGROUND_METRICS_PATH` at a file under a shared volume (for example `/app/tmp/queue/async-background.shm`). +--- + +## Step 2.5 — Mount the optional dashboard + +The dashboard is a separate, read-only Rack app over the same SQLite file. It never enqueues, +retries, deletes, or otherwise mutates jobs. Before mounting it, add its read-model indexes once +in the same release step as the queue migration: + +```ruby +# bin/migrate_async_background +require "async/background/queue/client" + +queue_path = ENV.fetch("QUEUE_DB_PATH") +Async::Background::Queue.migrate!(path: queue_path) +Async::Background::Queue.prepare_dashboard!(path: queue_path) +``` + +`prepare_dashboard!` is idempotent. It installs four dashboard-only indexes for done, failed, +claimed, and executing lists; the core pending index already exists. Normal queue-only deployments +do not pay their write cost. + +### Rack / Falcon + +Put this in the Rack app that serves the dashboard (for example, a dedicated `config.ru`): + +```ruby +# frozen_string_literal: true + +require "async/background/web" + +Async::Background::Web.configure do |config| + config.queue_path = ENV.fetch("QUEUE_DB_PATH", "/var/lib/app/queue.db") + config.auth = ->(env) { env["warden"]&.user&.admin? } + + # Optional: requires async-utilization and a shared path visible to workers. + config.metrics_path = ENV["ASYNC_BACKGROUND_METRICS_PATH"] + config.total_workers = ENV.fetch("BACKGROUND_FORKS", 1).to_i + + # Must match the Rack/Rails mount point below. + config.mount_path = "/admin/background" + + # Default transport. One SSE connection per open tab; no browser polling. + config.transport = :sse + config.stream_poll_seconds = 0.5 # one SQLite change check per Rack process + config.stream_heartbeat_seconds = 25.0 # keeps proxies from idling out the stream + config.stream_retry_ms = 5_000 +end + +run Async::Background::Web.app +``` + +Add `rack` to the application bundle when it is not already present: + +```ruby +gem "rack", "~> 3.0" +``` + +When metrics are not needed, omit both `metrics_path` and `total_workers`. + +### Rails + +Configure the dashboard once during boot: + +```ruby +# config/initializers/async_background_dashboard.rb +require "async/background/web" + +Async::Background::Web.configure do |config| + config.queue_path = ENV.fetch("QUEUE_DB_PATH", Rails.root.join("tmp/queue/background.db").to_s) + config.auth = ->(env) { env["warden"]&.user&.admin? } + config.metrics_path = ENV["ASYNC_BACKGROUND_METRICS_PATH"] + config.total_workers = ENV.fetch("BACKGROUND_FORKS", 1).to_i + config.mount_path = "/admin/background" + config.transport = :sse +end +``` + +Then mount the Rack app: + +```ruby +# config/routes.rb +mount Async::Background::Web.app => "/admin/background" +``` + +Use an application-specific authorization predicate. The gem intentionally has no permissive +default: a missing or falsey `auth` result returns `401`. Do not expose the dashboard publicly +without an authentication layer in front of it. + +### Live updates and rate limits + +SSE is the default transport. A dashboard tab opens one authenticated `GET /api/stream` request; +the server sends a complete overview snapshot after connect and after the queue changes. The browser +performs ordinary JSON requests only for the initial page and when it needs to redraw the *active* +list. It does **not** poll on a timer. + +Internally, each Rack process with at least one connected dashboard uses one long-lived SQLite read +connection and compares `PRAGMA data_version` every `stream_poll_seconds` (default `0.5`). This is a +single local database read per process, not per browser tab. SQLite documents `data_version` for +exactly this interactive-cache invalidation use case. The stream ships a heartbeat every 25 seconds, +uses a server-supplied 5-second reconnect delay, and each reconnect begins from a full current +snapshot; no event log or Redis is required. + +When the host application applies a generic Rack::Attack throttle to all `/admin` requests, exempt +**authenticated** dashboard reads or put the dashboard behind a separate admin throttle. Do not let +a long-lived stream and its initial list request count as abuse: + +```ruby +# config/initializers/rack_attack.rb +Rack::Attack.safelist("authenticated async-background dashboard") do |request| + request.path.start_with?("/admin/background") && + request.env["warden"]&.user(:admin_user).present? +end +``` + +The dashboard's own `config.auth` still runs for every request; this only prevents a generic rate +limit from treating an authenticated operator's live dashboard as a burst. Adapt the Warden scope to +your application. If the reverse proxy buffers streaming responses, disable buffering for +`/admin/background/api/stream`; the response already includes `X-Accel-Buffering: no` for nginx. + +Use `config.transport = :polling` only for a server that cannot keep an SSE response open. It is a +compatibility fallback, not the recommended production mode. +   ## Step 3 — Docker setup diff --git a/lib/async/background/metrics.rb b/lib/async/background/metrics.rb index 3a1fdfc..91d1064 100644 --- a/lib/async/background/metrics.rb +++ b/lib/async/background/metrics.rb @@ -191,10 +191,12 @@ def validate_worker!(worker_index, total_workers) def ensure_shm!(total_workers, path) required_size = self.class.segment_size * total_workers + page_size = IO::Buffer::PAGE_SIZE + mapped_size = ((required_size + page_size - 1) / page_size) * page_size File.open(path, File::CREAT | File::RDWR, 0o644) do |file| file.flock(File::LOCK_EX) - file.truncate(required_size) if file.size < required_size + file.truncate(mapped_size) if file.size < mapped_size ensure file.flock(File::LOCK_UN) rescue nil end diff --git a/lib/async/background/queue/schema.rb b/lib/async/background/queue/schema.rb index 3795963..a75d60b 100644 --- a/lib/async/background/queue/schema.rb +++ b/lib/async/background/queue/schema.rb @@ -13,7 +13,12 @@ module Schema VERSION = 1 MIGRATION_BUSY_TIMEOUT_MS = 30_000 CORE_INDEXES = %w[idx_jobs_pending].freeze - DASHBOARD_INDEXES = %w[idx_jobs_done_finished_at idx_jobs_failed_finished_at idx_jobs_running].freeze + DASHBOARD_INDEXES = %w[ + idx_jobs_done_finished_at + idx_jobs_failed_finished_at + idx_jobs_executing_started_at + idx_jobs_claimed_locked_at + ].freeze REQUIRED_INDEXES = CORE_INDEXES module_function diff --git a/lib/async/background/queue/sql.rb b/lib/async/background/queue/sql.rb index ca30e43..e629ae6 100644 --- a/lib/async/background/queue/sql.rb +++ b/lib/async/background/queue/sql.rb @@ -192,13 +192,24 @@ def self.add_column(name, sql_type) WHERE status = 'failed' SQL - CREATE_RUNNING_INDEX = <<~SQL.freeze - CREATE INDEX IF NOT EXISTS idx_jobs_running + CREATE_EXECUTING_INDEX = <<~SQL.freeze + CREATE INDEX IF NOT EXISTS idx_jobs_executing_started_at + ON jobs(started_at) + WHERE status = 'running' AND started_at IS NOT NULL + SQL + + CREATE_CLAIMED_INDEX = <<~SQL.freeze + CREATE INDEX IF NOT EXISTS idx_jobs_claimed_locked_at ON jobs(locked_at) - WHERE status = 'running' + WHERE status = 'running' AND started_at IS NULL SQL - CREATE_DASHBOARD_INDEXES = [CREATE_DONE_INDEX, CREATE_FAILED_INDEX, CREATE_RUNNING_INDEX].freeze + CREATE_DASHBOARD_INDEXES = [ + CREATE_DONE_INDEX, + CREATE_FAILED_INDEX, + CREATE_EXECUTING_INDEX, + CREATE_CLAIMED_INDEX + ].freeze end end end diff --git a/lib/async/background/runner.rb b/lib/async/background/runner.rb index f6ad02c..5fbb004 100644 --- a/lib/async/background/runner.rb +++ b/lib/async/background/runner.rb @@ -29,8 +29,11 @@ class Runner :metrics, :queue_store + # `config_path: nil` explicitly disables recurring jobs. This keeps the + # dynamic SQLite queue usable on its own; a supplied path remains strict + # so a typo cannot silently disable scheduled work. def initialize( - config_path:, + config_path: nil, job_count: 2, worker_index:, total_workers:, @@ -53,8 +56,9 @@ def initialize( @drain_barrier = ::Async::Barrier.new @semaphore = ::Async::Semaphore.new(job_count, parent: @drain_barrier) - @heap = build_heap(config_path) + @heap = config_path.nil? ? MinHeap.new : build_heap(config_path) setup_queue(queue_socket_dir, queue_db_path, queue_mmap) + validate_work_source!(config_path) end def run @@ -82,6 +86,11 @@ def running? = @running private def scheduler_loop(task) + # Queue-only workers have no heap entry to sleep on. Keep the runner + # alive until #stop / SIGTERM wakes this condition; the queue listener + # continues independently in its own Async task. + return shutdown.wait if heap.empty? && @listen_queue + loop do entry = heap.peek break unless entry @@ -93,6 +102,12 @@ def scheduler_loop(task) end end + def validate_work_source!(config_path) + return unless config_path.nil? && !@listen_queue + + raise ConfigError, 'Runner requires config_path or queue_socket_dir' + end + def wait_for_next_entry(task, entry) wait = [entry.next_run_at - monotonic_now, MIN_SLEEP_TIME].max wait_with_shutdown(task, wait) diff --git a/lib/async/background/runner/schedule.rb b/lib/async/background/runner/schedule.rb index 6acbc95..e0e1d2d 100644 --- a/lib/async/background/runner/schedule.rb +++ b/lib/async/background/runner/schedule.rb @@ -13,6 +13,8 @@ module Schedule private def build_heap(config_path) + return MinHeap.new if config_path.nil? + schedule = load_schedule(config_path) build_entries(schedule, monotonic_now) end diff --git a/lib/async/background/version.rb b/lib/async/background/version.rb index 233f4ac..1b9ef69 100644 --- a/lib/async/background/version.rb +++ b/lib/async/background/version.rb @@ -2,6 +2,6 @@ module Async module Background - VERSION = '0.7.2' + VERSION = '1.0.0' end end diff --git a/lib/async/background/web.rb b/lib/async/background/web.rb new file mode 100644 index 0000000..0cdba96 --- /dev/null +++ b/lib/async/background/web.rb @@ -0,0 +1,52 @@ +# frozen_string_literal: true + +begin + require 'rack' +rescue LoadError + raise LoadError, + "Async::Background::Web requires 'rack'. " \ + "Add `gem 'rack', '~> 3.0'` to your Gemfile." +end + +require_relative 'web/errors' +require_relative 'web/configuration' +require_relative 'web/sql' +require_relative 'web/cursor' +require_relative 'web/request' +require_relative 'web/response' +require_relative 'web/snapshot' +require_relative 'web/metrics_reader' +require_relative 'web/serializer' +require_relative 'web/auth' +require_relative 'web/router' +require_relative 'web/event_hub' +require_relative 'web/stream' +require_relative 'web/assets' +require_relative 'web/app' + +module Async + module Background + module Web + module_function + + def configure + @configuration ||= Configuration.new + yield @configuration if block_given? + @configuration + end + + def configuration + @configuration or raise NotConfiguredError, + 'Async::Background::Web is not configured. Call Async::Background::Web.configure.' + end + + def reset! + @configuration = nil + end + + def app + App.new(configuration) + end + end + end +end diff --git a/lib/async/background/web/app.rb b/lib/async/background/web/app.rb new file mode 100644 index 0000000..ac92c4d --- /dev/null +++ b/lib/async/background/web/app.rb @@ -0,0 +1,138 @@ +# frozen_string_literal: true + +module Async + module Background + module Web + class App + def initialize(config) + @config = config.validate! + @auth = Auth.new(@config.auth) + @snapshot = Snapshot.new(path: @config.queue_path, counts_cache_ttl: @config.counts_cache_ttl).open! + @metrics_reader = build_metrics_reader + @serializer = Serializer.new(@config) + @event_hub = build_event_hub + @router = Router.new + end + + def call(env) + return Response.unauthorized unless @auth.authorized?(env) + + route = @router.match(env) + return Response.not_found unless route + + dispatch(route, env) + rescue RequestError => error + Response.bad_request(error.message) + rescue UnavailableError, ClosedError + Response.unavailable + rescue StandardError + # Do not turn internal class names, paths or database errors into an + # unauthenticated information disclosure channel. + Response.internal_error + end + + def close + @event_hub&.close + @snapshot.close + self + end + + private + + def build_metrics_reader + return unless @config.metrics_enabled? + + MetricsReader.new(path: @config.metrics_path, total_workers: @config.total_workers) + end + + def build_event_hub + return unless @config.transport == :sse + + EventHub.new( + @snapshot, + @serializer, + metrics_reader: @metrics_reader, + poll_seconds: @config.stream_poll_seconds + ) + end + + def dispatch(route, env) + case route + when :index then Response.html(Assets.render_index(@config)) + when :javascript then Response.javascript(Assets::JS) + when :stylesheet then Response.stylesheet(Assets::CSS) + when :overview then overview_response + when :executing then in_flight_response(:executing, env) + when :claimed then in_flight_response(:claimed, env) + when :done then terminal_response(:done, env) + when :failed then terminal_response(:failed, env) + when :pending then pending_response(env) + when :metrics then metrics_response + when :config then config_response + when :stream then stream_response + else Response.not_found + end + end + + def overview_response + Response.json(@serializer.overview(@snapshot.overview, metrics_payload)) + end + + def in_flight_response(kind, env) + request = Request.new(env, @config) + rows = kind == :executing ? @snapshot.executing(limit: request.limit) : @snapshot.claimed(limit: request.limit) + payload = kind == :executing ? @serializer.executing(rows) : @serializer.claimed(rows) + Response.json({items: payload}) + end + + def terminal_response(kind, env) + request = Request.new(env, @config) + cursor = request.finished_cursor + rows = kind == :done ? @snapshot.recent_done(limit: request.limit, cursor: cursor) : + @snapshot.recent_failed(limit: request.limit, cursor: cursor) + payload = kind == :done ? @serializer.done(rows) : @serializer.failed(rows) + Response.json(payload) + end + + def pending_response(env) + request = Request.new(env, @config) + rows = @snapshot.pending(limit: request.limit, cursor: request.pending_cursor) + Response.json(@serializer.pending(rows)) + end + + def metrics_response + Response.json(metrics_payload || {available: false, workers: [], totals: MetricsReader::EMPTY_TOTALS}) + end + + def metrics_payload + @metrics_reader&.aggregated + end + + def config_response + Response.json( + { + title: @config.title, + poll_interval_ms: @config.poll_interval_ms, + transport: @config.transport.to_s, + expose_args: @config.expose_args, + list_limit: @config.list_limit, + mount_path: @config.mount_path + } + ) + end + + def stream_response + return Response.not_found unless @config.transport == :sse + + Response.sse( + Stream.new( + @event_hub, + heartbeat_seconds: @config.stream_heartbeat_seconds, + retry_ms: @config.stream_retry_ms + ) + ) + end + end + end + end +end diff --git a/lib/async/background/web/assets.rb b/lib/async/background/web/assets.rb new file mode 100644 index 0000000..2bc47da --- /dev/null +++ b/lib/async/background/web/assets.rb @@ -0,0 +1,726 @@ +# frozen_string_literal: true + +require 'cgi' +require 'digest' + +module Async + module Background + module Web + module Assets + CSS = <<~CSS + :root { + --bg: #0f1115; + --panel: #161a20; + --panel-soft: #1c2128; + --border: #2a313b; + --text: #e6e8ec; + --text-dim: #98a2b3; + --accent: #4f8ef7; + --green: #4ade80; + --amber: #fbbf24; + --red: #f87171; + --blue: #60a5fa; + --gray: #94a3b8; + } + * { box-sizing: border-box; } + body { + margin: 0; + font: 14px/1.5 -apple-system, BlinkMacSystemFont, "Segoe UI", Roboto, sans-serif; + background: var(--bg); + color: var(--text); + } + header { + display: flex; + align-items: center; + gap: 16px; + padding: 14px 24px; + border-bottom: 1px solid var(--border); + background: var(--panel); + } + header h1 { font-size: 16px; margin: 0; font-weight: 600; } + header .meta { color: var(--text-dim); font-size: 12px; } + header .status-dot { + width: 8px; height: 8px; border-radius: 50%; + background: var(--gray); + display: inline-block; margin-right: 6px; + } + header .status-dot.ok { background: var(--green); } + header .status-dot.stale { background: var(--amber); } + header .status-dot.error { background: var(--red); } + main { padding: 16px 24px; } + .counts { + display: grid; + grid-template-columns: repeat(auto-fit, minmax(140px, 1fr)); + gap: 12px; + margin-bottom: 20px; + } + .count-card { + background: var(--panel); + border: 1px solid var(--border); + border-radius: 8px; + padding: 12px 14px; + } + .count-card .label { color: var(--text-dim); font-size: 11px; text-transform: uppercase; letter-spacing: .04em; } + .count-card .value { font-size: 24px; font-weight: 600; margin-top: 4px; font-variant-numeric: tabular-nums; } + .count-card.executing .value { color: var(--blue); } + .count-card.claimed .value { color: var(--amber); } + .count-card.pending .value { color: var(--text); } + .count-card.done .value { color: var(--green); } + .count-card.failed .value { color: var(--red); } + .totals { + display: grid; + grid-template-columns: repeat(auto-fit, minmax(140px, 1fr)); + gap: 12px; + margin-bottom: 20px; + } + .total-card { + background: var(--panel-soft); + border: 1px solid var(--border); + border-radius: 8px; + padding: 10px 12px; + } + .total-card .label { color: var(--text-dim); font-size: 11px; } + .total-card .value { font-variant-numeric: tabular-nums; font-size: 18px; margin-top: 2px; } + nav.tabs { + display: flex; + gap: 2px; + border-bottom: 1px solid var(--border); + margin-bottom: 12px; + flex-wrap: wrap; + } + nav.tabs button { + background: transparent; + color: var(--text-dim); + border: none; + border-bottom: 2px solid transparent; + padding: 10px 14px; + font: inherit; + cursor: pointer; + border-radius: 0; + } + nav.tabs button:hover { color: var(--text); } + nav.tabs button.active { + color: var(--text); + border-bottom-color: var(--accent); + } + nav.tabs button .badge { + display: inline-block; + background: var(--panel-soft); + color: var(--text-dim); + border-radius: 10px; + padding: 1px 8px; + font-size: 11px; + margin-left: 6px; + font-variant-numeric: tabular-nums; + } + table { + width: 100%; + border-collapse: collapse; + background: var(--panel); + border: 1px solid var(--border); + border-radius: 8px; + overflow: hidden; + } + th, td { + text-align: left; + padding: 9px 12px; + border-bottom: 1px solid var(--border); + vertical-align: top; + font-variant-numeric: tabular-nums; + } + tbody tr:last-child td { border-bottom: none; } + th { + background: var(--panel-soft); + color: var(--text-dim); + font-weight: 500; + font-size: 11px; + text-transform: uppercase; + letter-spacing: .04em; + } + td.dim { color: var(--text-dim); } + td.mono { font-family: ui-monospace, SFMono-Regular, Menlo, monospace; font-size: 12px; } + .empty { + padding: 32px; + text-align: center; + color: var(--text-dim); + background: var(--panel); + border: 1px solid var(--border); + border-radius: 8px; + } + .pagination { + display: flex; + gap: 8px; + margin-top: 12px; + } + button.btn { + background: var(--panel-soft); + color: var(--text); + border: 1px solid var(--border); + border-radius: 6px; + padding: 6px 12px; + font: inherit; + cursor: pointer; + } + button.btn:hover { border-color: var(--accent); } + button.btn:disabled { opacity: .4; cursor: not-allowed; } + .err-msg { + color: var(--red); + font-family: ui-monospace, SFMono-Regular, Menlo, monospace; + font-size: 12px; + max-width: 480px; + white-space: pre-wrap; + word-break: break-word; + } + .err-class { color: var(--amber); font-weight: 600; } + .args-cell pre { + margin: 0; + font-family: ui-monospace, SFMono-Regular, Menlo, monospace; + font-size: 12px; + color: var(--text-dim); + white-space: pre-wrap; + word-break: break-all; + max-width: 380px; + } + .args-redacted { color: var(--text-dim); font-style: italic; } + CSS + + JS = <<~'JS' + (function () { + const state = { + config: null, + activeTab: 'executing', + data: { executing: [], claimed: [], pending: [], done: [], failed: [] }, + cursors: { pending: null, done: null, failed: null }, + counts: { executing: 0, claimed: 0, pending: 0, done: 0, failed: 0 }, + totals: null, + overview: null, + dataVersion: null, + lastUpdate: null, + connection: 'connecting', + stream: null, + pollingTimer: null, + pollingInFlight: false, + listAbort: null, + listRequestId: 0, + listRefreshTimer: null, + listRefreshQueued: false, + listError: null + }; + + function initialBasePath() { + if (document.body && document.body.dataset.mountPath !== undefined) { + return document.body.dataset.mountPath; + } + + if (document.currentScript && document.currentScript.src) { + const scriptUrl = new URL(document.currentScript.src, location.origin); + return scriptUrl.pathname.replace(/\/assets\/app\.js$/, ''); + } + + return location.pathname.replace(/\/$/, ''); + } + + const bootBasePath = initialBasePath(); + + function basePath() { + return state.config && state.config.mount_path !== undefined ? + state.config.mount_path : bootBasePath; + } + + class HttpError extends Error { + constructor(response) { + super('http ' + response.status); + this.name = 'HttpError'; + this.status = response.status; + } + } + + async function api(path, params, signal) { + const url = new URL(basePath() + path, location.origin); + if (params) { + Object.keys(params).forEach((key) => { + if (params[key] !== null && params[key] !== undefined) { + url.searchParams.set(key, params[key]); + } + }); + } + + const response = await fetch(url.toString(), { + credentials: 'same-origin', + headers: { accept: 'application/json' }, + signal: signal + }); + + if (!response.ok) throw new HttpError(response); + return response.json(); + } + + async function loadConfig() { + state.config = await api('/api/config'); + const title = document.getElementById('title'); + if (title) title.textContent = state.config.title; + document.title = state.config.title + ' dashboard'; + } + + async function refreshOverview() { + try { + applyOverview(await api('/api/overview')); + } catch (error) { + setConnection('error'); + return null; + } + } + + async function refreshActiveList(reset) { + if (!state.config) return; + + const tab = state.activeTab; + const requestId = ++state.listRequestId; + const cursor = !reset ? state.cursors[tab] : null; + const controller = new AbortController(); + + if (state.listAbort) state.listAbort.abort(); + state.listAbort = controller; + + try { + const params = { limit: state.config.list_limit }; + if (cursor) params.cursor = cursor; + const payload = await api('/api/' + tab, params, controller.signal); + + if (requestId !== state.listRequestId || tab !== state.activeTab) return; + + const items = Array.isArray(payload.items) ? payload.items : Array.isArray(payload) ? payload : []; + const nextCursor = Array.isArray(payload.items) ? payload.next_cursor || null : null; + state.data[tab] = cursor ? state.data[tab].concat(items) : items; + state.cursors[tab] = nextCursor; + state.listError = null; + renderList(); + } catch (error) { + if (error.name === 'AbortError') return; + if (requestId !== state.listRequestId || tab !== state.activeTab) return; + + state.listError = error; + setConnection('error'); + renderList(); + } finally { + if (requestId === state.listRequestId) state.listAbort = null; + } + } + + // Coalesce a burst of queue changes into at most one list request at a + // time. The data_version snapshot is authoritative, so skipped + // intermediate renders do not lose state. + function scheduleActiveListRefresh() { + state.listRefreshQueued = true; + if (state.listRefreshTimer) return; + + state.listRefreshTimer = setTimeout(async function () { + state.listRefreshTimer = null; + while (state.listRefreshQueued) { + state.listRefreshQueued = false; + await refreshActiveList(true); + } + }, 100); + } + + function applyOverview(overview) { + state.overview = overview; + state.counts = overview.counts || state.counts; + state.dataVersion = overview.data_version; + state.totals = overview.metrics || null; + state.lastUpdate = Date.now(); + setConnection('ok', false); + renderCounts(); + renderTotals(); + renderTabBadges(); + renderHeader(overview); + } + + function setConnection(connection, render) { + state.connection = connection; + if (render !== false) renderHeader(state.overview); + } + + function renderHeader(overview) { + const dot = document.getElementById('status-dot'); + const meta = document.getElementById('meta'); + if (!dot || !meta) return; + + const stateClass = state.connection === 'ok' ? 'ok' : state.connection === 'stale' ? 'stale' : 'error'; + dot.className = 'status-dot ' + stateClass; + + const parts = []; + if (state.connection === 'stale') parts.push('reconnecting'); + if (state.connection === 'error') parts.push('connection error'); + if (state.lastUpdate) parts.push('updated ' + relTime(state.lastUpdate) + ' ago'); + if (state.dataVersion !== null && state.dataVersion !== undefined) parts.push('data_version ' + state.dataVersion); + if (overview && overview.next_pending_run_at) { + parts.push('next pending in ' + formatDuration(overview.next_pending_run_at - (Date.now() / 1000))); + } + meta.textContent = parts.join(' · '); + } + + function renderCounts() { + const root = document.getElementById('counts'); + if (!root) return; + root.replaceChildren(); + + [ + ['executing', 'Executing'], + ['claimed', 'Claimed'], + ['pending', 'Pending'], + ['done', 'Done'], + ['failed', 'Failed'] + ].forEach(([key, label]) => { + const card = document.createElement('div'); + card.className = 'count-card ' + key; + const labelElement = document.createElement('div'); + labelElement.className = 'label'; + labelElement.textContent = label; + const value = document.createElement('div'); + value.className = 'value'; + value.textContent = (state.counts[key] || 0).toLocaleString(); + card.append(labelElement, value); + root.append(card); + }); + } + + function renderTotals() { + const root = document.getElementById('totals'); + if (!root) return; + root.replaceChildren(); + if (!state.totals || !state.totals.totals) { + root.style.display = 'none'; + return; + } + + root.style.display = ''; + const totals = state.totals.totals; + [ + ['total_runs', 'Runs'], + ['total_successes', 'Successes'], + ['total_failures', 'Failures'], + ['total_timeouts', 'Timeouts'], + ['total_skips', 'Skipped'], + ['active_jobs', 'Active workers'], + ['last_duration_ms', 'Last duration (ms)'] + ].forEach(([key, label]) => { + const card = document.createElement('div'); + card.className = 'total-card'; + const labelElement = document.createElement('div'); + labelElement.className = 'label'; + labelElement.textContent = label; + const value = document.createElement('div'); + value.className = 'value'; + value.textContent = totals[key] !== null && totals[key] !== undefined ? Number(totals[key]).toLocaleString() : '-'; + card.append(labelElement, value); + root.append(card); + }); + } + + function renderTabBadges() { + ['executing', 'claimed', 'pending', 'done', 'failed'].forEach((key) => { + const badge = document.querySelector('button[data-tab="' + key + '"] .badge'); + if (badge) badge.textContent = (state.counts[key] || 0).toLocaleString(); + }); + } + + function renderList() { + const root = document.getElementById('list'); + const pagination = document.getElementById('pagination'); + if (!root) return; + root.replaceChildren(); + if (pagination) pagination.replaceChildren(); + + if (state.listError) { + const error = document.createElement('div'); + error.className = 'empty'; + error.textContent = 'Unable to load jobs (' + state.listError.message + ')'; + root.append(error); + return; + } + + const items = state.data[state.activeTab] || []; + if (items.length === 0) { + const empty = document.createElement('div'); + empty.className = 'empty'; + empty.textContent = 'No jobs in this list'; + root.append(empty); + return; + } + + root.append(buildTable(state.activeTab, items)); + renderPagination(); + } + + function renderPagination() { + const root = document.getElementById('pagination'); + if (!root || !['pending', 'done', 'failed'].includes(state.activeTab)) return; + if (!state.cursors[state.activeTab]) return; + + const button = document.createElement('button'); + button.className = 'btn'; + button.type = 'button'; + button.textContent = 'Load more'; + button.addEventListener('click', () => refreshActiveList(false)); + root.append(button); + } + + function buildTable(tab, items) { + const table = document.createElement('table'); + const columns = tableColumns(tab); + const head = document.createElement('thead'); + const headRow = document.createElement('tr'); + columns.forEach((column) => { + const cell = document.createElement('th'); + cell.textContent = column.label; + headRow.append(cell); + }); + head.append(headRow); + table.append(head); + + const body = document.createElement('tbody'); + items.forEach((item) => { + const row = document.createElement('tr'); + columns.forEach((column) => { + const cell = document.createElement('td'); + column.render(cell, item); + row.append(cell); + }); + body.append(row); + }); + table.append(body); + return table; + } + + function tableColumns(tab) { + const id = { label: 'ID', render: (cell, item) => { cell.className = 'mono dim'; cell.textContent = item.id; } }; + const klass = { label: 'Class', render: (cell, item) => { cell.className = 'mono'; cell.textContent = item.class_name; } }; + const args = { + label: 'Args', + render: (cell, item) => { + cell.className = 'args-cell'; + if (state.config && state.config.expose_args) { + if (item.args === null || item.args === undefined) { + cell.textContent = '(' + (item.args_count || 0) + ')'; + } else { + const pre = document.createElement('pre'); + pre.textContent = JSON.stringify(item.args); + cell.append(pre); + } + } else { + const hidden = document.createElement('span'); + hidden.className = 'args-redacted'; + hidden.textContent = (item.args_count || 0) + ' args (hidden)'; + cell.append(hidden); + } + } + }; + const time = (key, label) => ({ + label: label, + render: (cell, item) => { cell.className = 'dim mono'; cell.textContent = formatTime(item[key]); } + }); + const duration = { + label: 'Duration', + render: (cell, item) => { cell.className = 'mono dim'; cell.textContent = item.duration_ms ? item.duration_ms + ' ms' : '-'; } + }; + const error = { + label: 'Error', + render: (cell, item) => { + cell.className = 'err-msg'; + if (!item.last_error_class) { + cell.textContent = '-'; + return; + } + const errorClass = document.createElement('span'); + errorClass.className = 'err-class'; + errorClass.textContent = item.last_error_class; + cell.append(errorClass, document.createTextNode(' ' + (item.last_error_message || ''))); + } + }; + + if (tab === 'executing') return [id, klass, args, time('started_at', 'Started'), { label: 'Worker', render: (cell, item) => { cell.className = 'mono dim'; cell.textContent = item.locked_by; } }]; + if (tab === 'claimed') return [id, klass, args, time('locked_at', 'Claimed'), { label: 'Worker', render: (cell, item) => { cell.className = 'mono dim'; cell.textContent = item.locked_by; } }]; + if (tab === 'done') return [id, klass, args, time('finished_at', 'Finished'), duration]; + if (tab === 'failed') return [id, klass, args, time('finished_at', 'Finished'), duration, error]; + return [id, klass, args, time('run_at', 'Run at')]; + } + + function setActiveTab(tab) { + if (tab === state.activeTab) return; + state.activeTab = tab; + state.cursors[tab] = null; + state.listError = null; + document.querySelectorAll('nav.tabs button').forEach((button) => { + button.classList.toggle('active', button.dataset.tab === tab); + }); + refreshActiveList(true); + } + + function attachTabs() { + document.querySelectorAll('nav.tabs button').forEach((button) => { + button.addEventListener('click', () => setActiveTab(button.dataset.tab)); + }); + } + + function streamUrl() { + return new URL(basePath() + '/api/stream', location.origin).toString(); + } + + function stopStream() { + if (state.stream) state.stream.close(); + state.stream = null; + } + + function startStream() { + stopStream(); + setConnection('stale'); + + const stream = new EventSource(streamUrl()); + state.stream = stream; + + stream.addEventListener('overview', (event) => { + if (state.stream !== stream) return; + try { + applyOverview(JSON.parse(event.data)); + scheduleActiveListRefresh(); + } catch (_) { + setConnection('error'); + } + }); + + stream.addEventListener('unavailable', () => { + if (state.stream === stream) setConnection('stale'); + }); + + stream.addEventListener('open', () => { + if (state.stream === stream) setConnection('ok'); + }); + + stream.addEventListener('error', () => { + if (state.stream !== stream) return; + setConnection(stream.readyState === EventSource.CLOSED ? 'error' : 'stale'); + }); + } + + async function pollingTick() { + if (state.pollingInFlight) return; + state.pollingInFlight = true; + try { + await refreshOverview(); + await refreshActiveList(true); + } finally { + state.pollingInFlight = false; + } + } + + function startPolling() { + pollingTick(); + state.pollingTimer = setInterval(pollingTick, state.config.poll_interval_ms); + } + + function formatTime(seconds) { + if (!seconds) return '-'; + const date = new Date(seconds * 1000); + return isNaN(date.getTime()) ? '-' : date.toLocaleString(); + } + + function formatDuration(seconds) { + if (!isFinite(seconds)) return '-'; + if (seconds <= 0) return 'now'; + if (seconds < 60) return Math.round(seconds) + 's'; + if (seconds < 3600) return Math.round(seconds / 60) + 'm'; + return Math.round(seconds / 3600) + 'h'; + } + + function relTime(timestamp) { + return formatDuration((Date.now() - timestamp) / 1000); + } + + async function boot() { + attachTabs(); + await loadConfig(); + await Promise.all([refreshOverview(), refreshActiveList(true)]); + + if (state.config.transport === 'sse' && typeof EventSource !== 'undefined') { + startStream(); + } else { + startPolling(); + } + + setInterval(() => renderHeader(state.overview), 1000); + } + + function start() { + boot().catch((error) => { + setConnection('error'); + if (window.console && console.error) console.error('Async::Background dashboard boot failed', error); + }); + } + + window.addEventListener('beforeunload', () => { + stopStream(); + if (state.pollingTimer) clearInterval(state.pollingTimer); + if (state.listRefreshTimer) clearTimeout(state.listRefreshTimer); + if (state.listAbort) state.listAbort.abort(); + }, { once: true }); + + if (document.readyState === 'loading') { + document.addEventListener('DOMContentLoaded', start, { once: true }); + } else { + start(); + } + })(); + JS + + INDEX_HTML = <<~HTML + + + + + + %<title>s + + + +
+

%s</h1> + <span class="meta"><span id="status-dot" class="status-dot"></span><span id="meta"></span></span> + </header> + <main> + <section id="counts" class="counts"></section> + <section id="totals" class="totals"></section> + <nav class="tabs"> + <button type="button" data-tab="executing" class="active">Executing <span class="badge">0</span></button> + <button type="button" data-tab="claimed">Claimed <span class="badge">0</span></button> + <button type="button" data-tab="pending">Pending <span class="badge">0</span></button> + <button type="button" data-tab="done">Done <span class="badge">0</span></button> + <button type="button" data-tab="failed">Failed <span class="badge">0</span></button> + </nav> + <section id="list"></section> + <section id="pagination" class="pagination"></section> + </main> + <script defer src="%<base>s/assets/app.js?v=%<asset_version>s"></script> + </body> + </html> + HTML + + module_function + + def asset_version + @asset_version ||= Digest::SHA256.hexdigest("#{JS}\0#{CSS}")[0, 12] + end + + def render_index(config) + base = config.mount_path.to_s.sub(%r{/\z}, '') + format( + INDEX_HTML, + title: CGI.escapeHTML(config.title.to_s), + base: CGI.escapeHTML(base), + asset_version: asset_version + ) + end + end + end + end +end diff --git a/lib/async/background/web/auth.rb b/lib/async/background/web/auth.rb new file mode 100644 index 0000000..a3bf31f --- /dev/null +++ b/lib/async/background/web/auth.rb @@ -0,0 +1,19 @@ +# frozen_string_literal: true + +module Async + module Background + module Web + class Auth + def initialize(callable) + @callable = callable + end + + def authorized?(env) + !!@callable.call(env) + rescue StandardError + false + end + end + end + end +end diff --git a/lib/async/background/web/configuration.rb b/lib/async/background/web/configuration.rb new file mode 100644 index 0000000..aba38a5 --- /dev/null +++ b/lib/async/background/web/configuration.rb @@ -0,0 +1,158 @@ +# frozen_string_literal: true + +require_relative '../queue/store' + +module Async + module Background + module Web + class Configuration + DEFAULT_LIST_LIMIT = 50 + MAX_LIST_LIMIT = 200 + DEFAULT_COUNTS_TTL = 3.0 + DEFAULT_POLL_INTERVAL_MS = 2000 + DEFAULT_STREAM_POLL_SECONDS = 0.5 + DEFAULT_STREAM_HEARTBEAT_SECONDS = 25.0 + DEFAULT_STREAM_RETRY_MS = 5000 + TRANSPORTS = %i[polling sse].freeze + DEFAULT_TRANSPORT = :sse + DEFAULT_REDACT = ->(args) { args.is_a?(Array) ? args.map { '***' } : args } + + attr_accessor :queue_path, + :auth, + :expose_args, + :redact_args, + :metrics_path, + :total_workers, + :counts_cache_ttl, + :list_limit, + :poll_interval_ms, + :transport, + :stream_poll_seconds, + :stream_heartbeat_seconds, + :stream_retry_ms, + :title, + :mount_path + + def initialize + @queue_path = Queue::Store.default_path + @auth = nil + @expose_args = false + @redact_args = DEFAULT_REDACT + @metrics_path = nil + @total_workers = nil + @counts_cache_ttl = DEFAULT_COUNTS_TTL + @list_limit = DEFAULT_LIST_LIMIT + @poll_interval_ms = DEFAULT_POLL_INTERVAL_MS + @transport = DEFAULT_TRANSPORT + @stream_poll_seconds = DEFAULT_STREAM_POLL_SECONDS + @stream_heartbeat_seconds = DEFAULT_STREAM_HEARTBEAT_SECONDS + @stream_retry_ms = DEFAULT_STREAM_RETRY_MS + @title = 'Async::Background' + @mount_path = '' + end + + def validate! + validate_queue_path! + validate_auth! + validate_list_limit! + validate_cache_ttl! + validate_poll_interval! + validate_transport! + validate_stream! + validate_redactor! + validate_metrics! + validate_mount_path! + self + end + + # Strict request-path parsing. Silently changing a malformed requested + # page size to the default makes API clients repeat or skip work. + def limit_for(requested) + return list_limit if requested.nil? || requested.empty? + + value = Integer(requested, 10) + raise RequestError, 'limit must be a positive integer' unless value.positive? + + [value, MAX_LIST_LIMIT].min + rescue ArgumentError, TypeError + raise RequestError, 'limit must be a positive integer' + end + + def metrics_enabled? + !metrics_path.nil? + end + + private + + def validate_queue_path! + raise ConfigurationError, 'queue_path must be set' if queue_path.nil? || queue_path.to_s.empty? + end + + def validate_auth! + raise ConfigurationError, 'auth must be configured (gem ships no permissive default)' if auth.nil? + + return if auth.respond_to?(:call) + + raise ConfigurationError, 'auth must respond to #call(env) and return truthy on success' + end + + def validate_list_limit! + return if list_limit.is_a?(Integer) && list_limit.between?(1, MAX_LIST_LIMIT) + + raise ConfigurationError, "list_limit must be an Integer between 1 and #{MAX_LIST_LIMIT}" + end + + def validate_cache_ttl! + return if counts_cache_ttl.is_a?(Numeric) && counts_cache_ttl >= 0 + + raise ConfigurationError, 'counts_cache_ttl must be a non-negative Numeric' + end + + def validate_poll_interval! + return if poll_interval_ms.is_a?(Integer) && poll_interval_ms >= 200 + + raise ConfigurationError, 'poll_interval_ms must be an Integer >= 200' + end + + def validate_transport! + return if TRANSPORTS.include?(transport) + + raise ConfigurationError, "transport must be one of #{TRANSPORTS.inspect}" + end + + def validate_stream! + unless stream_poll_seconds.is_a?(Numeric) && stream_poll_seconds >= 0.1 + raise ConfigurationError, 'stream_poll_seconds must be a Numeric >= 0.1' + end + + unless stream_heartbeat_seconds.is_a?(Numeric) && stream_heartbeat_seconds >= 5 + raise ConfigurationError, 'stream_heartbeat_seconds must be a Numeric >= 5' + end + + return if stream_retry_ms.is_a?(Integer) && stream_retry_ms >= 500 + + raise ConfigurationError, 'stream_retry_ms must be an Integer >= 500' + end + + def validate_redactor! + return unless expose_args && redact_args && !redact_args.respond_to?(:call) + + raise ConfigurationError, 'redact_args must respond to #call(args)' + end + + def validate_metrics! + return unless metrics_enabled? + return if total_workers.is_a?(Integer) && total_workers.positive? + + raise ConfigurationError, 'metrics_path requires total_workers to be a positive Integer' + end + + def validate_mount_path! + return if mount_path.is_a?(String) + + raise ConfigurationError, 'mount_path must be a String' + end + end + end + end +end diff --git a/lib/async/background/web/cursor.rb b/lib/async/background/web/cursor.rb new file mode 100644 index 0000000..deda526 --- /dev/null +++ b/lib/async/background/web/cursor.rb @@ -0,0 +1,58 @@ +# frozen_string_literal: true + +require 'base64' + +module Async + module Background + module Web + module Cursor + module_function + + def encode_finished(finished_at, id) + encode(finished_at, id) + end + + def encode_pending(run_at, id) + encode(run_at, id) + end + + def decode_finished(value) + timestamp, id = decode(value) + return unless timestamp + + {finished_at: timestamp, id: id} + end + + def decode_pending(value) + timestamp, id = decode(value) + return unless timestamp + + {run_at: timestamp, id: id} + end + + def encode(timestamp, id) + return if timestamp.nil? || id.nil? + + Base64.urlsafe_encode64("#{Float(timestamp)}:#{Integer(id)}", padding: false) + end + private_class_method :encode + + def decode(value) + return if value.nil? || value.to_s.empty? + + timestamp_raw, id_raw, extra = Base64.urlsafe_decode64(value.to_s).split(':', 3) + raise RequestError, 'invalid cursor' if timestamp_raw.nil? || id_raw.nil? || extra + + timestamp = Float(timestamp_raw) + id = Integer(id_raw) + raise RequestError, 'invalid cursor' unless timestamp.finite? && id.positive? + + [timestamp, id] + rescue ArgumentError, TypeError + raise RequestError, 'invalid cursor' + end + private_class_method :decode + end + end + end +end diff --git a/lib/async/background/web/errors.rb b/lib/async/background/web/errors.rb new file mode 100644 index 0000000..a691ec7 --- /dev/null +++ b/lib/async/background/web/errors.rb @@ -0,0 +1,14 @@ +# frozen_string_literal: true + +module Async + module Background + module Web + class Error < StandardError; end + class ConfigurationError < Error; end + class NotConfiguredError < Error; end + class RequestError < Error; end + class UnavailableError < Error; end + class ClosedError < Error; end + end + end +end diff --git a/lib/async/background/web/event_hub.rb b/lib/async/background/web/event_hub.rb new file mode 100644 index 0000000..4d389ab --- /dev/null +++ b/lib/async/background/web/event_hub.rb @@ -0,0 +1,194 @@ +# frozen_string_literal: true + +require 'json' + +module Async + module Background + module Web + class EventHub + HEARTBEAT_FRAME = ":keepalive\n\n" + UNAVAILABLE_FRAME = "event: unavailable\ndata: #{JSON.generate(error: 'unavailable')}\n\n".freeze + + class Subscription + def initialize(clock: nil) + @clock = clock || -> { Process.clock_gettime(Process::CLOCK_MONOTONIC) } + @mutex = Mutex.new + @condition = ConditionVariable.new + @frame = nil + @closed = false + end + + def publish(frame) + @mutex.synchronize do + return false if @closed + + @frame = frame + @condition.signal + true + end + end + + def pop(timeout:) + deadline = @clock.call + timeout + + @mutex.synchronize do + while @frame.nil? && !@closed + remaining = deadline - @clock.call + break if remaining <= 0 + + @condition.wait(@mutex, remaining) + end + + frame = @frame + @frame = nil + frame + end + end + + def close + @mutex.synchronize do + return if @closed + + @closed = true + @condition.broadcast + end + end + + def closed? + @mutex.synchronize { @closed } + end + end + + def initialize(snapshot, serializer, metrics_reader: nil, poll_seconds:, sleeper: nil) + @snapshot = snapshot + @serializer = serializer + @metrics_reader = metrics_reader + @poll_seconds = poll_seconds + @sleeper = sleeper || ->(seconds) { sleep(seconds) } + @mutex = Mutex.new + @condition = ConditionVariable.new + @subscribers = {} + @closed = false + @monitor = nil + @last_data_version = nil + @unavailable = false + end + + def subscribe + subscription = Subscription.new + frame, data_version = current_overview + + @mutex.synchronize do + raise ClosedError, 'event hub is closed' if @closed + + @subscribers[subscription.object_id] = subscription + @last_data_version ||= data_version + start_monitor_unless_running! + @condition.signal + end + + [subscription, frame] + end + + def unsubscribe(subscription) + @mutex.synchronize do + @subscribers.delete(subscription.object_id) + end + subscription.close + nil + end + + def close + monitor = nil + subscribers = nil + + @mutex.synchronize do + return if @closed + + @closed = true + subscribers = @subscribers.values + @subscribers.clear + monitor = @monitor + @condition.broadcast + end + + subscribers.each(&:close) + monitor&.join(1) unless monitor == Thread.current + nil + end + + private + + def start_monitor_unless_running! + return if @monitor&.alive? + + @monitor = Thread.new { monitor_loop } + @monitor.name = 'async-background-web-events' if @monitor.respond_to?(:name=) + @monitor.abort_on_exception = false + end + + def monitor_loop + loop do + break unless wait_for_subscribers + + begin + detect_change + rescue ClosedError, UnavailableError + notify_unavailable + end + + @sleeper.call(@poll_seconds) + end + ensure + @mutex.synchronize { @monitor = nil if @monitor == Thread.current } + end + + def wait_for_subscribers + @mutex.synchronize do + @condition.wait(@mutex) while !@closed && @subscribers.empty? + !@closed + end + end + + def detect_change + version = @snapshot.data_version + previous_version = @mutex.synchronize { @last_data_version } + if version == previous_version + @mutex.synchronize { @unavailable = false } + return + end + + frame, observed_version = current_overview + @mutex.synchronize do + @last_data_version = observed_version + @unavailable = false + end + broadcast(frame) + end + + def current_overview + overview = @snapshot.overview(force: true) + metrics = @metrics_reader&.aggregated + payload = @serializer.overview(overview, metrics) + ["event: overview\ndata: #{JSON.generate(payload)}\n\n", payload.fetch(:data_version)] + end + + def notify_unavailable + should_broadcast = @mutex.synchronize do + next false if @unavailable + + @unavailable = true + true + end + broadcast(UNAVAILABLE_FRAME) if should_broadcast + end + + def broadcast(frame) + subscribers = @mutex.synchronize { @subscribers.values.dup } + subscribers.each { |subscription| subscription.publish(frame) } + nil + end + end + end + end +end diff --git a/lib/async/background/web/metrics_reader.rb b/lib/async/background/web/metrics_reader.rb new file mode 100644 index 0000000..3c2c35c --- /dev/null +++ b/lib/async/background/web/metrics_reader.rb @@ -0,0 +1,96 @@ +# frozen_string_literal: true + +require_relative '../clock' +require_relative '../metrics' + +module Async + module Background + module Web + class MetricsReader + include Clock + + DEFAULT_TTL = 1.0 + EMPTY_WORKERS = [].freeze + EMPTY_TOTALS = { + total_runs: 0, + total_successes: 0, + total_failures: 0, + total_timeouts: 0, + total_skips: 0, + active_jobs: 0, + last_run_at: 0, + last_duration_ms: nil + }.freeze + + def initialize(path:, total_workers:, ttl: DEFAULT_TTL) + @path = path + @total_workers = total_workers + @ttl = ttl + @mutex = Mutex.new + @cache = nil + @cached_at = nil + end + + def aggregated + @mutex.synchronize do + now = monotonic_now + return @cache if cache_current?(now) + + @cache = read_metrics.freeze + @cached_at = now + @cache + end + end + + private + + def cache_current?(now) + @cache && @cached_at && (now - @cached_at) < @ttl + end + + def read_metrics + return unavailable unless Metrics.available? && File.file?(@path) + + workers = Metrics.read_all(total_workers: @total_workers, path: @path) + {available: true, workers: workers, totals: aggregate(workers)} + rescue StandardError + unavailable + end + + def unavailable + {available: false, workers: EMPTY_WORKERS, totals: EMPTY_TOTALS} + end + + def aggregate(workers) + totals = { + total_runs: 0, + total_successes: 0, + total_failures: 0, + total_timeouts: 0, + total_skips: 0, + active_jobs: 0, + last_run_at: 0, + last_duration_ms: nil + } + + workers.each do |worker| + totals[:total_runs] += worker[:total_runs].to_i + totals[:total_successes] += worker[:total_successes].to_i + totals[:total_failures] += worker[:total_failures].to_i + totals[:total_timeouts] += worker[:total_timeouts].to_i + totals[:total_skips] += worker[:total_skips].to_i + totals[:active_jobs] += worker[:active_jobs].to_i + + last_run_at = worker[:last_run_at].to_i + next unless last_run_at > totals[:last_run_at] + + totals[:last_run_at] = last_run_at + totals[:last_duration_ms] = worker[:last_duration_ms] + end + + totals.freeze + end + end + end + end +end diff --git a/lib/async/background/web/request.rb b/lib/async/background/web/request.rb new file mode 100644 index 0000000..c5b74f6 --- /dev/null +++ b/lib/async/background/web/request.rb @@ -0,0 +1,36 @@ +# frozen_string_literal: true + +module Async + module Background + module Web + class Request + def initialize(env, config) + @config = config + @params = parse(env['QUERY_STRING']) + end + + def limit + @config.limit_for(@params['limit']) + end + + def finished_cursor + Cursor.decode_finished(@params['cursor']) + end + + def pending_cursor + Cursor.decode_pending(@params['cursor']) + end + + private + + def parse(query) + return {} if query.nil? || query.empty? + + Rack::Utils.parse_query(query) + rescue StandardError + raise RequestError, 'invalid query string' + end + end + end + end +end diff --git a/lib/async/background/web/response.rb b/lib/async/background/web/response.rb new file mode 100644 index 0000000..0cf793a --- /dev/null +++ b/lib/async/background/web/response.rb @@ -0,0 +1,85 @@ +# frozen_string_literal: true + +require 'json' + +module Async + module Background + module Web + module Response + module_function + + JSON_TYPE = 'application/json; charset=utf-8' + HTML_TYPE = 'text/html; charset=utf-8' + TEXT_TYPE = 'text/plain; charset=utf-8' + JAVASCRIPT_TYPE = 'application/javascript; charset=utf-8' + CSS_TYPE = 'text/css; charset=utf-8' + NO_STORE = 'no-store' + ASSET_CACHE = 'public, max-age=31536000, immutable' + + UNAUTHORIZED_BODY = 'unauthorized' + NOT_FOUND_BODY = 'not found' + BAD_REQUEST_BODY = JSON.generate(error: 'invalid_request').freeze + UNAVAILABLE_BODY = JSON.generate(error: 'service_unavailable').freeze + INTERNAL_ERROR_BODY = JSON.generate(error: 'internal_error').freeze + EVENT_STREAM_TYPE = 'text/event-stream; charset=utf-8' + + def sse(body) + [200, sse_headers, body] + end + + def json(payload, status: 200) + [status, no_store_headers(JSON_TYPE), [JSON.generate(payload)]] + end + + def html(body) + [200, no_store_headers(HTML_TYPE), [body]] + end + + def javascript(body) + [200, asset_headers(JAVASCRIPT_TYPE), [body]] + end + + def stylesheet(body) + [200, asset_headers(CSS_TYPE), [body]] + end + + def unauthorized + [401, no_store_headers(TEXT_TYPE), [UNAUTHORIZED_BODY]] + end + + def not_found + [404, no_store_headers(TEXT_TYPE), [NOT_FOUND_BODY]] + end + + def bad_request(message = nil) + body = message.nil? ? BAD_REQUEST_BODY : JSON.generate(error: 'invalid_request', message: message) + [400, no_store_headers(JSON_TYPE), [body]] + end + + def unavailable + [503, no_store_headers(JSON_TYPE), [UNAVAILABLE_BODY]] + end + + def internal_error + [500, no_store_headers(JSON_TYPE), [INTERNAL_ERROR_BODY]] + end + + def no_store_headers(content_type) + {'content-type' => content_type, 'cache-control' => NO_STORE} + end + + def asset_headers(content_type) + {'content-type' => content_type, 'cache-control' => ASSET_CACHE} + end + + def sse_headers + { + 'content-type' => EVENT_STREAM_TYPE, + 'cache-control' => 'no-cache, no-transform', + 'x-accel-buffering' => 'no' + } + end + end + end + end +end diff --git a/lib/async/background/web/router.rb b/lib/async/background/web/router.rb new file mode 100644 index 0000000..fd2ee64 --- /dev/null +++ b/lib/async/background/web/router.rb @@ -0,0 +1,30 @@ +# frozen_string_literal: true + +module Async + module Background + module Web + class Router + GET_ROUTES = { + '/' => :index, + '/assets/app.js' => :javascript, + '/assets/app.css' => :stylesheet, + '/api/overview' => :overview, + '/api/executing' => :executing, + '/api/claimed' => :claimed, + '/api/done' => :done, + '/api/failed' => :failed, + '/api/pending' => :pending, + '/api/metrics' => :metrics, + '/api/config' => :config, + '/api/stream' => :stream + }.freeze + + def match(env) + return unless env['REQUEST_METHOD'] == 'GET' + + GET_ROUTES[env['PATH_INFO'] || '/'] + end + end + end + end +end diff --git a/lib/async/background/web/serializer.rb b/lib/async/background/web/serializer.rb new file mode 100644 index 0000000..af1b61c --- /dev/null +++ b/lib/async/background/web/serializer.rb @@ -0,0 +1,154 @@ +# frozen_string_literal: true + +require 'json' +require_relative 'cursor' + +module Async + module Background + module Web + class Serializer + EMPTY_OPTIONS = {}.freeze + EMPTY_ARGS = [].freeze + + def initialize(config) + @config = config + end + + def overview(snapshot_data, metrics_data = nil) + payload = { + counts: snapshot_data.fetch(:counts), + next_pending_run_at: snapshot_data[:next_pending_run_at], + data_version: snapshot_data.fetch(:data_version), + generated_at: snapshot_data.fetch(:generated_at) + } + payload[:metrics] = metrics_data if metrics_data + payload + end + + def executing(rows) + rows.map { |row| executing_item(row) } + end + + def claimed(rows) + rows.map { |row| claimed_item(row) } + end + + def done(rows) + page(rows.map { |row| done_item(row) }) { |item| Cursor.encode_finished(item[:finished_at], item[:id]) } + end + + def failed(rows) + page(rows.map { |row| failed_item(row) }) { |item| Cursor.encode_finished(item[:finished_at], item[:id]) } + end + + def pending(rows) + page(rows.map { |row| pending_item(row) }) { |item| Cursor.encode_pending(item[:run_at], item[:id]) } + end + + private + + def page(items) + {items: items, next_cursor: items.empty? ? nil : yield(items.last)} + end + + def executing_item(row) + args, args_count = args_for(row[:args_raw]) + { + id: row[:id], + class_name: row[:class_name], + args: args, + args_count: args_count, + options: parse_options(row[:options_raw]), + started_at: row[:started_at], + locked_by: row[:locked_by], + locked_at: row[:locked_at] + } + end + + def claimed_item(row) + args, args_count = args_for(row[:args_raw]) + { + id: row[:id], + class_name: row[:class_name], + args: args, + args_count: args_count, + options: parse_options(row[:options_raw]), + locked_at: row[:locked_at], + locked_by: row[:locked_by] + } + end + + def done_item(row) + args, args_count = args_for(row[:args_raw]) + { + id: row[:id], + class_name: row[:class_name], + args: args, + args_count: args_count, + options: parse_options(row[:options_raw]), + finished_at: row[:finished_at], + duration_ms: row[:duration_ms] + } + end + + def failed_item(row) + args, args_count = args_for(row[:args_raw]) + { + id: row[:id], + class_name: row[:class_name], + args: args, + args_count: args_count, + options: parse_options(row[:options_raw]), + finished_at: row[:finished_at], + duration_ms: row[:duration_ms], + last_error_class: row[:last_error_class], + last_error_message: row[:last_error_message] + } + end + + def pending_item(row) + args, args_count = args_for(row[:args_raw]) + { + id: row[:id], + class_name: row[:class_name], + args: args, + args_count: args_count, + options: parse_options(row[:options_raw]), + created_at: row[:created_at], + run_at: row[:run_at] + } + end + + def args_for(raw) + if raw.nil? || raw.empty? || raw == '[]' + return [@config.expose_args ? redact(EMPTY_ARGS) : nil, 0] + end + + parsed = parse_json(raw) + count = parsed.is_a?(Array) ? parsed.length : 0 + return [nil, count] unless @config.expose_args + + [redact(parsed), count] + end + + def redact(args) + redactor = @config.redact_args + redactor ? redactor.call(args) : args + end + + def parse_options(raw) + return EMPTY_OPTIONS if raw.nil? || raw.empty? || raw == '{}' + + parsed = parse_json(raw) + parsed.is_a?(Hash) ? parsed : EMPTY_OPTIONS + end + + def parse_json(raw) + JSON.parse(raw) + rescue JSON::ParserError + nil + end + end + end + end +end diff --git a/lib/async/background/web/snapshot.rb b/lib/async/background/web/snapshot.rb new file mode 100644 index 0000000..6572422 --- /dev/null +++ b/lib/async/background/web/snapshot.rb @@ -0,0 +1,247 @@ +# frozen_string_literal: true + +require 'uri' + +require_relative '../clock' +require_relative '../queue/sql' +require_relative 'sql' + +module Async + module Background + module Web + class Snapshot + include Clock + + CacheEntry = Data.define(:value, :created_at) + + def initialize(path:, counts_cache_ttl:) + @path = path + @overview_cache_ttl = counts_cache_ttl + @mutex = Mutex.new + @db = nil + @overview_cache = nil + end + + def open! + @mutex.synchronize do + return self if connected? + + db = open_database + configure_database(db) + @db = db + rescue StandardError + db&.close unless db&.closed? + raise + end + self + end + + def close + @mutex.synchronize do + @db&.close unless @db&.closed? + @db = nil + @overview_cache = nil + end + self + end + + def closed? + @mutex.synchronize { !connected? } + end + + def data_version + with_database { |db| db.get_first_value(Queue::SQL::DATA_VERSION).to_i } + end + + def overview(force: false) + with_database do |db| + now = monotonic_now + return @overview_cache.value if !force && overview_cache_current?(now) + + value = read_transaction(db) { overview_from(db) }.freeze + @overview_cache = CacheEntry.new(value, now) + value + end + end + + def executing(limit:) + read_rows(SQL::EXECUTING, [limit]).map { |row| executing_row(row) } + end + + def claimed(limit:) + read_rows(SQL::CLAIMED, [limit]).map { |row| claimed_row(row) } + end + + def recent_done(limit:, cursor: nil) + sql, binds = terminal_query(SQL::DONE, SQL::DONE_AFTER, limit, cursor) + read_rows(sql, binds).map { |row| done_row(row) } + end + + def recent_failed(limit:, cursor: nil) + sql, binds = terminal_query(SQL::FAILED, SQL::FAILED_AFTER, limit, cursor) + read_rows(sql, binds).map { |row| failed_row(row) } + end + + def pending(limit:, cursor: nil) + sql, binds = pending_query(limit, cursor) + read_rows(sql, binds).map { |row| pending_row(row) } + end + + private + + def connected? + @db && !@db.closed? + end + + def open_database + require_sqlite3 + SQLite3::Database.new(database_uri, uri: true) + rescue LoadError + raise + rescue StandardError => error + raise UnavailableError, "cannot open queue database: #{error.message}" + end + + def database_uri + path = URI::DEFAULT_PARSER.escape(File.expand_path(@path)).gsub('?', '%3F') + "file:#{path}?mode=ro" + end + + def configure_database(db) + db.execute(SQL::BUSY_TIMEOUT) + db.execute(SQL::QUERY_ONLY) + end + + def with_database + @mutex.synchronize do + raise ClosedError, 'snapshot is closed' unless connected? + + yield @db + end + rescue ClosedError, UnavailableError + raise + rescue StandardError + raise UnavailableError, 'queue database is unavailable' + end + + def read_rows(sql, binds) + with_database { |db| read_transaction(db) { db.execute(sql, binds) } } + end + + def read_transaction(db) + db.execute(SQL::BEGIN_READ_TRANSACTION) + result = yield + db.execute(SQL::COMMIT) + result + rescue StandardError + rollback(db) + raise + end + + def rollback(db) + db.execute(SQL::ROLLBACK) + rescue StandardError + nil + end + + def overview_cache_current?(now) + cache = @overview_cache + cache && (now - cache.created_at) < @overview_cache_ttl + end + + def overview_from(db) + { + counts: { + executing: db.get_first_value(SQL::OVERVIEW_EXECUTING).to_i, + claimed: db.get_first_value(SQL::OVERVIEW_CLAIMED).to_i, + pending: db.get_first_value(SQL::OVERVIEW_PENDING).to_i, + done: db.get_first_value(SQL::OVERVIEW_DONE).to_i, + failed: db.get_first_value(SQL::OVERVIEW_FAILED).to_i + }.freeze, + next_pending_run_at: db.get_first_value(SQL::OVERVIEW_NEXT_PENDING), + data_version: db.get_first_value(Queue::SQL::DATA_VERSION).to_i, + generated_at: realtime_now + } + end + + def terminal_query(first_page_sql, next_page_sql, limit, cursor) + return [first_page_sql, [limit]] unless cursor + + [next_page_sql, [cursor.fetch(:finished_at), cursor.fetch(:id), limit]] + end + + def pending_query(limit, cursor) + return [SQL::PENDING, [limit]] unless cursor + + [SQL::PENDING_AFTER, [cursor.fetch(:run_at), cursor.fetch(:id), limit]] + end + + def executing_row(row) + { + id: row[0], + class_name: row[1], + args_raw: row[2], + options_raw: row[3], + started_at: row[4], + locked_by: row[5], + locked_at: row[6] + } + end + + def claimed_row(row) + { + id: row[0], + class_name: row[1], + args_raw: row[2], + options_raw: row[3], + locked_at: row[4], + locked_by: row[5] + } + end + + def done_row(row) + { + id: row[0], + class_name: row[1], + args_raw: row[2], + options_raw: row[3], + finished_at: row[4], + duration_ms: row[5] + } + end + + def failed_row(row) + { + id: row[0], + class_name: row[1], + args_raw: row[2], + options_raw: row[3], + finished_at: row[4], + duration_ms: row[5], + last_error_class: row[6], + last_error_message: row[7] + } + end + + def pending_row(row) + { + id: row[0], + class_name: row[1], + args_raw: row[2], + options_raw: row[3], + created_at: row[4], + run_at: row[5] + } + end + + def require_sqlite3 + require 'sqlite3' + rescue LoadError + raise LoadError, + "sqlite3 gem is required for Async::Background::Web. " \ + "Add `gem 'sqlite3', '~> 2.0'` to your Gemfile." + end + end + end + end +end diff --git a/lib/async/background/web/sql.rb b/lib/async/background/web/sql.rb new file mode 100644 index 0000000..3538a2e --- /dev/null +++ b/lib/async/background/web/sql.rb @@ -0,0 +1,88 @@ +# frozen_string_literal: true + +module Async + module Background + module Web + module SQL + BEGIN_READ_TRANSACTION = 'BEGIN'.freeze + COMMIT = 'COMMIT'.freeze + ROLLBACK = 'ROLLBACK'.freeze + QUERY_ONLY = 'PRAGMA query_only = ON'.freeze + BUSY_TIMEOUT = 'PRAGMA busy_timeout = 2000'.freeze + + OVERVIEW_EXECUTING = "SELECT COUNT(*) FROM jobs WHERE status = 'running' AND started_at IS NOT NULL".freeze + OVERVIEW_CLAIMED = "SELECT COUNT(*) FROM jobs WHERE status = 'running' AND started_at IS NULL".freeze + OVERVIEW_PENDING = "SELECT COUNT(*) FROM jobs WHERE status = 'pending'".freeze + OVERVIEW_DONE = "SELECT COUNT(*) FROM jobs WHERE status = 'done'".freeze + OVERVIEW_FAILED = "SELECT COUNT(*) FROM jobs WHERE status = 'failed'".freeze + OVERVIEW_NEXT_PENDING = "SELECT MIN(run_at) FROM jobs WHERE status = 'pending'".freeze + + EXECUTING = <<~SQL.freeze + SELECT id, class_name, args, options, started_at, locked_by, locked_at + FROM jobs + WHERE status = 'running' AND started_at IS NOT NULL + ORDER BY started_at, id + LIMIT ? + SQL + + CLAIMED = <<~SQL.freeze + SELECT id, class_name, args, options, locked_at, locked_by + FROM jobs + WHERE status = 'running' AND started_at IS NULL + ORDER BY locked_at, id + LIMIT ? + SQL + + DONE = <<~SQL.freeze + SELECT id, class_name, args, options, finished_at, duration_ms + FROM jobs + WHERE status = 'done' + ORDER BY finished_at DESC, id DESC + LIMIT ? + SQL + + DONE_AFTER = <<~SQL.freeze + SELECT id, class_name, args, options, finished_at, duration_ms + FROM jobs + WHERE status = 'done' AND (finished_at, id) < (?, ?) + ORDER BY finished_at DESC, id DESC + LIMIT ? + SQL + + FAILED = <<~SQL.freeze + SELECT id, class_name, args, options, finished_at, duration_ms, + last_error_class, last_error_message + FROM jobs + WHERE status = 'failed' + ORDER BY finished_at DESC, id DESC + LIMIT ? + SQL + + FAILED_AFTER = <<~SQL.freeze + SELECT id, class_name, args, options, finished_at, duration_ms, + last_error_class, last_error_message + FROM jobs + WHERE status = 'failed' AND (finished_at, id) < (?, ?) + ORDER BY finished_at DESC, id DESC + LIMIT ? + SQL + + PENDING = <<~SQL.freeze + SELECT id, class_name, args, options, created_at, run_at + FROM jobs + WHERE status = 'pending' + ORDER BY run_at, id + LIMIT ? + SQL + + PENDING_AFTER = <<~SQL.freeze + SELECT id, class_name, args, options, created_at, run_at + FROM jobs + WHERE status = 'pending' AND (run_at, id) > (?, ?) + ORDER BY run_at, id + LIMIT ? + SQL + end + end + end +end diff --git a/lib/async/background/web/stream.rb b/lib/async/background/web/stream.rb new file mode 100644 index 0000000..309d382 --- /dev/null +++ b/lib/async/background/web/stream.rb @@ -0,0 +1,43 @@ +# frozen_string_literal: true + +module Async + module Background + module Web + class Stream + def initialize(hub, heartbeat_seconds:, retry_ms:) + @hub = hub + @heartbeat_seconds = heartbeat_seconds + @retry_ms = retry_ms + end + + def each + subscription, initial_frame = @hub.subscribe + yield "retry: #{@retry_ms}\n\n" + yield initial_frame + + loop do + frame = subscription.pop(timeout: @heartbeat_seconds) + break if frame.nil? && subscription.closed? + + yield(frame || EventHub::HEARTBEAT_FRAME) + end + rescue Errno::EPIPE, IOError + nil + rescue ClosedError, UnavailableError + safe_yield(EventHub::UNAVAILABLE_FRAME) { |frame| yield frame } + nil + ensure + @hub.unsubscribe(subscription) if subscription + end + + private + + def safe_yield(frame) + yield frame + rescue Errno::EPIPE, IOError + nil + end + end + end + end +end diff --git a/spec/async/background/queue/store_migration_spec.rb b/spec/async/background/queue/store_migration_spec.rb index 3ef57d8..f1277bd 100644 --- a/spec/async/background/queue/store_migration_spec.rb +++ b/spec/async/background/queue/store_migration_spec.rb @@ -109,7 +109,8 @@ def plan_details(db, sql, binds = []) 'idx_jobs_pending', 'idx_jobs_done_finished_at', 'idx_jobs_failed_finished_at', - 'idx_jobs_running' + 'idx_jobs_executing_started_at', + 'idx_jobs_claimed_locked_at' ) schema_version = @db.get_first_value('PRAGMA schema_version') @@ -125,7 +126,8 @@ def plan_details(db, sql, binds = []) 'idx_jobs_pending', 'idx_jobs_done_finished_at', 'idx_jobs_failed_finished_at', - 'idx_jobs_running' + 'idx_jobs_executing_started_at', + 'idx_jobs_claimed_locked_at' ) end @@ -167,15 +169,36 @@ def plan_details(db, sql, binds = []) "SELECT id FROM jobs WHERE status = 'failed' " \ 'ORDER BY finished_at DESC, id DESC LIMIT 50' ), - in_flight: plan_details( + done_after: plan_details( @db, - "SELECT id FROM jobs WHERE status = 'running' ORDER BY locked_at ASC, id ASC LIMIT 50" + "SELECT id FROM jobs WHERE status = 'done' AND (finished_at, id) < (?, ?) " \ + 'ORDER BY finished_at DESC, id DESC LIMIT 50', + [now + 10, 9_999] + ), + pending_after: plan_details( + @db, + "SELECT id FROM jobs WHERE status = 'pending' AND (run_at, id) > (?, ?) " \ + 'ORDER BY run_at ASC, id ASC LIMIT 50', + [now - 10, 0] + ), + executing: plan_details( + @db, + "SELECT id FROM jobs WHERE status = 'running' AND started_at IS NOT NULL " \ + 'ORDER BY started_at ASC, id ASC LIMIT 50' + ), + claimed: plan_details( + @db, + "SELECT id FROM jobs WHERE status = 'running' AND started_at IS NULL " \ + 'ORDER BY locked_at ASC, id ASC LIMIT 50' ) } expect(plans.fetch(:done)).to include('idx_jobs_done_finished_at') expect(plans.fetch(:failed)).to include('idx_jobs_failed_finished_at') - expect(plans.fetch(:in_flight)).to include('idx_jobs_running') + expect(plans.fetch(:done_after)).to include('idx_jobs_done_finished_at') + expect(plans.fetch(:pending_after)).to include('idx_jobs_pending') + expect(plans.fetch(:executing)).to include('idx_jobs_executing_started_at') + expect(plans.fetch(:claimed)).to include('idx_jobs_claimed_locked_at') plans.each_value { |plan| expect(plan).not_to include('USE TEMP B-TREE FOR ORDER BY') } end diff --git a/spec/async/background/queue/store_spec.rb b/spec/async/background/queue/store_spec.rb index b05a2ab..e90e390 100644 --- a/spec/async/background/queue/store_spec.rb +++ b/spec/async/background/queue/store_spec.rb @@ -107,7 +107,8 @@ def db 'idx_jobs_pending', 'idx_jobs_done_finished_at', 'idx_jobs_failed_finished_at', - 'idx_jobs_running' + 'idx_jobs_executing_started_at', + 'idx_jobs_claimed_locked_at' ) end end diff --git a/spec/async/background/runner_spec.rb b/spec/async/background/runner_spec.rb index ab1dae9..e91db18 100644 --- a/spec/async/background/runner_spec.rb +++ b/spec/async/background/runner_spec.rb @@ -2,6 +2,7 @@ require 'spec_helper' require 'yaml' +require 'tmpdir' RSpec.describe Async::Background::Runner, type: :unit do before(:all) do @@ -54,6 +55,45 @@ def build_runner(schedule: minimal_schedule, worker_index: 1, total_workers: 1) let(:runner) { build_runner } + describe 'queue-only initialization' do + it 'accepts config_path: nil when a queue listener is configured' do + socket_dir = Dir.mktmpdir('async-background-runner') + queue_path = temp_db_path + metrics_path = temp_file_path('.shm') + queue_only_runner = nil + + expect { + queue_only_runner = described_class.new( + config_path: nil, + job_count: 1, + worker_index: 1, + total_workers: 1, + queue_db_path: queue_path, + queue_socket_dir: socket_dir, + metrics_shm_path: metrics_path + ) + }.not_to raise_error + + expect(queue_only_runner.heap).to be_empty + expect(queue_only_runner.queue_store).to be_a(Async::Background::Queue::Store) + ensure + queue_only_runner&.queue_store&.close + queue_only_runner&.instance_variable_get(:@queue_waker)&.close + FileUtils.rm_rf(socket_dir) if socket_dir + end + + it 'rejects an empty runner with neither a schedule nor a queue listener' do + expect { + described_class.new( + config_path: nil, + job_count: 1, + worker_index: 1, + total_workers: 1 + ) + }.to raise_error(Async::Background::ConfigError, /config_path or queue_socket_dir/) + end + end + describe '#resolve_job_class (private)' do it 'resolves a top-level Job class by name' do klass = runner.send(:resolve_job_class, 'RunnerSpecJob') diff --git a/spec/async/background/web/app_spec.rb b/spec/async/background/web/app_spec.rb new file mode 100644 index 0000000..16552f9 --- /dev/null +++ b/spec/async/background/web/app_spec.rb @@ -0,0 +1,255 @@ +# frozen_string_literal: true + +require 'spec_helper' +require 'async/background/web' +require 'json' + +RSpec.describe Async::Background::Web::App do + let(:db_path) { temp_db_path } + + before do + Async::Background::Queue::Store.prepare_dashboard!(path: db_path) + end + + def build_app(overrides = {}) + config = Async::Background::Web::Configuration.new + config.queue_path = db_path + config.auth = ->(env) { env['HTTP_X_TOKEN'] == 'allow' } + overrides.each { |k, v| config.public_send("#{k}=", v) } + described_class.new(config) + end + + def env_for(method:, path:, headers: {}, query: '') + base = { + 'REQUEST_METHOD' => method, + 'PATH_INFO' => path, + 'QUERY_STRING' => query + } + headers.each { |k, v| base["HTTP_#{k.upcase.tr('-', '_')}"] = v } + base + end + + describe 'auth gate' do + it 'returns 401 when auth callable returns falsy' do + app = build_app + status, _, body = app.call(env_for(method: 'GET', path: '/api/overview')) + expect(status).to eq(401) + expect(body.first).to eq('unauthorized') + end + + it 'returns 401 when auth callable raises' do + app = build_app + app.instance_variable_set(:@auth, Async::Background::Web::Auth.new(->(_env) { raise 'boom' })) + status, = app.call(env_for(method: 'GET', path: '/api/overview')) + expect(status).to eq(401) + end + + it 'lets the request through when auth returns truthy' do + app = build_app + status, headers, = app.call(env_for(method: 'GET', path: '/api/overview', headers: { 'x-token' => 'allow' })) + expect(status).to eq(200) + expect(headers['content-type']).to start_with('application/json') + end + end + + describe '/api/overview' do + let(:app) { build_app } + + it 'returns json with counts and data_version' do + _, _, body = app.call(env_for(method: 'GET', path: '/api/overview', headers: { 'x-token' => 'allow' })) + payload = JSON.parse(body.first, symbolize_names: true) + expect(payload).to include(:counts, :data_version, :generated_at) + expect(payload[:counts]).to include(:executing, :claimed, :pending, :done, :failed) + end + end + + describe '/api/config' do + let(:app) { build_app(title: 'My Background', poll_interval_ms: 1500) } + + it 'exposes UI knobs' do + _, _, body = app.call(env_for(method: 'GET', path: '/api/config', headers: { 'x-token' => 'allow' })) + payload = JSON.parse(body.first, symbolize_names: true) + expect(payload[:title]).to eq('My Background') + expect(payload[:poll_interval_ms]).to eq(1500) + expect(payload[:expose_args]).to eq(false) + expect(payload[:transport]).to eq('sse') + end + + it 'reports sse when transport is sse' do + app = build_app(transport: :sse) + _, _, body = app.call(env_for(method: 'GET', path: '/api/config', headers: { 'x-token' => 'allow' })) + payload = JSON.parse(body.first, symbolize_names: true) + expect(payload[:transport]).to eq('sse') + end + end + + describe '/api/stream' do + it 'returns 404 when transport is explicitly polling' do + app = build_app(transport: :polling) + status, = app.call(env_for(method: 'GET', path: '/api/stream', headers: { 'x-token' => 'allow' })) + expect(status).to eq(404) + end + + it 'returns 200 text/event-stream when transport is sse' do + app = build_app(transport: :sse) + status, headers, body = app.call(env_for(method: 'GET', path: '/api/stream', headers: { 'x-token' => 'allow' })) + expect(status).to eq(200) + expect(headers['content-type']).to start_with('text/event-stream') + expect(headers['cache-control']).to include('no-cache') + expect(headers['x-accel-buffering']).to eq('no') + expect(body).to respond_to(:each) + end + + it 'requires auth like every other endpoint' do + app = build_app(transport: :sse) + status, = app.call(env_for(method: 'GET', path: '/api/stream')) + expect(status).to eq(401) + end + end + + describe 'in-flight routes' do + let(:app) { build_app } + + it 'returns an item envelope for executing jobs' do + status, _, body = app.call(env_for(method: 'GET', path: '/api/executing', headers: { 'x-token' => 'allow' })) + + expect(status).to eq(200) + expect(JSON.parse(body.first, symbolize_names: true)).to eq(items: []) + end + + it 'returns an item envelope for claimed jobs' do + status, _, body = app.call(env_for(method: 'GET', path: '/api/claimed', headers: { 'x-token' => 'allow' })) + + expect(status).to eq(200) + expect(JSON.parse(body.first, symbolize_names: true)).to eq(items: []) + end + end + + describe '/api/done with cursor' do + let(:app) { build_app } + + before do + store = Async::Background::Queue::Store.new(path: db_path) + 5.times do |i| + store.enqueue('CursorJob', [i], 1_700_000_000.0 - 100) + job = store.fetch(1) + store.complete(job[:id], claim_token: job[:claim_token], finished_at: 1_700_000_000.0 + i, duration_ms: 1) + end + store.close + end + + it 'returns items and next_cursor' do + _, _, body = app.call(env_for(method: 'GET', path: '/api/done', query: 'limit=2', headers: { 'x-token' => 'allow' })) + payload = JSON.parse(body.first, symbolize_names: true) + expect(payload[:items].length).to eq(2) + expect(payload[:next_cursor]).not_to be_nil + end + end + + describe '/api/metrics' do + it 'reports unavailable when metrics_path is not configured' do + app = build_app + _, _, body = app.call(env_for(method: 'GET', path: '/api/metrics', headers: { 'x-token' => 'allow' })) + payload = JSON.parse(body.first, symbolize_names: true) + expect(payload[:available]).to eq(false) + end + end + + describe '/' do + let(:app) { build_app } + + it 'serves the HTML shell' do + status, headers, body = app.call(env_for(method: 'GET', path: '/', headers: { 'x-token' => 'allow' })) + expect(status).to eq(200) + expect(headers['content-type']).to start_with('text/html') + expect(body.first).to include('<title>') + expect(body.first).to include('Async::Background') + end + + it 'serves the JS asset' do + status, headers, body = app.call(env_for(method: 'GET', path: '/assets/app.js', headers: { 'x-token' => 'allow' })) + expect(status).to eq(200) + expect(headers['content-type']).to start_with('application/javascript') + expect(body.first).to include('DOMContentLoaded') + expect(body.first).to include('bootBasePath') + expect(body.first).to include('document.currentScript') + expect(body.first).to include(%q{replace(/\/assets\/app\.js$/, '')}) + expect(body.first).to include(%q{replace(/\/$/, '')}) + expect(body.first).to include('scheduleActiveListRefresh') + expect(body.first).to include('new EventSource(streamUrl())') + end + + it 'embeds the configured mount path into the HTML shell' do + app = build_app(mount_path: '/admin/background') + status, _, body = app.call(env_for(method: 'GET', path: '/', headers: { 'x-token' => 'allow' })) + + expect(status).to eq(200) + expect(body.first).to include('data-mount-path="/admin/background"') + expect(body.first).to include('src="/admin/background/assets/app.js?v=') + expect(body.first).to include('app.css?v=') + end + + it 'serves the CSS asset' do + status, headers = app.call(env_for(method: 'GET', path: '/assets/app.css', headers: { 'x-token' => 'allow' })) + expect(status).to eq(200) + expect(headers['content-type']).to start_with('text/css') + end + end + + describe 'unknown routes' do + let(:app) { build_app } + + it 'returns 404' do + status, _, body = app.call(env_for(method: 'GET', path: '/nope', headers: { 'x-token' => 'allow' })) + expect(status).to eq(404) + expect(body.first).to eq('not found') + end + end + + + describe 'request errors' do + let(:app) { build_app } + + it 'returns 400 for a malformed cursor instead of silently restarting pagination' do + status, _, body = app.call( + env_for(method: 'GET', path: '/api/done', query: 'cursor=not-a-cursor', headers: { 'x-token' => 'allow' }) + ) + + expect(status).to eq(400) + expect(JSON.parse(body.first, symbolize_names: true)).to eq(error: 'invalid_request', message: 'invalid cursor') + end + + it 'returns 400 for a non-positive limit' do + status, _, body = app.call( + env_for(method: 'GET', path: '/api/pending', query: 'limit=0', headers: { 'x-token' => 'allow' }) + ) + + expect(status).to eq(400) + expect(JSON.parse(body.first, symbolize_names: true)).to include(error: 'invalid_request') + end + end + + describe 'lifecycle' do + it 'returns 503 after the read model is closed' do + app = build_app + app.close + + status, _, body = app.call(env_for(method: 'GET', path: '/api/overview', headers: { 'x-token' => 'allow' })) + expect(status).to eq(503) + expect(JSON.parse(body.first, symbolize_names: true)).to eq(error: 'service_unavailable') + end + end + + describe 'internal errors' do + let(:app) { build_app } + + it 'wraps internal exceptions in 500 JSON' do + allow_any_instance_of(Async::Background::Web::Snapshot).to receive(:overview).and_raise(RuntimeError, 'kaboom') + status, headers, body = app.call(env_for(method: 'GET', path: '/api/overview', headers: { 'x-token' => 'allow' })) + expect(status).to eq(500) + expect(headers['content-type']).to start_with('application/json') + payload = JSON.parse(body.first, symbolize_names: true) + expect(payload).to eq(error: 'internal_error') + end + end +end diff --git a/spec/async/background/web/configuration_spec.rb b/spec/async/background/web/configuration_spec.rb new file mode 100644 index 0000000..6823634 --- /dev/null +++ b/spec/async/background/web/configuration_spec.rb @@ -0,0 +1,126 @@ +# frozen_string_literal: true + +require 'spec_helper' +require 'async/background/web' + +RSpec.describe Async::Background::Web::Configuration do + let(:config) { described_class.new } + + describe '#validate!' do + it 'requires queue_path' do + config.queue_path = nil + config.auth = ->(_env) { true } + expect { config.validate! }.to raise_error(Async::Background::Web::ConfigurationError, /queue_path/) + end + + it 'rejects empty queue_path' do + config.queue_path = '' + config.auth = ->(_env) { true } + expect { config.validate! }.to raise_error(Async::Background::Web::ConfigurationError, /queue_path/) + end + + it 'requires auth to be configured' do + config.queue_path = '/tmp/q.db' + config.auth = nil + expect { config.validate! }.to raise_error(Async::Background::Web::ConfigurationError, /auth must be configured/) + end + + it 'requires auth to be callable' do + config.queue_path = '/tmp/q.db' + config.auth = 'not a proc' + expect { config.validate! }.to raise_error(Async::Background::Web::ConfigurationError, /respond to #call/) + end + + it 'requires list_limit in range' do + config.queue_path = '/tmp/q.db' + config.auth = ->(_env) { true } + config.list_limit = 0 + expect { config.validate! }.to raise_error(Async::Background::Web::ConfigurationError, /list_limit/) + config.list_limit = 500 + expect { config.validate! }.to raise_error(Async::Background::Web::ConfigurationError, /list_limit/) + end + + it 'requires non-negative counts_cache_ttl' do + config.queue_path = '/tmp/q.db' + config.auth = ->(_env) { true } + config.counts_cache_ttl = -1 + expect { config.validate! }.to raise_error(Async::Background::Web::ConfigurationError, /counts_cache_ttl/) + end + + it 'requires poll_interval_ms >= 200' do + config.queue_path = '/tmp/q.db' + config.auth = ->(_env) { true } + config.poll_interval_ms = 100 + expect { config.validate! }.to raise_error(Async::Background::Web::ConfigurationError, /poll_interval_ms/) + end + + it 'requires transport to be one of :polling or :sse' do + config.queue_path = '/tmp/q.db' + config.auth = ->(_env) { true } + config.transport = :ws + expect { config.validate! }.to raise_error(Async::Background::Web::ConfigurationError, /transport/) + end + + it 'accepts :sse transport' do + config.queue_path = '/tmp/q.db' + config.auth = ->(_env) { true } + config.transport = :sse + expect { config.validate! }.not_to raise_error + end + + it 'validates SSE timing knobs' do + config.queue_path = '/tmp/q.db' + config.auth = ->(_env) { true } + config.stream_poll_seconds = 0.05 + expect { config.validate! }.to raise_error(Async::Background::Web::ConfigurationError, /stream_poll_seconds/) + + config.stream_poll_seconds = 0.5 + config.stream_heartbeat_seconds = 4 + expect { config.validate! }.to raise_error(Async::Background::Web::ConfigurationError, /stream_heartbeat_seconds/) + + config.stream_heartbeat_seconds = 25 + config.stream_retry_ms = 100 + expect { config.validate! }.to raise_error(Async::Background::Web::ConfigurationError, /stream_retry_ms/) + end + + it 'requires total_workers when metrics_path is set' do + config.queue_path = '/tmp/q.db' + config.auth = ->(_env) { true } + config.metrics_path = '/tmp/m.shm' + config.total_workers = nil + expect { config.validate! }.to raise_error(Async::Background::Web::ConfigurationError, /total_workers/) + end + + it 'passes with minimal valid config' do + config.queue_path = '/tmp/q.db' + config.auth = ->(_env) { true } + expect(config.validate!).to eq(config) + end + end + + describe '#limit_for' do + it 'rejects malformed and non-positive HTTP values' do + expect { config.limit_for('0') }.to raise_error(Async::Background::Web::RequestError, /positive/) + expect { config.limit_for('oops') }.to raise_error(Async::Background::Web::RequestError, /positive/) + end + + it 'caps an oversized value at MAX_LIST_LIMIT' do + expect(config.limit_for('1000')).to eq(described_class::MAX_LIST_LIMIT) + end + end + + describe 'defaults' do + it 'sets safe defaults' do + expect(config.expose_args).to eq(false) + expect(config.list_limit).to eq(50) + expect(config.counts_cache_ttl).to eq(3.0) + expect(config.poll_interval_ms).to eq(2000) + expect(config.transport).to eq(:sse) + expect(config.stream_poll_seconds).to eq(0.5) + expect(config.stream_heartbeat_seconds).to eq(25.0) + expect(config.stream_retry_ms).to eq(5000) + expect(config.title).to eq('Async::Background') + expect(config.mount_path).to eq('') + end + end +end diff --git a/spec/async/background/web/cursor_spec.rb b/spec/async/background/web/cursor_spec.rb new file mode 100644 index 0000000..5b02ba3 --- /dev/null +++ b/spec/async/background/web/cursor_spec.rb @@ -0,0 +1,39 @@ +# frozen_string_literal: true + +require 'spec_helper' +require 'async/background/web' + +RSpec.describe Async::Background::Web::Cursor do + describe '.decode_finished' do + it 'roundtrips an opaque finished cursor' do + encoded = described_class.encode_finished(1234.5, 99) + expect(described_class.decode_finished(encoded)).to eq(finished_at: 1234.5, id: 99) + end + + it 'returns nil only when no cursor was supplied' do + expect(described_class.decode_finished(nil)).to be_nil + expect(described_class.decode_finished('')).to be_nil + end + + it 'rejects malformed, non-finite and non-positive cursor values' do + invalid = [ + 'not-base64', + Base64.urlsafe_encode64('NaN:1', padding: false), + Base64.urlsafe_encode64('1.0:0', padding: false), + Base64.urlsafe_encode64('1.0:1:extra', padding: false) + ] + + invalid.each do |value| + expect { described_class.decode_finished(value) } + .to raise_error(Async::Background::Web::RequestError, 'invalid cursor') + end + end + end + + describe '.decode_pending' do + it 'roundtrips an opaque pending cursor' do + encoded = described_class.encode_pending(555.25, 42) + expect(described_class.decode_pending(encoded)).to eq(run_at: 555.25, id: 42) + end + end +end diff --git a/spec/async/background/web/dashboard_query_plans_spec.rb b/spec/async/background/web/dashboard_query_plans_spec.rb new file mode 100644 index 0000000..74d6f3c --- /dev/null +++ b/spec/async/background/web/dashboard_query_plans_spec.rb @@ -0,0 +1,83 @@ +# frozen_string_literal: true + +require 'spec_helper' +require 'async/background/web' + +RSpec.describe 'dashboard SQL query plans', type: :unit do + let(:db_path) { temp_db_path } + let(:db) do + Async::Background::Queue::Store.prepare_dashboard!(path: db_path) + require 'sqlite3' + SQLite3::Database.new(db_path) + end + + after do + db&.close unless db&.closed? + end + + def plan_for(sql, binds = []) + db.execute("EXPLAIN QUERY PLAN #{sql}", binds).map { |row| row.last.to_s }.join("\n") + end + + shared_examples 'uses index without temp sort' do |sql_const, binds, expected_index| + it "uses #{expected_index} and avoids temp B-tree" do + sql = Async::Background::Web::SQL.const_get(sql_const) + plan = plan_for(sql, binds) + expect(plan).to include(expected_index), + "plan was:\n#{plan}\nexpected index: #{expected_index}" + expect(plan).not_to include('USE TEMP B-TREE'), + "plan should not require a temp B-tree sort:\n#{plan}" + end + end + + describe 'list queries' do + include_examples 'uses index without temp sort', :DONE, [50], 'idx_jobs_done_finished_at' + include_examples 'uses index without temp sort', :DONE_AFTER, [1_700_000_000.0, 999, 50], 'idx_jobs_done_finished_at' + include_examples 'uses index without temp sort', :FAILED, [50], 'idx_jobs_failed_finished_at' + include_examples 'uses index without temp sort', :FAILED_AFTER, [1_700_000_000.0, 999, 50], 'idx_jobs_failed_finished_at' + include_examples 'uses index without temp sort', :PENDING, [50], 'idx_jobs_pending' + include_examples 'uses index without temp sort', :PENDING_AFTER, [1_700_000_000.0, 0, 50], 'idx_jobs_pending' + end + + describe 'overview scalar queries' do + it 'pending count uses idx_jobs_pending' do + plan = plan_for(Async::Background::Web::SQL::OVERVIEW_PENDING) + expect(plan).to include('idx_jobs_pending') + end + + it 'done count uses the per-status covering index' do + plan = plan_for(Async::Background::Web::SQL::OVERVIEW_DONE) + expect(plan).to include('idx_jobs_done_finished_at') + end + + it 'failed count uses the per-status covering index' do + plan = plan_for(Async::Background::Web::SQL::OVERVIEW_FAILED) + expect(plan).to include('idx_jobs_failed_finished_at') + end + + it 'next_pending uses idx_jobs_pending' do + plan = plan_for(Async::Background::Web::SQL::OVERVIEW_NEXT_PENDING) + expect(plan).to include('idx_jobs_pending') + end + + it 'executing count uses the executing partial index' do + plan = plan_for(Async::Background::Web::SQL::OVERVIEW_EXECUTING) + expect(plan).to include('idx_jobs_executing_started_at') + end + + it 'claimed count uses the claimed partial index' do + plan = plan_for(Async::Background::Web::SQL::OVERVIEW_CLAIMED) + expect(plan).to include('idx_jobs_claimed_locked_at') + end + end + + describe 'no overview query does a full table scan' do + %i[OVERVIEW_PENDING OVERVIEW_DONE OVERVIEW_FAILED OVERVIEW_NEXT_PENDING].each do |const| + it "#{const} avoids SCAN jobs without an index" do + plan = plan_for(Async::Background::Web::SQL.const_get(const)) + expect(plan).not_to match(/SCAN jobs(?! USING)/), + "plan does a full scan:\n#{plan}" + end + end + end +end diff --git a/spec/async/background/web/event_hub_spec.rb b/spec/async/background/web/event_hub_spec.rb new file mode 100644 index 0000000..6aef2f5 --- /dev/null +++ b/spec/async/background/web/event_hub_spec.rb @@ -0,0 +1,91 @@ +# frozen_string_literal: true + +require 'spec_helper' +require 'async/background/web' +require 'timeout' + +RSpec.describe Async::Background::Web::EventHub do + class HubSnapshot + def initialize(version: 1) + @mutex = Mutex.new + @version = version + end + + def data_version + @mutex.synchronize { @version } + end + + def overview(force: false) + @mutex.synchronize do + { + counts: {executing: 0, claimed: 0, pending: @version, done: 0, failed: 0}, + next_pending_run_at: nil, + data_version: @version, + generated_at: 1.0 + } + end + end + + def advance! + @mutex.synchronize { @version += 1 } + end + end + + class HubSerializer + def overview(snapshot, _metrics) + snapshot + end + end + + def parse_overview(frame) + JSON.parse(frame.split("data: ", 2).last, symbolize_names: true) + end + + def wait_for(timeout: 1) + Timeout.timeout(timeout) do + loop do + value = yield + return value if value + + sleep(0.005) + end + end + end + + it 'fans one committed change out to every connected stream' do + snapshot = HubSnapshot.new + hub = described_class.new(snapshot, HubSerializer.new, poll_seconds: 0.01) + first, first_frame = hub.subscribe + second, second_frame = hub.subscribe + + expect(parse_overview(first_frame).fetch(:data_version)).to eq(1) + expect(parse_overview(second_frame).fetch(:data_version)).to eq(1) + + snapshot.advance! + + first_update = wait_for { first.pop(timeout: 0.02) } + second_update = wait_for { second.pop(timeout: 0.02) } + expect(parse_overview(first_update).fetch(:data_version)).to eq(2) + expect(parse_overview(second_update).fetch(:data_version)).to eq(2) + ensure + hub&.close + end + + it 'keeps only the newest pending frame for a slow subscriber' do + subscription = described_class::Subscription.new + subscription.publish('older') + subscription.publish('newest') + + expect(subscription.pop(timeout: 0)).to eq('newest') + end + + it 'unblocks a waiting subscriber when it is closed' do + subscription = described_class::Subscription.new + waiter = Thread.new { subscription.pop(timeout: 5) } + sleep(0.01) + subscription.close + + expect(waiter.join(1)).not_to be_nil + expect(waiter.value).to be_nil + end +end diff --git a/spec/async/background/web/metrics_reader_spec.rb b/spec/async/background/web/metrics_reader_spec.rb new file mode 100644 index 0000000..3bda9a1 --- /dev/null +++ b/spec/async/background/web/metrics_reader_spec.rb @@ -0,0 +1,45 @@ +# frozen_string_literal: true + +require 'spec_helper' +require 'async/background/web' + +RSpec.describe Async::Background::Web::MetricsReader do + let(:reader) { described_class.new(path: '/tmp/async-background-metrics-test.shm', total_workers: 2, ttl: 0) } + + it 'reports unavailable metrics explicitly when the optional integration is absent' do + allow(Async::Background::Metrics).to receive(:available?).and_return(false) + + expect(reader.aggregated).to eq( + available: false, + workers: [], + totals: described_class::EMPTY_TOTALS + ) + end + + it 'aggregates available worker snapshots without changing their per-worker values' do + workers = [ + {worker: 1, total_runs: 4, total_successes: 3, total_failures: 1, total_timeouts: 0, total_skips: 2, + active_jobs: 1, last_run_at: 10, last_duration_ms: 25}, + {worker: 2, total_runs: 5, total_successes: 5, total_failures: 0, total_timeouts: 1, total_skips: 0, + active_jobs: 2, last_run_at: 20, last_duration_ms: 50} + ].freeze + allow(File).to receive(:file?).with('/tmp/async-background-metrics-test.shm').and_return(true) + allow(Async::Background::Metrics).to receive(:available?).and_return(true) + allow(Async::Background::Metrics).to receive(:read_all).with(total_workers: 2, path: '/tmp/async-background-metrics-test.shm') + .and_return(workers) + + result = reader.aggregated + expect(result[:available]).to eq(true) + expect(result[:workers]).to eq(workers) + expect(result[:totals]).to include( + total_runs: 9, + total_successes: 8, + total_failures: 1, + total_timeouts: 1, + total_skips: 2, + active_jobs: 3, + last_run_at: 20, + last_duration_ms: 50 + ) + end +end diff --git a/spec/async/background/web/request_spec.rb b/spec/async/background/web/request_spec.rb new file mode 100644 index 0000000..5aba6f2 --- /dev/null +++ b/spec/async/background/web/request_spec.rb @@ -0,0 +1,35 @@ +# frozen_string_literal: true + +require 'spec_helper' +require 'async/background/web' + +RSpec.describe Async::Background::Web::Request do + let(:config) do + Async::Background::Web::Configuration.new.tap do |value| + value.queue_path = '/tmp/queue.db' + value.auth = ->(_) { true } + end + end + + def request(query = '') + described_class.new({'QUERY_STRING' => query}, config) + end + + it 'parses a bounded limit and terminal cursor once' do + cursor = Async::Background::Web::Cursor.encode_finished(10.5, 3) + value = request("limit=75&cursor=#{cursor}") + + expect(value.limit).to eq(75) + expect(value.finished_cursor).to eq(finished_at: 10.5, id: 3) + end + + it 'rejects an invalid limit instead of silently changing the requested page' do + expect { request('limit=zero').limit } + .to raise_error(Async::Background::Web::RequestError, /positive integer/) + end + + it 'rejects an invalid cursor instead of treating it as the first page' do + expect { request('cursor=nope').finished_cursor } + .to raise_error(Async::Background::Web::RequestError, 'invalid cursor') + end +end diff --git a/spec/async/background/web/serializer_spec.rb b/spec/async/background/web/serializer_spec.rb new file mode 100644 index 0000000..e3cb936 --- /dev/null +++ b/spec/async/background/web/serializer_spec.rb @@ -0,0 +1,97 @@ +# frozen_string_literal: true + +require 'spec_helper' +require 'async/background/web' + +RSpec.describe Async::Background::Web::Serializer do + let(:config) { Async::Background::Web::Configuration.new.tap { |c| c.queue_path = '/tmp/x.db'; c.auth = ->(_) { true } } } + let(:serializer) { described_class.new(config) } + + describe 'args redaction by default' do + let(:row) do + { + id: 1, + class_name: 'MyJob', + args_raw: JSON.generate(['secret-token', { 'email' => 'user@example.com' }]), + options_raw: nil, + finished_at: 1.0, + duration_ms: 10 + } + end + + it 'does not include args when expose_args is false' do + result = serializer.done([row])[:items].first + expect(result[:args]).to be_nil + expect(result[:args_count]).to eq(2) + end + + it 'exposes raw args when expose_args is true and no redactor is set' do + config.expose_args = true + config.redact_args = nil + result = serializer.done([row])[:items].first + expect(result[:args]).to eq(['secret-token', { 'email' => 'user@example.com' }]) + end + + it 'keeps zero-argument jobs visible as an empty array when args are exposed' do + config.expose_args = true + config.redact_args = nil + row = { id: 1, class_name: 'J', args_raw: '[]', options_raw: nil, finished_at: 1.0, duration_ms: 1 } + + result = serializer.done([row])[:items].first + expect(result[:args]).to eq([]) + expect(result[:args_count]).to eq(0) + end + + it 'applies custom redactor when expose_args is true' do + config.expose_args = true + config.redact_args = ->(args) { args.map { |_| '[REDACTED]' } } + result = serializer.done([row])[:items].first + expect(result[:args]).to eq(['[REDACTED]', '[REDACTED]']) + end + end + + describe 'paging shape' do + it 'returns items array and next_cursor for done' do + row = { id: 1, class_name: 'J', args_raw: '[]', options_raw: nil, finished_at: 99.0, duration_ms: 1 } + result = serializer.done([row]) + expect(result).to have_key(:items) + expect(result).to have_key(:next_cursor) + expect(result[:next_cursor]).not_to be_nil + end + + it 'next_cursor is nil when items is empty' do + result = serializer.done([]) + expect(result[:next_cursor]).to be_nil + end + end + + describe 'parsing args' do + it 'handles malformed JSON without raising' do + row = { id: 1, class_name: 'J', args_raw: 'not json', options_raw: nil, finished_at: 1.0, duration_ms: 1 } + expect { serializer.done([row]) }.not_to raise_error + end + + it 'treats empty string as no args' do + row = { id: 1, class_name: 'J', args_raw: '', options_raw: nil, finished_at: 1.0, duration_ms: 1 } + result = serializer.done([row])[:items].first + expect(result[:args_count]).to eq(0) + end + end + + describe 'overview shape' do + it 'includes metrics when provided' do + snap = { counts: { done: 1 }, next_pending_run_at: 99.0, data_version: 7, generated_at: 100.0 } + metrics = { workers: [], totals: { total_runs: 5 } } + result = serializer.overview(snap, metrics) + expect(result[:metrics]).to eq(metrics) + expect(result[:counts]).to eq(done: 1) + expect(result[:data_version]).to eq(7) + end + + it 'omits metrics when nil' do + snap = { counts: { done: 1 }, next_pending_run_at: nil, data_version: 7, generated_at: 100.0 } + result = serializer.overview(snap, nil) + expect(result).not_to have_key(:metrics) + end + end +end diff --git a/spec/async/background/web/snapshot_spec.rb b/spec/async/background/web/snapshot_spec.rb new file mode 100644 index 0000000..903cb55 --- /dev/null +++ b/spec/async/background/web/snapshot_spec.rb @@ -0,0 +1,221 @@ +# frozen_string_literal: true + +require 'spec_helper' +require 'async/background/web' + +RSpec.describe Async::Background::Web::Snapshot do + let(:db_path) { temp_db_path } + let(:store) { Async::Background::Queue::Store.new(path: db_path) } + let(:snapshot) { described_class.new(path: db_path, counts_cache_ttl: 0).open! } + + before do + Async::Background::Queue::Store.prepare_dashboard!(path: db_path) + end + + after do + snapshot.close + store.close + end + + def seed_done(n, base_time: 1_700_000_000.0) + n.times do |i| + id = store.enqueue('DoneJob', [i], base_time - 100) + job = store.fetch(1) + store.complete(job[:id], claim_token: job[:claim_token], finished_at: base_time - (n - i), duration_ms: 10 + i) + id + end + end + + def seed_failed(n, base_time: 1_700_000_000.0) + n.times do |i| + store.enqueue('FailJob', [i], base_time - 100) + job = store.fetch(1) + store.fail( + job[:id], + claim_token: job[:claim_token], + error_class: StandardError, + error_message: "boom #{i}", + finished_at: base_time - (n - i), + duration_ms: 5 + ) + end + end + + def seed_pending(n, base_time: 1_700_000_000.0) + n.times { |i| store.enqueue('PendingJob', [i], base_time + i) } + end + + def seed_executing(n, base_time: 1_700_000_000.0) + n.times do |i| + store.enqueue("ExecJob#{i}", [i], base_time - 100) + job = store.fetch(1) + store.mark_started!(job[:id], claim_token: job[:claim_token], started_at: base_time + i) + end + end + + def seed_claimed(n, base_time: 1_700_000_000.0) + n.times do |i| + store.enqueue("ClaimedJob#{i}", [i], base_time - 100) + store.fetch(1) + end + end + + describe 'mode=ro enforcement' do + it 'cannot write through the snapshot connection' do + db = snapshot.instance_variable_get(:@db) + expect { + db.execute("INSERT INTO jobs (class_name, args, status, created_at, run_at) VALUES ('X', '[]', 'pending', 1, 1)") + }.to raise_error(SQLite3::Exception) + end + end + + describe '#overview' do + it 'returns counts and data_version' do + seed_pending(2) + seed_done(3) + seed_failed(1) + result = snapshot.overview + expect(result[:counts]).to include(pending: 2, done: 3, failed: 1) + expect(result[:data_version]).to be_a(Integer) + expect(result[:generated_at]).to be_a(Numeric) + end + + it 'reports executing vs claimed separately' do + seed_executing(2) + seed_claimed(3) + result = snapshot.overview + expect(result[:counts][:executing]).to eq(2) + expect(result[:counts][:claimed]).to eq(3) + end + + it 'returns next_pending_run_at' do + seed_pending(2, base_time: 1_700_000_000.0) + expect(snapshot.overview[:next_pending_run_at]).to eq(1_700_000_000.0) + end + end + + describe 'counts cache' do + let(:snapshot) { described_class.new(path: db_path, counts_cache_ttl: 60).open! } + + it 'reuses cached counts within ttl' do + seed_done(2) + first = snapshot.overview + seed_done(3) + second = snapshot.overview + expect(second).to eq(first) + end + + it 'can bypass the cache for a committed-event refresh' do + seed_done(2) + snapshot.overview + seed_done(3) + + expect(snapshot.overview(force: true)[:counts][:done]).to eq(5) + end + end + + describe '#recent_done' do + it 'returns jobs in finished_at DESC order' do + seed_done(5) + rows = snapshot.recent_done(limit: 3) + finished = rows.map { |r| r[:finished_at] } + expect(finished).to eq(finished.sort.reverse) + expect(rows.length).to eq(3) + end + + it 'cursor pagination yields each row exactly once with no duplicates or gaps' do + seed_done(10, base_time: 1_700_000_000.0) + all_ids = [] + cursor = nil + 4.times do + page = snapshot.recent_done(limit: 3, cursor: cursor) + ids = page.map { |r| r[:id] } + all_ids.concat(ids) + break if page.empty? + + last = page.last + cursor = { finished_at: last[:finished_at], id: last[:id] } + end + expect(all_ids.uniq.length).to eq(all_ids.length) + expect(all_ids.length).to eq(10) + end + + it 'cursor pagination is stable when many rows share the same finished_at' do + same_time = 1_700_000_000.0 + 6.times { |i| store.enqueue("Same#{i}", [i], same_time - 100) } + 6.times do + job = store.fetch(1) + store.complete(job[:id], claim_token: job[:claim_token], finished_at: same_time, duration_ms: 1) + end + all_ids = [] + cursor = nil + 3.times do + page = snapshot.recent_done(limit: 2, cursor: cursor) + break if page.empty? + + all_ids.concat(page.map { |r| r[:id] }) + last = page.last + cursor = { finished_at: last[:finished_at], id: last[:id] } + end + expect(all_ids.uniq.length).to eq(all_ids.length) + expect(all_ids.length).to eq(6) + end + end + + describe '#recent_failed' do + it 'carries last_error_class and last_error_message' do + seed_failed(1) + rows = snapshot.recent_failed(limit: 1) + expect(rows.first[:last_error_class]).to eq('StandardError') + expect(rows.first[:last_error_message]).to include('boom 0') + end + end + + describe '#executing' do + it 'returns only running rows with started_at set' do + seed_executing(2) + seed_claimed(3) + rows = snapshot.executing(limit: 10) + expect(rows.length).to eq(2) + expect(rows).to all(satisfy { |r| !r[:started_at].nil? }) + end + end + + describe '#claimed' do + it 'returns only running rows with started_at NULL' do + seed_executing(2) + seed_claimed(3) + rows = snapshot.claimed(limit: 10) + expect(rows.length).to eq(3) + end + end + + describe '#pending' do + it 'returns pending rows ordered by run_at asc' do + seed_pending(5, base_time: 1_700_000_000.0) + rows = snapshot.pending(limit: 10) + run_ats = rows.map { |r| r[:run_at] } + expect(run_ats).to eq(run_ats.sort) + end + end + + describe '#data_version' do + it 'returns an integer' do + expect(snapshot.data_version).to be_a(Integer) + end + end + + describe 'read errors' do + it 'raises a typed error after close' do + snapshot.close + expect { snapshot.overview }.to raise_error(Async::Background::Web::ClosedError, /closed/) + end + end + + describe '#close' do + it 'is idempotent' do + snapshot.close + expect { snapshot.close }.not_to raise_error + end + end +end diff --git a/spec/async/background/web/stream_spec.rb b/spec/async/background/web/stream_spec.rb new file mode 100644 index 0000000..55494e4 --- /dev/null +++ b/spec/async/background/web/stream_spec.rb @@ -0,0 +1,55 @@ +# frozen_string_literal: true + +require 'spec_helper' +require 'async/background/web' + +RSpec.describe Async::Background::Web::Stream do + let(:subscription) { instance_double(Async::Background::Web::EventHub::Subscription) } + let(:hub) { instance_double(Async::Background::Web::EventHub) } + + before do + allow(hub).to receive(:subscribe).and_return([subscription, "event: overview\ndata: {\"data_version\":1}\n\n"]) + allow(hub).to receive(:unsubscribe) + end + + it 'starts with a retry directive and an authoritative overview' do + allow(subscription).to receive(:pop).and_return(nil) + allow(subscription).to receive(:closed?).and_return(true) + stream = described_class.new(hub, heartbeat_seconds: 30, retry_ms: 5000) + frames = [] + + stream.each { |frame| frames << frame } + + expect(frames).to eq( + [ + "retry: 5000\n\n", + "event: overview\ndata: {\"data_version\":1}\n\n" + ] + ) + expect(hub).to have_received(:unsubscribe).with(subscription) + end + + it 'sends a heartbeat while the connection is idle' do + allow(subscription).to receive(:pop).and_return(nil) + allow(subscription).to receive(:closed?).and_return(false) + stream = described_class.new(hub, heartbeat_seconds: 30, retry_ms: 5000) + frames = [] + + stream.each do |frame| + frames << frame + raise StopIteration if frames.length == 3 + end + rescue StopIteration + expect(frames.last).to eq(Async::Background::Web::EventHub::HEARTBEAT_FRAME) + end + + it 'exits cleanly when the client disconnects' do + allow(subscription).to receive(:pop).and_return('event: overview\ndata: {}\n\n') + allow(subscription).to receive(:closed?).and_return(false) + stream = described_class.new(hub, heartbeat_seconds: 30, retry_ms: 5000) + + expect { + stream.each { |_frame| raise Errno::EPIPE } + }.not_to raise_error + end +end