Fix replay queue spill-to-disk CodeRabbit review findings

- Fix memory leak: spill-read entries during replay were allocated in
  TopMemoryContext but never freed because entry_spilled was only set on
  the first-pass path. Add !entry->from_pq to the free condition so
  replay-path spill-read entries are freed after processing.

- Set GUC minimum for spock.exception_replay_queue_size from -1 to 0,
  removing an undocumented value that silently behaved the same as 0.

- Use explicit subscription name in spill_transaction test when polling
  sub_show_status to avoid ambiguity with multiple subscriptions.

Author: danolivo

src/spock.c

@@ -1159,7 +1159,7 @@ _PG_init(void)
 							"Set to 0 to disable spilling (unlimited memory).",
 							&spock_replay_queue_size,
 							4,
-							-1,
+							0,
 							MAX_KILOBYTES / 1024,
 							PGC_SIGHUP,
 							GUC_UNIT_MB,

src/spock_apply.c

@@ -3053,9 +3053,11 @@ apply_work(PGconn *streamConn)
 					 * In-memory entries (both first-pass and replay) live in
 					 * ApplyReplayContext and are freed by
 					 * MemoryContextReset inside apply_replay_queue_reset,
-					 * called from handle_commit.
+					 * called on applying a COMMIT.
+					 *
+					 * !from_pq catches spill-read entries during replay.
 					 */
-					if (entry_spilled)
+					if (entry_spilled || !entry->from_pq)
 						apply_replay_entry_free(entry);
 				}
 				else if (c == 'k')
@@ -3991,9 +3993,16 @@ apply_replay_spill_write_entry(int len, char *data)
 	Assert(len > 0);
 	Assert(data != NULL);
 
+	/*
+	 * Increment the count before writing so that a partial or failed write
+	 * (ERROR from BufFileWrite) leaves the count higher than the number of
+	 * complete records on disk.  During replay, apply_replay_spill_read_entry()
+	 * will attempt to read this record, hit EOF or a short read, and raise
+	 * ERROR — which triggers a clean worker restart via the outer PG_CATCH.
+	 */
+	apply_replay_spill_count++;
 	BufFileWrite(apply_replay_spill_file, &len, sizeof(int));
 	BufFileWrite(apply_replay_spill_file, data, len);
-	apply_replay_spill_count++;
 }
 
 /*
@@ -4195,11 +4204,6 @@ apply_replay_queue_append_entry(ApplyReplayEntry *entry, StringInfo msg)
 
 		apply_replay_spill_write_entry(msg->len, msg->data);
 
-		/* XXX: keep DEBUG1 logging until spill-to-disk code is proven stable */
-		if (xact_action_counter % 100 == 0)
-			elog(DEBUG1, "SPOCK %s: replay queue spill entry %u: ",
-				 MySubscription->name, xact_action_counter);
-
 		/*
 		 * Move the entry struct from ApplyReplayContext to TopMemoryContext.
 		 * handle_commit calls apply_replay_queue_reset which does
@@ -4209,9 +4213,8 @@ apply_replay_queue_append_entry(ApplyReplayEntry *entry, StringInfo msg)
 		 */
 		oldctx = MemoryContextSwitchTo(TopMemoryContext);
 		mc_entry = (ApplyReplayEntry *) palloc(sizeof(ApplyReplayEntry));
-		/* nosemgrep: palloc above allocates exactly sizeof(ApplyReplayEntry)
-		 * bytes — the same size memcpy copies — so no overflow is possible. */
-		memcpy(mc_entry, entry, sizeof(ApplyReplayEntry));
+		/* palloc and memcpy use the same sizeof — no overflow possible. */
+		memcpy(mc_entry, entry, sizeof(ApplyReplayEntry)); /* nosemgrep */
 		MemoryContextSwitchTo(oldctx);
 
 		pfree(entry);

tests/regress/expected/spill_transaction.out

@@ -267,7 +267,7 @@ COPY test_spill (id) FROM PROGRAM 'seq 1 50000' WITH (FORMAT text);
 DO $$
 BEGIN
   SET LOCAL statement_timeout = '30s';
-  WHILE (SELECT status FROM spock.sub_show_status()) <> 'disabled' LOOP
+  WHILE (SELECT status FROM spock.sub_show_status('test_subscription')) <> 'disabled' LOOP
     PERFORM pg_sleep(0.1);
   END LOOP;
 END;

tests/regress/sql/spill_transaction.sql

@@ -178,7 +178,7 @@ COPY test_spill (id) FROM PROGRAM 'seq 1 50000' WITH (FORMAT text);
 DO $$
 BEGIN
   SET LOCAL statement_timeout = '30s';
-  WHILE (SELECT status FROM spock.sub_show_status()) <> 'disabled' LOOP
+  WHILE (SELECT status FROM spock.sub_show_status('test_subscription')) <> 'disabled' LOOP
     PERFORM pg_sleep(0.1);
   END LOOP;
 END;