[Session Rollovers] Simplify & Consolidate Timeouts (#389)

Testing script:

```bash
#!/bin/bash

# Script to monitor XRPL EVM testnet block number every 0.2 seconds
# Usage: ./block_monitor.sh [--local]
#   --local: Use localhost endpoint instead of remote

# Parse command line arguments
USE_LOCAL=false
for arg in "$@"; do
    case $arg in
        --local)
            USE_LOCAL=true
            shift
            ;;
        *)
            echo "Unknown option: $arg"
            echo "Usage: $0 [--local]"
            exit 1
            ;;
    esac
done

# Print configuration mode at the beginning
if [ "$USE_LOCAL" = true ]; then
    echo "🔧 MODE: LOCAL"
else
    echo "🌐 MODE: REMOTE"
fi
echo ""

# Set endpoint based on flag
if [ "$USE_LOCAL" = true ]; then
    ENDPOINT="http://localhost:3070/v1"
    HEADERS='-H "Target-Service-Id: xrplevm" -H "Authorization: test_api_key"'
    echo "Starting XRPL EVM testnet block number monitor (LOCAL)..."
else
    ENDPOINT="https://xrpl-evm-testnet.rpc.grove.city/v1/6c5de5ff"
    HEADERS='-H "Content-Type: application/json"'
    echo "Starting XRPL EVM testnet block number monitor (REMOTE)..."
fi

echo "Press Ctrl+C to stop"
echo ""

# Initialize variables for delta calculation
last_request_time=0

while true; do
    # Get current timestamp for display and delta calculation
    current_time=$(date +%s.%N)
    timestamp=$(date '+%Y-%m-%d %H:%M:%S')

    # Calculate delta from last request
    if [ "$last_request_time" != "0" ]; then
        delta=$(echo "$current_time - $last_request_time" | bc -l)
        delta_formatted=$(printf "%.2f" "$delta")

        # Color code based on delta (assuming 1s target interval)
        # Green: <= 1.2s (quick), Yellow: 1.2-2.0s (moderate), Red: > 2.0s (slow)
        if (( $(echo "$delta <= 1" | bc -l) )); then
            delta_color="\033[32m"  # Green
            delta_status="⚡"
        elif (( $(echo "$delta <= 3" | bc -l) )); then
            delta_color="\033[33m"  # Yellow
            delta_status="⏱️"
        else
            delta_color="\033[31m"  # Red
            delta_status="🐌"
        fi
        reset_color="\033[0m"
        delta_display=" ${delta_color}[Δ${delta_formatted}s ${delta_status}]${reset_color}"
    else
        delta_display=""
    fi

    # Make the curl request and capture both response and error
    if [ "$USE_LOCAL" = true ]; then
        response=$(curl -s "$ENDPOINT" \
            -H "Target-Service-Id: xrplevm" \
            -H "Authorization: test_api_key" \
            -d '{"jsonrpc": "2.0", "id": 1, "method": "eth_blockNumber" }' 2>&1)
    else
        response=$(curl -s "$ENDPOINT" \
            -X POST \
            -H "Content-Type: application/json" \
            --data '{"method":"eth_blockNumber","params":[],"id":1,"jsonrpc":"2.0"}' 2>&1)
    fi

    # Check if curl command was successful
    curl_exit_code=$?

    if [ $curl_exit_code -ne 0 ]; then
        echo -e "❌ [$timestamp] ERROR: curl failed with exit code $curl_exit_code: $response$delta_display"
    else
        # Check if response contains an error
        if echo "$response" | grep -q '"error"'; then
            echo -e "❌ [$timestamp] API ERROR: $response$delta_display"
        else
            # Extract block number from response (remove 0x prefix and convert from hex)
            block_hex=$(echo "$response" | grep -o '"result":"0x[^"]*"' | cut -d'"' -f4)

            if [ -z "$block_hex" ]; then
                echo -e "❌ [$timestamp] PARSE ERROR: Could not extract block number from response: $response$delta_display"
            else
                # Convert hex to decimal, handle potential conversion errors
                if block_decimal=$((16#${block_hex#0x})) 2>/dev/null; then
                    echo -e "✅ [$timestamp] Block: $block_decimal (hex: $block_hex)$delta_display"
                else
                    echo -e "❌ [$timestamp] CONVERSION ERROR: Could not convert hex $block_hex to decimal$delta_display"
                fi
            fi
        fi
    fi

    # Update last request time for next iteration
    last_request_time="$current_time"

    # Wait 0.5 seconds
    sleep 0.5
done
```

---------

Co-authored-by: Arash Deshmeh <[email protected]>

Author: Olshansk

config/config_test.go

@@ -45,11 +45,12 @@ func Test_LoadGatewayConfigFromYAML(t *testing.T) {
 					},
 				},
 				Router: RouterConfig{
-					Port:                  defaultPort,
-					MaxRequestHeaderBytes: defaultMaxRequestHeaderBytes,
-					ReadTimeout:           defaultHTTPServerReadTimeout,
-					WriteTimeout:          defaultHTTPServerWriteTimeout,
-					IdleTimeout:           defaultHTTPServerIdleTimeout,
+					Port:                            defaultPort,
+					MaxRequestHeaderBytes:           defaultMaxRequestHeaderBytes,
+					ReadTimeout:                     defaultHTTPServerReadTimeout,
+					WriteTimeout:                    defaultHTTPServerWriteTimeout,
+					IdleTimeout:                     defaultHTTPServerIdleTimeout,
+					SystemOverheadAllowanceDuration: defaultSystemOverheadAllowanceDuration,
 				},
 				Logger: LoggerConfig{
 					Level: defaultLogLevel,
@@ -131,11 +132,12 @@ logger_config:
 					},
 				},
 				Router: RouterConfig{
-					Port:                  defaultPort,
-					MaxRequestHeaderBytes: defaultMaxRequestHeaderBytes,
-					ReadTimeout:           defaultHTTPServerReadTimeout,
-					WriteTimeout:          defaultHTTPServerWriteTimeout,
-					IdleTimeout:           defaultHTTPServerIdleTimeout,
+					Port:                            defaultPort,
+					MaxRequestHeaderBytes:           defaultMaxRequestHeaderBytes,
+					ReadTimeout:                     defaultHTTPServerReadTimeout,
+					WriteTimeout:                    defaultHTTPServerWriteTimeout,
+					IdleTimeout:                     defaultHTTPServerIdleTimeout,
+					SystemOverheadAllowanceDuration: defaultSystemOverheadAllowanceDuration,
 				},
 				Logger: LoggerConfig{
 					Level: "debug",

config/router.go

@@ -1,6 +1,7 @@
 package config
 
 import (
+	"fmt"
 	"time"
 )
 
@@ -14,25 +15,30 @@ const (
 	// defaultMaxRequestHeaderBytes is the default maximum size of the HTTP request header.
 	defaultMaxRequestHeaderBytes = 2 * 1e6 // 2 MB
 
-	// HTTP server's default timeout values.
-	defaultHTTPServerReadTimeout = 10 * time.Second
-	defaultHTTPServerIdleTimeout = 120 * time.Second
+	// Reserve time for system overhead, i.e. time spent on non-business logic operations.
+	// Examples:
+	// - Read HTTP Request body
+	// - Write HTTP Response
+	defaultSystemOverheadAllowanceDuration = 10 * time.Second
 
-	// HTTP request handler's WriteTimeout.
 	// https://pkg.go.dev/net/http#Server
-	defaultHTTPServerWriteTimeout = 20 * time.Second
+	// HTTP server's default timeout values.
+	defaultHTTPServerReadTimeout  = 60 * time.Second
+	defaultHTTPServerWriteTimeout = 120 * time.Second
+	defaultHTTPServerIdleTimeout  = 180 * time.Second
 )
 
 /* --------------------------------- Router Config Struct -------------------------------- */
 
 // RouterConfig contains server configuration settings.
 // See default values above.
 type RouterConfig struct {
-	Port                  int           `yaml:"port"`
-	MaxRequestHeaderBytes int           `yaml:"max_request_header_bytes"`
-	ReadTimeout           time.Duration `yaml:"read_timeout"`
-	WriteTimeout          time.Duration `yaml:"write_timeout"`
-	IdleTimeout           time.Duration `yaml:"idle_timeout"`
+	Port                            int           `yaml:"port"`
+	MaxRequestHeaderBytes           int           `yaml:"max_request_header_bytes"`
+	ReadTimeout                     time.Duration `yaml:"read_timeout"`
+	WriteTimeout                    time.Duration `yaml:"write_timeout"`
+	IdleTimeout                     time.Duration `yaml:"idle_timeout"`
+	SystemOverheadAllowanceDuration time.Duration `yaml:"system_overhead_allowance_duration"`
 }
 
 /* --------------------------------- Router Config Private Helpers -------------------------------- */
@@ -54,4 +60,10 @@ func (c *RouterConfig) hydrateRouterDefaults() {
 	if c.IdleTimeout == 0 {
 		c.IdleTimeout = defaultHTTPServerIdleTimeout
 	}
+	if c.SystemOverheadAllowanceDuration == 0 {
+		c.SystemOverheadAllowanceDuration = defaultSystemOverheadAllowanceDuration
+	}
+	if c.SystemOverheadAllowanceDuration >= c.ReadTimeout || c.SystemOverheadAllowanceDuration >= c.WriteTimeout {
+		panic(fmt.Sprintf("system overhead allowance duration %v must be less than read timeout %v and write timeout %v", c.SystemOverheadAllowanceDuration, c.ReadTimeout, c.WriteTimeout))
+	}
 }

config/router_test.go

@@ -59,11 +59,12 @@ func TestRouterConfig_hydrateRouterDefaults(t *testing.T) {
 			name: "should set all defaults",
 			cfg:  RouterConfig{},
 			want: RouterConfig{
-				Port:                  defaultPort,
-				MaxRequestHeaderBytes: defaultMaxRequestHeaderBytes,
-				ReadTimeout:           defaultHTTPServerReadTimeout,
-				WriteTimeout:          defaultHTTPServerWriteTimeout,
-				IdleTimeout:           defaultHTTPServerIdleTimeout,
+				Port:                            defaultPort,
+				MaxRequestHeaderBytes:           defaultMaxRequestHeaderBytes,
+				ReadTimeout:                     defaultHTTPServerReadTimeout,
+				WriteTimeout:                    defaultHTTPServerWriteTimeout,
+				IdleTimeout:                     defaultHTTPServerIdleTimeout,
+				SystemOverheadAllowanceDuration: defaultSystemOverheadAllowanceDuration,
 			},
 		},
 		{
@@ -72,11 +73,12 @@ func TestRouterConfig_hydrateRouterDefaults(t *testing.T) {
 				Port: 8080,
 			},
 			want: RouterConfig{
-				Port:                  8080,
-				MaxRequestHeaderBytes: defaultMaxRequestHeaderBytes,
-				ReadTimeout:           defaultHTTPServerReadTimeout,
-				WriteTimeout:          defaultHTTPServerWriteTimeout,
-				IdleTimeout:           defaultHTTPServerIdleTimeout,
+				Port:                            8080,
+				MaxRequestHeaderBytes:           defaultMaxRequestHeaderBytes,
+				ReadTimeout:                     defaultHTTPServerReadTimeout,
+				WriteTimeout:                    defaultHTTPServerWriteTimeout,
+				IdleTimeout:                     defaultHTTPServerIdleTimeout,
+				SystemOverheadAllowanceDuration: defaultSystemOverheadAllowanceDuration,
 			},
 		},
 	}

data/reporter_http.go

@@ -13,8 +13,8 @@ import (
 	"github.com/buildwithgrove/path/observation"
 )
 
-// defaultPostTimeoutMS defines the default timeout for HTTP POST operations in milliseconds (10 seconds)
-const defaultPostTimeoutMS = 10_000
+// defaultDataReporterPostTimeoutMillisec defines the default timeout for HTTP POST operations in milliseconds (10 seconds)
+const defaultDataReporterPostTimeoutMillisec = 20_000
 
 // DataReporterHTTP exports observations to an external components over HTTP (e.g. Flentd HTTP Plugin, a Messaging system, or a database)
 var _ gateway.RequestResponseReporter = &DataReporterHTTP{}
@@ -63,7 +63,7 @@ func (drh *DataReporterHTTP) sendRecordOverHTTP(serializedDataRecord []byte) err
 	// Determine the timeout to use
 	timeoutMS := drh.PostTimeoutMS
 	if timeoutMS <= 0 {
-		timeoutMS = defaultPostTimeoutMS // Default timeout
+		timeoutMS = defaultDataReporterPostTimeoutMillisec // Default timeout
 	}
 
 	// Create an HTTP client with the configured timeout

gateway/gateway.go

@@ -60,7 +60,7 @@ type Gateway struct {
 // Reference: https://en.wikipedia.org/wiki/Template_method_pattern
 //
 // TODO_FUTURE: Refactor when adding other protocols (e.g. gRPC):
-//   - Extract generic processing into common method  
+//   - Extract generic processing into common method
 //   - Keep HTTP-specific details separate
 func (g Gateway) HandleServiceRequest(
 	ctx context.Context,

gateway/request_context.go

@@ -32,8 +32,12 @@ const (
 	// - If all endpoints are sanctioned, send parallel requests by default
 	// - Make this configurable at the gateway level yaml config
 	// - Enable parallel requests for gateways that maintain their own backend nodes as a special config
-	maxParallelRequests    = 1
-	parallelRequestTimeout = 30 * time.Second
+	maxParallelRequests = 1
+
+	// RelayRequestTimeout is the timeout for relay requests
+	// TODO_TECHDEBT: Look into whether we can remove this variable altogether and consolidate
+	// it with HTTP level timeouts.
+	RelayRequestTimeout = 60 * time.Second
 )
 
 // requestContext is responsible for performing the steps necessary to complete a service request.

gateway/request_context_handle_request.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"fmt"
 	"strings"
+	"sync"
 	"time"
 
 	"github.com/pokt-network/poktroll/pkg/polylog"
@@ -89,7 +90,8 @@ func (rc *requestContext) handleParallelRelayRequests() error {
 		With("service_id", rc.serviceID)
 	logger.Debug().Msg("Starting parallel relay race")
 
-	ctx, cancel := context.WithTimeout(rc.context, parallelRequestTimeout)
+	// TODO_TECHDEBT: Make sure timed out parallel requests are also sanctioned.
+	ctx, cancel := context.WithTimeout(rc.context, RelayRequestTimeout)
 	defer cancel()
 
 	resultChan := rc.launchParallelRequests(ctx, logger)
@@ -112,33 +114,46 @@ func (rc *requestContext) updateParallelRequestMetrics(metrics *parallelRequestM
 func (rc *requestContext) launchParallelRequests(ctx context.Context, logger polylog.Logger) <-chan parallelRelayResult {
 	resultChan := make(chan parallelRelayResult, len(rc.protocolContexts))
 
+	// Ensures thread-safety of QoS context operations.
+	qosContextMutex := sync.Mutex{}
+
 	for protocolCtxIdx, protocolCtx := range rc.protocolContexts {
-		go rc.executeRelayRequest(ctx, logger, protocolCtx, protocolCtxIdx, resultChan)
+		go rc.executeOneOfParallelRequests(ctx, logger, protocolCtx, protocolCtxIdx, resultChan, &qosContextMutex)
 	}
 
 	return resultChan
 }
 
-// executeRelayRequest handles a single relay request in a goroutine
-func (rc *requestContext) executeRelayRequest(
+// executeOneOfParallelRequests handles a single relay request in a goroutine
+func (rc *requestContext) executeOneOfParallelRequests(
 	ctx context.Context,
 	logger polylog.Logger,
 	protocolCtx ProtocolRequestContext,
 	index int,
 	resultChan chan<- parallelRelayResult,
+	qosContextMutex *sync.Mutex,
 ) {
 	startTime := time.Now()
-	response, err := protocolCtx.HandleServiceRequest(rc.qosCtx.GetServicePayload())
+	endpointResponse, err := protocolCtx.HandleServiceRequest(rc.qosCtx.GetServicePayload())
 	duration := time.Since(startTime)
 
 	result := parallelRelayResult{
-		response:  response,
+		response:  endpointResponse,
 		err:       err,
 		index:     index,
 		duration:  duration,
 		startTime: startTime,
 	}
 
+	if err != nil {
+		// TODO_TECHDEBT(@adshmh): refactor the parallel requests feature:
+		// 1. Ensure parallel requests are handled correctly by the QoS layer: e.g. cannot use the most recent response as best anymore.
+		// 2. Simplify the parallel requests feature: it may be best to fully encapsulate it in the protocol/shannon package.
+		qosContextMutex.Lock()
+		rc.qosCtx.UpdateWithResponse(endpointResponse.EndpointAddr, endpointResponse.Bytes)
+		qosContextMutex.Unlock()
+	}
+
 	select {
 	case resultChan <- result:
 		// Result sent successfully

observation/protocol/shannon.pb.go

@@ -100,9 +100,21 @@ enum ShannonEndpointErrorType {
   // New error types for better classification (added without renumbering existing ones)
   // Error was recognized but specific classification failed.
   // Only general category: HTTP, Raw Payload, General, is known.
+
+  // ** SHANNON_ENDPOINT_ERROR_HTTP_UNKNOWN **
+  // PATH -> HTTP Request -> Endpoint
+  // HTTP Response is NOT received
+  // We cannot move forward with PATH business logic
+  // There is nothing to go to.
   SHANNON_ENDPOINT_ERROR_HTTP_UNKNOWN = 34;
+
   SHANNON_ENDPOINT_ERROR_RAW_PAYLOAD_UNKNOWN = 35;
   SHANNON_ENDPOINT_ERROR_UNKNOWN = 36;
+
+  // The relay request sent to the endpoint via HTTP returned a non 2XX HTTP status code.
+  SHANNON_ENDPOINT_ERROR_HTTP_NON_2XX_STATUS = 37;
+  // The relay request sent to the endpoint via HTTP failed with `context deadline exceeded` error
+  SHANNON_ENDPOINT_ERROR_HTTP_CONTEXT_DEADLINE_EXCEEDED = 38;
 }
 
 // ShannonSanctionType specifies the duration type for endpoint sanctions

proto/path/protocol/shannon.proto

@@ -8,11 +8,10 @@ import (
 // TODO_DOCUMENT(@adshmh): Add more examples (e.g. for RESTful services)
 // TODO_IMPROVE(@adshmh): Use an interface here that returns the serialized form of the request.
 type Payload struct {
-	Data            string
-	Method          string
-	Path            string
-	Headers         map[string]string
-	TimeoutMillisec int
+	Data    string
+	Method  string
+	Path    string
+	Headers map[string]string
 	// RPCType indicates the type of RPC protocol for routing to appropriate backend ports:
 	// - RPCType_REST: Cosmos SDK REST API (typically port 1317)
 	// - RPCType_COMET_BFT: CometBFT RPC (typically port 26657)
@@ -27,11 +26,10 @@ func EmptyErrorPayload() Payload {
 	return Payload{
 		// This is an intentional placeholder to distinguish errors in retrieving payloads
 		// from explicit empty payloads set by PATH.
-		Data:            "PATH_EmptyErrorPayload",
-		Method:          "",
-		Path:            "",
-		Headers:         map[string]string{},
-		TimeoutMillisec: 0,
-		RPCType:         sharedtypes.RPCType_UNKNOWN_RPC,
+		Data:    "PATH_EmptyErrorPayload",
+		Method:  "",
+		Path:    "",
+		Headers: map[string]string{},
+		RPCType: sharedtypes.RPCType_UNKNOWN_RPC,
 	}
 }