Add CDC file cleanup watchdog

.github/workflows/run-tests-tiered.yml

@@ -145,6 +145,7 @@ jobs:
           - skip-large-objects
           - cdc-low-level
           - cdc-test-decoding
+          - cdc-prune
           - cdc-endpos-between-transaction
           - cdc-filtering
           - cdc-wal2json

docs/include/clone.rst

@@ -49,4 +49,6 @@
      --defer-analyze               Defer ANALYZE until after post-data restore
      --defer-validate-fks          Create FK constraints as NOT VALID, skipping validation scan
      --use-copy-binary             Use the COPY BINARY format for COPY operations
+     --prune-threshold           Max size of applied CDC files to retain (e.g. 10GB, 0 to disable)
+     --prune-min-age             Min age before applied CDC files can be deleted (e.g. 15m, 2h)
 

docs/include/follow.rst

@@ -17,4 +17,6 @@
      --create-slot                 Create the replication slot
      --origin                      Use this Postgres replication origin node name
      --endpos                      Stop replaying changes when reaching this LSN
+     --prune-threshold           Max size of applied CDC files to retain (e.g. 10GB, 0 to disable)
+     --prune-min-age             Min age before applied CDC files can be deleted (e.g. 15m, 2h)
 

src/bin/pgcopydb/cli_clone_follow.c

@@ -72,6 +72,8 @@
 	"  --defer-analyze               Defer ANALYZE until after post-data restore\n" \
 	"  --defer-validate-fks          Create FK constraints as NOT VALID, skipping validation scan\n" \
 	"  --use-copy-binary             Use the COPY BINARY format for COPY operations\n" \
+	"  --prune-threshold           Max size of applied CDC files to retain (e.g. 10GB, 0 to disable)\n" \
+	"  --prune-min-age             Min age before applied CDC files can be deleted (e.g. 15m, 2h)\n" \
 
 CommandLine clone_command =
 	make_command(
@@ -110,7 +112,9 @@ CommandLine follow_command =
 		"  --slot-name                   Use this Postgres replication slot name\n"
 		"  --create-slot                 Create the replication slot\n"
 		"  --origin                      Use this Postgres replication origin node name\n"
-		"  --endpos                      Stop replaying changes when reaching this LSN\n",
+		"  --endpos                      Stop replaying changes when reaching this LSN\n"
+		"  --prune-threshold           Max size of applied CDC files to retain (e.g. 10GB, 0 to disable)\n"
+		"  --prune-min-age             Min age before applied CDC files can be deleted (e.g. 15m, 2h)\n",
 		cli_copy_db_getopts,
 		cli_follow);
 
@@ -225,7 +229,9 @@ clone_and_follow(CopyDataSpec *copySpecs)
 						   &(copySpecs->filters),
 						   copyDBoptions.stdIn,
 						   copyDBoptions.stdOut,
-						   logSQL))
+						   logSQL,
+						   copyDBoptions.pruneThresholdBytes,
+						   copyDBoptions.pruneMinAgeSeconds))
 	{
 		/* errors have already been logged */
 		exit(EXIT_CODE_INTERNAL_ERROR);
@@ -561,7 +567,9 @@ cli_follow(int argc, char **argv)
 						   &(copySpecs.filters),
 						   copyDBoptions.stdIn,
 						   copyDBoptions.stdOut,
-						   logSQL))
+						   logSQL,
+						   copyDBoptions.pruneThresholdBytes,
+						   copyDBoptions.pruneMinAgeSeconds))
 	{
 		/* errors have already been logged */
 		exit(EXIT_CODE_INTERNAL_ERROR);

src/bin/pgcopydb/cli_common.c

@@ -657,6 +657,8 @@ cli_copy_db_getopts(int argc, char **argv)
 		{ "defer-indexes", no_argument, NULL, 257 },
 		{ "defer-analyze", no_argument, NULL, 258 },
 		{ "defer-validate-fks", no_argument, NULL, 259 },
+		{ "prune-threshold", required_argument, NULL, 260 },
+		{ "prune-min-age", required_argument, NULL, 261 },
 		{ "help", no_argument, NULL, 'h' },
 		{ NULL, 0, NULL, 0 }
 	};
@@ -1159,6 +1161,45 @@ cli_copy_db_getopts(int argc, char **argv)
 				break;
 			}
 
+			case 260:
+			{
+				if (!cli_parse_bytes_pretty(
+						optarg,
+						&(options.pruneThresholdBytes),
+						(char *) &(options.pruneThresholdPretty),
+						sizeof(options.pruneThresholdPretty)))
+				{
+					log_fatal("Failed to parse --prune-threshold: \"%s\"",
+							  optarg);
+					++errors;
+				}
+
+				log_trace("--prune-threshold %s (%lld)",
+						  options.pruneThresholdPretty,
+						  (long long) options.pruneThresholdBytes);
+				break;
+			}
+
+			case 261:
+			{
+				if (!cli_parse_duration(
+						optarg,
+						&(options.pruneMinAgeSeconds)))
+				{
+					log_fatal("Failed to parse --prune-min-age: \"%s\"",
+							  optarg);
+					++errors;
+				}
+
+				strlcpy(options.pruneMinAgePretty, optarg,
+						sizeof(options.pruneMinAgePretty));
+
+				log_trace("--prune-min-age %s (%d seconds)",
+						  options.pruneMinAgePretty,
+						  options.pruneMinAgeSeconds);
+				break;
+			}
+
 			case '?':
 			default:
 			{
@@ -1210,6 +1251,23 @@ cli_copy_db_getopts(int argc, char **argv)
 		exit(EXIT_CODE_BAD_ARGS);
 	}
 
+	if (options.pruneThresholdBytes == 0 && options.pruneMinAgeSeconds > 0)
+	{
+		log_warn("--prune-min-age has no effect without --prune-threshold");
+	}
+
+	/*
+	 * When prune threshold is set but min-age wasn't explicitly provided,
+	 * default to 15 minutes (900 seconds) for safety.
+	 */
+	if (options.pruneThresholdBytes > 0 && options.pruneMinAgeSeconds == 0 &&
+		options.pruneMinAgePretty[0] == '\0')
+	{
+		options.pruneMinAgeSeconds = 900;
+		strlcpy(options.pruneMinAgePretty, "15m",
+				sizeof(options.pruneMinAgePretty));
+	}
+
 	if (errors > 0)
 	{
 		commandline_help(stderr);

src/bin/pgcopydb/cli_common.h

@@ -100,6 +100,12 @@ typedef struct CopyDBOptions
 
 	char filterFileName[MAXPGPATH];
 	char requirementsFileName[MAXPGPATH];
+
+	/* CDC file prune configuration */
+	uint64_t pruneThresholdBytes;
+	char pruneThresholdPretty[BUFSIZE];
+	int pruneMinAgeSeconds;
+	char pruneMinAgePretty[BUFSIZE];
 } CopyDBOptions;
 
 extern bool outputJSON;

src/bin/pgcopydb/cli_snapshot.c

@@ -326,7 +326,9 @@ cli_create_snapshot(int argc, char **argv)
 							   &(copySpecs.filters),
 							   createSNoptions.stdIn,
 							   createSNoptions.stdOut,
-							   logSQL))
+							   logSQL,
+							   0,
+							   0))
 		{
 			/* errors have already been logged */
 			exit(EXIT_CODE_INTERNAL_ERROR);

src/bin/pgcopydb/cli_stream.c

@@ -222,6 +222,8 @@ cli_stream_getopts(int argc, char **argv)
 		{ "debug", no_argument, NULL, 'd' },
 		{ "trace", no_argument, NULL, 'z' },
 		{ "quiet", no_argument, NULL, 'q' },
+		{ "prune-threshold", required_argument, NULL, 256 },
+		{ "prune-min-age", required_argument, NULL, 257 },
 		{ "help", no_argument, NULL, 'h' },
 		{ NULL, 0, NULL, 0 }
 	};
@@ -434,6 +436,45 @@ cli_stream_getopts(int argc, char **argv)
 				break;
 			}
 
+			case 256:
+			{
+				if (!cli_parse_bytes_pretty(
+						optarg,
+						&(options.pruneThresholdBytes),
+						(char *) &(options.pruneThresholdPretty),
+						sizeof(options.pruneThresholdPretty)))
+				{
+					log_fatal("Failed to parse --prune-threshold: \"%s\"",
+							  optarg);
+					++errors;
+				}
+
+				log_trace("--prune-threshold %s (%lld)",
+						  options.pruneThresholdPretty,
+						  (long long) options.pruneThresholdBytes);
+				break;
+			}
+
+			case 257:
+			{
+				if (!cli_parse_duration(
+						optarg,
+						&(options.pruneMinAgeSeconds)))
+				{
+					log_fatal("Failed to parse --prune-min-age: \"%s\"",
+							  optarg);
+					++errors;
+				}
+
+				strlcpy(options.pruneMinAgePretty, optarg,
+						sizeof(options.pruneMinAgePretty));
+
+				log_trace("--prune-min-age %s (%d seconds)",
+						  options.pruneMinAgePretty,
+						  options.pruneMinAgeSeconds);
+				break;
+			}
+
 			case '?':
 			default:
 			{
@@ -472,6 +513,23 @@ cli_stream_getopts(int argc, char **argv)
 		exit(EXIT_CODE_BAD_ARGS);
 	}
 
+	if (options.pruneThresholdBytes == 0 && options.pruneMinAgeSeconds > 0)
+	{
+		log_warn("--prune-min-age has no effect without --prune-threshold");
+	}
+
+	/*
+	 * When prune threshold is set but min-age wasn't explicitly provided,
+	 * default to 15 minutes (900 seconds) for safety.
+	 */
+	if (options.pruneThresholdBytes > 0 && options.pruneMinAgeSeconds == 0 &&
+		options.pruneMinAgePretty[0] == '\0')
+	{
+		options.pruneMinAgeSeconds = 900;
+		strlcpy(options.pruneMinAgePretty, "15m",
+				sizeof(options.pruneMinAgePretty));
+	}
+
 	if (errors > 0)
 	{
 		commandline_help(stderr);
@@ -585,7 +643,9 @@ cli_stream_setup(int argc, char **argv)
 						   &(copySpecs.filters),
 						   streamDBoptions.stdIn,
 						   streamDBoptions.stdOut,
-						   logSQL))
+						   logSQL,
+						   streamDBoptions.pruneThresholdBytes,
+						   streamDBoptions.pruneMinAgeSeconds))
 	{
 		/* errors have already been logged */
 		exit(EXIT_CODE_INTERNAL_ERROR);
@@ -729,7 +789,9 @@ cli_stream_catchup(int argc, char **argv)
 						   &(copySpecs.filters),
 						   streamDBoptions.stdIn,
 						   streamDBoptions.stdOut,
-						   logSQL))
+						   logSQL,
+						   streamDBoptions.pruneThresholdBytes,
+						   streamDBoptions.pruneMinAgeSeconds))
 	{
 		/* errors have already been logged */
 		exit(EXIT_CODE_INTERNAL_ERROR);
@@ -813,7 +875,9 @@ cli_stream_replay(int argc, char **argv)
 						   &(copySpecs.filters),
 						   true,  /* stdin */
 						   true, /* stdout */
-						   logSQL))
+						   logSQL,
+						   streamDBoptions.pruneThresholdBytes,
+						   streamDBoptions.pruneMinAgeSeconds))
 	{
 		/* errors have already been logged */
 		exit(EXIT_CODE_INTERNAL_ERROR);
@@ -939,7 +1003,9 @@ cli_stream_transform(int argc, char **argv)
 						   &(copySpecs.filters),
 						   streamDBoptions.stdIn,
 						   streamDBoptions.stdOut,
-						   logSQL))
+						   logSQL,
+						   streamDBoptions.pruneThresholdBytes,
+						   streamDBoptions.pruneMinAgeSeconds))
 	{
 		/* errors have already been logged */
 		exit(EXIT_CODE_INTERNAL_ERROR);
@@ -1102,7 +1168,9 @@ cli_stream_apply(int argc, char **argv)
 							   &(copySpecs.filters),
 							   true, /* streamDBoptions.stdIn */
 							   false, /* streamDBoptions.stdOut */
-							   logSQL))
+							   logSQL,
+							   streamDBoptions.pruneThresholdBytes,
+							   streamDBoptions.pruneMinAgeSeconds))
 		{
 			/* errors have already been logged */
 			exit(EXIT_CODE_INTERNAL_ERROR);
@@ -1215,7 +1283,9 @@ stream_start_in_mode(LogicalStreamMode mode)
 						   &(copySpecs.filters),
 						   streamDBoptions.stdIn,
 						   streamDBoptions.stdOut,
-						   logSQL))
+						   logSQL,
+						   streamDBoptions.pruneThresholdBytes,
+						   streamDBoptions.pruneMinAgeSeconds))
 	{
 		/* errors have already been logged */
 		exit(EXIT_CODE_INTERNAL_ERROR);

src/bin/pgcopydb/follow.c

@@ -12,6 +12,7 @@
 
 #include "cli_common.h"
 #include "cli_root.h"
+#include "ld_prune.h"
 #include "ld_stream.h"
 #include "log.h"
 #include "progress.h"
@@ -607,6 +608,23 @@ followDB(CopyDataSpec *copySpecs, StreamSpecs *streamSpecs)
 		}
 	}
 
+	/*
+	 * When prune threshold is configured, start the prune watchdog
+	 * to periodically remove old applied CDC files.
+	 */
+	if (streamSpecs->pruneThresholdBytes > 0)
+	{
+		FollowSubProcess *prune = &(streamSpecs->prune);
+
+		if (!follow_start_subprocess(streamSpecs, prune))
+		{
+			log_error("Failed to start the %s process", prune->name);
+
+			(void) follow_exit_early(streamSpecs);
+			return false;
+		}
+	}
+
 	/*
 	 * Close pipe ends which follow is not using. Otherwise the processes
 	 * like transform and apply which reads from the pipe during replay
@@ -830,6 +848,18 @@ follow_start_catchup(StreamSpecs *specs)
 }
 
 
+/*
+ * follow_start_prune starts a sub-process that prunes old CDC files.
+ * The catalog is already opened by follow_start_subprocess before this is
+ * called.
+ */
+bool
+follow_start_prune(StreamSpecs *specs)
+{
+	return cdc_prune_loop(specs);
+}
+
+
 /*
  * follow_start_subprocess forks a subprocess and calls the given function.
  */
@@ -946,7 +976,8 @@ follow_wait_subprocesses(StreamSpecs *specs)
 	FollowSubProcess *processArray[] = {
 		&(specs->prefetch),
 		&(specs->transform),
-		&(specs->catchup)
+		&(specs->catchup),
+		&(specs->prune)
 	};
 
 	int count = sizeof(processArray) / sizeof(processArray[0]);
@@ -1184,7 +1215,8 @@ follow_terminate_subprocesses(StreamSpecs *specs)
 	FollowSubProcess *processArray[] = {
 		&(specs->prefetch),
 		&(specs->transform),
-		&(specs->catchup)
+		&(specs->catchup),
+		&(specs->prune)
 	};
 	int count = sizeof(processArray) / sizeof(processArray[0]);