Merge pull request #2598 from Juliusan/prune-state-manager

State pruning in store

components/chains/component.go

@@ -102,6 +102,14 @@ func provide(c *dig.Container) error {
 				deps.ChainStateDatabaseManager.ChainStateKVStore,
 				ParamsWAL.Enabled,
 				ParamsWAL.Path,
+				ParamsStateManager.BlockCacheMaxSize,
+				ParamsStateManager.BlockCacheBlocksInCacheDuration,
+				ParamsStateManager.BlockCacheBlockCleaningPeriod,
+				ParamsStateManager.StateManagerGetBlockRetry,
+				ParamsStateManager.StateManagerRequestCleaningPeriod,
+				ParamsStateManager.StateManagerTimerTickPeriod,
+				ParamsStateManager.PruningMinStatesToKeep,
+				ParamsStateManager.PruningMaxStatesToDelete,
 				deps.ChainRecordRegistryProvider,
 				deps.DKShareRegistryProvider,
 				deps.NodeIdentityProvider,

components/chains/params.go

@@ -21,15 +21,28 @@ type ParametersWAL struct {
 	Path    string `default:"waspdb/wal" usage:"the path to the \"write-ahead logging\" folder"`
 }
 
+type ParametersStateManager struct {
+	BlockCacheMaxSize                 int           `default:"1000" usage:"how many blocks may be stored in cache before old ones start being deleted"`
+	BlockCacheBlocksInCacheDuration   time.Duration `default:"1h" usage:"how long should the block stay in block cache before being deleted"`
+	BlockCacheBlockCleaningPeriod     time.Duration `default:"1m" usage:"how often should the block cache be cleaned"`
+	StateManagerGetBlockRetry         time.Duration `default:"3s" usage:"how often get block requests should be repeated"`
+	StateManagerRequestCleaningPeriod time.Duration `default:"1s" usage:"how often requests waiting for response should be checked for expired context"`
+	StateManagerTimerTickPeriod       time.Duration `default:"1s" usage:"how often timer tick fires in state manager"`
+	PruningMinStatesToKeep            int           `default:"10000" usage:"this number of states will always be available in the store; if 0 - store pruning is disabled"`
+	PruningMaxStatesToDelete          int           `default:"1000" usage:"on single store pruning attempt at most this number of states will be deleted"`
+}
+
 var (
-	ParamsChains = &ParametersChains{}
-	ParamsWAL    = &ParametersWAL{}
+	ParamsChains       = &ParametersChains{}
+	ParamsWAL          = &ParametersWAL{}
+	ParamsStateManager = &ParametersStateManager{}
 )
 
 var params = &app.ComponentParams{
 	Params: map[string]any{
-		"chains": ParamsChains,
-		"wal":    ParamsWAL,
+		"chains":       ParamsChains,
+		"wal":          ParamsWAL,
+		"stateManager": ParamsStateManager,
 	},
 	Masked: nil,
 }

packages/chain/node.go

@@ -33,6 +33,7 @@ import (
 	consGR "github.com/iotaledger/wasp/packages/chain/cons/cons_gr"
 	"github.com/iotaledger/wasp/packages/chain/mempool"
 	"github.com/iotaledger/wasp/packages/chain/statemanager"
+	"github.com/iotaledger/wasp/packages/chain/statemanager/sm_gpa"
 	"github.com/iotaledger/wasp/packages/chain/statemanager/sm_gpa/sm_gpa_utils"
 	"github.com/iotaledger/wasp/packages/cryptolib"
 	"github.com/iotaledger/wasp/packages/gpa"
@@ -271,6 +272,7 @@ func New(
 	deriveAliasOutputByQuorum bool,
 	pipeliningLimit int,
 	consensusDelay time.Duration,
+	smParameters sm_gpa.StateManagerParameters,
 ) (Chain, error) {
 	log.Debugf("Starting the chain, chainID=%v", chainID)
 	if listener == nil {
@@ -398,6 +400,7 @@ func New(
 		chainMetrics,
 		chainMetrics,
 		cni.log.Named("SM"),
+		smParameters,
 	)
 	if err != nil {
 		return nil, fmt.Errorf("cannot create stateMgr: %w", err)

packages/chain/node_test.go

@@ -19,6 +19,7 @@ import (
 	iotago "github.com/iotaledger/iota.go/v3"
 	"github.com/iotaledger/wasp/contracts/native/inccounter"
 	"github.com/iotaledger/wasp/packages/chain"
+	"github.com/iotaledger/wasp/packages/chain/statemanager/sm_gpa"
 	"github.com/iotaledger/wasp/packages/chain/statemanager/sm_gpa/sm_gpa_utils"
 	"github.com/iotaledger/wasp/packages/cryptolib"
 	"github.com/iotaledger/wasp/packages/isc"
@@ -459,6 +460,7 @@ func newEnv(t *testing.T, n, f int, reliable bool) *testEnv {
 			true,
 			-1,
 			10*time.Millisecond,
+			sm_gpa.NewStateManagerParameters(),
 		)
 		require.NoError(t, err)
 		te.nodes[i].ServersUpdated(te.peerPubKeys)

packages/chain/statemanager/sm_gpa/sm_gpa_utils/block_cache_rapid_no_wal_test.go

@@ -178,7 +178,7 @@ func (bcnwtsmT *blockCacheNoWALTestSM) getAndCheckBlock(t *rapid.T, blockKey Blo
 	require.True(t, ok)
 	block := bcnwtsmT.bc.GetBlock(blockExpected.L1Commitment())
 	require.NotNil(t, block)
-	require.True(t, blockExpected.Hash().Equals(block.Hash())) // Should be Equals instead of Hash().Equals(); bwtsmT.blocks[blockHash]
+	CheckBlocksEqual(t, blockExpected, block)
 }
 
 func TestBlockCachePropBasedNoWAL(t *testing.T) {

packages/chain/statemanager/sm_gpa/sm_gpa_utils/block_wal_rapid_test.go

@@ -142,7 +142,9 @@ func (bwtsmT *blockWALTestSM) ReadGoodBlock(t *rapid.T) {
 	blockHash := rapid.SampledFrom(blockHashes).Example()
 	block, err := bwtsmT.bw.Read(blockHash)
 	require.NoError(t, err)
-	require.True(t, block.Hash().Equals(blockHash)) // Should be Equals instead of Hash().Equals(); bwtsmT.blocks[blockHash]
+	blockExpected, ok := bwtsmT.blocks[blockHash]
+	require.True(t, ok)
+	CheckBlocksEqual(t, blockExpected, block)
 	t.Logf("Block %s read", blockHash)
 }
 
@@ -153,7 +155,9 @@ func (bwtsmT *blockWALTestSM) ReadMovedBlock(t *rapid.T) {
 	blockHash := rapid.SampledFrom(bwtsmT.blocksMoved).Example()
 	block, err := bwtsmT.bw.Read(blockHash)
 	require.NoError(t, err)
-	require.False(t, block.Hash().Equals(blockHash)) // Should be Equals instead of Hash().Equals(); bwtsmT.blocks[blockHash]
+	blockExpected, ok := bwtsmT.blocks[blockHash]
+	require.True(t, ok)
+	CheckBlocksDifferent(t, blockExpected, block)
 	t.Logf("Moved block %s read", blockHash)
 }
 

packages/chain/statemanager/sm_gpa/sm_gpa_utils/block_wal_test.go

@@ -40,7 +40,7 @@ func TestBlockWALBasic(t *testing.T) {
 	for i := range blocksInWAL {
 		block, err2 := walGood.Read(blocks[i].Hash())
 		require.NoError(t, err2)
-		require.True(t, blocks[i].Hash().Equals(block.Hash())) // Should be Equals instead of Hash().Equals()
+		CheckBlocksEqual(t, blocks[i], block)
 		_, err2 = walBad.Read(blocks[i].Hash())
 		require.Error(t, err2)
 	}
@@ -88,8 +88,7 @@ func TestBlockWALOverwrite(t *testing.T) {
 	require.True(t, wal.Contains(blocks[0].Hash()))
 	block, err = wal.Read(blocks[0].Hash())
 	require.NoError(t, err)
-	require.True(t, blocks[0].Hash().Equals(block.Hash()))
-	// require.True(t, blocks[0].Equals(block))
+	CheckBlocksEqual(t, blocks[0], block)
 }
 
 // Check if after restart wal is functioning correctly
@@ -114,7 +113,7 @@ func TestBlockWALRestart(t *testing.T) {
 		require.True(t, wal.Contains(blocks[i].Hash()))
 		block, err := wal.Read(blocks[i].Hash())
 		require.NoError(t, err)
-		require.True(t, blocks[i].Hash().Equals(block.Hash())) // Should be Equals instead of Hash().Equals()
+		CheckBlocksEqual(t, blocks[i], block)
 	}
 }
 

packages/chain/statemanager/sm_gpa/sm_gpa_utils/test_block_factory.go

@@ -17,6 +17,7 @@ import (
 	"github.com/iotaledger/wasp/packages/isc/coreutil"
 	"github.com/iotaledger/wasp/packages/kv"
 	"github.com/iotaledger/wasp/packages/kv/codec"
+	"github.com/iotaledger/wasp/packages/kv/dict"
 	"github.com/iotaledger/wasp/packages/origin"
 	"github.com/iotaledger/wasp/packages/state"
 	"github.com/iotaledger/wasp/packages/testutil"
@@ -30,14 +31,21 @@ type BlockFactory struct {
 	aliasOutputs        map[state.BlockHash]*isc.AliasOutputWithID
 }
 
-func NewBlockFactory(t require.TestingT) *BlockFactory {
+func NewBlockFactory(t require.TestingT, chainInitParamsOpt ...dict.Dict) *BlockFactory {
+	var chainInitParams dict.Dict
+	if len(chainInitParamsOpt) > 0 {
+		chainInitParams = chainInitParamsOpt[0]
+	} else {
+		chainInitParams = nil
+	}
 	aliasOutput0ID := iotago.OutputIDFromTransactionIDAndIndex(getRandomTxID(t), 0)
 	chainID := isc.ChainIDFromAliasID(iotago.AliasIDFromOutputID(aliasOutput0ID))
 	stateAddress := cryptolib.NewKeyPair().GetPublicKey().AsEd25519Address()
+	originCommitment := origin.L1Commitment(chainInitParams, 0)
 	aliasOutput0 := &iotago.AliasOutput{
 		Amount:        tpkg.TestTokenSupply,
 		AliasID:       chainID.AsAliasID(), // NOTE: not very correct: origin output's AliasID should be empty; left here to make mocking transitions easier
-		StateMetadata: testutil.DummyStateMetadata(origin.L1Commitment(nil, 0)).Bytes(),
+		StateMetadata: testutil.DummyStateMetadata(originCommitment).Bytes(),
 		Conditions: iotago.UnlockConditions{
 			&iotago.StateControllerAddressUnlockCondition{Address: stateAddress},
 			&iotago.GovernorAddressUnlockCondition{Address: stateAddress},
@@ -49,16 +57,15 @@ func NewBlockFactory(t require.TestingT) *BlockFactory {
 		},
 	}
 	aliasOutputs := make(map[state.BlockHash]*isc.AliasOutputWithID)
-	originCommitment := origin.L1Commitment(nil, 0)
 	originOutput := isc.NewAliasOutputWithID(aliasOutput0, aliasOutput0ID)
 	aliasOutputs[originCommitment.BlockHash()] = originOutput
 	chainStore := state.NewStore(mapdb.NewMapDB())
-	origin.InitChain(chainStore, nil, 0)
+	origin.InitChain(chainStore, chainInitParams, 0)
 	return &BlockFactory{
 		t:                   t,
 		store:               chainStore,
 		chainID:             chainID,
-		lastBlockCommitment: origin.L1Commitment(nil, 0),
+		lastBlockCommitment: originCommitment,
 		aliasOutputs:        aliasOutputs,
 	}
 }

packages/chain/statemanager/sm_gpa/sm_gpa_utils/test_utils.go

@@ -0,0 +1,36 @@
+// Copyright 2020 IOTA Stiftung
+// SPDX-License-Identifier: Apache-2.0
+
+package sm_gpa_utils
+
+import (
+	"github.com/stretchr/testify/require"
+
+	"github.com/iotaledger/wasp/packages/state"
+)
+
+func CheckBlockInStore(t require.TestingT, store state.Store, origBlock state.Block) {
+	blockFromStore, err := store.BlockByTrieRoot(origBlock.TrieRoot())
+	require.NoError(t, err)
+	CheckBlocksEqual(t, origBlock, blockFromStore)
+}
+
+// NOTE: this function should not exist. state.Block should have Equals method
+func CheckBlocksEqual(t require.TestingT, block1, block2 state.Block) {
+	require.Equal(t, block1.StateIndex(), block2.StateIndex())
+	require.True(t, block1.PreviousL1Commitment().Equals(block2.PreviousL1Commitment()))
+	require.True(t, block1.L1Commitment().Equals(block2.L1Commitment()))
+	// NOTE: having separate sentences instead of require.True(t, BlocksEqual(block1, block2))
+	//       to have a more precise location of error in logs.
+}
+
+func BlocksEqual(block1, block2 state.Block) bool {
+	return block1.StateIndex() == block2.StateIndex() &&
+		block1.PreviousL1Commitment().Equals(block2.PreviousL1Commitment()) &&
+		block1.L1Commitment().Equals(block2.L1Commitment())
+}
+
+// NOTE: this function should not exist. state.Block should have Equals method
+func CheckBlocksDifferent(t require.TestingT, block1, block2 state.Block) {
+	require.False(t, block1.Hash().Equals(block2.Hash()))
+}

packages/chain/statemanager/sm_gpa/sm_messages/block_message_test.go

@@ -18,6 +18,6 @@ func TestMarshalUnmarshalBlockMessage(t *testing.T) {
 		unmarshaled := NewEmptyBlockMessage()
 		err = unmarshaled.UnmarshalBinary(marshaled)
 		require.NoError(t, err)
-		require.True(t, blocks[i].Hash().Equals(unmarshaled.GetBlock().Hash())) // Should be Equals instead of Hash().Equals()
+		sm_gpa_utils.CheckBlocksEqual(t, blocks[i], unmarshaled.GetBlock())
 	}
 }