go/consensus/cometbft: Serve service descriptor queries immediately

Author: peternose

.changelog/6209.trivial.md

@@ -310,29 +310,18 @@ type ServiceEvent struct {
 type ServiceDescriptor struct {
 	name      string
 	eventType string
-	queryCh   <-chan cmtpubsub.Query
+	queryCh   chan cmtpubsub.Query
 }
 
 // NewServiceDescriptor creates a new consensus service descriptor.
-func NewServiceDescriptor(name, eventType string, queryCh <-chan cmtpubsub.Query) *ServiceDescriptor {
+func NewServiceDescriptor(name, eventType string, capacity int) *ServiceDescriptor {
 	return &ServiceDescriptor{
 		name:      name,
 		eventType: eventType,
-		queryCh:   queryCh,
+		queryCh:   make(chan cmtpubsub.Query, capacity),
 	}
 }
 
-// NewStaticServiceDescriptor creates a new static consensus service descriptor.
-func NewStaticServiceDescriptor(name, eventType string, queries []cmtpubsub.Query) *ServiceDescriptor {
-	ch := make(chan cmtpubsub.Query)
-	go func() {
-		for _, q := range queries {
-			ch <- q
-		}
-	}()
-	return NewServiceDescriptor(name, eventType, ch)
-}
-
 // Name returns the name of this service.
 func (sd *ServiceDescriptor) Name() string {
 	return sd.name
@@ -348,6 +337,14 @@ func (sd *ServiceDescriptor) Queries() <-chan cmtpubsub.Query {
 	return sd.queryCh
 }
 
+// AddQuery enqueues query that need to be subscribed to.
+//
+// This method blocks if the query channel is full, which can occur when the
+// number of pending queries exceeds the channel's capacity.
+func (sd *ServiceDescriptor) AddQuery(query cmtpubsub.Query) {
+	sd.queryCh <- query
+}
+
 // ServiceClient is a consensus service client.
 type ServiceClient interface {
 	// ServiceDescriptor returns the consensus service descriptor.

go/consensus/cometbft/api/api.go

@@ -5,7 +5,6 @@ import (
 
 	"github.com/stretchr/testify/require"
 
-	cmtpubsub "github.com/cometbft/cometbft/libs/pubsub"
 	cmtquery "github.com/cometbft/cometbft/libs/pubsub/query"
 )
 
@@ -14,7 +13,8 @@ func TestServiceDescriptor(t *testing.T) {
 
 	q1 := cmtquery.MustParse("a='b'")
 
-	sd := NewStaticServiceDescriptor("test", "test_type", []cmtpubsub.Query{q1})
+	sd := NewServiceDescriptor("test", "test_type", 1)
+	sd.AddQuery(q1)
 	require.Equal("test", sd.Name())
 	require.Equal("test_type", sd.EventType())
 	recvQ1 := <-sd.Queries()

go/consensus/cometbft/api/api_test.go

@@ -9,7 +9,6 @@ import (
 	"sync"
 
 	cmtabcitypes "github.com/cometbft/cometbft/abci/types"
-	cmtpubsub "github.com/cometbft/cometbft/libs/pubsub"
 	cmttypes "github.com/cometbft/cometbft/types"
 	"github.com/eapache/channels"
 
@@ -56,7 +55,8 @@ type ServiceClient struct {
 
 // New constructs a new CometBFT backed beacon service client.
 func New(baseEpoch api.EpochTime, baseBlock int64, consensus consensus.Backend, querier *app.QueryFactory) *ServiceClient {
-	descriptor := cmtapi.NewStaticServiceDescriptor(api.ModuleName, app.EventType, []cmtpubsub.Query{app.QueryApp})
+	descriptor := cmtapi.NewServiceDescriptor(api.ModuleName, app.EventType, 1)
+	descriptor.AddQuery(app.QueryApp)
 
 	return &ServiceClient{
 		logger:            logging.GetLogger("cometbft/beacon"),

go/consensus/cometbft/beacon/beacon.go

@@ -6,7 +6,6 @@ import (
 	"fmt"
 
 	cmtabcitypes "github.com/cometbft/cometbft/abci/types"
-	cmtpubsub "github.com/cometbft/cometbft/libs/pubsub"
 	cmttypes "github.com/cometbft/cometbft/types"
 
 	"github.com/oasisprotocol/oasis-core/go/common/logging"
@@ -33,7 +32,8 @@ type ServiceClient struct {
 
 // New constructs a new CometBFT backed governance service client.
 func New(consensus consensus.Backend, querier *app.QueryFactory) *ServiceClient {
-	descriptor := tmapi.NewStaticServiceDescriptor(api.ModuleName, app.EventType, []cmtpubsub.Query{app.QueryApp})
+	descriptor := tmapi.NewServiceDescriptor(api.ModuleName, app.EventType, 1)
+	descriptor.AddQuery(app.QueryApp)
 
 	return &ServiceClient{
 		logger:        logging.GetLogger("cometbft/staking"),

go/consensus/cometbft/governance/governance.go

@@ -6,7 +6,6 @@ import (
 	"context"
 
 	cmtabcitypes "github.com/cometbft/cometbft/abci/types"
-	cmtpubsub "github.com/cometbft/cometbft/libs/pubsub"
 	cmttypes "github.com/cometbft/cometbft/types"
 
 	tmapi "github.com/oasisprotocol/oasis-core/go/consensus/cometbft/api"
@@ -30,7 +29,8 @@ type ServiceClient struct {
 
 // New constructs a new CometBFT backed key manager service client.
 func New(querier *app.QueryFactory) *ServiceClient {
-	descriptor := tmapi.NewStaticServiceDescriptor(api.ModuleName, app.EventType, []cmtpubsub.Query{app.QueryApp})
+	descriptor := tmapi.NewServiceDescriptor(api.ModuleName, app.EventType, 1)
+	descriptor.AddQuery(app.QueryApp)
 
 	return &ServiceClient{
 		descriptor:    descriptor,

go/consensus/cometbft/keymanager/keymanager.go

@@ -6,7 +6,6 @@ import (
 	"fmt"
 
 	cmtabcitypes "github.com/cometbft/cometbft/abci/types"
-	cmtpubsub "github.com/cometbft/cometbft/libs/pubsub"
 	cmttypes "github.com/cometbft/cometbft/types"
 	"github.com/eapache/channels"
 
@@ -39,7 +38,8 @@ type ServiceClient struct {
 
 // New constructs a new CometBFT backed registry service client.
 func New(consensus consensus.Backend, querier *app.QueryFactory) *ServiceClient {
-	descriptor := tmapi.NewStaticServiceDescriptor(api.ModuleName, app.EventType, []cmtpubsub.Query{app.QueryApp})
+	descriptor := tmapi.NewServiceDescriptor(api.ModuleName, app.EventType, 1)
+	descriptor.AddQuery(app.QueryApp)
 
 	return &ServiceClient{
 		logger:           logging.GetLogger("cometbft/registry"),

go/consensus/cometbft/registry/registry.go

@@ -9,7 +9,6 @@ import (
 	"sync"
 
 	cmtabcitypes "github.com/cometbft/cometbft/abci/types"
-	cmtpubsub "github.com/cometbft/cometbft/libs/pubsub"
 	cmttypes "github.com/cometbft/cometbft/types"
 
 	"github.com/oasisprotocol/oasis-core/go/common"
@@ -23,7 +22,7 @@ import (
 	"github.com/oasisprotocol/oasis-core/go/roothash/api/block"
 	"github.com/oasisprotocol/oasis-core/go/roothash/api/commitment"
 	"github.com/oasisprotocol/oasis-core/go/roothash/api/message"
-	runtimeRegistry "github.com/oasisprotocol/oasis-core/go/runtime/registry"
+	"github.com/oasisprotocol/oasis-core/go/runtime/registry"
 )
 
 const crashPointBlockBeforeIndex = "roothash.before_index"
@@ -56,14 +55,12 @@ type ServiceClient struct {
 	runtimeNotifiers map[common.Namespace]*runtimeBrokers
 	genesisBlocks    map[common.Namespace]*block.Block
 
-	queryCh         chan cmtpubsub.Query
 	trackedRuntimes map[common.Namespace]*trackedRuntime
 }
 
 // New constructs a new CometBFT-based roothash service client.
 func New(consensus consensus.Backend, querier *app.QueryFactory) *ServiceClient {
-	queryCh := make(chan cmtpubsub.Query, runtimeRegistry.MaxRuntimeCount)
-	descriptor := cmtapi.NewServiceDescriptor(api.ModuleName, app.EventType, queryCh)
+	descriptor := cmtapi.NewServiceDescriptor(api.ModuleName, app.EventType, registry.MaxRuntimeCount)
 
 	return &ServiceClient{
 		logger:           logging.GetLogger("cometbft/roothash"),
@@ -237,7 +234,8 @@ func (sc *ServiceClient) trackRuntime(id common.Namespace) {
 	}
 
 	// Request subscription to events for this runtime.
-	sc.queryCh <- app.QueryForRuntime(id)
+	query := app.QueryForRuntime(id)
+	sc.descriptor.AddQuery(query)
 }
 
 // StateToGenesis implements api.Backend.

go/consensus/cometbft/roothash/roothash.go

@@ -6,7 +6,6 @@ import (
 	"fmt"
 
 	cmtabcitypes "github.com/cometbft/cometbft/abci/types"
-	cmtpubsub "github.com/cometbft/cometbft/libs/pubsub"
 	cmttypes "github.com/cometbft/cometbft/types"
 	"github.com/eapache/channels"
 
@@ -32,7 +31,8 @@ type ServiceClient struct {
 
 // New constructs a new CometBFT-based scheduler service client.
 func New(querier *app.QueryFactory) *ServiceClient {
-	descriptor := tmapi.NewStaticServiceDescriptor(api.ModuleName, app.EventType, []cmtpubsub.Query{app.QueryApp})
+	descriptor := tmapi.NewServiceDescriptor(api.ModuleName, app.EventType, 1)
+	descriptor.AddQuery(app.QueryApp)
 
 	sc := &ServiceClient{
 		logger:     logging.GetLogger("cometbft/scheduler"),

go/consensus/cometbft/scheduler/scheduler.go

@@ -6,7 +6,6 @@ import (
 	"fmt"
 
 	cmtabcitypes "github.com/cometbft/cometbft/abci/types"
-	cmtpubsub "github.com/cometbft/cometbft/libs/pubsub"
 	cmttypes "github.com/cometbft/cometbft/types"
 
 	"github.com/oasisprotocol/oasis-core/go/common/logging"
@@ -33,7 +32,8 @@ type ServiceClient struct {
 
 // New constructs a new CometBFT backed staking service client.
 func New(consensus consensus.Backend, querier *app.QueryFactory) *ServiceClient {
-	descriptor := tmapi.NewStaticServiceDescriptor(api.ModuleName, app.EventType, []cmtpubsub.Query{app.QueryApp})
+	descriptor := tmapi.NewServiceDescriptor(api.ModuleName, app.EventType, 1)
+	descriptor.AddQuery(app.QueryApp)
 
 	return &ServiceClient{
 		logger:        logging.GetLogger("cometbft/staking"),