From 4f86b1f4d6fc9955fb8efa56dbe74a0acde5a1be Mon Sep 17 00:00:00 2001 From: Ljubisa Gacevic Date: Mon, 31 Aug 2026 16:12:35 +0200 Subject: [PATCH 01/12] docs: spec for the verify snapshot regression gate Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_014WoaRY6SZi2gUPPmQEhJgr --- .../2026-08-31-verify-snapshot-design.md | 140 ++++++++++++++++++ 1 file changed, 140 insertions(+) create mode 100644 docs/superpowers/specs/2026-08-31-verify-snapshot-design.md diff --git a/docs/superpowers/specs/2026-08-31-verify-snapshot-design.md b/docs/superpowers/specs/2026-08-31-verify-snapshot-design.md new file mode 100644 index 0000000..3a7fb8e --- /dev/null +++ b/docs/superpowers/specs/2026-08-31-verify-snapshot-design.md @@ -0,0 +1,140 @@ +# `batch-export verify` — snapshot regression gate + +- **Date:** 2026-08-31 +- **Status:** proposed +- **Scope:** one subcommand + its package, and one workflow step wiring it in + +## Problem + +The Batch Sync workflow resumes the postage-batch snapshot from a tag of +[ethersphere/batch-archive](https://github.com/ethersphere/batch-archive) and +publishes the refreshed file back as a commit on `main` plus the next patch +tag. Nothing verifies that the refreshed snapshot is a strict extension of the +one it resumed from. A regression in batch-export — in the resume logic, the +slim NDJSON encoding, or the gzip append handling — could silently drop, +mutate, or reorder historical entries, and the workflow would tag and publish +the corrupted snapshot as the new latest version. + +Separately, the workflow extracts the last block number for the commit title +with a `gunzip | tail | sed` pipeline whose regex re-encodes the slim JSON +shape by hand. Nothing ties that regex to the Go struct that defines the +format, so a format change passes every Go test and breaks only in CI. + +## Goal + +A `batch-export verify` subcommand that: + +1. proves the new snapshot contains everything the old one did, byte for byte + and in order, with only new entries appended; +2. validates the appended entries parse and are correctly ordered; +3. prints the new snapshot's last block number, replacing the sed pipeline. + +Wired into the Batch Sync workflow between the Export and Publish steps, so a +failed verification blocks the commit and the tag. + +## Non-goals + +- No chain-side validation: `verify` never queries an RPC endpoint. +- No repair: a bad snapshot is reported, never fixed. +- No comparison against batch-archive `main`'s current file: the contract is + old-input vs. new-output of one export run. (The workflow always resumes + the file it just fetched, so that is the meaningful pair.) +- No change to export behavior, scheduling, or bootstrap flows. + +## CLI contract + +``` +batch-export verify --old --new [--verbosity ] +``` + +- `--old` (required): the snapshot the export resumed from — in the workflow, + the file fetched from the batch-archive tag. +- `--new` (required): the freshly exported snapshot. +- Formats: plain NDJSON or gzip, each file detected independently by content + (the same detection `--resume` uses); any old/new combination is accepted. +- **stdout** on success: the new snapshot's last block number, decimal, one + line, nothing else — safe for `$(...)` capture in the workflow. +- **stderr**: all logging and diagnostics. +- **Exit code**: 0 = verified; non-zero = do not publish (verification + failure, I/O error, and parse error are deliberately not distinguished — + the workflow reacts identically to all of them). + +## Verification algorithm + +1. **Cursor of the old file** via the existing `resume.Read`: yields the last + complete entry's `(blockNumber, logIndex)`, the compressed flag, and the + clean size. `resume.Read`'s refusals fail verification with the same + meaning they have on resume: `ErrNotAnExport` (file was altered) and + `ErrNoLogs` (nothing to extend). +2. **Prefix property.** The new file's decompressed content must begin, byte + for byte, with the old file's *clean content* — the same bytes + `resume.PrepareOutput` copies (for a gzip old file: the compressed bytes up + to the cursor's clean size, decompressed; for plain NDJSON: the file up to + clean size). Comparison is streamed with constant memory; files are never + loaded whole. If the old file carries a truncated tail past its clean + size, the tail is excluded from the comparison and a warning is logged — + mirroring what resume itself discards. +3. **Appended tail.** Every line of the new file past the prefix must: + - stay under the same line-length cap resume enforces; + - parse as an export log entry (slim or full shape, as resume accepts); + - be strictly increasing in `(blockNumber, logIndex)` order, with the + first appended entry strictly greater than the old file's cursor — + matching resume's skip semantics, which re-query the cursor block but + never re-write entries at or before the cursor. +4. **Empty tail is valid**: a run that found no new logs verifies + successfully and prints the old cursor's block number. +5. **Output**: the last appended entry's block number (or the old cursor's, + when the tail is empty), printed in decimal on stdout. + +Failure messages name the location: byte offset and line number of the first +divergence for a prefix mismatch; line number and reason for a tail failure. + +## Code layout + +``` +cmd/verify.go thin cobra command: flags, logger, calls pkg/verify, + prints the block number (mirrors cmd/export.go style) +pkg/verify/verify.go Verify(oldPath, newPath) (Result, error); + Result{LastBlock uint64, Appended int} +pkg/verify/verify_test.go unit tests with generated fixture files +``` + +`pkg/verify` reuses `pkg/resume` for cursor reading and format handling. If a +helper it needs (e.g. the decompressing opener) is unexported in `pkg/resume`, +the minimal piece is exported from there rather than duplicated. + +## Workflow integration (same PR) + +`.github/workflows/batch-sync.yml` gains one step between Export and Publish: + +```yaml + - name: Verify snapshot + run: | + set -euo pipefail + last_block="$(./dist/batch-export verify --old resume.ndjson.gzip --new snapshot.ndjson.gzip)" + echo "LAST_BLOCK=${last_block}" >> "${GITHUB_ENV}" +``` + +The Publish step then uses `${LAST_BLOCK}` in the commit title and drops the +`gunzip | tail | sed` block together with its empty-result guard. A verify +failure stops the job before any commit or tag is created. + +## Testing + +- **Unit (`pkg/verify`)**, fixtures generated per test: identical files; + proper superset; corrupted byte inside the prefix; new file shorter than + old; non-monotonic appended entry; duplicate of the cursor entry; malformed + JSON tail line; old file with no complete entry; old file with a truncated + tail; plain/gzip in all four combinations. +- **Command (`cmd`)**: exit code and stdout shape for one passing and one + failing pair. +- **Workflow**: `actionlint` locally; end-to-end via a manual dispatch after + merge. + +Development is test-first (TDD), consistent with the repo's existing +`pkg/resume` test style. + +## Delivery + +Branch `feat/verify-snapshot` off `main`; conventional commits (`feat:` for +the subcommand, `ci:` for the workflow wiring); one PR to `main`. From e41cd0f524c3ecceccafc8a0e588b81b2e3ad70e Mon Sep 17 00:00:00 2001 From: Ljubisa Gacevic Date: Mon, 31 Aug 2026 16:44:03 +0200 Subject: [PATCH 02/12] docs: implementation plan for the verify snapshot gate Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_014WoaRY6SZi2gUPPmQEhJgr --- .../plans/2026-08-31-verify-snapshot.md | 992 ++++++++++++++++++ 1 file changed, 992 insertions(+) create mode 100644 docs/superpowers/plans/2026-08-31-verify-snapshot.md diff --git a/docs/superpowers/plans/2026-08-31-verify-snapshot.md b/docs/superpowers/plans/2026-08-31-verify-snapshot.md new file mode 100644 index 0000000..39ba7ba --- /dev/null +++ b/docs/superpowers/plans/2026-08-31-verify-snapshot.md @@ -0,0 +1,992 @@ +# `batch-export verify` Implementation Plan + +> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. + +**Goal:** A `verify` subcommand that proves a refreshed snapshot strictly extends the snapshot it was resumed from, printing the new last block number — wired into the Batch Sync workflow as a gate before publishing. + +**Architecture:** `pkg/verify` compares the two files using `pkg/resume`'s existing cursor/format machinery (three small helpers get exported from `resume` first). `cmd/verify.go` is a thin cobra wrapper. The workflow gains a `Verify snapshot` step whose stdout replaces the current `gunzip | tail | sed` block-number extraction. + +**Tech Stack:** Go (stdlib only — no new dependencies), cobra (already used), `github.com/ethersphere/bee/v2/pkg/log` (already used; its default sink is os.Stderr, so stdout stays clean). + +**Spec:** `docs/superpowers/specs/2026-08-31-verify-snapshot-design.md` + +## Global Constraints + +- Module path is `github.com/ethersphere/batch-export`; work happens on branch `feat/verify-snapshot`. +- Conventional commits (`feat:`, `test:`, `ci:`, `docs:`). +- stdout of `verify` carries ONLY the decimal last block number and a newline; everything else goes to stderr via the logger. Non-zero exit = do not publish. +- Streaming with constant memory: never load a snapshot whole (they are tens of MB). +- No behavior change to `export`, `resume`, or the workflow's publish semantics beyond swapping the block-number source. +- Run tests with `go test ./...` and `go vet ./...`; both must pass before every commit. +- Comment style: sparse, constraint-stating, matching `pkg/resume` (read it first). + +--- + +### Task 1: Export `MaxLineBytes`, `ParseEntry`, and `OpenClean` from `pkg/resume` + +`pkg/verify` needs three things resume already knows: the line-length cap, how to parse one NDJSON line into `(blockNumber, logIndex)`, and how to read a file's decompressed clean content. Export them here so verify never re-encodes the on-disk format. + +**Files:** +- Modify: `pkg/resume/resume.go` +- Test: `pkg/resume/resume_test.go` (append new test functions; its existing helpers `testLog`, `ndjson`, `gz`, `truncateLast`, `write`, and constants `plainFile`, `gzipFile` are reused) + +**Interfaces:** +- Consumes: existing `resume.Read`, internal `parseCursor`, internal `maxLineBytes`. +- Produces (later tasks rely on these exact signatures): + - `const MaxLineBytes = 1 << 20` (renamed from `maxLineBytes`) + - `func ParseEntry(line []byte) (*Cursor, error)` + - `func OpenClean(path string, c *Cursor) (io.ReadCloser, error)` + +- [ ] **Step 1: Write the failing tests** + +Append to `pkg/resume/resume_test.go`: + +```go +func TestParseEntry(t *testing.T) { + t.Parallel() + + cursor, err := resume.ParseEntry(ndjson(t, testLog(7, 3))) + if err != nil { + t.Fatalf("ParseEntry: %v", err) + } + if cursor.BlockNumber != 7 || cursor.LogIndex != 3 { + t.Fatalf("got (%d, %d), want (7, 3)", cursor.BlockNumber, cursor.LogIndex) + } + + if _, err := resume.ParseEntry([]byte(`{"foo":1}`)); err == nil { + t.Fatal("want an error for a line missing blockNumber and logIndex") + } +} + +func TestOpenClean(t *testing.T) { + t.Parallel() + + content := ndjson(t, testLog(1, 0), testLog(2, 0)) + + tests := []struct { + name string + file string + blob []byte + }{ + {nameFormatPlain, plainFile, content}, + {nameFormatGzip, gzipFile, gz(t, content)}, + {"plain with interrupted tail", plainFile, append(slices.Clone(content), `{"blockNu`...)}, + {"gzip with interrupted tail", gzipFile, truncateLast(append(gz(t, content), gz(t, ndjson(t, testLog(3, 0)))...), 4)}, + } + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + + path := write(t, tc.file, tc.blob) + cursor, err := resume.Read(path) + if err != nil { + t.Fatalf("Read: %v", err) + } + + r, err := resume.OpenClean(path, cursor) + if err != nil { + t.Fatalf("OpenClean: %v", err) + } + defer r.Close() + + got, err := io.ReadAll(r) + if err != nil { + t.Fatalf("reading clean content: %v", err) + } + if !bytes.Equal(got, content) { + t.Fatalf("clean content mismatch:\ngot %q\nwant %q", got, content) + } + }) + } +} +``` + +Notes for the implementer: `ndjson` returns newline-terminated lines and `ParseEntry` trims whitespace, so passing a full line works. The "gzip with interrupted tail" case appends a second gzip member and cuts 4 bytes off its trailer — `Read` then reports `CleanSize` at the first member's boundary, and `OpenClean` must return only that member's content. Check the file's existing imports; `slices` and `io` are already imported. + +- [ ] **Step 2: Run the tests to verify they fail** + +Run: `go test ./pkg/resume/ -run 'TestParseEntry|TestOpenClean' -v` +Expected: compile error — `undefined: resume.ParseEntry` and `undefined: resume.OpenClean`. + +- [ ] **Step 3: Implement in `pkg/resume/resume.go`** + +Rename the constant (find every use: `grep -n maxLineBytes pkg/resume/resume.go` — it appears only in this file) and update its comment: + +```go +const ( + // MaxLineBytes caps a single line. Exported log lines run to a few + // hundred bytes, so anything longer is not this tool's output. + MaxLineBytes = 1 << 20 + // bufferSize is how much of a gzip file is buffered per read. + bufferSize = 64 * 1024 +) +``` + +Add at the end of the file: + +```go +// ParseEntry builds a cursor from a single exported NDJSON line, accepting +// both the slim and the full log shape. It is the only place outside the +// writer where the on-disk field names are known, so consumers never +// re-encode them. +func ParseEntry(line []byte) (*Cursor, error) { + return parseCursor(line) +} + +// OpenClean returns the decompressed clean content of the export at path — +// the same bytes PrepareOutput would carry over. c must come from Read on +// the same, unmodified file. +func OpenClean(path string, c *Cursor) (io.ReadCloser, error) { + file, err := os.Open(path) + if err != nil { + return nil, fmt.Errorf("error opening export file: %w", err) + } + + limited := io.LimitReader(file, c.CleanSize) + if !c.Compressed { + return &cleanReader{Reader: limited, closer: file}, nil + } + + gz, err := gzip.NewReader(bufio.NewReaderSize(limited, bufferSize)) + if err != nil { + file.Close() + return nil, fmt.Errorf("error reading gzip export file: %w", err) + } + + return &cleanReader{Reader: gz, closer: file}, nil +} + +// cleanReader pairs the decompressed stream with the file it draws from. +type cleanReader struct { + io.Reader + closer io.Closer +} + +func (r *cleanReader) Close() error { return r.closer.Close() } +``` + +The gzip path relies on `gzip.Reader`'s default multistream mode to walk every member inside the `CleanSize` limit; `CleanSize` always ends on a member boundary, so the limited stream ends cleanly. + +- [ ] **Step 4: Run the full package tests** + +Run: `go test ./pkg/resume/ -v` and `go vet ./...` +Expected: PASS (the rename must not have missed a reference; `go vet` is the catch-all). + +- [ ] **Step 5: Commit** + +```bash +git add pkg/resume/resume.go pkg/resume/resume_test.go +git commit -m "feat: export clean-content helpers from pkg/resume" +``` + +--- + +### Task 2: `pkg/verify` — core check, happy paths + +**Files:** +- Create: `pkg/verify/verify.go` +- Create: `pkg/verify/verify_test.go` + +**Interfaces:** +- Consumes: `resume.Read`, `resume.OpenClean`, `resume.ParseEntry`, `resume.MaxLineBytes`, `resume.Cursor` (Task 1). +- Produces (Tasks 3–5 rely on these exact names): + - `type Result struct { LastBlock uint64; Appended int; OldTruncated bool }` + - `func Verify(oldPath, newPath string) (Result, error)` + - `var ErrMismatch, ErrOrder, ErrTruncatedNew error` + +- [ ] **Step 1: Write the failing tests** + +Create `pkg/verify/verify_test.go`: + +```go +package verify_test + +import ( + "bytes" + "compress/gzip" + "encoding/json" + "os" + "path/filepath" + "testing" + + "github.com/ethereum/go-ethereum/common" + "github.com/ethereum/go-ethereum/core/types" + "github.com/ethersphere/batch-export/pkg/filestore" + "github.com/ethersphere/batch-export/pkg/verify" +) + +// testLog builds a log shaped like the ones the exporter writes. +func testLog(blockNumber uint64, logIndex uint) types.Log { + return types.Log{ + Address: common.HexToAddress("0x45a1502382541cd610cc9068e88727426b696293"), + Topics: []common.Hash{common.HexToHash("0xae46785019700e30375a5d7b4f91e32f8060ef085111f896ebf889450aa2ab5a")}, + Data: bytes.Repeat([]byte{0xab}, 32), + BlockNumber: blockNumber, + TxHash: common.HexToHash("0xb08f07656eaafa8efc458e2aa90773648d95ec8119873d212b4377dea5190cc0"), + Index: logIndex, + } +} + +// ndjson renders logs the way the exporter's slim format does. +func ndjson(t *testing.T, logs ...types.Log) []byte { + t.Helper() + + var buf bytes.Buffer + enc := json.NewEncoder(&buf) + for _, l := range logs { + if err := enc.Encode(filestore.NewSlimLog(l)); err != nil { + t.Fatal(err) + } + } + + return buf.Bytes() +} + +// gz compresses b as a single gzip member. +func gz(t *testing.T, b []byte) []byte { + t.Helper() + + var buf bytes.Buffer + w := gzip.NewWriter(&buf) + if _, err := w.Write(b); err != nil { + t.Fatal(err) + } + if err := w.Close(); err != nil { + t.Fatal(err) + } + + return buf.Bytes() +} + +// write puts content into a fresh temp file and returns its path. +func write(t *testing.T, name string, content []byte) string { + t.Helper() + + path := filepath.Join(t.TempDir(), name) + if err := os.WriteFile(path, content, 0o644); err != nil { + t.Fatal(err) + } + + return path +} + +func TestVerifySuperset(t *testing.T) { + t.Parallel() + + // The new snapshot is the old member plus an appended member, exactly + // the shape a resumed gzip export produces. + oldBlob := gz(t, ndjson(t, testLog(1, 0), testLog(2, 0))) + newBlob := append(bytes.Clone(oldBlob), gz(t, ndjson(t, testLog(2, 1), testLog(3, 0)))...) + + res, err := verify.Verify( + write(t, "old.ndjson.gzip", oldBlob), + write(t, "new.ndjson.gzip", newBlob), + ) + if err != nil { + t.Fatalf("Verify: %v", err) + } + if res.LastBlock != 3 { + t.Errorf("LastBlock = %d, want 3", res.LastBlock) + } + if res.Appended != 2 { + t.Errorf("Appended = %d, want 2", res.Appended) + } + if res.OldTruncated { + t.Error("OldTruncated = true, want false") + } +} + +func TestVerifyIdentical(t *testing.T) { + t.Parallel() + + blob := gz(t, ndjson(t, testLog(1, 0), testLog(2, 0))) + + res, err := verify.Verify( + write(t, "old.ndjson.gzip", blob), + write(t, "new.ndjson.gzip", blob), + ) + if err != nil { + t.Fatalf("Verify: %v", err) + } + if res.LastBlock != 2 || res.Appended != 0 { + t.Errorf("got LastBlock=%d Appended=%d, want 2 and 0", res.LastBlock, res.Appended) + } +} +``` + +- [ ] **Step 2: Run the tests to verify they fail** + +Run: `go test ./pkg/verify/ -v` +Expected: compile error — package `verify` does not exist. + +- [ ] **Step 3: Implement `pkg/verify/verify.go`** + +```go +// Package verify checks that a refreshed snapshot is a strict extension of +// the snapshot it was resumed from: everything the old file held, byte for +// byte and in order, followed only by newer entries. +package verify + +import ( + "bufio" + "bytes" + "errors" + "fmt" + "io" + + "github.com/ethersphere/batch-export/pkg/resume" +) + +var ( + // ErrMismatch indicates the new snapshot does not begin with the old + // snapshot's content: an entry was dropped, mutated, or reordered. + ErrMismatch = errors.New("new snapshot does not extend the old snapshot") + // ErrOrder indicates an appended entry out of (blockNumber, logIndex) + // order relative to what precedes it. + ErrOrder = errors.New("appended entries out of order") + // ErrTruncatedNew indicates the new snapshot ends in an interrupted + // write; a completed export never does. + ErrTruncatedNew = errors.New("new snapshot ends in an interrupted write") +) + +// chunkSize is how many bytes of both snapshots are compared per read. +const chunkSize = 64 * 1024 + +// Result reports what a successful verification established. +type Result struct { + // LastBlock is the block number of the new snapshot's final entry. + LastBlock uint64 + // Appended is how many entries the new snapshot adds. + Appended int + // OldTruncated reports whether the old snapshot carried an interrupted + // tail, excluded from the comparison the way resuming excludes it from + // the copy. + OldTruncated bool +} + +// Verify checks that the snapshot at newPath extends the one at oldPath. +func Verify(oldPath, newPath string) (Result, error) { + oldCursor, err := resume.Read(oldPath) + if err != nil { + return Result{}, fmt.Errorf("old snapshot: %w", err) + } + newCursor, err := resume.Read(newPath) + if err != nil { + return Result{}, fmt.Errorf("new snapshot: %w", err) + } + if newCursor.Truncated { + return Result{}, ErrTruncatedNew + } + + oldClean, err := resume.OpenClean(oldPath, oldCursor) + if err != nil { + return Result{}, fmt.Errorf("old snapshot: %w", err) + } + defer oldClean.Close() + + newClean, err := resume.OpenClean(newPath, newCursor) + if err != nil { + return Result{}, fmt.Errorf("new snapshot: %w", err) + } + defer newClean.Close() + + newBuffered := bufio.NewReaderSize(newClean, chunkSize) + if err := comparePrefix(oldClean, newBuffered); err != nil { + return Result{}, err + } + + appended, err := checkTail(newBuffered, oldCursor) + if err != nil { + return Result{}, err + } + + return Result{ + LastBlock: newCursor.BlockNumber, + Appended: appended, + OldTruncated: oldCursor.Truncated, + }, nil +} + +// comparePrefix requires every byte of oldR to appear at the start of newR. +func comparePrefix(oldR io.Reader, newR *bufio.Reader) error { + oldBuf := make([]byte, chunkSize) + newBuf := make([]byte, chunkSize) + + var offset int64 + for { + n, err := oldR.Read(oldBuf) + if n > 0 { + if _, rerr := io.ReadFull(newR, newBuf[:n]); rerr != nil { + return fmt.Errorf("%w: new snapshot ends at byte %d of the old content", ErrMismatch, offset) + } + if !bytes.Equal(oldBuf[:n], newBuf[:n]) { + return fmt.Errorf("%w: content diverges within bytes %d..%d of the old content", ErrMismatch, offset, offset+int64(n)) + } + offset += int64(n) + } + if errors.Is(err, io.EOF) { + return nil + } + if err != nil { + return fmt.Errorf("error reading old snapshot: %w", err) + } + } +} + +// checkTail walks the entries the new snapshot appends after the old +// content, requiring each to advance the (blockNumber, logIndex) order. +func checkTail(r *bufio.Reader, prev *resume.Cursor) (int, error) { + last := *prev + + var ( + appended int + lineNum int + line []byte + ) + for { + chunk, err := r.ReadSlice('\n') + line = append(line, chunk...) + if len(line) > resume.MaxLineBytes { + return 0, fmt.Errorf("appended line %d exceeds %d bytes", lineNum+1, resume.MaxLineBytes) + } + + switch { + case errors.Is(err, bufio.ErrBufferFull): + continue + case errors.Is(err, io.EOF): + if len(line) > 0 { + return 0, fmt.Errorf("appended line %d ends without a newline", lineNum+1) + } + return appended, nil + case err != nil: + return 0, fmt.Errorf("error reading new snapshot: %w", err) + } + + lineNum++ + entry, err := resume.ParseEntry(line) + if err != nil { + return 0, fmt.Errorf("appended line %d is not a log entry: %w", lineNum, err) + } + if !after(entry, &last) { + return 0, fmt.Errorf("%w: appended line %d holds (block %d, log %d) after (block %d, log %d)", + ErrOrder, lineNum, entry.BlockNumber, entry.LogIndex, last.BlockNumber, last.LogIndex) + } + last = *entry + appended++ + line = line[:0] + } +} + +// after reports whether entry strictly follows prev in log order. +func after(entry, prev *resume.Cursor) bool { + if entry.BlockNumber != prev.BlockNumber { + return entry.BlockNumber > prev.BlockNumber + } + return entry.LogIndex > prev.LogIndex +} +``` + +- [ ] **Step 4: Run the tests to verify they pass** + +Run: `go test ./pkg/verify/ -v` and `go vet ./...` +Expected: PASS. + +- [ ] **Step 5: Commit** + +```bash +git add pkg/verify/ +git commit -m "feat: add pkg/verify snapshot extension check" +``` + +--- + +### Task 3: `pkg/verify` — failure modes and truncation handling + +The implementation from Task 2 should already handle these; this task pins each failure mode with a test and fixes anything that surfaces. + +**Files:** +- Modify: `pkg/verify/verify_test.go` (append) +- Modify: `pkg/verify/verify.go` (only if a test exposes a gap) + +**Interfaces:** +- Consumes: everything Task 2 produced, plus `resume.ErrNoLogs`. + +- [ ] **Step 1: Write the tests** + +Append to `pkg/verify/verify_test.go` (add `"slices"`, `"errors"`, and `"github.com/ethersphere/batch-export/pkg/resume"` to the imports): + +```go +func TestVerifyRefusals(t *testing.T) { + t.Parallel() + + oldContent := ndjson(t, testLog(1, 0), testLog(2, 0)) + + // mutated flips one byte inside the first entry's blockNumber. + mutated := bytes.Replace(slices.Clone(oldContent), []byte(`"blockNumber":"0x1"`), []byte(`"blockNumber":"0x9"`), 1) + + tests := []struct { + name string + old []byte + new []byte + wantErr error + }{ + { + name: "mutated entry inside the old content", + old: gz(t, oldContent), + new: gz(t, append(mutated, ndjson(t, testLog(10, 0))...)), + wantErr: verify.ErrMismatch, + }, + { + name: "dropped entry", + old: gz(t, oldContent), + new: gz(t, ndjson(t, testLog(1, 0), testLog(3, 0))), + wantErr: verify.ErrMismatch, + }, + { + name: "new shorter than old", + old: gz(t, oldContent), + new: gz(t, ndjson(t, testLog(1, 0))), + wantErr: verify.ErrMismatch, + }, + { + name: "appended entry repeats the cursor", + old: gz(t, oldContent), + new: append(gz(t, oldContent), gz(t, ndjson(t, testLog(2, 0)))...), + wantErr: verify.ErrOrder, + }, + { + name: "appended entries out of order", + old: gz(t, oldContent), + new: append(gz(t, oldContent), gz(t, ndjson(t, testLog(5, 0), testLog(4, 0)))...), + wantErr: verify.ErrOrder, + }, + { + name: "new ends in an interrupted write", + old: oldContent, + new: append(slices.Clone(oldContent), `{"blockNumber":"0x3"`...), + wantErr: verify.ErrTruncatedNew, + }, + { + name: "old holds no complete entry", + old: nil, + new: oldContent, + wantErr: resume.ErrNoLogs, + }, + } + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + + _, err := verify.Verify( + write(t, "old", tc.old), + write(t, "new", tc.new), + ) + if !errors.Is(err, tc.wantErr) { + t.Fatalf("Verify error = %v, want %v", err, tc.wantErr) + } + }) + } +} + +func TestVerifyMalformedAppendedLine(t *testing.T) { + t.Parallel() + + oldContent := ndjson(t, testLog(1, 0), testLog(2, 0)) + // The middle appended line is valid JSON but not a log entry; resume.Read + // on a plain file only inspects the tail, so only checkTail can catch it. + newContent := append(slices.Clone(oldContent), "{\"foo\":1}\n"...) + newContent = append(newContent, ndjson(t, testLog(3, 0))...) + + _, err := verify.Verify( + write(t, "old", oldContent), + write(t, "new", newContent), + ) + if err == nil { + t.Fatal("want an error for a malformed appended line") + } +} + +func TestVerifyOldTruncated(t *testing.T) { + t.Parallel() + + oldContent := ndjson(t, testLog(1, 0), testLog(2, 0)) + oldBlob := append(slices.Clone(oldContent), `{"block`...) + newBlob := append(slices.Clone(oldContent), ndjson(t, testLog(3, 0))...) + + res, err := verify.Verify( + write(t, "old", oldBlob), + write(t, "new", newBlob), + ) + if err != nil { + t.Fatalf("Verify: %v", err) + } + if !res.OldTruncated { + t.Error("OldTruncated = false, want true") + } + if res.LastBlock != 3 || res.Appended != 1 { + t.Errorf("got LastBlock=%d Appended=%d, want 3 and 1", res.LastBlock, res.Appended) + } +} +``` + +- [ ] **Step 2: Run the tests** + +Run: `go test ./pkg/verify/ -v` +Expected: PASS if Task 2's implementation is complete. Any FAIL marks a real gap — fix `verify.go` minimally until green, and keep the fix inside the semantics the spec defines. + +- [ ] **Step 3: Commit** + +```bash +git add pkg/verify/ +git commit -m "test: cover verify failure modes and truncation handling" +``` + +--- + +### Task 4: `pkg/verify` — format combination matrix + +**Files:** +- Modify: `pkg/verify/verify_test.go` (append) + +**Interfaces:** +- Consumes: everything from Tasks 2–3. + +- [ ] **Step 1: Write the test** + +```go +func TestVerifyFormatCombinations(t *testing.T) { + t.Parallel() + + oldContent := ndjson(t, testLog(1, 0), testLog(2, 0)) + newContent := append(slices.Clone(oldContent), ndjson(t, testLog(3, 0))...) + + tests := []struct { + name string + old []byte + new []byte + }{ + {"plain old, plain new", oldContent, newContent}, + {"plain old, gzip new", oldContent, gz(t, newContent)}, + {"gzip old, plain new", gz(t, oldContent), newContent}, + {"gzip old, gzip new", gz(t, oldContent), gz(t, newContent)}, + } + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + + res, err := verify.Verify( + write(t, "old", tc.old), + write(t, "new", tc.new), + ) + if err != nil { + t.Fatalf("Verify: %v", err) + } + if res.LastBlock != 3 || res.Appended != 1 { + t.Errorf("got LastBlock=%d Appended=%d, want 3 and 1", res.LastBlock, res.Appended) + } + }) + } +} +``` + +- [ ] **Step 2: Run the test** + +Run: `go test ./pkg/verify/ -v` +Expected: PASS (the format handling all lives in `resume`; a FAIL here means `OpenClean` or the buffering broke a combination — fix there, not with special cases in verify). + +- [ ] **Step 3: Commit** + +```bash +git add pkg/verify/verify_test.go +git commit -m "test: cover verify across plain and gzip snapshot combinations" +``` + +--- + +### Task 5: `verify` subcommand and README + +**Files:** +- Create: `cmd/verify.go` +- Create: `cmd/verify_test.go` +- Modify: `cmd/cmd.go` (register the command in `newCommand`) +- Modify: `README.md` (new subsection after "Continuing a previous snapshot") + +**Interfaces:** +- Consumes: `verify.Verify`, `verify.Result` (Task 2); the `command` struct and `c.log` from `cmd/cmd.go`. +- Produces: `batch-export verify --old --new ` printing the last block in decimal on stdout; `func (c *command) initVerifyCmd() error`. + +- [ ] **Step 1: Write the failing tests** + +Create `cmd/verify_test.go` (internal test — `package cmd`, like `export_test.go`): + +```go +package cmd + +import ( + "bytes" + "compress/gzip" + "context" + "encoding/json" + "os" + "path/filepath" + "testing" + + "github.com/ethereum/go-ethereum/core/types" + "github.com/ethersphere/batch-export/pkg/filestore" +) + +// writeSnapshot writes a slim-format gzip snapshot with one entry per +// (blockNumber, logIndex) pair and returns its path. +func writeSnapshot(t *testing.T, name string, entries ...[2]uint64) string { + t.Helper() + + var buf bytes.Buffer + w := gzip.NewWriter(&buf) + enc := json.NewEncoder(w) + for _, e := range entries { + if err := enc.Encode(filestore.NewSlimLog(types.Log{BlockNumber: e[0], Index: uint(e[1])})); err != nil { + t.Fatal(err) + } + } + if err := w.Close(); err != nil { + t.Fatal(err) + } + + path := filepath.Join(t.TempDir(), name) + if err := os.WriteFile(path, buf.Bytes(), 0o644); err != nil { + t.Fatal(err) + } + + return path +} + +func TestVerifyCmd(t *testing.T) { + t.Parallel() + + oldPath := writeSnapshot(t, "old.ndjson.gzip", [2]uint64{1, 0}, [2]uint64{2, 0}) + newPath := writeSnapshot(t, "new.ndjson.gzip", [2]uint64{1, 0}, [2]uint64{2, 0}, [2]uint64{3, 0}) + + c, err := newCommand() + if err != nil { + t.Fatal(err) + } + var out bytes.Buffer + c.root.SetOut(&out) + c.root.SetArgs([]string{"verify", "--old", oldPath, "--new", newPath}) + + if err := c.Execute(context.Background()); err != nil { + t.Fatalf("verify: %v", err) + } + if got, want := out.String(), "3\n"; got != want { + t.Errorf("stdout = %q, want %q", got, want) + } +} + +func TestVerifyCmdRefusal(t *testing.T) { + t.Parallel() + + oldPath := writeSnapshot(t, "old.ndjson.gzip", [2]uint64{1, 0}, [2]uint64{2, 0}) + newPath := writeSnapshot(t, "new.ndjson.gzip", [2]uint64{1, 0}, [2]uint64{3, 0}) + + c, err := newCommand() + if err != nil { + t.Fatal(err) + } + var out bytes.Buffer + c.root.SetOut(&out) + c.root.SetArgs([]string{"verify", "--old", oldPath, "--new", newPath}) + + if err := c.Execute(context.Background()); err == nil { + t.Fatal("want an error for a snapshot that drops an entry") + } + if out.Len() != 0 { + t.Errorf("stdout = %q, want empty", out.String()) + } +} +``` + +- [ ] **Step 2: Run the tests to verify they fail** + +Run: `go test ./cmd/ -run TestVerifyCmd -v` +Expected: FAIL — cobra reports `unknown command "verify"` (surfaced as an `Execute` error in the first test). + +- [ ] **Step 3: Implement `cmd/verify.go` and register it** + +Create `cmd/verify.go`: + +```go +package cmd + +import ( + "fmt" + + "github.com/ethersphere/batch-export/pkg/verify" + "github.com/spf13/cobra" +) + +func (c *command) initVerifyCmd() error { + var ( + oldFile string + newFile string + ) + + cmd := &cobra.Command{ + Use: "verify", + Short: "Verify that a refreshed snapshot extends the one it was resumed from", + Long: `Verifies that the snapshot at --new holds everything the snapshot at --old does, +byte for byte and in order, followed only by newer entries. On success the new +snapshot's last block number is printed to stdout in decimal; any failure exits +non-zero with nothing on stdout.`, + RunE: func(cmd *cobra.Command, args []string) error { + result, err := verify.Verify(oldFile, newFile) + if err != nil { + return err + } + + if result.OldTruncated { + c.log.Warning("old snapshot ends in an interrupted write; the truncated tail was excluded from the comparison", "oldFile", oldFile) + } + c.log.Info("snapshot verified", "oldFile", oldFile, "newFile", newFile, "appended", result.Appended, "lastBlock", result.LastBlock) + + _, err = fmt.Fprintln(cmd.OutOrStdout(), result.LastBlock) + return err + }, + } + + cmd.Flags().StringVar(&oldFile, "old", "", "Snapshot the export resumed from (.ndjson, .gz or .gzip)") + cmd.Flags().StringVar(&newFile, "new", "", "Freshly exported snapshot to check (.ndjson, .gz or .gzip)") + for _, name := range []string{"old", "new"} { + if err := cmd.MarkFlagRequired(name); err != nil { + return err + } + } + + c.root.AddCommand(cmd) + + return nil +} +``` + +In `cmd/cmd.go`, register it right after the export command: + +```go + if err := c.initExportCmd(); err != nil { + return nil, err + } + + if err := c.initVerifyCmd(); err != nil { + return nil, err + } +``` + +- [ ] **Step 4: Run the tests to verify they pass** + +Run: `go test ./cmd/ -v` and `go test ./...` and `go vet ./...` +Expected: PASS. + +- [ ] **Step 5: Add the README section** + +In `README.md`, directly after the "Continuing a previous snapshot" section's content (before the next `##` heading), add: + +```markdown +### Verifying a snapshot + +`verify` proves a refreshed snapshot still holds everything the original did — +byte for byte, in order — followed only by newer entries, and prints the new +snapshot's last block number in decimal: + +```sh +batch-export verify --old export.ndjson.gzip --new snapshot.ndjson.gzip +``` + +A non-zero exit means the new snapshot must not replace the old one. +``` + +- [ ] **Step 6: Commit** + +```bash +git add cmd/verify.go cmd/verify_test.go cmd/cmd.go README.md +git commit -m "feat: add verify command gating snapshot publication" +``` + +--- + +### Task 6: Workflow wiring and PR + +**Files:** +- Modify: `.github/workflows/batch-sync.yml` + +**Interfaces:** +- Consumes: the `verify` subcommand's stdout contract (Task 5); the workflow's existing `resume.ndjson.gzip` / `snapshot.ndjson.gzip` step outputs. +- Produces: `LAST_BLOCK` in `GITHUB_ENV`, consumed by the Publish step. + +- [ ] **Step 1: Add the Verify step** + +In `.github/workflows/batch-sync.yml`, insert between the `Export` and `Publish to batch-archive` steps: + +```yaml + # A failed verification stops the job before any commit or tag exists. + - name: Verify snapshot + run: | + set -euo pipefail + last_block="$(./dist/batch-export verify --old resume.ndjson.gzip --new snapshot.ndjson.gzip)" + echo "LAST_BLOCK=${last_block}" >> "${GITHUB_ENV}" +``` + +- [ ] **Step 2: Replace the sed extraction in the Publish step** + +Delete this block from `Publish to batch-archive`: + +```yaml + # blockNumber is hex ("0x...") in the slim NDJSON; the commit title + # uses decimal, matching the archive's history. + last_block_hex="$(gunzip -c archive/export.ndjson.gzip | tail -n 1 \ + | sed -nE 's/.*"blockNumber":"(0x[0-9a-fA-F]+)".*/\1/p')" + if [ -z "${last_block_hex}" ]; then + echo "::error::could not read blockNumber from the last snapshot entry" + exit 1 + fi + last_block="$(printf '%d' "${last_block_hex}")" +``` + +and change both remaining uses of `${last_block}` to `${LAST_BLOCK}` — one in `git commit -m "chore: update snapshot to block number ${last_block}"`, one in the final `::notice::published snapshot at block ${last_block} ...` line. + +- [ ] **Step 3: Validate** + +Run: +```bash +python3 -c "import yaml; yaml.safe_load(open('.github/workflows/batch-sync.yml')); print('YAML OK')" +actionlint .github/workflows/batch-sync.yml || true # only the known custom runner-label warning is acceptable +grep -n "last_block" .github/workflows/batch-sync.yml # expect: only the Verify step's local variable, no stale lowercase uses in Publish +``` +Expected: YAML OK; no actionlint findings beyond the `bee` runner label; grep shows no `${last_block}` left in the Publish step. + +- [ ] **Step 4: Full check and commit** + +Run: `go test ./...` and `go vet ./...` +Expected: PASS. + +```bash +git add .github/workflows/batch-sync.yml +git commit -m "ci: gate publishing on snapshot verification" +``` + +- [ ] **Step 5: Push and open the PR** + +```bash +git push -u origin feat/verify-snapshot +gh pr create --title "feat: verify refreshed snapshots before publishing" --body "$(cat <<'EOF' +Adds a `verify` subcommand and wires it into the Batch Sync workflow as a regression gate before any commit or tag reaches batch-archive. + +## What it does + +- `batch-export verify --old --new ` proves the new snapshot strictly extends the old one: the old content must appear byte for byte at the start of the new file, and every appended entry must parse and advance the (blockNumber, logIndex) order. On success it prints the new last block number in decimal on stdout; any failure exits non-zero. +- `pkg/verify` builds entirely on `pkg/resume`'s existing cursor and format handling; `resume` newly exports `MaxLineBytes`, `ParseEntry`, and `OpenClean` so no consumer re-encodes the on-disk format. +- The workflow runs `verify` between Export and Publish. Its stdout replaces the previous `gunzip | tail | sed` block-number extraction, so the commit title's block number now comes from the same Go code that defines the snapshot format. + +Spec: `docs/superpowers/specs/2026-08-31-verify-snapshot-design.md` + +🤖 Generated with [Claude Code](https://claude.com/claude-code) +EOF +)" +``` From 3cc3a2d0a8c8e05f5d038a4332ca54e7490d261e Mon Sep 17 00:00:00 2001 From: Ljubisa Gacevic Date: Mon, 31 Aug 2026 16:48:18 +0200 Subject: [PATCH 03/12] feat: export clean-content helpers from pkg/resume --- pkg/resume/resume.go | 53 ++++++++++++++++++++++++++++++----- pkg/resume/resume_test.go | 58 +++++++++++++++++++++++++++++++++++++++ 2 files changed, 104 insertions(+), 7 deletions(-) diff --git a/pkg/resume/resume.go b/pkg/resume/resume.go index a444615..f900d62 100644 --- a/pkg/resume/resume.go +++ b/pkg/resume/resume.go @@ -19,9 +19,9 @@ import ( ) const ( - // maxLineBytes caps a single line. Exported log lines run to a few + // MaxLineBytes caps a single line. Exported log lines run to a few // hundred bytes, so anything longer is not this tool's output. - maxLineBytes = 1 << 20 + MaxLineBytes = 1 << 20 // bufferSize is how much of a gzip file is buffered per read. bufferSize = 64 * 1024 ) @@ -206,7 +206,7 @@ func isGzip(file *os.File) (bool, error) { // line must parse as a log entry, and the only bytes allowed after it are a // single interrupted write — a trailing fragment with no newline. func lastCursorPlain(file *os.File, size int64) (*Cursor, error) { - window := min(size, 2*maxLineBytes) + window := min(size, 2*MaxLineBytes) offset := size - window buf := make([]byte, window) @@ -218,12 +218,12 @@ func lastCursorPlain(file *os.File, size int64) (*Cursor, error) { if nl < 0 { // No newline at all: an interrupted first write, unless the file is // longer than any single line the tool writes. - if size > maxLineBytes { + if size > MaxLineBytes { return nil, fmt.Errorf("%w: no newline in the final %d bytes (from offset %d)", ErrNotAnExport, window, offset) } return nil, ErrNoLogs } - if tail := window - int64(nl) - 1; tail > maxLineBytes { + if tail := window - int64(nl) - 1; tail > MaxLineBytes { return nil, fmt.Errorf("%w: %d bytes without a newline after offset %d", ErrNotAnExport, tail, offset+int64(nl)+1) } @@ -354,8 +354,8 @@ func scanMember(r io.Reader, member int) (*Cursor, error) { for { chunk, err := buffered.ReadSlice('\n') line = append(line, chunk...) - if len(line) > maxLineBytes { - return nil, fmt.Errorf("%w: member %d, line %d exceeds %d bytes", ErrNotAnExport, member, lineNum, maxLineBytes) + if len(line) > MaxLineBytes { + return nil, fmt.Errorf("%w: member %d, line %d exceeds %d bytes", ErrNotAnExport, member, lineNum, MaxLineBytes) } switch { @@ -429,3 +429,42 @@ func parseCursor(line []byte) (*Cursor, error) { LogIndex: uint(*parsed.LogIndex), }, nil } + +// ParseEntry builds a cursor from a single exported NDJSON line, accepting +// both the slim and the full log shape. It is the only place outside the +// writer where the on-disk field names are known, so consumers never +// re-encode them. +func ParseEntry(line []byte) (*Cursor, error) { + return parseCursor(line) +} + +// OpenClean returns the decompressed clean content of the export at path — +// the same bytes PrepareOutput would carry over. c must come from Read on +// the same, unmodified file. +func OpenClean(path string, c *Cursor) (io.ReadCloser, error) { + file, err := os.Open(path) + if err != nil { + return nil, fmt.Errorf("error opening export file: %w", err) + } + + limited := io.LimitReader(file, c.CleanSize) + if !c.Compressed { + return &cleanReader{Reader: limited, closer: file}, nil + } + + gz, err := gzip.NewReader(bufio.NewReaderSize(limited, bufferSize)) + if err != nil { + file.Close() + return nil, fmt.Errorf("error reading gzip export file: %w", err) + } + + return &cleanReader{Reader: gz, closer: file}, nil +} + +// cleanReader pairs the decompressed stream with the file it draws from. +type cleanReader struct { + io.Reader + closer io.Closer +} + +func (r *cleanReader) Close() error { return r.closer.Close() } diff --git a/pkg/resume/resume_test.go b/pkg/resume/resume_test.go index 2ce2bb8..16212d3 100644 --- a/pkg/resume/resume_test.go +++ b/pkg/resume/resume_test.go @@ -1256,3 +1256,61 @@ func TestPrepareOutputInvalidDirectory(t *testing.T) { t.Fatal("input was modified on failed PrepareOutput") } } + +func TestParseEntry(t *testing.T) { + t.Parallel() + + cursor, err := resume.ParseEntry(ndjson(t, testLog(7, 3))) + if err != nil { + t.Fatalf("ParseEntry: %v", err) + } + if cursor.BlockNumber != 7 || cursor.LogIndex != 3 { + t.Fatalf("got (%d, %d), want (7, 3)", cursor.BlockNumber, cursor.LogIndex) + } + + if _, err := resume.ParseEntry([]byte(`{"foo":1}`)); err == nil { + t.Fatal("want an error for a line missing blockNumber and logIndex") + } +} + +func TestOpenClean(t *testing.T) { + t.Parallel() + + content := ndjson(t, testLog(1, 0), testLog(2, 0)) + + tests := []struct { + name string + file string + blob []byte + }{ + {nameFormatPlain, plainFile, content}, + {nameFormatGzip, gzipFile, gz(t, content)}, + {"plain with interrupted tail", plainFile, append(slices.Clone(content), `{"blockNu`...)}, + {"gzip with interrupted tail", gzipFile, truncateLast(append(gz(t, content), gz(t, ndjson(t, testLog(3, 0)))...), 4)}, + } + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + + path := write(t, tc.file, tc.blob) + cursor, err := resume.Read(path) + if err != nil { + t.Fatalf("Read: %v", err) + } + + r, err := resume.OpenClean(path, cursor) + if err != nil { + t.Fatalf("OpenClean: %v", err) + } + defer r.Close() + + got, err := io.ReadAll(r) + if err != nil { + t.Fatalf("reading clean content: %v", err) + } + if !bytes.Equal(got, content) { + t.Fatalf("clean content mismatch:\ngot %q\nwant %q", got, content) + } + }) + } +} From 27d512af772b3d58f5229a34f9cd06d14ae37963 Mon Sep 17 00:00:00 2001 From: Ljubisa Gacevic Date: Mon, 31 Aug 2026 16:54:12 +0200 Subject: [PATCH 04/12] feat: add pkg/verify snapshot extension check --- pkg/verify/verify.go | 162 ++++++++++++++++++++++++++++++++++++++ pkg/verify/verify_test.go | 113 ++++++++++++++++++++++++++ 2 files changed, 275 insertions(+) create mode 100644 pkg/verify/verify.go create mode 100644 pkg/verify/verify_test.go diff --git a/pkg/verify/verify.go b/pkg/verify/verify.go new file mode 100644 index 0000000..44ffe37 --- /dev/null +++ b/pkg/verify/verify.go @@ -0,0 +1,162 @@ +// Package verify checks that a refreshed snapshot is a strict extension of +// the snapshot it was resumed from: everything the old file held, byte for +// byte and in order, followed only by newer entries. +package verify + +import ( + "bufio" + "bytes" + "errors" + "fmt" + "io" + + "github.com/ethersphere/batch-export/pkg/resume" +) + +var ( + // ErrMismatch indicates the new snapshot does not begin with the old + // snapshot's content: an entry was dropped, mutated, or reordered. + ErrMismatch = errors.New("new snapshot does not extend the old snapshot") + // ErrOrder indicates an appended entry out of (blockNumber, logIndex) + // order relative to what precedes it. + ErrOrder = errors.New("appended entries out of order") + // ErrTruncatedNew indicates the new snapshot ends in an interrupted + // write; a completed export never does. + ErrTruncatedNew = errors.New("new snapshot ends in an interrupted write") +) + +// chunkSize is how many bytes of both snapshots are compared per read. +const chunkSize = 64 * 1024 + +// Result reports what a successful verification established. +type Result struct { + // LastBlock is the block number of the new snapshot's final entry. + LastBlock uint64 + // Appended is how many entries the new snapshot adds. + Appended int + // OldTruncated reports whether the old snapshot carried an interrupted + // tail, excluded from the comparison the way resuming excludes it from + // the copy. + OldTruncated bool +} + +// Verify checks that the snapshot at newPath extends the one at oldPath. +func Verify(oldPath, newPath string) (Result, error) { + oldCursor, err := resume.Read(oldPath) + if err != nil { + return Result{}, fmt.Errorf("old snapshot: %w", err) + } + newCursor, err := resume.Read(newPath) + if err != nil { + return Result{}, fmt.Errorf("new snapshot: %w", err) + } + if newCursor.Truncated { + return Result{}, ErrTruncatedNew + } + + oldClean, err := resume.OpenClean(oldPath, oldCursor) + if err != nil { + return Result{}, fmt.Errorf("old snapshot: %w", err) + } + defer oldClean.Close() + + newClean, err := resume.OpenClean(newPath, newCursor) + if err != nil { + return Result{}, fmt.Errorf("new snapshot: %w", err) + } + defer newClean.Close() + + newBuffered := bufio.NewReaderSize(newClean, chunkSize) + if err := comparePrefix(oldClean, newBuffered); err != nil { + return Result{}, err + } + + appended, err := checkTail(newBuffered, oldCursor) + if err != nil { + return Result{}, err + } + + return Result{ + LastBlock: newCursor.BlockNumber, + Appended: appended, + OldTruncated: oldCursor.Truncated, + }, nil +} + +// comparePrefix requires every byte of oldR to appear at the start of newR. +func comparePrefix(oldR io.Reader, newR *bufio.Reader) error { + oldBuf := make([]byte, chunkSize) + newBuf := make([]byte, chunkSize) + + var offset int64 + for { + n, err := oldR.Read(oldBuf) + if n > 0 { + if _, rerr := io.ReadFull(newR, newBuf[:n]); rerr != nil { + return fmt.Errorf("%w: new snapshot ends at byte %d of the old content", ErrMismatch, offset) + } + if !bytes.Equal(oldBuf[:n], newBuf[:n]) { + return fmt.Errorf("%w: content diverges within bytes %d..%d of the old content", ErrMismatch, offset, offset+int64(n)) + } + offset += int64(n) + } + if errors.Is(err, io.EOF) { + return nil + } + if err != nil { + return fmt.Errorf("error reading old snapshot: %w", err) + } + } +} + +// checkTail walks the entries the new snapshot appends after the old +// content, requiring each to advance the (blockNumber, logIndex) order. +func checkTail(r *bufio.Reader, prev *resume.Cursor) (int, error) { + last := *prev + + var ( + appended int + lineNum int + line []byte + ) + for { + chunk, err := r.ReadSlice('\n') + line = append(line, chunk...) + if len(line) > resume.MaxLineBytes { + return 0, fmt.Errorf("appended line %d exceeds %d bytes", lineNum+1, resume.MaxLineBytes) + } + + switch { + case errors.Is(err, bufio.ErrBufferFull): + continue + case errors.Is(err, io.EOF): + if len(line) > 0 { + return 0, fmt.Errorf("appended line %d ends without a newline", lineNum+1) + } + return appended, nil + case err != nil: + return 0, fmt.Errorf("error reading new snapshot: %w", err) + } + + lineNum++ + entry, err := resume.ParseEntry(line) + if err != nil { + return 0, fmt.Errorf("appended line %d is not a log entry: %w", lineNum, err) + } + if !after(entry, &last) { + return 0, fmt.Errorf("%w: appended line %d holds (block %d, log %d) after (block %d, log %d)", + ErrOrder, lineNum, entry.BlockNumber, entry.LogIndex, last.BlockNumber, last.LogIndex) + } + last = *entry + appended++ + line = line[:0] + } +} + +// after reports whether entry strictly follows prev in log order. +func after(entry, prev *resume.Cursor) bool { + if entry.BlockNumber != prev.BlockNumber { + return entry.BlockNumber > prev.BlockNumber + } + return entry.LogIndex > prev.LogIndex +} diff --git a/pkg/verify/verify_test.go b/pkg/verify/verify_test.go new file mode 100644 index 0000000..38d1cb1 --- /dev/null +++ b/pkg/verify/verify_test.go @@ -0,0 +1,113 @@ +package verify_test + +import ( + "bytes" + "compress/gzip" + "encoding/json" + "os" + "path/filepath" + "testing" + + "github.com/ethereum/go-ethereum/common" + "github.com/ethereum/go-ethereum/core/types" + "github.com/ethersphere/batch-export/pkg/filestore" + "github.com/ethersphere/batch-export/pkg/verify" +) + +// testLog builds a log shaped like the ones the exporter writes. +func testLog(blockNumber uint64, logIndex uint) types.Log { + return types.Log{ + Address: common.HexToAddress("0x45a1502382541cd610cc9068e88727426b696293"), + Topics: []common.Hash{common.HexToHash("0xae46785019700e30375a5d7b4f91e32f8060ef085111f896ebf889450aa2ab5a")}, + Data: bytes.Repeat([]byte{0xab}, 32), + BlockNumber: blockNumber, + TxHash: common.HexToHash("0xb08f07656eaafa8efc458e2aa90773648d95ec8119873d212b4377dea5190cc0"), + Index: logIndex, + } +} + +// ndjson renders logs the way the exporter's slim format does. +func ndjson(t *testing.T, logs ...types.Log) []byte { + t.Helper() + + var buf bytes.Buffer + enc := json.NewEncoder(&buf) + for _, l := range logs { + if err := enc.Encode(filestore.NewSlimLog(l)); err != nil { + t.Fatal(err) + } + } + + return buf.Bytes() +} + +// gz compresses b as a single gzip member. +func gz(t *testing.T, b []byte) []byte { + t.Helper() + + var buf bytes.Buffer + w := gzip.NewWriter(&buf) + if _, err := w.Write(b); err != nil { + t.Fatal(err) + } + if err := w.Close(); err != nil { + t.Fatal(err) + } + + return buf.Bytes() +} + +// write puts content into a fresh temp file and returns its path. +func write(t *testing.T, name string, content []byte) string { + t.Helper() + + path := filepath.Join(t.TempDir(), name) + if err := os.WriteFile(path, content, 0o644); err != nil { + t.Fatal(err) + } + + return path +} + +func TestVerifySuperset(t *testing.T) { + t.Parallel() + + // The new snapshot is the old member plus an appended member, exactly + // the shape a resumed gzip export produces. + oldBlob := gz(t, ndjson(t, testLog(1, 0), testLog(2, 0))) + newBlob := append(bytes.Clone(oldBlob), gz(t, ndjson(t, testLog(2, 1), testLog(3, 0)))...) + + res, err := verify.Verify( + write(t, "old.ndjson.gzip", oldBlob), + write(t, "new.ndjson.gzip", newBlob), + ) + if err != nil { + t.Fatalf("Verify: %v", err) + } + if res.LastBlock != 3 { + t.Errorf("LastBlock = %d, want 3", res.LastBlock) + } + if res.Appended != 2 { + t.Errorf("Appended = %d, want 2", res.Appended) + } + if res.OldTruncated { + t.Error("OldTruncated = true, want false") + } +} + +func TestVerifyIdentical(t *testing.T) { + t.Parallel() + + blob := gz(t, ndjson(t, testLog(1, 0), testLog(2, 0))) + + res, err := verify.Verify( + write(t, "old.ndjson.gzip", blob), + write(t, "new.ndjson.gzip", blob), + ) + if err != nil { + t.Fatalf("Verify: %v", err) + } + if res.LastBlock != 2 || res.Appended != 0 { + t.Errorf("got LastBlock=%d Appended=%d, want 2 and 0", res.LastBlock, res.Appended) + } +} From c3f6df1e4cff34fbfc887b928a297b0ae3d7ba67 Mon Sep 17 00:00:00 2001 From: Ljubisa Gacevic Date: Mon, 31 Aug 2026 17:02:58 +0200 Subject: [PATCH 05/12] test: cover verify failure modes and truncation handling --- pkg/verify/verify_test.go | 115 ++++++++++++++++++++++++++++++++++++++ 1 file changed, 115 insertions(+) diff --git a/pkg/verify/verify_test.go b/pkg/verify/verify_test.go index 38d1cb1..07cfbf9 100644 --- a/pkg/verify/verify_test.go +++ b/pkg/verify/verify_test.go @@ -4,13 +4,16 @@ import ( "bytes" "compress/gzip" "encoding/json" + "errors" "os" "path/filepath" + "slices" "testing" "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/core/types" "github.com/ethersphere/batch-export/pkg/filestore" + "github.com/ethersphere/batch-export/pkg/resume" "github.com/ethersphere/batch-export/pkg/verify" ) @@ -111,3 +114,115 @@ func TestVerifyIdentical(t *testing.T) { t.Errorf("got LastBlock=%d Appended=%d, want 2 and 0", res.LastBlock, res.Appended) } } + +func TestVerifyRefusals(t *testing.T) { + t.Parallel() + + oldContent := ndjson(t, testLog(1, 0), testLog(2, 0)) + + // mutated flips one byte inside the first entry's blockNumber. + mutated := bytes.Replace(slices.Clone(oldContent), []byte(`"blockNumber":"0x1"`), []byte(`"blockNumber":"0x9"`), 1) + + tests := []struct { + name string + old []byte + new []byte + wantErr error + }{ + { + name: "mutated entry inside the old content", + old: gz(t, oldContent), + new: gz(t, append(mutated, ndjson(t, testLog(10, 0))...)), + wantErr: verify.ErrMismatch, + }, + { + name: "dropped entry", + old: gz(t, oldContent), + new: gz(t, ndjson(t, testLog(1, 0), testLog(3, 0))), + wantErr: verify.ErrMismatch, + }, + { + name: "new shorter than old", + old: gz(t, oldContent), + new: gz(t, ndjson(t, testLog(1, 0))), + wantErr: verify.ErrMismatch, + }, + { + name: "appended entry repeats the cursor", + old: gz(t, oldContent), + new: append(gz(t, oldContent), gz(t, ndjson(t, testLog(2, 0)))...), + wantErr: verify.ErrOrder, + }, + { + name: "appended entries out of order", + old: gz(t, oldContent), + new: append(gz(t, oldContent), gz(t, ndjson(t, testLog(5, 0), testLog(4, 0)))...), + wantErr: verify.ErrOrder, + }, + { + name: "new ends in an interrupted write", + old: oldContent, + new: append(slices.Clone(oldContent), `{"blockNumber":"0x3"`...), + wantErr: verify.ErrTruncatedNew, + }, + { + name: "old holds no complete entry", + old: nil, + new: oldContent, + wantErr: resume.ErrNoLogs, + }, + } + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + + _, err := verify.Verify( + write(t, "old", tc.old), + write(t, "new", tc.new), + ) + if !errors.Is(err, tc.wantErr) { + t.Fatalf("Verify error = %v, want %v", err, tc.wantErr) + } + }) + } +} + +func TestVerifyMalformedAppendedLine(t *testing.T) { + t.Parallel() + + oldContent := ndjson(t, testLog(1, 0), testLog(2, 0)) + // The middle appended line is valid JSON but not a log entry; resume.Read + // on a plain file only inspects the tail, so only checkTail can catch it. + newContent := append(slices.Clone(oldContent), "{\"foo\":1}\n"...) + newContent = append(newContent, ndjson(t, testLog(3, 0))...) + + _, err := verify.Verify( + write(t, "old", oldContent), + write(t, "new", newContent), + ) + if err == nil { + t.Fatal("want an error for a malformed appended line") + } +} + +func TestVerifyOldTruncated(t *testing.T) { + t.Parallel() + + oldContent := ndjson(t, testLog(1, 0), testLog(2, 0)) + oldBlob := append(slices.Clone(oldContent), `{"block`...) + newBlob := append(slices.Clone(oldContent), ndjson(t, testLog(3, 0))...) + + res, err := verify.Verify( + write(t, "old", oldBlob), + write(t, "new", newBlob), + ) + if err != nil { + t.Fatalf("Verify: %v", err) + } + if !res.OldTruncated { + t.Error("OldTruncated = false, want true") + } + if res.LastBlock != 3 || res.Appended != 1 { + t.Errorf("got LastBlock=%d Appended=%d, want 3 and 1", res.LastBlock, res.Appended) + } +} From eab6745f79d6ad922e934813cf3e6f38b9f4c6fb Mon Sep 17 00:00:00 2001 From: Ljubisa Gacevic Date: Mon, 31 Aug 2026 17:03:24 +0200 Subject: [PATCH 06/12] test: cover verify across plain and gzip snapshot combinations --- pkg/verify/verify_test.go | 34 ++++++++++++++++++++++++++++++++++ 1 file changed, 34 insertions(+) diff --git a/pkg/verify/verify_test.go b/pkg/verify/verify_test.go index 07cfbf9..c1165f6 100644 --- a/pkg/verify/verify_test.go +++ b/pkg/verify/verify_test.go @@ -226,3 +226,37 @@ func TestVerifyOldTruncated(t *testing.T) { t.Errorf("got LastBlock=%d Appended=%d, want 3 and 1", res.LastBlock, res.Appended) } } + +func TestVerifyFormatCombinations(t *testing.T) { + t.Parallel() + + oldContent := ndjson(t, testLog(1, 0), testLog(2, 0)) + newContent := append(slices.Clone(oldContent), ndjson(t, testLog(3, 0))...) + + tests := []struct { + name string + old []byte + new []byte + }{ + {"plain old, plain new", oldContent, newContent}, + {"plain old, gzip new", oldContent, gz(t, newContent)}, + {"gzip old, plain new", gz(t, oldContent), newContent}, + {"gzip old, gzip new", gz(t, oldContent), gz(t, newContent)}, + } + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + + res, err := verify.Verify( + write(t, "old", tc.old), + write(t, "new", tc.new), + ) + if err != nil { + t.Fatalf("Verify: %v", err) + } + if res.LastBlock != 3 || res.Appended != 1 { + t.Errorf("got LastBlock=%d Appended=%d, want 3 and 1", res.LastBlock, res.Appended) + } + }) + } +} From dac6d6f426fccc92de7fd22b49cc49bc935e8ebe Mon Sep 17 00:00:00 2001 From: Ljubisa Gacevic Date: Mon, 31 Aug 2026 17:10:57 +0200 Subject: [PATCH 07/12] feat: add verify command gating snapshot publication Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_014WoaRY6SZi2gUPPmQEhJgr --- README.md | 12 +++++++ cmd/cmd.go | 4 +++ cmd/verify.go | 50 ++++++++++++++++++++++++++++ cmd/verify_test.go | 83 ++++++++++++++++++++++++++++++++++++++++++++++ 4 files changed, 149 insertions(+) create mode 100644 cmd/verify.go create mode 100644 cmd/verify_test.go diff --git a/README.md b/README.md index ebca32c..feb93b6 100644 --- a/README.md +++ b/README.md @@ -118,6 +118,18 @@ touched — delete the incomplete `--output` file and rerun. The produced NDJSON is consumed by [batch-archive](https://github.com/ethersphere/batch-archive), which embeds it for use inside Bee. +### Verifying a snapshot + +`verify` proves a refreshed snapshot still holds everything the original did — +byte for byte, in order — followed only by newer entries, and prints the new +snapshot's last block number in decimal: + +```sh +batch-export verify --old export.ndjson.gzip --new snapshot.ndjson.gzip +``` + +A non-zero exit means the new snapshot must not replace the old one. + ## Maintainers - [Bee](https://github.com/orgs/ethersphere/teams/bee) team diff --git a/cmd/cmd.go b/cmd/cmd.go index 4ed1f11..95b11b4 100644 --- a/cmd/cmd.go +++ b/cmd/cmd.go @@ -52,6 +52,10 @@ func newCommand() (c *command, err error) { return nil, err } + if err := c.initVerifyCmd(); err != nil { + return nil, err + } + return c, nil } diff --git a/cmd/verify.go b/cmd/verify.go new file mode 100644 index 0000000..21cbb9b --- /dev/null +++ b/cmd/verify.go @@ -0,0 +1,50 @@ +package cmd + +import ( + "fmt" + + "github.com/ethersphere/batch-export/pkg/verify" + "github.com/spf13/cobra" +) + +func (c *command) initVerifyCmd() error { + var ( + oldFile string + newFile string + ) + + cmd := &cobra.Command{ + Use: "verify", + Short: "Verify that a refreshed snapshot extends the one it was resumed from", + Long: `Verifies that the snapshot at --new holds everything the snapshot at --old does, +byte for byte and in order, followed only by newer entries. On success the new +snapshot's last block number is printed to stdout in decimal; any failure exits +non-zero with nothing on stdout.`, + RunE: func(cmd *cobra.Command, args []string) error { + result, err := verify.Verify(oldFile, newFile) + if err != nil { + return err + } + + if result.OldTruncated { + c.log.Warning("old snapshot ends in an interrupted write; the truncated tail was excluded from the comparison", "oldFile", oldFile) + } + c.log.Info("snapshot verified", "oldFile", oldFile, "newFile", newFile, "appended", result.Appended, "lastBlock", result.LastBlock) + + _, err = fmt.Fprintln(cmd.OutOrStdout(), result.LastBlock) + return err + }, + } + + cmd.Flags().StringVar(&oldFile, "old", "", "Snapshot the export resumed from (.ndjson, .gz or .gzip)") + cmd.Flags().StringVar(&newFile, "new", "", "Freshly exported snapshot to check (.ndjson, .gz or .gzip)") + for _, name := range []string{"old", "new"} { + if err := cmd.MarkFlagRequired(name); err != nil { + return err + } + } + + c.root.AddCommand(cmd) + + return nil +} diff --git a/cmd/verify_test.go b/cmd/verify_test.go new file mode 100644 index 0000000..6b86b52 --- /dev/null +++ b/cmd/verify_test.go @@ -0,0 +1,83 @@ +package cmd + +import ( + "bytes" + "compress/gzip" + "context" + "encoding/json" + "os" + "path/filepath" + "testing" + + "github.com/ethereum/go-ethereum/core/types" + "github.com/ethersphere/batch-export/pkg/filestore" +) + +// writeSnapshot writes a slim-format gzip snapshot with one entry per +// (blockNumber, logIndex) pair and returns its path. +func writeSnapshot(t *testing.T, name string, entries ...[2]uint64) string { + t.Helper() + + var buf bytes.Buffer + w := gzip.NewWriter(&buf) + enc := json.NewEncoder(w) + for _, e := range entries { + if err := enc.Encode(filestore.NewSlimLog(types.Log{BlockNumber: e[0], Index: uint(e[1])})); err != nil { + t.Fatal(err) + } + } + if err := w.Close(); err != nil { + t.Fatal(err) + } + + path := filepath.Join(t.TempDir(), name) + if err := os.WriteFile(path, buf.Bytes(), 0o644); err != nil { + t.Fatal(err) + } + + return path +} + +func TestVerifyCmd(t *testing.T) { + t.Parallel() + + oldPath := writeSnapshot(t, "old.ndjson.gzip", [2]uint64{1, 0}, [2]uint64{2, 0}) + newPath := writeSnapshot(t, "new.ndjson.gzip", [2]uint64{1, 0}, [2]uint64{2, 0}, [2]uint64{3, 0}) + + c, err := newCommand() + if err != nil { + t.Fatal(err) + } + var out bytes.Buffer + c.root.SetOut(&out) + c.root.SetArgs([]string{"verify", "--old", oldPath, "--new", newPath}) + + if err := c.Execute(context.Background()); err != nil { + t.Fatalf("verify: %v", err) + } + if got, want := out.String(), "3\n"; got != want { + t.Errorf("stdout = %q, want %q", got, want) + } +} + +func TestVerifyCmdRefusal(t *testing.T) { + t.Parallel() + + oldPath := writeSnapshot(t, "old.ndjson.gzip", [2]uint64{1, 0}, [2]uint64{2, 0}) + newPath := writeSnapshot(t, "new.ndjson.gzip", [2]uint64{1, 0}, [2]uint64{3, 0}) + + c, err := newCommand() + if err != nil { + t.Fatal(err) + } + var out bytes.Buffer + c.root.SetOut(&out) + c.root.SetArgs([]string{"verify", "--old", oldPath, "--new", newPath}) + + if err := c.Execute(context.Background()); err == nil { + t.Fatal("want an error for a snapshot that drops an entry") + } + if out.Len() != 0 { + t.Errorf("stdout = %q, want empty", out.String()) + } +} From 0be9cb9865aa23902984a5c56ea0f01954a84808 Mon Sep 17 00:00:00 2001 From: Ljubisa Gacevic Date: Mon, 31 Aug 2026 17:18:11 +0200 Subject: [PATCH 08/12] ci: gate publishing on snapshot verification --- .github/workflows/batch-sync.yml | 21 +++++++++------------ 1 file changed, 9 insertions(+), 12 deletions(-) diff --git a/.github/workflows/batch-sync.yml b/.github/workflows/batch-sync.yml index 4a9fc78..8dd6024 100644 --- a/.github/workflows/batch-sync.yml +++ b/.github/workflows/batch-sync.yml @@ -151,6 +151,13 @@ jobs: --slim=true \ --verbosity "${VERBOSITY}" + # A failed verification stops the job before any commit or tag exists. + - name: Verify snapshot + run: | + set -euo pipefail + last_block="$(./dist/batch-export verify --old resume.ndjson.gzip --new snapshot.ndjson.gzip)" + echo "LAST_BLOCK=${last_block}" >> "${GITHUB_ENV}" + - name: Publish to batch-archive env: TRIGGERED_BY: ${{ github.actor }} @@ -164,16 +171,6 @@ jobs: exit 0 fi - # blockNumber is hex ("0x...") in the slim NDJSON; the commit title - # uses decimal, matching the archive's history. - last_block_hex="$(gunzip -c archive/export.ndjson.gzip | tail -n 1 \ - | sed -nE 's/.*"blockNumber":"(0x[0-9a-fA-F]+)".*/\1/p')" - if [ -z "${last_block_hex}" ]; then - echo "::error::could not read blockNumber from the last snapshot entry" - exit 1 - fi - last_block="$(printf '%d' "${last_block_hex}")" - # Bump the highest existing semver tag, not the resume tag, so an # older resume can't collide; non-semver tags are ignored. latest_tag="$(git tag | grep -E '^v[0-9]+\.[0-9]+\.[0-9]+$' | sort -V | tail -n 1 || true)" @@ -188,10 +185,10 @@ jobs: git config user.email "41898470+github-actions[bot]@users.noreply.github.com" git add archive/export.ndjson.gzip git commit \ - -m "chore: update snapshot to block number ${last_block}" \ + -m "chore: update snapshot to block number ${LAST_BLOCK}" \ -m "Resumed from tag ${ARCHIVE_TAG} and exported up to the latest finalized block. Triggered by @${TRIGGERED_BY} via ${GITHUB_SERVER_URL}/${GITHUB_REPOSITORY}/actions/runs/${GITHUB_RUN_ID}" git tag "${new_tag}" # --atomic: a rejected push to main rejects the tag too, so no # orphaned tag can become a later run's resume point. git push --atomic origin HEAD:main "refs/tags/${new_tag}" - echo "::notice::published snapshot at block ${last_block} as ${new_tag} (resumed from ${ARCHIVE_TAG})" + echo "::notice::published snapshot at block ${LAST_BLOCK} as ${new_tag} (resumed from ${ARCHIVE_TAG})" From ff2751bf93815045775b301758fa460e30920b72 Mon Sep 17 00:00:00 2001 From: Ljubisa Gacevic Date: Mon, 31 Aug 2026 17:36:54 +0200 Subject: [PATCH 09/12] fix: address final review findings on the verify gate MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - workflow: Verify now compares against batch-archive/archive/export.ndjson.gzip (the file Publish overwrites) instead of the resumed tag, closing the gap where a non-tip archive_tag plus a stale finalized head could publish a snapshot that loses entries main already has. - Makefile: `make test` now runs `go test -v ./...` so cmd/verify_test.go and cmd/export_test.go run in CI via go.yml, not just pkg/... - workflow: documented why the Publish step's `git diff --quiet` guard can never fire today (resumed gzip export always appends an empty member); behavior unchanged, comment only. - pkg/resume: fixed OpenClean's doc comment (it returns decompressed content, not the same raw bytes PrepareOutput copies) and switched its two new error messages from "export file" to "resume file" to match the rest of the package. - pkg/verify: comparePrefix now distinguishes a genuinely short new snapshot (io.EOF/io.ErrUnexpectedEOF, still ErrMismatch) from a real read failure (wrapped as "error reading new snapshot"); both paths stay fail-closed. Also documented why checkTail's missing-trailing-newline branch is unreachable in practice, without removing it. - README: verify example now uses the ./dist/ prefix like every other command example. - spec: narrowed the Problem section's claim — the gate does not catch a regression confined to the slim NDJSON encoding, since ParseEntry only requires blockNumber and logIndex; that encoding is pinned separately by pkg/filestore's own tests. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_014WoaRY6SZi2gUPPmQEhJgr --- .github/workflows/batch-sync.yml | 13 ++++++++++++- Makefile | 2 +- README.md | 2 +- .../specs/2026-08-31-verify-snapshot-design.md | 12 ++++++++---- pkg/resume/resume.go | 11 ++++++----- pkg/verify/verify.go | 11 ++++++++++- 6 files changed, 38 insertions(+), 13 deletions(-) diff --git a/.github/workflows/batch-sync.yml b/.github/workflows/batch-sync.yml index 8dd6024..9deffb2 100644 --- a/.github/workflows/batch-sync.yml +++ b/.github/workflows/batch-sync.yml @@ -152,10 +152,18 @@ jobs: --verbosity "${VERBOSITY}" # A failed verification stops the job before any commit or tag exists. + # + # --old is the file Publish overwrites (main's current archive), not the + # resumed tag: a non-tip archive_tag plus a stale finalized head could + # otherwise verify clean against the tag while still losing entries main + # already has. Identical to the resumed tag in the normal case (resumed + # tag == main's tip), so this is a no-op then. An archive version + # written before the slim format would fail closed here, which is the + # intended answer. - name: Verify snapshot run: | set -euo pipefail - last_block="$(./dist/batch-export verify --old resume.ndjson.gzip --new snapshot.ndjson.gzip)" + last_block="$(./dist/batch-export verify --old batch-archive/archive/export.ndjson.gzip --new snapshot.ndjson.gzip)" echo "LAST_BLOCK=${last_block}" >> "${GITHUB_ENV}" - name: Publish to batch-archive @@ -166,6 +174,9 @@ jobs: cp snapshot.ndjson.gzip batch-archive/archive/export.ndjson.gzip cd batch-archive + # Cannot fire today: a resumed gzip export always appends a ~20-byte + # empty gzip member even when zero new logs were fetched, so the + # file always differs. Kept in case that stops being true. if git diff --quiet; then echo "::notice::snapshot is unchanged; nothing to publish" exit 0 diff --git a/Makefile b/Makefile index 65ced99..d3e7ff3 100644 --- a/Makefile +++ b/Makefile @@ -40,6 +40,6 @@ clean: .PHONY: test test: - $(GO) test -v ./pkg/... + $(GO) test -v ./... FORCE: diff --git a/README.md b/README.md index feb93b6..b6896ad 100644 --- a/README.md +++ b/README.md @@ -125,7 +125,7 @@ byte for byte, in order — followed only by newer entries, and prints the new snapshot's last block number in decimal: ```sh -batch-export verify --old export.ndjson.gzip --new snapshot.ndjson.gzip +./dist/batch-export verify --old export.ndjson.gzip --new snapshot.ndjson.gzip ``` A non-zero exit means the new snapshot must not replace the old one. diff --git a/docs/superpowers/specs/2026-08-31-verify-snapshot-design.md b/docs/superpowers/specs/2026-08-31-verify-snapshot-design.md index 3a7fb8e..a5af0c9 100644 --- a/docs/superpowers/specs/2026-08-31-verify-snapshot-design.md +++ b/docs/superpowers/specs/2026-08-31-verify-snapshot-design.md @@ -10,10 +10,14 @@ The Batch Sync workflow resumes the postage-batch snapshot from a tag of [ethersphere/batch-archive](https://github.com/ethersphere/batch-archive) and publishes the refreshed file back as a commit on `main` plus the next patch tag. Nothing verifies that the refreshed snapshot is a strict extension of the -one it resumed from. A regression in batch-export — in the resume logic, the -slim NDJSON encoding, or the gzip append handling — could silently drop, -mutate, or reorder historical entries, and the workflow would tag and publish -the corrupted snapshot as the new latest version. +one it resumed from. A regression in batch-export — in the resume logic or the +gzip append handling — could silently drop, mutate, or reorder historical +entries, and the workflow would tag and publish the corrupted snapshot as the +new latest version. (The gate does not catch a regression confined to the +slim NDJSON encoding itself: `ParseEntry` only requires `blockNumber` and +`logIndex`, so an appended entry written in the full geth shape still +verifies clean. The encoding is pinned separately, by `pkg/filestore`'s own +tests.) Separately, the workflow extracts the last block number for the commit title with a `gunzip | tail | sed` pipeline whose regex re-encodes the slim JSON diff --git a/pkg/resume/resume.go b/pkg/resume/resume.go index f900d62..cb074ab 100644 --- a/pkg/resume/resume.go +++ b/pkg/resume/resume.go @@ -438,13 +438,14 @@ func ParseEntry(line []byte) (*Cursor, error) { return parseCursor(line) } -// OpenClean returns the decompressed clean content of the export at path — -// the same bytes PrepareOutput would carry over. c must come from Read on -// the same, unmodified file. +// OpenClean returns the decompressed form of the clean content at path — the +// same content PrepareOutput carries over, but decompressed rather than +// PrepareOutput's raw copy. c must come from Read on the same, unmodified +// file. func OpenClean(path string, c *Cursor) (io.ReadCloser, error) { file, err := os.Open(path) if err != nil { - return nil, fmt.Errorf("error opening export file: %w", err) + return nil, fmt.Errorf("error opening resume file: %w", err) } limited := io.LimitReader(file, c.CleanSize) @@ -455,7 +456,7 @@ func OpenClean(path string, c *Cursor) (io.ReadCloser, error) { gz, err := gzip.NewReader(bufio.NewReaderSize(limited, bufferSize)) if err != nil { file.Close() - return nil, fmt.Errorf("error reading gzip export file: %w", err) + return nil, fmt.Errorf("error reading gzip resume file: %w", err) } return &cleanReader{Reader: gz, closer: file}, nil diff --git a/pkg/verify/verify.go b/pkg/verify/verify.go index 44ffe37..c291def 100644 --- a/pkg/verify/verify.go +++ b/pkg/verify/verify.go @@ -92,8 +92,14 @@ func comparePrefix(oldR io.Reader, newR *bufio.Reader) error { for { n, err := oldR.Read(oldBuf) if n > 0 { - if _, rerr := io.ReadFull(newR, newBuf[:n]); rerr != nil { + _, rerr := io.ReadFull(newR, newBuf[:n]) + switch { + // A short new file is a mismatch; anything else is a read + // failure and must not be reported as if the snapshot were short. + case errors.Is(rerr, io.EOF), errors.Is(rerr, io.ErrUnexpectedEOF): return fmt.Errorf("%w: new snapshot ends at byte %d of the old content", ErrMismatch, offset) + case rerr != nil: + return fmt.Errorf("error reading new snapshot: %w", rerr) } if !bytes.Equal(oldBuf[:n], newBuf[:n]) { return fmt.Errorf("%w: content diverges within bytes %d..%d of the old content", ErrMismatch, offset, offset+int64(n)) @@ -130,6 +136,9 @@ func checkTail(r *bufio.Reader, prev *resume.Cursor) (int, error) { case errors.Is(err, bufio.ErrBufferFull): continue case errors.Is(err, io.EOF): + // Defensive: Verify rejects a truncated new file up front, and a + // clean gzip member ending mid-line is already refused by + // resume.Read, so this should be unreachable in practice. if len(line) > 0 { return 0, fmt.Errorf("appended line %d ends without a newline", lineNum+1) } From 035051ff423c6d1e9d164c2af675536b5da43701 Mon Sep 17 00:00:00 2001 From: Ljubisa Gacevic Date: Mon, 31 Aug 2026 17:40:48 +0200 Subject: [PATCH 10/12] docs: align spec and plan with the shipped verify target The final review's fix pointed verify --old at the archive file the publish step overwrites rather than the resumed tag's file; three spec passages and the plan's Task 6 snippet still described the old behaviour. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_014WoaRY6SZi2gUPPmQEhJgr --- .../plans/2026-08-31-verify-snapshot.md | 5 +++++ .../specs/2026-08-31-verify-snapshot-design.md | 17 +++++++++++------ 2 files changed, 16 insertions(+), 6 deletions(-) diff --git a/docs/superpowers/plans/2026-08-31-verify-snapshot.md b/docs/superpowers/plans/2026-08-31-verify-snapshot.md index 39ba7ba..12fabb5 100644 --- a/docs/superpowers/plans/2026-08-31-verify-snapshot.md +++ b/docs/superpowers/plans/2026-08-31-verify-snapshot.md @@ -933,6 +933,11 @@ In `.github/workflows/batch-sync.yml`, insert between the `Export` and `Publish echo "LAST_BLOCK=${last_block}" >> "${GITHUB_ENV}" ``` +> Superseded during the final review: `--old` ships as +> `batch-archive/archive/export.ndjson.gzip`, the file the publish step +> overwrites, because a non-tip `archive_tag` makes the resumed file and the +> replaced file diverge. See the spec's CLI contract. + - [ ] **Step 2: Replace the sed extraction in the Publish step** Delete this block from `Publish to batch-archive`: diff --git a/docs/superpowers/specs/2026-08-31-verify-snapshot-design.md b/docs/superpowers/specs/2026-08-31-verify-snapshot-design.md index a5af0c9..8c0dd0e 100644 --- a/docs/superpowers/specs/2026-08-31-verify-snapshot-design.md +++ b/docs/superpowers/specs/2026-08-31-verify-snapshot-design.md @@ -40,9 +40,6 @@ failed verification blocks the commit and the tag. - No chain-side validation: `verify` never queries an RPC endpoint. - No repair: a bad snapshot is reported, never fixed. -- No comparison against batch-archive `main`'s current file: the contract is - old-input vs. new-output of one export run. (The workflow always resumes - the file it just fetched, so that is the meaningful pair.) - No change to export behavior, scheduling, or bootstrap flows. ## CLI contract @@ -51,8 +48,10 @@ failed verification blocks the commit and the tag. batch-export verify --old --new [--verbosity ] ``` -- `--old` (required): the snapshot the export resumed from — in the workflow, - the file fetched from the batch-archive tag. +- `--old` (required): the snapshot the new one must extend — in the workflow, + the archive file the publish step overwrites, which is what a regression + would actually destroy. That is the same file the export resumed from + whenever the resumed tag is main's tip, and a stricter check when it is not. - `--new` (required): the freshly exported snapshot. - Formats: plain NDJSON or gzip, each file detected independently by content (the same detection `--resume` uses); any old/new combination is accepted. @@ -115,10 +114,16 @@ the minimal piece is exported from there rather than duplicated. - name: Verify snapshot run: | set -euo pipefail - last_block="$(./dist/batch-export verify --old resume.ndjson.gzip --new snapshot.ndjson.gzip)" + last_block="$(./dist/batch-export verify \ + --old batch-archive/archive/export.ndjson.gzip \ + --new snapshot.ndjson.gzip)" echo "LAST_BLOCK=${last_block}" >> "${GITHUB_ENV}" ``` +`--old` is the archive's current file rather than the resumed tag's: the +publish step overwrites it, so it is what a regression would destroy. The +two are the same bytes whenever the resumed tag is main's tip. + The Publish step then uses `${LAST_BLOCK}` in the commit title and drops the `gunzip | tail | sed` block together with its empty-result guard. A verify failure stops the job before any commit or tag is created. From 9fe15d2c3fd9befa27117c8e86a4f413706755c3 Mon Sep 17 00:00:00 2001 From: Ljubisa Gacevic Date: Mon, 31 Aug 2026 22:10:46 +0200 Subject: [PATCH 11/12] docs: drop the spec and plan from the repo The verify gate's design and implementation plan lived only to produce the change; the rationale that outlives them is at its point of use in the code and the workflow. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01LqPjhcvFr2LLuBdhyjQwCe --- .../plans/2026-08-31-verify-snapshot.md | 997 ------------------ .../2026-08-31-verify-snapshot-design.md | 149 --- 2 files changed, 1146 deletions(-) delete mode 100644 docs/superpowers/plans/2026-08-31-verify-snapshot.md delete mode 100644 docs/superpowers/specs/2026-08-31-verify-snapshot-design.md diff --git a/docs/superpowers/plans/2026-08-31-verify-snapshot.md b/docs/superpowers/plans/2026-08-31-verify-snapshot.md deleted file mode 100644 index 12fabb5..0000000 --- a/docs/superpowers/plans/2026-08-31-verify-snapshot.md +++ /dev/null @@ -1,997 +0,0 @@ -# `batch-export verify` Implementation Plan - -> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. - -**Goal:** A `verify` subcommand that proves a refreshed snapshot strictly extends the snapshot it was resumed from, printing the new last block number — wired into the Batch Sync workflow as a gate before publishing. - -**Architecture:** `pkg/verify` compares the two files using `pkg/resume`'s existing cursor/format machinery (three small helpers get exported from `resume` first). `cmd/verify.go` is a thin cobra wrapper. The workflow gains a `Verify snapshot` step whose stdout replaces the current `gunzip | tail | sed` block-number extraction. - -**Tech Stack:** Go (stdlib only — no new dependencies), cobra (already used), `github.com/ethersphere/bee/v2/pkg/log` (already used; its default sink is os.Stderr, so stdout stays clean). - -**Spec:** `docs/superpowers/specs/2026-08-31-verify-snapshot-design.md` - -## Global Constraints - -- Module path is `github.com/ethersphere/batch-export`; work happens on branch `feat/verify-snapshot`. -- Conventional commits (`feat:`, `test:`, `ci:`, `docs:`). -- stdout of `verify` carries ONLY the decimal last block number and a newline; everything else goes to stderr via the logger. Non-zero exit = do not publish. -- Streaming with constant memory: never load a snapshot whole (they are tens of MB). -- No behavior change to `export`, `resume`, or the workflow's publish semantics beyond swapping the block-number source. -- Run tests with `go test ./...` and `go vet ./...`; both must pass before every commit. -- Comment style: sparse, constraint-stating, matching `pkg/resume` (read it first). - ---- - -### Task 1: Export `MaxLineBytes`, `ParseEntry`, and `OpenClean` from `pkg/resume` - -`pkg/verify` needs three things resume already knows: the line-length cap, how to parse one NDJSON line into `(blockNumber, logIndex)`, and how to read a file's decompressed clean content. Export them here so verify never re-encodes the on-disk format. - -**Files:** -- Modify: `pkg/resume/resume.go` -- Test: `pkg/resume/resume_test.go` (append new test functions; its existing helpers `testLog`, `ndjson`, `gz`, `truncateLast`, `write`, and constants `plainFile`, `gzipFile` are reused) - -**Interfaces:** -- Consumes: existing `resume.Read`, internal `parseCursor`, internal `maxLineBytes`. -- Produces (later tasks rely on these exact signatures): - - `const MaxLineBytes = 1 << 20` (renamed from `maxLineBytes`) - - `func ParseEntry(line []byte) (*Cursor, error)` - - `func OpenClean(path string, c *Cursor) (io.ReadCloser, error)` - -- [ ] **Step 1: Write the failing tests** - -Append to `pkg/resume/resume_test.go`: - -```go -func TestParseEntry(t *testing.T) { - t.Parallel() - - cursor, err := resume.ParseEntry(ndjson(t, testLog(7, 3))) - if err != nil { - t.Fatalf("ParseEntry: %v", err) - } - if cursor.BlockNumber != 7 || cursor.LogIndex != 3 { - t.Fatalf("got (%d, %d), want (7, 3)", cursor.BlockNumber, cursor.LogIndex) - } - - if _, err := resume.ParseEntry([]byte(`{"foo":1}`)); err == nil { - t.Fatal("want an error for a line missing blockNumber and logIndex") - } -} - -func TestOpenClean(t *testing.T) { - t.Parallel() - - content := ndjson(t, testLog(1, 0), testLog(2, 0)) - - tests := []struct { - name string - file string - blob []byte - }{ - {nameFormatPlain, plainFile, content}, - {nameFormatGzip, gzipFile, gz(t, content)}, - {"plain with interrupted tail", plainFile, append(slices.Clone(content), `{"blockNu`...)}, - {"gzip with interrupted tail", gzipFile, truncateLast(append(gz(t, content), gz(t, ndjson(t, testLog(3, 0)))...), 4)}, - } - for _, tc := range tests { - t.Run(tc.name, func(t *testing.T) { - t.Parallel() - - path := write(t, tc.file, tc.blob) - cursor, err := resume.Read(path) - if err != nil { - t.Fatalf("Read: %v", err) - } - - r, err := resume.OpenClean(path, cursor) - if err != nil { - t.Fatalf("OpenClean: %v", err) - } - defer r.Close() - - got, err := io.ReadAll(r) - if err != nil { - t.Fatalf("reading clean content: %v", err) - } - if !bytes.Equal(got, content) { - t.Fatalf("clean content mismatch:\ngot %q\nwant %q", got, content) - } - }) - } -} -``` - -Notes for the implementer: `ndjson` returns newline-terminated lines and `ParseEntry` trims whitespace, so passing a full line works. The "gzip with interrupted tail" case appends a second gzip member and cuts 4 bytes off its trailer — `Read` then reports `CleanSize` at the first member's boundary, and `OpenClean` must return only that member's content. Check the file's existing imports; `slices` and `io` are already imported. - -- [ ] **Step 2: Run the tests to verify they fail** - -Run: `go test ./pkg/resume/ -run 'TestParseEntry|TestOpenClean' -v` -Expected: compile error — `undefined: resume.ParseEntry` and `undefined: resume.OpenClean`. - -- [ ] **Step 3: Implement in `pkg/resume/resume.go`** - -Rename the constant (find every use: `grep -n maxLineBytes pkg/resume/resume.go` — it appears only in this file) and update its comment: - -```go -const ( - // MaxLineBytes caps a single line. Exported log lines run to a few - // hundred bytes, so anything longer is not this tool's output. - MaxLineBytes = 1 << 20 - // bufferSize is how much of a gzip file is buffered per read. - bufferSize = 64 * 1024 -) -``` - -Add at the end of the file: - -```go -// ParseEntry builds a cursor from a single exported NDJSON line, accepting -// both the slim and the full log shape. It is the only place outside the -// writer where the on-disk field names are known, so consumers never -// re-encode them. -func ParseEntry(line []byte) (*Cursor, error) { - return parseCursor(line) -} - -// OpenClean returns the decompressed clean content of the export at path — -// the same bytes PrepareOutput would carry over. c must come from Read on -// the same, unmodified file. -func OpenClean(path string, c *Cursor) (io.ReadCloser, error) { - file, err := os.Open(path) - if err != nil { - return nil, fmt.Errorf("error opening export file: %w", err) - } - - limited := io.LimitReader(file, c.CleanSize) - if !c.Compressed { - return &cleanReader{Reader: limited, closer: file}, nil - } - - gz, err := gzip.NewReader(bufio.NewReaderSize(limited, bufferSize)) - if err != nil { - file.Close() - return nil, fmt.Errorf("error reading gzip export file: %w", err) - } - - return &cleanReader{Reader: gz, closer: file}, nil -} - -// cleanReader pairs the decompressed stream with the file it draws from. -type cleanReader struct { - io.Reader - closer io.Closer -} - -func (r *cleanReader) Close() error { return r.closer.Close() } -``` - -The gzip path relies on `gzip.Reader`'s default multistream mode to walk every member inside the `CleanSize` limit; `CleanSize` always ends on a member boundary, so the limited stream ends cleanly. - -- [ ] **Step 4: Run the full package tests** - -Run: `go test ./pkg/resume/ -v` and `go vet ./...` -Expected: PASS (the rename must not have missed a reference; `go vet` is the catch-all). - -- [ ] **Step 5: Commit** - -```bash -git add pkg/resume/resume.go pkg/resume/resume_test.go -git commit -m "feat: export clean-content helpers from pkg/resume" -``` - ---- - -### Task 2: `pkg/verify` — core check, happy paths - -**Files:** -- Create: `pkg/verify/verify.go` -- Create: `pkg/verify/verify_test.go` - -**Interfaces:** -- Consumes: `resume.Read`, `resume.OpenClean`, `resume.ParseEntry`, `resume.MaxLineBytes`, `resume.Cursor` (Task 1). -- Produces (Tasks 3–5 rely on these exact names): - - `type Result struct { LastBlock uint64; Appended int; OldTruncated bool }` - - `func Verify(oldPath, newPath string) (Result, error)` - - `var ErrMismatch, ErrOrder, ErrTruncatedNew error` - -- [ ] **Step 1: Write the failing tests** - -Create `pkg/verify/verify_test.go`: - -```go -package verify_test - -import ( - "bytes" - "compress/gzip" - "encoding/json" - "os" - "path/filepath" - "testing" - - "github.com/ethereum/go-ethereum/common" - "github.com/ethereum/go-ethereum/core/types" - "github.com/ethersphere/batch-export/pkg/filestore" - "github.com/ethersphere/batch-export/pkg/verify" -) - -// testLog builds a log shaped like the ones the exporter writes. -func testLog(blockNumber uint64, logIndex uint) types.Log { - return types.Log{ - Address: common.HexToAddress("0x45a1502382541cd610cc9068e88727426b696293"), - Topics: []common.Hash{common.HexToHash("0xae46785019700e30375a5d7b4f91e32f8060ef085111f896ebf889450aa2ab5a")}, - Data: bytes.Repeat([]byte{0xab}, 32), - BlockNumber: blockNumber, - TxHash: common.HexToHash("0xb08f07656eaafa8efc458e2aa90773648d95ec8119873d212b4377dea5190cc0"), - Index: logIndex, - } -} - -// ndjson renders logs the way the exporter's slim format does. -func ndjson(t *testing.T, logs ...types.Log) []byte { - t.Helper() - - var buf bytes.Buffer - enc := json.NewEncoder(&buf) - for _, l := range logs { - if err := enc.Encode(filestore.NewSlimLog(l)); err != nil { - t.Fatal(err) - } - } - - return buf.Bytes() -} - -// gz compresses b as a single gzip member. -func gz(t *testing.T, b []byte) []byte { - t.Helper() - - var buf bytes.Buffer - w := gzip.NewWriter(&buf) - if _, err := w.Write(b); err != nil { - t.Fatal(err) - } - if err := w.Close(); err != nil { - t.Fatal(err) - } - - return buf.Bytes() -} - -// write puts content into a fresh temp file and returns its path. -func write(t *testing.T, name string, content []byte) string { - t.Helper() - - path := filepath.Join(t.TempDir(), name) - if err := os.WriteFile(path, content, 0o644); err != nil { - t.Fatal(err) - } - - return path -} - -func TestVerifySuperset(t *testing.T) { - t.Parallel() - - // The new snapshot is the old member plus an appended member, exactly - // the shape a resumed gzip export produces. - oldBlob := gz(t, ndjson(t, testLog(1, 0), testLog(2, 0))) - newBlob := append(bytes.Clone(oldBlob), gz(t, ndjson(t, testLog(2, 1), testLog(3, 0)))...) - - res, err := verify.Verify( - write(t, "old.ndjson.gzip", oldBlob), - write(t, "new.ndjson.gzip", newBlob), - ) - if err != nil { - t.Fatalf("Verify: %v", err) - } - if res.LastBlock != 3 { - t.Errorf("LastBlock = %d, want 3", res.LastBlock) - } - if res.Appended != 2 { - t.Errorf("Appended = %d, want 2", res.Appended) - } - if res.OldTruncated { - t.Error("OldTruncated = true, want false") - } -} - -func TestVerifyIdentical(t *testing.T) { - t.Parallel() - - blob := gz(t, ndjson(t, testLog(1, 0), testLog(2, 0))) - - res, err := verify.Verify( - write(t, "old.ndjson.gzip", blob), - write(t, "new.ndjson.gzip", blob), - ) - if err != nil { - t.Fatalf("Verify: %v", err) - } - if res.LastBlock != 2 || res.Appended != 0 { - t.Errorf("got LastBlock=%d Appended=%d, want 2 and 0", res.LastBlock, res.Appended) - } -} -``` - -- [ ] **Step 2: Run the tests to verify they fail** - -Run: `go test ./pkg/verify/ -v` -Expected: compile error — package `verify` does not exist. - -- [ ] **Step 3: Implement `pkg/verify/verify.go`** - -```go -// Package verify checks that a refreshed snapshot is a strict extension of -// the snapshot it was resumed from: everything the old file held, byte for -// byte and in order, followed only by newer entries. -package verify - -import ( - "bufio" - "bytes" - "errors" - "fmt" - "io" - - "github.com/ethersphere/batch-export/pkg/resume" -) - -var ( - // ErrMismatch indicates the new snapshot does not begin with the old - // snapshot's content: an entry was dropped, mutated, or reordered. - ErrMismatch = errors.New("new snapshot does not extend the old snapshot") - // ErrOrder indicates an appended entry out of (blockNumber, logIndex) - // order relative to what precedes it. - ErrOrder = errors.New("appended entries out of order") - // ErrTruncatedNew indicates the new snapshot ends in an interrupted - // write; a completed export never does. - ErrTruncatedNew = errors.New("new snapshot ends in an interrupted write") -) - -// chunkSize is how many bytes of both snapshots are compared per read. -const chunkSize = 64 * 1024 - -// Result reports what a successful verification established. -type Result struct { - // LastBlock is the block number of the new snapshot's final entry. - LastBlock uint64 - // Appended is how many entries the new snapshot adds. - Appended int - // OldTruncated reports whether the old snapshot carried an interrupted - // tail, excluded from the comparison the way resuming excludes it from - // the copy. - OldTruncated bool -} - -// Verify checks that the snapshot at newPath extends the one at oldPath. -func Verify(oldPath, newPath string) (Result, error) { - oldCursor, err := resume.Read(oldPath) - if err != nil { - return Result{}, fmt.Errorf("old snapshot: %w", err) - } - newCursor, err := resume.Read(newPath) - if err != nil { - return Result{}, fmt.Errorf("new snapshot: %w", err) - } - if newCursor.Truncated { - return Result{}, ErrTruncatedNew - } - - oldClean, err := resume.OpenClean(oldPath, oldCursor) - if err != nil { - return Result{}, fmt.Errorf("old snapshot: %w", err) - } - defer oldClean.Close() - - newClean, err := resume.OpenClean(newPath, newCursor) - if err != nil { - return Result{}, fmt.Errorf("new snapshot: %w", err) - } - defer newClean.Close() - - newBuffered := bufio.NewReaderSize(newClean, chunkSize) - if err := comparePrefix(oldClean, newBuffered); err != nil { - return Result{}, err - } - - appended, err := checkTail(newBuffered, oldCursor) - if err != nil { - return Result{}, err - } - - return Result{ - LastBlock: newCursor.BlockNumber, - Appended: appended, - OldTruncated: oldCursor.Truncated, - }, nil -} - -// comparePrefix requires every byte of oldR to appear at the start of newR. -func comparePrefix(oldR io.Reader, newR *bufio.Reader) error { - oldBuf := make([]byte, chunkSize) - newBuf := make([]byte, chunkSize) - - var offset int64 - for { - n, err := oldR.Read(oldBuf) - if n > 0 { - if _, rerr := io.ReadFull(newR, newBuf[:n]); rerr != nil { - return fmt.Errorf("%w: new snapshot ends at byte %d of the old content", ErrMismatch, offset) - } - if !bytes.Equal(oldBuf[:n], newBuf[:n]) { - return fmt.Errorf("%w: content diverges within bytes %d..%d of the old content", ErrMismatch, offset, offset+int64(n)) - } - offset += int64(n) - } - if errors.Is(err, io.EOF) { - return nil - } - if err != nil { - return fmt.Errorf("error reading old snapshot: %w", err) - } - } -} - -// checkTail walks the entries the new snapshot appends after the old -// content, requiring each to advance the (blockNumber, logIndex) order. -func checkTail(r *bufio.Reader, prev *resume.Cursor) (int, error) { - last := *prev - - var ( - appended int - lineNum int - line []byte - ) - for { - chunk, err := r.ReadSlice('\n') - line = append(line, chunk...) - if len(line) > resume.MaxLineBytes { - return 0, fmt.Errorf("appended line %d exceeds %d bytes", lineNum+1, resume.MaxLineBytes) - } - - switch { - case errors.Is(err, bufio.ErrBufferFull): - continue - case errors.Is(err, io.EOF): - if len(line) > 0 { - return 0, fmt.Errorf("appended line %d ends without a newline", lineNum+1) - } - return appended, nil - case err != nil: - return 0, fmt.Errorf("error reading new snapshot: %w", err) - } - - lineNum++ - entry, err := resume.ParseEntry(line) - if err != nil { - return 0, fmt.Errorf("appended line %d is not a log entry: %w", lineNum, err) - } - if !after(entry, &last) { - return 0, fmt.Errorf("%w: appended line %d holds (block %d, log %d) after (block %d, log %d)", - ErrOrder, lineNum, entry.BlockNumber, entry.LogIndex, last.BlockNumber, last.LogIndex) - } - last = *entry - appended++ - line = line[:0] - } -} - -// after reports whether entry strictly follows prev in log order. -func after(entry, prev *resume.Cursor) bool { - if entry.BlockNumber != prev.BlockNumber { - return entry.BlockNumber > prev.BlockNumber - } - return entry.LogIndex > prev.LogIndex -} -``` - -- [ ] **Step 4: Run the tests to verify they pass** - -Run: `go test ./pkg/verify/ -v` and `go vet ./...` -Expected: PASS. - -- [ ] **Step 5: Commit** - -```bash -git add pkg/verify/ -git commit -m "feat: add pkg/verify snapshot extension check" -``` - ---- - -### Task 3: `pkg/verify` — failure modes and truncation handling - -The implementation from Task 2 should already handle these; this task pins each failure mode with a test and fixes anything that surfaces. - -**Files:** -- Modify: `pkg/verify/verify_test.go` (append) -- Modify: `pkg/verify/verify.go` (only if a test exposes a gap) - -**Interfaces:** -- Consumes: everything Task 2 produced, plus `resume.ErrNoLogs`. - -- [ ] **Step 1: Write the tests** - -Append to `pkg/verify/verify_test.go` (add `"slices"`, `"errors"`, and `"github.com/ethersphere/batch-export/pkg/resume"` to the imports): - -```go -func TestVerifyRefusals(t *testing.T) { - t.Parallel() - - oldContent := ndjson(t, testLog(1, 0), testLog(2, 0)) - - // mutated flips one byte inside the first entry's blockNumber. - mutated := bytes.Replace(slices.Clone(oldContent), []byte(`"blockNumber":"0x1"`), []byte(`"blockNumber":"0x9"`), 1) - - tests := []struct { - name string - old []byte - new []byte - wantErr error - }{ - { - name: "mutated entry inside the old content", - old: gz(t, oldContent), - new: gz(t, append(mutated, ndjson(t, testLog(10, 0))...)), - wantErr: verify.ErrMismatch, - }, - { - name: "dropped entry", - old: gz(t, oldContent), - new: gz(t, ndjson(t, testLog(1, 0), testLog(3, 0))), - wantErr: verify.ErrMismatch, - }, - { - name: "new shorter than old", - old: gz(t, oldContent), - new: gz(t, ndjson(t, testLog(1, 0))), - wantErr: verify.ErrMismatch, - }, - { - name: "appended entry repeats the cursor", - old: gz(t, oldContent), - new: append(gz(t, oldContent), gz(t, ndjson(t, testLog(2, 0)))...), - wantErr: verify.ErrOrder, - }, - { - name: "appended entries out of order", - old: gz(t, oldContent), - new: append(gz(t, oldContent), gz(t, ndjson(t, testLog(5, 0), testLog(4, 0)))...), - wantErr: verify.ErrOrder, - }, - { - name: "new ends in an interrupted write", - old: oldContent, - new: append(slices.Clone(oldContent), `{"blockNumber":"0x3"`...), - wantErr: verify.ErrTruncatedNew, - }, - { - name: "old holds no complete entry", - old: nil, - new: oldContent, - wantErr: resume.ErrNoLogs, - }, - } - for _, tc := range tests { - t.Run(tc.name, func(t *testing.T) { - t.Parallel() - - _, err := verify.Verify( - write(t, "old", tc.old), - write(t, "new", tc.new), - ) - if !errors.Is(err, tc.wantErr) { - t.Fatalf("Verify error = %v, want %v", err, tc.wantErr) - } - }) - } -} - -func TestVerifyMalformedAppendedLine(t *testing.T) { - t.Parallel() - - oldContent := ndjson(t, testLog(1, 0), testLog(2, 0)) - // The middle appended line is valid JSON but not a log entry; resume.Read - // on a plain file only inspects the tail, so only checkTail can catch it. - newContent := append(slices.Clone(oldContent), "{\"foo\":1}\n"...) - newContent = append(newContent, ndjson(t, testLog(3, 0))...) - - _, err := verify.Verify( - write(t, "old", oldContent), - write(t, "new", newContent), - ) - if err == nil { - t.Fatal("want an error for a malformed appended line") - } -} - -func TestVerifyOldTruncated(t *testing.T) { - t.Parallel() - - oldContent := ndjson(t, testLog(1, 0), testLog(2, 0)) - oldBlob := append(slices.Clone(oldContent), `{"block`...) - newBlob := append(slices.Clone(oldContent), ndjson(t, testLog(3, 0))...) - - res, err := verify.Verify( - write(t, "old", oldBlob), - write(t, "new", newBlob), - ) - if err != nil { - t.Fatalf("Verify: %v", err) - } - if !res.OldTruncated { - t.Error("OldTruncated = false, want true") - } - if res.LastBlock != 3 || res.Appended != 1 { - t.Errorf("got LastBlock=%d Appended=%d, want 3 and 1", res.LastBlock, res.Appended) - } -} -``` - -- [ ] **Step 2: Run the tests** - -Run: `go test ./pkg/verify/ -v` -Expected: PASS if Task 2's implementation is complete. Any FAIL marks a real gap — fix `verify.go` minimally until green, and keep the fix inside the semantics the spec defines. - -- [ ] **Step 3: Commit** - -```bash -git add pkg/verify/ -git commit -m "test: cover verify failure modes and truncation handling" -``` - ---- - -### Task 4: `pkg/verify` — format combination matrix - -**Files:** -- Modify: `pkg/verify/verify_test.go` (append) - -**Interfaces:** -- Consumes: everything from Tasks 2–3. - -- [ ] **Step 1: Write the test** - -```go -func TestVerifyFormatCombinations(t *testing.T) { - t.Parallel() - - oldContent := ndjson(t, testLog(1, 0), testLog(2, 0)) - newContent := append(slices.Clone(oldContent), ndjson(t, testLog(3, 0))...) - - tests := []struct { - name string - old []byte - new []byte - }{ - {"plain old, plain new", oldContent, newContent}, - {"plain old, gzip new", oldContent, gz(t, newContent)}, - {"gzip old, plain new", gz(t, oldContent), newContent}, - {"gzip old, gzip new", gz(t, oldContent), gz(t, newContent)}, - } - for _, tc := range tests { - t.Run(tc.name, func(t *testing.T) { - t.Parallel() - - res, err := verify.Verify( - write(t, "old", tc.old), - write(t, "new", tc.new), - ) - if err != nil { - t.Fatalf("Verify: %v", err) - } - if res.LastBlock != 3 || res.Appended != 1 { - t.Errorf("got LastBlock=%d Appended=%d, want 3 and 1", res.LastBlock, res.Appended) - } - }) - } -} -``` - -- [ ] **Step 2: Run the test** - -Run: `go test ./pkg/verify/ -v` -Expected: PASS (the format handling all lives in `resume`; a FAIL here means `OpenClean` or the buffering broke a combination — fix there, not with special cases in verify). - -- [ ] **Step 3: Commit** - -```bash -git add pkg/verify/verify_test.go -git commit -m "test: cover verify across plain and gzip snapshot combinations" -``` - ---- - -### Task 5: `verify` subcommand and README - -**Files:** -- Create: `cmd/verify.go` -- Create: `cmd/verify_test.go` -- Modify: `cmd/cmd.go` (register the command in `newCommand`) -- Modify: `README.md` (new subsection after "Continuing a previous snapshot") - -**Interfaces:** -- Consumes: `verify.Verify`, `verify.Result` (Task 2); the `command` struct and `c.log` from `cmd/cmd.go`. -- Produces: `batch-export verify --old --new ` printing the last block in decimal on stdout; `func (c *command) initVerifyCmd() error`. - -- [ ] **Step 1: Write the failing tests** - -Create `cmd/verify_test.go` (internal test — `package cmd`, like `export_test.go`): - -```go -package cmd - -import ( - "bytes" - "compress/gzip" - "context" - "encoding/json" - "os" - "path/filepath" - "testing" - - "github.com/ethereum/go-ethereum/core/types" - "github.com/ethersphere/batch-export/pkg/filestore" -) - -// writeSnapshot writes a slim-format gzip snapshot with one entry per -// (blockNumber, logIndex) pair and returns its path. -func writeSnapshot(t *testing.T, name string, entries ...[2]uint64) string { - t.Helper() - - var buf bytes.Buffer - w := gzip.NewWriter(&buf) - enc := json.NewEncoder(w) - for _, e := range entries { - if err := enc.Encode(filestore.NewSlimLog(types.Log{BlockNumber: e[0], Index: uint(e[1])})); err != nil { - t.Fatal(err) - } - } - if err := w.Close(); err != nil { - t.Fatal(err) - } - - path := filepath.Join(t.TempDir(), name) - if err := os.WriteFile(path, buf.Bytes(), 0o644); err != nil { - t.Fatal(err) - } - - return path -} - -func TestVerifyCmd(t *testing.T) { - t.Parallel() - - oldPath := writeSnapshot(t, "old.ndjson.gzip", [2]uint64{1, 0}, [2]uint64{2, 0}) - newPath := writeSnapshot(t, "new.ndjson.gzip", [2]uint64{1, 0}, [2]uint64{2, 0}, [2]uint64{3, 0}) - - c, err := newCommand() - if err != nil { - t.Fatal(err) - } - var out bytes.Buffer - c.root.SetOut(&out) - c.root.SetArgs([]string{"verify", "--old", oldPath, "--new", newPath}) - - if err := c.Execute(context.Background()); err != nil { - t.Fatalf("verify: %v", err) - } - if got, want := out.String(), "3\n"; got != want { - t.Errorf("stdout = %q, want %q", got, want) - } -} - -func TestVerifyCmdRefusal(t *testing.T) { - t.Parallel() - - oldPath := writeSnapshot(t, "old.ndjson.gzip", [2]uint64{1, 0}, [2]uint64{2, 0}) - newPath := writeSnapshot(t, "new.ndjson.gzip", [2]uint64{1, 0}, [2]uint64{3, 0}) - - c, err := newCommand() - if err != nil { - t.Fatal(err) - } - var out bytes.Buffer - c.root.SetOut(&out) - c.root.SetArgs([]string{"verify", "--old", oldPath, "--new", newPath}) - - if err := c.Execute(context.Background()); err == nil { - t.Fatal("want an error for a snapshot that drops an entry") - } - if out.Len() != 0 { - t.Errorf("stdout = %q, want empty", out.String()) - } -} -``` - -- [ ] **Step 2: Run the tests to verify they fail** - -Run: `go test ./cmd/ -run TestVerifyCmd -v` -Expected: FAIL — cobra reports `unknown command "verify"` (surfaced as an `Execute` error in the first test). - -- [ ] **Step 3: Implement `cmd/verify.go` and register it** - -Create `cmd/verify.go`: - -```go -package cmd - -import ( - "fmt" - - "github.com/ethersphere/batch-export/pkg/verify" - "github.com/spf13/cobra" -) - -func (c *command) initVerifyCmd() error { - var ( - oldFile string - newFile string - ) - - cmd := &cobra.Command{ - Use: "verify", - Short: "Verify that a refreshed snapshot extends the one it was resumed from", - Long: `Verifies that the snapshot at --new holds everything the snapshot at --old does, -byte for byte and in order, followed only by newer entries. On success the new -snapshot's last block number is printed to stdout in decimal; any failure exits -non-zero with nothing on stdout.`, - RunE: func(cmd *cobra.Command, args []string) error { - result, err := verify.Verify(oldFile, newFile) - if err != nil { - return err - } - - if result.OldTruncated { - c.log.Warning("old snapshot ends in an interrupted write; the truncated tail was excluded from the comparison", "oldFile", oldFile) - } - c.log.Info("snapshot verified", "oldFile", oldFile, "newFile", newFile, "appended", result.Appended, "lastBlock", result.LastBlock) - - _, err = fmt.Fprintln(cmd.OutOrStdout(), result.LastBlock) - return err - }, - } - - cmd.Flags().StringVar(&oldFile, "old", "", "Snapshot the export resumed from (.ndjson, .gz or .gzip)") - cmd.Flags().StringVar(&newFile, "new", "", "Freshly exported snapshot to check (.ndjson, .gz or .gzip)") - for _, name := range []string{"old", "new"} { - if err := cmd.MarkFlagRequired(name); err != nil { - return err - } - } - - c.root.AddCommand(cmd) - - return nil -} -``` - -In `cmd/cmd.go`, register it right after the export command: - -```go - if err := c.initExportCmd(); err != nil { - return nil, err - } - - if err := c.initVerifyCmd(); err != nil { - return nil, err - } -``` - -- [ ] **Step 4: Run the tests to verify they pass** - -Run: `go test ./cmd/ -v` and `go test ./...` and `go vet ./...` -Expected: PASS. - -- [ ] **Step 5: Add the README section** - -In `README.md`, directly after the "Continuing a previous snapshot" section's content (before the next `##` heading), add: - -```markdown -### Verifying a snapshot - -`verify` proves a refreshed snapshot still holds everything the original did — -byte for byte, in order — followed only by newer entries, and prints the new -snapshot's last block number in decimal: - -```sh -batch-export verify --old export.ndjson.gzip --new snapshot.ndjson.gzip -``` - -A non-zero exit means the new snapshot must not replace the old one. -``` - -- [ ] **Step 6: Commit** - -```bash -git add cmd/verify.go cmd/verify_test.go cmd/cmd.go README.md -git commit -m "feat: add verify command gating snapshot publication" -``` - ---- - -### Task 6: Workflow wiring and PR - -**Files:** -- Modify: `.github/workflows/batch-sync.yml` - -**Interfaces:** -- Consumes: the `verify` subcommand's stdout contract (Task 5); the workflow's existing `resume.ndjson.gzip` / `snapshot.ndjson.gzip` step outputs. -- Produces: `LAST_BLOCK` in `GITHUB_ENV`, consumed by the Publish step. - -- [ ] **Step 1: Add the Verify step** - -In `.github/workflows/batch-sync.yml`, insert between the `Export` and `Publish to batch-archive` steps: - -```yaml - # A failed verification stops the job before any commit or tag exists. - - name: Verify snapshot - run: | - set -euo pipefail - last_block="$(./dist/batch-export verify --old resume.ndjson.gzip --new snapshot.ndjson.gzip)" - echo "LAST_BLOCK=${last_block}" >> "${GITHUB_ENV}" -``` - -> Superseded during the final review: `--old` ships as -> `batch-archive/archive/export.ndjson.gzip`, the file the publish step -> overwrites, because a non-tip `archive_tag` makes the resumed file and the -> replaced file diverge. See the spec's CLI contract. - -- [ ] **Step 2: Replace the sed extraction in the Publish step** - -Delete this block from `Publish to batch-archive`: - -```yaml - # blockNumber is hex ("0x...") in the slim NDJSON; the commit title - # uses decimal, matching the archive's history. - last_block_hex="$(gunzip -c archive/export.ndjson.gzip | tail -n 1 \ - | sed -nE 's/.*"blockNumber":"(0x[0-9a-fA-F]+)".*/\1/p')" - if [ -z "${last_block_hex}" ]; then - echo "::error::could not read blockNumber from the last snapshot entry" - exit 1 - fi - last_block="$(printf '%d' "${last_block_hex}")" -``` - -and change both remaining uses of `${last_block}` to `${LAST_BLOCK}` — one in `git commit -m "chore: update snapshot to block number ${last_block}"`, one in the final `::notice::published snapshot at block ${last_block} ...` line. - -- [ ] **Step 3: Validate** - -Run: -```bash -python3 -c "import yaml; yaml.safe_load(open('.github/workflows/batch-sync.yml')); print('YAML OK')" -actionlint .github/workflows/batch-sync.yml || true # only the known custom runner-label warning is acceptable -grep -n "last_block" .github/workflows/batch-sync.yml # expect: only the Verify step's local variable, no stale lowercase uses in Publish -``` -Expected: YAML OK; no actionlint findings beyond the `bee` runner label; grep shows no `${last_block}` left in the Publish step. - -- [ ] **Step 4: Full check and commit** - -Run: `go test ./...` and `go vet ./...` -Expected: PASS. - -```bash -git add .github/workflows/batch-sync.yml -git commit -m "ci: gate publishing on snapshot verification" -``` - -- [ ] **Step 5: Push and open the PR** - -```bash -git push -u origin feat/verify-snapshot -gh pr create --title "feat: verify refreshed snapshots before publishing" --body "$(cat <<'EOF' -Adds a `verify` subcommand and wires it into the Batch Sync workflow as a regression gate before any commit or tag reaches batch-archive. - -## What it does - -- `batch-export verify --old --new ` proves the new snapshot strictly extends the old one: the old content must appear byte for byte at the start of the new file, and every appended entry must parse and advance the (blockNumber, logIndex) order. On success it prints the new last block number in decimal on stdout; any failure exits non-zero. -- `pkg/verify` builds entirely on `pkg/resume`'s existing cursor and format handling; `resume` newly exports `MaxLineBytes`, `ParseEntry`, and `OpenClean` so no consumer re-encodes the on-disk format. -- The workflow runs `verify` between Export and Publish. Its stdout replaces the previous `gunzip | tail | sed` block-number extraction, so the commit title's block number now comes from the same Go code that defines the snapshot format. - -Spec: `docs/superpowers/specs/2026-08-31-verify-snapshot-design.md` - -🤖 Generated with [Claude Code](https://claude.com/claude-code) -EOF -)" -``` diff --git a/docs/superpowers/specs/2026-08-31-verify-snapshot-design.md b/docs/superpowers/specs/2026-08-31-verify-snapshot-design.md deleted file mode 100644 index 8c0dd0e..0000000 --- a/docs/superpowers/specs/2026-08-31-verify-snapshot-design.md +++ /dev/null @@ -1,149 +0,0 @@ -# `batch-export verify` — snapshot regression gate - -- **Date:** 2026-08-31 -- **Status:** proposed -- **Scope:** one subcommand + its package, and one workflow step wiring it in - -## Problem - -The Batch Sync workflow resumes the postage-batch snapshot from a tag of -[ethersphere/batch-archive](https://github.com/ethersphere/batch-archive) and -publishes the refreshed file back as a commit on `main` plus the next patch -tag. Nothing verifies that the refreshed snapshot is a strict extension of the -one it resumed from. A regression in batch-export — in the resume logic or the -gzip append handling — could silently drop, mutate, or reorder historical -entries, and the workflow would tag and publish the corrupted snapshot as the -new latest version. (The gate does not catch a regression confined to the -slim NDJSON encoding itself: `ParseEntry` only requires `blockNumber` and -`logIndex`, so an appended entry written in the full geth shape still -verifies clean. The encoding is pinned separately, by `pkg/filestore`'s own -tests.) - -Separately, the workflow extracts the last block number for the commit title -with a `gunzip | tail | sed` pipeline whose regex re-encodes the slim JSON -shape by hand. Nothing ties that regex to the Go struct that defines the -format, so a format change passes every Go test and breaks only in CI. - -## Goal - -A `batch-export verify` subcommand that: - -1. proves the new snapshot contains everything the old one did, byte for byte - and in order, with only new entries appended; -2. validates the appended entries parse and are correctly ordered; -3. prints the new snapshot's last block number, replacing the sed pipeline. - -Wired into the Batch Sync workflow between the Export and Publish steps, so a -failed verification blocks the commit and the tag. - -## Non-goals - -- No chain-side validation: `verify` never queries an RPC endpoint. -- No repair: a bad snapshot is reported, never fixed. -- No change to export behavior, scheduling, or bootstrap flows. - -## CLI contract - -``` -batch-export verify --old --new [--verbosity ] -``` - -- `--old` (required): the snapshot the new one must extend — in the workflow, - the archive file the publish step overwrites, which is what a regression - would actually destroy. That is the same file the export resumed from - whenever the resumed tag is main's tip, and a stricter check when it is not. -- `--new` (required): the freshly exported snapshot. -- Formats: plain NDJSON or gzip, each file detected independently by content - (the same detection `--resume` uses); any old/new combination is accepted. -- **stdout** on success: the new snapshot's last block number, decimal, one - line, nothing else — safe for `$(...)` capture in the workflow. -- **stderr**: all logging and diagnostics. -- **Exit code**: 0 = verified; non-zero = do not publish (verification - failure, I/O error, and parse error are deliberately not distinguished — - the workflow reacts identically to all of them). - -## Verification algorithm - -1. **Cursor of the old file** via the existing `resume.Read`: yields the last - complete entry's `(blockNumber, logIndex)`, the compressed flag, and the - clean size. `resume.Read`'s refusals fail verification with the same - meaning they have on resume: `ErrNotAnExport` (file was altered) and - `ErrNoLogs` (nothing to extend). -2. **Prefix property.** The new file's decompressed content must begin, byte - for byte, with the old file's *clean content* — the same bytes - `resume.PrepareOutput` copies (for a gzip old file: the compressed bytes up - to the cursor's clean size, decompressed; for plain NDJSON: the file up to - clean size). Comparison is streamed with constant memory; files are never - loaded whole. If the old file carries a truncated tail past its clean - size, the tail is excluded from the comparison and a warning is logged — - mirroring what resume itself discards. -3. **Appended tail.** Every line of the new file past the prefix must: - - stay under the same line-length cap resume enforces; - - parse as an export log entry (slim or full shape, as resume accepts); - - be strictly increasing in `(blockNumber, logIndex)` order, with the - first appended entry strictly greater than the old file's cursor — - matching resume's skip semantics, which re-query the cursor block but - never re-write entries at or before the cursor. -4. **Empty tail is valid**: a run that found no new logs verifies - successfully and prints the old cursor's block number. -5. **Output**: the last appended entry's block number (or the old cursor's, - when the tail is empty), printed in decimal on stdout. - -Failure messages name the location: byte offset and line number of the first -divergence for a prefix mismatch; line number and reason for a tail failure. - -## Code layout - -``` -cmd/verify.go thin cobra command: flags, logger, calls pkg/verify, - prints the block number (mirrors cmd/export.go style) -pkg/verify/verify.go Verify(oldPath, newPath) (Result, error); - Result{LastBlock uint64, Appended int} -pkg/verify/verify_test.go unit tests with generated fixture files -``` - -`pkg/verify` reuses `pkg/resume` for cursor reading and format handling. If a -helper it needs (e.g. the decompressing opener) is unexported in `pkg/resume`, -the minimal piece is exported from there rather than duplicated. - -## Workflow integration (same PR) - -`.github/workflows/batch-sync.yml` gains one step between Export and Publish: - -```yaml - - name: Verify snapshot - run: | - set -euo pipefail - last_block="$(./dist/batch-export verify \ - --old batch-archive/archive/export.ndjson.gzip \ - --new snapshot.ndjson.gzip)" - echo "LAST_BLOCK=${last_block}" >> "${GITHUB_ENV}" -``` - -`--old` is the archive's current file rather than the resumed tag's: the -publish step overwrites it, so it is what a regression would destroy. The -two are the same bytes whenever the resumed tag is main's tip. - -The Publish step then uses `${LAST_BLOCK}` in the commit title and drops the -`gunzip | tail | sed` block together with its empty-result guard. A verify -failure stops the job before any commit or tag is created. - -## Testing - -- **Unit (`pkg/verify`)**, fixtures generated per test: identical files; - proper superset; corrupted byte inside the prefix; new file shorter than - old; non-monotonic appended entry; duplicate of the cursor entry; malformed - JSON tail line; old file with no complete entry; old file with a truncated - tail; plain/gzip in all four combinations. -- **Command (`cmd`)**: exit code and stdout shape for one passing and one - failing pair. -- **Workflow**: `actionlint` locally; end-to-end via a manual dispatch after - merge. - -Development is test-first (TDD), consistent with the repo's existing -`pkg/resume` test style. - -## Delivery - -Branch `feat/verify-snapshot` off `main`; conventional commits (`feat:` for -the subcommand, `ci:` for the workflow wiring); one PR to `main`. From 778ece7189fdb69deb7e352ff783b49ba0bfa6da Mon Sep 17 00:00:00 2001 From: Ljubisa Gacevic Date: Mon, 31 Aug 2026 22:10:46 +0200 Subject: [PATCH 12/12] refactor: share the line reader and ordering rule with pkg/verify checkTail had transcribed scanMember's line loop, which is what forced maxLineBytes to be exported; it now calls resume.ReadLine, and the cap is unexported again. after() restated Cursor.Skip inverted, so both go through the new Cursor.Before. parseCursor is renamed ParseEntry, dropping the pass-through wrapper, and OpenClean becomes a Cursor method so its precondition is in the signature. LAST_BLOCK now crosses steps through GITHUB_OUTPUT rather than the job-wide env, with the empty-value guard the old shell pipeline used to carry, and the step comments are trimmed to what the code cannot say itself. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01LqPjhcvFr2LLuBdhyjQwCe --- .github/workflows/batch-sync.yml | 21 +++---- cmd/verify.go | 7 +-- cmd/verify_test.go | 79 +++++++++++++----------- pkg/resume/resume.go | 103 +++++++++++++++++++------------ pkg/resume/resume_test.go | 2 +- pkg/verify/verify.go | 46 ++++---------- 6 files changed, 132 insertions(+), 126 deletions(-) diff --git a/.github/workflows/batch-sync.yml b/.github/workflows/batch-sync.yml index 9deffb2..b3024a4 100644 --- a/.github/workflows/batch-sync.yml +++ b/.github/workflows/batch-sync.yml @@ -151,32 +151,29 @@ jobs: --slim=true \ --verbosity "${VERBOSITY}" - # A failed verification stops the job before any commit or tag exists. - # # --old is the file Publish overwrites (main's current archive), not the - # resumed tag: a non-tip archive_tag plus a stale finalized head could - # otherwise verify clean against the tag while still losing entries main - # already has. Identical to the resumed tag in the normal case (resumed - # tag == main's tip), so this is a no-op then. An archive version - # written before the slim format would fail closed here, which is the - # intended answer. + # resumed tag: a non-tip archive_tag could otherwise verify clean against + # the tag while still losing entries main already has. A failure stops + # the job before any commit or tag exists. - name: Verify snapshot + id: verify run: | set -euo pipefail last_block="$(./dist/batch-export verify --old batch-archive/archive/export.ndjson.gzip --new snapshot.ndjson.gzip)" - echo "LAST_BLOCK=${last_block}" >> "${GITHUB_ENV}" + echo "last_block=${last_block}" >> "${GITHUB_OUTPUT}" - name: Publish to batch-archive env: TRIGGERED_BY: ${{ github.actor }} + LAST_BLOCK: ${{ steps.verify.outputs.last_block }} run: | set -euo pipefail + # Empty only if Verify was skipped or reordered; never commit without it. + : "${LAST_BLOCK:?the verify step reported no block number}" cp snapshot.ndjson.gzip batch-archive/archive/export.ndjson.gzip cd batch-archive - # Cannot fire today: a resumed gzip export always appends a ~20-byte - # empty gzip member even when zero new logs were fetched, so the - # file always differs. Kept in case that stops being true. + # Never fires today: a resumed gzip export always appends a member. if git diff --quiet; then echo "::notice::snapshot is unchanged; nothing to publish" exit 0 diff --git a/cmd/verify.go b/cmd/verify.go index 21cbb9b..64a0537 100644 --- a/cmd/verify.go +++ b/cmd/verify.go @@ -16,10 +16,9 @@ func (c *command) initVerifyCmd() error { cmd := &cobra.Command{ Use: "verify", Short: "Verify that a refreshed snapshot extends the one it was resumed from", - Long: `Verifies that the snapshot at --new holds everything the snapshot at --old does, -byte for byte and in order, followed only by newer entries. On success the new -snapshot's last block number is printed to stdout in decimal; any failure exits -non-zero with nothing on stdout.`, + Long: `Verifies that --new begins with the entire content of --old, byte for byte, +and that every entry after it is strictly newer. Prints the new snapshot's last +block number in decimal on stdout; any failure exits non-zero with empty stdout.`, RunE: func(cmd *cobra.Command, args []string) error { result, err := verify.Verify(oldFile, newFile) if err != nil { diff --git a/cmd/verify_test.go b/cmd/verify_test.go index 6b86b52..f73c908 100644 --- a/cmd/verify_test.go +++ b/cmd/verify_test.go @@ -41,43 +41,52 @@ func writeSnapshot(t *testing.T, name string, entries ...[2]uint64) string { func TestVerifyCmd(t *testing.T) { t.Parallel() - oldPath := writeSnapshot(t, "old.ndjson.gzip", [2]uint64{1, 0}, [2]uint64{2, 0}) - newPath := writeSnapshot(t, "new.ndjson.gzip", [2]uint64{1, 0}, [2]uint64{2, 0}, [2]uint64{3, 0}) - - c, err := newCommand() - if err != nil { - t.Fatal(err) - } - var out bytes.Buffer - c.root.SetOut(&out) - c.root.SetArgs([]string{"verify", "--old", oldPath, "--new", newPath}) - - if err := c.Execute(context.Background()); err != nil { - t.Fatalf("verify: %v", err) - } - if got, want := out.String(), "3\n"; got != want { - t.Errorf("stdout = %q, want %q", got, want) + tests := []struct { + name string + old [][2]uint64 + new [][2]uint64 + wantErr bool + wantOut string + }{ + { + name: "extension prints the last block", + old: [][2]uint64{{1, 0}, {2, 0}}, + new: [][2]uint64{{1, 0}, {2, 0}, {3, 0}}, + wantOut: "3\n", + }, + { + name: "a dropped entry is refused with empty stdout", + old: [][2]uint64{{1, 0}, {2, 0}}, + new: [][2]uint64{{1, 0}, {3, 0}}, + wantErr: true, + }, } -} + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() -func TestVerifyCmdRefusal(t *testing.T) { - t.Parallel() - - oldPath := writeSnapshot(t, "old.ndjson.gzip", [2]uint64{1, 0}, [2]uint64{2, 0}) - newPath := writeSnapshot(t, "new.ndjson.gzip", [2]uint64{1, 0}, [2]uint64{3, 0}) - - c, err := newCommand() - if err != nil { - t.Fatal(err) - } - var out bytes.Buffer - c.root.SetOut(&out) - c.root.SetArgs([]string{"verify", "--old", oldPath, "--new", newPath}) + c, err := newCommand() + if err != nil { + t.Fatal(err) + } + var out bytes.Buffer + c.root.SetOut(&out) + c.root.SetArgs([]string{ + "verify", + "--old", writeSnapshot(t, "old.ndjson.gzip", tc.old...), + "--new", writeSnapshot(t, "new.ndjson.gzip", tc.new...), + }) - if err := c.Execute(context.Background()); err == nil { - t.Fatal("want an error for a snapshot that drops an entry") - } - if out.Len() != 0 { - t.Errorf("stdout = %q, want empty", out.String()) + err = c.Execute(context.Background()) + if tc.wantErr && err == nil { + t.Fatal("verify: want an error") + } + if !tc.wantErr && err != nil { + t.Fatalf("verify: %v", err) + } + if got := out.String(); got != tc.wantOut { + t.Errorf("stdout = %q, want %q", got, tc.wantOut) + } + }) } } diff --git a/pkg/resume/resume.go b/pkg/resume/resume.go index cb074ab..d7b008d 100644 --- a/pkg/resume/resume.go +++ b/pkg/resume/resume.go @@ -19,9 +19,9 @@ import ( ) const ( - // MaxLineBytes caps a single line. Exported log lines run to a few + // maxLineBytes caps a single line. Exported log lines run to a few // hundred bytes, so anything longer is not this tool's output. - MaxLineBytes = 1 << 20 + maxLineBytes = 1 << 20 // bufferSize is how much of a gzip file is buffered per read. bufferSize = 64 * 1024 ) @@ -34,6 +34,11 @@ var ( // complete entry to resume from: it is empty, or holds only an // interrupted first write. The remedy is a fresh export. ErrNoLogs = errors.New("no complete log entry found") + + // errLineTooLong indicates a line over maxLineBytes. + errLineTooLong = errors.New("line exceeds the maximum length") + // errPartialLine indicates a stream ending without a final newline. + errPartialLine = errors.New("line ends without a newline") ) // Cursor marks the last log entry saved by a previous export, together with @@ -54,12 +59,18 @@ type Cursor struct { Truncated bool } +// Before reports whether the cursor's entry precedes the entry at +// (blockNumber, logIndex) in export order. +func (c *Cursor) Before(blockNumber uint64, logIndex uint) bool { + if blockNumber != c.BlockNumber { + return blockNumber > c.BlockNumber + } + return logIndex > c.LogIndex +} + // Skip reports whether l was already written to the file the cursor came from. func (c *Cursor) Skip(l types.Log) bool { - if l.BlockNumber != c.BlockNumber { - return l.BlockNumber < c.BlockNumber - } - return l.Index <= c.LogIndex + return !c.Before(l.BlockNumber, l.Index) } // PrepareOutput readies outputPath for appending the continuation of the @@ -206,7 +217,7 @@ func isGzip(file *os.File) (bool, error) { // line must parse as a log entry, and the only bytes allowed after it are a // single interrupted write — a trailing fragment with no newline. func lastCursorPlain(file *os.File, size int64) (*Cursor, error) { - window := min(size, 2*MaxLineBytes) + window := min(size, 2*maxLineBytes) offset := size - window buf := make([]byte, window) @@ -218,12 +229,12 @@ func lastCursorPlain(file *os.File, size int64) (*Cursor, error) { if nl < 0 { // No newline at all: an interrupted first write, unless the file is // longer than any single line the tool writes. - if size > MaxLineBytes { + if size > maxLineBytes { return nil, fmt.Errorf("%w: no newline in the final %d bytes (from offset %d)", ErrNotAnExport, window, offset) } return nil, ErrNoLogs } - if tail := window - int64(nl) - 1; tail > MaxLineBytes { + if tail := window - int64(nl) - 1; tail > maxLineBytes { return nil, fmt.Errorf("%w: %d bytes without a newline after offset %d", ErrNotAnExport, tail, offset+int64(nl)+1) } @@ -231,7 +242,7 @@ func lastCursorPlain(file *os.File, size int64) (*Cursor, error) { if start == 0 && offset > 0 { return nil, fmt.Errorf("%w: final line is over %d bytes long", ErrNotAnExport, nl) } - cursor, err := parseCursor(buf[start:nl]) + cursor, err := ParseEntry(buf[start:nl]) if err != nil { return nil, fmt.Errorf("%w: final complete line at offset %d is not a log entry: %w", ErrNotAnExport, offset+int64(start), err) } @@ -352,34 +363,57 @@ func scanMember(r io.Reader, member int) (*Cursor, error) { lineNum = 1 ) for { - chunk, err := buffered.ReadSlice('\n') - line = append(line, chunk...) - if len(line) > MaxLineBytes { - return nil, fmt.Errorf("%w: member %d, line %d exceeds %d bytes", ErrNotAnExport, member, lineNum, MaxLineBytes) - } - + var err error + line, err = ReadLine(line[:0], buffered) switch { - case errors.Is(err, bufio.ErrBufferFull): - continue + case errors.Is(err, errLineTooLong): + return nil, fmt.Errorf("%w: member %d, line %d exceeds %d bytes", ErrNotAnExport, member, lineNum, maxLineBytes) + case errors.Is(err, errPartialLine): + return nil, fmt.Errorf("%w: member %d, line %d ends mid-line", ErrNotAnExport, member, lineNum) case errors.Is(err, io.EOF): - if len(line) > 0 { - return nil, fmt.Errorf("%w: member %d, line %d ends mid-line", ErrNotAnExport, member, lineNum) - } return last, nil case err != nil: return last, err } - cursor, err := parseCursor(line) + cursor, err := ParseEntry(line) if err != nil { return nil, fmt.Errorf("%w: member %d, line %d is not a log entry: %w", ErrNotAnExport, member, lineNum, err) } last = cursor - line = line[:0] lineNum++ } } +// ReadLine appends the next newline-terminated line from r to dst and returns +// the extended buffer; pass dst[:0] to reuse it across calls. A clean end of +// stream returns io.EOF. A final line with no newline — an interrupted write — +// and a line longer than any this tool writes are both returned as errors for +// the caller to wrap with the sentinel that fits its context. +func ReadLine(dst []byte, r *bufio.Reader) ([]byte, error) { + for { + chunk, err := r.ReadSlice('\n') + dst = append(dst, chunk...) + if len(dst) > maxLineBytes { + return dst, errLineTooLong + } + + switch { + case errors.Is(err, bufio.ErrBufferFull): + continue + case errors.Is(err, io.EOF): + if len(dst) > 0 { + return dst, errPartialLine + } + return dst, io.EOF + case err != nil: + return dst, err + } + + return dst, nil + } +} + // countingReader counts the bytes consumed from the reader it wraps. // // It must keep implementing io.ByteReader: that is what makes gzip.Reader read @@ -408,9 +442,10 @@ func (c *countingReader) ReadByte() (byte, error) { return b, err } -// parseCursor builds a cursor from a single NDJSON line. Callers wrap the -// returned error with the sentinel that fits their context. -func parseCursor(line []byte) (*Cursor, error) { +// ParseEntry builds a cursor from a single exported NDJSON line, in either +// the slim or the full log shape. Callers wrap the returned error with the +// sentinel that fits their context. +func ParseEntry(line []byte) (*Cursor, error) { line = bytes.TrimSpace(line) if len(line) == 0 { return nil, errors.New("blank line") @@ -430,19 +465,9 @@ func parseCursor(line []byte) (*Cursor, error) { }, nil } -// ParseEntry builds a cursor from a single exported NDJSON line, accepting -// both the slim and the full log shape. It is the only place outside the -// writer where the on-disk field names are known, so consumers never -// re-encode them. -func ParseEntry(line []byte) (*Cursor, error) { - return parseCursor(line) -} - -// OpenClean returns the decompressed form of the clean content at path — the -// same content PrepareOutput carries over, but decompressed rather than -// PrepareOutput's raw copy. c must come from Read on the same, unmodified -// file. -func OpenClean(path string, c *Cursor) (io.ReadCloser, error) { +// OpenClean returns the clean content at path, decompressed. The cursor must +// come from Read on that same, unmodified file. +func (c *Cursor) OpenClean(path string) (io.ReadCloser, error) { file, err := os.Open(path) if err != nil { return nil, fmt.Errorf("error opening resume file: %w", err) diff --git a/pkg/resume/resume_test.go b/pkg/resume/resume_test.go index 16212d3..5657b05 100644 --- a/pkg/resume/resume_test.go +++ b/pkg/resume/resume_test.go @@ -1298,7 +1298,7 @@ func TestOpenClean(t *testing.T) { t.Fatalf("Read: %v", err) } - r, err := resume.OpenClean(path, cursor) + r, err := cursor.OpenClean(path) if err != nil { t.Fatalf("OpenClean: %v", err) } diff --git a/pkg/verify/verify.go b/pkg/verify/verify.go index c291def..8a5eb06 100644 --- a/pkg/verify/verify.go +++ b/pkg/verify/verify.go @@ -34,9 +34,8 @@ type Result struct { LastBlock uint64 // Appended is how many entries the new snapshot adds. Appended int - // OldTruncated reports whether the old snapshot carried an interrupted - // tail, excluded from the comparison the way resuming excludes it from - // the copy. + // OldTruncated reports whether the old snapshot's interrupted tail was + // excluded from the comparison. OldTruncated bool } @@ -54,13 +53,13 @@ func Verify(oldPath, newPath string) (Result, error) { return Result{}, ErrTruncatedNew } - oldClean, err := resume.OpenClean(oldPath, oldCursor) + oldClean, err := oldCursor.OpenClean(oldPath) if err != nil { return Result{}, fmt.Errorf("old snapshot: %w", err) } defer oldClean.Close() - newClean, err := resume.OpenClean(newPath, newCursor) + newClean, err := newCursor.OpenClean(newPath) if err != nil { return Result{}, fmt.Errorf("new snapshot: %w", err) } @@ -121,30 +120,17 @@ func checkTail(r *bufio.Reader, prev *resume.Cursor) (int, error) { last := *prev var ( - appended int - lineNum int - line []byte + lineNum int + line []byte ) for { - chunk, err := r.ReadSlice('\n') - line = append(line, chunk...) - if len(line) > resume.MaxLineBytes { - return 0, fmt.Errorf("appended line %d exceeds %d bytes", lineNum+1, resume.MaxLineBytes) - } - + var err error + line, err = resume.ReadLine(line[:0], r) switch { - case errors.Is(err, bufio.ErrBufferFull): - continue case errors.Is(err, io.EOF): - // Defensive: Verify rejects a truncated new file up front, and a - // clean gzip member ending mid-line is already refused by - // resume.Read, so this should be unreachable in practice. - if len(line) > 0 { - return 0, fmt.Errorf("appended line %d ends without a newline", lineNum+1) - } - return appended, nil + return lineNum, nil case err != nil: - return 0, fmt.Errorf("error reading new snapshot: %w", err) + return 0, fmt.Errorf("appended line %d: %w", lineNum+1, err) } lineNum++ @@ -152,20 +138,10 @@ func checkTail(r *bufio.Reader, prev *resume.Cursor) (int, error) { if err != nil { return 0, fmt.Errorf("appended line %d is not a log entry: %w", lineNum, err) } - if !after(entry, &last) { + if !last.Before(entry.BlockNumber, entry.LogIndex) { return 0, fmt.Errorf("%w: appended line %d holds (block %d, log %d) after (block %d, log %d)", ErrOrder, lineNum, entry.BlockNumber, entry.LogIndex, last.BlockNumber, last.LogIndex) } last = *entry - appended++ - line = line[:0] - } -} - -// after reports whether entry strictly follows prev in log order. -func after(entry, prev *resume.Cursor) bool { - if entry.BlockNumber != prev.BlockNumber { - return entry.BlockNumber > prev.BlockNumber } - return entry.LogIndex > prev.LogIndex }