Minor updates

Author: jakeloo

internal/orchestrator/chain_tracker.go

@@ -9,7 +9,7 @@ import (
 	"github.com/thirdweb-dev/indexer/internal/rpc"
 )
 
-const DEFAULT_CHAIN_TRACKER_POLL_INTERVAL = 300000 // 5 minutes
+const DEFAULT_CHAIN_TRACKER_POLL_INTERVAL = 60 * 1000 // 1 minutes
 
 type ChainTracker struct {
 	rpc               rpc.IRPCClient

internal/orchestrator/orchestrator.go

@@ -51,9 +51,6 @@ func (o *Orchestrator) Start() {
 		o.cancel()
 	}()
 
-	// Create the work mode monitor first
-	workModeMonitor := NewWorkModeMonitor(o.rpc, o.storage)
-
 	o.initializeWorkerAndPoller()
 
 	o.wg.Add(1)
@@ -91,14 +88,6 @@ func (o *Orchestrator) Start() {
 		}()
 	}
 
-	o.wg.Add(1)
-	go func() {
-		defer o.wg.Done()
-		workModeMonitor.Start(ctx)
-
-		log.Info().Msg("Work mode monitor completed")
-	}()
-
 	// The chain tracker is always running
 	o.wg.Add(1)
 	go func() {

internal/storage/badger.go

@@ -25,7 +25,16 @@ type BadgerConnector struct {
 	gcTicker *time.Ticker
 	stopGC   chan struct{}
 
+	// Configuration
+	stagingDataTTL        time.Duration // TTL for staging data entries
+	gcInterval            time.Duration // Interval for running garbage collection
+	cacheRefreshInterval  time.Duration // Interval for refreshing range cache
+	cacheStalenessTimeout time.Duration // Timeout before considering cache entry stale
+
 	// In-memory block range cache
+	// NOTE: Staging data has a TTL. The cache is refreshed periodically
+	// to detect expired entries and update min/max ranges accordingly.
+	// Badger doesn't provide expiry notifications, so we rely on periodic scanning.
 	rangeCache      map[string]*blockRange // chainId -> range
 	rangeCacheMu    sync.RWMutex
 	rangeUpdateChan chan string // channel for triggering background updates
@@ -71,15 +80,19 @@ func NewBadgerConnector(cfg *config.BadgerConfig) (*BadgerConnector, error) {
 	}
 
 	bc := &BadgerConnector{
-		db:              db,
-		stopGC:          make(chan struct{}),
-		rangeCache:      make(map[string]*blockRange),
-		rangeUpdateChan: make(chan string, 5),
-		stopRangeUpdate: make(chan struct{}),
+		db:                    db,
+		stopGC:                make(chan struct{}),
+		rangeCache:            make(map[string]*blockRange),
+		rangeUpdateChan:       make(chan string, 5),
+		stopRangeUpdate:       make(chan struct{}),
+		stagingDataTTL:        10 * time.Minute,
+		gcInterval:            60 * time.Second,
+		cacheRefreshInterval:  60 * time.Second,
+		cacheStalenessTimeout: 120 * time.Second,
 	}
 
 	// Start GC routine
-	bc.gcTicker = time.NewTicker(time.Duration(60) * time.Second)
+	bc.gcTicker = time.NewTicker(bc.gcInterval)
 	go bc.runGC()
 
 	// Start range cache update routine
@@ -104,7 +117,7 @@ func (bc *BadgerConnector) runGC() {
 
 // runRangeCacheUpdater runs in the background to validate cache entries
 func (bc *BadgerConnector) runRangeCacheUpdater() {
-	ticker := time.NewTicker(time.Minute)
+	ticker := time.NewTicker(bc.cacheRefreshInterval)
 	defer ticker.Stop()
 
 	for {
@@ -184,7 +197,7 @@ func (bc *BadgerConnector) refreshStaleRanges() {
 	staleChains := []string{}
 	now := time.Now()
 	for chainId, r := range bc.rangeCache {
-		if now.Sub(r.lastUpdated) > 3*time.Minute {
+		if now.Sub(r.lastUpdated) > bc.cacheStalenessTimeout {
 			staleChains = append(staleChains, chainId)
 		}
 	}
@@ -391,7 +404,9 @@ func (bc *BadgerConnector) InsertStagingData(data []common.BlockData) error {
 				return err
 			}
 
-			if err := txn.Set(key, buf.Bytes()); err != nil {
+			// Set with configured TTL for staging data
+			entry := badger.NewEntry(key, buf.Bytes()).WithTTL(bc.stagingDataTTL)
+			if err := txn.SetEntry(entry); err != nil {
 				return err
 			}
 

internal/worker/worker.go

@@ -431,11 +431,8 @@ func (w *Worker) Run(ctx context.Context, blockNumbers []*big.Int) []rpc.GetFull
 		}
 
 		log.Debug().
-			Int("total", len(results)).
-			Str("first_block", blockNumbers[0].String()).
-			Str("last_block", blockNumbers[len(results)-1].String()).
-			Str("first_block_result", results[0].BlockNumber.String()).
-			Str("last_block_result", results[len(results)-1].BlockNumber.String()).
+			Str("first_block", results[0].BlockNumber.String()).
+			Str("last_block", results[len(results)-1].BlockNumber.String()).
 			Int("successful", successful).
 			Int("failed", failed).
 			Str("source", sourceType.String()).