feat(relay): enforce RelayCountOnNodeError and add separate protocol error retries

When quorum is disabled (no lava-quorum-* headers), RelayCountOnNodeError
now acts as a hard limit on the total number of retry batches. Setting
RelayCountOnNodeError=0 effectively disables all retries on node errors.

When quorum is enabled, quorumParams.Max is used as the retry limit instead,
allowing quorum functionality to work independently of RelayCountOnNodeError.

Additionally, introduces a new `--set-retry-count-on-protocol-error` CLI flag
that controls protocol error retries independently from node error retries.

Changes:
- Modified retryCondition() in both smart router and consumer state machines
  to respect RelayCountOnNodeError when quorum is disabled
- Updated HasRequiredNodeResults() to count node errors as "results" when
  RelayCountOnNodeError=0 to prevent quorum logic from triggering retries
- Added HasRequiredNodeResults() return value for protocol error count
- Update retryCondition to apply appropriate limit based on error type:
  - Only node errors: use RelayCountOnNodeError
  - Only protocol errors: use RelayCountOnProtocolError
  - Both: use max of both limits
- Add RelayCountOnProtocolError variable and CLI flags
- Added comprehensive tests for retry limit behavior

This allows users to disable node error retries (--set-retry-count-on-node-error=0)
while still allowing protocol error retries for transient connection issues.

Author: nimrod-teich

protocol/chainlib/chainproxy/rpcInterfaceMessages/jsonRPCMessage.go

@@ -15,11 +15,6 @@ import (
 
 var ErrFailedToConvertMessage = sdkerrors.New("RPC error", 1000, "failed to convert a message")
 
-// BatchNodeErrorOnAny controls batch request error detection:
-// - false (default): batch is an error only if ALL sub-requests failed
-// - true: batch is an error if ANY sub-request failed (strict mode)
-var BatchNodeErrorOnAny = false
-
 type JsonrpcMessage struct {
 	Version                string               `json:"jsonrpc,omitempty"`
 	ID                     json.RawMessage      `json:"id,omitempty"`
@@ -233,48 +228,28 @@ func NewBatchMessage(msgs []JsonrpcMessage) (JsonrpcBatchMessage, error) {
 }
 
 // returns if error exists and
-// Behavior controlled by BatchNodeErrorOnAny flag:
-// - false (default): batch is an error only if ALL sub-requests failed
-// - true: batch is an error if ANY sub-request failed (strict mode)
 func CheckResponseErrorForJsonRpcBatch(data []byte, httpStatusCode int) (hasError bool, errorMessage string) {
-	// Use a temporary struct that checks for both error and result presence
+	// Use a temporary struct that omits the Result field to avoid allocating memory for large results
 	var result []struct {
-		Error  *rpcclient.JsonError `json:"error,omitempty"`
-		Result json.RawMessage      `json:"result,omitempty"`
+		Error *rpcclient.JsonError `json:"error,omitempty"`
 	}
 	err := json.Unmarshal(data, &result)
 	if err != nil {
 		utils.LavaFormatWarning("Failed unmarshalling CheckError", err, utils.LogAttr("data", string(data)))
 		return false, ""
 	}
-
-	hasAnySuccess := false
-	hasAnyError := false
-	aggregatedErrors := ""
-
-	for _, batchResult := range result {
-		if batchResult.Error == nil && len(batchResult.Result) > 0 {
-			hasAnySuccess = true
+	aggregatedResults := ""
+	numberOfBatchElements := len(result)
+	for idx, batchResult := range result {
+		if batchResult.Error == nil {
+			continue
 		}
-		if batchResult.Error != nil && batchResult.Error.Message != "" {
-			hasAnyError = true
-			if aggregatedErrors != "" {
-				aggregatedErrors += ",-," // add a unique comma separator between results
+		if batchResult.Error.Message != "" {
+			aggregatedResults += batchResult.Error.Message
+			if idx < numberOfBatchElements-1 {
+				aggregatedResults += ",-," // add a unique comma separator between results
 			}
-			aggregatedErrors += batchResult.Error.Message
 		}
 	}
-
-	// BatchNodeErrorOnAny=true (strict): error if ANY sub-request failed
-	// BatchNodeErrorOnAny=false (default): error only if ALL sub-requests failed
-	if BatchNodeErrorOnAny {
-		// Strict mode: any error means batch failed
-		return hasAnyError, aggregatedErrors
-	}
-
-	// Default mode: only error if no successes at all
-	if hasAnySuccess {
-		return false, ""
-	}
-	return aggregatedErrors != "", aggregatedErrors
+	return aggregatedResults != "", aggregatedResults
 }

protocol/chainlib/chainproxy/rpcInterfaceMessages/jsonRPCMessage_test.go

@@ -151,135 +151,3 @@ func TestParseJsonRPCBatch(t *testing.T) {
 		}
 	}
 }
-
-func TestCheckResponseErrorForJsonRpcBatch(t *testing.T) {
-	t.Run("all_success_no_error", func(t *testing.T) {
-		// All sub-requests succeeded
-		data := []byte(`[
-			{"jsonrpc":"2.0","id":1,"result":{"blockNumber":"0x123"}},
-			{"jsonrpc":"2.0","id":2,"result":{"blockNumber":"0x124"}},
-			{"jsonrpc":"2.0","id":3,"result":{"blockNumber":"0x125"}}
-		]`)
-		hasError, errorMsg := CheckResponseErrorForJsonRpcBatch(data, 200)
-		require.False(t, hasError, "All success should not have error")
-		require.Empty(t, errorMsg)
-	})
-
-	t.Run("all_errors_should_return_error", func(t *testing.T) {
-		// All sub-requests failed
-		data := []byte(`[
-			{"jsonrpc":"2.0","id":1,"error":{"code":-32000,"message":"Slot was skipped"}},
-			{"jsonrpc":"2.0","id":2,"error":{"code":-32000,"message":"Slot was skipped"}},
-			{"jsonrpc":"2.0","id":3,"error":{"code":-32000,"message":"Slot was skipped"}}
-		]`)
-		hasError, errorMsg := CheckResponseErrorForJsonRpcBatch(data, 200)
-		require.True(t, hasError, "All errors should return error")
-		require.Contains(t, errorMsg, "Slot was skipped")
-	})
-
-	t.Run("partial_success_should_not_error", func(t *testing.T) {
-		// Some sub-requests succeeded, some failed - should NOT be considered an error
-		data := []byte(`[
-			{"jsonrpc":"2.0","id":1,"result":{"blockNumber":"0x123"}},
-			{"jsonrpc":"2.0","id":2,"error":{"code":-32000,"message":"Slot was skipped"}},
-			{"jsonrpc":"2.0","id":3,"result":{"blockNumber":"0x125"}}
-		]`)
-		hasError, errorMsg := CheckResponseErrorForJsonRpcBatch(data, 200)
-		require.False(t, hasError, "Partial success should not be considered an error")
-		require.Empty(t, errorMsg)
-	})
-
-	t.Run("single_success_among_errors_should_not_error", func(t *testing.T) {
-		// Only one success among multiple errors - should NOT be considered an error
-		data := []byte(`[
-			{"jsonrpc":"2.0","id":1,"error":{"code":-32000,"message":"Slot was skipped"}},
-			{"jsonrpc":"2.0","id":2,"error":{"code":-32000,"message":"Slot was skipped"}},
-			{"jsonrpc":"2.0","id":3,"result":{"blockNumber":"0x125"}}
-		]`)
-		hasError, errorMsg := CheckResponseErrorForJsonRpcBatch(data, 200)
-		require.False(t, hasError, "Single success should prevent error classification")
-		require.Empty(t, errorMsg)
-	})
-
-	t.Run("empty_batch_no_error", func(t *testing.T) {
-		data := []byte(`[]`)
-		hasError, errorMsg := CheckResponseErrorForJsonRpcBatch(data, 200)
-		require.False(t, hasError)
-		require.Empty(t, errorMsg)
-	})
-
-	t.Run("invalid_json_no_error", func(t *testing.T) {
-		data := []byte(`not valid json`)
-		hasError, errorMsg := CheckResponseErrorForJsonRpcBatch(data, 200)
-		require.False(t, hasError, "Invalid JSON should not return error (fail-open)")
-		require.Empty(t, errorMsg)
-	})
-
-	t.Run("null_result_with_no_error_is_success", func(t *testing.T) {
-		// null result without error is still a valid response (API returned successfully)
-		data := []byte(`[
-			{"jsonrpc":"2.0","id":1,"result":null},
-			{"jsonrpc":"2.0","id":2,"error":{"code":-32000,"message":"Some error"}}
-		]`)
-		// null result is a valid JSON-RPC response - it means the method executed successfully
-		// and returned null. This counts as a success.
-		hasError, _ := CheckResponseErrorForJsonRpcBatch(data, 200)
-		require.False(t, hasError, "null result is a valid success response")
-	})
-
-	t.Run("real_world_solana_batch_partial_error", func(t *testing.T) {
-		// Real-world scenario: Solana batch request where some slots are skipped
-		data := []byte(`[
-			{"jsonrpc":"2.0","id":1,"result":{"slot":123,"blockhash":"abc"}},
-			{"jsonrpc":"2.0","id":2,"error":{"code":-32007,"message":"Slot 124 was skipped, or missing in long-term storage"}},
-			{"jsonrpc":"2.0","id":3,"result":{"slot":125,"blockhash":"def"}}
-		]`)
-		hasError, errorMsg := CheckResponseErrorForJsonRpcBatch(data, 200)
-		require.False(t, hasError, "Solana batch with partial skipped slots should not trigger retry")
-		require.Empty(t, errorMsg)
-	})
-}
-
-func TestCheckResponseErrorForJsonRpcBatch_StrictMode(t *testing.T) {
-	// Save and restore original flag value
-	originalValue := BatchNodeErrorOnAny
-	defer func() { BatchNodeErrorOnAny = originalValue }()
-
-	t.Run("strict_mode_partial_success_is_error", func(t *testing.T) {
-		BatchNodeErrorOnAny = true // Enable strict mode
-
-		data := []byte(`[
-			{"jsonrpc":"2.0","id":1,"result":{"blockNumber":"0x123"}},
-			{"jsonrpc":"2.0","id":2,"error":{"code":-32000,"message":"Slot was skipped"}},
-			{"jsonrpc":"2.0","id":3,"result":{"blockNumber":"0x125"}}
-		]`)
-		hasError, errorMsg := CheckResponseErrorForJsonRpcBatch(data, 200)
-		require.True(t, hasError, "Strict mode: partial success should be an error")
-		require.Contains(t, errorMsg, "Slot was skipped")
-	})
-
-	t.Run("strict_mode_all_success_no_error", func(t *testing.T) {
-		BatchNodeErrorOnAny = true // Enable strict mode
-
-		data := []byte(`[
-			{"jsonrpc":"2.0","id":1,"result":{"blockNumber":"0x123"}},
-			{"jsonrpc":"2.0","id":2,"result":{"blockNumber":"0x124"}},
-			{"jsonrpc":"2.0","id":3,"result":{"blockNumber":"0x125"}}
-		]`)
-		hasError, errorMsg := CheckResponseErrorForJsonRpcBatch(data, 200)
-		require.False(t, hasError, "Strict mode: all success should not have error")
-		require.Empty(t, errorMsg)
-	})
-
-	t.Run("default_mode_partial_success_no_error", func(t *testing.T) {
-		BatchNodeErrorOnAny = false // Default mode
-
-		data := []byte(`[
-			{"jsonrpc":"2.0","id":1,"result":{"blockNumber":"0x123"}},
-			{"jsonrpc":"2.0","id":2,"error":{"code":-32000,"message":"Slot was skipped"}},
-			{"jsonrpc":"2.0","id":3,"result":{"blockNumber":"0x125"}}
-		]`)
-		hasError, _ := CheckResponseErrorForJsonRpcBatch(data, 200)
-		require.False(t, hasError, "Default mode: partial success should not be an error")
-	})
-}

protocol/common/cobra_common.go

@@ -30,7 +30,8 @@ const (
 	SharedStateFlag         = "shared-state"
 	// Disable relay retries when we get node errors.
 	// This feature is suppose to help with successful relays in some chains that return node errors on rare race conditions on the serviced chains.
-	SetRelayCountOnNodeErrorFlag = "set-retry-count-on-node-error"
+	SetRelayCountOnNodeErrorFlag     = "set-retry-count-on-node-error"
+	SetRelayCountOnProtocolErrorFlag = "set-retry-count-on-protocol-error"
 	UseStaticSpecFlag            = "use-static-spec" // allows the user to manually load a spec providing a path, this is useful to test spec changes before they hit the blockchain
 	GitHubTokenFlag              = "github-token"    // GitHub personal access token for accessing private repositories and higher API rate limits
 	EpochDurationFlag            = "epoch-duration"  // duration of each epoch for time-based epoch system (standalone mode)

protocol/relaycore/interfaces.go

@@ -25,7 +25,7 @@ type RelayStateMachine interface {
 // ResultsCheckerInf interface for checking results
 type ResultsCheckerInf interface {
 	WaitForResults(ctx context.Context) error
-	HasRequiredNodeResults(tries int) (bool, int)
+	HasRequiredNodeResults(tries int) (bool, int, int) // returns (done, nodeErrors, protocolErrors)
 	GetQuorumParams() common.QuorumParams
 }
 

protocol/relaycore/relay_processor.go

@@ -251,9 +251,9 @@ func (rp *RelayProcessor) shouldRetryRelay(resultsCount int, hashErr error, node
 	return false
 }
 
-func (rp *RelayProcessor) HasRequiredNodeResults(tries int) (bool, int) {
+func (rp *RelayProcessor) HasRequiredNodeResults(tries int) (bool, int, int) {
 	if rp == nil {
-		return false, 0
+		return false, 0, 0
 	}
 	rp.lock.RLock()
 	defer rp.lock.RUnlock()
@@ -293,7 +293,7 @@ func (rp *RelayProcessor) HasRequiredNodeResults(tries int) (bool, int) {
 				utils.LogAttr("currentQourumEqualResults", rp.currentQourumEqualResults),
 			)
 		}
-		return true, nodeErrors
+		return true, nodeErrors, protocolErrors
 	}
 	if rp.selection == Quorum {
 		// We need a quorum of all node results
@@ -319,7 +319,14 @@ func (rp *RelayProcessor) HasRequiredNodeResults(tries int) (bool, int) {
 			}
 		} else {
 			// Quorum feature disabled: check if we have enough results for quorum
-			retryForQuorumNeeded = !(resultsCount >= neededForQuorum)
+			// When RelayCountOnNodeError is 0, treat node errors as "results" for quorum purposes
+			// This ensures that when retries are disabled, we don't retry to replace node errors
+			if RelayCountOnNodeError == 0 {
+				totalNodeResults := resultsCount + nodeErrors
+				retryForQuorumNeeded = !(totalNodeResults >= neededForQuorum)
+			} else {
+				retryForQuorumNeeded = !(resultsCount >= neededForQuorum)
+			}
 		}
 
 		if !retryForQuorumNeeded {
@@ -338,7 +345,7 @@ func (rp *RelayProcessor) HasRequiredNodeResults(tries int) (bool, int) {
 					utils.LogAttr("currentQourumEqualResults", rp.currentQourumEqualResults),
 				)
 			}
-			return !shouldRetry, nodeErrors
+			return !shouldRetry, nodeErrors, protocolErrors
 		}
 	}
 	// on BestResult we want to retry if there is no success
@@ -354,7 +361,7 @@ func (rp *RelayProcessor) HasRequiredNodeResults(tries int) (bool, int) {
 			utils.LogAttr("currentQourumEqualResults", rp.currentQourumEqualResults),
 		)
 	}
-	return false, nodeErrors
+	return false, nodeErrors, protocolErrors
 }
 
 func (rp *RelayProcessor) handleResponse(response *RelayResponse) {