Poller S3 support

Author: jakeloo

cmd/root.go

@@ -56,6 +56,19 @@ func init() {
 	rootCmd.PersistentFlags().Bool("poller-force-from-block", false, "Force the poller to start from the block specified in `poller-from-block`")
 	rootCmd.PersistentFlags().Int("poller-until-block", 0, "Until which block to poll")
 	rootCmd.PersistentFlags().Int("poller-parallel-pollers", 5, "Maximum number of parallel pollers")
+	rootCmd.PersistentFlags().String("poller-s3-bucket", "", "S3 bucket for poller archive source")
+	rootCmd.PersistentFlags().String("poller-s3-region", "", "S3 region for poller archive source")
+	rootCmd.PersistentFlags().String("poller-s3-prefix", "", "S3 prefix for poller archive source")
+	rootCmd.PersistentFlags().String("poller-s3-accessKeyId", "", "S3 access key ID for poller archive source")
+	rootCmd.PersistentFlags().String("poller-s3-secretAccessKey", "", "S3 secret access key for poller archive source")
+	rootCmd.PersistentFlags().String("poller-s3-endpoint", "", "S3 endpoint for poller archive source (for S3-compatible services)")
+	rootCmd.PersistentFlags().String("poller-s3-format", "parquet", "S3 storage format for poller archive source")
+	rootCmd.PersistentFlags().String("poller-s3-cacheDir", "/tmp/insight-archive", "Local cache directory for poller archive source")
+	rootCmd.PersistentFlags().Int("poller-s3-metadataTTL", 0, "Metadata cache TTL in seconds for poller archive source")
+	rootCmd.PersistentFlags().Int("poller-s3-fileCacheTTL", 0, "File cache TTL in seconds for poller archive source")
+	rootCmd.PersistentFlags().Int64("poller-s3-maxCacheSize", 0, "Max cache size in bytes for poller archive source (default 5GB)")
+	rootCmd.PersistentFlags().Int("poller-s3-cleanupInterval", 0, "Cache cleanup interval in seconds for poller archive source")
+	rootCmd.PersistentFlags().Int("poller-s3-maxConcurrentDownloads", 3, "Max concurrent downloads for poller archive source")
 	rootCmd.PersistentFlags().Bool("committer-enabled", true, "Toggle committer")
 	rootCmd.PersistentFlags().Int("committer-blocks-per-commit", 10, "How many blocks to commit each interval")
 	rootCmd.PersistentFlags().Int("committer-interval", 1000, "How often to commit blocks in milliseconds")
@@ -247,6 +260,18 @@ func init() {
 	viper.BindPFlag("poller.forceFromBlock", rootCmd.PersistentFlags().Lookup("poller-force-from-block"))
 	viper.BindPFlag("poller.untilBlock", rootCmd.PersistentFlags().Lookup("poller-until-block"))
 	viper.BindPFlag("poller.parallelPollers", rootCmd.PersistentFlags().Lookup("poller-parallel-pollers"))
+	viper.BindPFlag("poller.s3.endpoint", rootCmd.PersistentFlags().Lookup("poller-s3-endpoint"))
+	viper.BindPFlag("poller.s3.accessKeyId", rootCmd.PersistentFlags().Lookup("poller-s3-accessKeyId"))
+	viper.BindPFlag("poller.s3.secretAccessKey", rootCmd.PersistentFlags().Lookup("poller-s3-secretAccessKey"))
+	viper.BindPFlag("poller.s3.bucket", rootCmd.PersistentFlags().Lookup("poller-s3-bucket"))
+	viper.BindPFlag("poller.s3.region", rootCmd.PersistentFlags().Lookup("poller-s3-region"))
+	viper.BindPFlag("poller.s3.prefix", rootCmd.PersistentFlags().Lookup("poller-s3-prefix"))
+	viper.BindPFlag("poller.s3.cacheDir", rootCmd.PersistentFlags().Lookup("poller-s3-cacheDir"))
+	viper.BindPFlag("poller.s3.metadataTTL", rootCmd.PersistentFlags().Lookup("poller-s3-metadataTTL"))
+	viper.BindPFlag("poller.s3.fileCacheTTL", rootCmd.PersistentFlags().Lookup("poller-s3-fileCacheTTL"))
+	viper.BindPFlag("poller.s3.maxCacheSize", rootCmd.PersistentFlags().Lookup("poller-s3-maxCacheSize"))
+	viper.BindPFlag("poller.s3.cleanupInterval", rootCmd.PersistentFlags().Lookup("poller-s3-cleanupInterval"))
+	viper.BindPFlag("poller.s3.maxConcurrentDownloads", rootCmd.PersistentFlags().Lookup("poller-s3-maxConcurrentDownloads"))
 	viper.BindPFlag("committer.enabled", rootCmd.PersistentFlags().Lookup("committer-enabled"))
 	viper.BindPFlag("committer.blocksPerCommit", rootCmd.PersistentFlags().Lookup("committer-blocks-per-commit"))
 	viper.BindPFlag("committer.interval", rootCmd.PersistentFlags().Lookup("committer-interval"))

configs/config.go

@@ -5,6 +5,7 @@ import (
 	"fmt"
 	"os"
 	"strings"
+	"time"
 
 	"github.com/rs/zerolog/log"
 	"github.com/spf13/viper"
@@ -16,13 +17,14 @@ type LogConfig struct {
 }
 
 type PollerConfig struct {
-	Enabled         bool `mapstructure:"enabled"`
-	Interval        int  `mapstructure:"interval"`
-	BlocksPerPoll   int  `mapstructure:"blocksPerPoll"`
-	FromBlock       int  `mapstructure:"fromBlock"`
-	ForceFromBlock  bool `mapstructure:"forceFromBlock"`
-	UntilBlock      int  `mapstructure:"untilBlock"`
-	ParallelPollers int  `mapstructure:"parallelPollers"`
+	Enabled         bool            `mapstructure:"enabled"`
+	Interval        int             `mapstructure:"interval"`
+	BlocksPerPoll   int             `mapstructure:"blocksPerPoll"`
+	FromBlock       int             `mapstructure:"fromBlock"`
+	ForceFromBlock  bool            `mapstructure:"forceFromBlock"`
+	UntilBlock      int             `mapstructure:"untilBlock"`
+	ParallelPollers int             `mapstructure:"parallelPollers"`
+	S3              *S3SourceConfig `mapstructure:"s3"`
 }
 
 type CommitterConfig struct {
@@ -74,22 +76,26 @@ type StorageMainConfig struct {
 	Postgres   *PostgresConfig   `mapstructure:"postgres"`
 	Kafka      *KafkaConfig      `mapstructure:"kafka"`
 	Badger     *BadgerConfig     `mapstructure:"badger"`
-	S3         *S3Config         `mapstructure:"s3"`
+	S3         *S3StorageConfig  `mapstructure:"s3"`
 }
 
 type BadgerConfig struct {
 	Path string `mapstructure:"path"`
 }
 
 type S3Config struct {
-	Bucket          string         `mapstructure:"bucket"`
-	Region          string         `mapstructure:"region"`
-	Prefix          string         `mapstructure:"prefix"`
-	AccessKeyID     string         `mapstructure:"accessKeyId"`
-	SecretAccessKey string         `mapstructure:"secretAccessKey"`
-	Endpoint        string         `mapstructure:"endpoint"`
-	Format          string         `mapstructure:"format"`
-	Parquet         *ParquetConfig `mapstructure:"parquet"`
+	Bucket          string `mapstructure:"bucket"`
+	Region          string `mapstructure:"region"`
+	Prefix          string `mapstructure:"prefix"`
+	AccessKeyID     string `mapstructure:"accessKeyId"`
+	SecretAccessKey string `mapstructure:"secretAccessKey"`
+	Endpoint        string `mapstructure:"endpoint"`
+}
+
+type S3StorageConfig struct {
+	S3Config `mapstructure:",squash"`
+	Format   string         `mapstructure:"format"`
+	Parquet  *ParquetConfig `mapstructure:"parquet"`
 	// Buffering configuration
 	BufferSize       int64 `mapstructure:"bufferSizeMB"`         // Target buffer size in MB before flush (default 512 MB)
 	BufferTimeout    int   `mapstructure:"bufferTimeoutSeconds"` // Max time in seconds before flush (default 300 = 5 min)
@@ -237,6 +243,16 @@ type PublisherConfig struct {
 	Events       EventPublisherConfig       `mapstructure:"events"`
 }
 
+type S3SourceConfig struct {
+	S3Config               `mapstructure:",squash"`
+	CacheDir               string        `mapstructure:"cacheDir"`
+	MetadataTTL            time.Duration `mapstructure:"metadataTTL"`
+	FileCacheTTL           time.Duration `mapstructure:"fileCacheTTL"`
+	MaxCacheSize           int64         `mapstructure:"maxCacheSize"`
+	CleanupInterval        time.Duration `mapstructure:"cleanupInterval"`
+	MaxConcurrentDownloads int           `mapstructure:"maxConcurrentDownloads"`
+}
+
 type WorkModeConfig struct {
 	CheckIntervalMinutes int   `mapstructure:"checkIntervalMinutes"`
 	LiveModeThreshold    int64 `mapstructure:"liveModeThreshold"`

internal/orchestrator/orchestrator.go

@@ -22,7 +22,6 @@ type Orchestrator struct {
 	reorgHandlerEnabled     bool
 	cancel                  context.CancelFunc
 	wg                      sync.WaitGroup
-	shutdownOnce            sync.Once
 }
 
 func NewOrchestrator(rpc rpc.IRPCClient) (*Orchestrator, error) {
@@ -65,7 +64,10 @@ func (o *Orchestrator) Start() {
 			workModeMonitor.RegisterChannel(pollerWorkModeChan)
 			defer workModeMonitor.UnregisterChannel(pollerWorkModeChan)
 
-			poller := NewPoller(o.rpc, o.storage, WithPollerWorkModeChan(pollerWorkModeChan))
+			poller := NewPoller(o.rpc, o.storage,
+				WithPollerWorkModeChan(pollerWorkModeChan),
+				WithPollerS3Source(config.Cfg.Poller.S3),
+			)
 			poller.Start(ctx)
 
 			log.Info().Msg("Poller completed")

internal/orchestrator/poller.go

@@ -12,6 +12,7 @@ import (
 	"github.com/thirdweb-dev/indexer/internal/common"
 	"github.com/thirdweb-dev/indexer/internal/metrics"
 	"github.com/thirdweb-dev/indexer/internal/rpc"
+	"github.com/thirdweb-dev/indexer/internal/source"
 	"github.com/thirdweb-dev/indexer/internal/storage"
 	"github.com/thirdweb-dev/indexer/internal/worker"
 )
@@ -21,6 +22,7 @@ const DEFAULT_TRIGGER_INTERVAL = 1000
 
 type Poller struct {
 	rpc                  rpc.IRPCClient
+	worker               *worker.Worker
 	blocksPerPoll        int64
 	triggerIntervalMs    int64
 	storage              storage.IStorage
@@ -47,15 +49,33 @@ func WithPollerWorkModeChan(ch chan WorkMode) PollerOption {
 	}
 }
 
+func WithPollerS3Source(cfg *config.S3SourceConfig) PollerOption {
+	return func(p *Poller) {
+		if cfg == nil || cfg.Region == "" || cfg.Bucket == "" {
+			return
+		}
+
+		source, err := source.NewS3Source(cfg, p.rpc.GetChainID())
+		if err != nil {
+			log.Fatal().Err(err).Msg("Failed to create S3 source")
+		}
+
+		log.Info().Msg("Poller S3 source configuration detected, setting up S3 source for poller")
+		p.worker = worker.NewWorkerWithArchive(p.rpc, source)
+	}
+}
+
 func NewBoundlessPoller(rpc rpc.IRPCClient, storage storage.IStorage, opts ...PollerOption) *Poller {
 	blocksPerPoll := config.Cfg.Poller.BlocksPerPoll
 	if blocksPerPoll == 0 {
 		blocksPerPoll = DEFAULT_BLOCKS_PER_POLL
 	}
+
 	triggerInterval := config.Cfg.Poller.Interval
 	if triggerInterval == 0 {
 		triggerInterval = DEFAULT_TRIGGER_INTERVAL
 	}
+
 	poller := &Poller{
 		rpc:               rpc,
 		triggerIntervalMs: int64(triggerInterval),
@@ -68,6 +88,10 @@ func NewBoundlessPoller(rpc rpc.IRPCClient, storage storage.IStorage, opts ...Po
 		opt(poller)
 	}
 
+	if poller.worker == nil {
+		poller.worker = worker.NewWorker(poller.rpc)
+	}
+
 	return poller
 }
 
@@ -235,8 +259,7 @@ func (p *Poller) PollWithoutSaving(ctx context.Context, blockNumbers []*big.Int)
 	endBlockNumberFloat, _ := endBlock.Float64()
 	metrics.PollerLastTriggeredBlock.Set(endBlockNumberFloat)
 
-	worker := worker.NewWorker(p.rpc)
-	results := worker.Run(ctx, blockNumbers)
+	results := p.worker.Run(ctx, blockNumbers)
 	blockData, failedResults := p.convertPollResultsToBlockData(results)
 	return blockData, failedResults
 }