fix: silent-stall RPC failures (DEVOP-594)

.beads/.gitignore

@@ -0,0 +1,46 @@
+# Database
+*.db
+*.db-journal
+*.db-shm
+*.db-wal
+
+# Lock files
+*.lock
+
+# Temporary
+last-touched
+*.tmp
+
+# Local history backups
+.br_history/
+
+# DB-family recovery artifacts (truncated WAL/SHM, quarantined sidecars)
+# — same lifecycle as .br_history/, written by recovery paths and
+# `br doctor --repair`. Filename suffix `.truncated-wal` slips past the
+# generic `*.db-wal` glob above, so it needs an explicit entry (#271).
+.br_recovery/
+
+# Sync state (local-only, per-machine)
+.sync.lock
+sync_base.jsonl
+
+# Merge artifacts (temporary files from 3-way merge)
+beads.base.jsonl
+beads.base.meta.json
+beads.left.jsonl
+beads.left.meta.json
+beads.right.jsonl
+beads.right.meta.json
+
+# Daemon runtime files
+daemon.lock
+daemon.log
+daemon.pid
+bd.sock
+sync-state.json
+
+# Worktree redirect file
+redirect
+
+# bv (beads viewer) lock file
+.bv.lock

.beads/config.yaml

@@ -0,0 +1,4 @@
+# Beads Project Configuration
+# issue_prefix: allora-offchain-node
+# default_priority: 2
+# default_type: task

.beads/metadata.json

@@ -0,0 +1,4 @@
+{
+  "database": "beads.db",
+  "jsonl_export": "issues.jsonl"
+}

.gitignore

@@ -29,4 +29,6 @@ go.work.sum
 config.json
 
 .DS_Store
-**/__debug*
+**/__debug*
+
+.sirene/

lib/connection_manager.go

@@ -194,13 +194,22 @@ func (connectionManager *ConnectionManager) GetCurrentTxNode() (*NodeConfig, err
 	return &connectionManager.txNodes[connectionManager.txIdx], nil
 }
 
-// internal function, switches to a node assuming a lock has been acquired
+// internal function, switches to a node assuming a lock has been acquired.
+// When the configured list has only one node, "switching" cannot rotate to a
+// different endpoint — but we still want to force a fresh dial so that a stale
+// connection (e.g. half-closed by an upstream CDN/LB) is replaced. The
+// single-node short-circuit is delegated to forceReconnectLocked.
 func (connectionManager *ConnectionManager) switchToNodeLocked(mode, index int, nodes []NodeConfig) (*NodeConfig, error) {
 	if len(nodes) == 0 || index < 0 || index >= len(nodes) {
 		return nil, fmt.Errorf("invalid node index, not switching")
 	}
 	if len(nodes) == 1 {
-		return &nodes[0], nil
+		// Only one endpoint configured: rotating index is a no-op. Force the
+		// underlying connection to be re-dialed so the next call hits a fresh
+		// gRPC stream / RPC HTTP transport instead of the (likely dead) one
+		// we just failed on. This is the only recovery path available when no
+		// failover endpoint exists.
+		return connectionManager.forceReconnectLocked(mode, 0, nodes)
 	}
 	var oldIndex int
 	if mode == GRPC_MODE {
@@ -222,6 +231,91 @@ func (connectionManager *ConnectionManager) switchToNodeLocked(mode, index int,
 	return &nodes[index], nil
 }
 
+// forceReconnectLocked rebuilds the underlying chain client for the node at
+// `index` and writes the new NodeConfig back into the slice in place. The
+// caller MUST hold the appropriate write lock (queryMu for GRPC_MODE, txMu
+// for RPC_MODE) before calling this.
+//
+// Why this exists: the existing monitorGRPCConnection goroutine in
+// lib/grpcclient only forces a re-dial when the grpc.ClientConn observes a
+// TransientFailure or Shutdown state. When a CDN/LB upstream half-closes an
+// HTTP/2 stream, the client side often stays stuck in Ready until the next
+// RPC fails — so the monitor never reacts and every subsequent call returns
+// the same Unavailable error forever. Calling forceReconnectLocked from the
+// error-classification path breaks that loop.
+//
+// Behaviour:
+//   - On success, the old chain client (gRPC or HTTP) is closed (best-effort)
+//     and replaced with a freshly initialized one bound to the same endpoint.
+//   - On failure, the old NodeConfig is left intact and the original error
+//     is returned. Callers should treat that as "still broken, try again
+//     later" — not a fatal condition.
+func (connectionManager *ConnectionManager) forceReconnectLocked(mode, index int, nodes []NodeConfig) (*NodeConfig, error) {
+	if len(nodes) == 0 || index < 0 || index >= len(nodes) {
+		return nil, fmt.Errorf("invalid node index, not reconnecting")
+	}
+	endpoint := nodes[index].ServerAddress
+	walletCfg := connectionManager.walletConfig
+	if walletCfg == nil {
+		return nil, fmt.Errorf("wallet config not initialized, cannot reconnect")
+	}
+
+	// Construct a minimal UserConfig view for the factory. We only need the
+	// wallet portion populated; GenerateNodeConfig does not read worker/reputer.
+	factoryConfig := &UserConfig{ // nolint: exhaustruct
+		Wallet: *walletCfg,
+	}
+
+	log.Warn().Str("endpoint", endpoint).Int("mode", mode).Msg("Forcing reconnect to chain endpoint")
+	newNode, err := factoryConfig.GenerateNodeConfig(context.Background(), connectionManager.wallet, mode, endpoint)
+	if err != nil {
+		log.Error().Err(err).Str("endpoint", endpoint).Msg("Force reconnect failed, retaining previous (likely-stale) connection")
+		return &nodes[index], err
+	}
+	newNode.ConnectionManager = connectionManager
+
+	// Best-effort close of the old underlying client. The new connection has
+	// already been created, so in-flight calls on the old one need to bleed
+	// out (or fail fast) rather than block forever on a dead stream.
+	closeOld(&nodes[index], mode)
+
+	// Replace in place so any caller that has cached &nodes[index] sees the
+	// new chain client on the next dereference.
+	nodes[index] = *newNode
+	metrics.GetMetrics().IncrementMetricsCounterWithLabels(metrics.GRPCReconnectionCount, endpoint)
+	log.Info().Str("endpoint", endpoint).Int("mode", mode).Msg("Force reconnect complete")
+	return &nodes[index], nil
+}
+
+// closeOld closes the chain client embedded in a NodeConfig in best-effort
+// fashion. Errors are logged at debug since we may be calling this on an
+// already-broken connection where Close itself errors out.
+func closeOld(n *NodeConfig, mode int) {
+	if n == nil {
+		return
+	}
+	if mode == GRPC_MODE && n.Chain.GRPCClient != nil {
+		// Stop the per-connection monitor goroutine first, otherwise it will
+		// keep ticking against the closed conn until parent ctx is cancelled
+		// (i.e. process exit) and quietly leak.
+		if n.Chain.GRPCMonitorCancel != nil {
+			n.Chain.GRPCMonitorCancel()
+			n.Chain.GRPCMonitorCancel = nil
+		}
+		if err := n.Chain.GRPCClient.Close(); err != nil {
+			log.Debug().Err(err).Str("endpoint", n.ServerAddress).Msg("Closing old gRPC client returned error (likely already broken)")
+		}
+		n.Chain.GRPCClient = nil
+	}
+	if mode == RPC_MODE && n.Chain.RPCClient != nil {
+		// AlloraRPCClient wraps *cometrpc.HTTP, which uses pooled net/http
+		// transports under the hood. The cleanest way to release pooled
+		// connections is to drop the reference and let Go GC the transport
+		// after pending requests drain. No explicit Close exists.
+		n.Chain.RPCClient = nil
+	}
+}
+
 // SwitchToNextNode switches to the next node in the list.
 // Node change is persistent, so it will be used again in the next call
 // Returns current node if error

lib/domain_config.go

@@ -2,6 +2,7 @@ package lib
 
 import (
 	"allora_offchain_node/lib/rpcclient"
+	"context"
 	"errors"
 	"fmt"
 
@@ -79,6 +80,7 @@ type WalletConfig struct {
 type ChainConfig struct {
 	RPCClient            *rpcclient.AlloraRPCClient // A custom wrapper around the cometrpc.HTTP client
 	GRPCClient           *grpc.ClientConn           // Basic type to be used to init module-based clients
+	GRPCMonitorCancel    context.CancelFunc         // Stops the per-conn monitor goroutine; non-nil only for GRPC_MODE nodes
 	EmissionsQueryClient emissions.QueryServiceClient
 	BankQueryClient      bank.QueryClient
 	AuthQueryClient      auth.QueryClient

lib/errors.go

@@ -48,6 +48,7 @@ const ErrCodeNotPermittedToSubmitPayload = 23
 const ErrCodeNotPermittedToAddStake = 24
 const ErrCodeReadFlatPanic = 25
 const ErrCodeReadPerBytePanic = 26
+const ErrCodeGRPCTransport = 27
 const ErrCodeUnexpectedError = 100
 
 var (
@@ -70,6 +71,7 @@ var (
 	ErrReputerNonceWindowNotAvailable = errorsmod.Register(ErrorCodespace, ErrCodeReputerNonceWindowNotAvailable, "reputer nonce window not available")
 	ErrWorkerNonceWindowNotAvailable  = errorsmod.Register(ErrorCodespace, ErrCodeWorkerNonceWindowNotAvailable, "worker nonce window not available")
 	ErrNoInferencesFoundForTopic      = errorsmod.Register(ErrorCodespace, ErrCodeNoInferencesFoundForTopic, "no inferences found for topic")
+	ErrGRPCTransport                  = errorsmod.Register(ErrorCodespace, ErrCodeGRPCTransport, "grpc transport failure")
 )
 
 // Errors substrings that are not ABCI errors and do not have a specific error code
@@ -83,6 +85,11 @@ const ErrorMessageNotPermittedToAddStake = "not permitted to add stake"
 const ErrorMessageReadFlatPanic = "{ReadFlat}: panic"
 const ErrorMessageReadPerBytePanic = "{ReadPerByte}: panic"
 const ErrorMessageConnectionRefused = "connection refused"
+const ErrorMessageConnectionReset = "connection reset by peer"
+const ErrorMessageConnectionTimedOut = "read: connection timed out"
+const ErrorMessageGRPCUnavailableTransport = "code = Unavailable desc ="
+const ErrorMessageReadingFromServer = "error reading from server"
+const ErrorMessageGRPCEOF = "code = Unavailable desc = unexpected EOF"
 const ErrorMessageNoInferencesFoundForTopic = "no inferences found for topic"
 const ErrorContextDeadlineExceeded = "context deadline exceeded"
 const ErrorReputerNonceWindowNotAvailable = "reputer nonce window not available"
@@ -303,6 +310,17 @@ func triageStringMatchingError(ctx context.Context, err error, infoMsg string, n
 		log.Warn().Err(err).Str("rpc", node.ServerAddress).Str("msg", infoMsg).Msg("Connection refused, switching to next node")
 		metrics.GetMetrics().IncrementMetricsCounterWithLabels(metrics.ActorTxErrorCount, node.ConnectionManager.wallet.Address, strconv.Itoa(ErrCodeConnectionRefused))
 		return ErrorProcessingSwitchingNode, ErrConnectionRefused
+	} else if isGRPCTransportError(err) {
+		// Catches mid-stream gRPC transport failures that don't carry a parseable
+		// HTTP status: connection-reset-by-peer, "unexpected EOF", and the broad
+		// "code = Unavailable desc = ... transport:" family. These typically come
+		// from a CDN/load balancer in front of the chain gRPC endpoint half-closing
+		// long-lived HTTP/2 streams. Without an explicit handler these previously
+		// fell through to the info-level catch-all and the node-switching path
+		// (which is what actually forces a fresh dial) never fired.
+		log.Warn().Err(err).Str("rpc", node.ServerAddress).Str("msg", infoMsg).Msg("gRPC transport failure, switching to next node")
+		metrics.GetMetrics().IncrementMetricsCounterWithLabels(metrics.ActorTxErrorCount, node.ConnectionManager.wallet.Address, strconv.Itoa(ErrCodeGRPCTransport))
+		return ErrorProcessingSwitchingNode, ErrGRPCTransport
 	} else if strings.Contains(err.Error(), ErrorReputerNonceWindowNotAvailable) {
 		metrics.GetMetrics().IncrementMetricsCounterWithLabels(metrics.ActorTxErrorCount, node.ConnectionManager.wallet.Address, strconv.Itoa(ErrCodeReputerNonceWindowNotAvailable))
 		log.Warn().
@@ -326,7 +344,7 @@ func triageStringMatchingError(ctx context.Context, err error, infoMsg string, n
 		}
 		return ErrorProcessingContinue, nil
 	}
-	log.Info().Err(err).Str("rpc", node.ServerAddress).Str("msg", infoMsg).Msg("Unknown error")
+	log.Error().Err(err).Str("rpc", node.ServerAddress).Str("msg", infoMsg).Msg("Unknown error - no specific handler matched")
 	metrics.GetMetrics().IncrementMetricsCounterWithLabels(metrics.ActorTxErrorCount, node.ConnectionManager.wallet.Address, strconv.Itoa(ErrCodeUnexpectedError))
 	return ErrorProcessingError, errorsmod.Wrap(ErrUnexpectedError, err.Error())
 }
@@ -356,10 +374,23 @@ func triageHTTPStatusError(err error, node *NodeConfig, infoMsg string) (string,
 	return "", nil
 }
 
-// ParseHTTPStatus extracts HTTP status code and message from an error string
+// ParseHTTPStatus extracts HTTP status code and message from an error string.
+//
+// Recognizes two phrasings observed in the wild:
+//
+//  1. Legacy / HTTP-client style:  "Status: 404 Not Found"
+//  2. gRPC transport style:        "rpc error: code = Unavailable desc =
+//     unexpected HTTP status code received from server: 502 (Bad Gateway); ..."
+//
+// The gRPC phrasing matters because the chain endpoints are typically fronted by
+// a CDN/load balancer (e.g. Cloudflare), and transient 502/503/504s from the
+// edge surface through grpc-go in exactly the second form. Without matching it
+// here, ProcessErrorTx falls through to the generic catch-all which logs at
+// info-level and never triggers node switching, producing silent stalls.
 func ParseHTTPStatus(input string) (int, string, error) {
-	// Updated regex to be less greedy and handle the standard HTTP status format
-	re := regexp.MustCompile(`(?i)Status:\s*(\d+)(?:\s+([^-]+))?`)
+	// Match either "Status: NNN [Reason]" or
+	// "HTTP status code received from server: NNN [(Reason)]"
+	re := regexp.MustCompile(`(?i)(?:Status:|HTTP status code received from server:)\s*(\d+)(?:\s+\(?([^);,\n]+?)\)?(?:[);,\n]|$))?`)
 	matches := re.FindStringSubmatch(input)
 
 	if len(matches) < 2 {
@@ -380,12 +411,37 @@ func ParseHTTPStatus(input string) (int, string, error) {
 	return code, message, nil
 }
 
+// isGRPCTransportError returns true if err looks like a grpc-go transport-layer
+// failure that warrants treating the connection as dead and forcing a re-dial.
+//
+// The grpc-go client surfaces several flavours of mid-stream failure with no
+// parseable HTTP status code:
+//   - "connection reset by peer" (TCP RST from peer or middlebox)
+//   - "read: connection timed out" (idle close)
+//   - "code = Unavailable desc = unexpected EOF" (half-closed HTTP/2 stream)
+//   - "code = Unavailable desc = error reading from server: ..." (generic read failure)
+//
+// All of these mean the client's persistent grpc.ClientConn is in a state where
+// the next call will fail the same way until the connection is replaced.
+func isGRPCTransportError(err error) bool {
+	if err == nil {
+		return false
+	}
+	s := err.Error()
+	return strings.Contains(s, ErrorMessageConnectionReset) ||
+		strings.Contains(s, ErrorMessageConnectionTimedOut) ||
+		strings.Contains(s, ErrorMessageGRPCEOF) ||
+		(strings.Contains(s, ErrorMessageGRPCUnavailableTransport) &&
+			strings.Contains(s, ErrorMessageReadingFromServer))
+}
+
 // Returns true if the error is a switching-node error
 func IsErrorSwitchingNode(err error) bool {
 	return errors.Is(err, ErrHTTP) ||
 		errors.Is(err, ErrFullMempool) ||
 		errors.Is(err, ErrReadPanic) ||
 		errors.Is(err, ErrConnectionRefused) ||
+		errors.Is(err, ErrGRPCTransport) ||
 		errors.Is(err, ErrUnexpectedError)
 }