From 263f6a7fe0d3db63b818d8eced32f15d57e1bebe Mon Sep 17 00:00:00 2001 From: miiu Date: Wed, 22 Oct 2025 16:52:30 +0300 Subject: [PATCH 1/6] update history repository --- common/common.go | 5 + dblookupext/common.go | 114 +++++++ dblookupext/common_test.go | 301 ++++++++++++++++++ .../factory/historyRepositoryFactory.go | 7 + .../factory/historyRepositoryFactory_test.go | 2 + dblookupext/historyRepository.go | 141 ++++++-- dblookupext/historyRepository_test.go | 2 + process/block/baseProcess.go | 40 ++- process/block/shardblockProposal.go | 12 + 9 files changed, 590 insertions(+), 34 deletions(-) create mode 100644 dblookupext/common.go create mode 100644 dblookupext/common_test.go diff --git a/common/common.go b/common/common.go index 4921d7d9c97..6fbb0fc5bd5 100644 --- a/common/common.go +++ b/common/common.go @@ -33,6 +33,11 @@ type chainParametersHandler interface { IsInterfaceNil() bool } +// PrepareLogEventsKey will prepare logs key for cacher +func PrepareLogEventsKey(headerHash []byte) []byte { + return append([]byte("logs"), headerHash...) +} + // IsValidRelayedTxV3 returns true if the provided transaction is a valid transaction of type relayed v3 func IsValidRelayedTxV3(tx data.TransactionHandler) bool { relayedTx, isRelayedV3 := tx.(data.RelayedTransactionHandler) diff --git a/dblookupext/common.go b/dblookupext/common.go new file mode 100644 index 00000000000..85a595b611c --- /dev/null +++ b/dblookupext/common.go @@ -0,0 +1,114 @@ +package dblookupext + +import ( + "encoding/hex" + "fmt" + + "github.com/multiversx/mx-chain-core-go/data" + "github.com/multiversx/mx-chain-core-go/data/block" + "github.com/multiversx/mx-chain-core-go/marshal" + "github.com/multiversx/mx-chain-go/common" + "github.com/multiversx/mx-chain-go/process" + "github.com/multiversx/mx-chain-go/storage" +) + +func getIntermediateTxs(cache storage.Cacher, headerHash []byte) (map[string]data.TransactionHandler, map[string]data.TransactionHandler, error) { + cachedIntermediateTxs, ok := cache.Get(headerHash) + if !ok { + log.Warn("intermediateTxs not found in dataPool", "hash", headerHash) + return nil, nil, fmt.Errorf("%w for header %s", process.ErrMissingHeader, hex.EncodeToString(headerHash)) + } + + cachedIntermediateTxsMap, ok := cachedIntermediateTxs.(map[block.Type]map[string]data.TransactionHandler) + if !ok { + return make(map[string]data.TransactionHandler), make(map[string]data.TransactionHandler), nil + } + + scrs := cachedIntermediateTxsMap[block.SmartContractResultBlock] + receipts := cachedIntermediateTxsMap[block.ReceiptBlock] + + return scrs, receipts, nil +} + +func getLogs(cache storage.Cacher, headerHash []byte) ([]*data.LogData, error) { + logsKey := common.PrepareLogEventsKey(headerHash) + cachedLogs, ok := cache.Get(logsKey) + if !ok { + log.Warn("logs not found in dataPool", "hash", headerHash) + return nil, fmt.Errorf("%w for header %s", process.ErrMissingHeader, hex.EncodeToString(headerHash)) + } + cachedLogsSlice, ok := cachedLogs.([]*data.LogData) + if !ok { + return []*data.LogData{}, nil + } + return cachedLogsSlice, nil +} + +func getIntraMbs(cache storage.Cacher, marshaller marshal.Marshalizer, headerHash []byte) ([]*block.MiniBlock, error) { + cachedIntraMBs, ok := cache.Get(headerHash) + if !ok { + log.Warn("intra miniblocks not found in dataPool", "hash", headerHash) + return nil, fmt.Errorf("%w for header %s", process.ErrMissingHeader, hex.EncodeToString(headerHash)) + } + cachedLogsBuff := cachedIntraMBs.([]byte) + var intraMBs []*block.MiniBlock + errUnmarshal := marshaller.Unmarshal(&intraMBs, cachedLogsBuff) + if errUnmarshal != nil { + return nil, fmt.Errorf("%w getIntraMbs: cannot unmarshall", errUnmarshal) + } + + return intraMBs, nil +} + +func getBody(cache storage.Cacher, marshaller marshal.Marshalizer, baseExecResult data.BaseExecutionResultHandler, shardID uint32) (*block.Body, error) { + miniBlockHeaderHandlers, err := extractMiniBlocksHeaderHandlersFromExecResult(baseExecResult, shardID) + if err != nil { + return nil, err + } + + var miniBlocks block.MiniBlockSlice + for _, miniBlockHeaderHandler := range miniBlockHeaderHandlers { + mbHash := miniBlockHeaderHandler.GetHash() + cachedMiniBlock, found := cache.Get(mbHash) + if !found { + log.Warn("mini block from execution result not cached after execution", + "mini block hash", mbHash) + return nil, process.ErrMissingMiniBlock + } + + cachedMiniBlockBytes := cachedMiniBlock.([]byte) + + var miniBlock *block.MiniBlock + err = marshaller.Unmarshal(&miniBlock, cachedMiniBlockBytes) + if err != nil { + return nil, err + } + + miniBlocks = append(miniBlocks, miniBlock) + } + + return &block.Body{MiniBlocks: miniBlocks}, nil +} + +func extractMiniBlocksHeaderHandlersFromExecResult( + baseExecResult data.BaseExecutionResultHandler, + headerShard uint32, +) ([]data.MiniBlockHeaderHandler, error) { + if headerShard == common.MetachainShardId { + metaExecResult, ok := baseExecResult.(data.MetaExecutionResultHandler) + if !ok { + log.Warn("extractMiniBlocksHeaderHandlersFromExecResult assert failed to MetaExecutionResultHandler") + return nil, process.ErrWrongTypeAssertion + } + + return metaExecResult.GetMiniBlockHeadersHandlers(), nil + } + + execResult, ok := baseExecResult.(data.ExecutionResultHandler) + if !ok { + log.Warn("extractMiniBlocksHeaderHandlersFromExecResult assert failed to ExecutionResultHandler") + return nil, process.ErrWrongTypeAssertion + } + + return execResult.GetMiniBlockHeadersHandlers(), nil +} diff --git a/dblookupext/common_test.go b/dblookupext/common_test.go new file mode 100644 index 00000000000..3f56c4f064a --- /dev/null +++ b/dblookupext/common_test.go @@ -0,0 +1,301 @@ +package dblookupext + +import ( + "errors" + "strings" + "testing" + + "github.com/multiversx/mx-chain-core-go/core" + "github.com/multiversx/mx-chain-core-go/data" + "github.com/multiversx/mx-chain-core-go/data/block" + "github.com/multiversx/mx-chain-core-go/data/receipt" + "github.com/multiversx/mx-chain-core-go/data/smartContractResult" + "github.com/multiversx/mx-chain-core-go/data/transaction" + "github.com/multiversx/mx-chain-go/common" + "github.com/multiversx/mx-chain-go/process" + "github.com/multiversx/mx-chain-go/testscommon/cache" + "github.com/multiversx/mx-chain-go/testscommon/marshallerMock" + "github.com/stretchr/testify/require" +) + +func TestGetIntermediateTxs(t *testing.T) { + t.Parallel() + + t.Run("getIntermediateTxs cannot find in cache", func(t *testing.T) { + cacher := cache.NewCacherMock() + + headerHash := []byte("h") + + _, _, err := getIntermediateTxs(cacher, headerHash) + require.True(t, errors.Is(err, process.ErrMissingHeader)) + }) + + t.Run("getIntermediateTxs wrong type in cache should return empty maps", func(t *testing.T) { + cacher := cache.NewCacherMock() + + headerHash := []byte("h") + cacher.Put(headerHash, []byte("wrong"), 0) + + scrs, receipts, err := getIntermediateTxs(cacher, headerHash) + require.NoError(t, err) + require.Len(t, scrs, 0) + require.Len(t, receipts, 0) + }) + + t.Run("getIntermediateTxs should work", func(t *testing.T) { + cachedIntermediateTxsMap := map[block.Type]map[string]data.TransactionHandler{} + cachedIntermediateTxsMap[block.SmartContractResultBlock] = map[string]data.TransactionHandler{ + "h1": &smartContractResult.SmartContractResult{}, + } + cachedIntermediateTxsMap[block.ReceiptBlock] = map[string]data.TransactionHandler{ + "r1": &receipt.Receipt{}, + } + + cacher := cache.NewCacherMock() + + headerHash := []byte("h") + cacher.Put(headerHash, cachedIntermediateTxsMap, 0) + + scrs, receipts, err := getIntermediateTxs(cacher, headerHash) + require.NoError(t, err) + require.Len(t, scrs, 1) + require.Len(t, receipts, 1) + require.Equal(t, cachedIntermediateTxsMap[block.SmartContractResultBlock], scrs) + require.Equal(t, cachedIntermediateTxsMap[block.ReceiptBlock], receipts) + }) + +} + +func TestGetLogs(t *testing.T) { + t.Parallel() + + t.Run("getLogs cannot find in cache", func(t *testing.T) { + cacher := cache.NewCacherMock() + + headerHash := []byte("h") + + _, err := getLogs(cacher, headerHash) + require.True(t, errors.Is(err, process.ErrMissingHeader)) + }) + + t.Run("getLogs wrong type in cache should return empty slice", func(t *testing.T) { + cacher := cache.NewCacherMock() + + headerHash := []byte("h") + logsKey := common.PrepareLogEventsKey(headerHash) + cacher.Put(logsKey, "wrong type", 0) + + logs, err := getLogs(cacher, headerHash) + require.Nil(t, err) + require.Len(t, logs, 0) + }) + + t.Run("getLogs should work", func(t *testing.T) { + cacher := cache.NewCacherMock() + + headerHash := []byte("h") + expectedLogs := []*data.LogData{ + { + LogHandler: &transaction.Log{}, + TxHash: "t1", + }, + { + LogHandler: &transaction.Log{}, + TxHash: "t2", + }, + } + logsKey := common.PrepareLogEventsKey(headerHash) + + cacher.Put(logsKey, expectedLogs, 0) + + logs, err := getLogs(cacher, headerHash) + require.Nil(t, err) + require.Len(t, logs, 2) + require.Equal(t, expectedLogs, logs) + }) + +} + +func TestGetIntraMbs(t *testing.T) { + t.Parallel() + + t.Run("getIntraMbs cannot find in cache", func(t *testing.T) { + cacher := cache.NewCacherMock() + marshaller := &marshallerMock.MarshalizerMock{} + + headerHash := []byte("h") + + _, err := getIntraMbs(cacher, marshaller, headerHash) + require.True(t, errors.Is(err, process.ErrMissingHeader)) + }) + + t.Run("getIntraMbs wrong type should error", func(t *testing.T) { + cacher := cache.NewCacherMock() + marshaller := &marshallerMock.MarshalizerMock{} + + headerHash := []byte("h") + cacher.Put(headerHash, []byte("wrong type"), 0) + + intraMBs, err := getIntraMbs(cacher, marshaller, headerHash) + require.Nil(t, intraMBs) + require.NotNil(t, err) + require.True(t, strings.Contains(err.Error(), "getIntraMbs: cannot unmarshall")) + }) + + t.Run("getIntraMbs should work", func(t *testing.T) { + cacher := cache.NewCacherMock() + marshaller := &marshallerMock.MarshalizerMock{} + + headerHash := []byte("h") + expectedMbs := []*block.MiniBlock{ + {SenderShardID: 0}, + {SenderShardID: 0}, + } + intraMbsBytes, _ := marshaller.Marshal(expectedMbs) + + cacher.Put(headerHash, intraMbsBytes, 0) + + intraMBs, err := getIntraMbs(cacher, marshaller, headerHash) + require.Nil(t, err) + require.Equal(t, expectedMbs, intraMBs) + }) + +} + +func TestExtractMiniBlocksHeaderHandlersFromExecResult(t *testing.T) { + t.Run("wrong type shard", func(t *testing.T) { + executionResult := &block.BaseExecutionResult{} + + _, err := extractMiniBlocksHeaderHandlersFromExecResult(executionResult, 0) + require.Equal(t, process.ErrWrongTypeAssertion, err) + }) + t.Run("should work shard", func(t *testing.T) { + executionResult := &block.ExecutionResult{ + MiniBlockHeaders: []block.MiniBlockHeader{ + {SenderShardID: 1}, + {SenderShardID: 0}, + }, + } + + res, err := extractMiniBlocksHeaderHandlersFromExecResult(executionResult, 0) + require.Nil(t, err) + require.Len(t, res, 2) + }) + t.Run("wrong type meta", func(t *testing.T) { + executionResult := &block.BaseExecutionResult{} + _, err := extractMiniBlocksHeaderHandlersFromExecResult(executionResult, core.MetachainShardId) + require.Equal(t, process.ErrWrongTypeAssertion, err) + }) + t.Run("should work meta", func(t *testing.T) { + executionResult := &block.MetaExecutionResult{ + MiniBlockHeaders: []block.MiniBlockHeader{ + {SenderShardID: 0}, + {SenderShardID: 1}, + }, + } + + res, err := extractMiniBlocksHeaderHandlersFromExecResult(executionResult, core.MetachainShardId) + require.Nil(t, err) + require.Len(t, res, 2) + }) +} + +func TestGetBody(t *testing.T) { + t.Parallel() + + t.Run("cannot get mb headers should error", func(t *testing.T) { + executionResult := &block.BaseExecutionResult{} + + marshaller := &marshallerMock.MarshalizerMock{} + cacher := cache.NewCacherMock() + + _, err := getBody(cacher, marshaller, executionResult, 0) + require.NotNil(t, err) + }) + + t.Run("missing miniblock should error", func(t *testing.T) { + mb1 := &block.MiniBlock{ + SenderShardID: 1, + } + h1 := []byte("h1") + h2 := []byte("h2") + + executionResult := &block.ExecutionResult{ + MiniBlockHeaders: []block.MiniBlockHeader{ + { + Hash: h1, + }, + { + Hash: h2, + }, + }, + } + + marshaller := &marshallerMock.MarshalizerMock{} + mb1Bytes, _ := marshaller.Marshal(mb1) + + cacher := cache.NewCacherMock() + cacher.Put(h1, mb1Bytes, 0) + + _, err := getBody(cacher, marshaller, executionResult, 0) + require.Equal(t, process.ErrMissingMiniBlock, err) + }) + + t.Run("marshaller error", func(t *testing.T) { + h1 := []byte("h1") + h2 := []byte("h2") + + executionResult := &block.ExecutionResult{ + MiniBlockHeaders: []block.MiniBlockHeader{ + { + Hash: h1, + }, + { + Hash: h2, + }, + }, + } + + marshaller := &marshallerMock.MarshalizerMock{} + + cacher := cache.NewCacherMock() + cacher.Put(h1, []byte("wrong"), 0) + + _, err := getBody(cacher, marshaller, executionResult, 0) + require.NotNil(t, err) + }) + + t.Run("getBody should work", func(t *testing.T) { + mb1 := &block.MiniBlock{ + SenderShardID: 1, + } + mb2 := &block.MiniBlock{ + SenderShardID: 2, + } + h1 := []byte("h1") + h2 := []byte("h2") + + executionResult := &block.ExecutionResult{ + MiniBlockHeaders: []block.MiniBlockHeader{ + { + Hash: h1, + }, + { + Hash: h2, + }, + }, + } + + marshaller := &marshallerMock.MarshalizerMock{} + mb1Bytes, _ := marshaller.Marshal(mb1) + mb2Bytes, _ := marshaller.Marshal(mb2) + + cacher := cache.NewCacherMock() + cacher.Put(h1, mb1Bytes, 0) + cacher.Put(h2, mb2Bytes, 0) + + res, err := getBody(cacher, marshaller, executionResult, 0) + require.Nil(t, err) + require.Equal(t, &block.Body{MiniBlocks: []*block.MiniBlock{mb1, mb2}}, res) + }) +} diff --git a/dblookupext/factory/historyRepositoryFactory.go b/dblookupext/factory/historyRepositoryFactory.go index 5302d363c70..81a2021e26c 100644 --- a/dblookupext/factory/historyRepositoryFactory.go +++ b/dblookupext/factory/historyRepositoryFactory.go @@ -23,6 +23,7 @@ type ArgsHistoryRepositoryFactory struct { Marshalizer marshal.Marshalizer Hasher hashing.Hasher Uint64ByteSliceConverter typeConverters.Uint64ByteSliceConverter + DataPool dataRetriever.PoolsHolder } type historyRepositoryFactory struct { @@ -32,6 +33,7 @@ type historyRepositoryFactory struct { marshalizer marshal.Marshalizer hasher hashing.Hasher uInt64ByteSliceConverter typeConverters.Uint64ByteSliceConverter + dataPool dataRetriever.PoolsHolder } // NewHistoryRepositoryFactory creates an instance of historyRepositoryFactory @@ -48,6 +50,9 @@ func NewHistoryRepositoryFactory(args *ArgsHistoryRepositoryFactory) (dblookupex if check.IfNil(args.Uint64ByteSliceConverter) { return nil, process.ErrNilUint64Converter } + if check.IfNil(args.DataPool) { + return nil, process.ErrNilDataPoolHolder + } return &historyRepositoryFactory{ selfShardID: args.SelfShardID, @@ -56,6 +61,7 @@ func NewHistoryRepositoryFactory(args *ArgsHistoryRepositoryFactory) (dblookupex marshalizer: args.Marshalizer, hasher: args.Hasher, uInt64ByteSliceConverter: args.Uint64ByteSliceConverter, + dataPool: args.DataPool, }, nil } @@ -126,6 +132,7 @@ func (hpf *historyRepositoryFactory) Create() (dblookupext.HistoryRepository, er EventsHashesByTxHashStorer: resultsHashesByTxHashStorer, ESDTSuppliesHandler: esdtSuppliesHandler, ExecutionResultsStorer: executionResultsStorer, + DataPool: hpf.dataPool, } return dblookupext.NewHistoryRepository(historyRepArgs) } diff --git a/dblookupext/factory/historyRepositoryFactory_test.go b/dblookupext/factory/historyRepositoryFactory_test.go index 04419bbc093..1449142b6ad 100644 --- a/dblookupext/factory/historyRepositoryFactory_test.go +++ b/dblookupext/factory/historyRepositoryFactory_test.go @@ -14,6 +14,7 @@ import ( "github.com/multiversx/mx-chain-go/process" processMock "github.com/multiversx/mx-chain-go/process/mock" "github.com/multiversx/mx-chain-go/storage" + dataRetrieverMock "github.com/multiversx/mx-chain-go/testscommon/dataRetriever" "github.com/multiversx/mx-chain-go/testscommon/hashingMocks" storageStubs "github.com/multiversx/mx-chain-go/testscommon/storage" "github.com/stretchr/testify/require" @@ -120,5 +121,6 @@ func getArgs() *factory.ArgsHistoryRepositoryFactory { Marshalizer: &mock.MarshalizerMock{}, Hasher: &hashingMocks.HasherMock{}, Uint64ByteSliceConverter: &processMock.Uint64ByteSliceConverterMock{}, + DataPool: &dataRetrieverMock.PoolsHolderMock{}, } } diff --git a/dblookupext/historyRepository.go b/dblookupext/historyRepository.go index 75ded784b2f..f1dc7426dc2 100644 --- a/dblookupext/historyRepository.go +++ b/dblookupext/historyRepository.go @@ -15,6 +15,7 @@ import ( "github.com/multiversx/mx-chain-core-go/hashing" "github.com/multiversx/mx-chain-core-go/marshal" "github.com/multiversx/mx-chain-go/common/logging" + "github.com/multiversx/mx-chain-go/dataRetriever" "github.com/multiversx/mx-chain-go/dblookupext/esdtSupply" "github.com/multiversx/mx-chain-go/process" "github.com/multiversx/mx-chain-go/storage" @@ -39,6 +40,7 @@ type HistoryRepositoryArguments struct { Marshalizer marshal.Marshalizer Hasher hashing.Hasher ESDTSuppliesHandler SuppliesHandler + DataPool dataRetriever.PoolsHolder } type historyRepository struct { @@ -53,6 +55,7 @@ type historyRepository struct { marshalizer marshal.Marshalizer hasher hashing.Hasher esdtSuppliesHandler SuppliesHandler + dataPool dataRetriever.PoolsHolder // These maps temporarily hold notifications of "notarized at source or destination", to deal with unwanted concurrency effects // The unwanted concurrency effects could be accentuated by the fast db-replay-validate mechanism. @@ -102,6 +105,9 @@ func NewHistoryRepository(arguments HistoryRepositoryArguments) (*historyReposit if check.IfNil(arguments.ExecutionResultsStorer) { return nil, process.ErrNilStore } + if check.IfNil(arguments.DataPool) { + return nil, process.ErrNilDataPoolHolder + } hashToEpochIndex := newHashToEpochIndex(arguments.EpochByHashStorer, arguments.Marshalizer) deduplicationCacheForInsertMiniblockMetadata, _ := cache.NewLRUCache(sizeOfDeduplicationCache) @@ -125,6 +131,7 @@ func NewHistoryRepository(arguments HistoryRepositoryArguments) (*historyReposit esdtSuppliesHandler: arguments.ESDTSuppliesHandler, uint64ByteSliceConverter: arguments.Uint64ByteSliceConverter, executionResultsProcessor: executionResultsProc, + dataPool: arguments.DataPool, }, nil } @@ -136,63 +143,143 @@ func (hr *historyRepository) RecordBlock(blockHeaderHash []byte, scrResultsFromPool map[string]data.TransactionHandler, receiptsFromPool map[string]data.TransactionHandler, createdIntraShardMiniBlocks []*block.MiniBlock, - logs []*data.LogData) error { + logs []*data.LogData, +) error { hr.recordBlockMutex.Lock() defer hr.recordBlockMutex.Unlock() - log.Debug("RecordBlock()", "nonce", blockHeader.GetNonce(), "blockHeaderHash", blockHeaderHash, "header type", fmt.Sprintf("%T", blockHeader)) - body, ok := blockBody.(*block.Body) if !ok { return errCannotCastToBlockBody } - epoch := blockHeader.GetEpoch() + err := hr.recordBlock(blockHeaderHash, blockHeader.GetEpoch(), blockHeader.GetNonce(), blockHeader.GetRound(), body.MiniBlocks) + if err != nil { + return err + } - err := hr.epochByHashIndex.saveEpochByHash(blockHeaderHash, epoch) + err = hr.recordExtraData(blockHeaderHash, blockHeader, scrResultsFromPool, receiptsFromPool, createdIntraShardMiniBlocks, logs) if err != nil { - return newErrCannotSaveEpochByHash("block header", blockHeaderHash, err) + return err } - for _, miniblock := range body.MiniBlocks { - if miniblock.Type == block.PeerBlock { - continue + err = hr.putHashByRound(blockHeaderHash, blockHeader) + if err != nil { + return err + } + + err = hr.executionResultsProcessor.saveExecutionResultsFromHeader(blockHeader) + if err != nil { + return err + } + + return nil +} + +func (hr *historyRepository) recordExtraData( + blockHeaderHash []byte, + blockHeader data.HeaderHandler, + scrResultsFromPool map[string]data.TransactionHandler, + receiptsFromPool map[string]data.TransactionHandler, + createdIntraShardMiniBlocks []*block.MiniBlock, + logs []*data.LogData, +) error { + if blockHeader.IsHeaderV3() { + return hr.recordDataBasedOnExecutionResults(blockHeader) + } + + return hr.recordExecutionData(blockHeaderHash, blockHeader.GetEpoch(), blockHeader.GetNonce(), blockHeader.GetRound(), scrResultsFromPool, receiptsFromPool, createdIntraShardMiniBlocks, logs) +} + +func (hr *historyRepository) recordDataBasedOnExecutionResults(blockHeader data.HeaderHandler) error { + for _, executionResult := range blockHeader.GetExecutionResultsHandlers() { + headerHash := executionResult.GetHeaderHash() + + scrResultsFromPool, receiptsFromPool, err := getIntermediateTxs(hr.dataPool.PostProcessTransactions(), headerHash) + if err != nil { + return err + } + logs, err := getLogs(hr.dataPool.PostProcessTransactions(), headerHash) + if err != nil { + return err + } + body, err := getBody(hr.dataPool.ExecutedMiniBlocks(), hr.marshalizer, executionResult, hr.selfShardID) + if err != nil { + return err + } + intraMbs, err := getIntraMbs(hr.dataPool.ExecutedMiniBlocks(), hr.marshalizer, headerHash) + if err != nil { + return err } - err = hr.recordMiniblock(blockHeaderHash, blockHeader, miniblock, epoch) + headerNonce := executionResult.GetHeaderNonce() + headerEpoch := executionResult.GetHeaderEpoch() + headerRound := executionResult.GetHeaderRound() + err = hr.recordBlock(headerHash, headerEpoch, headerNonce, headerRound, body.MiniBlocks) if err != nil { - logging.LogErrAsErrorExceptAsDebugIfClosingError(log, err, "cannot record miniblock", - "type", miniblock.Type, "error", err) - continue + return err + } + + err = hr.recordExecutionData(headerHash, headerEpoch, headerNonce, headerRound, scrResultsFromPool, receiptsFromPool, intraMbs, logs) + if err != nil { + return err } } + return nil +} + +func (hr *historyRepository) recordExecutionData(blockHeaderHash []byte, + epoch uint32, + nonce uint64, + round uint64, + scrResultsFromPool map[string]data.TransactionHandler, + receiptsFromPool map[string]data.TransactionHandler, + createdIntraShardMiniBlocks []*block.MiniBlock, + logs []*data.LogData, +) error { + for _, miniBlock := range createdIntraShardMiniBlocks { - err = hr.recordMiniblock(blockHeaderHash, blockHeader, miniBlock, epoch) + err := hr.recordMiniblock(blockHeaderHash, nonce, round, miniBlock, epoch) if err != nil { logging.LogErrAsErrorExceptAsDebugIfClosingError(log, err, "cannot record in shard miniblock", "type", miniBlock.Type, "error", err) } } - err = hr.eventsHashesByTxHashIndex.saveResultsHashes(epoch, scrResultsFromPool, receiptsFromPool) + err := hr.eventsHashesByTxHashIndex.saveResultsHashes(epoch, scrResultsFromPool, receiptsFromPool) if err != nil { return err } - err = hr.esdtSuppliesHandler.ProcessLogs(blockHeader.GetNonce(), logs) - if err != nil { - return err - } + return hr.esdtSuppliesHandler.ProcessLogs(nonce, logs) +} - err = hr.putHashByRound(blockHeaderHash, blockHeader) +func (hr *historyRepository) recordBlock( + blockHeaderHash []byte, + epoch uint32, + nonce uint64, + round uint64, + miniBlocks []*block.MiniBlock, +) error { + log.Debug("RecordBlock()", "nonce", nonce, "blockHeaderHash", blockHeaderHash) + + err := hr.epochByHashIndex.saveEpochByHash(blockHeaderHash, epoch) if err != nil { - return err + return newErrCannotSaveEpochByHash("block header", blockHeaderHash, err) } - err = hr.executionResultsProcessor.saveExecutionResultsFromHeader(blockHeader) - if err != nil { - return err + for _, miniblock := range miniBlocks { + if miniblock.Type == block.PeerBlock { + continue + } + + err = hr.recordMiniblock(blockHeaderHash, nonce, round, miniblock, epoch) + if err != nil { + logging.LogErrAsErrorExceptAsDebugIfClosingError(log, err, "cannot record miniblock", + "type", miniblock.Type, "error", err) + continue + } } return nil @@ -203,7 +290,7 @@ func (hr *historyRepository) putHashByRound(blockHeaderHash []byte, header data. return hr.blockHashByRound.Put(roundToByteSlice, blockHeaderHash) } -func (hr *historyRepository) recordMiniblock(blockHeaderHash []byte, blockHeader data.HeaderHandler, miniblock *block.MiniBlock, epoch uint32) error { +func (hr *historyRepository) recordMiniblock(blockHeaderHash []byte, nonce uint64, round uint64, miniblock *block.MiniBlock, epoch uint32) error { miniblockHash, err := hr.computeMiniblockHash(miniblock) if err != nil { return err @@ -223,8 +310,8 @@ func (hr *historyRepository) recordMiniblock(blockHeaderHash []byte, blockHeader Epoch: epoch, HeaderHash: blockHeaderHash, MiniblockHash: miniblockHash, - Round: blockHeader.GetRound(), - HeaderNonce: blockHeader.GetNonce(), + Round: round, + HeaderNonce: nonce, SourceShardID: miniblock.GetSenderShardID(), DestinationShardID: miniblock.GetReceiverShardID(), } diff --git a/dblookupext/historyRepository_test.go b/dblookupext/historyRepository_test.go index 200a18decc0..7602e9d054b 100644 --- a/dblookupext/historyRepository_test.go +++ b/dblookupext/historyRepository_test.go @@ -13,6 +13,7 @@ import ( epochStartMocks "github.com/multiversx/mx-chain-go/epochStart/mock" "github.com/multiversx/mx-chain-go/process" "github.com/multiversx/mx-chain-go/storage" + dataRetrieverMock "github.com/multiversx/mx-chain-go/testscommon/dataRetriever" "github.com/multiversx/mx-chain-go/testscommon/genericMocks" "github.com/multiversx/mx-chain-go/testscommon/hashingMocks" storageStubs "github.com/multiversx/mx-chain-go/testscommon/storage" @@ -39,6 +40,7 @@ func createMockHistoryRepoArgs(epoch uint32) HistoryRepositoryArguments { Hasher: &hashingMocks.HasherMock{}, ESDTSuppliesHandler: sp, Uint64ByteSliceConverter: &epochStartMocks.Uint64ByteSliceConverterMock{}, + DataPool: &dataRetrieverMock.PoolsHolderMock{}, } return args diff --git a/process/block/baseProcess.go b/process/block/baseProcess.go index d19642ea7b2..6e77709a216 100644 --- a/process/block/baseProcess.go +++ b/process/block/baseProcess.go @@ -1295,7 +1295,7 @@ func (bp *baseProcessor) getFinalMiniBlocksFromExecutionResults( cachedMiniBlockBytes := cachedMiniBlock.([]byte) var miniBlock *block.MiniBlock - err := bp.marshalizer.Unmarshal(&miniBlock, cachedMiniBlockBytes) + err = bp.marshalizer.Unmarshal(&miniBlock, cachedMiniBlockBytes) if err != nil { return nil, err } @@ -2464,6 +2464,9 @@ func (bp *baseProcessor) saveExecutedData(header data.HeaderHandler, headerHash return err } + // cleanup intra shard mini-blocks + bp.dataPool.MiniBlocks().Remove(headerHash) + return bp.saveIntermediateTxs(headerHash) } @@ -2554,6 +2557,29 @@ func (bp *baseProcessor) putMiniBlocksIntoStorage(miniBlockHeaderHandlers []data return nil } +func (bp *baseProcessor) cacheIntraShardMiniBlocks(headerHash []byte, mbs []*block.MiniBlock) error { + marshalledMbs, err := bp.marshalizer.Marshal(mbs) + if err != nil { + return err + } + + bp.dataPool.ExecutedMiniBlocks().Put(headerHash, marshalledMbs, len(marshalledMbs)) + + return nil +} + +func (bp *baseProcessor) cacheLogEvents(headerHash []byte, logs []*data.LogData) error { + logsMarshalled, err := bp.marshalizer.Marshal(logs) + if err != nil { + return err + } + + key := common.PrepareLogEventsKey(headerHash) + bp.dataPool.PostProcessTransactions().Put(key, logs, len(logsMarshalled)) + + return nil +} + func (bp *baseProcessor) cacheExecutedMiniBlocks(body *block.Body, miniBlockHeaders []data.MiniBlockHeaderHandler) error { for i, mbHeader := range miniBlockHeaders { miniBlockHash := mbHeader.GetHash() @@ -2575,7 +2601,7 @@ func (bp *baseProcessor) cacheIntermediateTxsForHeader(headerHash []byte) error return err } - bp.dataPool.PostProcessTransactions().Put(headerHash, buff, len(buff)) + bp.dataPool.PostProcessTransactions().Put(headerHash, intermediateTxs, len(buff)) return nil } @@ -2588,11 +2614,9 @@ func (bp *baseProcessor) saveIntermediateTxs(headerHash []byte) error { return fmt.Errorf("%w for header %s", process.ErrMissingHeader, hex.EncodeToString(headerHash)) } - cachedIntermediateTxsBuff := cachedIntermediateTxs.([]byte) - cachedIntermediateTxsMap := map[block.Type]map[string]data.TransactionHandler{} - errUnmarshal := bp.marshalizer.Unmarshal(cachedIntermediateTxsMap, cachedIntermediateTxsBuff) - if errUnmarshal != nil { - return errUnmarshal + cachedIntermediateTxsMap, ok := cachedIntermediateTxs.(map[block.Type]map[string]data.TransactionHandler) + if !ok { + log.Warn("saveIntermediateTxs: intermediateTxs cannot cast to concrete type", "hash", headerHash) } for blockType, cachedTransactionsMap := range cachedIntermediateTxsMap { @@ -2604,6 +2628,8 @@ func (bp *baseProcessor) saveIntermediateTxs(headerHash []byte) error { // all transactions moved, cleaning the cache postProcessTxsCache.Remove(headerHash) + // cleanup all log events + postProcessTxsCache.Remove(common.PrepareLogEventsKey(headerHash)) return nil } diff --git a/process/block/shardblockProposal.go b/process/block/shardblockProposal.go index 2cd0005525d..0fbd5d7a854 100644 --- a/process/block/shardblockProposal.go +++ b/process/block/shardblockProposal.go @@ -837,6 +837,12 @@ func (sp *shardProcessor) collectExecutionResults(headerHash []byte, header data return nil, err } + intraMiniBlocks := sp.txCoordinator.GetCreatedInShardMiniBlocks() + err = sp.cacheIntraShardMiniBlocks(headerHash, intraMiniBlocks) + if err != nil { + return nil, err + } + err = sp.cacheExecutedMiniBlocks(sanitizedBodyAfterExecution, miniBlockHeaderHandlers) if err != nil { return nil, err @@ -862,6 +868,12 @@ func (sp *shardProcessor) collectExecutionResults(headerHash []byte, header data return nil, err } + logs := sp.txCoordinator.GetAllCurrentLogs() + err = sp.cacheLogEvents(headerHash, logs) + if err != nil { + return nil, err + } + err = sp.cacheIntermediateTxsForHeader(headerHash) if err != nil { return nil, err From 9dada4acfd0de90f348da98e201b4b399369aa9b Mon Sep 17 00:00:00 2001 From: miiu Date: Thu, 23 Oct 2025 14:28:16 +0300 Subject: [PATCH 2/6] happy path --- dblookupext/historyRepository_test.go | 95 +++++++++++++++++++++++++++ 1 file changed, 95 insertions(+) diff --git a/dblookupext/historyRepository_test.go b/dblookupext/historyRepository_test.go index 7602e9d054b..29e25218b8d 100644 --- a/dblookupext/historyRepository_test.go +++ b/dblookupext/historyRepository_test.go @@ -8,6 +8,10 @@ import ( "github.com/multiversx/mx-chain-core-go/core" "github.com/multiversx/mx-chain-core-go/data" "github.com/multiversx/mx-chain-core-go/data/block" + "github.com/multiversx/mx-chain-core-go/data/receipt" + "github.com/multiversx/mx-chain-core-go/data/smartContractResult" + "github.com/multiversx/mx-chain-core-go/data/transaction" + "github.com/multiversx/mx-chain-go/common" "github.com/multiversx/mx-chain-go/common/mock" "github.com/multiversx/mx-chain-go/dblookupext/esdtSupply" epochStartMocks "github.com/multiversx/mx-chain-go/epochStart/mock" @@ -744,3 +748,94 @@ func TestHistoryRepository_ConcurrentlyRecordAndNotarizeSameBlockMultipleTimes(t require.Equal(t, 4001, int(metadata.NotarizedAtDestinationInMetaNonce)) require.Equal(t, []byte("metablockFoo"), metadata.NotarizedAtDestinationInMetaHash) } + +func TestRecordHeaderV3(t *testing.T) { + t.Parallel() + + t.Run("record block v3 should work no execution results", func(t *testing.T) { + args := createMockHistoryRepoArgs(42) + repo, err := NewHistoryRepository(args) + require.Nil(t, err) + + header := &block.HeaderV3{} + body := &block.Body{} + + headerHash := []byte("headerHash") + err = repo.RecordBlock(headerHash, header, body, nil, nil, nil, nil) + require.Nil(t, err) + }) + + t.Run("record block v3 should work", func(t *testing.T) { + args := createMockHistoryRepoArgs(42) + args.DataPool = dataRetrieverMock.NewPoolsHolderMock() + repo, err := NewHistoryRepository(args) + require.Nil(t, err) + + executionResultHeaderHash := []byte("executionResultHeaderHash") + mb := &block.MiniBlock{SenderShardID: 0} + mbHash1, _ := repo.computeMiniblockHash(mb) + header := &block.HeaderV3{ + Nonce: 100, + Round: 101, + Epoch: 42, + ExecutionResults: []*block.ExecutionResult{ + { + + BaseExecutionResult: &block.BaseExecutionResult{ + HeaderHash: executionResultHeaderHash, + HeaderNonce: 99, + HeaderRound: 100, + HeaderEpoch: 42, + }, + MiniBlockHeaders: []block.MiniBlockHeader{ + { + Hash: mbHash1, + }, + }, + }, + }, + } + + mbBytes, _ := repo.marshalizer.Marshal(mb) + args.DataPool.ExecutedMiniBlocks().Put(mbHash1, mbBytes, 1) + + cachedIntermediateTxsMap := map[block.Type]map[string]data.TransactionHandler{} + cachedIntermediateTxsMap[block.SmartContractResultBlock] = map[string]data.TransactionHandler{ + "h1": &smartContractResult.SmartContractResult{}, + } + cachedIntermediateTxsMap[block.ReceiptBlock] = map[string]data.TransactionHandler{ + "r1": &receipt.Receipt{}, + } + args.DataPool.PostProcessTransactions().Put(executionResultHeaderHash, cachedIntermediateTxsMap, 1) + + expectedLogs := []*data.LogData{ + { + LogHandler: &transaction.Log{}, + TxHash: "t1", + }, + } + logsKey := common.PrepareLogEventsKey(executionResultHeaderHash) + args.DataPool.PostProcessTransactions().Put(logsKey, expectedLogs, 1) + + expectedMbs := []*block.MiniBlock{ + {SenderShardID: 0}, + } + intraMbsBytes, _ := repo.marshalizer.Marshal(expectedMbs) + + args.DataPool.ExecutedMiniBlocks().Put(executionResultHeaderHash, intraMbsBytes, 0) + + body := &block.Body{} + + headerHash := []byte("headerHash") + err = repo.RecordBlock(headerHash, header, body, nil, nil, nil, nil) + require.Nil(t, err) + + epoch, err := repo.GetEpochByHash(executionResultHeaderHash) + require.Nil(t, err) + require.Equal(t, 42, int(epoch)) + + epoch, err = repo.GetEpochByHash(mbHash1) + require.Nil(t, err) + require.Equal(t, 42, int(epoch)) + }) +} From dffeac2f0bcb8c9d71cbcd26d19e2930e2b66f27 Mon Sep 17 00:00:00 2001 From: miiu Date: Thu, 23 Oct 2025 14:41:09 +0300 Subject: [PATCH 3/6] fix unit tests --- process/block/baseProcessHeaderV3_test.go | 123 ++++++---------------- 1 file changed, 31 insertions(+), 92 deletions(-) diff --git a/process/block/baseProcessHeaderV3_test.go b/process/block/baseProcessHeaderV3_test.go index df30929dbb3..f99c9f1c24f 100644 --- a/process/block/baseProcessHeaderV3_test.go +++ b/process/block/baseProcessHeaderV3_test.go @@ -452,48 +452,6 @@ func TestBaseProcessor_saveExecutedData(t *testing.T) { err := bp.saveExecutedData(header, headerHash) require.True(t, errors.Is(err, process.ErrMissingHeader)) }) - t.Run("unmarshall error", func(t *testing.T) { - t.Parallel() - - bp := &baseProcessor{ - store: &commonStorage.ChainStorerStub{ - GetStorerCalled: func(unitType dataRetriever.UnitType) (storage.Storer, error) { - require.Fail(t, "should not be called") - return nil, nil - }, - }, - dataPool: &dataRetrieverMock.PoolsHolderStub{ - PostProcessTransactionsCalled: func() storage.Cacher { - return &cache.CacherStub{ - GetCalled: func(key []byte) (value interface{}, ok bool) { - return []byte("marshalled map"), true - }, - } - }, - }, - marshalizer: &marshallerMock.MarshalizerStub{ - UnmarshalCalled: func(obj interface{}, buff []byte) error { - return errExpected - }, - }, - shardCoordinator: &testscommon.ShardsCoordinatorMock{}, - } - header := &testscommon.HeaderHandlerStub{ - IsHeaderV3Called: func() bool { - return true - }, - GetExecutionResultsHandlersCalled: func() []data.BaseExecutionResultHandler { - return []data.BaseExecutionResultHandler{ - &block.ExecutionResult{ - MiniBlockHeaders: []block.MiniBlockHeader{}, - }, - } - }, - } - - err := bp.saveExecutedData(header, headerHash) - require.Equal(t, errExpected, err) - }) t.Run("putTransactionsIntoStorage fails due to invalid block type", func(t *testing.T) { t.Parallel() @@ -508,18 +466,14 @@ func TestBaseProcessor_saveExecutedData(t *testing.T) { PostProcessTransactionsCalled: func() storage.Cacher { return &cache.CacherStub{ GetCalled: func(key []byte) (value interface{}, ok bool) { - return []byte("marshalled map"), true + txsMap := make(map[block.Type]map[string]data.TransactionHandler) + txsMap[block.PeerBlock] = map[string]data.TransactionHandler{} // should never have PeerBlock + return txsMap, true }, } }, }, - marshalizer: &marshallerMock.MarshalizerStub{ - UnmarshalCalled: func(obj interface{}, buff []byte) error { - txsMap := obj.(map[block.Type]map[string]data.TransactionHandler) - txsMap[block.PeerBlock] = map[string]data.TransactionHandler{} // should never have PeerBlock - return nil - }, - }, + marshalizer: &marshallerMock.MarshalizerMock{}, shardCoordinator: &testscommon.ShardsCoordinatorMock{}, } header := &testscommon.HeaderHandlerStub{ @@ -555,18 +509,14 @@ func TestBaseProcessor_saveExecutedData(t *testing.T) { PostProcessTransactionsCalled: func() storage.Cacher { return &cache.CacherStub{ GetCalled: func(key []byte) (value interface{}, ok bool) { - return []byte("marshalled map"), true + txsMap := make(map[block.Type]map[string]data.TransactionHandler) + txsMap[block.TxBlock] = map[string]data.TransactionHandler{} // force TransactionUnit + return txsMap, true }, } }, }, - marshalizer: &marshallerMock.MarshalizerStub{ - UnmarshalCalled: func(obj interface{}, buff []byte) error { - txsMap := obj.(map[block.Type]map[string]data.TransactionHandler) - txsMap[block.TxBlock] = map[string]data.TransactionHandler{} // force TransactionUnit - return nil - }, - }, + marshalizer: &marshallerMock.MarshalizerMock{}, shardCoordinator: &testscommon.ShardsCoordinatorMock{}, } header := &testscommon.HeaderHandlerStub{ @@ -598,20 +548,16 @@ func TestBaseProcessor_saveExecutedData(t *testing.T) { PostProcessTransactionsCalled: func() storage.Cacher { return &cache.CacherStub{ GetCalled: func(key []byte) (value interface{}, ok bool) { - return []byte("marshalled map"), true + txsMap := make(map[block.Type]map[string]data.TransactionHandler) + txsMap[block.TxBlock] = map[string]data.TransactionHandler{ + "hash": nil, + } + return txsMap, true }, } }, }, - marshalizer: &marshallerMock.MarshalizerStub{ - UnmarshalCalled: func(obj interface{}, buff []byte) error { - txsMap := obj.(map[block.Type]map[string]data.TransactionHandler) - txsMap[block.TxBlock] = map[string]data.TransactionHandler{ - "hash": nil, - } - return nil - }, - }, + marshalizer: &marshallerMock.MarshalizerMock{}, shardCoordinator: &testscommon.ShardsCoordinatorMock{}, } header := &testscommon.HeaderHandlerStub{ @@ -643,19 +589,16 @@ func TestBaseProcessor_saveExecutedData(t *testing.T) { PostProcessTransactionsCalled: func() storage.Cacher { return &cache.CacherStub{ GetCalled: func(key []byte) (value interface{}, ok bool) { - return []byte("marshalled map"), true + txsMap := make(map[block.Type]map[string]data.TransactionHandler) + txsMap[block.TxBlock] = map[string]data.TransactionHandler{ + "hash": &transaction.Transaction{}, + } + return txsMap, true }, } }, }, marshalizer: &marshallerMock.MarshalizerStub{ - UnmarshalCalled: func(obj interface{}, buff []byte) error { - txsMap := obj.(map[block.Type]map[string]data.TransactionHandler) - txsMap[block.TxBlock] = map[string]data.TransactionHandler{ - "hash": &transaction.Transaction{}, - } - return nil - }, MarshalCalled: func(obj interface{}) ([]byte, error) { return nil, errExpected }, @@ -700,7 +643,17 @@ func TestBaseProcessor_saveExecutedData(t *testing.T) { PostProcessTransactionsCalled: func() storage.Cacher { return &cache.CacherStub{ GetCalled: func(key []byte) (value interface{}, ok bool) { - return []byte("marshalled map"), true + txsMap := make(map[block.Type]map[string]data.TransactionHandler) + txsMap[block.SmartContractResultBlock] = map[string]data.TransactionHandler{ + "hashSCR": &transaction.Transaction{}, + } + txsMap[block.RewardsBlock] = map[string]data.TransactionHandler{ + "hashReward": &transaction.Transaction{}, // for coverage + } + txsMap[block.ReceiptBlock] = map[string]data.TransactionHandler{ + "hashReward": &transaction.Transaction{}, // for coverage + } + return txsMap, true }, RemoveCalled: func(key []byte) { wasRemoveCalledForTxs = true @@ -718,21 +671,7 @@ func TestBaseProcessor_saveExecutedData(t *testing.T) { } }, }, - marshalizer: &marshallerMock.MarshalizerStub{ - UnmarshalCalled: func(obj interface{}, buff []byte) error { - txsMap := obj.(map[block.Type]map[string]data.TransactionHandler) - txsMap[block.SmartContractResultBlock] = map[string]data.TransactionHandler{ - "hashSCR": &transaction.Transaction{}, - } - txsMap[block.RewardsBlock] = map[string]data.TransactionHandler{ - "hashReward": &transaction.Transaction{}, // for coverage - } - txsMap[block.ReceiptBlock] = map[string]data.TransactionHandler{ - "hashReward": &transaction.Transaction{}, // for coverage - } - return nil - }, - }, + marshalizer: &marshallerMock.MarshalizerMock{}, shardCoordinator: &testscommon.ShardsCoordinatorMock{}, } header := &testscommon.HeaderHandlerStub{ From 2f271f699e1b66bbdab0f593419e471f2f7c75e1 Mon Sep 17 00:00:00 2001 From: miiu Date: Thu, 23 Oct 2025 16:34:56 +0300 Subject: [PATCH 4/6] fixes --- integrationTests/realcomponents/processorRunner.go | 1 + node/chainSimulator/components/processComponents.go | 1 + node/nodeRunner.go | 1 + 3 files changed, 3 insertions(+) diff --git a/integrationTests/realcomponents/processorRunner.go b/integrationTests/realcomponents/processorRunner.go index 8bdb94cf2e3..8530566d85d 100644 --- a/integrationTests/realcomponents/processorRunner.go +++ b/integrationTests/realcomponents/processorRunner.go @@ -394,6 +394,7 @@ func (pr *ProcessorRunner) createProcessComponents(tb testing.TB) { Marshalizer: pr.CoreComponents.InternalMarshalizer(), Store: pr.DataComponents.StorageService(), Uint64ByteSliceConverter: pr.CoreComponents.Uint64ByteSliceConverter(), + DataPool: pr.DataComponents.Datapool(), } historyRepositoryFactory, err := dbLookupFactory.NewHistoryRepositoryFactory(historyRepoFactoryArgs) require.Nil(tb, err) diff --git a/node/chainSimulator/components/processComponents.go b/node/chainSimulator/components/processComponents.go index c7cce6e2beb..85feb496238 100644 --- a/node/chainSimulator/components/processComponents.go +++ b/node/chainSimulator/components/processComponents.go @@ -150,6 +150,7 @@ func CreateProcessComponents(args ArgsProcessComponentsHolder) (*processComponen Marshalizer: args.CoreComponents.InternalMarshalizer(), Store: args.DataComponents.StorageService(), Uint64ByteSliceConverter: args.CoreComponents.Uint64ByteSliceConverter(), + DataPool: args.DataComponents.Datapool(), } historyRepositoryFactory, err := dbLookupFactory.NewHistoryRepositoryFactory(historyRepoFactoryArgs) if err != nil { diff --git a/node/nodeRunner.go b/node/nodeRunner.go index e9b3d622706..2637d8d113b 100644 --- a/node/nodeRunner.go +++ b/node/nodeRunner.go @@ -1225,6 +1225,7 @@ func (nr *nodeRunner) CreateManagedProcessComponents( Marshalizer: coreComponents.InternalMarshalizer(), Store: dataComponents.StorageService(), Uint64ByteSliceConverter: coreComponents.Uint64ByteSliceConverter(), + DataPool: dataComponents.Datapool(), } historyRepositoryFactory, err := dbLookupFactory.NewHistoryRepositoryFactory(historyRepoFactoryArgs) if err != nil { From 2d3db6130c18cbd39623798de7efa47e9529bdaf Mon Sep 17 00:00:00 2001 From: miiu Date: Fri, 24 Oct 2025 15:15:45 +0300 Subject: [PATCH 5/6] fixes after review --- dblookupext/common.go | 5 +++-- dblookupext/common_test.go | 43 ++++++++++++++++++++++++++++++------ process/block/baseProcess.go | 11 ++++----- 3 files changed, 45 insertions(+), 14 deletions(-) diff --git a/dblookupext/common.go b/dblookupext/common.go index 85a595b611c..79be6bc0066 100644 --- a/dblookupext/common.go +++ b/dblookupext/common.go @@ -21,7 +21,7 @@ func getIntermediateTxs(cache storage.Cacher, headerHash []byte) (map[string]dat cachedIntermediateTxsMap, ok := cachedIntermediateTxs.(map[block.Type]map[string]data.TransactionHandler) if !ok { - return make(map[string]data.TransactionHandler), make(map[string]data.TransactionHandler), nil + return nil, nil, fmt.Errorf("%w for cached intermediate transaction %s", process.ErrWrongTypeAssertion, hex.EncodeToString(headerHash)) } scrs := cachedIntermediateTxsMap[block.SmartContractResultBlock] @@ -39,7 +39,7 @@ func getLogs(cache storage.Cacher, headerHash []byte) ([]*data.LogData, error) { } cachedLogsSlice, ok := cachedLogs.([]*data.LogData) if !ok { - return []*data.LogData{}, nil + return nil, fmt.Errorf("%w for cached logs %s", process.ErrWrongTypeAssertion, hex.EncodeToString(headerHash)) } return cachedLogsSlice, nil } @@ -90,6 +90,7 @@ func getBody(cache storage.Cacher, marshaller marshal.Marshalizer, baseExecResul return &block.Body{MiniBlocks: miniBlocks}, nil } +// TODO reuse the method that was moved into common after PR #7337 is merged func extractMiniBlocksHeaderHandlersFromExecResult( baseExecResult data.BaseExecutionResultHandler, headerShard uint32, diff --git a/dblookupext/common_test.go b/dblookupext/common_test.go index 3f56c4f064a..b6cee3c493c 100644 --- a/dblookupext/common_test.go +++ b/dblookupext/common_test.go @@ -22,6 +22,8 @@ func TestGetIntermediateTxs(t *testing.T) { t.Parallel() t.Run("getIntermediateTxs cannot find in cache", func(t *testing.T) { + t.Parallel() + cacher := cache.NewCacherMock() headerHash := []byte("h") @@ -31,18 +33,20 @@ func TestGetIntermediateTxs(t *testing.T) { }) t.Run("getIntermediateTxs wrong type in cache should return empty maps", func(t *testing.T) { + t.Parallel() + cacher := cache.NewCacherMock() headerHash := []byte("h") cacher.Put(headerHash, []byte("wrong"), 0) - scrs, receipts, err := getIntermediateTxs(cacher, headerHash) - require.NoError(t, err) - require.Len(t, scrs, 0) - require.Len(t, receipts, 0) + _, _, err := getIntermediateTxs(cacher, headerHash) + require.True(t, errors.Is(err, process.ErrWrongTypeAssertion)) }) t.Run("getIntermediateTxs should work", func(t *testing.T) { + t.Parallel() + cachedIntermediateTxsMap := map[block.Type]map[string]data.TransactionHandler{} cachedIntermediateTxsMap[block.SmartContractResultBlock] = map[string]data.TransactionHandler{ "h1": &smartContractResult.SmartContractResult{}, @@ -70,6 +74,8 @@ func TestGetLogs(t *testing.T) { t.Parallel() t.Run("getLogs cannot find in cache", func(t *testing.T) { + t.Parallel() + cacher := cache.NewCacherMock() headerHash := []byte("h") @@ -79,18 +85,21 @@ func TestGetLogs(t *testing.T) { }) t.Run("getLogs wrong type in cache should return empty slice", func(t *testing.T) { + t.Parallel() + cacher := cache.NewCacherMock() headerHash := []byte("h") logsKey := common.PrepareLogEventsKey(headerHash) cacher.Put(logsKey, "wrong type", 0) - logs, err := getLogs(cacher, headerHash) - require.Nil(t, err) - require.Len(t, logs, 0) + _, err := getLogs(cacher, headerHash) + require.True(t, errors.Is(err, process.ErrWrongTypeAssertion)) }) t.Run("getLogs should work", func(t *testing.T) { + t.Parallel() + cacher := cache.NewCacherMock() headerHash := []byte("h") @@ -120,6 +129,8 @@ func TestGetIntraMbs(t *testing.T) { t.Parallel() t.Run("getIntraMbs cannot find in cache", func(t *testing.T) { + t.Parallel() + cacher := cache.NewCacherMock() marshaller := &marshallerMock.MarshalizerMock{} @@ -130,6 +141,8 @@ func TestGetIntraMbs(t *testing.T) { }) t.Run("getIntraMbs wrong type should error", func(t *testing.T) { + t.Parallel() + cacher := cache.NewCacherMock() marshaller := &marshallerMock.MarshalizerMock{} @@ -143,6 +156,8 @@ func TestGetIntraMbs(t *testing.T) { }) t.Run("getIntraMbs should work", func(t *testing.T) { + t.Parallel() + cacher := cache.NewCacherMock() marshaller := &marshallerMock.MarshalizerMock{} @@ -164,6 +179,8 @@ func TestGetIntraMbs(t *testing.T) { func TestExtractMiniBlocksHeaderHandlersFromExecResult(t *testing.T) { t.Run("wrong type shard", func(t *testing.T) { + t.Parallel() + executionResult := &block.BaseExecutionResult{} _, err := extractMiniBlocksHeaderHandlersFromExecResult(executionResult, 0) @@ -182,11 +199,15 @@ func TestExtractMiniBlocksHeaderHandlersFromExecResult(t *testing.T) { require.Len(t, res, 2) }) t.Run("wrong type meta", func(t *testing.T) { + t.Parallel() + executionResult := &block.BaseExecutionResult{} _, err := extractMiniBlocksHeaderHandlersFromExecResult(executionResult, core.MetachainShardId) require.Equal(t, process.ErrWrongTypeAssertion, err) }) t.Run("should work meta", func(t *testing.T) { + t.Parallel() + executionResult := &block.MetaExecutionResult{ MiniBlockHeaders: []block.MiniBlockHeader{ {SenderShardID: 0}, @@ -204,6 +225,8 @@ func TestGetBody(t *testing.T) { t.Parallel() t.Run("cannot get mb headers should error", func(t *testing.T) { + t.Parallel() + executionResult := &block.BaseExecutionResult{} marshaller := &marshallerMock.MarshalizerMock{} @@ -214,6 +237,8 @@ func TestGetBody(t *testing.T) { }) t.Run("missing miniblock should error", func(t *testing.T) { + t.Parallel() + mb1 := &block.MiniBlock{ SenderShardID: 1, } @@ -242,6 +267,8 @@ func TestGetBody(t *testing.T) { }) t.Run("marshaller error", func(t *testing.T) { + t.Parallel() + h1 := []byte("h1") h2 := []byte("h2") @@ -266,6 +293,8 @@ func TestGetBody(t *testing.T) { }) t.Run("getBody should work", func(t *testing.T) { + t.Parallel() + mb1 := &block.MiniBlock{ SenderShardID: 1, } diff --git a/process/block/baseProcess.go b/process/block/baseProcess.go index 30d1e9f535d..6f6edae1e4d 100644 --- a/process/block/baseProcess.go +++ b/process/block/baseProcess.go @@ -2468,9 +2468,6 @@ func (bp *baseProcessor) saveExecutedData(header data.HeaderHandler, headerHash return err } - // cleanup intra shard mini-blocks - bp.dataPool.MiniBlocks().Remove(headerHash) - return bp.saveIntermediateTxs(headerHash) } @@ -2490,6 +2487,12 @@ func (bp *baseProcessor) saveMiniBlocksFromExecutionResults(header data.HeaderHa if err != nil { return err } + + executionResultHeaderHash := baseExecutionResult.GetHeaderHash() + // cleanup all intra shard miniblocks + bp.dataPool.MiniBlocks().Remove(executionResultHeaderHash) + // cleanup all log events + bp.dataPool.PostProcessTransactions().Remove(common.PrepareLogEventsKey(executionResultHeaderHash)) } return nil @@ -2632,8 +2635,6 @@ func (bp *baseProcessor) saveIntermediateTxs(headerHash []byte) error { // all transactions moved, cleaning the cache postProcessTxsCache.Remove(headerHash) - // cleanup all log events - postProcessTxsCache.Remove(common.PrepareLogEventsKey(headerHash)) return nil } From e7a51f20b05e7a07fb54066999e3539778600f89 Mon Sep 17 00:00:00 2001 From: miiu Date: Wed, 29 Oct 2025 13:10:11 +0200 Subject: [PATCH 6/6] fixes after second review --- dblookupext/historyRepository_test.go | 4 ++++ process/block/baseProcess.go | 14 ++++++++------ 2 files changed, 12 insertions(+), 6 deletions(-) diff --git a/dblookupext/historyRepository_test.go b/dblookupext/historyRepository_test.go index 29e25218b8d..69913abc9c7 100644 --- a/dblookupext/historyRepository_test.go +++ b/dblookupext/historyRepository_test.go @@ -753,6 +753,8 @@ func TestRecordHeaderV3(t *testing.T) { t.Parallel() t.Run("record block v3 should work no execution results", func(t *testing.T) { + t.Parallel() + args := createMockHistoryRepoArgs(42) repo, err := NewHistoryRepository(args) require.Nil(t, err) @@ -766,6 +768,8 @@ func TestRecordHeaderV3(t *testing.T) { }) t.Run("record block v3 should work", func(t *testing.T) { + t.Parallel() + args := createMockHistoryRepoArgs(42) args.DataPool = dataRetrieverMock.NewPoolsHolderMock() repo, err := NewHistoryRepository(args) diff --git a/process/block/baseProcess.go b/process/block/baseProcess.go index 29e9e3e454f..365b401c087 100644 --- a/process/block/baseProcess.go +++ b/process/block/baseProcess.go @@ -1139,6 +1139,14 @@ func (bp *baseProcessor) cleanupPools(headerHandler data.HeaderHandler) { bp.cleanupPoolsForCrossShard(core.MetachainShardId, noncesToPrevFinal) } + for _, executionResult := range headerHandler.GetExecutionResultsHandlers() { + executionResultHeaderHash := executionResult.GetHeaderHash() + // cleanup all intra shard miniblocks + bp.dataPool.MiniBlocks().Remove(executionResultHeaderHash) + // cleanup all log events + bp.dataPool.PostProcessTransactions().Remove(common.PrepareLogEventsKey(executionResultHeaderHash)) + } + } func (bp *baseProcessor) cleanupPoolsForCrossShard( @@ -2513,12 +2521,6 @@ func (bp *baseProcessor) saveMiniBlocksFromExecutionResults(header data.HeaderHa if err != nil { return err } - - executionResultHeaderHash := baseExecutionResult.GetHeaderHash() - // cleanup all intra shard miniblocks - bp.dataPool.MiniBlocks().Remove(executionResultHeaderHash) - // cleanup all log events - bp.dataPool.PostProcessTransactions().Remove(common.PrepareLogEventsKey(executionResultHeaderHash)) } return nil