Celestia ws methods

docs/method-specs.md

@@ -71,7 +71,7 @@ Each method object:
   "local": false,
   "dispatch": "broadcast",
   "sticky": { "send-sticky": false, "create-sticky": false },
-  "subscription": { "is-subscribe": false, "method": "subscribe", "unsubscribe-method": "unsubscribe" },
+  "subscription": { "is-subscribe": false, "type": "base", "method": "subscribe", "unsubscribe-method": "unsubscribe" },
   "grpc": { "call-type": "server-stream-finite" }
 }
 ```
@@ -85,7 +85,10 @@ Each method object:
   - The two flags are mutually exclusive.
 - `subscription` (object) — only relevant on WebSocket-style methods:
   - `is-subscribe: true` — declares this method as a subscription open call.
-  - `method` (string) — for sub helpers; the underlying JSON-RPC method name when it differs from the entry's `name`.
+  - `type` (string) — the wire model of the subscription. **_Default_**: `base`.
+    - `base` — the JSON-RPC subscription model (`eth_subscribe`, Solana and Substrate subscriptions): the ack result is a subscription id, events are notifications named `method` carrying `params.subscription` and `params.result`, and `unsubscribe-method` takes the subscription id.
+    - `channel` — the go-jsonrpc channel model (celestia-node): the ack result is a per-connection channel id, events are `xrpc.ch.val` notifications with `params: [channelId, value]`, the node closes a channel with `xrpc.ch.close`, and `unsubscribe-method` (`xrpc.cancel`) takes the id of the *original subscribe request* and is never answered.
+  - `method` (string) — the notification method name the node emits for this subscription's events (`eth_subscription`, `xrpc.ch.val`).
   - `unsubscribe-method` (string) — the paired unsubscribe method.
 - `dispatch` (string) — optional fan-out execution policy for unary methods. Supported values:
   - `broadcast` — the service sends the same request to every matching available upstream, waits for fan-out to complete, and returns the first successful response in selected-upstream order (not the fastest response). This is intended for transaction propagation methods such as `eth_sendRawTransaction`. If all upstreams fail, the service returns a deterministic upstream/protocol error. It is gated by `chain-defaults.<chain>.dispatch.broadcast`.
@@ -242,7 +245,7 @@ The `specs` package embeds the specs below (see [`pkg/methods/specs/`](../pkg/me
 | `polkadot` | `polkadot-json-rpc`, `polkadot-websocket` |
 | `astar` | `eth`, `polkadot` |
 | `cosmos-evm` | `eth`, `cosmos` |
-| `celestia` | `celestia-json-rpc`, `cosmos` |
+| `celestia` | `celestia-json-rpc`, `celestia-websocket`, `cosmos` |
 
 ### Plain specs
 
@@ -252,7 +255,7 @@ Grouped by the transports they declare:
 | --- | --- |
 | `json-rpc`, `websocket` | `arbitrum`, `avail`, `cronos_zkevm`, `eth-json-rpc`, `fantom`, `filecoin`, `harmony_0`, `harmony_1`, `hyperliquid-eth`, `klaytn-json-rpc`, `linea`, `mantle`, `optimism`, `polkadot-json-rpc`, `polygon`, `polygon_zkevm`, `rootstock`, `scroll`, `sei`, `solana-json-rpc`, `viction`, `zk` |
 | `json-rpc` | `algorand-json-rpc`, `aztec`, `bitcoin-json-rpc`, `celestia-json-rpc`, `near-json-rpc`, `starknet-json-rpc`, `stellar-json-rpc`, `tron-json-rpc` |
-| `websocket` | `eth-websocket`, `klaytn-websocket`, `polkadot-websocket`, `solana-websocket` |
+| `websocket` | `celestia-websocket`, `eth-websocket`, `klaytn-websocket`, `polkadot-websocket`, `solana-websocket` |
 | `tendermint` | `cosmos-tendermint` |
 | `rest` | `algorand-rest`, `aptos`, `cosmos-rest`, `eth-beacon-chain`, `stellar-horizon`, `ton-http-v2`, `tron-rest` |
 | `rest-indexer` | `ton-index-v3` |

pkg/methods/celestia_spec_test.go

@@ -0,0 +1,97 @@
+package specs_test
+
+import (
+	"os"
+	"testing"
+
+	specs "github.com/drpcorg/public/pkg/methods"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// celestia-node speaks the go-jsonrpc channel protocol: a subscribe call is
+// acknowledged with a channel id, events arrive as xrpc.ch.val notifications,
+// the node closes a channel with xrpc.ch.close and the client cancels one with
+// xrpc.cancel naming the original request. The spec pins those names and the
+// channel subscription type; a wrong value fails silently at runtime because
+// frames are matched on the channel id, never on the method name.
+func TestCelestiaWebsocketChannelSubscriptions(t *testing.T) {
+	require.NoError(t, specs.NewMethodSpecLoader().Load())
+
+	assert.Contains(t, specs.GetSpecConnectors("celestia"), specs.WebsocketConnector)
+	assert.ElementsMatch(t, []string{"header.Subscribe", "blob.Subscribe"}, specs.GetSubMethods("celestia").ToSlice())
+
+	for _, name := range []string{"header.Subscribe", "blob.Subscribe"} {
+		assert.True(t, specs.IsSubscribeMethod("celestia", name), "%s is not a subscribe method", name)
+
+		method := specs.GetSpecMethod("celestia", name)
+		require.NotNil(t, method, "no spec method %s", name)
+		require.NotNil(t, method.Subscription, "%s has no subscription settings", name)
+		assert.Equal(t, specs.SubscriptionTypeChannel, method.Subscription.Type, "%s: wrong subscription type", name)
+		assert.Equal(t, "xrpc.ch.val", method.Subscription.Method, "%s: wrong notification name", name)
+		assert.False(t, method.IsCacheable(), "%s must not be cacheable", name)
+		assert.Equal(t, []specs.ApiConnectorType{specs.WebsocketConnector}, method.GetApiConnectorTypes(), "%s: wrong connectors", name)
+
+		unsub, ok := specs.GetUnsubscribeMethod("celestia", name)
+		require.True(t, ok, "no unsubscribe method for %s", name)
+		assert.Equal(t, "xrpc.cancel", unsub, "%s: wrong unsubscribe name", name)
+	}
+
+	cancel := specs.GetSpecMethod("celestia", "xrpc.cancel")
+	require.NotNil(t, cancel)
+	assert.True(t, cancel.IsLocal())
+	assert.False(t, cancel.IsCacheable())
+	assert.False(t, cancel.IsSubscribe())
+
+	assert.Contains(t, wsMethods(t, "celestia"), "header.Subscribe")
+	assert.NotContains(t, jsonRpcMethods(t, "celestia"), "header.Subscribe")
+	assert.NotContains(t, wsMethods(t, "celestia"), "header.LocalHead")
+}
+
+// An absent type means the JSON-RPC subscription model every other websocket
+// spec uses, so those specs need no change and consumers never see "".
+func TestSubscriptionTypeDefaultsToBase(t *testing.T) {
+	require.NoError(t, specs.NewMethodSpecLoader().Load())
+
+	for specName, methodName := range map[string]string{
+		"eth":      "eth_subscribe",
+		"solana":   "slotSubscribe",
+		"polkadot": "chain_subscribeNewHeads",
+	} {
+		method := specs.GetSpecMethod(specName, methodName)
+		require.NotNil(t, method, "%s: no spec method %s", specName, methodName)
+		require.NotNil(t, method.Subscription, "%s: %s has no subscription settings", specName, methodName)
+		assert.Equal(t, specs.SubscriptionTypeBase, method.Subscription.Type, "%s.%s", specName, methodName)
+	}
+}
+
+func TestLoadSpecSubscriptionTypes(t *testing.T) {
+	require.NoError(t, specs.NewMethodSpecLoaderWithFs(os.DirFS("test_specs/subscription_types")).Load())
+
+	for methodName, want := range map[string]specs.SubscriptionType{
+		"eth_subscribe":           specs.SubscriptionTypeBase,
+		"chain_subscribeNewHeads": specs.SubscriptionTypeBase,
+		"header.Subscribe":        specs.SubscriptionTypeChannel,
+	} {
+		method := specs.GetSpecMethod("test", methodName)
+		require.NotNil(t, method, "no spec method %s", methodName)
+		require.NotNil(t, method.Subscription, "%s has no subscription settings", methodName)
+		assert.Equal(t, want, method.Subscription.Type, methodName)
+	}
+}
+
+func TestLoadSpecWrongSubscriptionTypeThenError(t *testing.T) {
+	err := specs.NewMethodSpecLoaderWithFs(os.DirFS("test_specs/wrong_subscription_type")).Load()
+
+	assert.ErrorContains(t, err, "couldn't read method specs: error during method 'header.Subscribe' of 'spec.json' validation, cause: unknown subscription type - stream")
+}
+
+// A Go-constructed method must see the same default as a loaded one.
+func TestMethodWithSettingsDefaultsSubscriptionType(t *testing.T) {
+	method := specs.MethodWithSettings("eth_subscribe", []specs.ApiConnectorType{specs.WebsocketConnector}, &specs.MethodSettings{
+		Subscription: &specs.Subscription{IsSubscribe: true, Method: "eth_subscription", UnsubMethod: "eth_unsubscribe"},
+	}, nil)
+	require.NotNil(t, method)
+	require.NotNil(t, method.Subscription)
+	assert.Equal(t, specs.SubscriptionTypeBase, method.Subscription.Type)
+}

pkg/methods/cosmos_spec_test.go

@@ -265,14 +265,15 @@ func TestCosmosEvmBundleCarriesEthAndCosmos(t *testing.T) {
 }
 
 // Celestia is a Cosmos SDK chain whose DA node also serves its own JSON-RPC
-// (header.*, blob.*, share.*), so the celestia bundle is the union of
-// celestia-json-rpc and cosmos.
+// (header.*, blob.*, share.*) and websocket subscriptions, so the celestia
+// bundle is the union of celestia-json-rpc, celestia-websocket and cosmos.
 func TestCelestiaBundleCarriesDaRpcAndCosmos(t *testing.T) {
 	require.NoError(t, specs.NewMethodSpecLoader().Load())
 
 	assert.ElementsMatch(t,
 		[]specs.ApiConnectorType{
 			specs.JsonRpcConnector,
+			specs.WebsocketConnector,
 			specs.TendermintConnector,
 			specs.RestConnector,
 			specs.GrpcConnector,

pkg/methods/data.go

@@ -102,9 +102,39 @@ type Sticky struct {
 }
 
 type Subscription struct {
-	IsSubscribe bool   `json:"is-subscribe"`
-	Method      string `json:"method"`
-	UnsubMethod string `json:"unsubscribe-method"`
+	IsSubscribe bool             `json:"is-subscribe"`
+	Type        SubscriptionType `json:"type"`
+	Method      string           `json:"method"`
+	UnsubMethod string           `json:"unsubscribe-method"`
+}
+
+// SubscriptionType names the wire model of a subscription: how the subscribe
+// call is acknowledged, how events are framed and what the unsubscribe method
+// takes. Absent in JSON means SubscriptionTypeBase.
+type SubscriptionType string
+
+const (
+	// SubscriptionTypeBase is the JSON-RPC subscription model (eth_subscribe,
+	// Solana and Substrate subscriptions): the ack result is a subscription id,
+	// events are notifications named Subscription.Method with
+	// params.subscription and params.result, and the unsubscribe method takes
+	// the subscription id.
+	SubscriptionTypeBase SubscriptionType = "base"
+	// SubscriptionTypeChannel is the go-jsonrpc channel model (celestia-node):
+	// the ack result is a per-connection channel id, events are xrpc.ch.val
+	// notifications with params [channelId, value], the node closes a channel
+	// with xrpc.ch.close, and the unsubscribe method (xrpc.cancel) takes the id
+	// of the original subscribe request and is never answered.
+	SubscriptionTypeChannel SubscriptionType = "channel"
+)
+
+func (s SubscriptionType) validate() error {
+	switch s {
+	case "", SubscriptionTypeBase, SubscriptionTypeChannel:
+		return nil
+	default:
+		return fmt.Errorf("unknown subscription type - %s", s)
+	}
 }
 
 type ParserReturnType string
@@ -219,6 +249,11 @@ func (m *MethodSettings) validate() error {
 			return errors.New("both 'create-sticky' and 'send-sticky' are enabled")
 		}
 	}
+	if m.Subscription != nil {
+		if err := m.Subscription.Type.validate(); err != nil {
+			return err
+		}
+	}
 	if m.Grpc != nil {
 		if err := m.Grpc.CallType.validate(); err != nil {
 			return err

pkg/methods/method.go

@@ -174,7 +174,10 @@ func fromMethodData(methodData *MethodData, apiConnectorTypes []ApiConnectorType
 			cacheable = *methodData.Settings.Cacheable
 		}
 		if methodData.Settings.Subscription != nil {
-			sub = methodData.Settings.Subscription
+			sub = new(*methodData.Settings.Subscription)
+			if sub.Type == "" {
+				sub.Type = SubscriptionTypeBase
+			}
 		}
 		enforceIntegrity = methodData.Settings.EnforceIntegrity
 		local = methodData.Settings.Local

pkg/methods/specs/celestia-json-rpc.json

@@ -228,6 +228,14 @@
         "cacheable": false
       },
       "params": []
+    },
+    {
+      "name": "xrpc.cancel",
+      "settings": {
+        "cacheable": false,
+        "local": true
+      },
+      "params": []
     }
   ]
 }

pkg/methods/specs/celestia-websocket.json

@@ -0,0 +1,42 @@
+{
+  "openrpc": "1.0.0",
+  "info": {
+    "title": "Celestia DA node websocket subscription methods (go-jsonrpc channels)",
+    "version": "1.0.0"
+  },
+  "spec": {
+    "name": "celestia-websocket",
+    "api-connectors": [
+      "websocket"
+    ],
+    "type": "plain"
+  },
+  "methods": [
+    {
+      "name": "header.Subscribe",
+      "settings": {
+        "cacheable": false,
+        "subscription": {
+          "is-subscribe": true,
+          "type": "channel",
+          "method": "xrpc.ch.val",
+          "unsubscribe-method": "xrpc.cancel"
+        }
+      },
+      "params": []
+    },
+    {
+      "name": "blob.Subscribe",
+      "settings": {
+        "cacheable": false,
+        "subscription": {
+          "is-subscribe": true,
+          "type": "channel",
+          "method": "xrpc.ch.val",
+          "unsubscribe-method": "xrpc.cancel"
+        }
+      },
+      "params": []
+    }
+  ]
+}

pkg/methods/specs/celestia.json

@@ -1,7 +1,7 @@
 {
   "openrpc": "1.0.0",
   "info": {
-    "title": "Celestia chain methods - DA node JSON-RPC plus Cosmos Tendermint RPC, LCD REST and gRPC",
+    "title": "Celestia chain methods - DA node JSON-RPC and websocket plus Cosmos Tendermint RPC, LCD REST and gRPC",
     "version": "1.0.0"
   },
   "spec": {
@@ -10,6 +10,7 @@
   },
   "spec-imports": [
     "celestia-json-rpc",
+    "celestia-websocket",
     "cosmos"
   ]
 }

pkg/methods/test_specs/subscription_types/spec.json

@@ -0,0 +1,47 @@
+{
+  "spec": {
+    "api-connectors": ["websocket"],
+    "name": "test",
+    "type": "plain"
+  },
+  "methods": [
+    {
+      "name": "eth_subscribe",
+      "params": [],
+      "settings": {
+        "cacheable": false,
+        "subscription": {
+          "is-subscribe": true,
+          "method": "eth_subscription",
+          "unsubscribe-method": "eth_unsubscribe"
+        }
+      }
+    },
+    {
+      "name": "chain_subscribeNewHeads",
+      "params": [],
+      "settings": {
+        "cacheable": false,
+        "subscription": {
+          "is-subscribe": true,
+          "type": "base",
+          "method": "chain_newHead",
+          "unsubscribe-method": "chain_unsubscribeNewHeads"
+        }
+      }
+    },
+    {
+      "name": "header.Subscribe",
+      "params": [],
+      "settings": {
+        "cacheable": false,
+        "subscription": {
+          "is-subscribe": true,
+          "type": "channel",
+          "method": "xrpc.ch.val",
+          "unsubscribe-method": "xrpc.cancel"
+        }
+      }
+    }
+  ]
+}

pkg/methods/test_specs/wrong_subscription_type/spec.json

@@ -0,0 +1,22 @@
+{
+  "spec": {
+    "api-connectors": ["websocket"],
+    "name": "test",
+    "type": "plain"
+  },
+  "methods": [
+    {
+      "name": "header.Subscribe",
+      "params": [],
+      "settings": {
+        "cacheable": false,
+        "subscription": {
+          "is-subscribe": true,
+          "type": "stream",
+          "method": "xrpc.ch.val",
+          "unsubscribe-method": "xrpc.cancel"
+        }
+      }
+    }
+  ]
+}