From ef109fe9e155da177390b9db084a4ce70b67b399 Mon Sep 17 00:00:00 2001
From: Chris Lu <chrislusf@users.noreply.github.com>
Date: Wed, 24 Jun 2026 14:24:22 -0700
Subject: [PATCH] mount: don't hang close() when a writer is killed during
 flush (#10090)

* operation: bound AssignVolume with a deadline

AssignVolume ran on context.Background(), so when the filer is overwhelmed
the RPC could block indefinitely and wedge every caller holding the
connection. Give it a 30s deadline so a stuck assign fails and the caller's
retry/error path runs instead of hanging forever.

* mount: abort flush when the FUSE request is interrupted

On close(), a killed process blocks in fuse_flush waiting for the mount to
answer. doFlush ran its metadata CreateEntry on context.Background() and
ignored the kernel interrupt channel, so against an overwhelmed filer the
flush never completed and the process stayed in uninterruptible sleep --
making the pod un-killable.

Derive a context from the FUSE cancel channel in Flush/Fsync and thread it
through doFlush -> flushMetadataToFiler -> streamCreateEntry; the retry loop
stops as soon as the context is cancelled. Release and the pre-rename flush
keep a non-cancellable context since they must finish regardless.

* operation: harden the AssignVolume timeout test

Make the test double's signal send non-blocking and bound the receive with a
timeout so a regression can't wedge the test instead of failing it.
---
 weed/mount/error_classifier_test.go     |  9 ++--
 weed/mount/metadata_flush_retry.go      | 39 +++++++++-----
 weed/mount/metadata_flush_retry_test.go | 45 ++++++++++++++--
 weed/mount/weedfs_async_flush.go        |  5 +-
 weed/mount/weedfs_file_io.go            |  6 ++-
 weed/mount/weedfs_file_mkrm.go          |  2 +-
 weed/mount/weedfs_file_sync.go          | 40 +++++++++++---
 weed/mount/weedfs_file_sync_test.go     | 50 ++++++++++++++++-
 weed/mount/weedfs_rename.go             |  3 +-
 weed/mount/wfs_save.go                  |  2 +-
 weed/operation/upload_content.go        |  8 ++-
 weed/operation/upload_content_test.go   | 72 +++++++++++++++++++++++++
 12 files changed, 245 insertions(+), 36 deletions(-)

diff --git a/weed/mount/error_classifier_test.go b/weed/mount/error_classifier_test.go
index 1414cf2e8..b49db4f99 100644
--- a/weed/mount/error_classifier_test.go
+++ b/weed/mount/error_classifier_test.go
@@ -1,6 +1,7 @@
 package mount
 
 import (
+	"context"
 	"errors"
 	"fmt"
 	"syscall"
@@ -82,13 +83,13 @@ func TestRetryMetadataFlushIfShortCircuitsOnPermanentError(t *testing.T) {
 	t.Cleanup(func() {
 		metadataFlushSleep = originalSleep
 	})
-	metadataFlushSleep = func(_ time.Duration) {
+	metadataFlushSleep = func(_ context.Context, _ time.Duration) {
 		t.Fatal("sleep should not be called when shouldRetry returns false")
 	}
 
 	attempts := 0
 	permanent := status.Error(codes.NotFound, "entry missing")
-	err := retryMetadataFlushIf(func() error {
+	err := retryMetadataFlushIf(context.Background(), func() error {
 		attempts++
 		return permanent
 	}, isRetryableFilerError, nil)
@@ -108,11 +109,11 @@ func TestRetryMetadataFlushIfRetriesTransientErrors(t *testing.T) {
 	t.Cleanup(func() {
 		metadataFlushSleep = originalSleep
 	})
-	metadataFlushSleep = func(_ time.Duration) {}
+	metadataFlushSleep = func(_ context.Context, _ time.Duration) {}
 
 	attempts := 0
 	transient := status.Error(codes.Canceled, "grpc: the client connection is closing")
-	err := retryMetadataFlushIf(func() error {
+	err := retryMetadataFlushIf(context.Background(), func() error {
 		attempts++
 		return transient
 	}, isRetryableFilerError, nil)
diff --git a/weed/mount/metadata_flush_retry.go b/weed/mount/metadata_flush_retry.go
index 5a4f3c8d8..8e14cffee 100644
--- a/weed/mount/metadata_flush_retry.go
+++ b/weed/mount/metadata_flush_retry.go
@@ -1,21 +1,30 @@
 package mount
 
-import "time"
+import (
+	"context"
+	"time"
+)
 
 const metadataFlushRetries = 3
 
-var metadataFlushSleep = time.Sleep
-
-func retryMetadataFlush(flush func() error, onRetry func(nextAttempt, totalAttempts int, backoff time.Duration, err error)) error {
-	return retryMetadataFlushIf(flush, nil, onRetry)
+// metadataFlushSleep waits for d or until ctx is cancelled. Overridable in tests.
+var metadataFlushSleep = func(ctx context.Context, d time.Duration) {
+	timer := time.NewTimer(d)
+	defer timer.Stop()
+	select {
+	case <-ctx.Done():
+	case <-timer.C:
+	}
 }
 
-// retryMetadataFlushIf is retryMetadataFlush with an optional shouldRetry
-// predicate. If shouldRetry is nil or returns true, the flush is retried with
-// exponential backoff; if it returns false, the error is returned immediately
-// so callers don't pay retry latency on clearly permanent errors (e.g.
-// ENOENT/EACCES/EINVAL from a synchronous setattr).
-func retryMetadataFlushIf(flush func() error, shouldRetry func(error) bool, onRetry func(nextAttempt, totalAttempts int, backoff time.Duration, err error)) error {
+func retryMetadataFlush(ctx context.Context, flush func() error, onRetry func(nextAttempt, totalAttempts int, backoff time.Duration, err error)) error {
+	return retryMetadataFlushIf(ctx, flush, nil, onRetry)
+}
+
+// retryMetadataFlushIf retries flush with exponential backoff, stopping early
+// when shouldRetry returns false (clearly permanent errors) or when ctx is
+// cancelled (the FUSE request was interrupted, e.g. the process was killed).
+func retryMetadataFlushIf(ctx context.Context, flush func() error, shouldRetry func(error) bool, onRetry func(nextAttempt, totalAttempts int, backoff time.Duration, err error)) error {
 	totalAttempts := metadataFlushRetries + 1
 	var err error
 	for attempt := 1; attempt <= totalAttempts; attempt++ {
@@ -29,12 +38,18 @@ func retryMetadataFlushIf(flush func() error, shouldRetry func(error) bool, onRe
 		if shouldRetry != nil && !shouldRetry(err) {
 			break
 		}
+		if ctx.Err() != nil {
+			break
+		}
 
 		backoff := time.Duration(1<<uint(attempt-1)) * time.Second
 		if onRetry != nil {
 			onRetry(attempt+1, totalAttempts, backoff, err)
 		}
-		metadataFlushSleep(backoff)
+		metadataFlushSleep(ctx, backoff)
+		if ctx.Err() != nil {
+			break
+		}
 	}
 	return err
 }
diff --git a/weed/mount/metadata_flush_retry_test.go b/weed/mount/metadata_flush_retry_test.go
index 815bca4ed..7bf3a23ff 100644
--- a/weed/mount/metadata_flush_retry_test.go
+++ b/weed/mount/metadata_flush_retry_test.go
@@ -1,6 +1,7 @@
 package mount
 
 import (
+	"context"
 	"errors"
 	"testing"
 	"time"
@@ -13,12 +14,12 @@ func TestRetryMetadataFlushEventuallySucceeds(t *testing.T) {
 	})
 
 	var sleeps []time.Duration
-	metadataFlushSleep = func(d time.Duration) {
+	metadataFlushSleep = func(ctx context.Context, d time.Duration) {
 		sleeps = append(sleeps, d)
 	}
 
 	attempts := 0
-	err := retryMetadataFlush(func() error {
+	err := retryMetadataFlush(context.Background(), func() error {
 		attempts++
 		if attempts < 3 {
 			return errors.New("temporary failure")
@@ -51,13 +52,13 @@ func TestRetryMetadataFlushReturnsLastError(t *testing.T) {
 	})
 
 	var sleeps []time.Duration
-	metadataFlushSleep = func(d time.Duration) {
+	metadataFlushSleep = func(ctx context.Context, d time.Duration) {
 		sleeps = append(sleeps, d)
 	}
 
 	expectedErr := errors.New("permanent failure")
 	attempts := 0
-	err := retryMetadataFlush(func() error {
+	err := retryMetadataFlush(context.Background(), func() error {
 		attempts++
 		return expectedErr
 	}, nil)
@@ -79,3 +80,39 @@ func TestRetryMetadataFlushReturnsLastError(t *testing.T) {
 		}
 	}
 }
+
+// TestRetryMetadataFlushStopsOnCancel verifies that an interrupted flush (the
+// calling process was killed, so the FUSE cancel channel fired) abandons its
+// retries immediately instead of sleeping out the backoff, so the killed
+// process is not held in close() while the filer is overwhelmed.
+func TestRetryMetadataFlushStopsOnCancel(t *testing.T) {
+	originalSleep := metadataFlushSleep
+	t.Cleanup(func() {
+		metadataFlushSleep = originalSleep
+	})
+
+	var sleeps []time.Duration
+	metadataFlushSleep = func(ctx context.Context, d time.Duration) {
+		sleeps = append(sleeps, d)
+	}
+
+	ctx, cancel := context.WithCancel(context.Background())
+	cancel() // process already killed before the first attempt completes
+
+	attempts := 0
+	flushErr := errors.New("filer overwhelmed")
+	err := retryMetadataFlush(ctx, func() error {
+		attempts++
+		return flushErr
+	}, nil)
+	if !errors.Is(err, flushErr) {
+		t.Fatalf("retryMetadataFlush error = %v, want %v", err, flushErr)
+	}
+
+	if attempts != 1 {
+		t.Fatalf("attempts = %d, want 1 (no retries after cancel)", attempts)
+	}
+	if len(sleeps) != 0 {
+		t.Fatalf("sleeps = %v, want none (no backoff after cancel)", sleeps)
+	}
+}
diff --git a/weed/mount/weedfs_async_flush.go b/weed/mount/weedfs_async_flush.go
index 64b8f929d..4be34245c 100644
--- a/weed/mount/weedfs_async_flush.go
+++ b/weed/mount/weedfs_async_flush.go
@@ -1,6 +1,7 @@
 package mount
 
 import (
+	"context"
 	"time"
 
 	"github.com/seaweedfs/go-fuse/v2/fuse"
@@ -114,8 +115,8 @@ func (wfs *WFS) completeAsyncFlush(fh *FileHandle) {
 // with exponential backoff on transient errors. The chunk data is already on the
 // volume servers at this point; only the filer metadata reference needs persisting.
 func (wfs *WFS) flushMetadataWithRetry(fh *FileHandle, dir, name string, fileFullPath util.FullPath) {
-	err := retryMetadataFlush(func() error {
-		return wfs.flushMetadataToFiler(fh, dir, name, fh.asyncFlushUid, fh.asyncFlushGid)
+	err := retryMetadataFlush(context.Background(), func() error {
+		return wfs.flushMetadataToFiler(context.Background(), fh, dir, name, fh.asyncFlushUid, fh.asyncFlushGid)
 	}, func(nextAttempt, totalAttempts int, backoff time.Duration, err error) {
 		glog.Warningf("completeAsyncFlush %s: retrying metadata flush (attempt %d/%d) after %v: %v",
 			fileFullPath, nextAttempt, totalAttempts, backoff, err)
diff --git a/weed/mount/weedfs_file_io.go b/weed/mount/weedfs_file_io.go
index 959d05849..b40a489a4 100644
--- a/weed/mount/weedfs_file_io.go
+++ b/weed/mount/weedfs_file_io.go
@@ -1,6 +1,8 @@
 package mount
 
 import (
+	"context"
+
 	"github.com/seaweedfs/go-fuse/v2/fuse"
 	"github.com/seaweedfs/seaweedfs/weed/glog"
 )
@@ -143,7 +145,9 @@ func (wfs *WFS) Release(cancel <-chan struct{}, in *fuse.ReleaseIn) {
 	// the clean case, so the duplicate call after a normal Flush is cheap.
 	if fh := wfs.GetHandle(FileHandleId(in.Fh)); fh != nil {
 		allowAsync := in.ReleaseFlags&fuse.FUSE_RELEASE_FLOCK_UNLOCK == 0
-		if status := wfs.doFlush(fh, in.Uid, in.Gid, allowAsync); status != fuse.OK {
+		// Release is the last chance to persist the handle, so it must finish
+		// even if the triggering syscall was interrupted: non-cancellable context.
+		if status := wfs.doFlush(context.Background(), fh, in.Uid, in.Gid, allowAsync); status != fuse.OK {
 			glog.Warningf("release fh %d inode %d: fallback flush failed: %v", in.Fh, in.NodeId, status)
 		}
 	}
diff --git a/weed/mount/weedfs_file_mkrm.go b/weed/mount/weedfs_file_mkrm.go
index 5fc2d63a7..d7d4b7043 100644
--- a/weed/mount/weedfs_file_mkrm.go
+++ b/weed/mount/weedfs_file_mkrm.go
@@ -447,7 +447,7 @@ func (wfs *WFS) asyncCreateEntry(dirFullPath util.FullPath, entry *filer_pb.Entr
 			Signatures:               []int32{wfs.signature},
 			SkipCheckParentDirectory: true,
 		}
-		err := retryMetadataFlush(func() error {
+		err := retryMetadataFlush(context.Background(), func() error {
 			resp, createErr := wfs.streamCreateEntry(context.Background(), request)
 			if createErr != nil {
 				return createErr
diff --git a/weed/mount/weedfs_file_sync.go b/weed/mount/weedfs_file_sync.go
index 2a589c909..0082178a2 100644
--- a/weed/mount/weedfs_file_sync.go
+++ b/weed/mount/weedfs_file_sync.go
@@ -71,13 +71,36 @@ func (wfs *WFS) Flush(cancel <-chan struct{}, in *fuse.FlushIn) fuse.Status {
 	// would silently degrade to a blocking flush for ordinary close().
 	hasPosixLocks := wfs.hasPosixOwner(in.NodeId, in.LockOwner)
 	allowAsync := !hasPosixLocks
-	status := wfs.doFlush(fh, in.Uid, in.Gid, allowAsync)
+
+	// Abort the flush when the kernel interrupts the request (the calling
+	// process was killed); otherwise close() hangs in uninterruptible sleep
+	// while the metadata flush retries against an overwhelmed filer.
+	ctx, cancelFunc := fuseInterruptContext(cancel)
+	defer cancelFunc()
+
+	status := wfs.doFlush(ctx, fh, in.Uid, in.Gid, allowAsync)
 	if in.LockOwner != 0 {
 		wfs.releasePosixOwner(in.NodeId, in.LockOwner)
 	}
 	return status
 }
 
+// fuseInterruptContext returns a context cancelled when the FUSE cancel channel
+// fires (request interrupted). The caller must call the returned func.
+func fuseInterruptContext(cancel <-chan struct{}) (context.Context, context.CancelFunc) {
+	ctx, cancelFunc := context.WithCancel(context.Background())
+	if cancel != nil {
+		go func() {
+			select {
+			case <-cancel:
+				cancelFunc()
+			case <-ctx.Done():
+			}
+		}()
+	}
+	return ctx, cancelFunc
+}
+
 /**
  * Synchronize file contents
  *
@@ -104,12 +127,15 @@ func (wfs *WFS) Fsync(cancel <-chan struct{}, in *fuse.FsyncIn) (code fuse.Statu
 		return fuse.ENOENT
 	}
 
+	ctx, cancelFunc := fuseInterruptContext(cancel)
+	defer cancelFunc()
+
 	// Fsync is an explicit sync request — always flush synchronously
-	return wfs.doFlush(fh, in.Uid, in.Gid, false)
+	return wfs.doFlush(ctx, fh, in.Uid, in.Gid, false)
 
 }
 
-func (wfs *WFS) doFlush(fh *FileHandle, uid, gid uint32, allowAsync bool) fuse.Status {
+func (wfs *WFS) doFlush(ctx context.Context, fh *FileHandle, uid, gid uint32, allowAsync bool) fuse.Status {
 
 	// flush works at fh level
 	fileFullPath := fh.FullPath()
@@ -160,8 +186,8 @@ func (wfs *WFS) doFlush(fh *FileHandle, uid, gid uint32, allowAsync bool) fuse.S
 		return fuse.Status(syscall.ENOSPC)
 	}
 
-	if err := retryMetadataFlush(func() error {
-		return wfs.flushMetadataToFiler(fh, dir, name, uid, gid)
+	if err := retryMetadataFlush(ctx, func() error {
+		return wfs.flushMetadataToFiler(ctx, fh, dir, name, uid, gid)
 	}, func(nextAttempt, totalAttempts int, backoff time.Duration, err error) {
 		glog.Warningf("%v fh %d flush: retrying metadata flush (attempt %d/%d) after %v: %v",
 			fileFullPath, fh.fh, nextAttempt, totalAttempts, backoff, err)
@@ -183,7 +209,7 @@ func (wfs *WFS) doFlush(fh *FileHandle, uid, gid uint32, allowAsync bool) fuse.S
 // When -dlm is enabled, the distributed lock is already held by the FileHandle
 // from open-for-write through close, so no additional distributed lock is
 // needed here. The local fhLockTable lock below serializes within this mount.
-func (wfs *WFS) flushMetadataToFiler(fh *FileHandle, dir, name string, uid, gid uint32) error {
+func (wfs *WFS) flushMetadataToFiler(ctx context.Context, fh *FileHandle, dir, name string, uid, gid uint32) error {
 	fileFullPath := fh.FullPath()
 	glog.V(4).Infof("flushMetadataToFiler %s/%s inode %d fh %d", dir, name, fh.inode, fh.fh)
 
@@ -240,7 +266,7 @@ func (wfs *WFS) flushMetadataToFiler(fh *FileHandle, dir, name string, uid, gid
 
 	wfs.mapPbIdFromLocalToFiler(request.Entry)
 
-	resp, err := wfs.streamCreateEntry(context.Background(), request)
+	resp, err := wfs.streamCreateEntry(ctx, request)
 	if err != nil {
 		glog.Errorf("fh flush create %s: %v", fileFullPath, err)
 		return fmt.Errorf("fh flush create %s: %v", fileFullPath, err)
diff --git a/weed/mount/weedfs_file_sync_test.go b/weed/mount/weedfs_file_sync_test.go
index d9c9b38a3..044115646 100644
--- a/weed/mount/weedfs_file_sync_test.go
+++ b/weed/mount/weedfs_file_sync_test.go
@@ -6,6 +6,7 @@ import (
 	"math"
 	"math/rand/v2"
 	"testing"
+	"time"
 
 	"github.com/seaweedfs/seaweedfs/weed/filer"
 	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
@@ -327,9 +328,9 @@ func TestFlushCycleManifestAccumulation(t *testing.T) {
 		for slot := 0; slot < numSlots; slot++ {
 			entryChunks = append(entryChunks, &filer_pb.FileChunk{
 				Offset: int64(slot) * int64(chunkSize), Size: chunkSize,
-				FileId: fmt.Sprintf("%d,%x00000000", cycle+1, nextKey),
+				FileId:       fmt.Sprintf("%d,%x00000000", cycle+1, nextKey),
 				ModifiedTsNs: nextTs,
-				Fid: &filer_pb.FileId{VolumeId: uint32(cycle + 1), FileKey: nextKey, Cookie: 0},
+				Fid:          &filer_pb.FileId{VolumeId: uint32(cycle + 1), FileKey: nextKey, Cookie: 0},
 			})
 			nextTs++
 			nextKey++
@@ -442,3 +443,48 @@ func TestVisibleContentPreservedAfterCompact(t *testing.T) {
 		}
 	}
 }
+
+// TestFuseInterruptContextCancelsOnInterrupt verifies the interrupt wiring:
+// when the kernel interrupts a flush/fsync (the calling process was killed)
+// go-fuse closes the cancel channel, and the derived context must be cancelled
+// so the in-flight metadata RPC and its retries abort instead of leaving the
+// killed process stuck in uninterruptible sleep inside close().
+func TestFuseInterruptContextCancelsOnInterrupt(t *testing.T) {
+	cancel := make(chan struct{})
+	ctx, cancelFunc := fuseInterruptContext(cancel)
+	defer cancelFunc()
+
+	select {
+	case <-ctx.Done():
+		t.Fatal("context cancelled before the FUSE request was interrupted")
+	default:
+	}
+
+	close(cancel) // kernel sent FUSE_INTERRUPT (process killed)
+
+	select {
+	case <-ctx.Done():
+	case <-time.After(5 * time.Second):
+		t.Fatal("context not cancelled after FUSE interrupt")
+	}
+}
+
+// A nil cancel channel (no interrupt plumbing) must still yield a usable,
+// cancellable context that does not fire on its own.
+func TestFuseInterruptContextNilChannel(t *testing.T) {
+	ctx, cancelFunc := fuseInterruptContext(nil)
+	defer cancelFunc()
+
+	select {
+	case <-ctx.Done():
+		t.Fatal("context cancelled with no interrupt and no explicit cancel")
+	default:
+	}
+
+	cancelFunc()
+	select {
+	case <-ctx.Done():
+	case <-time.After(5 * time.Second):
+		t.Fatal("context not cancelled after explicit cancelFunc")
+	}
+}
diff --git a/weed/mount/weedfs_rename.go b/weed/mount/weedfs_rename.go
index d14eddb79..0fdcd1820 100644
--- a/weed/mount/weedfs_rename.go
+++ b/weed/mount/weedfs_rename.go
@@ -246,7 +246,8 @@ func (wfs *WFS) Rename(cancel <-chan struct{}, in *fuse.RenameIn, oldName string
 		// BEFORE any async flush interference.
 		if fh, ok := wfs.fhMap.FindFileHandle(inode); ok && fh.dirtyMetadata {
 			glog.V(4).Infof("dir Rename %s: flushing deferred metadata before rename", oldPath)
-			if flushStatus := wfs.doFlush(fh, oldEntry.Attributes.Uid, oldEntry.Attributes.Gid, false); flushStatus != fuse.OK {
+			// Prerequisite for the rename, so it must complete: non-cancellable context.
+			if flushStatus := wfs.doFlush(context.Background(), fh, oldEntry.Attributes.Uid, oldEntry.Attributes.Gid, false); flushStatus != fuse.OK {
 				glog.Warningf("dir Rename %s: flush before rename failed: %v", oldPath, flushStatus)
 				return flushStatus
 			}
diff --git a/weed/mount/wfs_save.go b/weed/mount/wfs_save.go
index be848fc2d..24f692ae2 100644
--- a/weed/mount/wfs_save.go
+++ b/weed/mount/wfs_save.go
@@ -28,7 +28,7 @@ func (wfs *WFS) saveEntry(path util.FullPath, entry *filer_pb.Entry) (code fuse.
 	glog.V(1).Infof("save entry: %v", request)
 
 	var resp *filer_pb.UpdateEntryResponse
-	err := retryMetadataFlushIf(func() error {
+	err := retryMetadataFlushIf(context.Background(), func() error {
 		var callErr error
 		resp, callErr = wfs.streamUpdateEntry(context.Background(), request)
 		return callErr
diff --git a/weed/operation/upload_content.go b/weed/operation/upload_content.go
index 075087ee0..99a8a57d6 100644
--- a/weed/operation/upload_content.go
+++ b/weed/operation/upload_content.go
@@ -109,6 +109,10 @@ var uploadRetryableAssignErrList = []string{
 	"Volume Size ",
 }
 
+// assignVolumeTimeout bounds a single AssignVolume RPC so an overwhelmed filer
+// can't block the caller forever. Overridable in tests.
+var assignVolumeTimeout = 30 * time.Second
+
 // HTTPClient interface for testing
 type HTTPClient interface {
 	Do(req *http.Request) (*http.Response, error)
@@ -210,7 +214,9 @@ func (uploader *Uploader) UploadWithRetry(filerClient filer_pb.FilerClient, assi
 	fileId, uploadResult, err = uploader.uploadWithRetryData(func() (fileId string, host string, auth security.EncodedJwt, err error) {
 		// grpc assign volume
 		if grpcAssignErr := filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
-			resp, assignErr := client.AssignVolume(context.Background(), assignRequest)
+			assignCtx, assignCancel := context.WithTimeout(context.Background(), assignVolumeTimeout)
+			defer assignCancel()
+			resp, assignErr := client.AssignVolume(assignCtx, assignRequest)
 			if assignErr != nil {
 				glog.V(0).Infof("assign volume failure %v: %v", assignRequest, assignErr)
 				return assignErr
diff --git a/weed/operation/upload_content_test.go b/weed/operation/upload_content_test.go
index b9d925e89..bd1e10e50 100644
--- a/weed/operation/upload_content_test.go
+++ b/weed/operation/upload_content_test.go
@@ -13,8 +13,10 @@ import (
 	"testing"
 	"time"
 
+	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
 	"github.com/seaweedfs/seaweedfs/weed/security"
 	"github.com/seaweedfs/seaweedfs/weed/storage/needle"
+	"google.golang.org/grpc"
 )
 
 type scriptedHTTPResponse struct {
@@ -163,6 +165,76 @@ func (c *bodyCapturingHTTPClient) Do(req *http.Request) (*http.Response, error)
 	}, nil
 }
 
+// hangingAssignSeaweedClient is a SeaweedFilerClient whose AssignVolume blocks
+// until its context is done, so a test can prove UploadWithRetry bounds the RPC.
+type hangingAssignSeaweedClient struct {
+	filer_pb.SeaweedFilerClient
+	sawDeadline chan bool
+}
+
+func (c *hangingAssignSeaweedClient) AssignVolume(ctx context.Context, in *filer_pb.AssignVolumeRequest, opts ...grpc.CallOption) (*filer_pb.AssignVolumeResponse, error) {
+	_, ok := ctx.Deadline()
+	// Non-blocking so a retry that calls this more than once can't wedge here.
+	select {
+	case c.sawDeadline <- ok:
+	default:
+	}
+	<-ctx.Done() // simulate an overwhelmed filer that never answers
+	return nil, ctx.Err()
+}
+
+type hangingAssignFilerClient struct {
+	inner *hangingAssignSeaweedClient
+}
+
+func (c *hangingAssignFilerClient) WithFilerClient(streamingMode bool, fn func(filer_pb.SeaweedFilerClient) error) error {
+	return fn(c.inner)
+}
+func (c *hangingAssignFilerClient) AdjustedUrl(loc *filer_pb.Location) string { return loc.GetUrl() }
+func (c *hangingAssignFilerClient) GetDataCenter() string                     { return "" }
+
+// TestUploadWithRetryBoundsAssignVolume covers the case where an AssignVolume
+// against an overwhelmed filer carried no deadline, so the upload (and the FUSE
+// flush driving it) blocked forever. UploadWithRetry must give the RPC a
+// deadline and return once it expires instead of hanging.
+func TestUploadWithRetryBoundsAssignVolume(t *testing.T) {
+	original := assignVolumeTimeout
+	assignVolumeTimeout = 200 * time.Millisecond
+	t.Cleanup(func() { assignVolumeTimeout = original })
+
+	client := &hangingAssignFilerClient{inner: &hangingAssignSeaweedClient{sawDeadline: make(chan bool, 1)}}
+	uploader := newUploader(&scriptedHTTPClient{responses: map[string][]scriptedHTTPResponse{}})
+
+	done := make(chan error, 1)
+	go func() {
+		_, _, err, _ := uploader.UploadWithRetry(client,
+			&filer_pb.AssignVolumeRequest{Count: 1},
+			&UploadOption{Filename: "test.bin"},
+			func(host, fileId string) string { return "http://" + host + "/" + fileId },
+			bytes.NewReader([]byte("abc")),
+		)
+		done <- err
+	}()
+
+	select {
+	case sawDeadline := <-client.inner.sawDeadline:
+		if !sawDeadline {
+			t.Fatal("AssignVolume received a context with no deadline")
+		}
+	case <-time.After(10 * time.Second):
+		t.Fatal("AssignVolume was never called")
+	}
+
+	select {
+	case err := <-done:
+		if err == nil {
+			t.Fatal("expected UploadWithRetry to fail once the assign deadline expired")
+		}
+	case <-time.After(10 * time.Second):
+		t.Fatal("UploadWithRetry hung well past the assign timeout")
+	}
+}
+
 // TestUploadRewindsBodyOnConnectionReset reproduces issue #9139 follow-up:
 // when the inner Do retry fires on a "connection reset" / "closed network"
 // error, the *bytes.Reader body has already been consumed, so without an