Bound the libpq pipeline backlog with byte-based syncs

src/bin/pgcopydb/ld_apply.c

@@ -32,14 +32,14 @@
 #include "summary.h"
 
 /*
- * libpq's output buffer uses a signed int for its size, so it can only grow
- * to ~1 GB before the doubling arithmetic overflows.  When the CDC apply
- * pipelines many EXECUTE statements with large parameter values (e.g. rows
- * containing 300 MB email bodies), the accumulated data can exceed that
- * limit.  We force a pipeline sync at transaction boundaries once the
- * estimated parameter data reaches this threshold.
+ * libpq memmoves its whole outstanding buffer contents on every partial send
+ * and read, making apply quadratic in the pipeline backlog size.  Sync the
+ * pipeline every time this many bytes were sent, mid-transaction included.
  */
-#define PIPELINE_BYTES_SYNC_THRESHOLD (512ULL * 1024 * 1024)
+#define PIPELINE_BYTES_SYNC_THRESHOLD (1024 * 1024)
+
+/* per-statement allowance for protocol framing and result traffic */
+#define PIPELINE_STMT_OVERHEAD 64
 
 GUC applySettingsSync[] = {
 	COMMON_GUC_SETTINGS,
@@ -583,25 +583,17 @@ stream_apply_file(StreamApplyContext *context)
 
 		/*
 		 * Sync the pipeline at transaction boundaries (COMMIT or
-		 * KEEPALIVE) when either the 1-second timer has elapsed or
-		 * when accumulated parameter data approaches libpq's output
-		 * buffer limit.
-		 *
-		 * libpq's output buffer size is tracked with a signed int,
-		 * so the buffer can grow to ~1 GB before the doubling logic
-		 * overflows.  We sync well before that at 512 MB to leave
-		 * headroom for wire-protocol framing and PREPARE overhead.
+		 * KEEPALIVE) when the 1-second timer has elapsed, bounding
+		 * replay latency.  The byte-based sync that bounds the
+		 * pipeline backlog size lives in stream_apply_sql.
 		 */
 		if (metadata->action == STREAM_ACTION_COMMIT ||
 			metadata->action == STREAM_ACTION_KEEPALIVE)
 		{
 			bool timeToSync =
 				1 < (time(NULL) - context->applyPgConn.pipelineSyncTime);
 
-			bool bufferNearFull =
-				context->pipelineBytes >= PIPELINE_BYTES_SYNC_THRESHOLD;
-
-			if (timeToSync || bufferNearFull)
+			if (timeToSync)
 			{
 				/* fetch results until done */
 				if (!pgsql_sync_pipeline(&(context->applyPgConn)))
@@ -655,6 +647,9 @@ stream_apply_sql(StreamApplyContext *context,
 {
 	PGSQL *applyPgConn = &(context->applyPgConn);
 
+	/* bytes this line hands to libpq; branches that send nothing zero it */
+	uint64_t sentBytes = strlen(sql);
+
 	switch (metadata->action)
 	{
 		case STREAM_ACTION_SWITCH:
@@ -668,6 +663,7 @@ stream_apply_sql(StreamApplyContext *context,
 			 * .sql file to apply.
 			 */
 			context->switchLSN = metadata->lsn;
+			sentBytes = 0;
 
 			break;
 		}
@@ -1204,6 +1200,8 @@ stream_apply_sql(StreamApplyContext *context,
 				return true;
 			}
 
+			sentBytes = 0;
+
 			/* Only prepare if we haven't already */
 			if (stmt != NULL && !stmt->prepared)
 			{
@@ -1218,6 +1216,7 @@ stream_apply_sql(StreamApplyContext *context,
 				}
 
 				stmt->prepared = true;
+				sentBytes = strlen(metadata->stmt);
 			}
 
 			break;
@@ -1294,40 +1293,6 @@ stream_apply_sql(StreamApplyContext *context,
 					/* errors have already been logged */
 					return false;
 				}
-
-				/*
-				 * Track accumulated parameter bytes in the pipeline.
-				 * libpq output buffer uses int (~1 GB effective max
-				 * due to doubling), force sync before overflow.
-				 */
-				for (int j = 0; j < count; j++)
-				{
-					if (paramValues[j] != NULL)
-					{
-						context->pipelineBytes += strlen(paramValues[j]);
-					}
-				}
-
-				/*
-				 * When a single transaction contains many large rows
-				 * (e.g. 330 MB email bodies), the pipeline buffer can
-				 * exceed libpq's ~1 GB limit before reaching COMMIT.
-				 * Sync mid-transaction to drain the buffer.  This is
-				 * safe: we use explicit BEGIN/COMMIT, so a pipeline
-				 * sync only flushes pending results without affecting
-				 * the transaction.
-				 */
-				if (context->pipelineBytes >= PIPELINE_BYTES_SYNC_THRESHOLD)
-				{
-					if (!pgsql_sync_pipeline(applyPgConn))
-					{
-						log_error("Failed to sync the pipeline, "
-								  "see previous error for details");
-						return false;
-					}
-
-					context->pipelineBytes = 0;
-				}
 			}
 
 
@@ -1379,6 +1344,29 @@ stream_apply_sql(StreamApplyContext *context,
 		}
 	}
 
+	/*
+	 * Sync the pipeline as soon as the data sent since the last sync
+	 * exceeds the threshold, even in the middle of a transaction: a
+	 * pipeline sync only fetches pending results, the explicit
+	 * BEGIN/COMMIT transaction is unaffected.
+	 */
+	if (sentBytes > 0)
+	{
+		context->pipelineBytes += sentBytes + PIPELINE_STMT_OVERHEAD;
+	}
+
+	if (context->pipelineBytes >= PIPELINE_BYTES_SYNC_THRESHOLD)
+	{
+		if (!pgsql_sync_pipeline(applyPgConn))
+		{
+			log_error("Failed to sync the pipeline, "
+					  "see previous error for details");
+			return false;
+		}
+
+		context->pipelineBytes = 0;
+	}
+
 	return true;
 }
 

src/bin/pgcopydb/ld_replay.c

@@ -418,6 +418,8 @@ stream_replay_line(void *ctx, const char *line, bool *stop)
 							  "error for details");
 					return false;
 				}
+
+				context->pipelineBytes = 0;
 			}
 			break;
 		}
@@ -482,6 +484,8 @@ stream_replay_line(void *ctx, const char *line, bool *stop)
 					  "details");
 			return false;
 		}
+
+		context->pipelineBytes = 0;
 	}
 
 	return true;

tests/cdc-low-level/copydb.sh

@@ -129,5 +129,34 @@ pgcopydb stream replay --resume --endpos "${lsn}"
 psql -At -d ${PGCOPYDB_TARGET_PGURI} -c "select pg_replication_origin_advance('pgcopydb', '0/0');"
 pgcopydb stream apply /usr/src/pgcopydb/pipeline-deadlock.sql
 
+# bounded pipeline sync test: a single multi-MB transaction must trigger
+# byte-based pipeline syncs before its COMMIT, bounding libpq's buffers
+BOUNDED=/tmp/bounded-sync.sql
+
+awk 'BEGIN {
+  printf "BEGIN; -- {\"xid\":99000001,\"lsn\":\"E0/10000000\",\"timestamp\":\"2026-07-25 10:00:00.000000+0000\",\"commit_lsn\":\"E0/20000000\"}\n";
+  for (i = 1; i <= 25000; i++) {
+    printf "PREPARE ab12cd34 AS INSERT INTO \"public\".\"bounded_sync\" (\"id\", \"payload\") overriding system value VALUES ($1, $2);\n";
+    printf "EXECUTE ab12cd34[\"%d\",\"payload-%d-0123456789abcdefghijklmnopqrstuvwxyz0123456789abcdefghijklmnopqrstuvwxyz\"];\n", i, i;
+  }
+  printf "COMMIT; -- {\"xid\":99000001,\"lsn\":\"E0/20000000\",\"timestamp\":\"2026-07-25 10:00:00.000000+0000\"}\n";
+  printf "-- SWITCH WAL {\"lsn\":\"E0/21000000\"}\n";
+}' > ${BOUNDED}
+
+psql -At -d ${PGCOPYDB_TARGET_PGURI} -c "select pg_replication_origin_advance('pgcopydb', '0/0');"
+
+pgcopydb stream apply --trace ${BOUNDED} > /tmp/bounded-sync.log 2>&1 || \
+    { tail -50 /tmp/bounded-sync.log; exit 1; }
+
+syncs=$(grep -c "Start pipeline sync" /tmp/bounded-sync.log || true)
+rows=$(psql -At -d ${PGCOPYDB_TARGET_PGURI} -c "select count(*) from bounded_sync")
+
+echo "bounded sync test: ${syncs} pipeline syncs, ${rows} rows applied"
+
+test "${rows}" -eq 25000
+
+# unpatched pgcopydb only syncs at COMMIT/EOF, which fails this assertion
+test "${syncs}" -ge 5
+
 # cleanup
 pgcopydb stream cleanup --verbose

tests/cdc-low-level/ddl.sql

@@ -37,4 +37,7 @@ EXECUTE FUNCTION notify_on_insert();
 -- Lets enable the trigger for the replica mode
 ALTER TABLE metrics ENABLE REPLICA TRIGGER insert_notice_trigger;
 
+-- bounded pipeline sync test table; keep trigger-free (no NOTICE traffic)
+create table bounded_sync(id int, payload text);
+
 commit;