Skip to content

Optimize snapshot unpack by using multiple accounts directories #245

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
palmerlao wants to merge 4 commits into
base: dev
Choose a base branch
from
Open

Optimize snapshot unpack by using multiple accounts directories #245

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
c57a5a5
Store accountsdb append-vecs as per-disk big files
palmerlao Jul 2, 2026
03211ac
Rewrite the snapshot index flush as a read-sort-write pipeline
palmerlao Jul 3, 2026
da241cd
Add a flush-sort-workers flag and tuning setting
palmerlao Jul 3, 2026
46157e1
Add a flush correctness test
palmerlao Jul 3, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
64 changes: 39 additions & 25 deletions cmd/mithril/node/node.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -59,7 +59,8 @@ var (
bootstrapMode string // "auto", "snapshot", or "accountsdb"
snapshotArchivePath string
incrementalSnapshotFilename string
accountsPath string
accountsPath string // primary accounts/metadata dir (accountsPaths[0])
accountsPaths []string // one dir per disk shard
scratchDirectory string
rpcEndpoints []string
cluster string // "mainnet-beta", "testnet", "devnet"
Expand Down Expand Up @@ -208,7 +209,7 @@ func init() {
Run.Flags().StringVar(&incrementalSnapshotFilename, "incremental-snapshot", "", "Path to specific incremental snapshot file (bypasses auto-discovery)")

// [ledger] section flags
Run.Flags().StringVarP(&accountsPath, "accounts-path", "o", "", "Output path for writing AccountsDB data to")
Run.Flags().StringSliceVarP(&accountsPaths, "accounts-path", "o", []string{}, "Output path(s) for writing AccountsDB data to - one dir per disk shard, can specify multiple")
Run.Flags().StringVar(&blockstorePath, "ledger-path", "/tmp/blocks", "Path containing slot.json files")

// [network] section flags
Expand All @@ -228,6 +229,8 @@ func init() {
Run.Flags().Uint64Var(&borrowedAccountArenaSize, "borrowed-account-arena-size", 1024, "Number of borrowed accounts to preallocate in arena (0 to disable)")
Run.Flags().IntVar(&snapshot.ZstdDecoderConcurrency, "zstd-decoder-concurrency", runtime.NumCPU(), "Zstd decoder concurrency")
Run.Flags().IntVar(&snapshot.MaxConcurrentFlushers, "max-concurrent-flushers", snapshot.DefaultSnapshotMaxConcurrentFlushers, "Bound for number of log shards to flush to Accounts DB Index at once")
Run.Flags().IntVar(&snapshot.SnapshotFlushSortWorkers, "flush-sort-workers", snapshot.DefaultSnapshotFlushSortWorkers, "Snapshot index-flush concurrent sort workers (0 = one per CPU)")
Run.Flags().BoolVar(&snapshot.SnapshotDirectIO, "snapshot-directio", snapshot.DefaultSnapshotDirectIO, "Write snapshot big files with O_DIRECT (bypasses the page cache; helps decode throughput on low-RAM boxes)")
Run.Flags().IntVar(&snapshot.SnapshotAppendVecCopyingWorkers, "snapshot-append-vec-workers", snapshot.DefaultSnapshotAppendVecCopyingWorkers, "Snapshot bootstrap appendvec write workers")
Run.Flags().IntVar(&snapshot.SnapshotIndexEntryBuilderWorkers, "snapshot-index-builder-workers", snapshot.DefaultSnapshotIndexEntryBuilderWorkers, "Snapshot bootstrap account-index parser workers")
Run.Flags().IntVar(&snapshot.SnapshotIndexEntryCommitterWorkers, "snapshot-index-committer-workers", snapshot.DefaultSnapshotIndexEntryCommitterWorkers, "Snapshot bootstrap account-index shard enqueue workers")
Expand Down Expand Up @@ -445,13 +448,22 @@ func initConfigAndBindFlags(cmd *cobra.Command) error {
if incrementalSnapshotFilename == "" {
incrementalSnapshotFilename = getString("incremental-snapshot", "ledger.incremental_snapshot")
}
accountsPath = getString("accounts-path", "storage.accounts")
if accountsPath == "" {
accountsPath = getString("accounts-path", "ledger.accounts_path")
// storage.accounts is a list of one dir per disk shard; the single-disk case
// is just a slice of one. A scalar string in TOML is accepted too (viper casts
// it to a 1-element slice), and the --accounts-path CLI flag takes 1+ dirs.
accountsPaths = getStringSlice("accounts-path", "storage.accounts")
if len(accountsPaths) == 0 {
accountsPaths = getStringSlice("accounts-path", "ledger.accounts_path")
}
if len(accountsPaths) == 0 {
return fmt.Errorf("no accounts path configured (set storage.accounts)")
}
accountsPath = accountsPaths[0]
// Check write permission early to fail fast with helpful error
if err := checkDirWritable(accountsPath, "AccountsDB"); err != nil {
return err
for _, p := range accountsPaths {
if err := checkDirWritable(p, "AccountsDB"); err != nil {
return err
}
}
blockstorePath = getString("ledger-path", "storage.shredstore")
if blockstorePath == "" && config.IsSet("storage.blockstore") {
Expand Down Expand Up @@ -672,6 +684,8 @@ func initConfigAndBindFlags(cmd *cobra.Command) error {

snapshot.ZstdDecoderConcurrency = getInt("zstd-decoder-concurrency", "tuning.zstd_decoder_concurrency")
snapshot.MaxConcurrentFlushers = getInt("max-concurrent-flushers", "tuning.max_concurrent_flushers")
snapshot.SnapshotFlushSortWorkers = getInt("flush-sort-workers", "tuning.flush_sort_workers")
snapshot.SnapshotDirectIO = getBool("snapshot-directio", "tuning.snapshot_directio")
snapshot.SnapshotAppendVecCopyingWorkers = getInt("snapshot-append-vec-workers", "tuning.snapshot_append_vec_workers")
snapshot.SnapshotIndexEntryBuilderWorkers = getInt("snapshot-index-builder-workers", "tuning.snapshot_index_builder_workers")
snapshot.SnapshotIndexEntryCommitterWorkers = getInt("snapshot-index-committer-workers", "tuning.snapshot_index_committer_workers")
Expand Down Expand Up @@ -1040,7 +1054,7 @@ func runLive(c *cobra.Command, args []string) {
// Build directly from the specified files (BuildAccountsDbPaths handles AccountsDB cleanup internally)
// NOTE: We do NOT clean snapshot files in explicit mode - user wants to keep their explicit snapshots
dp := progress.NewDualProgress()
accountsDb, manifest, err = snapshot.BuildAccountsDbPaths(ctx, snapshotArchivePath, incrementalSnapshotFilename, accountsPath, dp)
accountsDb, manifest, err = snapshot.BuildAccountsDbPaths(ctx, snapshotArchivePath, incrementalSnapshotFilename, accountsPaths, dp)
if err != nil {
klog.Fatalf("failed to build AccountsDB from snapshot: %v", err)
}
Expand All @@ -1067,7 +1081,7 @@ func runLive(c *cobra.Command, args []string) {
mlog.Log.Infof("WARNING: no state file found, AccountsDB may be from incomplete build")
}
mlog.Log.Infof("Resuming from existing AccountsDB at slot %d", accountsDBSlot)
accountsDb, err = accountsdb.OpenDb(accountsPath)
accountsDb, err = accountsdb.OpenDb(accountsPaths)
if err != nil {
klog.Fatalf("failed to open AccountsDB at %s: %v", accountsPath, err)
}
Expand Down Expand Up @@ -1098,7 +1112,7 @@ func runLive(c *cobra.Command, args []string) {
state.RecordRebuild(accountsPath, 0, "", getVersion(), getCommit(), getBranch(), "new-snapshot mode (no prior state)")
}
mlog.Log.Infof("Cleaning up previous AccountsDB artifacts in %s", accountsPath)
snapshot.CleanAccountsDbDir(accountsPath)
snapshot.CleanAccountsDbDir(accountsPaths)
}
// Clean existing snapshots (respecting retention setting)
if snapshotDownloadPath != "" {
Expand All @@ -1109,7 +1123,7 @@ func runLive(c *cobra.Command, args []string) {
mlog.Log.Infof("Cleaning up existing snapshot files in %s (keeping %d)", snapshotDownloadPath, maxSnapshots)
snapshot.CleanSnapshotDownloadDir(snapshotDownloadPath, maxSnapshots)
}
accountsDb, manifest, err = downloadAndBuildFromSnapshot(ctx, rpcEndpoints, snapshotDownloadPath, accountsPath, blockstorePath)
accountsDb, manifest, err = downloadAndBuildFromSnapshot(ctx, rpcEndpoints, snapshotDownloadPath, accountsPaths, blockstorePath)
if err != nil {
klog.Fatalf("failed to build AccountsDB from snapshot: %v", err)
}
Expand Down Expand Up @@ -1138,7 +1152,7 @@ func runLive(c *cobra.Command, args []string) {
state.RecordRebuild(accountsPath, 0, "", getVersion(), getCommit(), getBranch(), "snapshot mode (no prior state)")
}
mlog.Log.Infof("Cleaning up previous AccountsDB artifacts in %s", accountsPath)
snapshot.CleanAccountsDbDir(accountsPath)
snapshot.CleanAccountsDbDir(accountsPaths)
}

// Check for existing fresh snapshot
Expand All @@ -1151,7 +1165,7 @@ func runLive(c *cobra.Command, args []string) {
if existingSnap != nil {
// Reuse existing snapshot
mlog.Log.Infof("Reusing existing snapshot file at slot %d", existingSnap.slot)
accountsDb, manifest, err = buildFromExistingSnapshot(ctx, existingSnap, snapshotDownloadPath, accountsPath, blockstorePath, rpcEndpoints)
accountsDb, manifest, err = buildFromExistingSnapshot(ctx, existingSnap, snapshotDownloadPath, accountsPaths, blockstorePath, rpcEndpoints)
} else {
// Download fresh
mlog.Log.Infof("no fresh snapshot file found, downloading new one")
Expand All @@ -1163,7 +1177,7 @@ func runLive(c *cobra.Command, args []string) {
}
snapshot.CleanSnapshotDownloadDir(snapshotDownloadPath, maxSnapshots)
}
accountsDb, manifest, err = downloadAndBuildFromSnapshot(ctx, rpcEndpoints, snapshotDownloadPath, accountsPath, blockstorePath)
accountsDb, manifest, err = downloadAndBuildFromSnapshot(ctx, rpcEndpoints, snapshotDownloadPath, accountsPaths, blockstorePath)
}
if err != nil {
klog.Fatalf("failed to build AccountsDB from snapshot: %v", err)
Expand Down Expand Up @@ -1217,13 +1231,13 @@ func runLive(c *cobra.Command, args []string) {
// mithrilState is guaranteed non-nil here (we prompted because it was stale)
state.RecordRebuild(accountsPath, mithrilState.LastSlot, mithrilState.LastBankhash, getVersion(), getCommit(), getBranch(), "user chose rebuild (stale AccountsDB)")
mlog.Log.Infof("Cleaning up previous AccountsDB artifacts in %s", accountsPath)
snapshot.CleanAccountsDbDir(accountsPath)
snapshot.CleanAccountsDbDir(accountsPaths)
}
// Check for existing fresh snapshot
existingSnap := detectFreshSnapshot(snapshotDownloadPath, fullThreshold, rpcEndpoints, ctx)
if existingSnap != nil {
mlog.Log.Infof("Reusing existing snapshot file at slot %d", existingSnap.slot)
accountsDb, manifest, err = buildFromExistingSnapshot(ctx, existingSnap, snapshotDownloadPath, accountsPath, blockstorePath, rpcEndpoints)
accountsDb, manifest, err = buildFromExistingSnapshot(ctx, existingSnap, snapshotDownloadPath, accountsPaths, blockstorePath, rpcEndpoints)
} else {
// Clean up old snapshot files
if snapshotDownloadPath != "" {
Expand All @@ -1233,7 +1247,7 @@ func runLive(c *cobra.Command, args []string) {
}
snapshot.CleanSnapshotDownloadDir(snapshotDownloadPath, maxSnapshots)
}
accountsDb, manifest, err = downloadAndBuildFromSnapshot(ctx, rpcEndpoints, snapshotDownloadPath, accountsPath, blockstorePath)
accountsDb, manifest, err = downloadAndBuildFromSnapshot(ctx, rpcEndpoints, snapshotDownloadPath, accountsPaths, blockstorePath)
}
if err != nil {
klog.Fatalf("failed to build AccountsDB from snapshot: %v", err)
Expand All @@ -1255,7 +1269,7 @@ func runLive(c *cobra.Command, args []string) {
mlog.Log.Infof("mode=auto: Resuming from existing AccountsDB at slot %d", accountsDBSlot)
// Record resume in history
state.RecordResume(accountsPath, mithrilState.LastSlot, mithrilState.LastBankhash, replay.CurrentRunID, getVersion(), getCommit(), getBranch())
accountsDb, err = accountsdb.OpenDb(accountsPath)
accountsDb, err = accountsdb.OpenDb(accountsPaths)
if err != nil {
klog.Fatalf("failed to open AccountsDB at %s: %v", accountsPath, err)
}
Expand Down Expand Up @@ -1311,22 +1325,22 @@ func runLive(c *cobra.Command, args []string) {
state.RecordRebuild(accountsPath, 0, "", getVersion(), getCommit(), getBranch(), reason)
}
mlog.Log.Infof("Cleaning up previous AccountsDB artifacts in %s", accountsPath)
snapshot.CleanAccountsDbDir(accountsPath)
snapshot.CleanAccountsDbDir(accountsPaths)
}

// Check for existing fresh snapshot
existingSnap := detectFreshSnapshot(snapshotDownloadPath, fullThreshold, rpcEndpoints, ctx)
if existingSnap != nil {
mlog.Log.Infof("Reusing existing snapshot file at slot %d", existingSnap.slot)
accountsDb, manifest, err = buildFromExistingSnapshot(ctx, existingSnap, snapshotDownloadPath, accountsPath, blockstorePath, rpcEndpoints)
accountsDb, manifest, err = buildFromExistingSnapshot(ctx, existingSnap, snapshotDownloadPath, accountsPaths, blockstorePath, rpcEndpoints)
} else {
// Clean up old snapshot files based on retention settings
maxSnapshots := config.GetInt("snapshot.max_full_snapshots")
if maxSnapshots == 0 {
maxSnapshots = 1 // default: keep 1 snapshot
}
snapshot.CleanSnapshotDownloadDir(snapshotDownloadPath, maxSnapshots)
accountsDb, manifest, err = downloadAndBuildFromSnapshot(ctx, rpcEndpoints, snapshotDownloadPath, accountsPath, blockstorePath)
accountsDb, manifest, err = downloadAndBuildFromSnapshot(ctx, rpcEndpoints, snapshotDownloadPath, accountsPaths, blockstorePath)
}
if err != nil {
klog.Fatalf("failed to build AccountsDB from snapshot: %v", err)
Expand Down Expand Up @@ -2222,7 +2236,7 @@ func queryLatestSnapshotSlot(ctx context.Context, rpcEndpoints []string) (uint64
}

// buildFromExistingSnapshot builds AccountsDB from an existing downloaded snapshot file.
func buildFromExistingSnapshot(ctx context.Context, snap *snapshotInfo, snapshotDir, accountsPath, blockstorePath string, rpcEndpoints []string) (*accountsdb.AccountsDb, *snapshot.SnapshotManifest, error) {
func buildFromExistingSnapshot(ctx context.Context, snap *snapshotInfo, snapshotDir string, accountsPaths []string, blockstorePath string, rpcEndpoints []string) (*accountsdb.AccountsDb, *snapshot.SnapshotManifest, error) {
snapCfg := buildSnapshotConfig(rpcEndpoints)

// Construct full path to snapshot file
Expand All @@ -2232,7 +2246,7 @@ func buildFromExistingSnapshot(ctx context.Context, snap *snapshotInfo, snapshot
// Create progress display for extract
dp := progress.NewDualProgress()

accountsDb, manifest, err := snapshot.BuildAccountsDbAuto(ctx, fullSnapshotPath, snapshotDir, int(snap.slot), int(snap.slot), accountsPath, rpcEndpoints, blockstorePath, snapCfg, dp)
accountsDb, manifest, err := snapshot.BuildAccountsDbAuto(ctx, fullSnapshotPath, snapshotDir, int(snap.slot), int(snap.slot), accountsPaths, rpcEndpoints, blockstorePath, snapCfg, dp)
if err != nil {
return nil, nil, fmt.Errorf("failed to build AccountsDB from snapshot: %w", err)
}
Expand All @@ -2242,7 +2256,7 @@ func buildFromExistingSnapshot(ctx context.Context, snap *snapshotInfo, snapshot
}

// downloadAndBuildFromSnapshot finds, downloads, and builds AccountsDB from a snapshot
func downloadAndBuildFromSnapshot(ctx context.Context, rpcEndpoints []string, snapshotDownloadPath, accountsPath, blockstorePath string) (*accountsdb.AccountsDb, *snapshot.SnapshotManifest, error) {
func downloadAndBuildFromSnapshot(ctx context.Context, rpcEndpoints []string, snapshotDownloadPath string, accountsPaths []string, blockstorePath string) (*accountsdb.AccountsDb, *snapshot.SnapshotManifest, error) {
snapCfg := buildSnapshotConfig(rpcEndpoints)
fullSnapshotDlStart := time.Now()
fullSnapshotInfo, err := snapshotdl.GetSnapshotURLWithInfo(ctx, snapCfg)
Expand All @@ -2266,7 +2280,7 @@ func downloadAndBuildFromSnapshot(ctx context.Context, rpcEndpoints []string, sn
// Create progress display for snapshot download and extract
dp := progress.NewDualProgress()

accountsDb, manifest, err := snapshot.BuildAccountsDbAuto(ctx, fullSnapshotURL, snapshotDownloadPath, fullSnapshotSlot, fullSnapshotSlot, accountsPath, rpcEndpoints, blockstorePath, snapCfg, dp)
accountsDb, manifest, err := snapshot.BuildAccountsDbAuto(ctx, fullSnapshotURL, snapshotDownloadPath, fullSnapshotSlot, fullSnapshotSlot, accountsPaths, rpcEndpoints, blockstorePath, snapCfg, dp)
if err != nil {
return nil, nil, fmt.Errorf("failed to build AccountsDB from snapshot: %w", err)
}
Expand Down
24 changes: 24 additions & 0 deletions config.example.toml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -71,6 +71,18 @@ name = "mithril"
# Put this on your fastest NVMe due to heavy random I/O.
accounts = "/mnt/mithril-accounts"

# For multi-disk setups, accounts may instead be a list of directories, one
# per physical disk. The snapshot unpack shards append-vecs across them so
# aggregate write throughput approaches the sum of the disks. The first entry
# also holds the shared metadata (account index, manifest, state); the others
# hold only their disk's shard of the account data. The number of directories
# is fixed at build time - changing it requires rebuilding the AccountsDB.
# accounts = [
# "/mnt/nvme0/mithril-accounts", # primary: metadata + shard 0
# "/mnt/nvme1/mithril-accounts", # shard 1
# "/mnt/nvme2/mithril-accounts", # shard 2
# ]

# Shredstore - Lightbringer stores received shreds here
# Used for block streaming and potential repair serving.
shredstore = "/mnt/mithril-ledger/shredstore"
Expand Down Expand Up @@ -362,6 +374,18 @@ name = "mithril"
# Bound for number of index shards to convert to SSTs at once.
max_concurrent_flushers = 8

# Concurrent sort workers used during the index flush.
# 0 (default) = one per CPU.
# flush_sort_workers = 0

# Write the snapshot append-vec big files with O_DIRECT, bypassing the OS page
# cache. On a ~460GB unpack the buffered path churns hundreds of GB of dirty
# pages, which evicts the zstd-decode working set; O_DIRECT avoids that and
# gives a modest (~6%) unpack speedup that grows on low-RAM boxes. Off by
# default (buffered is the safe choice). Requires a filesystem that supports
# O_DIRECT (ext4/xfs; not tmpfs).
# snapshot_directio = false

# Size in MB for serialized parameter arena (0 to disable)
param_arena_size_mb = 512

Expand Down
Loading
Loading