dbft: refactor transactions management

CHANGELOG.md

@@ -7,10 +7,11 @@ This document outlines major changes between releases.
 New features:
 
 Behaviour changes:
+ * transactions management refactoring implied library API changes (#164)
 
 Improvements:
  * minimum required Go version is 1.26 (#144, #156, #162)
- * `go.uber.org/zap` dependency upgrade from v1.27.0 to v1.27.1 (#156) 
+ * `go.uber.org/zap` dependency upgrade from v1.27.0 to v1.27.1 (#156)
 
 Bugs fixed:
 

config.go

@@ -2,6 +2,7 @@ package dbft
 
 import (
 	"errors"
+	"iter"
 	"time"
 
 	"go.uber.org/zap"
@@ -42,7 +43,7 @@ type Config[H Hash] struct {
 	// RequestTx is a callback which is called when transaction contained
 	// in current block can't be found in memory pool. The slice received by
 	// this callback MUST NOT be changed.
-	RequestTx func(h ...H)
+	RequestTx func(h iter.Seq[H])
 	// SubscribeForTxs is a callback which is called when dBFT needs to track incoming
 	// mempool transactions. Subscription is supposed to be single-use, no unsubscription
 	// is initiated by dBFT, hence it's the user's duty to manage and release resources.
@@ -51,8 +52,6 @@ type Config[H Hash] struct {
 	// StopTxFlow is a callback which is called when the process no longer needs
 	// any transactions.
 	StopTxFlow func()
-	// GetTx returns a transaction from memory pool.
-	GetTx func(h H) Transaction[H]
 	// GetVerified returns a slice of verified transactions
 	// to be proposed in a new block.
 	GetVerified func() []Transaction[H]
@@ -82,7 +81,7 @@ type Config[H Hash] struct {
 	// NewConsensusPayload is a constructor for payload.ConsensusPayload.
 	NewConsensusPayload func(*Context[H], MessageType, any) ConsensusPayload[H]
 	// NewPrepareRequest is a constructor for payload.PrepareRequest.
-	NewPrepareRequest func(ts uint64, nonce uint64, transactionHashes []H) PrepareRequest[H]
+	NewPrepareRequest func(ts uint64, nonce uint64, txes []Transaction[H]) PrepareRequest[H]
 	// NewPrepareResponse is a constructor for payload.PrepareResponse.
 	NewPrepareResponse func(preparationHash H) PrepareResponse[H]
 	// NewChangeView is a constructor for payload.ChangeView.
@@ -120,9 +119,8 @@ func defaultConfig[H Hash]() *Config[H] {
 		TimePerBlock:       func() time.Duration { return defaultSecondsPerBlock },
 		TimestampIncrement: defaultTimestampIncrement,
 		GetKeyPair:         nil,
-		RequestTx:          func(...H) {},
+		RequestTx:          func(iter.Seq[H]) {},
 		StopTxFlow:         func() {},
-		GetTx:              func(H) Transaction[H] { return nil },
 		GetVerified:        func() []Transaction[H] { return make([]Transaction[H], 0) },
 		VerifyBlock:        func(Block[H]) bool { return true },
 		Broadcast:          func(ConsensusPayload[H]) {},
@@ -275,7 +273,7 @@ func WithNewBlockFromContext[H Hash](f func(ctx *Context[H]) Block[H]) func(conf
 }
 
 // WithRequestTx sets RequestTx.
-func WithRequestTx[H Hash](f func(h ...H)) func(config *Config[H]) {
+func WithRequestTx[H Hash](f func(hs iter.Seq[H])) func(config *Config[H]) {
 	return func(cfg *Config[H]) {
 		cfg.RequestTx = f
 	}
@@ -295,13 +293,6 @@ func WithStopTxFlow[H Hash](f func()) func(config *Config[H]) {
 	}
 }
 
-// WithGetTx sets GetTx.
-func WithGetTx[H Hash](f func(h H) Transaction[H]) func(config *Config[H]) {
-	return func(cfg *Config[H]) {
-		cfg.GetTx = f
-	}
-}
-
 // WithGetVerified sets GetVerified.
 func WithGetVerified[H Hash](f func() []Transaction[H]) func(config *Config[H]) {
 	return func(cfg *Config[H]) {
@@ -388,7 +379,7 @@ func WithNewConsensusPayload[H Hash](f func(ctx *Context[H], typ MessageType, ms
 }
 
 // WithNewPrepareRequest sets NewPrepareRequest.
-func WithNewPrepareRequest[H Hash](f func(ts uint64, nonce uint64, transactionHashes []H) PrepareRequest[H]) func(config *Config[H]) {
+func WithNewPrepareRequest[H Hash](f func(ts uint64, nonce uint64, transactionHashes []Transaction[H]) PrepareRequest[H]) func(config *Config[H]) {
 	return func(cfg *Config[H]) {
 		cfg.NewPrepareRequest = f
 	}

context.go

@@ -55,12 +55,12 @@ type Context[H Hash] struct {
 	// Timestamp is a nanosecond-precision timestamp
 	Timestamp uint64
 	Nonce     uint64
-	// TransactionHashes is a slice of hashes of proposed transactions in the current block.
-	TransactionHashes []H
-	// MissingTransactions is a slice of hashes containing missing transactions for the current block.
-	MissingTransactions []H
-	// Transactions is a map containing actual transactions for the current block.
-	Transactions map[H]Transaction[H]
+	// Transactions is a slice containing actual transactions for the current block.
+	Transactions []Transaction[H]
+	// MissingTransactions is a map of a missing transaction hash to its index in
+	// the slice of proposed transactions for the current block. This map is
+	// managed by dBFT and should not be changed by the user.
+	MissingTransactions map[H]int
 
 	// PreparationPayloads stores consensus Prepare* payloads for the current epoch.
 	PreparationPayloads []ConsensusPayload[H]
@@ -287,15 +287,8 @@ func (c *Context[H]) reset(view byte, ts uint64) {
 	}
 	c.PreparationPayloads = emptyReusableSlice(c.PreparationPayloads, n)
 
-	if c.Transactions == nil { // Init.
-		c.Transactions = make(map[H]Transaction[H])
-	} else { // Regular use.
-		clear(c.Transactions)
-	}
-	c.TransactionHashes = nil
-	if c.MissingTransactions != nil {
-		c.MissingTransactions = c.MissingTransactions[:0]
-	}
+	c.Transactions = nil        // overwritten by the user on every proposal, no need for manual clear management.
+	c.MissingTransactions = nil // overwritten by the user on every proposal, no need for manual clear management.
 	c.PrimaryIndex = c.GetPrimaryIndex(view)
 	c.ViewNumber = view
 
@@ -325,13 +318,7 @@ func (c *Context[H]) Fill(force bool) bool {
 	_, _ = rand.Read(b)
 
 	c.Nonce = binary.LittleEndian.Uint64(b)
-	c.TransactionHashes = make([]H, len(txx))
-
-	for i := range txx {
-		h := txx[i].Hash()
-		c.TransactionHashes[i] = h
-		c.Transactions[h] = txx[i]
-	}
+	c.Transactions = txx
 
 	c.Timestamp = c.lastBlockTimestamp + c.Config.TimestampIncrement
 	if now := c.getTimestamp(); now > c.Timestamp {
@@ -353,18 +340,12 @@ func (c *Context[H]) CreateBlock() Block[H] {
 			return nil
 		}
 
-		txx := make([]Transaction[H], len(c.TransactionHashes))
-
-		for i, h := range c.TransactionHashes {
-			txx[i] = c.Transactions[h]
-		}
-
 		// Anti-MEV extension properly sets PreBlock transactions once during PreBlock
 		// construction and then never updates these transactions in the dBFT context.
-		// Thus, user must not reuse txx if anti-MEV extension is enabled. However,
+		// Thus, user must not reuse c.Transactions if anti-MEV extension is enabled. However,
 		// we don't skip a call to Block.SetTransactions since it may be used as a
 		// signal to the user's code to finalize the block.
-		c.block.SetTransactions(txx)
+		c.block.SetTransactions(c.Transactions)
 	}
 
 	return c.block
@@ -377,13 +358,7 @@ func (c *Context[H]) CreatePreBlock() PreBlock[H] {
 			return nil
 		}
 
-		txx := make([]Transaction[H], len(c.TransactionHashes))
-
-		for i, h := range c.TransactionHashes {
-			txx[i] = c.Transactions[h]
-		}
-
-		c.preBlock.SetTransactions(txx)
+		c.preBlock.SetTransactions(c.Transactions)
 	}
 
 	return c.preBlock
@@ -432,7 +407,7 @@ func (c *Context[H]) MakePreHeader() PreBlock[H] {
 // hasAllTransactions returns true iff all transactions were received
 // for the proposed block.
 func (c *Context[H]) hasAllTransactions() bool {
-	return len(c.TransactionHashes) == len(c.Transactions)
+	return len(c.MissingTransactions) == 0
 }
 
 func (c *Context[H]) subscribeForTransactions() {

dbft.go

@@ -2,7 +2,7 @@ package dbft
 
 import (
 	"fmt"
-	"slices"
+	"maps"
 	"sync"
 	"time"
 
@@ -51,7 +51,12 @@ func New[H Hash](options ...func(config *Config[H])) (*DBFT[H], error) {
 }
 
 func (d *DBFT[H]) addTransaction(tx Transaction[H]) {
-	d.Transactions[tx.Hash()] = tx
+	i, ok := d.MissingTransactions[tx.Hash()]
+	if !ok {
+		return
+	}
+	d.Transactions[i] = tx
+	delete(d.MissingTransactions, tx.Hash())
 	if d.hasAllTransactions() {
 		if d.IsPrimary() || d.Context.WatchOnly() {
 			return
@@ -170,21 +175,11 @@ func (d *DBFT[H]) OnTransaction(tx Transaction[H]) {
 	// 	zap.Bool("block_sent", d.BlockSent()))
 	if !d.IsBackup() || d.NotAcceptingPayloadsDueToViewChanging() ||
 		!d.RequestSentOrReceived() || d.ResponseSent() || d.PreCommitSent() ||
-		d.CommitSent() || d.BlockSent() || len(d.MissingTransactions) == 0 {
+		d.CommitSent() || d.BlockSent() || d.hasAllTransactions() {
 		return
 	}
 
-	i := slices.Index(d.MissingTransactions, tx.Hash())
-	if i < 0 {
-		return
-	}
 	d.addTransaction(tx)
-	// `addTransaction` checks for responses and commits. If this was the last transaction
-	// Context could be initialized on a new height, clearing this field.
-	if len(d.MissingTransactions) == 0 {
-		return
-	}
-	d.MissingTransactions = slices.Delete(d.MissingTransactions, i, i+1)
 }
 
 // OnTimeout advances state machine as if timeout was fired.
@@ -349,12 +344,12 @@ func (d *DBFT[H]) onPrepareRequest(msg ConsensusPayload[H]) {
 
 	d.Timestamp = p.Timestamp()
 	d.Nonce = p.Nonce()
-	d.TransactionHashes = p.TransactionHashes()
+	d.Transactions, d.MissingTransactions = p.Transactions()
 
-	d.Logger.Info("received PrepareRequest", zap.Uint16("validator", msg.ValidatorIndex()), zap.Int("tx", len(d.TransactionHashes)))
-	d.processMissingTx()
+	d.Logger.Info("received PrepareRequest", zap.Uint16("validator", msg.ValidatorIndex()), zap.Int("tx", len(d.Transactions)))
 	d.updateExistingPayloads(msg)
 	d.PreparationPayloads[msg.ValidatorIndex()] = msg
+	d.processMissingTx()
 
 	if !d.hasAllTransactions() || !d.createAndCheckBlock() || d.Context.WatchOnly() {
 		return
@@ -364,22 +359,12 @@ func (d *DBFT[H]) onPrepareRequest(msg ConsensusPayload[H]) {
 	d.checkPrepare()
 }
 
+// processMissingTx fills in the map of missing transactions and requests them.
 func (d *DBFT[H]) processMissingTx() {
-	for _, h := range d.TransactionHashes {
-		if _, ok := d.Transactions[h]; ok {
-			continue
-		}
-		if tx := d.GetTx(h); tx == nil {
-			d.MissingTransactions = append(d.MissingTransactions, h)
-		} else {
-			d.Transactions[h] = tx
-		}
-	}
-
 	if len(d.MissingTransactions) != 0 {
 		d.Logger.Info("missing tx",
 			zap.Int("count", len(d.MissingTransactions)))
-		d.RequestTx(d.MissingTransactions...)
+		d.RequestTx(maps.Keys(d.MissingTransactions))
 	}
 }