backfill working

Author: nischitpra

cmd/migrate_valid.go

@@ -181,7 +181,7 @@ func createMigratedBlockRangesTable(psql *storage.PostgresConnector) {
 			chain_id BIGINT NOT NULL,
 			block_number BIGINT NOT NULL,
 			created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
-			updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
+			updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
 		) WITH (fillfactor = 80, autovacuum_vacuum_scale_factor = 0.1, autovacuum_analyze_scale_factor = 0.05)
 	`
 
@@ -242,7 +242,7 @@ func (m *Migrator) DetermineMigrationBoundaries() (*big.Int, *big.Int) {
 
 func (m *Migrator) getAbsStartAndEndBlock() (*big.Int, *big.Int) {
 	// get latest block from main storage
-	latestBlockStored, err := m.storage.MainStorage.GetMaxBlockNumber(m.rpcClient.GetChainID())
+	latestBlockStored, err := m.rpcClient.GetLatestBlockNumber(context.Background())
 	if err != nil {
 		log.Fatal().Err(err).Msg("Failed to get latest block from main storage")
 	}
@@ -304,13 +304,14 @@ func (m *Migrator) GetMaxBlockNumberInRange(startBlock *big.Int, endBlock *big.I
 
 	// Get the maximum end_block for the given chain_id
 	maxBlock, err := m.getMaxMigratedBlock(chainID, startBlock, endBlock)
-	if err != nil {
-		log.Warn().Err(err).Msg("Failed to get last migrated block, returning start block")
-		return startBlock, err
+	if err != nil || maxBlock.Cmp(startBlock) < 0 {
+		log.Warn().Err(err).Msg("Failed to get last migrated block, returning start block - 1")
+		// Return startBlock - 1 so that the next block to migrate is startBlock
+		return new(big.Int).Sub(startBlock, big.NewInt(1)), nil
 	}
 
-	// Return maxBlock + 1 as the next block to migrate
-	return new(big.Int).Add(maxBlock, big.NewInt(1)), nil
+	// Return the actual maxBlock (not +1) since this represents the last migrated block
+	return maxBlock, nil
 }
 
 // upsertMigratedBlockRange upserts a row for the given chain_id and block range

configs/kafka_config.yml

@@ -0,0 +1,86 @@
+rpc:
+  url: https://eth.llamarpc.com
+  chainId: "1"
+  blockReceipts:
+    enabled: true
+
+log:
+  level: debug
+  prettify: true
+
+poller:
+  enabled: true
+  interval: 2000
+  blocksPerPoll: 100
+
+committer:
+  enabled: true
+  interval: 2000
+  blocksPerCommit: 100
+
+storage:
+  main:
+    clickhouse:
+      host: localhost
+      port: 9440
+      username: admin
+      password: password
+      database: default
+      disableTLS: true
+      asyncInsert: true
+      maxRowsPerInsert: 1000
+      maxOpenConns: 50
+      maxIdleConns: 10
+  
+  staging:
+    postgres:
+      host: localhost
+      port: 5432
+      username: admin
+      password: password
+      database: insight
+      sslMode: disable
+      maxOpenConns: 50
+      maxIdleConns: 10
+      maxConnLifetime: 300
+      connectTimeout: 10
+  
+  orchestrator:
+    postgres:
+      host: localhost
+      port: 5432
+      username: admin
+      password: password
+      database: insight
+      sslMode: disable
+      maxOpenConns: 50
+      maxIdleConns: 10
+      maxConnLifetime: 300
+      connectTimeout: 10
+
+api:
+  host: localhost:3000
+  basicAuth:
+    username: admin
+    password: admin
+
+publisher:
+  enabled: false
+  mode: default
+
+# New Kafka configuration
+newKafka:
+  brokers: "localhost:9092"
+  username: ""
+  password: ""
+
+validation:
+  mode: minimal 
+
+# Work mode configuration - Controls system behavior based on blockchain state
+workMode:
+  # Interval in minutes to check if system should switch between live/historical mode
+  checkIntervalMinutes: 10
+  # Block number threshold to determine if system is in "live mode" (near chain head)
+  # Setting this very high forces backfill mode for testing
+  liveModeThreshold: 1000000

internal/storage/clickhouse.go

@@ -2016,6 +2016,11 @@ func (c *ClickHouseConnector) FindMissingBlockNumbers(chainId *big.Int, startBlo
 }
 
 func (c *ClickHouseConnector) GetFullBlockData(chainId *big.Int, blockNumbers []*big.Int) (blocks []common.BlockData, err error) {
+	// For migration purposes, we don't need ForceConsistentData which causes FINAL keyword issues
+	return c.GetFullBlockDataWithOptions(chainId, blockNumbers, true)
+}
+
+func (c *ClickHouseConnector) GetFullBlockDataWithOptions(chainId *big.Int, blockNumbers []*big.Int, forceConsistentData bool) (blocks []common.BlockData, err error) {
 	// Get blocks, logs and transactions concurrently
 	type blockResult struct {
 		blocks []common.Block
@@ -2047,7 +2052,7 @@ func (c *ClickHouseConnector) GetFullBlockData(chainId *big.Int, blockNumbers []
 		blocksResult, err := c.GetBlocks(QueryFilter{
 			ChainId:             chainId,
 			BlockNumbers:        blockNumbers,
-			ForceConsistentData: true,
+			ForceConsistentData: forceConsistentData,
 		})
 		blocksChan <- blockResult{blocks: blocksResult.Data, err: err}
 	}()
@@ -2056,7 +2061,7 @@ func (c *ClickHouseConnector) GetFullBlockData(chainId *big.Int, blockNumbers []
 		logsResult, err := c.GetLogs(QueryFilter{
 			ChainId:             chainId,
 			BlockNumbers:        blockNumbers,
-			ForceConsistentData: true,
+			ForceConsistentData: forceConsistentData,
 		})
 		if err != nil {
 			logsChan <- logResult{err: err}
@@ -2076,7 +2081,7 @@ func (c *ClickHouseConnector) GetFullBlockData(chainId *big.Int, blockNumbers []
 		transactionsResult, err := c.GetTransactions(QueryFilter{
 			ChainId:             chainId,
 			BlockNumbers:        blockNumbers,
-			ForceConsistentData: true,
+			ForceConsistentData: forceConsistentData,
 		})
 		if err != nil {
 			txsChan <- txResult{err: err}
@@ -2096,7 +2101,7 @@ func (c *ClickHouseConnector) GetFullBlockData(chainId *big.Int, blockNumbers []
 		tracesResult, err := c.GetTraces(QueryFilter{
 			ChainId:             chainId,
 			BlockNumbers:        blockNumbers,
-			ForceConsistentData: true,
+			ForceConsistentData: forceConsistentData,
 		})
 		if err != nil {
 			tracesChan <- traceResult{err: err}