From b36dc0b54b03c5b392e2da80821bddb3392422e8 Mon Sep 17 00:00:00 2001 From: Marcus Pasell <3690498+rickyrombo@users.noreply.github.com> Date: Thu, 8 Oct 2026 18:15:31 -0700 Subject: [PATCH 1/2] feat(genesis-writer): emit a portable chain dump and manifest After a successful write, pg_dump the chain database (directory format, parallel) to /chain.dump and write /MANIFEST.json, so the artifact is self-contained instead of a live database operators must seal and export by hand. The dump runs before the writer exits, so nothing else has connected first. --no-dump skips it; the manifest is still written. The manifest records chain_id, end/first-live height, the source snapshot's last indexed old-chain block and new_chain_flush_from_block (= that + 1), genesis sha256 and validator address, pg_dump/server versions, the writer commit, and the per-node identity files not to seed onto a bootstrap node. Pin Postgres versions to the node's bundled major (15) via --target-pg-major: the managed cluster runs exactly that major, and a --dst-dsn server newer than it is refused before writing unless --allow-newer-postgres is given. A pg_dump in range is resolved up front. Run cmd/genesis-writer unit tests in make test-unit, and update ROLLOUT.md (steps 1, 4, 7, 8, 11; Appendices B, C, E) and README for the new flow. Co-Authored-By: Claude Opus 5.5 --- cmd/genesis-writer/README.md | 22 +- cmd/genesis-writer/ROLLOUT.md | 160 ++++++++--- cmd/genesis-writer/artifact.go | 401 ++++++++++++++++++++++++++++ cmd/genesis-writer/artifact_test.go | 247 +++++++++++++++++ cmd/genesis-writer/main.go | 60 ++++- cmd/genesis-writer/pgversion.go | 222 +++++++++++++++ cmd/genesis-writer/postgres.go | 58 ++-- cmd/genesis-writer/writer.go | 54 +++- dev/docker-compose.yml | 1 + 9 files changed, 1160 insertions(+), 65 deletions(-) create mode 100644 cmd/genesis-writer/artifact.go create mode 100644 cmd/genesis-writer/artifact_test.go create mode 100644 cmd/genesis-writer/pgversion.go diff --git a/cmd/genesis-writer/README.md b/cmd/genesis-writer/README.md index 3b205b40..57f03607 100644 --- a/cmd/genesis-writer/README.md +++ b/cmd/genesis-writer/README.md @@ -58,6 +58,9 @@ This will: - Write all entities as synthetic blocks to `/core//` - Write CometBFT `state.db` and `blockstore.db` - Update `genesis.json` with the migration address and end height +- `pg_dump` the chain database to `/chain.dump` and write + `/MANIFEST.json` — the portable artifact to ship (see + [ROLLOUT.md Appendix B](ROLLOUT.md#b-the-artifact-chaindump-and-manifestjson)) - Print next-steps instructions for running a node ## Usage @@ -77,6 +80,8 @@ genesis-writer write \ [--batch-size 1000] \ # default: 1000 [--run-migrations] \ # auto-enabled for managed postgres [--resume] \ # resume interrupted run + [--no-dump] \ # skip chain.dump (MANIFEST.json still written) + [--target-pg-major 15] \ # node's postgres major; pins managed pg + pg_dump [--skip-users] [--skip-wallets] [--skip-tracks] [--skip-playlists] \ [--skip-social] [--skip-plays] [--skip-apps] [--skip-comments] \ [--skip-emails] @@ -99,6 +104,11 @@ genesis-writer write \ | `--batch-size` | `GENESIS_BATCH_SIZE` | `1000` | Rows fetched from source DB per query | | `--run-migrations` | — | false | Apply the Core chain schema before writing (auto-enabled for managed postgres) | | `--resume` | — | false | Resume from the last completed step of a previous run | +| `--dump` / `--no-dump` | `GENESIS_DUMP` | true | `pg_dump --format=directory` the chain database to `/chain.dump` after a successful write. Requires `--data-dir` | +| `--dump-jobs` | `GENESIS_DUMP_JOBS` | half the CPUs, ≤ 8 | `pg_dump --jobs` | +| `--target-pg-major` | `GENESIS_TARGET_PG_MAJOR` | `15` | Postgres major of the node that restores the dump (the node image bundles 15). The managed postgres runs exactly this major; a newer `--dst-dsn` server is refused | +| `--allow-newer-postgres` | — | false | Proceed with a `--dst-dsn` server or `pg_dump` newer than `--target-pg-major`. The output then cannot be restored on a node as-is | +| `--source-chain-id` | `GENESIS_SOURCE_CHAIN_ID` | `audius-mainnet-alpha-beta` | Old chain whose `core_indexed_blocks` in the source snapshot give `source_last_indexed_block` in the manifest | | `--skip-users` | `GENESIS_SKIP_USERS` | false | Skip user migration | | `--skip-wallets` | `GENESIS_SKIP_WALLETS` | false | Skip associated wallets and dashboard wallet users | | `--skip-tracks` | `GENESIS_SKIP_TRACKS` | false | Skip track migration | @@ -125,14 +135,24 @@ layout as a production node: │ ├── blockstore.db/ │ ├── state.db/ │ └── priv_validator_state.json +├── chain.dump/ # pg_dump --format=directory of the chain database +├── MANIFEST.json # heights, hashes, versions, files not to seed └── postgres/ └── ``` +`config/priv_validator_key.json` and `data/priv_validator_state.json` (and +`node_key.json` / `addrbook.json`, if a node later runs against the directory) +are per-node identity. `MANIFEST.json` lists them under +`exclude_from_bootstrap`; do not seed them onto another node. + ### Managed postgres When `--dst-dsn` is omitted, the genesis-writer starts its own PostgreSQL -instance at `/postgres/` using the system's `pg_ctl`. It: +instance at `/postgres/` using the `pg_ctl` of exactly +`--target-pg-major` (default 15, the node's bundled version — e.g. Homebrew +`postgresql@15`). A newer major is never picked up, because its dump could not +be restored on a node. It: - Initializes a new cluster if none exists (`initdb`) - Starts an existing cluster if stopped diff --git a/cmd/genesis-writer/ROLLOUT.md b/cmd/genesis-writer/ROLLOUT.md index c5c4bd4d..10a86328 100644 --- a/cmd/genesis-writer/ROLLOUT.md +++ b/cmd/genesis-writer/ROLLOUT.md @@ -82,7 +82,7 @@ cleanup the migration unblocks, and must wait for step 15. | | | |---|---| | [A. Source snapshot and writer inputs](#a-source-snapshot-and-writer-inputs) | step 1 | -| [B. Sealing the artifact](#b-sealing-the-artifact) | step 1 | +| [B. The artifact: `chain.dump` and `MANIFEST.json`](#b-the-artifact-chaindump-and-manifestjson) | steps 1, 4 | | [C. Verifying: replay, parity, calibration](#c-verifying-replay-parity-calibration) | step 1 | | [D. Expected parity divergences](#d-expected-parity-divergences) | step 1 | | [E. Node identity and the peer lists](#e-node-identity-and-the-peer-lists) | steps 3, 4 | @@ -114,9 +114,12 @@ cleanup the migration unblocks, and must wait for step 15. 1. Restore a production snapshot → [Appendix A](#a-source-snapshot-and-writer-inputs). 2. Run `genesis-writer` with `--priv-validator-key-file` pointing at the bootstrap validator key. Inputs and preconditions → [Appendix A](#a-source-snapshot-and-writer-inputs). -3. **Seal the artifact before anything connects to it** → [Appendix B](#b-sealing-the-artifact). -4. Verify: replay into a scratch ETL database, then run parity against the - source snapshot → [Appendix C](#c-verifying-replay-parity-calibration). +3. The run ends by writing `/chain.dump` and `/MANIFEST.json`. + **Those are the artifact** — serve and ship from them, never from the + writer's live database → [Appendix B](#b-the-artifact-chaindump-and-manifestjson). +4. Verify: `pg_restore` the dump into a scratch database, serve it from a node, + replay into a scratch ETL database, then run parity against the source + snapshot → [Appendix C](#c-verifying-replay-parity-calibration). 5. Judge the parity output against the known divergences → [Appendix D](#d-expected-parity-divergences). ### 2. Register the second state-sync RPC @@ -161,16 +164,43 @@ so nothing else needs flipping. Coverage of the seeding is verified in Same delegate key as `audius.rickyrombo.com`, different chain directory and a fresh database. -1. Preseed from the writer output: the core files and the database. - **Exclude `node_key.json`** → [Appendix E](#e-node-identity-and-the-peer-lists). -2. Point its storage at the existing node's, so both hold all blobs. -3. Environment: +1. Preseed the core files: copy `core//` from the writer output, + **minus every path in `MANIFEST.json`'s `exclude_from_bootstrap`** — + `node_key.json`, `priv_validator_key.json`, `addrbook.json`, + `priv_validator_state.json` → [Appendix E](#e-node-identity-and-the-peer-lists). + Check `sha256sum config/genesis.json` against the manifest's `genesis_sha256`. +2. Preseed the database, **before the node's first start**: + 1. Copy `chain.dump/` to the host and create a new database on the node's + Postgres. Restore with the node's bundled `pg_restore` (Postgres 15); + the manifest's `dump.restore_min_pg_major` is the oldest that reads it. + + ``` + createdb -U postgres openaudio_beta + pg_restore -U postgres -d openaudio_beta --no-owner --no-privileges \ + --jobs 8 --exit-on-error chain.dump + ``` + 2. Restore the mediorum tables from the old node's database into the same + database: `uploads`, `audio_previews`, `qm_audio_analyses`, + `blob_presence`, `blob_presence_state`. The chain dump holds none of + them, and without them the node regenerates previews and analyses and + re-walks the bucket. + + ``` + pg_dump -U postgres -d -Fc -f mediorum.dump \ + -t uploads -t audio_previews -t qm_audio_analyses \ + -t blob_presence -t blob_presence_state + pg_restore -U postgres -d openaudio_beta --no-owner --no-privileges mediorum.dump + ``` + 3. Point the node at it: `POSTGRES_DB=openaudio_beta` (or a full + `OPENAUDIO_DB_URL`). +3. Point its storage at the existing node's, so both hold all blobs. +4. Environment: - `OPENAUDIO_ARCHIVE=true` - `OPENAUDIO_STATE_SYNC_SERVE_SNAPSHOTS=true` - `OPENAUDIO_STATE_SYNC_ENABLE=false` — it is the source, not a consumer - `OPENAUDIO_PERSISTENT_PEERS=` — empty; the second node does not exist yet - `BlockInterval=20000`, `Keep=2` → [Appendix F](#f-snapshot-interval-and-retention) -4. Confirm ≥80 GiB free, or snapshots silently skip. +5. Confirm ≥80 GiB free, or snapshots silently skip. ### 5. Route plays through old-chain hosts @@ -188,17 +218,17 @@ Why this is necessary → [Appendix G](#g-why-plays-need-routing). ### 7. Stand up the second snapshot RPC `v.audius.rickyrombo.com`, same artifacts and environment as step 4, except its -own delegate wallet — and exclude `node_key.json`, `priv_validator_key.json`, -and `priv_validator_state.json`. +own delegate wallet. The same `exclude_from_bootstrap` paths stay behind; skip +the mediorum restore, since storage is off here. 1. Disable storage: no mediorum churn, and storage proofs are irrelevant for it. 2. Confirm ≥80 GiB free. ### 8. Confirm blocks -The node should produce the block after the writer's end height. Take that -height from the writer's completion output or the run's `BUILT_FROM.txt` — **do -not hardcode it here**, every run ends somewhere different. +The node should produce the block after the writer's end height — +`first_live_height` in the run's `MANIFEST.json` (`end_height` + 1). **Do not +hardcode it here**, every run ends somewhere different. If it does not, the validator key does not match genesis. That is the dead-chain failure mode and it looks like "waiting for comet to catch up" at height 0. @@ -254,8 +284,12 @@ Requires [api#1018](https://github.com/AudiusProject/api/pull/1018) merged first 1. `newChainFlushEnabled=true` 2. `NewChainURL` → bootstrap -3. `NewChainFlushFromBlock` → the last block of the snapshot the genesis was - built from +3. `NewChainFlushFromBlock` → `new_chain_flush_from_block` in `MANIFEST.json`: + the source snapshot's last indexed old-chain block **plus one** + (`source_last_indexed_block` + 1). The flusher deletes rows with + `confirmed_block < NewChainFlushFromBlock` and replays the rest, so setting + it to the snapshot's own last block replays that block onto the new chain a + second time. Confirm the flusher keeps up with the relay's write rate before continuing; it is serial, and step 12 depends on it draining. @@ -376,16 +410,64 @@ The validator set is hashed into every block header, so the key cannot be swappe afterwards — a different key means rebuilding the chain. Whoever runs the writer must hold the bootstrap's delegate key at write time. -## B. Sealing the artifact +## B. The artifact: `chain.dump` and `MANIFEST.json` -**Seal the artifact before pointing anything at it.** The verification node, -the ETL, and every other service take the chain DB as `OPENAUDIO_DB_URL`, and -any of them can destroy it. A node whose binary does not embed *this* -artifact's genesis does not fail -- it takes its "generate a new genesis" path -and runs the core migrations **down**, dropping every table the writer just -filled. `OPENAUDIO_ENV=dev` loads the embedded `dev.json`, so a binary built -before the genesis was copied in looks completely normal until it wipes the -write. This has happened once already, on 2026-08-25. +A successful run leaves: + +``` +/ + core// CometBFT home: genesis, blockstore, state + chain.dump/ pg_dump --format=directory of the chain database + MANIFEST.json chain_id, end_height, first_live_height, + source_last_indexed_block, new_chain_flush_from_block, + genesis_sha256, genesis_validator_address, + dump.{pg_dump_version, server_major, restore_min_pg_major}, + writer_commit, exclude_from_bootstrap + postgres/ the writer's managed cluster (only without --dst-dsn) +``` + +`MANIFEST.json` replaces the hand-written `BUILT_FROM.txt`: read heights, hashes +and the flush-from block from it instead of from log lines. + +**Why a dump, not the live database.** The verification node, the ETL, and every +other service take the chain DB as `OPENAUDIO_DB_URL`, and any of them can +destroy it. A node whose binary does not embed *this* artifact's genesis does +not fail -- it takes its "generate a new genesis" path and runs the core +migrations **down**, dropping every table the writer just filled. +`OPENAUDIO_ENV=dev` loads the embedded `dev.json`, so a binary built before the +genesis was copied in looks completely normal until it wipes the write. This +happened on 2026-08-25 and cost a 3h36m write. The writer now dumps before it +exits, so the dump predates anything else connecting, and every consumer +restores its own copy: + +``` +createdb +pg_restore -d --no-owner --no-privileges --jobs 8 --exit-on-error /chain.dump +``` + +A wiped copy costs a re-restore, not a re-run. `--no-dump` skips the dump; the +manifest is still written with `"dump": null`. + +**Version pinning.** `pg_restore` cannot read an archive from a newer +`pg_dump`, and `pg_dump` cannot dump a newer server, so server ≤ `pg_dump` ≤ +node. The node image bundles Postgres 15. `--target-pg-major` (default 15) +therefore: + +- pins the managed cluster to exactly that major, rather than whatever is + newest on the machine (a Homebrew `postgresql@15` on macOS); +- refuses a `--dst-dsn` server newer than it before writing anything + (`--allow-newer-postgres` overrides, loudly, for output you will move some + other way); +- picks a `pg_dump` in range, and fails up front if there is none. + +The 2026-08-25 artifact predates this: it sits in a PG17 cluster on macOS, so +no PG15 tool can dump it and no PG15 `pg_restore` can read a PG17 dump. Moving it +took a manual `pg_dump` → `rsync` → `pg_restore`; that is still the procedure +for that artifact. + +**Sealing is now optional.** If you serve from the writer's live database +anyway — a `--no-dump` run, or the 2026-08-25 artifact — seal it first so +nothing can connect to the original: ```sql ALTER DATABASE WITH ALLOW_CONNECTIONS false; @@ -393,16 +475,17 @@ CREATE DATABASE _serve TEMPLATE ; ``` A sealed database still works as a `TEMPLATE` -- that is how `template0` -works -- so sealing first leaves no window where it is reachable, and the copy -is file-level (minutes, not a re-run). Point the node, the replay, and parity -at `_serve`. Verify the artifact **after** the services are up: a -count taken before you start a node certifies a state the node may then change. +works -- and the copy is file-level (minutes, not a re-run). + +Either way, verify **after** the services are up: a count taken before you +start a node certifies a state the node may then change. ## C. Verifying: replay, parity, calibration **Running the verification.** Serve the artifact from a node, replay it into a -scratch ETL database, then compare that against the source snapshot. Point the -replay at the **serve copy**, never the sealed original, and point `--db` at an +scratch ETL database, then compare that against the source snapshot. Serve +from a database **restored from `chain.dump`** (or the serve copy of a sealed +database, [Appendix B](#b-the-artifact-chaindump-and-manifestjson)), and point `--db` at an ETL database, never the chain database -- the replay creates ETL schema and drops serving indexes in whatever it is given. @@ -505,6 +588,21 @@ one left there belongs to whichever node last ran against that directory — on P2P identity unrelated to its delegate key and unrelated to the id in `ProdPersistentPeers`. Delete it and let the node derive its own. +The same holds for every per-node file a CometBFT home accumulates. The writer +lists them in `MANIFEST.json` under `exclude_from_bootstrap`, with whether each +existed when it finished: + +| Path under `core//` | Why it stays behind | +|---|---| +| `config/node_key.json` | P2P identity of whichever node last ran there | +| `config/priv_validator_key.json` | the node derives it from `OPENAUDIO_DELEGATE_PRIVATE_KEY`; it must match `genesis_validator_address` ([J §3](#3-the-two-keys--they-are-different-and-only-one-is-ephemeral)) | +| `config/addrbook.json` | peer book of whichever node last ran there | +| `data/priv_validator_state.json` | double-sign guard of whichever node last signed there | + +Anything that ran against the output after the writer (a verification node) +can add files the manifest marks absent, so exclude every listed path, not just +the ones marked present. + `v.audius.rickyrombo.com` cannot be listed until it exists and its delegate key is known. Until then the second node reaches the network by dialling the first, and PEX propagates from there. Fill it in before step 15. diff --git a/cmd/genesis-writer/artifact.go b/cmd/genesis-writer/artifact.go new file mode 100644 index 00000000..fc49b045 --- /dev/null +++ b/cmd/genesis-writer/artifact.go @@ -0,0 +1,401 @@ +package main + +import ( + "context" + "crypto/sha256" + "encoding/hex" + "encoding/json" + "fmt" + "io/fs" + "net/url" + "os" + "os/exec" + "path/filepath" + "runtime" + "runtime/debug" + "strconv" + "strings" + "time" + + corecfg "github.com/OpenAudio/go-openaudio/pkg/core/config" + cmttypes "github.com/cometbft/cometbft/types" + "github.com/jackc/pgx/v5/pgxpool" + "go.uber.org/zap" +) + +// The writer's output is a portable artifact rather than a live database: +// +// / +// core// CometBFT home (genesis, blockstore, state) +// chain.dump/ pg_dump -Fd of the chain database +// MANIFEST.json what was written, from what, and how to seed a node +// +// A live database invites anything with its DSN to connect, and a node whose +// binary does not embed this genesis runs the core migrations down on connect. +// That destroyed a 3h36m write on 2026-08-25. The dump is taken before the +// writer exits, so nothing else has had a chance to touch the database. +const ( + chainDumpDirName = "chain.dump" + manifestFileName = "MANIFEST.json" + + // defaultSourceChainID is the old chain the production snapshot indexed. + defaultSourceChainID = "audius-mainnet-alpha-beta" +) + +// ArtifactManifest is MANIFEST.json. Operators and runbooks read values from +// it instead of copying them out of log lines. +type ArtifactManifest struct { + ChainID string `json:"chain_id"` + // EndHeight is the last block the writer produced, and the genesis + // migration end height. The bootstrap node's first live block is + // EndHeight+1. + EndHeight int64 `json:"end_height"` + FirstLiveHeight int64 `json:"first_live_height"` + + // SourceLastIndexedBlock is MAX(height) in the source snapshot's + // core_indexed_blocks for SourceChainID: the last old-chain block whose + // effects the migration carries. + SourceChainID string `json:"source_chain_id"` + SourceLastIndexedBlock int64 `json:"source_last_indexed_block"` + // NewChainFlushFromBlock is the value for the API's flusher. It deletes + // rows with confirmed_block < this value, so it must be one past the + // snapshot's last block or that block is replayed onto the new chain twice. + NewChainFlushFromBlock int64 `json:"new_chain_flush_from_block"` + + GenesisSHA256 string `json:"genesis_sha256"` + GenesisValidatorAddress string `json:"genesis_validator_address"` + GenesisMigrationAddress string `json:"genesis_migration_address"` + + Dump *DumpInfo `json:"dump"` + + WriterCommit string `json:"writer_commit"` + WrittenAt time.Time `json:"written_at"` + + // ExcludeFromBootstrap lists per-node identity under the CometBFT home that + // must not be seeded onto another node. Paths are relative to the data dir. + ExcludeFromBootstrap []ExcludedFile `json:"exclude_from_bootstrap"` +} + +// DumpInfo records the chain dump and the Postgres versions it is tied to. +type DumpInfo struct { + Path string `json:"path"` // relative to the data dir + Format string `json:"format"` + PgDumpVersion string `json:"pg_dump_version"` + PgDumpMajor int `json:"pg_dump_major"` + ServerVersion string `json:"server_version"` + ServerMajor int `json:"server_major"` + RestoreMinMajor int `json:"restore_min_pg_major"` // oldest pg_restore that reads it + SizeBytes int64 `json:"size_bytes"` +} + +// ExcludedFile is one path operators must not copy onto a bootstrap node. +type ExcludedFile struct { + Path string `json:"path"` + Reason string `json:"reason"` + Present bool `json:"present"` // whether it existed when the manifest was written +} + +// bootstrapExclusions is the per-node identity a CometBFT home accumulates. +// None of it belongs to the chain; all of it belongs to whichever process last +// ran against the directory. See ROLLOUT.md Appendix E. +func bootstrapExclusions(chainID string) []ExcludedFile { + home := filepath.Join("core", chainID) + return []ExcludedFile{ + { + Path: filepath.Join(home, "config", "node_key.json"), + Reason: "P2P identity of whichever node last ran here; a node derives its own from its delegate key", + }, + { + Path: filepath.Join(home, "config", "priv_validator_key.json"), + Reason: "consensus key; the bootstrap derives it from OPENAUDIO_DELEGATE_PRIVATE_KEY, and it must match genesis_validator_address", + }, + { + Path: filepath.Join(home, "config", "addrbook.json"), + Reason: "peer address book of whichever node last ran here", + }, + { + Path: filepath.Join(home, "data", "priv_validator_state.json"), + Reason: "double-sign guard of whichever node last signed here; a stale one can stop a node from starting", + }, + } +} + +// manifestInputs is everything newManifest needs, gathered by the writer. +type manifestInputs struct { + DataDir string + ChainID string + EndHeight int64 + SourceChainID string + SourceLastIndexedBlock int64 + GenesisFile string // the emitted genesis.json + MigrationAddress string + Dump *DumpInfo + WriterCommit string + Now time.Time +} + +func newManifest(in manifestInputs) (*ArtifactManifest, error) { + raw, err := os.ReadFile(in.GenesisFile) + if err != nil { + return nil, fmt.Errorf("read genesis: %w", err) + } + sum := sha256.Sum256(raw) + + genDoc, err := cmttypes.GenesisDocFromJSON(raw) + if err != nil { + return nil, fmt.Errorf("parse genesis: %w", err) + } + if len(genDoc.Validators) != 1 { + return nil, fmt.Errorf("genesis lists %d validators; the writer signs every block with one key, so it must list exactly one", len(genDoc.Validators)) + } + + excl := bootstrapExclusions(in.ChainID) + for i := range excl { + _, err := os.Stat(filepath.Join(in.DataDir, excl[i].Path)) + excl[i].Present = err == nil + } + + return &ArtifactManifest{ + ChainID: in.ChainID, + EndHeight: in.EndHeight, + FirstLiveHeight: in.EndHeight + 1, + SourceChainID: in.SourceChainID, + SourceLastIndexedBlock: in.SourceLastIndexedBlock, + NewChainFlushFromBlock: in.SourceLastIndexedBlock + 1, + GenesisSHA256: hex.EncodeToString(sum[:]), + GenesisValidatorAddress: genDoc.Validators[0].Address.String(), + GenesisMigrationAddress: in.MigrationAddress, + Dump: in.Dump, + WriterCommit: in.WriterCommit, + WrittenAt: in.Now.UTC(), + ExcludeFromBootstrap: excl, + }, nil +} + +// writeFileAtomic writes via a temp file and rename so a reader never sees a +// partial manifest. +func writeFileAtomic(path string, data []byte, perm os.FileMode) error { + tmp := path + ".tmp" + if err := os.WriteFile(tmp, data, perm); err != nil { + return err + } + return os.Rename(tmp, path) +} + +func writeManifest(path string, m *ArtifactManifest) error { + out, err := json.MarshalIndent(m, "", " ") + if err != nil { + return err + } + return writeFileAtomic(path, append(out, '\n'), 0o644) +} + +// querySourceLastIndexedBlock returns the last old-chain block the source +// snapshot indexed. Zero rows is an error: a snapshot that indexed nothing is +// not a production snapshot, and a zero here would become a flush-from value +// that replays the old chain's entire history. +func querySourceLastIndexedBlock(ctx context.Context, src *pgxpool.Pool, chainID string) (int64, error) { + var h *int64 + if err := src.QueryRow(ctx, + `SELECT MAX(height) FROM core_indexed_blocks WHERE chain_id = $1`, chainID).Scan(&h); err != nil { + return 0, fmt.Errorf("query core_indexed_blocks: %w", err) + } + if h == nil { + return 0, fmt.Errorf("source core_indexed_blocks has no rows for chain %q (set --source-chain-id)", chainID) + } + return *h, nil +} + +// writerCommit identifies the writer build: the VCS revision Go stamps into +// `go build` binaries, else the ldflags version, else "unknown". +func writerCommit() string { + if bi, ok := debug.ReadBuildInfo(); ok { + var rev string + var dirty bool + for _, s := range bi.Settings { + switch s.Key { + case "vcs.revision": + rev = s.Value + case "vcs.modified": + dirty = s.Value == "true" + } + } + if rev != "" { + if dirty { + rev += "-dirty" + } + return rev + } + } + if corecfg.Version != "" { + return corecfg.Version + } + return "unknown" +} + +// pgDumpConn splits a DSN into the connection string pg_dump receives on argv +// and the environment it needs, so a password never lands in argv. +func pgDumpConn(dsn string) (connStr string, env []string) { + u, err := url.Parse(dsn) + if err != nil || (u.Scheme != "postgres" && u.Scheme != "postgresql") || u.User == nil { + return dsn, nil + } + pw, ok := u.User.Password() + if !ok { + return dsn, nil + } + u.User = url.User(u.User.Username()) + return u.String(), []string{"PGPASSWORD=" + pw} +} + +// defaultDumpJobs is the pg_dump -j used when --dump-jobs is unset. Each job +// is a server connection and a backend process, so stay well under the +// default max_connections and leave CPUs for the server itself. +func defaultDumpJobs() int { + return min(max(runtime.NumCPU()/2, 1), 8) +} + +// pgDumpArgs builds the pg_dump invocation. The writer's progress table is +// internal resume state, not chain state, so it stays out. Owners and ACLs are +// dropped because the node restores as its own role. +func pgDumpArgs(connStr, outDir string, jobs int) []string { + return []string{ + "--format=directory", + "--jobs=" + strconv.Itoa(jobs), + "--file=" + outDir, + "--no-owner", + "--no-privileges", + "--exclude-table=genesis_writer_progress", + "--dbname=" + connStr, + } +} + +// dumpChainDB writes the destination database to /chain.dump. It +// dumps into a staging directory and renames on success, so chain.dump only +// ever holds a complete archive. A chain.dump left by an earlier run of the +// same data dir is replaced: the database just dumped supersedes it. +func dumpChainDB(ctx context.Context, tool *pgDumpTool, dsn, dataDir string, jobs int, logger *zap.Logger) (string, int64, error) { + final := filepath.Join(dataDir, chainDumpDirName) + staging := final + ".partial" + if err := os.RemoveAll(staging); err != nil { + return "", 0, fmt.Errorf("remove stale %s: %w", staging, err) + } + + connStr, env := pgDumpConn(dsn) + cmd := exec.CommandContext(ctx, tool.Path, pgDumpArgs(connStr, staging, jobs)...) + cmd.Env = append(os.Environ(), env...) + cmd.Stdout = os.Stdout + cmd.Stderr = os.Stderr + + start := time.Now() + logger.Info("dumping chain database", + zap.String("pg_dump", tool.Version), zap.String("out", final), zap.Int("jobs", jobs)) + if err := cmd.Run(); err != nil { + return "", 0, fmt.Errorf("pg_dump: %w", err) + } + + if _, err := os.Stat(final); err == nil { + logger.Warn("replacing chain dump from an earlier run", zap.String("path", final)) + if err := os.RemoveAll(final); err != nil { + return "", 0, fmt.Errorf("remove old %s: %w", final, err) + } + } + if err := os.Rename(staging, final); err != nil { + return "", 0, fmt.Errorf("rename %s: %w", staging, err) + } + + size, err := dirSize(final) + if err != nil { + return "", 0, fmt.Errorf("size %s: %w", final, err) + } + logger.Info("chain database dumped", + zap.String("path", final), zap.Int64("bytes", size), zap.Duration("elapsed", time.Since(start))) + return final, size, nil +} + +func dirSize(dir string) (int64, error) { + var total int64 + err := filepath.WalkDir(dir, func(_ string, d fs.DirEntry, err error) error { + if err != nil || d.IsDir() { + return err + } + info, err := d.Info() + if err != nil { + return err + } + total += info.Size() + return nil + }) + return total, err +} + +// emitArtifact dumps the chain database and writes MANIFEST.json. It runs at +// the very end of Run, after the indexes are rebuilt so the dump carries them. +func (w *Writer) emitArtifact(ctx context.Context) error { + dataDir := w.cfg.ArtifactDir + + var dump *DumpInfo + if w.cfg.Dump { + pt := w.cfg.Postgres + if pt == nil || pt.Dump == nil { + return fmt.Errorf("dump requested without a resolved pg_dump") + } + jobs := w.cfg.DumpJobs + if jobs <= 0 { + jobs = defaultDumpJobs() + } + path, size, err := dumpChainDB(ctx, pt.Dump, w.cfg.DstDSN, dataDir, jobs, w.logger) + if err != nil { + return err + } + rel, _ := filepath.Rel(dataDir, path) + dump = &DumpInfo{ + Path: rel, + Format: "directory", + PgDumpVersion: pt.Dump.Version, + PgDumpMajor: pt.Dump.Major, + ServerVersion: pt.ServerVersion, + ServerMajor: pt.ServerMajor, + RestoreMinMajor: pt.Dump.Major, + SizeBytes: size, + } + } else { + w.logger.Warn("chain dump disabled (--no-dump); the chain database is only in the live postgres") + } + + m, err := newManifest(manifestInputs{ + DataDir: dataDir, + ChainID: w.cfg.ChainID, + EndHeight: w.finalHeight, + SourceChainID: w.cfg.SourceChainID, + SourceLastIndexedBlock: w.sourceLastIndexedBlock, + GenesisFile: filepath.Join(w.cfg.CMTHome, "config", "genesis.json"), + MigrationAddress: w.signerAddr, + Dump: dump, + WriterCommit: writerCommit(), + Now: time.Now(), + }) + if err != nil { + return fmt.Errorf("build manifest: %w", err) + } + path := filepath.Join(dataDir, manifestFileName) + if err := writeManifest(path, m); err != nil { + return fmt.Errorf("write manifest: %w", err) + } + + var present []string + for _, e := range m.ExcludeFromBootstrap { + if e.Present { + present = append(present, e.Path) + } + } + w.logger.Info("wrote manifest", + zap.String("path", path), + zap.Int64("end_height", m.EndHeight), + zap.Int64("new_chain_flush_from_block", m.NewChainFlushFromBlock), + zap.String("genesis_sha256", m.GenesisSHA256), + zap.String("genesis_validator", m.GenesisValidatorAddress), + zap.String("do_not_seed", strings.Join(present, ", ")), + ) + return nil +} diff --git a/cmd/genesis-writer/artifact_test.go b/cmd/genesis-writer/artifact_test.go new file mode 100644 index 00000000..dd856204 --- /dev/null +++ b/cmd/genesis-writer/artifact_test.go @@ -0,0 +1,247 @@ +package main + +import ( + "crypto/sha256" + "encoding/hex" + "encoding/json" + "os" + "path/filepath" + "slices" + "strings" + "testing" + "time" + + "github.com/cometbft/cometbft/privval" + "go.uber.org/zap" +) + +// The manifest is what operators read heights and hashes from instead of log +// lines, so every field has to come from the artifact it describes. +func TestManifestDescribesTheArtifact(t *testing.T) { + dataDir := t.TempDir() + chainID := "audius-mainnet-beta" + home := cmtHome(dataDir, chainID) + genesisFile := filepath.Join(home, "config", "genesis.json") + keyFile := filepath.Join(home, "config", "priv_validator_key.json") + if err := ensureGenesisFiles(genesisFile, keyFile, chainID, time.Unix(1_700_000_000, 0).UTC(), zap.NewNop()); err != nil { + t.Fatal(err) + } + + dump := &DumpInfo{Path: chainDumpDirName, Format: "directory", PgDumpVersion: "pg_dump (PostgreSQL) 15.8", PgDumpMajor: 15, ServerVersion: "15.8", ServerMajor: 15, RestoreMinMajor: 15} + now := time.Date(2026, 8, 25, 12, 0, 0, 0, time.FixedZone("PDT", -7*3600)) + m, err := newManifest(manifestInputs{ + DataDir: dataDir, + ChainID: chainID, + EndHeight: 5839, + SourceChainID: defaultSourceChainID, + SourceLastIndexedBlock: 12_345_678, + GenesisFile: genesisFile, + MigrationAddress: "0x7D01Cd0A89cc73F5a6DBEd10992AA472A2312D5F", + Dump: dump, + WriterCommit: "abc123", + Now: now, + }) + if err != nil { + t.Fatal(err) + } + + raw, _ := os.ReadFile(genesisFile) + sum := sha256.Sum256(raw) + pv := privval.LoadFilePVEmptyState(keyFile, "") + + checks := []struct { + name string + got, want any + }{ + {"chain_id", m.ChainID, chainID}, + {"end_height", m.EndHeight, int64(5839)}, + {"first_live_height", m.FirstLiveHeight, int64(5840)}, + {"source_chain_id", m.SourceChainID, "audius-mainnet-alpha-beta"}, + {"source_last_indexed_block", m.SourceLastIndexedBlock, int64(12_345_678)}, + // One past the snapshot: the flusher deletes confirmed_block < value, + // so the snapshot's own last block would otherwise replay twice. + {"new_chain_flush_from_block", m.NewChainFlushFromBlock, int64(12_345_679)}, + {"genesis_sha256", m.GenesisSHA256, hex.EncodeToString(sum[:])}, + {"genesis_validator_address", m.GenesisValidatorAddress, pv.Key.Address.String()}, + {"genesis_migration_address", m.GenesisMigrationAddress, "0x7D01Cd0A89cc73F5a6DBEd10992AA472A2312D5F"}, + {"writer_commit", m.WriterCommit, "abc123"}, + {"written_at is UTC", m.WrittenAt, now.UTC()}, + {"dump", m.Dump, dump}, + } + for _, c := range checks { + if c.got != c.want { + t.Errorf("%s = %v, want %v", c.name, c.got, c.want) + } + } + + // Per-node identity is listed, and presence reflects the directory: the + // writer itself created the validator key and state, nothing created the + // node key or address book. + present := map[string]bool{} + for _, e := range m.ExcludeFromBootstrap { + if e.Reason == "" { + t.Errorf("%s has no reason", e.Path) + } + present[e.Path] = e.Present + } + wantPresent := map[string]bool{ + "core/audius-mainnet-beta/config/node_key.json": false, + "core/audius-mainnet-beta/config/priv_validator_key.json": true, + "core/audius-mainnet-beta/config/addrbook.json": false, + "core/audius-mainnet-beta/data/priv_validator_state.json": true, + } + if len(present) != len(wantPresent) { + t.Errorf("exclude_from_bootstrap = %v, want keys of %v", present, wantPresent) + } + for p, want := range wantPresent { + got, ok := present[p] + if !ok { + t.Errorf("exclude_from_bootstrap is missing %s", p) + } else if got != want { + t.Errorf("%s present = %v, want %v", p, got, want) + } + } + + // The JSON keys are the contract runbooks read; pin them. + path := filepath.Join(dataDir, manifestFileName) + if err := writeManifest(path, m); err != nil { + t.Fatal(err) + } + b, err := os.ReadFile(path) + if err != nil { + t.Fatal(err) + } + var doc map[string]any + if err := json.Unmarshal(b, &doc); err != nil { + t.Fatal(err) + } + for _, k := range []string{ + "chain_id", "end_height", "first_live_height", "source_chain_id", + "source_last_indexed_block", "new_chain_flush_from_block", "genesis_sha256", + "genesis_validator_address", "genesis_migration_address", "dump", + "writer_commit", "written_at", "exclude_from_bootstrap", + } { + if _, ok := doc[k]; !ok { + t.Errorf("MANIFEST.json missing key %q", k) + } + } + d := doc["dump"].(map[string]any) + for _, k := range []string{"path", "format", "pg_dump_version", "pg_dump_major", "server_version", "server_major", "restore_min_pg_major"} { + if _, ok := d[k]; !ok { + t.Errorf("MANIFEST.json dump missing key %q", k) + } + } + if _, err := os.Stat(path + ".tmp"); !os.IsNotExist(err) { + t.Errorf("temp manifest left behind: %v", err) + } +} + +func TestManifestWithoutDumpRecordsNull(t *testing.T) { + dataDir := t.TempDir() + home := cmtHome(dataDir, "c") + genesisFile := filepath.Join(home, "config", "genesis.json") + if err := ensureGenesisFiles(genesisFile, filepath.Join(home, "config", "priv_validator_key.json"), "c", time.Now(), zap.NewNop()); err != nil { + t.Fatal(err) + } + m, err := newManifest(manifestInputs{DataDir: dataDir, ChainID: "c", EndHeight: 1, GenesisFile: genesisFile, Now: time.Now()}) + if err != nil { + t.Fatal(err) + } + b, _ := json.Marshal(m) + if !strings.Contains(string(b), `"dump":null`) { + t.Errorf("want explicit null dump, got %s", b) + } +} + +func TestParsePgToolMajor(t *testing.T) { + cases := map[string]int{ + "pg_dump (PostgreSQL) 15.8 (Debian 15.8-1.pgdg120+1)": 15, + "pg_dump (PostgreSQL) 17.2 (Homebrew)\n": 17, + "pg_ctl (PostgreSQL) 16.4": 16, + "pg_dump (PostgreSQL) 18beta1": 18, + } + for in, want := range cases { + got, err := parsePgToolMajor(in) + if err != nil || got != want { + t.Errorf("parsePgToolMajor(%q) = %d, %v; want %d", in, got, err, want) + } + } + if _, err := parsePgToolMajor("garbage"); err == nil { + t.Error("want error for unrecognized output") + } + if got := serverMajorFromNum(150008); got != 15 { + t.Errorf("serverMajorFromNum(150008) = %d", got) + } + if got := serverMajorFromNum(170002); got != 17 { + t.Errorf("serverMajorFromNum(170002) = %d", got) + } +} + +// server <= pg_dump <= node, or the output cannot reach the node. +func TestPostgresVersionChecks(t *testing.T) { + const node = nodePostgresMajor + if node != 15 { + t.Fatalf("nodePostgresMajor = %d; the node image (cmd/openaudio/Dockerfile) bundles postgresql-15 — update both together", node) + } + + // Server major. + if err := checkServerMajor(15, node, false); err != nil { + t.Errorf("same major refused: %v", err) + } + if err := checkServerMajor(14, node, false); err != nil { + t.Errorf("older server refused: %v", err) + } + err := checkServerMajor(17, node, false) + if err == nil { + t.Fatal("newer server (the 2026-08-25 PG17 case) accepted without override") + } + if !strings.Contains(err.Error(), "--allow-newer-postgres") { + t.Errorf("refusal does not name the override: %v", err) + } + if err := checkServerMajor(17, node, true); err != nil { + t.Errorf("override ignored: %v", err) + } + + // pg_dump major. + cases := []struct { + dump, server int + allow, want bool + }{ + {15, 15, false, true}, + {15, 14, false, true}, // newer client, older server: fine + {14, 15, false, false}, // pg_dump refuses a newer server + {17, 15, false, false}, // archive unreadable by pg_restore 15 + {17, 15, true, true}, + {16, 17, true, false}, // still cannot dump a newer server + } + for _, c := range cases { + if got := pgDumpFits(c.dump, c.server, node, c.allow); got != c.want { + t.Errorf("pgDumpFits(dump=%d, server=%d, allow=%v) = %v, want %v", c.dump, c.server, c.allow, got, c.want) + } + } +} + +func TestPgDumpKeepsPasswordOffArgv(t *testing.T) { + conn, env := pgDumpConn("postgres://writer:s3cret@localhost:5440/openaudio?sslmode=disable") + if strings.Contains(conn, "s3cret") { + t.Errorf("password on argv: %s", conn) + } + if conn != "postgres://writer@localhost:5440/openaudio?sslmode=disable" { + t.Errorf("conn = %s", conn) + } + if !slices.Equal(env, []string{"PGPASSWORD=s3cret"}) { + t.Errorf("env = %v", env) + } + + conn, env = pgDumpConn("postgres://writer@localhost:5440/openaudio") + if conn != "postgres://writer@localhost:5440/openaudio" || env != nil { + t.Errorf("passwordless DSN changed: %s %v", conn, env) + } + + args := pgDumpArgs("CONN", "/out/chain.dump.partial", 4) + for _, want := range []string{"--format=directory", "--jobs=4", "--file=/out/chain.dump.partial", "--exclude-table=genesis_writer_progress", "--dbname=CONN"} { + if !slices.Contains(args, want) { + t.Errorf("pg_dump args %v missing %s", args, want) + } + } +} diff --git a/cmd/genesis-writer/main.go b/cmd/genesis-writer/main.go index 2c70adb8..5a905eb3 100644 --- a/cmd/genesis-writer/main.go +++ b/cmd/genesis-writer/main.go @@ -20,6 +20,7 @@ import ( "os" "os/signal" "path/filepath" + "strconv" "strings" "syscall" "time" @@ -118,6 +119,37 @@ func writeCmd() *cli.Command { Name: "run-migrations", Usage: "Apply the Core chain schema to dst-dsn before writing (for fresh databases)", }, + &cli.BoolFlag{ + Name: "dump", + Usage: "After a successful write, pg_dump the chain database to /" + chainDumpDirName + " (directory format, parallel). The dump and " + manifestFileName + " are the artifact to ship; see ROLLOUT.md", + EnvVars: []string{"GENESIS_DUMP"}, + Value: true, + }, + &cli.BoolFlag{ + Name: "no-dump", + Usage: "Skip the chain dump (same as --dump=false). " + manifestFileName + " is still written", + }, + &cli.IntFlag{ + Name: "dump-jobs", + Usage: "pg_dump --jobs for the chain dump (default: half the CPUs, at most 8)", + EnvVars: []string{"GENESIS_DUMP_JOBS"}, + }, + &cli.IntFlag{ + Name: "target-pg-major", + Usage: "Postgres major version of the node that restores the dump (the node image bundles " + strconv.Itoa(nodePostgresMajor) + "). The managed postgres runs exactly this major, and a --dst-dsn server newer than it is refused", + EnvVars: []string{"GENESIS_TARGET_PG_MAJOR"}, + Value: nodePostgresMajor, + }, + &cli.BoolFlag{ + Name: "allow-newer-postgres", + Usage: "Proceed even though the --dst-dsn server (or the only pg_dump found) is newer than --target-pg-major. The output then cannot be restored on a node as-is", + }, + &cli.StringFlag{ + Name: "source-chain-id", + Usage: "Old chain ID whose core_indexed_blocks rows in the source snapshot mark its last indexed block, recorded in " + manifestFileName, + EnvVars: []string{"GENESIS_SOURCE_CHAIN_ID"}, + Value: defaultSourceChainID, + }, &cli.BoolFlag{ Name: "resume", Usage: "Resume from the last completed step of a previous run", @@ -189,24 +221,42 @@ func writeCmd() *cli.Command { } } + dump := c.Bool("dump") && !c.Bool("no-dump") + if dump && dataDir == "" { + return fmt.Errorf("--dump writes to /%s; set --data-dir or pass --no-dump", chainDumpDirName) + } + targetPgMajor := c.Int("target-pg-major") + // Resolve destination DSN — start a managed postgres if needed. dstDSN := c.String("dst-dsn") runMigrations := c.Bool("run-migrations") var pg *managedPostgres + var pgBinHint string if dstDSN == "" { if dataDir == "" { return fmt.Errorf("either --dst-dsn or --data-dir must be set") } - pg, dstDSN, err = startManagedPostgres(dataDir, logger) + pg, dstDSN, err = startManagedPostgres(dataDir, targetPgMajor, logger) if err != nil { return fmt.Errorf("managed postgres: %w", err) } defer pg.Stop() + pgBinHint = pg.binDir // Always run migrations for managed postgres — they're idempotent // and fast, so safe on both fresh runs and resume. runMigrations = true } + ctx, cancel := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) + defer cancel() + + // Check versions before writing anything: an output the node + // cannot restore should cost seconds, not a full run. + pgInfo, err := checkDestinationPostgres(ctx, dstDSN, targetPgMajor, c.Bool("allow-newer-postgres"), dump, pgBinHint, logger) + if err != nil { + return err + } + cfg := &WriterConfig{ SrcDSN: c.String("src-dsn"), DstDSN: dstDSN, @@ -234,6 +284,11 @@ func writeCmd() *cli.Command { SkipRewards: c.Bool("skip-rewards"), CoreDSN: c.String("core-dsn"), LaunchpadMintsFile: c.String("launchpad-mints"), + ArtifactDir: dataDir, + Dump: dump, + DumpJobs: c.Int("dump-jobs"), + Postgres: pgInfo, + SourceChainID: c.String("source-chain-id"), } w, err := NewWriter(cfg, logger) @@ -242,9 +297,6 @@ func writeCmd() *cli.Command { } defer w.Close() - ctx, cancel := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) - defer cancel() - if err := w.Run(ctx); err != nil { return err } diff --git a/cmd/genesis-writer/pgversion.go b/cmd/genesis-writer/pgversion.go new file mode 100644 index 00000000..f3eb235c --- /dev/null +++ b/cmd/genesis-writer/pgversion.go @@ -0,0 +1,222 @@ +package main + +import ( + "context" + "fmt" + "os" + "os/exec" + "path/filepath" + "regexp" + "strconv" + "strings" + + "github.com/jackc/pgx/v5" + "go.uber.org/zap" +) + +// nodePostgresMajor is the Postgres major version the node image bundles +// (cmd/openaudio/Dockerfile installs postgresql-15). The chain dump is restored +// there, so it is the default target for everything the writer produces. +// +// Two compatibility rules pin the whole pipeline to it: +// +// - pg_restore cannot read an archive written by a newer pg_dump: each major +// bumps the archive format (1.14 through 15, 1.15 in 16, 1.16 in 17), and +// an older pg_restore refuses with "unsupported version in file header". +// - pg_dump refuses to dump a server newer than itself. +// +// So server <= pg_dump <= node. On 2026-08-25 the artifact sat in a PG17 +// cluster on macOS, which no PG15 pg_dump can read and whose PG17 dump no PG15 +// pg_restore can load; moving it to the node took a hand-built workaround. +const nodePostgresMajor = 15 + +// pgDumpTool is a pg_dump binary and the version it reports. +type pgDumpTool struct { + Path string + Version string // first line of `pg_dump --version` + Major int +} + +// pgTarget describes the destination database's server and the pg_dump that +// will archive it. +type pgTarget struct { + ServerVersion string // SHOW server_version + ServerMajor int + TargetMajor int // the node's Postgres major, where the dump is restored + Dump *pgDumpTool // nil when dumping is disabled +} + +var pgVersionRe = regexp.MustCompile(`\(PostgreSQL\)\s+(\d+)`) + +// parsePgToolMajor extracts the major version from a Postgres client tool's +// --version output, e.g. "pg_dump (PostgreSQL) 15.8 (Debian 15.8-1.pgdg120+1)" +// or "pg_dump (PostgreSQL) 17beta1". +func parsePgToolMajor(out string) (int, error) { + m := pgVersionRe.FindStringSubmatch(out) + if m == nil { + return 0, fmt.Errorf("unrecognized version output %q", strings.TrimSpace(out)) + } + return strconv.Atoi(m[1]) +} + +// serverMajorFromNum converts server_version_num (e.g. 150008) to its major +// version. Every supported server is >= 10, where the major is the leading +// component. +func serverMajorFromNum(num int) int { + return num / 10000 +} + +// checkServerMajor enforces server <= target. A newer server produces a +// database no node can restore, so it is refused unless explicitly allowed. +func checkServerMajor(server, target int, allowNewer bool) error { + if server <= target { + return nil + } + if allowNewer { + return nil + } + return fmt.Errorf("destination postgres is major %d but the node restores into postgres %d: "+ + "pg_dump cannot dump a newer server with an older client, and a pg_dump %d archive "+ + "is unreadable by pg_restore %d. Write into a postgres %d server (omit --dst-dsn to "+ + "use the managed one), or pass --allow-newer-postgres if you will move the data some other way", + server, target, server, target, target) +} + +// pgDumpFits reports whether a pg_dump of the given major can dump the server +// and produce an archive the target's pg_restore reads. +func pgDumpFits(dumpMajor, server, target int, allowNewer bool) bool { + if dumpMajor < server { + return false + } + return dumpMajor <= target || allowNewer +} + +// pgBinCandidates lists directories that conventionally hold the client tools +// for one Postgres major, most specific first. +func pgBinCandidates(major int) []string { + m := strconv.Itoa(major) + return []string{ + // macOS Homebrew + "/opt/homebrew/opt/postgresql@" + m + "/bin", + "/usr/local/opt/postgresql@" + m + "/bin", + // macOS Postgres.app + "/Applications/Postgres.app/Contents/Versions/" + m + "/bin", + // Debian / Ubuntu + "/usr/lib/postgresql/" + m + "/bin", + // RHEL / Fedora PGDG + "/usr/pgsql-" + m + "/bin", + } +} + +func probePgTool(path string) (version string, major int, err error) { + out, err := exec.Command(path, "--version").Output() + if err != nil { + return "", 0, err + } + version = strings.TrimSpace(strings.SplitN(string(out), "\n", 2)[0]) + major, err = parsePgToolMajor(version) + return version, major, err +} + +// findPgDump picks a pg_dump that satisfies server <= pg_dump <= target. It +// prefers the target major, then the managed cluster's bin dir, then PATH. +func findPgDump(binDirHint string, server, target int, allowNewer bool) (*pgDumpTool, error) { + var dirs []string + dirs = append(dirs, pgBinCandidates(target)...) + if binDirHint != "" { + dirs = append([]string{binDirHint}, dirs...) + } + for m := target - 1; m >= server; m-- { + dirs = append(dirs, pgBinCandidates(m)...) + } + + var paths []string + for _, d := range dirs { + paths = append(paths, filepath.Join(d, "pg_dump")) + } + if p, err := exec.LookPath("pg_dump"); err == nil { + paths = append(paths, p) + } + + seen := map[string]bool{} + var rejected []string + for _, p := range paths { + if seen[p] { + continue + } + seen[p] = true + if _, err := os.Stat(p); err != nil { + continue + } + version, major, err := probePgTool(p) + if err != nil { + rejected = append(rejected, fmt.Sprintf("%s (%v)", p, err)) + continue + } + if pgDumpFits(major, server, target, allowNewer) { + return &pgDumpTool{Path: p, Version: version, Major: major}, nil + } + rejected = append(rejected, fmt.Sprintf("%s (major %d)", p, major)) + } + + msg := fmt.Sprintf("no pg_dump with server(%d) <= major <= node(%d) found; install postgres %d client tools (e.g. `brew install postgresql@%d`)", + server, target, target, target) + if len(rejected) > 0 { + msg += "; rejected: " + strings.Join(rejected, ", ") + } + return nil, fmt.Errorf("%s, or pass --no-dump", msg) +} + +// checkDestinationPostgres reads the destination server's version, enforces +// that the node can restore it, and resolves the pg_dump to archive it with. +// It runs before any writing, so a mismatch costs seconds rather than a +// multi-hour run that cannot be moved. +func checkDestinationPostgres(ctx context.Context, dsn string, target int, allowNewer, dump bool, binDirHint string, logger *zap.Logger) (*pgTarget, error) { + conn, err := pgx.Connect(ctx, dsn) + if err != nil { + return nil, fmt.Errorf("connect dst db for version check: %w", err) + } + defer conn.Close(ctx) + + var numStr, version string + if err := conn.QueryRow(ctx, "SHOW server_version_num").Scan(&numStr); err != nil { + return nil, fmt.Errorf("read server_version_num: %w", err) + } + if err := conn.QueryRow(ctx, "SHOW server_version").Scan(&version); err != nil { + return nil, fmt.Errorf("read server_version: %w", err) + } + num, err := strconv.Atoi(numStr) + if err != nil { + return nil, fmt.Errorf("parse server_version_num %q: %w", numStr, err) + } + + t := &pgTarget{ServerVersion: version, ServerMajor: serverMajorFromNum(num), TargetMajor: target} + if err := checkServerMajor(t.ServerMajor, target, allowNewer); err != nil { + return nil, err + } + if t.ServerMajor > target { + logger.Error("!!! DESTINATION POSTGRES IS NEWER THAN THE NODE'S — the output cannot be restored on a node as-is !!!", + zap.Int("server_major", t.ServerMajor), + zap.Int("node_major", target), + zap.String("override", "--allow-newer-postgres")) + } + + if dump { + tool, err := findPgDump(binDirHint, t.ServerMajor, target, allowNewer) + if err != nil { + return nil, err + } + t.Dump = tool + if tool.Major > target { + logger.Error("!!! pg_dump IS NEWER THAN THE NODE'S postgres — its archive will not load with the node's pg_restore !!!", + zap.String("pg_dump", tool.Version), zap.Int("node_major", target)) + } + } + + logger.Info("destination postgres version checked", + zap.String("server_version", t.ServerVersion), + zap.Int("node_major", target), + zap.Any("pg_dump", t.Dump), + ) + return t, nil +} diff --git a/cmd/genesis-writer/postgres.go b/cmd/genesis-writer/postgres.go index 253cac6a..e1764206 100644 --- a/cmd/genesis-writer/postgres.go +++ b/cmd/genesis-writer/postgres.go @@ -6,6 +6,7 @@ import ( "os/exec" "os/user" "path/filepath" + "strconv" "strings" "time" @@ -66,32 +67,27 @@ type managedPostgres struct { started bool } -// findPgBinDir searches common locations for the pg_ctl binary directory. -func findPgBinDir() (string, error) { - candidates := []string{ - // macOS Homebrew - "/opt/homebrew/opt/postgresql@17/bin", - "/opt/homebrew/opt/postgresql@16/bin", - "/opt/homebrew/opt/postgresql@15/bin", - "/usr/local/opt/postgresql@17/bin", - "/usr/local/opt/postgresql@16/bin", - "/usr/local/opt/postgresql@15/bin", - // Linux - "/usr/lib/postgresql/17/bin", - "/usr/lib/postgresql/16/bin", - "/usr/lib/postgresql/15/bin", - "/usr/bin", - } - for _, dir := range candidates { +// findPgBinDir finds the server binaries for exactly one Postgres major. +// +// The managed cluster's major is pinned rather than "newest installed" because +// the database leaves this machine as a pg_dump archive restored on a node, and +// the node's pg_restore cannot read an archive from a newer pg_dump (see +// nodePostgresMajor). Picking up a Homebrew postgresql@17 because it happened +// to be installed is how the 2026-08-25 artifact ended up unmovable. +func findPgBinDir(major int) (string, error) { + for _, dir := range pgBinCandidates(major) { if _, err := os.Stat(filepath.Join(dir, "pg_ctl")); err == nil { return dir, nil } } - // Fall back to PATH + // Fall back to PATH, but only if it is the right major. if path, err := exec.LookPath("pg_ctl"); err == nil { - return filepath.Dir(path), nil + if _, m, err := probePgTool(path); err == nil && m == major { + return filepath.Dir(path), nil + } } - return "", fmt.Errorf("could not find pg_ctl; install PostgreSQL or set --dst-dsn") + return "", fmt.Errorf("could not find PostgreSQL %d server binaries (pg_ctl); install them "+ + "(e.g. `brew install postgresql@%d`), change --target-pg-major, or set --dst-dsn", major, major) } // startManagedPostgres ensures a local PostgreSQL instance is running with its @@ -100,8 +96,8 @@ func findPgBinDir() (string, error) { // - Fresh run: initdb → start → create database // - Resume (stopped): start existing cluster → verify database exists // - Resume (already running): verify database exists -func startManagedPostgres(dataDir string, logger *zap.Logger) (*managedPostgres, string, error) { - binDir, err := findPgBinDir() +func startManagedPostgres(dataDir string, major int, logger *zap.Logger) (*managedPostgres, string, error) { + binDir, err := findPgBinDir(major) if err != nil { return nil, "", err } @@ -114,6 +110,14 @@ func startManagedPostgres(dataDir string, logger *zap.Logger) (*managedPostgres, logger: logger, } + // A cluster from an earlier run must be the same major: postgres cannot + // start a data directory from another major, and a resume would otherwise + // fail with a less helpful error from pg_ctl. + if v, ok := pg.clusterMajor(); ok && v != major { + return nil, "", fmt.Errorf("existing cluster at %s is postgres %d but --target-pg-major is %d; "+ + "use a fresh --data-dir or pass --target-pg-major=%d", pgDataDir, v, major, v) + } + if pg.isRunning() { logger.Info("postgres already running", zap.String("dataDir", pgDataDir)) } else if pg.isInitialized() { @@ -163,6 +167,16 @@ func (pg *managedPostgres) isInitialized() bool { return err == nil } +// clusterMajor reads PG_VERSION from an initialized data directory. +func (pg *managedPostgres) clusterMajor() (int, bool) { + b, err := os.ReadFile(filepath.Join(pg.dataDir, "PG_VERSION")) + if err != nil { + return 0, false + } + v, err := strconv.Atoi(strings.TrimSpace(string(b))) + return v, err == nil +} + func (pg *managedPostgres) isRunning() bool { cmd := pg.pgCtl("status", "-D", pg.dataDir) return cmd.Run() == nil diff --git a/cmd/genesis-writer/writer.go b/cmd/genesis-writer/writer.go index bb6386bd..46b3f223 100644 --- a/cmd/genesis-writer/writer.go +++ b/cmd/genesis-writer/writer.go @@ -69,6 +69,20 @@ type WriterConfig struct { RunMigrations bool // Resume picks up from the last completed step if a previous run was interrupted. Resume bool + + // ArtifactDir is the data dir that receives MANIFEST.json and, when Dump + // is set, chain.dump. Empty skips both (tests that only inspect the DB). + ArtifactDir string + // Dump archives the chain database to /chain.dump. + Dump bool + // DumpJobs is pg_dump --jobs; <= 0 picks a default. + DumpJobs int + // Postgres is the destination server's version and the pg_dump resolved + // for it, checked before the write starts. + Postgres *pgTarget + // SourceChainID is the old chain whose core_indexed_blocks bound the + // source snapshot. + SourceChainID string } // Writer reads Audius DP entities, signs them, and writes real CometBFT blocks @@ -88,6 +102,10 @@ type Writer struct { // projectBlockAuthState — and reported at the end of Run. authProjectionSkips int + // sourceLastIndexedBlock is the last old-chain block in the source + // snapshot, read at the start of Run for the manifest. + sourceLastIndexedBlock int64 + // current block being assembled height int64 blockTxs [][]byte @@ -451,6 +469,19 @@ func (w *Writer) Run(ctx context.Context) error { return fmt.Errorf("create progress table: %w", err) } + // Read the manifest's source bound up front: a source without it should + // fail now, not after the write. + if w.cfg.ArtifactDir != "" { + h, err := querySourceLastIndexedBlock(ctx, w.srcDB, w.cfg.SourceChainID) + if err != nil { + return fmt.Errorf("source snapshot bound: %w", err) + } + w.sourceLastIndexedBlock = h + w.logger.Info("source snapshot bound", + zap.String("source_chain_id", w.cfg.SourceChainID), + zap.Int64("last_indexed_block", h)) + } + // If resuming, load the last known chain state from the database. if w.cfg.Resume { var maxHeight int64 @@ -683,24 +714,33 @@ func (w *Writer) Run(ctx context.Context) error { return fmt.Errorf("create indexes: %w", err) } + // Dump and manifest last: the dump must carry the rebuilt indexes, and it + // runs before this process exits so nothing else has connected first. + if w.cfg.ArtifactDir != "" && w.cfg.CMTHome != "" && w.finalHeight > 0 { + if err := w.emitArtifact(ctx); err != nil { + return fmt.Errorf("emit artifact: %w", err) + } + } + // Print next-steps instructions. if w.cfg.CMTHome != "" && w.finalHeight > 0 { genesisPath := filepath.Join(w.cfg.CMTHome, "config", "genesis.json") + next := fmt.Sprintf("%d", w.finalHeight+1) w.logger.Info("genesis write finished — next steps:\n\n" + " 1. Copy the genesis file into the source tree and rebuild:\n" + " cp " + genesisPath + " pkg/core/config/genesis/prod.json\n" + " Then add it to pkg/core/config/genesis/genesis.go and rebuild the binary.\n\n" + - " 2. Start the bootstrap node with the genesis-writer output as the data dir.\n" + - " In docker-compose.yml, mount it to /data:\n" + - " volumes:\n" + - " - " + filepath.Dir(w.cfg.CMTHome) + ":/data\n" + - " The node will pick up at height " + fmt.Sprintf("%d", w.finalHeight+1) + " and begin live consensus.\n\n" + + " 2. Seed the bootstrap node (ROLLOUT.md step 4):\n" + + " - copy " + w.cfg.CMTHome + " except the files MANIFEST.json lists\n" + + " under exclude_from_bootstrap\n" + + " - pg_restore " + chainDumpDirName + " into a new database on the node\n" + + " The node will pick up at height " + next + " and begin live consensus.\n\n" + " 3. Once the bootstrap node is running, other nodes can state-sync from it:\n" + " [statesync]\n" + " enable = true\n" + " rpc_servers = \":26657,:26657\"\n" + - " trust_height = " + fmt.Sprintf("%d", w.finalHeight+1) + "\n" + - " trust_hash = \"\"\n") + " trust_height = " + next + "\n" + + " trust_hash = \"\"\n") } return nil diff --git a/dev/docker-compose.yml b/dev/docker-compose.yml index 024b384d..3cb1765f 100644 --- a/dev/docker-compose.yml +++ b/dev/docker-compose.yml @@ -224,6 +224,7 @@ services: - ./pkg/lifecycle/... - ./pkg/rewards/... - ./pkg/common/... + - ./cmd/genesis-writer/... volumes: - ./cmd:/app/cmd - ./pkg:/app/pkg From aa469af8f397db7884573e55fc39eac48dd0362c Mon Sep 17 00:00:00 2001 From: Marcus Pasell <3690498+rickyrombo@users.noreply.github.com> Date: Fri, 23 Oct 2026 00:42:22 -0700 Subject: [PATCH 2/2] fix(genesis-writer): make a failed dump recoverable with --resume The dump and manifest run after every step is checkpointed, so a failed pg_dump costs nothing but the dump. Re-running with --resume is the way back, but it did not quite work: with every step already done, no block is written, flushBlock never runs, and finalTime stays zero. writeCMTState then rewrote state.db with a zero LastBlockTime. Resume now restores the tail block's header time along with its BlockID and commit, through a resumeTail helper that reads all three from the blockstore. The emit-artifact error says the write is intact and to re-run with --resume, and the README documents the recovery. Co-Authored-By: Claude Opus 5.5 --- cmd/genesis-writer/README.md | 10 +++ cmd/genesis-writer/blockstore_resume_test.go | 38 +++++++- cmd/genesis-writer/writer.go | 91 +++++++++++++------- 3 files changed, 107 insertions(+), 32 deletions(-) diff --git a/cmd/genesis-writer/README.md b/cmd/genesis-writer/README.md index 57f03607..2cd61963 100644 --- a/cmd/genesis-writer/README.md +++ b/cmd/genesis-writer/README.md @@ -186,6 +186,16 @@ event subscriptions and 63 comments with it. The cost is that each step ends its block early, so blocks no longer pack to exactly `--max-txs-per-block`. +#### If the dump fails + +The dump and manifest run after every step is checkpointed, so a failure +there (`pg_dump` missing or killed, disk full) does not cost the write. Fix +the cause and re-run the same command with `--resume`. Every step is skipped; +the run rewrites `state.db` and `genesis.json` from the database and +blockstore (the same values as before), finds the indexes already built, and +redoes only `chain.dump` and `MANIFEST.json`. A half-written +`chain.dump.partial` from the failed attempt is removed first. + ## Indexer integration Indexers that consume the Core chain must handle `ManageEntityLegacyMigration` diff --git a/cmd/genesis-writer/blockstore_resume_test.go b/cmd/genesis-writer/blockstore_resume_test.go index 796d35d8..d7caea25 100644 --- a/cmd/genesis-writer/blockstore_resume_test.go +++ b/cmd/genesis-writer/blockstore_resume_test.go @@ -25,7 +25,11 @@ import ( // The resulting failure is indistinguishable from data loss: "commit not found" // against a blockstore that is completely intact. That is what it looked like // when an interrupted five-hour run refused to resume. -func TestSeenCommitResolvesAtTheTailAndBlockCommitDoesNot(t *testing.T) { +// saveTailBlock writes a single block at height 1 with a seen commit, the +// state resume finds at the end of a run, and returns the store, the block and +// its parts. +func saveTailBlock(t *testing.T) (*cmtstore.BlockStore, *cmttypes.Block, *cmttypes.PartSet) { + t.Helper() db, err := dbm.NewDB("blockstore", dbm.MemDBBackend, t.TempDir()) if err != nil { t.Fatalf("open blockstore: %v", err) @@ -66,6 +70,38 @@ func TestSeenCommitResolvesAtTheTailAndBlockCommitDoesNot(t *testing.T) { Signature: bytes.Repeat([]byte{2}, 64), }}, }) + return store, block, parts +} + +// A resume that writes no new blocks -- retrying a failed dump, where every +// step is already checkpointed -- never reaches flushBlock, so the tail's time +// has to come from the blockstore. Without it writeCMTState rewrote state.db +// with a zero LastBlockTime. +func TestResumeTailRestoresTheLastBlock(t *testing.T) { + store, block, parts := saveTailBlock(t) + + tail, err := resumeTail(store, store.Height()) + if err != nil { + t.Fatalf("resumeTail: %v", err) + } + if !tail.time.Equal(block.Time) { + t.Errorf("time = %v, want the tail block's header time %v", tail.time, block.Time) + } + want := cmttypes.BlockID{Hash: block.Hash(), PartSetHeader: parts.Header()} + if !tail.blockID.Equals(want) { + t.Errorf("blockID = %v, want %v", tail.blockID, want) + } + if tail.commit == nil || tail.commit.Height != block.Height { + t.Errorf("commit = %v, want the seen commit for height %d", tail.commit, block.Height) + } + + if _, err := resumeTail(store, store.Height()+1); err == nil { + t.Error("resumeTail past the tail returned no error") + } +} + +func TestSeenCommitResolvesAtTheTailAndBlockCommitDoesNot(t *testing.T) { + store, _, _ := saveTailBlock(t) tail := store.Height() if tail != 1 { diff --git a/cmd/genesis-writer/writer.go b/cmd/genesis-writer/writer.go index 46b3f223..094f6ac9 100644 --- a/cmd/genesis-writer/writer.go +++ b/cmd/genesis-writer/writer.go @@ -456,6 +456,50 @@ func (w *Writer) stepNames() []string { return names } +// blockstoreTail is the last written block's state that a resume restores. +type blockstoreTail struct { + blockID cmttypes.BlockID + commit *cmttypes.Commit + time time.Time +} + +// resumeTail reads the last written block back out of the blockstore: its +// BlockID, its commit, and its header time, which are exactly the values a +// continuous run would hold after writing it. +func resumeTail(store *cmtstore.BlockStore, height int64) (blockstoreTail, error) { + lastBlock, _ := store.LoadBlock(height) + if lastBlock == nil { + return blockstoreTail{}, fmt.Errorf("load resume block %d from blockstore: not found", height) + } + blockParts, err := lastBlock.MakePartSet(cmttypes.BlockPartSizeBytes) + if err != nil { + return blockstoreTail{}, fmt.Errorf("make part set for resume block %d: %w", height, err) + } + // LoadSeenCommit, not LoadBlockCommit. SaveBlock stores the commit it is + // handed under the SEEN-commit key for that height, and separately stores + // block.LastCommit under the commit key for height-1. So + // LoadBlockCommit(h) only returns a value once block h+1 has been saved -- + // which is never true for the last block written. + // + // Resume always lands on the last block, so this could only ever fail, + // reporting a missing commit for a blockstore that is perfectly intact. + // The error read like data loss and sent me looking for a truncated write + // that had not happened. + // + // The seen commit is also what the non-resume path assigns to w.lastCommit + // after each block, so this restores exactly the value a continuous run + // would have held at that point. + commit := store.LoadSeenCommit(height) + if commit == nil { + return blockstoreTail{}, fmt.Errorf("load resume seen commit %d from blockstore: not found", height) + } + return blockstoreTail{ + blockID: cmttypes.BlockID{Hash: lastBlock.Hash(), PartSetHeader: blockParts.Header()}, + commit: commit, + time: lastBlock.Time, + }, nil +} + func (w *Writer) Run(ctx context.Context) error { start := time.Now() w.logger.Info("starting genesis write") @@ -509,38 +553,18 @@ func (w *Writer) Run(ctx context.Context) error { if w.blockStore == nil { return fmt.Errorf("cannot resume without blockstore (--cmt-home required)") } - lastBlock, _ := w.blockStore.LoadBlock(maxHeight) - if lastBlock == nil { - return fmt.Errorf("load resume block %d from blockstore: not found", maxHeight) - } - blockParts, err := lastBlock.MakePartSet(cmttypes.BlockPartSizeBytes) + tail, err := resumeTail(w.blockStore, maxHeight) if err != nil { - return fmt.Errorf("make part set for resume block %d: %w", maxHeight, err) - } - w.prevBlockID = cmttypes.BlockID{ - Hash: lastBlock.Hash(), - PartSetHeader: blockParts.Header(), - } - // LoadSeenCommit, not LoadBlockCommit. SaveBlock stores the commit - // it is handed under the SEEN-commit key for that height, and - // separately stores block.LastCommit under the commit key for - // height-1. So LoadBlockCommit(h) only returns a value once block - // h+1 has been saved -- which is never true for the last block - // written. - // - // Resume always lands on the last block, so this could only ever - // fail, reporting a missing commit for a blockstore that is - // perfectly intact. The error read like data loss and sent me - // looking for a truncated write that had not happened. - // - // The seen commit is also what the non-resume path assigns to - // w.lastCommit after each block, so this restores exactly the value - // a continuous run would have held at that point. - lastCommit := w.blockStore.LoadSeenCommit(maxHeight) - if lastCommit == nil { - return fmt.Errorf("load resume seen commit %d from blockstore: not found", maxHeight) + return err } - w.lastCommit = lastCommit + w.prevBlockID = tail.blockID + w.lastCommit = tail.commit + // A resume that writes no new blocks (every step already done, e.g. + // retrying a failed dump) never reaches flushBlock, which is what + // sets finalTime. writeCMTState still runs and stores it as the + // state's LastBlockTime, so restore it here or state.db is rewritten + // with the zero time. + w.finalTime = tail.time w.logger.Info("resuming from height", zap.Int64("height", w.height)) } @@ -718,7 +742,12 @@ func (w *Writer) Run(ctx context.Context) error { // runs before this process exits so nothing else has connected first. if w.cfg.ArtifactDir != "" && w.cfg.CMTHome != "" && w.finalHeight > 0 { if err := w.emitArtifact(ctx); err != nil { - return fmt.Errorf("emit artifact: %w", err) + // Everything before this point is checkpointed, so a failed dump + // does not cost the write. Say so: the bare error reads like the + // whole run failed. + return fmt.Errorf("emit artifact: %w\n\nThe chain write is complete and intact; only the dump or manifest failed. "+ + "Fix the cause and re-run the same command with --resume: every step is skipped, and it "+ + "redoes only the CometBFT state, the index check, the dump and %s", err, manifestFileName) } }