fix(rpc/subscription): subscribe runtime version notify when version changes

dot/core/service.go

@@ -18,8 +18,10 @@ package core
 import (
 	"bytes"
 	"context"
+	"errors"
 	"fmt"
 	"math/big"
+	"math/rand"
 	"os"
 	"sync"
 
@@ -69,6 +71,9 @@ type Service struct {
 
 	// Keystore
 	keys *keystore.GlobalKeystore
+
+	runtimeChangedLock sync.RWMutex
+	runtimeChanged     map[byte]chan<- runtime.Version
 }
 
 // Config holds the configuration for the core Service.
@@ -151,6 +156,7 @@ func NewService(cfg *Config) (*Service, error) {
 		codeSubstitute:       cfg.CodeSubstitutes,
 		codeSubstitutedState: cfg.CodeSubstitutedState,
 		digestHandler:        cfg.DigestHandler,
+		runtimeChanged:       make(map[byte]chan<- runtime.Version),
 	}
 
 	return srv, nil
@@ -308,6 +314,11 @@ func (s *Service) handleRuntimeChanges(newState *rtstorage.TrieState) error {
 		return err
 	}
 
+	ver, err := s.rt.Version()
+	if err == nil {
+		go s.notifyRuntimeUpdated(ver)
+	}
+
 	s.codeHash = currCodeHash
 
 	err = s.codeSubstitutedState.StoreCodeSubstitutedBlockHash(common.Hash{})
@@ -462,6 +473,25 @@ func (s *Service) handleChainReorg(prev, curr common.Hash) error {
 	return nil
 }
 
+func (s *Service) notifyRuntimeUpdated(version runtime.Version) {
+	s.runtimeChangedLock.RLock()
+	defer s.runtimeChangedLock.RUnlock()
+
+	if len(s.runtimeChanged) == 0 {
+		return
+	}
+
+	logger.Info("notifying runtime updated chans...", "chans", s.runtimeChanged)
+	for _, ch := range s.runtimeChanged {
+		go func(ch chan<- runtime.Version) {
+			select {
+			case ch <- version:
+			default:
+			}
+		}(ch)
+	}
+}
+
 // maintainTransactionPool removes any transactions that were included in the new block, revalidates the transactions in the pool,
 // and moves them to the queue if valid.
 // See https://github.com/paritytech/substrate/blob/74804b5649eccfb83c90aec87bdca58e5d5c8789/client/transaction-pool/src/lib.rs#L545
@@ -579,3 +609,40 @@ func (s *Service) GetMetadata(bhash *common.Hash) ([]byte, error) {
 	s.rt.SetContextStorage(ts)
 	return s.rt.Metadata()
 }
+
+// RegisterRuntimeUpdatedChannel function to register chan that is notified when runtime version changes
+func (s *Service) RegisterRuntimeUpdatedChannel(ch chan<- runtime.Version) (byte, error) {
+	s.runtimeChangedLock.Lock()
+	defer s.runtimeChangedLock.Unlock()
+
+	if len(s.runtimeChanged) == 256 {
+		return 0, errors.New("channel limit reached")
+	}
+
+	var id byte
+	for {
+		id = generateID()
+		if s.runtimeChanged[id] == nil {
+			break
+		}
+	}
+
+	s.runtimeChanged[id] = ch
+	return id, nil
+}
+
+func (s *Service) UnregisterRuntimeUpdatedChannel (id byte) bool {
+	ch := s.runtimeChanged[id]
+	if ch != nil {
+		close(ch)
+		s.runtimeChanged[id] = nil
+		return true
+	}
+	return false
+}
+
+func generateID() byte {
+	// skipcq: GSC-G404
+	id := rand.Intn(256) //nolint
+	return byte(id)
+}

dot/rpc/modules/api.go

@@ -74,6 +74,8 @@ type CoreAPI interface {
 	GetRuntimeVersion(bhash *common.Hash) (runtime.Version, error)
 	HandleSubmittedExtrinsic(types.Extrinsic) error
 	GetMetadata(bhash *common.Hash) ([]byte, error)
+	RegisterRuntimeUpdatedChannel(ch chan<- runtime.Version) (byte, error)
+	UnregisterRuntimeUpdatedChannel(id byte) bool
 }
 
 // RPCAPI is the interface for methods related to RPC service

dot/rpc/modules/api_mocks.go

@@ -3,6 +3,7 @@ package modules
 import (
 	modulesmocks "github.com/ChainSafe/gossamer/dot/rpc/modules/mocks"
 	"github.com/ChainSafe/gossamer/lib/common"
+	runtimemocks "github.com/ChainSafe/gossamer/lib/runtime/mocks"
 	"github.com/stretchr/testify/mock"
 )
 
@@ -42,9 +43,23 @@ func NewMockCoreAPI() *modulesmocks.MockCoreAPI {
 	m := new(modulesmocks.MockCoreAPI)
 	m.On("InsertKey", mock.AnythingOfType("crypto.Keypair"))
 	m.On("HasKey", mock.AnythingOfType("string"), mock.AnythingOfType("string")).Return(false, nil)
-	m.On("GetRuntimeVersion", mock.AnythingOfType("*common.Hash")).Return(nil, nil)
+	m.On("GetRuntimeVersion", mock.AnythingOfType("*common.Hash")).Return(NewMockVersion(), nil)
 	m.On("IsBlockProducer").Return(false)
 	m.On("HandleSubmittedExtrinsic", mock.AnythingOfType("types.Extrinsic")).Return(nil)
 	m.On("GetMetadata", mock.AnythingOfType("*common.Hash")).Return(nil, nil)
+	m.On("RegisterRuntimeUpdatedChannel", mock.AnythingOfType("chan<- runtime.Version")).Return(byte(0), nil)
+	return m
+}
+
+// NewMockVersion creates and returns an runtime Version interface mock
+func NewMockVersion() *runtimemocks.MockVersion {
+	m := new(runtimemocks.MockVersion)
+	m.On("SpecName").Return([]byte(`mock-spec`))
+	m.On("ImplName").Return(nil)
+	m.On("AuthoringVersion").Return(uint32(0))
+	m.On("SpecVersion").Return(uint32(0))
+	m.On("ImplVersion").Return(uint32(0))
+	m.On("TransactionVersion").Return(uint32(0))
+	m.On("APIItems").Return(nil)
 	return m
 }

dot/rpc/subscription/listeners.go

@@ -24,6 +24,7 @@ import (
 	"github.com/ChainSafe/gossamer/dot/state"
 	"github.com/ChainSafe/gossamer/dot/types"
 	"github.com/ChainSafe/gossamer/lib/common"
+	"github.com/ChainSafe/gossamer/lib/runtime"
 )
 
 // Listener interface for functions that define Listener related functions
@@ -258,20 +259,25 @@ func (l *ExtrinsicSubmitListener) Stop() { l.cancel() }
 
 // RuntimeVersionListener to handle listening for Runtime Version
 type RuntimeVersionListener struct {
-	wsconn *WSConn
-	subID  uint
+	wsconn        WSConnAPI
+	subID         uint
+	runtimeUpdate chan runtime.Version
+	channelID byte
+	coreAPI       modules.CoreAPI
+}
+
+type VersionListener interface {
+	GetID() byte
 }
 
 // Listen implementation of Listen interface to listen for runtime version changes
 func (l *RuntimeVersionListener) Listen() {
 	// This sends current runtime version once when subscription is created
-	// TODO (ed) add logic to send updates when runtime version changes
-	rtVersion, err := l.wsconn.CoreAPI.GetRuntimeVersion(nil)
+	rtVersion, err := l.coreAPI.GetRuntimeVersion(nil)
 	if err != nil {
 		return
 	}
 	ver := modules.StateRuntimeVersionResponse{}
-
 	ver.SpecName = string(rtVersion.SpecName())
 	ver.ImplName = string(rtVersion.ImplName())
 	ver.AuthoringVersion = rtVersion.AuthoringVersion()
@@ -280,9 +286,30 @@ func (l *RuntimeVersionListener) Listen() {
 	ver.TransactionVersion = rtVersion.TransactionVersion()
 	ver.Apis = modules.ConvertAPIs(rtVersion.APIItems())
 
-	l.wsconn.safeSend(newSubscriptionResponse("state_runtimeVersion", l.subID, ver))
+	go l.wsconn.safeSend(newSubscriptionResponse("state_runtimeVersion", l.subID, ver))
+
+	// listen for runtime updates
+	go func() {
+		for info := range l.runtimeUpdate {
+			ver := modules.StateRuntimeVersionResponse{}
+
+			ver.SpecName = string(info.SpecName())
+			ver.ImplName = string(info.ImplName())
+			ver.AuthoringVersion = info.AuthoringVersion()
+			ver.SpecVersion = info.SpecVersion()
+			ver.ImplVersion = info.ImplVersion()
+			ver.TransactionVersion = info.TransactionVersion()
+			ver.Apis = modules.ConvertAPIs(info.APIItems())
+
+			l.wsconn.safeSend(newSubscriptionResponse("state_runtimeVersion", l.subID, ver))
+		}
+	}()
 }
 
 // Stop to runtimeVersionListener not implemented yet because the listener
-// does not need to be stoped
+// does not need to be stopped
 func (l *RuntimeVersionListener) Stop() {}
+
+func (l *RuntimeVersionListener) GetID() byte {
+	return l.channelID
+}

dot/rpc/subscription/listeners_test.go

@@ -17,14 +17,18 @@
 package subscription
 
 import (
+	"io/ioutil"
 	"math/big"
+	"path/filepath"
 	"testing"
 	"time"
 
 	"github.com/ChainSafe/gossamer/dot/rpc/modules"
 	"github.com/ChainSafe/gossamer/dot/state"
 	"github.com/ChainSafe/gossamer/dot/types"
 	"github.com/ChainSafe/gossamer/lib/common"
+	"github.com/ChainSafe/gossamer/lib/runtime"
+	"github.com/ChainSafe/gossamer/lib/runtime/wasmer"
 	"github.com/stretchr/testify/require"
 )
 
@@ -159,3 +163,58 @@ func TestExtrinsicSubmitListener_Listen(t *testing.T) {
 	expectedFinalizedRespones := newSubscriptionResponse(AuthorExtrinsicUpdates, esl.subID, resFinalised)
 	require.Equal(t, expectedFinalizedRespones, mockConnection.lastMessage)
 }
+
+func TestRuntimeChannelListener_Listen(t *testing.T) {
+	notifyChan := make(chan runtime.Version)
+	mockConnection := &MockWSConnAPI{}
+	rvl := RuntimeVersionListener{
+		wsconn:        mockConnection,
+		subID:         0,
+		runtimeUpdate: notifyChan,
+		coreAPI:       modules.NewMockCoreAPI(),
+	}
+
+	expectedInitialVersion := modules.StateRuntimeVersionResponse{
+		SpecName: "mock-spec",
+		Apis:     modules.ConvertAPIs(nil),
+	}
+
+	expectedInitialResponse := newSubcriptionBaseResponseJSON()
+	expectedInitialResponse.Method = "state_runtimeVersion"
+	expectedInitialResponse.Params.Result = expectedInitialVersion
+
+	instance := wasmer.NewTestInstance(t, runtime.NODE_RUNTIME)
+	_, err := runtime.GetRuntimeBlob(runtime.POLKADOT_RUNTIME_FP, runtime.POLKADOT_RUNTIME_URL)
+	require.NoError(t, err)
+	fp, err := filepath.Abs(runtime.POLKADOT_RUNTIME_FP)
+	require.NoError(t, err)
+	code, err := ioutil.ReadFile(fp)
+	require.NoError(t, err)
+	version, err := instance.CheckRuntimeVersion(code)
+	require.NoError(t, err)
+
+	expectedUpdatedVersion := modules.StateRuntimeVersionResponse{
+		SpecName:           "polkadot",
+		ImplName:           "parity-polkadot",
+		AuthoringVersion:   0,
+		SpecVersion:        25,
+		ImplVersion:        0,
+		TransactionVersion: 5,
+		Apis:               modules.ConvertAPIs(version.APIItems()),
+	}
+
+	expectedUpdateResponse := newSubcriptionBaseResponseJSON()
+	expectedUpdateResponse.Method = "state_runtimeVersion"
+	expectedUpdateResponse.Params.Result = expectedUpdatedVersion
+
+	go rvl.Listen()
+
+	//check initial response
+	time.Sleep(time.Millisecond * 10)
+	require.Equal(t, expectedInitialResponse, mockConnection.lastMessage)
+
+	// check response after update
+	notifyChan <- version
+	time.Sleep(time.Millisecond * 10)
+	require.Equal(t, expectedUpdateResponse, mockConnection.lastMessage)
+}

dot/rpc/subscription/websocket.go

@@ -31,6 +31,7 @@ import (
 	"github.com/ChainSafe/gossamer/dot/state"
 	"github.com/ChainSafe/gossamer/dot/types"
 	"github.com/ChainSafe/gossamer/lib/common"
+	"github.com/ChainSafe/gossamer/lib/runtime"
 	log "github.com/ChainSafe/log15"
 	"github.com/gorilla/websocket"
 )
@@ -345,17 +346,25 @@ func (c *WSConn) initExtrinsicWatch(reqID float64, params interface{}) (Listener
 }
 
 func (c *WSConn) initRuntimeVersionListener(reqID float64, _ interface{}) (Listener, error) {
-	rvl := &RuntimeVersionListener{
-		wsconn: c,
-	}
-
 	if c.CoreAPI == nil {
 		c.safeSendError(reqID, nil, "error CoreAPI not set")
 		return nil, fmt.Errorf("error CoreAPI not set")
 	}
 
+	rvl := &RuntimeVersionListener{
+		wsconn:        c,
+		runtimeUpdate: make(chan runtime.Version),
+		coreAPI:       c.CoreAPI,
+	}
+
+	chanID, err := c.CoreAPI.RegisterRuntimeUpdatedChannel(rvl.runtimeUpdate)
+	if err != nil {
+		return nil, err
+	}
+
 	c.mu.Lock()
 
+	rvl.channelID = chanID
 	c.qtyListeners++
 	rvl.subID = c.qtyListeners
 	c.Subscriptions[rvl.subID] = rvl
@@ -367,6 +376,19 @@ func (c *WSConn) initRuntimeVersionListener(reqID float64, _ interface{}) (Liste
 	return rvl, nil
 }
 
+func (c *WSConn) unsubscribeRuntimeVersionListener(reqID float64, l Listener, _ interface{}) {
+	observer, ok := l.(VersionListener)
+	if !ok {
+		initRes := newBooleanResponseJSON(false, reqID)
+		c.safeSend(initRes)
+		return
+	}
+	id := observer.GetID()
+
+	res := c.CoreAPI.UnregisterRuntimeUpdatedChannel(id)
+	c.safeSend(newBooleanResponseJSON(res, reqID))
+}
+
 func (c *WSConn) safeSend(msg interface{}) {
 	c.mu.Lock()
 	defer c.mu.Unlock()