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
This commit is contained in:
Chris Lu
2026-04-03 16:15:24 -07:00
parent d49c2a7364
commit abbd0207ba
2 changed files with 173 additions and 0 deletions
+38
View File
@@ -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)
}
+135
View File
@@ -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()))
}