diff --git a/weed/mount/content_flush_condition_test.go b/weed/mount/content_flush_condition_test.go index 7c7e4d3b4c..68a5817670 100644 --- a/weed/mount/content_flush_condition_test.go +++ b/weed/mount/content_flush_condition_test.go @@ -196,6 +196,51 @@ func TestContentFlushRepeatsOnOneHandle(t *testing.T) { } } +type refusedThenStalledFiler struct { + *contentFlushConditionFiler +} + +func (s *refusedThenStalledFiler) LookupDirectoryEntry(ctx context.Context, r *filer_pb.LookupDirectoryEntryRequest) (*filer_pb.LookupDirectoryEntryResponse, error) { + <-ctx.Done() + return nil, status.FromContextError(ctx.Err()).Err() +} + +// A refused create stays refused when reading the stored entry stalls: the +// flush must not resend the create as if the filer had timed out. +func TestContentFlushRefusalWithStalledLookup(t *testing.T) { + wfs := newInvalidateTestWFS(t) + wfs.filerCallTimeout = 200 * time.Millisecond + fake := &contentFlushConditionFiler{entry: flushConditionEntry(1, 2)} + startFakeFiler(t, wfs, &refusedThenStalledFiler{fake}) + oldSleep := metadataFlushSleep + metadataFlushSleep = func(context.Context, time.Duration) {} + t.Cleanup(func() { metadataFlushSleep = oldSleep }) + observed := flushConditionEntry(1) + wfs.mapPbIdFromFilerToLocal(observed) + live := flushConditionEntry(1, 3) + live.Chunks[1].Offset = 2 + wfs.mapPbIdFromFilerToLocal(live) + fh := newFileHandle(wfs, 1, 42, live) + t.Cleanup(func() { fh.dirtyPages.Destroy() }) + fh.baseEntry.Store(observed) + fh.dirtyMetadata = true + fh.RememberPath(util.FullPath("/dir/file")) + start := time.Now() + code := wfs.doFlush(context.Background(), fh, 1000, 20, false) + elapsed := time.Since(start) + fake.mu.Lock() + defer fake.mu.Unlock() + if code == fuse.OK || !fh.dirtyMetadata { + t.Fatalf("refused flush reported %v, dirty=%v", code, fh.dirtyMetadata) + } + if fake.calls != 1 || elapsed > 2*time.Second { + t.Fatalf("refused create sent %d times over %v", fake.calls, elapsed) + } + if got := chunkFileKeys(fake.entry); !slices.Equal(got, []uint64{1, 2}) { + t.Fatalf("stored chunks %v", got) + } +} + func chunkFileKeys(e *filer_pb.Entry) (keys []uint64) { for _, c := range e.GetChunks() { keys = append(keys, c.GetFid().GetFileKey()) diff --git a/weed/mount/weedfs_file_sync.go b/weed/mount/weedfs_file_sync.go index ef905d2095..491b2f0aec 100644 --- a/weed/mount/weedfs_file_sync.go +++ b/weed/mount/weedfs_file_sync.go @@ -371,26 +371,22 @@ func (wfs *WFS) createContentEntry(ctx context.Context, fh *FileHandle, request sent := request.Entry for attempt := 0; ; attempt++ { request.Condition = &filer_pb.WriteCondition{Clauses: []*filer_pb.WriteCondition_Clause{guard}} - var stored *filer_pb.Entry err = wfs.withFilerClient(ctx, false, wfs.filerRPCWait(), func(ctx context.Context, client filer_pb.SeaweedFilerClient) error { var callErr error - stored = nil resp, callErr = filer_pb.CreateEntryWithResponse(ctx, client, request) - if base == nil || attempt >= maxContentFlushMerges || !errors.Is(callErr, filer_pb.ErrPreconditionFailed) { - return callErr - } - if found, lookupErr := filer_pb.LookupEntry(ctx, client, &filer_pb.LookupDirectoryEntryRequest{Directory: request.Directory, Name: request.Entry.Name}); lookupErr == nil { - stored = found.Entry - } return callErr }) if err == nil && request.Entry != sent { merged = proto.Clone(request.Entry).(*filer_pb.Entry) wfs.mapPbIdFromFilerToLocal(merged) } - if err == nil || stored == nil { + if err == nil || base == nil || attempt >= maxContentFlushMerges || !errors.Is(err, filer_pb.ErrPreconditionFailed) { return resp, merged, err } + stored := wfs.lookupRefusedEntry(ctx, request.Directory, request.Entry.Name) + if stored == nil { + return resp, nil, err + } mark := fh.markSince(base) next, ok := mergeContentEntry(stored, base, local, sent, mark.truncatedTo, mark.rewrote) if !ok || buffered && !mergeKeepsBuffered(next, local) { @@ -402,6 +398,21 @@ func (wfs *WFS) createContentEntry(ctx context.Context, fh *FileHandle, request } } +// lookupRefusedEntry reads the entry a refused create ran into, within one +// filer call: a read that stalls must not hold or replay the refusal. +func (wfs *WFS) lookupRefusedEntry(ctx context.Context, dir, name string) (stored *filer_pb.Entry) { + ctx, cancel := context.WithTimeout(ctx, wfs.filerRPCWait()) + defer cancel() + _ = wfs.withFilerClient(ctx, false, wfs.filerRPCWait(), func(ctx context.Context, client filer_pb.SeaweedFilerClient) error { + found, err := filer_pb.LookupEntry(ctx, client, &filer_pb.LookupDirectoryEntryRequest{Directory: dir, Name: name}) + if err == nil { + stored = found.Entry + } + return err + }) + return stored +} + // maxContentFlushMerges bounds how often one flush rebases onto a moving entry. const maxContentFlushMerges = 3