Merge pull request #6209 from oasisprotocol/peternose/trivial/refactor-service-descriptor

go/consensus/cometbft: Refactor service descriptor

Author: peternose

.changelog/6209.trivial.md

@@ -307,59 +307,48 @@ type ServiceEvent struct {
 }
 
 // ServiceDescriptor is a CometBFT consensus service descriptor.
-type ServiceDescriptor interface {
-	// Name returns the name of this service.
-	Name() string
-
-	// EventType returns the event type associated with the consensus service.
-	EventType() string
-
-	// Queries returns a channel that emits queries that need to be subscribed to.
-	Queries() <-chan cmtpubsub.Query
-}
-
-type serviceDescriptor struct {
+type ServiceDescriptor struct {
 	name      string
 	eventType string
-	queryCh   <-chan cmtpubsub.Query
+	queryCh   chan cmtpubsub.Query
 }
 
-func (sd *serviceDescriptor) Name() string {
+// NewServiceDescriptor creates a new consensus service descriptor.
+func NewServiceDescriptor(name, eventType string, capacity int) *ServiceDescriptor {
+	return &ServiceDescriptor{
+		name:      name,
+		eventType: eventType,
+		queryCh:   make(chan cmtpubsub.Query, capacity),
+	}
+}
+
+// Name returns the name of this service.
+func (sd *ServiceDescriptor) Name() string {
 	return sd.name
 }
 
-func (sd *serviceDescriptor) EventType() string {
+// EventType returns the event type associated with the consensus service.
+func (sd *ServiceDescriptor) EventType() string {
 	return sd.eventType
 }
 
-func (sd *serviceDescriptor) Queries() <-chan cmtpubsub.Query {
+// Queries returns a channel that emits queries that need to be subscribed to.
+func (sd *ServiceDescriptor) Queries() <-chan cmtpubsub.Query {
 	return sd.queryCh
 }
 
-// NewServiceDescriptor creates a new consensus service descriptor.
-func NewServiceDescriptor(name, eventType string, queryCh <-chan cmtpubsub.Query) ServiceDescriptor {
-	return &serviceDescriptor{
-		name:      name,
-		eventType: eventType,
-		queryCh:   queryCh,
-	}
-}
-
-// 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)
+// 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.
-	ServiceDescriptor() ServiceDescriptor
+	ServiceDescriptor() *ServiceDescriptor
 
 	// DeliverHeight delivers a new block height.
 	DeliverHeight(ctx context.Context, height int64) error

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,18 +9,17 @@ 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"
 
-	beaconAPI "github.com/oasisprotocol/oasis-core/go/beacon/api"
+	"github.com/oasisprotocol/oasis-core/go/beacon/api"
 	"github.com/oasisprotocol/oasis-core/go/common/cache/lru"
 	"github.com/oasisprotocol/oasis-core/go/common/crypto/hash"
 	"github.com/oasisprotocol/oasis-core/go/common/logging"
 	"github.com/oasisprotocol/oasis-core/go/common/pubsub"
 	consensus "github.com/oasisprotocol/oasis-core/go/consensus/api"
 	"github.com/oasisprotocol/oasis-core/go/consensus/api/events"
-	tmapi "github.com/oasisprotocol/oasis-core/go/consensus/cometbft/api"
+	cmtapi "github.com/oasisprotocol/oasis-core/go/consensus/cometbft/api"
 	app "github.com/oasisprotocol/oasis-core/go/consensus/cometbft/apps/beacon"
 )
 
@@ -30,45 +29,50 @@ const epochCacheCapacity = 128
 // ServiceClient is the beacon service client.
 type ServiceClient struct {
 	sync.RWMutex
-	tmapi.BaseServiceClient
+	cmtapi.BaseServiceClient
 
 	logger *logging.Logger
 
-	consensus consensus.Backend
-	querier   *app.QueryFactory
+	consensus  consensus.Backend
+	querier    *app.QueryFactory
+	descriptor *cmtapi.ServiceDescriptor
 
 	epochNotifier     *pubsub.Broker
-	epochLastNotified beaconAPI.EpochTime
-	epoch             beaconAPI.EpochTime
+	epochLastNotified api.EpochTime
+	epoch             api.EpochTime
 	epochCurrentBlock int64
 	epochCache        *lru.Cache
 
 	vrfNotifier     *pubsub.Broker
 	vrfLastNotified hash.Hash
-	vrfEvent        *beaconAPI.VRFEvent
+	vrfEvent        *api.VRFEvent
 
 	initialNotify bool
 
-	baseEpoch beaconAPI.EpochTime
+	baseEpoch api.EpochTime
 	baseBlock int64
 }
 
 // New constructs a new CometBFT backed beacon service client.
-func New(baseEpoch beaconAPI.EpochTime, baseBlock int64, consensus consensus.Backend, querier *app.QueryFactory) *ServiceClient {
+func New(baseEpoch api.EpochTime, baseBlock int64, consensus consensus.Backend, querier *app.QueryFactory) *ServiceClient {
+	descriptor := cmtapi.NewServiceDescriptor(api.ModuleName, app.EventType, 1)
+	descriptor.AddQuery(app.QueryApp)
+
 	return &ServiceClient{
 		logger:            logging.GetLogger("cometbft/beacon"),
 		consensus:         consensus,
 		querier:           querier,
+		descriptor:        descriptor,
 		epochNotifier:     pubsub.NewBroker(false),
-		epochLastNotified: beaconAPI.EpochInvalid,
+		epochLastNotified: api.EpochInvalid,
 		epochCache:        lru.New(lru.Capacity(epochCacheCapacity, false)),
 		vrfNotifier:       pubsub.NewBroker(false),
 		baseEpoch:         baseEpoch,
 		baseBlock:         baseBlock,
 	}
 }
 
-func (sc *ServiceClient) StateToGenesis(ctx context.Context, height int64) (*beaconAPI.Genesis, error) {
+func (sc *ServiceClient) StateToGenesis(ctx context.Context, height int64) (*api.Genesis, error) {
 	q, err := sc.querier.QueryAt(ctx, height)
 	if err != nil {
 		return nil, err
@@ -77,7 +81,7 @@ func (sc *ServiceClient) StateToGenesis(ctx context.Context, height int64) (*bea
 	return q.Genesis(ctx)
 }
 
-func (sc *ServiceClient) ConsensusParameters(ctx context.Context, height int64) (*beaconAPI.ConsensusParameters, error) {
+func (sc *ServiceClient) ConsensusParameters(ctx context.Context, height int64) (*api.ConsensusParameters, error) {
 	q, err := sc.querier.QueryAt(ctx, height)
 	if err != nil {
 		return nil, fmt.Errorf("beacon: genesis query failed: %w", err)
@@ -86,21 +90,21 @@ func (sc *ServiceClient) ConsensusParameters(ctx context.Context, height int64)
 	return q.ConsensusParameters(ctx)
 }
 
-func (sc *ServiceClient) GetBaseEpoch(context.Context) (beaconAPI.EpochTime, error) {
+func (sc *ServiceClient) GetBaseEpoch(context.Context) (api.EpochTime, error) {
 	return sc.baseEpoch, nil
 }
 
-func (sc *ServiceClient) GetEpoch(ctx context.Context, height int64) (beaconAPI.EpochTime, error) {
+func (sc *ServiceClient) GetEpoch(ctx context.Context, height int64) (api.EpochTime, error) {
 	q, err := sc.querier.QueryAt(ctx, height)
 	if err != nil {
-		return beaconAPI.EpochInvalid, err
+		return api.EpochInvalid, err
 	}
 
 	epoch, _, err := q.Epoch(ctx)
 	return epoch, err
 }
 
-func (sc *ServiceClient) GetFutureEpoch(ctx context.Context, height int64) (*beaconAPI.EpochTimeState, error) {
+func (sc *ServiceClient) GetFutureEpoch(ctx context.Context, height int64) (*api.EpochTimeState, error) {
 	q, err := sc.querier.QueryAt(ctx, height)
 	if err != nil {
 		return nil, err
@@ -109,7 +113,7 @@ func (sc *ServiceClient) GetFutureEpoch(ctx context.Context, height int64) (*bea
 	return q.FutureEpoch(ctx)
 }
 
-func (sc *ServiceClient) GetEpochBlock(ctx context.Context, epoch beaconAPI.EpochTime) (int64, error) {
+func (sc *ServiceClient) GetEpochBlock(ctx context.Context, epoch api.EpochTime) (int64, error) {
 	now, currentBlk := sc.currentEpochBlock()
 	switch {
 	case epoch == now:
@@ -140,15 +144,15 @@ func (sc *ServiceClient) GetEpochBlock(ctx context.Context, epoch beaconAPI.Epoc
 
 	// Find historic epoch with bounded bisection.
 	const maxIterations = 20 // Should be good enough for most use cases.
-	var prevEpoch beaconAPI.EpochTime
+	var prevEpoch api.EpochTime
 	for range maxIterations {
 		q, err := sc.querier.QueryAt(ctx, height)
 		if err != nil {
 			return 0, fmt.Errorf("failed to query epoch: %w", err)
 		}
 
 		var (
-			curEpoch    beaconAPI.EpochTime
+			curEpoch    api.EpochTime
 			epochHeight int64
 		)
 		curEpoch, epochHeight, err = q.Epoch(ctx)
@@ -176,7 +180,7 @@ func (sc *ServiceClient) GetEpochBlock(ctx context.Context, epoch beaconAPI.Epoc
 	return 0, fmt.Errorf("failed to find historic epoch")
 }
 
-func (sc *ServiceClient) WaitEpoch(ctx context.Context, epoch beaconAPI.EpochTime) error {
+func (sc *ServiceClient) WaitEpoch(ctx context.Context, epoch api.EpochTime) error {
 	ch, sub, err := sc.WatchEpochs(ctx)
 	if err != nil {
 		return err
@@ -198,18 +202,18 @@ func (sc *ServiceClient) WaitEpoch(ctx context.Context, epoch beaconAPI.EpochTim
 	}
 }
 
-func (sc *ServiceClient) WatchEpochs(context.Context) (<-chan beaconAPI.EpochTime, pubsub.ClosableSubscription, error) {
+func (sc *ServiceClient) WatchEpochs(context.Context) (<-chan api.EpochTime, pubsub.ClosableSubscription, error) {
 	hook := sc.epochNotifierHook()
-	ch := make(chan beaconAPI.EpochTime)
+	ch := make(chan api.EpochTime)
 	sub := sc.epochNotifier.SubscribeEx(hook)
 	sub.Unwrap(ch)
 
 	return ch, sub, nil
 }
 
-func (sc *ServiceClient) WatchLatestEpoch(context.Context) (<-chan beaconAPI.EpochTime, pubsub.ClosableSubscription, error) {
+func (sc *ServiceClient) WatchLatestEpoch(context.Context) (<-chan api.EpochTime, pubsub.ClosableSubscription, error) {
 	hook := sc.epochNotifierHook()
-	ch := make(chan beaconAPI.EpochTime)
+	ch := make(chan api.EpochTime)
 	sub := sc.epochNotifier.SubscribeBufferedEx(1, hook)
 	sub.Unwrap(ch)
 
@@ -225,7 +229,7 @@ func (sc *ServiceClient) GetBeacon(ctx context.Context, height int64) ([]byte, e
 	return q.Beacon(ctx)
 }
 
-func (sc *ServiceClient) GetVRFState(ctx context.Context, height int64) (*beaconAPI.VRFState, error) {
+func (sc *ServiceClient) GetVRFState(ctx context.Context, height int64) (*api.VRFState, error) {
 	q, err := sc.querier.QueryAt(ctx, height)
 	if err != nil {
 		return nil, err
@@ -234,17 +238,17 @@ func (sc *ServiceClient) GetVRFState(ctx context.Context, height int64) (*beacon
 	return q.VRFState(ctx)
 }
 
-func (sc *ServiceClient) WatchLatestVRFEvent(context.Context) (<-chan *beaconAPI.VRFEvent, *pubsub.Subscription, error) {
+func (sc *ServiceClient) WatchLatestVRFEvent(context.Context) (<-chan *api.VRFEvent, *pubsub.Subscription, error) {
 	hook := sc.vrfNotifierHook()
-	ch := make(chan *beaconAPI.VRFEvent)
+	ch := make(chan *api.VRFEvent)
 	sub := sc.vrfNotifier.SubscribeEx(hook)
 	sub.Unwrap(ch)
 
 	return ch, sub, nil
 }
 
-func (sc *ServiceClient) ServiceDescriptor() tmapi.ServiceDescriptor {
-	return tmapi.NewStaticServiceDescriptor("beacon", app.EventType, []cmtpubsub.Query{app.QueryApp})
+func (sc *ServiceClient) ServiceDescriptor() *cmtapi.ServiceDescriptor {
+	return sc.descriptor
 }
 
 func (sc *ServiceClient) DeliverHeight(ctx context.Context, height int64) error {
@@ -266,13 +270,13 @@ func (sc *ServiceClient) DeliverHeight(ctx context.Context, height int64) error
 		sc.epochNotifier.Broadcast(epoch)
 	}
 
-	var vrfState *beaconAPI.VRFState
+	var vrfState *api.VRFState
 	vrfState, err = q.VRFState(ctx)
 	if err != nil {
 		return fmt.Errorf("beacon: failed to query VRF state: %w", err)
 	}
 	if vrfState != nil {
-		var event beaconAPI.VRFEvent
+		var event api.VRFEvent
 		event.FromState(vrfState)
 
 		if sc.updateCachedVRFEvent(&event) {
@@ -289,8 +293,8 @@ func (sc *ServiceClient) DeliverEvent(_ context.Context, height int64, _ cmttype
 		key := pair.GetKey()
 		val := pair.GetValue()
 
-		if events.IsAttributeKind(key, &beaconAPI.EpochEvent{}) {
-			var event beaconAPI.EpochEvent
+		if events.IsAttributeKind(key, &api.EpochEvent{}) {
+			var event api.EpochEvent
 			if err := events.DecodeValue(val, &event); err != nil {
 				sc.logger.Error("epochtime: malformed epoch event value",
 					"err", err,
@@ -302,8 +306,8 @@ func (sc *ServiceClient) DeliverEvent(_ context.Context, height int64, _ cmttype
 				sc.epochNotifier.Broadcast(event.Epoch)
 			}
 		}
-		if events.IsAttributeKind(key, &beaconAPI.VRFEvent{}) {
-			var event beaconAPI.VRFEvent
+		if events.IsAttributeKind(key, &api.VRFEvent{}) {
+			var event api.VRFEvent
 			if err := events.DecodeValue(val, &event); err != nil {
 				sc.logger.Error("beacon: malformed VRF event",
 					"err", err,
@@ -318,7 +322,7 @@ func (sc *ServiceClient) DeliverEvent(_ context.Context, height int64, _ cmttype
 	return nil
 }
 
-func (sc *ServiceClient) updateCachedEpoch(height int64, epoch beaconAPI.EpochTime) bool {
+func (sc *ServiceClient) updateCachedEpoch(height int64, epoch api.EpochTime) bool {
 	sc.Lock()
 	defer sc.Unlock()
 
@@ -338,7 +342,7 @@ func (sc *ServiceClient) updateCachedEpoch(height int64, epoch beaconAPI.EpochTi
 	return false
 }
 
-func (sc *ServiceClient) updateCachedVRFEvent(event *beaconAPI.VRFEvent) bool {
+func (sc *ServiceClient) updateCachedVRFEvent(event *api.VRFEvent) bool {
 	sc.Lock()
 	defer sc.Unlock()
 
@@ -358,7 +362,7 @@ func (sc *ServiceClient) updateCachedVRFEvent(event *beaconAPI.VRFEvent) bool {
 	return false
 }
 
-func (sc *ServiceClient) currentEpochBlock() (beaconAPI.EpochTime, int64) {
+func (sc *ServiceClient) currentEpochBlock() (api.EpochTime, int64) {
 	sc.RLock()
 	defer sc.RUnlock()