Address review feedback: add pause timeout recovery and restrict privileges

Add best-effort resume_apply_workers call in PG_CATCH error path so
workers are unblocked if slot creation fails.  Guard with
PQstatus check to avoid hanging on a broken connection.

Add ConditionVariableTimedSleep to both pause check points so workers
self-recover after spock.pause_timeout seconds if resume is never
called (e.g., caller crashed or connection lost).

Extract maybe_pause_for_slot_creation() helper to deduplicate the
pause logic in begin_replication_step and handle_commit.

Revoke EXECUTE on pause_apply_workers and resume_apply_workers from
PUBLIC to prevent unprivileged users from blocking apply workers.

Author: mason-sharp

sql/spock--5.0.6--6.0.0-devel.sql

@@ -35,11 +35,15 @@ RETURNS void
 AS 'MODULE_PATHNAME', 'spock_pause_apply_workers'
 LANGUAGE C VOLATILE;
 
+REVOKE ALL ON FUNCTION spock.pause_apply_workers() FROM PUBLIC;
+
 CREATE FUNCTION spock.resume_apply_workers()
 RETURNS void
 AS 'MODULE_PATHNAME', 'spock_resume_apply_workers'
 LANGUAGE C VOLATILE;
 
+REVOKE ALL ON FUNCTION spock.resume_apply_workers() FROM PUBLIC;
+
 -- Read peer progress (ros.remote_lsn) for all peer subscriptions.
 -- Called while apply workers are paused and the slot's snapshot is imported.
 -- Row 0: header (lsn + snapshot placeholder).  Rows 1+: one progress entry per peer.

sql/spock--6.0.0-devel.sql

@@ -542,11 +542,15 @@ RETURNS void
 AS 'MODULE_PATHNAME', 'spock_pause_apply_workers'
 LANGUAGE C VOLATILE;
 
+REVOKE ALL ON FUNCTION spock.pause_apply_workers() FROM PUBLIC;
+
 CREATE FUNCTION spock.resume_apply_workers()
 RETURNS void
 AS 'MODULE_PATHNAME', 'spock_resume_apply_workers'
 LANGUAGE C VOLATILE;
 
+REVOKE ALL ON FUNCTION spock.resume_apply_workers() FROM PUBLIC;
+
 CREATE PROCEDURE spock.wait_for_sync_event(
 	OUT result          bool,
 	origin_id           oid,

src/spock_apply.c

@@ -438,6 +438,35 @@ action_error_callback(void *arg)
  * existing transaction).
  * Also provide a global snapshot and ensure we run in ApplyMessageContext.
  */
+
+/*
+ * If the pause flag is set (slot creation in progress for add_node),
+ * sleep on the ConditionVariable until resumed or timed out.
+ */
+static void
+maybe_pause_for_slot_creation(void)
+{
+	if (pg_atomic_read_u32(&SpockCtx->pause_apply) != 0)
+	{
+		MyApplyWorker->paused = true;
+		ConditionVariablePrepareToSleep(&SpockCtx->pause_cv);
+		while (pg_atomic_read_u32(&SpockCtx->pause_apply) != 0)
+		{
+			if (ConditionVariableTimedSleep(&SpockCtx->pause_cv,
+										    spock_pause_timeout * 1000L,
+										    WAIT_EVENT_LOGICAL_APPLY_MAIN))
+			{
+				elog(WARNING, "SPOCK: apply worker pause timed out after %ds, resuming",
+					 spock_pause_timeout);
+				pg_atomic_write_u32(&SpockCtx->pause_apply, 0);
+				break;
+			}
+		}
+		ConditionVariableCancelSleep();
+		MyApplyWorker->paused = false;
+	}
+}
+
 static bool
 begin_replication_step(void)
 {
@@ -464,16 +493,7 @@ begin_replication_step(void)
 		 * xid polling.  The previous transaction's commit is fully
 		 * complete, so ros.remote_lsn reflects only committed state.
 		 */
-		if (pg_atomic_read_u32(&SpockCtx->pause_apply) != 0)
-		{
-			MyApplyWorker->paused = true;
-			ConditionVariablePrepareToSleep(&SpockCtx->pause_cv);
-			while (pg_atomic_read_u32(&SpockCtx->pause_apply) != 0)
-				ConditionVariableSleep(&SpockCtx->pause_cv,
-									  WAIT_EVENT_LOGICAL_APPLY_MAIN);
-			ConditionVariableCancelSleep();
-			MyApplyWorker->paused = false;
-		}
+		maybe_pause_for_slot_creation();
 
 		StartTransactionCommand();
 		spock_apply_heap_begin();
@@ -1011,21 +1031,7 @@ handle_commit(StringInfo s)
 
 	in_remote_transaction = false;
 
-	/*
-	 * If slot creation (add_node) is waiting for us, pause here.  The
-	 * commit is fully complete (ros.remote_lsn updated, xid cleared),
-	 * so the snapshot and origin will be consistent.
-	 */
-	if (pg_atomic_read_u32(&SpockCtx->pause_apply) != 0)
-	{
-		MyApplyWorker->paused = true;
-		ConditionVariablePrepareToSleep(&SpockCtx->pause_cv);
-		while (pg_atomic_read_u32(&SpockCtx->pause_apply) != 0)
-			ConditionVariableSleep(&SpockCtx->pause_cv,
-								  WAIT_EVENT_LOGICAL_APPLY_MAIN);
-		ConditionVariableCancelSleep();
-		MyApplyWorker->paused = false;
-	}
+	maybe_pause_for_slot_creation();
 
 	/*
 	 * Stop replay if we're doing limited replay and we've replayed up to the

src/spock_sync.c

@@ -1472,6 +1472,17 @@ spock_sync_subscription(SpockSubscription *sub)
 				 sub->name, sub->slot_name,
 				 edata->message ? edata->message : "");
 
+			/* Best-effort resume of apply workers on the remote node.
+			 * If the connection is broken this will fail silently —
+			 * the workers' CV timeout will recover them. */
+			if (origin_conn && PQstatus(origin_conn) == CONNECTION_OK)
+			{
+				PGresult *rres = PQexec(origin_conn,
+										"SELECT spock.resume_apply_workers()");
+				if (rres)
+					PQclear(rres);
+			}
+
 			FreeErrorData(edata);
 			PG_RE_THROW();
 		}