Enhance UTXOsChanged notifications (#1522)

* In PropagateUTXOsChangedNotifications, add the given addresses to the address list instead of replacing them.

* Add StopNotifyingUtxosChangedRequestMessage to rpc.proto.

* Implement StopNotifyingUTXOsChanged.

* Optimize convertUTXOChangesToUTXOsChangedNotification.

app/appmessage/message.go

@@ -117,6 +117,8 @@ const (
 	CmdNotifyUTXOsChangedRequestMessage
 	CmdNotifyUTXOsChangedResponseMessage
 	CmdUTXOsChangedNotificationMessage
+	CmdStopNotifyingUTXOsChangedRequestMessage
+	CmdStopNotifyingUTXOsChangedResponseMessage
 	CmdGetUTXOsByAddressesRequestMessage
 	CmdGetUTXOsByAddressesResponseMessage
 	CmdGetVirtualSelectedParentBlueScoreRequestMessage
@@ -221,6 +223,8 @@ var RPCMessageCommandToString = map[MessageCommand]string{
 	CmdNotifyUTXOsChangedRequestMessage:                           "NotifyUTXOsChangedRequest",
 	CmdNotifyUTXOsChangedResponseMessage:                          "NotifyUTXOsChangedResponse",
 	CmdUTXOsChangedNotificationMessage:                            "UTXOsChangedNotification",
+	CmdStopNotifyingUTXOsChangedRequestMessage:                    "StopNotifyingUTXOsChangedRequest",
+	CmdStopNotifyingUTXOsChangedResponseMessage:                   "StopNotifyingUTXOsChangedResponse",
 	CmdGetUTXOsByAddressesRequestMessage:                          "GetUTXOsByAddressesRequest",
 	CmdGetUTXOsByAddressesResponseMessage:                         "GetUTXOsByAddressesResponse",
 	CmdGetVirtualSelectedParentBlueScoreRequestMessage:            "GetVirtualSelectedParentBlueScoreRequest",

app/appmessage/rpc_stop_notifying_utxos_changed.go

@@ -0,0 +1,37 @@
+package appmessage
+
+// StopNotifyingUTXOsChangedRequestMessage is an appmessage corresponding to
+// its respective RPC message
+type StopNotifyingUTXOsChangedRequestMessage struct {
+	baseMessage
+	Addresses []string
+}
+
+// Command returns the protocol command string for the message
+func (msg *StopNotifyingUTXOsChangedRequestMessage) Command() MessageCommand {
+	return CmdStopNotifyingUTXOsChangedRequestMessage
+}
+
+// NewStopNotifyingUTXOsChangedRequestMessage returns a instance of the message
+func NewStopNotifyingUTXOsChangedRequestMessage(addresses []string) *StopNotifyingUTXOsChangedRequestMessage {
+	return &StopNotifyingUTXOsChangedRequestMessage{
+		Addresses: addresses,
+	}
+}
+
+// StopNotifyingUTXOsChangedResponseMessage is an appmessage corresponding to
+// its respective RPC message
+type StopNotifyingUTXOsChangedResponseMessage struct {
+	baseMessage
+	Error *RPCError
+}
+
+// Command returns the protocol command string for the message
+func (msg *StopNotifyingUTXOsChangedResponseMessage) Command() MessageCommand {
+	return CmdStopNotifyingUTXOsChangedResponseMessage
+}
+
+// NewStopNotifyingUTXOsChangedResponseMessage returns a instance of the message
+func NewStopNotifyingUTXOsChangedResponseMessage() *StopNotifyingUTXOsChangedResponseMessage {
+	return &StopNotifyingUTXOsChangedResponseMessage{}
+}

app/rpc/rpc.go

@@ -35,6 +35,7 @@ var handlers = map[appmessage.MessageCommand]handler{
 	appmessage.CmdShutDownRequestMessage:                                    rpchandlers.HandleShutDown,
 	appmessage.CmdGetHeadersRequestMessage:                                  rpchandlers.HandleGetHeaders,
 	appmessage.CmdNotifyUTXOsChangedRequestMessage:                          rpchandlers.HandleNotifyUTXOsChanged,
+	appmessage.CmdStopNotifyingUTXOsChangedRequestMessage:                   rpchandlers.HandleStopNotifyingUTXOsChanged,
 	appmessage.CmdGetUTXOsByAddressesRequestMessage:                         rpchandlers.HandleGetUTXOsByAddresses,
 	appmessage.CmdGetVirtualSelectedParentBlueScoreRequestMessage:           rpchandlers.HandleGetVirtualSelectedParentBlueScore,
 	appmessage.CmdNotifyVirtualSelectedParentBlueScoreChangedRequestMessage: rpchandlers.HandleNotifyVirtualSelectedParentBlueScoreChanged,

app/rpc/rpccontext/notificationmanager.go

@@ -31,7 +31,7 @@ type NotificationListener struct {
 	propagateUTXOsChangedNotifications                          bool
 	propagateVirtualSelectedParentBlueScoreChangedNotifications bool
 
-	propagateUTXOsChangedNotificationAddresses []*UTXOsChangedNotificationAddress
+	propagateUTXOsChangedNotificationAddresses map[utxoindex.ScriptPublicKeyString]*UTXOsChangedNotificationAddress
 }
 
 // NewNotificationManager creates a new NotificationManager
@@ -216,34 +216,70 @@ func (nl *NotificationListener) PropagateFinalityConflictResolvedNotifications()
 }
 
 // PropagateUTXOsChangedNotifications instructs the listener to send UTXOs changed notifications
-// to the remote listener
+// to the remote listener for the given addresses. Subsequent calls instruct the listener to
+// send UTXOs changed notifications for those addresses along with the old ones. Duplicate addresses
+// are ignored.
 func (nl *NotificationListener) PropagateUTXOsChangedNotifications(addresses []*UTXOsChangedNotificationAddress) {
+	if !nl.propagateUTXOsChangedNotifications {
 		nl.propagateUTXOsChangedNotifications = true
-	nl.propagateUTXOsChangedNotificationAddresses = addresses
+		nl.propagateUTXOsChangedNotificationAddresses =
+			make(map[utxoindex.ScriptPublicKeyString]*UTXOsChangedNotificationAddress, len(addresses))
+	}
+
+	for _, address := range addresses {
+		nl.propagateUTXOsChangedNotificationAddresses[address.ScriptPublicKeyString] = address
+	}
+}
+
+// StopPropagatingUTXOsChangedNotifications instructs the listener to stop sending UTXOs
+// changed notifications to the remote listener for the given addresses. Addresses for which
+// notifications are not currently sent are ignored.
+func (nl *NotificationListener) StopPropagatingUTXOsChangedNotifications(addresses []*UTXOsChangedNotificationAddress) {
+	if !nl.propagateUTXOsChangedNotifications {
+		return
+	}
+
+	for _, address := range addresses {
+		delete(nl.propagateUTXOsChangedNotificationAddresses, address.ScriptPublicKeyString)
+	}
 }
 
 func (nl *NotificationListener) convertUTXOChangesToUTXOsChangedNotification(
 	utxoChanges *utxoindex.UTXOChanges) *appmessage.UTXOsChangedNotificationMessage {
 
+	// As an optimization, we iterate over the smaller set (O(n)) among the two below
+	// and check existence over the larger set (O(1))
+	utxoChangesSize := len(utxoChanges.Added) + len(utxoChanges.Removed)
+	addressesSize := len(nl.propagateUTXOsChangedNotificationAddresses)
+
 	notification := &appmessage.UTXOsChangedNotificationMessage{}
+	if utxoChangesSize < addressesSize {
+		for scriptPublicKeyString, addedPairs := range utxoChanges.Added {
+			if listenerAddress, ok := nl.propagateUTXOsChangedNotificationAddresses[scriptPublicKeyString]; ok {
+				utxosByAddressesEntries := ConvertUTXOOutpointEntryPairsToUTXOsByAddressesEntries(listenerAddress.Address, addedPairs)
+				notification.Added = append(notification.Added, utxosByAddressesEntries...)
+			}
+		}
+		for scriptPublicKeyString, removedOutpoints := range utxoChanges.Removed {
+			if listenerAddress, ok := nl.propagateUTXOsChangedNotificationAddresses[scriptPublicKeyString]; ok {
+				utxosByAddressesEntries := convertUTXOOutpointsToUTXOsByAddressesEntries(listenerAddress.Address, removedOutpoints)
+				notification.Removed = append(notification.Removed, utxosByAddressesEntries...)
+			}
+		}
+	} else {
 		for _, listenerAddress := range nl.propagateUTXOsChangedNotificationAddresses {
 			listenerScriptPublicKeyString := listenerAddress.ScriptPublicKeyString
 			if addedPairs, ok := utxoChanges.Added[listenerScriptPublicKeyString]; ok {
-			notification.Added = append(notification.Added,
+				utxosByAddressesEntries := ConvertUTXOOutpointEntryPairsToUTXOsByAddressesEntries(listenerAddress.Address, addedPairs)
-				ConvertUTXOOutpointEntryPairsToUTXOsByAddressesEntries(listenerAddress.Address, addedPairs)...)
+				notification.Added = append(notification.Added, utxosByAddressesEntries...)
 			}
 			if removedOutpoints, ok := utxoChanges.Removed[listenerScriptPublicKeyString]; ok {
-			for outpoint := range removedOutpoints {
+				utxosByAddressesEntries := convertUTXOOutpointsToUTXOsByAddressesEntries(listenerAddress.Address, removedOutpoints)
-				notification.Removed = append(notification.Removed, &appmessage.UTXOsByAddressesEntry{
+				notification.Removed = append(notification.Removed, utxosByAddressesEntries...)
-					Address: listenerAddress.Address,
-					Outpoint: &appmessage.RPCOutpoint{
-						TransactionID: outpoint.TransactionID.String(),
-						Index:         outpoint.Index,
-					},
-				})
 			}
 		}
 	}
+
 	return notification
 }
 

app/rpc/rpccontext/utxos_by_addresses.go

@@ -2,6 +2,9 @@ package rpccontext
 
 import (
 	"encoding/hex"
+	"github.com/kaspanet/kaspad/domain/consensus/utils/txscript"
+	"github.com/kaspanet/kaspad/util"
+	"github.com/pkg/errors"
 
 	"github.com/kaspanet/kaspad/app/appmessage"
 	"github.com/kaspanet/kaspad/domain/utxoindex"
@@ -28,3 +31,43 @@ func ConvertUTXOOutpointEntryPairsToUTXOsByAddressesEntries(address string, pair
 	}
 	return utxosByAddressesEntries
 }
+
+// convertUTXOOutpointsToUTXOsByAddressesEntries converts
+// UTXOOutpoints to a slice of UTXOsByAddressesEntry
+func convertUTXOOutpointsToUTXOsByAddressesEntries(address string, outpoints utxoindex.UTXOOutpoints) []*appmessage.UTXOsByAddressesEntry {
+	utxosByAddressesEntries := make([]*appmessage.UTXOsByAddressesEntry, 0, len(outpoints))
+	for outpoint := range outpoints {
+		utxosByAddressesEntries = append(utxosByAddressesEntries, &appmessage.UTXOsByAddressesEntry{
+			Address: address,
+			Outpoint: &appmessage.RPCOutpoint{
+				TransactionID: outpoint.TransactionID.String(),
+				Index:         outpoint.Index,
+			},
+		})
+	}
+	return utxosByAddressesEntries
+}
+
+// ConvertAddressStringsToUTXOsChangedNotificationAddresses converts address strings
+// to UTXOsChangedNotificationAddresses
+func (ctx *Context) ConvertAddressStringsToUTXOsChangedNotificationAddresses(
+	addressStrings []string) ([]*UTXOsChangedNotificationAddress, error) {
+
+	addresses := make([]*UTXOsChangedNotificationAddress, len(addressStrings))
+	for i, addressString := range addressStrings {
+		address, err := util.DecodeAddress(addressString, ctx.Config.ActiveNetParams.Prefix)
+		if err != nil {
+			return nil, errors.Errorf("Could not decode address '%s': %s", addressString, err)
+		}
+		scriptPublicKey, err := txscript.PayToAddrScript(address)
+		if err != nil {
+			return nil, errors.Errorf("Could not create a scriptPublicKey for address '%s': %s", addressString, err)
+		}
+		scriptPublicKeyString := utxoindex.ConvertScriptPublicKeyToString(scriptPublicKey)
+		addresses[i] = &UTXOsChangedNotificationAddress{
+			Address:               addressString,
+			ScriptPublicKeyString: scriptPublicKeyString,
+		}
+	}
+	return addresses, nil
+}

app/rpc/rpchandlers/notify_utxos_changed.go

@@ -3,10 +3,7 @@ package rpchandlers
 import (
 	"github.com/kaspanet/kaspad/app/appmessage"
 	"github.com/kaspanet/kaspad/app/rpc/rpccontext"
-	"github.com/kaspanet/kaspad/domain/consensus/utils/txscript"
-	"github.com/kaspanet/kaspad/domain/utxoindex"
 	"github.com/kaspanet/kaspad/infrastructure/network/netadapter/router"
-	"github.com/kaspanet/kaspad/util"
 )
 
 // HandleNotifyUTXOsChanged handles the respectively named RPC command
@@ -18,27 +15,12 @@ func HandleNotifyUTXOsChanged(context *rpccontext.Context, router *router.Router
 	}
 
 	notifyUTXOsChangedRequest := request.(*appmessage.NotifyUTXOsChangedRequestMessage)
-
+	addresses, err := context.ConvertAddressStringsToUTXOsChangedNotificationAddresses(notifyUTXOsChangedRequest.Addresses)
-	addresses := make([]*rpccontext.UTXOsChangedNotificationAddress, len(notifyUTXOsChangedRequest.Addresses))
-	for i, addressString := range notifyUTXOsChangedRequest.Addresses {
-		address, err := util.DecodeAddress(addressString, context.Config.ActiveNetParams.Prefix)
 	if err != nil {
 		errorMessage := appmessage.NewNotifyUTXOsChangedResponseMessage()
-			errorMessage.Error = appmessage.RPCErrorf("Could not decode address '%s': %s", addressString, err)
+		errorMessage.Error = appmessage.RPCErrorf("Parsing error: %s", err)
 		return errorMessage, nil
 	}
-		scriptPublicKey, err := txscript.PayToAddrScript(address)
-		if err != nil {
-			errorMessage := appmessage.NewNotifyUTXOsChangedResponseMessage()
-			errorMessage.Error = appmessage.RPCErrorf("Could not create a scriptPublicKey for address '%s': %s", addressString, err)
-			return errorMessage, nil
-		}
-		scriptPublicKeyString := utxoindex.ConvertScriptPublicKeyToString(scriptPublicKey)
-		addresses[i] = &rpccontext.UTXOsChangedNotificationAddress{
-			Address:               addressString,
-			ScriptPublicKeyString: scriptPublicKeyString,
-		}
-	}
 
 	listener, err := context.NotificationManager.Listener(router)
 	if err != nil {

app/rpc/rpchandlers/stop_notifying_utxos_changed.go

@@ -0,0 +1,33 @@
+package rpchandlers
+
+import (
+	"github.com/kaspanet/kaspad/app/appmessage"
+	"github.com/kaspanet/kaspad/app/rpc/rpccontext"
+	"github.com/kaspanet/kaspad/infrastructure/network/netadapter/router"
+)
+
+// HandleStopNotifyingUTXOsChanged handles the respectively named RPC command
+func HandleStopNotifyingUTXOsChanged(context *rpccontext.Context, router *router.Router, request appmessage.Message) (appmessage.Message, error) {
+	if !context.Config.UTXOIndex {
+		errorMessage := appmessage.NewStopNotifyingUTXOsChangedResponseMessage()
+		errorMessage.Error = appmessage.RPCErrorf("Method unavailable when kaspad is run without --utxoindex")
+		return errorMessage, nil
+	}
+
+	stopNotifyingUTXOsChangedRequest := request.(*appmessage.StopNotifyingUTXOsChangedRequestMessage)
+	addresses, err := context.ConvertAddressStringsToUTXOsChangedNotificationAddresses(stopNotifyingUTXOsChangedRequest.Addresses)
+	if err != nil {
+		errorMessage := appmessage.NewNotifyUTXOsChangedResponseMessage()
+		errorMessage.Error = appmessage.RPCErrorf("Parsing error: %s", err)
+		return errorMessage, nil
+	}
+
+	listener, err := context.NotificationManager.Listener(router)
+	if err != nil {
+		return nil, err
+	}
+	listener.StopPropagatingUTXOsChangedNotifications(addresses)
+
+	response := appmessage.NewStopNotifyingUTXOsChangedResponseMessage()
+	return response, nil
+}

infrastructure/network/netadapter/server/grpcserver/protowire/messages.pb.go

@@ -127,6 +127,8 @@ type KaspadMessage struct {
 	//	*KaspadMessage_UnbanResponse
 	//	*KaspadMessage_GetInfoRequest
 	//	*KaspadMessage_GetInfoResponse
+	//	*KaspadMessage_StopNotifyingUtxosChangedRequest
+	//	*KaspadMessage_StopNotifyingUtxosChangedResponse
 	Payload isKaspadMessage_Payload `protobuf_oneof:"payload"`
 }
 
@@ -841,6 +843,20 @@ func (x *KaspadMessage) GetGetInfoResponse() *GetInfoResponseMessage {
 	return nil
 }
 
+func (x *KaspadMessage) GetStopNotifyingUtxosChangedRequest() *StopNotifyingUtxosChangedRequestMessage {
+	if x, ok := x.GetPayload().(*KaspadMessage_StopNotifyingUtxosChangedRequest); ok {
+		return x.StopNotifyingUtxosChangedRequest
+	}
+	return nil
+}
+
+func (x *KaspadMessage) GetStopNotifyingUtxosChangedResponse() *StopNotifyingUtxosChangedResponseMessage {
+	if x, ok := x.GetPayload().(*KaspadMessage_StopNotifyingUtxosChangedResponse); ok {
+		return x.StopNotifyingUtxosChangedResponse
+	}
+	return nil
+}
+
 type isKaspadMessage_Payload interface {
 	isKaspadMessage_Payload()
 }
@@ -1229,6 +1245,14 @@ type KaspadMessage_GetInfoResponse struct {
 	GetInfoResponse *GetInfoResponseMessage `protobuf:"bytes,1064,opt,name=getInfoResponse,proto3,oneof"`
 }
 
+type KaspadMessage_StopNotifyingUtxosChangedRequest struct {
+	StopNotifyingUtxosChangedRequest *StopNotifyingUtxosChangedRequestMessage `protobuf:"bytes,1065,opt,name=stopNotifyingUtxosChangedRequest,proto3,oneof"`
+}
+
+type KaspadMessage_StopNotifyingUtxosChangedResponse struct {
+	StopNotifyingUtxosChangedResponse *StopNotifyingUtxosChangedResponseMessage `protobuf:"bytes,1066,opt,name=stopNotifyingUtxosChangedResponse,proto3,oneof"`
+}
+
 func (*KaspadMessage_Addresses) isKaspadMessage_Payload() {}
 
 func (*KaspadMessage_Block) isKaspadMessage_Payload() {}
@@ -1421,13 +1445,17 @@ func (*KaspadMessage_GetInfoRequest) isKaspadMessage_Payload() {}
 
 func (*KaspadMessage_GetInfoResponse) isKaspadMessage_Payload() {}
 
+func (*KaspadMessage_StopNotifyingUtxosChangedRequest) isKaspadMessage_Payload() {}
+
+func (*KaspadMessage_StopNotifyingUtxosChangedResponse) isKaspadMessage_Payload() {}
+
 var File_messages_proto protoreflect.FileDescriptor
 
 var file_messages_proto_rawDesc = []byte{
 	0x0a, 0x0e, 0x6d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x73, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f,
 	0x12, 0x09, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x77, 0x69, 0x72, 0x65, 0x1a, 0x09, 0x70, 0x32, 0x70,
 	0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x1a, 0x09, 0x72, 0x70, 0x63, 0x2e, 0x70, 0x72, 0x6f, 0x74,
-	0x6f, 0x22, 0xe0, 0x4c, 0x0a, 0x0d, 0x4b, 0x61, 0x73, 0x70, 0x61, 0x64, 0x4d, 0x65, 0x73, 0x73,
+	0x6f, 0x22, 0xeb, 0x4e, 0x0a, 0x0d, 0x4b, 0x61, 0x73, 0x70, 0x61, 0x64, 0x4d, 0x65, 0x73, 0x73,
 	0x61, 0x67, 0x65, 0x12, 0x3b, 0x0a, 0x09, 0x61, 0x64, 0x64, 0x72, 0x65, 0x73, 0x73, 0x65, 0x73,
 	0x18, 0x01, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x1b, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x77, 0x69,
 	0x72, 0x65, 0x2e, 0x41, 0x64, 0x64, 0x72, 0x65, 0x73, 0x73, 0x65, 0x73, 0x4d, 0x65, 0x73, 0x73,
@@ -2040,21 +2068,38 @@ var file_messages_proto_rawDesc = []byte{
 	0x01, 0x28, 0x0b, 0x32, 0x21, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x77, 0x69, 0x72, 0x65, 0x2e,
 	0x47, 0x65, 0x74, 0x49, 0x6e, 0x66, 0x6f, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x4d,
 	0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x48, 0x00, 0x52, 0x0f, 0x67, 0x65, 0x74, 0x49, 0x6e, 0x66,
-	0x6f, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x42, 0x09, 0x0a, 0x07, 0x70, 0x61, 0x79,
+	0x6f, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x12, 0x81, 0x01, 0x0a, 0x20, 0x73, 0x74,
-	0x6c, 0x6f, 0x61, 0x64, 0x32, 0x50, 0x0a, 0x03, 0x50, 0x32, 0x50, 0x12, 0x49, 0x0a, 0x0d, 0x4d,
+	0x6f, 0x70, 0x4e, 0x6f, 0x74, 0x69, 0x66, 0x79, 0x69, 0x6e, 0x67, 0x55, 0x74, 0x78, 0x6f, 0x73,
-	0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x53, 0x74, 0x72, 0x65, 0x61, 0x6d, 0x12, 0x18, 0x2e, 0x70,
+	0x43, 0x68, 0x61, 0x6e, 0x67, 0x65, 0x64, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x18, 0xa9,
-	0x72, 0x6f, 0x74, 0x6f, 0x77, 0x69, 0x72, 0x65, 0x2e, 0x4b, 0x61, 0x73, 0x70, 0x61, 0x64, 0x4d,
+	0x08, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x32, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x77, 0x69, 0x72,
-	0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x1a, 0x18, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x77, 0x69,
+	0x65, 0x2e, 0x53, 0x74, 0x6f, 0x70, 0x4e, 0x6f, 0x74, 0x69, 0x66, 0x79, 0x69, 0x6e, 0x67, 0x55,
-	0x72, 0x65, 0x2e, 0x4b, 0x61, 0x73, 0x70, 0x61, 0x64, 0x4d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65,
+	0x74, 0x78, 0x6f, 0x73, 0x43, 0x68, 0x61, 0x6e, 0x67, 0x65, 0x64, 0x52, 0x65, 0x71, 0x75, 0x65,
-	0x22, 0x00, 0x28, 0x01, 0x30, 0x01, 0x32, 0x50, 0x0a, 0x03, 0x52, 0x50, 0x43, 0x12, 0x49, 0x0a,
+	0x73, 0x74, 0x4d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x48, 0x00, 0x52, 0x20, 0x73, 0x74, 0x6f,
-	0x0d, 0x4d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x53, 0x74, 0x72, 0x65, 0x61, 0x6d, 0x12, 0x18,
+	0x70, 0x4e, 0x6f, 0x74, 0x69, 0x66, 0x79, 0x69, 0x6e, 0x67, 0x55, 0x74, 0x78, 0x6f, 0x73, 0x43,
-	0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x77, 0x69, 0x72, 0x65, 0x2e, 0x4b, 0x61, 0x73, 0x70, 0x61,
+	0x68, 0x61, 0x6e, 0x67, 0x65, 0x64, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x12, 0x84, 0x01,
-	0x64, 0x4d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x1a, 0x18, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f,
+	0x0a, 0x21, 0x73, 0x74, 0x6f, 0x70, 0x4e, 0x6f, 0x74, 0x69, 0x66, 0x79, 0x69, 0x6e, 0x67, 0x55,
-	0x77, 0x69, 0x72, 0x65, 0x2e, 0x4b, 0x61, 0x73, 0x70, 0x61, 0x64, 0x4d, 0x65, 0x73, 0x73, 0x61,
+	0x74, 0x78, 0x6f, 0x73, 0x43, 0x68, 0x61, 0x6e, 0x67, 0x65, 0x64, 0x52, 0x65, 0x73, 0x70, 0x6f,
-	0x67, 0x65, 0x22, 0x00, 0x28, 0x01, 0x30, 0x01, 0x42, 0x26, 0x5a, 0x24, 0x67, 0x69, 0x74, 0x68,
+	0x6e, 0x73, 0x65, 0x18, 0xaa, 0x08, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x33, 0x2e, 0x70, 0x72, 0x6f,
-	0x75, 0x62, 0x2e, 0x63, 0x6f, 0x6d, 0x2f, 0x6b, 0x61, 0x73, 0x70, 0x61, 0x6e, 0x65, 0x74, 0x2f,
+	0x74, 0x6f, 0x77, 0x69, 0x72, 0x65, 0x2e, 0x53, 0x74, 0x6f, 0x70, 0x4e, 0x6f, 0x74, 0x69, 0x66,
-	0x6b, 0x61, 0x73, 0x70, 0x61, 0x64, 0x2f, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x77, 0x69, 0x72, 0x65,
+	0x79, 0x69, 0x6e, 0x67, 0x55, 0x74, 0x78, 0x6f, 0x73, 0x43, 0x68, 0x61, 0x6e, 0x67, 0x65, 0x64,
-	0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33,
+	0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x4d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x48,
+	0x00, 0x52, 0x21, 0x73, 0x74, 0x6f, 0x70, 0x4e, 0x6f, 0x74, 0x69, 0x66, 0x79, 0x69, 0x6e, 0x67,
+	0x55, 0x74, 0x78, 0x6f, 0x73, 0x43, 0x68, 0x61, 0x6e, 0x67, 0x65, 0x64, 0x52, 0x65, 0x73, 0x70,
+	0x6f, 0x6e, 0x73, 0x65, 0x42, 0x09, 0x0a, 0x07, 0x70, 0x61, 0x79, 0x6c, 0x6f, 0x61, 0x64, 0x32,
+	0x50, 0x0a, 0x03, 0x50, 0x32, 0x50, 0x12, 0x49, 0x0a, 0x0d, 0x4d, 0x65, 0x73, 0x73, 0x61, 0x67,
+	0x65, 0x53, 0x74, 0x72, 0x65, 0x61, 0x6d, 0x12, 0x18, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x77,
+	0x69, 0x72, 0x65, 0x2e, 0x4b, 0x61, 0x73, 0x70, 0x61, 0x64, 0x4d, 0x65, 0x73, 0x73, 0x61, 0x67,
+	0x65, 0x1a, 0x18, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x77, 0x69, 0x72, 0x65, 0x2e, 0x4b, 0x61,
+	0x73, 0x70, 0x61, 0x64, 0x4d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x22, 0x00, 0x28, 0x01, 0x30,
+	0x01, 0x32, 0x50, 0x0a, 0x03, 0x52, 0x50, 0x43, 0x12, 0x49, 0x0a, 0x0d, 0x4d, 0x65, 0x73, 0x73,
+	0x61, 0x67, 0x65, 0x53, 0x74, 0x72, 0x65, 0x61, 0x6d, 0x12, 0x18, 0x2e, 0x70, 0x72, 0x6f, 0x74,
+	0x6f, 0x77, 0x69, 0x72, 0x65, 0x2e, 0x4b, 0x61, 0x73, 0x70, 0x61, 0x64, 0x4d, 0x65, 0x73, 0x73,
+	0x61, 0x67, 0x65, 0x1a, 0x18, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x77, 0x69, 0x72, 0x65, 0x2e,
+	0x4b, 0x61, 0x73, 0x70, 0x61, 0x64, 0x4d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x22, 0x00, 0x28,
+	0x01, 0x30, 0x01, 0x42, 0x26, 0x5a, 0x24, 0x67, 0x69, 0x74, 0x68, 0x75, 0x62, 0x2e, 0x63, 0x6f,
+	0x6d, 0x2f, 0x6b, 0x61, 0x73, 0x70, 0x61, 0x6e, 0x65, 0x74, 0x2f, 0x6b, 0x61, 0x73, 0x70, 0x61,
+	0x64, 0x2f, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x77, 0x69, 0x72, 0x65, 0x62, 0x06, 0x70, 0x72, 0x6f,
+	0x74, 0x6f, 0x33,
 }
 
 var (
@@ -2167,6 +2212,8 @@ var file_messages_proto_goTypes = []interface{}{
 	(*UnbanResponseMessage)(nil),                                       // 93: protowire.UnbanResponseMessage
 	(*GetInfoRequestMessage)(nil),                                      // 94: protowire.GetInfoRequestMessage
 	(*GetInfoResponseMessage)(nil),                                     // 95: protowire.GetInfoResponseMessage
+	(*StopNotifyingUtxosChangedRequestMessage)(nil),                    // 96: protowire.StopNotifyingUtxosChangedRequestMessage
+	(*StopNotifyingUtxosChangedResponseMessage)(nil),                   // 97: protowire.StopNotifyingUtxosChangedResponseMessage
 }
 var file_messages_proto_depIdxs = []int32{
 	1,   // 0: protowire.KaspadMessage.addresses:type_name -> protowire.AddressesMessage
@@ -2265,15 +2312,17 @@ var file_messages_proto_depIdxs = []int32{
 	93,  // 93: protowire.KaspadMessage.unbanResponse:type_name -> protowire.UnbanResponseMessage
 	94,  // 94: protowire.KaspadMessage.getInfoRequest:type_name -> protowire.GetInfoRequestMessage
 	95,  // 95: protowire.KaspadMessage.getInfoResponse:type_name -> protowire.GetInfoResponseMessage
-	0,  // 96: protowire.P2P.MessageStream:input_type -> protowire.KaspadMessage
+	96,  // 96: protowire.KaspadMessage.stopNotifyingUtxosChangedRequest:type_name -> protowire.StopNotifyingUtxosChangedRequestMessage
-	0,  // 97: protowire.RPC.MessageStream:input_type -> protowire.KaspadMessage
+	97,  // 97: protowire.KaspadMessage.stopNotifyingUtxosChangedResponse:type_name -> protowire.StopNotifyingUtxosChangedResponseMessage
-	0,  // 98: protowire.P2P.MessageStream:output_type -> protowire.KaspadMessage
+	0,   // 98: protowire.P2P.MessageStream:input_type -> protowire.KaspadMessage
-	0,  // 99: protowire.RPC.MessageStream:output_type -> protowire.KaspadMessage
+	0,   // 99: protowire.RPC.MessageStream:input_type -> protowire.KaspadMessage
-	98, // [98:100] is the sub-list for method output_type
+	0,   // 100: protowire.P2P.MessageStream:output_type -> protowire.KaspadMessage
-	96, // [96:98] is the sub-list for method input_type
+	0,   // 101: protowire.RPC.MessageStream:output_type -> protowire.KaspadMessage
-	96, // [96:96] is the sub-list for extension type_name
+	100, // [100:102] is the sub-list for method output_type
-	96, // [96:96] is the sub-list for extension extendee
+	98,  // [98:100] is the sub-list for method input_type
-	0,  // [0:96] is the sub-list for field type_name
+	98,  // [98:98] is the sub-list for extension type_name
+	98,  // [98:98] is the sub-list for extension extendee
+	0,   // [0:98] is the sub-list for field type_name
 }
 
 func init() { file_messages_proto_init() }
@@ -2394,6 +2443,8 @@ func file_messages_proto_init() {
 		(*KaspadMessage_UnbanResponse)(nil),
 		(*KaspadMessage_GetInfoRequest)(nil),
 		(*KaspadMessage_GetInfoResponse)(nil),
+		(*KaspadMessage_StopNotifyingUtxosChangedRequest)(nil),
+		(*KaspadMessage_StopNotifyingUtxosChangedResponse)(nil),
 	}
 	type x struct{}
 	out := protoimpl.TypeBuilder{

infrastructure/network/netadapter/server/grpcserver/protowire/messages.proto

@@ -105,6 +105,8 @@ message KaspadMessage {
     UnbanResponseMessage unbanResponse = 1062;
     GetInfoRequestMessage getInfoRequest = 1063;
     GetInfoResponseMessage getInfoResponse = 1064;
+    StopNotifyingUtxosChangedRequestMessage stopNotifyingUtxosChangedRequest = 1065;
+    StopNotifyingUtxosChangedResponseMessage stopNotifyingUtxosChangedResponse = 1066;
   }
 }
 

infrastructure/network/netadapter/server/grpcserver/protowire/rpc.md

@@ -68,6 +68,8 @@
   - [NotifyUtxosChangedResponseMessage](#protowire.NotifyUtxosChangedResponseMessage)
   - [UtxosChangedNotificationMessage](#protowire.UtxosChangedNotificationMessage)
   - [UtxosByAddressesEntry](#protowire.UtxosByAddressesEntry)
+  - [StopNotifyingUtxosChangedRequestMessage](#protowire.StopNotifyingUtxosChangedRequestMessage)
+  - [StopNotifyingUtxosChangedResponseMessage](#protowire.StopNotifyingUtxosChangedResponseMessage)
   - [RpcTransaction](#protowire.RpcTransaction)
   - [RpcTransactionInput](#protowire.RpcTransactionInput)
   - [RpcScriptPublicKey](#protowire.RpcScriptPublicKey)
@@ -857,7 +859,6 @@ kaspad's current virtual.
 | Field | Type | Label | Description |
 | ----- | ---- | ----- | ----------- |
 | blockHashes | [string](#string) | repeated |  |
-| blockHexes | [string](#string) | repeated |  |
 | blockVerboseData | [BlockVerboseData](#protowire.BlockVerboseData) | repeated |  |
 | error | [RPCError](#protowire.RPCError) |  |  |
 
@@ -1128,25 +1129,39 @@ See: NotifyUtxosChangedRequestMessage
 
 ### UtxosByAddressesEntry
 
-
-
 | Field | Type | Label | Description |
 | ----- | ---- | ----- | ----------- |
 | address | [string](#string) |  |  |
 | outpoint | [RpcOutpoint](#protowire.RpcOutpoint) |  |  |
 | utxoEntry | [RpcUtxoEntry](#protowire.RpcUtxoEntry) |  |  |
 
+<a name="protowire.StopNotifyingUtxosChangedRequestMessage"></a>
 
+### StopNotifyingUtxosChangedRequestMessage
 
+StopNotifyingUtxosChangedRequestMessage unregisters this connection for utxoChanged notifications for the given
+addresses.
 
+This call is only available when this kaspad was started with `--utxoindex`
 
+See: UtxosChangedNotificationMessage
+
+| Field | Type | Label | Description |
+| ----- | ---- | ----- | ----------- |
+| addresses | [string](#string) | repeated |  |
+
+<a name="protowire.StopNotifyingUtxosChangedResponseMessage"></a>
+
+### StopNotifyingUtxosChangedResponseMessage
+
+| Field | Type | Label | Description |
+| ----- | ---- | ----- | ----------- |
+| error | [RPCError](#protowire.RPCError) |  |  |
 
 <a name="protowire.RpcTransaction"></a>
 
 ### RpcTransaction
 
-
-
 | Field | Type | Label | Description |
 | ----- | ---- | ----- | ----------- |
 | version | [uint32](#uint32) |  |  |

infrastructure/network/netadapter/server/grpcserver/protowire/rpc.proto

@@ -450,6 +450,20 @@ message UtxosByAddressesEntry {
   RpcUtxoEntry utxoEntry = 3;
 }
 
+// StopNotifyingUtxosChangedRequestMessage unregisters this connection for utxoChanged notifications
+// for the given addresses.
+//
+// This call is only available when this kaspad was started with `--utxoindex`
+//
+// See: UtxosChangedNotificationMessage
+message StopNotifyingUtxosChangedRequestMessage {
+  repeated string addresses = 1;
+}
+
+message StopNotifyingUtxosChangedResponseMessage {
+  RPCError error = 1000;
+}
+
 message RpcTransaction {
   uint32 version = 1;
   repeated RpcTransactionInput inputs = 2;

infrastructure/network/netadapter/server/grpcserver/protowire/rpc_stop_notifying_utxos_changed.go

@@ -0,0 +1,39 @@
+package protowire
+
+import (
+	"github.com/kaspanet/kaspad/app/appmessage"
+)
+
+func (x *KaspadMessage_StopNotifyingUtxosChangedRequest) toAppMessage() (appmessage.Message, error) {
+	return &appmessage.StopNotifyingUTXOsChangedRequestMessage{
+		Addresses: x.StopNotifyingUtxosChangedRequest.Addresses,
+	}, nil
+}
+
+func (x *KaspadMessage_StopNotifyingUtxosChangedRequest) fromAppMessage(message *appmessage.StopNotifyingUTXOsChangedRequestMessage) error {
+	x.StopNotifyingUtxosChangedRequest = &StopNotifyingUtxosChangedRequestMessage{
+		Addresses: message.Addresses,
+	}
+	return nil
+}
+
+func (x *KaspadMessage_StopNotifyingUtxosChangedResponse) toAppMessage() (appmessage.Message, error) {
+	var err *appmessage.RPCError
+	if x.StopNotifyingUtxosChangedResponse.Error != nil {
+		err = &appmessage.RPCError{Message: x.StopNotifyingUtxosChangedResponse.Error.Message}
+	}
+	return &appmessage.StopNotifyingUTXOsChangedResponseMessage{
+		Error: err,
+	}, nil
+}
+
+func (x *KaspadMessage_StopNotifyingUtxosChangedResponse) fromAppMessage(message *appmessage.StopNotifyingUTXOsChangedResponseMessage) error {
+	var err *RPCError
+	if message.Error != nil {
+		err = &RPCError{Message: message.Error.Message}
+	}
+	x.StopNotifyingUtxosChangedResponse = &StopNotifyingUtxosChangedResponseMessage{
+		Error: err,
+	}
+	return nil
+}

infrastructure/network/netadapter/server/grpcserver/protowire/wire.go

@@ -643,6 +643,20 @@ func toRPCPayload(message appmessage.Message) (isKaspadMessage_Payload, error) {
 			return nil, err
 		}
 		return payload, nil
+	case *appmessage.StopNotifyingUTXOsChangedRequestMessage:
+		payload := new(KaspadMessage_StopNotifyingUtxosChangedRequest)
+		err := payload.fromAppMessage(message)
+		if err != nil {
+			return nil, err
+		}
+		return payload, nil
+	case *appmessage.StopNotifyingUTXOsChangedResponseMessage:
+		payload := new(KaspadMessage_StopNotifyingUtxosChangedResponse)
+		err := payload.fromAppMessage(message)
+		if err != nil {
+			return nil, err
+		}
+		return payload, nil
 	case *appmessage.GetUTXOsByAddressesRequestMessage:
 		payload := new(KaspadMessage_GetUtxosByAddressesRequest)
 		err := payload.fromAppMessage(message)