diff --git a/CHANGELOG.md b/CHANGELOG.md index 5f8e8a9c..0ce930f2 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,10 @@ Pre-1.0 note: while `pg_durable` is in major version `0`, minor releases may inc ## [0.2.6] - Unreleased +### Added + +- **`df.wait_for_condition(condition, max_check_interval, notify_key)`:** wait until a SQL predicate is true instead of on a cron schedule. The predicate is evaluated as `() IS TRUE` under `default_transaction_read_only`, so ordinary scalar-subquery rules apply and a condition with side effects fails rather than performing them repeatedly. `max_check_interval` is required (one-second floor) and is the guaranteed worst-case latency. An optional `notify_key` lets a producer `pg_notify('pg_durable_condition', key)` to be picked up sooner; the interval remains the guarantee, so a missed notification costs latency, not correctness. + ### Fixed - **Deep workflow composition (#327):** workflow graphs deeper than serde_json's 127-level recursion limit no longer silently collapse into SQL text. Nested children are deserialized one graph level at a time, and `df.explain()` now enforces the configured graph-depth limit before traversal. diff --git a/USER_GUIDE.md b/USER_GUIDE.md index 053be94f..7ba81cb0 100644 --- a/USER_GUIDE.md +++ b/USER_GUIDE.md @@ -53,6 +53,7 @@ pg_durable enables you to define and execute **durable SQL functions** entirely | **Conditional Logic** | Branch with `?>` `!>` operators or `df.if()` | | **Timers & Delays** | Sleep with `df.sleep()` | | **Cron Scheduling** | Schedule with `df.wait_for_schedule()` | +| **Condition Triggers** | Wait for a SQL predicate with `df.wait_for_condition()` | | **Eternal Loops** | Create forever-running jobs with `@>` operator or `df.loop()` | | **Signals** | Wait for external events with `df.wait_for_signal()` | | **Variable Substitution** | Pass results between steps using `$name` | @@ -247,6 +248,7 @@ df.sql('SELECT 1') ~> df.sql('SELECT 2') |----------|-------------|---------| | `df.sleep(seconds)` | Pause for N seconds | `df.sleep(60)` | | `df.wait_for_schedule(cron)` | Wait until cron matches | `df.wait_for_schedule('0 * * * *')` | +| `df.wait_for_condition(condition, max_check_interval, notify_key)` | Wait until a SQL predicate is true | `df.wait_for_condition('SELECT count(*) > 8 FROM segments', '1min')` | | `df.http(url, method, body, headers, timeout)` | Make HTTP request | `df.http('https://api.example.com', 'POST', '{"key": "value"}')` | | `df.join(a, b)` | Execute in parallel, wait for all | `df.join('SELECT 1', 'SELECT 2')` | | `df.join3(a, b, c)` | Three in parallel | `df.join3(a, b, c)` | @@ -403,6 +405,7 @@ SELECT df.result('a1b2c3d4')::jsonb->'rows'->0->>'answer'; - A SQL query returning no rows produces: `{"rows": [], "row_count": 0}` - `df.sleep()` returns a top-level JSON object like `{"slept": true, "seconds": 60}` - `df.wait_for_schedule()` returns a top-level JSON object: `{"scheduled": true}` +- `df.wait_for_condition()` returns a top-level JSON object: `{"condition_met": true}` - `df.http()` and `df.http_multipart()` return a top-level JSON object with `status`, `body`, `encoding`, `headers`, `ok`, and `duration_ms` fields - `df.break('value')` stores the literal value as the loop result (not wrapped in `rows`) @@ -1224,6 +1227,59 @@ SELECT df.start( ); ``` +### Condition-Based Triggering + +`df.wait_for_condition()` blocks until a SQL predicate is true. Use it when the +work is driven by the state of your data rather than by the clock. + +```sql +-- Compact whenever the segment count crosses a threshold. +SELECT df.start( + @> ( + df.wait_for_condition( + 'SELECT count(*) > 8 FROM playground.segments', + '1min' + ) + ~> 'SELECT playground.compact_segments()' + ), + 'segment-compactor' +); +``` + +The predicate is evaluated as `() IS TRUE`, so ordinary scalar +subquery rules apply: it must produce a single boolean column, no rows or NULL +read as false, and more than one row is an error. Use `SELECT EXISTS (...)` for +existence checks. It runs read-only, so a condition with side effects fails +rather than performing them repeatedly. + +`max_check_interval` is required and has a one-second floor. It is the worst +case latency between the condition becoming true and the workflow noticing. + +Add a `notify_key` to be woken sooner. The producer of the data notifies the +fixed channel `pg_durable_condition` with the key as the payload: + +```sql +SELECT df.start( + @> ( + df.wait_for_condition( + 'SELECT count(*) > 8 FROM playground.segments', + '5min', + notify_key => 'segments_changed' + ) + ~> 'SELECT playground.compact_segments()' + ), + 'segment-compactor' +); + +-- In the writer, after committing new segments: +SELECT pg_notify('pg_durable_condition', 'segments_changed'); +``` + +The interval remains the guarantee; the notification is only an accelerator, so +a missed notification costs latency rather than correctness. `notify_key` is +not a permission — any role can send any payload, and the worst that does is +re-evaluate a predicate early. + ### While Loops Use `df.loop(body, condition)` to repeat while a condition is true: diff --git a/docs/spec-wait-for-condition.md b/docs/spec-wait-for-condition.md new file mode 100644 index 00000000..3b112d6c --- /dev/null +++ b/docs/spec-wait-for-condition.md @@ -0,0 +1,374 @@ +# Wait For Condition Specification + +**Status:** Proposal +**Date:** 2026-08-19 +**Target version:** 0.2.6 (unreleased) +**Related:** [spec-failure-policy.md](spec-failure-policy.md) + +## Overview + +Today the only way to trigger recurring work is `df.wait_for_schedule()` with a +cron expression, so you have to guess a rate. + +Take the background compactor for a bm25 index. Writes land as small segments, +and queries slow down until something merges them. The work is due when the +segment count crosses a threshold, which has nothing to do with the clock. Run +the merge every minute and almost every run finds nothing to do. Run it hourly +and the index stays bloated in between. + +`df.wait_for_condition()` waits on a SQL predicate instead of a clock. It +re-checks on a backstop interval, and an optional `NOTIFY` from whatever +changes the data makes it fire sooner. + +## Background + +PostgreSQL has one push mechanism reachable from SQL: `LISTEN`/`NOTIFY`, sent +with `pg_notify()`. It is transactional, delivered at commit and discarded on +rollback. It reaches only the sessions listening at that moment and is never +replayed, so a notification sent while nothing is listening is gone. Triggers +are the usual place to call it from, having no way to signal anything +themselves. + +The durable alternative is an ordinary table. A producer can record the change +in the same transaction that makes it, and the row survives a restart, but +nothing pushes, so something has to poll for it. Nothing else in core closes +the gap: logical replication slots are durable and also pull-based, and neither +advisory locks nor background worker latches can carry a notification from +ordinary SQL. No primitive is both durable and prompt. + +## API + +The compactor written as a condition instead of a schedule. The `tp_` functions +belong to pg_textsearch; pg_durable only ever sees opaque SQL. + +```sql +SELECT df.start( + @> ( df.wait_for_condition( + $$SELECT count(*) > 8 FROM tp_segments('docs_idx')$$, + max_check_interval => '1min', + notify_key => 'tp:segments_changed:docs_idx') + ~> df.sql($$SELECT tp_force_merge('docs_idx')$$) ), + 'compactor'); +``` + +**Parameters:** +- `condition` - SQL returning exactly one boolean column. See Predicate rules + below. +- `max_check_interval` - Required. The longest the node will go without + re-checking. A notification triggers a check sooner. Must be at least + 1 second. +- `notify_key` - Optional. Any string the waiter and the producer agree on. + Without it the timer is the only thing that triggers a check. + +To make it fire sooner, notify from wherever the data changes: + +```sql +SELECT pg_notify('pg_durable_condition', 'tp:segments_changed:docs_idx'); +``` + +Both sides just have to use the same string. Naming it after the data lets +unrelated waiters share one key. Several workflows might watch a `jobs` table, +one for a high-priority job appearing and another for the backlog passing a few +hundred; a single `jobs_changed` on the insert path wakes both, and each +applies its own predicate. Naming it after the condition works too. + +Nothing validates the string. If the producer never notifies, or spells the key +differently, the wait falls back to `max_check_interval` with no error. +pg_durable can't derive the key or install the trigger for you, because the +predicate is opaque SQL. + +### Filtering on the producer side + +Notifying isn't free on either end. Committing a transaction that issued +`NOTIFY` takes a cluster-wide `AccessExclusiveLock` on PostgreSQL 15 through +18, serializing it against every other notifying commit in the instance. A +per-row trigger on a heavily written table puts every write behind that lock. +And every notification costs the waiter a predicate evaluation, so notifying +per flush during ingest checks far more often than the backstop would. + +So notify where segments are flushed rather than per row, and filter before +notifying. + +```sql +IF (SELECT count(*) FROM tp_segments('docs_idx')) > 8 THEN + PERFORM pg_notify('pg_durable_condition', 'tp:segments_changed:docs_idx'); +END IF; +``` + +That duplicates the threshold, which is safe as long as the producer's filter +is looser than or equal to the predicate. Notify too often and you pay a wasted +re-evaluation; notify too rarely and the condition waits for +`max_check_interval`. + +A producer that filters is reporting the condition rather than the data, so +`tp:compaction_due:docs_idx` fits better at that point, at the cost of no +longer suiting a waiter with a different threshold. Where several waiters share +a key, filter on the loosest of their conditions. + +### Predicate rules + +The predicate is evaluated as `() IS TRUE`, so PostgreSQL's scalar +subquery rules apply: one boolean column, no rows or NULL reads as false, more +than one row is an error. Use `SELECT EXISTS (...)` for existence checks rather +than `SELECT true FROM t WHERE ...`, which starts erroring as soon as two rows +match. + +That also rejects what `df.if()` would coerce. `df.if()` reads +`SELECT count(*)` as true for any nonzero count, tolerable for a branch +evaluated once but not for a predicate that would then fire on every check and +never stop. `IS TRUE` rejects it during parse analysis, before the condition +runs. + +The predicate runs with `default_transaction_read_only`, so a condition with +side effects fails instead of performing them a million times. The activity +uses the extended query protocol, one statement per call, and the subquery +wrapper narrows it further: anything that isn't a single `SELECT` fails to +parse. + +The condition has to describe a state that persists until the work runs. Every +check re-reads the predicate, so a condition that goes true and then false +again between checks is missed entirely, notification or not. "More than eight +segments" persists until something compacts them. "A segment was just written" +doesn't. + +### Choosing max_check_interval + +There is no default. You have to state how stale you're willing to let the +condition get, because we can't guess it and a bad guess is invisible. The +floor is one second, which `LOOP_MIN_ITER_DURATION` enforces anyway by holding +every loop iteration to a second of wall clock. + +A check is one timer plus one `execute_sql`. At 5 seconds a single waiter runs +17k checks a day, negligible for a row-existence test. Raise it if your +predicate is expensive. + +With a `notify_key` the interval is only a safety net for conditions that +become true without anyone notifying: another writer, a restore, a producer +whose filter is stricter than the predicate. Minutes are reasonable there, and +cost fewer checks and fewer continuations. + +## Behavior + +The node registers and subscribes before it evaluates, so the two overlap: +evaluation catches anything that became true earlier, the subscription catches +anything from then on, and nothing falls between them. + +A condition that is already true fires without waiting. If it's false, the node +waits for whichever comes first: an event raised from a matching `NOTIFY`, or +`max_check_interval`. Then it evaluates again. It returns +`{"condition_met": true}`, in the same style as `df.wait_for_schedule()` +returning `{"scheduled": true}`. + +Two properties of duroxide 0.1.30 make that ordering work. +`ctx.dequeue_event()` emits its action when the future is created rather than +when it's awaited, so the subscription is durable before the predicate runs. +And it's a mailbox, so an event arriving while the instance isn't parked is +buffered until consumed, and unconsumed events survive `continue_as_new` (up to +100, past which the oldest are dropped with a warning). `schedule_wait` would +discard both. + +A third property decides how the loop holds its subscription. Naively, +recreating the subscription each iteration also works: the one created before a +predicate that returns true is dropped at `break`, and the one that loses +`select2` to the timer is dropped too, and neither loses a buffered +notification, because `DurableFuture::drop` marks the token cancelled and +duroxide skips cancelled subscriptions when matching arrivals in FIFO order. +But it's the wrong shape. Each recreation writes a subscribe/cancel pair to +history, and duroxide computes an arrival index by scanning every prior +subscription for that name, so a long wait becomes quadratic in the number of +checks. Instead the node holds one subscription across iterations and recreates +it only after it has actually delivered. Measured over 34 checks, that takes a +condition wait from 6 history events per check to 4 and from 35 subscriptions +to 1. + +`DurableFuture` is `Unpin`, so `&mut fut` is itself a future and the +subscription survives the `select2` that borrows it. + +One loss window remains: a notification sent while the worker is reconnecting +is gone before duroxide sees it. So the contract is the backstop. The condition +fires within `max_check_interval` of becoming true, which is the latency to set +for the case where no notification arrives, whether because the worker was +reconnecting, the producer filtered too aggressively, or whatever made the +condition true doesn't notify at all. + +The predicate runs as `submitted_by` through the existing `execute_sql` +activity. A predicate that errors is an ordinary node failure, so the +`on_failure` policy in the failure policy spec applies unchanged. + +### Bounded history + +Each check appends a timer and an activity to duroxide history, and history is +replayed on every event, so waiting in place forever at a short interval would +grow history without bound. + +So it doesn't. After 100 checks the node abandons the current loop iteration, +using the same unwind the failure policy spec introduces. The loop continues as +new, history is truncated, and the node re-evaluates on entry. At a 5-second +interval that's one continuation every 8 minutes. + +The cost is a short window during the continuation with nothing registered to +receive a notification. Nothing is lost: the node registers and evaluates again +on re-entry, overlapping the same way it does on first entry. Continuations are +unbounded, since the failure policy spec removes `MAX_LOOP_ITERATIONS`. + +## Registry and worker + +Waiting itself needs no state: `ctx.dequeue_event()` is a single history entry +that can block indefinitely without holding resources. State is needed only to +route a notification to the instances waiting on it. + +```sql +CREATE TABLE df.condition_waiters ( + instance_id TEXT NOT NULL, + node_id TEXT NOT NULL, + notify_key TEXT NOT NULL, + created_at TIMESTAMPTZ NOT NULL DEFAULT pg_catalog.now(), + PRIMARY KEY (instance_id, node_id) +); + +CREATE INDEX idx_condition_waiters_notify_key ON df.condition_waiters(notify_key); +``` + +`instance_id` is the duroxide instance id from `ctx.instance_id()`, not the +8-char `df` instance id. A node inside a loop body runs in a subtree child +instance, whose id `subtree_instance_id()` composes as +`{parent}::{execution}::{root_node}`. The `df` instance id is the first `::` +segment, which is what the cleanup sweep matches on. + +A row exists only while a wait with a `notify_key` is outstanding, and a wait +without one registers nothing. That is also what keeps the mailbox short: while +the loop body is running there is no row, so no event is raised. + +Registration and removal are two new activities. The node registers, creates +the wait future, then evaluates, and removes the row once the predicate is +true. Both are idempotent (`ON CONFLICT DO NOTHING`, delete by primary key), so +at-least-once execution is harmless. + +### Listener + +The worker holds one `sqlx::postgres::PgListener` on the single channel +`pg_durable_condition`, for the extension database. One fixed channel means the +worker never has to issue `LISTEN` or `UNLISTEN` as waiters come and go; it +filters on the payload instead. + +`PgListener` is a protocol-level client over host/port, which is what makes +this work at all: a background worker that issued `LISTEN` through SPI would +get only a latch wake, with the payload logged rather than delivered. + +On a notification with payload `K`, the worker selects waiters where +`notify_key = K` and calls `client.enqueue_event(instance_id, 'df:condition:' +|| K, '')` for each. It must be `enqueue_event`, not `raise_event`: +`ctx.dequeue_event()` consumes `WorkItem::QueueMessage`, which only +`enqueue_event` produces, and only that path gives the mailbox buffering the +ordering argument above depends on. `raise_external_event()` is doubly wrong, +since it also fans out to child instances. That's signal semantics. + +The worker wakes on the first notification for a key, then suppresses that key +for one second. An idle system gets its wake immediately; a system notifying +thousands of times a second still evaluates each predicate at most once a +second. That costs no trigger latency, since `LOOP_MIN_ITER_DURATION` already +stops a loop from firing more than once a second, and it bounds the cost of an +unfiltered producer. The suppression map drops entries older than the window, +so an unbounded stream of distinct keys doesn't grow it. + +On reconnect the worker wakes every registered waiter. +Notifications during the gap can't be recovered, and this bounds the catch-up +to the number of waiters rather than leaving every one of them to its backstop. + +### Cleanup + +The existing reconcile sweep (`src/worker.rs`, `pg_durable.reconcile_interval`) +deletes waiter rows whose instance is no longer running, so a cancelled or +crashed instance can't leak rows. + +### notify_key is not a permission + +Any role can `pg_notify` any payload. The worst that does is re-evaluate a +predicate early, which is what the backstop does on its own schedule anyway. +Don't treat `notify_key` as an access control boundary. + +### Out of scope + +One listener, on the extension database. Condition waits against other +databases are not covered here; see `docs/multi-database.md` for the existing +constraint that the duroxide runtime is tied to a single database. + +## Upgrade & Migration + +Unlike the failure policy spec, this one changes the `df` schema. Add to +`sql/pg_durable--0.2.5--0.2.6.sql`: + +1. `CREATE TABLE df.condition_waiters` and its index, copied from the + pgrx-generated fresh-install DDL so Scenario A sees identical fresh-install + and upgrade schemas. +2. `CREATE FUNCTION df.wait_for_condition(...)`. It's a new function, so + nothing needs dropping. +3. Drop and re-add `nodes_node_type_chk` and `nodes_structure_chk` to admit + `WAIT_CONDITION`. Both are `NOT VALID`, so this doesn't rewrite existing + rows. + +The node type also has to be registered in `VALID_NODE_TYPES` (`src/types.rs`), +which is the canonical list the DDL mirrors, and in `src/explain.rs`. Missing it +from `VALID_NODE_TYPES` fails quietly rather than loudly: `Durofut::ensure()` +falls back to treating an unrecognized node's JSON as a plain SQL string, so the +graph builds and only misbehaves at run time. + +**B1 (new `.so`, un-upgraded schema):** a pre-0.2.6 schema has no +`df.condition_waiters`. The registration activity probes +`information_schema.tables` (caching the result, re-probing while absent so an +in-place `ALTER EXTENSION UPDATE` is picked up) and continues without +registering, and the listener and cleanup sweep tolerate the table being +absent. `notify_key` then does nothing and the wait falls back to the interval +alone. The backstop is the contract, so a customer who never upgrades gets a +working, slower trigger. + +**Replay of in-flight instances:** `df.wait_for_condition()` doesn't exist +before 0.2.6, so no in-flight instance can contain one. Nothing to preserve. +The predicate does reuse the existing `execute_sql` activity, though, so the +`read_only` flag it adds to that activity's input must be omitted from the +serialized form when false. Otherwise every pre-existing history would replay +against a changed activity input. + +## Testing + +**Unit** (`./scripts/test-unit.sh`): +- `max_check_interval` is required; omitting it is an error and a value below + 1 second is rejected. A valid interval round-trips into the node config. +- `notify_key` is optional and absent from the config when omitted. +- The node config serializes and deserializes with the predicate intact. +- `df.condition_waiters` has the expected columns and an index on `notify_key`. +- The `read_only` flag is absent from a serialized `execute_sql` input when + false and defaults to false when missing. +- Notification suppression: the first notification for a key wakes, a repeat + inside the window does not, a repeat after it does, keys are independent, and + stale entries are evicted. +- The check interval rejects zero and negative values, which a hand-written + node could carry and which would otherwise widen into an infinite timer. +- `condition_met` reads the wrapped boolean and rejects a missing, null, or + non-boolean value, since `() IS TRUE` can never produce one. + +**E2E** (`tests/e2e/sql/67_wait_for_condition.sql`): +1. **Already true.** The predicate is true at start. The node completes without + waiting a full interval. +2. **Becomes true.** Another statement makes the predicate true. The instance + parks first, then completes, with no notification involved. +3. **Notification accelerates.** `max_check_interval => '5min'` with a + `notify_key`. The waiter row appears while parked, a `pg_notify` completes + the instance in seconds — proving the notification path fired rather than + the backstop — and the waiter row is gone afterwards. +4. **Second notification.** The same shape, but the first `pg_notify` arrives + with the predicate still false. The instance must stay parked, then complete + on a second `pg_notify` — the case that fails if the node consumes its + subscription without recreating it. +5. **Writing predicate.** A condition containing `INSERT` fails the node rather + than writing a row. +6. **Non-boolean predicate.** `SELECT count(*)` fails the node rather than + being coerced the way `df.if()` would coerce it. + +The producer's write and its `pg_notify` have to be their own top-level +statements. Inside the `DO` block that polls for completion they would not +commit until the block ended, so the worker would never see either. + +Deferred with the failure-policy work: a broken predicate under +`on_failure => 'continue'`, and the bounded-history unwind, both of which need +that spec's mechanism. diff --git a/docs/upgrade-testing.md b/docs/upgrade-testing.md index 294fcae0..a5aef9c6 100644 --- a/docs/upgrade-testing.md +++ b/docs/upgrade-testing.md @@ -205,6 +205,13 @@ what the upgrade script handles, and any backward compatibility considerations. ### v0.2.5 → v0.2.6 +#### Add `df.wait_for_condition()` and the condition waiter registry +- **DDL change (df schema):** Adds the table `df.condition_waiters` (`instance_id TEXT`, `node_id TEXT`, `notify_key TEXT`, `created_at TIMESTAMPTZ`, PK `(instance_id, node_id)`) with `idx_condition_waiters_notify_key`, a new node type `WAIT_CONDITION`, and a new `#[pg_extern(schema = "df")]` function `df.wait_for_condition(text, interval, text)`. `instance_id` is the duroxide instance id (composite for loop subtrees), not the 8-char `df` instance id, so it is `TEXT`. +- **Upgrade script:** `sql/pg_durable--0.2.5--0.2.6.sql` creates the table and index verbatim from the pgrx-generated fresh-install DDL, and drops/re-adds `nodes_node_type_chk` and `nodes_structure_chk` with `WAIT_CONDITION` admitted (`WAIT_CONDITION` is a leaf: no children, `query` required). Both constraints are `NOT VALID`, so re-adding them does not rewrite existing rows. `df.ensure_durofut()` is dropped by this same script, so there is no PL/pgSQL validator to keep in sync — `VALID_NODE_TYPES` in `src/types.rs` remains the canonical list. The new function needs no drop. +- **Scenario A considerations:** Fresh and upgraded schemas both carry the table, the index, and the two re-added constraints in identical form. +- **Scenario B1 considerations:** A pre-0.2.6 schema has no `df.condition_waiters`. The register/unregister activities probe `information_schema.tables` (caching a positive result, re-probing while absent so an in-place `ALTER EXTENSION UPDATE` is picked up without a worker restart) and skip registration when the table is missing; the NOTIFY listener and the reconcile sweep tolerate the same absence. `notify_key` then does nothing and the wait falls back to `max_check_interval`, which is the documented guarantee. No existing symbol is removed or renamed. The `WAIT_CONDITION` node type cannot appear on an un-upgraded schema, since `df.wait_for_condition()` has no catalog entry there. +- **Scenario B2 considerations:** The predicate reuses the existing `execute_sql` activity, whose input gains a `read_only` flag. That field is skipped during serialization when false, so activity inputs recorded by pre-0.2.6 binaries are byte-identical to what the new binary produces and in-flight histories replay unchanged. No in-flight instance can contain a `WAIT_CONDITION` node. + #### Remove `df.ensure_durofut()` - **DDL change:** Fresh installs no longer create the undocumented `df.ensure_durofut(text)` PL/pgSQL helper. `df.if_then_op()` now stores its condition and then-branch operands as text in the partial marker; `df.if_else_op()` extracts those operands and passes all three directly to the Rust-backed `df.if()`, which already performs Durofut normalization. - **Upgrade script:** `sql/pg_durable--0.2.5--0.2.6.sql` replaces both operator helpers before dropping `df.ensure_durofut(text)` with `RESTRICT`. The new `df.if_else_op()` uses JSON text extraction, which accepts both new string-valued partial markers and object-valued markers emitted before the upgrade. `RESTRICT` deliberately aborts rather than silently removing a customer-owned object that depends on the undocumented helper. diff --git a/scripts/test-e2e-local.sh b/scripts/test-e2e-local.sh index 5bfdd5cb..e0eb81b3 100755 --- a/scripts/test-e2e-local.sh +++ b/scripts/test-e2e-local.sh @@ -76,7 +76,7 @@ ALL_PHASES=( "http-allow-all" ) -PGRX_HOME="$HOME/.pgrx" +PGRX_HOME="${PGRX_HOME:-$HOME/.pgrx}" PG_USER="postgres" PG_DB="postgres" E2E_USER="df_e2e_user" @@ -262,7 +262,7 @@ if ! [[ "$REPEAT_COUNT" =~ ^[0-9]+$ ]] || [ "$REPEAT_COUNT" -lt 1 ]; then exit 1 fi -PG_PORT="$((28800 + PG_VERSION))" +PG_PORT="${PG_PORT:-$((28800 + PG_VERSION))}" DATA_DIR="$PGRX_HOME/data-$PG_VERSION" LOG_FILE="$PGRX_HOME/$PG_VERSION.log" CONF_FILE="$DATA_DIR/postgresql.conf" diff --git a/scripts/test-epoch-race.sh b/scripts/test-epoch-race.sh index 97503898..b40094f3 100755 --- a/scripts/test-epoch-race.sh +++ b/scripts/test-epoch-race.sh @@ -54,8 +54,8 @@ while [[ $# -gt 0 ]]; do esac done -PG_PORT="$((28800 + PG_VERSION))" -PGRX_HOME="$HOME/.pgrx" +PG_PORT="${PG_PORT:-$((28800 + PG_VERSION))}" +PGRX_HOME="${PGRX_HOME:-$HOME/.pgrx}" DATA_DIR="$PGRX_HOME/data-$PG_VERSION" LOG_FILE="$PGRX_HOME/$PG_VERSION.log" diff --git a/scripts/test-shutdown.sh b/scripts/test-shutdown.sh index 93c6d91c..d8109bba 100755 --- a/scripts/test-shutdown.sh +++ b/scripts/test-shutdown.sh @@ -55,8 +55,8 @@ while [[ $# -gt 0 ]]; do esac done -PG_PORT="$((28800 + PG_VERSION))" -PGRX_HOME="$HOME/.pgrx" +PG_PORT="${PG_PORT:-$((28800 + PG_VERSION))}" +PGRX_HOME="${PGRX_HOME:-$HOME/.pgrx}" DATA_DIR="$PGRX_HOME/data-$PG_VERSION" LOG_FILE="$PGRX_HOME/$PG_VERSION.log" diff --git a/scripts/test-upgrade.sh b/scripts/test-upgrade.sh index 6975e314..298098d2 100755 --- a/scripts/test-upgrade.sh +++ b/scripts/test-upgrade.sh @@ -59,8 +59,8 @@ while [[ $# -gt 0 ]]; do done # pgrx settings -PGRX_HOME="$HOME/.pgrx" -PG_PORT="$((28800 + PG_VERSION))" +PGRX_HOME="${PGRX_HOME:-$HOME/.pgrx}" +PG_PORT="${PG_PORT:-$((28800 + PG_VERSION))}" # Find pgrx binaries PGRX_BIN=$(ls -d "$PGRX_HOME/$PG_VERSION."*/pgrx-install/bin 2>/dev/null | head -1) diff --git a/sql/pg_durable--0.2.5--0.2.6.sql b/sql/pg_durable--0.2.5--0.2.6.sql index 1b265794..0cb39eb0 100644 --- a/sql/pg_durable--0.2.5--0.2.6.sql +++ b/sql/pg_durable--0.2.5--0.2.6.sql @@ -46,3 +46,60 @@ $$ LANGUAGE plpgsql IMMUTABLE SET search_path = pg_catalog, pg_temp; -- RESTRICT is intentional: do not silently remove customer-owned objects that -- depend on this undocumented helper. DROP FUNCTION df.ensure_durofut(text) RESTRICT; + +-- df.wait_for_condition(): registry of orchestration nodes currently blocked on +-- a predicate. The background worker's NOTIFY listener joins incoming payloads +-- against notify_key to decide which waiters to wake early. +-- +-- instance_id is the *duroxide* instance id, not the 8-char df instance id: a +-- node inside a loop body runs in a subtree child instance whose id is the +-- composite "{parent}::{execution}::{root_node}". +CREATE TABLE df.condition_waiters ( + instance_id TEXT NOT NULL, + node_id TEXT NOT NULL, + notify_key TEXT NOT NULL, + created_at TIMESTAMPTZ NOT NULL DEFAULT pg_catalog.now(), + PRIMARY KEY (instance_id, node_id) +); + +CREATE INDEX idx_condition_waiters_notify_key ON df.condition_waiters(notify_key); + +-- Admit the WAIT_CONDITION node type into the schema constraints. These mirror +-- VALID_NODE_TYPES in src/types.rs (the Rust constant is the canonical source). +-- The constraints are NOT VALID, so re-adding them does not rewrite existing rows. +ALTER TABLE df.nodes DROP CONSTRAINT nodes_node_type_chk; +ALTER TABLE df.nodes + ADD CONSTRAINT nodes_node_type_chk + CHECK (node_type OPERATOR(pg_catalog.=) ANY (ARRAY['SQL', 'THEN', 'IF', 'JOIN', 'LOOP', 'BREAK', 'RACE', 'SLEEP', 'WAIT_SCHEDULE', 'WAIT_CONDITION', 'HTTP', 'HTTP_MULTIPART', 'SIGNAL'])) NOT VALID; + +ALTER TABLE df.nodes DROP CONSTRAINT nodes_structure_chk; +ALTER TABLE df.nodes + ADD CONSTRAINT nodes_structure_chk + CHECK ( + CASE + WHEN node_type OPERATOR(pg_catalog.=) ANY (ARRAY['SQL', 'SLEEP', 'WAIT_SCHEDULE', 'WAIT_CONDITION', 'BREAK', 'HTTP', 'HTTP_MULTIPART', 'SIGNAL']) + THEN left_node IS NULL AND right_node IS NULL AND query IS NOT NULL + WHEN node_type OPERATOR(pg_catalog.=) 'THEN' + THEN left_node IS NOT NULL AND right_node IS NOT NULL AND query IS NULL + WHEN node_type OPERATOR(pg_catalog.=) 'IF' + THEN left_node IS NOT NULL AND right_node IS NOT NULL AND query IS NOT NULL + WHEN node_type OPERATOR(pg_catalog.=) 'LOOP' + THEN left_node IS NOT NULL AND right_node IS NULL + WHEN node_type OPERATOR(pg_catalog.=) 'JOIN' + THEN left_node IS NOT NULL AND right_node IS NOT NULL + WHEN node_type OPERATOR(pg_catalog.=) 'RACE' + THEN left_node IS NOT NULL AND right_node IS NOT NULL AND query IS NULL + ELSE FALSE + END + ) NOT VALID; + +-- df.wait_for_condition(): copied verbatim from the pgrx-generated fresh-install +-- DDL so the upgraded and fresh schemas match. New function, nothing to drop. +CREATE FUNCTION df."wait_for_condition"( + "condition" TEXT, /* &str */ + "max_check_interval" interval, /* pgrx::datum::interval::Interval */ + "notify_key" TEXT DEFAULT NULL /* core::option::Option<&str> */ +) RETURNS TEXT /* alloc::string::String */ + +LANGUAGE c /* Rust */ +AS 'MODULE_PATHNAME', 'wait_for_condition_wrapper'; diff --git a/src/activities/execute_sql.rs b/src/activities/execute_sql.rs index 29e7edeb..a3e63fc8 100644 --- a/src/activities/execute_sql.rs +++ b/src/activities/execute_sql.rs @@ -64,6 +64,19 @@ pub struct ExecuteSqlInput { /// Target database (None = extension database) #[serde(skip_serializing_if = "Option::is_none")] pub database: Option, + /// Run the statement with `default_transaction_read_only` so it cannot + /// perform writes. Used by `df.wait_for_condition()` predicates, which are + /// re-evaluated indefinitely and must not have side effects. + /// + /// Skipped when false so the serialized activity input is byte-identical to + /// what pre-`read_only` binaries recorded, keeping in-flight histories + /// replayable. + #[serde(default, skip_serializing_if = "is_false")] + pub read_only: bool, +} + +fn is_false(v: &bool) -> bool { + !*v } /// Decode a single column value from a PostgreSQL row into a `serde_json::Value`. @@ -226,6 +239,13 @@ pub async fn execute( let mut conn = connect_as_user(&input.submitted_by, input.database.as_deref()).await?; + if input.read_only { + sqlx::query("SET default_transaction_read_only = on") + .execute(&mut conn) + .await + .map_err(|e| format!("SET default_transaction_read_only failed: {e}"))?; + } + // SECURITY: Dynamic SQL is intentional. The query is authored by the submitting // user via df.sql() and executes under their own role via connect_as_user(). // This is equivalent to the user running SQL directly. @@ -260,3 +280,42 @@ pub async fn execute( } } } + +#[cfg(test)] +mod tests { + use super::*; + + /// Replay safety: histories written before `read_only` existed serialized + /// the input without the field. Emitting it unconditionally would change + /// the recorded activity input and break replay of in-flight instances. + #[test] + fn read_only_is_omitted_when_false() { + let input = ExecuteSqlInput { + query: "SELECT 1".to_string(), + submitted_by: "alice".to_string(), + database: None, + read_only: false, + }; + let json = serde_json::to_string(&input).unwrap(); + assert_eq!(json, r#"{"query":"SELECT 1","submitted_by":"alice"}"#); + } + + #[test] + fn read_only_is_emitted_when_true() { + let input = ExecuteSqlInput { + query: "SELECT true".to_string(), + submitted_by: "alice".to_string(), + database: None, + read_only: true, + }; + let json = serde_json::to_string(&input).unwrap(); + assert!(json.contains(r#""read_only":true"#), "got {json}"); + } + + #[test] + fn read_only_defaults_to_false_when_absent() { + let input: ExecuteSqlInput = + serde_json::from_str(r#"{"query":"SELECT 1","submitted_by":"alice"}"#).unwrap(); + assert!(!input.read_only); + } +} diff --git a/src/activities/mod.rs b/src/activities/mod.rs index 6937dc5c..3dad5400 100644 --- a/src/activities/mod.rs +++ b/src/activities/mod.rs @@ -11,5 +11,7 @@ pub mod execute_multipart; pub mod execute_sql; pub mod http_response; pub mod load_function_graph; +pub mod register_condition_waiter; +pub mod unregister_condition_waiter; pub mod update_instance_status; pub mod update_node_status; diff --git a/src/activities/register_condition_waiter.rs b/src/activities/register_condition_waiter.rs new file mode 100644 index 00000000..b179b0ca --- /dev/null +++ b/src/activities/register_condition_waiter.rs @@ -0,0 +1,98 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the PostgreSQL License. + +//! RegisterConditionWaiter activity - records a node blocked in +//! `df.wait_for_condition()` so the worker's NOTIFY listener can wake it. +//! +//! A row exists only while a wait carrying a `notify_key` is outstanding. A +//! wait without a key registers nothing and relies purely on its interval. +//! +//! The insert is idempotent (`ON CONFLICT DO NOTHING` on the primary key), so +//! duroxide's at-least-once activity execution is harmless. + +use duroxide::ActivityContext; +use serde::{Deserialize, Serialize}; +use sqlx::PgPool; +use std::sync::atomic::{AtomicU8, Ordering}; +use std::sync::Arc; + +/// Activity name for registration and scheduling +pub const NAME: &str = "pg_durable::activity::register-condition-waiter"; + +/// Input for the register/unregister condition waiter activities. +#[derive(Debug, Serialize, Deserialize)] +pub struct ConditionWaiterInput { + /// The duroxide instance id (`ctx.instance_id()`), which for a node inside + /// a loop body is a composite subtree id, not the 8-char df instance id. + pub instance_id: String, + pub node_id: String, + /// Only set on register; unregister deletes by primary key. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub notify_key: Option, +} + +/// Process-global cache for whether `df.condition_waiters` exists. +/// +/// 0 = unknown, 1 = present, 2 = absent. The table is added by the +/// 0.2.5 → 0.2.6 upgrade; a binary newer than the schema (Scenario B1) must run +/// against an older schema that lacks it. We cache "present" permanently once +/// seen, but re-probe on "unknown"/"absent" so an in-place ALTER EXTENSION +/// UPDATE that adds the table is picked up without a worker restart. +static WAITERS_TABLE: AtomicU8 = AtomicU8::new(0); + +pub(crate) async fn waiters_table_present(pool: &PgPool) -> bool { + if WAITERS_TABLE.load(Ordering::Relaxed) == 1 { + return true; + } + let present = sqlx::query_scalar::<_, bool>( + "SELECT EXISTS (SELECT 1 FROM information_schema.tables \ + WHERE table_schema = 'df' AND table_name = 'condition_waiters')", + ) + .fetch_one(pool) + .await + .unwrap_or(false); + WAITERS_TABLE.store(if present { 1 } else { 2 }, Ordering::Relaxed); + present +} + +pub async fn execute( + ctx: ActivityContext, + pool: Arc, + input_json: String, +) -> Result { + let input: ConditionWaiterInput = serde_json::from_str(&input_json) + .map_err(|e| format!("Invalid register_condition_waiter input: {e}"))?; + + let notify_key = match input.notify_key { + Some(k) => k, + None => return Ok("{}".to_string()), + }; + + // Missing table means the .so is newer than the schema. The interval + // backstop still fires, so degrade to polling rather than failing the node. + if !waiters_table_present(&pool).await { + ctx.trace_info( + "df.condition_waiters is absent (schema predates 0.2.6); \ + condition wait will rely on max_check_interval only", + ); + return Ok("{}".to_string()); + } + + sqlx::query( + "INSERT INTO df.condition_waiters (instance_id, node_id, notify_key) \ + VALUES ($1, $2, $3) ON CONFLICT (instance_id, node_id) DO NOTHING", + ) + .bind(&input.instance_id) + .bind(&input.node_id) + .bind(¬ify_key) + .execute(&*pool) + .await + .map_err(|e| format!("Failed to register condition waiter: {e}"))?; + + ctx.trace_info(format!( + "Registered condition waiter {}/{} on key '{}'", + input.instance_id, input.node_id, notify_key + )); + + Ok("{}".to_string()) +} diff --git a/src/activities/unregister_condition_waiter.rs b/src/activities/unregister_condition_waiter.rs new file mode 100644 index 00000000..d71e2016 --- /dev/null +++ b/src/activities/unregister_condition_waiter.rs @@ -0,0 +1,44 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the PostgreSQL License. + +//! UnregisterConditionWaiter activity - removes the row written by +//! [`crate::activities::register_condition_waiter`] once the predicate is true. +//! +//! Deleting by primary key is idempotent, so a replayed or duplicated execution +//! is harmless. + +use duroxide::ActivityContext; +use sqlx::PgPool; +use std::sync::Arc; + +use super::register_condition_waiter::{waiters_table_present, ConditionWaiterInput}; + +/// Activity name for registration and scheduling +pub const NAME: &str = "pg_durable::activity::unregister-condition-waiter"; + +pub async fn execute( + ctx: ActivityContext, + pool: Arc, + input_json: String, +) -> Result { + let input: ConditionWaiterInput = serde_json::from_str(&input_json) + .map_err(|e| format!("Invalid unregister_condition_waiter input: {e}"))?; + + if !waiters_table_present(&pool).await { + return Ok("{}".to_string()); + } + + sqlx::query("DELETE FROM df.condition_waiters WHERE instance_id = $1 AND node_id = $2") + .bind(&input.instance_id) + .bind(&input.node_id) + .execute(&*pool) + .await + .map_err(|e| format!("Failed to unregister condition waiter: {e}"))?; + + ctx.trace_info(format!( + "Unregistered condition waiter {}/{}", + input.instance_id, input.node_id + )); + + Ok("{}".to_string()) +} diff --git a/src/condition_listener.rs b/src/condition_listener.rs new file mode 100644 index 00000000..352dd3fa --- /dev/null +++ b/src/condition_listener.rs @@ -0,0 +1,226 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the PostgreSQL License. + +//! NOTIFY listener that wakes `df.wait_for_condition()` waiters early. +//! +//! One `PgListener` holds a single `LISTEN pg_durable_condition` for the +//! extension database, so the worker never issues `LISTEN`/`UNLISTEN` as +//! waiters come and go. Routing is done on the payload instead: the payload is +//! the `notify_key`, and `df.condition_waiters` maps it to the instances +//! parked on it. +//! +//! Waking is `Client::enqueue_event`, not `raise_event`. `dequeue_event` (what +//! the orchestration node subscribes with) consumes `WorkItem::QueueMessage`, +//! which only `enqueue_event` produces, and it is a mailbox — an event that +//! lands while the instance is not parked is buffered rather than dropped. +//! `raise_external_event` is doubly wrong here: it also fans out to child +//! instances, which is signal semantics. +//! +//! The listener runs inside the background worker, which connects over +//! host/port with sqlx. That matters: a background worker that issued `LISTEN` +//! through SPI would receive only a latch wake, with the payload logged rather +//! than delivered. + +use std::collections::HashMap; +use std::sync::Arc; +use std::time::{Duration, Instant}; + +use duroxide::Client; +use pgrx::log; +use sqlx::postgres::PgListener; + +use crate::types::condition_queue_name; + +/// The single channel every condition producer notifies. +pub const CHANNEL: &str = "pg_durable_condition"; + +/// Minimum spacing between two wakes for the same `notify_key`. +/// +/// An idle system gets its wake immediately; a producer notifying thousands of +/// times a second still causes at most one re-evaluation per key per second. +/// That costs no trigger latency, since `LOOP_MIN_ITER_DURATION` already stops +/// a loop from firing more than once a second. +const SUPPRESS_WINDOW: Duration = Duration::from_secs(1); + +/// Whether a notification for `key` should wake its waiters, given the last +/// wake time per key. Records the decision in `seen`. +fn should_raise(key: &str, now: Instant, seen: &mut HashMap) -> bool { + // Drop entries that can no longer suppress anything, so an unbounded stream + // of distinct keys does not grow the map forever. + seen.retain(|_, last| now.duration_since(*last) < SUPPRESS_WINDOW); + + match seen.get(key) { + Some(last) if now.duration_since(*last) < SUPPRESS_WINDOW => false, + _ => { + seen.insert(key.to_string(), now); + true + } + } +} + +async fn waiters_for_key(pool: &sqlx::PgPool, key: &str) -> Result, sqlx::Error> { + sqlx::query_scalar::<_, String>( + "SELECT instance_id FROM df.condition_waiters WHERE notify_key = $1", + ) + .bind(key) + .fetch_all(pool) + .await +} + +async fn all_waiters(pool: &sqlx::PgPool) -> Result, sqlx::Error> { + sqlx::query_as::<_, (String, String)>( + "SELECT instance_id, notify_key FROM df.condition_waiters", + ) + .fetch_all(pool) + .await +} + +async fn wake(client: &Client, instance: &str, key: &str) { + if let Err(e) = client + .enqueue_event(instance, condition_queue_name(key), "") + .await + { + log!("pg_durable: waking condition waiter {instance} on '{key}' failed: {e}"); + } +} + +/// Notifications sent while nothing was listening are gone, so on every +/// (re)connect wake every registered waiter once. That bounds catch-up to the +/// number of waiters instead of leaving each one to its interval backstop. +async fn resync(pool: &sqlx::PgPool, client: &Client) { + match all_waiters(pool).await { + Ok(waiters) => { + for (instance, key) in &waiters { + wake(client, instance, key).await; + } + if !waiters.is_empty() { + log!( + "pg_durable: condition listener resynced {} waiter(s)", + waiters.len() + ); + } + } + // The table is absent on a schema older than 0.2.6 (Scenario B1). + // Waiters then fall back to their interval, which is the documented + // backstop, so this is not fatal. + Err(e) => log!("pg_durable: condition listener resync failed: {e}"), + } +} + +/// Listen for condition notifications until the task is aborted. +/// +/// Reconnects on error rather than returning, so a terminated backend or a +/// restarted server does not permanently disable early wakeups. +pub async fn run(conn_str: String, pool: Arc, client: Client) { + /// Backoff between reconnect attempts. + const RECONNECT_DELAY: Duration = Duration::from_secs(5); + + let mut seen: HashMap = HashMap::new(); + + loop { + let mut listener = match PgListener::connect(&conn_str).await { + Ok(l) => l, + Err(e) => { + log!("pg_durable: condition listener connect failed: {e}"); + tokio::time::sleep(RECONNECT_DELAY).await; + continue; + } + }; + + if let Err(e) = listener.listen(CHANNEL).await { + log!("pg_durable: LISTEN {CHANNEL} failed: {e}"); + tokio::time::sleep(RECONNECT_DELAY).await; + continue; + } + + log!("pg_durable: condition listener attached to '{CHANNEL}'"); + seen.clear(); + resync(&pool, &client).await; + + loop { + match listener.recv().await { + Ok(notification) => { + let key = notification.payload(); + if key.is_empty() || !should_raise(key, Instant::now(), &mut seen) { + continue; + } + match waiters_for_key(&pool, key).await { + Ok(instances) => { + for instance in &instances { + wake(&client, instance, key).await; + } + } + Err(e) => { + log!("pg_durable: looking up condition waiters for '{key}' failed: {e}") + } + } + } + Err(e) => { + log!("pg_durable: condition listener disconnected: {e}"); + break; + } + } + } + + tokio::time::sleep(RECONNECT_DELAY).await; + } +} + +#[cfg(test)] +mod tests { + use super::*; + use std::collections::HashMap; + use std::time::{Duration, Instant}; + + #[test] + fn first_notification_for_a_key_raises() { + let mut seen = HashMap::new(); + let now = Instant::now(); + assert!(should_raise("k", now, &mut seen)); + } + + #[test] + fn repeat_within_window_is_suppressed() { + let mut seen = HashMap::new(); + let now = Instant::now(); + assert!(should_raise("k", now, &mut seen)); + assert!(!should_raise( + "k", + now + Duration::from_millis(999), + &mut seen + )); + } + + #[test] + fn repeat_after_window_raises() { + let mut seen = HashMap::new(); + let now = Instant::now(); + assert!(should_raise("k", now, &mut seen)); + assert!(should_raise( + "k", + now + Duration::from_millis(1_000), + &mut seen + )); + } + + #[test] + fn suppression_is_per_key() { + let mut seen = HashMap::new(); + let now = Instant::now(); + assert!(should_raise("a", now, &mut seen)); + assert!(should_raise("b", now, &mut seen)); + } + + /// A busy producer must not grow the suppression map without bound. + #[test] + fn stale_keys_are_evicted() { + let mut seen = HashMap::new(); + let start = Instant::now(); + for i in 0..10 { + should_raise(&format!("k{i}"), start, &mut seen); + } + assert_eq!(seen.len(), 10); + should_raise("fresh", start + Duration::from_secs(60), &mut seen); + assert_eq!(seen.len(), 1, "expected stale keys to be evicted"); + } +} diff --git a/src/dsl.rs b/src/dsl.rs index 7070e52b..e448e6f5 100644 --- a/src/dsl.rs +++ b/src/dsl.rs @@ -306,6 +306,51 @@ pub fn wait_for_schedule(cron_expr: &str) -> String { .to_json() } +/// Creates a wait-for-condition node that waits until a SQL predicate is true. +/// +/// The predicate is re-evaluated on `max_check_interval`, and, when a +/// `notify_key` is supplied, sooner if a matching `pg_notify` arrives on the +/// `pg_durable_condition` channel. Only the raw predicate is stored; it is +/// wrapped in `IS TRUE` at execution time so `df.nodes` shows what the user +/// wrote. +#[pg_extern(schema = "df")] +pub fn wait_for_condition( + condition: &str, + max_check_interval: pgrx::datum::Interval, + notify_key: default!(Option<&str>, "NULL"), +) -> String { + if condition.trim().is_empty() { + pgrx::error!("Condition must not be empty"); + } + + // Reject sub-second intervals: LOOP_MIN_ITER_DURATION already holds a loop + // iteration to a second, so anything shorter buys no responsiveness. + let micros = max_check_interval.as_micros(); + if micros < 1_000_000 { + pgrx::error!("max_check_interval must be at least 1 second"); + } + let secs = (micros / 1_000_000) as i64; + + let mut config = serde_json::json!({ + "condition": condition, + "max_check_interval_secs": secs, + }); + + if let Some(key) = notify_key { + if key.trim().is_empty() { + pgrx::error!("notify_key must not be empty when supplied"); + } + config["notify_key"] = serde_json::Value::String(key.to_string()); + } + + Durofut { + node_type: "WAIT_CONDITION".to_string(), + query: Some(config.to_string()), + ..Default::default() + } + .to_json() +} + /// Creates a loop node. /// /// With one argument: repeats the body indefinitely (infinite loop). diff --git a/src/explain.rs b/src/explain.rs index ae1e4d44..ebb51493 100644 --- a/src/explain.rs +++ b/src/explain.rs @@ -445,6 +445,7 @@ fn build_tree_recursive( && seq_node.node_type != "SQL" && seq_node.node_type != "SLEEP" && seq_node.node_type != "WAIT_SCHEDULE" + && seq_node.node_type != "WAIT_CONDITION" && seq_node.node_type != "HTTP" && seq_node.node_type != "HTTP_MULTIPART" { @@ -666,6 +667,25 @@ fn format_node_display(node: &ExplainNode) -> String { .unwrap_or_else(|| "?".to_string()); format!("WAIT '{cron}'{name_suffix}") } + "WAIT_CONDITION" => { + let cfg = node + .query + .as_ref() + .and_then(|q| serde_json::from_str::(q).ok()) + .unwrap_or(serde_json::Value::Null); + let condition = cfg["condition"].as_str().unwrap_or("?"); + let truncated = if condition.len() > 40 { + format!("{}...", &condition[..37]) + } else { + condition.to_string() + }; + let key = cfg["notify_key"] + .as_str() + .map(|k| format!(", notify_key: {k}")) + .unwrap_or_default(); + let interval = cfg["max_check_interval_secs"].as_i64().unwrap_or(0); + format!("WAIT UNTIL {truncated} (every {interval}s{key}){name_suffix}") + } "HTTP" | "HTTP_MULTIPART" => { // Parse config to get method and URL let (method, url) = node diff --git a/src/lib.rs b/src/lib.rs index 11736117..067be5eb 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -54,6 +54,7 @@ pub static RECONCILE_INTERVAL: GucSetting = GucSetting::::new(3600); // Module declarations pub mod activities; pub mod client; +pub mod condition_listener; pub mod dsl; pub mod explain; pub mod monitoring; @@ -361,6 +362,23 @@ CREATE TABLE df._worker_epoch ( last_seen_at TIMESTAMPTZ DEFAULT pg_catalog.now() ); +-- Registry of orchestration nodes currently blocked in df.wait_for_condition(). +-- The background worker's NOTIFY listener joins incoming payloads against +-- notify_key to decide which waiters to wake early. +-- +-- instance_id is the *duroxide* instance id, not the 8-char df instance id: a +-- node inside a loop body runs in a subtree child instance whose id is the +-- composite "{parent}::{execution}::{root_node}". +CREATE TABLE df.condition_waiters ( + instance_id TEXT NOT NULL, + node_id TEXT NOT NULL, + notify_key TEXT NOT NULL, + created_at TIMESTAMPTZ NOT NULL DEFAULT pg_catalog.now(), + PRIMARY KEY (instance_id, node_id) +); + +CREATE INDEX idx_condition_waiters_notify_key ON df.condition_waiters(notify_key); + ALTER TABLE df.instances ADD CONSTRAINT instances_id_format_chk -- Operators (OPERATOR(pg_catalog.)) and functions (e.g. pg_catalog.now) @@ -391,7 +409,7 @@ ALTER TABLE df.nodes ADD CONSTRAINT nodes_right_node_format_chk CHECK (right_node IS NULL OR right_node OPERATOR(pg_catalog.~) '^[0-9a-f]{8}$') NOT VALID, ADD CONSTRAINT nodes_node_type_chk - CHECK (node_type OPERATOR(pg_catalog.=) ANY (ARRAY['SQL', 'THEN', 'IF', 'JOIN', 'LOOP', 'BREAK', 'RACE', 'SLEEP', 'WAIT_SCHEDULE', 'HTTP', 'HTTP_MULTIPART', 'SIGNAL'])) NOT VALID, + CHECK (node_type OPERATOR(pg_catalog.=) ANY (ARRAY['SQL', 'THEN', 'IF', 'JOIN', 'LOOP', 'BREAK', 'RACE', 'SLEEP', 'WAIT_SCHEDULE', 'WAIT_CONDITION', 'HTTP', 'HTTP_MULTIPART', 'SIGNAL'])) NOT VALID, ADD CONSTRAINT nodes_result_name_chk CHECK (result_name IS NULL OR result_name OPERATOR(pg_catalog.~) '^[A-Za-z_][A-Za-z0-9_]*$') NOT VALID, ADD CONSTRAINT nodes_status_chk @@ -401,7 +419,7 @@ ALTER TABLE df.nodes ADD CONSTRAINT nodes_structure_chk CHECK ( CASE - WHEN node_type OPERATOR(pg_catalog.=) ANY (ARRAY['SQL', 'SLEEP', 'WAIT_SCHEDULE', 'BREAK', 'HTTP', 'HTTP_MULTIPART', 'SIGNAL']) + WHEN node_type OPERATOR(pg_catalog.=) ANY (ARRAY['SQL', 'SLEEP', 'WAIT_SCHEDULE', 'WAIT_CONDITION', 'BREAK', 'HTTP', 'HTTP_MULTIPART', 'SIGNAL']) THEN left_node IS NULL AND right_node IS NULL AND query IS NOT NULL WHEN node_type OPERATOR(pg_catalog.=) 'THEN' THEN left_node IS NOT NULL AND right_node IS NOT NULL AND query IS NULL @@ -1430,6 +1448,109 @@ mod tests { assert_eq!(config["timeout_seconds"], 60); } + // ======================================================================== + // df.wait_for_condition() + // ======================================================================== + + #[pg_test] + fn test_wait_for_condition_via_sql() { + let result = Spi::get_one::( + "SELECT df.wait_for_condition('SELECT count(*) > 8 FROM t', '1min')", + ) + .unwrap() + .unwrap(); + let fut = Durofut::from_json(&result); + assert_eq!(fut.node_type, "WAIT_CONDITION"); + + let config: serde_json::Value = serde_json::from_str(fut.query.as_ref().unwrap()).unwrap(); + assert_eq!(config["condition"], "SELECT count(*) > 8 FROM t"); + assert_eq!(config["max_check_interval_secs"], 60); + assert!( + config.get("notify_key").is_none(), + "notify_key must be absent when not supplied" + ); + } + + #[pg_test] + fn test_wait_for_condition_with_notify_key_via_sql() { + let result = Spi::get_one::( + "SELECT df.wait_for_condition('SELECT true', '30s', 'seg_changed')", + ) + .unwrap() + .unwrap(); + let fut = Durofut::from_json(&result); + + let config: serde_json::Value = serde_json::from_str(fut.query.as_ref().unwrap()).unwrap(); + assert_eq!(config["max_check_interval_secs"], 30); + assert_eq!(config["notify_key"], "seg_changed"); + } + + #[pg_test(error = "max_check_interval must be at least 1 second")] + fn test_wait_for_condition_rejects_sub_second_interval() { + Spi::get_one::("SELECT df.wait_for_condition('SELECT true', '500ms')").ok(); + } + + #[pg_test(error = "Condition must not be empty")] + fn test_wait_for_condition_rejects_empty_condition() { + Spi::get_one::("SELECT df.wait_for_condition(' ', '1min')").ok(); + } + + #[pg_test] + fn test_wait_for_condition_wraps_predicate_as_is_true() { + // The raw predicate is stored unwrapped; wrapping happens at execution + // time so df.nodes stays readable and EXPLAIN shows what the user wrote. + let wrapped = crate::types::wrap_condition_sql("SELECT count(*) > 8 FROM t"); + assert_eq!( + wrapped, + "SELECT (SELECT count(*) > 8 FROM t) IS TRUE AS condition_met" + ); + } + + #[pg_test] + fn test_wait_for_condition_in_sequence() { + let cond = Spi::get_one::("SELECT df.wait_for_condition('SELECT true', '1min')") + .unwrap() + .unwrap(); + let sql_node = crate::dsl::sql("SELECT 1"); + let seq = crate::dsl::then_fn(&cond, &sql_node); + let fut = Durofut::from_json(&seq); + assert_eq!(fut.node_type, "THEN"); + assert!(fut.left_node.is_some()); + } + + #[pg_test] + fn test_condition_waiters_table_shape() { + // instance_id holds the duroxide instance id, which for a loop subtree is + // a composite like "{parent}::{execution}::{node}", so it cannot be the + // 8-char df instance id type used elsewhere. + let cols = Spi::get_one::( + "SELECT pg_catalog.string_agg( + a.attname || ':' || pg_catalog.format_type(a.atttypid, a.atttypmod), + ',' ORDER BY a.attnum) + FROM pg_catalog.pg_attribute a + WHERE a.attrelid = 'df.condition_waiters'::pg_catalog.regclass + AND a.attnum > 0 AND NOT a.attisdropped", + ) + .unwrap() + .unwrap(); + assert_eq!( + cols, + "instance_id:text,node_id:text,notify_key:text,created_at:timestamp with time zone" + ); + } + + #[pg_test] + fn test_condition_waiters_indexed_by_notify_key() { + let n = Spi::get_one::( + "SELECT pg_catalog.count(*) FROM pg_catalog.pg_indexes + WHERE schemaname = 'df' AND tablename = 'condition_waiters' + AND indexdef LIKE '%notify_key%'", + ) + .unwrap() + .unwrap(); + assert!(n >= 1, "expected an index on notify_key, found {n}"); + } + #[pg_test] fn test_wait_for_signal_in_sequence() { let sql_node = crate::dsl::sql("SELECT 1"); diff --git a/src/orchestrations/execute_function_graph.rs b/src/orchestrations/execute_function_graph.rs index d2e42dbb..7581306d 100644 --- a/src/orchestrations/execute_function_graph.rs +++ b/src/orchestrations/execute_function_graph.rs @@ -564,6 +564,9 @@ async fn execute_node_inner( "then" => execute_then_node(ctx, graph, node, node_id, results, exec_ctx).await, "sleep" => execute_sleep_node(ctx, node, node_id).await, "wait_schedule" => execute_wait_schedule_node(ctx, node, node_id).await, + "wait_condition" => { + execute_wait_condition_node(ctx, node, node_id, results, exec_ctx, &sys_vars).await + } "loop" => execute_loop_node(ctx, graph, node, node_id, results, exec_ctx).await, "if" => execute_if_node(ctx, graph, node, node_id, results, exec_ctx).await, "join" => execute_join_node(ctx, graph, node, node_id, results, exec_ctx).await, @@ -1681,6 +1684,176 @@ async fn execute_http_multipart_node( Ok(result) } +/// Maximum predicate evaluations in a single `df.wait_for_condition()` node +/// before it warns about unbounded history growth. +/// +/// The design calls for abandoning the loop iteration here so `continue_as_new` +/// truncates history. That unwind arrives with the failure-policy work; until +/// then the node keeps waiting and only traces. +const CONDITION_CHECK_WARN_THRESHOLD: u64 = 100; + +async fn execute_wait_condition_node( + ctx: &OrchestrationContext, + node: &FunctionNode, + node_id: &str, + results: &mut HashMap, + exec_ctx: &ExecutionContext, + sys_vars: &SystemVars, +) -> NodeResult { + let config_str = node + .query + .as_ref() + .ok_or_else(|| format!("WAIT_CONDITION node {node_id} has no config"))?; + + let config: serde_json::Value = serde_json::from_str(config_str) + .map_err(|e| format!("Invalid WAIT_CONDITION config: {e}"))?; + + let condition = config["condition"] + .as_str() + .ok_or_else(|| "WAIT_CONDITION missing condition".to_string())?; + let interval_secs = config["max_check_interval_secs"] + .as_i64() + .ok_or_else(|| "WAIT_CONDITION missing max_check_interval_secs".to_string())?; + let interval = check_interval(interval_secs)?; + let notify_key = config["notify_key"].as_str(); + + let substituted = substitute_all(condition, results, &exec_ctx.vars, sys_vars)?; + let predicate_sql = crate::types::wrap_condition_sql(&substituted); + + let instance_id = ctx.instance_id(); + let waiter_input = serde_json::json!({ + "instance_id": instance_id, + "node_id": node_id, + "notify_key": notify_key, + }) + .to_string(); + + // Register before the first evaluation so the two overlap: the evaluation + // catches anything already true, the subscription catches anything from + // then on, and nothing falls between them. + if notify_key.is_some() { + ctx.schedule_activity( + activities::register_condition_waiter::NAME, + waiter_input.clone(), + ) + .await?; + } + + // Must agree with the listener's enqueue_event target; both go through + // types::condition_queue_name so the two can never drift apart. + let queue = notify_key.map(crate::types::condition_queue_name); + let predicate_input = serde_json::json!({ + "query": predicate_sql, + "submitted_by": node.submitted_by, + "database": node.database, + "read_only": true, + }) + .to_string(); + + ctx.trace_info(format!( + "Waiting for condition (max_check_interval: {interval_secs}s{}): {predicate_sql}", + notify_key + .map(|k| format!(", notify_key: {k}")) + .unwrap_or_default() + )); + + let mut checks: u64 = 0; + // One subscription is held across iterations and only recreated once it has + // actually delivered. Recreating it every iteration would work — a dropped + // DurableFuture is marked cancelled and duroxide skips cancelled + // subscriptions when matching arrivals in FIFO order, so nothing is lost — + // but it writes a QueueSubscribed/QueueSubscriptionCancelled pair per check. + // On a wait that polls for hours that churn dominates history, and duroxide + // rescans every prior subscription to compute an arrival index, making the + // wait quadratic in the number of checks. + let mut event_fut = queue.as_ref().map(|q| ctx.dequeue_event(q.clone())); + loop { + // The subscription must exist before the predicate runs, otherwise a + // notification landing during evaluation is lost. dequeue_event emits + // its action when the future is created, and it is a mailbox, so an + // event that arrives while nothing is parked is buffered rather than + // dropped. + if event_fut.is_none() { + if let Some(q) = queue.as_ref() { + event_fut = Some(ctx.dequeue_event(q.clone())); + } + } + + let raw = ctx + .schedule_activity(activities::execute_sql::NAME, predicate_input.clone()) + .await?; + checks += 1; + + if condition_met(&raw)? { + break; + } + + if checks == CONDITION_CHECK_WARN_THRESHOLD { + ctx.trace_info(format!( + "Condition node {node_id} has evaluated {checks} times without firing; \ + history will keep growing until the iteration unwinds" + )); + } + + let timer_fut = ctx.schedule_timer(interval); + match event_fut.as_mut() { + // select2 polls in argument order, so a notification that is already + // buffered wins over a timer that fired in the same turn. + Some(event) => match ctx.select2(event, timer_fut).await { + duroxide::Either2::First(_) => { + // Consumed. The next iteration subscribes again. + event_fut = None; + ctx.trace_info("Condition notified; re-evaluating"); + } + duroxide::Either2::Second(()) => { + // The subscription is untouched and stays live. + ctx.trace_info("Condition check interval elapsed; re-evaluating") + } + }, + None => timer_fut.await, + } + } + + if notify_key.is_some() { + ctx.schedule_activity(activities::unregister_condition_waiter::NAME, waiter_input) + .await?; + } + + let result = r#"{"condition_met": true}"#.to_string(); + if let Some(name) = &node.result_name { + results.insert(name.clone(), result.clone()); + } + Ok(result) +} + +/// Validate a `WAIT_CONDITION` check interval and convert it to a `Duration`. +/// +/// `df.wait_for_condition()` already rejects sub-second intervals, but a graph +/// can also be hand-written as raw node JSON. Rejecting here keeps a negative +/// value from widening through `as u64` into an effectively infinite timer. +fn check_interval(secs: i64) -> Result { + if secs < 1 { + return Err(format!( + "WAIT_CONDITION max_check_interval_secs must be at least 1, got {secs}" + )); + } + Ok(Duration::from_secs(secs as u64)) +} + +/// Read the single `condition_met` boolean out of an `execute_sql` result. +/// +/// The predicate is wrapped as `SELECT () IS TRUE AS condition_met`, +/// so PostgreSQL has already reduced every case to one boolean row: no rows and +/// NULL are false, more than one row and a non-boolean expression are errors +/// raised before we get here. +fn condition_met(raw: &str) -> Result { + let parsed: serde_json::Value = + serde_json::from_str(raw).map_err(|e| format!("Invalid condition result: {e}"))?; + parsed["rows"][0]["condition_met"] + .as_bool() + .ok_or_else(|| format!("Condition predicate returned no boolean condition_met: {raw}")) +} + async fn execute_signal_node( ctx: &OrchestrationContext, node: &FunctionNode, @@ -1764,6 +1937,37 @@ async fn execute_signal_node( mod tests { use super::*; + #[test] + fn check_interval_accepts_one_second_and_above() { + assert_eq!(check_interval(1).unwrap(), Duration::from_secs(1)); + assert_eq!(check_interval(300).unwrap(), Duration::from_secs(300)); + } + + /// A hand-written node could carry a negative interval, which `as u64` + /// would widen into an effectively infinite timer. + #[test] + fn check_interval_rejects_zero_and_negative() { + assert!(check_interval(0).is_err()); + assert!(check_interval(-1).is_err()); + assert!(check_interval(i64::MIN).is_err()); + } + + #[test] + fn condition_met_reads_the_wrapped_boolean() { + assert!(condition_met(r#"{"rows":[{"condition_met":true}]}"#).unwrap()); + assert!(!condition_met(r#"{"rows":[{"condition_met":false}]}"#).unwrap()); + } + + /// `() IS TRUE` never yields NULL or an empty result, so anything + /// else means the predicate did not go through the wrapper. + #[test] + fn condition_met_rejects_a_missing_or_non_boolean_value() { + assert!(condition_met(r#"{"rows":[]}"#).is_err()); + assert!(condition_met(r#"{"rows":[{"condition_met":null}]}"#).is_err()); + assert!(condition_met(r#"{"rows":[{"count":1}]}"#).is_err()); + assert!(condition_met("not json").is_err()); + } + /// Build an envelope JSON string the way `execute_subtree` serializes a `SubtreeEnvelope`. /// When `control` is `None` the field is omitted entirely, reproducing an envelope recorded /// by a pre-#148 binary (<= v0.2.2) that had no `control` field. diff --git a/src/registry.rs b/src/registry.rs index 8bcc79b7..e5e34282 100644 --- a/src/registry.rs +++ b/src/registry.rs @@ -20,6 +20,8 @@ pub fn create_activity_registry(pool: Arc, semaphore: Arc) -> let node_status_pool = pool.clone(); let http_pool = pool.clone(); let multipart_pool = pool.clone(); + let register_pool = pool.clone(); + let unregister_pool = pool.clone(); ActivityRegistry::builder() .register(activities::execute_sql::NAME, move |ctx: ActivityContext, input_json: String| { @@ -46,6 +48,14 @@ pub fn create_activity_registry(pool: Arc, semaphore: Arc) -> let pool = multipart_pool.clone(); async move { activities::execute_multipart::execute(ctx, pool, config_json).await } }) + .register(activities::register_condition_waiter::NAME, move |ctx: ActivityContext, input_json: String| { + let pool = register_pool.clone(); + async move { activities::register_condition_waiter::execute(ctx, pool, input_json).await } + }) + .register(activities::unregister_condition_waiter::NAME, move |ctx: ActivityContext, input_json: String| { + let pool = unregister_pool.clone(); + async move { activities::unregister_condition_waiter::execute(ctx, pool, input_json).await } + }) .build() } diff --git a/src/types.rs b/src/types.rs index 39becd3a..65010688 100644 --- a/src/types.rs +++ b/src/types.rs @@ -447,6 +447,25 @@ pub fn evaluate_condition(result: &str) -> Result { Ok(is_truthy(&serde_json::Value::String(result.to_string()))) } +/// Wrap a user predicate so it is evaluated with SQL's own truth rules. +/// +/// The predicate becomes a scalar subquery tested with `IS TRUE`, which gives +/// PostgreSQL's existing semantics for free: exactly one column, zero rows or +/// NULL read as false, and more than one row is a cardinality error. It also +/// constrains the predicate to a single `SELECT`, since nothing else parses in +/// subquery position. +pub fn wrap_condition_sql(condition: &str) -> String { + let trimmed = condition.trim().trim_end_matches(';').trim_end(); + format!("SELECT ({trimmed}) IS TRUE AS condition_met") +} + +/// The duroxide queue a `df.wait_for_condition()` node subscribes to for a +/// given `notify_key`. Shared by the orchestration and the NOTIFY listener so +/// the two cannot drift. +pub fn condition_queue_name(notify_key: &str) -> String { + format!("df:condition:{notify_key}") +} + pub fn is_truthy(value: &serde_json::Value) -> bool { match value { serde_json::Value::Bool(b) => *b, @@ -1329,6 +1348,7 @@ pub const VALID_NODE_TYPES: &[&str] = &[ "RACE", "SLEEP", "WAIT_SCHEDULE", + "WAIT_CONDITION", "HTTP", "HTTP_MULTIPART", "SIGNAL", @@ -1696,6 +1716,14 @@ mod tests { use super::*; use serde_json::json; + #[test] + fn condition_queue_name_is_namespaced() { + assert_eq!( + condition_queue_name("segments:docs_idx"), + "df:condition:segments:docs_idx" + ); + } + #[test] fn durofut_raw_children_preserve_wire_format() { let json = r#"{"node_type":"THEN","left_node":{"node_type":"SQL","query":"SELECT 1"},"right_node":{"node_type":"SQL","query":"SELECT 2"}}"#; diff --git a/src/worker.rs b/src/worker.rs index b99b1481..5ae54b34 100644 --- a/src/worker.rs +++ b/src/worker.rs @@ -1000,6 +1000,14 @@ async fn run_until_extension_dropped_or_shutdown( let client = Client::new(duroxide_store.clone()); + // Wakes df.wait_for_condition() waiters on NOTIFY. Purely an accelerator: + // if it dies, every waiter still fires within its max_check_interval. + let condition_listener = tokio::spawn(crate::condition_listener::run( + postgres_connection_string(), + Arc::new(maintenance_pool.clone()), + client.clone(), + )); + let mut drop_check = tokio::time::interval(drop_poll_interval); drop_check.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay); @@ -1083,13 +1091,47 @@ async fn run_until_extension_dropped_or_shutdown( Ok(_) => {} Err(e) => log!("pg_durable: reclaiming orphaned engine records failed: {e}"), } + + sweep_condition_waiters(maintenance_pool).await; } } } + condition_listener.abort(); teardown_runtime(duroxide_runtime, duroxide_store).await; } +/// Delete waiter rows whose df instance is no longer active. +/// +/// A cancelled or crashed instance never runs its unregister activity, so +/// without this the row would leak. The stored id is the duroxide instance id; +/// its first `::` segment is the df instance id for both root and subtree +/// instances, since `subtree_instance_id` composes onto the parent's id. +async fn sweep_condition_waiters(pool: &sqlx::PgPool) { + let result = sqlx::query( + "DELETE FROM df.condition_waiters w \ + WHERE NOT EXISTS ( \ + SELECT 1 FROM df.instances i \ + WHERE i.id = pg_catalog.split_part(w.instance_id, '::', 1) \ + AND i.status IN ('pending', 'running') \ + )", + ) + .execute(pool) + .await; + + match result { + Ok(r) if r.rows_affected() > 0 => { + log!( + "pg_durable: removed {} orphaned condition waiter row(s)", + r.rows_affected() + ); + } + Ok(_) => {} + // Absent on a schema older than 0.2.6 (Scenario B1); nothing to sweep. + Err(e) => log!("pg_durable: sweeping condition waiters failed: {e}"), + } +} + /// Shut down a duroxide runtime and close its store pool. /// /// Two callers, and the branch below distinguishes them by cause rather than by diff --git a/tests/e2e/sql/67_wait_for_condition.sql b/tests/e2e/sql/67_wait_for_condition.sql new file mode 100644 index 00000000..9348a509 --- /dev/null +++ b/tests/e2e/sql/67_wait_for_condition.sql @@ -0,0 +1,368 @@ +-- Copyright (c) Microsoft Corporation. +-- Licensed under the PostgreSQL License. + +-- Test: df.wait_for_condition() +-- Verifies that a predicate that is already true fires immediately, that one +-- that becomes true later fires on the interval backstop, that a pg_notify on +-- the declared notify_key beats a long backstop, that a second notification is +-- still delivered after the first one was consumed, that the waiter registry is +-- populated while parked and cleared afterwards, and that a predicate with +-- side effects fails instead of writing. + +DROP TABLE IF EXISTS test_cond_gate; +CREATE TABLE test_cond_gate (name TEXT PRIMARY KEY, ready BOOLEAN NOT NULL); +INSERT INTO test_cond_gate VALUES ('already', true), ('later', false), ('notified', false), ('twice', false); + +DROP TABLE IF EXISTS test_cond_done; +CREATE TABLE test_cond_done (name TEXT PRIMARY KEY); + +CREATE TEMP TABLE _cond_state (name TEXT PRIMARY KEY, instance_id TEXT, elapsed NUMERIC); + +-- --------------------------------------------------------------------------- +-- 1. Already true: fires without waiting out the interval. +-- --------------------------------------------------------------------------- +INSERT INTO _cond_state (name, instance_id) +SELECT 'already', df.start( + df.wait_for_condition( + 'SELECT ready FROM test_cond_gate WHERE name = ''already''', + '30s' + ) ~> 'INSERT INTO test_cond_done VALUES (''already'')', + 'test-cond-already' +); + +DO $$ +DECLARE + inst_id TEXT; + status TEXT; + started TIMESTAMPTZ := clock_timestamp(); + attempts INT := 0; +BEGIN + SELECT instance_id INTO inst_id FROM _cond_state WHERE name = 'already'; + LOOP + SELECT s INTO status FROM df.status(inst_id) s; + EXIT WHEN lower(status) IN ('completed', 'failed', 'cancelled') OR attempts > 300; + PERFORM pg_sleep(0.1); + attempts := attempts + 1; + END LOOP; + + IF lower(status) != 'completed' THEN + RAISE EXCEPTION 'TEST FAILED (already true): status = %', status; + END IF; + + -- A full interval is 30s; anything under that proves the first evaluation + -- fired rather than the backstop. + UPDATE _cond_state + SET elapsed = extract(epoch FROM clock_timestamp() - started) + WHERE name = 'already'; + + IF (SELECT elapsed FROM _cond_state WHERE name = 'already') > 20 THEN + RAISE EXCEPTION 'TEST FAILED (already true): waited % seconds, expected immediate', + (SELECT elapsed FROM _cond_state WHERE name = 'already'); + END IF; + + IF NOT EXISTS (SELECT 1 FROM test_cond_done WHERE name = 'already') THEN + RAISE EXCEPTION 'TEST FAILED (already true): body did not run'; + END IF; +END $$; + +-- --------------------------------------------------------------------------- +-- 2. Becomes true later, no notification: the interval backstop fires it. +-- --------------------------------------------------------------------------- +INSERT INTO _cond_state (name, instance_id) +SELECT 'later', df.start( + df.wait_for_condition( + 'SELECT ready FROM test_cond_gate WHERE name = ''later''', + '1s' + ) ~> 'INSERT INTO test_cond_done VALUES (''later'')', + 'test-cond-later' +); + +DO $$ +DECLARE + inst_id TEXT; + status TEXT; +BEGIN + SELECT instance_id INTO inst_id FROM _cond_state WHERE name = 'later'; + + -- Let it park on a false predicate first, so this is not case 1 again. + PERFORM pg_sleep(2); + SELECT s INTO status FROM df.status(inst_id) s; + IF lower(status) NOT IN ('pending', 'running') THEN + RAISE EXCEPTION 'TEST FAILED (becomes true): instance left the wait early, status = %', status; + END IF; +END $$; + +-- Must be its own statement: a write inside the polling block would not be +-- visible to the worker's separate connection until that block committed. +UPDATE test_cond_gate SET ready = true WHERE name = 'later'; + +DO $$ +DECLARE + inst_id TEXT; + status TEXT; + attempts INT := 0; +BEGIN + SELECT instance_id INTO inst_id FROM _cond_state WHERE name = 'later'; + LOOP + SELECT s INTO status FROM df.status(inst_id) s; + EXIT WHEN lower(status) IN ('completed', 'failed', 'cancelled') OR attempts > 300; + PERFORM pg_sleep(0.1); + attempts := attempts + 1; + END LOOP; + + IF lower(status) != 'completed' THEN + RAISE EXCEPTION 'TEST FAILED (becomes true): status = %', status; + END IF; + + IF NOT EXISTS (SELECT 1 FROM test_cond_done WHERE name = 'later') THEN + RAISE EXCEPTION 'TEST FAILED (becomes true): body did not run'; + END IF; +END $$; + +-- --------------------------------------------------------------------------- +-- 3. Notification beats a 5-minute backstop, and the waiter registry is +-- populated while parked and cleared once the predicate fires. +-- --------------------------------------------------------------------------- +INSERT INTO _cond_state (name, instance_id) +SELECT 'notified', df.start( + df.wait_for_condition( + 'SELECT ready FROM test_cond_gate WHERE name = ''notified''', + '5min', + notify_key => 'test_cond_key' + ) ~> 'INSERT INTO test_cond_done VALUES (''notified'')', + 'test-cond-notified' +); + +DO $$ +DECLARE + inst_id TEXT; + attempts INT := 0; + waiters INT; +BEGIN + SELECT instance_id INTO inst_id FROM _cond_state WHERE name = 'notified'; + + -- Wait for the node to register before notifying. + LOOP + SELECT count(*) INTO waiters + FROM df.condition_waiters + WHERE notify_key = 'test_cond_key' + AND split_part(instance_id, '::', 1) = inst_id; + EXIT WHEN waiters > 0 OR attempts > 200; + PERFORM pg_sleep(0.1); + attempts := attempts + 1; + END LOOP; + + IF waiters = 0 THEN + RAISE EXCEPTION 'TEST FAILED (notify): no waiter row registered for %', inst_id; + END IF; +END $$; + +-- Own statements again: both the gate write and the notification are only +-- visible/delivered at commit. +UPDATE test_cond_gate SET ready = true WHERE name = 'notified'; +SELECT pg_notify('pg_durable_condition', 'test_cond_key'); + +DO $$ +DECLARE + inst_id TEXT; + status TEXT; + started TIMESTAMPTZ := clock_timestamp(); + attempts INT := 0; + waiters INT; +BEGIN + SELECT instance_id INTO inst_id FROM _cond_state WHERE name = 'notified'; + + LOOP + SELECT s INTO status FROM df.status(inst_id) s; + EXIT WHEN lower(status) IN ('completed', 'failed', 'cancelled') OR attempts > 300; + PERFORM pg_sleep(0.1); + attempts := attempts + 1; + END LOOP; + + IF lower(status) != 'completed' THEN + RAISE EXCEPTION 'TEST FAILED (notify): status = %', status; + END IF; + + -- The backstop is 5 minutes, so completing at all within the 30s poll + -- budget proves the notification woke it. + IF extract(epoch FROM clock_timestamp() - started) > 30 THEN + RAISE EXCEPTION 'TEST FAILED (notify): took % seconds, backstop must not have been beaten', + extract(epoch FROM clock_timestamp() - started); + END IF; + + SELECT count(*) INTO waiters + FROM df.condition_waiters + WHERE split_part(instance_id, '::', 1) = inst_id; + IF waiters > 0 THEN + RAISE EXCEPTION 'TEST FAILED (notify): % waiter row(s) left registered', waiters; + END IF; +END $$; + +-- --------------------------------------------------------------------------- +-- 4. A second notification is still delivered after the first was consumed. +-- The node parks on one subscription, wakes on notification #1 with the +-- predicate still false, and must re-subscribe so notification #2 lands. +-- The backstop is 5 minutes, so only the notification path can complete it. +-- --------------------------------------------------------------------------- +INSERT INTO _cond_state (name, instance_id) +SELECT 'twice', df.start( + df.wait_for_condition( + 'SELECT ready FROM test_cond_gate WHERE name = ''twice''', + '5min', + notify_key => 'test_cond_key2' + ) ~> 'INSERT INTO test_cond_done VALUES (''twice'')', + 'test-cond-twice' +); + +DO $$ +DECLARE + inst_id TEXT; + attempts INT := 0; + waiters INT; +BEGIN + SELECT instance_id INTO inst_id FROM _cond_state WHERE name = 'twice'; + + LOOP + SELECT count(*) INTO waiters + FROM df.condition_waiters + WHERE notify_key = 'test_cond_key2' + AND split_part(instance_id, '::', 1) = inst_id; + EXIT WHEN waiters > 0 OR attempts > 200; + PERFORM pg_sleep(0.1); + attempts := attempts + 1; + END LOOP; + + IF waiters = 0 THEN + RAISE EXCEPTION 'TEST FAILED (twice): no waiter row registered for %', inst_id; + END IF; +END $$; + +-- Notification #1: the gate is still false, so this only burns the node's +-- current subscription. +SELECT pg_notify('pg_durable_condition', 'test_cond_key2'); + +-- Give the node time to consume #1, re-check, and re-subscribe. This also +-- clears the listener's per-key suppression window so #2 is not deduplicated. +SELECT pg_sleep(3); + +DO $$ +DECLARE + inst_id TEXT; + status TEXT; +BEGIN + SELECT instance_id INTO inst_id FROM _cond_state WHERE name = 'twice'; + SELECT s INTO status FROM df.status(inst_id) s; + IF lower(status) != 'running' THEN + RAISE EXCEPTION 'TEST FAILED (twice): woke early on notification #1, status = %', status; + END IF; +END $$; + +-- Notification #2, now with the predicate satisfiable. +UPDATE test_cond_gate SET ready = true WHERE name = 'twice'; +SELECT pg_notify('pg_durable_condition', 'test_cond_key2'); + +DO $$ +DECLARE + inst_id TEXT; + status TEXT; + started TIMESTAMPTZ := clock_timestamp(); + attempts INT := 0; +BEGIN + SELECT instance_id INTO inst_id FROM _cond_state WHERE name = 'twice'; + + LOOP + SELECT s INTO status FROM df.status(inst_id) s; + EXIT WHEN lower(status) IN ('completed', 'failed', 'cancelled') OR attempts > 300; + PERFORM pg_sleep(0.1); + attempts := attempts + 1; + END LOOP; + + IF lower(status) != 'completed' THEN + RAISE EXCEPTION 'TEST FAILED (twice): status = %', status; + END IF; + + IF extract(epoch FROM clock_timestamp() - started) > 30 THEN + RAISE EXCEPTION 'TEST FAILED (twice): took % seconds, notification #2 was not delivered', + extract(epoch FROM clock_timestamp() - started); + END IF; + + IF NOT EXISTS (SELECT 1 FROM test_cond_done WHERE name = 'twice') THEN + RAISE EXCEPTION 'TEST FAILED (twice): body did not run'; + END IF; +END $$; + +-- --------------------------------------------------------------------------- +-- 5. A predicate with side effects fails the node instead of writing. +-- --------------------------------------------------------------------------- +DROP TABLE IF EXISTS test_cond_sideeffect; +CREATE TABLE test_cond_sideeffect (n INT); + +CREATE TEMP TABLE _cond_write AS +SELECT df.start( + df.wait_for_condition( + 'SELECT true FROM (INSERT INTO test_cond_sideeffect VALUES (1) RETURNING n) w', + '1s' + ), + 'test-cond-readonly' +) AS instance_id; + +DO $$ +DECLARE + inst_id TEXT; + status TEXT; + attempts INT := 0; +BEGIN + SELECT instance_id INTO inst_id FROM _cond_write; + LOOP + SELECT s INTO status FROM df.status(inst_id) s; + EXIT WHEN lower(status) IN ('completed', 'failed', 'cancelled') OR attempts > 300; + PERFORM pg_sleep(0.1); + attempts := attempts + 1; + END LOOP; + + IF lower(status) != 'failed' THEN + RAISE EXCEPTION 'TEST FAILED (read-only): expected failed, got %', status; + END IF; + + IF EXISTS (SELECT 1 FROM test_cond_sideeffect) THEN + RAISE EXCEPTION 'TEST FAILED (read-only): predicate wrote a row'; + END IF; +END $$; + +-- --------------------------------------------------------------------------- +-- 6. A non-boolean predicate is rejected rather than coerced. +-- --------------------------------------------------------------------------- +CREATE TEMP TABLE _cond_count AS +SELECT df.start( + df.wait_for_condition('SELECT count(*) FROM test_cond_gate', '1s'), + 'test-cond-count' +) AS instance_id; + +DO $$ +DECLARE + inst_id TEXT; + status TEXT; + attempts INT := 0; +BEGIN + SELECT instance_id INTO inst_id FROM _cond_count; + LOOP + SELECT s INTO status FROM df.status(inst_id) s; + EXIT WHEN lower(status) IN ('completed', 'failed', 'cancelled') OR attempts > 300; + PERFORM pg_sleep(0.1); + attempts := attempts + 1; + END LOOP; + + IF lower(status) != 'failed' THEN + RAISE EXCEPTION 'TEST FAILED (non-boolean): expected failed, got %', status; + END IF; +END $$; + +-- Cleanup +DROP TABLE _cond_state; +DROP TABLE _cond_write; +DROP TABLE _cond_count; +DROP TABLE test_cond_gate; +DROP TABLE test_cond_done; +DROP TABLE test_cond_sideeffect; + +SELECT 'TEST PASSED' AS result;