fix(filer.sync): scope filesystem key sanitization to the local sink (#9894)

* fix(filer.sync): scope filesystem key sanitization to the local sink

destKey ran every sink key through escapeKey, whose Windows build strips
colons. Colons are illegal in NTFS filenames so the local sink needs that,
but s3/filer/azure/gcs/b2 accept them as ordinary key bytes — stripping
them silently diverged the destination key (a source a:b replicated as ab).

Move the sanitization into the local sink behind a Windows build tag,
applied at every entry point so the previously-unescaped in-place-update
paths stay consistent. Non-local sinks now keep the raw key; non-Windows
builds are unchanged; a leading drive-letter colon is preserved.

* test(filer.sync): cover incremental destKey and localsink update/delete sanitization

Lock the colon-preserving behavior for the incremental destKey branch, and
extend the Windows local-sink test to assert UpdateEntry and DeleteEntry also
sanitize the key, not just CreateEntry.

weed/command/filer_sync_process_test.go

@@ -1,6 +1,7 @@
 package command
 
 import (
+	"strings"
 	"testing"
 
 	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
@@ -12,9 +13,10 @@ import (
 var _ sink.ReplicationSink = (*recordingSyncSink)(nil)
 
 type recordingSyncSink struct {
-	deleteKeys []string
-	createKeys []string
-	updateKeys []string
+	deleteKeys  []string
+	createKeys  []string
+	updateKeys  []string
+	incremental bool
 }
 
 func (s *recordingSyncSink) GetName() string { return "recording" }
@@ -35,7 +37,27 @@ func (s *recordingSyncSink) UpdateEntry(key string, oldEntry *filer_pb.Entry, ne
 }
 func (s *recordingSyncSink) GetSinkToDirectory() string         { return "/dest" }
 func (s *recordingSyncSink) SetSourceFiler(*source.FilerSource) {}
-func (s *recordingSyncSink) IsIncremental() bool                { return false }
+func (s *recordingSyncSink) IsIncremental() bool                { return s.incremental }
+
+// TestDestKeyPreservesColonForNonLocalSink guards against the command layer
+// stripping colons from keys destined for non-local sinks. Colon
+// sanitization belongs to the local filesystem sink only.
+func TestDestKeyPreservesColonForNonLocalSink(t *testing.T) {
+	t.Run("non-incremental", func(t *testing.T) {
+		got := destKey(&recordingSyncSink{}, "/backup", "/src", util.FullPath("/src/2024:01:02/file:name.txt"), 0)
+		want := "/backup/2024:01:02/file:name.txt"
+		if got != want {
+			t.Errorf("destKey() = %q, want %q", got, want)
+		}
+	})
+	t.Run("incremental", func(t *testing.T) {
+		got := destKey(&recordingSyncSink{incremental: true}, "/backup", "/src", util.FullPath("/src/file:name.txt"), 0)
+		// the date partition varies by timezone; assert the colon-bearing name survives.
+		if !strings.HasPrefix(got, "/backup/") || !strings.HasSuffix(got, "/file:name.txt") {
+			t.Errorf("destKey() = %q, want /backup/<date>/file:name.txt with colon preserved", got)
+		}
+	})
+}
 
 func TestPathIsEqualOrUnderUsesDirectoryBoundaries(t *testing.T) {
 	tests := []struct {