[WebSockets] Clarify WebSockets implementation package responsibilities and add Observations to Metrics and Data (#386)

## 🌿 Summary

Clarify WebSockets implementation package responsibilities and add
Observations to Metrics and Data

### 🌱 Primary Changes:
- Add WebSocket connection and message observations to protocol
observations for metrics and data reporting
- Create dedicated `gateway` package websocketRequestContext to handle
WebSocket-specific request processing separate from HTTP
- Implement WebSocket message processing in `gateway` package to
orchestrate `protocol` and `QoS`-level message processing
- Add WebSocket-specific metrics counters and error tracking in Shannon
metrics

### 🍃 Secondary changes:
- Update documentation with WebSocket connection examples using wscat
- Add WebSocket-specific error types to gateway and protocol error
classifications
- Refactor bridge implementation to be protocol-agnostic with message
processor interface
- Rename HTTP-specific request context files for clarity

## 💡 New TODOs 

New TODOs introduced in this PR:
- TODO_TECHDEBT(@adshmh,@commoddity): Use ParseHTTPRequest as the single
entry point to QoS, including for a WebSocket request. -
gateway/websocket_request_context.go
- TODO_TECHDEBT(@commoddity): process message using QoS context and
update the message observations. messageObservations.Qos =
wrc.qosCtx.ProcessProtocolEndpointWebsocketMessage(msgData) -
gateway/websocket_request_context.go
- TODO_NEXT(@commoddity): Introduce correct error classification for
WebSocket errors. - Error creating a WebSocket connection. - Error
signing the relay request. - Error validating the relay response. -
protocol/shannon/sanctions.go
- TODO_IMPROVE(@commoddity,@adshmh): Cleanly separate fallback endpoint
handling from the protocol package. -
protocol/shannon/websocket_context.go

## 🛠️ Type of change

Select one or more from the following:

- [x] New feature, functionality or library

## 🤯 Sanity Checklist

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

---------

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

Author: 3 people

data/legacy_gateway.go

@@ -63,7 +63,9 @@ func setLegacyFieldsFromGatewayObservations(
 	legacyRecord.RequestTimestamp = formatTimestampPbForBigQueryJSON(observations.ReceivedTime)
 
 	// Request processing time, in seconds.
-	legacyRecord.RequestRoundTripTime = float64(observations.CompletedTime.AsTime().Sub(observations.ReceivedTime.AsTime()).Milliseconds())
+	//   - For HTTP requests, this is the round-trip time.
+	//   - For WebSocket requests, this is the total elapsed time the WebSocket connection was open.
+	legacyRecord.RequestRoundTripTime = float64(observations.CompletedTime.AsTime().Sub(observations.ReceivedTime.AsTime()).Milliseconds()) / 1000
 
 	return legacyRecord
 }

data/legacy_protocol_shannon.go

@@ -50,7 +50,36 @@ func setLegacyFieldsFromShannonProtocolObservations(
 		return legacyRecord
 	}
 
-	endpointObservations := observations.GetEndpointObservations()
+	// Handle different observation types based on the oneof field
+	switch obsData := observations.GetObservationData().(type) {
+
+	// HTTP observations
+	case *protocolobservation.ShannonRequestObservations_HttpObservations:
+		return setLegacyFieldsFromHTTPObservations(logger, legacyRecord, obsData.HttpObservations)
+
+	// WebSocket connection observations
+	case *protocolobservation.ShannonRequestObservations_WebsocketConnectionObservation:
+		return setLegacyFieldsFromWebsocketConnectionObservation(logger, legacyRecord, obsData.WebsocketConnectionObservation)
+
+	// WebSocket message observations
+	case *protocolobservation.ShannonRequestObservations_WebsocketMessageObservation:
+		return setLegacyFieldsFromWebsocketMessageObservation(logger, legacyRecord, obsData.WebsocketMessageObservation)
+
+	// Unknown observation type
+	default:
+		logger.Warn().Msg("Unknown observation type received for legacy record processing")
+		return legacyRecord
+	}
+}
+
+// setLegacyFieldsFromHTTPObservations populates legacy record with HTTP endpoint observation data.
+// This handles the original HTTP relay processing logic.
+func setLegacyFieldsFromHTTPObservations(
+	logger polylog.Logger,
+	legacyRecord *legacyRecord,
+	httpObservations *protocolobservation.ShannonHTTPEndpointObservations,
+) *legacyRecord {
+	endpointObservations := httpObservations.GetEndpointObservations()
 	// No endpoint observations: this should not happen as the request has not error set.
 	// Log a warning entry.
 	if len(endpointObservations) == 0 {
@@ -81,19 +110,160 @@ func setLegacyFieldsFromShannonProtocolObservations(
 	// Will be "fallback" in the case of a request sent to a fallback endpoint.
 	legacyRecord.NodeAddress = endpointObservation.GetSupplier()
 
-	// Extract the endpoint's domain from its URL.
+	// Extract and set the endpoint's domain from its URL.
+	// Empty value if parsing the URL above failed.
 	endpointDomain, err := shannonmetrics.ExtractDomainOrHost(endpointObservation.GetEndpointUrl())
 	if err != nil {
 		logger.With("endpoint_url", endpointObservation.EndpointUrl).Warn().Err(err).Msg("Could not extract domain from Shannon endpoint URL")
 		return legacyRecord
 	}
+	legacyRecord.NodeDomain = endpointDomain
 
-	// Set the endpoint domain field: empty value if parsing the URL above failed.
+	return legacyRecord
+}
+
+// setLegacyFieldsFromWebsocketConnectionObservation populates legacy record with WebSocket connection observation data.
+// This handles WebSocket connection lifecycle (not individual messages).
+func setLegacyFieldsFromWebsocketConnectionObservation(
+	logger polylog.Logger,
+	legacyRecord *legacyRecord,
+	wsConnectionObs *protocolobservation.ShannonWebsocketConnectionObservation,
+) *legacyRecord {
+	// Update error fields if a connection error has occurred.
+	legacyRecord = setLegacyErrFieldsFromWebsocketConnectionError(legacyRecord, wsConnectionObs)
+
+	// Set application address
+	legacyRecord.ProtocolAppPublicKey = wsConnectionObs.GetEndpointAppAddress()
+
+	// WebSocket connections don't have separate query/response timestamps at the protocol level.
+	// Connection timing is tracked at the gateway level instead.
+	// Set both timestamps to empty strings to indicate they don't apply.
+	legacyRecord.NodeQueryTimestamp = ""
+	legacyRecord.NodeReceiveTimestamp = ""
+	legacyRecord.endpointTripTime = 0
+
+	// Set endpoint address to the supplier address.
+	legacyRecord.NodeAddress = wsConnectionObs.GetSupplier()
+
+	// Extract and set the endpoint's domain from its URL.
+	// Empty value if parsing the URL above failed.
+	endpointDomain, err := shannonmetrics.ExtractDomainOrHost(wsConnectionObs.GetEndpointUrl())
+	if err != nil {
+		logger.With("endpoint_url", wsConnectionObs.EndpointUrl).Warn().Err(err).Msg("Could not extract domain from WebSocket endpoint URL")
+		return legacyRecord
+	}
 	legacyRecord.NodeDomain = endpointDomain
 
 	return legacyRecord
 }
 
+// setLegacyFieldsFromWebsocketMessageObservation populates legacy record with WebSocket message observation data.
+// This handles individual WebSocket messages sent over an established connection.
+func setLegacyFieldsFromWebsocketMessageObservation(
+	logger polylog.Logger,
+	legacyRecord *legacyRecord,
+	wsMessageObs *protocolobservation.ShannonWebsocketMessageObservation,
+) *legacyRecord {
+	// Update error fields if a message error has occurred.
+	legacyRecord = setLegacyErrFieldsFromWebsocketMessageError(legacyRecord, wsMessageObs)
+
+	// Set application address
+	legacyRecord.ProtocolAppPublicKey = wsMessageObs.GetEndpointAppAddress()
+
+	// WebSocket messages lack separate request/response cycles - timestamps don't apply
+	// Set both timestamps to empty strings as requested by @fredteumer
+	legacyRecord.NodeQueryTimestamp = ""
+	// TODO_REVISIT: Can individual websocket message have a receive timestamp?
+	legacyRecord.NodeReceiveTimestamp = ""
+
+	// WebSocket messages have no request/response latency - set to 0 as it doesn't apply
+	legacyRecord.endpointTripTime = 0
+
+	// Set endpoint address to the supplier address.
+	legacyRecord.NodeAddress = wsMessageObs.GetSupplier()
+
+	// Extract and set the endpoint's domain from its URL.
+	// Empty value if parsing the URL above failed.
+	endpointDomain, err := shannonmetrics.ExtractDomainOrHost(wsMessageObs.GetEndpointUrl())
+	if err != nil {
+		logger.With("endpoint_url", wsMessageObs.EndpointUrl).Warn().Err(err).Msg("Could not extract domain from WebSocket message endpoint URL")
+		return legacyRecord
+	}
+	legacyRecord.NodeDomain = endpointDomain
+
+	// WebSocket messages lack HTTP-style methods and JSON-RPC extraction is QoS-level - using identifier for analytics
+	// TODO_TECHDEBT(@adshmh,@commoddity): When QoS observations for WebSocket messages are added,
+	// use the method from the QoS observations and move this to a new method in the `legacy_qos.go` file.
+	legacyRecord.ChainMethod = "websocket_message"
+
+	// Using MessagePayloadSize as closest equivalent to HTTP request size for bandwidth analytics
+	legacyRecord.RequestDataSize = float64(wsMessageObs.GetMessagePayloadSize())
+
+	return legacyRecord
+}
+
+// setLegacyErrFieldsFromWebsocketConnectionError populates error fields in legacy record from WebSocket connection error data.
+func setLegacyErrFieldsFromWebsocketConnectionError(
+	legacyRecord *legacyRecord,
+	wsConnectionObs *protocolobservation.ShannonWebsocketConnectionObservation,
+) *legacyRecord {
+	endpointErr := wsConnectionObs.ErrorType
+	// No endpoint error has occurred: no error processing required.
+	if endpointErr == nil {
+		return legacyRecord
+	}
+
+	// Update ErrorType using the observed endpoint error.
+	legacyRecord.ErrorType = endpointErr.String()
+
+	// Build the endpoint error details, including any sanctions.
+	var errMsg string
+	if errDetails := wsConnectionObs.GetErrorDetails(); errDetails != "" {
+		errMsg = fmt.Sprintf("error details: %s", errDetails)
+	}
+
+	// Add the sanction details to the error message.
+	if endpointSanction := wsConnectionObs.RecommendedSanction; endpointSanction != nil {
+		errMsg = fmt.Sprintf("%s, sanction: %s", errMsg, endpointSanction.String())
+	}
+
+	// Set the error message field.
+	legacyRecord.ErrorMessage = errMsg
+
+	return legacyRecord
+}
+
+// setLegacyErrFieldsFromWebsocketMessageError populates error fields in legacy record from WebSocket message error data.
+func setLegacyErrFieldsFromWebsocketMessageError(
+	legacyRecord *legacyRecord,
+	wsMessageObs *protocolobservation.ShannonWebsocketMessageObservation,
+) *legacyRecord {
+	endpointErr := wsMessageObs.ErrorType
+	// No endpoint error has occurred: no error processing required.
+	if endpointErr == nil {
+		return legacyRecord
+	}
+
+	// Update ErrorType using the observed endpoint error.
+	legacyRecord.ErrorType = endpointErr.String()
+
+	// Build the endpoint error details, including any sanctions.
+	var errMsg string
+	if errDetails := wsMessageObs.GetErrorDetails(); errDetails != "" {
+		errMsg = fmt.Sprintf("error details: %s", errDetails)
+	}
+
+	// Add the sanction details to the error message.
+	if endpointSanction := wsMessageObs.RecommendedSanction; endpointSanction != nil {
+		errMsg = fmt.Sprintf("%s, sanction: %s", errMsg, endpointSanction.String())
+	}
+
+	// Set the error message field.
+	legacyRecord.ErrorMessage = errMsg
+
+	return legacyRecord
+}
+
 // setLegacyErrFieldsFromShannonEndpointError populates error fields in legacy record from endpoint error data.
 // It handles:
 // - Error type mapping

docusaurus/docs/develop/path/2_cheatsheet_shannon.md

@@ -47,6 +47,7 @@ Skip to [2.1 Generate Shannon Config](#21-generate-shannon-config).
   - [3.2 Check configured services](#32-check-configured-services)
 - [4. Test Relays](#4-test-relays)
   - [Test Relay with `curl`](#test-relay-with-curl)
+  - [Test WebSockets with `wscat`](#test-websockets-with-wscat)
   - [Load Testing with `relay-util`](#load-testing-with-relay-util)
 - [5. Stop PATH](#5-stop-path)
 
@@ -268,6 +269,47 @@ Expected response:
 { "id": 1, "jsonrpc": "2.0", "result": "0x2f01a" }
 ```
 
+### Test WebSockets with `wscat`
+
+:::tip
+
+For `wscat` installation instructions, see [here](https://github.com/ArtiomL/wscat?tab=readme-ov-file#installation).
+
+:::
+
+```bash
+wscat -c ws://localhost:3070/v1 \
+ -H "Authorization: test_api_key" \
+ -H "Target-Service-Id: xrplevm"
+```
+
+Expected terminal prompt:
+
+```bash
+Connected (press CTRL+C to quit)
+>
+```
+
+Sample WebSocket request/response:
+
+```bash
+> {"jsonrpc": "2.0", "id": 1, "method": "eth_blockNumber" }
+< {"id":1,"jsonrpc":"2.0","result":"0x17cbc3"}
+
+> {"jsonrpc": "2.0", "id": 1, "method": "eth_blockNumber" }
+< {"id":1,"jsonrpc":"2.0","result":"0x17cbc4"}
+```
+
+:::info
+
+This is a simple terminal-based WebSocket example and does not contain reconnection logic.
+
+Connections will drop on session rollover, which is expected behavior.
+
+In production environments, you should implement reconnection logic and handle errors gracefully.
+
+:::
+
 ### Load Testing with `relay-util`
 
 Send 100 requests with performance metrics:

e2e/config/e2e_load_test.config.tmpl.yaml

@@ -46,7 +46,7 @@ e2e_load_test_config:
 # Default configuration applied to all test cases (unless overridden)
 default_service_config:
   global_rps: 100 # Requests per second (shared across all methods)
-  requests_per_method: 300 # Number of requests per method
+  requests_per_method: 30 # Number of requests per method
   success_rate: 0.95 # Minimum required success rate (80%)
   max_p50_latency_ms: 10000ms # Max allowed P50 latency (ms)
   max_p95_latency_ms: 20000ms # Max allowed P95 latency (ms)

gateway/errors.go

@@ -14,4 +14,12 @@ var (
 
 	// Error building protocol contexts from HTTP request.
 	errBuildProtocolContextsFromHTTPRequest = errors.New("error building protocol contexts from HTTP request")
+
+	// WebSocket request was rejected by QoS instance.
+	// e.g. WebSocket subscription request validation failed.
+	errWebsocketRequestRejectedByQoS = errors.New("WebSocket request rejected by QoS instance")
+
+	// WebSocket connection establishment failed.
+	// e.g. Failed to upgrade HTTP connection to WebSocket or connect to endpoint.
+	errWebsocketConnectionFailed = errors.New("WebSocket connection establishment failed")
 )