diff --git a/.sqlx/query-26e3cc708d766ff096f19c7af4537d82b9d5664f6f43478ec6ab10b591a591b0.json b/.sqlx/query-26e3cc708d766ff096f19c7af4537d82b9d5664f6f43478ec6ab10b591a591b0.json new file mode 100644 index 0000000..97bbc1f --- /dev/null +++ b/.sqlx/query-26e3cc708d766ff096f19c7af4537d82b9d5664f6f43478ec6ab10b591a591b0.json @@ -0,0 +1,28 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT id AS \"id!\", attempt AS \"attempt!\" FROM queue.event_deliveries\n WHERE endpoint = $1 ORDER BY id DESC LIMIT 1", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "id!", + "type_info": "Int8" + }, + { + "ordinal": 1, + "name": "attempt!", + "type_info": "Int4" + } + ], + "parameters": { + "Left": [ + "Text" + ] + }, + "nullable": [ + false, + false + ] + }, + "hash": "26e3cc708d766ff096f19c7af4537d82b9d5664f6f43478ec6ab10b591a591b0" +} diff --git a/.sqlx/query-2e9b41722e7cc1dcdf7068a7b3900412e40fc9aee82e1913a4d881c0ccdb629a.json b/.sqlx/query-2e9b41722e7cc1dcdf7068a7b3900412e40fc9aee82e1913a4d881c0ccdb629a.json new file mode 100644 index 0000000..557a738 --- /dev/null +++ b/.sqlx/query-2e9b41722e7cc1dcdf7068a7b3900412e40fc9aee82e1913a4d881c0ccdb629a.json @@ -0,0 +1,20 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT EXTRACT(EPOCH FROM (MIN(next_attempt_at) - now()))::float8 AS \"secs\"\n FROM queue.event_deliveries\n WHERE attempt < max_attempts", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "secs", + "type_info": "Float8" + } + ], + "parameters": { + "Left": [] + }, + "nullable": [ + null + ] + }, + "hash": "2e9b41722e7cc1dcdf7068a7b3900412e40fc9aee82e1913a4d881c0ccdb629a" +} diff --git a/.sqlx/query-a4981444643a40ea7b2c8359335a041ebbd1650888b17af984dddb5ccc0fa95c.json b/.sqlx/query-a4981444643a40ea7b2c8359335a041ebbd1650888b17af984dddb5ccc0fa95c.json new file mode 100644 index 0000000..0a57287 --- /dev/null +++ b/.sqlx/query-a4981444643a40ea7b2c8359335a041ebbd1650888b17af984dddb5ccc0fa95c.json @@ -0,0 +1,20 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT EXTRACT(EPOCH FROM (MIN(next_fire_at) - now()))::float8 AS \"secs\"\n FROM queue.schedule\n WHERE status = 'active'", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "secs", + "type_info": "Float8" + } + ], + "parameters": { + "Left": [] + }, + "nullable": [ + null + ] + }, + "hash": "a4981444643a40ea7b2c8359335a041ebbd1650888b17af984dddb5ccc0fa95c" +} diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index 37b4aad..afe4ddb 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -194,7 +194,7 @@ The signature follows the SNS v2 spec: alphabetically sorted field name/value pa Both steps happen inside the same HTTP handler (`src/ops/event.rs`). The SQS fan-out is synchronous; HTTP deliveries are asynchronous (picked up by the delivery worker). -The delivery worker (`src/ops/delivery.rs`) polls `http_deliveries` in a background task. Each poll opens a transaction, fetches pending rows with `FOR UPDATE SKIP LOCKED`, POSTs to each endpoint, and deletes on success or updates `attempt` / `next_attempt_at` on failure. Backoff schedule: 10s → 30s → 60s → 300s. Exhausted rows (`attempt >= max_attempts`) stay for inspection; no automatic reaping. +The delivery worker (`src/ops/delivery.rs`) is event-driven. Each cycle it opens a transaction, fetches pending rows with `FOR UPDATE SKIP LOCKED`, POSTs to each endpoint, and deletes on success or updates `attempt` / `next_attempt_at` on failure. When a batch comes back empty it sleeps until the earliest pending row's `next_attempt_at` (or parks if the table is empty), woken early in-process when a publish inserts new deliveries — so an idle queue issues no queries and can scale to zero. Backoff schedule: 10s → 30s → 60s → 300s. Rows that exhaust all attempts (`attempt >= max_attempts`) are swept at the end of a delivery batch — deleted, logged, and counted via `delivery_exhausted_total` — so dead letters surface in logs/metrics instead of silently accumulating. **REST API subscriptions** default to `raw_delivery=true` (raw payload). Opt in to the SNS envelope with `"envelope": true` in the subscribe body. **SNS wire protocol subscriptions** default to `raw_delivery=false` (envelope); set `RawMessageDelivery=true` via `SetSubscriptionAttributes` to switch. @@ -205,7 +205,7 @@ Bindings are stored in `queue.topic_subscriptions` with columns `protocol`, `end ### Schedules (time-based triggers) -`queue.schedule` is a single table indexed by `(next_fire_at) WHERE status = 'active'`. The scheduler worker (`src/ops/schedule_worker.rs`) polls it every `QUEUE_SCHEDULE_POLL_MS` (default 1000ms): +`queue.schedule` is a single table indexed by `(next_fire_at) WHERE status = 'active'`. The scheduler worker (`src/ops/schedule_worker.rs`) is event-driven: each cycle it fires everything due, then sleeps until the earliest active schedule's `next_fire_at` (capped at `KEEPALIVE_CAP`, under the instance light-sleep window so the VM stays awake while schedules are active and the fire lands on time), woken early in-process by any `/schedules` mutation. With no active schedules it parks and issues no queries, so the queue and Postgres scale to zero. Each firing cycle: 1. Open transaction, claim up to `QUEUE_SCHEDULE_BATCH_SIZE` due rows with `FOR UPDATE SKIP LOCKED ORDER BY next_fire_at`. 2. For each row, open a `SAVEPOINT`, then dispatch by `target_kind`: @@ -220,7 +220,7 @@ Every fired message carries `headers._schedule = { name, scheduled_for, out_of_b Schedule expressions accept four shapes (`cron`, `every`, `when`, `fire_at`); the parser layer (`src/schedule/`) canonicalizes each to either a cron string + timezone or a single UTC instant before storage. Dry-runs via `POST /v1/previews` round-trip the canonical form, a `human_readable` description, and the next N fire times — the agent-affordance that catches the "I wrote `0 9 *` and it doesn't do what I think" class of mistake before commit. -The worker is always on and trivially cheap when idle: an empty partial index probe per poll cycle, sub-millisecond. +The worker runs for the process lifetime but does no work when idle: with no active schedules it parks on an in-process notification and issues zero queries, letting the VM sleep. `QUEUE_SCHEDULE_POLL_MS` is now only the retry backoff after a transient error, not a steady-state poll interval. ### Write coalescer @@ -289,27 +289,27 @@ Within the selected group, messages are delivered in `msg_id ASC` order (FIFO). ## Configuration -| Variable | Default | What It Controls | -| ---------------------------------- | ------------------------ | -------------------------------------------------------------------------------------------------------------------------------- | -| `DATABASE_URL` | (required) | PostgreSQL connection string passed to sqlx `PgPoolOptions`. | -| `QUEUE_ADDRESS` | `0.0.0.0:9324` | TCP bind address for the HTTP server. | -| `QUEUE_DEFAULT_VISIBILITY_TIMEOUT` | `30` | Seconds applied when a `ReceiveMessage` request omits `VisibilityTimeout`. | -| `QUEUE_MAX_CONNECTIONS` | `10` | Hard cap on the sqlx connection pool. Excess operations wait for a free slot. | -| `LOG_LEVEL` | `info` | `EnvFilter` directive (e.g. `beyond_queue=debug,info`). JSON-structured output. | -| `OTLP_ENABLED` | `false` | Enable OpenTelemetry OTLP trace export over gRPC. | -| `OTLP_ENDPOINT` | `http://localhost:4317` | gRPC OTLP collector. Used when `OTLP_ENABLED=true`. | -| `OTLP_SAMPLE_RATE` | `0.1` | Fraction of traces sampled (0.0 = never, 1.0 = always). Only effective when `OTLP_ENABLED=true`. | -| `QUEUE_LINGER_MS` | `0` | Write coalescer window (ms). Non-FIFO sends are held up to this duration and flushed as a single batch. `0` disables coalescing. | -| `QUEUE_BASE_URL` | `http://{QUEUE_ADDRESS}` | Base URL for SQS queue URLs returned to clients (`{QUEUE_BASE_URL}/000000000000/{name}`). Override when behind a proxy. | -| `QUEUE_HTTP_DELIVERY_ENABLED` | `true` | Enable the background HTTP/HTTPS delivery worker. | -| `QUEUE_HTTP_DELIVERY_POLL_MS` | `1000` | Delivery worker poll interval (ms). Lower values increase responsiveness at the cost of idle DB load. | -| `QUEUE_HTTP_DELIVERY_TIMEOUT_SECS` | `5` | Per-request timeout for outbound webhook POSTs. | -| `QUEUE_HTTP_DELIVERY_BATCH_SIZE` | `50` | Maximum rows the delivery worker claims per poll cycle. Tune up under high SNS fanout load. | -| `QUEUE_SCHEDULE_ENABLED` | `true` | Enable the background schedule worker (cron / every / when triggers). | -| `QUEUE_SCHEDULE_POLL_MS` | `1000` | Schedule worker poll interval (ms). Floor on fire latency; idle cost is one sub-millisecond partial-index probe per cycle. | -| `QUEUE_SCHEDULE_BATCH_SIZE` | `32` | Maximum rows the schedule worker claims per poll cycle. | -| `QUEUE_SCHEDULE_PREVIEW_COUNT` | `5` | Number of upcoming fire timestamps projected in `next_fires` on schedule and preview responses. | -| `QUEUE_SCHEDULE_LIST_MAX` | `1000` | Hard cap on `GET /v1/schedules` response size. | +| Variable | Default | What It Controls | +| ---------------------------------- | ------------------------ | ----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | +| `DATABASE_URL` | (required) | PostgreSQL connection string passed to sqlx `PgPoolOptions`. | +| `QUEUE_ADDRESS` | `0.0.0.0:9324` | TCP bind address for the HTTP server. | +| `QUEUE_DEFAULT_VISIBILITY_TIMEOUT` | `30` | Seconds applied when a `ReceiveMessage` request omits `VisibilityTimeout`. | +| `QUEUE_MAX_CONNECTIONS` | `10` | Hard cap on the sqlx connection pool. Excess operations wait for a free slot. | +| `LOG_LEVEL` | `info` | `EnvFilter` directive (e.g. `beyond_queue=debug,info`). JSON-structured output. | +| `OTLP_ENABLED` | `false` | Enable OpenTelemetry OTLP trace export over gRPC. | +| `OTLP_ENDPOINT` | `http://localhost:4317` | gRPC OTLP collector. Used when `OTLP_ENABLED=true`. | +| `OTLP_SAMPLE_RATE` | `0.1` | Fraction of traces sampled (0.0 = never, 1.0 = always). Only effective when `OTLP_ENABLED=true`. | +| `QUEUE_LINGER_MS` | `0` | Write coalescer window (ms). Non-FIFO sends are held up to this duration and flushed as a single batch. `0` disables coalescing. | +| `QUEUE_BASE_URL` | `http://{QUEUE_ADDRESS}` | Base URL for SQS queue URLs returned to clients (`{QUEUE_BASE_URL}/000000000000/{name}`). Override when behind a proxy. | +| `QUEUE_HTTP_DELIVERY_ENABLED` | `true` | Enable the background HTTP/HTTPS delivery worker. | +| `QUEUE_HTTP_DELIVERY_POLL_MS` | `1000` | Retry backoff (ms) after a transient delivery-batch error. Not a steady-state poll: the worker is event-driven (sleeps until the next `next_attempt_at`, woken in-process on publish). | +| `QUEUE_HTTP_DELIVERY_TIMEOUT_SECS` | `5` | Per-request timeout for outbound webhook POSTs. | +| `QUEUE_HTTP_DELIVERY_BATCH_SIZE` | `50` | Maximum rows the delivery worker claims per poll cycle. Tune up under high SNS fanout load. | +| `QUEUE_SCHEDULE_ENABLED` | `true` | Enable the background schedule worker (cron / every / when triggers). | +| `QUEUE_SCHEDULE_POLL_MS` | `1000` | Retry backoff (ms) after a transient schedule-poll error. Not a steady-state poll: the worker is event-driven (sleeps until the next `next_fire_at`, capped under the light-sleep window, woken in-process on `/schedules` writes). | +| `QUEUE_SCHEDULE_BATCH_SIZE` | `32` | Maximum rows the schedule worker claims per poll cycle. | +| `QUEUE_SCHEDULE_PREVIEW_COUNT` | `5` | Number of upcoming fire timestamps projected in `next_fires` on schedule and preview responses. | +| `QUEUE_SCHEDULE_LIST_MAX` | `1000` | Hard cap on `GET /v1/schedules` response size. | --- diff --git a/SCHEDULES.md b/SCHEDULES.md index 5372fa2..70451a2 100644 --- a/SCHEDULES.md +++ b/SCHEDULES.md @@ -23,9 +23,10 @@ already uses. One new table, zero new pgrx functions, one new SDK verb. built. - **Forks with the rest of the platform.** Schedules live in user Postgres, on the user's GlideFS volume. -- **Trivial idle cost.** The scheduler worker is always on but polls a - partial index `WHERE status = 'active'`; an empty schedule table costs - one sub-millisecond index probe per poll cycle. +- **Zero idle cost.** The scheduler worker is event-driven: with no active + schedules it parks and issues no queries at all, so the queue (and Postgres) + scale to zero. With schedules active it sleeps until the next `next_fire_at` + (a partial-index probe on `WHERE status = 'active'`), not on a fixed poll. - The minimum effective surface. One table, one worker, three expression forms, three target kinds. @@ -57,33 +58,44 @@ into a workflow is the same fanout a topic does. ## The runtime model -The queue server is platform infrastructure. It runs alongside the user's -Postgres on their GlideFS volume and does not scale to zero — same as the -database. Users don't pay for it any more than they pay for Postgres being up. +The queue server scales to zero with the app. The platform sleeps an idle VM +(no network traffic for ~5 min → light sleep → deep sleep), and the queue +server is built so an idle app generates no traffic: the schedule worker only +talks to Postgres when there is work to do (see _Worker lifecycle_). When no +schedules are active and no deliveries are pending, the queue VM goes idle and +sleeps — and because the queue is the only thing querying it, Postgres sleeps +too. + +While any schedule **is** active, the worker re-checks Postgres at least every +~`KEEPALIVE_CAP` (kept under the light-sleep window). Those checks are network +traffic to Postgres, so both VMs stay awake and the fire lands on time — the +keepalive is just the re-check the worker has to do anyway, not a separate +mechanism. So: idle app → both sleep; app with schedules → both awake until the +schedules are paused/deleted. Function VMs — user code — sleep when idle. When a schedule fires, the scheduler worker (running in the queue server) calls `queue.send` or `queue.publish_event`. The fan-out sends HTTP to the function VM's endpoint. The gateway sees a sleeping VM, wakes it, delivers the request. The function -processes it and goes back to sleep. - -No special wakeup mechanism. No external scheduler. HTTP fan-out handles -sleeping function VMs as a natural consequence of the existing wake-on-traffic -machinery — the same path any other inbound request takes. +processes it and goes back to sleep — the same wake-on-traffic path any other +inbound request takes. ### Worker lifecycle -The scheduler worker starts at queue server boot and runs for the lifetime -of the process. Its idle cost is one partial-index probe per poll cycle — -sub-millisecond on an empty `schedule_due_idx` — so chasing "zero overhead" -by stopping the worker when no schedules exist would add intra-process -signaling complexity without measurable benefit. Same pattern as the HTTP -delivery worker. +The scheduler worker starts at queue server boot and runs for the lifetime of +the process, but it is **event-driven, not a fixed poll**. Each cycle it fires +everything currently due, then sleeps until the earliest active schedule's +`next_fire_at` (capped at `KEEPALIVE_CAP`), woken early in-process when a +`/schedules` mutation changes the table. When there are no active schedules it +parks entirely — zero queries, zero traffic — so the VM can sleep. The HTTP +delivery worker follows the same shape, keyed on `event_deliveries`. On boot, the worker runs one catchup pass (firing any missed occurrences for active schedules subject to each schedule's `catchup` and -`catchup_limit` settings) before entering the normal poll loop. This -covers fires missed while the queue server was down. +`catchup_limit` settings) before entering the normal loop. This covers fires +missed while the queue server was down or asleep. (Light sleep is a +pause/resume, not a restart, so it does not re-run catchup; while schedules are +active the VM stays awake anyway, so no fire is missed.) --- diff --git a/src/db.rs b/src/db.rs index 0b9f6a1..22d63f9 100644 --- a/src/db.rs +++ b/src/db.rs @@ -8,6 +8,11 @@ const CONNECT_RETRY_BUDGET: Duration = Duration::from_secs(60); /// First backoff after a failed connect; doubles up to [`CONNECT_BACKOFF_MAX`]. const CONNECT_BACKOFF_START: Duration = Duration::from_millis(250); const CONNECT_BACKOFF_MAX: Duration = Duration::from_secs(3); +/// Pool acquire timeout. Sized to tolerate Postgres waking from deep sleep +/// (re-export of its volume from S3 + boot can take seconds): when the whole app +/// has been idle and Postgres has scaled to zero, the queue's first query holds +/// while Postgres wakes — via eBPF wake-on-traffic — rather than failing fast. +const ACQUIRE_TIMEOUT: Duration = Duration::from_secs(30); /// Connect to postgres, retrying with capped exponential backoff while it's /// unreachable. @@ -23,7 +28,7 @@ pub async fn connect(database_url: &str, max_connections: u32) -> anyhow::Result loop { let attempt = PgPoolOptions::new() .max_connections(max_connections) - .acquire_timeout(Duration::from_secs(5)) + .acquire_timeout(ACQUIRE_TIMEOUT) .connect(database_url) .await; match attempt { diff --git a/src/handoff.rs b/src/handoff.rs index 8ef4d48..f889cb9 100644 --- a/src/handoff.rs +++ b/src/handoff.rs @@ -63,6 +63,8 @@ pub struct Rebuild { pub signer: Arc, pub base_url: Arc, pub metrics: Arc, + pub delivery_notify: Arc, + pub schedule_notify: Arc, } impl Rebuild { @@ -89,6 +91,8 @@ impl Rebuild { coalescer, signer: self.signer.clone(), metrics: self.metrics.clone(), + delivery_notify: self.delivery_notify.clone(), + schedule_notify: self.schedule_notify.clone(), }; (state, coalescer_jh) } diff --git a/src/lib.rs b/src/lib.rs index 64665ea..49dafe4 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -56,6 +56,13 @@ pub struct AppState { /// RSA-2048 signer for SNS notification envelopes. pub signer: Arc, pub metrics: Arc, + /// In-process wakeup for the HTTP-delivery worker. Poked after a publish + /// inserts `event_deliveries` rows so the worker re-checks immediately + /// instead of waiting out its sleep-until-next-due timer. + pub delivery_notify: Arc, + /// In-process wakeup for the schedule worker. Poked after any `/schedules` + /// mutation so the worker recomputes its next-fire deadline immediately. + pub schedule_notify: Arc, } /// Parse an AWS service request body: returns `Ok((is_json, action_name, parsed_body))`. @@ -163,6 +170,13 @@ pub async fn serve(config: Config) -> anyhow::Result<()> { let pool = db::connect(&config.database_url, config.max_connections).await?; let metrics = Arc::new(Metrics::new()); + // In-process wakeups: route handlers poke these after writing the tables the + // workers watch, so the workers re-check immediately instead of polling. + // Primitives run single-instance (max=1), so one process = one set of + // workers — no cross-replica notification needed. + let delivery_notify = Arc::new(tokio::sync::Notify::new()); + let schedule_notify = Arc::new(tokio::sync::Notify::new()); + let delivery_handle = if config.http_delivery_enabled { tracing::info!("HTTP delivery worker enabled"); Some(ops::delivery::start( @@ -173,6 +187,7 @@ pub async fn serve(config: Config) -> anyhow::Result<()> { batch_size: config.http_delivery_batch_size, }, metrics.clone(), + delivery_notify.clone(), )?) } else { None @@ -187,13 +202,13 @@ pub async fn serve(config: Config) -> anyhow::Result<()> { batch_size: config.schedule_batch_size, }, metrics.clone(), + schedule_notify.clone(), + delivery_notify.clone(), )) } else { None }; - let scrape_handle = start_queue_depth_scrape(pool.clone(), metrics.clone()); - // 5. TLS validation (up-front; misconfig surfaces immediately). let tls_parts = match (&config.tls_cert, &config.tls_key, &config.tls_ca) { (Some(c), Some(k), Some(ca)) => Some(handoff::TlsParts { @@ -219,6 +234,8 @@ pub async fn serve(config: Config) -> anyhow::Result<()> { signer: signer.clone(), base_url: base_url.clone(), metrics: metrics.clone(), + delivery_notify: delivery_notify.clone(), + schedule_notify: schedule_notify.clone(), }; let rt = Handle::current(); let (initial_state, coalescer_handle) = rebuild.build_state(&rt); @@ -371,8 +388,6 @@ pub async fn serve(config: Config) -> anyhow::Result<()> { jh.abort(); let _ = jh.await; } - scrape_handle.abort(); - let _ = scrape_handle.await; tracing::info!("shutdown complete"); Ok(()) @@ -658,6 +673,28 @@ async fn record_metrics(State(state): State, req: Request, next: Next) } async fn metrics_handler(State(state): State) -> impl IntoResponse { + // Refresh queue-depth gauges on demand rather than from a background poll. + // A background scrape loop would generate continuous queue→Postgres traffic + // and pin both VMs awake; computing here means an idle (unscraped) VM stays + // silent and can scale to zero. Best-effort: if Postgres is unreachable, + // serve the rest of the metrics anyway. + match ops::queue_admin::all_queue_depths(&state.pool).await { + Ok(snapshots) => { + for s in snapshots { + state + .metrics + .queue_depth + .with_label_values(&[&s.queue_name]) + .set(s.visible as f64); + state + .metrics + .queue_in_flight + .with_label_values(&[&s.queue_name]) + .set(s.in_flight as f64); + } + } + Err(e) => tracing::warn!(error = %e, "on-demand queue depth refresh failed"), + } ( StatusCode::OK, [( @@ -723,26 +760,3 @@ async fn healthz_ready(State(state): State) -> impl IntoResponse { .into_response() } } - -fn start_queue_depth_scrape(pool: PgPool, metrics: Arc) -> tokio::task::JoinHandle<()> { - tokio::spawn(async move { - loop { - match ops::queue_admin::all_queue_depths(&pool).await { - Ok(snapshots) => { - for s in snapshots { - metrics - .queue_depth - .with_label_values(&[&s.queue_name]) - .set(s.visible as f64); - metrics - .queue_in_flight - .with_label_values(&[&s.queue_name]) - .set(s.in_flight as f64); - } - } - Err(e) => tracing::warn!(error = %e, "queue depth scrape failed"), - } - tokio::time::sleep(Duration::from_secs(15)).await; - } - }) -} diff --git a/src/ops/delivery.rs b/src/ops/delivery.rs index be8b0a2..9783705 100644 --- a/src/ops/delivery.rs +++ b/src/ops/delivery.rs @@ -4,11 +4,18 @@ use std::time::{Duration, Instant}; use chrono::Utc; use reqwest::Client; use sqlx::PgPool; +use tokio::sync::Notify; use tokio::task::JoinHandle; use tracing::Instrument as _; use crate::metrics::Metrics; +/// When no delivery is pending, park this long before a defensive re-probe. New +/// deliveries are signalled in-process via `notify`, so this is only a backstop; +/// it is deliberately longer than the instance light-sleep window so an idle VM +/// sleeps in the gap rather than waking itself to poll. +const IDLE_PARK: Duration = Duration::from_secs(3600); + pub struct DeliveryConfig { pub poll_interval_ms: u64, pub delivery_timeout_secs: u64, @@ -29,28 +36,71 @@ pub fn start( pool: PgPool, config: DeliveryConfig, metrics: Arc, + notify: Arc, ) -> anyhow::Result> { let client = Client::builder() .timeout(Duration::from_secs(config.delivery_timeout_secs)) .build()?; - Ok(tokio::spawn(run(pool, client, config, metrics))) + Ok(tokio::spawn(run(pool, client, config, metrics, notify))) } -async fn run(pool: PgPool, client: Client, config: DeliveryConfig, metrics: Arc) { +/// Event-driven delivery loop. Drains all due deliveries, then sleeps until the +/// earliest pending retry (or parks if none) — woken early by `notify` when a +/// publish inserts new deliveries. When the table is empty it generates no +/// traffic, letting the VM (and Postgres) scale to zero. +async fn run( + pool: PgPool, + client: Client, + config: DeliveryConfig, + metrics: Arc, + notify: Arc, +) { loop { match deliver_batch(&pool, &client, &config, &metrics).await { - Ok(0) => { - tokio::time::sleep(Duration::from_millis(config.poll_interval_ms)).await; - } + // A full-ish batch may mean more is due right now — loop immediately. + Ok(n) if n > 0 => continue, + // Nothing due — fall through to wait for the next due time / a poke. Ok(_) => {} Err(e) => { tracing::error!(error = %e, "http delivery batch error"); - tokio::time::sleep(Duration::from_millis(config.poll_interval_ms)).await; + wait_or_notified(Duration::from_millis(config.poll_interval_ms), ¬ify).await; + continue; } } + + let wait = match earliest_pending_in(&pool).await { + Ok(Some(secs)) => Duration::from_secs_f64(secs.max(0.0)), + Ok(None) => IDLE_PARK, + Err(e) => { + tracing::warn!(error = %e, "delivery next-due probe failed"); + Duration::from_millis(config.poll_interval_ms) + } + }; + wait_or_notified(wait, ¬ify).await; + } +} + +/// Sleep for `dur`, returning early if `notify` is poked. +async fn wait_or_notified(dur: Duration, notify: &Notify) { + tokio::select! { + _ = tokio::time::sleep(dur) => {} + _ = notify.notified() => {} } } +/// Seconds until the earliest deliverable row becomes due (`None` if there are +/// no rows with attempts remaining). Negative values mean already-due. +async fn earliest_pending_in(pool: &PgPool) -> anyhow::Result> { + let row = sqlx::query!( + r#"SELECT EXTRACT(EPOCH FROM (MIN(next_attempt_at) - now()))::float8 AS "secs" + FROM queue.event_deliveries + WHERE attempt < max_attempts"# + ) + .fetch_one(pool) + .await?; + Ok(row.secs) +} + async fn deliver_batch( pool: &PgPool, client: &Client, diff --git a/src/ops/schedule_worker.rs b/src/ops/schedule_worker.rs index e78b131..9804308 100644 --- a/src/ops/schedule_worker.rs +++ b/src/ops/schedule_worker.rs @@ -1,8 +1,12 @@ -//! Schedule worker — polls `queue.schedule` for due rows and fires them. +//! Schedule worker — fires due rows in `queue.schedule`. //! -//! Mirrors `ops::delivery` in shape: pure polling loop, no LISTEN/NOTIFY, -//! configured via env, gracefully aborted on shutdown. Always on; an -//! empty schedule table costs one partial-index probe per poll. +//! Event-driven: it fires everything currently due, then sleeps until the +//! earliest active schedule's `next_fire_at` (capped — see [`KEEPALIVE_CAP`]), +//! woken early in-process via `notify` when a `/schedules` mutation changes the +//! table. When there are no active schedules it parks and generates no traffic, +//! letting the VM (and Postgres) scale to zero. When schedules *do* exist, the +//! periodic re-check is itself a query to Postgres, which keeps both VMs awake +//! so fires land on time — no separate keepalive mechanism. //! //! Per-row failure isolation via `SAVEPOINT`: an individual dispatch //! failure (target queue missing, malformed payload) rolls back only @@ -17,12 +21,31 @@ use std::time::Duration; use chrono::{DateTime, Utc}; use sqlx::{PgPool, Postgres, Transaction}; +use tokio::sync::Notify; use tokio::task::JoinHandle; use crate::metrics::Metrics; use crate::ops::schedule::{self, ScheduleRow}; use crate::schedule::expression::{Canonical, Expression}; +/// Upper bound on how long the worker sleeps while any active schedule exists. +/// Kept safely under the instance light-sleep window (300s for primitives), so a +/// far-off `next_fire_at` still produces a re-check every ~cap — traffic that +/// keeps the queue and Postgres VMs awake so the fire lands on time. +const KEEPALIVE_CAP: Duration = Duration::from_secs(240); + +/// When no schedule is active, park this long before a defensive re-probe. +/// Mutations are signalled in-process via `notify`, so this is only a backstop; +/// longer than the light-sleep window so an idle VM sleeps in the gap. +const IDLE_PARK: Duration = Duration::from_secs(3600); + +/// Floor on the sleep between fire cycles. A fire that *fails* does not advance +/// the row's `next_fire_at`, so the row stays due and would otherwise be +/// re-claimed with zero delay — a hot loop that hammers the pool until the +/// schedule hits `failure_threshold`. Flooring the re-check rate-limits that +/// (and back-to-back catch-up batches) to a sane cadence. +const MIN_RECHECK: Duration = Duration::from_millis(250); + pub struct ScheduleWorkerConfig { pub poll_interval_ms: u64, pub batch_size: i64, @@ -37,27 +60,72 @@ impl Default for ScheduleWorkerConfig { } } -pub fn start(pool: PgPool, config: ScheduleWorkerConfig, _metrics: Arc) -> JoinHandle<()> { - tokio::spawn(run(pool, config)) +pub fn start( + pool: PgPool, + config: ScheduleWorkerConfig, + _metrics: Arc, + notify: Arc, + delivery_notify: Arc, +) -> JoinHandle<()> { + tokio::spawn(run(pool, config, notify, delivery_notify)) } -async fn run(pool: PgPool, config: ScheduleWorkerConfig) { +async fn run( + pool: PgPool, + config: ScheduleWorkerConfig, + notify: Arc, + delivery_notify: Arc, +) { loop { match poll_once(&pool, &config).await { - Ok(fired) => { - if fired == 0 { - tokio::time::sleep(Duration::from_millis(config.poll_interval_ms)).await; - } - // If we fired any, loop immediately — there may be more. - } + // A topic target may have created deliveries — wake that worker. + Ok(fired) if fired > 0 => delivery_notify.notify_one(), + Ok(_) => {} Err(e) => { tracing::error!(error = %e, "schedule worker poll failed"); - tokio::time::sleep(Duration::from_millis(config.poll_interval_ms)).await; + wait_or_notified(Duration::from_millis(config.poll_interval_ms), ¬ify).await; + continue; } } + + // Sleep until the next fire. A still-due row (more catch-up work, or a + // failed fire that didn't advance) reports ~0s; the MIN_RECHECK floor + // turns that into a paced retry instead of a hot loop. + let wait = match earliest_active_fire_in(&pool).await { + Ok(Some(secs)) => { + Duration::from_secs_f64(secs.max(0.0)).clamp(MIN_RECHECK, KEEPALIVE_CAP) + } + Ok(None) => IDLE_PARK, + Err(e) => { + tracing::warn!(error = %e, "schedule next-due probe failed"); + Duration::from_millis(config.poll_interval_ms) + } + }; + wait_or_notified(wait, ¬ify).await; } } +/// Sleep for `dur`, returning early if `notify` is poked. +async fn wait_or_notified(dur: Duration, notify: &Notify) { + tokio::select! { + _ = tokio::time::sleep(dur) => {} + _ = notify.notified() => {} + } +} + +/// Seconds until the earliest active schedule is due (`None` if no active +/// schedules). Negative values mean already-due. +async fn earliest_active_fire_in(pool: &PgPool) -> anyhow::Result> { + let row = sqlx::query!( + r#"SELECT EXTRACT(EPOCH FROM (MIN(next_fire_at) - now()))::float8 AS "secs" + FROM queue.schedule + WHERE status = 'active'"# + ) + .fetch_one(pool) + .await?; + Ok(row.secs) +} + /// One poll cycle: claim due rows, fire each in a savepoint, commit. /// Returns the number of rows processed. async fn poll_once(pool: &PgPool, config: &ScheduleWorkerConfig) -> anyhow::Result { diff --git a/src/routes/events.rs b/src/routes/events.rs index 2ba51c9..5d8703e 100644 --- a/src/routes/events.rs +++ b/src/routes/events.rs @@ -92,6 +92,10 @@ pub async fn publish_event( event::queue_event_deliveries(&state.pool, &routing_key, &body.message, Some(&envelope)) .await?; + // Wake the delivery worker so it dispatches immediately instead of waiting + // out its sleep-until-next-due timer. + state.delivery_notify.notify_one(); + Ok(( StatusCode::CREATED, Json(TopicSendResponse { diff --git a/src/routes/schedules.rs b/src/routes/schedules.rs index d80c027..4dbaeac 100644 --- a/src/routes/schedules.rs +++ b/src/routes/schedules.rs @@ -39,6 +39,7 @@ pub async fn create_schedule( let preview_count = state.config.schedule_preview_count; let name = spec.name.clone(); let sched = schedule::create(&state.pool, spec, preview_count).await?; + state.schedule_notify.notify_one(); Ok((StatusCode::CREATED, location_header(&name), Json(sched))) } @@ -123,6 +124,7 @@ pub async fn upsert_schedule( state.config.schedule_preview_count, ) .await?; + state.schedule_notify.notify_one(); let status = if created { StatusCode::CREATED } else { @@ -179,6 +181,7 @@ pub async fn patch_schedule( state.config.schedule_preview_count, ) .await?; + state.schedule_notify.notify_one(); return Ok(Json(sched)); } @@ -189,6 +192,7 @@ pub async fn patch_schedule( state.config.schedule_preview_count, ) .await?; + state.schedule_notify.notify_one(); Ok(Json(sched)) } @@ -210,6 +214,7 @@ pub async fn delete_schedule( Path(name): Path, ) -> Result { schedule::delete(&state.pool, &name).await?; + state.schedule_notify.notify_one(); Ok(StatusCode::NO_CONTENT) } @@ -234,6 +239,8 @@ pub async fn run_schedule( Path(name): Path, ) -> Result { let run = schedule::run_now(&state.pool, &name).await?; + // A topic target may have created deliveries; wake the delivery worker. + state.delivery_notify.notify_one(); Ok((StatusCode::ACCEPTED, Json(run))) } diff --git a/src/test_support.rs b/src/test_support.rs index faaa86e..97c2e91 100644 --- a/src/test_support.rs +++ b/src/test_support.rs @@ -58,6 +58,8 @@ pub async fn start_with_coalescer(pool: PgPool, linger_ms: u64) -> anyhow::Resul coalescer: Some(coalescer), signer, metrics, + delivery_notify: Arc::new(tokio::sync::Notify::new()), + schedule_notify: Arc::new(tokio::sync::Notify::new()), }; let app = crate::build_router(state); @@ -107,6 +109,11 @@ pub async fn start(pool: PgPool, database_url: String) -> anyhow::Result anyhow::Result anyhow::Result = config.base_url().into(); @@ -137,6 +147,8 @@ pub async fn start(pool: PgPool, database_url: String) -> anyhow::Result(); let sub_id = sub["id"].as_i64().expect("subscription id"); - // Publish — creates an http_deliveries row + // Publish — creates an event_deliveries row; the worker attempts it promptly. client .post( "/v1/events/cancel.me", @@ -120,18 +125,31 @@ async fn test_http_delivery_unsubscribe_cancels_pending() { .await .assert_status(201); - // Unsubscribe — CASCADE deletes http_deliveries rows + // Wait for the first (failing) attempt so the row is now in backoff, pending. + webhook.wait_for(1, std::time::Duration::from_secs(5)).await; + + // Unsubscribe — CASCADE deletes the pending (backing-off) delivery row. client .delete(&format!("/v1/events/cancel.*/subscriptions/{sub_id}")) .await .assert_status(204); - // Give worker a moment to see the empty table — delivery should NOT arrive - tokio::time::sleep(std::time::Duration::from_millis(300)).await; + // The pending row is gone, so the ~10s retry never fires. Scope the count to + // this test's endpoint — the integration DB is shared across tests. + let remaining: i64 = + sqlx::query_scalar("SELECT count(*) FROM queue.event_deliveries WHERE endpoint = $1") + .bind(&webhook.url) + .fetch_one(&env.pool) + .await + .expect("count event_deliveries"); + assert_eq!( + remaining, 0, + "unsubscribe should CASCADE-delete pending deliveries" + ); assert_eq!( webhook.received_count(), - 0, - "delivery should have been cancelled" + 1, + "only the initial failed attempt should have fired; the retry was cancelled" ); } @@ -161,47 +179,69 @@ async fn test_http_delivery_dead_letter() { .await .assert_status(201); - // Wait for the first delivery attempt to be recorded. - tokio::time::sleep(std::time::Duration::from_millis(500)).await; - - // The row should exist with attempt >= 1. - let row = sqlx::query!( - r#"SELECT id AS "id!", attempt AS "attempt!" FROM queue.event_deliveries - WHERE endpoint = $1 ORDER BY id DESC LIMIT 1"#, - webhook.url, - ) - .fetch_optional(&env.pool) - .await - .unwrap() - .expect("expected an http_deliveries row after first attempt"); - assert!( - row.attempt >= 1, - "delivery worker should have attempted at least once" - ); + // Poll until the first failed attempt is recorded (attempt incremented), + // scoped to this test's endpoint (the integration DB is shared across tests). + // wait_for() alone is too early — it fires when the webhook receives the POST, + // before the worker commits the failure. + let mut row_id: Option = None; + for _ in 0..50 { + let r = sqlx::query!( + r#"SELECT id AS "id!", attempt AS "attempt!" FROM queue.event_deliveries + WHERE endpoint = $1 ORDER BY id DESC LIMIT 1"#, + webhook.url, + ) + .fetch_optional(&env.pool) + .await + .unwrap(); + if let Some(r) = r + && r.attempt >= 1 + { + row_id = Some(r.id); + break; + } + tokio::time::sleep(std::time::Duration::from_millis(100)).await; + } + let row_id = row_id.expect("delivery worker should have recorded a failed attempt"); - // Fast-forward: set attempt = max_attempts to simulate exhaustion. + // Fast-forward this row to exhaustion. sqlx::query!( "UPDATE queue.event_deliveries SET attempt = max_attempts WHERE id = $1", - row.id, + row_id, ) .execute(&env.pool) .await .unwrap(); - // Give the worker a couple of poll cycles. - tokio::time::sleep(std::time::Duration::from_millis(200)).await; + // Publish a fresh event to the same endpoint to drive another delivery batch. + // The exhausted-row sweep runs at the end of each batch and discards + // dead-lettered rows (deleted + logged + counted) so they don't accumulate. + client + .post( + "/v1/events/deadletter.test", + &serde_json::json!({ "message": { "fail": true } }), + ) + .await + .assert_status(201); - // Row must still exist — exhausted rows are retained for inspection, not deleted. - let still_there = sqlx::query!( - r#"SELECT id AS "id!" FROM queue.event_deliveries WHERE id = $1"#, - row.id, - ) - .fetch_optional(&env.pool) - .await - .unwrap(); + // The dead-lettered row is swept; poll until it's gone. + let mut swept = false; + for _ in 0..50 { + let still = sqlx::query!( + r#"SELECT id AS "id!" FROM queue.event_deliveries WHERE id = $1"#, + row_id, + ) + .fetch_optional(&env.pool) + .await + .unwrap(); + if still.is_none() { + swept = true; + break; + } + tokio::time::sleep(std::time::Duration::from_millis(100)).await; + } assert!( - still_there.is_some(), - "dead-lettered row must remain for inspection" + swept, + "exhausted (dead-lettered) row should be swept by the delivery worker" ); } diff --git a/tests/integration_test/tls.rs b/tests/integration_test/tls.rs index a19ebe1..022eb63 100644 --- a/tests/integration_test/tls.rs +++ b/tests/integration_test/tls.rs @@ -108,6 +108,8 @@ async fn start_tls_server(certs: &CertBundle) -> String { coalescer: None, signer: Arc::new(beyond_queue::signing::Signer::generate().unwrap()), metrics: Arc::new(beyond_queue::metrics::Metrics::new()), + delivery_notify: Arc::new(tokio::sync::Notify::new()), + schedule_notify: Arc::new(tokio::sync::Notify::new()), }); let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();