diff --git a/weed/replication/sink/filersink/fetch_write.go b/weed/replication/sink/filersink/fetch_write.go index 47b381cbc8..1dbc12b4a5 100644 --- a/weed/replication/sink/filersink/fetch_write.go +++ b/weed/replication/sink/filersink/fetch_write.go @@ -104,7 +104,7 @@ func (g *missingSourceChunkGate) wrap(err error) error { // topLevelId anchors nested chunks to the entry-level chunk that contains them: // supersession is judged by whether the live entry still lists that chunk, since // a manifest's id is the identity of its contents. -func (fs *FilerSink) replicateChunks(ctx context.Context, sourceChunks []*filer_pb.FileChunk, path string, sourceMtimeNs int64, topLevelId string) (replicatedChunks []*filer_pb.FileChunk, err error) { +func (fs *FilerSink) replicateChunks(ctx context.Context, sourceChunks []*filer_pb.FileChunk, path string, replayed *filer_pb.Entry, topLevelId string) (replicatedChunks []*filer_pb.FileChunk, err error) { if len(sourceChunks) == 0 { return } @@ -155,7 +155,7 @@ func (fs *FilerSink) replicateChunks(ctx context.Context, sourceChunks []*filer_ if anchor == "" { anchor = sourceChunk.GetFileIdString() } - replicatedChunk, replicateErr := fs.replicateOneManifestChunk(ctx, sourceChunk, path, sourceMtimeNs, anchor) + replicatedChunk, replicateErr := fs.replicateOneManifestChunk(ctx, sourceChunk, path, replayed, anchor) if replicateErr != nil { setError(replicateErr) break @@ -173,7 +173,7 @@ func (fs *FilerSink) replicateChunks(ctx context.Context, sourceChunks []*filer_ defer wg.Done() var replicatedChunk *filer_pb.FileChunk retryErr := util.Retry("replicate chunks", func() error { - chunk, e := fs.replicateOneChunk(source, path, sourceMtimeNs, topLevelId) + chunk, e := fs.replicateOneChunk(source, path, replayed, topLevelId) if e != nil { return e } @@ -196,9 +196,9 @@ func (fs *FilerSink) replicateChunks(ctx context.Context, sourceChunks []*filer_ return } -func (fs *FilerSink) replicateOneChunk(sourceChunk *filer_pb.FileChunk, path string, sourceMtimeNs int64, topLevelId string) (*filer_pb.FileChunk, error) { +func (fs *FilerSink) replicateOneChunk(sourceChunk *filer_pb.FileChunk, path string, replayed *filer_pb.Entry, topLevelId string) (*filer_pb.FileChunk, error) { - fileId, err := fs.fetchAndWrite(sourceChunk, path, sourceMtimeNs, topLevelId) + fileId, err := fs.fetchAndWrite(sourceChunk, path, replayed, topLevelId) if err != nil { return nil, fmt.Errorf("copy %s: %w", sourceChunk.GetFileIdString(), err) } @@ -217,7 +217,7 @@ func (fs *FilerSink) replicateOneChunk(sourceChunk *filer_pb.FileChunk, path str }, nil } -func (fs *FilerSink) replicateOneManifestChunk(ctx context.Context, sourceChunk *filer_pb.FileChunk, path string, sourceMtimeNs int64, topLevelId string) (*filer_pb.FileChunk, error) { +func (fs *FilerSink) replicateOneManifestChunk(ctx context.Context, sourceChunk *filer_pb.FileChunk, path string, replayed *filer_pb.Entry, topLevelId string) (*filer_pb.FileChunk, error) { // LookupFileId can transiently fail during a write burst (volume not yet // registered). Retry like the data-chunk path does; the gate stops on // supersession or, when that cannot be checked, after a few attempts. @@ -231,12 +231,12 @@ func (fs *FilerSink) replicateOneManifestChunk(ctx context.Context, sourceChunk } resolvedChunks = rc return nil - }, fs.manifestResolveRetryGate(path, sourceMtimeNs, topLevelId, missingGate)) + }, fs.manifestResolveRetryGate(path, replayed, topLevelId, missingGate)) if err != nil { return nil, fmt.Errorf("resolve manifest %s: %w", sourceChunk.GetFileIdString(), missingGate.wrap(err)) } - replicatedResolvedChunks, err := fs.replicateChunks(ctx, resolvedChunks, path, sourceMtimeNs, topLevelId) + replicatedResolvedChunks, err := fs.replicateChunks(ctx, resolvedChunks, path, replayed, topLevelId) if err != nil { return nil, fmt.Errorf("replicate manifest data chunks %s: %w", sourceChunk.GetFileIdString(), err) } @@ -253,7 +253,7 @@ func (fs *FilerSink) replicateOneManifestChunk(ctx context.Context, sourceChunk return nil, fmt.Errorf("marshal manifest %s: %w", sourceChunk.GetFileIdString(), err) } - manifestFileId, err := fs.uploadManifestChunk(path, sourceMtimeNs, sourceChunk.GetFileIdString(), manifestData) + manifestFileId, err := fs.uploadManifestChunk(path, replayed, sourceChunk.GetFileIdString(), manifestData) if err != nil { return nil, err } @@ -282,11 +282,11 @@ const maxUnverifiableResolveAttempts = 3 // dated target keys don't map back to a source path) — propagating lets // filer.backup decide with the event's real source key instead of spinning // here forever. -func (fs *FilerSink) manifestResolveRetryGate(path string, sourceMtimeNs int64, chunkName string, missingGate *missingSourceChunkGate) func(error) bool { +func (fs *FilerSink) manifestResolveRetryGate(path string, replayed *filer_pb.Entry, chunkName string, missingGate *missingSourceChunkGate) func(error) bool { _, canCheckSupersession := fs.targetPathToSourcePath(path) attempts := 0 return func(resolveErr error) (shouldContinue bool) { - if fs.hasSourceNewerVersion(path, sourceMtimeNs) { + if fs.sourceMovedOn(path, replayed) { glog.V(1).Infof("skip retrying stale source manifest %s for %s: %v", chunkName, path, resolveErr) return false } @@ -295,7 +295,7 @@ func (fs *FilerSink) manifestResolveRetryGate(path string, sourceMtimeNs int64, chunkName, path, missingSourceChunkGrace, resolveErr) return false } - if isSourceChunkMissing(resolveErr) && fs.sourceDroppedChunk(path, chunkName) { + if isSourceChunkMissing(resolveErr) && fs.sourceDroppedChunk(path, chunkName, replayed) { glog.V(0).Infof("source no longer lists manifest %s for %s, skipping the obsolete version: %v", chunkName, path, resolveErr) return false @@ -330,7 +330,7 @@ func isTransientResolveError(err error) bool { (strings.Contains(errStr, "volume id") && strings.Contains(errStr, "not found")) } -func (fs *FilerSink) uploadManifestChunk(path string, sourceMtimeNs int64, sourceFileId string, manifestData []byte) (fileId string, err error) { +func (fs *FilerSink) uploadManifestChunk(path string, replayed *filer_pb.Entry, sourceFileId string, manifestData []byte) (fileId string, err error) { uploader, err := fs.getUploader() if err != nil { glog.V(0).Infof("upload manifest data %v: %v", sourceFileId, err) @@ -375,7 +375,7 @@ func (fs *FilerSink) uploadManifestChunk(path string, sourceMtimeNs int64, sourc fileId = currentFileId return nil }, func(uploadErr error) (shouldContinue bool) { - if fs.hasSourceNewerVersion(path, sourceMtimeNs) { + if fs.sourceMovedOn(path, replayed) { glog.V(1).Infof("skip retrying stale source manifest %s for %s: %v", sourceFileId, path, uploadErr) return false } @@ -389,7 +389,7 @@ func (fs *FilerSink) uploadManifestChunk(path string, sourceMtimeNs int64, sourc return fileId, nil } -func (fs *FilerSink) fetchAndWrite(sourceChunk *filer_pb.FileChunk, path string, sourceMtimeNs int64, topLevelId string) (fileId string, err error) { +func (fs *FilerSink) fetchAndWrite(sourceChunk *filer_pb.FileChunk, path string, replayed *filer_pb.Entry, topLevelId string) (fileId string, err error) { uploader, err := fs.getUploader() if err != nil { glog.V(0).Infof("upload source data %v: %v", sourceChunk.GetFileIdString(), err) @@ -505,7 +505,7 @@ func (fs *FilerSink) fetchAndWrite(sourceChunk *filer_pb.FileChunk, path string, transferStatus.mu.Unlock() return false } - if fs.hasSourceNewerVersion(path, sourceMtimeNs) { + if fs.sourceMovedOn(path, replayed) { glog.V(1).Infof("skip retrying stale source %s for %s: %v", sourceChunk.GetFileIdString(), path, retryErr) return false } @@ -522,7 +522,7 @@ func (fs *FilerSink) fetchAndWrite(sourceChunk *filer_pb.FileChunk, path string, if anchorId == "" { anchorId = sourceChunk.GetFileIdString() } - if fs.sourceDroppedChunk(path, anchorId) { + if fs.sourceDroppedChunk(path, anchorId, replayed) { glog.V(0).Infof("source no longer lists %s for %s, skipping the obsolete version: %v", anchorId, path, retryErr) return false @@ -653,32 +653,19 @@ func (fs *FilerSink) sourceStillServesChunks() bool { return true } -// hasSourceNewerVersion reports whether the source's current entry for targetPath -// has moved past sourceMtimeNs — gone, or a strictly-newer mtime — meaning the -// version being replayed is stale. The lookup runs regardless of sourceMtimeNs so -// a deleted source is detected even when the replayed mtime is epoch/unset. -func (fs *FilerSink) hasSourceNewerVersion(targetPath string, sourceMtimeNs int64) bool { - sourcePath, ok := fs.targetPathToSourcePath(targetPath) - if !ok { - return false - } - ctx, cancel := context.WithTimeout(context.Background(), sourceLookupTimeout) - defer cancel() - return SourceSupersedes(ctx, fs.filerSource, sourcePath, sourceMtimeNs) -} - // sourceDroppedChunk reports whether the live source entry still exists but no // longer references fileId — a version rewritten in place, which the mtime // supersession check cannot see. The live entry's mtime does not matter: a // client that preserves timestamps stamps a rewrite with an older mtime, and -// the chunk it stopped referencing is just as gone. -func (fs *FilerSink) sourceDroppedChunk(targetPath string, fileId string) bool { +// the chunk it stopped referencing is just as gone. A live entry committed +// before the replayed version lags the event and proves nothing. +func (fs *FilerSink) sourceDroppedChunk(targetPath string, fileId string, replayed *filer_pb.Entry) bool { sourcePath, ok := fs.targetPathToSourcePath(targetPath) if !ok { return false } live, err := fs.sourceEntry(sourcePath) - if err != nil || live == nil { + if err != nil || live == nil || committedBefore(live, replayed) { return false } return !chunkFileIds(live.GetChunks())[fileId] diff --git a/weed/replication/sink/filersink/fetch_write_test.go b/weed/replication/sink/filersink/fetch_write_test.go index 75dc53c061..f9e4f11d6a 100644 --- a/weed/replication/sink/filersink/fetch_write_test.go +++ b/weed/replication/sink/filersink/fetch_write_test.go @@ -233,7 +233,7 @@ func TestFetchAndWriteRejectsZeroByteSource(t *testing.T) { err error }, 1) go func() { - gotFileId, gotErr := fs.fetchAndWrite(sourceChunk, "/dst/index.bin", 0, "") + gotFileId, gotErr := fs.fetchAndWrite(sourceChunk, "/dst/index.bin", nil, "") done <- struct { fileId string err error @@ -331,7 +331,7 @@ func TestReplicateChunksPreservesSizeMismatchSentinel(t *testing.T) { sourceChunks := []*filer_pb.FileChunk{{FileId: fid, Size: expectedSize}} - _, err := fs.replicateChunks(nil, sourceChunks, "/dst/index.bin", 0, "") + _, err := fs.replicateChunks(nil, sourceChunks, "/dst/index.bin", nil, "") if err == nil { t.Fatal("expected error from replicateChunks, got nil") } @@ -396,7 +396,7 @@ func TestSourceSupersedesEpochMtime(t *testing.T) { // attempts and propagate instead of spinning forever. func TestManifestResolveRetryGateUnverifiableSupersessionBounded(t *testing.T) { fs := &FilerSink{isIncremental: true, dir: "/backup"} - gate := fs.manifestResolveRetryGate("/backup/2026-07-10/buckets/x/f.pt", 123, "3,01abc", fs.newMissingSourceChunkGate("3,01abc")) + gate := fs.manifestResolveRetryGate("/backup/2026-07-10/buckets/x/f.pt", entryWithMtimeNs(123), "3,01abc", fs.newMissingSourceChunkGate("3,01abc")) resolveErr := errors.New("LookupFileId volume id 3: not found") for i := 1; i < maxUnverifiableResolveAttempts; i++ { if !gate(resolveErr) { @@ -414,7 +414,7 @@ func TestManifestResolveRetryGateUnverifiableSupersessionBounded(t *testing.T) { // retrying until the source is superseded. func TestManifestResolveRetryGateNonTransientPropagates(t *testing.T) { fs := &FilerSink{dir: "/backup"} - gate := fs.manifestResolveRetryGate("/backup/buckets/x/f.pt", 123, "3,01abc", fs.newMissingSourceChunkGate("3,01abc")) + gate := fs.manifestResolveRetryGate("/backup/buckets/x/f.pt", entryWithMtimeNs(123), "3,01abc", fs.newMissingSourceChunkGate("3,01abc")) permanentErrs := []error{ errors.New("fail to unmarshal manifest 3,01abc: proto: cannot parse invalid wire-format data"), errors.New("invalid fileId abc"), @@ -616,7 +616,7 @@ func TestFetchAndWriteStopsOnMissingSourceChunk(t *testing.T) { done := make(chan error, 1) go func() { - _, fetchErr := fs.fetchAndWrite(&filer_pb.FileChunk{FileId: "5617,01abc", Size: 10}, "/dst/x.bin", 5*int64(time.Second), "") + _, fetchErr := fs.fetchAndWrite(&filer_pb.FileChunk{FileId: "5617,01abc", Size: 10}, "/dst/x.bin", live, "") done <- fetchErr }() @@ -800,7 +800,7 @@ func TestFetchAndWriteMissingSourceChunkUnverifiableSupersession(t *testing.T) { done := make(chan error, 1) go func() { _, fetchErr := fs.fetchAndWrite(&filer_pb.FileChunk{FileId: "5617,01abc", Size: 10}, - "/backup/2026-07-10/x.bin", 5*int64(time.Second), "") + "/backup/2026-07-10/x.bin", entryWithMtimeNs(5*int64(time.Second)), "") done <- fetchErr }() @@ -823,17 +823,34 @@ func TestSourceDroppedChunk(t *testing.T) { Chunks: []*filer_pb.FileChunk{{FileId: "5617,02def", Size: 10}}, } fs := &FilerSink{filerSource: startSourceFilerWithEntry(t, live, ""), dir: "/dst"} + replayed := entryWithMtimeNs(5 * int64(time.Second)) - if !fs.sourceDroppedChunk("/dst/x.bin", "5617,01abc") { + if !fs.sourceDroppedChunk("/dst/x.bin", "5617,01abc", replayed) { t.Fatal("live entry no longer referencing the chunk must count as dropped") } - if fs.sourceDroppedChunk("/dst/x.bin", "5617,02def") { + if fs.sourceDroppedChunk("/dst/x.bin", "5617,02def", replayed) { t.Fatal("a chunk the live entry still lists must not count as dropped") } live.Attributes.Mtime = 4 - if !fs.sourceDroppedChunk("/dst/x.bin", "5617,01abc") { + if !fs.sourceDroppedChunk("/dst/x.bin", "5617,01abc", replayed) { t.Fatal("a live entry stamped older than the replayed version still dropped the chunk") } + + replayed.CommitTsNs, live.CommitTsNs = 100, 200 + if !fs.sourceDroppedChunk("/dst/x.bin", "5617,01abc", replayed) { + t.Fatal("a live entry committed after the replay dropped the chunk") + } + live.CommitTsNs, live.Attributes.Mtime = 50, 6 + if fs.sourceDroppedChunk("/dst/x.bin", "5617,01abc", replayed) { + t.Fatal("a live entry committed before the replay lags the event and cannot prove the chunk dropped") + } +} + +func entryWithMtimeNs(mtimeNs int64) *filer_pb.Entry { + return &filer_pb.Entry{Attributes: &filer_pb.FuseAttributes{ + Mtime: mtimeNs / int64(time.Second), + MtimeNs: int32(mtimeNs % int64(time.Second)), + }} } func TestSourceMovedOn(t *testing.T) { @@ -937,7 +954,7 @@ func TestFetchAndWriteStopsWhenSourceDroppedChunk(t *testing.T) { done := make(chan error, 1) go func() { _, fetchErr := fs.fetchAndWrite(&filer_pb.FileChunk{FileId: "5617,01abc", Size: 10}, - "/dst/x.bin", 5*int64(time.Second), "") + "/dst/x.bin", entryWithMtimeNs(5*int64(time.Second)), "") done <- fetchErr }() @@ -984,7 +1001,7 @@ func TestFetchAndWriteStopsWhenOlderLiveEntryDroppedChunk(t *testing.T) { done := make(chan error, 1) go func() { _, fetchErr := fs.fetchAndWrite(&filer_pb.FileChunk{FileId: "5617,01abc", Size: 10}, - "/dst/x.bin", 5*int64(time.Second), "") + "/dst/x.bin", entryWithMtimeNs(5*int64(time.Second)), "") done <- fetchErr }() @@ -1035,7 +1052,7 @@ func TestFetchAndWriteManifestChildKeepsRetrying(t *testing.T) { done := make(chan error, 1) go func() { _, fetchErr := fs.fetchAndWrite(&filer_pb.FileChunk{FileId: "5617,01abc", Size: 10}, - "/dst/x.bin", 5*int64(time.Second), "5617,09fed") + "/dst/x.bin", live, "5617,09fed") done <- fetchErr }() @@ -1049,6 +1066,165 @@ func TestFetchAndWriteManifestChildKeepsRetrying(t *testing.T) { } } +// A transient read error must not keep retrying a version the source has +// since committed over, even when the later version carries an older mtime. +func TestFetchAndWriteStopsOnTransientErrorWhenCommittedOver(t *testing.T) { + refused, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("listen: %v", err) + } + refusedAddr := refused.Addr().String() + refused.Close() + + live, replayed := committedOverWithOlderMtime() + fs := &FilerSink{ + filerSource: startSourceFilerWithEntry(t, live, refusedAddr, "5617"), + dir: "/dst", + executor: util.NewLimitedConcurrentExecutor(1), + } + fs.SetUploader(operation.NewUploaderWithHttpClient(http.DefaultClient)) + + done := make(chan error, 1) + go func() { + _, fetchErr := fs.fetchAndWrite(replayed.Chunks[0], "/dst/x.bin", replayed, "") + done <- fetchErr + }() + + select { + case fetchErr := <-done: + if fetchErr == nil { + t.Fatal("a read against a refused volume server cannot succeed") + } + case <-time.After(5 * time.Second): + t.Fatal("fetchAndWrite kept retrying a version the source committed over") + } +} + +// The resolve gate stops on the same supersession check, and like it ignores a +// live lookup that lags the replay: an entry committed earlier proves nothing, +// even with a newer mtime or without the replayed manifest. +func TestManifestResolveRetryGateOrdersByCommit(t *testing.T) { + transient := errors.New("dial tcp: connection refused") + live, replayed := committedOverWithOlderMtime() + manifestId := replayed.Chunks[0].GetFileIdString() + + fs := &FilerSink{filerSource: startSourceFilerWithEntry(t, live, ""), dir: "/dst"} + gate := fs.manifestResolveRetryGate("/dst/x.bin", replayed, manifestId, fs.newMissingSourceChunkGate(manifestId)) + if gate(transient) { + t.Fatal("a transient resolve error must not keep retrying a version the source committed over") + } + + lagging := &filer_pb.Entry{ + Attributes: &filer_pb.FuseAttributes{Mtime: 6}, + Chunks: []*filer_pb.FileChunk{{FileId: "5617,00aaa", Size: 10}}, + CommitTsNs: 50, + } + fs = &FilerSink{filerSource: startSourceFilerWithEntry(t, lagging, ""), dir: "/dst"} + gate = fs.manifestResolveRetryGate("/dst/x.bin", replayed, manifestId, fs.newMissingSourceChunkGate(manifestId)) + if !gate(transient) { + t.Fatal("a live lookup committed before the replay must not stop its retries") + } + + volumeGone := fmt.Errorf("LookupFileId locate volume id 5617: %w", source.ErrVolumeNotFound) + gate = fs.manifestResolveRetryGate("/dst/x.bin", replayed, manifestId, fs.newMissingSourceChunkGate(manifestId)) + if !gate(volumeGone) { + t.Fatal("a live lookup committed before the replay must not cut a missing volume's wait short") + } +} + +// A replicate failure on a version the source has moved past is skipped: the +// later event carries the current content. The error here is transient, not a +// missing chunk, so only the supersession check can turn it into a skip. +func TestOnReplicateChunkErrorSkipsSupersededVersion(t *testing.T) { + transient := errors.New("copy 5617,01abc: dial tcp: connection refused") + live, replayed := committedOverWithOlderMtime() + + t.Run("source entry gone", func(t *testing.T) { + fs := &FilerSink{filerSource: startSourceFilerServer(t, &sourceFilerServer{ + lookupFn: func(dir, name string) (*filer_pb.Entry, error) { + return nil, filer_pb.ErrNotFound + }, + }), dir: "/dst"} + if err := fs.onReplicateChunkError("/dst/x.bin", replayed, transient); err != nil { + t.Fatalf("a deleted source entry must skip the replay, got %v", err) + } + }) + + t.Run("committed over with an older mtime", func(t *testing.T) { + fs := &FilerSink{filerSource: startSourceFilerWithEntry(t, live, ""), dir: "/dst"} + if err := fs.onReplicateChunkError("/dst/x.bin", replayed, transient); err != nil { + t.Fatalf("a later commit that dropped the replayed chunks must skip the replay, got %v", err) + } + }) + + t.Run("source unchanged", func(t *testing.T) { + fs := &FilerSink{filerSource: startSourceFilerWithEntry(t, replayed, ""), dir: "/dst"} + if err := fs.onReplicateChunkError("/dst/x.bin", replayed, transient); !errors.Is(err, transient) { + t.Fatalf("a failure on the current version must propagate, got %v", err) + } + }) +} + +func committedOverWithOlderMtime() (live, replayed *filer_pb.Entry) { + live = &filer_pb.Entry{ + Attributes: &filer_pb.FuseAttributes{Mtime: 4}, + Chunks: []*filer_pb.FileChunk{{FileId: "5617,02def", Size: 10}}, + CommitTsNs: 200, + } + replayed = &filer_pb.Entry{ + Attributes: &filer_pb.FuseAttributes{Mtime: 5}, + Chunks: []*filer_pb.FileChunk{{FileId: "5617,01abc", Size: 10}}, + CommitTsNs: 100, + } + return live, replayed +} + +// A peer filer's event can reach the sink before the source store applies it, +// so the live lookup returns the version the replay replaces, which never +// listed the replay's chunk. That lagging entry cannot prove the chunk dropped: +// the 404 must be waited out like any other, not cut short. +func TestFetchAndWriteLaggingLookupKeepsRetrying(t *testing.T) { + prevRetryWaitTime, prevGrace := util.RetryWaitTime, missingSourceChunkGrace + util.RetryWaitTime = 100 * time.Millisecond + missingSourceChunkGrace = 500 * time.Millisecond + t.Cleanup(func() { + util.RetryWaitTime, missingSourceChunkGrace = prevRetryWaitTime, prevGrace + }) + + gone := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + http.Error(w, "not found", http.StatusNotFound) + })) + defer gone.Close() + + lagging := &filer_pb.Entry{ + Attributes: &filer_pb.FuseAttributes{Mtime: 4}, + Chunks: []*filer_pb.FileChunk{{FileId: "5617,00aaa", Size: 10}}, + CommitTsNs: 50, + } + _, replayed := committedOverWithOlderMtime() + fs := &FilerSink{ + filerSource: startSourceFilerWithEntry(t, lagging, strings.TrimPrefix(gone.URL, "http://"), "5617"), + dir: "/dst", + executor: util.NewLimitedConcurrentExecutor(1), + } + fs.SetUploader(operation.NewUploaderWithHttpClient(http.DefaultClient)) + + done := make(chan error, 1) + go func() { + _, fetchErr := fs.fetchAndWrite(replayed.Chunks[0], "/dst/x.bin", replayed, "") + done <- fetchErr + }() + + select { + case fetchErr := <-done: + if !errors.Is(fetchErr, errSourceChunkMissing) { + t.Fatalf("a lagging live lookup must not cut the missing-chunk wait short, got %v", fetchErr) + } + case <-time.After(10 * time.Second): + t.Fatal("the missing chunk must still expire via the gate") + } +} + // A replayed version whose source entry is gone must not cost a single chunk // read or destination write: on a busy directory every skipped fetch is one // that would have 404'd or, worse, uploaded bytes a later event replaces. diff --git a/weed/replication/sink/filersink/filer_sink.go b/weed/replication/sink/filersink/filer_sink.go index 6ecc92f3f0..7773c4bcca 100644 --- a/weed/replication/sink/filersink/filer_sink.go +++ b/weed/replication/sink/filersink/filer_sink.go @@ -338,7 +338,7 @@ func (fs *FilerSink) createEntry(key string, entry *filer_pb.Entry, signatures [ } } - replicatedChunks, err := fs.replicateChunks(context.Background(), entry.GetChunks(), key, getEntryMtimeNs(entry), "") + replicatedChunks, err := fs.replicateChunks(context.Background(), entry.GetChunks(), key, entry, "") if err != nil { // don't swallow: committing mismatched bytes would propagate corruption @@ -439,7 +439,7 @@ func (fs *FilerSink) UpdateEntry(key string, oldEntry *filer_pb.Entry, newParent case updateRepair: glog.Warningf("repair %s: destination diverged from source (dst %d chunks, src %d chunks); re-replicating full source content", key, len(existingEntry.GetChunks()), len(newEntry.GetChunks())) - replicatedChunks, err := fs.replicateChunks(context.Background(), newEntry.GetChunks(), targetKey, getEntryMtimeNs(newEntry), "") + replicatedChunks, err := fs.replicateChunks(context.Background(), newEntry.GetChunks(), targetKey, newEntry, "") if err != nil { if errors.Is(err, errChunkSizeMismatch) { return true, fs.onCorruptChunk(targetKey, newEntry, err) @@ -467,7 +467,7 @@ func (fs *FilerSink) UpdateEntry(key string, oldEntry *filer_pb.Entry, newParent } // replicate the chunks that are new in the source - replicatedChunks, err := fs.replicateChunks(context.Background(), newChunks, targetKey, getEntryMtimeNs(newEntry), "") + replicatedChunks, err := fs.replicateChunks(context.Background(), newChunks, targetKey, newEntry, "") if err != nil { if errors.Is(err, errChunkSizeMismatch) { return true, fs.onCorruptChunk(targetKey, newEntry, err) diff --git a/weed/replication/sink/filersink/filer_sink_test.go b/weed/replication/sink/filersink/filer_sink_test.go index c276fad0b1..f54467ebbd 100644 --- a/weed/replication/sink/filersink/filer_sink_test.go +++ b/weed/replication/sink/filersink/filer_sink_test.go @@ -108,7 +108,7 @@ func TestGetEntryMtimeNs(t *testing.T) { // onCorruptChunk must surface the error, not skip, when it can't confirm // supersession (no source filer here) — otherwise it could drop a live file. func TestOnCorruptChunkRefusesWhenSupersessionUnconfirmed(t *testing.T) { - fs := &FilerSink{} // filerSource nil => hasSourceNewerVersion is always false + fs := &FilerSink{} // filerSource nil => sourceMovedOn is always false sentinel := errors.New("corrupt chunk: read 0 bytes, source says 107") got := fs.onCorruptChunk("/buckets/x/config", @@ -123,14 +123,14 @@ func TestOnCorruptChunkRefusesWhenSupersessionUnconfirmed(t *testing.T) { // onReplicateChunkError must NOT swallow a transient/partial replicate failure // (a volume-id lookup race or unretried manifest-chunk resolve during a write // burst) when it cannot confirm the source has moved on. Here filerSource is -// nil, so hasSourceNewerVersion is always false — the "supersession unconfirmed" +// nil, so sourceMovedOn is always false — the "supersession unconfirmed" // case. Returning nil would let the metadata subscription advance its offset // past an event whose file was never written, silently dropping it (the // filer.backup partial-landing bug). It must propagate so the offset stays put -// and the event is reprocessed. Genuine source-404 (gone/superseded) skipping is -// covered by TestSourceSupersedes, which gates the return-nil branch. +// and the event is reprocessed. The gone or superseded skip is covered by +// TestOnReplicateChunkErrorSkipsSupersededVersion. func TestOnReplicateChunkErrorPropagatesWhenSupersessionUnconfirmed(t *testing.T) { - fs := &FilerSink{} // filerSource nil => hasSourceNewerVersion always false + fs := &FilerSink{} // filerSource nil => sourceMovedOn always false sentinel := errors.New("replicate manifest data chunks 7,02def: volume id 7 not found") got := fs.onReplicateChunkError("/buckets/x/model.pt",