diff --git a/weed/command/filer_remote_sync_dir.go b/weed/command/filer_remote_sync_dir.go index 1942950f08..5213412598 100644 --- a/weed/command/filer_remote_sync_dir.go +++ b/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 diff --git a/weed/command/filer_remote_sync_dir_test.go b/weed/command/filer_remote_sync_dir_test.go index 5b004bd068..911ccd6033 100644 --- a/weed/command/filer_remote_sync_dir_test.go +++ b/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 }