feat: add --resume for incremental snapshot exports

README.md

@@ -11,10 +11,11 @@ batch-export is a tool to retrieve Ethereum event logs for specific contracts, p
 - Saves retrieved logs to a specified output file (default: `export.ndjson`) in NDJSON format.
 - Exports up to the latest **finalized** block by default (`--end=0`), so a snapshot never contains logs from blocks that can still be reorged.
 - Graceful shutdown on interrupt signals (Ctrl+C).
+- Continue a previous export from where it stopped (incremental snapshots).
 
 ## Requirements
 
-- Go 1.24 or later
+- Go 1.25 or later
 
 ## Installation
 
@@ -47,16 +48,72 @@ The primary command is export.
   -b, --block-range-limit uint32   Max blocks per log query (default 5)
   -c, --compress                   Compress to GZIP
       --end uint                   End block (optional, uses latest finalized block if 0)
-  -e, --endpoint string            Ethereum RPC endpoint URL
+  -e, --endpoint string            Ethereum based RPC endpoint URL (default "https://rpc.gnosis.gateway.fm")
   -h, --help                       help for export
   -m, --max-request int            Max RPC requests/sec (default 15)
   -o, --output string              Output file path (NDJSON) (default "export.ndjson")
+  -r, --resume string              Continue a previous export file (.ndjson, .gz or .gzip); combine with --output to write a new snapshot instead of appending in place
       --retry-delay duration       Delay before the first retry, doubling per retry up to 30s (default 1s)
       --retry-max int              Max retries per RPC request on transient network errors (0 disables retrying) (default 5)
       --start uint                 Start block (optional, uses contract start block if 0) (default 31306381)
   -v, --verbosity string           Log verbosity (silent, error, warn, info, debug) (default "info")
 ```
 
+### Continuing a previous snapshot
+
+Instead of re-exporting every block, point `--resume` at the previous
+snapshot and name the new one with `--output`. The previous file is read,
+never modified; the new file holds everything the previous one did plus the
+blocks exported since:
+
+```sh
+./dist/batch-export export --resume snapshots/2026-07.ndjson.gzip --output snapshots/2026-08.ndjson.gzip
+```
+
+If `--output` names a different existing file, it is overwritten — the same
+`os.Create` semantics as a fresh export — so pick a new name for each
+snapshot. (Naming the input itself under another spelling — absolute vs
+relative, a symlink, a case difference — is detected and appends in place
+instead.)
+
+Formats are detected by content, not extension: `.ndjson`, `.gz` and `.gzip`
+all work, and the output's format always matches the input's. Gzip stores no
+filename inside, so decompressing names the result after the archive minus
+its extension — name snapshots with the double extension, as above, and
+extraction yields a `.ndjson` file. Omitting
+`--output` (or naming the input) appends to the previous file in place — the
+space-saving variant:
+
+```sh
+./dist/batch-export export --resume export.ndjson.gzip
+```
+
+`--start` is ignored (with a warning) when `--resume` is set: the cursor in
+the previous file decides where the fetch resumes. The last exported block is
+re-queried and entries already present are skipped, so continuing neither
+duplicates nor drops a log. Each continuation of a compressed snapshot adds a
+gzip member — standard tools (`gzcat`, `gunzip`, Go, Python) read
+multi-member files as one stream, a year of monthly continuations costs about
+0.02% in size, and `gzcat old.gzip | gzip > fresh.gzip` consolidates the
+members any time.
+
+Keep one canonical snapshot file. `--compress` is ignored when resuming:
+regenerating a `.gzip` from a plain twin is how an independently continued
+archive gets overwritten. Resume a compressed file to get a compressed
+result.
+
+Resume only files this tool produced. The file's tail is validated before
+anything is written: content the tool never writes — a non-log line, foreign
+data, an alien gzip member — is refused rather than repaired. The one
+exception is the tool's own interrupted final write (a run killed
+mid-export): in copy mode it is simply not copied, in place it is truncated
+away with a warning, and its entries are re-fetched. Note that resuming does
+not detect a file from a different chain; pairing the snapshot with the
+right `--endpoint` is the operator's contract.
+
+If a copy-mode run itself is interrupted or fails, the input was never
+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.
 
 ## Maintainers

cmd/export.go

@@ -4,13 +4,17 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"io"
+	"path/filepath"
 	"sync"
 	"time"
 
+	"github.com/ethereum/go-ethereum/core/types"
 	ethclient "github.com/ethersphere/batch-export/pkg/ethclientwrapper"
 	"github.com/ethersphere/batch-export/pkg/eventfetcher"
 	"github.com/ethersphere/batch-export/pkg/filestore"
 	"github.com/ethersphere/batch-export/pkg/gzipstore"
+	"github.com/ethersphere/batch-export/pkg/resume"
 	"github.com/ethersphere/bee/v2/pkg/config"
 	"github.com/ethersphere/bee/v2/pkg/util/abiutil"
 	"github.com/spf13/cobra"
@@ -25,6 +29,7 @@ func (c *command) initExportCmd() (err error) {
 		blockRangeLimit uint32
 		outputFile      string
 		compress        bool
+		resumeFile      string
 		retryMax        int
 		retryDelay      time.Duration
 	)
@@ -40,7 +45,38 @@ with an exponential backoff (--retry-max, --retry-delay).
 The retrieved logs are saved to the specified output file (default: 'export.ndjson') in NDJSON format.
 The process can be interrupted at any time (Ctrl+C), and it will attempt to save already retrieved logs before exiting.`,
 		RunE: func(cmd *cobra.Command, args []string) (err error) {
-			ctx := cmd.Context()
+			ctx, cancel := context.WithCancel(cmd.Context())
+			defer cancel()
+
+			var cursor *resume.Cursor
+			if resumeFile != "" {
+				cursor, err = resume.Read(resumeFile)
+				if err != nil {
+					return fmt.Errorf("failed to read resume file %q: %w", resumeFile, err)
+				}
+
+				if cmd.Flags().Changed("start") {
+					c.log.Warning("--start is ignored when --resume is set", "resumeFile", resumeFile)
+				}
+				if compress {
+					c.log.Warning("--compress is ignored when resuming; resume a compressed file to get a compressed result", "resumeFile", resumeFile)
+					compress = false
+				}
+				// An unset --output means in-place; so does naming the input.
+				if !cmd.Flags().Changed("output") || filepath.Clean(outputFile) == filepath.Clean(resumeFile) {
+					outputFile = resumeFile
+				}
+
+				startBlock = cursor.BlockNumber
+
+				c.log.Info("Resuming export",
+					"resumeFile", resumeFile,
+					"outputFile", outputFile,
+					"startBlock", startBlock,
+					"lastLogIndex", cursor.LogIndex,
+					"compressed", cursor.Compressed,
+				)
+			}
 
 			if retryMax < 0 {
 				return fmt.Errorf("invalid --retry-max %d: must not be negative", retryMax)
@@ -77,6 +113,28 @@ The process can be interrupted at any time (Ctrl+C), and it will attempt to save
 				startBlock = chainCfg.PostageStampStartBlock
 			}
 
+			if cursor != nil {
+				discarded, err := resume.PrepareOutput(cursor, resumeFile, outputFile)
+				if err != nil {
+					return err
+				}
+				if discarded > 0 {
+					c.log.Warning("previous export ends with an interrupted write, leaving it out",
+						"resumeFile", resumeFile,
+						"offset", cursor.CleanSize,
+						"discardedBytes", discarded,
+					)
+				}
+			}
+
+			// Opened before the first log is fetched: from inside the saving
+			// goroutine, a failure here would leave the fetcher pushing into a
+			// channel nobody drains.
+			w, err := openOutput(outputFile, cursor)
+			if err != nil {
+				return fmt.Errorf("failed to open output file: %w", err)
+			}
+
 			c.log.Info("Retrieving logs", "startBlock", startBlock, "endBlock", endBlock)
 
 			logChan, errorChan := client.GetLogs(ctx, &eventfetcher.Request{
@@ -91,14 +149,20 @@ The process can be interrupted at any time (Ctrl+C), and it will attempt to save
 			ticker := time.NewTicker(15 * time.Second)
 			defer ticker.Stop()
 
+			var saveErr error
 			go func() {
 				defer wg.Done()
-				if err := filestore.SaveLogsAsync(ctx, logChan, outputFile); err != nil {
-					if errors.Is(err, context.Canceled) {
+
+				if err := saveLogs(ctx, logChan, w, cursor); err != nil {
+					if solelyCanceled(err) {
 						c.log.Error(err, "context canceled while saving logs")
 						return
 					}
 					c.log.Error(err, "error saving logs")
+					// Stop the fetcher too: with the saver gone, logChan
+					// would fill and block it forever.
+					saveErr = err
+					cancel()
 					return
 				}
 				c.log.Info("all logs have been saved", "outputFile", outputFile)
@@ -120,12 +184,20 @@ The process can be interrupted at any time (Ctrl+C), and it will attempt to save
 					if !ok {
 						errorChan = nil
 					} else {
-						return fmt.Errorf("error retrieving logs: %w", err)
+						wg.Wait()
+						if saveErr != nil && errors.Is(err, context.Canceled) {
+							return saveErr
+						}
+						return errors.Join(fmt.Errorf("error retrieving logs: %w", err), saveErr)
 					}
 				case <-ticker.C:
 					c.log.Info("still retrieving logs...")
 				case <-ctx.Done():
 					c.log.Info("context canceled, waiting for logs to be saved...")
+					wg.Wait()
+					if saveErr != nil {
+						return saveErr
+					}
 					if err := compressFunc(); err != nil {
 						return errors.Join(fmt.Errorf("error compressing file: %w", err), ctx.Err())
 					}
@@ -138,6 +210,9 @@ The process can be interrupted at any time (Ctrl+C), and it will attempt to save
 			}
 
 			wg.Wait()
+			if saveErr != nil {
+				return saveErr
+			}
 			if err := compressFunc(); err != nil {
 				return fmt.Errorf("error compressing file: %w", err)
 			}
@@ -153,10 +228,58 @@ The process can be interrupted at any time (Ctrl+C), and it will attempt to save
 	cmd.Flags().Uint32VarP(&blockRangeLimit, "block-range-limit", "b", 5, "Max blocks per log query")
 	cmd.Flags().StringVarP(&outputFile, "output", "o", "export.ndjson", "Output file path (NDJSON)")
 	cmd.Flags().BoolVarP(&compress, "compress", "c", false, "Compress to GZIP")
+	cmd.Flags().StringVarP(&resumeFile, "resume", "r", "", "Continue a previous export file (.ndjson, .gz or .gzip); combine with --output to write a new snapshot instead of appending in place")
 	cmd.Flags().IntVarP(&retryMax, "retry-max", "", 5, "Max retries per RPC request on transient network errors (0 disables retrying)")
 	cmd.Flags().DurationVarP(&retryDelay, "retry-delay", "", ethclient.DefaultRetryDelay, "Delay before the first retry, doubling per retry up to 30s")
 
 	c.root.AddCommand(cmd)
 
 	return nil
 }
+
+// openOutput opens the destination for a run's logs: a fresh file when cursor
+// is nil, or a writer that appends to the file the cursor came from.
+func openOutput(outputFile string, cursor *resume.Cursor) (io.WriteCloser, error) {
+	switch {
+	case cursor == nil:
+		return filestore.CreateWriter(outputFile)
+	case cursor.Compressed:
+		return gzipstore.AppendWriter(outputFile)
+	default:
+		return filestore.AppendWriter(outputFile)
+	}
+}
+
+// saveLogs writes logs to w, dropping any entry a resumed export already holds.
+// A nil cursor means the destination starts empty, so every log is kept.
+func saveLogs(ctx context.Context, logChan <-chan types.Log, w io.WriteCloser, cursor *resume.Cursor) error {
+	var skip func(types.Log) bool
+	if cursor != nil {
+		skip = cursor.Skip
+	}
+
+	return filestore.AppendLogsAsync(ctx, logChan, w, skip)
+}
+
+// solelyCanceled reports whether err contains nothing beyond context
+// cancellation, unwrapping joined and wrapped errors along the way. It is
+// stricter than errors.Is(err, context.Canceled): AppendLogsAsync joins the
+// save error with the destination's Close error, and a SIGINT racing a
+// failing gzip-member flush must not be reported as pure cancellation.
+func solelyCanceled(err error) bool {
+	if err == nil {
+		return false
+	}
+	if u, ok := err.(interface{ Unwrap() []error }); ok {
+		for _, e := range u.Unwrap() {
+			if !solelyCanceled(e) {
+				return false
+			}
+		}
+		return true
+	}
+	if u := errors.Unwrap(err); u != nil {
+		return solelyCanceled(u)
+	}
+	return errors.Is(err, context.Canceled)
+}

cmd/export_test.go

@@ -0,0 +1,75 @@
+package cmd
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"io"
+	"testing"
+)
+
+func TestSolelyCanceled(t *testing.T) {
+	t.Parallel()
+
+	tests := []struct {
+		name string
+		err  error
+		want bool
+	}{
+		{
+			name: "nil error",
+			err:  nil,
+			want: false,
+		},
+		{
+			name: "pure context.Canceled",
+			err:  context.Canceled,
+			want: true,
+		},
+		{
+			name: "wrapped context.Canceled",
+			err:  fmt.Errorf("wrap: %w", context.Canceled),
+			want: true,
+		},
+		{
+			name: "joined context.Canceled with context.Canceled",
+			err:  errors.Join(context.Canceled, context.Canceled),
+			want: true,
+		},
+		{
+			name: "joined context.Canceled with wrapped context.Canceled",
+			err:  errors.Join(context.Canceled, fmt.Errorf("wrap: %w", context.Canceled)),
+			want: true,
+		},
+		{
+			name: "joined context.Canceled with real io error",
+			err:  errors.Join(context.Canceled, io.ErrUnexpectedEOF),
+			want: false,
+		},
+		{
+			name: "joined real error with context.Canceled",
+			err:  errors.Join(errors.New("disk full"), context.Canceled),
+			want: false,
+		},
+		{
+			name: "pure io error",
+			err:  io.ErrClosedPipe,
+			want: false,
+		},
+		{
+			name: "context deadline exceeded",
+			err:  context.DeadlineExceeded,
+			want: false,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			t.Parallel()
+
+			if got := solelyCanceled(tt.err); got != tt.want {
+				t.Errorf("solelyCanceled(%v) = %v, want %v", tt.err, got, tt.want)
+			}
+		})
+	}
+}