cmd, core, eth, triedb/pathdb: improve journal file flushing

Signed-off-by: jsvisa <delweng@gmail.com>

cmd/geth/main.go

@@ -79,6 +79,7 @@ var (
 		utils.BlobPoolDataDirFlag,
 		utils.BlobPoolDataCapFlag,
 		utils.BlobPoolPriceBumpFlag,
+		utils.TrieDBJournalFlag,
 		utils.SyncModeFlag,
 		utils.SyncTargetFlag,
 		utils.ExitWhenSyncedFlag,
@@ -97,7 +98,7 @@ var (
 		utils.CacheFlag,
 		utils.CacheDatabaseFlag,
 		utils.CacheTrieFlag,
-		utils.CacheTrieJournalFlag,
+		utils.CacheTrieJournalFlag,   // deprecated
 		utils.CacheTrieRejournalFlag, // deprecated
 		utils.CacheGCFlag,
 		utils.CacheSnapshotFlag,

cmd/utils/flags.go

@@ -446,6 +446,15 @@ var (
 		Value:    ethconfig.Defaults.BlobPool.PriceBump,
 		Category: flags.BlobPoolCategory,
 	}
+
+	// Trie database settings
+	TrieDBJournalFlag = &cli.StringFlag{
+		Name:     "triedb.journal",
+		Usage:    "Path to the journal used for persisting trie data across node restarts",
+		Value:    ethconfig.Defaults.TrieDBJournal,
+		Category: flags.TrieDatabaseCategory,
+	}
+
 	// Performance tuning settings
 	CacheFlag = &cli.IntFlag{
 		Name:     "cache",
@@ -465,11 +474,6 @@ var (
 		Value:    15,
 		Category: flags.PerfCategory,
 	}
-	CacheTrieJournalFlag = &cli.StringFlag{
-		Name:     "cache.trie.journal",
-		Usage:    "Disk journal directory for trie cache to survive node restarts (experimental, default is disabled)",
-		Category: flags.PerfCategory,
-	}
 	CacheGCFlag = &cli.IntFlag{
 		Name:     "cache.gc",
 		Usage:    "Percentage of cache memory allowance to use for trie pruning (default = 25% full mode, 0% archive mode)",
@@ -988,6 +992,7 @@ var (
 		DataDirFlag,
 		AncientFlag,
 		EraFlag,
+		TrieDBJournalFlag,
 		RemoteDBFlag,
 		DBEngineFlag,
 		StateSchemeFlag,
@@ -1685,8 +1690,8 @@ func SetEthConfig(ctx *cli.Context, stack *node.Node, cfg *ethconfig.Config) {
 	if ctx.IsSet(CacheFlag.Name) || ctx.IsSet(CacheSnapshotFlag.Name) {
 		cfg.SnapshotCache = ctx.Int(CacheFlag.Name) * ctx.Int(CacheSnapshotFlag.Name) / 100
 	}
-	if journal := ctx.String(CacheTrieJournalFlag.Name); journal != "" {
+	if ctx.IsSet(TrieDBJournalFlag.Name) {
-		cfg.TrieJournal = stack.ResolvePath(journal)
+		cfg.TrieDBJournal = stack.ResolvePath(ctx.String(TrieDBJournalFlag.Name))
 	}
 	if ctx.IsSet(CacheLogSizeFlag.Name) {
 		cfg.FilterLogCacheSize = ctx.Int(CacheLogSizeFlag.Name)
@@ -2211,8 +2216,8 @@ func MakeChain(ctx *cli.Context, stack *node.Node, readonly bool) (*core.BlockCh
 		options.Preimages = true
 		log.Info("Enabling recording of key preimages since archive mode is used")
 	}
-	if journal := ctx.String(CacheTrieJournalFlag.Name); journal != "" {
+	if ctx.IsSet(TrieDBJournalFlag.Name) {
-		options.TrieJournal = stack.ResolvePath(journal)
+		options.TrieDBJournal = stack.ResolvePath(ctx.String(TrieDBJournalFlag.Name))
 	}
 	if !ctx.Bool(SnapshotFlag.Name) {
 		options.SnapshotLimit = 0 // Disabled

cmd/utils/flags_legacy.go

@@ -35,6 +35,7 @@ var ShowDeprecated = &cli.Command{
 var DeprecatedFlags = []cli.Flag{
 	NoUSBFlag,
 	LegacyWhitelistFlag,
+	CacheTrieJournalFlag,
 	CacheTrieRejournalFlag,
 	LegacyDiscoveryV5Flag,
 	TxLookupLimitFlag,
@@ -59,6 +60,11 @@ var (
 		Category: flags.DeprecatedCategory,
 	}
 	// Deprecated July 2023
+	CacheTrieJournalFlag = &cli.StringFlag{
+		Name:     "cache.trie.journal",
+		Usage:    "Disk journal directory for trie cache to survive node restarts",
+		Category: flags.DeprecatedCategory,
+	}
 	CacheTrieRejournalFlag = &cli.DurationFlag{
 		Name:     "cache.trie.rejournal",
 		Usage:    "Time interval to regenerate the trie cache journal",

core/blockchain.go

@@ -166,7 +166,7 @@ type BlockChainConfig struct {
 	TrieDirtyLimit   int           // Memory limit (MB) at which to start flushing dirty trie nodes to disk
 	TrieTimeLimit    time.Duration // Time limit after which to flush the current in-memory trie to disk
 	TrieNoAsyncFlush bool          // Whether the asynchronous buffer flushing is disallowed
-	TrieJournal      string        // Path used to store the trie pathdb journal
+	TrieDBJournal    string        // Path to the journal used for persisting trie data across node restarts
 
 	Preimages    bool   // Whether to store preimage of trie key to the disk
 	StateHistory uint64 // Number of blocks from head whose state histories are reserved.
@@ -247,6 +247,7 @@ func (cfg *BlockChainConfig) triedbConfig(isVerkle bool) *triedb.Config {
 			EnableStateIndexing: cfg.ArchiveMode,
 			TrieCleanSize:       cfg.TrieCleanLimit * 1024 * 1024,
 			StateCleanSize:      cfg.SnapshotLimit * 1024 * 1024,
+			JournalPath:         cfg.TrieDBJournal,
 
 			// TODO(rjl493456442): The write buffer represents the memory limit used
 			// for flushing both trie data and state data to disk. The config name

eth/backend.go

@@ -227,7 +227,6 @@ func New(stack *node.Node, config *ethconfig.Config) (*Ethereum, error) {
 			TrieDirtyLimit:   config.TrieDirtyCache,
 			ArchiveMode:      config.NoPruning,
 			TrieTimeLimit:    config.TrieTimeout,
-			TrieJournal:      config.TrieJournal,
 			SnapshotLimit:    config.SnapshotCache,
 			Preimages:        config.Preimages,
 			StateHistory:     config.StateHistory,
@@ -239,6 +238,11 @@ func New(stack *node.Node, config *ethconfig.Config) (*Ethereum, error) {
 			},
 		}
 	)
+	if config.TrieDBJournal != "" {
+		options.TrieDBJournal = stack.ResolvePath(config.TrieDBJournal)
+	} else {
+		log.Warn("Trie database journal is persisted within the database")
+	}
 
 	if config.VMTrace != "" {
 		traceConfig := json.RawMessage("{}")

eth/ethconfig/config.go

@@ -60,6 +60,7 @@ var Defaults = Config{
 	TrieCleanCache:     154,
 	TrieDirtyCache:     256,
 	TrieTimeout:        60 * time.Minute,
+	TrieDBJournal:      "triedb.journal",
 	SnapshotCache:      102,
 	FilterLogCacheSize: 32,
 	Miner:              miner.DefaultConfig,
@@ -103,12 +104,6 @@ type Config struct {
 	LogHistory           uint64 `toml:",omitempty"` // The maximum number of blocks from head where a log search index is maintained.
 	LogNoHistory         bool   `toml:",omitempty"` // No log search index is maintained.
 	LogExportCheckpoints string // export log index checkpoints to file
-	StateHistory         uint64 `toml:",omitempty"` // The maximum number of blocks from head whose state histories are reserved.
-
-	// State scheme represents the scheme used to store ethereum states and trie
-	// nodes on top. It can be 'hash', 'path', or none which means use the scheme
-	// consistent with persistent state.
-	StateScheme string `toml:",omitempty"`
 
 	// RequiredBlocks is a set of block number -> hash mappings which must be in the
 	// canonical chain of all remote peers. Setting the option makes geth verify the
@@ -122,12 +117,20 @@ type Config struct {
 	DatabaseFreezer    string
 	DatabaseEra        string
 
+	// Trie database options. TODO(rjl493456442) move all trie database options
+	// into a separate config structure.
 	TrieCleanCache int
 	TrieDirtyCache int
 	TrieTimeout    time.Duration
 	SnapshotCache  int
 	Preimages      bool
-	TrieJournal    string
+	StateHistory   uint64 `toml:",omitempty"` // The maximum number of blocks from head whose state histories are reserved.
+	TrieDBJournal  string // Path to the journal used for persisting trie data across node restarts
+
+	// State scheme represents the scheme used to store ethereum states and trie
+	// nodes on top. It can be 'hash', 'path', or none which means use the scheme
+	// consistent with persistent state.
+	StateScheme string `toml:",omitempty"`
 
 	// This is the number of blocks for which logs will be cached in the filter system.
 	FilterLogCacheSize int

internal/flags/categories.go

@@ -19,24 +19,25 @@ package flags
 import "github.com/urfave/cli/v2"
 
 const (
-	EthCategory        = "ETHEREUM"
+	EthCategory          = "ETHEREUM"
-	BeaconCategory     = "BEACON CHAIN"
+	BeaconCategory       = "BEACON CHAIN"
-	DevCategory        = "DEVELOPER CHAIN"
+	DevCategory          = "DEVELOPER CHAIN"
-	StateCategory      = "STATE HISTORY MANAGEMENT"
+	StateCategory        = "STATE HISTORY MANAGEMENT"
-	TxPoolCategory     = "TRANSACTION POOL (EVM)"
+	TxPoolCategory       = "TRANSACTION POOL (EVM)"
-	BlobPoolCategory   = "TRANSACTION POOL (BLOB)"
+	BlobPoolCategory     = "TRANSACTION POOL (BLOB)"
-	PerfCategory       = "PERFORMANCE TUNING"
+	PerfCategory         = "PERFORMANCE TUNING"
-	AccountCategory    = "ACCOUNT"
+	AccountCategory      = "ACCOUNT"
-	APICategory        = "API AND CONSOLE"
+	APICategory          = "API AND CONSOLE"
-	NetworkingCategory = "NETWORKING"
+	NetworkingCategory   = "NETWORKING"
-	MinerCategory      = "MINER"
+	MinerCategory        = "MINER"
-	GasPriceCategory   = "GAS PRICE ORACLE"
+	TrieDatabaseCategory = "TRIE DATABASE"
-	VMCategory         = "VIRTUAL MACHINE"
+	GasPriceCategory     = "GAS PRICE ORACLE"
-	LoggingCategory    = "LOGGING AND DEBUGGING"
+	VMCategory           = "VIRTUAL MACHINE"
-	MetricsCategory    = "METRICS AND STATS"
+	LoggingCategory      = "LOGGING AND DEBUGGING"
-	MiscCategory       = "MISC"
+	MetricsCategory      = "METRICS AND STATS"
-	TestingCategory    = "TESTING"
+	MiscCategory         = "MISC"
-	DeprecatedCategory = "ALIASED (deprecated)"
+	TestingCategory      = "TESTING"
+	DeprecatedCategory   = "ALIASED (deprecated)"
 )
 
 func init() {

triedb/pathdb/database.go

@@ -120,6 +120,7 @@ type Config struct {
 	StateCleanSize      int    // Maximum memory allowance (in bytes) for caching clean state data
 	WriteBufferSize     int    // Maximum memory allowance (in bytes) for write buffer
 	ReadOnly            bool   // Flag whether the database is opened in read only mode
+	JournalPath         string // Absolute path of journal file (null means the journal data is persisted in key-value store)
 
 	// Testing configurations
 	SnapshotNoBuild   bool // Flag Whether the state generation is allowed
@@ -156,6 +157,9 @@ func (c *Config) fields() []interface{} {
 	} else {
 		list = append(list, "history", fmt.Sprintf("last %d blocks", c.StateHistory))
 	}
+	if c.JournalPath != "" {
+		list = append(list, "journal", c.JournalPath)
+	}
 	return list
 }
 

triedb/pathdb/database_test.go

@@ -21,12 +21,15 @@ import (
 	"errors"
 	"fmt"
 	"math/rand"
+	"os"
+	"path/filepath"
 	"testing"
 
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/core/rawdb"
 	"github.com/ethereum/go-ethereum/core/types"
 	"github.com/ethereum/go-ethereum/crypto"
+	"github.com/ethereum/go-ethereum/ethdb"
 	"github.com/ethereum/go-ethereum/internal/testrand"
 	"github.com/ethereum/go-ethereum/rlp"
 	"github.com/ethereum/go-ethereum/trie"
@@ -121,7 +124,7 @@ type tester struct {
 	snapStorages map[common.Hash]map[common.Hash]map[common.Hash][]byte // Keyed by the hash of account address and the hash of storage key
 }
 
-func newTester(t *testing.T, historyLimit uint64, isVerkle bool, layers int, enableIndex bool) *tester {
+func newTester(t *testing.T, historyLimit uint64, isVerkle bool, layers int, enableIndex bool, journal string) *tester {
 	var (
 		disk, _ = rawdb.Open(rawdb.NewMemoryDatabase(), rawdb.OpenOptions{Ancient: t.TempDir()})
 		db      = New(disk, &Config{
@@ -131,6 +134,7 @@ func newTester(t *testing.T, historyLimit uint64, isVerkle bool, layers int, ena
 			StateCleanSize:      256 * 1024,
 			WriteBufferSize:     256 * 1024,
 			NoAsyncFlush:        true,
+			JournalPath:         journal,
 		}, isVerkle)
 
 		obj = &tester{
@@ -466,7 +470,7 @@ func TestDatabaseRollback(t *testing.T) {
 	}()
 
 	// Verify state histories
-	tester := newTester(t, 0, false, 32, false)
+	tester := newTester(t, 0, false, 32, false, "")
 	defer tester.release()
 
 	if err := tester.verifyHistory(); err != nil {
@@ -500,7 +504,7 @@ func TestDatabaseRecoverable(t *testing.T) {
 	}()
 
 	var (
-		tester = newTester(t, 0, false, 12, false)
+		tester = newTester(t, 0, false, 12, false, "")
 		index  = tester.bottomIndex()
 	)
 	defer tester.release()
@@ -544,7 +548,7 @@ func TestDisable(t *testing.T) {
 		maxDiffLayers = 128
 	}()
 
-	tester := newTester(t, 0, false, 32, false)
+	tester := newTester(t, 0, false, 32, false, "")
 	defer tester.release()
 
 	stored := crypto.Keccak256Hash(rawdb.ReadAccountTrieNode(tester.db.diskdb, nil))
@@ -586,7 +590,7 @@ func TestCommit(t *testing.T) {
 		maxDiffLayers = 128
 	}()
 
-	tester := newTester(t, 0, false, 12, false)
+	tester := newTester(t, 0, false, 12, false, "")
 	defer tester.release()
 
 	if err := tester.db.Commit(tester.lastHash(), false); err != nil {
@@ -610,13 +614,18 @@ func TestCommit(t *testing.T) {
 }
 
 func TestJournal(t *testing.T) {
+	testJournal(t, "")
+	testJournal(t, t.TempDir())
+}
+
+func testJournal(t *testing.T, journal string) {
 	// Redefine the diff layer depth allowance for faster testing.
 	maxDiffLayers = 4
 	defer func() {
 		maxDiffLayers = 128
 	}()
 
-	tester := newTester(t, 0, false, 12, false)
+	tester := newTester(t, 0, false, 12, false, journal)
 	defer tester.release()
 
 	if err := tester.db.Journal(tester.lastHash()); err != nil {
@@ -640,13 +649,30 @@ func TestJournal(t *testing.T) {
 }
 
 func TestCorruptedJournal(t *testing.T) {
+	testCorruptedJournal(t, "", func(db ethdb.Database) {
+		// Mutate the journal in disk, it should be regarded as invalid
+		blob := rawdb.ReadTrieJournal(db)
+		blob[0] = 0xa
+		rawdb.WriteTrieJournal(db, blob)
+	})
+
+	path := filepath.Join(t.TempDir(), "journal")
+	testCorruptedJournal(t, path, func(_ ethdb.Database) {
+		f, _ := os.OpenFile(path, os.O_WRONLY, 0644)
+		f.WriteAt([]byte{0xa}, 0)
+		f.Sync()
+		f.Close()
+	})
+}
+
+func testCorruptedJournal(t *testing.T, journal string, modifyFn func(database ethdb.Database)) {
 	// Redefine the diff layer depth allowance for faster testing.
 	maxDiffLayers = 4
 	defer func() {
 		maxDiffLayers = 128
 	}()
 
-	tester := newTester(t, 0, false, 12, false)
+	tester := newTester(t, 0, false, 12, false, journal)
 	defer tester.release()
 
 	if err := tester.db.Journal(tester.lastHash()); err != nil {
@@ -655,13 +681,10 @@ func TestCorruptedJournal(t *testing.T) {
 	tester.db.Close()
 	root := crypto.Keccak256Hash(rawdb.ReadAccountTrieNode(tester.db.diskdb, nil))
 
-	// Mutate the journal in disk, it should be regarded as invalid
+	modifyFn(tester.db.diskdb)
-	blob := rawdb.ReadTrieJournal(tester.db.diskdb)
-	blob[0] = 0xa
-	rawdb.WriteTrieJournal(tester.db.diskdb, blob)
 
 	// Verify states, all not-yet-written states should be discarded
-	tester.db = New(tester.db.diskdb, nil, false)
+	tester.db = New(tester.db.diskdb, tester.db.config, false)
 	for i := 0; i < len(tester.roots); i++ {
 		if tester.roots[i] == root {
 			if err := tester.verifyState(root); err != nil {
@@ -694,7 +717,7 @@ func TestTailTruncateHistory(t *testing.T) {
 		maxDiffLayers = 128
 	}()
 
-	tester := newTester(t, 10, false, 12, false)
+	tester := newTester(t, 10, false, 12, false, "")
 	defer tester.release()
 
 	tester.db.Close()

triedb/pathdb/fileutils.go

@@ -0,0 +1,54 @@
+// Copyright 2025 The go-ethereum Authors
+// This file is part of the go-ethereum library.
+//
+// The go-ethereum library is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Lesser General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// The go-ethereum library is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU Lesser General Public License for more details.
+//
+// You should have received a copy of the GNU Lesser General Public License
+// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
+
+package pathdb
+
+import (
+	"errors"
+	"os"
+	"syscall"
+)
+
+func isErrInvalid(err error) bool {
+	if errors.Is(err, os.ErrInvalid) {
+		return true
+	}
+	// Go >= 1.8 returns *os.PathError instead
+	if patherr, ok := err.(*os.PathError); ok && patherr.Err == syscall.EINVAL {
+		return true
+	}
+	return false
+}
+
+func syncDir(name string) error {
+	// As per fsync manpage, Linux seems to expect fsync on directory, however
+	// some system don't support this, so we will ignore syscall.EINVAL.
+	//
+	// From fsync(2):
+	//   Calling fsync() does not necessarily ensure that the entry in the
+	//   directory containing the file has also reached disk. For that an
+	//   explicit fsync() on a file descriptor for the directory is also needed.
+	f, err := os.Open(name)
+	if err != nil {
+		return err
+	}
+	defer f.Close()
+
+	if err := f.Sync(); err != nil && !isErrInvalid(err) {
+		return err
+	}
+	return nil
+}

triedb/pathdb/journal.go

@@ -22,6 +22,7 @@ import (
 	"fmt"
 	"io"
 	"os"
+	"path/filepath"
 	"time"
 
 	"github.com/ethereum/go-ethereum/common"
@@ -53,16 +54,17 @@ const journalVersion uint64 = 3
 // loadJournal tries to parse the layer journal from the disk.
 func (db *Database) loadJournal(diskRoot common.Hash) (layer, error) {
 	var reader io.Reader
-	if db.config.Journal != "" && common.FileExist(db.config.Journal) {
+	if db.config.JournalPath != "" && common.FileExist(db.config.JournalPath) {
-		log.Info("Load pathdb disklayer journal", "path", db.config.Journal)
 		// If a journal file is specified, read it from there
-		f, err := os.OpenFile(db.config.Journal, os.O_RDONLY, 0644)
+		log.Info("Load database journal from file", "path", db.config.JournalPath)
+		f, err := os.OpenFile(db.config.JournalPath, os.O_RDONLY, 0644)
 		if err != nil {
-			return nil, fmt.Errorf("failed to read journal file %s: %w", db.config.Journal, err)
+			return nil, fmt.Errorf("failed to read journal file %s: %w", db.config.JournalPath, err)
 		}
 		defer f.Close()
 		reader = f
 	} else {
+		log.Info("Load database journal from database")
 		journal := rawdb.ReadTrieJournal(db.diskdb)
 		if len(journal) == 0 {
 			return nil, errMissJournal
@@ -331,21 +333,24 @@ func (db *Database) Journal(root common.Hash) error {
 		return errDatabaseReadOnly
 	}
 
-	var journal io.Writer
-	var file *os.File
 	// Store the journal into the database and return
-	if db.config.Journal != "" {
+	var (
+		file    *os.File
+		journal io.Writer
+		tmpName = db.config.JournalPath + ".tmp"
+	)
+	if db.config.JournalPath != "" {
 		// Write into a temp file first
 		var err error
-		file, err = os.OpenFile(db.config.Journal+".new", os.O_WRONLY|os.O_CREATE, 0644)
+		file, err = os.OpenFile(tmpName, os.O_WRONLY|os.O_CREATE|os.O_TRUNC, 0644)
 		if err != nil {
-			return fmt.Errorf("failed to open journal file %s: %w", db.config.Journal, err)
+			return fmt.Errorf("failed to open journal file %s: %w", db.config.JournalPath, err)
 		}
 		defer func() {
 			if file != nil {
 				file.Close()
-				// Clean up temp file if we didn't successfully rename it
+				os.Remove(tmpName) // Clean up temp file if we didn't successfully rename it
-				os.Remove(db.config.Journal + ".new")
+				log.Warn("Removed leftover temporary journal file", "path", tmpName)
 			}
 		}()
 		journal = file
@@ -373,21 +378,28 @@ func (db *Database) Journal(root common.Hash) error {
 
 	// Store the journal into the database and return
 	var size int
-	if db.config.Journal == "" {
+	if db.config.JournalPath == "" {
 		data := journal.(*bytes.Buffer)
 		size = data.Len()
 		rawdb.WriteTrieJournal(db.diskdb, data.Bytes())
 	} else {
-		// Close the temporary file and atomically rename it
-		if err := file.Close(); err != nil {
-			return fmt.Errorf("failed to close temporary journal file: %w", err)
-		}
-		// Replace the live journal with the newly generated one
-		if err := os.Rename(db.config.Journal+".new", db.config.Journal); err != nil {
-			return fmt.Errorf("failed to rename temporary journal file: %w", err)
-		}
 		stat, _ := journal.(*os.File).Stat()
 		size = int(stat.Size())
+
+		// Close the temporary file and atomically rename it
+		if err := file.Sync(); err != nil {
+			return fmt.Errorf("failed to fsync the journal, %v", err)
+		}
+		if err := file.Close(); err != nil {
+			return fmt.Errorf("failed to close the journal: %v", err)
+		}
+		// Replace the live journal with the newly generated one
+		if err := os.Rename(tmpName, db.config.JournalPath); err != nil {
+			return fmt.Errorf("failed to rename the journal: %v", err)
+		}
+		if err := syncDir(filepath.Dir(db.config.JournalPath)); err != nil {
+			return fmt.Errorf("failed to fsync the dir: %v", err)
+		}
 		file = nil
 	}