feat(consensus): persist standalone WAL with crash-safe replay (#3763)

* feat(consensus): persist standalone WAL with crash-safe replay

* fix: handle WAL file close errors

* test: cover WAL record and prune watermark decode paths

* fix(db/memory): avoid panic when temp path creation fails

* refactor: use constant for estimatedRecordBytes

* refactor: pass WalRecordEnvelope by reference

* refactor: remove impossible WAL payload size check

* refactor: remove impossible WAL payload nil checks

* refactor: pass WAL proposal value by reference

* fix: clarify remaining bytes error

* refactor: rename consensus db package to walstore

* refactor: combine duplicate WAL number checks

* refactor: pass WAL uint64 arrays by reference

* refactor: remove redundant WAL entry checks

* test: assert memory DB copy path after copy

* refactor: clarify WAL batch record indexing

Author: ongyimeng

consensus/consensus.go

@@ -5,7 +5,6 @@ import (
 
 	"github.com/NethermindEth/juno/blockchain"
 	"github.com/NethermindEth/juno/builder"
-	consensusDB "github.com/NethermindEth/juno/consensus/db"
 	"github.com/NethermindEth/juno/consensus/driver"
 	"github.com/NethermindEth/juno/consensus/p2p"
 	"github.com/NethermindEth/juno/consensus/p2p/config"
@@ -16,6 +15,7 @@ import (
 	"github.com/NethermindEth/juno/consensus/tendermint"
 	"github.com/NethermindEth/juno/consensus/types"
 	"github.com/NethermindEth/juno/consensus/votecounter"
+	"github.com/NethermindEth/juno/consensus/walstore"
 	"github.com/NethermindEth/juno/core/felt"
 	"github.com/NethermindEth/juno/db"
 	"github.com/NethermindEth/juno/p2p/sync"
@@ -33,6 +33,15 @@ type ConsensusServices struct {
 	CommitListener driver.CommitListener[starknet.Value, starknet.Hash]
 }
 
+type initOptions struct {
+	wrapWALStore func(
+		walstore.TendermintWALStore[starknet.Value, starknet.Hash, starknet.Address],
+	) walstore.TendermintWALStore[starknet.Value, starknet.Hash, starknet.Address]
+	wrapBroadcasters func(
+		p2p.Broadcasters[starknet.Value, starknet.Hash, starknet.Address],
+	) p2p.Broadcasters[starknet.Value, starknet.Hash, starknet.Address]
+}
+
 func Init(
 	host host.Host,
 	logger *log.ZapLogger,
@@ -45,16 +54,53 @@ func Init(
 	timeoutFn driver.TimeoutFn,
 	bootstrapPeersFn func() []peer.AddrInfo,
 	compiler compiler.Compiler,
+) (ConsensusServices, error) {
+	return initWithOptions(
+		host,
+		logger,
+		database,
+		blockchain,
+		vm,
+		blockFetcher,
+		nodeAddress,
+		validators,
+		timeoutFn,
+		bootstrapPeersFn,
+		compiler,
+		initOptions{},
+	)
+}
+
+func initWithOptions(
+	host host.Host,
+	logger *log.ZapLogger,
+	database db.KeyValueStore,
+	blockchain *blockchain.Blockchain,
+	vm vm.VM,
+	blockFetcher *sync.BlockFetcher,
+	nodeAddress *starknet.Address,
+	validators votecounter.Validators[starknet.Address],
+	timeoutFn driver.TimeoutFn,
+	bootstrapPeersFn func() []peer.AddrInfo,
+	compiler compiler.Compiler,
+	options initOptions,
 ) (ConsensusServices, error) {
 	chainHeight, err := blockchain.Height()
 	if err != nil && !errors.Is(err, db.ErrKeyNotFound) {
 		return ConsensusServices{}, err
 	}
 	currentHeight := types.Height(chainHeight + 1)
-
-	tendermintDB := consensusDB.NewTendermintDB[
-		starknet.Value, starknet.Hash, starknet.Address,
+	tendermintWALStore, err := walstore.NewTendermintWALStore[
+		starknet.Value,
+		starknet.Hash,
+		starknet.Address,
 	](database)
+	if err != nil {
+		return ConsensusServices{}, err
+	}
+	if options.wrapWALStore != nil {
+		tendermintWALStore = options.wrapWALStore(tendermintWALStore)
+	}
 
 	executor := builder.NewExecutor(blockchain, vm, logger, false, false)
 	builder := builder.New(blockchain, executor)
@@ -77,13 +123,17 @@ func Init(
 	commitListener := driver.NewCommitListener(logger, &proposalStore, proposer, p2p)
 
 	messageExtractor := consensusSync.New(validators, toValue, &proposalStore)
+	broadcasters := p2p.Broadcasters()
+	if options.wrapBroadcasters != nil {
+		broadcasters = options.wrapBroadcasters(broadcasters)
+	}
 
 	driver := driver.New(
 		logger,
-		tendermintDB,
+		tendermintWALStore,
 		stateMachine,
 		commitListener,
-		p2p.Broadcasters(),
+		broadcasters,
 		p2p.Listeners(),
 		blockFetcher,
 		&messageExtractor,

consensus/consensus_test.go

@@ -206,7 +206,7 @@ func writeBlock(
 			require.NoError(t, err)
 			require.NoError(t, bc.Store(committedBlock.Block, commitments, committedBlock.StateUpdate, committedBlock.NewClasses))
 
-			close(committedBlock.Persisted)
+			committedBlock.Persisted <- nil
 
 			commit := commit{
 				nodeIndex:      index,