diff --git a/pkg/core/blockchain.go b/pkg/core/blockchain.go index 11d0d78de..3014f87fb 100644 --- a/pkg/core/blockchain.go +++ b/pkg/core/blockchain.go @@ -394,7 +394,7 @@ func (bc *Blockchain) Run() { persistTimer := time.NewTimer(persistInterval) defer func() { persistTimer.Stop() - if err := bc.persist(); err != nil { + if _, err := bc.persist(); err != nil { bc.log.Warn("failed to persist", zap.Error(err)) } if err := bc.dao.Store.Close(); err != nil { @@ -408,13 +408,15 @@ func (bc *Blockchain) Run() { case <-bc.stopCh: return case <-persistTimer.C: - go func() { - err := bc.persist() - if err != nil { - bc.log.Warn("failed to persist blockchain", zap.Error(err)) - } - persistTimer.Reset(persistInterval) - }() + dur, err := bc.persist() + if err != nil { + bc.log.Warn("failed to persist blockchain", zap.Error(err)) + } + interval := persistInterval - dur + if interval <= 0 { + interval = time.Microsecond // Reset doesn't work with zero value + } + persistTimer.Reset(interval) } } } @@ -1175,41 +1177,43 @@ func (bc *Blockchain) LastBatch() *storage.MemBatch { } // persist flushes current in-memory Store contents to the persistent storage. -func (bc *Blockchain) persist() error { +func (bc *Blockchain) persist() (time.Duration, error) { var ( start = time.Now() + duration time.Duration persisted int err error ) persisted, err = bc.dao.Persist() if err != nil { - return err + return 0, err } if persisted > 0 { bHeight, err := bc.dao.GetCurrentBlockHeight() if err != nil { - return err + return 0, err } oldHeight := atomic.SwapUint32(&bc.persistedHeight, bHeight) diff := bHeight - oldHeight storedHeaderHeight, _, err := bc.dao.GetCurrentHeaderHeight() if err != nil { - return err + return 0, err } + duration = time.Since(start) bc.log.Info("persisted to disk", zap.Uint32("blocks", diff), zap.Int("keys", persisted), zap.Uint32("headerHeight", storedHeaderHeight), zap.Uint32("blockHeight", bHeight), - zap.Duration("took", time.Since(start))) + zap.Duration("took", duration)) // update monitoring metrics. updatePersistedHeightMetric(bHeight) } - return nil + return duration, nil } // GetTransaction returns a TX and its height by the given hash. The height is MaxUint32 if tx is in the mempool. diff --git a/pkg/core/blockchain_test.go b/pkg/core/blockchain_test.go index d3a2b7ead..8f0d00d51 100644 --- a/pkg/core/blockchain_test.go +++ b/pkg/core/blockchain_test.go @@ -118,7 +118,8 @@ func TestAddBlock(t *testing.T) { assert.Equal(t, lastBlock.Hash(), bc.CurrentHeaderHash()) // This one tests persisting blocks, so it does need to persist() - require.NoError(t, bc.persist()) + _, err = bc.persist() + require.NoError(t, err) for _, block := range blocks { key := storage.AppendPrefix(storage.DataBlock, block.Hash().BytesBE()) @@ -236,7 +237,8 @@ func TestGetHeader(t *testing.T) { b2 := bc.newBlock() _, err = bc.GetHeader(b2.Hash()) assert.Error(t, err) - assert.NoError(t, bc.persist()) + _, err = bc.persist() + assert.NoError(t, err) } } @@ -253,7 +255,8 @@ func TestGetBlock(t *testing.T) { assert.Equal(t, blocks[i].Index, block.Index) assert.Equal(t, blocks[i].Hash(), block.Hash()) } - assert.NoError(t, bc.persist()) + _, err = bc.persist() + assert.NoError(t, err) } t.Run("store only header", func(t *testing.T) { @@ -1305,7 +1308,8 @@ func TestHasBlock(t *testing.T) { } newBlock := bc.newBlock() assert.False(t, bc.HasBlock(newBlock.Hash())) - assert.NoError(t, bc.persist()) + _, err = bc.persist() + assert.NoError(t, err) } } @@ -1340,7 +1344,8 @@ func TestGetTransaction(t *testing.T) { assert.Equal(t, block.Transactions[0], tx) assert.Equal(t, 1, io.GetVarSize(tx.Attributes)) assert.Equal(t, 1, io.GetVarSize(tx.Scripts)) - assert.NoError(t, bc.persist()) + _, err = bc.persist() + assert.NoError(t, err) } } diff --git a/pkg/core/native_contract_test.go b/pkg/core/native_contract_test.go index 9c7b81fcb..eb2403b4b 100644 --- a/pkg/core/native_contract_test.go +++ b/pkg/core/native_contract_test.go @@ -196,7 +196,8 @@ func TestNativeContract_Invoke(t *testing.T) { res, err := invokeContractMethod(chain, price, tn.Metadata().Hash, "sum", int64(14), int64(28)) require.NoError(t, err) checkResult(t, res, stackitem.Make(42)) - require.NoError(t, chain.persist()) + _, err = chain.persist() + require.NoError(t, err) select { case index := <-tn.blocks: diff --git a/pkg/core/native_policy_test.go b/pkg/core/native_policy_test.go index 3b315b28c..631a1718f 100644 --- a/pkg/core/native_policy_test.go +++ b/pkg/core/native_policy_test.go @@ -36,7 +36,8 @@ func testGetSet(t *testing.T, chain *Blockchain, hash util.Uint160, name string, res, err := invokeContractMethod(chain, 100000000, hash, getName) require.NoError(t, err) checkResult(t, res, stackitem.Make(defaultValue)) - require.NoError(t, chain.persist()) + _, err = chain.persist() + require.NoError(t, err) }) t.Run("set, too small value", func(t *testing.T) { @@ -68,13 +69,15 @@ func testGetSet(t *testing.T, chain *Blockchain, hash util.Uint160, name string, if name != "GasPerBlock" { // GasPerBlock is set on the next block checkResult(t, aers[1], stackitem.Make(defaultValue+1)) } - require.NoError(t, chain.persist()) + _, err = chain.persist() + require.NoError(t, err) // Get in the next block. res, err := invokeContractMethod(chain, 100000000, hash, getName) require.NoError(t, err) checkResult(t, res, stackitem.Make(defaultValue+1)) - require.NoError(t, chain.persist()) + _, err = chain.persist() + require.NoError(t, err) }) } @@ -128,7 +131,8 @@ func TestBlockedAccounts(t *testing.T) { res, err := invokeContractMethod(chain, 100000000, policyHash, "isBlocked", random.Uint160()) require.NoError(t, err) checkResult(t, res, stackitem.NewBool(false)) - require.NoError(t, chain.persist()) + _, err = chain.persist() + require.NoError(t, err) }) t.Run("block-unblock account", func(t *testing.T) { @@ -138,7 +142,8 @@ func TestBlockedAccounts(t *testing.T) { isBlocked := chain.contracts.Policy.IsBlockedInternal(chain.dao, account) require.Equal(t, isBlocked, true) - require.NoError(t, chain.persist()) + _, err = chain.persist() + require.NoError(t, err) res, err = invokeContractMethodGeneric(chain, 100000000, policyHash, "unblockAccount", true, account.BytesBE()) require.NoError(t, err) @@ -146,7 +151,8 @@ func TestBlockedAccounts(t *testing.T) { isBlocked = chain.contracts.Policy.IsBlockedInternal(chain.dao, account) require.Equal(t, false, isBlocked) - require.NoError(t, chain.persist()) + _, err = chain.persist() + require.NoError(t, err) }) t.Run("double-block", func(t *testing.T) { @@ -154,25 +160,29 @@ func TestBlockedAccounts(t *testing.T) { res, err := invokeContractMethodGeneric(chain, 100000000, policyHash, "blockAccount", true, account.BytesBE()) require.NoError(t, err) checkResult(t, res, stackitem.NewBool(true)) - require.NoError(t, chain.persist()) + _, err = chain.persist() + require.NoError(t, err) // double-block should fail res, err = invokeContractMethodGeneric(chain, 100000000, policyHash, "blockAccount", true, account.BytesBE()) require.NoError(t, err) checkResult(t, res, stackitem.NewBool(false)) - require.NoError(t, chain.persist()) + _, err = chain.persist() + require.NoError(t, err) // unblock res, err = invokeContractMethodGeneric(chain, 100000000, policyHash, "unblockAccount", true, account.BytesBE()) require.NoError(t, err) checkResult(t, res, stackitem.NewBool(true)) - require.NoError(t, chain.persist()) + _, err = chain.persist() + require.NoError(t, err) // unblock the same account should fail as we don't have it blocked res, err = invokeContractMethodGeneric(chain, 100000000, policyHash, "unblockAccount", true, account.BytesBE()) require.NoError(t, err) checkResult(t, res, stackitem.NewBool(false)) - require.NoError(t, chain.persist()) + _, err = chain.persist() + require.NoError(t, err) }) t.Run("not signed by committee", func(t *testing.T) {