diff --git a/README.md b/README.md index 8b7c167..a221c5f 100644 --- a/README.md +++ b/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 diff --git a/cmd/export.go b/cmd/export.go index 1c54273..0962a3c 100644 --- a/cmd/export.go +++ b/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,6 +228,7 @@ 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") @@ -160,3 +236,50 @@ The process can be interrupted at any time (Ctrl+C), and it will attempt to save 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) +} diff --git a/cmd/export_test.go b/cmd/export_test.go new file mode 100644 index 0000000..f3cf096 --- /dev/null +++ b/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) + } + }) + } +} diff --git a/pkg/filestore/filestore.go b/pkg/filestore/filestore.go index b07134d..56132de 100644 --- a/pkg/filestore/filestore.go +++ b/pkg/filestore/filestore.go @@ -11,25 +11,46 @@ import ( "github.com/ethereum/go-ethereum/core/types" ) -// SaveLogsAsync writes logs to a file asynchronously. -func SaveLogsAsync(ctx context.Context, logChan <-chan types.Log, filePath string) error { +// CreateWriter creates the file at filePath, replacing anything already there. +// It is separate from writing so a caller can fail before it starts producing +// logs it would have nowhere to put. +func CreateWriter(filePath string) (io.WriteCloser, error) { file, err := os.Create(filePath) if err != nil { - return fmt.Errorf("error creating file: %w", err) + return nil, fmt.Errorf("error creating file: %w", err) } - return saveLogs(ctx, logChan, file) + + return file, nil } -func saveLogs(ctx context.Context, logChan <-chan types.Log, w io.WriteCloser) (err error) { - // OS-buffered writes can surface a failure only when the file is flushed - // at close time, so a swallowed Close error would report an incomplete - // export as success. - defer func() { - if cerr := w.Close(); cerr != nil { - err = errors.Join(err, fmt.Errorf("error closing file: %w", cerr)) - } - }() +// AppendWriter opens an existing NDJSON file for appending. +func AppendWriter(filePath string) (io.WriteCloser, error) { + file, err := os.OpenFile(filePath, os.O_WRONLY|os.O_APPEND, 0o644) + if err != nil { + return nil, fmt.Errorf("error opening file for appending: %w", err) + } + + return file, nil +} +// AppendLogsAsync writes logs to w asynchronously, keeping whatever the +// destination already holds. Logs for which skip reports true are dropped; a +// nil skip writes every log. w is closed before returning, cancellation +// included, so a buffered destination is always flushed. +// +// The close error is joined rather than discarded: on a compressed destination +// Close writes the terminator and footer, so a failure there leaves a truncated +// member that must not be reported as a completed save. errors.Join keeps +// errors.Is working, so a cancelled context still reads as context.Canceled. +func AppendLogsAsync(ctx context.Context, logChan <-chan types.Log, w io.WriteCloser, skip func(types.Log) bool) (err error) { + defer func() { err = errors.Join(err, w.Close()) }() + + return writeLogs(ctx, logChan, w, skip) +} + +// writeLogs encodes logs from logChan to w as NDJSON until the channel is +// closed or the context is cancelled. +func writeLogs(ctx context.Context, logChan <-chan types.Log, w io.Writer, skip func(types.Log) bool) error { encoder := json.NewEncoder(w) for { @@ -41,6 +62,10 @@ func saveLogs(ctx context.Context, logChan <-chan types.Log, w io.WriteCloser) ( return nil } + if skip != nil && skip(logObj) { + continue + } + if err := encoder.Encode(logObj); err != nil { return fmt.Errorf("error encoding log: %w", err) } diff --git a/pkg/filestore/filestore_internal_test.go b/pkg/filestore/filestore_internal_test.go deleted file mode 100644 index 3cd0653..0000000 --- a/pkg/filestore/filestore_internal_test.go +++ /dev/null @@ -1,44 +0,0 @@ -package filestore - -import ( - "context" - "errors" - "testing" - - "github.com/ethereum/go-ethereum/core/types" -) - -// closeFailWriter accepts all writes but fails on Close, mimicking a file -// whose OS-buffered data cannot be flushed (disk full, quota, network fs). -type closeFailWriter struct{ closeErr error } - -func (w *closeFailWriter) Write(p []byte) (int, error) { return len(p), nil } -func (w *closeFailWriter) Close() error { return w.closeErr } - -func TestSaveLogsReportsCloseError(t *testing.T) { - closeErr := errors.New("flush to disk failed") - - logChan := make(chan types.Log, 1) - logChan <- types.Log{BlockNumber: 1} - close(logChan) - - err := saveLogs(context.Background(), logChan, &closeFailWriter{closeErr: closeErr}) - if !errors.Is(err, closeErr) { - t.Fatalf("got %v, want close error %v", err, closeErr) - } -} - -func TestSaveLogsKeepsContextErrorOnCloseFailure(t *testing.T) { - closeErr := errors.New("flush to disk failed") - - ctx, cancel := context.WithCancel(context.Background()) - cancel() - - err := saveLogs(ctx, make(chan types.Log), &closeFailWriter{closeErr: closeErr}) - if !errors.Is(err, context.Canceled) { - t.Fatalf("got %v, want context.Canceled", err) - } - if !errors.Is(err, closeErr) { - t.Fatalf("got %v, want close error %v", err, closeErr) - } -} diff --git a/pkg/filestore/filestore_test.go b/pkg/filestore/filestore_test.go index 8add67e..6ae1cf2 100644 --- a/pkg/filestore/filestore_test.go +++ b/pkg/filestore/filestore_test.go @@ -4,8 +4,11 @@ import ( "bufio" "context" "encoding/json" + "errors" "os" "path/filepath" + "slices" + "strings" "testing" "github.com/ethereum/go-ethereum/common" @@ -13,49 +16,174 @@ import ( "github.com/ethersphere/batch-export/pkg/filestore" ) -func TestSaveLogsAsyncWritesNDJSON(t *testing.T) { - path := filepath.Join(t.TempDir(), "export.ndjson") - - logChan := make(chan types.Log, 2) - for i := uint64(1); i <= 2; i++ { - logChan <- types.Log{ - Address: common.HexToAddress("0x000000000000000000000000000000000000bEEF"), - Topics: []common.Hash{common.HexToHash("0x11")}, - Data: []byte{0xde, 0xad}, - BlockNumber: i, - } - } - close(logChan) - - if err := filestore.SaveLogsAsync(context.Background(), logChan, path); err != nil { - t.Fatalf("SaveLogsAsync: %v", err) - } +// blocksIn returns the block number of every log line in the file at path. +func blocksIn(t *testing.T, path string) []uint64 { + t.Helper() file, err := os.Open(path) if err != nil { - t.Fatalf("open output: %v", err) + t.Fatalf("open %s: %v", path, err) } defer file.Close() - var got []types.Log + var blocks []uint64 scanner := bufio.NewScanner(file) for scanner.Scan() { + if strings.TrimSpace(scanner.Text()) == "" { + continue + } var l types.Log if err := json.Unmarshal(scanner.Bytes(), &l); err != nil { - t.Fatalf("decode line %d: %v", len(got)+1, err) + t.Fatalf("unmarshal %q: %v", scanner.Text(), err) } - got = append(got, l) + blocks = append(blocks, l.BlockNumber) } if err := scanner.Err(); err != nil { - t.Fatalf("scan output: %v", err) + t.Fatalf("scan %s: %v", path, err) + } + + return blocks +} + +// seed writes logs for the given blocks to a fresh file at path. +func seed(t *testing.T, path string, blocks ...uint64) { + t.Helper() + + w, err := filestore.CreateWriter(path) + if err != nil { + t.Fatalf("CreateWriter() error = %v", err) + } + if err := filestore.AppendLogsAsync(t.Context(), feed(blocks...), w, nil); err != nil { + t.Fatalf("AppendLogsAsync() error = %v", err) } +} - if len(got) != 2 { - t.Fatalf("got %d logs, want 2", len(got)) +// feed returns a closed channel already holding logs for the given blocks. +// Topics must stay a non-nil empty slice: go-ethereum's generated +// Log.UnmarshalJSON rejects a null "topics" as a missing required field. +func feed(blocks ...uint64) <-chan types.Log { + ch := make(chan types.Log, len(blocks)) + for _, b := range blocks { + ch <- types.Log{BlockNumber: b, Topics: []common.Hash{}} } - for i, l := range got { - if l.BlockNumber != uint64(i+1) { - t.Errorf("log %d: blockNumber got %d want %d", i, l.BlockNumber, i+1) - } + close(ch) + + return ch +} + +func TestCreateWriterReplacesExistingFile(t *testing.T) { + t.Parallel() + + path := filepath.Join(t.TempDir(), "export.ndjson") + if err := os.WriteFile(path, []byte("stale\n"), 0o644); err != nil { + t.Fatalf("seed %s: %v", path, err) + } + + seed(t, path, 1, 2) + + want := []uint64{1, 2} + if got := blocksIn(t, path); !slices.Equal(got, want) { + t.Errorf("blocks = %v, want %v", got, want) + } +} + +func TestAppendLogsAsyncKeepsExistingContent(t *testing.T) { + t.Parallel() + + path := filepath.Join(t.TempDir(), "export.ndjson") + seed(t, path, 1, 2) + + w, err := filestore.AppendWriter(path) + if err != nil { + t.Fatalf("AppendWriter() error = %v", err) + } + if err := filestore.AppendLogsAsync(t.Context(), feed(3, 4), w, nil); err != nil { + t.Fatalf("AppendLogsAsync() error = %v", err) + } + + want := []uint64{1, 2, 3, 4} + if got := blocksIn(t, path); !slices.Equal(got, want) { + t.Errorf("blocks = %v, want %v", got, want) + } +} + +func TestAppendLogsAsyncSkipsFilteredLogs(t *testing.T) { + t.Parallel() + + path := filepath.Join(t.TempDir(), "export.ndjson") + seed(t, path, 1, 2) + + w, err := filestore.AppendWriter(path) + if err != nil { + t.Fatalf("AppendWriter() error = %v", err) + } + skip := func(l types.Log) bool { return l.BlockNumber <= 2 } + if err := filestore.AppendLogsAsync(t.Context(), feed(1, 2, 3, 4), w, skip); err != nil { + t.Fatalf("AppendLogsAsync() error = %v", err) + } + + want := []uint64{1, 2, 3, 4} + if got := blocksIn(t, path); !slices.Equal(got, want) { + t.Errorf("blocks = %v, want %v", got, want) + } +} + +func TestAppendLogsAsyncClosesWriterOnCancel(t *testing.T) { + t.Parallel() + + path := filepath.Join(t.TempDir(), "export.ndjson") + seed(t, path, 1) + + w, err := filestore.AppendWriter(path) + if err != nil { + t.Fatalf("AppendWriter() error = %v", err) + } + + ctx, cancel := context.WithCancel(t.Context()) + cancel() + + // An open channel that never delivers, so cancellation is the only exit. + if err := filestore.AppendLogsAsync(ctx, make(chan types.Log), w, nil); !errors.Is(err, context.Canceled) { + t.Fatalf("AppendLogsAsync() error = %v, want context.Canceled", err) + } + + // The writer must already be closed; closing it again must fail. + if err := w.Close(); err == nil { + t.Error("writer was left open after cancellation") + } +} + +// closeFailWriter accepts all writes but fails on Close, mimicking a file +// whose OS-buffered data cannot be flushed (disk full, quota, network fs). +type closeFailWriter struct{ closeErr error } + +func (w *closeFailWriter) Write(p []byte) (int, error) { return len(p), nil } +func (w *closeFailWriter) Close() error { return w.closeErr } + +func TestAppendLogsAsyncReportsCloseError(t *testing.T) { + t.Parallel() + + closeErr := errors.New("flush to disk failed") + + err := filestore.AppendLogsAsync(t.Context(), feed(1), &closeFailWriter{closeErr: closeErr}, nil) + if !errors.Is(err, closeErr) { + t.Fatalf("AppendLogsAsync() error = %v, want close error %v", err, closeErr) + } +} + +func TestAppendLogsAsyncKeepsContextErrorOnCloseFailure(t *testing.T) { + t.Parallel() + + closeErr := errors.New("flush to disk failed") + + ctx, cancel := context.WithCancel(t.Context()) + cancel() + + err := filestore.AppendLogsAsync(ctx, make(chan types.Log), &closeFailWriter{closeErr: closeErr}, nil) + if !errors.Is(err, context.Canceled) { + t.Fatalf("AppendLogsAsync() error = %v, want context.Canceled", err) + } + if !errors.Is(err, closeErr) { + t.Fatalf("AppendLogsAsync() error = %v, want close error %v", err, closeErr) } } diff --git a/pkg/gzipstore/gzipstore.go b/pkg/gzipstore/gzipstore.go index 2b079ed..fbdc5ff 100644 --- a/pkg/gzipstore/gzipstore.go +++ b/pkg/gzipstore/gzipstore.go @@ -10,14 +10,12 @@ import ( // CompressFile compresses the specified input file into the specified output gzip file. func CompressFile(inputFilePath string, outputFilePath string) error { - // open the input file for reading inputFile, err := os.Open(inputFilePath) if err != nil { return fmt.Errorf("failed to open input file '%s': %w", inputFilePath, err) } defer inputFile.Close() - // create the output file for writing outputFile, err := os.Create(outputFilePath) if err != nil { return fmt.Errorf("failed to create output file '%s': %w", outputFilePath, err) @@ -44,3 +42,36 @@ func compress(dst io.Writer, src io.Reader) (err error) { return nil } + +// AppendWriter opens an existing gzip file and returns a writer that adds a new +// gzip member to it. Concatenated members form a valid gzip stream, so readers +// see one continuous file and the existing bytes are never rewritten. The +// caller must close the writer to flush the member. +func AppendWriter(filePath string) (io.WriteCloser, error) { + file, err := os.OpenFile(filePath, os.O_WRONLY|os.O_APPEND, 0o644) + if err != nil { + return nil, fmt.Errorf("failed to open gzip file '%s' for appending: %w", filePath, err) + } + + return &memberWriter{file: file, gzip: gzip.NewWriter(file)}, nil +} + +// memberWriter writes one gzip member and owns the file it was opened from. +type memberWriter struct { + file *os.File + gzip *gzip.Writer +} + +func (w *memberWriter) Write(p []byte) (int, error) { + return w.gzip.Write(p) +} + +// Close finishes the gzip member, then closes the file. Both are attempted even +// if the first fails, so the descriptor is never leaked. +func (w *memberWriter) Close() error { + if err := errors.Join(w.gzip.Close(), w.file.Close()); err != nil { + return fmt.Errorf("failed to close gzip file '%s': %w", w.file.Name(), err) + } + + return nil +} diff --git a/pkg/gzipstore/gzipstore_test.go b/pkg/gzipstore/gzipstore_test.go index 61a173a..b5501af 100644 --- a/pkg/gzipstore/gzipstore_test.go +++ b/pkg/gzipstore/gzipstore_test.go @@ -11,6 +11,105 @@ import ( "github.com/ethersphere/batch-export/pkg/gzipstore" ) +// writeGzip creates a gzip file holding content and returns its path. +func writeGzip(t *testing.T, content string) string { + t.Helper() + + path := filepath.Join(t.TempDir(), "export.ndjson.gzip") + + var buf bytes.Buffer + w := gzip.NewWriter(&buf) + if _, err := io.WriteString(w, content); err != nil { + t.Fatalf("gzip write: %v", err) + } + if err := w.Close(); err != nil { + t.Fatalf("gzip close: %v", err) + } + if err := os.WriteFile(path, buf.Bytes(), 0o644); err != nil { + t.Fatalf("write %s: %v", path, err) + } + + return path +} + +// readGzip decompresses the whole file, following every member. +func readGzip(t *testing.T, path string) string { + t.Helper() + + file, err := os.Open(path) + if err != nil { + t.Fatalf("open %s: %v", path, err) + } + defer file.Close() + + reader, err := gzip.NewReader(file) + if err != nil { + t.Fatalf("gzip reader: %v", err) + } + defer reader.Close() + + got, err := io.ReadAll(reader) + if err != nil { + t.Fatalf("read gzip: %v", err) + } + + return string(got) +} + +func TestAppendWriterAddsReadableMember(t *testing.T) { + t.Parallel() + + path := writeGzip(t, "first\nsecond\n") + + w, err := gzipstore.AppendWriter(path) + if err != nil { + t.Fatalf("AppendWriter() error = %v", err) + } + if _, err := io.WriteString(w, "third\nfourth\n"); err != nil { + t.Fatalf("write: %v", err) + } + if err := w.Close(); err != nil { + t.Fatalf("Close() error = %v", err) + } + + const want = "first\nsecond\nthird\nfourth\n" + if got := readGzip(t, path); got != want { + t.Errorf("content = %q, want %q", got, want) + } +} + +func TestAppendWriterRepeatedAppends(t *testing.T) { + t.Parallel() + + path := writeGzip(t, "a\n") + + for _, line := range []string{"b\n", "c\n"} { + w, err := gzipstore.AppendWriter(path) + if err != nil { + t.Fatalf("AppendWriter() error = %v", err) + } + if _, err := io.WriteString(w, line); err != nil { + t.Fatalf("write: %v", err) + } + if err := w.Close(); err != nil { + t.Fatalf("Close() error = %v", err) + } + } + + const want = "a\nb\nc\n" + if got := readGzip(t, path); got != want { + t.Errorf("content = %q, want %q", got, want) + } +} + +func TestAppendWriterMissingFile(t *testing.T) { + t.Parallel() + + if _, err := gzipstore.AppendWriter(filepath.Join(t.TempDir(), "nope.gzip")); err == nil { + t.Fatal("AppendWriter() error = nil, want an error") + } +} + func TestCompressFileRoundTrip(t *testing.T) { dir := t.TempDir() inputPath := filepath.Join(dir, "in.ndjson") diff --git a/pkg/resume/resume.go b/pkg/resume/resume.go new file mode 100644 index 0000000..a444615 --- /dev/null +++ b/pkg/resume/resume.go @@ -0,0 +1,431 @@ +// Package resume locates the point at which a previous export stopped so that +// a new run can continue from there. +package resume + +import ( + "bufio" + "bytes" + "compress/flate" + "compress/gzip" + "encoding/json" + "errors" + "fmt" + "io" + "os" + "path/filepath" + + "github.com/ethereum/go-ethereum/common/hexutil" + "github.com/ethereum/go-ethereum/core/types" +) + +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 +) + +var ( + // ErrNotAnExport indicates content this tool never writes. The file was + // altered after export, so resuming it is refused rather than repaired. + ErrNotAnExport = errors.New("not an untouched batch-export file") + // ErrNoLogs indicates a file consistent with tool output that holds no + // 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") +) + +// Cursor marks the last log entry saved by a previous export, together with +// the point up to which that export's file is known to be complete. +type Cursor struct { + // BlockNumber is the block of the last saved log. A resumed export + // re-queries it, because an interrupted run may have saved only part of it. + BlockNumber uint64 + // LogIndex is the index of the last saved log within BlockNumber. + LogIndex uint + // Compressed reports whether the file holds gzip rather than plain NDJSON. + Compressed bool + // CleanSize is the offset just past the entry the cursor points at: past + // that line's newline, or past the gzip member it ends in. Bytes beyond it + // are a partial write and must be discarded before anything is appended. + CleanSize int64 + // Truncated reports whether any bytes follow CleanSize. + Truncated bool +} + +// 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 +} + +// PrepareOutput readies outputPath for appending the continuation of the +// export at inputPath. The same file is prepared in place: an interrupted +// final write, if any, is truncated away. "Same file" is not lexical — a +// relative and an absolute spelling of one path, a symlink or hardlink, or +// two names a case-insensitive filesystem folds together all count, detected +// via os.SameFile (device+inode), because os.Create on a distinct-looking +// spelling of the input would truncate it before it could be copied from. +// With genuinely distinct files the input is never modified: its clean +// content is copied raw into outputPath, replacing whatever was there, and +// the interrupted tail is simply not copied. Either way it returns how many +// trailing bytes were left out — every entry they held falls at or after the +// cursor, so the resumed query fetches it again. +func PrepareOutput(c *Cursor, inputPath, outputPath string) (int64, error) { + if filepath.Clean(inputPath) == filepath.Clean(outputPath) { + return prepareInPlace(c, inputPath) + } + + in, err := os.Open(inputPath) + if err != nil { + return 0, fmt.Errorf("error opening resume file: %w", err) + } + defer in.Close() + + info, err := in.Stat() + if err != nil { + return 0, fmt.Errorf("error inspecting resume file: %w", err) + } + + // The same file under two names — relative vs absolute, a symlink or + // hardlink, a case-insensitive filesystem — is in-place, not copy mode: + // os.Create would truncate the input we are about to read. + if outInfo, err := os.Stat(outputPath); err == nil && os.SameFile(info, outInfo) { + return prepareInPlace(c, inputPath) + } + + out, err := os.Create(outputPath) + if err != nil { + return 0, fmt.Errorf("error creating output file: %w", err) + } + + _, err = io.CopyN(out, in, c.CleanSize) + if cerr := out.Close(); err == nil { + err = cerr + } + if err != nil { + return 0, fmt.Errorf("error copying clean content to output file: %w", err) + } + + return info.Size() - c.CleanSize, nil +} + +// prepareInPlace drops the interrupted tail so appending continues from the +// clean boundary. It never truncates past the offset the reader identified. +func prepareInPlace(c *Cursor, path string) (int64, error) { + if !c.Truncated { + return 0, nil + } + + info, err := os.Stat(path) + if err != nil { + return 0, fmt.Errorf("error inspecting resume file: %w", err) + } + if info.Size() <= c.CleanSize { + return 0, nil + } + + if err := os.Truncate(path, c.CleanSize); err != nil { + return 0, fmt.Errorf("error truncating resume file: %w", err) + } + + return info.Size() - c.CleanSize, nil +} + +// cursorLine is the part of an exported log line a cursor is built from. Both +// fields are pointers so a line missing either one can be rejected. +type cursorLine struct { + BlockNumber *hexutil.Uint64 `json:"blockNumber"` + LogIndex *hexutil.Uint `json:"logIndex"` +} + +// Read returns a cursor for the last complete log entry in the file at path, +// detecting plain NDJSON or gzip from the leading bytes rather than the +// extension. +// +// Only a properly terminated entry counts: a plain line must end in a newline, +// a compressed one must sit in a member that decoded to a clean end of stream. +// A caller that appends must first discard everything past CleanSize. +// +// It returns ErrNoLogs when there is no complete entry, and ErrNotAnExport +// when the file holds content this tool did not write. +func Read(path string) (*Cursor, error) { + file, err := os.Open(path) + if err != nil { + return nil, fmt.Errorf("error opening resume file: %w", err) + } + defer file.Close() + + info, err := file.Stat() + if err != nil { + return nil, fmt.Errorf("error inspecting resume file: %w", err) + } + + compressed, err := isGzip(file) + if err != nil { + return nil, fmt.Errorf("error reading resume file: %w", err) + } + + var cursor *Cursor + if compressed { + cursor, err = lastCursorGzip(file) + } else { + cursor, err = lastCursorPlain(file, info.Size()) + } + if err != nil { + return nil, err + } + + cursor.Compressed = compressed + cursor.Truncated = cursor.CleanSize < info.Size() + + return cursor, nil +} + +// isGzip reports whether file starts with the gzip magic bytes. +func isGzip(file *os.File) (bool, error) { + var magic [2]byte + + n, err := file.ReadAt(magic[:], 0) + if err != nil && !errors.Is(err, io.EOF) { + return false, err + } + if n < len(magic) { + return false, nil + } + + return magic[0] == 0x1f && magic[1] == 0x8b, nil +} + +// lastCursorPlain validates the tail of a plain NDJSON file and returns a +// cursor for its last complete line. The tool writes one entry per line in a +// single call, so only the tail needs examining: the last newline-terminated +// 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) + offset := size - window + + buf := make([]byte, window) + if _, err := file.ReadAt(buf, offset); err != nil { + return nil, fmt.Errorf("error reading resume file: %w", err) + } + + nl := bytes.LastIndexByte(buf, '\n') + 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 { + 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 { + return nil, fmt.Errorf("%w: %d bytes without a newline after offset %d", ErrNotAnExport, tail, offset+int64(nl)+1) + } + + start := bytes.LastIndexByte(buf[:nl], '\n') + 1 + 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]) + 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) + } + cursor.CleanSize = offset + int64(nl) + 1 + + return cursor, nil +} + +// lastCursorGzip walks a gzip file one member at a time and returns a cursor +// for the last entry inside the cleanly terminated prefix. Gzip cannot be +// seeked, so the whole stream is decompressed; every complete line is +// validated on the way. A member cut short by an interrupted write ends the +// walk: CleanSize stays at the last clean member boundary, and the truncated +// member's content — about to be discarded and re-fetched — never advances +// the cursor. +func lastCursorGzip(file *os.File) (*Cursor, error) { + counter := &countingReader{reader: bufio.NewReaderSize(file, bufferSize)} + + reader, err := gzip.NewReader(counter) + if err != nil { + // isGzip already confirmed the magic bytes, so failure here means the + // file was cut inside the 10-byte header (truncation-shaped, like the + // plain format's interrupted-first-write case) or the header past the + // magic bytes is corrupt (tampering). + if truncationShaped(err) { + return nil, ErrNoLogs + } + return nil, fmt.Errorf("%w: gzip header: %w", ErrNotAnExport, err) + } + defer reader.Close() + + var ( + cursor *Cursor + cleanSize int64 + sawClean bool + member = 1 + ) + for { + // Reset turns multistream back on, so it must be switched off per + // member, not just for the first. + reader.Multistream(false) + + last, err := scanMember(reader, member) + switch { + case errors.Is(err, ErrNotAnExport): + return nil, err + case err != nil && !truncationShaped(err): + return nil, fmt.Errorf("%w: member %d: %w", ErrNotAnExport, member, err) + case err != nil: + // The member never got its trailer: an interrupted final write. + return gzipResult(cursor, cleanSize, sawClean) + } + + if last != nil { + cursor = last + } + cleanSize, sawClean = counter.read, true + + // A clean end of file makes Reset report io.EOF; a partly written + // next member fails to parse as a header and also ends the walk. + if err := reader.Reset(counter); err != nil { + if errors.Is(err, io.EOF) || truncationShaped(err) { + return gzipResult(cursor, cleanSize, sawClean) + } + return nil, fmt.Errorf("error reading gzip resume file: %w", err) + } + member++ + } +} + +// gzipResult finalizes the walk: a file with no clean member, or none holding +// an entry, has nothing to resume from. +func gzipResult(cursor *Cursor, cleanSize int64, sawClean bool) (*Cursor, error) { + if !sawClean || cursor == nil { + return nil, ErrNoLogs + } + cursor.CleanSize = cleanSize + + return cursor, nil +} + +// truncationShaped reports whether err is what an interrupted write produces, +// as opposed to a real read failure that must not be mistaken for one. +// +// gzip.ErrChecksum is deliberately excluded: an interrupted write cannot +// produce a full-length member body plus a complete 8-byte trailer that then +// holds the wrong CRC — that shape is corruption or tampering, not +// truncation, so callers classify it ErrNotAnExport instead. One side effect: +// a genuine I/O error from the underlying file that happens to look like +// corruption is also classified ErrNotAnExport, since it is rare and +// indistinguishable at this layer; §5's strictness favors refusal over +// guessing. +func truncationShaped(err error) bool { + var corrupt flate.CorruptInputError + + return errors.Is(err, io.ErrUnexpectedEOF) || + errors.Is(err, gzip.ErrHeader) || + errors.As(err, &corrupt) +} + +// scanMember reads one gzip member's NDJSON and returns a cursor for its last +// entry, nil when the member is empty. A nil error means the member decoded +// to a clean end of stream on a line boundary. The tool writes members +// holding whole log lines only, so a complete line that does not parse, or a +// clean member ending mid-line, is foreign content (ErrNotAnExport); any +// other read error is returned as-is for the caller to classify. +// +// member is this member's 1-based position in the file, threaded through +// purely so a refusal can name where the offending line lives — a file with +// tens of thousands of lines split across gzip members otherwise gives the +// operator nothing to search for. +func scanMember(r io.Reader, member int) (*Cursor, error) { + buffered := bufio.NewReaderSize(r, bufferSize) + + var ( + last *Cursor + line []byte + 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) + } + + switch { + case errors.Is(err, bufio.ErrBufferFull): + continue + 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) + 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++ + } +} + +// countingReader counts the bytes consumed from the reader it wraps. +// +// It must keep implementing io.ByteReader: that is what makes gzip.Reader read +// from it directly instead of wrapping it in a bufio.Reader of its own. Without +// it the count runs ahead of the real position and member boundaries can no +// longer be observed exactly. +type countingReader struct { + reader *bufio.Reader + // read is the number of bytes handed out so far. + read int64 +} + +func (c *countingReader) Read(p []byte) (int, error) { + n, err := c.reader.Read(p) + c.read += int64(n) + + return n, err +} + +func (c *countingReader) ReadByte() (byte, error) { + b, err := c.reader.ReadByte() + if err == nil { + c.read++ + } + + 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) { + line = bytes.TrimSpace(line) + if len(line) == 0 { + return nil, errors.New("blank line") + } + + var parsed cursorLine + if err := json.Unmarshal(line, &parsed); err != nil { + return nil, err + } + if parsed.BlockNumber == nil || parsed.LogIndex == nil { + return nil, errors.New("missing blockNumber or logIndex") + } + + return &Cursor{ + BlockNumber: uint64(*parsed.BlockNumber), + LogIndex: uint(*parsed.LogIndex), + }, nil +} diff --git a/pkg/resume/resume_test.go b/pkg/resume/resume_test.go new file mode 100644 index 0000000..ad458f4 --- /dev/null +++ b/pkg/resume/resume_test.go @@ -0,0 +1,1258 @@ +package resume_test + +import ( + "bytes" + "compress/gzip" + "encoding/json" + "errors" + "fmt" + "io" + "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/gzipstore" + "github.com/ethersphere/batch-export/pkg/resume" +) + +// Export file names. The format is detected from the file's leading bytes, so +// the name a case uses never decides how it is read. +const ( + plainFile = "export.ndjson" + gzipFile = "export.ndjson.gzip" +) + +// Format names reused across several tables of test cases. +const ( + nameFormatPlain = "plain ndjson" + nameFormatGzip = "gzip" +) + +// testLog builds a log shaped like the ones the exporter writes. types.Log +// always marshals blockNumber and logIndex, even at zero. +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"), + TxIndex: 9, + BlockHash: common.HexToHash("0x86dc5f9da5fcba5191f6b3d2ba995bd75532ef369a7baa3970b3fb292ae91324"), + Index: logIndex, + Removed: false, + } +} + +// ndjson renders logs as newline-delimited JSON, the exporter's output format. +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(l); err != nil { + t.Fatalf("encode log: %v", err) + } + } + return buf.Bytes() +} + +// gz compresses b into 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.Fatalf("gzip write: %v", err) + } + if err := w.Close(); err != nil { + t.Fatalf("gzip close: %v", err) + } + return buf.Bytes() +} + +// truncateLast drops the final n bytes of b, standing in for a write that a +// hard kill cut short. +func truncateLast(b []byte, n int) []byte { + return b[:len(b)-n] +} + +// write puts content in a 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.Fatalf("write %s: %v", path, err) + } + return path +} + +func TestReadCursor(t *testing.T) { + t.Parallel() + + threeLogs := ndjson(t, testLog(100, 0), testLog(101, 1), testLog(102, 7)) + + // many is large enough to span several of the gzip path's 64 KiB member + // buffers, exercising the buffered scan across bufio.ReadSlice refills + // rather than a single in-memory read. + manyLogs := make([]types.Log, 0, 2000) + for i := range 2000 { + manyLogs = append(manyLogs, testLog(uint64(1000+i), uint(i%16))) + } + many := ndjson(t, manyLogs...) + + // twoLogsEnd is the offset just past the newline closing the second of + // threeLogs: the last clean boundary once the third line loses its own. + twoLogsEnd := int64(len(ndjson(t, testLog(100, 0), testLog(101, 1)))) + threeLogsEnd := int64(len(threeLogs)) + + tests := []struct { + name string + content []byte + wantBlock uint64 + wantIndex uint + wantCompressed bool + // wantTruncated says whether bytes follow the last clean boundary. + // When it is false the boundary must be the end of the file, so + // wantCleanSize is only consulted for the truncated cases. + wantTruncated bool + wantCleanSize int64 + }{ + { + name: nameFormatPlain, + content: threeLogs, + wantBlock: 102, + wantIndex: 7, + }, + { + name: "single line", + content: ndjson(t, testLog(55, 3)), + wantBlock: 55, + wantIndex: 3, + }, + { + // A line cut short by an interrupted write. The cursor is the + // line before it and the partial bytes are reported, so a caller + // discards them instead of appending onto them. + name: "truncated trailing line is excluded from the clean boundary", + content: append(append([]byte{}, threeLogs...), []byte(`{"address":"0x45a15","topics":["0xae`)...), + wantBlock: 102, + wantIndex: 7, + wantTruncated: true, + wantCleanSize: threeLogsEnd, + }, + { + // json.Encoder writes a value and its newline in one call, so a + // last line that parses but has no newline is still a partial + // write: the cursor must fall back to the line before it, or the + // unterminated log would be skipped on resume and lost. + name: "trailing line without a newline is not a clean end", + content: bytes.TrimSuffix(threeLogs, []byte("\n")), + wantBlock: 101, + wantIndex: 1, + wantTruncated: true, + wantCleanSize: twoLogsEnd, + }, + { + name: "many logs", + content: many, + wantBlock: 2999, + wantIndex: 15, + }, + { + name: nameFormatGzip, + content: gz(t, threeLogs), + wantBlock: 102, + wantIndex: 7, + wantCompressed: true, + }, + { + name: "gzip spanning many logs", + content: gz(t, many), + wantBlock: 2999, + wantIndex: 15, + wantCompressed: true, + }, + { + name: "multi member gzip reads through to the last member", + content: append(gz(t, threeLogs), gz(t, ndjson(t, testLog(200, 2)))...), + wantBlock: 200, + wantIndex: 2, + wantCompressed: true, + }, + { + // A resume interrupted before the second member was flushed: the + // header is present but carries no decodable data. The clean + // boundary is the end of the first member, so the half-written + // header is reported rather than appended onto. + name: "multi member gzip with unflushed final member", + content: append(gz(t, threeLogs), []byte{0x1f, 0x8b, 0x08, 0, 0, 0, 0, 0, 0, 0xff}...), + wantBlock: 102, + wantIndex: 7, + wantCompressed: true, + wantTruncated: true, + wantCleanSize: int64(len(gz(t, threeLogs))), + }, + { + // Only the last member's trailer is lost. Everything before it is + // still a member boundary, so the file stays recoverable. + name: "multi member gzip with a truncated final member", + content: truncateLast(append(gz(t, threeLogs), gz(t, ndjson(t, testLog(200, 2)))...), 6), + wantBlock: 102, + wantIndex: 7, + wantCompressed: true, + wantTruncated: true, + wantCleanSize: int64(len(gz(t, threeLogs))), + }, + { + name: "empty trailing gzip member is valid", + content: append(gz(t, threeLogs), gz(t, nil)...), + wantBlock: 102, + wantIndex: 7, + wantCompressed: true, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + t.Parallel() + + got, err := resume.Read(write(t, plainFile, tt.content)) + if err != nil { + t.Fatalf("Read() error = %v, want nil", err) + } + if got.BlockNumber != tt.wantBlock { + t.Errorf("BlockNumber = %d, want %d", got.BlockNumber, tt.wantBlock) + } + if got.LogIndex != tt.wantIndex { + t.Errorf("LogIndex = %d, want %d", got.LogIndex, tt.wantIndex) + } + if got.Compressed != tt.wantCompressed { + t.Errorf("Compressed = %t, want %t", got.Compressed, tt.wantCompressed) + } + if got.Truncated != tt.wantTruncated { + t.Errorf("Truncated = %t, want %t", got.Truncated, tt.wantTruncated) + } + + // A file reported as untruncated must be clean all the way to its + // end, which is the invariant an appending caller relies on. + wantCleanSize := int64(len(tt.content)) + if tt.wantTruncated { + wantCleanSize = tt.wantCleanSize + } + if got.CleanSize != wantCleanSize { + t.Errorf("CleanSize = %d, want %d", got.CleanSize, wantCleanSize) + } + }) + } +} + +// TestReadRefusals covers §5's strict contract: the only irregularity +// tolerated is the tool's own interrupted final write. Content the tool never +// writes is ErrNotAnExport; a file consistent with tool output but holding no +// complete entry is ErrNoLogs. +func TestReadRefusals(t *testing.T) { + t.Parallel() + + logs := ndjson(t, testLog(100, 0), testLog(101, 0)) + + tests := []struct { + name string + content []byte + wantErr error + }{ + { + name: "empty file", + content: []byte{}, + wantErr: resume.ErrNoLogs, + }, + { + name: "plain file holding one unterminated line", + content: bytes.TrimSuffix(ndjson(t, testLog(100, 0)), []byte("\n")), + wantErr: resume.ErrNoLogs, + }, + { + // The tool never writes a blank line. + name: "only a newline", + content: []byte("\n"), + wantErr: resume.ErrNotAnExport, + }, + { + name: "trailing blank line after valid logs", + content: append(append([]byte{}, logs...), '\n'), + wantErr: resume.ErrNotAnExport, + }, + { + // A complete line that is not a log entry means the file was + // altered after export; refusing beats guessing what to cut. + name: "trailing non-log line after valid logs", + content: append(append([]byte{}, logs...), []byte("not-json\n")...), + wantErr: resume.ErrNotAnExport, + }, + { + name: "trailing log line missing blockNumber", + content: append(append([]byte{}, logs...), []byte("{\"address\":\"0x1\",\"topics\":[],\"data\":\"0x\"}\n")...), + wantErr: resume.ErrNotAnExport, + }, + { + name: "only garbage lines", + content: bytes.Repeat([]byte("not-json\n"), 10), + wantErr: resume.ErrNotAnExport, + }, + { + // Finding #4's shape: refused from a single tail read, no + // backward scan through the junk. + name: "newline-free tail longer than a line can be", + content: append(append([]byte{}, logs...), bytes.Repeat([]byte("x"), 1<<20+1)...), + wantErr: resume.ErrNotAnExport, + }, + { + name: "single unterminated line larger than the cap", + content: bytes.Repeat([]byte("x"), 2<<20), + wantErr: resume.ErrNotAnExport, + }, + { + // A cleanly terminated member whose content stops mid-line + // cannot come from this tool: members hold whole lines, and an + // interrupted write cannot produce a valid trailer. + name: "gzip member ending mid line", + content: gz(t, bytes.TrimSuffix(logs, []byte("\n"))), + wantErr: resume.ErrNotAnExport, + }, + { + // Finding #6's shape: foreign data concatenated as its own valid + // member. Refused rather than treated as a removable tail. + name: "gzip junk member after valid member", + content: append(gz(t, logs), gz(t, []byte("not-json\nalso-not\n"))...), + wantErr: resume.ErrNotAnExport, + }, + { + name: "gzip member holding a non-log line between logs", + content: gz(t, append(append([]byte{}, logs...), []byte("not-json\n")...)), + wantErr: resume.ErrNotAnExport, + }, + { + // A sole member without its trailer is an interrupted first + // write: nothing to resume from, so a fresh export is the remedy. + name: "gzip with a single truncated member", + content: truncateLast(gz(t, logs), 6), + wantErr: resume.ErrNoLogs, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + t.Parallel() + + if _, err := resume.Read(write(t, plainFile, tt.content)); !errors.Is(err, tt.wantErr) { + t.Fatalf("Read() error = %v, want %v", err, tt.wantErr) + } + }) + } +} + +func TestReadMissingFile(t *testing.T) { + t.Parallel() + + _, err := resume.Read(filepath.Join(t.TempDir(), "does-not-exist.ndjson")) + if !errors.Is(err, os.ErrNotExist) { + t.Fatalf("Read() error = %v, want os.ErrNotExist", err) + } +} + +func TestCursorSkip(t *testing.T) { + t.Parallel() + + cursor := &resume.Cursor{BlockNumber: 100, LogIndex: 5} + + tests := []struct { + name string + log types.Log + want bool + }{ + {name: "earlier block", log: testLog(99, 0), want: true}, + {name: "same block earlier index", log: testLog(100, 4), want: true}, + {name: "same block same index", log: testLog(100, 5), want: true}, + {name: "same block later index", log: testLog(100, 6), want: false}, + {name: "later block index zero", log: testLog(101, 0), want: false}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + t.Parallel() + + if got := cursor.Skip(tt.log); got != tt.want { + t.Errorf("Skip() = %t, want %t", got, tt.want) + } + }) + } +} + +// decompressAll reads every byte of a (possibly multi-member) gzip stream. +// Multistream is on by default, so concatenated members read as one stream. +func decompressAll(t *testing.T, b []byte) []byte { + t.Helper() + + reader, err := gzip.NewReader(bytes.NewReader(b)) + if err != nil { + t.Fatalf("gzip reader: %v", err) + } + defer reader.Close() + + got, err := io.ReadAll(reader) + if err != nil { + t.Fatalf("decompress: %v", err) + } + + return got +} + +// TestAppendResumeRoundTrip exercises resume, filestore and gzipstore together +// with no RPC: read the cursor, replay the boundary block plus new blocks +// through AppendLogsAsync filtered by Skip, check the result byte-for-byte +// against the original plus exactly the new logs, then resume again to confirm +// the appended file is itself resumable. +func TestAppendResumeRoundTrip(t *testing.T) { + t.Parallel() + + tests := []struct { + name string + compressed bool + }{ + {name: nameFormatPlain}, + {name: nameFormatGzip, compressed: true}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + t.Parallel() + + // original spans three blocks, the last of which carries two + // logs, so resuming must handle a genuinely partial block. + original := []types.Log{ + testLog(100, 0), + testLog(101, 0), + testLog(102, 0), + testLog(102, 1), + } + // boundaryHigher shares the cursor's block but was never saved, + // so it must be written rather than skipped. + boundaryHigher := testLog(102, 2) + newer := []types.Log{ + testLog(103, 0), + testLog(104, 0), + } + + plain := ndjson(t, original...) + + var ( + originalBytes []byte + name string + ) + if tt.compressed { + originalBytes = gz(t, plain) + name = "export.ndjson.gz" + } else { + originalBytes = plain + name = plainFile + } + path := write(t, name, originalBytes) + + cursor, err := resume.Read(path) + if err != nil { + t.Fatalf("Read() error = %v", err) + } + if cursor.BlockNumber != 102 || cursor.LogIndex != 1 { + t.Fatalf("cursor = {%d,%d}, want {102,1}", cursor.BlockNumber, cursor.LogIndex) + } + if cursor.Compressed != tt.compressed { + t.Fatalf("Compressed = %t, want %t", cursor.Compressed, tt.compressed) + } + + var w io.WriteCloser + if cursor.Compressed { + w, err = gzipstore.AppendWriter(path) + } else { + w, err = filestore.AppendWriter(path) + } + if err != nil { + t.Fatalf("AppendWriter() error = %v", err) + } + + // replay mimics a resumed export re-querying the boundary block: + // its already-written logs must be skipped, its never-saved + // higher-index log must not be, and the later blocks are new. + replay := make([]types.Log, 0, 3+len(newer)) + replay = append(replay, testLog(102, 0), testLog(102, 1), boundaryHigher) + replay = append(replay, newer...) + + ch := make(chan types.Log, len(replay)) + for _, l := range replay { + ch <- l + } + close(ch) + + if err := filestore.AppendLogsAsync(t.Context(), ch, w, cursor.Skip); err != nil { + t.Fatalf("AppendLogsAsync() error = %v", err) + } + + gotRaw, err := os.ReadFile(path) + if err != nil { + t.Fatalf("read %s: %v", path, err) + } + + // The original bytes must survive as an unchanged prefix: + // appending must never rewrite existing content. For gzip this + // also proves a genuine second member was added rather than the + // archive being decompressed and rewritten. + if len(gotRaw) < len(originalBytes) || !bytes.Equal(gotRaw[:len(originalBytes)], originalBytes) { + t.Fatalf("original bytes are not an unchanged prefix of the appended file") + } + + gotPlain := gotRaw + if tt.compressed { + gotPlain = decompressAll(t, gotRaw) + } + + all := make([]types.Log, 0, len(original)+1+len(newer)) + all = append(all, original...) + all = append(all, boundaryHigher) + all = append(all, newer...) + want := ndjson(t, all...) + + if !bytes.Equal(gotPlain, want) { + t.Fatalf("content = %s, want %s", gotPlain, want) + } + + cursor2, err := resume.Read(path) + if err != nil { + t.Fatalf("second Read() error = %v", err) + } + if cursor2.BlockNumber != 104 || cursor2.LogIndex != 0 { + t.Errorf("resumed cursor = {%d,%d}, want {104,0}", cursor2.BlockNumber, cursor2.LogIndex) + } + if cursor2.Compressed != tt.compressed { + t.Errorf("resumed Compressed = %t, want %t", cursor2.Compressed, tt.compressed) + } + }) + } +} + +// logsIn returns every log an export file holds, decompressing gzip first. +// Anything that is not a whole, newline-terminated NDJSON line fails the test, +// so a file that merely looks recovered is caught here rather than downstream. +func logsIn(t *testing.T, path string) []types.Log { + t.Helper() + + raw, err := os.ReadFile(path) + if err != nil { + t.Fatalf("read %s: %v", path, err) + } + if bytes.HasPrefix(raw, []byte{0x1f, 0x8b}) { + raw = decompressAll(t, raw) + } + if len(raw) == 0 { + return nil + } + if raw[len(raw)-1] != '\n' { + t.Fatalf("file does not end on a line boundary, last 80 bytes: %q", raw[max(0, len(raw)-80):]) + } + + var logs []types.Log + for i, line := range bytes.Split(bytes.TrimSuffix(raw, []byte("\n")), []byte("\n")) { + var l types.Log + if err := json.Unmarshal(line, &l); err != nil { + t.Fatalf("line %d does not parse: %v: %q", i+1, err, line) + } + logs = append(logs, l) + } + + return logs +} + +// ids renders logs as block/index pairs, the identity in which "no log +// duplicated and none lost" is measured. +func ids(logs []types.Log) []string { + out := make([]string, 0, len(logs)) + for _, l := range logs { + out = append(out, fmt.Sprintf("%d/%d", l.BlockNumber, l.Index)) + } + + return out +} + +// appendWriter opens the append writer that matches the cursor's format. +func appendWriter(t *testing.T, path string, cursor *resume.Cursor) io.WriteCloser { + t.Helper() + + var ( + w io.WriteCloser + err error + ) + if cursor.Compressed { + w, err = gzipstore.AppendWriter(path) + } else { + w, err = filestore.AppendWriter(path) + } + if err != nil { + t.Fatalf("AppendWriter() error = %v", err) + } + + return w +} + +// feed returns a closed channel already holding logs. +func feed(logs ...types.Log) <-chan types.Log { + ch := make(chan types.Log, len(logs)) + for _, l := range logs { + ch <- l + } + close(ch) + + return ch +} + +// TestCopyResumeRoundTrip is the spec's recommended workflow (§4): resume an +// archived snapshot into a NEW file. The input must come out byte-identical; +// the output must hold the input's content plus exactly the new entries and +// be itself resumable. +func TestCopyResumeRoundTrip(t *testing.T) { + t.Parallel() + + tests := []struct { + name string + compressed bool + }{ + {name: nameFormatPlain}, + {name: nameFormatGzip, compressed: true}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + t.Parallel() + + original := []types.Log{testLog(100, 0), testLog(101, 0), testLog(102, 0), testLog(102, 1)} + boundaryHigher := testLog(102, 2) + newer := []types.Log{testLog(103, 0), testLog(104, 0)} + + inputBytes := ndjson(t, original...) + if tt.compressed { + inputBytes = gz(t, inputBytes) + } + dir := t.TempDir() + input := filepath.Join(dir, "prev.snapshot") + output := filepath.Join(dir, "next.snapshot") + if err := os.WriteFile(input, inputBytes, 0o644); err != nil { + t.Fatalf("write input: %v", err) + } + + cursor, err := resume.Read(input) + if err != nil { + t.Fatalf("Read() error = %v", err) + } + + discarded, err := resume.PrepareOutput(cursor, input, output) + if err != nil { + t.Fatalf("PrepareOutput() error = %v", err) + } + if discarded != 0 { + t.Errorf("discarded = %d, want 0 for a clean input", discarded) + } + + w := appendWriter(t, output, cursor) + replay := []types.Log{testLog(102, 0), testLog(102, 1), boundaryHigher} + replay = append(replay, newer...) + if err := filestore.AppendLogsAsync(t.Context(), feed(replay...), w, cursor.Skip); err != nil { + t.Fatalf("AppendLogsAsync() error = %v", err) + } + + // The input is untouched, byte for byte. + gotInput, err := os.ReadFile(input) + if err != nil { + t.Fatalf("read input: %v", err) + } + if !bytes.Equal(gotInput, inputBytes) { + t.Fatal("input file was modified by a copy-mode resume") + } + + // The output begins with the input's exact bytes (raw prefix + // copy, no recompression) and holds the full sequence. + gotOutput, err := os.ReadFile(output) + if err != nil { + t.Fatalf("read output: %v", err) + } + if len(gotOutput) < len(inputBytes) || !bytes.Equal(gotOutput[:len(inputBytes)], inputBytes) { + t.Fatal("input bytes are not an unchanged prefix of the output") + } + all := append(append(append([]types.Log{}, original...), boundaryHigher), newer...) + if got, want := ids(logsIn(t, output)), ids(all); !slices.Equal(got, want) { + t.Fatalf("output logs = %v, want %v", got, want) + } + + cursor2, err := resume.Read(output) + if err != nil { + t.Fatalf("Read(output) error = %v", err) + } + if cursor2.BlockNumber != 104 || cursor2.LogIndex != 0 || cursor2.Truncated { + t.Errorf("output cursor = {%d,%d,truncated=%t}, want {104,0,false}", cursor2.BlockNumber, cursor2.LogIndex, cursor2.Truncated) + } + }) + } +} + +// TestCopyResumeFromInterruptedInput: copy mode never repairs the input — the +// interrupted tail stays in it — while the output gets only clean content +// plus the re-fetched entries. +func TestCopyResumeFromInterruptedInput(t *testing.T) { + t.Parallel() + + saved := ndjson(t, testLog(100, 0), testLog(101, 0)) + + tests := []struct { + name string + content []byte + compressed bool + }{ + { + name: "plain with a truncated last line", + content: append(append([]byte{}, saved...), []byte(`{"address":"0x45a15`)...), + }, + { + name: "gzip with a truncated final member", + content: append(gz(t, saved), truncateLast(gz(t, ndjson(t, testLog(102, 0))), 6)...), + compressed: true, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + t.Parallel() + + dir := t.TempDir() + input := filepath.Join(dir, "prev.snapshot") + output := filepath.Join(dir, "next.snapshot") + if err := os.WriteFile(input, tt.content, 0o644); err != nil { + t.Fatalf("write input: %v", err) + } + + cursor, err := resume.Read(input) + if err != nil { + t.Fatalf("Read() error = %v", err) + } + if !cursor.Truncated { + t.Fatal("test setup: input should be truncated") + } + + discarded, err := resume.PrepareOutput(cursor, input, output) + if err != nil { + t.Fatalf("PrepareOutput() error = %v", err) + } + if want := int64(len(tt.content)) - cursor.CleanSize; discarded != want { + t.Errorf("discarded = %d, want %d", discarded, want) + } + + w := appendWriter(t, output, cursor) + replay := []types.Log{testLog(101, 0), testLog(102, 0), testLog(103, 0)} + if err := filestore.AppendLogsAsync(t.Context(), feed(replay...), w, cursor.Skip); err != nil { + t.Fatalf("AppendLogsAsync() error = %v", err) + } + + gotInput, err := os.ReadFile(input) + if err != nil { + t.Fatalf("read input: %v", err) + } + if !bytes.Equal(gotInput, tt.content) { + t.Fatal("input was modified, interrupted tail included it must stay") + } + + want := []types.Log{testLog(100, 0), testLog(101, 0), testLog(102, 0), testLog(103, 0)} + if got := ids(logsIn(t, output)); !slices.Equal(got, ids(want)) { + t.Fatalf("output logs = %v, want %v", got, ids(want)) + } + }) + } +} + +// TestPrepareOutputSameFileUnderDifferentSpellings pins Fix 1: two spellings +// of the SAME underlying file — an absolute path against a relative one, or a +// symlink pointed at the input — must be treated as in-place, not copy mode. +// Before the fix, PrepareOutput compared filepath.Clean strings, missed both +// cases, and its os.Create(outputPath) truncated the input to zero bytes +// before io.CopyN ever got to read from it, destroying the file it was +// supposed to leave untouched. +// +// Each case covers both a clean input (must survive byte for byte) and one +// with an interrupted final write (must be truncated to exactly CleanSize, +// the same result prepareInPlace produces for a same-path call). +func TestPrepareOutputSameFileUnderDifferentSpellings(t *testing.T) { + logs := ndjson(t, testLog(100, 0), testLog(101, 0)) + interrupted := append(append([]byte{}, logs...), []byte(`{"address":"0x45a1502382541`)...) + + contents := []struct { + name string + content []byte + }{ + {name: "clean input", content: logs}, + {name: "interrupted write", content: interrupted}, + } + + // spelling produces an outputPath that names the same file as input by + // some route other than an identical string. + spellings := []struct { + name string + outPath func(t *testing.T, dir, input string) string + }{ + { + // t.Chdir affects the whole process, so this case (and therefore + // the whole test) cannot run in parallel with anything else. + name: "absolute input, relative output", + outPath: func(t *testing.T, dir, input string) string { + t.Chdir(dir) + return filepath.Base(input) + }, + }, + { + name: "symlink to the input", + outPath: func(t *testing.T, dir, input string) string { + link := filepath.Join(dir, "alias-of-input") + if err := os.Symlink(input, link); err != nil { + t.Skipf("symlinks unsupported on this filesystem: %v", err) + } + return link + }, + }, + } + + for _, ct := range contents { + for _, sp := range spellings { + t.Run(ct.name+"/"+sp.name, func(t *testing.T) { + dir := t.TempDir() + input := filepath.Join(dir, "snapshot.ndjson") + if err := os.WriteFile(input, ct.content, 0o644); err != nil { + t.Fatalf("write input: %v", err) + } + + cursor, err := resume.Read(input) + if err != nil { + t.Fatalf("Read() error = %v", err) + } + + output := sp.outPath(t, dir, input) + + discarded, err := resume.PrepareOutput(cursor, input, output) + if err != nil { + t.Fatalf("PrepareOutput() error = %v", err) + } + + wantDiscarded := int64(len(ct.content)) - cursor.CleanSize + if discarded != wantDiscarded { + t.Errorf("discarded = %d, want %d", discarded, wantDiscarded) + } + + got, err := os.ReadFile(input) + if err != nil { + t.Fatalf("read input: %v", err) + } + want := ct.content[:cursor.CleanSize] + if !bytes.Equal(got, want) { + t.Fatalf("input holds %d bytes, want the %d-byte clean prefix preserved (copy mode must not have truncated the input)", len(got), len(want)) + } + }) + } + } +} + +// TestReadRejectsCorruptedChecksum pins Fix 3: a member whose body is intact +// but whose trailing CRC has been altered is corruption or tampering, not the +// tool's own interrupted write, and must be refused with ErrNotAnExport +// rather than silently treated as a truncated final member. +func TestReadRejectsCorruptedChecksum(t *testing.T) { + t.Parallel() + + content := gz(t, ndjson(t, testLog(100, 0), testLog(101, 0))) + // The trailer is the last 8 bytes: a 4-byte CRC32 followed by a 4-byte + // ISIZE. Flipping a byte within the first 4 corrupts the CRC alone, + // leaving the member body and its length untouched. + content[len(content)-8] ^= 0xff + + _, err := resume.Read(write(t, gzipFile, content)) + if !errors.Is(err, resume.ErrNotAnExport) { + t.Fatalf("Read() error = %v, want ErrNotAnExport", err) + } +} + +// TestReadGzipCutInsideHeader pins Fix 4: a file cut short before the gzip +// header is even complete is an interrupted first write, exactly like the +// plain format's equivalent, and must report ErrNoLogs rather than a bare +// wrapped error that escapes the two-sentinel taxonomy. +func TestReadGzipCutInsideHeader(t *testing.T) { + t.Parallel() + + full := gz(t, ndjson(t, testLog(100, 0))) + + _, err := resume.Read(write(t, gzipFile, full[:5])) + if !errors.Is(err, resume.ErrNoLogs) { + t.Fatalf("Read() error = %v, want ErrNoLogs", err) + } +} + +// TestResumeAfterInterruptedWrite covers the three shapes a run killed +// mid-write leaves behind: a plain line cut in half, a plain line that parses +// but never got its newline, and a gzip member that never got its trailer. +// Appending onto any of them blindly corrupts the file. +// +// Each case walks the whole recovery — read the cursor, discard past the clean +// boundary, append what a resumed query returns — and requires the file to +// parse end to end holding exactly the four logs, in order, none duplicated +// and none lost. +func TestResumeAfterInterruptedWrite(t *testing.T) { + t.Parallel() + + // saved is what the interrupted run had managed to write whole. + saved := ndjson(t, testLog(100, 0), testLog(101, 0)) + savedFirst := ndjson(t, testLog(100, 0)) + + // want is the full export: what the file must hold once resumed. + want := []types.Log{testLog(100, 0), testLog(101, 0), testLog(102, 0), testLog(103, 0)} + + tests := []struct { + name string + // content is the file the interrupted run left behind. + content []byte + fileName string + compressed bool + // wantBlock and wantIndex are the last entry that is complete and + // properly terminated. + wantBlock uint64 + wantIndex uint + wantCleanSize int64 + // replay is what the resumed query returns from wantBlock onwards. + replay []types.Log + }{ + { + // Appending onto the half line glues the next log onto it, + // destroying block 102. + name: "plain file with a truncated last line", + content: append(append([]byte{}, saved...), []byte(`{"address":"0x45a1502382541cd610cc9068e88727426b6`)...), + fileName: plainFile, + wantBlock: 101, + wantIndex: 0, + wantCleanSize: int64(len(saved)), + replay: []types.Log{testLog(101, 0), testLog(102, 0), testLog(103, 0)}, + }, + { + // The nastiest: the line parses, so a cursor naming it would make + // the resumed query skip block 101 while the append fuses the next + // log onto it — unparseable, and never re-fetched. The cursor must + // name block 100 so that 101 is fetched again. + name: "plain file with a valid but unterminated last line", + content: bytes.TrimSuffix(saved, []byte("\n")), + fileName: plainFile, + wantBlock: 100, + wantIndex: 0, + wantCleanSize: int64(len(savedFirst)), + replay: []types.Log{testLog(100, 0), testLog(101, 0), testLog(102, 0), testLog(103, 0)}, + }, + { + // The final member lost its trailer, so a member appended after + // it sits behind corrupt data and can never be read back, while + // every later resume appends more of the same and reports + // success. + name: "gzip file with a truncated final member", + content: append(gz(t, savedFirst), truncateLast(gz(t, ndjson(t, testLog(101, 0))), 6)...), + fileName: gzipFile, + compressed: true, + wantBlock: 100, + wantIndex: 0, + wantCleanSize: int64(len(gz(t, savedFirst))), + replay: []types.Log{testLog(100, 0), testLog(101, 0), testLog(102, 0), testLog(103, 0)}, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + t.Parallel() + + path := write(t, tt.fileName, tt.content) + + cursor, err := resume.Read(path) + if err != nil { + t.Fatalf("Read() error = %v", err) + } + if cursor.BlockNumber != tt.wantBlock || cursor.LogIndex != tt.wantIndex { + t.Fatalf("cursor = {%d,%d}, want {%d,%d}", cursor.BlockNumber, cursor.LogIndex, tt.wantBlock, tt.wantIndex) + } + if cursor.Compressed != tt.compressed { + t.Fatalf("Compressed = %t, want %t", cursor.Compressed, tt.compressed) + } + if !cursor.Truncated { + t.Fatalf("Truncated = false, want true: the partial write went unreported") + } + if cursor.CleanSize != tt.wantCleanSize { + t.Fatalf("CleanSize = %d, want %d", cursor.CleanSize, tt.wantCleanSize) + } + + // Recovery: drop the partial tail, exactly as the export command + // does, and never further than the reported boundary. + discarded, err := resume.PrepareOutput(cursor, path, path) + if err != nil { + t.Fatalf("PrepareOutput() error = %v", err) + } + if want := int64(len(tt.content)) - cursor.CleanSize; discarded != want { + t.Errorf("discarded = %d, want %d", discarded, want) + } + + if err := filestore.AppendLogsAsync(t.Context(), feed(tt.replay...), appendWriter(t, path, cursor), cursor.Skip); err != nil { + t.Fatalf("AppendLogsAsync() error = %v", err) + } + + if got := ids(logsIn(t, path)); !slices.Equal(got, ids(want)) { + t.Fatalf("logs = %v, want %v", got, ids(want)) + } + + // The recovered file must itself resume cleanly, or a second + // interruption would start the corruption over again. + cursor2, err := resume.Read(path) + if err != nil { + t.Fatalf("second Read() error = %v", err) + } + if cursor2.BlockNumber != 103 || cursor2.LogIndex != 0 { + t.Errorf("resumed cursor = {%d,%d}, want {103,0}", cursor2.BlockNumber, cursor2.LogIndex) + } + if cursor2.Truncated { + t.Errorf("Truncated = true, want false: the recovered file is not clean") + } + }) + } +} + +// TestReadReportsGzipReadErrors pins Fix 2: a gzip stream that stops early +// must not be reported as a clean read just because some lines decoded. The +// old code returned the last cursor it had seen with a nil error, which let +// every later resume append more data behind the corruption, for ever. +func TestReadReportsGzipReadErrors(t *testing.T) { + t.Parallel() + + clean := gz(t, ndjson(t, testLog(100, 0))) + content := append(append([]byte{}, clean...), truncateLast(gz(t, ndjson(t, testLog(101, 0), testLog(102, 0))), 6)...) + + cursor, err := resume.Read(write(t, gzipFile, content)) + if err != nil { + t.Fatalf("Read() error = %v", err) + } + if !cursor.Truncated { + t.Error("Truncated = false, want true: the unclean stream was reported as clean") + } + if cursor.CleanSize != int64(len(clean)) { + t.Errorf("CleanSize = %d, want %d", cursor.CleanSize, len(clean)) + } + // The cursor must not name a log that only the broken member holds: it is + // about to be discarded, and a cursor past it would skip the refetch. + if cursor.BlockNumber != 100 || cursor.LogIndex != 0 { + t.Errorf("cursor = {%d,%d}, want {100,0}", cursor.BlockNumber, cursor.LogIndex) + } +} + +// TestGzipCleanSizeIsAMemberBoundary checks the offset the gzip walk reports +// is exactly a member boundary and not a byte off, which is what makes +// truncating to it safe. Reading the file back after cutting it there must +// yield the members that came before, whole. +func TestGzipCleanSizeIsAMemberBoundary(t *testing.T) { + t.Parallel() + + first := ndjson(t, testLog(100, 0), testLog(101, 0)) + second := ndjson(t, testLog(102, 0)) + + members := append(gz(t, first), gz(t, second)...) + boundary := int64(len(members)) + content := append(append([]byte{}, members...), truncateLast(gz(t, ndjson(t, testLog(103, 0))), 4)...) + + path := write(t, gzipFile, content) + + cursor, err := resume.Read(path) + if err != nil { + t.Fatalf("Read() error = %v", err) + } + if cursor.CleanSize != boundary { + t.Fatalf("CleanSize = %d, want %d", cursor.CleanSize, boundary) + } + + if err := os.Truncate(path, cursor.CleanSize); err != nil { + t.Fatalf("truncate %s: %v", path, err) + } + + raw, err := os.ReadFile(path) + if err != nil { + t.Fatalf("read %s: %v", path, err) + } + if got, want := decompressAll(t, raw), append(append([]byte{}, first...), second...); !bytes.Equal(got, want) { + t.Fatalf("content = %s, want %s", got, want) + } +} + +// TestResumeMultipleEmptyAppends checks that resuming multiple times when no +// new logs exist on chain (e.g. producing empty writes or empty gzip members) +// preserves existing data and allows subsequent appends with new logs. +func TestResumeMultipleEmptyAppends(t *testing.T) { + t.Parallel() + + for _, tc := range []struct { + name string + fileName string + compressed bool + }{ + {name: nameFormatPlain, fileName: plainFile, compressed: false}, + {name: nameFormatGzip, fileName: gzipFile, compressed: true}, + } { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + + path := filepath.Join(t.TempDir(), tc.fileName) + + // Initial write of 2 logs + initLogs := []types.Log{testLog(100, 0), testLog(101, 0)} + var initialContent []byte + if tc.compressed { + initialContent = gz(t, ndjson(t, initLogs...)) + } else { + initialContent = ndjson(t, initLogs...) + } + if err := os.WriteFile(path, initialContent, 0o644); err != nil { + t.Fatalf("write initial file: %v", err) + } + + // First empty resume: 0 new logs + cursor, err := resume.Read(path) + if err != nil { + t.Fatalf("first Read() error = %v", err) + } + if _, err := resume.PrepareOutput(cursor, path, path); err != nil { + t.Fatalf("first PrepareOutput() error = %v", err) + } + if err := filestore.AppendLogsAsync(t.Context(), feed(), appendWriter(t, path, cursor), cursor.Skip); err != nil { + t.Fatalf("first AppendLogsAsync() error = %v", err) + } + + // Second empty resume: 0 new logs + cursor, err = resume.Read(path) + if err != nil { + t.Fatalf("second Read() error = %v", err) + } + if _, err := resume.PrepareOutput(cursor, path, path); err != nil { + t.Fatalf("second PrepareOutput() error = %v", err) + } + if err := filestore.AppendLogsAsync(t.Context(), feed(), appendWriter(t, path, cursor), cursor.Skip); err != nil { + t.Fatalf("second AppendLogsAsync() error = %v", err) + } + + // Third resume: 2 new logs + newLogs := []types.Log{testLog(101, 0), testLog(102, 0), testLog(103, 0)} + cursor, err = resume.Read(path) + if err != nil { + t.Fatalf("third Read() error = %v", err) + } + if _, err := resume.PrepareOutput(cursor, path, path); err != nil { + t.Fatalf("third PrepareOutput() error = %v", err) + } + if err := filestore.AppendLogsAsync(t.Context(), feed(newLogs...), appendWriter(t, path, cursor), cursor.Skip); err != nil { + t.Fatalf("third AppendLogsAsync() error = %v", err) + } + + // Verify all 4 logs are present and correctly ordered + want := []types.Log{testLog(100, 0), testLog(101, 0), testLog(102, 0), testLog(103, 0)} + got := logsIn(t, path) + if !slices.Equal(ids(got), ids(want)) { + t.Fatalf("logs = %v, want %v", ids(got), ids(want)) + } + }) + } +} + +// TestCursorSkipWithIndexGaps tests that Cursor.Skip correctly skips +// non-consecutive log indices within the same block and preceding blocks. +func TestCursorSkipWithIndexGaps(t *testing.T) { + t.Parallel() + + cursor := &resume.Cursor{ + BlockNumber: 500, + LogIndex: 7, + } + + tests := []struct { + name string + log types.Log + want bool + }{ + {name: "earlier block", log: testLog(499, 100), want: true}, + {name: "same block lower index", log: testLog(500, 3), want: true}, + {name: "same block exact index", log: testLog(500, 7), want: true}, + {name: "same block higher index", log: testLog(500, 8), want: false}, + {name: "same block much higher index", log: testLog(500, 20), want: false}, + {name: "later block index 0", log: testLog(501, 0), want: false}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + t.Parallel() + if got := cursor.Skip(tt.log); got != tt.want { + t.Errorf("Skip(%+v) = %v, want %v", tt.log, got, tt.want) + } + }) + } +} + +// TestCRLFLineEndings ensures NDJSON files with CRLF (\r\n) line endings +// are parsed and resumed without errors. +func TestCRLFLineEndings(t *testing.T) { + t.Parallel() + + raw := ndjson(t, testLog(100, 0), testLog(101, 5)) + crlf := bytes.ReplaceAll(raw, []byte("\n"), []byte("\r\n")) + + path := write(t, plainFile, crlf) + cursor, err := resume.Read(path) + if err != nil { + t.Fatalf("Read() error = %v", err) + } + if cursor.BlockNumber != 101 || cursor.LogIndex != 5 { + t.Errorf("cursor = {%d,%d}, want {101,5}", cursor.BlockNumber, cursor.LogIndex) + } + if cursor.CleanSize != int64(len(crlf)) { + t.Errorf("CleanSize = %d, want %d", cursor.CleanSize, len(crlf)) + } +} + +// TestPrepareOutputInvalidDirectory asserts that copying to a non-existent +// target directory reports an error and leaves the input file untouched. +func TestPrepareOutputInvalidDirectory(t *testing.T) { + t.Parallel() + + content := ndjson(t, testLog(100, 0)) + input := write(t, plainFile, content) + + cursor, err := resume.Read(input) + if err != nil { + t.Fatalf("Read() error = %v", err) + } + + output := filepath.Join(t.TempDir(), "nonexistent", "nested", "out.ndjson") + if _, err := resume.PrepareOutput(cursor, input, output); err == nil { + t.Fatal("PrepareOutput() succeeded for non-existent dir, want error") + } + + // Assert input remains intact + got, err := os.ReadFile(input) + if err != nil { + t.Fatalf("ReadFile() error = %v", err) + } + if !bytes.Equal(got, content) { + t.Fatal("input was modified on failed PrepareOutput") + } +}