feat: verify refreshed snapshots before publishing

.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})"

Makefile

@@ -40,6 +40,6 @@ clean:
 
 .PHONY: test
 test:
-	$(GO) test -v ./pkg/...
+	$(GO) test -v ./...
 
 FORCE:

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

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
 }
 

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
+}

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)
+			}
+		})
+	}
+}