feat(filer.backup): -initialSnapshot seeds destination from live tree (#9126)

* feat(filer.backup): -initialSnapshot seeds destination from live tree

Replaying the metadata event log on a fresh sync only leaves files that
still exist on the source at replay time: any entry that was created and
later deleted is replayed as a create/delete pair and never materializes
on the destination. Users who wipe the destination and re-run
filer.backup therefore see "only new files" instead of a full backup,
even when -timeAgo=876000h is passed and the subscription genuinely
starts from epoch (ref discussion #8672).

Add a -initialSnapshot opt-in flag: when set on a fresh sync (no prior
checkpoint, -timeAgo unset), walk the live filer tree under -filerPath
via TraverseBfs and seed the destination through sink.CreateEntry, then
persist the walk-start timestamp as the checkpoint and subscribe from
there. Capturing the timestamp before the walk lets the subscription
catch any create/update/delete racing with the walk — sink CreateEntry
is idempotent across the builtin sinks so replay is safe.

Honors existing -filerExcludePaths / -filerExcludeFileNames /
-filerExcludePathPatterns filters and skips /topics/.system/log the
same way the subscription path does.

Also log "starting from <t> (no prior checkpoint)" instead of a
misleading "resuming from 1970-01-01" when the KV has no stored offset.

* fix(filer.backup): guard initialSnapshot counters under TraverseBfs workers

TraverseBfs fans the callback out across 5 worker goroutines, so the
entryCount / byteCount updates and the 5-second progress-log gate in
runInitialSnapshot were racing. Switch the counters to atomic.Int64 and
protect the lastLog check/update with a short-scoped mutex so the heavy
sink.CreateEntry call stays outside the critical section.

Flagged by gemini-code-assist on #9126; verified with go test -race.

* fix(filer.backup): harden initialSnapshot against transient errors and path edge cases

Three review items from CodeRabbit on #9126:

1. getOffset errors no longer leave isFreshSync=true. Before, a transient
   KV read failure would cause runFilerBackup's retry loop to redo the
   full -initialSnapshot walk on every retry. Treat any offset-read
   error as "not fresh" so the snapshot only runs when we've verified
   there really is no prior checkpoint.

2. initialSnapshotTargetKey now normalizes sourcePath to a trailing-
   slash base before stripping the prefix, so edge cases where
   sourceKey equals sourcePath (trailing-slash mismatch or root-entry
   emission) no longer index past the end. Unit tests cover both
   forms.

3. Documented the TraverseBfs-enumerates-excluded-subtrees performance
   characteristic on runInitialSnapshot, since pruning requires a
   separate change to TraverseBfs itself.

* fix(filer.backup): retry setOffset after initialSnapshot to avoid full re-walks

If the snapshot walk finishes but the subsequent setOffset fails, the
retry loop in runFilerBackup will re-enter doFilerBackup with an empty
checkpoint and run the full BFS again — on a multi-million-entry tree
that's hours of wasted work over a 100-byte KV write. Retry the write a
handful of times with exponential backoff before giving up, and log
loudly at the final failure (with snapshotTsNs + sinkId) so operators
recognize the symptom instead of guessing at mysterious repeated walks.

Nitpick raised by CodeRabbit on #9126.

* fix(filer.backup): initialSnapshot ignore404, skew margin, exclude dir-entry itself

Three review items from CodeRabbit on #9126:

1. ignore404Error now threads into runInitialSnapshot. If a file is listed
   by TraverseBfs and then deleted before CreateEntry reads its chunks,
   the follow path already ignores 404s — the snapshot path was aborting
   and triggering a full re-walk. Treat an ignorable 404 as "skip this
   entry, continue."

2. snapshotTsNs now uses `time.Now() - 1min` instead of `time.Now()`.
   Metadata events are stamped server-side, so a fast backup-host clock
   could skip events that fire during or right after the walk. Matches
   the 1-minute margin meta_aggregator.go applies on initial peer
   traversal; duplicate replay is harmless because CreateEntry is
   idempotent.

3. Exclude checks now run against the entry's own full path, not just
   its parent. A walked directory whose full path matches SystemLogDir
   or -filerExcludePaths was being seeded to the destination; only its
   descendants were being skipped. Verified with a manual repro where
   -filerExcludePaths=/data/skipdir now keeps the skipdir entry itself
   off the destination.

* refactor(filer): share destKey helper between buildKey and initialSnapshot

Extract destKey(dataSink, targetPath, sourcePath, sourceKey, mTime) from
buildKey in filer_sync.go. Both the event-log path (buildKey) and the
initialSnapshot walk (initialSnapshotTargetKey) now go through the same
helper, so a walk-seeded file and an event-replayed file always resolve
to the same destination key.

As a bonus, buildKey picks up the defensive trailing-slash normalization
that initialSnapshotTargetKey introduced — no more index-past-end risk
when sourceKey happens to equal sourcePath. Also tightens the mTime
lookup to guard against nil Attributes (caught by an existing test
against buildKey when I first moved the lookup out of the incremental
branch).

weed/command/filer_backup.go

@@ -1,17 +1,22 @@
 package command
 
 import (
+	"context"
 	"errors"
 	"fmt"
 	nethttp "net/http"
 	"regexp"
 	"strings"
+	"sync"
+	"sync/atomic"
 	"time"
 
+	"github.com/seaweedfs/seaweedfs/weed/filer"
 	"github.com/seaweedfs/seaweedfs/weed/glog"
 	"github.com/seaweedfs/seaweedfs/weed/pb"
 	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
 	"github.com/seaweedfs/seaweedfs/weed/replication/repl_util"
+	"github.com/seaweedfs/seaweedfs/weed/replication/sink"
 	"github.com/seaweedfs/seaweedfs/weed/replication/source"
 	"github.com/seaweedfs/seaweedfs/weed/security"
 	"github.com/seaweedfs/seaweedfs/weed/util"
@@ -35,6 +40,7 @@ type FilerBackupOptions struct {
 	ignore404Error      *bool
 	timeAgo             *time.Duration
 	retentionDays       *int
+	initialSnapshot     *bool
 }
 
 var (
@@ -57,6 +63,7 @@ func init() {
 	filerBackupOptions.retentionDays = cmdFilerBackup.Flag.Int("retentionDays", 0, "incremental backup retention days")
 	filerBackupOptions.disableErrorRetry = cmdFilerBackup.Flag.Bool("disableErrorRetry", false, "disables errors retry, only logs will print")
 	filerBackupOptions.ignore404Error = cmdFilerBackup.Flag.Bool("ignore404Error", true, "ignore 404 errors from filer")
+	filerBackupOptions.initialSnapshot = cmdFilerBackup.Flag.Bool("initialSnapshot", false, "before subscribing to metadata updates, walk the live filer tree under -filerPath and seed the destination. Only runs on a fresh sync (no prior checkpoint and -timeAgo is 0). After the walk, subscription starts from the walk-start timestamp so concurrent changes are still captured.")
 }
 
 var cmdFilerBackup = &Command{
@@ -70,6 +77,11 @@ var cmdFilerBackup = &Command{
 	If restarted and "-timeAgo" is not set, the synchronization will resume from the previous checkpoints, persisted every minute.
 	A fresh sync will start from the earliest metadata logs. To reset the checkpoints, just set "-timeAgo" to a high value.
 
+	On a fresh sync the metadata event log only re-materializes files that still exist on the source; entries that were
+	created and later deleted are replayed as a create-then-delete pair and therefore never appear on the destination.
+	Pass "-initialSnapshot" to walk the live filer tree first and seed the destination with the current tree, then
+	subscribe from the walk-start timestamp. The walk only runs when there is no prior checkpoint.
+
 `,
 }
 
@@ -123,17 +135,27 @@ func doFilerBackup(grpcDialOption grpc.DialOption, backupOption *FilerBackupOpti
 
 	// get start time for the data sink
 	startFrom := time.Unix(0, 0)
+	isFreshSync := true
 	sinkId := util.HashStringToLong(dataSink.GetName() + dataSink.GetSinkToDirectory())
 	if timeAgo.Milliseconds() == 0 {
 		lastOffsetTsNs, err := getOffset(grpcDialOption, sourceFiler, BackupKeyPrefix, int32(sinkId))
 		if err != nil {
-			glog.V(0).Infof("starting from %v", startFrom)
-		} else {
+			// A KV read failure is ambiguous — a checkpoint may well exist but the
+			// source filer is temporarily unreachable. Don't treat that as a fresh
+			// sync; otherwise runFilerBackup's retry loop would redo the full
+			// -initialSnapshot walk on every transient error.
+			isFreshSync = false
+			glog.V(0).Infof("starting from %v (offset read failed: %v)", startFrom, err)
+		} else if lastOffsetTsNs > 0 {
 			startFrom = time.Unix(0, lastOffsetTsNs)
+			isFreshSync = false
 			glog.V(0).Infof("resuming from %v", startFrom)
+		} else {
+			glog.V(0).Infof("starting from %v (no prior checkpoint)", startFrom)
 		}
 	} else {
 		startFrom = time.Now().Add(-timeAgo)
+		isFreshSync = false
 		glog.V(0).Infof("start time is set to %v", startFrom)
 	}
 
@@ -150,6 +172,28 @@ func doFilerBackup(grpcDialOption grpc.DialOption, backupOption *FilerBackupOpti
 	}
 	dataSink.SetSourceFiler(filerSource)
 
+	// When the destination has no prior checkpoint and the user opted in to an
+	// initial snapshot, walk the live filer tree first and seed the destination
+	// with the current entries. This avoids the "only new files appear" pitfall
+	// of replaying the metadata event log: entries created-then-deleted before
+	// the walk leave no trace, so a re-backup after wiping the destination
+	// reflects what is actually live on the source instead of an empty tree.
+	if *backupOption.initialSnapshot && isFreshSync {
+		snapshotTsNs, err := runInitialSnapshot(sourceFiler.ToGrpcAddress(), filerSource, sourcePath, targetPath, excludePaths, reExcludeFileName, excludeFileNames, excludePathPatterns, dataSink, *backupOption.ignore404Error)
+		if err != nil {
+			return fmt.Errorf("initial snapshot: %w", err)
+		}
+		// The walk can take hours on large trees; retry the tiny KV write a
+		// handful of times before giving up so a flaky filer KV doesn't force
+		// the whole walk to repeat on the next retry loop iteration.
+		if err := persistSnapshotOffset(grpcDialOption, sourceFiler, int32(sinkId), snapshotTsNs); err != nil {
+			glog.Errorf("initialSnapshot: FAILED to persist offset %d for sinkId %d after retries: %v — the next retry will redo the full walk", snapshotTsNs, sinkId, err)
+			return fmt.Errorf("persist initial snapshot offset: %w", err)
+		}
+		startFrom = time.Unix(0, snapshotTsNs)
+		glog.V(0).Infof("initialSnapshot done; subscribing from %v", startFrom)
+	}
+
 	var processEventFn func(*filer_pb.SubscribeMetadataResponse) error
 	if *backupOption.ignore404Error {
 		processEventFnGenerated := genProcessFunction(sourcePath, targetPath, excludePaths, reExcludeFileName, excludeFileNames, excludePathPatterns, dataSink, *backupOption.doDeleteFiles, debug)
@@ -241,3 +285,151 @@ func isIgnorable404(err error) bool {
 		strings.Contains(errStr, "LookupFileId") ||
 		(strings.Contains(errStr, "volume id") && strings.Contains(errStr, "not found"))
 }
+
+// persistSnapshotOffset writes the snapshot high-water mark to the source
+// filer's KV, retrying a few times with exponential backoff on transient
+// errors. Losing this write forces a full re-walk on the next retry loop
+// iteration, so a small retry budget here is far cheaper than paying the
+// walk cost again on a large tree.
+func persistSnapshotOffset(grpcDialOption grpc.DialOption, sourceFiler pb.ServerAddress, sinkId int32, tsNs int64) error {
+	const attempts = 4
+	backoff := 500 * time.Millisecond
+	var lastErr error
+	for i := 1; i <= attempts; i++ {
+		if err := setOffset(grpcDialOption, sourceFiler, BackupKeyPrefix, sinkId, tsNs); err == nil {
+			return nil
+		} else {
+			lastErr = err
+			if i == attempts {
+				break
+			}
+			glog.V(0).Infof("initialSnapshot: setOffset attempt %d/%d failed: %v (retrying in %v)", i, attempts, err, backoff)
+			time.Sleep(backoff)
+			backoff *= 2
+		}
+	}
+	return lastErr
+}
+
+// runInitialSnapshot walks the live filer tree under sourcePath and seeds the
+// destination via the sink. The snapshot timestamp is captured before the walk
+// starts so any create/update/delete that races with the walk is still caught
+// by the subscription that runs afterward (sink CreateEntry is idempotent for
+// all builtin sinks, so replaying a concurrent create from the subscription
+// over an entry already written by the walk is safe).
+//
+// Note on excludes: TraverseBfs enumerates every directory and enqueues its
+// children unconditionally before the callback runs, so the exclude filters
+// below only prevent *processing* of excluded entries — they do not prune the
+// listing RPCs for excluded subtrees. For small system excludes (SystemLogDir)
+// that is fine; for large user-supplied excludes (say an archive directory
+// with millions of files), the listing cost can dominate the snapshot. A
+// proper prune signal through TraverseBfs is a separate change.
+func runInitialSnapshot(
+	grpcAddress string,
+	filerSource *source.FilerSource,
+	sourcePath string,
+	targetPath string,
+	excludePaths []string,
+	reExcludeFileName *regexp.Regexp,
+	excludeFileNames []*wildcard.WildcardMatcher,
+	excludePathPatterns []*wildcard.WildcardMatcher,
+	dataSink sink.ReplicationSink,
+	ignore404Error bool,
+) (int64, error) {
+	// Metadata events are stamped server-side, so the backup host clock may be
+	// ahead of the filer's. Take `now - 1min` as the subscription watermark so
+	// a fast client clock can't skip events that fire during/right-after the
+	// walk. meta_aggregator.go uses the same 1-minute margin on initial peer
+	// traversal; see the note there on why duplicate replay is harmless.
+	snapshotTsNs := time.Now().Add(-time.Minute).UnixNano()
+	glog.V(0).Infof("initialSnapshot: walking %s on %s -> %s (snapshotTsNs=%d, -1m skew margin applied)", sourcePath, grpcAddress, targetPath, snapshotTsNs)
+
+	// TraverseBfs fans the callback out across 5 worker goroutines, so counter
+	// updates and the progress-log gate need to be safe under concurrent access.
+	var entryCount, byteCount atomic.Int64
+	start := time.Now()
+	var logMu sync.Mutex
+	lastLog := start
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	err := filer_pb.TraverseBfs(ctx, filerSource, util.FullPath(sourcePath), func(parentPath util.FullPath, entry *filer_pb.Entry) error {
+		parent := string(parentPath)
+		sourceKey := parentPath.Child(entry.Name)
+		source := string(sourceKey)
+		// Check exclusion on the entry's own path too — otherwise a walked
+		// directory whose full path is SystemLogDir or a user-excluded path
+		// still gets seeded (only its children would be skipped by the parent
+		// check).
+		if parent == filer.SystemLogDir || strings.HasPrefix(parent, filer.SystemLogDir+"/") ||
+			source == filer.SystemLogDir || strings.HasPrefix(source, filer.SystemLogDir+"/") {
+			return nil
+		}
+		if matchesExcludePath(parent, excludePaths) || matchesExcludePath(source, excludePaths) {
+			return nil
+		}
+		if isEntryExcluded(parent, entry, reExcludeFileName, excludeFileNames, excludePathPatterns) {
+			return nil
+		}
+
+		if !util.IsEqualOrUnder(source, sourcePath) {
+			return nil
+		}
+
+		targetKey := initialSnapshotTargetKey(dataSink, targetPath, sourcePath, sourceKey, entry)
+		if err := dataSink.CreateEntry(targetKey, entry, nil); err != nil {
+			// A file can be listed by TraverseBfs and then deleted before
+			// CreateEntry reads its chunks. The follow phase ignores 404s when
+			// -ignore404Error is set; apply the same policy here so the walk
+			// doesn't abort (and trigger a full re-walk on retry) just because
+			// a single entry disappeared mid-snapshot.
+			if ignore404Error && isIgnorable404(err) {
+				glog.V(0).Infof("initialSnapshot: source entry %s disappeared, ignore it: %s", sourceKey, err.Error())
+				return nil
+			}
+			return fmt.Errorf("seed %s: %w", targetKey, err)
+		}
+
+		curEntries := entryCount.Add(1)
+		var curBytes int64
+		if entry.Attributes != nil {
+			curBytes = byteCount.Add(int64(entry.Attributes.FileSize))
+		} else {
+			curBytes = byteCount.Load()
+		}
+
+		now := time.Now()
+		logMu.Lock()
+		shouldLog := now.Sub(lastLog) >= 5*time.Second
+		if shouldLog {
+			lastLog = now
+		}
+		logMu.Unlock()
+		if shouldLog {
+			elapsed := now.Sub(start).Seconds()
+			if elapsed == 0 {
+				elapsed = 1
+			}
+			glog.V(0).Infof("initialSnapshot: %d entries / %d bytes seeded (%.1f/sec)", curEntries, curBytes, float64(curEntries)/elapsed)
+		}
+		return nil
+	})
+	if err != nil {
+		return 0, err
+	}
+	glog.V(0).Infof("initialSnapshot: done — %d entries / %d bytes in %v", entryCount.Load(), byteCount.Load(), time.Since(start))
+	return snapshotTsNs, nil
+}
+
+// initialSnapshotTargetKey pulls the entry mtime and delegates to the shared
+// destKey helper in filer_sync.go so the walk and the event-log path produce
+// the same destination key for the same source entry.
+func initialSnapshotTargetKey(dataSink sink.ReplicationSink, targetPath, sourcePath string, sourceKey util.FullPath, entry *filer_pb.Entry) string {
+	var mTime int64
+	if entry.Attributes != nil {
+		mTime = entry.Attributes.Mtime
+	}
+	return destKey(dataSink, targetPath, sourcePath, sourceKey, mTime)
+}

weed/command/filer_backup_test.go

@@ -7,7 +7,12 @@ import (
 	"net/http/httptest"
 	"os"
 	"testing"
+	"time"
 
+	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
+	"github.com/seaweedfs/seaweedfs/weed/replication/sink"
+	"github.com/seaweedfs/seaweedfs/weed/replication/source"
+	"github.com/seaweedfs/seaweedfs/weed/util"
 	util_http "github.com/seaweedfs/seaweedfs/weed/util/http"
 )
 
@@ -71,3 +76,66 @@ func TestIsIgnorable404_NonIgnorableError(t *testing.T) {
 		t.Errorf("expected not ignorable, got ignorable: %v", genErr)
 	}
 }
+
+// stubSink is a minimal ReplicationSink used to exercise initialSnapshotTargetKey
+// without standing up a real sink. Only the two methods read by the key builder
+// (GetName, IsIncremental) need meaningful behavior; the rest satisfy the interface.
+type stubSink struct {
+	name          string
+	isIncremental bool
+}
+
+func (s *stubSink) GetName() string                           { return s.name }
+func (s *stubSink) Initialize(util.Configuration, string) error { return nil }
+func (s *stubSink) DeleteEntry(string, bool, bool, []int32) error {
+	return nil
+}
+func (s *stubSink) CreateEntry(string, *filer_pb.Entry, []int32) error { return nil }
+func (s *stubSink) UpdateEntry(string, *filer_pb.Entry, string, *filer_pb.Entry, bool, []int32) (bool, error) {
+	return false, nil
+}
+func (s *stubSink) GetSinkToDirectory() string       { return "" }
+func (s *stubSink) SetSourceFiler(*source.FilerSource) {}
+func (s *stubSink) IsIncremental() bool              { return s.isIncremental }
+
+var _ sink.ReplicationSink = (*stubSink)(nil)
+
+func TestInitialSnapshotTargetKey(t *testing.T) {
+	// Mirror the non-incremental path of buildKey so a refactor of one without
+	// the other will fail this test.
+	mirror := &stubSink{name: "mirror", isIncremental: false}
+	got := initialSnapshotTargetKey(mirror, "/backup", "/data", util.FullPath("/data/sub/file.txt"), &filer_pb.Entry{})
+	if got != "/backup/sub/file.txt" {
+		t.Errorf("mirror sink: got %q, want %q", got, "/backup/sub/file.txt")
+	}
+
+	// Incremental sinks partition by entry mtime, so the seed must use the same
+	// YYYY-MM-DD prefix a replayed CreateEntry would produce. buildKey in
+	// filer_sync.go formats the date in local time, so compute the expected
+	// key the same way to keep the test timezone-independent.
+	inc := &stubSink{name: "inc", isIncremental: true}
+	mtime := int64(1704196800) // 2024-01-02T12:00:00 UTC — unambiguously Jan 2 in nearly all timezones
+	gotInc := initialSnapshotTargetKey(inc, "/backup", "/data", util.FullPath("/data/sub/file.txt"), &filer_pb.Entry{
+		Attributes: &filer_pb.FuseAttributes{Mtime: mtime},
+	})
+	wantInc := "/backup/" + time.Unix(mtime, 0).Format("2006-01-02") + "/sub/file.txt"
+	if gotInc != wantInc {
+		t.Errorf("incremental sink: got %q, want %q", gotInc, wantInc)
+	}
+
+	// Trailing-slash sourcePath still produces a clean relative key.
+	gotTrail := initialSnapshotTargetKey(mirror, "/backup", "/data/", util.FullPath("/data/file.txt"), &filer_pb.Entry{})
+	if gotTrail != "/backup/file.txt" {
+		t.Errorf("trailing-slash sourcePath: got %q, want %q", gotTrail, "/backup/file.txt")
+	}
+
+	// Edge cases CodeRabbit called out: sourceKey equal to sourcePath
+	// (non-trailing and trailing variants). Real TraverseBfs walks never emit
+	// the root itself, but the helper must not panic if something else does.
+	if got := initialSnapshotTargetKey(mirror, "/backup", "/data", util.FullPath("/data"), &filer_pb.Entry{}); got != "/backup" {
+		t.Errorf("sourceKey == sourcePath (no slash): got %q, want %q", got, "/backup")
+	}
+	if got := initialSnapshotTargetKey(mirror, "/backup", "/data/", util.FullPath("/data"), &filer_pb.Entry{}); got != "/backup" {
+		t.Errorf("sourceKey == sourcePath (trailing slash mismatch): got %q, want %q", got, "/backup")
+	}
+}

weed/command/filer_sync.go

@@ -653,21 +653,39 @@ func genProcessFunction(sourcePath string, targetPath string, excludePaths []str
 	return processEventFn
 }
 
-func buildKey(dataSink sink.ReplicationSink, message *filer_pb.EventNotification, targetPath string, sourceKey util.FullPath, sourcePath string) (key string) {
-	if !dataSink.IsIncremental() {
-		key = util.Join(targetPath, string(sourceKey)[len(sourcePath):])
-	} else {
-		var mTime int64
-		if message.NewEntry != nil {
-			mTime = message.NewEntry.Attributes.Mtime
-		} else if message.OldEntry != nil {
-			mTime = message.OldEntry.Attributes.Mtime
-		}
-		dateKey := time.Unix(mTime, 0).Format("2006-01-02")
-		key = util.Join(targetPath, dateKey, string(sourceKey)[len(sourcePath):])
+func buildKey(dataSink sink.ReplicationSink, message *filer_pb.EventNotification, targetPath string, sourceKey util.FullPath, sourcePath string) string {
+	var mTime int64
+	if message.NewEntry != nil && message.NewEntry.Attributes != nil {
+		mTime = message.NewEntry.Attributes.Mtime
+	} else if message.OldEntry != nil && message.OldEntry.Attributes != nil {
+		mTime = message.OldEntry.Attributes.Mtime
 	}
+	return destKey(dataSink, targetPath, sourcePath, sourceKey, mTime)
+}
 
-	return escapeKey(key)
+// destKey derives the sink-side key for a source entry. Shared between the
+// event-log path (buildKey) and the initialSnapshot walk (both paths need the
+// same target layout so a walk-seeded file and an event-replayed file resolve
+// to the same destination key). Normalizing to a trailing-slash base avoids
+// indexing past the end of sourceKey when callers differ on trailing-slash
+// conventions or when sourceKey equals sourcePath exactly.
+func destKey(dataSink sink.ReplicationSink, targetPath, sourcePath string, sourceKey util.FullPath, mTime int64) string {
+	base := strings.TrimSuffix(sourcePath, "/") + "/"
+	sk := string(sourceKey)
+	var relative string
+	switch {
+	case strings.HasPrefix(sk, base):
+		relative = sk[len(base):]
+	case sk == strings.TrimSuffix(sourcePath, "/"):
+		relative = ""
+	default:
+		relative = strings.TrimPrefix(sk, "/")
+	}
+	if !dataSink.IsIncremental() {
+		return escapeKey(util.Join(targetPath, relative))
+	}
+	dateKey := time.Unix(mTime, 0).Format("2006-01-02")
+	return escapeKey(util.Join(targetPath, dateKey, relative))
 }
 
 // isEntryExcluded checks whether a single side (old or new) of an event is excluded