feat: implement mini-batch & continue-as-new for replicator workflow

Signed-off-by: bestmike007 <i@bestmike007.com>

Author: bestmike007

config/chainstorage/aptos/mainnet/base.yml

@@ -167,10 +167,12 @@ workflows:
     workflow_execution_timeout: 24h
     workflow_identity: workflow.event_backfiller
   replicator:
-    activity_retry_maximum_attempts: 3
+    activity_retry_maximum_attempts: 5
     activity_schedule_to_start_timeout: 5m
     activity_start_to_close_timeout: 10m
     batch_size: 1000
+    mini_batch_size: 100
+    checkpoint_size: 10000
     parallelism: 10
     task_list: default
     workflow_decision_timeout: 2m

config/chainstorage/arbitrum/mainnet/base.yml

@@ -210,6 +210,8 @@ type (
 	ReplicatorWorkflowConfig struct {
 		WorkflowConfig `mapstructure:",squash"`
 		BatchSize      uint64 `mapstructure:"batch_size" validate:"required"`
+		MiniBatchSize  uint64 `mapstructure:"mini_batch_size" validate:"required"`
+		CheckpointSize uint64 `mapstructure:"checkpoint_size" validate:"required,gtfield=BatchSize"`
 		Parallelism    int    `mapstructure:"parallelism" validate:"required"`
 	}
 

config/chainstorage/avacchain/mainnet/base.yml

@@ -27,6 +27,7 @@ const (
 	ActivityEventReconciler = "activity.event_reconciler"
 	ActivityEventLoader     = "activity.event_loader"
 	ActivityReplicator      = "activity.replicator"
+	ActivityUpdateWatermark = "activity.update_watermark"
 
 	loggerMsg = "activity.request"
 

config/chainstorage/base/goerli/base.yml

@@ -18,4 +18,5 @@ var Module = fx.Options(
 	fx.Provide(NewEventReconciler),
 	fx.Provide(NewEventLoader),
 	fx.Provide(NewReplicator),
+	fx.Provide(NewUpdateWatermark),
 )