feat: on start fetch onchain state of oracle

Author: nnn-gif

services/bridge/internal/bridge/bridge.go

@@ -365,6 +365,22 @@ func (b *Bridge) Start(ctx context.Context) error {
 
 	logger.Info("Starting bridge service")
 
+	// Load chain state per oracle per symbol
+	if b.routerRegistry != nil {
+		logger.Info("Fetching router state from on-chain...")
+		ethClients := make(map[int64]rpc.EthClient)
+		for chainID, writeClient := range b.writeClients {
+			ethClients[chainID] = writeClient.GetEthClient()
+		}
+		routers := b.routerRegistry.GetActiveRouters()
+		for _, router := range routers {
+			if err := router.FetchOracleStateFromOnChain(ctx, ethClients); err != nil {
+				logger.Warnf("Router state fetch failed for %s: %v", router.ID(), err)
+			}
+		}
+		logger.Info("Router state fetch completed")
+	}
+
 	// Start transaction queue manager
 	b.queueManager.Start()
 
@@ -450,6 +466,7 @@ func (b *Bridge) getOrCreateOraclePool(routerID string) *worker.WorkerPool {
 		b.workerPoolConfig.TaskTimeout.Duration())
 
 	pool = worker.NewWorkerPool(
+		routerID,
 		b.workerPoolConfig.MaxWorkers,
 		b.workerPoolConfig.TaskQueueSize,
 		b.workerPoolConfig.TaskTimeout.Duration(),

services/bridge/internal/leader/onchain_monitor.go

@@ -1,17 +1,13 @@
 package leader
 
 import (
-	"bytes"
 	"context"
-	"fmt"
 	"math/big"
 	"strconv"
 	"strings"
 	"sync"
 	"time"
 
-	"github.com/ethereum/go-ethereum"
-	"github.com/ethereum/go-ethereum/accounts/abi"
 	"github.com/ethereum/go-ethereum/common"
 
 	"github.com/diadata.org/Spectra-interoperability/pkg/logger"
@@ -288,58 +284,7 @@ func (m *OnChainMonitor) generateKey(chainID int64, contractAddress common.Addre
 }
 
 func (m *OnChainMonitor) getValueFromContract(dest *DestinationMonitor, symbol string) (*big.Int, uint64, error) {
-
-	const getValueABI = `[{
-        "inputs": [{"internalType": "string", "name": "key", "type": "string"}],
-        "name": "getValue",
-        "outputs": [
-            {"internalType": "uint128", "name": "value", "type": "uint128"},
-            {"internalType": "uint128", "name": "timestamp", "type": "uint128"}
-        ],
-        "stateMutability": "view",
-        "type": "function"
-    }]`
-
-	parsedABI, err := abi.JSON(bytes.NewReader([]byte(getValueABI)))
-	if err != nil {
-		return nil, 0, fmt.Errorf("failed to parse ABI: %w", err)
-	}
-
-	data, err := parsedABI.Pack("getValue", symbol)
-	if err != nil {
-		return nil, 0, fmt.Errorf("failed to pack input data: %w", err)
-	}
-
-	callMsg := ethereum.CallMsg{
-		To:   &dest.ContractAddress,
-		Data: data,
-	}
-
-	resultBytes, err := dest.Client.CallContract(m.ctx, callMsg, nil)
-	if err != nil {
-		return nil, 0, fmt.Errorf("contract call failed: %w", err)
-	}
-
-	outputs, err := parsedABI.Unpack("getValue", resultBytes)
-	if err != nil {
-		return nil, 0, fmt.Errorf("failed to unpack result: %w", err)
-	}
-
-	if len(outputs) != 2 {
-		return nil, 0, fmt.Errorf("unexpected number of outputs: got %d, want 2", len(outputs))
-	}
-
-	value, ok := outputs[0].(*big.Int)
-	if !ok {
-		return nil, 0, fmt.Errorf("failed to convert value to big.Int, got type %T: %v", outputs[0], outputs[0])
-	}
-
-	timestamp, ok := outputs[1].(*big.Int)
-	if !ok {
-		return nil, 0, fmt.Errorf("failed to convert timestamp to big.Int, got type %T: %v", outputs[1], outputs[1])
-	}
-
-	return value, timestamp.Uint64(), nil
+	return utils.GetValueFromOracleContract(m.ctx, dest.Client, dest.ContractAddress, symbol)
 }
 
 func formatValue(v *big.Int) string {

services/bridge/internal/utils/contract.go

@@ -0,0 +1,69 @@
+package utils
+
+import (
+	"context"
+	"fmt"
+	"math/big"
+	"strings"
+
+	"github.com/ethereum/go-ethereum"
+	"github.com/ethereum/go-ethereum/accounts/abi"
+	"github.com/ethereum/go-ethereum/common"
+
+	"github.com/diadata.org/Spectra-interoperability/pkg/rpc"
+)
+
+// GetValueFromOracleContract calls the getValue method on an oracle contract
+func GetValueFromOracleContract(ctx context.Context, client rpc.EthClient, contractAddress common.Address, symbol string) (*big.Int, uint64, error) {
+	const getValueABI = `[{
+        "inputs": [{"internalType": "string", "name": "key", "type": "string"}],
+        "name": "getValue",
+        "outputs": [
+            {"internalType": "uint128", "name": "value", "type": "uint128"},
+            {"internalType": "uint128", "name": "timestamp", "type": "uint128"}
+        ],
+        "stateMutability": "view",
+        "type": "function"
+    }]`
+
+	parsedABI, err := abi.JSON(strings.NewReader(getValueABI))
+	if err != nil {
+		return nil, 0, fmt.Errorf("failed to parse ABI: %w", err)
+	}
+
+	data, err := parsedABI.Pack("getValue", symbol)
+	if err != nil {
+		return nil, 0, fmt.Errorf("failed to pack input data: %w", err)
+	}
+
+	callMsg := ethereum.CallMsg{
+		To:   &contractAddress,
+		Data: data,
+	}
+
+	resultBytes, err := client.CallContract(ctx, callMsg, nil)
+	if err != nil {
+		return nil, 0, fmt.Errorf("contract call failed: %w", err)
+	}
+
+	outputs, err := parsedABI.Unpack("getValue", resultBytes)
+	if err != nil {
+		return nil, 0, fmt.Errorf("failed to unpack result: %w", err)
+	}
+
+	if len(outputs) != 2 {
+		return nil, 0, fmt.Errorf("unexpected number of outputs: got %d, want 2", len(outputs))
+	}
+
+	value, ok := outputs[0].(*big.Int)
+	if !ok {
+		return nil, 0, fmt.Errorf("failed to convert value to big.Int, got type %T: %v", outputs[0], outputs[0])
+	}
+
+	timestamp, ok := outputs[1].(*big.Int)
+	if !ok {
+		return nil, 0, fmt.Errorf("failed to convert timestamp to big.Int, got type %T: %v", outputs[1], outputs[1])
+	}
+
+	return value, timestamp.Uint64(), nil
+}

services/bridge/internal/worker/worker_pool.go

@@ -20,6 +20,7 @@ type WorkerTask struct {
 
 // WorkerPool manages a pool of workers for processing update requests
 type WorkerPool struct {
+	routerID         string            // Router/oracle identifier for logging
 	maxWorkers       int
 	taskQueue        chan *WorkerTask
 	workers          []*Worker
@@ -43,7 +44,7 @@ type Worker struct {
 }
 
 // NewWorkerPool creates a new worker pool
-func NewWorkerPool(maxWorkers int, taskQueueSize int, taskTimeout time.Duration) *WorkerPool {
+func NewWorkerPool(routerID string, maxWorkers int, taskQueueSize int, taskTimeout time.Duration) *WorkerPool {
 	// Use taskQueueSize if provided, otherwise fallback to maxWorkers*2
 	queueSize := taskQueueSize
 	if queueSize <= 0 {
@@ -54,9 +55,10 @@ func NewWorkerPool(maxWorkers int, taskQueueSize int, taskTimeout time.Duration)
 		taskTimeout = 6 * time.Minute
 	}
 
-	logger.Infof("Creating worker pool: maxWorkers=%d, taskQueueSize=%d, taskTimeout=%v", maxWorkers, queueSize, taskTimeout)
+	logger.Infof("Creating worker pool for router %s: maxWorkers=%d, taskQueueSize=%d, taskTimeout=%v", routerID, maxWorkers, queueSize, taskTimeout)
 
 	return &WorkerPool{
+		routerID:     routerID,
 		maxWorkers:   maxWorkers,
 		taskQueue:    make(chan *WorkerTask, queueSize),
 		shutdownChan: make(chan struct{}),
@@ -167,14 +169,14 @@ func (wp *WorkerPool) healthMonitor(ctx context.Context) {
 
 			// Log warning if queue is getting full (>80% capacity)
 			if queueCap > 0 && float64(queueSize)/float64(queueCap) > 0.8 {
-				logger.Warnf("Worker pool queue nearing capacity: %d/%d (%.1f%%), active workers: %d/%d",
-					queueSize, queueCap, float64(queueSize)/float64(queueCap)*100, activeCount, wp.maxWorkers)
+				logger.Warnf("Worker pool queue nearing capacity for %s: %d/%d (%.1f%%), active workers: %d/%d",
+					wp.routerID, queueSize, queueCap, float64(queueSize)/float64(queueCap)*100, activeCount, wp.maxWorkers)
 			}
 
 			// Log warning if all workers are busy and queue has items
 			if int(activeCount) >= wp.maxWorkers && queueSize > 0 {
-				logger.Warnf("All %d workers busy with %d tasks queued - consider increasing worker count",
-					wp.maxWorkers, queueSize)
+				logger.Warnf("All %d workers busy for %s with %d tasks queued - consider increasing worker count",
+					wp.maxWorkers, wp.routerID, queueSize)
 			}
 
 			logger.Debugf("Worker pool health: active=%d/%d, queue=%d/%d",

services/bridge/pkg/router/generic_interface.go

@@ -1,6 +1,9 @@
 package router
 
 import (
+	"context"
+
+	"github.com/diadata.org/Spectra-interoperability/pkg/rpc"
 	"github.com/diadata.org/Spectra-interoperability/services/bridge/config"
 )
 
@@ -17,4 +20,5 @@ type GenericRouterInterface interface {
 	GetStats() GenericRouterStats
 	UpdateDestinationTime(dest config.RouterDestination, symbol string, data ...*config.ExtractedData)
 	GetSymbolFromData(data *config.ExtractedData) string
+	FetchOracleStateFromOnChain(ctx context.Context, clients map[int64]rpc.EthClient) error
 }

services/bridge/pkg/router/generic_router.go

@@ -1,6 +1,7 @@
 package router
 
 import (
+	"context"
 	"crypto/ecdsa"
 	"fmt"
 	"math/big"
@@ -14,6 +15,7 @@ import (
 	"github.com/ethereum/go-ethereum/crypto"
 
 	"github.com/diadata.org/Spectra-interoperability/pkg/logger"
+	"github.com/diadata.org/Spectra-interoperability/pkg/rpc"
 	"github.com/diadata.org/Spectra-interoperability/services/bridge/config"
 	"github.com/diadata.org/Spectra-interoperability/services/bridge/internal/utils"
 )
@@ -778,6 +780,66 @@ func (gr *GenericRouter) GetStats() GenericRouterStats {
 	return gr.stats
 }
 
+// FetchOracleStateFromOnChain fetches the latest state from on-chain contracts
+func (gr *GenericRouter) FetchOracleStateFromOnChain(ctx context.Context, clients map[int64]rpc.EthClient) error {
+	logger.Infof("Fetching oracle state from on-chain for router: %s", gr.config.ID)
+
+	startTime := time.Now()
+	totalDestinations := 0
+	successfulFetches := 0
+
+	// Get symbols from router config
+	symbols := GetSymbolsFromConfig(gr.config)
+	if len(symbols) == 0 {
+		logger.Infof("No symbols configured for router %s, skipping fetch", gr.config.ID)
+		return nil
+	}
+
+	logger.Infof("Found %d symbols in router config for fetch: %v", len(symbols), symbols)
+
+	for _, dest := range gr.config.Destinations {
+		client, exists := clients[dest.ChainID]
+		if !exists {
+			logger.Warnf("No client found for chain %d, skipping fetch", dest.ChainID)
+			continue
+		}
+
+		totalDestinations += len(symbols)
+
+		for _, symbol := range symbols {
+			value, timestamp, err := utils.GetValueFromOracleContract(ctx, client, common.HexToAddress(dest.Contract), symbol)
+			if err != nil {
+				logger.Warnf("Failed to fetch on-chain state for %s on chain %d: %v", symbol, dest.ChainID, err)
+				continue
+			}
+
+			// Update in-memory cache
+			destKey := gr.generateDestinationKey(dest, symbol)
+			state := gr.getOrCreateDestinationState(destKey)
+
+			state.mu.Lock()
+			if timestamp > 0 {
+				state.lastUpdate = time.Unix(int64(timestamp), 0)
+				state.lastTimestamp = timestamp
+			}
+			if value != nil && value.Sign() != 0 {
+				state.lastPrice = value.String()
+			}
+			state.mu.Unlock()
+
+			successfulFetches++
+			logger.Infof("Fetched state for %s on chain %d: timestamp=%d, price=%s",
+				symbol, dest.ChainID, timestamp, value.String())
+		}
+	}
+
+	duration := time.Since(startTime)
+	logger.Infof("Router state fetch completed for %s: %d/%d states fetched in %v",
+		gr.config.ID, successfulFetches, totalDestinations, duration)
+
+	return nil
+}
+
 // ProcessingConfig returns the router's processing configuration
 func (gr *GenericRouter) ProcessingConfig() *config.ProcessingConfig {
 	return &gr.config.Processing