Add df.wait_for_condition() for condition-based triggering

[Todd J. Green (tjgreen42)]

Implements the wait-for-condition design, included in this PR.

Problem

Durable functions could only be triggered on a wall-clock schedule via df.wait_for_schedule(). A workflow that needs to react to data changes — the motivating case is background compaction of a pg_textsearch BM25 index — had to poll on a cron cadence, trading latency against wasted wakeups.

What this adds

df.wait_for_condition(condition, max_check_interval, notify_key)

Blocks a durable instance until a SQL predicate becomes true. The predicate is evaluated:

1. when the instance starts waiting,
2. whenever a matching pg_notify arrives, and
3. at least once per max_check_interval as a backstop.

notify_key is optional — omit it and the node degenerates to pure interval polling, which is still correct, just higher-latency.

SELECT df.start(
    df.loop(
        df.wait_for_condition(
            'SELECT count(*) > 10 FROM ts.segments WHERE index_id = 1',
            INTERVAL '5 minutes',
            'ts_segments'
        ) ~> df.sql('SELECT ts.compact(1)')
    ),
    'bm25-compactor'
);

The producer side is just pg_notify('pg_durable_condition', 'ts_segments').

Implementation notes

• WAIT_CONDITION node type. Registered in VALID_NODE_TYPES, in both nodes CHECK constraints, and in df.explain(). Worth calling out: missing any one of these fails silently — Durofut::ensure() falls back to treating the node's JSON as a raw SQL string, so the graph builds, df.start() succeeds, and the instance just sits pending. Cost me a debugging cycle; it's now written down in the spec.
• df.condition_waiters registry, indexed on notify_key, populated by register/unregister activities that bracket each wait.
• Listener task in the background worker LISTENs on the pg_durable_condition channel and enqueues a duroxide event to each waiter registered for the payload key. Per-key suppression (1s) collapses notification storms; the listener resyncs on reconnect so nothing is missed across a dropped connection.
• Predicate evaluation runs under default_transaction_read_only, wrapped as SELECT (<condition>) IS TRUE. This delegates to PostgreSQL's own boolean rules rather than inventing new ones: no rows or NULL is false, and a multi-row or non-boolean predicate is an error rather than a silent false.
• Replay safety. ExecuteSqlInput.read_only is skip_serializing_if = "is_false", so pre-existing duroxide histories serialize byte-identically.
• Cleanup. The reconcile tick sweeps waiter rows whose owning instance is no longer running.

dequeue_event/enqueue_event (not raise_event) is the right duroxide primitive here: the action is emitted at future-creation time, so the subscription is durable before the predicate first runs, and it behaves as a mailbox — events that arrive while the instance is unparked are buffered rather than lost.

• One subscription held across checks. The wait loop originally created a fresh dequeue_event per iteration. That is correct — a dropped DurableFuture is marked cancelled and duroxide skips cancelled subscriptions when matching arrivals in FIFO order — but it writes a subscribe/cancel pair to history per check, and duroxide derives an arrival index by scanning every prior subscription for that queue name, so a long wait goes quadratic. The node now holds one subscription across iterations and recreates it only once it has actually delivered (DurableFuture is Unpin, so &mut fut is a future and survives the select2 that borrows it). Measured over a 34-check probe: 6 history events per check → 4, and 35 subscriptions → 1.

Testing

• tests/e2e/sql/67_wait_for_condition.sql — 6 cases: already-true, becomes-true on the interval backstop, NOTIFY beating a 5-minute backstop (plus waiter-row lifecycle), a second NOTIFY after the first was consumed with the predicate still false — the case that fails if the subscription is not recreated — a writing predicate, and a non-boolean predicate.
• 8 new #[pg_test] cases in src/lib.rs (DSL shape, table shape, index), plus unit tests for wrap_condition_sql, condition_queue_name, the read_only serialization contract, and the listener's suppression logic.
• Full local suites: 54/54 E2E, 304/304 unit, 70/70 upgrade (Scenario A schema comparison + B1 backward compat against v0.2.3/v0.2.4/v0.2.5 schemas).
• cargo fmt clean. Clippy adds zero new findings (verified against the main baseline: 27 pre-existing uninlined_format_args before and after).

Deferred

Bounded history — unwinding the orchestration after ~100 condition checks — is intentionally left out. It depends on the unwind mechanism from the sibling failure policy spec, which is being implemented separately. The on_failure interaction cases are deferred for the same reason.

Flag for review

scripts/test-e2e-local.sh and scripts/test-upgrade.sh now honor PGRX_HOME / PG_PORT env overrides (2 lines each). I needed this to run an isolated cluster while another workstream shared ~/.pgrx and port 28817. It's defensible on its own as a parallel-work enabler, but it is scope creep — happy to drop it if you'd rather keep this PR focused.