filer.remote.sync: skip an event the filer has written past, not only rewritten (#12176)

isSuperseded only looked for chunks the event named that the filer no
longer holds. A file copied in through a mount logs a create while the
copy is still running: the mount flushes metadata with chunk uploads in
flight, so that chunk list can have holes. ChunkStreamReader stops at
the first hole, the multipart upload sends a short part ("ContentLength=
8388608 with Body length 4194304"), and since every chunk the event
names is still live the event was never superseded. The daemon retried
it forever and the offset never moved past it.

Treat chunks the filer has added since the event the same way. The
event that added them follows in the log and uploads the finished file.

weed/command/filer_remote_sync_dir.go

@@ -936,18 +936,22 @@ func readRemoteStorageConf(filerClient filer_pb.FilerClient, storageName string)
 }
 
 // isSuperseded reports whether the filer has moved past the entry an event
-// described: it is deleted, or it no longer references every chunk the event
-// named. Those are the chunks the filer deletes when an entry is updated, so
-// they are gone from the volume servers, or about to be, and no retry of the
-// upload can succeed. Chunks are compared by file id, not content: a rewrite
-// stores even identical bytes under new ids and drops the old ones. A replay
-// from an earlier offset (-timeAgo) re-emits such events. Failing one holds the
-// sync offset before it, so every restart of the subscription replays it into
-// the same dead chunks, and progress on everything after it in the log is never
-// persisted. Skipping is safe: the event that superseded this one follows in
-// the log, and the delete removes the remote object or the rewrite uploads the
-// current content. A lookup that fails for any other reason keeps the write
-// failure, so the event is retried.
+// described: it is deleted, it no longer references every chunk the event
+// named, or it references chunks the event did not. Dropped chunks are the ones
+// the filer deletes when an entry is updated, so they are gone from the volume
+// servers, or about to be, and no retry of the upload can succeed. Added chunks
+// mean the event was a snapshot of a write still in progress. A mount flushes
+// metadata while it is still uploading chunks in parallel, so such a snapshot
+// can have holes, and the stream reader stops at the first one: every retry
+// sends a short body and fails the same way. Chunks are compared by file id,
+// not content: a rewrite stores even identical bytes under new ids and drops
+// the old ones. A replay from an earlier offset (-timeAgo) re-emits such
+// events. Failing one holds the sync offset before it, so every restart of the
+// subscription replays it into the same failure, and progress on everything
+// after it in the log is never persisted. Skipping is safe: the event that
+// superseded this one follows in the log, and the delete removes the remote
+// object or the later write uploads the current content. A lookup that fails
+// for any other reason keeps the write failure, so the event is retried.
 func isSuperseded(filerClient filer_pb.FilerClient, dir string, entry *filer_pb.Entry) bool {
 	current, _, _, err := filer_pb.GetEntry(context.Background(), filerClient, util.NewFullPath(dir, entry.Name))
 	if errors.Is(err, filer_pb.ErrNotFound) {
@@ -965,7 +969,8 @@ func isSuperseded(filerClient filer_pb.FilerClient, dir string, entry *filer_pb.
 	if len(entry.Content) > 0 || len(current.Content) > 0 {
 		return !bytes.Equal(entry.Content, current.Content)
 	}
-	return len(filer.DoMinusChunks(entry.GetChunks(), current.GetChunks())) > 0
+	return len(filer.DoMinusChunks(entry.GetChunks(), current.GetChunks())) > 0 ||
+		len(filer.DoMinusChunks(current.GetChunks(), entry.GetChunks())) > 0
 }
 
 // errSuperseded marks an upload that failed for an entry the filer has since

weed/command/filer_remote_sync_dir_test.go

@@ -517,6 +517,10 @@ func chunk(fileId, etag string) *filer_pb.FileChunk {
 	return &filer_pb.FileChunk{FileId: fileId, Size: 1024, ETag: etag}
 }
 
+func chunkAt(fileId, etag string, offset int64) *filer_pb.FileChunk {
+	return &filer_pb.FileChunk{FileId: fileId, Offset: offset, Size: 1024, ETag: etag}
+}
+
 func entryWith(name string, remote *filer_pb.RemoteEntry, chunks ...*filer_pb.FileChunk) *filer_pb.Entry {
 	return &filer_pb.Entry{Name: name, Attributes: &filer_pb.FuseAttributes{Mtime: 1786096669}, RemoteEntry: remote, Chunks: chunks}
 }
@@ -532,6 +536,11 @@ func entryWith(name string, remote *filer_pb.RemoteEntry, chunks ...*filer_pb.Fi
 // by chunk identity, the way the filer itself decides which chunks an update
 // leaves for deletion. Content fingerprints would miss a delete-and-recreate
 // of identical bytes and fail the event forever on its dead chunks.
+//
+// Chunks the filer has added since count too. The event was then a snapshot
+// of a write in progress: a mount flushes metadata while chunk uploads are
+// still running, so the snapshot can have holes, and the stream reader stops
+// at the first one on every retry.
 func TestIsSuperseded(t *testing.T) {
 	const dir = "/buckets/ingest"
 	event := entryWith("tmpjt9req69__Alan_Doe_Resume-1.pdf", nil, chunk("3,01", "e1"), chunk("3,02", "e2"))
@@ -549,7 +558,7 @@ func TestIsSuperseded(t *testing.T) {
 		{name: "truncated since", filer: &stubFilerClient{entry: entryWith(event.Name, nil, chunk("3,01", "e1"))}, want: true},
 		{name: "still as described", filer: &stubFilerClient{entry: entryWith(event.Name, nil, chunk("3,01", "e1"), chunk("3,02", "e2"))}, want: false},
 		{name: "still as described, since replicated", filer: &stubFilerClient{entry: entryWith(event.Name, stamped, chunk("3,01", "e1"), chunk("3,02", "e2"))}, want: false},
-		{name: "appended since, chunks still live", filer: &stubFilerClient{entry: entryWith(event.Name, nil, chunk("3,01", "e1"), chunk("3,02", "e2"), chunk("3,03", "e3"))}, want: false},
+		{name: "appended since", filer: &stubFilerClient{entry: entryWith(event.Name, nil, chunk("3,01", "e1"), chunk("3,02", "e2"), chunk("3,03", "e3"))}, want: true},
 		{name: "filer lookup failed", filer: &stubFilerClient{err: errors.New("rpc error: code = Unavailable")}, want: false},
 	}
 	for _, tt := range tests {
@@ -580,6 +589,17 @@ func TestIsSuperseded(t *testing.T) {
 		}
 	})
 
+	t.Run("mid-write snapshot with a hole, filled in since", func(t *testing.T) {
+		snapshot := entryWith("IMG_6526.mov", nil, chunkAt("222,01", "e1", 0), chunkAt("220,03", "e3", 2048))
+		filled := entryWith("IMG_6526.mov", nil, chunkAt("222,01", "e1", 0), chunkAt("216,02", "e2", 1024), chunkAt("220,03", "e3", 2048))
+		if !isSuperseded(&stubFilerClient{entry: filled}, dir, snapshot) {
+			t.Error("isSuperseded = false, want true: the filer holds the finished write, the snapshot can never be read past its hole")
+		}
+		if isSuperseded(&stubFilerClient{entry: filled}, dir, filled) {
+			t.Error("isSuperseded = true, want false for the event that describes the finished write")
+		}
+	})
+
 	t.Run("inline content rewritten since", func(t *testing.T) {
 		event := &filer_pb.Entry{Name: "note.txt", Content: []byte("v1")}
 		if !isSuperseded(&stubFilerClient{entry: &filer_pb.Entry{Name: "note.txt", Content: []byte("v2")}}, dir, event) {
@@ -652,6 +672,22 @@ func TestRetriedWriteFileStopsWhenSuperseded(t *testing.T) {
 		}
 	})
 
+	t.Run("mid-write snapshot filled in during the attempt, one write", func(t *testing.T) {
+		snapshot := entryWith("IMG_6526.mov", nil, chunkAt("222,01", "e1", 0), chunkAt("220,03", "e3", 2048))
+		filled := entryWith("IMG_6526.mov", nil, chunkAt("222,01", "e1", 0), chunkAt("216,02", "e2", 1024), chunkAt("220,03", "e3", 2048))
+		// What the S3 SDK reports when the stream reader stops at the hole.
+		shortBody := errors.New("RequestError: send request failed\ncaused by: Put \"https://s3.example.com/tier/IMG_6526.mov?partNumber=12\": net/http: HTTP/1.x transport connection broken: http: ContentLength=8388608 with Body length 4194304")
+		remote := &failingRemote{err: shortBody}
+		filerClient := &supersedingFilerClient{live: snapshot, then: filled}
+		_, err := retriedWriteFile(remote, filerClient, dir, snapshot, dest)
+		if !errors.Is(err, errSuperseded) {
+			t.Errorf("err = %v, want errSuperseded", err)
+		}
+		if remote.writes != 1 {
+			t.Errorf("wrote %d times, want 1: the snapshot reads short on every retry", remote.writes)
+		}
+	})
+
 	t.Run("still as described, transient failure keeps retrying", func(t *testing.T) {
 		prev := util.RetryWaitTime
 		util.RetryWaitTime = 1600 * time.Millisecond // two attempts: 1s, then 1.5s
@@ -993,6 +1029,7 @@ func TestIsFailedPrecondition(t *testing.T) {
 type supersedingFilerClient struct {
 	filer_pb.SeaweedFilerClient
 	live    *filer_pb.Entry
+	then    *filer_pb.Entry // returned after the first lookup; nil means deleted
 	lookups int
 }
 
@@ -1001,6 +1038,9 @@ func (c *supersedingFilerClient) LookupDirectoryEntry(context.Context, *filer_pb
 	if c.lookups == 1 {
 		return &filer_pb.LookupDirectoryEntryResponse{Entry: c.live}, nil
 	}
+	if c.then != nil {
+		return &filer_pb.LookupDirectoryEntryResponse{Entry: c.then}, nil
+	}
 	return nil, filer_pb.ErrNotFound
 }