filer.meta.backup: stream from the start of an unfinished walk (#12458)

* filer.meta.backup: stream from the start of an unfinished walk

A walk only inserts entries, so when the first walk is interrupted the
next start walks again and streams from the second walk's start: deletes
and renames between the two walks never reach the backup store. Keep the
first walk's start time in the store until a walk completes and stream
from it.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* filer.meta.backup: start fresh once the stream has passed an earlier walk's start

A walk start left in the store by a crash between saving the offset and
clearing it, or by an interrupted -restart followed by a plain restart,
stops mattering once the stream saves an offset past it. Ignore it then
instead of replaying from it on the next walk.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* filer.meta.backup: fail the walk on an unreadable offset

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>

weed/command/filer_meta_backup.go

@@ -2,6 +2,7 @@ package command
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"reflect"
 	"strings"
@@ -109,17 +110,10 @@ func runFilerMetaBackup(cmd *Command, args []string) bool {
 	}
 
 	if *metaBackup.restart || missingPreviousBackup {
-		glog.V(0).Infof("traversing metadata tree...")
-		startTime := time.Now()
-		if err := metaBackup.traverseMetadata(); err != nil {
-			glog.Errorf("traverse meta data: %v", err)
+		if err := metaBackup.copyMetadata(metaBackup.traverseMetadata); err != nil {
+			glog.Exitf("%v", err)
 			return true
 		}
-		if err := metaBackup.setOffset(startTime); err != nil {
-			glog.Exitf("set metadata backup offset: %v", err)
-			return true
-		}
-		glog.V(0).Infof("metadata copied up to %v", startTime)
 	}
 
 	for {
@@ -162,6 +156,52 @@ func (metaBackup *FilerMetaBackupOptions) shouldInclude(fullpath string) bool {
 	return metaBackup.pathFilter.ShouldInclude(fullpath)
 }
 
+// copyMetadata walks the tree, then saves the offset the stream resumes from.
+// A walk only inserts, so the stream has to replay every delete and rename
+// since the first walk that did not finish: its start time stays in the store
+// until a walk completes.
+func (metaBackup *FilerMetaBackupOptions) copyMetadata(traverse func() error) error {
+	startTime, err := metaBackup.walkStartTime()
+	if err != nil {
+		return err
+	}
+	glog.V(0).Infof("traversing metadata tree...")
+	if err := traverse(); err != nil {
+		return fmt.Errorf("traverse meta data: %w", err)
+	}
+	if err := metaBackup.setOffset(startTime); err != nil {
+		return fmt.Errorf("set metadata backup offset: %w", err)
+	}
+	if err := metaBackup.store.KvDelete(context.Background(), MetaBackupWalkKey); err != nil {
+		glog.Warningf("clear metadata walk start: %v", err)
+	}
+	glog.V(0).Infof("metadata copied up to %v", startTime)
+	return nil
+}
+
+// walkStartTime is the start of an unfinished earlier walk, unless the stream
+// has saved an offset past it since, or else now, saved as this walk's start.
+func (metaBackup *FilerMetaBackupOptions) walkStartTime() (time.Time, error) {
+	earlier, err := metaBackup.getTime(MetaBackupWalkKey)
+	if err == nil {
+		offset, offsetErr := metaBackup.getOffset()
+		if offsetErr != nil && !errors.Is(offsetErr, filer.ErrKvNotFound) {
+			return time.Time{}, fmt.Errorf("read metadata backup offset: %w", offsetErr)
+		}
+		if offsetErr != nil || !offset.After(earlier) {
+			glog.V(0).Infof("an earlier metadata walk started at %v did not finish; streaming from then", earlier)
+			return earlier, nil
+		}
+	} else if !errors.Is(err, filer.ErrKvNotFound) {
+		return time.Time{}, fmt.Errorf("read metadata walk start: %w", err)
+	}
+	now := time.Now()
+	if err := metaBackup.setTime(MetaBackupWalkKey, now); err != nil {
+		return time.Time{}, fmt.Errorf("save metadata walk start: %w", err)
+	}
+	return now, nil
+}
+
 func (metaBackup *FilerMetaBackupOptions) traverseMetadata() (err error) {
 	return filer_pb.TraverseBfs(context.Background(), metaBackup, util.FullPath(*metaBackup.filerDirectory), func(parentPath util.FullPath, entry *filer_pb.Entry) error {
 		fullpath := string(parentPath.Child(entry.Name))
@@ -178,7 +218,8 @@ func (metaBackup *FilerMetaBackupOptions) traverseMetadata() (err error) {
 }
 
 var (
-	MetaBackupKey = []byte("metaBackup")
+	MetaBackupKey     = []byte("metaBackup")
+	MetaBackupWalkKey = []byte("metaBackupWalk")
 )
 
 func (metaBackup *FilerMetaBackupOptions) streamMetadataBackup(startTime time.Time) error {
@@ -274,23 +315,27 @@ func (metaBackup *FilerMetaBackupOptions) streamMetadataBackup(startTime time.Ti
 }
 
 func (metaBackup *FilerMetaBackupOptions) getOffset() (lastWriteTime time.Time, err error) {
-	value, err := metaBackup.store.KvGet(context.Background(), MetaBackupKey)
+	return metaBackup.getTime(MetaBackupKey)
+}
+
+func (metaBackup *FilerMetaBackupOptions) setOffset(lastWriteTime time.Time) error {
+	return metaBackup.setTime(MetaBackupKey, lastWriteTime)
+}
+
+func (metaBackup *FilerMetaBackupOptions) getTime(key []byte) (time.Time, error) {
+	value, err := metaBackup.store.KvGet(context.Background(), key)
 	if err != nil {
-		return
+		return time.Time{}, err
 	}
 	tsNs := util.BytesToUint64(value)
 
 	return time.Unix(0, int64(tsNs)), nil
 }
 
-func (metaBackup *FilerMetaBackupOptions) setOffset(lastWriteTime time.Time) error {
+func (metaBackup *FilerMetaBackupOptions) setTime(key []byte, t time.Time) error {
 	valueBuf := make([]byte, 8)
-	util.Uint64toBytes(valueBuf, uint64(lastWriteTime.UnixNano()))
-
-	if err := metaBackup.store.KvPut(context.Background(), MetaBackupKey, valueBuf); err != nil {
-		return err
-	}
-	return nil
+	util.Uint64toBytes(valueBuf, uint64(t.UnixNano()))
+	return metaBackup.store.KvPut(context.Background(), key, valueBuf)
 }
 
 var _ = filer_pb.FilerClient(&FilerMetaBackupOptions{})

weed/command/filer_meta_backup_test.go

@@ -0,0 +1,120 @@
+package command
+
+import (
+	"context"
+	"errors"
+	"testing"
+	"time"
+
+	"github.com/seaweedfs/seaweedfs/weed/filer"
+)
+
+type kvOnlyStore struct {
+	filer.FilerStore
+	kv     map[string][]byte
+	getErr error
+}
+
+func (s *kvOnlyStore) KvGet(_ context.Context, key []byte) ([]byte, error) {
+	if s.getErr != nil && string(key) == string(MetaBackupKey) {
+		return nil, s.getErr
+	}
+	value, found := s.kv[string(key)]
+	if !found {
+		return nil, filer.ErrKvNotFound
+	}
+	return value, nil
+}
+
+func (s *kvOnlyStore) KvPut(_ context.Context, key []byte, value []byte) error {
+	s.kv[string(key)] = value
+	return nil
+}
+
+func (s *kvOnlyStore) KvDelete(_ context.Context, key []byte) error {
+	delete(s.kv, string(key))
+	return nil
+}
+
+// TestCopyMetadataStreamsFromAnUnfinishedWalk: a walk only inserts, so a walk
+// that restarts after an interrupted one must stream from the first walk's
+// start, or the deletes between the two walks never reach the backup.
+func TestCopyMetadataStreamsFromAnUnfinishedWalk(t *testing.T) {
+	backup := &FilerMetaBackupOptions{store: &kvOnlyStore{kv: map[string][]byte{}}}
+
+	beforeFirst := time.Now()
+	interrupted := errors.New("killed")
+	if err := backup.copyMetadata(func() error { return interrupted }); !errors.Is(err, interrupted) {
+		t.Fatalf("interrupted walk: err = %v, want %v", err, interrupted)
+	}
+	if _, err := backup.getOffset(); !errors.Is(err, filer.ErrKvNotFound) {
+		t.Fatalf("offset after an interrupted walk: err = %v, want none saved", err)
+	}
+	firstStart, err := backup.getTime(MetaBackupWalkKey)
+	if err != nil || firstStart.Before(beforeFirst) {
+		t.Fatalf("walk start = %v, %v; want the first walk's start", firstStart, err)
+	}
+
+	time.Sleep(time.Millisecond)
+	if err := backup.copyMetadata(func() error { return interrupted }); err == nil {
+		t.Fatal("second interrupted walk: err = nil")
+	}
+	if got, _ := backup.getTime(MetaBackupWalkKey); !got.Equal(firstStart) {
+		t.Fatalf("walk start after a second interruption = %v, want the first walk's %v", got, firstStart)
+	}
+
+	if err := backup.copyMetadata(func() error { return nil }); err != nil {
+		t.Fatalf("finished walk: %v", err)
+	}
+	offset, err := backup.getOffset()
+	if err != nil || !offset.Equal(firstStart) {
+		t.Fatalf("offset = %v, %v; want the first walk's start %v", offset, err, firstStart)
+	}
+	if _, err := backup.getTime(MetaBackupWalkKey); !errors.Is(err, filer.ErrKvNotFound) {
+		t.Fatalf("walk start after a finished walk: err = %v, want it cleared", err)
+	}
+
+	beforeRestart := time.Now()
+	if err := backup.copyMetadata(func() error { return nil }); err != nil {
+		t.Fatalf("walk after a finished one: %v", err)
+	}
+	if offset, _ := backup.getOffset(); offset.Before(beforeRestart) {
+		t.Fatalf("offset = %v, want a fresh walk's start after %v", offset, beforeRestart)
+	}
+}
+
+// TestCopyMetadataIgnoresAWalkStartTheStreamPassed: a walk start left behind
+// (a crash between saving the offset and clearing it, or an interrupted
+// -restart followed by a plain restart) only matters until the stream saves an
+// offset past it; after that a new walk starts fresh.
+func TestCopyMetadataIgnoresAWalkStartTheStreamPassed(t *testing.T) {
+	backup := &FilerMetaBackupOptions{store: &kvOnlyStore{kv: map[string][]byte{}}}
+	walkStart := time.Now().Add(-time.Hour)
+	if err := backup.setTime(MetaBackupWalkKey, walkStart); err != nil {
+		t.Fatal(err)
+	}
+
+	if err := backup.setOffset(walkStart); err != nil {
+		t.Fatal(err)
+	}
+	if got, err := backup.walkStartTime(); err != nil || !got.Equal(walkStart) {
+		t.Fatalf("walk start with the offset at it = %v, %v; want %v", got, err, walkStart)
+	}
+
+	if err := backup.setOffset(walkStart.Add(time.Minute)); err != nil {
+		t.Fatal(err)
+	}
+	store := backup.store.(*kvOnlyStore)
+	store.getErr = errors.New("connection reset")
+	if _, err := backup.walkStartTime(); !errors.Is(err, store.getErr) {
+		t.Fatalf("walk start with an unreadable offset: err = %v, want the read error", err)
+	}
+	store.getErr = nil
+	beforeWalk := time.Now()
+	if err := backup.copyMetadata(func() error { return nil }); err != nil {
+		t.Fatal(err)
+	}
+	if offset, _ := backup.getOffset(); offset.Before(beforeWalk) {
+		t.Fatalf("offset = %v, want a fresh walk's start after %v", offset, beforeWalk)
+	}
+}