filer.remote.sync: skip a holed snapshot even while the filer still holds it

A mid-write snapshot the writer never finished is neither deleted nor
written past: the filer holds it unchanged, so the superseded check never
fires and every retry reads short at the same gap, pinning the sync offset
forever. The upload can never succeed for that entry — skip it like a
superseded one. A finishing write logs its own event and uploads the full
content; a dead write has nothing readable to lose.

The check also covers a chunkless entry that is not a dropped local copy
(e.g. a truncated remote-only entry): its reader yields zero bytes against
a nonzero size, which can silently clobber the remote object with an empty
body where the backend does not verify the length.

weed/command/filer_remote_sync_dir.go

@@ -2,11 +2,13 @@ package command
 
 import (
 	"bytes"
+	"cmp"
 	"context"
 	"errors"
 	"fmt"
 	"io"
 	"os"
+	"slices"
 	"strings"
 	"time"
 
@@ -973,9 +975,43 @@ func isSuperseded(filerClient filer_pb.FilerClient, dir string, entry *filer_pb.
 		len(filer.DoMinusChunks(current.GetChunks(), entry.GetChunks())) > 0
 }
 
-// errSuperseded marks an upload that failed for an entry the filer has since
-// moved past (isSuperseded). The caller skips the event instead of failing it.
-var errSuperseded = errors.New("deleted or rewritten since the event was logged")
+// hasReadHole reports an entry whose chunks do not cover its declared size: a
+// snapshot of a write still in progress, flushed while chunk uploads were
+// still running. The stream reader stops at the gap, so the upload fails on
+// every retry whether or not the filer ever moves past the entry — a write
+// that never finishes leaves the gap and wedges the event. Skipping is safe
+// either way: a finishing write logs its own event and uploads the full
+// content, and a dead one has nothing readable to lose.
+func hasReadHole(entry *filer_pb.Entry) bool {
+	size := int64(entry.GetAttributes().GetFileSize())
+	if size == 0 || len(entry.Content) > 0 {
+		return false
+	}
+	chunks := slices.Clone(entry.GetChunks())
+	slices.SortFunc(chunks, func(a, b *filer_pb.FileChunk) int {
+		return cmp.Compare(a.GetOffset(), b.GetOffset())
+	})
+	var next int64
+	for _, chunk := range chunks {
+		if chunk.GetOffset() >= size {
+			break
+		}
+		if chunk.GetOffset() > next {
+			return true
+		}
+		if end := chunk.GetOffset() + int64(chunk.GetSize()); end > next {
+			next = end
+		}
+	}
+	return next < size
+}
+
+// errSuperseded marks an upload that cannot produce the entry an event
+// described: the filer moved past it (isSuperseded), or the entry is a
+// mid-write snapshot with a gap its chunks do not cover (hasReadHole). The
+// caller skips the event instead of failing it; a write completing logs a
+// later event that carries the full content.
+var errSuperseded = errors.New("superseded or incomplete since the event was logged")
 
 // retriedWriteFile uploads the entry, retrying transient failures. The filer is
 // asked whether the entry is superseded before the first attempt and after
@@ -986,6 +1022,9 @@ var errSuperseded = errors.New("deleted or rewritten since the event was logged"
 // still holds is worth sending. After: a dead chunk reads as a transient
 // "RequestError" from the SDK, and waiting out the backoff on it buys nothing.
 func retriedWriteFile(client remote_storage.RemoteStorageClient, filerSource filer_pb.FilerClient, dir string, newEntry *filer_pb.Entry, dest *remote_pb.RemoteStorageLocation) (remoteEntry *filer_pb.RemoteEntry, err error) {
+	if hasReadHole(newEntry) {
+		return nil, fmt.Errorf("%s %w", util.NewFullPath(dir, newEntry.Name), errSuperseded)
+	}
 	if isSuperseded(filerSource, dir, newEntry) {
 		return nil, fmt.Errorf("%s %w", util.NewFullPath(dir, newEntry.Name), errSuperseded)
 	}

weed/command/filer_remote_sync_dir_test.go

@@ -688,6 +688,23 @@ func TestRetriedWriteFileStopsWhenSuperseded(t *testing.T) {
 		}
 	})
 
+	t.Run("mid-write snapshot whose hole never fills, no write", func(t *testing.T) {
+		snapshot := entryWith("IMG_6526.mov", nil, chunkAt("222,01", "e1", 0), chunkAt("220,03", "e3", 2048))
+		snapshot.Attributes.FileSize = 3072
+		remote := &failingRemote{err: errors.New("RequestError: send request failed")}
+		filerClient := &stubFilerClient{entry: snapshot}
+		_, err := retriedWriteFile(remote, filerClient, dir, snapshot, dest)
+		if !errors.Is(err, errSuperseded) {
+			t.Errorf("err = %v, want errSuperseded", err)
+		}
+		if remote.writes != 0 {
+			t.Errorf("wrote %d times, want 0: a snapshot with a hole reads short on every retry", remote.writes)
+		}
+		if filerClient.lookups != 0 {
+			t.Errorf("looked up the filer %d times, want 0: no state makes this uploadable", filerClient.lookups)
+		}
+	})
+
 	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