Add heartbeat to background worker epoch sentinel

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

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:

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
 

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",

src/worker.rs

@@ -260,7 +260,7 @@ async fn write_epoch_sentinel(pool: &sqlx::PgPool) -> Result<String, sqlx::Error
     sqlx::query("DELETE FROM df._worker_epoch")
         .execute(pool)
         .await?;
-    sqlx::query("INSERT INTO df._worker_epoch (epoch_id) VALUES ($1::uuid)")
+    sqlx::query("INSERT INTO df._worker_epoch (epoch_id, started_at, last_seen_at) VALUES ($1::uuid, now(), now())")
         .bind(&epoch_id)
         .execute(pool)
         .await?;
@@ -273,14 +273,16 @@ async fn write_epoch_sentinel(pool: &sqlx::PgPool) -> Result<String, sqlx::Error
 /// `false` when it is missing or the query fails (extension dropped
 /// or drop+recreated).
 async fn check_epoch_sentinel(pool: &sqlx::PgPool, epoch_id: &str) -> 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(

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;