fix(clickhouse): add opt-out for staging cleanup after QRep flow (PeerDB-io#4466)

## Summary

Restores the ability to retain ClickHouse staging avro files after a
snapshot/QRep flow completes, addressing a silent behavior change
introduced in PeerDB-io#4200.

## Background — the regression

PeerDB-io#4200 (`StagingStore` abstraction) refactored the ClickHouse connector
so avro **upload** and staging **cleanup** both key off the staging
store's real bucket prefix (`staging.KeyPrefix()`). The PR was described
as "zero behavior change" for the S3 path, but the cleanup path actually
changed behavior:

- **Before:** `CleanupQRepFlow` keyed off `config.StagingPath` (=
`SnapshotStagingPath`), guarded by `strings.HasPrefix(stagingPath,
"s3://")`. For ClickHouse mirrors, `SnapshotStagingPath` is typically
empty (the bucket comes from `PEERDB_CLICKHOUSE_AWS_S3_BUCKET_NAME`), so
the guard was false and **cleanup was a no-op** — staging avro files
were retained in the bucket.
- **After:** cleanup keys off the same real prefix that uploads use, so
it now always deletes `<staging-prefix>/<flowJobName>` once the flow
completes.

Deployments that upgraded from 0.36.18 → 0.36.24 and had downstream
pipelines consuming those staging avro artifacts saw the files disappear
as soon as the mirror's snapshot completed.

## Change

Cleanup remains the default — it is the correct behavior and avoids
unbounded staging growth. This PR adds an opt-out:

- New dynconf setting `PEERDB_CLICKHOUSE_SKIP_STAGING_CLEANUP` (bool,
default `false`).
- When enabled, `CleanupQRepFlow` logs and returns early without
deleting staging objects.

## Notes

- The staging bucket is PeerDB-internal scratch space; relying on
retained files is fragile. The recommended long-term path for consuming
avro artifacts is a dedicated PG→S3 QRep mirror. This flag is an escape
hatch for affected deployments.
- Separately, there is a latent mismatch when `PEERDB_S3_UUID_PREFIX` is
enabled: upload keys become `<prefix>/<uuid>/<flowJobName>/...` while
cleanup deletes `<prefix>/<flowJobName>`, so cleanup misses those
objects. Out of scope here but worth tracking.

## Test plan

- [x] `go build ./connectors/clickhouse/... ./internal/...`
- [x] `go vet ./connectors/clickhouse/`
- [ ] Manual: run a CH snapshot with
`PEERDB_CLICKHOUSE_SKIP_STAGING_CLEANUP=true` and confirm staging avro
files remain after completion; with default, confirm they are deleted.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: dtunikov

flow/connectors/clickhouse/qrep.go

@@ -6,6 +6,7 @@ import (
 	"log/slog"
 
 	"github.com/PeerDB-io/peerdb/flow/generated/protos"
+	"github.com/PeerDB-io/peerdb/flow/internal"
 	"github.com/PeerDB-io/peerdb/flow/model"
 	"github.com/PeerDB-io/peerdb/flow/shared"
 )
@@ -44,6 +45,17 @@ func (c *ClickHouseConnector) ConsolidateQRepPartitions(_ context.Context, confi
 // CleanupQRepFlow function for clickhouse connector
 func (c *ClickHouseConnector) CleanupQRepFlow(ctx context.Context, config *protos.QRepConfig) error {
 	flowName := config.FlowJobName
+
+	skipCleanup, err := internal.PeerDBClickHouseSkipStagingCleanup(ctx, config.Env)
+	if err != nil {
+		return fmt.Errorf("failed to read staging cleanup config: %w", err)
+	}
+	if skipCleanup {
+		c.logger.Info("Skipping staging cleanup after QRepFlow (PEERDB_CLICKHOUSE_SKIP_STAGING_CLEANUP enabled)",
+			slog.String("stagingPath", c.staging.BucketPath()), slog.String("flowName", flowName))
+		return nil
+	}
+
 	c.logger.Info("Cleaning up stage after QRepFlow",
 		slog.String("stagingPath", c.staging.BucketPath()), slog.String("flowName", flowName))
 

flow/internal/dynamicconf.go

@@ -198,6 +198,15 @@ var DynamicSettings = [...]*protos.DynamicSetting{
 		ApplyMode:        protos.DynconfApplyMode_APPLY_MODE_IMMEDIATE,
 		TargetForSetting: protos.DynconfTarget_CLICKHOUSE,
 	},
+	{
+		Name: "PEERDB_CLICKHOUSE_SKIP_STAGING_CLEANUP",
+		Description: "Skip deleting ClickHouse staging avro files after a QRep/snapshot flow completes. " +
+			"Enable to retain staging artifacts for downstream pipelines that consume them",
+		DefaultValue:     "false",
+		ValueType:        protos.DynconfValueType_BOOL,
+		ApplyMode:        protos.DynconfApplyMode_APPLY_MODE_IMMEDIATE,
+		TargetForSetting: protos.DynconfTarget_CLICKHOUSE,
+	},
 	{
 		Name:             "PEERDB_S3_UUID_PREFIX",
 		Description:      "Use random UUID as prefix instead of flow name, can help partitioning on non-AWS based s3 providers",
@@ -758,6 +767,10 @@ func PeerDBClickHouseStagingBucketName(ctx context.Context, env map[string]strin
 	return dynLookup(ctx, env, "PEERDB_CLICKHOUSE_STAGING_BUCKET_NAME")
 }
 
+func PeerDBClickHouseSkipStagingCleanup(ctx context.Context, env map[string]string) (bool, error) {
+	return dynamicConfBool(ctx, env, "PEERDB_CLICKHOUSE_SKIP_STAGING_CLEANUP")
+}
+
 func PeerDBClickHouseAWSS3BucketName(ctx context.Context, env map[string]string) (string, error) {
 	return dynLookup(ctx, env, "PEERDB_CLICKHOUSE_AWS_S3_BUCKET_NAME")
 }