native: add committee change events

Port neo-project/neo#3158.

Close #3326

Signed-off-by: Ekaterina Pavlova <ekt@morphbits.io>

pkg/core/native/native_neo.go

@@ -265,6 +265,10 @@ func newNEO(cfg config.ProtocolConfiguration) *NEO {
 		manifest.NewParameter("to", smartcontract.PublicKeyType),
 		manifest.NewParameter("amount", smartcontract.IntegerType),
 	)
+	n.AddEvent("CommitteeChanged",
+		manifest.NewParameter("old", smartcontract.ArrayType),
+		manifest.NewParameter("new", smartcontract.ArrayType),
+	)
 
 	return n
 }
@@ -425,11 +429,36 @@ func (n *NEO) OnPersist(ic *interop.Context) error {
 		cache := ic.DAO.GetRWCache(n.ID).(*NeoCache)
 		// Cached newEpoch* values always have proper value set (either by PostPersist
 		// during the last epoch block handling or by initialization code).
+		prevCommitteeKeys := make([]stackitem.Item, len(cache.committee))
+		for i, member := range cache.committee {
+			pub, err := member.PublicKey()
+			if err != nil {
+				return fmt.Errorf("failed to get public key: %w", err)
+			}
+			prevCommitteeKeys[i] = stackitem.NewByteArray(pub.Bytes())
+		}
+		prevCommitteeStackItem := stackitem.NewArray(prevCommitteeKeys)
+
 		cache.nextValidators = cache.newEpochNextValidators
 		cache.committee = cache.newEpochCommittee
 		cache.committeeHash = cache.newEpochCommitteeHash
 		cache.votesChanged = false
 
+		newCommitteeKeys := make([]stackitem.Item, len(cache.committee))
+		for i, member := range cache.committee {
+			pub, err := member.PublicKey()
+			if err != nil {
+				return fmt.Errorf("failed to get public key: %w", err)
+			}
+			newCommitteeKeys[i] = stackitem.NewByteArray(pub.Bytes())
+		}
+		newCommitteeStackItem := stackitem.NewArray(newCommitteeKeys)
+		// Check if the committee has changed and notify subscribers.
+		if !prevCommitteeStackItem.Equals(newCommitteeStackItem) {
+			ic.AddNotification(n.Hash, "CommitteeChanged", stackitem.NewArray([]stackitem.Item{
+				prevCommitteeStackItem, newCommitteeStackItem,
+			}))
+		}
 		// We need to put in storage anyway, as it affects dumps
 		ic.DAO.PutStorageItem(n.ID, prefixCommittee, cache.committee.Bytes(ic.DAO.GetItemCtx()))
 	}

pkg/core/native/native_test/neo_test.go

@@ -64,6 +64,8 @@ func TestNEO_RegisterPriceCache(t *testing.T) {
 }
 
 func TestNEO_CandidateEvents(t *testing.T) {
+	neoCommitteeInvoker := newNeoCommitteeClient(t, 100_0000_0000)
+	neoValidatorsInvoker := neoCommitteeInvoker.WithSigners(neoCommitteeInvoker.Validator)
 	c := newNativeClient(t, nativenames.Neo)
 	singleSigner := c.Signers[0].(neotest.MultiSigner).Single(0)
 	cc := c.WithSigners(c.Signers[0], singleSigner)
@@ -116,6 +118,17 @@ func TestNEO_CandidateEvents(t *testing.T) {
 	tx = cc.Invoke(t, true, "unregisterCandidate", pkb)
 	aer = e.GetTxExecResult(t, tx)
 	require.Equal(t, 0, len(aer.Events))
+	neoValidatorsInvoker.AddNewBlock(t)
+	e.CheckHalt(t, tx, stackitem.Make(true))
+	e.CheckTxNotificationEvent(t, tx, 0, state.NotificationEvent{
+		ScriptHash: c.Hash,
+		Name:       "CommitteeChanged",
+		Item: stackitem.NewArray([]stackitem.Item{
+			stackitem.NewByteArray(pkb),
+			stackitem.NewBool(true),
+			stackitem.Make(0),
+		}),
+	})
 }
 
 func TestNEO_Vote(t *testing.T) {

pkg/rpcclient/neo/neo.go

@@ -75,6 +75,12 @@ type CandidateStateEvent struct {
 	Votes      *big.Int
 }
 
+// CommitteeChangedEvent represents a CommitteeChanged NEO event.
+type CommitteeChangedEvent struct {
+	Old []keys.PublicKey
+	New []keys.PublicKey
+}
+
 // VoteEvent represents a Vote NEO event.
 type VoteEvent struct {
 	Account util.Uint160

pkg/rpcclient/wsclient_test.go

@@ -155,11 +155,13 @@ func TestWSClientEvents(t *testing.T) {
 		fmt.Sprintf(`{"jsonrpc":"2.0","method":"block_added","params":[%s]}`, b1Verbose),
 		`{"jsonrpc":"2.0","method":"event_missed","params":[]}`, // the last one, will trigger receiver channels closing.
 	}
+	startSending := make(chan struct{})
 	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {
 		if req.URL.Path == "/ws" && req.Method == "GET" {
 			var upgrader = websocket.Upgrader{}
 			ws, err := upgrader.Upgrade(w, req, nil)
 			require.NoError(t, err)
+			<-startSending
 			for _, event := range events {
 				err = ws.SetWriteDeadline(time.Now().Add(2 * time.Second))
 				require.NoError(t, err)
@@ -209,6 +211,7 @@ func TestWSClientEvents(t *testing.T) {
 	// MissedEvent must close the channels above.
 
 	wsc.subscriptionsLock.Unlock()
+	close(startSending)
 
 	var (
 		b1Cnt, b2Cnt                                      int
@@ -297,11 +300,13 @@ func TestWSClientNonBlockingEvents(t *testing.T) {
 	require.True(t, chCap < len(events))
 
 	var blocksSent atomic.Bool
+	startSending := make(chan struct{})
 	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {
 		if req.URL.Path == "/ws" && req.Method == "GET" {
 			var upgrader = websocket.Upgrader{}
 			ws, err := upgrader.Upgrade(w, req, nil)
 			require.NoError(t, err)
+			<-startSending
 			for _, event := range events {
 				err = ws.SetWriteDeadline(time.Now().Add(2 * time.Second))
 				require.NoError(t, err)
@@ -331,6 +336,7 @@ func TestWSClientNonBlockingEvents(t *testing.T) {
 	wsc.receivers[chan<- *block.Block(bCh)] = []string{"0", "1"}
 	wsc.subscriptionsLock.Unlock()
 
+	close(startSending)
 	// Check that events are sent to WSClient.
 	require.Eventually(t, func() bool {
 		return blocksSent.Load()