Pipelines: R2 Data Catalog sinks enforce minimum write interval of 60 seconds (#13951)

Author: sejoker

.changeset/pipelines-r2-data-catalog-min-interval.md

@@ -0,0 +1,9 @@
+---
+"wrangler": patch
+---
+
+Enforce minimum 60 second interval for R2 Data Catalog sinks
+
+R2 Data Catalog sinks now require a minimum `--roll-interval` of 60 seconds to prevent compaction issues in the R2 Data Catalog. This validation is applied when creating sinks via `wrangler pipelines sinks create` with type `r2-data-catalog`, and during the interactive `wrangler pipelines setup` flow.
+
+Regular R2 sinks are not affected and can still use intervals as low as 10 seconds.

packages/wrangler/src/__tests__/pipelines.test.ts

@@ -1720,6 +1720,34 @@ describe("wrangler pipelines", () => {
 				`[Error: --namespace is required for r2-data-catalog sinks]`
 			);
 		});
+
+		it("should error when r2-data-catalog has interval below minimum", async ({
+			expect,
+		}) => {
+			await expect(
+				runWrangler(
+					"pipelines sinks create my_sink --type r2-data-catalog --bucket catalog-bucket --namespace default --table my-table --catalog-token token123 --roll-interval 30"
+				)
+			).rejects.toThrowErrorMatchingInlineSnapshot(
+				`[Error: Pipeline frequency must be at least 60 seconds for R2 Data Catalog sinks to prevent compaction issues. Current value: 30 seconds.]`
+			);
+		});
+
+		it("should allow r2 sinks with interval below 60 seconds", async ({
+			expect,
+		}) => {
+			const createRequest = mockCreateSinkRequest(expect, {
+				name: "my_sink",
+				type: "r2",
+			});
+
+			await runWrangler(
+				"pipelines sinks create my_sink --type r2 --bucket my-bucket --access-key-id mykey --secret-access-key mysecret --roll-interval 30"
+			);
+
+			expect(createRequest.count).toBe(1);
+			expect(std.err).toMatchInlineSnapshot(`""`);
+		});
 	});
 
 	describe("pipelines sinks list", () => {

packages/wrangler/src/pipelines/cli/setup.ts

@@ -191,10 +191,14 @@ async function promptCompression(): Promise<string> {
 	});
 }
 
-async function promptRollingPolicy(): Promise<{
+async function promptRollingPolicy(options?: {
+	minIntervalSeconds?: number;
+}): Promise<{
 	fileSizeBytes: number;
 	intervalSeconds: number;
 }> {
+	const minInterval = options?.minIntervalSeconds ?? 10;
+
 	const fileSizeMB = await promptWithRetry(
 		() => "File size",
 		() =>
@@ -214,13 +218,13 @@ async function promptRollingPolicy(): Promise<{
 	const intervalSeconds = await promptWithRetry(
 		() => "Interval",
 		() =>
-			prompt("Roll file when time reaches (seconds, minimum 10):", {
+			prompt(`Roll file when time reaches (seconds, minimum ${minInterval}):`, {
 				defaultValue: String(SINK_DEFAULTS.rolling_policy.interval_seconds),
 			}),
 		(value) => {
 			const num = parseInt(value, 10);
-			if (isNaN(num) || num < 10) {
-				throw new UserError("Interval must be a number >= 10", {
+			if (isNaN(num) || num < minInterval) {
+				throw new UserError(`Interval must be a number >= ${minInterval}`, {
 					telemetryMessage: "pipelines setup invalid interval",
 				});
 			}
@@ -960,7 +964,10 @@ async function setupDataCatalogSink(
 	const token = await promptCatalogToken(config, accountId, bucket);
 
 	const compression = await promptCompression();
-	const rollingPolicy = await promptRollingPolicy();
+	// R2 Data Catalog sinks require minimum 60 second intervals to prevent compaction issues
+	const rollingPolicy = await promptRollingPolicy({
+		minIntervalSeconds: SINK_DEFAULTS.rolling_policy.min_interval_seconds,
+	});
 
 	setupConfig.sinkConfig = {
 		name: setupConfig.sinkName,

packages/wrangler/src/pipelines/cli/sinks/create.ts

@@ -144,6 +144,19 @@ export const pipelinesSinksCreateCommand = createCommand({
 					{ telemetryMessage: "pipelines sinks create invalid format" }
 				);
 			}
+			// Enforce minimum interval for R2 Data Catalog to prevent compaction issues
+			if (
+				args.rollInterval !== undefined &&
+				args.rollInterval < SINK_DEFAULTS.rolling_policy.min_interval_seconds
+			) {
+				throw new CommandLineArgsError(
+					`Pipeline frequency must be at least ${SINK_DEFAULTS.rolling_policy.min_interval_seconds} seconds for R2 Data Catalog sinks to prevent compaction issues. Current value: ${args.rollInterval} seconds.`,
+					{
+						telemetryMessage:
+							"pipelines r2 data catalog interval below minimum threshold",
+					}
+				);
+			}
 		}
 	},
 	async handler(args, { config }) {

packages/wrangler/src/pipelines/defaults.ts

@@ -9,6 +9,7 @@ export const SINK_DEFAULTS = {
 	rolling_policy: {
 		file_size_bytes: undefined,
 		interval_seconds: 300,
+		min_interval_seconds: 60,
 	},
 	r2: {
 		path: "",