From a0d067cad94fa03a155e24a9b5d7f322ae918eae Mon Sep 17 00:00:00 2001 From: nnn-gif Date: Thu, 28 May 2026 21:38:31 +0530 Subject: [PATCH 1/4] feat: make waitforreceipt async --- services/bridge/internal/api/server.go | 25 + services/bridge/internal/bridge/server.go | 1 + .../internal/bridge/transaction_handler.go | 120 +++-- .../bridge/internal/bridge/write_client.go | 13 + .../internal/contracts/nonce_manager.go | 4 +- .../bridge/internal/transaction/client.go | 21 +- .../bridge/internal/transaction/executor.go | 10 + services/bridge/internal/transaction/queue.go | 120 ++++- .../internal/transaction/queue_manager.go | 77 ++-- .../internal/worker/realistic_failing_test.go | 434 +++++------------- .../bridge/internal/worker/worker_pool.go | 108 ++++- 11 files changed, 533 insertions(+), 400 deletions(-) diff --git a/services/bridge/internal/api/server.go b/services/bridge/internal/api/server.go index e0d8ceb..b316cbf 100644 --- a/services/bridge/internal/api/server.go +++ b/services/bridge/internal/api/server.go @@ -45,6 +45,11 @@ type PoolLister interface { ListPools() []PoolInfo } +// QueueStatter provides read-only access to transaction queue stats +type QueueStatter interface { + GetAllQueueStats() []map[string]interface{} +} + // Server represents the API server type Server struct { config *config.APIConfig @@ -58,6 +63,7 @@ type Server struct { httpServer *http.Server failoverHandler *FailoverHandler poolLister PoolLister + queueStatter QueueStatter } // NewServer creates a new API server @@ -147,6 +153,11 @@ func (s *Server) SetPoolLister(lister PoolLister) { s.poolLister = lister } +// SetQueueStatter sets the queue stats provider for transaction queue inspection +func (s *Server) SetQueueStatter(statter QueueStatter) { + s.queueStatter = statter +} + // setupRoutes configures all API routes func (s *Server) setupRoutes() { s.router.HandleFunc("/health", s.handleHealth).Methods("GET") @@ -196,6 +207,7 @@ func (s *Server) setupRoutes() { // Worker pool endpoints v1.HandleFunc("/pools", s.handleGetPools).Methods("GET") v1.HandleFunc("/pools/{router_id}/tasks", s.handleGetPoolTasks).Methods("GET") + v1.HandleFunc("/queue", s.handleGetQueue).Methods("GET") // Failover endpoints (if available) if s.failoverHandler != nil { @@ -482,6 +494,19 @@ func (s *Server) handleGetPoolTasks(w http.ResponseWriter, r *http.Request) { s.writeError(w, http.StatusNotFound, "Pool not found for router: "+routerID, nil) } +func (s *Server) handleGetQueue(w http.ResponseWriter, r *http.Request) { + if s.queueStatter == nil { + s.writeError(w, http.StatusNotFound, "Queue stats not available", nil) + return + } + + stats := s.queueStatter.GetAllQueueStats() + s.writeJSON(w, map[string]interface{}{ + "queues": stats, + "count": len(stats), + }) +} + // Middleware func (s *Server) loggingMiddleware(next http.Handler) http.Handler { diff --git a/services/bridge/internal/bridge/server.go b/services/bridge/internal/bridge/server.go index e307ea2..950a0e8 100644 --- a/services/bridge/internal/bridge/server.go +++ b/services/bridge/internal/bridge/server.go @@ -34,6 +34,7 @@ func (b *Bridge) startMetricsServer(ctx context.Context) { apiServer := api.NewServer(b.configService, b.db, metricsCollector, b.routerRegistry, priceCache) apiServer.SetPoolLister(b) + apiServer.SetQueueStatter(b.queueManager) go func() { if err := apiServer.Start(ctx); err != nil { diff --git a/services/bridge/internal/bridge/transaction_handler.go b/services/bridge/internal/bridge/transaction_handler.go index f3682c2..095d974 100644 --- a/services/bridge/internal/bridge/transaction_handler.go +++ b/services/bridge/internal/bridge/transaction_handler.go @@ -4,6 +4,7 @@ import ( "context" "fmt" "math/big" + "sync" "time" "github.com/ethereum/go-ethereum/common" @@ -20,6 +21,9 @@ import ( // Constants for gas configuration const ( DefaultGasLimit = uint64(300000) + + // AsyncReceiptTimeout is the max time to wait for a receipt in the background + AsyncReceiptTimeout = 5 * time.Minute ) // TransactionContext encapsulates all data needed for transaction processing @@ -38,6 +42,7 @@ type TransactionHandler struct { routerRegistry *router.GenericRegistry metricsTracker *MetricsTracker onChainMonitor *leader.OnChainMonitor // Optional: for replica monitoring info + wg sync.WaitGroup // tracks in-flight async confirmations } // NewTransactionHandler creates a new transaction handler @@ -50,17 +55,29 @@ func NewTransactionHandler(writeClients map[int64]*WriteClient, registry *router } } -// Process handles the complete transaction lifecycle +// WaitAsyncConfirmations blocks until all in-flight async confirmations finish. +// Called during bridge shutdown to avoid orphan goroutines. +func (h *TransactionHandler) WaitAsyncConfirmations() { + h.wg.Wait() +} + +// Process handles the complete transaction lifecycle. +// After execute() succeeds, confirmation (waitForReceipt + updateState) runs +// asynchronously in a background goroutine. The worker is freed immediately. func (h *TransactionHandler) Process(ctx context.Context, updateReq *bridgetypes.UpdateRequest) error { startTime := time.Now() - logger.Infof("[TX-HANDLER] Starting transaction processing: router=%s, chain=%d, contract=%s", - updateReq.RouterID, updateReq.DestinationChain.ChainID, updateReq.Contract.Address) + deadline, _ := ctx.Deadline() + logger.Infof("[TX-HANDLER] Starting transaction processing: router=%s, chain=%d, contract=%s, deadline_in=%v", + updateReq.RouterID, updateReq.DestinationChain.ChainID, updateReq.Contract.Address, + time.Until(deadline).Round(time.Millisecond)) + t0 := time.Now() txCtx, err := h.buildContext(ctx, updateReq) if err != nil { - logger.Errorf("[TX-HANDLER] Failed to build context: router=%s, error=%v", updateReq.RouterID, err) + logger.Errorf("[TX-HANDLER] Failed to build context: router=%s, error=%v, took=%v", updateReq.RouterID, err, time.Since(t0)) return err } + logger.Infof("[TX-HANDLER] buildContext done: took=%v, deadline_remaining=%v", time.Since(t0), time.Until(deadline).Round(time.Millisecond)) logger.Infof("[TX-HANDLER] Processing update for %s on chain %d (elapsed=%v)", txCtx.Identifier, txCtx.UpdateRequest.DestinationChain.ChainID, time.Since(startTime)) @@ -68,7 +85,9 @@ func (h *TransactionHandler) Process(ctx context.Context, updateReq *bridgetypes return err } + t1 := time.Now() tx, err := h.execute(txCtx) + logger.Infof("[TX-HANDLER] execute done: took=%v, deadline_remaining=%v", time.Since(t1), time.Until(deadline).Round(time.Millisecond)) if err != nil { h.recordFailure(txCtx, "submission", "transaction_failed") return fmt.Errorf("failed to send transaction: %w", err) @@ -83,7 +102,17 @@ func (h *TransactionHandler) Process(ctx context.Context, updateReq *bridgetypes tx.Hash().Hex(), txCtx.Identifier, txCtx.UpdateRequest.DestinationChain.ChainID, txCtx.UpdateRequest.RouterID, txCtx.Symbol, triggeredByMonitoring) - return h.confirm(txCtx, tx) + // Mark update time immediately so the bridge staleness check doesn't re-queue + // this symbol while the receipt is pending. updateState() in asyncConfirm will + // also call this (idempotent) plus OnRouted. + if txCtx.UpdateRequest.Intent != nil && txCtx.UpdateRequest.Contract != nil { + txCtx.DestClient.updateLastUpdate(txCtx.UpdateRequest.Intent.Symbol, txCtx.UpdateRequest.Contract.Address) + } + + // Launch async confirmation — frees the worker immediately + h.asyncConfirm(txCtx, tx) + + return nil } // buildContext creates the transaction context with all necessary data @@ -158,37 +187,66 @@ func (h *TransactionHandler) executeWithMethodConfig(txCtx *TransactionContext) return tx, nil } -// confirm waits for transaction confirmation and updates state -func (h *TransactionHandler) confirm(txCtx *TransactionContext, tx *types.Transaction) error { +// asyncConfirm launches waitForReceipt + updateState in a background goroutine. +// The worker is freed immediately after this returns. +func (h *TransactionHandler) asyncConfirm(txCtx *TransactionContext, tx *types.Transaction) { h.recordSubmission(txCtx, tx.Hash().Hex()) - logger.Infof("[TX-CONFIRM] Waiting for receipt: tx=%s, router=%s, symbol=%s, chain=%d", - tx.Hash().Hex(), txCtx.UpdateRequest.RouterID, txCtx.Symbol, txCtx.UpdateRequest.DestinationChain.ChainID) - - confirmStartTime := time.Now() - receipt, err := h.waitForReceipt(txCtx.Ctx, txCtx.DestClient.client, tx.Hash(), txCtx.UpdateRequest.RouterID) - if err != nil { - h.recordFailure(txCtx, "confirmation", "receipt_timeout") - logger.Errorf("[TX-CONFIRM] Failed to get receipt after %v: tx=%s, router=%s, symbol=%s, chain=%d, error=%v", - time.Since(confirmStartTime), tx.Hash().Hex(), txCtx.UpdateRequest.RouterID, txCtx.Symbol, - txCtx.UpdateRequest.DestinationChain.ChainID, err) - return fmt.Errorf("failed to get transaction receipt: %w", err) - } - - if receipt.Status == 0 { - h.recordFailure(txCtx, "confirmation", "transaction_reverted") - logRevertedTransaction(tx, receipt, txCtx) - return fmt.Errorf("transaction reverted (status: 0): hash=%s, symbol=%s, gas=%d", - tx.Hash().Hex(), txCtx.Symbol, receipt.GasUsed) - } + // Capture values for the goroutine — txCtx may go out of scope + symbol := txCtx.Symbol + routerID := txCtx.UpdateRequest.RouterID + chainID := txCtx.UpdateRequest.DestinationChain.ChainID + txHash := tx.Hash().Hex() + client := txCtx.DestClient.client + destClient := txCtx.DestClient + var contractAddr string + if txCtx.UpdateRequest.Contract != nil { + contractAddr = txCtx.UpdateRequest.Contract.Address + } + + h.wg.Add(1) + go func() { + defer h.wg.Done() + + // Detached context — worker's context expires with taskTimeout (1m), + // but receipt waiting can take longer. Capped at AsyncReceiptTimeout. + confirmCtx, cancel := context.WithTimeout(context.Background(), AsyncReceiptTimeout) + defer cancel() + + logger.Infof("[TX-CONFIRM-ASYNC] Waiting for receipt: tx=%s, router=%s, symbol=%s, chain=%d", + txHash, routerID, symbol, chainID) + + confirmStartTime := time.Now() + receipt, err := h.waitForReceipt(confirmCtx, client, tx.Hash(), routerID) + if err != nil { + h.recordFailure(txCtx, "confirmation", "receipt_timeout") + logger.Errorf("[TX-CONFIRM-ASYNC] Failed to get receipt after %v: tx=%s, router=%s, symbol=%s, chain=%d, error=%v", + time.Since(confirmStartTime), txHash, routerID, symbol, chainID, err) + // Roll back lastUpdate so the bridge re-queues this symbol + if contractAddr != "" { + destClient.clearLastUpdate(symbol, contractAddr) + } + return + } - h.recordConfirmation(txCtx, tx.Hash().Hex(), receipt.GasUsed) - h.updateState(txCtx) + if receipt.Status == 0 { + h.recordFailure(txCtx, "confirmation", "transaction_reverted") + logRevertedTransaction(tx, receipt, txCtx) + logger.Errorf("[TX-CONFIRM-ASYNC] Transaction reverted: hash=%s, symbol=%s, gas=%d", + txHash, symbol, receipt.GasUsed) + // Roll back lastUpdate so the bridge re-queues this symbol + if contractAddr != "" { + destClient.clearLastUpdate(symbol, contractAddr) + } + return + } - logger.Infof("Transaction confirmed: %s, status: %d, gas used: %d, router=%s, symbol=%s, confirm_time=%v", - tx.Hash().Hex(), receipt.Status, receipt.GasUsed, txCtx.UpdateRequest.RouterID, txCtx.Symbol, time.Since(confirmStartTime)) + h.recordConfirmation(txCtx, txHash, receipt.GasUsed) + h.updateState(txCtx) - return nil + logger.Infof("[TX-CONFIRM-ASYNC] Confirmed: %s, status: %d, gas: %d, router=%s, symbol=%s, confirm_time=%v", + txHash, receipt.Status, receipt.GasUsed, routerID, symbol, time.Since(confirmStartTime)) + }() } // recordSubmission records transaction submission metrics diff --git a/services/bridge/internal/bridge/write_client.go b/services/bridge/internal/bridge/write_client.go index c7f7d2e..1a74164 100644 --- a/services/bridge/internal/bridge/write_client.go +++ b/services/bridge/internal/bridge/write_client.go @@ -88,6 +88,16 @@ func (wc *WriteClient) updateLastUpdate(symbol, contract string) { logger.Debugf("Updated lastUpdate for %s on chain %d", key, wc.chainConfig.ChainID) } +// clearLastUpdate removes the lastUpdate entry so the bridge re-queues the symbol. +// Called when an async confirmation fails (receipt timeout or tx reverted). +func (wc *WriteClient) clearLastUpdate(symbol, contract string) { + wc.mu.Lock() + defer wc.mu.Unlock() + key := fmt.Sprintf("%d-%s-%s", wc.chainConfig.ChainID, symbol, contract) + delete(wc.lastUpdate, key) + logger.Debugf("Cleared lastUpdate for %s on chain %d (tx failed, allowing re-queue)", key, wc.chainConfig.ChainID) +} + // getLastUpdate returns the last update time for a specific symbol and contract, or zero time if not found func (wc *WriteClient) getLastUpdate(symbol, contract string) time.Time { wc.mu.RLock() @@ -99,10 +109,13 @@ func (wc *WriteClient) getLastUpdate(symbol, contract string) time.Time { // getGasPrice gets the current gas price for a destination chain func (wc *WriteClient) getGasPrice(ctx context.Context) (*big.Int, error) { + t0 := time.Now() gasPrice, err := wc.client.SuggestGasPrice(ctx) if err != nil { + logger.Errorf("[RPC] SuggestGasPrice failed: took=%v, error=%v", time.Since(t0), err) return nil, err } + logger.Infof("[RPC] SuggestGasPrice: took=%v, gas_price=%s wei", time.Since(t0), gasPrice.String()) multiplier := wc.chainConfig.GasMultiplier if multiplier == 0 { diff --git a/services/bridge/internal/contracts/nonce_manager.go b/services/bridge/internal/contracts/nonce_manager.go index da530e6..f0d4fdf 100644 --- a/services/bridge/internal/contracts/nonce_manager.go +++ b/services/bridge/internal/contracts/nonce_manager.go @@ -137,11 +137,13 @@ func (nm *NonceManager) GetNextNonce(ctx context.Context) (uint64, error) { defer nm.mu.Unlock() // Get confirmed nonce + t0 := time.Now() chainNonce, err := nm.client.NonceAt(ctx, nm.address, nil) if err != nil { - logger.Errorf("NonceManager: Failed to get confirmed nonce, wallet=%s, chain=%d, rpc=%s, error=%v", nm.address.Hex(), nm.chainID, nm.getRPCURL(), err) + logger.Errorf("[RPC] NonceAt failed: took=%v, wallet=%s, chain=%d, error=%v", time.Since(t0), nm.address.Hex(), nm.chainID, err) return 0, fmt.Errorf("failed to get confirmed nonce: %w", err) } + logger.Infof("[RPC] NonceAt done: took=%v, nonce=%d, wallet=%s", time.Since(t0), chainNonce, nm.address.Hex()) if !nm.initialized { logger.Infof("NonceManager: Initializing with confirmed nonce %d for wallet=%s, chain=%d, rpc=%s", chainNonce, nm.address.Hex(), nm.chainID, nm.getRPCURL()) } else if chainNonce > nm.localNonce { diff --git a/services/bridge/internal/transaction/client.go b/services/bridge/internal/transaction/client.go index 5cbb0ad..da84c7d 100644 --- a/services/bridge/internal/transaction/client.go +++ b/services/bridge/internal/transaction/client.go @@ -5,6 +5,7 @@ import ( "fmt" "math/big" "strings" + "time" "github.com/ethereum/go-ethereum/accounts/abi" "github.com/ethereum/go-ethereum/common" @@ -57,7 +58,25 @@ func (c *Client) CallMethod(ctx context.Context, contractAddr, methodName, abiJS return c.executor.Execute(execCtx, req) } - return queue.Submit(ctx, executorFunc) + // Extract metadata for queue visibility + meta := SubmitMeta{ + ChainID: c.chainID, + } + if updateReq != nil { + meta.RouterID = updateReq.RouterID + meta.Contract = contractAddr + if updateReq.Intent != nil { + meta.Symbol = updateReq.Intent.Symbol + if updateReq.Intent.Timestamp != nil { + meta.OnchainTS = updateReq.Intent.Timestamp.Int64() + } + } + } + + t0 := time.Now() + tx, err := queue.Submit(ctx, executorFunc, meta) + logger.Infof("[QUEUE-SUBMIT] CallMethod %s on %s: took=%v", methodName, contractAddr, time.Since(t0)) + return tx, err } func (c *Client) BuildParams(methodConfig *config.DestinationMethodConfig, updateReq *bridgetypes.UpdateRequest) ([]interface{}, error) { diff --git a/services/bridge/internal/transaction/executor.go b/services/bridge/internal/transaction/executor.go index fb6f30e..9a9790c 100644 --- a/services/bridge/internal/transaction/executor.go +++ b/services/bridge/internal/transaction/executor.go @@ -6,6 +6,7 @@ import ( "fmt" "math/big" "strings" + "time" "github.com/ethereum/go-ethereum" "github.com/ethereum/go-ethereum/accounts/abi" @@ -98,7 +99,9 @@ func (e *Executor) Execute(ctx context.Context, req *Request) (*types.Transactio Data: callData, } + t0 := time.Now() if _, err := e.ethClient.CallContract(ctx, callMsg, nil); err != nil { + logger.Errorf("[RPC] CallContract (simulate) failed: took=%v, error=%v", time.Since(t0), err) // Log raw error for debugging - MUST be Error level to see in production logger.Errorf("Raw simulation error for chain %d: %v (Type: %T)", e.chainID, err, err) @@ -122,12 +125,16 @@ func (e *Executor) Execute(ctx context.Context, req *Request) (*types.Transactio routerID = req.UpdateRequest.RouterID } logger.Infof("Transaction simulation successful, proceeding to send transaction, router=%s", routerID) + logger.Infof("[RPC] CallContract (simulate) done: took=%v", time.Since(t0)) // CRITICAL: Allocate nonce immediately before sending to minimize staleness window // This happens AFTER simulation to reduce the time between allocation and sending + t1 := time.Now() if err := e.receiverClient.UpdateAuth(ctx, req.GasPrice); err != nil { + logger.Errorf("[RPC] UpdateAuth (nonce) failed: took=%v, error=%v", time.Since(t1), err) return nil, fmt.Errorf("failed to update auth: %w", err) } + logger.Infof("[RPC] UpdateAuth (nonce) done: took=%v", time.Since(t1)) // Refresh auth to get the newly allocated nonce auth = e.receiverClient.GetAuth() @@ -139,8 +146,10 @@ func (e *Executor) Execute(ctx context.Context, req *Request) (*types.Transactio usedNonce := auth.Nonce.Uint64() // Send transaction immediately (within ~1ms of nonce allocation) + t2 := time.Now() tx, err := bind.NewBoundContract(contractAddress, parsedABI, e.ethClient, e.ethClient, e.ethClient).Transact(auth, req.MethodName, req.Params...) if err != nil { + logger.Errorf("[RPC] Transact (send) failed: took=%v, nonce=%d, error=%v", time.Since(t2), usedNonce, err) logger.Errorf("Transaction failed: %v", err) // CRITICAL: Notify NonceManager about the error so it can resync if needed e.receiverClient.HandleTransactionError(ctx, err, usedNonce) @@ -149,6 +158,7 @@ func (e *Executor) Execute(ctx context.Context, req *Request) (*types.Transactio // Mark nonce as successfully sent to mempool e.receiverClient.MarkNonceSent(usedNonce, tx.Hash().Hex()) + logger.Infof("[RPC] Transact (send) done: took=%v, tx_hash=%s, nonce=%d", time.Since(t2), tx.Hash().Hex(), usedNonce) symbol := "unknown" if req.UpdateRequest.Intent != nil { diff --git a/services/bridge/internal/transaction/queue.go b/services/bridge/internal/transaction/queue.go index b7e6bf2..021a62e 100644 --- a/services/bridge/internal/transaction/queue.go +++ b/services/bridge/internal/transaction/queue.go @@ -22,6 +22,17 @@ type Queue struct { mu sync.Mutex running bool metrics *metrics.Collector + + // Stats tracking + totalSubmitted int64 + totalCompleted int64 + totalFailed int64 + lastSubmitTime time.Time + lastCompleteAt time.Time + avgExecTime time.Duration + + // Pending item tracking for API visibility + pendingItems []SubmitMeta } type queuedRequest struct { @@ -29,6 +40,17 @@ type queuedRequest struct { executor ExecutorFunc resultCh chan *Result enqueueTime time.Time + meta SubmitMeta +} + +// SubmitMeta carries caller-provided metadata about what the tx is for +type SubmitMeta struct { + Symbol string `json:"symbol"` + Contract string `json:"contract"` + ChainID int64 `json:"chain_id"` + RouterID string `json:"router_id"` + Enqueued time.Time `json:"enqueued"` + OnchainTS int64 `json:"onchain_ts"` } func NewQueue(queueKey string, queueSize int, metrics *metrics.Collector) *Queue { @@ -74,7 +96,7 @@ func (q *Queue) Stop() { logger.Infof("Transaction queue stopped: %s", q.queueKey) } -func (q *Queue) Submit(ctx context.Context, executor ExecutorFunc) (*types.Transaction, error) { +func (q *Queue) Submit(ctx context.Context, executor ExecutorFunc, meta SubmitMeta) (*types.Transaction, error) { q.mu.Lock() if !q.running { q.mu.Unlock() @@ -82,29 +104,47 @@ func (q *Queue) Submit(ctx context.Context, executor ExecutorFunc) (*types.Trans } q.mu.Unlock() + deadline, _ := ctx.Deadline() + meta.Enqueued = time.Now() resultCh := make(chan *Result, 1) req := &queuedRequest{ ctx: ctx, executor: executor, resultCh: resultCh, - enqueueTime: time.Now(), + enqueueTime: meta.Enqueued, + meta: meta, } + t0 := time.Now() select { case q.queue <- req: + q.mu.Lock() + q.totalSubmitted++ + q.lastSubmitTime = time.Now() + q.pendingItems = append(q.pendingItems, meta) + q.mu.Unlock() if q.metrics != nil { q.metrics.SetQueueLength(q.queueKey, len(q.queue)) } + logger.Infof("[QUEUE] Submitted to serial queue: symbol=%s, queue_depth=%d, enqueue_took=%v, deadline_remaining=%v", + meta.Symbol, len(q.queue), time.Since(t0), time.Until(deadline).Round(time.Millisecond)) case <-ctx.Done(): + logger.Warnf("[QUEUE] Submit cancelled before enqueue: symbol=%s, took=%v, err=%v", meta.Symbol, time.Since(t0), ctx.Err()) return nil, ctx.Err() case <-time.After(30 * time.Second): + logger.Warnf("[QUEUE] Submit timed out enqueueing: symbol=%s, queue_depth=%d", meta.Symbol, len(q.queue)) return nil, fmt.Errorf("timeout submitting transaction to queue") } + t1 := time.Now() select { case result := <-resultCh: + logger.Infof("[QUEUE] Got result from serial queue: symbol=%s, wait_took=%v, deadline_remaining=%v", + meta.Symbol, time.Since(t1), time.Until(deadline).Round(time.Millisecond)) return result.Tx, result.Err case <-ctx.Done(): + logger.Warnf("[QUEUE] Submit cancelled while waiting for result: symbol=%s, waited=%v, deadline_remaining=%v, err=%v", + meta.Symbol, time.Since(t1), time.Until(deadline).Round(time.Millisecond), ctx.Err()) return nil, ctx.Err() } } @@ -127,9 +167,16 @@ func (q *Queue) processQueue() { } func (q *Queue) processRequest(req *queuedRequest) { + // Dequeue from pending tracking (FIFO) + q.mu.Lock() + if len(q.pendingItems) > 0 { + q.pendingItems = q.pendingItems[1:] + } + q.mu.Unlock() + select { case <-req.ctx.Done(): - logger.Warnf("Transaction request cancelled before execution: %s", q.queueKey) + logger.Warnf("Transaction request cancelled before execution: %s symbol=%s", q.queueKey, req.meta.Symbol) req.resultCh <- &Result{ Tx: nil, Err: req.ctx.Err(), @@ -146,8 +193,10 @@ func (q *Queue) processRequest(req *queuedRequest) { } startTime := time.Now() + logger.Infof("[QUEUE-EXEC] Starting execution: symbol=%s, waited_in_queue=%v", req.meta.Symbol, time.Since(req.enqueueTime)) tx, err := req.executor(req.ctx) duration := time.Since(startTime) + logger.Infof("[QUEUE-EXEC] Execution done: symbol=%s, took=%v, err=%v", req.meta.Symbol, duration, err) if q.metrics != nil { q.metrics.ObserveQueueProcessingDuration(q.queueKey, duration.Seconds()) @@ -156,12 +205,28 @@ func (q *Queue) processRequest(req *queuedRequest) { if err != nil { errorDetail := extractErrorDetail(err) logger.Errorf("Transaction execution failed for queue [%s] after %v: %s", q.queueKey, duration, errorDetail) + q.mu.Lock() + q.totalFailed++ + q.mu.Unlock() } else if tx != nil { logger.Infof("Transaction executed successfully for queue [%s] in %v: %s", q.queueKey, duration, tx.Hash().Hex()) } else { logger.Infof("Transaction executed successfully for queue [%s] in %v (no tx returned)", q.queueKey, duration) } + q.mu.Lock() + if err == nil { + q.totalCompleted++ + // Running average: avg = avg * 0.9 + current * 0.1 + if q.avgExecTime == 0 { + q.avgExecTime = duration + } else { + q.avgExecTime = (q.avgExecTime*9 + duration*1) / 10 + } + } + q.lastCompleteAt = time.Now() + q.mu.Unlock() + select { case req.resultCh <- &Result{Tx: tx, Err: err}: case <-time.After(5 * time.Second): @@ -173,6 +238,55 @@ func (q *Queue) GetQueueLength() int { return len(q.queue) } +// QueueStats holds snapshot statistics for a queue +type QueueStats struct { + Key string `json:"key"` + Pending int `json:"pending"` + Capacity int `json:"capacity"` + Running bool `json:"running"` + TotalSubmitted int64 `json:"total_submitted"` + TotalCompleted int64 `json:"total_completed"` + TotalFailed int64 `json:"total_failed"` + AvgExecTime string `json:"avg_exec_time"` + LastSubmitTime time.Time `json:"last_submit_time"` + LastCompleteAt time.Time `json:"last_complete_at"` + ThroughputPerMin float64 `json:"throughput_per_min"` + PendingItems []SubmitMeta `json:"pending_items"` +} + +func (q *Queue) GetStats() QueueStats { + q.mu.Lock() + defer q.mu.Unlock() + + // Copy pending items + pending := make([]SubmitMeta, len(q.pendingItems)) + copy(pending, q.pendingItems) + + stats := QueueStats{ + Key: q.queueKey, + Pending: len(q.queue), + Capacity: cap(q.queue), + Running: q.running, + TotalSubmitted: q.totalSubmitted, + TotalCompleted: q.totalCompleted, + TotalFailed: q.totalFailed, + AvgExecTime: q.avgExecTime.Round(time.Millisecond).String(), + LastSubmitTime: q.lastSubmitTime, + LastCompleteAt: q.lastCompleteAt, + PendingItems: pending, + } + + // Calculate throughput from uptime + if !q.lastCompleteAt.IsZero() && q.totalCompleted > 0 { + uptime := time.Since(q.lastSubmitTime).Minutes() + if uptime > 0 { + stats.ThroughputPerMin = float64(q.totalCompleted) / uptime + } + } + + return stats +} + // extractErrorDetail extracts detailed error information including exact revert reasons func extractErrorDetail(err error) string { if err == nil { diff --git a/services/bridge/internal/transaction/queue_manager.go b/services/bridge/internal/transaction/queue_manager.go index 076db4e..b17c1a6 100644 --- a/services/bridge/internal/transaction/queue_manager.go +++ b/services/bridge/internal/transaction/queue_manager.go @@ -17,9 +17,6 @@ type QueueManager struct { } func NewQueueManager(queueSize int, metrics *metrics.Collector) *QueueManager { - if queueSize <= 0 { - queueSize = 100 - } return &QueueManager{ queues: make(map[string]*Queue), queueSize: queueSize, @@ -30,73 +27,91 @@ func NewQueueManager(queueSize int, metrics *metrics.Collector) *QueueManager { func (qm *QueueManager) Start() { qm.mu.Lock() defer qm.mu.Unlock() - if qm.running { return } - qm.running = true - logger.Info("Transaction queue manager started") + logger.Infof("Transaction queue manager started with queue size %d", qm.queueSize) } func (qm *QueueManager) Stop() { qm.mu.Lock() defer qm.mu.Unlock() - if !qm.running { return } - - logger.Info("Stopping transaction queue manager") + qm.running = false for key, queue := range qm.queues { queue.Stop() logger.Infof("Stopped queue: %s", key) } - qm.queues = make(map[string]*Queue) - qm.running = false - - logger.Info("Transaction queue manager stopped") + logger.Infof("Transaction queue manager stopped") } -func (qm *QueueManager) GetOrCreateQueue(walletAddress string, chainID int64) (*Queue, error) { - queueKey := fmt.Sprintf("%s-%d", walletAddress, chainID) - - qm.mu.RLock() - if queue, exists := qm.queues[queueKey]; exists { - qm.mu.RUnlock() - return queue, nil - } - qm.mu.RUnlock() - +func (qm *QueueManager) GetOrCreateQueue(walletAddr string, chainID int64) (*Queue, error) { qm.mu.Lock() defer qm.mu.Unlock() + queueKey := fmt.Sprintf("%s-%d", walletAddr, chainID) + if queue, exists := qm.queues[queueKey]; exists { return queue, nil } - if !qm.running { - return nil, fmt.Errorf("queue manager is not running") - } - queue := NewQueue(queueKey, qm.queueSize, qm.metrics) queue.Start() qm.queues[queueKey] = queue logger.Infof("Created new transaction queue: %s", queueKey) - return queue, nil } -func (qm *QueueManager) GetQueueStats() map[string]int { +func (qm *QueueManager) GetQueueCount() int { + qm.mu.RLock() + defer qm.mu.RUnlock() + return len(qm.queues) +} + +func (qm *QueueManager) GetQueueLengths() map[string]int { qm.mu.RLock() defer qm.mu.RUnlock() - stats := make(map[string]int) + lengths := make(map[string]int, len(qm.queues)) for key, queue := range qm.queues { - stats[key] = queue.GetQueueLength() + lengths[key] = queue.GetQueueLength() + } + return lengths +} + +// GetQueueStats returns queue statistics (alias for GetQueueLengths) +func (qm *QueueManager) GetQueueStats() map[string]int { + return qm.GetQueueLengths() +} + +// GetAllQueueStats returns detailed stats for all queues as maps for API consumption +func (qm *QueueManager) GetAllQueueStats() []map[string]interface{} { + qm.mu.RLock() + defer qm.mu.RUnlock() + + stats := make([]map[string]interface{}, 0, len(qm.queues)) + for _, queue := range qm.queues { + s := queue.GetStats() + stats = append(stats, map[string]interface{}{ + "key": s.Key, + "pending": s.Pending, + "capacity": s.Capacity, + "running": s.Running, + "total_submitted": s.TotalSubmitted, + "total_completed": s.TotalCompleted, + "total_failed": s.TotalFailed, + "avg_exec_time": s.AvgExecTime, + "last_submit_time": s.LastSubmitTime, + "last_complete_at": s.LastCompleteAt, + "throughput_per_min": s.ThroughputPerMin, + "pending_items": s.PendingItems, + }) } return stats } diff --git a/services/bridge/internal/worker/realistic_failing_test.go b/services/bridge/internal/worker/realistic_failing_test.go index ff2a34a..5b535f1 100644 --- a/services/bridge/internal/worker/realistic_failing_test.go +++ b/services/bridge/internal/worker/realistic_failing_test.go @@ -4,7 +4,6 @@ import ( "context" "fmt" "math/big" - "sync" "sync/atomic" "testing" "time" @@ -16,68 +15,45 @@ import ( ) // ============================================================================ -// REALISTIC FAILING TESTS - Simulating Production Issues +// Worker pool tests for async-confirm architecture +// Handlers return quickly (~1s simulate+send). Receipt confirmation is async. // ============================================================================ -// HangingRPCHandler simulates a real RPC handler that doesn't respect context -// This is what happens when go-ethereum RPC client blocks on network I/O -type HangingRPCHandler struct { - callCount atomic.Int64 - shouldHang bool - hangDuration time.Duration - hangStarted chan struct{} - hangCompleted chan struct{} +// CountingHandler tracks calls and can inject errors or delays +type CountingHandler struct { + callCount atomic.Int64 + errorCount atomic.Int64 + shouldError bool + delay time.Duration } -func (h *HangingRPCHandler) Execute(ctx context.Context, task *WorkerTask) error { +func (h *CountingHandler) Execute(ctx context.Context, task *WorkerTask) error { h.callCount.Add(1) - - if h.shouldHang { - if h.hangStarted != nil { - h.hangStarted <- struct{}{} - } - - // Simulate RPC call that blocks without checking context - // This is what happens when RPC client doesn't respect context - time.Sleep(h.hangDuration) - - if h.hangCompleted != nil { - h.hangCompleted <- struct{}{} + if h.delay > 0 { + select { + case <-time.After(h.delay): + case <-ctx.Done(): + return ctx.Err() } } - + if h.shouldError { + h.errorCount.Add(1) + return fmt.Errorf("simulated RPC error") + } return nil } -// TestProcessTask_RealRPCTimeoutScenario -// -// REAL PRODUCTION ISSUE: When RPC client (go-ethereum) makes a call that hangs due to network issues, -// the handler blocks forever because: -// 1. RPC client doesn't check context during I/O operations -// 2. Task timeout expires but handler is still blocked on network I/O -// 3. processTask doesn't forcefully interrupt the handler -// -// This test creates a realistic scenario where handler simulates a slow RPC call. -func TestProcessTask_RealRPCTimeoutScenario(t *testing.T) { - // Create handler that simulates slow RPC (10 seconds) - handler := &HangingRPCHandler{ - shouldHang: true, - hangDuration: 10 * time.Second, // Long-running RPC - hangStarted: make(chan struct{}), - hangCompleted: make(chan struct{}), - } - - // Task timeout is only 2 seconds - shortTimeout := 2 * time.Second - ctx, cancel := context.WithCancel(context.Background()) - defer cancel() +func makeTestTask(id string, handler func(context.Context, *WorkerTask) error) *WorkerTask { + return makeTestTaskWithSymbol(id, "BTC/USD", handler) +} - task := &WorkerTask{ - ID: "real-rpc-timeout", +func makeTestTaskWithSymbol(id, symbol string, handler func(context.Context, *WorkerTask) error) *WorkerTask { + return &WorkerTask{ + ID: id, Request: &bridgetypes.UpdateRequest{ RouterID: "test-router", Intent: &bridgetypes.OracleIntent{ - Symbol: "BTC/USD", + Symbol: symbol, Price: big.NewInt(50000000000000000), Timestamp: big.NewInt(time.Now().Unix()), Expiry: big.NewInt(time.Now().Add(time.Hour).Unix()), @@ -92,325 +68,159 @@ func TestProcessTask_RealRPCTimeoutScenario(t *testing.T) { GasLimit: 300000, }, }, - Handler: handler.Execute, + Handler: handler, } +} + +// TestProcessTask_Success tests that a fast handler completes immediately +func TestProcessTask_Success(t *testing.T) { + handler := &CountingHandler{} + task := makeTestTask("fast-success", handler.Execute) - pool := createTestWorkerPool("test-router", 1, shortTimeout) + pool := createTestWorkerPool("test-router", 1, 30*time.Second) worker := &Worker{ - id: 0, - taskQueue: make(chan *WorkerTask, 1), - quit: make(chan struct{}), - wg: &sync.WaitGroup{}, - metricsCollector: nil, - pool: pool, + id: 0, + taskQueue: make(chan *WorkerTask, 1), + quit: make(chan struct{}), + pool: pool, } - startTime := time.Now() done := make(chan struct{}) - go func() { - worker.processTask(ctx, task) + worker.processTask(context.Background(), task) close(done) }() - // Wait for handler to start hanging - select { - case <-handler.hangStarted: - t.Logf("Handler started hanging at %v", time.Since(startTime)) - case <-time.After(500 * time.Millisecond): - t.Fatal("Handler did not start hanging") - } - - // Wait for task completion (should timeout at 2 seconds) select { case <-done: - duration := time.Since(startTime) - t.Logf("Task completed after %v", duration) - - // BUG: If duration is close to handler hang duration (10s) instead of timeout (2s), - // then taskCtx timeout didn't work - if duration > 5*time.Second { - t.Errorf("BUG CONFIRMED: Task took %v (expected ~%v for timeout) - Handler continued despite task timeout", - duration, shortTimeout) - } - - case <-time.After(shortTimeout + 12*time.Second): - // Check if handler actually completed - select { - case <-handler.hangCompleted: - t.Errorf("BUG CONFIRMED: Handler completed its full %s hang despite %s task timeout", - handler.hangDuration, shortTimeout) - case <-time.After(100 * time.Millisecond): - t.Errorf("BUG CONFIRMED: Task never completed - handler is still hanging after timeout") - } + assert.Equal(t, int64(1), handler.callCount.Load(), "handler should be called once") + case <-time.After(5 * time.Second): + t.Fatal("task didn't complete within 5s") } - // Verify active workers cleaned up assert.Eventually(t, func() bool { - return atomic.LoadInt32(&pool.activeWorkers) == int32(0) - }, 1*time.Second, 50*time.Millisecond, - "Active workers should be 0 after task completion") + return atomic.LoadInt32(&pool.activeWorkers) == 0 + }, 1*time.Second, 50*time.Millisecond, "active workers should be 0") } -// TestProcessTask_NetworkTimeoutScenario simulates actual network timeout -func TestProcessTask_NetworkTimeoutScenario(t *testing.T) { - handler := &HangingRPCHandler{ - shouldHang: true, - hangDuration: 30 * time.Second, - hangStarted: make(chan struct{}), - hangCompleted: make(chan struct{}), +// TestProcessTask_RetriesOnError tests retry logic when handler fails +func TestProcessTask_RetriesOnError(t *testing.T) { + handler := &CountingHandler{shouldError: true} + task := makeTestTask("retry-test", handler.Execute) + + pool := createTestWorkerPool("test-router", 1, 30*time.Second) + worker := &Worker{ + id: 0, + taskQueue: make(chan *WorkerTask, 1), + quit: make(chan struct{}), + pool: pool, } - // Very short timeout to simulate network issue - networkTimeout := 500 * time.Millisecond - ctx, cancel := context.WithCancel(context.Background()) - defer cancel() + done := make(chan struct{}) + go func() { + worker.processTask(context.Background(), task) + close(done) + }() - task := &WorkerTask{ - ID: "network-timeout", - Request: &bridgetypes.UpdateRequest{ - RouterID: "test-router", - Intent: &bridgetypes.OracleIntent{ - Symbol: "ETH/USD", - Price: big.NewInt(3000000000000000), - Timestamp: big.NewInt(time.Now().Unix()), - Expiry: big.NewInt(time.Now().Add(time.Hour).Unix()), - Nonce: big.NewInt(54321), - Signer: common.HexToAddress("0xabcdef1234567890abcdef1234567890abcdef12"), - }, - DestinationChain: &config.DestinationConfig{ - ChainID: 1, - }, - DestinationMethodConfig: &config.DestinationMethodConfig{ - Name: "updatePrice", - GasLimit: 300000, - }, - }, - Handler: handler.Execute, + select { + case <-done: + // 3 attempts (loop runs for retry=0,1,2) + assert.Equal(t, int64(3), handler.callCount.Load(), "handler should be called 3 times (maxRetries)") + case <-time.After(30 * time.Second): + t.Fatal("task didn't complete within 30s") } +} + +// TestProcessTask_ContextCancelStopsRetries tests that context cancellation prevents retries +func TestProcessTask_ContextCancelStopsRetries(t *testing.T) { + handler := &CountingHandler{shouldError: true} + task := makeTestTask("cancel-test", handler.Execute) - pool := createTestWorkerPool("test-router", 1, networkTimeout) + // Short timeout to trigger context expiry + pool := createTestWorkerPool("test-router", 1, 1*time.Second) worker := &Worker{ - id: 0, - taskQueue: make(chan *WorkerTask, 1), - quit: make(chan struct{}), - wg: &sync.WaitGroup{}, - metricsCollector: nil, - pool: pool, + id: 0, + taskQueue: make(chan *WorkerTask, 1), + quit: make(chan struct{}), + pool: pool, } startTime := time.Now() done := make(chan struct{}) - go func() { - worker.processTask(ctx, task) + worker.processTask(context.Background(), task) close(done) }() - // Wait for hang - <-handler.hangStarted - - // Task should timeout quickly select { case <-done: duration := time.Since(startTime) - t.Logf("Task completed in %v", duration) - - if duration > 2*time.Second { - t.Errorf("BUG: Task took too long (%v vs expected %v timeout)", duration, networkTimeout) - } - - case <-time.After(networkTimeout + 3*time.Second): - t.Errorf("BUG: Task didn't timeout within expected time") + t.Logf("Task completed in %v after %d handler calls", duration, handler.callCount.Load()) + // Should have stopped after context expired, not all 3 retries + assert.Less(t, handler.callCount.Load(), int64(4), "should not have retried 3 times after context expiry") + case <-time.After(15 * time.Second): + t.Fatal("task didn't complete within 15s") } } -// TestProcessTask_MultipleWorkersTimeout tests multiple workers with hanging handlers -func TestProcessTask_MultipleWorkersTimeout(t *testing.T) { - numWorkers := 3 - taskTimeout := 1 * time.Second - handlerHangDuration := 10 * time.Second - - ctx, cancel := context.WithCancel(context.Background()) - defer cancel() - - pool := createTestWorkerPool("test-router", numWorkers, taskTimeout) - pool.Start(ctx) - defer pool.Stop(ctx) - - handlers := make([]*HangingRPCHandler, numWorkers) - - // Submit tasks that will hang - for i := 0; i < numWorkers; i++ { - handler := &HangingRPCHandler{ - shouldHang: true, - hangDuration: handlerHangDuration, - hangStarted: make(chan struct{}), - hangCompleted: make(chan struct{}), - } - handlers[i] = handler - - task := &WorkerTask{ - ID: fmt.Sprintf("multi-worker-task-%d", i), - Request: &bridgetypes.UpdateRequest{ - RouterID: "test-router", - Intent: &bridgetypes.OracleIntent{ - Symbol: "BTC/USD", - Price: big.NewInt(50000000000000000), - Timestamp: big.NewInt(time.Now().Unix()), - Expiry: big.NewInt(time.Now().Add(time.Hour).Unix()), - Nonce: big.NewInt(int64(i)), - Signer: common.HexToAddress("0x1234567890123456789012345678901234567890"), - }, - DestinationChain: &config.DestinationConfig{ - ChainID: 1, - }, - DestinationMethodConfig: &config.DestinationMethodConfig{ - Name: "updatePrice", - GasLimit: 300000, - }, - }, - Handler: handler.Execute, - } - - pool.Submit(task) - } +// TestProcessTask_ContextRespectedDuringDelay tests handler respects context during slow ops +func TestProcessTask_ContextRespectedDuringDelay(t *testing.T) { + handler := &CountingHandler{delay: 10 * time.Second} + task := makeTestTask("delay-cancel", handler.Execute) - // Wait for all handlers to start hanging - for _, handler := range handlers { - select { - case <-handler.hangStarted: - case <-time.After(1 * time.Second): - t.Errorf("Handler didn't start hanging") - } + // 2s timeout, handler wants to sleep 10s + pool := createTestWorkerPool("test-router", 1, 2*time.Second) + worker := &Worker{ + id: 0, + taskQueue: make(chan *WorkerTask, 1), + quit: make(chan struct{}), + pool: pool, } - t.Logf("All %d handlers have started hanging", numWorkers) - - // All tasks should timeout within taskTimeout + margin - timeoutThreshold := taskTimeout + 2*time.Second startTime := time.Now() + done := make(chan struct{}) + go func() { + worker.processTask(context.Background(), task) + close(done) + }() - // Monitor pool until all tasks complete - for i := 0; i < 20; i++ { - time.Sleep(200 * time.Millisecond) - - stats := pool.GetStats() - t.Logf("Progress: Queue=%d, Active=%d, Expected=%d", stats.PendingTasks, stats.ActiveTasks, numWorkers) - - if stats.PendingTasks == 0 && stats.ActiveTasks == 0 { - duration := time.Since(startTime) - t.Logf("All tasks completed in %v", duration) - - if duration > timeoutThreshold { - t.Errorf("BUG: Tasks took %v to complete (expected < %v) - workers didn't respect timeout", - duration, timeoutThreshold) - } - - // Check if handlers actually completed their hang - completedHangs := 0 - for _, handler := range handlers { - select { - case <-handler.hangCompleted: - completedHangs++ - default: - } - } - - if completedHangs > 0 { - t.Errorf("BUG: %d/%d handlers completed their %s hang despite %s task timeout", - completedHangs, numWorkers, handlerHangDuration, taskTimeout) - } - - return - } + select { + case <-done: + duration := time.Since(startTime) + t.Logf("Task completed in %v", duration) + // Handler respects context (select on ctx.Done), so should complete in ~2s + assert.Less(t, duration, 5*time.Second, "should complete near timeout, not wait full 10s delay") + case <-time.After(10 * time.Second): + t.Fatal("task didn't complete - handler ignored context") } - - t.Errorf("BUG: Tasks didn't complete within timeout threshold of %v", timeoutThreshold) } -// TestProcessTask_ConcurrentSubmissionWithTimeout tests realistic load scenario -func TestProcessTask_ConcurrentSubmissionWithTimeout(t *testing.T) { - numWorkers := 2 - numTasks := 5 - taskTimeout := 1 * time.Second +// TestProcessTask_MultipleWorkers tests pool with multiple workers processing tasks +func TestProcessTask_MultipleWorkers(t *testing.T) { + numWorkers := 3 + numTasks := 6 ctx, cancel := context.WithCancel(context.Background()) defer cancel() - pool := createTestWorkerPool("test-router", numWorkers, taskTimeout) + pool := createTestWorkerPool("test-router", numWorkers, 30*time.Second) pool.Start(ctx) defer pool.Stop(ctx) - handlers := make([]*HangingRPCHandler, numTasks) + // Use a handler that completes quickly + handler := &CountingHandler{delay: 100 * time.Millisecond} - // Submit tasks concurrently + symbols := []string{"BTC/USD", "ETH/USD", "SOL/USD", "DOGE/USD", "ARB/USD", "MNT/USD"} for i := 0; i < numTasks; i++ { - handler := &HangingRPCHandler{ - shouldHang: true, - hangDuration: 15 * time.Second, - hangStarted: make(chan struct{}), - } - handlers[i] = handler - - task := &WorkerTask{ - ID: fmt.Sprintf("concurrent-task-%d", i), - Request: &bridgetypes.UpdateRequest{ - RouterID: "test-router", - Intent: &bridgetypes.OracleIntent{ - Symbol: "BTC/USD", - Price: big.NewInt(50000000000000000), - Timestamp: big.NewInt(time.Now().Unix()), - Expiry: big.NewInt(time.Now().Add(time.Hour).Unix()), - Nonce: big.NewInt(int64(i)), - Signer: common.HexToAddress("0x1234567890123456789012345678901234567890"), - }, - DestinationChain: &config.DestinationConfig{ - ChainID: 1, - }, - DestinationMethodConfig: &config.DestinationMethodConfig{ - Name: "updatePrice", - GasLimit: 300000, - }, - }, - Handler: handler.Execute, - } - + task := makeTestTaskWithSymbol(fmt.Sprintf("multi-task-%d", i), symbols[i], handler.Execute) pool.Submit(task) - time.Sleep(10 * time.Millisecond) // Small delay between submissions } - t.Logf("Submitted %d tasks to %d workers", numTasks, numWorkers) - - // All tasks should complete within reasonable time - maxWaitTime := time.Duration(numTasks)*taskTimeout + 5*time.Second - startTime := time.Now() - - // Monitor completion - completed := false - for !completed { - time.Sleep(200 * time.Millisecond) - + // All tasks should complete quickly: 6 tasks, 3 workers, 100ms each = ~200ms + assert.Eventually(t, func() bool { stats := pool.GetStats() - if stats.PendingTasks == 0 && stats.ActiveTasks == 0 { - completed = true - duration := time.Since(startTime) - t.Logf("All %d tasks completed in %v", numTasks, duration) + return stats.PendingTasks == 0 && stats.ActiveTasks == 0 + }, 5*time.Second, 100*time.Millisecond, "all tasks should complete") - // Calculate expected completion time: - // - numWorkers tasks start immediately - // - Remaining tasks wait in queue - expectedMaxTime := time.Duration((numTasks+numWorkers-1)/numWorkers)*taskTimeout + 2*time.Second - - if duration > expectedMaxTime*2 { // Allow 2x margin - t.Errorf("BUG: Tasks took %v (expected < %v) - queue or timeout issue", - duration, expectedMaxTime) - } - } - - if time.Since(startTime) > maxWaitTime { - t.Errorf("BUG: Tasks didn't complete within %v timeout", maxWaitTime) - break - } - } -} \ No newline at end of file + assert.Equal(t, int64(numTasks), handler.callCount.Load(), "all tasks should be processed") +} diff --git a/services/bridge/internal/worker/worker_pool.go b/services/bridge/internal/worker/worker_pool.go index e411f84..2b76927 100644 --- a/services/bridge/internal/worker/worker_pool.go +++ b/services/bridge/internal/worker/worker_pool.go @@ -2,6 +2,7 @@ package worker import ( "context" + "fmt" "sync" "sync/atomic" "time" @@ -43,6 +44,8 @@ type WorkerPool struct { taskTimeout time.Duration pendingMu sync.RWMutex pendingTasks []*WorkerTask + inflightMu sync.Mutex + inflightSymbols map[string]string // "SYMBOL/CHAINID" -> taskID, tracks active+queued tasks } // Worker represents a single worker in the pool @@ -70,11 +73,12 @@ func NewWorkerPool(routerID string, maxWorkers int, taskQueueSize int, taskTimeo 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{}), - taskTimeout: taskTimeout, + routerID: routerID, + maxWorkers: maxWorkers, + taskQueue: make(chan *WorkerTask, queueSize), + shutdownChan: make(chan struct{}), + taskTimeout: taskTimeout, + inflightSymbols: make(map[string]string), } } @@ -191,7 +195,7 @@ func (wp *WorkerPool) healthMonitor(ctx context.Context) { wp.maxWorkers, wp.routerID, queueSize, activeCount) } - logger.Debugf("[router=%s] Worker pool health: active=%d/%d, queue=%d/%d", + logger.Infof("[router=%s] Worker pool health: active=%d/%d, queue=%d/%d", wp.routerID, activeCount, wp.maxWorkers, queueSize, queueCap) } } @@ -212,13 +216,35 @@ func (wp *WorkerPool) Submit(task *WorkerTask) { return } + // Dedup: skip if this symbol is already in-flight (active worker or queued) + symbol := "unknown" + chainID := int64(0) + if task.Request != nil { + if task.Request.Intent != nil { + symbol = task.Request.Intent.Symbol + } + if task.Request.DestinationChain != nil { + chainID = task.Request.DestinationChain.ChainID + } + } + dedupeKey := fmt.Sprintf("%s-%d", symbol, chainID) + + wp.inflightMu.Lock() + if existingID, exists := wp.inflightSymbols[dedupeKey]; exists { + wp.inflightMu.Unlock() + logger.Debugf("[router=%s] Skipping duplicate task %s for %s (already in-flight as %s)", wp.routerID, task.ID, dedupeKey, existingID) + return + } + wp.inflightSymbols[dedupeKey] = task.ID + wp.inflightMu.Unlock() + select { case wp.taskQueue <- task: wp.pendingMu.Lock() wp.pendingTasks = append(wp.pendingTasks, task) wp.pendingMu.Unlock() queueSize := len(wp.taskQueue) - logger.Debugf("[router=%s] Task %s queued (queue: %d/%d)", wp.routerID, task.ID, queueSize, cap(wp.taskQueue)) + logger.Infof("[router=%s] Task %s queued (queue: %d/%d)", wp.routerID, task.ID, queueSize, cap(wp.taskQueue)) // Update queue size metric if wp.metricsCollector != nil { wp.metricsCollector.SetTaskQueueSize(int32(queueSize)) @@ -226,12 +252,12 @@ func (wp *WorkerPool) Submit(task *WorkerTask) { default: queueLen := len(wp.taskQueue) queueCap := cap(wp.taskQueue) - symbol := "unknown" - if task.Request != nil && task.Request.Intent != nil { - symbol = task.Request.Intent.Symbol - } logger.Errorf("[router=%s] CRITICAL: Task queue full (%d/%d), DROPPING task %s for symbol %s - consider increasing queue size or worker count", wp.routerID, queueLen, queueCap, task.ID, symbol) + // Remove from inflight since we dropped it + wp.inflightMu.Lock() + delete(wp.inflightSymbols, dedupeKey) + wp.inflightMu.Unlock() // Record dropped task metric if wp.metricsCollector != nil { wp.metricsCollector.IncWorkerTasksDropped() @@ -243,18 +269,18 @@ func (wp *WorkerPool) Submit(task *WorkerTask) { func (w *Worker) start(ctx context.Context) { defer w.wg.Done() - logger.Debugf("[router=%s][WORKER-%d] started", w.pool.routerID, w.id) + logger.Infof("[router=%s][WORKER-%d] started", w.pool.routerID, w.id) for { select { case <-ctx.Done(): - logger.Debugf("[router=%s][WORKER-%d] stopped due to context cancellation", w.pool.routerID, w.id) + logger.Infof("[router=%s][WORKER-%d] stopped due to context cancellation", w.pool.routerID, w.id) return case <-w.quit: - logger.Debugf("[router=%s][WORKER-%d] stopped due to quit signal", w.pool.routerID, w.id) + logger.Infof("[router=%s][WORKER-%d] stopped due to quit signal", w.pool.routerID, w.id) return case task := <-w.taskQueue: - logger.Debugf("[router=%s][WORKER-%d] picked up task %s", w.pool.routerID, w.id, task.ID) + logger.Infof("[router=%s][WORKER-%d] picked up task %s", w.pool.routerID, w.id, task.ID) w.pool.removePending(task.ID) w.processTask(ctx, task) } @@ -287,7 +313,7 @@ func (wp *WorkerPool) removePending(taskID string) { for i, t := range wp.pendingTasks { if t.ID == taskID { wp.pendingTasks = append(wp.pendingTasks[:i], wp.pendingTasks[i+1:]...) - logger.Debugf("[router=%s] Removed task %s from pending list (remaining: %d)", wp.routerID, taskID, len(wp.pendingTasks)) + logger.Infof("[router=%s] Removed task %s from pending list (remaining: %d)", wp.routerID, taskID, len(wp.pendingTasks)) return } } @@ -348,23 +374,48 @@ func (w *Worker) processTask(ctx context.Context, task *WorkerTask) { } logger.Infof("[router=%s][WORKER-%d] Starting task: %s, symbol=%s, chain=%d, task_router=%s, active_workers=%d", w.pool.routerID, w.id, task.ID, symbol, chainID, routerID, atomic.LoadInt32(&w.pool.activeWorkers)) - // Create timeout context to prevent workers from blocking forever taskCtx, cancel := context.WithTimeout(ctx, w.pool.taskTimeout) defer cancel() - - // Process the task with retry logic + deadline, _ := taskCtx.Deadline() + + // Process the task with retry logic. + // Handler runs in a goroutine so that taskCtx timeout can enforce a hard + // upper bound even if the handler (or RPC calls inside it) ignores context + // cancellation. Without this, a stalled TCP connection to the RPC endpoint + // can block the worker permanently (go-ethereum's Transact may not respect + // context cancellation for stalled connections). var err error maxRetries := 3 retryCount := 0 for retry := 0; retry < maxRetries; retry++ { if retry > 0 { retryCount++ - logger.Debugf("[router=%s][WORKER-%d] retrying task %s (attempt %d/%d)", w.pool.routerID, w.id, task.ID, retry+1, maxRetries) + logger.Infof("[router=%s][WORKER-%d] retrying task %s (attempt %d/%d)", w.pool.routerID, w.id, task.ID, retry+1, maxRetries) time.Sleep(time.Second * time.Duration(retry)) } - err = task.Handler(taskCtx, task) + type handlerResult struct { + err error + } + ch := make(chan handlerResult, 1) + handlerStart := time.Now() + go func() { + ch <- handlerResult{err: task.Handler(taskCtx, task)} + }() + + select { + case res := <-ch: + err = res.err + logger.Infof("[router=%s][WORKER-%d] Handler returned: took=%v, err=%v", w.pool.routerID, w.id, time.Since(handlerStart), err) + case <-taskCtx.Done(): + err = taskCtx.Err() + logger.Warnf("[router=%s][WORKER-%d] Task %s context expired during handler execution: took=%v, deadline_remaining=%v, err=%v", + w.pool.routerID, w.id, task.ID, time.Since(handlerStart), time.Until(deadline).Round(time.Millisecond), taskCtx.Err()) + // Goroutine may still be running -- buffered channel (cap 1) prevents leak + // when it eventually completes + } + if err == nil { break } @@ -378,6 +429,21 @@ func (w *Worker) processTask(ctx context.Context, task *WorkerTask) { logger.Errorf("[router=%s][WORKER-%d] Task %s failed (attempt %d/%d): %v", w.pool.routerID, w.id, task.ID, retry+1, maxRetries, err) } + // Clear inflight tracking so the next update for this symbol can be queued + if task.Request != nil { + clearSym := "unknown" + clearCID := int64(0) + if task.Request.Intent != nil { + clearSym = task.Request.Intent.Symbol + } + if task.Request.DestinationChain != nil { + clearCID = task.Request.DestinationChain.ChainID + } + key := fmt.Sprintf("%s-%d", clearSym, clearCID) + w.pool.inflightMu.Lock() + delete(w.pool.inflightSymbols, key) + w.pool.inflightMu.Unlock() + } duration := time.Since(startTime) From 319d4f2e4b0d6e218223d61c23ab036f890b0c31 Mon Sep 17 00:00:00 2001 From: nnn-gif Date: Mon, 22 Jun 2026 14:19:46 +0530 Subject: [PATCH 2/4] feat: attestor forward same timestamp --- services/attestor/config.yaml.example | 2 ++ services/attestor/pkg/config/config.go | 3 +++ services/attestor/pkg/intent/intent.go | 28 +++++++++++++++------- services/attestor/pkg/interfaces/signer.go | 10 ++++---- services/attestor/pkg/service/attestor.go | 21 ++++++++++++---- 5 files changed, 48 insertions(+), 16 deletions(-) diff --git a/services/attestor/config.yaml.example b/services/attestor/config.yaml.example index fbdb8cd..0eda3b6 100644 --- a/services/attestor/config.yaml.example +++ b/services/attestor/config.yaml.example @@ -22,6 +22,7 @@ attestor: - ETH/USD polling_time: 300ms batch_mode: true + use_oracle_timestamp: false # When true, embed the oracle contract's timestamp in the intent instead of time.Now() # Logging Configuration logging: @@ -46,6 +47,7 @@ api: # - ATTESTOR_ATTESTOR_SYMBOLS (comma-separated) # - ATTESTOR_ATTESTOR_POLLING_TIME (e.g., "5s", "300ms") # - ATTESTOR_ATTESTOR_BATCH_MODE (true/false) +# - ATTESTOR_ATTESTOR_USE_ORACLE_TIMESTAMP (true/false) — use oracle contract timestamp instead of time.Now() # - ATTESTOR_ATTESTOR_MODE (prime/replica) # - ATTESTOR_LOGGING_LEVEL # - ATTESTOR_METRICS_PORT diff --git a/services/attestor/pkg/config/config.go b/services/attestor/pkg/config/config.go index e86d8df..19bf67f 100644 --- a/services/attestor/pkg/config/config.go +++ b/services/attestor/pkg/config/config.go @@ -110,6 +110,7 @@ type Config struct { DeviationTrigger bool `mapstructure:"deviation_trigger"` DeviationThreshold int `mapstructure:"deviation_threshold"` ForceUpdateInterval time.Duration `mapstructure:"force_update_interval"` + UseOracleTimestamp bool `mapstructure:"use_oracle_timestamp"` Guardian GuardianConfig `mapstructure:"guardian"` } `mapstructure:"attestor"` @@ -187,6 +188,7 @@ func Init(configPath string) (*Config, error) { v.BindEnv("attestor.deviation_trigger", "ATTESTOR_ATTESTOR_DEVIATION_TRIGGER") v.BindEnv("attestor.deviation_threshold", "ATTESTOR_ATTESTOR_DEVIATION_THRESHOLD") v.BindEnv("attestor.force_update_interval", "ATTESTOR_ATTESTOR_FORCE_UPDATE_INTERVAL") + v.BindEnv("attestor.use_oracle_timestamp", "ATTESTOR_ATTESTOR_USE_ORACLE_TIMESTAMP") v.BindEnv("oracle.client_type", "ATTESTOR_ORACLE_CLIENT_TYPE") v.BindEnv("attestor.guardian.default.max_deviation_bips", "ATTESTOR_GUARDIAN_MAX_DEVIATION_BIPS") v.BindEnv("attestor.guardian.default.max_timestamp_age", "ATTESTOR_GUARDIAN_MAX_TIMESTAMP_AGE") @@ -208,6 +210,7 @@ func Init(configPath string) (*Config, error) { v.SetDefault("attestor.deviation_trigger", false) v.SetDefault("attestor.deviation_threshold", 50) v.SetDefault("attestor.force_update_interval", "0s") + v.SetDefault("attestor.use_oracle_timestamp", false) v.SetDefault("attestor.guardian.default.max_deviation_bips", 500) v.SetDefault("attestor.guardian.default.max_timestamp_age", 3600) v.SetDefault("attestor.guardian.default.min_guardian_matches", 1) diff --git a/services/attestor/pkg/intent/intent.go b/services/attestor/pkg/intent/intent.go index 8e31486..fd01984 100644 --- a/services/attestor/pkg/intent/intent.go +++ b/services/attestor/pkg/intent/intent.go @@ -42,7 +42,7 @@ func generateNonce() *big.Int { return big.NewInt(nonce) } -func AttestValue(ctx context.Context, multiClient *multirpc.MultiClient, privateKey string, fromAddress string, price *big.Int, volume *big.Int, symbol string) (string, error) { +func AttestValue(ctx context.Context, multiClient *multirpc.MultiClient, privateKey string, fromAddress string, price *big.Int, volume *big.Int, symbol string, oracleTimestamp *big.Int) (string, error) { if privateKey == "" { return "", fmt.Errorf("private key not provided") } @@ -50,12 +50,17 @@ func AttestValue(ctx context.Context, multiClient *multirpc.MultiClient, private return "", fmt.Errorf("multiClient is required") } - now := time.Now().Unix() - nowBig := big.NewInt(now) + // Use oracle timestamp if provided, otherwise fall back to current time + var nowBig *big.Int + if oracleTimestamp != nil && oracleTimestamp.Sign() > 0 { + nowBig = new(big.Int).Set(oracleTimestamp) + } else { + nowBig = big.NewInt(time.Now().Unix()) + } // Generate unique nonce using atomic counter to prevent collisions nonce := generateNonce() - expiry := big.NewInt(now + 3600) + expiry := big.NewInt(time.Now().Unix() + 3600) cfg := config.Get() intentType := cfg.Attestor.IntentType @@ -218,9 +223,10 @@ func AttestValue(ctx context.Context, multiClient *multirpc.MultiClient, private // SymbolData represents price data for a single symbol type SymbolData struct { - Symbol string - Price *big.Int - Volume *big.Int + Symbol string + Price *big.Int + Volume *big.Int + Timestamp *big.Int // Oracle timestamp; nil means use time.Now() at signing } func AttestMultipleValues(ctx context.Context, multiClient *multirpc.MultiClient, privateKey string, fromAddress string, symbolsData []SymbolData) (string, error) { @@ -282,6 +288,12 @@ func AttestMultipleValues(ctx context.Context, multiClient *multirpc.MultiClient // Generate unique nonce for each intent in the batch to prevent collisions nonce := generateNonce() + // Use per-symbol oracle timestamp if provided, otherwise fall back to current time + ts := nowBig + if data.Timestamp != nil && data.Timestamp.Sign() > 0 { + ts = data.Timestamp + } + intent := types.OracleIntent{ IntentType: intentType, Version: intentVersion, @@ -290,7 +302,7 @@ func AttestMultipleValues(ctx context.Context, multiClient *multirpc.MultiClient Expiry: expiry, Symbol: data.Symbol, Price: data.Price, - Timestamp: nowBig, + Timestamp: ts, Source: "DIA Oracle", } diff --git a/services/attestor/pkg/interfaces/signer.go b/services/attestor/pkg/interfaces/signer.go index fb39998..0dc5d3d 100644 --- a/services/attestor/pkg/interfaces/signer.go +++ b/services/attestor/pkg/interfaces/signer.go @@ -8,7 +8,8 @@ import ( // IntentSigner defines the interface for signing intents type IntentSigner interface { // SignIntent creates an EIP-712 signed intent for a single value - SignIntent(ctx context.Context, price, volume *big.Int, symbol string) ([]byte, error) + // timestamp is the oracle timestamp to embed in the intent; pass nil to use time.Now() + SignIntent(ctx context.Context, price, volume, timestamp *big.Int, symbol string) ([]byte, error) // SignBatchIntent creates an EIP-712 signed intent for multiple values SignBatchIntent(ctx context.Context, values []SymbolData) ([]byte, error) @@ -16,7 +17,8 @@ type IntentSigner interface { // SymbolData represents data for a single symbol in a batch type SymbolData struct { - Symbol string - Price *big.Int - Volume *big.Int + Symbol string + Price *big.Int + Volume *big.Int + Timestamp *big.Int // Oracle timestamp; nil means use time.Now() at signing } diff --git a/services/attestor/pkg/service/attestor.go b/services/attestor/pkg/service/attestor.go index 5245227..449461b 100644 --- a/services/attestor/pkg/service/attestor.go +++ b/services/attestor/pkg/service/attestor.go @@ -348,8 +348,14 @@ func (s *AttestorService) processSingleAttestation(ctx context.Context, symbol s "MinGuardianMatches": guardianParams.MinGuardianMatches, }).Debug("Retrieved oracle value") + // Determine timestamp to embed in the intent + var intentTimestamp *big.Int + if s.config.Attestor.UseOracleTimestamp && timestamp != nil && timestamp.Sign() > 0 { + intentTimestamp = timestamp + } + // Sign intent - signedIntent, err := s.signer.SignIntent(ctx, price, volume, symbol) + signedIntent, err := s.signer.SignIntent(ctx, price, volume, intentTimestamp, symbol) if err != nil { s.metrics.RecordIntentCreated(symbol, false) return errors.NewSignerError("sign intent", symbol, err) @@ -447,10 +453,17 @@ symbolLoop: } logger.WithFields(logFields).Debug("Retrieved oracle value") + // Set oracle timestamp if configured + var ts *big.Int + if s.config.Attestor.UseOracleTimestamp && timestamp != nil && timestamp.Sign() > 0 { + ts = timestamp + } + symbolData = append(symbolData, interfaces.SymbolData{ - Symbol: symbol, - Price: price, - Volume: volume, + Symbol: symbol, + Price: price, + Volume: volume, + Timestamp: ts, }) } From 5a215e610d2c2a8de98a449047e51ec0331d80f5 Mon Sep 17 00:00:00 2001 From: nnn-gif Date: Mon, 22 Jun 2026 14:29:56 +0530 Subject: [PATCH 3/4] feat: eIP712Signer add timestamp as param --- services/attestor/pkg/signer/eip712.go | 11 ++++++----- 1 file changed, 6 insertions(+), 5 deletions(-) diff --git a/services/attestor/pkg/signer/eip712.go b/services/attestor/pkg/signer/eip712.go index c30b99f..4b038c0 100644 --- a/services/attestor/pkg/signer/eip712.go +++ b/services/attestor/pkg/signer/eip712.go @@ -51,7 +51,7 @@ func NewEIP712Signer(privateKeyHex string, rpcURLs []string) (*EIP712Signer, err } // SignIntent creates an EIP-712 signed intent for a single value -func (s *EIP712Signer) SignIntent(ctx context.Context, price, volume *big.Int, symbol string) ([]byte, error) { +func (s *EIP712Signer) SignIntent(ctx context.Context, price, volume, timestamp *big.Int, symbol string) ([]byte, error) { // Validate inputs if price == nil || price.Sign() <= 0 { return nil, errors.NewValidationError("price", price, "must be positive") @@ -63,7 +63,7 @@ func (s *EIP712Signer) SignIntent(ctx context.Context, price, volume *big.Int, s return nil, errors.NewValidationError("symbol", symbol, "must not be empty") } - signedIntentJSON, err := intent.AttestValue(ctx, s.signingClient, s.privateKeyHex, s.address.Hex(), price, volume, symbol) + signedIntentJSON, err := intent.AttestValue(ctx, s.signingClient, s.privateKeyHex, s.address.Hex(), price, volume, symbol, timestamp) if err != nil { return nil, errors.NewSignerError("sign intent", symbol, err) } @@ -104,9 +104,10 @@ func (s *EIP712Signer) SignBatchIntent(ctx context.Context, values []interfaces. } symbolData = append(symbolData, intent.SymbolData{ - Symbol: v.Symbol, - Price: v.Price, - Volume: v.Volume, + Symbol: v.Symbol, + Price: v.Price, + Volume: v.Volume, + Timestamp: v.Timestamp, }) } From 13a7223285f5dc02e5266e2f916d90050f86743a Mon Sep 17 00:00:00 2001 From: nnn-gif Date: Wed, 24 Jun 2026 10:42:02 +0530 Subject: [PATCH 4/4] feat: enable separate wallet per router --- services/bridge/internal/bridge/bridge.go | 176 ++++++++++++++++-- services/bridge/internal/bridge/health.go | 8 +- .../internal/bridge/transaction_handler.go | 25 ++- 3 files changed, 185 insertions(+), 24 deletions(-) diff --git a/services/bridge/internal/bridge/bridge.go b/services/bridge/internal/bridge/bridge.go index aec64b5..2597d22 100644 --- a/services/bridge/internal/bridge/bridge.go +++ b/services/bridge/internal/bridge/bridge.go @@ -4,6 +4,7 @@ import ( "context" "fmt" "math/big" + "strings" "sync" "time" @@ -30,7 +31,8 @@ type Bridge struct { configService *config.ConfigService db *database.DB readClient rpc.EthClient - writeClients map[int64]*WriteClient + chainClients map[int64]*WriteClient + routerClients map[string]*WriteClient // Channels for communication updateChan chan *bridgetypes.UpdateRequest @@ -124,14 +126,18 @@ func NewBridge(modularCfg *config.ModularConfig, cfgService *config.ConfigServic logger.Errorf("Failed to load routers: %v", err) } - destClients := make(map[int64]*WriteClient) + chainClients := make(map[int64]*WriteClient) enabledChains := cfgService.GetEnabledChains() + chainConfigMap := make(map[int64]*config.ChainConfig) + for _, chainCfg := range enabledChains { + chainConfigMap[chainCfg.ChainID] = chainCfg + } if len(enabledChains) == 0 { return nil, fmt.Errorf("no enabled chains found in configuration - check that chains have 'enabled: true'") } - logger.Infof("Found %d enabled chain(s), attempting to create destination clients...", len(enabledChains)) + logger.Infof("Found %d enabled chain(s), attempting to create chain destination clients...", len(enabledChains)) var failedChains []string for _, chainConfig := range enabledChains { @@ -142,7 +148,7 @@ func NewBridge(modularCfg *config.ModularConfig, cfgService *config.ConfigServic continue } - // Check if there's at least one enabled + // Check if there's at least one enabled receiver/pushoracle hasReceiverContract := false for _, contract := range contracts { if (contract.Type == "receiver" || contract.Type == "pushoracle") && contract.Enabled { @@ -165,16 +171,16 @@ func NewBridge(modularCfg *config.ModularConfig, cfgService *config.ConfigServic destClient, err := NewWriteClient(chainConfig, contracts, cfgService.GetInfrastructure().PrivateKey, queueManager, maxSafeGap) if err != nil { - logger.Errorf("Failed to create destination client for chain %d (%s): %v", chainConfig.ChainID, chainConfig.Name, err) + logger.Errorf("Failed to create chain client for chain %d (%s): %v", chainConfig.ChainID, chainConfig.Name, err) failedChains = append(failedChains, fmt.Sprintf("%d (%s): %v", chainConfig.ChainID, chainConfig.Name, err)) continue } - destClients[chainConfig.ChainID] = destClient - logger.Infof("Successfully created destination client for chain %d (%s)", chainConfig.ChainID, chainConfig.Name) + chainClients[chainConfig.ChainID] = destClient + logger.Infof("Successfully created chain client for chain %d (%s)", chainConfig.ChainID, chainConfig.Name) } - if len(destClients) == 0 { - errorMsg := "no destination clients available. Reasons:\n" + if len(chainClients) == 0 { + errorMsg := "no chain clients available. Reasons:\n" if len(failedChains) > 0 { for _, reason := range failedChains { errorMsg += fmt.Sprintf(" - Chain %s\n", reason) @@ -185,6 +191,61 @@ func NewBridge(modularCfg *config.ModularConfig, cfgService *config.ConfigServic return nil, fmt.Errorf(errorMsg) } + routerClients := make(map[string]*WriteClient) + chainToRouterCount := make(map[int64]int) + + for _, router := range enabledRouters { + if len(router.Destinations) == 0 { + continue + } + chainID := router.Destinations[0].ChainID + chainToRouterCount[chainID]++ + } + + logger.Infof("Found %d enabled router(s), checking for router-specific private keys...", len(enabledRouters)) + + for _, router := range enabledRouters { + if router.PrivateKey == "" { + logger.Debugf("Router %s: will use chain client (no private key configured)", router.ID) + continue + } + + if len(router.Destinations) == 0 { + logger.Warnf("Router %s: has private key but no destinations - skipping client creation", router.ID) + continue + } + + dest := router.Destinations[0] + chainID := dest.ChainID + + chainConfig := chainConfigMap[chainID] + if chainConfig == nil { + logger.Errorf("Router %s: chain %d not found in enabled chains configuration", router.ID, chainID) + continue + } + + contracts := cfgService.GetContractsForChain(chainID) + if len(contracts) == 0 { + logger.Warnf("Router %s: chain %d has no enabled contracts - skipping client creation", router.ID, chainID) + continue + } + + maxSafeGap := calculateMaxSafeGap(chainToRouterCount[chainID]) + + routerClient, err := NewWriteClient(chainConfig, contracts, router.PrivateKey, queueManager, maxSafeGap) + if err != nil { + logger.Errorf("Failed to create router client for router %s: %v", router.ID, err) + continue + } + + routerClients[router.ID] = routerClient + logger.Infof("Successfully created router client for router %s on chain %d (using router private key)", router.ID, chainID) + } + + if len(routerClients) > 0 { + logger.Infof("Created %d router-specific client(s)", len(routerClients)) + } + // Store worker pool config for dynamic pool creation workerPoolConfig := cfgService.GetInfrastructure().WorkerPool @@ -195,7 +256,7 @@ func NewBridge(modularCfg *config.ModularConfig, cfgService *config.ConfigServic metricsManager := NewMetricsManager(metricsCollector) ethClients := make(map[int64]rpc.EthClient) - for chainID, writeClient := range destClients { + for chainID, writeClient := range chainClients { ethClients[chainID] = writeClient.GetEthClient() } @@ -205,7 +266,8 @@ func NewBridge(modularCfg *config.ModularConfig, cfgService *config.ConfigServic configService: cfgService, db: db, readClient: readClient, - writeClients: destClients, + chainClients: chainClients, // chain-based clients (infrastructure key) + routerClients: routerClients, // router-specific clients (router key) updateChan: make(chan *bridgetypes.UpdateRequest, 1000), errorChan: errorChan, shutdownChan: make(chan struct{}), @@ -386,7 +448,7 @@ func (b *Bridge) Start(ctx context.Context) error { if b.routerRegistry != nil { logger.Info("Fetching router state from on-chain...") ethClients := make(map[int64]rpc.EthClient) - for chainID, writeClient := range b.writeClients { + for chainID, writeClient := range b.chainClients { ethClients[chainID] = writeClient.GetEthClient() } routers := b.routerRegistry.GetActiveRouters() @@ -568,7 +630,10 @@ func (b *Bridge) Stop(ctx context.Context) error { // Close connections b.readClient.Close() - for _, destClient := range b.writeClients { + for _, destClient := range b.chainClients { + destClient.client.Close() + } + for _, destClient := range b.routerClients { destClient.client.Close() } @@ -656,7 +721,10 @@ func (b *Bridge) processUpdates(ctx context.Context) { // Check if update is stale based on last update in cache if updateReq.Intent != nil && !updateReq.CreatedAt.IsZero() && updateReq.Contract != nil { - destClient := b.writeClients[updateReq.DestinationChain.ChainID] + destClient := b.routerClients[updateReq.RouterID] + if destClient == nil { + destClient = b.chainClients[updateReq.DestinationChain.ChainID] + } if destClient != nil { lastUpdateTime := destClient.getLastUpdate(updateReq.Intent.Symbol, updateReq.Contract.Address) if !lastUpdateTime.IsZero() && updateReq.CreatedAt.Before(lastUpdateTime) { @@ -751,7 +819,7 @@ func (b *Bridge) handleUpdateRequest(ctx context.Context, task *worker.WorkerTas } }() - handler := NewTransactionHandler(b.writeClients, b.routerRegistry, b.metricsManager.GetTracker(), b.onChainMonitor) + handler := NewTransactionHandler(b.chainClients, b.routerClients, b.routerRegistry, b.metricsManager.GetTracker(), b.onChainMonitor) return handler.Process(ctx, task.Request) } @@ -788,3 +856,81 @@ func (b *Bridge) ListPools() []api.PoolInfo { } return pools } + +// resolveParameterValue resolves a template parameter source to its actual value +func (b *Bridge) resolveParameterValue(source string, updateReq *bridgetypes.UpdateRequest) (interface{}, error) { + if strings.HasPrefix(source, "${") && strings.HasSuffix(source, "}") { + templateVar := strings.TrimSuffix(strings.TrimPrefix(source, "${"), "}") + + switch { + case strings.HasPrefix(templateVar, "enrichment."): + enrichmentKey := strings.TrimPrefix(templateVar, "enrichment.") + if updateReq.ExtractedData != nil && updateReq.ExtractedData.Enrichment != nil { + if value, exists := updateReq.ExtractedData.Enrichment[enrichmentKey]; exists { + if enrichmentKey == "fullIntent" { + if intent, ok := value.(*bridgetypes.OracleIntent); ok { + logger.Debugf("Retrieved fullIntent from enrichment: symbol=%s price=%s timestamp=%s nonce=%s expiry=%s signer=%s source=%s", + intent.Symbol, + intent.Price.String(), + intent.Timestamp.String(), + intent.Nonce.String(), + intent.Expiry.String(), + intent.Signer.Hex(), + intent.Source) + return intent, nil + } + return nil, fmt.Errorf("fullIntent has unexpected type %T", value) + } + return value, nil + } + return nil, fmt.Errorf("enrichment key %s not found", enrichmentKey) + } + return nil, fmt.Errorf("enrichment data not available") + + case strings.HasPrefix(templateVar, "event."): + eventField := strings.TrimPrefix(templateVar, "event.") + if updateReq.Event == nil { + return nil, fmt.Errorf("event data not available") + } + + switch eventField { + case "requestId": + if updateReq.Event.RequestId != nil { + return updateReq.Event.RequestId, nil + } + return nil, fmt.Errorf("event requestId not found") + default: + return nil, fmt.Errorf("unsupported event field: %s", eventField) + } + + case strings.HasPrefix(templateVar, "intent."): + if updateReq.Intent == nil { + return nil, fmt.Errorf("intent data not available") + } + return updateReq.Intent, nil + + default: + return nil, fmt.Errorf("unsupported template variable: %s", templateVar) + } + } + + return source, nil +} + +// buildMethodParams builds the method parameters for a destination method call +func (b *Bridge) buildMethodParams(methodConfig *config.DestinationMethodConfig, updateReq *bridgetypes.UpdateRequest) ([]interface{}, error) { + if methodConfig == nil || methodConfig.Params == nil { + return nil, nil + } + + params := make([]interface{}, 0, len(methodConfig.Params)) + for paramName, paramValue := range methodConfig.Params { + value, err := b.resolveParameterValue(paramValue, updateReq) + if err != nil { + return nil, fmt.Errorf("failed to resolve parameter %s: %w", paramName, err) + } + params = append(params, value) + } + + return params, nil +} diff --git a/services/bridge/internal/bridge/health.go b/services/bridge/internal/bridge/health.go index 6c6ed6e..7b25592 100644 --- a/services/bridge/internal/bridge/health.go +++ b/services/bridge/internal/bridge/health.go @@ -33,13 +33,14 @@ func (b *Bridge) initializeChainStats() { } // Destination chain stats - for _, destClient := range b.writeClients { + for _, destClient := range b.chainClients { b.stats.ChainStats[destClient.chainConfig.ChainID] = &bridgetypes.ChainStatus{ ChainID: destClient.chainConfig.ChainID, Name: destClient.chainConfig.Name, Connected: true, } } + // Router-specific clients use same chains as chain clients, so no need to duplicate stats } // healthCheck performs periodic health checks @@ -98,12 +99,13 @@ func (b *Bridge) performHealthCheck(ctx context.Context) { logger.Errorf("Source chain health check failed: %v", err) } - // Check destination chains - for _, destClient := range b.writeClients { + // Check destination chains (chain-based clients) + for _, destClient := range b.chainClients { if err := b.checkChainHealth(ctx, destClient.client, destClient.chainConfig.ChainID); err != nil { logger.Errorf("Destination chain %d health check failed: %v", destClient.chainConfig.ChainID, err) } } + // Note: router-specific clients use same chains as chain clients, so no need to check again } // checkChainHealth checks the health of a single chain diff --git a/services/bridge/internal/bridge/transaction_handler.go b/services/bridge/internal/bridge/transaction_handler.go index 095d974..ea5651f 100644 --- a/services/bridge/internal/bridge/transaction_handler.go +++ b/services/bridge/internal/bridge/transaction_handler.go @@ -38,7 +38,8 @@ type TransactionContext struct { // TransactionHandler handles the complete lifecycle of a transaction type TransactionHandler struct { - writeClients map[int64]*WriteClient + chainClients map[int64]*WriteClient // Chain-based clients (infrastructure key) + routerClients map[string]*WriteClient // Router-specific clients (router key) routerRegistry *router.GenericRegistry metricsTracker *MetricsTracker onChainMonitor *leader.OnChainMonitor // Optional: for replica monitoring info @@ -46,9 +47,10 @@ type TransactionHandler struct { } // NewTransactionHandler creates a new transaction handler -func NewTransactionHandler(writeClients map[int64]*WriteClient, registry *router.GenericRegistry, tracker *MetricsTracker, monitor *leader.OnChainMonitor) *TransactionHandler { +func NewTransactionHandler(chainClients map[int64]*WriteClient, routerClients map[string]*WriteClient, registry *router.GenericRegistry, tracker *MetricsTracker, monitor *leader.OnChainMonitor) *TransactionHandler { return &TransactionHandler{ - writeClients: writeClients, + chainClients: chainClients, + routerClients: routerClients, routerRegistry: registry, metricsTracker: tracker, onChainMonitor: monitor, @@ -125,9 +127,20 @@ func (h *TransactionHandler) buildContext(ctx context.Context, updateReq *bridge return nil, fmt.Errorf("destination chain is nil") } - destClient := h.writeClients[updateReq.DestinationChain.ChainID] - if destClient == nil { - return nil, fmt.Errorf("destination client not found for chain %d", updateReq.DestinationChain.ChainID) + chainID := updateReq.DestinationChain.ChainID + routerID := updateReq.RouterID + + // Try router-specific client first + destClient, exists := h.routerClients[routerID] + if !exists { + // Fall back to chain-based client + destClient, exists = h.chainClients[chainID] + if !exists { + return nil, fmt.Errorf("no write client for router %s (router client not found) and no chain client for chain %d", routerID, chainID) + } + logger.Debugf("[TX-HANDLER] Using chain client for router %s on chain %d (no router-specific client)", routerID, chainID) + } else { + logger.Debugf("[TX-HANDLER] Using router-specific client for router %s on chain %d", routerID, chainID) } gasPrice, err := destClient.getGasPrice(ctx)