Rc/barnard patch 3

cmd/notifier/config/config.toml

@@ -3,6 +3,10 @@
     # Requires a redis instance/cluster and should be used when multiple observers push from the same shard
     CheckDuplicates = true
 
+    # WithReadStateChanges signals if read state changes operation will be handled
+    # It depends also if read state changes are enabled from observer nodes
+    WithReadStateChanges = false
+
     # ExternalMarshaller is used for handling incoming/outcoming api requests 
     [General.ExternalMarshaller]
         Type = "json"
@@ -114,3 +118,8 @@
     [RabbitMQ.BlockEventsExchange]
         Name = "block_events"
         Type = "fanout"
+
+    # The exchange which holds state accesses
+    [RabbitMQ.StateAccessesExchange]
+        Name = "state_accesses"
+        Type = "fanout"

common/constants.go

@@ -34,6 +34,9 @@ const (
 
 	// BlockScrs defines the subscription event type for block scrs
 	BlockScrs string = "block_scrs"
+
+	// BlockStateAccesses defines the subscription event type for block state accesses
+	BlockStateAccesses string = "block_state_accesses"
 )
 
 const (

config/config.go

@@ -20,9 +20,10 @@ type MainConfig struct {
 
 // GeneralConfig maps the general config section
 type GeneralConfig struct {
-	ExternalMarshaller MarshallerConfig
-	AddressConverter   AddressConverterConfig
-	CheckDuplicates    bool
+	ExternalMarshaller   MarshallerConfig
+	AddressConverter     AddressConverterConfig
+	CheckDuplicates      bool
+	WithReadStateChanges bool
 }
 
 // MarshallerConfig maps the marshaller configuration
@@ -80,6 +81,7 @@ type RabbitMQConfig struct {
 	BlockTxsExchange        RabbitMQExchangeConfig
 	BlockScrsExchange       RabbitMQExchangeConfig
 	BlockEventsExchange     RabbitMQExchangeConfig
+	StateAccessesExchange   RabbitMQExchangeConfig
 }
 
 // RabbitMQExchangeConfig holds the configuration for a rabbitMQ exchange

config/tomlConfig_test.go

@@ -17,6 +17,7 @@ func TestMainConfig(t *testing.T) {
 	adrConverterPrefix := "erd"
 	adrConverterLength := 32
 	checkDuplicates := true
+	withReadStateChanges := true
 
 	connectorAPIHost := "5000"
 	connectorAPIUsername := "guest"
@@ -50,7 +51,8 @@ func TestMainConfig(t *testing.T) {
 				Prefix: adrConverterPrefix,
 				Length: adrConverterLength,
 			},
-			CheckDuplicates: checkDuplicates,
+			CheckDuplicates:      checkDuplicates,
+			WithReadStateChanges: withReadStateChanges,
 		},
 		WebSocketConnector: config.WebSocketConfig{
 			Enabled:                    true,
@@ -95,6 +97,10 @@ func TestMainConfig(t *testing.T) {
     # Requires a redis instance/cluster and should be used when multiple observers push from the same shard
     CheckDuplicates = true
 
+    # WithReadStateChanges signals if read state changes operation will be handled
+    # It depends also if read state changes are enabled from observer nodes
+    WithReadStateChanges = true
+
     # ExternalMarshaller is used for handling incoming/outcoming api requests 
     [General.ExternalMarshaller]
         Type = "` + generalMarshallerType + `"

data/block.go

@@ -9,6 +9,7 @@ import (
 	"github.com/multiversx/mx-chain-core-go/data/receipt"
 	"github.com/multiversx/mx-chain-core-go/data/rewardTx"
 	"github.com/multiversx/mx-chain-core-go/data/smartContractResult"
+	"github.com/multiversx/mx-chain-core-go/data/stateChange"
 	"github.com/multiversx/mx-chain-core-go/data/transaction"
 )
 
@@ -22,14 +23,15 @@ type SaveBlockData struct {
 
 // InterceptorBlockData holds the block data needed for processing
 type InterceptorBlockData struct {
-	Hash          string
-	Body          nodeData.BodyHandler
-	Header        nodeData.HeaderHandler
-	Txs           map[string]*transaction.Transaction
-	TxsWithOrder  map[string]*outport.TxInfo
-	Scrs          map[string]*smartContractResult.SmartContractResult
-	ScrsWithOrder map[string]*outport.SCRInfo
-	LogEvents     []Event
+	Hash                     string
+	Body                     nodeData.BodyHandler
+	Header                   nodeData.HeaderHandler
+	Txs                      map[string]*transaction.Transaction
+	TxsWithOrder             map[string]*outport.TxInfo
+	Scrs                     map[string]*smartContractResult.SmartContractResult
+	ScrsWithOrder            map[string]*outport.SCRInfo
+	LogEvents                []Event
+	StateAccessesPerAccounts map[string]*stateChange.StateAccesses
 }
 
 // ArgsSaveBlockData holds the block data that will be received on push events
@@ -44,6 +46,7 @@ type ArgsSaveBlockData struct {
 	AlteredAccounts        map[string]*alteredAccount.AlteredAccount
 	NumberOfShards         uint32
 	HeaderTimeStampMs      uint64
+	StateAccesses          map[string]*stateChange.StateAccesses
 }
 
 // OutportBlockDataOld holds the block data that will be received on push events

data/outport.go

@@ -7,6 +7,7 @@ import (
 	"github.com/multiversx/mx-chain-core-go/data/receipt"
 	"github.com/multiversx/mx-chain-core-go/data/rewardTx"
 	"github.com/multiversx/mx-chain-core-go/data/smartContractResult"
+	"github.com/multiversx/mx-chain-core-go/data/stateChange"
 	"github.com/multiversx/mx-chain-core-go/data/transaction"
 )
 
@@ -73,6 +74,15 @@ type BlockEventsWithOrder struct {
 	Events      []Event                     `json:"events"`
 }
 
+// BlockStateAccesses holds the block state accesses
+type BlockStateAccesses struct {
+	Hash                     string                                `json:"hash"`
+	ShardID                  uint32                                `json:"shardID"`
+	TimeStampMs              uint64                                `json:"timestampMs"`
+	Nonce                    uint64                                `json:"nonce"`
+	StateAccessesPerAccounts map[string]*stateChange.StateAccesses `json:"stateAccessesPerAccounts"`
+}
+
 // NotifierTransaction defines a wrapper over transaction
 type NotifierTransaction struct {
 	*transaction.Transaction

disabled/disabledHub.go

@@ -33,6 +33,10 @@ func (h *Hub) PublishScrs(blockScrs data.BlockScrs) {
 func (h *Hub) PublishBlockEventsWithOrder(blockTxs data.BlockEventsWithOrder) {
 }
 
+// PublishStateAccesses does nothing
+func (h *Hub) PublishStateAccesses(blockTxs data.BlockStateAccesses) {
+}
+
 // RegisterEvent does nothing
 func (h *Hub) RegisterEvent(_ dispatcher.EventDispatcher) {
 }

dispatcher/hub/commonHub.go

@@ -96,6 +96,10 @@ func (ch *commonHub) handlePushBlockEvents(blockEvents data.BlockEvents, subscri
 // PublishRevert will publish revert event to dispatcher
 func (ch *commonHub) PublishRevert(revertBlock data.RevertBlock) {
 	subscriptions := ch.subscriptionMapper.Subscriptions()
+	_, ok := subscriptions[common.RevertBlockEvents]
+	if !ok {
+		return
+	}
 
 	dispatchersMap := make(map[uuid.UUID]data.RevertBlock)
 
@@ -115,6 +119,10 @@ func (ch *commonHub) PublishRevert(revertBlock data.RevertBlock) {
 // PublishFinalized will publish finalized event to dispatcher
 func (ch *commonHub) PublishFinalized(finalizedBlock data.FinalizedBlock) {
 	subscriptions := ch.subscriptionMapper.Subscriptions()
+	_, ok := subscriptions[common.FinalizedBlockEvents]
+	if !ok {
+		return
+	}
 
 	dispatchersMap := make(map[uuid.UUID]data.FinalizedBlock)
 
@@ -134,6 +142,10 @@ func (ch *commonHub) PublishFinalized(finalizedBlock data.FinalizedBlock) {
 // PublishTxs will publish txs event to dispatcher
 func (ch *commonHub) PublishTxs(blockTxs data.BlockTxs) {
 	subscriptions := ch.subscriptionMapper.Subscriptions()
+	_, ok := subscriptions[common.BlockTxs]
+	if !ok {
+		return
+	}
 
 	dispatchersMap := make(map[uuid.UUID]data.BlockTxs)
 
@@ -153,6 +165,10 @@ func (ch *commonHub) PublishTxs(blockTxs data.BlockTxs) {
 // PublishBlockEventsWithOrder will publish block events with order to dispatcher
 func (ch *commonHub) PublishBlockEventsWithOrder(blockTxs data.BlockEventsWithOrder) {
 	subscriptions := ch.subscriptionMapper.Subscriptions()
+	_, ok := subscriptions[common.BlockEvents]
+	if !ok {
+		return
+	}
 
 	dispatchersMap := make(map[uuid.UUID]data.BlockEventsWithOrder)
 
@@ -172,6 +188,10 @@ func (ch *commonHub) PublishBlockEventsWithOrder(blockTxs data.BlockEventsWithOr
 // PublishScrs will publish scrs events to dispatcher
 func (ch *commonHub) PublishScrs(blockScrs data.BlockScrs) {
 	subscriptions := ch.subscriptionMapper.Subscriptions()
+	_, ok := subscriptions[common.BlockScrs]
+	if !ok {
+		return
+	}
 
 	dispatchersMap := make(map[uuid.UUID]data.BlockScrs)
 
@@ -188,6 +208,29 @@ func (ch *commonHub) PublishScrs(blockScrs data.BlockScrs) {
 	}
 }
 
+// PublishStateAccesses will publish state accesses to dispatcher
+func (ch *commonHub) PublishStateAccesses(stateAccesses data.BlockStateAccesses) {
+	subscriptions := ch.subscriptionMapper.Subscriptions()
+	_, ok := subscriptions[common.BlockStateAccesses]
+	if !ok {
+		return
+	}
+
+	dispatchersMap := make(map[uuid.UUID]data.BlockStateAccesses)
+
+	for _, subscription := range subscriptions[common.BlockStateAccesses] {
+		dispatchersMap[subscription.DispatcherID] = stateAccesses
+	}
+
+	ch.mutDispatchers.RLock()
+	defer ch.mutDispatchers.RUnlock()
+	for id, event := range dispatchersMap {
+		if d, ok := ch.dispatchers[id]; ok {
+			d.StateAccessesEvent(event)
+		}
+	}
+}
+
 func (ch *commonHub) registerDispatcher(d dispatcher.EventDispatcher) {
 	ch.mutDispatchers.Lock()
 	defer ch.mutDispatchers.Unlock()

dispatcher/interface.go

@@ -19,6 +19,7 @@ type EventDispatcher interface {
 	TxsEvent(event data.BlockTxs)
 	BlockEvents(event data.BlockEventsWithOrder)
 	ScrsEvent(event data.BlockScrs)
+	StateAccessesEvent(event data.BlockStateAccesses)
 }
 
 // Hub defines the behaviour of a component which should be able to receive events

dispatcher/subscription.go

@@ -145,7 +145,8 @@ func getEventType(subEntry data.SubscriptionEntry) string {
 		subEntry.EventType == common.RevertBlockEvents ||
 		subEntry.EventType == common.BlockTxs ||
 		subEntry.EventType == common.BlockScrs ||
-		subEntry.EventType == common.BlockEvents {
+		subEntry.EventType == common.BlockEvents ||
+		subEntry.EventType == common.BlockStateAccesses {
 		return subEntry.EventType
 	}
 

dispatcher/ws/wsDispatcher.go

@@ -193,6 +193,26 @@ func (wd *websocketDispatcher) ScrsEvent(event data.BlockScrs) {
 	wd.send <- wsEventBytes
 }
 
+// StateAccessesEvent receives a block state accesses event and process it before pushing to socket
+func (wd *websocketDispatcher) StateAccessesEvent(event data.BlockStateAccesses) {
+	eventBytes, err := wd.marshaller.Marshal(event)
+	if err != nil {
+		log.Error("failure marshalling events", "err", err.Error())
+		return
+	}
+	wsEvent := &data.WebSocketEvent{
+		Type: common.BlockStateAccesses,
+		Data: eventBytes,
+	}
+	wsEventBytes, err := wd.marshaller.Marshal(wsEvent)
+	if err != nil {
+		log.Error("failure marshalling events", "err", err.Error())
+		return
+	}
+
+	wd.send <- wsEventBytes
+}
+
 // writePump listens on the send-channel and pushes data on the socket stream
 func (wd *websocketDispatcher) writePump() {
 	ticker := time.NewTicker(pingPeriod)

factory/processFactory.go

@@ -24,7 +24,8 @@ func CreateEventsInterceptor(cfg config.GeneralConfig) (process.EventsIntercepto
 	}
 
 	argsEventsInterceptor := process.ArgsEventsInterceptor{
-		PubKeyConverter: pubKeyConverter,
+		PubKeyConverter:      pubKeyConverter,
+		WithReadStateChanges: cfg.WithReadStateChanges,
 	}
 
 	return process.NewEventsInterceptor(argsEventsInterceptor)

go.mod

@@ -9,7 +9,7 @@ require (
 	github.com/google/uuid v1.6.0
 	github.com/gorilla/websocket v1.5.3
 	github.com/multiversx/mx-chain-communication-go v1.2.0
-	github.com/multiversx/mx-chain-core-go v1.4.0
+	github.com/multiversx/mx-chain-core-go v1.4.1-0.20250924085417-30f7c94f7d5d
 	github.com/multiversx/mx-chain-logger-go v1.1.0
 	github.com/pelletier/go-toml v1.9.3
 	github.com/prometheus/client_model v0.6.1

go.sum

@@ -130,8 +130,8 @@ github.com/mr-tron/base58 v1.2.0 h1:T/HDJBh4ZCPbU39/+c3rRvE0uKBQlU27+QI8LJ4t64o=
 github.com/mr-tron/base58 v1.2.0/go.mod h1:BinMc/sQntlIE1frQmRFPUoPA1Zkr8VRgBdjWI2mNwc=
 github.com/multiversx/mx-chain-communication-go v1.2.0 h1:0wOoLldiRbvaOPxwICbnRCqCpLqPewg8M/FMbC/0OXY=
 github.com/multiversx/mx-chain-communication-go v1.2.0/go.mod h1:wS3aAwkmHbC9mlzQdvL6p7l8Rqw3vmzhj7WZW1dTveA=
-github.com/multiversx/mx-chain-core-go v1.4.0 h1:p6FbfCzvMXF54kpS0B5mrjNWYpq4SEQqo0UvrMF7YVY=
-github.com/multiversx/mx-chain-core-go v1.4.0/go.mod h1:IO+vspNan+gT0WOHnJ95uvWygiziHZvfXpff6KnxV7g=
+github.com/multiversx/mx-chain-core-go v1.4.1-0.20250924085417-30f7c94f7d5d h1:DLZo0iqbsu7iMPWu2948XvkWE9jCly5A6IFFgeGf7MQ=
+github.com/multiversx/mx-chain-core-go v1.4.1-0.20250924085417-30f7c94f7d5d/go.mod h1:IO+vspNan+gT0WOHnJ95uvWygiziHZvfXpff6KnxV7g=
 github.com/multiversx/mx-chain-crypto-go v1.2.13-0.20250218161752-9482d9a22234 h1:NNI7kYxzsq+4mTPSUJo0cK1+iPxjUX+gRJDaBRwEQ7M=
 github.com/multiversx/mx-chain-crypto-go v1.2.13-0.20250218161752-9482d9a22234/go.mod h1:QZAw2bZcOxGQRgYACTrmP8pfTa3NyxENIL+00G6nM5E=
 github.com/multiversx/mx-chain-logger-go v1.1.0 h1:97x84A6L4RfCa6YOx1HpAFxZp1cf/WI0Qh112whgZNM=