[RelayMiner] Add websockets support

Tiltfile

@@ -375,9 +375,10 @@ k8s_resource(
     "validator",
     labels=["pocket_network"],
     port_forwards=[
-        "26657",  # RPC
+        "26657",  # CometBFT JSON-RPC
         "9090",  # the gRPC server address
         "40004",  # use with `dlv` when it's turned on in `localnet_config.yaml`
+        "1317", # CosmosSDK REST API
         # Use with pprof like this: `go tool pprof -http=:3333 http://localhost:6050/debug/pprof/goroutine`
         "6050:6060",
     ],

config.yml

@@ -96,6 +96,8 @@ validators:
           # "high" produces a lot of timeseries.
           # ONLY suitable for small networks such as LocalNet.
           cardinality-level: high
+      api:
+        swagger: true
     config:
       moniker: "validator1"
       consensus:
@@ -208,6 +210,14 @@ genesis:
             # `application1_stake_config.yaml` so that the stake command causes a state change.
             amount: "100000068" # ~100 POKT
             denom: upokt
+        - address: pokt1lqyu4v88vp8tzc86eaqr4lq8rwhssyn6rfwzex
+          delegatee_gateway_addresses:
+            [pokt15vzxjqklzjtlz7lahe8z2dfe9nm5vxwwmscne4]
+          service_configs:
+            - service_id: anvilws
+          stake:
+            amount: "100000068" # ~100 POKT
+            denom: upokt
     supplier:
       params:
         # TODO_BETA(@bryanchriswhite): Determine realistic amount for minimum gateway stake amount.
@@ -230,6 +240,14 @@ genesis:
               rev_share:
                 - address: pokt19a3t4yunp0dlpfjrp7qwnzwlrzd5fzs2gjaaaj
                   rev_share_percentage: 100
+            - service_id: anvilws
+              endpoints:
+                - configs: []
+                  rpc_type: WEBSOCKET
+                  url: ws://relayminer1:8545
+              rev_share:
+                - address: pokt19a3t4yunp0dlpfjrp7qwnzwlrzd5fzs2gjaaaj
+                  rev_share_percentage: 100
             - service_id: rest
               endpoints:
                 - configs: []
@@ -275,6 +293,10 @@ genesis:
           name: "anvil"
           compute_units_per_relay: 1
           owner_address: pokt1cwnu460557x0z78jv3hhc7356hhkrgc86c87q5
+        - id: anvilws
+          name: "anvilws"
+          compute_units_per_relay: 1
+          owner_address: pokt1cwnu460557x0z78jv3hhc7356hhkrgc86c87q5
         - id: ollama
           name: "ollama"
           compute_units_per_relay: 1

docusaurus/docs/protocol/primitives/async_service_mining.md

@@ -0,0 +1,54 @@
+# Mining Asynchronous Services
+
+::: warning
+
+Note: This documentation describes the behavior as of PR #1073. Implementation
+details may change as the protocol evolves.
+
+:::
+
+The bridge represents a WebSocket bridge between the gateway and the service
+backend. It handles the forwarding of relay requests from the gateway to the
+service backend and relay responses from the service backend to the gateway.
+
+## Asynchronous Message Handling
+
+Due to the asynchronous nature of WebSockets, there isn't always a 1:1 mapping
+between requests and responses. The bridge must handle two common scenarios:
+
+### 1. Many Responses for Few Requests (M-resp >> N-req)
+
+In this scenario, a single request can trigger multiple responses over time.
+For example:
+- A client subscribes once to an event stream (eth_subscribe)
+- The client receives many event notifications over time through that single
+  subscription
+
+### 2. Many Requests for Few Responses (N-req >> M-resp)
+
+In this scenario, multiple requests may be associated with fewer responses.
+For example:
+- A client uploads a large file in chunks, sending many requests
+- The server only occasionally sends progress updates
+
+## Design Implications
+
+This asynchronous design has two important implications:
+
+1. **Reward Eligibility**: Each message (inbound or outbound) is treated as a
+   reward-eligible relay. For example, with eth_subscribe, both the initial
+   subscription request and each received event would be eligible for rewards.
+
+2. **Message Pairing**: To maintain protocol compatibility, the bridge must always
+   pair messages when submitting to the miner. It does this by combining the most
+   recent request with the most recent response.
+
+## Future Considerations
+
+Currently, the RelayMiner is paid for each incoming and outgoing message
+transmitted. While this is the most common and trivial use case, future services
+might have different payable units of work (e.g. packet size, specific packet or
+data delimiter...).
+
+To support these use cases, the bridge should be extensible to allow for custom
+units of work to be metered and paid.

go.mod

@@ -42,7 +42,7 @@ require (
 	github.com/cosmos/ibc-go/v8 v8.2.0
 	github.com/go-kit/kit v0.13.0
 	github.com/gogo/status v1.1.0
-	github.com/golang/mock v1.6.0 // indirect
+	github.com/golang/mock v1.6.0
 	github.com/golang/protobuf v1.5.4
 	github.com/gorilla/mux v1.8.1
 	github.com/gorilla/websocket v1.5.3

localnet/kubernetes/values-relayminer-1.yaml

@@ -11,6 +11,12 @@ config:
         backend_url: http://anvil:8547/
         publicly_exposed_endpoints:
           - relayminer1
+    - service_id: anvilws
+      listen_url: http://0.0.0.0:8545
+      service_config:
+        backend_url: ws://anvil:8547/
+        publicly_exposed_endpoints:
+          - relayminer1
     - service_id: ollama
       listen_url: http://0.0.0.0:8545
       service_config:

localnet/kubernetes/values-relayminer-2.yaml

@@ -11,6 +11,12 @@ config:
         backend_url: http://anvil:8547/
         publicly_exposed_endpoints:
           - relayminer2
+    - service_id: anvilws
+      listen_url: http://0.0.0.0:8545
+      service_config:
+        backend_url: ws://anvil:8547/
+        publicly_exposed_endpoints:
+          - relayminer1
     - service_id: ollama
       listen_url: http://0.0.0.0:8545
       service_config:

localnet/kubernetes/values-relayminer-3.yaml

@@ -11,6 +11,12 @@ config:
         backend_url: http://anvil:8547/
         publicly_exposed_endpoints:
           - relayminer3
+    - service_id: anvilws
+      listen_url: http://0.0.0.0:8545
+      service_config:
+        backend_url: ws://anvil:8547/
+        publicly_exposed_endpoints:
+          - relayminer1
     - service_id: ollama
       listen_url: http://0.0.0.0:8545
       service_config:

localnet/poktrolld/config/app.toml

@@ -20,7 +20,7 @@ query-gas-limit = "0"
   rpc-max-body-bytes = 1000000
   rpc-read-timeout = 10
   rpc-write-timeout = 0
-  swagger = false
+  swagger = true
 
 [grpc]
   address = "localhost:9090"

localnet/poktrolld/config/application3_stake_config.yaml

@@ -1,3 +1,3 @@
 stake_amount: 100000070upokt
 service_ids:
-  - anvil
+  - anvilws

localnet/poktrolld/config/config.toml

@@ -8,7 +8,7 @@
 
 # The version of the CometBFT binary that created or
 # last modified the config file. Do not modify this.
-version = "0.38.10"
+version = "0.38.12"
 
 #######################################################################
 ###                   Main Base Config Options                      ###

localnet/poktrolld/config/relayminer_config.yaml

@@ -14,6 +14,12 @@ suppliers:
       backend_url: http://localhost:8081
       publicly_exposed_endpoints:
         - localhost
+  - service_id: anvilws
+    listen_url: http://localhost:6942
+    service_config:
+      backend_url: ws://localhost:8081
+      publicly_exposed_endpoints:
+        - localhost
 pprof:
   enabled: false
   addr: localhost:6060

localnet/poktrolld/config/relayminer_config_localnet_vscode.yaml

@@ -32,6 +32,12 @@ suppliers:
       backend_url: http://localhost:8547
       publicly_exposed_endpoints:
         - relayminer1
+  - service_id: anvilws
+    listen_url: http://0.0.0.0:8545
+    service_config:
+      backend_url: ws://localhost:8547
+      publicly_exposed_endpoints:
+        - relayminer1
   - service_id: ollama
     listen_url: http://0.0.0.0:8545
     service_config:

localnet/poktrolld/config/supplier1_stake_config.yaml

@@ -9,6 +9,10 @@ services:
     endpoints:
       - publicly_exposed_url: http://relayminer1:8545
         rpc_type: JSON_RPC
+  - service_id: anvilws
+    endpoints:
+      - publicly_exposed_url: ws://relayminer1:8545
+        rpc_type: WEBSOCKET
   - service_id: ollama
     rev_share_percent:
       pokt1mrqt5f7qh8uxs27cjm9t7v9e74a9vvdnq5jva4: 50

localnet/poktrolld/config/supplier2_stake_config.yaml

@@ -6,6 +6,10 @@ services:
     endpoints:
       - publicly_exposed_url: http://relayminer2:8545
         rpc_type: JSON_RPC
+  - service_id: anvilws
+    endpoints:
+      - publicly_exposed_url: ws://relayminer2:8545
+        rpc_type: WEBSOCKET
   - service_id: ollama
     endpoints:
       - publicly_exposed_url: http://relayminer2:8545

localnet/poktrolld/config/supplier3_stake_config.yaml

@@ -6,6 +6,10 @@ services:
     endpoints:
       - publicly_exposed_url: http://relayminer3:8545
         rpc_type: JSON_RPC
+  - service_id: anvilws
+    endpoints:
+      - publicly_exposed_url: ws://relayminer3:8545
+        rpc_type: WEBSOCKET
   - service_id: ollama
     endpoints:
       - publicly_exposed_url: http://relayminer3:8545

makefiles/relay.mk

@@ -19,5 +19,15 @@ send_relay_path_REST: acc_initialize_pubkeys ## Send a REST relay through PATH t
 # --data '{"model": "qwen:0.5b", "stream": false, "messages": [{"role": "user", "content":"count from 1 to 10"}]}' \
 # $(subst http://,http://ollama.,$(PATH_URL))/api/chat
 
+.PHONY: send_relay_path_WEBSOCKET
+send_relay_path_WEBSOCKET: test_e2e_env ## Send a WEBSOCKET relay through PATH to a local anvil (test ETH) node
+	@echo "Opening WebSocket connection...."
+	@echo "After the connection opens, copy & paste this to subscribe to new blocks:"
+	@echo '{"id":1,"jsonrpc":"2.0","method":"eth_subscribe","params":["newHeads"]}'
+	@echo "You should receive a subscription ID and subsequent block headers"
+	wscat -c ws://localhost:3000/v1/ \
+		-H "App-Address: pokt1lqyu4v88vp8tzc86eaqr4lq8rwhssyn6rfwzex" \
+		-H "Target-Service-Id: anvilws"
+
 # TODO_MAINNET(@olshansk): Add all the permissionless/delegated/centralized variations once
 # the following documentation is ready: https://www.notion.so/buildwithgrove/Different-Modes-of-Operation-PATH-LocalNet-Discussions-122a36edfff6805e9090c9a14f72f3b5?pvs=4#151a36edfff680d681a2dd7f4e5fee55

pkg/relayer/config/supplier_hydrator.go

@@ -73,7 +73,7 @@ func (supplierConfig *RelayMinerSupplierConfig) HydrateSupplier(
 	// by their own functions.
 	supplierConfig.ServiceConfig = &RelayMinerSupplierServiceConfig{}
 	switch backendUrl.Scheme {
-	case "http", "https":
+	case "http", "https", "ws", "wss":
 		supplierConfig.ServerType = RelayMinerServerTypeHTTP
 		if err := supplierConfig.ServiceConfig.
 			parseSupplierBackendUrl(yamlSupplierConfig.ServiceConfig); err != nil {

pkg/relayer/http_request.go

@@ -0,0 +1,103 @@
+package relayer
+
+import (
+	"bytes"
+	"encoding/base64"
+	"io"
+	"net/http"
+	"net/url"
+	"path"
+
+	sdktypes "github.com/pokt-network/shannon-sdk/types"
+
+	"github.com/pokt-network/poktroll/pkg/relayer/config"
+	"github.com/pokt-network/poktroll/x/service/types"
+)
+
+// BuildServiceBackendRequest builds the service backend request from the
+// relay request and the service configuration.
+func BuildServiceBackendRequest(
+	relayRequest *types.RelayRequest,
+	serviceConfig *config.RelayMinerSupplierServiceConfig,
+) (*http.Request, error) {
+	// Deserialize the relay request payload to get the upstream HTTP request.
+	poktHTTPRequest, err := sdktypes.DeserializeHTTPRequest(relayRequest.Payload)
+	if err != nil {
+		return nil, err
+	}
+
+	requestUrl, err := url.Parse(poktHTTPRequest.Url)
+	if err != nil {
+		return nil, err
+	}
+
+	requestUrl.Host = serviceConfig.BackendUrl.Host
+	requestUrl.Scheme = serviceConfig.BackendUrl.Scheme
+
+	// Prepend the service's backend URL path to the upstream request path to ensure
+	// proper routing while  preserving the original request structure. For RESTful APIs,
+	// this maintains resource identification.
+	//
+	// Example:
+	// - Backend URL: http://host:8080/api/v1
+	// - Upstream path: /users
+	// - Final path: http://host:8080/api/v1/users
+	requestUrl.Path = path.Join(serviceConfig.BackendUrl.Path, requestUrl.Path)
+
+	// Merge query parameters from both the upstream request and service's backend URL
+	// to maintain filtering and pagination functionality.
+	//
+	// Example:
+	// - Backend URL: http://host:8080/api/v1?key=abc
+	// - Upstream params: page=1
+	// - Final URL: http://host:8080/api/v1?key=abc&page=1
+	query := requestUrl.Query()
+	for key, values := range serviceConfig.BackendUrl.Query() {
+		for _, value := range values {
+			query.Add(key, value)
+		}
+	}
+	requestUrl.RawQuery = query.Encode()
+
+	// Create the HTTP header for the request by converting the RelayRequest's
+	// POKTHTTPRequest.Header to an http.Header.
+	header := http.Header{}
+	poktHTTPRequest.CopyToHTTPHeader(header)
+
+	// Basic HTTP Authentication.
+	// https://developer.mozilla.org/en-US/docs/Web/HTTP/Authentication
+	if serviceConfig.Authentication != nil {
+		auth := serviceConfig.Authentication.Username + ":" + serviceConfig.Authentication.Password
+		encodedAuth := base64.StdEncoding.EncodeToString([]byte(auth))
+		header.Set("Authorization", "Basic "+encodedAuth)
+	}
+
+	// Add service-specific configuration headers (e.g. auth/authz),
+	// overriding any matching upstream headers (i.e. same key).
+	for key, value := range serviceConfig.Headers {
+		header.Set(key, value)
+	}
+
+	// Create the HTTP request out of the RelayRequest's payload.
+	httpRequest := &http.Request{
+		Method: poktHTTPRequest.Method,
+		URL:    requestUrl,
+		Header: header,
+		Body:   io.NopCloser(bytes.NewReader(poktHTTPRequest.BodyBz)),
+	}
+
+	// TODO_TEST(red0ne): Test the request URL construction with different upstream
+	// request paths and query parameters.
+	// Use the same method, headers, and body as the original request to query the
+	// backend URL.
+	httpRequest.Host = serviceConfig.BackendUrl.Host
+
+	if serviceConfig.Authentication != nil {
+		httpRequest.SetBasicAuth(
+			serviceConfig.Authentication.Username,
+			serviceConfig.Authentication.Password,
+		)
+	}
+
+	return httpRequest, nil
+}