From bae2d0f68235747c4fd538947ae9c3056c19d739 Mon Sep 17 00:00:00 2001 From: Yevgeniy Firsov Date: Mon, 28 Sep 2026 13:07:14 -0700 Subject: [PATCH 1/4] fix: backport four correctness fixes from geesefs (#51) * fix: register for the flush wakeup before checking inode state SyncFile checked flushPending and waited on flusherCond without holding the inode lock, so a flush could complete between the state check and the wait and the wakeup would be missed. The sync then blocks until some unrelated flush happens to signal, or forever if none does. Take the inode lock to read the state, acquire flusherMu before releasing it, and only then wait, so completion cannot slip through the gap. Cherry-picked from geesefs fbf052a, with its test. Our fork predates it. The test fails against our current code with a one-second timeout and passes with the fix. (cherry picked from commit fbf052a89e2bc0ba21d7ca8db8b2d0e8b5f4d0a1) * fix: bound the part count instead of indexing past the last part tier completeMultipart derived the part count by taking partNum of the final size and incrementing when the part started below it. At a size that exactly fills the configured tiers that asks partRange for one part beyond the last, which panics with "Part number too large". Introduce numParts and maxParts, and use them, so the count saturates at the maximum instead of running off the end. Cherry-picked from geesefs 8fdec0d, with its tests; the module path in the test is retargeted. Note this supersedes their earlier attempt at the same bug (a5cc92e), which loosened the comparison in partRange and was reverted here for allowing out-of-range parts. We have neither, so only this one should be taken. Not reachable at ordinary file sizes: it needs a file that exactly fills every part tier. (cherry picked from commit 8fdec0de5a9c4b1e0f2b3d4c5a6b7c8d9e0f1a2b) * fix: bound read retries instead of retrying forever by default The read retry loop treats an attempt limit below one as unlimited, and the flag defaulted to zero, so a persistently failing read retried without end and the operation never returned an error to the caller. Default the limit to ten and reject a value below one at startup, matching the validation style already used for part sizes in the same function. Cherry-picked from geesefs ae2c4ed. Behaviour change worth calling out at review: unlimited retries were previously reachable by passing zero, and that now aborts at startup rather than being silently accepted. Anyone deliberately running unlimited has to change the flag. The unlimited branch itself stays in the read loop, since the S3 backend still lowers and restores the limit around its own calls. (cherry picked from commit ae2c4edfcb1a2b3c4d5e6f708192a3b4c5d6e7f8) * fix: do not take a lookup reference for . and .. in readdirplus The readdirplus path referenced every entry it wrote, including the . and .. entries. The kernel does not count those toward nlookup, so each listing leaked a reference on the directory and its parent, and those inodes were never forgotten. Skip the reference for the two self entries. Cherry-picked from geesefs bbedeb0. Our fork predates it. (cherry picked from commit bbedeb027a1b2c3d4e5f60718293a4b5c6d7e8f9) * fix: identify . and .. by directory position, not inode identity The backported reference fix compares the entry against dh.inode and dh.inode.Parent, but ReadDir releases the parent lock before returning, so a rename can change dh.inode.Parent in between. The comparison then misses, the .. entry is referenced, and no forget ever matches it: the leak the fix targets. ReadDir yields . and .. at internal offsets 0 and 1, and dh.Next only advances the position after this block, so the position still identifies this entry and is held under dh.mu throughout. It is also correct for the root, where ReadDir returns the directory itself for .. because there is no parent. Divergence from geesefs bbedeb0 rather than a port; worth sending upstream, which carries the same window. Raised in review on #51. --- core/cfg/flags.go | 12 +++-- core/file.go | 40 ++++++++++++---- core/file_sync_test.go | 88 ++++++++++++++++++++++++++++++++++ core/goofys_fuse.go | 10 +++- core/part_size_test.go | 104 +++++++++++++++++++++++++++++++++++++++++ 5 files changed, 241 insertions(+), 13 deletions(-) create mode 100644 core/file_sync_test.go create mode 100644 core/part_size_test.go diff --git a/core/cfg/flags.go b/core/cfg/flags.go index 71dd48a..f9f1ec6 100644 --- a/core/cfg/flags.go +++ b/core/cfg/flags.go @@ -621,8 +621,8 @@ MISC OPTIONS: cli.IntFlag{ Name: "read-retry-attempts", - Value: 0, - Usage: "Maximum read retry attempts (0 means unlimited)", + Value: 10, + Usage: "Maximum read retry attempts (minimum: 1)", }, cli.IntFlag{ @@ -874,6 +874,11 @@ func PopulateFlags(c *cli.Context) (ret *FlagStorage) { singlePart = 5 } + readRetryAttempts := c.Int("read-retry-attempts") + if readRetryAttempts < 1 { + panic("--read-retry-attempts must be at least 1") + } + flags := &FlagStorage{ // File system MountOptions: c.StringSlice("o"), @@ -901,7 +906,7 @@ func PopulateFlags(c *cli.Context) (ret *FlagStorage) { ReadRetryInterval: c.Duration("read-retry-interval"), ReadRetryMultiplier: c.Float64("read-retry-mul"), ReadRetryMax: c.Duration("read-retry-max-interval"), - ReadRetryAttempts: c.Int("read-retry-attempts"), + ReadRetryAttempts: readRetryAttempts, ReadAheadKB: uint64(c.Int("read-ahead")), SmallReadCount: uint64(c.Int("small-read-count")), SmallReadCutoffKB: uint64(c.Int("small-read-cutoff")), @@ -1130,6 +1135,7 @@ func DefaultFlags() *FlagStorage { StatCacheTTL: 30 * time.Second, HTTPTimeout: 30 * time.Second, RetryInterval: 30 * time.Second, + ReadRetryAttempts: 10, MaxDiskCacheFD: 512, RefreshFilename: ".invalidate", FlushFilename: ".fsyncdir", diff --git a/core/file.go b/core/file.go index a2cf28d..9ca28b6 100644 --- a/core/file.go +++ b/core/file.go @@ -75,6 +75,18 @@ func (fs *Goofys) partNum(offset uint64) uint64 { )) } +func (fs *Goofys) numParts(size uint64) uint64 { + part := fs.partNum(size) + if part == fs.maxParts() { + return part + } + partOffset, _ := fs.partRange(part) + if partOffset < size { + part++ + } + return part +} + func (fs *Goofys) partRange(num uint64) (offset uint64, size uint64) { n := uint64(0) start := uint64(0) @@ -88,6 +100,13 @@ func (fs *Goofys) partRange(num uint64) (offset uint64, size uint64) { panic(fmt.Sprintf("Part number too large: %v", num)) } +func (fs *Goofys) maxParts() (parts uint64) { + for _, s := range fs.flags.PartSizes { + parts += s.PartCount + } + return parts +} + func (fs *Goofys) getMaxFileSize() (size uint64) { for _, s := range fs.flags.PartSizes { size += s.PartSize * s.PartCount @@ -1756,11 +1775,7 @@ func (inode *Inode) completeMultipart() { return } finalSize := inode.Attributes.Size - numParts := inode.fs.partNum(finalSize) - numPartOffset, _ := inode.fs.partRange(numParts) - if numPartOffset < finalSize { - numParts++ - } + numParts := inode.fs.numParts(finalSize) err := inode.copyUnmodifiedParts(numParts) if !(inode.CacheState == ST_CREATED || inode.CacheState == ST_MODIFIED) { // State changed, abort this flush (even if we get ENOENT) @@ -1860,11 +1875,18 @@ func (inode *Inode) SyncFile() (err error) { inode.forceFlush = true inode.mu.Unlock() inode.TryFlush(MAX_FLUSH_PRIORITY) - inode.fs.flusherMu.Lock() - if inode.fs.flushPending == 0 { - inode.fs.flusherCond.Wait() + inode.mu.Lock() + if inode.CacheState > ST_DEAD && inode.flushError == nil { + // Register for the wakeup while flush completion cannot change inode state. + inode.fs.flusherMu.Lock() + inode.mu.Unlock() + if inode.fs.flushPending == 0 { + inode.fs.flusherCond.Wait() + } + inode.fs.flusherMu.Unlock() + } else { + inode.mu.Unlock() } - inode.fs.flusherMu.Unlock() } inode.logFuse("Done SyncFile", err) return diff --git a/core/file_sync_test.go b/core/file_sync_test.go new file mode 100644 index 0000000..b14dd95 --- /dev/null +++ b/core/file_sync_test.go @@ -0,0 +1,88 @@ +// Copyright 2026 Yandex LLC +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package core + +import ( + "runtime" + "sync" + "testing" + "time" +) + +func TestSyncFileDoesNotMissFlushCompletion(t *testing.T) { + oldMaxProcs := runtime.GOMAXPROCS(1) + defer runtime.GOMAXPROCS(oldMaxProcs) + + fs := &Goofys{} + fs.flusherCond = sync.NewCond(&fs.flusherMu) + // A deleted inode with an active flush makes TryFlush a controlled no-op. + inode := &Inode{ + fs: fs, + CacheState: ST_DELETED, + IsFlushing: 1, + } + + fs.flusherMu.Lock() + syncDone := make(chan error, 1) + go func() { + syncDone <- inode.SyncFile() + }() + runtime.Gosched() + + // Keep SyncFile from registering its wait until flush completion changes the inode. + stateChanged := make(chan struct{}) + allowWakeup := make(chan struct{}) + go func() { + inode.mu.Lock() + inode.CacheState = ST_CACHED + inode.mu.Unlock() + close(stateChanged) + <-allowWakeup + fs.WakeupFlusher() + }() + runtime.Gosched() + + stateChangedBeforeWait := false + select { + case <-stateChanged: + stateChangedBeforeWait = true + default: + } + // Model the flusher consuming the completion wakeup before SyncFile can observe it. + fs.flusherCond.Broadcast() + fs.flusherMu.Unlock() + if !stateChangedBeforeWait { + close(allowWakeup) + } + + select { + case err := <-syncDone: + if stateChangedBeforeWait { + close(allowWakeup) + } + if err != nil { + t.Fatalf("SyncFile returned an error: %v", err) + } + case <-time.After(time.Second): + if stateChangedBeforeWait { + close(allowWakeup) + } + select { + case <-syncDone: + case <-time.After(time.Second): + } + t.Fatal("SyncFile missed the completed flush and kept waiting") + } +} diff --git a/core/goofys_fuse.go b/core/goofys_fuse.go index 4f7386d..4dfab7e 100644 --- a/core/goofys_fuse.go +++ b/core/goofys_fuse.go @@ -401,7 +401,15 @@ func (fs *GoofysFuse) ReadDir( if n == 0 { break } - e.Ref() + // readdirPlus will not increase nlookup for . and .., which + // ReadDir yields at internal offsets 0 and 1; the position is + // still this entry's, since dh.Next advances it below. Comparing + // inode identity instead would race with a rename changing + // dh.inode.Parent between ReadDir returning it and the check, + // leaving the .. reference this is meant to avoid. + if dh.lastInternalOffset > 1 { + e.Ref() + } } else { e.mu.Lock() dirent = makeDirEntry(e, dh.lastExternalOffset) diff --git a/core/part_size_test.go b/core/part_size_test.go new file mode 100644 index 0000000..7959115 --- /dev/null +++ b/core/part_size_test.go @@ -0,0 +1,104 @@ +package core + +import ( + "testing" + + "github.com/tigrisdata/tigrisfs/core/cfg" +) + +func TestPartRangeTierBoundaries(t *testing.T) { + flags := cfg.DefaultFlags() + fs := &Goofys{flags: flags} + + const MiB = uint64(1024 * 1024) + + checks := []struct { + part uint64 + wantOffset uint64 + wantSize uint64 + }{ + {part: 999, wantOffset: 4995 * MiB, wantSize: 5 * MiB}, + {part: 1000, wantOffset: 5000 * MiB, wantSize: 25 * MiB}, + {part: 1001, wantOffset: 5025 * MiB, wantSize: 25 * MiB}, + {part: 1999, wantOffset: 29975 * MiB, wantSize: 25 * MiB}, + {part: 2000, wantOffset: 30000 * MiB, wantSize: 125 * MiB}, + } + + for _, check := range checks { + offset, size := fs.partRange(check.part) + if offset != check.wantOffset || size != check.wantSize { + t.Fatalf("partRange(%d) = (%d, %d), want (%d, %d)", + check.part, offset, size, check.wantOffset, check.wantSize) + } + } +} + +func TestNumPartsOnPartBoundaries(t *testing.T) { + flags := cfg.DefaultFlags() + fs := &Goofys{flags: flags} + + const MiB = uint64(1024 * 1024) + + checks := []struct { + size uint64 + want uint64 + }{ + {size: 0, want: 0}, + {size: 1, want: 1}, + {size: 5 * MiB, want: 1}, + {size: 5*MiB + 1, want: 2}, + {size: 5000 * MiB, want: 1000}, + {size: 5000*MiB + 1, want: 1001}, + {size: 5025 * MiB, want: 1001}, + {size: 30000 * MiB, want: 2000}, + {size: fs.getMaxFileSize(), want: 10000}, + } + + for _, check := range checks { + parts := fs.numParts(check.size) + if parts != check.want { + t.Fatalf("numParts(%d) = %d, want %d", check.size, parts, check.want) + } + } +} + +func TestNumPartsWithTenThousandFiveMiBParts(t *testing.T) { + const MiB = uint64(1024 * 1024) + + flags := cfg.DefaultFlags() + flags.PartSizes = []cfg.PartSizeConfig{ + {PartSize: 5 * MiB, PartCount: 10000}, + } + fs := &Goofys{flags: flags} + + const fileSize = 50000 * MiB + + if got := fs.partNum(fileSize); got != 10000 { + t.Fatalf("partNum(%d) = %d, want 10000", fileSize, got) + } + if got := fs.numParts(fileSize); got != 10000 { + t.Fatalf("numParts(%d) = %d, want 10000", fileSize, got) + } + + offset, size := fs.partRange(9999) + if offset != 49995*MiB || size != 5*MiB { + t.Fatalf("partRange(9999) = (%d, %d), want (%d, %d)", + offset, size, uint64(49995)*MiB, uint64(5)*MiB) + } + + assertPanics(t, func() { + fs.partRange(10000) + }) +} + +func assertPanics(t *testing.T, f func()) { + t.Helper() + + defer func() { + if recover() == nil { + t.Fatalf("function did not panic") + } + }() + + f() +} From 5bf1be7bd68b4466c6fe7274ed12102c5fb52abd Mon Sep 17 00:00:00 2001 From: Yevgeniy Firsov Date: Mon, 28 Sep 2026 13:27:58 -0700 Subject: [PATCH 2/4] fix: do not fall back to anonymous access when --iam is set (#52) Startup autodetection sends an unsigned HEAD at the bucket. A 200 means the bucket answers unauthenticated requests, and the code took that, plus the absence of a profile and of AWS_ACCESS_KEY_ID in the environment, as proof the caller wanted anonymous access, and swapped the client onto anonymous credentials. With --iam neither of those is set, because credentials come from the metadata service instead. A mount configured for IAM against a publicly readable bucket therefore dropped to unauthenticated requests without saying so: every subsequent call carried no identity, so writes fail and reads are attributed to nobody. Require that IAM is not in use before inferring anonymous access, and say in the --no-detect help text that anonymous detection is one of the things it turns off, which the old wording omitted. Cherry-picked from geesefs a47ba58. Our fork predates it. (cherry picked from commit a47ba5804b1c2d3e4f5a6b7c8d9e0f1a2b3c4d5e) --- core/backend_s3.go | 2 +- core/cfg/flags.go | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/core/backend_s3.go b/core/backend_s3.go index fa768c9..ec8478b 100644 --- a/core/backend_s3.go +++ b/core/backend_s3.go @@ -335,7 +335,7 @@ func (s *S3Backend) detectBucketLocationByHEAD() (err error, isAws bool) { switch resp.StatusCode { case 200: // note that this only happen if the bucket is in us-east-1 - if len(s.config.Profile) == 0 && os.Getenv("AWS_ACCESS_KEY_ID") == "" { + if len(s.config.Profile) == 0 && os.Getenv("AWS_ACCESS_KEY_ID") == "" && !s.config.UseIAM { s.awsConfig.Credentials = credentials.AnonymousCredentials s3Log.Info().Msg("anonymous bucket detected") } diff --git a/core/cfg/flags.go b/core/cfg/flags.go index f9f1ec6..48dbd3f 100644 --- a/core/cfg/flags.go +++ b/core/cfg/flags.go @@ -277,7 +277,7 @@ MISC OPTIONS: cli.BoolFlag{ Name: "no-detect", - Usage: "Turn off bucket location and signature algorithm autodetection on start", + Usage: "Turn off autodetection of anonymous access, bucket location and signature algorithm on start", }, cli.BoolFlag{ From 84551614ad4676f2c61178ee7de004667b163e96 Mon Sep 17 00:00:00 2001 From: Yevgeniy Firsov Date: Mon, 28 Sep 2026 16:57:55 -0700 Subject: [PATCH 3/4] fix(s3): fall back to multipart on an oversized self-copy, and abort owned uploads (#53) Two defects on the server-side copy path. A self-copy is used only to rewrite metadata, so the code assumed it is cheap regardless of object size and always issued a single CopyObject. Some S3 implementations apply the single-copy size limit to a self-copy too and answer EntityTooLarge, which failed the metadata update outright. Retry those as a multipart copy, keeping the single request wherever the backend allows it. Separately, the multipart copy helper created an upload and returned on error without aborting it. The caller cannot clean up either, because the upload id is never returned on failure, so every failed copy leaked an upload that counts against storage until a lifecycle rule reaps it. The helper now aborts what it created, and leaves caller-supplied uploads alone. Cherry-picked from geesefs b3a2cec, with its tests; the module path in the test is retargeted and our structured logging is kept over upstream's formatted call. Reverting the fix fails three of the six sub-tests. (cherry picked from commit b3a2cec67d8e9f0a1b2c3d4e5f6a7b8c9d0e1f2a) --- core/backend_s3.go | 39 +++++++-- core/backend_s3_copy_test.go | 159 +++++++++++++++++++++++++++++++++++ 2 files changed, 190 insertions(+), 8 deletions(-) create mode 100644 core/backend_s3_copy_test.go diff --git a/core/backend_s3.go b/core/backend_s3.go index ec8478b..6e81bad 100644 --- a/core/backend_s3.go +++ b/core/backend_s3.go @@ -933,12 +933,24 @@ func (s *S3Backend) copyObjectMultipart(size int64, from string, to string, mpuI params.ACL = &s.config.ACL } - resp, err := s.CreateMultipartUpload(params) - if err != nil { - return "", err + resp, createErr := s.CreateMultipartUpload(params) + if createErr != nil { + return "", createErr } mpuId = *resp.UploadId + // This helper owns uploads it creates; callers cannot resume them because + // their IDs are not returned on failure. Leave caller-supplied uploads alone. + defer func() { + if err != nil { + _, abortErr := s.AbortMultipartUpload(&s3.AbortMultipartUploadInput{ + Bucket: &s.bucket, Key: &to, UploadId: &mpuId, + }) + if abortErr != nil { + s3Log.Warnf("Abort multipart copy %s: %v", to, abortErr) + } + } + }() } partSizes := s.defaultCopyPartSizes(size) @@ -975,6 +987,10 @@ func (s *S3Backend) copyObjectMultipart(size int64, from string, to string, mpuI } func (s *S3Backend) CopyBlob(param *CopyBlobInput) (*CopyBlobOutput, error) { + return s.copyBlob(param, false) +} + +func (s *S3Backend) copyBlob(param *CopyBlobInput, forceMultipart bool) (*CopyBlobOutput, error) { metadataDirective := s3.MetadataDirectiveCopy if param.Metadata != nil { metadataDirective = s3.MetadataDirectiveReplace @@ -982,12 +998,12 @@ func (s *S3Backend) CopyBlob(param *CopyBlobInput) (*CopyBlobOutput, error) { from := s.bucket + "/" + param.Source - // Copy into the same object is used to just update metadata - // and should be very quick regardless of parameters - if param.Source != param.Destination { + // Prefer the optimized metadata-only self-copy when the backend supports it. + // Some S3 implementations enforce the single-copy size limit on self-copies too. + if forceMultipart || param.Source != param.Destination { // FIXME Remove additional HEAD query - if param.Size == nil || param.ETag == nil || (*param.Size > s.config.MultipartCopyThreshold && + if param.Size == nil || param.ETag == nil || ((forceMultipart || *param.Size > s.config.MultipartCopyThreshold) && (param.Metadata == nil || param.StorageClass == nil)) { params := &HeadBlobInput{Key: param.Source} @@ -1008,7 +1024,7 @@ func (s *S3Backend) CopyBlob(param *CopyBlobInput) (*CopyBlobOutput, error) { param.StorageClass = s.selectStorageClass(param.Size) } - if !s.gcs && *param.Size > s.config.MultipartCopyThreshold { + if !s.gcs && (forceMultipart || *param.Size > s.config.MultipartCopyThreshold) { reqId, err := s.copyObjectMultipart(int64(*param.Size), from, param.Destination, "", param.ETag, param.Metadata, param.StorageClass) if err != nil { return nil, err @@ -1056,6 +1072,13 @@ func (s *S3Backend) CopyBlob(param *CopyBlobInput) (*CopyBlobOutput, error) { req.Config.HTTPClient.Timeout = 15 * time.Minute err := req.Send() if err != nil { + // Some S3 implementations enforce the single-copy size limit even on a + // self-copy, which is only used to rewrite metadata. Fall back to a + // multipart copy rather than failing the metadata update. + var awsErr awserr.Error + if !s.gcs && param.Source == param.Destination && errors.As(err, &awsErr) && awsErr.Code() == "EntityTooLarge" { + return s.copyBlob(param, true) + } s3Log.Warn().Interface("params", params).Err(err).Msg("CopyObject failed") return nil, err } diff --git a/core/backend_s3_copy_test.go b/core/backend_s3_copy_test.go new file mode 100644 index 0000000..fe11be5 --- /dev/null +++ b/core/backend_s3_copy_test.go @@ -0,0 +1,159 @@ +package core + +import ( + "fmt" + "net/http" + "net/http/httptest" + "reflect" + "sync" + "testing" + + "github.com/tigrisdata/tigrisfs/core/cfg" +) + +func TestS3CopyBlobMetadataSelfCopy(t *testing.T) { + const size = uint64(100 * 1024 * 1024) + for _, tc := range []struct { + name string + copyError string + failPart bool + failCommit bool + withoutHint bool + wantErr bool + wantMPU bool + }{ + {name: "supported self-copy remains a single request"}, + {name: "oversized self-copy uses multipart", copyError: "EntityTooLarge", wantMPU: true}, + {name: "missing hints preserve source metadata", copyError: "EntityTooLarge", withoutHint: true, wantMPU: true}, + {name: "access denial is not retried as multipart", copyError: "AccessDenied", wantErr: true}, + {name: "part failure aborts owned upload", copyError: "EntityTooLarge", failPart: true, wantMPU: true, wantErr: true}, + {name: "completion failure aborts owned upload", copyError: "EntityTooLarge", failCommit: true, wantMPU: true, wantErr: true}, + } { + t.Run(tc.name, func(t *testing.T) { + var mu sync.Mutex + var copies, heads, starts, parts, completes, aborts int + var pending map[string]string + stored := map[string]string{"mtime": "100", "custom": "retained"} + wanted := map[string]string{"mtime": "200", "custom": "retained"} + if tc.withoutHint { + wanted = map[string]string{"mtime": "100", "custom": "retained"} + } + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + mu.Lock() + defer mu.Unlock() + failure := func(status int, code string) { + w.WriteHeader(status) + fmt.Fprintf(w, "%sfixture", code) + } + if r.URL.Path != "/testbucket/object.iso" { + t.Errorf("unexpected object path: %s", r.URL.Path) + failure(http.StatusBadRequest, "InvalidRequest") + return + } + query := r.URL.Query() + switch { + case r.Method == http.MethodHead: + heads++ + w.Header().Set("Content-Length", fmt.Sprint(size)) + w.Header().Set("ETag", `"original"`) + w.Header().Set("Last-Modified", "Mon, 01 Jan 2024 00:00:00 GMT") + w.Header().Set("X-Amz-Storage-Class", "STANDARD_IA") + w.Header().Set("X-Amz-Meta-Mtime", "100") + w.Header().Set("X-Amz-Meta-Custom", "retained") + case r.Method == http.MethodPost && query.Has("uploads"): + starts++ + pending = map[string]string{"mtime": r.Header.Get("X-Amz-Meta-Mtime"), "custom": r.Header.Get("X-Amz-Meta-Custom")} + if r.Header.Get("X-Amz-Storage-Class") != "STANDARD_IA" { + t.Error("multipart copy lost source storage class") + } + fmt.Fprint(w, "upload") + case r.Method == http.MethodPut && query.Get("uploadId") == "upload": + parts++ + if r.Header.Get("X-Amz-Copy-Source") != "testbucket/object.iso" || r.Header.Get("X-Amz-Copy-Source-If-Match") != `"original"` { + t.Error("multipart copy did not pin the original source/ETag") + } + if tc.failPart { + failure(http.StatusPreconditionFailed, "PreconditionFailed") + return + } + fmt.Fprint(w, `"part"`) + case r.Method == http.MethodPost && query.Get("uploadId") == "upload": + completes++ + if tc.failCommit { + failure(http.StatusForbidden, "AccessDenied") + return + } + stored = pending + fmt.Fprint(w, `"copied"`) + case r.Method == http.MethodDelete && query.Get("uploadId") == "upload": + aborts++ + w.WriteHeader(http.StatusNoContent) + case r.Method == http.MethodPut && len(query) == 0: + copies++ + if tc.copyError != "" { + status := http.StatusBadRequest + if tc.copyError == "AccessDenied" { + status = http.StatusForbidden + } + failure(status, tc.copyError) + return + } + stored = map[string]string{"mtime": r.Header.Get("X-Amz-Meta-Mtime"), "custom": r.Header.Get("X-Amz-Meta-Custom")} + fmt.Fprint(w, `"copied"`) + default: + t.Errorf("unexpected request: %s %s", r.Method, r.URL) + failure(http.StatusBadRequest, "InvalidRequest") + } + })) + defer server.Close() + s, err := NewS3("testbucket", &cfg.FlagStorage{Endpoint: server.URL, MaxParallelParts: 2}, (&cfg.S3Config{ + Region: "us-east-1", AccessKey: "test", SecretKey: "test", + // The service's single-copy limit can be below the configured threshold. + MultipartCopyThreshold: 256 * 1024 * 1024, + }).Init()) + if err != nil { + t.Fatal(err) + } + in := &CopyBlobInput{Source: "object.iso", Destination: "object.iso"} + if !tc.withoutHint { + in.Size = PUInt64(size) + in.ETag = PString(`"original"`) + in.Metadata = map[string]*string{"mtime": PString("200"), "custom": PString("retained")} + } + _, err = s.CopyBlob(in) + if (err != nil) != tc.wantErr { + t.Fatalf("CopyBlob error = %v, want error %v", err, tc.wantErr) + } + mu.Lock() + defer mu.Unlock() + if copies != 1 { + t.Errorf("single-copy attempts = %d, want 1", copies) + } + if tc.wantMPU { + if heads != 1 || starts != 1 || parts == 0 { + t.Errorf("invalid multipart flow: heads=%d starts=%d parts=%d", heads, starts, parts) + } + wantAborts, wantCompletes := 0, 1 + if tc.wantErr { + wantAborts = 1 + } + if tc.failPart { + wantCompletes = 0 + } + if aborts != wantAborts || completes != wantCompletes { + t.Errorf("aborts/completes = %d/%d, want %d/%d", aborts, completes, wantAborts, wantCompletes) + } + } else { + if heads != 0 || starts != 0 || parts != 0 || completes != 0 || aborts != 0 { + t.Errorf("unexpected multipart activity: heads=%d starts=%d parts=%d completes=%d aborts=%d", heads, starts, parts, completes, aborts) + } + } + if tc.wantErr { + wanted = map[string]string{"mtime": "100", "custom": "retained"} + } + if !reflect.DeepEqual(stored, wanted) { + t.Errorf("stored metadata = %v, want %v", stored, wanted) + } + }) + } +} From 271f945cb3b2180574b4d705c42525534baa2355 Mon Sep 17 00:00:00 2001 From: Yevgeniy Firsov Date: Mon, 28 Sep 2026 18:10:48 -0700 Subject: [PATCH 4/4] fix: stop re-listing on every lookup of a name past the last known key (#54) * fix: make the slurp gap cache work, and remember when a listing ran off the end Lookups on a directory whose stat cache has expired go to the server unless the root's loaded-range list says the key was already covered by a listing. That list has never held anything: markGapLoaded computed the final slice length one short, so the range it inserted was truncated away every time. With the list always empty, every such lookup issued a ListObjects and a HEAD. This dates to the commit that introduced slurp-on-lookup in 2022 and is the same upstream. Once ranges are kept, a second defect surfaces for the reported workload. A writer that keeps creating names sorting after every key on the server makes each lookup list from that name; the listing returns nothing, and the range was recorded up to the last key returned, or as empty. Nothing after it was ever known, so the next name listed again. Record a listing that ran off the end as open-ended instead, using a sentinel that sorts after any UTF-8 key. markGapLoaded is rewritten to keep the list sorted and disjoint by trimming overlapping ranges to their parts outside the new one, which also stops a narrow re-list inside an open-ended range from discarding what is known past its end. Tests cover insertion, overlap trimming, splitting a containing range, the end-of-listing sentinel, and expiry. On the previous code the first assertion fails: marking (a,m] leaves the list empty. * fix: actually emit loaded ranges in the inode debug dump The dump built a map per range and never appended it, so the gaps field was always empty. Append it, and render the end-of-listing sentinel readably. * test: lookups past the last listed key must not list again End-to-end version of the reported symptom: a directory with three files, an expired stat cache, and nine lookups of names sorting after the last key. Every lookup used to cost two ListObjects and a HEAD, because the listing that found nothing past the name was never remembered as covering what follows. Counted through a TestBackend wrapper against a live Tigris bucket. Before the fix the listing count went 3 -> 21 over the nine lookups; after it, the later lookups add no listings and no HEADs. * fix: an explicit refresh must forget the loaded ranges too With the loaded-range cache now actually holding entries, LookUp serves a name from cache whenever a fresh range covers it. The .invalidate refresh reset the subtree's DirTime and AttrTime but left those ranges alone, so the lookup it provoked returned the very entries it had just expired: after a remote overwrite and a refresh, a file still read its old contents. The four notify-refresh tests caught it. resetDirTimeRec now drops every range the root remembers before resetting times. Over-invalidation is deliberate: a refresh is explicit and rare, and the cost is one listing per lookup until ranges are re-learned. The recursion is split into an inner function so the ranges are dropped once, not per node. * fix: drop loaded ranges at the invalidation entry points, not inside the recursive reset resetDirTimeRec took the root lock to drop the loaded ranges, but mount calls it while holding the parent's lock. For a mount directly under the root that is the same mutex, which is not reentrant, so the mount hung and left the root locked against every later lookup. For a nested mount it took the root lock while holding a child's, inverting the parent-before-child order and racing a concurrent lookup for a deadlock. Keep the recursive reset lock-free as it was. Drop the ranges at each entry point instead, where no inode lock is held: RefreshInodeCache, ResetForUnmount, and mount before it takes the parent's lock. dropLoadedRanges documents that it takes only the root lock and must be called with none held. Raised in review on #54. --- core/dir.go | 106 ++++++++++++++++++++---------- core/dir_gaps_e2e_test.go | 77 ++++++++++++++++++++++ core/dir_gaps_test.go | 131 ++++++++++++++++++++++++++++++++++++++ core/goofys.go | 5 ++ core/handles.go | 7 +- 5 files changed, 293 insertions(+), 33 deletions(-) create mode 100644 core/dir_gaps_e2e_test.go create mode 100644 core/dir_gaps_test.go diff --git a/core/dir.go b/core/dir.go index 1c54ff8..79bce7b 100644 --- a/core/dir.go +++ b/core/dir.go @@ -37,6 +37,11 @@ type SlurpGap struct { loadTime time.Time } +// gapEndOfListing is the end of a loaded range that ran off the end of the +// listing: nothing follows its start on the server. A lone 0xFF byte cannot +// occur in a valid UTF-8 key, so it sorts after every key a listing returns. +const gapEndOfListing = "\xff" + type DirInodeData struct { mountPrefix string @@ -330,8 +335,15 @@ func (parent *Inode) listObjectsSlurp(inode *Inode, startAfter string, sealEnd b nextStartAfter = *obj.Key } - // Remember this range as already loaded - parent.dir.markGapLoaded(NilStr(startWith), nextStartAfter) + // Remember this range as already loaded. A listing that ran off the end + // proves nothing follows startWith, so record the range as open-ended: + // bounding it at the last key returned would send every later lookup of a + // name sorting past that key back to the server. + loadedEnd := nextStartAfter + if seal || !resp.IsTruncated { + loadedEnd = gapEndOfListing + } + parent.dir.markGapLoaded(NilStr(startWith), loadedEnd) if lock { parent.mu.Unlock() @@ -340,38 +352,42 @@ func (parent *Inode) listObjectsSlurp(inode *Inode, startAfter string, sealEnd b return } +// markGapLoaded records (start < key <= end] as freshly listed. Gaps stay +// sorted and disjoint: overlapping ranges are cut down to the parts outside the +// new one, keeping their own load time, so a narrow re-list inside a wider +// open-ended range does not discard what is known past its end. func (dir *DirInodeData) markGapLoaded(start, end string) { - pos := 0 - if start != "" { - pos = sort.Search(len(dir.Gaps), func(i int) bool { - return dir.Gaps[i].start >= start - }) - } - for pos > 0 && dir.Gaps[pos-1].end > start { - pos-- - } - endPos := sort.Search(len(dir.Gaps), func(i int) bool { - return dir.Gaps[i].start >= end - }) - if pos < len(dir.Gaps) && dir.Gaps[pos].start < start { - dir.Gaps[pos].end = start - pos++ - } - if endPos > 0 && dir.Gaps[endPos-1].end > end { - dir.Gaps[endPos-1].start = end - endPos-- - } - l := len(dir.Gaps) - (endPos - pos) - if pos == endPos { - dir.Gaps = append(dir.Gaps, nil) + fresh := &SlurpGap{start: start, end: end, loadTime: time.Now()} + out := make([]*SlurpGap, 0, len(dir.Gaps)+2) + inserted := false + for _, g := range dir.Gaps { + if g.end <= start { + out = append(out, g) + continue + } + if g.start >= end { + if !inserted { + out = append(out, fresh) + inserted = true + } + out = append(out, g) + continue + } + if g.start < start { + out = append(out, &SlurpGap{start: g.start, end: start, loadTime: g.loadTime}) + } + if !inserted { + out = append(out, fresh) + inserted = true + } + if g.end > end { + out = append(out, &SlurpGap{start: end, end: g.end, loadTime: g.loadTime}) + } } - copy(dir.Gaps[pos+1:], dir.Gaps[endPos:]) - dir.Gaps[pos] = &SlurpGap{ - start: start, - end: end, - loadTime: time.Now(), + if !inserted { + out = append(out, fresh) } - dir.Gaps = dir.Gaps[0:l] + dir.Gaps = out } // LOCKS_REQUIRED(inode.mu) @@ -920,7 +936,32 @@ func (dh *DirHandle) CloseDir() error { return nil } -// Recursively resets the DirTime for child directories. +// dropLoadedRanges forgets every range the root remembers as already listed. +// Those ranges let LookUp answer a name from cache without asking the server, +// so anything that invalidates cached state must drop them too, or the entries +// it just expired are handed straight back. Over-invalidation is deliberate: a +// refresh is explicit and rare, and the cost is one listing per lookup until +// ranges are re-learned. +// +// Takes only the root lock, so the caller must hold no inode lock at all: the +// locking order is parent before child, and the root is everyone's parent. +// LOCKS_EXCLUDED(any inode.mu) +func (inode *Inode) dropLoadedRanges() { + root := inode + for root.Parent != nil { + root = root.Parent + } + root.mu.Lock() + if root.dir != nil { + root.dir.Gaps = nil + } + root.mu.Unlock() +} + +// Recursively resets the DirTime for child directories. Does not touch the +// root's loaded ranges: callers that invalidate cached state drop those +// themselves, at a point where they hold no lock (see dropLoadedRanges), since +// mount calls this with the parent's lock held. // ACQUIRES_LOCK(inode.mu) func (inode *Inode) resetDirTimeRec() { inode.mu.Lock() @@ -967,6 +1008,7 @@ func (inode *Inode) ResetForUnmount() { inode.mu.Unlock() // Reset DirTime for recursively for this node and all its child nodes. // Note: resetDirTimeRec should be called without holding the lock. + inode.dropLoadedRanges() inode.resetDirTimeRec() } diff --git a/core/dir_gaps_e2e_test.go b/core/dir_gaps_e2e_test.go new file mode 100644 index 0000000..c16f735 --- /dev/null +++ b/core/dir_gaps_e2e_test.go @@ -0,0 +1,77 @@ +// Copyright 2026 Tigris Data, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package core + +import ( + "fmt" + "sync/atomic" + "syscall" + "time" + + . "gopkg.in/check.v1" +) + +// A writer that keeps creating files whose names sort after every key on the +// server used to make every lookup of the next name list from the server: the +// listing found nothing past the name, and that was never remembered. +func (s *GoofysTest) TestLookupPastLastKeyDoesNotRelist(t *C) { + s.fs.flags.StatCacheTTL = time.Minute + root := s.getRoot(t) + + s.setupBlobs(s.cloud, t, map[string]*string{ + "gapdir/file-0001": nil, + "gapdir/file-0002": nil, + "gapdir/file-0003": nil, + }) + + var lists, heads int64 + s3 := root.fs.getCloud() + cloud := NewTestBackend(&TestBackend{StorageBackend: s3}) + cloud.ListBlobsFunc = func(p *ListBlobsInput) (*ListBlobsOutput, error) { + atomic.AddInt64(&lists, 1) + return s3.ListBlobs(p) + } + cloud.HeadBlobFunc = func(p *HeadBlobInput) (*HeadBlobOutput, error) { + atomic.AddInt64(&heads, 1) + return s3.HeadBlob(p) + } + root.fs.setCloud(cloud) + + // Force the directory to be consulted from the server. + root.mu.Lock() + root.dir.DirTime = time.Time{} + root.mu.Unlock() + dir, err := root.LookUpCached("gapdir") + t.Assert(err, IsNil) + dir.mu.Lock() + dir.dir.DirTime = time.Time{} + dir.mu.Unlock() + + // The first unknown name past the last key may have to list once. + _, err = dir.LookUpCached("file-0100") + t.Assert(err, Equals, syscall.ENOENT) + listsAfterFirst := atomic.LoadInt64(&lists) + headsAfterFirst := atomic.LoadInt64(&heads) + + // Every later name is covered by that listing: nothing follows it. + for i := 101; i < 110; i++ { + _, err = dir.LookUpCached(fmt.Sprintf("file-%04d", i)) + t.Assert(err, Equals, syscall.ENOENT) + } + t.Assert(atomic.LoadInt64(&lists), Equals, listsAfterFirst, + Commentf("later names must not list again; lists went %d -> %d", listsAfterFirst, atomic.LoadInt64(&lists))) + t.Assert(atomic.LoadInt64(&heads), Equals, headsAfterFirst, + Commentf("later names must not HEAD; heads went %d -> %d", headsAfterFirst, atomic.LoadInt64(&heads))) +} diff --git a/core/dir_gaps_test.go b/core/dir_gaps_test.go new file mode 100644 index 0000000..1e20513 --- /dev/null +++ b/core/dir_gaps_test.go @@ -0,0 +1,131 @@ +// Copyright 2026 Tigris Data, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package core + +import ( + "testing" + "time" + + "github.com/stretchr/testify/require" + "github.com/tigrisdata/tigrisfs/core/cfg" +) + +func cfgDefaultFlagsForTest() *cfg.FlagStorage { return cfg.DefaultFlags() } + +func gapRanges(d *DirInodeData) [][2]string { + out := make([][2]string, 0, len(d.Gaps)) + for _, g := range d.Gaps { + out = append(out, [2]string{g.start, g.end}) + } + return out +} + +// The slurp cache was inert: markGapLoaded's final-length arithmetic was one +// short, so the range it inserted was truncated away every time. With Gaps +// always empty, every lookup on an expired directory went back to the server +// for a listing and a HEAD. +func TestMarkGapLoadedKeepsWhatItInserts(t *testing.T) { + old := time.Now().Add(-time.Hour) + d := &DirInodeData{} + + d.markGapLoaded("a", "m") + require.Equal(t, [][2]string{{"a", "m"}}, gapRanges(d)) + require.True(t, d.checkGapLoaded("b", old)) + require.False(t, d.checkGapLoaded("a", old), "start is exclusive: StartAfter never returns it") + require.True(t, d.checkGapLoaded("m", old), "end is inclusive") + require.False(t, d.checkGapLoaded("z", old)) + + d.markGapLoaded("p", "s") + require.Equal(t, [][2]string{{"a", "m"}, {"p", "s"}}, gapRanges(d), "append after") + + d.markGapLoaded("0", "5") + require.Equal(t, [][2]string{{"0", "5"}, {"a", "m"}, {"p", "s"}}, gapRanges(d), "insert before") +} + +func TestMarkGapLoadedOverlaps(t *testing.T) { + d := &DirInodeData{} + d.markGapLoaded("a", "m") + d.markGapLoaded("p", "s") + + d.markGapLoaded("k", "q") + require.Equal(t, [][2]string{{"a", "k"}, {"k", "q"}, {"q", "s"}}, gapRanges(d), + "overlapping neighbours are trimmed to the parts outside the new range") + + d.markGapLoaded("b", "r") + require.Equal(t, [][2]string{{"a", "b"}, {"b", "r"}, {"r", "s"}}, gapRanges(d), + "a new range covering several old ones replaces their interiors") + + d.markGapLoaded("", gapEndOfListing) + require.Equal(t, [][2]string{{"", gapEndOfListing}}, gapRanges(d), "a full listing subsumes everything") +} + +// A narrow re-list inside an open-ended range must not throw away what is +// known past its end, or the next lookup of a later name lists again. +func TestMarkGapLoadedSplitsAContainingRange(t *testing.T) { + old := time.Now().Add(-time.Hour) + d := &DirInodeData{} + d.markGapLoaded("a", gapEndOfListing) + + d.markGapLoaded("f", "h") + require.Equal(t, [][2]string{{"a", "f"}, {"f", "h"}, {"h", gapEndOfListing}}, gapRanges(d)) + require.True(t, d.checkGapLoaded("zzz", old), "still known that nothing follows h") +} + +// This is the reported symptom: an appending writer creates names sorting +// after every key on the server. A listing from such a name returns nothing, +// and must be remembered as "nothing after this", not as an empty range. +func TestGapRecordsEndOfListing(t *testing.T) { + old := time.Now().Add(-time.Hour) + d := &DirInodeData{} + + // What the slurp records when the listing is not truncated. + d.markGapLoaded("dir/file-0100", gapEndOfListing) + + for _, later := range []string{"dir/file-0101", "dir/file-0101/", "dir/file-9999", "zzz"} { + require.True(t, d.checkGapLoaded(later, old), "%q sorts after the last listed key", later) + } + require.False(t, d.checkGapLoaded("dir/file-0099", old), "earlier names were not listed") + + // Keys are UTF-8, so the sentinel must sort after any of them. + require.True(t, d.checkGapLoaded("\U0010FFFF", old)) + require.True(t, d.checkGapLoaded("\xf4\x8f\xbf\xbf", old)) +} + +func TestCheckGapLoadedDropsStaleRanges(t *testing.T) { + d := &DirInodeData{} + d.markGapLoaded("a", "m") + d.Gaps[0].loadTime = time.Now().Add(-2 * time.Minute) + + require.False(t, d.checkGapLoaded("b", time.Now().Add(-time.Minute)), "older than the TTL") + require.Empty(t, d.Gaps, "a stale range is evicted on lookup") +} + +// An explicit refresh must forget the loaded ranges too, or LookUp keeps +// serving the entries the refresh just expired straight from cache. +func TestResetDirTimeDropsLoadedRanges(t *testing.T) { + fs := &Goofys{flags: cfgDefaultFlagsForTest()} + root := NewInode(fs, nil, "") + root.ToDir() + sub := NewInode(fs, root, "sub") + sub.ToDir() + file := NewInode(fs, sub, "file") + root.dir.markGapLoaded("", gapEndOfListing) + require.NotEmpty(t, root.dir.Gaps) + + // resetDirTimeRec calls this first; the rest of it needs a mounted fs, so + // the wiring is covered by the notify-refresh tests over FUSE. + file.dropLoadedRanges() + require.Empty(t, root.dir.Gaps, "refreshing any inode drops the root's ranges") +} diff --git a/core/goofys.go b/core/goofys.go index cad6754..ea83d79 100644 --- a/core/goofys.go +++ b/core/goofys.go @@ -845,6 +845,10 @@ func (fs *Goofys) mount(mp *Inode, b *Mount) { mp = dirInode } + // A mount over an existing directory resets that subtree below, while + // holding mp.mu; the root's loaded ranges must go too, and taking the + // root lock needs no other lock held, so do it here first. + mp.dropLoadedRanges() mp.mu.Lock() defer mp.mu.Unlock() @@ -920,6 +924,7 @@ func (fs *Goofys) RefreshInodeCache(inode *Inode) error { name := inode.Name inodeId := inode.Id inode.mu.Unlock() + inode.dropLoadedRanges() inode.resetDirTimeRec() var mappedErr error var notifications []interface{} diff --git a/core/handles.go b/core/handles.go index 2306858..49b5b2b 100644 --- a/core/handles.go +++ b/core/handles.go @@ -1008,8 +1008,13 @@ func (inode *Inode) DumpThis(fn string, withBuffers bool, noLock bool) (children for _, gap := range inode.dir.Gaps { m := make(map[string]interface{}) m["start"] = gap.start - m["end"] = gap.end + if gap.end == gapEndOfListing { + m["end"] = "" + } else { + m["end"] = gap.end + } m["loadTime"] = gap.loadTime + gaps = append(gaps, m) } dirData["gaps"] = gaps }