refactor: split catchup state into dedicated store

Author: vietddude

internal/worker/base.go

@@ -16,6 +16,7 @@ import (
 	"github.com/fystack/multichain-indexer/pkg/infra"
 	"github.com/fystack/multichain-indexer/pkg/retry"
 	"github.com/fystack/multichain-indexer/pkg/store/blockstore"
+	"github.com/fystack/multichain-indexer/pkg/store/catchupstore"
 	"github.com/fystack/multichain-indexer/pkg/store/pubkeystore"
 )
 
@@ -43,6 +44,7 @@ type BaseWorker struct {
 	chain          indexer.Indexer
 	kvstore        infra.KVStore
 	blockStore     blockstore.Store
+	catchupStore   catchupstore.Store
 	pubkeyStore    pubkeystore.Store
 	emitter        events.Emitter
 	failedChan     chan FailedBlockEvent
@@ -63,6 +65,7 @@ func newWorkerWithMode(
 	cfg config.ChainConfig,
 	kv infra.KVStore,
 	blockStore blockstore.Store,
+	catchupStore catchupstore.Store,
 	emitter events.Emitter,
 	pubkeyStore pubkeystore.Store,
 	mode WorkerMode,
@@ -84,6 +87,7 @@ func newWorkerWithMode(
 		chain:          chain,
 		kvstore:        kv,
 		blockStore:     blockStore,
+		catchupStore:   catchupStore,
 		pubkeyStore:    pubkeyStore,
 		emitter:        emitter,
 		failedChan:     failedChan,

internal/worker/catchup.go

@@ -13,6 +13,7 @@ import (
 	"github.com/fystack/multichain-indexer/pkg/events"
 	"github.com/fystack/multichain-indexer/pkg/infra"
 	"github.com/fystack/multichain-indexer/pkg/store/blockstore"
+	"github.com/fystack/multichain-indexer/pkg/store/catchupstore"
 	"github.com/fystack/multichain-indexer/pkg/store/pubkeystore"
 )
 
@@ -36,6 +37,7 @@ func NewCatchupWorker(
 	cfg config.ChainConfig,
 	kv infra.KVStore,
 	blockStore blockstore.Store,
+	catchupStore catchupstore.Store,
 	emitter events.Emitter,
 	pubkeyStore pubkeystore.Store,
 	failedChan chan FailedBlockEvent,
@@ -47,6 +49,7 @@ func NewCatchupWorker(
 		cfg,
 		kv,
 		blockStore,
+		catchupStore,
 		emitter,
 		pubkeyStore,
 		ModeCatchup,
@@ -116,26 +119,22 @@ func (cw *CatchupWorker) runCatchup() {
 }
 
 func (cw *CatchupWorker) loadCatchupProgress() []blockstore.CatchupRange {
-	registry := status.EnsureStatusRegistry(cw.statusRegistry)
-
 	// Load existing catchup ranges from the store. The catchup worker only loads
 	// ranges; creating new ranges is the responsibility of the regular worker
 	// (via determineStartingBlock or skipAheadIfLagging).
-	progress, err := cw.blockStore.GetCatchupProgress(cw.chain.GetNetworkInternalCode())
+	progress, err := cw.catchupStore.GetProgress(cw.ctx, cw.chain.GetNetworkInternalCode())
 	if err != nil {
 		cw.logger.Warn("Failed to load catchup progress",
 			"chain", cw.chain.GetName(),
 			"error", err,
 		)
-		registry.SetCatchupRanges(cw.chain.GetName(), nil)
 		return nil
 	}
 
 	cw.logger.Info("Loaded catchup progress",
 		"chain", cw.chain.GetName(),
 		"progress_ranges", len(progress),
 	)
-	registry.SetCatchupRanges(cw.chain.GetName(), progress)
 	return progress
 }
 
@@ -319,14 +318,13 @@ func (cw *CatchupWorker) processRange(r blockstore.CatchupRange, workerID int) e
 func (cw *CatchupWorker) saveProgress(r blockstore.CatchupRange, current uint64) {
 	cw.progressMu.Lock()
 	defer cw.progressMu.Unlock()
-	registry := status.EnsureStatusRegistry(cw.statusRegistry)
 	cw.logger.Debug("Saving catchup progress",
 		"chain", cw.chain.GetName(),
 		"range", fmt.Sprintf("%d-%d", r.Start, r.End),
 		"current", current,
 	)
 	current = min(current, r.End)
-	if err := cw.blockStore.SaveCatchupProgress(cw.chain.GetNetworkInternalCode(), r.Start, r.End, current); err != nil {
+	if err := cw.catchupStore.SaveProgress(cw.ctx, cw.chain.GetNetworkInternalCode(), r.Start, r.End, current); err != nil {
 		cw.logger.Warn("Failed to save catchup progress",
 			"chain", cw.chain.GetName(),
 			"range", fmt.Sprintf("%d-%d", r.Start, r.End),
@@ -335,11 +333,6 @@ func (cw *CatchupWorker) saveProgress(r blockstore.CatchupRange, current uint64)
 		)
 		return
 	}
-	registry.UpsertCatchupRanges(cw.chain.GetName(), []blockstore.CatchupRange{{
-		Start:   r.Start,
-		End:     r.End,
-		Current: current,
-	}})
 	for i := range cw.blockRanges {
 		if cw.blockRanges[i].Start == r.Start && cw.blockRanges[i].End == r.End {
 			cw.blockRanges[i].Current = current
@@ -351,23 +344,20 @@ func (cw *CatchupWorker) saveProgress(r blockstore.CatchupRange, current uint64)
 func (cw *CatchupWorker) completeRange(r blockstore.CatchupRange) error {
 	cw.progressMu.Lock()
 	defer cw.progressMu.Unlock()
-	registry := status.EnsureStatusRegistry(cw.statusRegistry)
 
 	cw.logger.Info("Completing catchup range",
 		"chain", cw.chain.GetName(),
 		"range", fmt.Sprintf("%d-%d", r.Start, r.End),
 	)
 
-	if err := cw.blockStore.DeleteCatchupRange(cw.chain.GetNetworkInternalCode(), r.Start, r.End); err != nil {
+	if err := cw.catchupStore.DeleteRange(cw.ctx, cw.chain.GetNetworkInternalCode(), r.Start, r.End); err != nil {
 		cw.logger.Warn("Failed to delete catchup range",
 			"chain", cw.chain.GetName(),
 			"range", fmt.Sprintf("%d-%d", r.Start, r.End),
 			"error", err,
 		)
 		return err
 	}
-	registry.DeleteCatchupRange(cw.chain.GetName(), r.Start, r.End)
-
 	// Remove from local ranges
 	for i, existing := range cw.blockRanges {
 		if existing.Start == r.Start && existing.End == r.End {
@@ -406,15 +396,12 @@ func (cw *CatchupWorker) Close() error {
 		)
 	}
 
-	if err := cw.blockStore.SaveCatchupRanges(cw.chain.GetNetworkInternalCode(), rangesToSave); err != nil {
+	if err := cw.catchupStore.SaveRanges(cw.ctx, cw.chain.GetNetworkInternalCode(), rangesToSave); err != nil {
 		cw.logger.Error("Failed to batch save progress on close",
 			"chain", cw.chain.GetName(),
 			"ranges", len(rangesToSave),
 			"error", err,
 		)
-	} else {
-		registry := status.EnsureStatusRegistry(cw.statusRegistry)
-		registry.UpsertCatchupRanges(cw.chain.GetName(), rangesToSave)
 	}
 
 	return nil

internal/worker/catchup_test.go

@@ -25,6 +25,7 @@ func TestCatchupWorkerPollsForNewRangesInsteadOfExiting(t *testing.T) {
 	})
 
 	store := &stubBlockStore{}
+	catchupStore := &stubCatchupStore{}
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
 
@@ -39,6 +40,7 @@ func TestCatchupWorkerPollsForNewRangesInsteadOfExiting(t *testing.T) {
 			},
 			chain:          &stubIndexer{name: "aptos", internalCode: "APTOS_TESTNET", networkType: enum.NetworkTypeApt, latest: 100},
 			blockStore:     store,
+			catchupStore:   catchupStore,
 			statusRegistry: statusRegistry,
 		},
 		blockRanges: []blockstore.CatchupRange{},
@@ -51,7 +53,7 @@ func TestCatchupWorkerPollsForNewRangesInsteadOfExiting(t *testing.T) {
 	go func() {
 		time.Sleep(100 * time.Millisecond)
 		// Simulate regular worker creating catchup ranges in the store
-		store.catchupProgress = []blockstore.CatchupRange{
+		catchupStore.catchupProgress = []blockstore.CatchupRange{
 			{Start: 50, End: 69, Current: 49},
 			{Start: 70, End: 89, Current: 69},
 		}

internal/worker/factory.go

@@ -29,6 +29,7 @@ import (
 	"github.com/fystack/multichain-indexer/pkg/ratelimiter"
 	"github.com/fystack/multichain-indexer/pkg/repository"
 	"github.com/fystack/multichain-indexer/pkg/store/blockstore"
+	"github.com/fystack/multichain-indexer/pkg/store/catchupstore"
 	"github.com/fystack/multichain-indexer/pkg/store/pubkeystore"
 	tonaddr "github.com/xssnick/tonutils-go/address"
 	"gorm.io/gorm"
@@ -39,6 +40,7 @@ type WorkerDeps struct {
 	Ctx            context.Context
 	KVStore        infra.KVStore
 	BlockStore     blockstore.Store
+	CatchupStore   catchupstore.Store
 	Emitter        events.Emitter
 	Pubkey         pubkeystore.Store
 	Redis          infra.RedisClient
@@ -104,6 +106,7 @@ func BuildWorkers(
 				cfg,
 				deps.KVStore,
 				deps.BlockStore,
+				deps.CatchupStore,
 				deps.Emitter,
 				deps.Pubkey,
 				deps.FailedChan,
@@ -118,6 +121,7 @@ func BuildWorkers(
 				cfg,
 				deps.KVStore,
 				deps.BlockStore,
+				deps.CatchupStore,
 				deps.Emitter,
 				deps.Pubkey,
 				deps.FailedChan,
@@ -778,10 +782,11 @@ func CreateManagerWithWorkers(
 ) *Manager {
 	// Shared stores
 	blockStore := blockstore.NewBlockStore(kvstore)
+	catchupStore := catchupstore.New(redisClient)
 	pubkeyStore := pubkeystore.NewPublicKeyStore(addressBF)
 	statusRegistry := status.NewRegistry()
 
-	manager := NewManager(ctx, kvstore, blockStore, emitter, pubkeyStore)
+	manager := NewManager(ctx, kvstore, blockStore, catchupStore, emitter, pubkeyStore)
 	manager.registry = statusRegistry
 
 	// Loop each chain
@@ -818,15 +823,6 @@ func CreateManagerWithWorkers(
 		if existingFailed, err := blockStore.GetFailedBlocks(idxr.GetNetworkInternalCode()); err == nil {
 			statusRegistry.SetFailedBlocks(idxr.GetName(), existingFailed)
 		}
-		if existingCatchup, err := blockStore.GetCatchupProgress(idxr.GetNetworkInternalCode()); err == nil {
-			statusRegistry.SetCatchupRanges(idxr.GetName(), existingCatchup)
-		} else {
-			logger.Warn("Failed to load catchup progress for status registry",
-				"chain", chainName,
-				"internal_code", idxr.GetNetworkInternalCode(),
-				"error", err,
-			)
-		}
 
 		failedChan := make(chan FailedBlockEvent, 100)
 
@@ -835,6 +831,7 @@ func CreateManagerWithWorkers(
 			Ctx:            ctx,
 			KVStore:        kvstore,
 			BlockStore:     blockStore,
+			CatchupStore:   catchupStore,
 			Emitter:        emitter,
 			Pubkey:         pubkeyStore,
 			Redis:          redisClient,

internal/worker/factory_test.go

@@ -3,19 +3,17 @@ package worker
 import (
 	"context"
 	"errors"
-	"fmt"
 	"log/slog"
 	"testing"
 	"time"
 
 	"github.com/fystack/multichain-indexer/pkg/common/config"
-	"github.com/fystack/multichain-indexer/pkg/common/constant"
 	"github.com/fystack/multichain-indexer/pkg/common/enum"
 	commonlogger "github.com/fystack/multichain-indexer/pkg/common/logger"
 	"github.com/fystack/multichain-indexer/pkg/events"
 	"github.com/fystack/multichain-indexer/pkg/infra"
-	"github.com/fystack/multichain-indexer/pkg/store/blockstore"
 	"github.com/hashicorp/consul/api"
+	"github.com/redis/go-redis/v9"
 	"github.com/stretchr/testify/require"
 )
 
@@ -173,21 +171,18 @@ func TestCreateManagerWithWorkersBootstrapsCatchupRangesIntoStatusRegistry(t *te
 		},
 	}
 
-	kv := &listKVStore{
-		pairs: []*infra.KVPair{{
-			Key:   fmt.Sprintf("%s/%s/%s/%d-%d", blockstore.BlockStates, "a", constant.KVPrefixProgressCatchup, 1, 20),
-			Value: []byte("10"),
-		}},
-	}
+	redisClient, cleanup := setupFactoryTestRedis(t)
+	defer cleanup()
+	require.NoError(t, redisClient.HSet(context.Background(), "catchup_progress:a", "1-20", "10").Err())
 
 	manager := CreateManagerWithWorkers(
 		context.Background(),
 		cfg,
-		kv,
+		noopKVStore{},
 		nil,
 		nil,
 		events.Emitter(nil),
-		nil,
+		&factoryTestRedisClient{client: redisClient},
 		ManagerConfig{
 			Chains: []string{"chain-a"},
 		},
@@ -197,6 +192,14 @@ func TestCreateManagerWithWorkersBootstrapsCatchupRangesIntoStatusRegistry(t *te
 	require.Len(t, resp.Networks, 1)
 	require.Equal(t, 1, resp.Networks[0].CatchupRanges)
 	require.Equal(t, uint64(10), resp.Networks[0].CatchupPendingBlocks)
+
+	require.NoError(t, redisClient.Del(context.Background(), "catchup_progress:a").Err())
+	require.NoError(t, redisClient.HSet(context.Background(), "catchup_progress:a", "31-40", "35").Err())
+
+	resp = manager.StatusSnapshot("1.0.0")
+	require.Len(t, resp.Networks, 1)
+	require.Equal(t, 1, resp.Networks[0].CatchupRanges)
+	require.Equal(t, uint64(5), resp.Networks[0].CatchupPendingBlocks)
 }
 
 type noopKVStore struct{}
@@ -244,3 +247,56 @@ func (s *listKVStore) List(prefix string) ([]*infra.KVPair, error) {
 	}
 	return out, nil
 }
+
+type factoryTestRedisClient struct {
+	client *redis.Client
+}
+
+func (r *factoryTestRedisClient) GetClient() *redis.Client { return r.client }
+func (r *factoryTestRedisClient) Set(key string, value any, expiration time.Duration) error {
+	return r.client.Set(context.Background(), key, value, expiration).Err()
+}
+func (r *factoryTestRedisClient) Get(key string) (string, error) {
+	return r.client.Get(context.Background(), key).Result()
+}
+func (r *factoryTestRedisClient) Del(keys ...string) error {
+	return r.client.Del(context.Background(), keys...).Err()
+}
+func (r *factoryTestRedisClient) ZAdd(key string, members ...redis.Z) error {
+	return r.client.ZAdd(context.Background(), key, members...).Err()
+}
+func (r *factoryTestRedisClient) ZRem(key string, members ...interface{}) error {
+	return r.client.ZRem(context.Background(), key, members...).Err()
+}
+func (r *factoryTestRedisClient) ZRange(key string, start, stop int64) ([]string, error) {
+	return r.client.ZRange(context.Background(), key, start, stop).Result()
+}
+func (r *factoryTestRedisClient) ZRangeWithScores(key string, start, stop int64) ([]redis.Z, error) {
+	return r.client.ZRangeWithScores(context.Background(), key, start, stop).Result()
+}
+func (r *factoryTestRedisClient) ZRevRangeWithScores(key string, start, stop int64) ([]redis.Z, error) {
+	return r.client.ZRevRangeWithScores(context.Background(), key, start, stop).Result()
+}
+func (r *factoryTestRedisClient) Close() error { return r.client.Close() }
+
+func setupFactoryTestRedis(t *testing.T) (*redis.Client, func()) {
+	t.Helper()
+
+	client := redis.NewClient(&redis.Options{
+		Addr: "localhost:6379",
+		DB:   13,
+	})
+
+	ctx := context.Background()
+	if _, err := client.Ping(ctx).Result(); err != nil {
+		t.Skip("Redis not available")
+	}
+	require.NoError(t, client.FlushDB(ctx).Err())
+
+	cleanup := func() {
+		_ = client.FlushDB(ctx).Err()
+		_ = client.Close()
+	}
+
+	return client, cleanup
+}