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

consensus/consensus.go

@@ -5,7 +5,6 @@
 
 	"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 @@
 	"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 @@
 	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 @@
 	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 @@
 	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,