From 16d7b4be592075372c751d5ce75edcb0cd1fb71d Mon Sep 17 00:00:00 2001 From: Michael Habinsky Date: Fri, 23 Oct 2026 16:01:56 -0600 Subject: [PATCH 1/6] Release handle resources in all components and simplify logging (push it up to libfuse) --- component/block_cache/block_cache.go | 8 ++-- component/file_cache/file_cache.go | 41 ++++-------------- component/libfuse/libfuse2_handler.go | 8 +++- .../libfuse/libfuse2_handler_test_wrapper.go | 4 ++ component/loopback/loopback_fs.go | 10 +---- component/stream/read_write.go | 43 +++---------------- component/stream/read_write_filename.go | 11 +++-- component/xload/xload.go | 2 +- 8 files changed, 37 insertions(+), 90 deletions(-) diff --git a/component/block_cache/block_cache.go b/component/block_cache/block_cache.go index 0da2b9f72..9775920e7 100644 --- a/component/block_cache/block_cache.go +++ b/component/block_cache/block_cache.go @@ -621,21 +621,21 @@ func (bc *BlockCache) releaseFileInternal(options internal.ReleaseFileOptions) e defer bc.fileCloseOpt.Done() + var err error if options.Handle.Dirty() { log.Info( "BlockCache::CloseFile : name=%s, handle=%d dirty. Flushing the file.", options.Handle.Path, options.Handle.ID, ) - err := bc.FlushFile( + err = bc.FlushFile( internal.FlushFileOptions{Handle: options.Handle, CloseInProgress: true}, - ) //nolint + ) if err != nil { log.Err( "BlockCache::ReleaseFileInternal : failed to flush file %s", options.Handle.Path, ) - return err } } @@ -669,7 +669,7 @@ func (bc *BlockCache) releaseFileInternal(options internal.ReleaseFileOptions) e } options.Handle.Buffers.Cooked = nil - return nil + return err } func (bc *BlockCache) getBlockSize(fileSize uint64, block *Block) uint64 { diff --git a/component/file_cache/file_cache.go b/component/file_cache/file_cache.go index 5b4976b79..c2913f7c5 100644 --- a/component/file_cache/file_cache.go +++ b/component/file_cache/file_cache.go @@ -1833,40 +1833,17 @@ func (fc *FileCache) releaseFileInternal( // if file has not been interactively read or written to by end user, then there is no cached file to close. _, noCachedHandle := options.Handle.GetValue("openFileOptions") + var err error if !noCachedHandle { // update the cache policy _ = fc.FileUsed(options.Handle.Path) - // sync - if options.Handle.Dirty() { - // flush the local file - err := fc.flushFileLocal(options.Handle) - if err != nil { - return err - } - // upload - // flock is already locked - flushOptions := internal.FlushFileOptions{Handle: options.Handle, CloseInProgress: true} - err = fc.flushFileCloud(flushOptions) - if err != nil { - return err - } - } - // close - f := options.Handle.GetFileObject() - if f == nil { - log.Err("FileCache::releaseFileInternal : %s missing fd in handle", options.Handle.Path) - return syscall.EBADF - } - err := f.Close() - if err != nil { - log.Err( - "FileCache::releaseFileInternal : %s (%d) close failed [%v]", - options.Handle.Path, - int(f.Fd()), - err, - ) - return err - } + // upload (flock is already locked) + flushOptions := internal.FlushFileOptions{Handle: options.Handle, CloseInProgress: true} + err = fc.flushFileCloud(flushOptions) + // clears the flag if the upload failed, since the handle is going away + fc.clearHandleDirty(options.Handle) + // close the local file + err = errors.Join(err, options.Handle.GetFileObject().Close()) } flock.Dec() @@ -1876,7 +1853,7 @@ func (fc *FileCache) releaseFileInternal( flock.LazyOpen = false } - return nil + return err } // ReadInBuffer: Read the local file into a buffer diff --git a/component/libfuse/libfuse2_handler.go b/component/libfuse/libfuse2_handler.go index 5d507a63d..368b5d4ab 100644 --- a/component/libfuse/libfuse2_handler.go +++ b/component/libfuse/libfuse2_handler.go @@ -887,6 +887,11 @@ func fuseErrnoFromError(err error) int { return -fuse.EACCES case errors.Is(err, fs.ErrExist): return -fuse.EEXIST + // os.File methods return fs.ErrInvalid on a nil receiver and fs.ErrClosed after Close + case errors.Is(err, syscall.EBADF), + errors.Is(err, fs.ErrInvalid), + errors.Is(err, fs.ErrClosed): + return -fuse.EBADF default: return -fuse.EIO } @@ -1005,7 +1010,6 @@ func (cf *CgofuseFS) Release(path string, fh uint64) int { handle.ID, err.Error(), ) - return fuseErrnoFromError(err) } handlemap.Delete(handle.ID) @@ -1013,7 +1017,7 @@ func (cf *CgofuseFS) Release(path string, fh uint64) int { // decrement open file handles count libfuseStatsCollector.UpdateStats(stats_manager.Decrement, openHandles, (int64)(1)) - return 0 + return fuseErrnoFromError(err) } // Unlink deletes a file. diff --git a/component/libfuse/libfuse2_handler_test_wrapper.go b/component/libfuse/libfuse2_handler_test_wrapper.go index 3f1328158..f7cfe921c 100644 --- a/component/libfuse/libfuse2_handler_test_wrapper.go +++ b/component/libfuse/libfuse2_handler_test_wrapper.go @@ -356,6 +356,10 @@ func testFuseErrnoFromError(suite *libfuseTestSuite) { {err: syscall.EEXIST, expected: -fuse.EEXIST}, {err: fs.ErrExist, expected: -fuse.EEXIST}, {err: os.ErrExist, expected: -fuse.EEXIST}, + {err: syscall.EBADF, expected: -fuse.EBADF}, + {err: os.ErrInvalid, expected: -fuse.EBADF}, + {err: (*os.File)(nil).Close(), expected: -fuse.EBADF}, + {err: &fs.PathError{Op: "close", Err: os.ErrClosed}, expected: -fuse.EBADF}, {err: syscall.EIO, expected: -fuse.EIO}, {err: errors.New("boom"), expected: -fuse.EIO}, } diff --git a/component/loopback/loopback_fs.go b/component/loopback/loopback_fs.go index 4e0287747..e745e0e15 100644 --- a/component/loopback/loopback_fs.go +++ b/component/loopback/loopback_fs.go @@ -33,7 +33,6 @@ import ( "path/filepath" "runtime" "strings" - "syscall" "github.com/Seagate/cloudfuse/common" "github.com/Seagate/cloudfuse/common/config" @@ -252,14 +251,9 @@ func (lfs *LoopbackFS) OpenFile(options internal.OpenFileOptions) (*handlemap.Ha func (lfs *LoopbackFS) ReleaseFile(options internal.ReleaseFileOptions) error { log.Trace("LoopbackFS::ReleaseFile : name=%s", options.Handle.Path) - f := options.Handle.GetFileObject() - if f == nil { - log.Err("LoopbackFS::ReleaseFile : error [file not available]") - return syscall.EBADF - } - handlemap.Delete(options.Handle.ID) - return f.Close() + + return options.Handle.GetFileObject().Close() } func (lfs *LoopbackFS) RenameFile(options internal.RenameFileOptions) error { diff --git a/component/stream/read_write.go b/component/stream/read_write.go index 656e63f5f..76f9f9294 100644 --- a/component/stream/read_write.go +++ b/component/stream/read_write.go @@ -271,34 +271,12 @@ func (rw *ReadWriteCache) FlushFile(options internal.FlushFileOptions) error { func (rw *ReadWriteCache) ReleaseFile(options internal.ReleaseFileOptions) error { log.Trace("Stream::ReleaseFile : name=%s, handle=%d", options.Handle.Path, options.Handle.ID) // try to flush again to make sure it's cleaned up - err := rw.FlushFile(internal.FlushFileOptions{Handle: options.Handle}) - if err != nil { - log.Err( - "Stream::ReleaseFile : error flushing file %s [%s]", - options.Handle.Path, - err.Error(), - ) - return err - } + flushErr := rw.FlushFile(internal.FlushFileOptions{Handle: options.Handle}) if !rw.StreamOnly && !options.Handle.CacheObj.StreamOnly { - err = rw.purge(options.Handle, -1) - if err != nil { - log.Err( - "Stream::ReleaseFile : error purging file %s [%s]", - options.Handle.Path, - err.Error(), - ) - } - } - err = rw.NextComponent().ReleaseFile(options) - if err != nil { - log.Err( - "Stream::ReleaseFile : error releasing file %s [%s]", - options.Handle.Path, - err.Error(), - ) + // TODO: if the flush failed, this discards dirty blocks and the unflushed data is lost + rw.purge(options.Handle, -1) } - return err + return errors.Join(flushErr, rw.NextComponent().ReleaseFile(options)) } func (rw *ReadWriteCache) DeleteFile(options internal.DeleteFileOptions) error { @@ -396,15 +374,7 @@ func (rw *ReadWriteCache) Stop() error { handleMap.Range(func(key, value any) bool { handle := value.(*handlemap.Handle) if handle.CacheObj != nil && !handle.CacheObj.StreamOnly { - err := rw.purge(handle, -1) - if err != nil { - log.Err( - "Stream::Stop : failed to purge handle cache %s [%s]", - handle.Path, - err.Error(), - ) - return false - } + rw.purge(handle, -1) } return true }) @@ -417,7 +387,7 @@ func (rw *ReadWriteCache) GetAttr(options internal.GetAttrOptions) (*internal.Ob return rw.NextComponent().GetAttr(options) } -func (rw *ReadWriteCache) purge(handle *handlemap.Handle, size int64) error { +func (rw *ReadWriteCache) purge(handle *handlemap.Handle, size int64) { handle.CacheObj.Lock() defer handle.CacheObj.Unlock() handle.CacheObj.Purge() @@ -427,7 +397,6 @@ func (rw *ReadWriteCache) purge(handle *handlemap.Handle, size int64) error { } handle.CacheObj.StreamOnly = true atomic.AddInt32(&rw.CachedObjects, -1) - return nil } func (rw *ReadWriteCache) createHandleCache(handle *handlemap.Handle) error { diff --git a/component/stream/read_write_filename.go b/component/stream/read_write_filename.go index 7d5aa9f24..2e8dc71e3 100644 --- a/component/stream/read_write_filename.go +++ b/component/stream/read_write_filename.go @@ -216,19 +216,18 @@ func (rw *ReadWriteFilenameCache) RenameFile(options internal.RenameFileOptions) func (rw *ReadWriteFilenameCache) ReleaseFile(options internal.ReleaseFileOptions) error { log.Trace("Stream::ReleaseFile : name=%s, handle=%d", options.Handle.Path, options.Handle.ID) // try to flush again to make sure it's cleaned up - err := rw.FlushFile(internal.FlushFileOptions{Handle: options.Handle}) - if err != nil { + flushErr := rw.FlushFile(internal.FlushFileOptions{Handle: options.Handle}) + if flushErr != nil { log.Err( "Stream::ReleaseFile : error flushing file %s [%s]", options.Handle.Path, - err.Error(), + flushErr.Error(), ) - return err } if !rw.StreamOnly { rw.purge(options.Handle.Path, true) } - err = rw.NextComponent().ReleaseFile(options) + err := rw.NextComponent().ReleaseFile(options) if err != nil { log.Err( "Stream::ReleaseFile : error releasing file %s [%s]", @@ -236,7 +235,7 @@ func (rw *ReadWriteFilenameCache) ReleaseFile(options internal.ReleaseFileOption err.Error(), ) } - return err + return errors.Join(flushErr, err) } func (rw *ReadWriteFilenameCache) FlushFile(options internal.FlushFileOptions) error { diff --git a/component/xload/xload.go b/component/xload/xload.go index 18331d064..e46d35d3b 100644 --- a/component/xload/xload.go +++ b/component/xload/xload.go @@ -517,7 +517,7 @@ func (xl *Xload) ReleaseFile(options internal.ReleaseFileOptions) error { defer flock.Unlock() flock.Dec() - return nil + return options.Handle.GetFileObject().Close() } // ------------------------- Factory ------------------------------------------- From d19acb755ac17d30137e95df3e1b3e5dece4e115 Mon Sep 17 00:00:00 2001 From: Michael Habinsky Date: Fri, 23 Oct 2026 23:05:01 -0600 Subject: [PATCH 2/6] Retry failed uploads on close --- component/file_cache/async.go | 53 +++++++++++++++++++----------- component/file_cache/file_cache.go | 23 +++++++------ 2 files changed, 44 insertions(+), 32 deletions(-) diff --git a/component/file_cache/async.go b/component/file_cache/async.go index 847593484..87d6fb810 100644 --- a/component/file_cache/async.go +++ b/component/file_cache/async.go @@ -30,6 +30,7 @@ import ( "fmt" "os" "path/filepath" + "syscall" "time" "github.com/Seagate/cloudfuse/common" @@ -233,12 +234,6 @@ func (fc *FileCache) addPendingOp(name string, value pendingFlags) { } } -// syncsPendingOps : whether servicePendingOps is running. Queueing an op without it -// would pin the cached file forever, since eviction skips pending ops. -func (fc *FileCache) syncsPendingOps() bool { - return len(fc.schedule) > 0 || fc.offlineAccess -} - // persistent background thread function func (fc *FileCache) servicePendingOps() { // grows while upload cycles keep failing, resets as soon as one succeeds @@ -307,12 +302,11 @@ func (fc *FileCache) runPendingOpCycle() (int, error) { case <-fc.uploadWindow(): name := key.(string) numFilesProcessed++ - if !fc.updateObject(name, value.(pendingFlags)) { - // record the failure so the caller backs off instead of busy-retrying - cycleErr = fmt.Errorf("sync failed for %s", name) - return false - } - return true + cycleErr = errors.Join(cycleErr) + err := fc.updateObject(name, value.(pendingFlags)) + // record the failure so the caller backs off instead of busy-retrying + cycleErr = errors.Join(cycleErr, err) + return continueOnError(err) default: return false // upload window ended } @@ -321,9 +315,18 @@ func (fc *FileCache) runPendingOpCycle() (int, error) { return numFilesProcessed, cycleErr } +// continueOnError : whether err only affects one object, so the rest of the queue can still sync +func continueOnError(err error) bool { + return err == nil || + // ENOSPC won't be returned on deletions which could be queued + errors.Is(err, syscall.ENOSPC) || + // s3storage KeyTooLongError: permanent for this key only + errors.Is(err, syscall.ENAMETOOLONG) +} + // synchronize pending operation with cloud storage -// returns false if we should stop iterating (offline, etc.) -func (fc *FileCache) updateObject(name string, flags pendingFlags) bool { +// returns the cloud error if the op stays pending +func (fc *FileCache) updateObject(name string, flags pendingFlags) error { log.Trace("FileCache::updateObject : %s", name) // lock the file @@ -334,7 +337,7 @@ func (fc *FileCache) updateObject(name string, flags pendingFlags) bool { // don't double upload _, stillPending := fc.pendingOps.Load(name) if !stillPending { - return true + return nil } // look up file (or folder!) @@ -347,7 +350,7 @@ func (fc *FileCache) updateObject(name string, flags pendingFlags) bool { // there is no file to upload, so consider the inconsistency resolved fc.pendingOps.Delete(name) // this file failed, but we can continue with the rest of the pending ops - return true + return nil } if flags.isDeletion && !localMissing { log.Err("FileCache::updateObject : %s exists. Ignoring deletion flag!", name) @@ -364,7 +367,7 @@ func (fc *FileCache) updateObject(name string, flags pendingFlags) bool { if flags.isDeletion && fc.notInCloud(name) { log.Info("FileCache::updateObject : %s skipping cloud deletion (not in cloud)", name) fc.pendingOps.Delete(name) - return true + return nil } if flags.isDir { // delete folder @@ -376,29 +379,39 @@ func (fc *FileCache) updateObject(name string, flags pendingFlags) bool { options := internal.DeleteFileOptions{Name: name} cloudErr = fc.NextComponent().DeleteFile(options) } + // ignore ENOENT + if isNotExist(cloudErr) { + cloudErr = nil + } } else { op = "creation/update" if info.IsDir() { // upload folder options := internal.CreateDirOptions{Name: name, Mode: info.Mode()} cloudErr = fc.NextComponent().CreateDir(options) + // ignore EEXIST + if errors.Is(cloudErr, os.ErrExist) { + cloudErr = nil + } } else { // upload file objType = "file" cloudErr = fc.uploadFile(name) } } + // handle errors if cloudErr != nil { - log.Err("FileCache::updateObject : %s %s %s failed [%v]", name, objType, op, cloudErr) - return false + err := fmt.Errorf("%s %s %s failed [%w]", name, objType, op, cloudErr) + log.Err("FileCache::updateObject : %v", err) + return err } // update state log.Info("FileCache::updateObject : %s sync successful", name) fc.pendingOps.Delete(name) - return true + return nil } // returns true if we *know* that this entity does not exist in cloud storage diff --git a/component/file_cache/file_cache.go b/component/file_cache/file_cache.go index c2913f7c5..3209b5d53 100644 --- a/component/file_cache/file_cache.go +++ b/component/file_cache/file_cache.go @@ -68,7 +68,7 @@ type FileCache struct { policyTrace bool missedChmodList sync.Map // uses object name (common.JoinUnixFilepath) pendingOps sync.Map // uses object name (common.JoinUnixFilepath) - pendingOpAdded chan struct{} // signals when an offline operation is queued + pendingOpAdded chan struct{} // signals when a pending operation is queued mountPath string // uses os.Separator (filepath.Join) allowOther bool offloadIO bool @@ -194,10 +194,8 @@ func (fc *FileCache) Start(ctx context.Context) error { } else { close(fc.startScheduledUploads) } - if fc.syncsPendingOps() { - fc.pendingOpAdded = make(chan struct{}, 1) - fc.pendingOpsWorker.Go(fc.servicePendingOps) - } + fc.pendingOpAdded = make(chan struct{}, 1) + fc.pendingOpsWorker.Go(fc.servicePendingOps) return nil } @@ -1840,6 +1838,14 @@ func (fc *FileCache) releaseFileInternal( // upload (flock is already locked) flushOptions := internal.FlushFileOptions{Handle: options.Handle, CloseInProgress: true} err = fc.flushFileCloud(flushOptions) + if err != nil { + // the handle can't carry the retry past close, so the queue must (this also blocks eviction) + log.Warn( + "FileCache::releaseFileInternal : %s upload queued for retry", + options.Handle.Path, + ) + fc.addPendingOp(options.Handle.Path, pendingFlags{}) + } // clears the flag if the upload failed, since the handle is going away fc.clearHandleDirty(options.Handle) // close the local file @@ -2082,18 +2088,11 @@ func (fc *FileCache) flushFileCloud(options internal.FlushFileOptions) error { case err == nil: fc.clearHandleDirty(options.Handle) fc.pendingOps.Delete(options.Handle.Path) - case errors.Is(err, syscall.ENOSPC): - log.Err("FileCache::flushFileCloud : %s upload failed [%v]", options.Handle.Path, err) case isOffline(err) && fc.offlineAccess: log.Warn("FileCache::flushFileCloud : %s upload delayed (offline)", options.Handle.Path) // add file to upload queue fc.addPendingOp(options.Handle.Path, pendingFlags{}) err = nil - case fc.syncsPendingOps(): - log.Err("FileCache::flushFileCloud : %s upload failed [%v]", options.Handle.Path, err) - // queue for retry, which also protects the cached data from eviction. - // the error is still returned - the caller must not assume the data is safe - fc.addPendingOp(options.Handle.Path, pendingFlags{}) default: log.Err("FileCache::flushFileCloud : %s upload failed [%v]", options.Handle.Path, err) } From 08bc161422b5d3e29edc46ea5aa27851dde0c4d7 Mon Sep 17 00:00:00 2001 From: Michael Habinsky Date: Fri, 23 Oct 2026 23:09:13 -0600 Subject: [PATCH 3/6] Fixup --- component/file_cache/async.go | 1 - 1 file changed, 1 deletion(-) diff --git a/component/file_cache/async.go b/component/file_cache/async.go index 87d6fb810..91c008617 100644 --- a/component/file_cache/async.go +++ b/component/file_cache/async.go @@ -302,7 +302,6 @@ func (fc *FileCache) runPendingOpCycle() (int, error) { case <-fc.uploadWindow(): name := key.(string) numFilesProcessed++ - cycleErr = errors.Join(cycleErr) err := fc.updateObject(name, value.(pendingFlags)) // record the failure so the caller backs off instead of busy-retrying cycleErr = errors.Join(cycleErr, err) From 37a08b625d0db9781695c51e7c690e2c9c654342 Mon Sep 17 00:00:00 2001 From: Michael Habinsky Date: Fri, 23 Oct 2026 23:45:57 -0600 Subject: [PATCH 4/6] Let ENOSPC kill uploads. Simplify implementation. --- component/file_cache/async.go | 17 +- component/file_cache/file_cache.go | 3 +- component/file_cache/file_cache_test.go | 203 +++++++++++++++++++++++- 3 files changed, 208 insertions(+), 15 deletions(-) diff --git a/component/file_cache/async.go b/component/file_cache/async.go index 91c008617..a12bce52e 100644 --- a/component/file_cache/async.go +++ b/component/file_cache/async.go @@ -305,7 +305,7 @@ func (fc *FileCache) runPendingOpCycle() (int, error) { err := fc.updateObject(name, value.(pendingFlags)) // record the failure so the caller backs off instead of busy-retrying cycleErr = errors.Join(cycleErr, err) - return continueOnError(err) + return err == nil default: return false // upload window ended } @@ -314,15 +314,6 @@ func (fc *FileCache) runPendingOpCycle() (int, error) { return numFilesProcessed, cycleErr } -// continueOnError : whether err only affects one object, so the rest of the queue can still sync -func continueOnError(err error) bool { - return err == nil || - // ENOSPC won't be returned on deletions which could be queued - errors.Is(err, syscall.ENOSPC) || - // s3storage KeyTooLongError: permanent for this key only - errors.Is(err, syscall.ENAMETOOLONG) -} - // synchronize pending operation with cloud storage // returns the cloud error if the op stays pending func (fc *FileCache) updateObject(name string, flags pendingFlags) error { @@ -396,6 +387,12 @@ func (fc *FileCache) updateObject(name string, flags pendingFlags) error { // upload file objType = "file" cloudErr = fc.uploadFile(name) + if errors.Is(cloudErr, syscall.ENAMETOOLONG) { + // the key is too long for the backend, so retrying can never succeed + log.Err("FileCache::updateObject : %s key too long, abandoning upload", name) + fc.pendingOps.Delete(name) + return nil + } } } diff --git a/component/file_cache/file_cache.go b/component/file_cache/file_cache.go index 3209b5d53..3443f105e 100644 --- a/component/file_cache/file_cache.go +++ b/component/file_cache/file_cache.go @@ -1838,7 +1838,8 @@ func (fc *FileCache) releaseFileInternal( // upload (flock is already locked) flushOptions := internal.FlushFileOptions{Handle: options.Handle, CloseInProgress: true} err = fc.flushFileCloud(flushOptions) - if err != nil { + // the host owns capacity, so ENOSPC isn't retried behind its back + if err != nil && !errors.Is(err, syscall.ENOSPC) { // the handle can't carry the retry past close, so the queue must (this also blocks eviction) log.Warn( "FileCache::releaseFileInternal : %s upload queued for retry", diff --git a/component/file_cache/file_cache_test.go b/component/file_cache/file_cache_test.go index 2798caa7b..a2381f481 100644 --- a/component/file_cache/file_cache_test.go +++ b/component/file_cache/file_cache_test.go @@ -160,6 +160,22 @@ func (suite *fileCacheTestSuite) setupTestHelper(configuration string) { } +// setupMockWithConnection : mock-backed file cache whose cloud connection the test controls +func (suite *fileCacheTestSuite) setupMockWithConnection(configuration string) *atomic.Bool { + suite.assert = assert.New(suite.T()) + err := config.ReadConfigFromReader(strings.NewReader(configuration)) + suite.assert.NoError(err) + suite.mockCtrl = gomock.NewController(suite.T()) + suite.mock = internal.NewMockComponent(suite.mockCtrl) + suite.fileCache = newTestFileCache(suite.mock) + suite.useMock = true + connected := &atomic.Bool{} + connected.Store(true) + suite.mock.EXPECT().CloudConnected().AnyTimes().DoAndReturn(connected.Load) + suite.assert.NoError(suite.fileCache.Start(context.Background())) + return connected +} + func (suite *fileCacheTestSuite) cleanupTest() { err := suite.fileCache.Stop() if err != nil { @@ -2011,6 +2027,103 @@ func (suite *fileCacheTestSuite) TestReleaseFileOffline() { suite.assert.True(pending, "dirty file should be queued for upload when released offline") } +func (suite *fileCacheTestSuite) TestReleaseFileUploadFailureQueued() { + suite.cleanupTest() + connected := suite.setupMockWithConnection(fmt.Sprintf( + "file_cache:\n path: %s\n offload-io: true\n block-offline-access: true", + suite.cache_path, + )) + defer suite.cleanupTest() + + name := "release-upload-failure" + handle, err := suite.fileCache.CreateFile(internal.CreateFileOptions{Name: name, Mode: 0777}) + suite.assert.NoError(err) + _, err = suite.fileCache.WriteFile( + &internal.WriteFileOptions{Handle: handle, Data: []byte("unsynced data")}, + ) + suite.assert.NoError(err) + suite.assert.True(handle.Dirty()) + + // keep the pending-op worker from retrying the upload + connected.Store(false) + suite.mock.EXPECT().CopyFromFile(gomock.Any()).Return(syscall.EIO) + err = suite.fileCache.ReleaseFile(internal.ReleaseFileOptions{Handle: handle}) + suite.assert.ErrorIs(err, syscall.EIO) + + _, pending := suite.fileCache.pendingOps.Load(name) + suite.assert.True(pending, "failed upload on close should be queued for retry") + flock := suite.fileCache.fileLocks.Get(name) + suite.assert.Zero(flock.Count()) + suite.assert.Zero(flock.DirtyCount()) + suite.assert.ErrorIs(handle.GetFileObject().Close(), os.ErrClosed) +} + +func (suite *fileCacheTestSuite) TestReleaseFileEnospcNotQueued() { + suite.cleanupTest() + connected := suite.setupMockWithConnection( + fmt.Sprintf("file_cache:\n path: %s\n offload-io: true", suite.cache_path), + ) + defer suite.cleanupTest() + + name := "release-enospc" + handle, err := suite.fileCache.CreateFile(internal.CreateFileOptions{Name: name, Mode: 0777}) + suite.assert.NoError(err) + _, err = suite.fileCache.WriteFile( + &internal.WriteFileOptions{Handle: handle, Data: []byte("unsynced data")}, + ) + suite.assert.NoError(err) + suite.assert.True(handle.Dirty()) + + connected.Store(false) + suite.mock.EXPECT().CopyFromFile(gomock.Any()).Return(syscall.ENOSPC) + err = suite.fileCache.ReleaseFile(internal.ReleaseFileOptions{Handle: handle}) + suite.assert.ErrorIs(err, syscall.ENOSPC) + + _, pending := suite.fileCache.pendingOps.Load(name) + suite.assert.False(pending, "ENOSPC on close should be returned, not retried") + flock := suite.fileCache.fileLocks.Get(name) + suite.assert.Zero(flock.Count()) + suite.assert.Zero(flock.DirtyCount()) + suite.assert.ErrorIs(handle.GetFileObject().Close(), os.ErrClosed) +} + +func (suite *fileCacheTestSuite) TestSyncFileUploadFailureNotQueued() { + // enable mock component, with offline access disabled + suite.cleanupTest() + defaultConfig := fmt.Sprintf( + "file_cache:\n path: %s\n offload-io: true\n block-offline-access: true", + suite.cache_path, + ) + suite.useMock = true + suite.setupTestHelper(defaultConfig) + defer suite.cleanupTest() + + name := "sync-upload-failure" + f, err := os.OpenFile( + filepath.Join(suite.cache_path, name), + os.O_RDWR|os.O_CREATE|os.O_TRUNC, + 0777, + ) + suite.assert.NoError(err) + defer f.Close() + _, err = f.Write([]byte("unsynced data")) + suite.assert.NoError(err) + + handle := handlemap.NewHandle(name) + handle.SetFileObject(f) + suite.fileCache.setHandleDirty(handle) + + suite.mock.EXPECT().SyncFile(gomock.Any()).Return(nil) + suite.mock.EXPECT().CopyFromFile(gomock.Any()).Return(&common.CloudUnreachableError{}) + + err = suite.fileCache.SyncFile(internal.SyncFileOptions{Handle: handle}) + suite.assert.Error(err) + + _, pending := suite.fileCache.pendingOps.Load(name) + suite.assert.False(pending, "fsync failure should not queue a retry while the handle is open") + suite.assert.True(handle.Dirty(), "handle should stay dirty so close retries the upload") +} + func (suite *fileCacheTestSuite) TestOfflineToConnected() { // Set up mock manually so CloudConnected can change during the test suite.cleanupTest() @@ -3105,8 +3218,8 @@ func (suite *fileCacheTestSuite) TestUpdateObjectOfflineErrorKeepsPending() { suite.fileCache.addPendingOp(name, pendingFlags{}) suite.mock.EXPECT().CopyFromFile(gomock.Any()).Return(&common.CloudUnreachableError{}) - success := suite.fileCache.updateObject(name, pendingFlags{}) - suite.assert.False(success) + err = suite.fileCache.updateObject(name, pendingFlags{}) + suite.assert.Error(err) _, stillPending := suite.fileCache.pendingOps.Load(name) suite.assert.True( @@ -3131,13 +3244,71 @@ func (suite *fileCacheTestSuite) TestUpdateObjectDeletionMissingLocalFile() { suite.mock.EXPECT().GetAttr(internal.GetAttrOptions{Name: name}).Return(nil, nil) suite.mock.EXPECT().DeleteFile(internal.DeleteFileOptions{Name: name}).Return(nil) - success := suite.fileCache.updateObject(name, pendingFlags{isDeletion: true}) - suite.assert.True(success) + err := suite.fileCache.updateObject(name, pendingFlags{isDeletion: true}) + suite.assert.NoError(err) _, stillPending := suite.fileCache.pendingOps.Load(name) suite.assert.False(stillPending, "delete should be synced and removed from pendingOps") } +func (suite *fileCacheTestSuite) TestUpdateObjectDeletionAlreadyGone() { + // enable mock component + suite.cleanupTest() + defaultConfig := fmt.Sprintf( + "file_cache:\n path: %s\n offload-io: true", + suite.cache_path, + ) + suite.useMock = true + suite.setupTestHelper(defaultConfig) + defer suite.cleanupTest() + + fileName := "async-delete-gone-file" + suite.fileCache.addPendingOp(fileName, pendingFlags{isDeletion: true}) + suite.mock.EXPECT().GetAttr(internal.GetAttrOptions{Name: fileName}).Return(nil, nil) + suite.mock.EXPECT(). + DeleteFile(internal.DeleteFileOptions{Name: fileName}). + Return(syscall.ENOENT) + + dirName := "async-delete-gone-dir" + suite.fileCache.addPendingOp(dirName, pendingFlags{isDir: true, isDeletion: true}) + suite.mock.EXPECT().GetAttr(internal.GetAttrOptions{Name: dirName}).Return(nil, nil) + suite.mock.EXPECT(). + DeleteDir(internal.DeleteDirOptions{Name: dirName}). + Return(syscall.ENOENT) + + for _, name := range []string{fileName, dirName} { + flags, _ := suite.fileCache.pendingOps.Load(name) + err := suite.fileCache.updateObject(name, flags.(pendingFlags)) + suite.assert.NoError(err, name) + _, stillPending := suite.fileCache.pendingOps.Load(name) + suite.assert.False(stillPending, "%s: already deleted, so op should clear", name) + } +} + +func (suite *fileCacheTestSuite) TestUpdateObjectCreateDirAlreadyExists() { + // enable mock component + suite.cleanupTest() + defaultConfig := fmt.Sprintf( + "file_cache:\n path: %s\n offload-io: true", + suite.cache_path, + ) + suite.useMock = true + suite.setupTestHelper(defaultConfig) + defer suite.cleanupTest() + + name := "async-create-existing-dir" + err := os.MkdirAll(filepath.Join(suite.cache_path, name), 0777) + suite.assert.NoError(err) + suite.fileCache.addPendingOp(name, pendingFlags{isDir: true}) + suite.mock.EXPECT().CreateDir(gomock.Any()).Return(os.ErrExist) + + err = suite.fileCache.updateObject(name, pendingFlags{isDir: true}) + suite.assert.NoError(err) + + _, stillPending := suite.fileCache.pendingOps.Load(name) + suite.assert.False(stillPending, "existing directory should clear the op") +} + func (suite *fileCacheTestSuite) TestServicePendingOpsOfflineKeepsPending() { // enable mock component suite.cleanupTest() @@ -3182,6 +3353,30 @@ func (suite *fileCacheTestSuite) TestServicePendingOpsProcessesPendingOnline() { suite.assert.False(pending, "pending op should be processed by async uploader when online") } +func (suite *fileCacheTestSuite) TestRunPendingOpCycleReportsFailure() { + // enable mock component + suite.cleanupTest() + defaultConfig := fmt.Sprintf( + "file_cache:\n path: %s\n offload-io: true", + suite.cache_path, + ) + suite.useMock = true + suite.setupTestHelper(defaultConfig) + defer suite.cleanupTest() + + name := "cycle-upload-failure" + err := os.WriteFile(filepath.Join(suite.cache_path, name), []byte("data"), 0777) + suite.assert.NoError(err) + suite.fileCache.addPendingOp(name, pendingFlags{}) + suite.mock.EXPECT().CopyFromFile(gomock.Any()).Return(syscall.ENOSPC) + + _, err = suite.fileCache.runPendingOpCycle() + suite.assert.ErrorIs(err, syscall.ENOSPC, "failures must be reported so the worker backs off") + + _, pending := suite.fileCache.pendingOps.Load(name) + suite.assert.True(pending) +} + func (suite *fileCacheTestSuite) TestCreateEmptyFileEqualTrue() { defer suite.cleanupTest() suite.cleanupTest() // teardown the default file cache generated From 75079365031c471c9a8d7e341ca1004d1733f1ff Mon Sep 17 00:00:00 2001 From: Michael Habinsky Date: Tue, 27 Oct 2026 13:37:24 -0600 Subject: [PATCH 5/6] Improve readability and error handling in pending op code. --- component/file_cache/async.go | 81 ++++++++++++++++++----------------- 1 file changed, 42 insertions(+), 39 deletions(-) diff --git a/component/file_cache/async.go b/component/file_cache/async.go index a12bce52e..2098323e7 100644 --- a/component/file_cache/async.go +++ b/component/file_cache/async.go @@ -291,10 +291,7 @@ func (fc *FileCache) servicePendingOps() { // runPendingOpCycle : sync pending ops with cloud storage until the upload window // closes or an operation fails. Returns the number of ops attempted, and the error // that ended the cycle, if any. -func (fc *FileCache) runPendingOpCycle() (int, error) { - numFilesProcessed := 0 - var cycleErr error - +func (fc *FileCache) runPendingOpCycle() (numFilesProcessed int, err error) { fc.pendingOps.Range(func(key, value any) bool { select { case <-fc.componentStopping: @@ -302,22 +299,25 @@ func (fc *FileCache) runPendingOpCycle() (int, error) { case <-fc.uploadWindow(): name := key.(string) numFilesProcessed++ - err := fc.updateObject(name, value.(pendingFlags)) - // record the failure so the caller backs off instead of busy-retrying - cycleErr = errors.Join(cycleErr, err) + err = fc.updateObject(name, value.(pendingFlags)) return err == nil default: return false // upload window ended } }) - return numFilesProcessed, cycleErr + return numFilesProcessed, err } // synchronize pending operation with cloud storage -// returns the cloud error if the op stays pending +// returns the cloud error *if and only if* the op stays pending func (fc *FileCache) updateObject(name string, flags pendingFlags) error { - log.Trace("FileCache::updateObject : %s", name) + log.Trace( + "FileCache::updateObject : %s, isDir=%t, isDeletion=%t", + name, + flags.isDir, + flags.isDeletion, + ) // lock the file flock := fc.fileLocks.Get(name) @@ -336,11 +336,16 @@ func (fc *FileCache) updateObject(name string, flags pendingFlags) error { localMissing := os.IsNotExist(localErr) // in case of inconsistency, local state takes precedence (except to prevent incorrect deletions) if !flags.isDeletion && localErr != nil { - log.Err("FileCache::updateObject : %s stat failed. Here's why: %v", name, localErr) - // there is no file to upload, so consider the inconsistency resolved - fc.pendingOps.Delete(name) - // this file failed, but we can continue with the rest of the pending ops - return nil + if localMissing { + log.Err("FileCache::updateObject : %s not found. Pending operation dropped.", localPath) + fc.pendingOps.Delete(name) + // do not interrupt other pending operations + return nil + } else { + // other errors (EIO, EACCESS, out of resources, etc.) + log.Err("FileCache::updateObject : %s %v", name, localErr) + return localErr + } } if flags.isDeletion && !localMissing { log.Err("FileCache::updateObject : %s exists. Ignoring deletion flag!", name) @@ -350,63 +355,61 @@ func (fc *FileCache) updateObject(name string, flags pendingFlags) error { } // update cloud - op := "deletion" - objType := "directory" + var op, objType string var cloudErr error if localMissing { + op = "deletion" if flags.isDeletion && fc.notInCloud(name) { + // nothing to do log.Info("FileCache::updateObject : %s skipping cloud deletion (not in cloud)", name) - fc.pendingOps.Delete(name) - return nil - } - if flags.isDir { - // delete folder + } else if flags.isDir { + objType = "directory" options := internal.DeleteDirOptions{Name: name} cloudErr = fc.NextComponent().DeleteDir(options) } else { - // delete file objType = "file" options := internal.DeleteFileOptions{Name: name} cloudErr = fc.NextComponent().DeleteFile(options) } // ignore ENOENT if isNotExist(cloudErr) { + log.Debug("FileCache::updateObject : %s does not exist in cloud", name) cloudErr = nil } } else { op = "creation/update" if info.IsDir() { - // upload folder + objType = "directory" options := internal.CreateDirOptions{Name: name, Mode: info.Mode()} cloudErr = fc.NextComponent().CreateDir(options) - // ignore EEXIST - if errors.Is(cloudErr, os.ErrExist) { - cloudErr = nil - } } else { - // upload file objType = "file" cloudErr = fc.uploadFile(name) - if errors.Is(cloudErr, syscall.ENAMETOOLONG) { - // the key is too long for the backend, so retrying can never succeed - log.Err("FileCache::updateObject : %s key too long, abandoning upload", name) - fc.pendingOps.Delete(name) - return nil - } + } + // ignore EEXIST + if errors.Is(cloudErr, os.ErrExist) { + log.Debug("FileCache::updateObject : %s already exists in cloud", name) + cloudErr = nil } } - // handle errors + // drop any ops that will never succeed + if errors.Is(cloudErr, syscall.ENAMETOOLONG) { + // the key is too long for the backend, so retrying can never succeed + log.Err("FileCache::updateObject : %s key too long, abandoning upload", name) + fc.pendingOps.Delete(name) + // do not interrupt other pending operations + return nil + } + + // any remaining errors should interrupt all pending operations and delay retries if cloudErr != nil { err := fmt.Errorf("%s %s %s failed [%w]", name, objType, op, cloudErr) log.Err("FileCache::updateObject : %v", err) return err } - // update state - log.Info("FileCache::updateObject : %s sync successful", name) fc.pendingOps.Delete(name) - return nil } From c05d24ec32abf13bf2535b450ca041b847ce7fa8 Mon Sep 17 00:00:00 2001 From: Michael Habinsky Date: Tue, 27 Oct 2026 13:42:44 -0600 Subject: [PATCH 6/6] Allow ENOSPC to be returned to the application! --- component/libfuse/libfuse2_handler.go | 2 ++ component/libfuse/libfuse2_handler_test_wrapper.go | 2 ++ 2 files changed, 4 insertions(+) diff --git a/component/libfuse/libfuse2_handler.go b/component/libfuse/libfuse2_handler.go index 368b5d4ab..a73ee6031 100644 --- a/component/libfuse/libfuse2_handler.go +++ b/component/libfuse/libfuse2_handler.go @@ -873,6 +873,8 @@ func fuseErrnoFromError(err error) int { return -fuse.EBUSY case errors.Is(err, syscall.EROFS): return -fuse.EROFS + case errors.Is(err, syscall.ENOSPC): + return -fuse.ENOSPC case errors.Is(err, syscall.ETXTBSY): return -fuse.ETXTBSY case errors.Is(err, syscall.ENOSYS): diff --git a/component/libfuse/libfuse2_handler_test_wrapper.go b/component/libfuse/libfuse2_handler_test_wrapper.go index f7cfe921c..d893bd992 100644 --- a/component/libfuse/libfuse2_handler_test_wrapper.go +++ b/component/libfuse/libfuse2_handler_test_wrapper.go @@ -357,6 +357,8 @@ func testFuseErrnoFromError(suite *libfuseTestSuite) { {err: fs.ErrExist, expected: -fuse.EEXIST}, {err: os.ErrExist, expected: -fuse.EEXIST}, {err: syscall.EBADF, expected: -fuse.EBADF}, + {err: syscall.ENOSPC, expected: -fuse.ENOSPC}, + {err: fmt.Errorf("upload: %w", syscall.ENOSPC), expected: -fuse.ENOSPC}, {err: os.ErrInvalid, expected: -fuse.EBADF}, {err: (*os.File)(nil).Close(), expected: -fuse.EBADF}, {err: &fs.PathError{Op: "close", Err: os.ErrClosed}, expected: -fuse.EBADF},