core/txpool/locals: improve lock management

core/txpool/locals/journal.go

@@ -71,6 +71,7 @@ func (journal *journal) load(add func([]*types.Transaction) []error) error {
 
 	// Temporarily discard any journal additions (don't double add on load)
 	journal.writer = new(devNull)
+	defer func() { journal.writer = nil }()
 
 	// Inject all transactions from the journal into the pool
 	stream := rlp.NewStream(input, 0)
@@ -112,6 +113,7 @@ func (journal *journal) load(add func([]*types.Transaction) []error) error {
 		}
 	}
 	log.Info("Loaded local transaction journal", "transactions", total, "dropped", dropped)
+
 	return failure
 }
 
@@ -194,7 +196,6 @@ func (journal *journal) rotate(all map[common.Address]types.Transactions) error
 // close flushes the transaction journal contents to disk and closes the file.
 func (journal *journal) close() error {
 	var err error
-
 	if journal.writer != nil {
 		err = journal.writer.Close()
 		journal.writer = nil

core/txpool/locals/tx_tracker.go

@@ -114,13 +114,14 @@ func (tracker *TxTracker) TrackAll(txs []*types.Transaction) {
 }
 
 // recheck checks and returns any transactions that needs to be resubmitted.
-func (tracker *TxTracker) recheck(journalCheck bool) (resubmits []*types.Transaction, rejournal map[common.Address]types.Transactions) {
+func (tracker *TxTracker) recheck(journalCheck bool) []*types.Transaction {
 	tracker.mu.Lock()
 	defer tracker.mu.Unlock()
 
 	var (
 		numStales = 0
 		numOk     = 0
+		resubmits []*types.Transaction
 	)
 	for sender, txs := range tracker.byAddr {
 		// Wipe the stales
@@ -141,7 +142,7 @@ func (tracker *TxTracker) recheck(journalCheck bool) (resubmits []*types.Transac
 	}
 
 	if journalCheck { // rejournal
-		rejournal = make(map[common.Address]types.Transactions)
+		rejournal := make(map[common.Address]types.Transactions)
 		for _, tx := range tracker.all {
 			addr, _ := types.Sender(tracker.signer, tx)
 			rejournal[addr] = append(rejournal[addr], tx)
@@ -153,10 +154,16 @@ func (tracker *TxTracker) recheck(journalCheck bool) (resubmits []*types.Transac
 				return int(a.Nonce() - b.Nonce())
 			})
 		}
+		// Rejournal the tracker while holding the lock. No new transactions will
+		// be added to the old journal during this period, preventing any potential
+		// transaction loss.
+		if err := tracker.journal.rotate(rejournal); err != nil {
+			log.Warn("Transaction journal rotation failed", "err", err)
+		}
 	}
 	localGauge.Update(int64(len(tracker.all)))
 	log.Debug("Tx tracker status", "need-resubmit", len(resubmits), "stale", numStales, "ok", numOk)
-	return resubmits, rejournal
+	return resubmits
 }
 
 // Start implements node.Lifecycle interface
@@ -186,16 +193,11 @@ func (tracker *TxTracker) loop() {
 			return nil
 		})
 
-		// Setup the writer for the upcoming transactions. Lock to prevent
+		// Setup the writer for the upcoming transactions
-		// journal.setupWriter <-> journal.insert (via TrackAll) conflicts.
-		tracker.mu.Lock()
 		if err := tracker.journal.setupWriter(); err != nil {
 			log.Error("Failed to setup the journal writer", "err", err)
-			tracker.mu.Unlock()
 			return
 		}
-		tracker.mu.Unlock()
-
 		defer tracker.journal.close()
 	}
 	var (
@@ -207,20 +209,15 @@ func (tracker *TxTracker) loop() {
 		case <-tracker.shutdownCh:
 			return
 		case <-timer.C:
-			checkJournal := tracker.journal != nil && time.Since(lastJournal) > tracker.rejournal
+			var rejournal bool
-			resubmits, rejournal := tracker.recheck(checkJournal)
+			if tracker.journal != nil && time.Since(lastJournal) > tracker.rejournal {
+				rejournal, lastJournal = true, time.Now()
+				log.Debug("Rejournal the transaction tracker")
+			}
+			resubmits := tracker.recheck(rejournal)
 			if len(resubmits) > 0 {
 				tracker.pool.Add(resubmits, false)
 			}
-			if checkJournal {
-				// Lock to prevent journal.rotate <-> journal.insert (via TrackAll) conflicts
-				tracker.mu.Lock()
-				lastJournal = time.Now()
-				if err := tracker.journal.rotate(rejournal); err != nil {
-					log.Warn("Transaction journal rotation failed", "err", err)
-				}
-				tracker.mu.Unlock()
-			}
 			timer.Reset(recheckInterval)
 		}
 	}

core/txpool/locals/tx_tracker_test.go

@@ -17,7 +17,11 @@
 package locals
 
 import (
+	"fmt"
+	"maps"
 	"math/big"
+	"math/rand"
+	"path/filepath"
 	"testing"
 	"time"
 
@@ -146,20 +150,59 @@ func TestResubmit(t *testing.T) {
 	txsA := txs[:len(txs)/2]
 	txsB := txs[len(txs)/2:]
 	env.pool.Add(txsA, true)
+
 	pending, queued := env.pool.ContentFrom(address)
 	if len(pending) != len(txsA) || len(queued) != 0 {
 		t.Fatalf("Unexpected txpool content: %d, %d", len(pending), len(queued))
 	}
 	env.tracker.TrackAll(txs)
 
-	resubmit, all := env.tracker.recheck(true)
+	resubmit := env.tracker.recheck(true)
 	if len(resubmit) != len(txsB) {
 		t.Fatalf("Unexpected transactions to resubmit, got: %d, want: %d", len(resubmit), len(txsB))
 	}
-	if len(all) == 0 || len(all[address]) == 0 {
+	env.tracker.mu.Lock()
-		t.Fatalf("Unexpected transactions being tracked, got: %d, want: %d", 0, len(txs))
+	allCopy := maps.Clone(env.tracker.all)
-	}
+	env.tracker.mu.Unlock()
-	if len(all[address]) != len(txs) {
+
-		t.Fatalf("Unexpected transactions being tracked, got: %d, want: %d", len(all[address]), len(txs))
+	if len(allCopy) != len(txs) {
+		t.Fatalf("Unexpected transactions being tracked, got: %d, want: %d", len(allCopy), len(txs))
+	}
+}
+
+func TestJournal(t *testing.T) {
+	journalPath := filepath.Join(t.TempDir(), fmt.Sprintf("%d", rand.Int63()))
+	env := newTestEnv(t, 10, 0, journalPath)
+	defer env.close()
+
+	env.tracker.Start()
+	defer env.tracker.Stop()
+
+	txs := env.makeTxs(10)
+	txsA := txs[:len(txs)/2]
+	txsB := txs[len(txs)/2:]
+	env.pool.Add(txsA, true)
+
+	pending, queued := env.pool.ContentFrom(address)
+	if len(pending) != len(txsA) || len(queued) != 0 {
+		t.Fatalf("Unexpected txpool content: %d, %d", len(pending), len(queued))
+	}
+	env.tracker.TrackAll(txsA)
+	env.tracker.TrackAll(txsB)
+	env.tracker.recheck(true) // manually rejournal the tracker
+
+	// Make sure all the transactions are properly journalled
+	trackerB := New(journalPath, time.Minute, gspec.Config, env.pool)
+	trackerB.journal.load(func(transactions []*types.Transaction) []error {
+		trackerB.TrackAll(transactions)
+		return nil
+	})
+
+	trackerB.mu.Lock()
+	allCopy := maps.Clone(trackerB.all)
+	trackerB.mu.Unlock()
+
+	if len(allCopy) != len(txs) {
+		t.Fatalf("Unexpected transactions being tracked, got: %d, want: %d", len(allCopy), len(txs))
 	}
 }