diff --git a/.github/workflows/batch-sync.yml b/.github/workflows/batch-sync.yml index 4a9fc78..b3024a4 100644 --- a/.github/workflows/batch-sync.yml +++ b/.github/workflows/batch-sync.yml @@ -151,29 +151,34 @@ jobs: --slim=true \ --verbosity "${VERBOSITY}" + # --old is the file Publish overwrites (main's current archive), not the + # 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_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 + # 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 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 +193,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})" 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 ebca32c..b6896ad 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 +./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. + ## 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..64a0537 --- /dev/null +++ b/cmd/verify.go @@ -0,0 +1,49 @@ +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 --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 { + 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..f73c908 --- /dev/null +++ b/cmd/verify_test.go @@ -0,0 +1,92 @@ +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() + + 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() + + 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...), + }) + + 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 a444615..d7b008d 100644 --- a/pkg/resume/resume.go +++ b/pkg/resume/resume.go @@ -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 @@ -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") @@ -429,3 +464,33 @@ func parseCursor(line []byte) (*Cursor, error) { LogIndex: uint(*parsed.LogIndex), }, nil } + +// 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) + } + + 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 resume 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..5657b05 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 := cursor.OpenClean(path) + 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) + } + }) + } +} diff --git a/pkg/verify/verify.go b/pkg/verify/verify.go new file mode 100644 index 0000000..8a5eb06 --- /dev/null +++ b/pkg/verify/verify.go @@ -0,0 +1,147 @@ +// 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's interrupted tail was + // excluded from the comparison. + 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 := oldCursor.OpenClean(oldPath) + if err != nil { + return Result{}, fmt.Errorf("old snapshot: %w", err) + } + defer oldClean.Close() + + newClean, err := newCursor.OpenClean(newPath) + 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 { + _, 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)) + } + 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 ( + lineNum int + line []byte + ) + for { + var err error + line, err = resume.ReadLine(line[:0], r) + switch { + case errors.Is(err, io.EOF): + return lineNum, nil + case err != nil: + return 0, fmt.Errorf("appended line %d: %w", lineNum+1, 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 !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 + } +} diff --git a/pkg/verify/verify_test.go b/pkg/verify/verify_test.go new file mode 100644 index 0000000..c1165f6 --- /dev/null +++ b/pkg/verify/verify_test.go @@ -0,0 +1,262 @@ +package verify_test + +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" +) + +// 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) + } +} + +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) + } +} + +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) + } + }) + } +}