diff --git a/weed/mount/filehandle.go b/weed/mount/filehandle.go index cb530614d0..fa26379c7c 100644 --- a/weed/mount/filehandle.go +++ b/weed/mount/filehandle.go @@ -147,11 +147,13 @@ func (fh *FileHandle) SetEntry(entry *filer_pb.Entry) { // installAckedEntry installs filer-acknowledged state under the handle lock // when it outranks the handle. A version never advances without its value: // stamping alone would fence out the events carrying what the handle lacks. -// Dirty handles are skipped — local writes supersede the ack. +// Dirty handles are skipped — local writes supersede the ack. Deleted ones +// too: their path may already name a different file, and no acknowledged +// state belongs to this handle anymore. func (fh *FileHandle) installAckedEntry(entry *filer_pb.Entry, versionTsNs int64, signature int32) { fhActiveLock := fh.wfs.fhLockTable.AcquireLock("installAckedEntry", fh.fh, util.ExclusiveLock) defer fh.wfs.fhLockTable.ReleaseLock(fh.fh, fhActiveLock) - if versionTsNs == 0 || fh.dirtyMetadata || entry == fh.GetEntry().GetEntry() { + if versionTsNs == 0 || fh.dirtyMetadata || fh.isDeleted || entry == fh.GetEntry().GetEntry() { return } // Refuse only what is provably older. Two known, differing filer diff --git a/weed/mount/filehandle_read.go b/weed/mount/filehandle_read.go index 862a6e63de..a00e2d606f 100644 --- a/weed/mount/filehandle_read.go +++ b/weed/mount/filehandle_read.go @@ -111,6 +111,60 @@ func (fh *FileHandle) readFromChunksWithContext(ctx context.Context, buff []byte return int64(totalRead), ts, err } +// refreshEntryForRead re-reads the entry from the filer and installs it when it +// outranks what the handle holds. A read can resolve chunk views a concurrent +// overwrite already replaced — the filer queues the replaced file ids for +// deletion at commit, while the handle only learns the new list when the +// metadata event lands — so a fetch can name data that is already gone. The +// filer is authoritative once the commit lands; installing its entry lets a +// retry reach data that exists. Reports whether the handle advanced: an +// unchanged version means the failure is genuine and a retry would repeat it. +func (fh *FileHandle) refreshEntryForRead(ctx context.Context) bool { + // An in-flight async flush is mid-commit: wait it out before judging what + // the committed entry is. Dirty and deleted handles are fenced inside + // installAckedEntry, under the exclusive handle lock that owns those flags. + if !fh.wfs.waitForPendingAsyncFlush(ctx, fh.inode) { + return false + } + fileFullPath := fh.FullPath() + dir, name := fileFullPath.DirAndName() + var fetched *filer_pb.Entry + var versionTsNs int64 + var signature int32 + err := fh.wfs.withFilerClient(ctx, false, fh.wfs.filerRPCWait(), func(ctx context.Context, client filer_pb.SeaweedFilerClient) error { + resp, lookupErr := filer_pb.LookupEntry(ctx, client, &filer_pb.LookupDirectoryEntryRequest{ + Directory: dir, + Name: name, + }) + if lookupErr != nil { + return lookupErr + } + fetched, versionTsNs, signature = resp.Entry, resp.LogTsNs, resp.GetLogSignature() + return nil + }) + if err != nil || fetched == nil || versionTsNs == 0 { + return false + } + // The saved path may name a different file by now — an unlinked handle + // keeps answering under the reused name. Only an entry that still carries + // the handle's inode is its own content; zero means unverifiable, not + // different. + fh.entry.RLock() + curInode := fh.entry.Entry.GetAttributes().GetInode() + fh.entry.RUnlock() + newInode := fetched.GetAttributes().GetInode() + if curInode != 0 && newInode != 0 && curInode != newInode { + return false + } + fetched = proto.Clone(fetched).(*filer_pb.Entry) + if fh.wfs.option.UidGidMapper != nil { + fh.wfs.mapPbIdFromFilerToLocal(fetched) + } + before := fh.entryVersionTsNs.Load() + fh.installAckedEntry(fetched, versionTsNs, signature) + return fh.entryVersionTsNs.Load() != before +} + func (fh *FileHandle) downloadRemoteEntry(ctx context.Context, entry *LockedEntry) error { fileFullPath := fh.FullPath() diff --git a/weed/mount/meta_cache/meta_cache_ctx_poison_test.go b/weed/mount/meta_cache/meta_cache_ctx_poison_test.go new file mode 100644 index 0000000000..8191ad8cdf --- /dev/null +++ b/weed/mount/meta_cache/meta_cache_ctx_poison_test.go @@ -0,0 +1,180 @@ +package meta_cache + +import ( + "context" + "errors" + "io" + "sync" + "testing" + "time" + + "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" + "github.com/seaweedfs/seaweedfs/weed/util" + "google.golang.org/grpc" + "google.golang.org/grpc/metadata" +) + +// The shared directory build behind EnsureVisited is a singleflight call: the +// first requester for a path owns the build, and every later requester for the +// same path waits on that same build. If the build runs on the owner's request +// context, the owner walking away (a FUSE request returning, the kernel +// interrupting it) kills the build for everybody else. +// +// A waiter whose own context is still alive then receives somebody else's +// cancellation. weed/mount/weedfs_dir_read.go classifies a canceled read as +// EINTR only when the *local* context is done, so the waiter falls through to +// fuse.EIO — the kernel hands EIO to readdirent, kubelet's applyFSGroup fails, +// and the volume never mounts. That is the 2026-10-28 gitea outage. +// +// The mock below reproduces it deterministically: the filer stream blocks +// mid-listing, the owner's context is canceled while the build is in flight, +// and a second requester joins that same build. + +// ctxAwareListStream models a filer listing that fails with the request +// context's error once that context is canceled, the way a real gRPC stream +// does. +type ctxAwareListStream struct { + ctx context.Context + responses []*filer_pb.ListEntriesResponse + entered chan struct{} + release chan struct{} + once sync.Once + index int +} + +func (s *ctxAwareListStream) Recv() (*filer_pb.ListEntriesResponse, error) { + s.once.Do(func() { + close(s.entered) + <-s.release + }) + if err := s.ctx.Err(); err != nil { + return nil, err + } + if s.index >= len(s.responses) { + return nil, io.EOF + } + resp := s.responses[s.index] + s.index++ + return resp, nil +} + +func (s *ctxAwareListStream) Header() (metadata.MD, error) { return metadata.MD{}, nil } +func (s *ctxAwareListStream) Trailer() metadata.MD { return metadata.MD{} } +func (s *ctxAwareListStream) CloseSend() error { return nil } +func (s *ctxAwareListStream) Context() context.Context { return s.ctx } +func (s *ctxAwareListStream) SendMsg(any) error { return nil } +func (s *ctxAwareListStream) RecvMsg(any) error { return nil } + +// ctxAwareListClient counts how many listings were started. The shared build +// must be listed exactly once no matter how many requesters join it. +type ctxAwareListClient struct { + filer_pb.SeaweedFilerClient + stream *ctxAwareListStream + + mu sync.Mutex + calls int +} + +func (c *ctxAwareListClient) ListEntries(ctx context.Context, in *filer_pb.ListEntriesRequest, opts ...grpc.CallOption) (grpc.ServerStreamingClient[filer_pb.ListEntriesResponse], error) { + c.mu.Lock() + c.calls++ + c.mu.Unlock() + c.stream.ctx = ctx + return c.stream, nil +} + +func (c *ctxAwareListClient) listCalls() int { + c.mu.Lock() + defer c.mu.Unlock() + return c.calls +} + +func TestEnsureVisitedSharedBuildSurvivesOwnerRequestCancel(t *testing.T) { + mc, _, _, _ := newTestMetaCache(t, map[util.FullPath]bool{ + "/": true, + }) + defer mc.Shutdown() + + entered := make(chan struct{}) + release := make(chan struct{}) + client := &ctxAwareListClient{ + stream: &ctxAwareListStream{ + responses: []*filer_pb.ListEntriesResponse{ + { + Entry: &filer_pb.Entry{ + Name: "base.txt", + Attributes: &filer_pb.FuseAttributes{ + Crtime: 1, + Mtime: 1, + FileMode: 0100644, + FileSize: 3, + }, + }, + SnapshotTsNs: 100, + }, + }, + entered: entered, + release: release, + }, + } + accessor := &buildFilerAccessor{client: client} + + // The owner starts the shared build. + ownerCtx, cancelOwner := context.WithCancel(context.Background()) + ownerErr := make(chan error, 1) + go func() { + ownerErr <- EnsureVisited(ownerCtx, mc, accessor, util.FullPath("/dir"), 0) + }() + + select { + case <-entered: + case <-time.After(10 * time.Second): + t.Fatal("shared build never reached the filer listing") + } + + // The owner's FUSE request goes away while the build is still in flight. + cancelOwner() + if err := <-ownerErr; !errors.Is(err, context.Canceled) { + t.Fatalf("owner error = %v, want context.Canceled", err) + } + + // A second requester arrives and joins the still-running shared build. + waiterErr := make(chan error, 1) + go func() { + waiterErr <- EnsureVisited(context.Background(), mc, accessor, util.FullPath("/dir"), 0) + }() + + // Give the waiter time to join. If it started its own build instead, the + // listing is requested twice — the repro would not be the production path. + deadline := time.Now().Add(2 * time.Second) + for client.listCalls() < 2 && time.Now().Before(deadline) { + select { + case err := <-waiterErr: + t.Fatalf("waiter returned before the shared build finished (%v); it started a second listing (%d calls) instead of joining the shared build", + err, client.listCalls()) + case <-time.After(10 * time.Millisecond): + } + } + if calls := client.listCalls(); calls != 1 { + t.Fatalf("listing calls = %d, want 1 (the waiter must join the shared build)", calls) + } + + // Let the build finish. + close(release) + + select { + case err := <-waiterErr: + if err != nil { + t.Fatalf("waiter inherited the owner's cancellation: %v", err) + } + case <-time.After(30 * time.Second): + t.Fatal("waiter never returned") + } + + if calls := client.listCalls(); calls != 1 { + t.Fatalf("listing calls = %d, want 1", calls) + } + if !mc.IsDirectoryCached(util.FullPath("/dir")) { + t.Fatal("directory /dir should be cached after the shared build completes") + } +} diff --git a/weed/mount/meta_cache/meta_cache_init.go b/weed/mount/meta_cache/meta_cache_init.go index f4bdf9b216..946086cff0 100644 --- a/weed/mount/meta_cache/meta_cache_init.go +++ b/weed/mount/meta_cache/meta_cache_init.go @@ -91,13 +91,26 @@ const ( ) func doEnsureVisited(ctx context.Context, mc *MetaCache, client filer_pb.FilerClient, path util.FullPath, maxCacheableEntries int) error { + // The shared build must not run on the requester's context. A FUSE request + // that returns — or that the kernel interrupts — cancels that context, and + // every other requester waiting on the same build would inherit the + // cancellation. weed/mount/weedfs_dir_read.go only reports EINTR when the + // *local* context is done, so such a waiter falls through to fuse.EIO and + // the kernel hands EIO to readdirent: kubelet's applyFSGroup fails and the + // volume never mounts. + // + // Build on a context that keeps the request's values but not its + // cancellation. Waiters still abandon on their own context below, so a + // canceled request stops waiting immediately. + buildCtx := context.WithoutCancel(ctx) + // Use singleflight to deduplicate concurrent requests for the same path. // DoChan so a canceled request stops waiting even when another request // owns the in-flight build. resultCh := mc.visitGroup.DoChan(string(path), func() (interface{}, error) { // Check for cancellation before starting - if ctx.Err() != nil { - return nil, ctx.Err() + if buildCtx.Err() != nil { + return nil, buildCtx.Err() } // Double-check if already cached (another goroutine may have completed) @@ -128,7 +141,7 @@ func doEnsureVisited(ctx context.Context, mc *MetaCache, client filer_pb.FilerCl } } defer func() { - if !cleanupDone && ctx.Err() != nil { + if !cleanupDone && buildCtx.Err() != nil { cleanupBuild("canceled") } }() @@ -139,11 +152,11 @@ func doEnsureVisited(ctx context.Context, mc *MetaCache, client filer_pb.FilerCl entryCount = 0 sections = sectionBoundsCollector{} var batch []*filer.Entry // reset on retry, allow GC of previous entries - if err := mc.deleteFolderChildrenForRebuild(ctx, path); err != nil { + if err := mc.deleteFolderChildrenForRebuild(buildCtx, path); err != nil { return fmt.Errorf("clear existing entries for %s: %w", path, err) } var listErr error - snapshotTsNs, listErr = filer_pb.ReadDirAllEntriesWithSnapshot(ctx, client, path, "", func(pbEntry *filer_pb.Entry, isLast bool) error { + snapshotTsNs, listErr = filer_pb.ReadDirAllEntriesWithSnapshot(buildCtx, client, path, "", func(pbEntry *filer_pb.Entry, isLast bool) error { entry := filer.FromPbEntry(string(path), pbEntry) if !mc.includeSystemEntries && IsHiddenSystemEntry(string(path), entry.Name()) { return nil @@ -158,7 +171,7 @@ func doEnsureVisited(ctx context.Context, mc *MetaCache, client filer_pb.FilerCl // flush by size, not isLast: hidden entries can return early if len(batch) >= batchInsertSize { - if err := mc.doBatchInsertEntries(ctx, batch); err != nil { + if err := mc.doBatchInsertEntries(buildCtx, batch); err != nil { return fmt.Errorf("batch insert for %s: %w", path, err) } batch = make([]*filer.Entry, 0, batchInsertSize) @@ -169,7 +182,7 @@ func doEnsureVisited(ctx context.Context, mc *MetaCache, client filer_pb.FilerCl return listErr } if len(batch) > 0 { - if err := mc.doBatchInsertEntries(ctx, batch); err != nil { + if err := mc.doBatchInsertEntries(buildCtx, batch); err != nil { return fmt.Errorf("batch insert remaining for %s: %w", path, err) } } @@ -198,14 +211,14 @@ func doEnsureVisited(ctx context.Context, mc *MetaCache, client filer_pb.FilerCl // immediate (a clean-EOF stream glitch clears at once), later ones space out. // On cancellation the deferred cleanup aborts the build. for attempt := 0; entryCount == 0 && attempt < emptyRebuildConfirmations; attempt++ { - if ctx.Err() != nil { - return nil, ctx.Err() + if buildCtx.Err() != nil { + return nil, buildCtx.Err() } if attempt > 0 { select { case <-time.After(emptyRebuildConfirmDelay): - case <-ctx.Done(): - return nil, ctx.Err() + case <-buildCtx.Done(): + return nil, buildCtx.Err() } } if entryCount, snapshotTsNs, sections, fetchErr = reloadFromFiler(); fetchErr != nil { diff --git a/weed/mount/meta_cache/meta_cache_sections.go b/weed/mount/meta_cache/meta_cache_sections.go index 05781b1505..0918a124a8 100644 --- a/weed/mount/meta_cache/meta_cache_sections.go +++ b/weed/mount/meta_cache/meta_cache_sections.go @@ -285,21 +285,33 @@ func (mc *MetaCache) staleRangeOf(dirPath util.FullPath, name string) (nameRange } func (mc *MetaCache) refreshSection(ctx context.Context, client filer_pb.FilerClient, dirPath util.FullPath, lo, hi string) error { - _, err, _ := mc.visitGroup.Do(string(dirPath)+"\x00section\x00"+lo, func() (interface{}, error) { + // DoChan so a requester that leaves mid-refresh stops waiting on its own + // context while the shared refresh continues for the remaining waiters. + resultCh := mc.visitGroup.DoChan(string(dirPath)+"\x00section\x00"+lo, func() (interface{}, error) { if !mc.rangeStale(dirPath, lo) { return nil, nil } - entries, snapshotTsNs, err := mc.listFilerRange(ctx, client, dirPath, lo, hi) + // Shared like the directory build: the requester that started this + // flight can leave mid-refresh, so re-apply the bound on a context it + // cannot cancel rather than failing every waiter on its cancellation. + flightCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), sectionRefreshTimeout) + defer cancel() + entries, snapshotTsNs, err := mc.listFilerRange(flightCtx, client, dirPath, lo, hi) if err != nil { return nil, err } - return nil, mc.enqueueAndWait(ctx, metadataApplyRequest{ + return nil, mc.enqueueAndWait(flightCtx, metadataApplyRequest{ kind: metadataSectionRefresh, buildPath: dirPath, refresh: §ionRefresh{lo: lo, hi: hi, entries: entries, snapshotTsNs: snapshotTsNs}, }) }) - return err + select { + case result := <-resultCh: + return result.Err + case <-ctx.Done(): + return ctx.Err() + } } // listFilerRange reads [lo, hi) from the filer at one snapshot, paging by diff --git a/weed/mount/weedfs_dir_read.go b/weed/mount/weedfs_dir_read.go index 6e7a391592..b61d8bb0fc 100644 --- a/weed/mount/weedfs_dir_read.go +++ b/weed/mount/weedfs_dir_read.go @@ -13,6 +13,8 @@ import ( "github.com/seaweedfs/seaweedfs/weed/mount/meta_cache" "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" "github.com/seaweedfs/seaweedfs/weed/util" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" ) type DirectoryHandleId uint64 @@ -303,11 +305,7 @@ func (wfs *WFS) doReadDirectory(ctx context.Context, input *fuse.ReadIn, out Dir if errors.As(err, &tooLarge) { return wfs.readDirectoryDirect(ctx, input, out, dh, dirPath, processEachEntryFn) } - if ctx.Err() != nil { - return fuse.EINTR - } - glog.Errorf("dir ReadDirAll %s: %v", dirPath, err) - return fuse.EIO + return dirReadStatus(ctx, err, "dir ReadDirAll %s: %v", dirPath, err) } // Serve the maintained-but-unverified cache if the filer is unreachable. @@ -327,11 +325,7 @@ func (wfs *WFS) doReadDirectory(ctx context.Context, input *fuse.ReadIn, out Dir }) dh.lastListedName = storeLastName if loadErr != nil { - if ctx.Err() != nil { - return fuse.EINTR - } - glog.Errorf("list meta cache: %v", loadErr) - return fuse.EIO + return dirReadStatus(ctx, loadErr, "list meta cache: %v", loadErr) } } @@ -368,11 +362,7 @@ func (wfs *WFS) doReadDirectory(ctx context.Context, input *fuse.ReadIn, out Dir // carries on from wherever the cached walk reached. return wfs.readDirectoryDirect(ctx, input, out, dh, dirPath, processEachEntryFn) } - if ctx.Err() != nil { - return fuse.EINTR - } - glog.Errorf("dir ReadDirAll %s: %v", dirPath, err) - return fuse.EIO + return dirReadStatus(ctx, err, "dir ReadDirAll %s: %v", dirPath, err) } // Page from where the store itself reached, not from the last entry the @@ -404,11 +394,7 @@ func (wfs *WFS) doReadDirectory(ctx context.Context, input *fuse.ReadIn, out Dir return true, nil }) if loadErr != nil { - if ctx.Err() != nil { - return fuse.EINTR - } - glog.Errorf("list meta cache: %v", loadErr) - return fuse.EIO + return dirReadStatus(ctx, loadErr, "list meta cache: %v", loadErr) } dh.lastListedName = storeLastName @@ -423,6 +409,25 @@ func (wfs *WFS) doReadDirectory(ctx context.Context, input *fuse.ReadIn, out Dir return fuse.OK } +// dirReadStatus maps a failed directory read to a FUSE status. +// +// A canceled read is retryable and must be reported as EINTR, never as EIO: +// the kernel hands EIO from readdirent straight to the caller, and kubelet's +// applyFSGroup then fails the volume mount instead of retrying. A cancellation +// that arrives with the local context still alive belongs to a shared build +// that another requester abandoned — the read itself can still be retried, and +// a gRPC cancellation of the listing classifies the same way. Only real +// failures are worth an error log; interruptions are ordinary. +func dirReadStatus(ctx context.Context, err error, format string, args ...any) fuse.Status { + if ctx.Err() != nil || + errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) || + status.Code(err) == codes.Canceled || status.Code(err) == codes.DeadlineExceeded { + return fuse.EINTR + } + glog.Errorf(format, args...) + return fuse.EIO +} + // ensureDirectoryVisited pulls the directory into the local cache, unless it is // too large to cache: then the directory is marked read-through, so later // listings go straight to the filer without re-asking. @@ -444,11 +449,7 @@ func (wfs *WFS) readDirectoryDirect(ctx context.Context, input *fuse.ReadIn, out skipCount := uint32(input.Offset-dh.entryStreamOffset) + batchSize entries, snapshotTs, err := loadDirectoryEntriesDirect(listingCtx, wfs, wfs.option.UidGidMapper, dirPath, "", false, skipCount, dh.snapshotTsNs, wfs.option.IncludeSystemEntries) if err != nil { - if ctx.Err() != nil { - return fuse.EINTR - } - glog.Errorf("list filer directory: %v", err) - return fuse.EIO + return dirReadStatus(ctx, err, "list filer directory: %v", err) } dh.entryStream = append(dh.entryStream, entries...) if dh.snapshotTsNs == 0 { @@ -481,11 +482,7 @@ func (wfs *WFS) readDirectoryDirect(ctx context.Context, input *fuse.ReadIn, out entries, snapshotTs, err := loadDirectoryEntriesDirect(listingCtx, wfs, wfs.option.UidGidMapper, dirPath, lastEntryName, false, batchSize, dh.snapshotTsNs, wfs.option.IncludeSystemEntries) if err != nil { - if ctx.Err() != nil { - return fuse.EINTR - } - glog.Errorf("list filer directory: %v", err) - return fuse.EIO + return dirReadStatus(ctx, err, "list filer directory: %v", err) } if dh.snapshotTsNs == 0 { dh.snapshotTsNs = snapshotTs diff --git a/weed/mount/weedfs_dir_read_test.go b/weed/mount/weedfs_dir_read_test.go index b2c792da0f..356384f80c 100644 --- a/weed/mount/weedfs_dir_read_test.go +++ b/weed/mount/weedfs_dir_read_test.go @@ -2,14 +2,20 @@ package mount import ( "context" + "errors" + "fmt" "io" "testing" + "github.com/seaweedfs/go-fuse/v2/fuse" + "github.com/seaweedfs/seaweedfs/weed/mount/meta_cache" "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" "github.com/seaweedfs/seaweedfs/weed/util" "google.golang.org/grpc" + "google.golang.org/grpc/codes" "google.golang.org/grpc/metadata" + "google.golang.org/grpc/status" ) type directoryListStream struct { @@ -120,3 +126,28 @@ func TestLoadDirectoryEntriesDirectShowsSystemEntriesWhenEnabled(t *testing.T) { t.Fatalf("entry names = %q, %q, want topics, visible", entries[0].Name(), entries[1].Name()) } } + +func TestDirReadStatus(t *testing.T) { + canceledCtx, cancel := context.WithCancel(context.Background()) + cancel() + + tests := []struct { + name string + ctx context.Context + err error + want fuse.Status + }{ + {"local context canceled", canceledCtx, io.EOF, fuse.EINTR}, + {"context canceled error", context.Background(), context.Canceled, fuse.EINTR}, + {"context deadline error", context.Background(), context.DeadlineExceeded, fuse.EINTR}, + {"grpc canceled", context.Background(), status.Error(codes.Canceled, "stream canceled"), fuse.EINTR}, + {"grpc deadline exceeded", context.Background(), status.Error(codes.DeadlineExceeded, "deadline exceeded"), fuse.EINTR}, + {"wrapped grpc canceled", context.Background(), fmt.Errorf("list: %w", status.Error(codes.Canceled, "stream canceled")), fuse.EINTR}, + {"ordinary failure", context.Background(), errors.New("filer broke"), fuse.EIO}, + } + for _, tt := range tests { + if got := dirReadStatus(tt.ctx, tt.err, "dir read: %v", tt.err); got != tt.want { + t.Errorf("%s: dirReadStatus = %v, want %v", tt.name, got, tt.want) + } + } +} diff --git a/weed/mount/weedfs_file_read.go b/weed/mount/weedfs_file_read.go index 33f5b6bb09..c5cf9aa5ad 100644 --- a/weed/mount/weedfs_file_read.go +++ b/weed/mount/weedfs_file_read.go @@ -52,6 +52,19 @@ func (wfs *WFS) Read(cancel <-chan struct{}, in *fuse.ReadIn, buff []byte) (fuse offset := int64(in.Offset) totalRead, err := readDataByFileHandleWithContext(ctx, buff, fh, offset) + if err != nil { + // The chunk views a read resolves can name data a concurrent overwrite + // already replaced: the filer queues replaced file ids for deletion at + // commit while this handle only learns the new list when the metadata + // event lands. Refreshing the entry installs the chunks that exist; + // it takes the exclusive handle lock, so drop the shared one first. + fh.wfs.fhLockTable.ReleaseLock(fh.fh, fhActiveLock) + retry := fh.refreshEntryForRead(ctx) + fhActiveLock = fh.wfs.fhLockTable.AcquireLock("Read", fh.fh, util.SharedLock) + if retry { + totalRead, err = readDataByFileHandleWithContext(ctx, buff, fh, offset) + } + } if err != nil { glog.Warningf("file handle read %s %d: %v", fh.FullPath(), totalRead, err) return nil, fuse.EIO