From abbd0207ba50a33e70c0ad8869d3d68858386ab2 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Fri, 3 Apr 2026 16:15:24 -0700 Subject: [PATCH] fix(filer): add pre-cancellation checks to write paths, leave RollbackTransaction unguarded Refactors PR #8909 to: - Extract a checkContextCancelled() helper to eliminate repeated 3-line pattern - Guard 10 write/mutate methods against already-cancelled contexts - Intentionally leave RollbackTransaction unguarded since it is a cleanup operation that must succeed even after cancellation - Add deadline-exceeded test coverage alongside cancellation tests - Simplify tests from ~230 lines to ~130 lines with clearer structure --- weed/filer/filerstore_wrapper.go | 38 ++++++++ weed/filer/filerstore_wrapper_test.go | 135 ++++++++++++++++++++++++++ 2 files changed, 173 insertions(+) diff --git a/weed/filer/filerstore_wrapper.go b/weed/filer/filerstore_wrapper.go index 0c8d692f2..381aba66b 100644 --- a/weed/filer/filerstore_wrapper.go +++ b/weed/filer/filerstore_wrapper.go @@ -127,7 +127,18 @@ func (fsw *FilerStoreWrapper) Initialize(configuration util.Configuration, prefi return fsw.getDefaultStore().Initialize(configuration, prefix) } +// checkContextCancelled returns the context error if the context is already +// cancelled or expired. This is checked before calling context.WithoutCancel +// on write paths to prevent orphaned metadata when the originating request +// (e.g. S3 CopyObject, CompleteMultipartUpload) has already been abandoned. +func checkContextCancelled(ctx context.Context) error { + return ctx.Err() +} + func (fsw *FilerStoreWrapper) InsertEntry(ctx context.Context, entry *Entry) error { + if err := checkContextCancelled(ctx); err != nil { + return err + } ctx = context.WithoutCancel(ctx) actualStore := fsw.getActualStore(entry.FullPath) stats.FilerStoreCounter.WithLabelValues(actualStore.GetName(), "insert").Inc() @@ -155,6 +166,9 @@ func (fsw *FilerStoreWrapper) InsertEntry(ctx context.Context, entry *Entry) err // InsertEntryKnownAbsent skips the pre-insert FindEntry path when the caller has // already established that the target path does not exist. func (fsw *FilerStoreWrapper) InsertEntryKnownAbsent(ctx context.Context, entry *Entry) error { + if err := checkContextCancelled(ctx); err != nil { + return err + } ctx = context.WithoutCancel(ctx) actualStore := fsw.getActualStore(entry.FullPath) stats.FilerStoreCounter.WithLabelValues(actualStore.GetName(), "insert").Inc() @@ -178,6 +192,9 @@ func (fsw *FilerStoreWrapper) InsertEntryKnownAbsent(ctx context.Context, entry } func (fsw *FilerStoreWrapper) UpdateEntry(ctx context.Context, entry *Entry) error { + if err := checkContextCancelled(ctx); err != nil { + return err + } ctx = context.WithoutCancel(ctx) actualStore := fsw.getActualStore(entry.FullPath) stats.FilerStoreCounter.WithLabelValues(actualStore.GetName(), "update").Inc() @@ -236,6 +253,9 @@ func (fsw *FilerStoreWrapper) FindEntry(ctx context.Context, fp util.FullPath) ( } func (fsw *FilerStoreWrapper) DeleteEntry(ctx context.Context, fp util.FullPath) (err error) { + if err := checkContextCancelled(ctx); err != nil { + return err + } ctx = context.WithoutCancel(ctx) actualStore := fsw.getActualStore(fp) stats.FilerStoreCounter.WithLabelValues(actualStore.GetName(), "delete").Inc() @@ -264,6 +284,9 @@ func (fsw *FilerStoreWrapper) DeleteEntry(ctx context.Context, fp util.FullPath) } func (fsw *FilerStoreWrapper) DeleteOneEntry(ctx context.Context, existingEntry *Entry) (err error) { + if err := checkContextCancelled(ctx); err != nil { + return err + } ctx = context.WithoutCancel(ctx) actualStore := fsw.getActualStore(existingEntry.FullPath) stats.FilerStoreCounter.WithLabelValues(actualStore.GetName(), "delete").Inc() @@ -288,6 +311,9 @@ func (fsw *FilerStoreWrapper) DeleteOneEntry(ctx context.Context, existingEntry } func (fsw *FilerStoreWrapper) DeleteFolderChildren(ctx context.Context, fp util.FullPath) (err error) { + if err := checkContextCancelled(ctx); err != nil { + return err + } ctx = context.WithoutCancel(ctx) actualStore := fsw.getActualStore(fp + "/") stats.FilerStoreCounter.WithLabelValues(actualStore.GetName(), "deleteFolderChildren").Inc() @@ -394,11 +420,17 @@ func (fsw *FilerStoreWrapper) prefixFilterEntries(ctx context.Context, dirPath u } func (fsw *FilerStoreWrapper) BeginTransaction(ctx context.Context) (context.Context, error) { + if err := checkContextCancelled(ctx); err != nil { + return nil, err + } ctx = context.WithoutCancel(ctx) return fsw.getDefaultStore().BeginTransaction(ctx) } func (fsw *FilerStoreWrapper) CommitTransaction(ctx context.Context) error { + if err := checkContextCancelled(ctx); err != nil { + return err + } ctx = context.WithoutCancel(ctx) return fsw.getDefaultStore().CommitTransaction(ctx) } @@ -413,6 +445,9 @@ func (fsw *FilerStoreWrapper) Shutdown() { } func (fsw *FilerStoreWrapper) KvPut(ctx context.Context, key []byte, value []byte) (err error) { + if err := checkContextCancelled(ctx); err != nil { + return err + } ctx = context.WithoutCancel(ctx) return fsw.getDefaultStore().KvPut(ctx, key, value) } @@ -421,6 +456,9 @@ func (fsw *FilerStoreWrapper) KvGet(ctx context.Context, key []byte) (value []by return fsw.getDefaultStore().KvGet(ctx, key) } func (fsw *FilerStoreWrapper) KvDelete(ctx context.Context, key []byte) (err error) { + if err := checkContextCancelled(ctx); err != nil { + return err + } ctx = context.WithoutCancel(ctx) return fsw.getDefaultStore().KvDelete(ctx, key) } diff --git a/weed/filer/filerstore_wrapper_test.go b/weed/filer/filerstore_wrapper_test.go index c3ef32018..b17d590eb 100644 --- a/weed/filer/filerstore_wrapper_test.go +++ b/weed/filer/filerstore_wrapper_test.go @@ -2,8 +2,10 @@ package filer import ( "context" + "errors" "os" "testing" + "time" "github.com/seaweedfs/seaweedfs/weed/util" "github.com/stretchr/testify/assert" @@ -69,3 +71,136 @@ func TestFilerStoreWrapperMimeNormalization(t *testing.T) { } } } + +// cancelledCtx returns a context that is already cancelled. +func cancelledCtx() context.Context { + ctx, cancel := context.WithCancel(context.Background()) + cancel() + return ctx +} + +// expiredCtx returns a context whose deadline has already passed. +func expiredCtx() context.Context { + ctx, cancel := context.WithDeadline(context.Background(), time.Now().Add(-time.Second)) + _ = cancel // already expired, but keep the cancel func from leaking + return ctx +} + +func TestFilerStoreWrapperWriteOpsRejectCancelledContext(t *testing.T) { + newEntry := func(path string) *Entry { + return &Entry{ + FullPath: util.FullPath(path), + Attr: Attr{Mode: 0o660, Mime: "application/octet-stream"}, + } + } + + // Each write operation that should be guarded. + writeOps := []struct { + name string + run func(*FilerStoreWrapper, context.Context) error + }{ + {"InsertEntry", func(fsw *FilerStoreWrapper, ctx context.Context) error { + return fsw.InsertEntry(ctx, newEntry("/test/a")) + }}, + {"InsertEntryKnownAbsent", func(fsw *FilerStoreWrapper, ctx context.Context) error { + return fsw.InsertEntryKnownAbsent(ctx, newEntry("/test/b")) + }}, + {"UpdateEntry", func(fsw *FilerStoreWrapper, ctx context.Context) error { + _ = fsw.InsertEntry(context.Background(), newEntry("/test/c")) + return fsw.UpdateEntry(ctx, newEntry("/test/c")) + }}, + {"DeleteEntry", func(fsw *FilerStoreWrapper, ctx context.Context) error { + _ = fsw.InsertEntry(context.Background(), newEntry("/test/d")) + return fsw.DeleteEntry(ctx, "/test/d") + }}, + {"DeleteOneEntry", func(fsw *FilerStoreWrapper, ctx context.Context) error { + e := newEntry("/test/e") + _ = fsw.InsertEntry(context.Background(), e) + return fsw.DeleteOneEntry(ctx, e) + }}, + {"DeleteFolderChildren", func(fsw *FilerStoreWrapper, ctx context.Context) error { + _ = fsw.InsertEntry(context.Background(), newEntry("/test/folder/child")) + return fsw.DeleteFolderChildren(ctx, "/test/folder") + }}, + {"BeginTransaction", func(fsw *FilerStoreWrapper, ctx context.Context) error { + _, err := fsw.BeginTransaction(ctx) + return err + }}, + {"CommitTransaction", func(fsw *FilerStoreWrapper, ctx context.Context) error { + return fsw.CommitTransaction(ctx) + }}, + {"KvPut", func(fsw *FilerStoreWrapper, ctx context.Context) error { + return fsw.KvPut(ctx, []byte("k"), []byte("v")) + }}, + {"KvDelete", func(fsw *FilerStoreWrapper, ctx context.Context) error { + _ = fsw.KvPut(context.Background(), []byte("k"), []byte("v")) + return fsw.KvDelete(ctx, []byte("k")) + }}, + } + + badContexts := []struct { + name string + ctx context.Context + wantError error + }{ + {"cancelled", cancelledCtx(), context.Canceled}, + {"deadline exceeded", expiredCtx(), context.DeadlineExceeded}, + } + + for _, op := range writeOps { + for _, bc := range badContexts { + t.Run(op.name+"/"+bc.name, func(t *testing.T) { + wrapper := NewFilerStoreWrapper(newStubFilerStore()) + err := op.run(wrapper, bc.ctx) + require.Error(t, err) + assert.True(t, errors.Is(err, bc.wantError), "got %v, want %v", err, bc.wantError) + }) + } + } +} + +func TestFilerStoreWrapperWriteOpsSucceedWithActiveContext(t *testing.T) { + wrapper := NewFilerStoreWrapper(newStubFilerStore()) + ctx := context.Background() + entry := &Entry{ + FullPath: util.FullPath("/test/obj"), + Attr: Attr{Mode: 0o660}, + } + + require.NoError(t, wrapper.InsertEntry(ctx, entry)) + require.NoError(t, wrapper.UpdateEntry(ctx, entry)) + require.NoError(t, wrapper.DeleteOneEntry(ctx, entry)) + require.NoError(t, wrapper.InsertEntryKnownAbsent(ctx, entry)) + require.NoError(t, wrapper.DeleteEntry(ctx, entry.FullPath)) + require.NoError(t, wrapper.KvPut(ctx, []byte("k"), []byte("v"))) + require.NoError(t, wrapper.KvDelete(ctx, []byte("k"))) + + txCtx, err := wrapper.BeginTransaction(ctx) + require.NoError(t, err) + require.NoError(t, wrapper.CommitTransaction(txCtx)) +} + +func TestFilerStoreWrapperReadOpsSucceedWithCancelledContext(t *testing.T) { + wrapper := NewFilerStoreWrapper(newStubFilerStore()) + entry := &Entry{ + FullPath: util.FullPath("/test/readable"), + Attr: Attr{Mode: 0o660}, + } + require.NoError(t, wrapper.InsertEntry(context.Background(), entry)) + require.NoError(t, wrapper.KvPut(context.Background(), []byte("rk"), []byte("rv"))) + + ctx := cancelledCtx() + + _, err := wrapper.FindEntry(ctx, entry.FullPath) + assert.NoError(t, err) + + _, err = wrapper.KvGet(ctx, []byte("rk")) + assert.NoError(t, err) +} + +// RollbackTransaction must succeed even when the context is cancelled, +// because it is a cleanup operation called after failures. +func TestFilerStoreWrapperRollbackSucceedsWithCancelledContext(t *testing.T) { + wrapper := NewFilerStoreWrapper(newStubFilerStore()) + assert.NoError(t, wrapper.RollbackTransaction(cancelledCtx())) +}