storage: split runtime events related accounts into a separate table

oasisprotocol · Dec 23, 2024 · be7cc48 · be7cc48
1 parent 0d27e93
commit be7cc48
Show file tree

Hide file tree

Showing 16 changed files with 257 additions and 179 deletions.
diff --git a/analyzer/consensus/consensus.go b/analyzer/consensus/consensus.go
@@ -797,12 +797,15 @@ func (m *processor) queueRootHashMessageUpserts(batch *storage.QueryBatch, data
 				// The runtime has its own staking account, which is what
 				// performs these actions, e.g. when sending or receiving the
 				// consensus token. Register that as related to the message.
-				if _, err := addresses.RegisterRelatedRuntimeAddress(messageData.addressPreimages, messageData.relatedAddresses, event.RoothashExecutorCommitted.RuntimeID); err != nil {
+				if runtimeAddr, err := addresses.RegisterRuntimeAddress(messageData.addressPreimages, event.RoothashExecutorCommitted.RuntimeID); err != nil {
 					logger.Info("register runtime address failed",
 						"runtime_id", event.RoothashExecutorCommitted.RuntimeID,
 						"err", err,
 					)
+				} else {
+					messageData.relatedAddresses[runtimeAddr] = struct{}{}
 				}
+
 				for addr, preimageData := range messageData.addressPreimages {
 					batch.Queue(queries.AddressPreimageInsert,
 						addr,

diff --git a/analyzer/consensus/messages.go b/analyzer/consensus/messages.go
@@ -35,13 +35,14 @@ func extractMessageData(logger *log.Logger, m message.Message) MessageData {
 				break
 			}
 			messageData.body = body
-			_, err = addresses.RegisterRelatedOCSAddress(messageData.relatedAddresses, m.Staking.Transfer.To)
+			to, err := addresses.FromOCSAddress(m.Staking.Transfer.To)
 			if err != nil {
 				logger.Info("register related address 'to' failed",
 					"message_type", messageData.messageType,
 					"err", err,
 				)
 			}
+			messageData.relatedAddresses[to] = struct{}{}
 		case m.Staking.Withdraw != nil:
 			messageData.messageType = apiTypes.RoothashMessageTypeStakingWithdraw
 			body, err := json.Marshal(m.Staking.Withdraw)
@@ -53,13 +54,14 @@ func extractMessageData(logger *log.Logger, m message.Message) MessageData {
 				break
 			}
 			messageData.body = body
-			_, err = addresses.RegisterRelatedOCSAddress(messageData.relatedAddresses, m.Staking.Withdraw.From)
+			from, err := addresses.FromOCSAddress(m.Staking.Withdraw.From)
 			if err != nil {
 				logger.Info("register related address 'from' failed",
 					"message_type", messageData.messageType,
 					"err", err,
 				)
 			}
+			messageData.relatedAddresses[from] = struct{}{}
 		case m.Staking.AddEscrow != nil:
 			messageData.messageType = apiTypes.RoothashMessageTypeStakingAddEscrow
 			body, err := json.Marshal(m.Staking.AddEscrow)
@@ -71,13 +73,14 @@ func extractMessageData(logger *log.Logger, m message.Message) MessageData {
 				break
 			}
 			messageData.body = body
-			_, err = addresses.RegisterRelatedOCSAddress(messageData.relatedAddresses, m.Staking.AddEscrow.Account)
+			account, err := addresses.FromOCSAddress(m.Staking.AddEscrow.Account)
 			if err != nil {
 				logger.Info("register related address 'account' failed",
 					"message_type", messageData.messageType,
 					"err", err,
 				)
 			}
+			messageData.relatedAddresses[account] = struct{}{}
 		case m.Staking.ReclaimEscrow != nil:
 			messageData.messageType = apiTypes.RoothashMessageTypeStakingReclaimEscrow
 			body, err := json.Marshal(m.Staking.ReclaimEscrow)
@@ -89,13 +92,14 @@ func extractMessageData(logger *log.Logger, m message.Message) MessageData {
 				break
 			}
 			messageData.body = body
-			_, err = addresses.RegisterRelatedOCSAddress(messageData.relatedAddresses, m.Staking.ReclaimEscrow.Account)
+			account, err := addresses.FromOCSAddress(m.Staking.ReclaimEscrow.Account)
 			if err != nil {
 				logger.Info("register related address 'account' failed",
 					"message_type", messageData.messageType,
 					"err", err,
 				)
 			}
+			messageData.relatedAddresses[account] = struct{}{}
 		default:
 			logger.Info("unhandled staking message",
 				"staking_message", m.Staking,

diff --git a/analyzer/evmabibackfill/evm_abi_backfill.go b/analyzer/evmabibackfill/evm_abi_backfill.go
@@ -42,8 +42,13 @@ type abiEncodedTx struct {
 }
 
 type abiEncodedEvent struct {
+	// Event primary key.
+	Runtime   common.Runtime
 	Round     uint64
 	TxIndex   *int
+	Type      string
+	TypeIndex int
+
 	EventBody sdkEVM.Event
 }
 
@@ -122,8 +127,11 @@ func (p *processor) GetItems(ctx context.Context, limit uint64) ([]*abiEncodedIt
 		if err = eventRows.Scan(
 			&item.ContractAddr,
 			&item.Abi,
+			&ev.Runtime,
 			&ev.Round,
 			&ev.TxIndex,
+			&ev.Type,
+			&ev.TypeIndex,
 			&ev.EventBody,
 		); err != nil {
 			return nil, fmt.Errorf("scanning verified contract event: %w", err)
@@ -236,10 +244,11 @@ func (p *processor) ProcessItem(ctx context.Context, batch *storage.QueryBatch,
 		if item.Event != nil {
 			batch.Queue(
 				queries.RuntimeEventEvmParsedFieldsUpdate,
-				p.runtime,
+				item.Event.Runtime,
 				item.Event.Round,
 				item.Event.TxIndex,
-				item.Event.EventBody,
+				item.Event.Type,
+				item.Event.TypeIndex,
 				nil,
 				nil,
 				nil,
@@ -267,10 +276,11 @@ func (p *processor) ProcessItem(ctx context.Context, batch *storage.QueryBatch,
 		}
 		batch.Queue(
 			queries.RuntimeEventEvmParsedFieldsUpdate,
-			p.runtime,
+			item.Event.Runtime,
 			item.Event.Round,
 			item.Event.TxIndex,
-			item.Event.EventBody,
+			item.Event.Type,
+			item.Event.TypeIndex,
 			eventName,
 			eventArgs,
 			eventSig,

diff --git a/analyzer/queries/queries.go b/analyzer/queries/queries.go
@@ -636,22 +636,27 @@ var (
       tx_hash = $2`
 
 	RuntimeEventInsert = `
-    INSERT INTO chain.runtime_events (runtime, round, tx_index, tx_hash, tx_eth_hash, timestamp, type, body, related_accounts, evm_log_name, evm_log_params, evm_log_signature)
+    INSERT INTO chain.runtime_events (runtime, round, tx_index, type, type_index, tx_hash, tx_eth_hash, timestamp, body, evm_log_name, evm_log_params, evm_log_signature)
       VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12)`
 
+	RuntimeEventRelatedAccountsInsert = `
+    INSERT INTO chain.runtime_events_related_accounts (runtime, round, tx_index, type, type_index, account_address)
+      SELECT $1, $2, $3, $4, $5, unnest($6::text[])`
+
 	// We use COALESCE here to avoid overwriting existing data with null values.
 	RuntimeEventEvmParsedFieldsUpdate = `
     UPDATE chain.runtime_events
     SET
-      evm_log_name = COALESCE($5, evm_log_name),
-      evm_log_params = COALESCE($6, evm_log_params),
-      evm_log_signature = COALESCE($7, evm_log_signature),
+      evm_log_name = COALESCE($6, evm_log_name),
+      evm_log_params = COALESCE($7, evm_log_params),
+      evm_log_signature = COALESCE($8, evm_log_signature),
       abi_parsed_at = CURRENT_TIMESTAMP
     WHERE
       runtime = $1 AND
       round = $2 AND
       tx_index = $3 AND
-      body = $4`
+      type = $4 AND
+      type_index = $5`
 
 	RuntimeMintInsert = `
     INSERT INTO chain.runtime_transfers (runtime, round, sender, receiver, symbol, amount)
@@ -1134,8 +1139,11 @@ var (
     SELECT
       abi_contracts.addr,
       abi_contracts.abi,
+      evs.runtime,
       evs.round,
       evs.tx_index,
+      evs.type,
+      evs.type_index,
       evs.body
     FROM abi_contracts
     JOIN chain.address_preimages as preimages ON