mirror of
https://github.com/seaweedfs/seaweedfs.git
synced 2026-11-15 00:09:00 +01:00
mount: keep a refused content create refused when the stored-entry read stalls (#12629)
The read of the stored entry ran in the create's filer callback, so a read that hit the call deadline turned the refusal into a filer timeout, and both retry layers resent the refused create. Read it in its own call, bounded to one filer call. Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
1 parent
6eb2fb4842
commit
e0890391ac
2 files changed
+65
-9
No files matched your search
@@ -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())
|
||||
|
||||
@@ -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
|
||||
|
||||
|
||||
Reference in new issue
Block a user