diff --git a/USER_GUIDE.md b/USER_GUIDE.md index ec924fa8..297f04c9 100644 --- a/USER_GUIDE.md +++ b/USER_GUIDE.md @@ -1329,6 +1329,21 @@ SELECT df.status('a1b2c3d4'); SELECT df.result('a1b2c3d4'); ``` +### Worker Liveness + +Check whether the background worker is alive and healthy: + +```sql +SELECT epoch_id, started_at, last_seen_at, + now() - last_seen_at AS time_since_last_heartbeat + FROM df._worker_epoch; +``` + +- `time_since_last_heartbeat < 15 seconds` → worker is alive (recent heartbeat) +- No rows in `df._worker_epoch` or large `time_since_last_heartbeat` → worker is likely down or hasn't initialized yet + +The background worker updates `last_seen_at` every ~5 seconds as part of its normal operation. + --- ## User Isolation & Privileges diff --git a/docs/extension_lifecycle.md b/docs/extension_lifecycle.md index b353953a..41aeaf6e 100644 --- a/docs/extension_lifecycle.md +++ b/docs/extension_lifecycle.md @@ -206,7 +206,7 @@ Pure `pg_extension` polling can miss a DROP → CREATE cycle if both happen betw The **epoch sentinel** solves this: -- Extension SQL declares `df._worker_epoch (epoch_id UUID PRIMARY KEY, started_at TIMESTAMPTZ)`. +- Extension SQL declares `df._worker_epoch (epoch_id UUID PRIMARY KEY, started_at TIMESTAMPTZ, last_seen_at TIMESTAMPTZ)`. - After init, the BGW inserts a row with a fresh UUID. - The running-state poll checks: "does my UUID still exist?" instead of "does the extension exist?" - Three outcomes: diff --git a/scripts/test-e2e-local.sh b/scripts/test-e2e-local.sh index bd2c4450..9c001c5b 100755 --- a/scripts/test-e2e-local.sh +++ b/scripts/test-e2e-local.sh @@ -313,6 +313,7 @@ for run in $(seq 1 $REPEAT_COUNT); do # 28 drops/creates the extension # 29 uses dblink and creates pg_durable in a different database # 34 creates/drops a database for multi-database testing + # 35 reads df._worker_epoch (internal table) PSQL_USER="$E2E_USER" if [[ "$test_name" == "00_requires_shared_preload" \ || "$test_name" == "22_cross_connection" \ @@ -322,7 +323,8 @@ for run in $(seq 1 $REPEAT_COUNT); do || "$test_name" == "27_user_isolation" \ || "$test_name" == "28_bgw_lifecycle" \ || "$test_name" == "29_database_validation" \ - || "$test_name" == "34_multi_database" ]]; then + || "$test_name" == "34_multi_database" \ + || "$test_name" == "35_heartbeat_liveness" ]]; then PSQL_USER="$PG_USER" fi diff --git a/src/lib.rs b/src/lib.rs index 6b4a301e..c63b0766 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -141,7 +141,8 @@ CREATE TABLE IF NOT EXISTS df.vars ( -- recreation even though the extension is always "present" in pg_extension. CREATE TABLE IF NOT EXISTS df._worker_epoch ( epoch_id UUID PRIMARY KEY, - started_at TIMESTAMPTZ DEFAULT now() + started_at TIMESTAMPTZ DEFAULT now(), + last_seen_at TIMESTAMPTZ DEFAULT now() ); "#, name = "create_tables", diff --git a/src/worker.rs b/src/worker.rs index 59b2390c..8aaab496 100644 --- a/src/worker.rs +++ b/src/worker.rs @@ -260,7 +260,7 @@ async fn write_epoch_sentinel(pool: &sqlx::PgPool) -> Result Result bool { - let result: Result<(bool,), sqlx::Error> = - sqlx::query_as("SELECT EXISTS(SELECT 1 FROM df._worker_epoch WHERE epoch_id = $1::uuid)") - .bind(epoch_id) - .fetch_one(pool) - .await; + let result = sqlx::query( + "UPDATE df._worker_epoch SET last_seen_at = now() WHERE epoch_id = $1::uuid RETURNING epoch_id", + ) + .bind(epoch_id) + .fetch_optional(pool) + .await; // Query error (table/schema gone) ⇒ treat as "dropped" - result.map(|(exists,)| exists).unwrap_or(false) + // None ⇒ row missing (drop+recreated) + matches!(result, Ok(Some(_))) } async fn run_until_extension_dropped_or_shutdown( diff --git a/tests/e2e/sql/35_heartbeat_liveness.sql b/tests/e2e/sql/35_heartbeat_liveness.sql new file mode 100644 index 00000000..c45675d4 --- /dev/null +++ b/tests/e2e/sql/35_heartbeat_liveness.sql @@ -0,0 +1,41 @@ +-- Test: Worker heartbeat liveness (last_seen_at advances over time) +-- Validates that the background worker updates df._worker_epoch.last_seen_at +-- on each poll tick (~5 seconds). +-- Requires superuser: reads internal df._worker_epoch table. + +DO $$ +DECLARE + ts1 TIMESTAMPTZ; + ts2 TIMESTAMPTZ; + attempts INT := 0; +BEGIN + -- Wait for sentinel row to appear (worker may still be initializing) + LOOP + SELECT last_seen_at INTO ts1 FROM df._worker_epoch LIMIT 1; + EXIT WHEN ts1 IS NOT NULL OR attempts >= 150; + PERFORM pg_sleep(0.1); + attempts := attempts + 1; + END LOOP; + + IF ts1 IS NULL THEN + RAISE EXCEPTION 'TEST FAILED: no sentinel row after 15s — worker not running'; + END IF; + + -- Wait for last_seen_at to advance (poll tick is ~5s, allow up to 15s) + attempts := 0; + LOOP + PERFORM pg_sleep(1); + attempts := attempts + 1; + + SELECT last_seen_at INTO ts2 FROM df._worker_epoch LIMIT 1; + EXIT WHEN ts2 > ts1 OR attempts >= 15; + END LOOP; + + IF ts2 <= ts1 THEN + RAISE EXCEPTION 'TEST FAILED: last_seen_at did not advance after 15s (ts1=%, ts2=%)', ts1, ts2; + END IF; + + RAISE NOTICE 'PASSED: last_seen_at advanced from % to % after % seconds', ts1, ts2, attempts; +END $$; + +SELECT 'TEST PASSED' AS result;