Refactor the db processors

Author: boqiu

blockchain/sync/process/db/processor.go

@@ -0,0 +1,53 @@
+package db
+
+import (
+	"context"
+	"time"
+
+	"github.com/Conflux-Chain/go-conflux-util/ctxutil"
+	"github.com/Conflux-Chain/go-conflux-util/health"
+	"github.com/mcuadros/go-defaults"
+	"gorm.io/gorm"
+)
+
+type Option struct {
+	RetryInterval time.Duration `default:"3s"`
+
+	Health health.TimedCounterConfig
+}
+
+// RetriableProcessor operates on database till succeeded.
+type RetriableProcessor struct {
+	option Option
+	db     *gorm.DB
+	health *health.TimedCounter
+}
+
+func NewRetriableProcessor(db *gorm.DB, option Option) *RetriableProcessor {
+	defaults.SetDefaults(&option)
+
+	return &RetriableProcessor{
+		option: option,
+		db:     db,
+		health: health.NewTimedCounter(option.Health),
+	}
+}
+
+// Write executes the given op in a transaction. If failed, it will try again till succeeded.
+func (processor *RetriableProcessor) Write(ctx context.Context, op Operation) {
+	for {
+		err := processor.db.Transaction(func(tx *gorm.DB) error {
+			return op.Exec(tx)
+		})
+
+		processor.health.LogOnError(err, "Process blockchain data in Database")
+
+		if err == nil {
+			return
+		}
+
+		if err = ctxutil.Sleep(ctx, processor.option.RetryInterval); err != nil {
+			return
+		}
+	}
+}

blockchain/sync/process/db/processor_agg.go

@@ -2,11 +2,7 @@ package db
 
 import (
 	"context"
-	"time"
 
-	"github.com/Conflux-Chain/go-conflux-util/ctxutil"
-	"github.com/Conflux-Chain/go-conflux-util/health"
-	"github.com/mcuadros/go-defaults"
 	"gorm.io/gorm"
 )
 
@@ -15,28 +11,17 @@ type Processor[T any] interface {
 	Process(data T) Operation
 }
 
-type Option struct {
-	RetryInterval time.Duration `default:"3s"`
-
-	Health health.TimedCounterConfig
-}
-
 // AggregateProcessor aggregates multiple processor to process blockchain data in batch.
 type AggregateProcessor[T any] struct {
-	option     Option
-	db         *gorm.DB
+	*RetriableProcessor
+
 	processors []Processor[T]
-	health     *health.TimedCounter
 }
 
 func NewAggregateProcessor[T any](option Option, db *gorm.DB, processors ...Processor[T]) *AggregateProcessor[T] {
-	defaults.SetDefaults(&option)
-
 	return &AggregateProcessor[T]{
-		option:     option,
-		db:         db,
-		processors: processors,
-		health:     health.NewTimedCounter(option.Health),
+		RetriableProcessor: NewRetriableProcessor(db, option),
+		processors:         processors,
 	}
 }
 
@@ -49,24 +34,5 @@ func (processor *AggregateProcessor[T]) Process(ctx context.Context, data T) {
 		ops = append(ops, op)
 	}
 
-	processor.blockingWrite(ctx, ComposeOperation(ops...))
-}
-
-// blockingWrite executes the given op in a transaction. If failed, it will try again till succeeded.
-func (processor *AggregateProcessor[T]) blockingWrite(ctx context.Context, op Operation) {
-	for {
-		err := processor.db.Transaction(func(tx *gorm.DB) error {
-			return op.Exec(tx)
-		})
-
-		processor.health.LogOnError(err, "Process blockchain data in Database")
-
-		if err == nil {
-			return
-		}
-
-		if err = ctxutil.Sleep(ctx, processor.option.RetryInterval); err != nil {
-			return
-		}
-	}
+	processor.Write(ctx, ComposeOperation(ops...))
 }

blockchain/sync/process/db/processor_batch.go

@@ -13,8 +13,6 @@ import (
 // Note, thread-safe is not required in the implementations, since batch
 // related methods are executed in a single thread.
 type BatchProcessor[T any] interface {
-	Processor[T]
-
 	// BatchProcess processes the given data and returns the number of SQLs to be executed in batch.
 	BatchProcess(data T) int
 
@@ -37,7 +35,7 @@ type BatchOption struct {
 //
 // Generally, it is used during catch up phase.
 type BatchAggregateProcessor[T any] struct {
-	*AggregateProcessor[T]
+	*RetriableProcessor
 
 	option        BatchOption
 	processors    []BatchProcessor[T]
@@ -48,13 +46,8 @@ type BatchAggregateProcessor[T any] struct {
 func NewBatchAggregateProcessor[T any](option BatchOption, db *gorm.DB, processors ...BatchProcessor[T]) *BatchAggregateProcessor[T] {
 	defaults.SetDefaults(&option)
 
-	innerProcessors := make([]Processor[T], 0, len(processors))
-	for _, v := range processors {
-		innerProcessors = append(innerProcessors, v)
-	}
-
 	return &BatchAggregateProcessor[T]{
-		AggregateProcessor: NewAggregateProcessor(option.Option, db, innerProcessors...),
+		RetriableProcessor: NewRetriableProcessor(db, option.Option),
 		option:             option,
 		processors:         processors,
 		lastBatchTime:      time.Now(),
@@ -82,7 +75,7 @@ func (processor *BatchAggregateProcessor[T]) Process(ctx context.Context, data T
 }
 
 func (processor *BatchAggregateProcessor[T]) write(ctx context.Context) {
-	processor.blockingWrite(ctx, processor)
+	processor.Write(ctx, processor)
 
 	// reset
 	processor.lastBatchTime = time.Now()

blockchain/sync/process/db/processor_revertable.go

@@ -53,7 +53,7 @@ func (processor *RevertableAggregateProcessor[T]) Process(ctx context.Context, d
 			ops = append(ops, op)
 		}
 
-		processor.blockingWrite(ctx, ComposeOperation(ops...))
+		processor.Write(ctx, ComposeOperation(ops...))
 	}
 
 	processor.AggregateProcessor.Process(ctx, data.Data)