From cdb1501082c6a8e3116d65679542533665705a56 Mon Sep 17 00:00:00 2001 From: kirill Date: Mon, 21 Sep 2026 15:09:52 +0400 Subject: [PATCH] Celestia ws methods --- docs/method-specs.md | 11 ++- pkg/methods/celestia_spec_test.go | 97 +++++++++++++++++++ pkg/methods/cosmos_spec_test.go | 5 +- pkg/methods/data.go | 41 +++++++- pkg/methods/method.go | 5 +- pkg/methods/specs/celestia-json-rpc.json | 8 ++ pkg/methods/specs/celestia-websocket.json | 42 ++++++++ pkg/methods/specs/celestia.json | 3 +- .../test_specs/subscription_types/spec.json | 47 +++++++++ .../wrong_subscription_type/spec.json | 22 +++++ 10 files changed, 270 insertions(+), 11 deletions(-) create mode 100644 pkg/methods/celestia_spec_test.go create mode 100644 pkg/methods/specs/celestia-websocket.json create mode 100644 pkg/methods/test_specs/subscription_types/spec.json create mode 100644 pkg/methods/test_specs/wrong_subscription_type/spec.json diff --git a/docs/method-specs.md b/docs/method-specs.md index a2f4825..cd44af0 100644 --- a/docs/method-specs.md +++ b/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..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` | diff --git a/pkg/methods/celestia_spec_test.go b/pkg/methods/celestia_spec_test.go new file mode 100644 index 0000000..7673a94 --- /dev/null +++ b/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) +} diff --git a/pkg/methods/cosmos_spec_test.go b/pkg/methods/cosmos_spec_test.go index 7a70a36..be74be3 100644 --- a/pkg/methods/cosmos_spec_test.go +++ b/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, diff --git a/pkg/methods/data.go b/pkg/methods/data.go index f0607f5..5a9dc37 100644 --- a/pkg/methods/data.go +++ b/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 diff --git a/pkg/methods/method.go b/pkg/methods/method.go index d5ff190..b3736cc 100644 --- a/pkg/methods/method.go +++ b/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 diff --git a/pkg/methods/specs/celestia-json-rpc.json b/pkg/methods/specs/celestia-json-rpc.json index 604ede3..0abf5ab 100644 --- a/pkg/methods/specs/celestia-json-rpc.json +++ b/pkg/methods/specs/celestia-json-rpc.json @@ -228,6 +228,14 @@ "cacheable": false }, "params": [] + }, + { + "name": "xrpc.cancel", + "settings": { + "cacheable": false, + "local": true + }, + "params": [] } ] } diff --git a/pkg/methods/specs/celestia-websocket.json b/pkg/methods/specs/celestia-websocket.json new file mode 100644 index 0000000..2d8414c --- /dev/null +++ b/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": [] + } + ] +} diff --git a/pkg/methods/specs/celestia.json b/pkg/methods/specs/celestia.json index d21bc1c..1b75b12 100644 --- a/pkg/methods/specs/celestia.json +++ b/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" ] } diff --git a/pkg/methods/test_specs/subscription_types/spec.json b/pkg/methods/test_specs/subscription_types/spec.json new file mode 100644 index 0000000..e925ff5 --- /dev/null +++ b/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" + } + } + } + ] +} diff --git a/pkg/methods/test_specs/wrong_subscription_type/spec.json b/pkg/methods/test_specs/wrong_subscription_type/spec.json new file mode 100644 index 0000000..9be7fa4 --- /dev/null +++ b/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" + } + } + } + ] +}