[Batch Requests] Introduce handling for batch requests for EVM services (#388)

## 🌿 Summary

Introduce support for handling batch requests in EVM services, including
parallel processing and proper response correlation.

### 🌱 Primary Changes:
- Implement batch request handling in EVM services with proper JSON-RPC
2.0 spec compliance
- Add parallel processing for batch requests to improve performance
- Update data pipeline to handle multiple records for batch requests

### 🍃 Secondary changes:
- Refactor protocol interfaces to support batch payloads
- Add new metrics field to track batch requests
- Improve error handling and logging for batch scenarios


## 💡 New TODOs 

New TODOs introduced in this PR:
- TODO_TECHDEBT(@commoddity, @adshmh): add special handling for Cosmos
SDK observations. - data/legacy_qos.go

## 🛠️ Type of change

Select one or more from the following:

- [x] New feature, functionality or library
- [ ] Bug fix
- [ ] Code health or cleanup
- [ ] Documentation
- [ ] Other (specify)

## 🤯 Sanity Checklist

- [ ] I have updated the GitHub Issue 'assignees', 'reviewers',
'labels', 'project', 'iteration' and 'milestone'
- [ ] For docs, I have run 'make docusaurus_start'
- [x] For code, I have run 'make test_all'
- [ ] For configurations, I have update the documentation
- [ ] I added TODOs where applicable

Author: commoddity

data/legacy.go

@@ -39,49 +39,56 @@ type legacyRecord struct {
 	endpointTripTime float64 // endpoint response timestamp - endpoint query time, in seconds.
 }
 
-// converts a data record to legacy format, for compatibility with the existing data pipeline.
-func buildLegacyDataRecord(
+// converts data records to legacy format, for compatibility with the existing data pipeline.
+// Returns multiple legacy records for scenarios like EVM batch requests where each method
+// needs its own record.
+func buildLegacyDataRecords(
 	logger polylog.Logger,
 	observations *observation.RequestResponseObservations,
-) *legacyRecord {
-	// initialize the legacy-compatible data record.
-	legacyRecord := &legacyRecord{}
+) []*legacyRecord {
+	// initialize the base legacy-compatible data record.
+	baseLegacyRecord := &legacyRecord{}
 
 	// Update the legacy data record from Gateway observations.
 	if gatewayObservations := observations.GetGateway(); gatewayObservations != nil {
-		legacyRecord = setLegacyFieldsFromGatewayObservations(logger, legacyRecord, gatewayObservations)
+		baseLegacyRecord = setLegacyFieldsFromGatewayObservations(logger, baseLegacyRecord, gatewayObservations)
 	}
 
 	// TODO_MVP(@adshmh): Set legacy fields from Shannon observations.
 	//
 	// Extract protocol observations
 	protocolObservations := observations.GetProtocol()
 
-	// Update the data record from Shannonprotocol data
+	// Update the data record from Shannon protocol data
 	if shannonObservations := protocolObservations.GetShannon(); shannonObservations != nil {
-		legacyRecord = setLegacyFieldsFromShannonProtocolObservations(logger, legacyRecord, shannonObservations)
+		baseLegacyRecord = setLegacyFieldsFromShannonProtocolObservations(logger, baseLegacyRecord, shannonObservations)
 	}
 
-	// Update the legacy data record from QoS observations.
+	// Update the legacy data records from QoS observations.
+	// This may return multiple records for EVM batch requests.
+	var legacyRecords []*legacyRecord
 	if qosObservations := observations.GetQos(); qosObservations != nil {
-		legacyRecord = setLegacyFieldsFromQoSObservations(logger, legacyRecord, qosObservations)
+		legacyRecords = setLegacyFieldsFromQoSObservations(logger, baseLegacyRecord, qosObservations)
+	} else {
+		legacyRecords = []*legacyRecord{baseLegacyRecord}
 	}
 
-	// Set constant/calculated/inferred fields' values.
-	//
-	if legacyRecord.ErrorType != "" {
-		// Redundant value, set to comply with the legacy data pipeline.
-		legacyRecord.IsError = true
-	}
+	// Set constant/calculated/inferred fields' values for all records.
+	for _, legacyRecord := range legacyRecords {
+		if legacyRecord.ErrorType != "" {
+			// Redundant value, set to comply with the legacy data pipeline.
+			legacyRecord.IsError = true
+		}
 
-	// Hardcoded to "PATH" to inform the legacy data pipeline.
-	legacyRecord.ErrorSource = "PATH"
+		// Hardcoded to "PATH" to inform the legacy data pipeline.
+		legacyRecord.ErrorSource = "PATH"
 
-	// Time spent waiting for the endpoint's response, in seconds.
-	legacyRecord.NodeTripTime = legacyRecord.endpointTripTime
+		// Time spent waiting for the endpoint's response, in seconds.
+		legacyRecord.NodeTripTime = legacyRecord.endpointTripTime
 
-	// Total request processing time - time spent waiting for the endpoint, measured in seconds.
-	legacyRecord.PortalTripTime = legacyRecord.RequestRoundTripTime - legacyRecord.endpointTripTime
+		// Total request processing time - time spent waiting for the endpoint, measured in seconds.
+		legacyRecord.PortalTripTime = legacyRecord.RequestRoundTripTime - legacyRecord.endpointTripTime
+	}
 
-	return legacyRecord
+	return legacyRecords
 }

data/legacy_qos.go

@@ -12,73 +12,110 @@ import (
 	qosobservation "github.com/buildwithgrove/path/observation/qos"
 )
 
-// setLegacyFieldsFromQoSObservations populates legacy record with QoS observation data.
+// setLegacyFieldsFromQoSObservations populates legacy records with QoS observation data.
 // Currently supports:
-// - EVM observations
-// Future support planned for Solana and CometBFT
+// - EVM observations (returns multiple records based on RequestObservations)
+// - Solana observations (returns single record)
+// Future support planned for Cosmos SDK
 //
 // Parameters:
 // - logger: logging interface
-// - legacyRecord: the record to populate
+// - baseLegacyRecord: the base record to populate
 // - observations: QoS observations data
-// Returns: the populated legacy record
+//
+// Returns: slice of populated legacy records
 func setLegacyFieldsFromQoSObservations(
 	logger polylog.Logger,
-	legacyRecord *legacyRecord,
+	baseLegacyRecord *legacyRecord,
 	observations *qosobservation.Observations,
-) *legacyRecord {
+) []*legacyRecord {
+	// EVM observations may contains multiple records in the case of batch requests.
 	if evmObservations := observations.GetEvm(); evmObservations != nil {
-		return setLegacyFieldsFromQoSEVMObservations(logger, legacyRecord, evmObservations)
+		return setLegacyFieldsFromQoSEVMObservations(logger, baseLegacyRecord, evmObservations)
 	}
 
 	// Use Solana observations to update the legacy record's fields.
 	if solanaObservations := observations.GetSolana(); solanaObservations != nil {
-		return setLegacyFieldsFromQoSSolanaObservations(logger, legacyRecord, solanaObservations)
+		populatedRecord := setLegacyFieldsFromQoSSolanaObservations(logger, baseLegacyRecord, solanaObservations)
+		// Solana does not support batch requests so expect a single record.
+		return []*legacyRecord{populatedRecord}
 	}
 
-	return legacyRecord
+	// For all other services, expect a single record.
+	// TODO_TECHDEBT(@commoddity, @adshmh): add special handling for Cosmos SDK observations.
+	return []*legacyRecord{baseLegacyRecord}
 }
 
 // qosEVMErrorTypeStr defines the prefix for EVM QoS error types in legacy records
 const qosEVMErrorTypeStr = "QOS_EVM"
 
-// setLegacyFieldsFromQoSEVMObservations populates legacy record with EVM-specific QoS data.
+// setLegacyFieldsFromQoSEVMObservations populates legacy records with EVM-specific QoS data.
 // It captures:
 // - Request payload size
 // - JSONRPC method information
 // - Error details (when applicable)
+// Creates one legacy record per RequestObservation
 //
 // Parameters:
 // - logger: logging interface
-// - legacyRecord: the record to populate
+// - baseLegacyRecord: the base record to copy for each method
 // - observations: EVM-specific QoS observations
-// Returns: the populated legacy record
+//
+// Returns: slice of populated legacy records
+// EVM batch requests are supported as of PR #388.
 func setLegacyFieldsFromQoSEVMObservations(
-	logger polylog.Logger,
-	legacyRecord *legacyRecord,
+	_ polylog.Logger,
+	baseLegacyRecord *legacyRecord,
 	observations *qosobservation.EVMRequestObservations,
-) *legacyRecord {
-	// In bytes: the length of the request: float64 type is for compatibility with the legacy data pipeline.
-	legacyRecord.RequestDataSize = float64(observations.RequestPayloadLength)
+) []*legacyRecord {
+	// Set common fields from observations
+	baseLegacyRecord.RequestDataSize = float64(observations.RequestPayloadLength)
 
 	evmInterpreter := &qosobservation.EVMObservationInterpreter{
 		Observations: observations,
 	}
 
-	// Extract the JSONRPC request's method.
-	jsonrpcRequestMethod, _ := evmInterpreter.GetRequestMethod()
-	legacyRecord.ChainMethod = jsonrpcRequestMethod
+	// Extract all JSONRPC request methods
+	jsonrpcRequestMethods, ok := evmInterpreter.GetRequestMethods()
+	if !ok || len(jsonrpcRequestMethods) == 0 {
+		// If no methods found, return single record with base data
+		populateEVMErrorFields(baseLegacyRecord, evmInterpreter)
+		return []*legacyRecord{baseLegacyRecord}
+	}
+
+	// Create a separate legacy record for each method
+	// 	 - In the case of EVM batch requests, this will create multiple records.
+	// 	 - Non-EVM batch requests will create a single record.
+	var legacyRecords []*legacyRecord
+	for _, method := range jsonrpcRequestMethods {
+		// Create a copy of the base record
+		recordCopy := *baseLegacyRecord
+		legacyRecord := &recordCopy
+
+		// Set the method for this record
+		legacyRecord.ChainMethod = method
+
+		// Populate error fields if needed
+		populateEVMErrorFields(legacyRecord, evmInterpreter)
+
+		legacyRecords = append(legacyRecords, legacyRecord)
+	}
 
+	return legacyRecords
+}
+
+// populateEVMErrorFields sets error-related fields in the legacy record based on QoS observations
+func populateEVMErrorFields(legacyRecord *legacyRecord, evmInterpreter *qosobservation.EVMObservationInterpreter) {
 	// ErrorType is already set at gateway or protocol level.
 	// Skip updating the error fields to preserve the original error.
 	if legacyRecord.ErrorType != "" {
-		return legacyRecord
+		return
 	}
 
 	_, requestErr, err := evmInterpreter.GetRequestStatus()
 	// Could not extract request error details, skip the rest of the updates.
 	if err != nil || requestErr == nil {
-		return legacyRecord
+		return
 	}
 
 	legacyRecord.ErrorMessage = requestErr.String()
@@ -91,8 +128,6 @@ func setLegacyFieldsFromQoSEVMObservations(
 	default:
 		legacyRecord.ErrorType = fmt.Sprintf("%s_UNKNOWN_ERROR", qosEVMErrorTypeStr)
 	}
-
-	return legacyRecord
 }
 
 // setLegacyFieldsFromQoSSolanaObservations populates legacy record with Solana-specific QoS data.

data/reporter_http.go

@@ -16,7 +16,7 @@ import (
 // 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)
+// DataReporterHTTP exports observations to an external components over HTTP (e.g. Fluentd HTTP Plugin, a Messaging system, or a database)
 var _ gateway.RequestResponseReporter = &DataReporterHTTP{}
 
 // DataReporterHTTP sends the observation for each handled request to an HTTP endpoint.
@@ -37,25 +37,34 @@ type DataReporterHTTP struct {
 }
 
 // Publish the supplied observations:
-// - Build the expected data record.
-// - Send to the configured URL.
+// - Build the expected data records.
+// - Send each record to the configured URL.
 func (drh *DataReporterHTTP) Publish(observations *observation.RequestResponseObservations) {
 	logger := drh.hydrateLogger(observations)
 
 	// TODO_MVP(@adshmh): Replace this with the new DataRecord struct once the data pipeline is updated.
-	// convert to legacy-formatted data record
-	legacyDataRecord := buildLegacyDataRecord(logger, observations)
-
-	// Marshal the data record.
-	serializedRecord, err := json.Marshal(legacyDataRecord)
-	if err != nil {
-		logger.Warn().Err(err).Msg("Failed to serialize the data record. Skip reporting.")
-		return
-	}
-
-	// Send the marshaled data record to the data processor.
-	if err := drh.sendRecordOverHTTP(serializedRecord); err != nil {
-		logger.Warn().Err(err).Msg("Failed to send the data record over HTTP. Skip reporting.")
+	// convert to legacy-formatted data records (may be multiple for EVM batch requests)
+	legacyDataRecords := buildLegacyDataRecords(logger, observations)
+
+	// Process each legacy data record as a single relay for data pipeline and metering purposes.
+	//
+	// If the observations are for an EVM batch request, legacyDataRecords will contain multiple records.
+	// As of PR #388 all other QoS observations are expected to be single records.
+	// Reference: https://github.com/buildwithgrove/path/pull/388
+	for i, legacyDataRecord := range legacyDataRecords {
+		recordLogger := logger.With("record_index", i, "total_records", len(legacyDataRecords))
+
+		// Marshal the data record.
+		serializedRecord, err := json.Marshal(legacyDataRecord)
+		if err != nil {
+			recordLogger.Warn().Err(err).Msg("Failed to serialize the data record. Skip reporting this record.")
+			continue
+		}
+
+		// Send the marshaled data record to the data processor.
+		if err := drh.sendRecordOverHTTP(serializedRecord); err != nil {
+			recordLogger.Warn().Err(err).Msg("Failed to send the data record over HTTP. Skip reporting this record.")
+		}
 	}
 }
 

docusaurus/package.json

@@ -14,9 +14,9 @@
     "write-heading-ids": "docusaurus write-heading-ids"
   },
   "dependencies": {
-    "@docusaurus/core": "^3.8.0",
-    "@docusaurus/preset-classic": "^3.8.0",
-    "@docusaurus/theme-mermaid": "^3.8.0",
+    "@docusaurus/core": "^3.8.1",
+    "@docusaurus/preset-classic": "^3.8.1",
+    "@docusaurus/theme-mermaid": "^3.8.1",
     "@easyops-cn/docusaurus-search-local": "^0.46.1",
     "@mdx-js/react": "^3.0.0",
     "clsx": "^2.0.0",
@@ -31,8 +31,8 @@
     "remark-mermaid-plugin": "^1.0.2"
   },
   "devDependencies": {
-    "@docusaurus/module-type-aliases": "^3.8.0",
-    "@docusaurus/types": "^3.7.0"
+    "@docusaurus/module-type-aliases": "^3.8.1",
+    "@docusaurus/types": "^3.8.1"
   },
   "browserslist": {
     "production": [