PCSM-273: Implement document level parallel replication

config/config.go

@@ -30,6 +30,7 @@ type Config struct {
 
 	UseCollectionBulkWrite bool `mapstructure:"use-collection-bulk-write"`
 
+	Repl  ReplConfig  `mapstructure:",squash"`
 	Clone CloneConfig `mapstructure:",squash"`
 
 	// hidden startup flags
@@ -51,6 +52,26 @@ type MongoDBConfig struct {
 	TargetCompressors []string      `mapstructure:"dev-target-client-compressors"`
 }
 
+// ReplConfig holds replication operation configuration.
+type ReplConfig struct {
+	// NumWorkers is the number of replication workers.
+	// 0 means auto (defaults to runtime.NumCPU()).
+	NumWorkers int `mapstructure:"repl-num-workers"`
+	// ChangeStreamBatchSize is the batch size for MongoDB change streams.
+	// 0 means auto (defaults to config.ChangeStreamBatchSize).
+	ChangeStreamBatchSize int `mapstructure:"repl-change-stream-batch-size"`
+	// EventQueueSize is the buffer size of the channel between the change stream
+	// reader and the dispatcher.
+	// 0 means auto (defaults to config.ReplQueueSize).
+	EventQueueSize int `mapstructure:"repl-event-queue-size"`
+	// WorkerQueueSize is the per-worker routed event channel buffer size.
+	// 0 means auto (defaults to config.ReplQueueSize).
+	WorkerQueueSize int `mapstructure:"repl-worker-queue-size"`
+	// BulkOpsSize is the maximum number of operations per bulk write.
+	// 0 means auto (defaults to config.BulkOpsSize).
+	BulkOpsSize int `mapstructure:"repl-bulk-ops-size"`
+}
+
 // CloneConfig holds clone operation configuration.
 type CloneConfig struct {
 	// NumParallelCollections is the number of collections to clone in parallel.
@@ -143,6 +164,12 @@ func bindEnvVars() {
 
 	_ = viper.BindEnv("use-collection-bulk-write", "PCSM_USE_COLLECTION_BULK_WRITE")
 
+	_ = viper.BindEnv("repl-num-workers", "PCSM_REPL_NUM_WORKERS")
+	_ = viper.BindEnv("repl-change-stream-batch-size", "PCSM_REPL_CHANGE_STREAM_BATCH_SIZE")
+	_ = viper.BindEnv("repl-event-queue-size", "PCSM_REPL_EVENT_QUEUE_SIZE")
+	_ = viper.BindEnv("repl-worker-queue-size", "PCSM_REPL_WORKER_QUEUE_SIZE")
+	_ = viper.BindEnv("repl-bulk-ops-size", "PCSM_REPL_BULK_OPS_SIZE")
+
 	_ = viper.BindEnv("dev-target-client-compressors", "PCSM_DEV_TARGET_CLIENT_COMPRESSORS")
 
 	_ = viper.BindEnv("clone-num-parallel-collections", "PCSM_CLONE_NUM_PARALLEL_COLLECTIONS")

config/const.go

@@ -52,16 +52,20 @@ const (
 // Change stream and replication settings.
 const (
 	// ChangeStreamBatchSize is the batch size for MongoDB change streams.
-	ChangeStreamBatchSize = 1000
+	// Larger values reduce getMore round-trips to the source.
+	ChangeStreamBatchSize = 10_000
 	// ChangeStreamAwaitTime is the maximum amount of time to wait for new change event.
 	ChangeStreamAwaitTime = time.Second
 	// ReplQueueSize defines the buffer size of the internal channel used to transfer
-	// events between the change stream read and the change replication.
-	ReplQueueSize = ChangeStreamBatchSize
+	// events between the change stream reader and the dispatcher, and also the buffer
+	// size of each worker's routed event channel. Larger values absorb flush latency
+	// spikes and prevent head-of-line blocking across workers.
+	ReplQueueSize = 5_000
 	// BulkOpsSize is the maximum number of operations in a bulk write.
-	BulkOpsSize = ChangeStreamBatchSize
-	// BulkOpsInterval is the maximum interval between bulk write operations.
-	BulkOpsInterval = ChangeStreamAwaitTime
+	// Sized to allow efficient large batches when workers drain accumulated events.
+	BulkOpsSize = 5_000
+	// WorkerFlushInterval is the maximum interval between worker bulk write flushes.
+	WorkerFlushInterval = time.Second
 	// InitialSyncCheckInterval is the interval for checking the initial sync status.
 	InitialSyncCheckInterval = 10 * time.Second
 	// PrintLagTimeInterval is the interval at which the lag time is printed to the logs.

hack/metrics/docker-compose.yml

@@ -14,7 +14,7 @@ services:
     container_name: pcsm-grafana
     volumes:
       - ./provisioning:/etc/grafana/provisioning:ro
-      - ./graphana-board.json:/var/lib/grafana/dashboards/pcsm.json:ro
+      - ./grafana-board.json:/var/lib/grafana/dashboards/pcsm.json:ro
     environment:
       - GF_SECURITY_ADMIN_USER=admin
       - GF_SECURITY_ADMIN_PASSWORD=admin