New committer (#271)

* Initial kafka committer

* Update config

* Error on uninitialize brokers

* Update queries

* Option to disable TLS for kafka

* Add projection mode in blocks

* Fix publish parallel mode

* Gofmt

* Update schema

* Fix backfill table

* Update kafka storage producer

* Kafka + Redis

* Update schema payload

* Update kafka-postgres -> kafka-redis config

* Update schema to use replacing merge tree

* Fix schema

* Badger & S3

* Until block for committer

* terminate when poller or committer exit

* Fix commit until block

* Don't cancel active tasks in poller

* migrate with destination storage

* Remove RPC batch config in migrate

* Cleanup

* Add from_address, to_address to schema

* Retry with RPC batch size reduction

* Shuffle Orchestrator and Staging interface

* store poller in committer

* Simplified storage. Split kafka and redis

* kafka requires orchestrator

* Fix orchestrator flag

* Fix badger keys

* Fix backfill missing blocks in staging

* Revert "Fix backfill missing blocks in staging"

This reverts commit 6233232.

* block buffer

* Poller S3 support

* Fix boundaries for migration

* Badger for caching in s3 connector

* optimize s3 insertion

* redis tls. erc1155 batch mv

* fix projections, use _part_offset projections

* gofmt

* Fix test

Author: jakeloo

cmd/migrate_valid.go

@@ -32,12 +32,19 @@ func RunOrchestrator(cmd *cobra.Command, args []string) {
 	if err != nil {
 		log.Fatal().Err(err).Msg("Failed to create orchestrator")
 	}
+
 	// Start Prometheus metrics server
 	log.Info().Msg("Starting Metrics Server on port 2112")
 	go func() {
 		http.Handle("/metrics", promhttp.Handler())
-		http.ListenAndServe(":2112", nil)
+		if err := http.ListenAndServe(":2112", nil); err != nil {
+			log.Error().Err(err).Msg("Metrics server error")
+		}
 	}()
 
+	// Start orchestrator (blocks until shutdown)
+	// The orchestrator handles signals internally and coordinates shutdown
 	orchestrator.Start()
+
+	log.Info().Msg("Shutdown complete")
 }

cmd/orchestrator.go

@@ -5,6 +5,7 @@ import (
 	"fmt"
 	"os"
 	"strings"
+	"time"
 
 	"github.com/rs/zerolog/log"
 	"github.com/spf13/viper"
@@ -16,20 +17,22 @@ 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 {
 	Enabled         bool `mapstructure:"enabled"`
 	Interval        int  `mapstructure:"interval"`
 	BlocksPerCommit int  `mapstructure:"blocksPerCommit"`
 	FromBlock       int  `mapstructure:"fromBlock"`
+	UntilBlock      int  `mapstructure:"untilBlock"`
 }
 
 type ReorgHandlerConfig struct {
@@ -47,21 +50,62 @@ type FailureRecovererConfig struct {
 }
 
 type StorageConfig struct {
-	Staging      StorageConnectionConfig `mapstructure:"staging"`
-	Main         StorageConnectionConfig `mapstructure:"main"`
-	Orchestrator StorageConnectionConfig `mapstructure:"orchestrator"`
+	Orchestrator StorageOrchestratorConfig `mapstructure:"orchestrator"`
+	Staging      StorageStagingConfig      `mapstructure:"staging"`
+	Main         StorageMainConfig         `mapstructure:"main"`
 }
-type StorageType string
 
-const (
-	StorageTypeMain         StorageType = "main"
-	StorageTypeStaging      StorageType = "staging"
-	StorageTypeOrchestrator StorageType = "orchestrator"
-)
+type StorageOrchestratorConfig struct {
+	Type       string            `mapstructure:"type"`
+	Clickhouse *ClickhouseConfig `mapstructure:"clickhouse"`
+	Postgres   *PostgresConfig   `mapstructure:"postgres"`
+	Redis      *RedisConfig      `mapstructure:"redis"`
+	Badger     *BadgerConfig     `mapstructure:"badger"`
+}
 
-type StorageConnectionConfig struct {
+type StorageStagingConfig struct {
+	Type       string            `mapstructure:"type"`
 	Clickhouse *ClickhouseConfig `mapstructure:"clickhouse"`
 	Postgres   *PostgresConfig   `mapstructure:"postgres"`
+	Badger     *BadgerConfig     `mapstructure:"badger"`
+}
+
+type StorageMainConfig struct {
+	Type       string            `mapstructure:"type"`
+	Clickhouse *ClickhouseConfig `mapstructure:"clickhouse"`
+	Postgres   *PostgresConfig   `mapstructure:"postgres"`
+	Kafka      *KafkaConfig      `mapstructure:"kafka"`
+	Badger     *BadgerConfig     `mapstructure:"badger"`
+	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"`
+}
+
+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)
+	MaxBlocksPerFile int   `mapstructure:"maxBlocksPerFile"`     // Max blocks per parquet file (0 = no limit, only size/timeout triggers)
+}
+
+type ParquetConfig struct {
+	Compression  string `mapstructure:"compression"`
+	RowGroupSize int64  `mapstructure:"rowGroupSize"`
+	PageSize     int64  `mapstructure:"pageSize"`
 }
 
 type TableConfig struct {
@@ -86,6 +130,7 @@ type ClickhouseConfig struct {
 	EnableParallelViewProcessing bool                           `mapstructure:"enableParallelViewProcessing"`
 	MaxQueryTime                 int                            `mapstructure:"maxQueryTime"`
 	MaxMemoryUsage               int                            `mapstructure:"maxMemoryUsage"`
+	EnableCompression            bool                           `mapstructure:"enableCompression"`
 }
 
 type PostgresConfig struct {
@@ -101,6 +146,21 @@ type PostgresConfig struct {
 	ConnectTimeout  int    `mapstructure:"connectTimeout"`
 }
 
+type RedisConfig struct {
+	Host      string `mapstructure:"host"`
+	Port      int    `mapstructure:"port"`
+	Password  string `mapstructure:"password"`
+	DB        int    `mapstructure:"db"`
+	EnableTLS bool   `mapstructure:"enableTLS"`
+}
+
+type KafkaConfig struct {
+	Brokers   string `mapstructure:"brokers"`
+	Username  string `mapstructure:"username"`
+	Password  string `mapstructure:"password"`
+	EnableTLS bool   `mapstructure:"enableTLS"`
+}
+
 type RPCBatchRequestConfig struct {
 	BlocksPerRequest int `mapstructure:"blocksPerRequest"`
 	BatchDelay       int `mapstructure:"batchDelay"`
@@ -177,12 +237,23 @@ type PublisherConfig struct {
 	Brokers      string                     `mapstructure:"brokers"`
 	Username     string                     `mapstructure:"username"`
 	Password     string                     `mapstructure:"password"`
+	EnableTLS    bool                       `mapstructure:"enableTLS"`
 	Blocks       BlockPublisherConfig       `mapstructure:"blocks"`
 	Transactions TransactionPublisherConfig `mapstructure:"transactions"`
 	Traces       TracePublisherConfig       `mapstructure:"traces"`
 	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"`
@@ -192,6 +263,14 @@ type ValidationConfig struct {
 	Mode string `mapstructure:"mode"` // "disabled", "minimal", "strict"
 }
 
+type MigratorConfig struct {
+	Destination StorageMainConfig `mapstructure:"destination"`
+	StartBlock  uint              `mapstructure:"startBlock"`
+	EndBlock    uint              `mapstructure:"endBlock"`
+	BatchSize   uint              `mapstructure:"batchSize"`
+	WorkerCount uint              `mapstructure:"workerCount"`
+}
+
 type Config struct {
 	RPC              RPCConfig              `mapstructure:"rpc"`
 	Log              LogConfig              `mapstructure:"log"`
@@ -204,6 +283,7 @@ type Config struct {
 	Publisher        PublisherConfig        `mapstructure:"publisher"`
 	WorkMode         WorkModeConfig         `mapstructure:"workMode"`
 	Validation       ValidationConfig       `mapstructure:"validation"`
+	Migrator         MigratorConfig         `mapstructure:"migrator"`
 }
 
 var Cfg Config

cmd/root.go

@@ -4,14 +4,20 @@ go 1.23.0
 
 require (
 	github.com/ClickHouse/clickhouse-go/v2 v2.36.0
+	github.com/aws/aws-sdk-go-v2 v1.38.0
+	github.com/aws/aws-sdk-go-v2/config v1.31.0
+	github.com/aws/aws-sdk-go-v2/service/s3 v1.87.0
+	github.com/dgraph-io/badger/v4 v4.8.0
 	github.com/ethereum/go-ethereum v1.15.11
 	github.com/gin-gonic/gin v1.10.0
 	github.com/gorilla/schema v1.4.1
 	github.com/holiman/uint256 v1.3.2
 	github.com/lib/pq v1.10.9
+	github.com/parquet-go/parquet-go v0.25.1
 	github.com/prometheus/client_golang v1.20.4
+	github.com/redis/go-redis/v9 v9.12.1
 	github.com/rs/zerolog v1.33.0
-	github.com/spf13/cobra v1.8.1
+	github.com/spf13/cobra v1.9.1
 	github.com/spf13/viper v1.18.0
 	github.com/stretchr/testify v1.10.0
 	github.com/swaggo/files v1.0.1
@@ -25,6 +31,21 @@ require (
 	github.com/KyleBanks/depth v1.2.1 // indirect
 	github.com/Microsoft/go-winio v0.6.2 // indirect
 	github.com/andybalholm/brotli v1.1.1 // indirect
+	github.com/aws/aws-sdk-go-v2/aws/protocol/eventstream v1.7.0 // indirect
+	github.com/aws/aws-sdk-go-v2/credentials v1.18.4 // indirect
+	github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.18.3 // indirect
+	github.com/aws/aws-sdk-go-v2/internal/configsources v1.4.3 // indirect
+	github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.7.3 // indirect
+	github.com/aws/aws-sdk-go-v2/internal/ini v1.8.3 // indirect
+	github.com/aws/aws-sdk-go-v2/internal/v4a v1.4.3 // indirect
+	github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.13.0 // indirect
+	github.com/aws/aws-sdk-go-v2/service/internal/checksum v1.8.3 // indirect
+	github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.13.3 // indirect
+	github.com/aws/aws-sdk-go-v2/service/internal/s3shared v1.19.3 // indirect
+	github.com/aws/aws-sdk-go-v2/service/sso v1.28.0 // indirect
+	github.com/aws/aws-sdk-go-v2/service/ssooidc v1.33.0 // indirect
+	github.com/aws/aws-sdk-go-v2/service/sts v1.37.0 // indirect
+	github.com/aws/smithy-go v1.22.5 // indirect
 	github.com/beorn7/perks v1.0.1 // indirect
 	github.com/bits-and-blooms/bitset v1.20.0 // indirect
 	github.com/bytedance/sonic v1.12.6 // indirect
@@ -39,13 +60,18 @@ require (
 	github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc // indirect
 	github.com/deckarep/golang-set/v2 v2.6.0 // indirect
 	github.com/decred/dcrd/dcrec/secp256k1/v4 v4.0.1 // indirect
+	github.com/dgraph-io/ristretto/v2 v2.2.0 // indirect
+	github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f // indirect
+	github.com/dustin/go-humanize v1.0.1 // indirect
 	github.com/ethereum/c-kzg-4844/v2 v2.1.0 // indirect
 	github.com/ethereum/go-verkle v0.2.2 // indirect
 	github.com/fsnotify/fsnotify v1.7.0 // indirect
 	github.com/gabriel-vasile/mimetype v1.4.7 // indirect
 	github.com/gin-contrib/sse v0.1.0 // indirect
 	github.com/go-faster/city v1.0.1 // indirect
 	github.com/go-faster/errors v0.7.1 // indirect
+	github.com/go-logr/logr v1.4.3 // indirect
+	github.com/go-logr/stdr v1.2.2 // indirect
 	github.com/go-ole/go-ole v1.3.0 // indirect
 	github.com/go-openapi/jsonpointer v0.21.0 // indirect
 	github.com/go-openapi/jsonreference v0.21.0 // indirect
@@ -57,6 +83,7 @@ require (
 	github.com/goccy/go-json v0.10.4 // indirect
 	github.com/gofrs/flock v0.8.1 // indirect
 	github.com/golang/snappy v0.0.5-0.20220116011046-fa5810519dcb // indirect
+	github.com/google/flatbuffers v25.2.10+incompatible // indirect
 	github.com/google/uuid v1.6.0 // indirect
 	github.com/gorilla/websocket v1.4.2 // indirect
 	github.com/hashicorp/hcl v1.0.0 // indirect
@@ -94,7 +121,7 @@ require (
 	github.com/sourcegraph/conc v0.3.0 // indirect
 	github.com/spf13/afero v1.11.0 // indirect
 	github.com/spf13/cast v1.6.0 // indirect
-	github.com/spf13/pflag v1.0.5 // indirect
+	github.com/spf13/pflag v1.0.6 // indirect
 	github.com/stretchr/objx v0.5.2 // indirect
 	github.com/subosito/gotenv v1.6.0 // indirect
 	github.com/supranational/blst v0.3.14 // indirect
@@ -104,18 +131,20 @@ require (
 	github.com/twmb/franz-go/pkg/kmsg v1.9.0 // indirect
 	github.com/ugorji/go/codec v1.2.12 // indirect
 	github.com/yusufpapurcu/wmi v1.2.4 // indirect
-	go.opentelemetry.io/otel v1.36.0 // indirect
-	go.opentelemetry.io/otel/trace v1.36.0 // indirect
+	go.opentelemetry.io/auto/sdk v1.1.0 // indirect
+	go.opentelemetry.io/otel v1.37.0 // indirect
+	go.opentelemetry.io/otel/metric v1.37.0 // indirect
+	go.opentelemetry.io/otel/trace v1.37.0 // indirect
 	go.uber.org/multierr v1.11.0 // indirect
 	golang.org/x/arch v0.12.0 // indirect
-	golang.org/x/crypto v0.38.0 // indirect
+	golang.org/x/crypto v0.39.0 // indirect
 	golang.org/x/exp v0.0.0-20240325151524-a685a6edb6d8 // indirect
-	golang.org/x/net v0.40.0 // indirect
-	golang.org/x/sync v0.14.0 // indirect
-	golang.org/x/sys v0.33.0 // indirect
-	golang.org/x/text v0.25.0 // indirect
-	golang.org/x/tools v0.30.0 // indirect
-	google.golang.org/protobuf v1.36.1 // indirect
+	golang.org/x/net v0.41.0 // indirect
+	golang.org/x/sync v0.15.0 // indirect
+	golang.org/x/sys v0.34.0 // indirect
+	golang.org/x/text v0.26.0 // indirect
+	golang.org/x/tools v0.33.0 // indirect
+	google.golang.org/protobuf v1.36.6 // indirect
 	gopkg.in/ini.v1 v1.67.0 // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect
 	rsc.io/tmplfunc v0.0.3 // indirect