diff --git a/test/fuse_integration/write_buffer_cap_test.go b/test/fuse_integration/write_buffer_cap_test.go
new file mode 100644
index 000000000..0f7320db2
--- /dev/null
+++ b/test/fuse_integration/write_buffer_cap_test.go
@@ -0,0 +1,314 @@
+package fuse_test
+
+import (
+	"bytes"
+	"crypto/rand"
+	"fmt"
+	"io"
+	"net/http"
+	"os"
+	"path/filepath"
+	"runtime"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+)
+
+// mountDebugPort holds the debug/pprof port the test passed to the
+// mount process via -debug.port. It is set once at TestWriteBufferCap
+// entry and consulted from the write-timeout paths to fetch the mount
+// process's goroutine dump (the test's own dumpAllGoroutines only
+// covers the test process).
+var mountDebugPort int
+
+// fetchMountGoroutines pulls a full goroutine dump from the mount
+// process's pprof endpoint. If the mount debug port isn't configured
+// or the HTTP call fails, a short explanation is returned instead of
+// an error — this is diagnostic best-effort, not a test assertion.
+func fetchMountGoroutines() string {
+	if mountDebugPort == 0 {
+		return "(mount debug port not configured)"
+	}
+	url := fmt.Sprintf("http://127.0.0.1:%d/debug/pprof/goroutine?debug=2", mountDebugPort)
+	client := &http.Client{Timeout: 10 * time.Second}
+	resp, err := client.Get(url)
+	if err != nil {
+		return fmt.Sprintf("(failed to reach mount pprof at %s: %v)", url, err)
+	}
+	defer resp.Body.Close()
+	body, err := io.ReadAll(resp.Body)
+	if err != nil {
+		return fmt.Sprintf("(failed to read mount pprof body: %v)", err)
+	}
+	return string(body)
+}
+
+// dumpAllGoroutines returns a full stack trace of every live goroutine.
+// Used on write-timeout to give CI actionable diagnosis if the write
+// buffer cap ever re-regresses into a hang.
+func dumpAllGoroutines() string {
+	buf := make([]byte, 1<<20)
+	for {
+		n := runtime.Stack(buf, true)
+		if n < len(buf) {
+			return string(buf[:n])
+		}
+		buf = make([]byte, 2*len(buf))
+	}
+}
+
+// writeBufferCapConfig returns a TestConfig that exercises the new
+// -writeBufferSizeMB flag. The cap is set below the aggregate in-flight
+// write demand of the subtests below, so every new chunk has to pass
+// through the Reserve/Release backpressure path at least some of the
+// time. The cap is intentionally NOT minimal — over-tight settings
+// interact with the per-file writable-chunk limit and the FUSE MaxWrite
+// batching in ways that starve single-handle writers on slow CI.
+//
+// Also enables the mount's pprof debug endpoint so the test can fetch
+// mount-process goroutine dumps on write-timeout, which is the only
+// way to actually diagnose a backpressure deadlock (the test process
+// itself is just blocked in syscall.Write waiting on FUSE).
+func writeBufferCapConfig(debugPort int) *TestConfig {
+	return &TestConfig{
+		Collection:  "",
+		Replication: "000",
+		ChunkSizeMB: 2,   // 2 MiB chunks
+		CacheSizeMB: 100, // read cache (unrelated)
+		NumVolumes:  3,
+		EnableDebug: false,
+		MountOptions: []string{
+			// 16 MiB total write buffer ⇒ up to 8 chunks in flight
+			// across every open file handle on this mount. Large
+			// enough to avoid starving a single handle on a slow
+			// CI runner, small enough that the concurrent test
+			// below still has to drain through it.
+			"-writeBufferSizeMB=16",
+			"-debug=true",
+			fmt.Sprintf("-debug.port=%d", debugPort),
+			// Route glog to stderr so the framework's process log
+			// capture actually contains something — by default weed
+			// sends glog to /tmp/weed.* files which the CI artifact
+			// upload step never sees. Critical for diagnosing
+			// upload/saveToStorage errors on Linux runs.
+			"-logtostderr=true",
+			"-v=2",
+		},
+		SkipCleanup: false,
+	}
+}
+
+// writeWithTimeout wraps os.WriteFile with a hard deadline so a stuck
+// write fails the test fast instead of consuming the full job budget.
+// This is belt-and-braces around the 45-minute workflow timeout and
+// makes write-buffer regressions surface as an actionable failure.
+func writeWithTimeout(t *testing.T, path string, data []byte, timeout time.Duration) {
+	t.Helper()
+	done := make(chan error, 1)
+	go func() { done <- os.WriteFile(path, data, 0644) }()
+	select {
+	case err := <-done:
+		if err != nil {
+			// Dump mount goroutines on any write error, not just
+			// timeout — upload failures that surface via close()
+			// as EIO leave the mount process running but in an
+			// informative state (pending sealed chunks, error
+			// counters, etc).
+			t.Logf("write %s failed (%v) — dumping MOUNT goroutines:\n%s", path, err, fetchMountGoroutines())
+		}
+		require.NoError(t, err, "write %s", path)
+	case <-time.After(timeout):
+		t.Logf("write %s did not finish within %v — dumping TEST goroutines:\n%s", path, timeout, dumpAllGoroutines())
+		t.Logf("dumping MOUNT goroutines:\n%s", fetchMountGoroutines())
+		t.Fatalf("write %s timed out — write buffer cap is likely leaking or deadlocking", path)
+	}
+}
+
+// runSubtestWithWatchdog runs body on the current (subtest main)
+// goroutine and starts a watcher goroutine that logs diagnostics and
+// fails the test if body doesn't return within timeout.
+//
+// body must run on the main goroutine because test helpers inside it
+// (require.NoError, writeWithTimeout's own t.Fatalf on its internal
+// timeout) need t.Fatal / t.FailNow, which Go's testing docs restrict
+// to the goroutine running the test function. The watcher goroutine
+// only calls goroutine-safe t methods (t.Log, t.Logf, t.Errorf) so it
+// can mark the test failed and dump diagnostics without violating
+// that contract. If body is stuck past timeout the watcher still
+// surfaces the wedge (test + mount goroutine dumps + a FAIL mark);
+// body itself gets unblocked either by its own inner writeWithTimeout
+// firing t.Fatalf or by Go test's global -timeout.
+func runSubtestWithWatchdog(t *testing.T, timeout time.Duration, body func(t *testing.T)) {
+	t.Helper()
+	stop := make(chan struct{})
+	defer close(stop)
+	go func() {
+		select {
+		case <-stop:
+			return
+		case <-time.After(timeout):
+			t.Logf("subtest exceeded %v watchdog — dumping TEST goroutines:\n%s", timeout, dumpAllGoroutines())
+			t.Logf("dumping MOUNT goroutines:\n%s", fetchMountGoroutines())
+			t.Errorf("subtest exceeded %v watchdog — see goroutine dumps above", timeout)
+		}
+	}()
+	body(t)
+}
+
+// TestWriteBufferCap exercises the end-to-end write-buffer cap on a
+// real FUSE mount. Without the cap, a volume-server stall would let
+// the swap file grow without bound (issue #8777). With the cap, writers
+// must serialize through a bounded budget while still producing correct
+// output — that correctness (and the absence of deadlocks) is what
+// this test verifies.
+//
+// Note: this test deliberately does not assert that Reserve *blocked*
+// at some observed used-byte peak. The mount runs as a subprocess so
+// its in-process WriteBufferAccountant state is not reachable from the
+// test without adding a metrics/RPC surface to the mount binary. The
+// deterministic peak-vs-cap assertion instead lives in the in-package
+// unit test TestWriteBufferCap_SharedAcrossPipelines, which drives a
+// controlled gated uploader and samples Used() throughout the run.
+func TestWriteBufferCap(t *testing.T) {
+	mountDebugPort = freePort(t)
+	config := writeBufferCapConfig(mountDebugPort)
+	framework := NewFuseTestFramework(t, config)
+	defer framework.Cleanup()
+
+	require.NoError(t, framework.Setup(config))
+
+	const subtestTimeout = 3 * time.Minute
+
+	t.Run("ConcurrentWritesUnderCap", func(t *testing.T) {
+		runSubtestWithWatchdog(t, subtestTimeout, func(t *testing.T) {
+			testConcurrentWritesUnderCap(t, framework)
+		})
+	})
+
+	t.Run("LargeFileUnderCap", func(t *testing.T) {
+		runSubtestWithWatchdog(t, subtestTimeout, func(t *testing.T) {
+			testLargeFileUnderCap(t, framework)
+		})
+	})
+
+	t.Run("DoesNotDeadlockAfterPressure", func(t *testing.T) {
+		runSubtestWithWatchdog(t, subtestTimeout, func(t *testing.T) {
+			testWriteBufferNoDeadlockAfterPressure(t, framework)
+		})
+	})
+}
+
+// testConcurrentWritesUnderCap opens several files in parallel with
+// aggregate demand that exceeds the 16 MiB write buffer cap, then
+// verifies every byte survived the round trip.
+func testConcurrentWritesUnderCap(t *testing.T, framework *FuseTestFramework) {
+	const (
+		numFiles = 4
+		fileSize = 8 * 1024 * 1024 // 8 MiB per file ⇒ 32 MiB total vs 16 MiB cap
+	)
+
+	dir := "write_buffer_cap_concurrent"
+	framework.CreateTestDir(dir)
+
+	payloads := make([][]byte, numFiles)
+	for i := range payloads {
+		buf := make([]byte, fileSize)
+		_, err := rand.Read(buf)
+		require.NoError(t, err)
+		payloads[i] = buf
+	}
+
+	start := time.Now()
+	var wg sync.WaitGroup
+	errs := make(chan error, numFiles)
+	timedOut := make(chan struct{}, numFiles)
+	for i := 0; i < numFiles; i++ {
+		i := i
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			name := fmt.Sprintf("file_%02d.bin", i)
+			path := filepath.Join(framework.GetMountPoint(), dir, name)
+			done := make(chan error, 1)
+			go func() { done <- os.WriteFile(path, payloads[i], 0644) }()
+			select {
+			case err := <-done:
+				if err != nil {
+					errs <- fmt.Errorf("writer %d: %w", i, err)
+				}
+			case <-time.After(90 * time.Second):
+				timedOut <- struct{}{}
+				errs <- fmt.Errorf("writer %d: timed out after 90s", i)
+			}
+		}()
+	}
+	wg.Wait()
+	close(errs)
+	// If any writer timed out, dump every live goroutine so CI shows the
+	// wedge instead of just a walltime.
+	select {
+	case <-timedOut:
+		t.Logf("at least one concurrent writer timed out — dumping goroutines:\n%s", dumpAllGoroutines())
+	default:
+	}
+	for err := range errs {
+		t.Fatal(err)
+	}
+	t.Logf("wrote %d × %d MiB under 16 MiB cap in %v", numFiles, fileSize/(1024*1024), time.Since(start))
+
+	for i := 0; i < numFiles; i++ {
+		name := fmt.Sprintf("file_%02d.bin", i)
+		path := filepath.Join(framework.GetMountPoint(), dir, name)
+		got, err := os.ReadFile(path)
+		require.NoError(t, err, "read %s", name)
+		require.Equal(t, len(payloads[i]), len(got), "size mismatch for %s", name)
+		if !bytes.Equal(payloads[i], got) {
+			t.Fatalf("content mismatch for %s", name)
+		}
+	}
+}
+
+// testLargeFileUnderCap writes a single file whose size exceeds the
+// 16 MiB cap through a single handle, verifying that the pipeline
+// drains its own earlier chunks and makes forward progress rather than
+// self-deadlocking when the global budget is already full of its own
+// earlier sealed chunks.
+func testLargeFileUnderCap(t *testing.T, framework *FuseTestFramework) {
+	const fileSize = 20 * 1024 * 1024 // 20 MiB ⇒ 10 chunks vs 8-slot budget
+
+	payload := make([]byte, fileSize)
+	_, err := rand.Read(payload)
+	require.NoError(t, err)
+
+	name := "write_buffer_cap_large.bin"
+	path := filepath.Join(framework.GetMountPoint(), name)
+
+	start := time.Now()
+	writeWithTimeout(t, path, payload, 90*time.Second)
+	t.Logf("wrote %d MiB through one handle under 16 MiB cap in %v", fileSize/(1024*1024), time.Since(start))
+
+	got, err := os.ReadFile(path)
+	require.NoError(t, err)
+	require.Equal(t, len(payload), len(got))
+	if !bytes.Equal(payload, got) {
+		t.Fatal("content mismatch on large single-handle write")
+	}
+}
+
+// testWriteBufferNoDeadlockAfterPressure verifies the mount is still
+// healthy after being driven against the cap. A budget-slot leak would
+// eventually cause every new chunk allocation to hang; a quick canary
+// write catches that as a hard failure.
+func testWriteBufferNoDeadlockAfterPressure(t *testing.T, framework *FuseTestFramework) {
+	name := "write_buffer_cap_canary.txt"
+	path := filepath.Join(framework.GetMountPoint(), name)
+	content := []byte("write buffer cap canary — mount still healthy")
+
+	writeWithTimeout(t, path, content, 30*time.Second)
+
+	got, err := os.ReadFile(path)
+	require.NoError(t, err)
+	require.Equal(t, content, got)
+}
diff --git a/weed/command/fuse_std.go b/weed/command/fuse_std.go
index 3fb7236af..7a5400dc5 100644
--- a/weed/command/fuse_std.go
+++ b/weed/command/fuse_std.go
@@ -189,6 +189,12 @@ func runFuse(cmd *Command, args []string) bool {
 			}
 		case "cacheDirWrite":
 			mountOptions.cacheDirForWrite = &parameter.value
+		case "writeBufferSizeMB":
+			if parsed, err := strconv.ParseInt(parameter.value, 0, 64); err == nil {
+				mountOptions.writeBufferSizeMB = &parsed
+			} else {
+				panic(fmt.Errorf("writeBufferSizeMB: %s", err))
+			}
 		case "dataCenter":
 			mountOptions.dataCenter = &parameter.value
 		case "allowOthers":
diff --git a/weed/command/mount.go b/weed/command/mount.go
index aa0589c7c..5803cd2e5 100644
--- a/weed/command/mount.go
+++ b/weed/command/mount.go
@@ -22,6 +22,7 @@ type MountOptions struct {
 	cacheDirForRead      *string
 	cacheDirForWrite     *string
 	cacheSizeMBForRead   *int64
+	writeBufferSizeMB    *int64
 	dataCenter           *string
 	allowOthers          *bool
 	defaultPermissions   *bool
@@ -97,6 +98,7 @@ func init() {
 	mountOptions.cacheDirForRead = cmdMount.Flag.String("cacheDir", os.TempDir(), "local cache directory for file chunks and meta data")
 	mountOptions.cacheSizeMBForRead = cmdMount.Flag.Int64("cacheCapacityMB", 128, "file chunk read cache capacity in MB")
 	mountOptions.cacheDirForWrite = cmdMount.Flag.String("cacheDirWrite", "", "buffer writes mostly for large files")
+	mountOptions.writeBufferSizeMB = cmdMount.Flag.Int64("writeBufferSizeMB", 0, "global cap on the per-mount write buffer (memory + swap) in MB, 0 means unlimited. Bounds /tmp growth when volume uploads stall")
 	mountOptions.cacheMetaTtlSec = cmdMount.Flag.Int("cacheMetaTtlSec", 60, "metadata cache validity seconds")
 	mountOptions.dataCenter = cmdMount.Flag.String("dataCenter", "", "prefer to write to the data center")
 	mountOptions.allowOthers = cmdMount.Flag.Bool("allowOthers", true, "allows other users to access the file system")
diff --git a/weed/command/mount_std.go b/weed/command/mount_std.go
index ef23e8210..0ca6bb654 100644
--- a/weed/command/mount_std.go
+++ b/weed/command/mount_std.go
@@ -328,6 +328,7 @@ func RunMount(option *MountOptions, umask os.FileMode) bool {
 		CacheDirForRead:             *option.cacheDirForRead,
 		CacheSizeMBForRead:          *option.cacheSizeMBForRead,
 		CacheDirForWrite:            cacheDirForWrite,
+		WriteBufferSizeMB:           *option.writeBufferSizeMB,
 		CacheMetaTTlSec:             *option.cacheMetaTtlSec,
 		DataCenter:                  *option.dataCenter,
 		Quota:                       int64(*option.collectionQuota) * 1024 * 1024,
diff --git a/weed/mount/dirty_pages_chunked.go b/weed/mount/dirty_pages_chunked.go
index 52bc8c80a..0440d2057 100644
--- a/weed/mount/dirty_pages_chunked.go
+++ b/weed/mount/dirty_pages_chunked.go
@@ -33,7 +33,8 @@ func newMemoryChunkPages(fh *FileHandle, chunkSize int64) *ChunkedDirtyPages {
 	swapFileDir := fh.wfs.option.getUniqueCacheDirForWrite()
 
 	dirtyPages.uploadPipeline = page_writer.NewUploadPipeline(fh.wfs.concurrentWriters, chunkSize,
-		dirtyPages.saveChunkedFileIntervalToStorage, fh.wfs.option.ConcurrentWriters, swapFileDir)
+		dirtyPages.saveChunkedFileIntervalToStorage, fh.wfs.option.ConcurrentWriters, swapFileDir,
+		fh.wfs.writeBufferAccountant)
 
 	return dirtyPages
 }
diff --git a/weed/mount/page_writer/chunk_interval_list.go b/weed/mount/page_writer/chunk_interval_list.go
index 005385c1a..524d398c7 100644
--- a/weed/mount/page_writer/chunk_interval_list.go
+++ b/weed/mount/page_writer/chunk_interval_list.go
@@ -55,8 +55,29 @@ func (list *ChunkWrittenIntervalList) MarkWritten(startOffset, stopOffset, tsNs
 	list.addInterval(interval)
 }
 
+// IsComplete reports whether every byte of [0, chunkSize) has been
+// written, possibly via multiple adjacent or overlapping intervals.
+// addInterval does not merge adjacent intervals — a chunk filled by
+// two 1 MiB writes ends up as {[0,1M], [1M,2M]}, not one [0,2M] —
+// so checking list.size()==1 misses the "filled by adjacent writes"
+// case, leaving the chunk pinned in writableChunks even though all
+// its bytes are present. That latent bug became a hard deadlock once
+// -writeBufferSizeMB started reserving a global slot per writable
+// chunk: the chunks never got sealed, no uploader ran, no slot was
+// ever released, and the FUSE writer blocked in Reserve forever
+// (seaweedfs issue #8777 / PR #9066). Walking the list and tracking
+// the furthest covered offset detects adjacency correctly.
 func (list *ChunkWrittenIntervalList) IsComplete(chunkSize int64) bool {
-	return list.size() == 1 && list.head.next.isComplete(chunkSize)
+	var covered int64
+	for t := list.head.next; t != list.tail; t = t.next {
+		if t.StartOffset > covered {
+			return false // gap before this interval
+		}
+		if t.stopOffset > covered {
+			covered = t.stopOffset
+		}
+	}
+	return covered >= chunkSize
 }
 func (list *ChunkWrittenIntervalList) WrittenSize() (writtenByteCount int64) {
 	for t := list.head; t != nil; t = t.next {
diff --git a/weed/mount/page_writer/chunk_interval_list_test.go b/weed/mount/page_writer/chunk_interval_list_test.go
index 802130d6f..558a1f448 100644
--- a/weed/mount/page_writer/chunk_interval_list_test.go
+++ b/weed/mount/page_writer/chunk_interval_list_test.go
@@ -80,3 +80,79 @@ func hasData(usage *ChunkWrittenIntervalList, chunkStartOffset, x int64) bool {
 	}
 	return false
 }
+
+func TestIsComplete_AdjacentIntervals(t *testing.T) {
+	// Linux FUSE delivers writes up to FUSE_MAX_PAGES_PER_REQ
+	// (typically 1 MiB) per op, so a 2 MiB chunk filled by sequential
+	// writes arrives as two adjacent 1 MiB writes. addInterval does
+	// not merge adjacent intervals into one, but IsComplete must
+	// still report the chunk as fully covered — otherwise
+	// maybeMoveToSealed never fires, the chunk stays writable
+	// forever, and the write buffer cap cannot drain. Regression for
+	// the TestWriteBufferCap Linux CI deadlock on PR #9066.
+	const chunkSize int64 = 2 * 1024 * 1024
+
+	cases := []struct {
+		name     string
+		writes   [][2]int64
+		expected bool
+	}{
+		{
+			name:     "single full write",
+			writes:   [][2]int64{{0, chunkSize}},
+			expected: true,
+		},
+		{
+			name:     "two adjacent halves in order",
+			writes:   [][2]int64{{0, chunkSize / 2}, {chunkSize / 2, chunkSize}},
+			expected: true,
+		},
+		{
+			name:     "two adjacent halves out of order",
+			writes:   [][2]int64{{chunkSize / 2, chunkSize}, {0, chunkSize / 2}},
+			expected: true,
+		},
+		{
+			name: "eight adjacent eighths",
+			writes: func() [][2]int64 {
+				const n = 8
+				step := chunkSize / n
+				out := make([][2]int64, 0, n)
+				for i := int64(0); i < n; i++ {
+					out = append(out, [2]int64{i * step, (i + 1) * step})
+				}
+				return out
+			}(),
+			expected: true,
+		},
+		{
+			name:     "gap in the middle",
+			writes:   [][2]int64{{0, chunkSize / 4}, {chunkSize / 2, chunkSize}},
+			expected: false,
+		},
+		{
+			name:     "missing last byte",
+			writes:   [][2]int64{{0, chunkSize - 1}},
+			expected: false,
+		},
+		{
+			name:     "missing first byte",
+			writes:   [][2]int64{{1, chunkSize}},
+			expected: false,
+		},
+		{
+			name:     "overlapping intervals covering everything",
+			writes:   [][2]int64{{0, chunkSize*3/4 + 1}, {chunkSize / 2, chunkSize}},
+			expected: true,
+		},
+	}
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			list := newChunkWrittenIntervalList()
+			for i, w := range tc.writes {
+				list.MarkWritten(w[0], w[1], int64(i+1))
+			}
+			assert.Equal(t, tc.expected, list.IsComplete(chunkSize), "IsComplete mismatch")
+		})
+	}
+}
diff --git a/weed/mount/page_writer/upload_pipeline.go b/weed/mount/page_writer/upload_pipeline.go
index 4a0f1867c..a6dcb38f3 100644
--- a/weed/mount/page_writer/upload_pipeline.go
+++ b/weed/mount/page_writer/upload_pipeline.go
@@ -25,22 +25,43 @@ type UploadPipeline struct {
 	sealedChunks       map[LogicChunkIndex]*SealedChunk
 	activeReadChunks   map[LogicChunkIndex]int
 	readerCountCond    *sync.Cond
+	accountant         *WriteBufferAccountant
 }
 
 type SealedChunk struct {
 	chunk            PageChunk
 	referenceCounter int // track uploading or reading processes
+	// accountant and chunkSize are captured when the chunk is sealed so
+	// FreeReference can release the global write-budget slot exactly once
+	// regardless of which code path triggers the final deref (normal
+	// upload completion, Shutdown, or replace-in-place in moveToSealed).
+	accountant *WriteBufferAccountant
+	chunkSize  int64
 }
 
 func (sc *SealedChunk) FreeReference(messageOnFree string) {
+	// Early-return guard so repeated calls (Shutdown racing the async
+	// uploader, or any future caller that loses track of ownership) are
+	// strict no-ops rather than driving referenceCounter negative.
+	if sc.referenceCounter <= 0 {
+		return
+	}
 	sc.referenceCounter--
 	if sc.referenceCounter == 0 {
 		glog.V(4).Infof("Free sealed chunk: %s", messageOnFree)
 		sc.chunk.FreeResource()
+		sc.accountant.Release(sc.chunkSize)
 	}
 }
 
-func NewUploadPipeline(writers *util.LimitedConcurrentExecutor, chunkSize int64, saveToStorageFn SaveToStorageFunc, bufferChunkLimit int, swapFileDir string) *UploadPipeline {
+// NewUploadPipeline constructs an UploadPipeline. accountant may be nil,
+// in which case no global write-buffer cap is enforced. When non-nil,
+// creating a new page chunk (memory or swap) first reserves ChunkSize
+// bytes against it, blocking the writer if the global cap is reached.
+// The accountant is captured at construction so the pipeline's hot paths
+// (SaveDataAt, moveToSealed, Shutdown) can read up.accountant without
+// any synchronization.
+func NewUploadPipeline(writers *util.LimitedConcurrentExecutor, chunkSize int64, saveToStorageFn SaveToStorageFunc, bufferChunkLimit int, swapFileDir string, accountant *WriteBufferAccountant) *UploadPipeline {
 	t := &UploadPipeline{
 		ChunkSize:          chunkSize,
 		writableChunks:     make(map[LogicChunkIndex]PageChunk),
@@ -51,6 +72,7 @@ func NewUploadPipeline(writers *util.LimitedConcurrentExecutor, chunkSize int64,
 		activeReadChunks:   make(map[LogicChunkIndex]int),
 		writableChunkLimit: bufferChunkLimit,
 		swapFile:           NewSwapFile(swapFileDir, chunkSize),
+		accountant:         accountant,
 	}
 	t.readerCountCond = sync.NewCond(&t.chunksLock)
 	return t
@@ -64,6 +86,22 @@ func (up *UploadPipeline) SaveDataAt(p []byte, off int64, isSequential bool, tsN
 	logicChunkIndex := LogicChunkIndex(off / up.ChunkSize)
 
 	pageChunk, found := up.writableChunks[logicChunkIndex]
+	if !found {
+		// Reserve a chunk-sized slot against the global write budget before
+		// allocating. Reserve may block when volumes are full and sealed
+		// chunks can't drain, so we must release chunksLock first — the
+		// uploader goroutines that eventually call Release take chunksLock.
+		if up.accountant != nil {
+			up.chunksLock.Unlock()
+			up.accountant.Reserve(up.ChunkSize)
+			up.chunksLock.Lock()
+			// Re-check: another writer on the same file may have created
+			// the chunk while we were blocked. If so, give the slot back.
+			if pageChunk, found = up.writableChunks[logicChunkIndex]; found {
+				up.accountant.Release(up.ChunkSize)
+			}
+		}
+	}
 	if !found {
 		if len(up.writableChunks) > up.writableChunkLimit {
 			// if current file chunks is over the per file buffer count limit
@@ -105,6 +143,7 @@ func (up *UploadPipeline) SaveDataAt(p []byte, off int64, isSequential bool, tsN
 			pageChunk = up.swapFile.NewSwapFileChunk(logicChunkIndex)
 			// fmt.Printf(" create file chunk %d\n", logicChunkIndex)
 			if pageChunk == nil {
+				up.accountant.Release(up.ChunkSize)
 				return 0, fmt.Errorf("failed to create swap file chunk")
 			}
 		}
@@ -171,8 +210,8 @@ func (up *UploadPipeline) maybeMoveToSealed(memChunk PageChunk, logicChunkIndex
 }
 
 func (up *UploadPipeline) moveToSealed(memChunk PageChunk, logicChunkIndex LogicChunkIndex) {
-	atomic.AddInt32(&up.uploaderCount, 1)
-	glog.V(4).Infof("%s uploaderCount %d ++> %d", up.filepath, up.uploaderCount-1, up.uploaderCount)
+	newCount := atomic.AddInt32(&up.uploaderCount, 1)
+	glog.V(4).Infof("%s uploaderCount %d ++> %d", up.filepath, newCount-1, newCount)
 
 	if oldMemChunk, found := up.sealedChunks[logicChunkIndex]; found {
 		oldMemChunk.FreeReference(fmt.Sprintf("%s replace chunk %d", up.filepath, logicChunkIndex))
@@ -180,6 +219,8 @@ func (up *UploadPipeline) moveToSealed(memChunk PageChunk, logicChunkIndex Logic
 	sealedChunk := &SealedChunk{
 		chunk:            memChunk,
 		referenceCounter: 1, // default 1 is for uploading process
+		accountant:       up.accountant,
+		chunkSize:        up.ChunkSize,
 	}
 	up.sealedChunks[logicChunkIndex] = sealedChunk
 	delete(up.writableChunks, logicChunkIndex)
@@ -191,8 +232,8 @@ func (up *UploadPipeline) moveToSealed(memChunk PageChunk, logicChunkIndex Logic
 		sealedChunk.chunk.SaveContent(up.saveToStorageFn)
 
 		// notify waiting process
-		atomic.AddInt32(&up.uploaderCount, -1)
-		glog.V(4).Infof("%s uploaderCount %d --> %d", up.filepath, up.uploaderCount+1, up.uploaderCount)
+		newCount := atomic.AddInt32(&up.uploaderCount, -1)
+		glog.V(4).Infof("%s uploaderCount %d --> %d", up.filepath, newCount+1, newCount)
 		// Lock and Unlock are not required,
 		// but it may signal multiple times during one wakeup,
 		// and the waiting goroutine may miss some of them!
@@ -220,7 +261,20 @@ func (up *UploadPipeline) Shutdown() {
 
 	up.chunksLock.Lock()
 	defer up.chunksLock.Unlock()
+	// Free any writable chunks that were reserved but never sealed — on the
+	// Destroy() path (truncate / metadata invalidation) there is no
+	// preceding FlushData(), so dirty writable chunks would otherwise leak
+	// both their memory and their write-budget slots.
+	for logicChunkIndex, writableChunk := range up.writableChunks {
+		glog.V(4).Infof("%s uploadpipeline shutdown writable chunk %d", up.filepath, logicChunkIndex)
+		writableChunk.FreeResource()
+		up.accountant.Release(up.ChunkSize)
+		delete(up.writableChunks, logicChunkIndex)
+	}
 	for logicChunkIndex, sealedChunk := range up.sealedChunks {
+		// FreeReference releases the accountant slot on the refcount-zero
+		// transition; a racing async uploader will call FreeReference again
+		// and be a no-op, so there is no double-release.
 		sealedChunk.FreeReference(fmt.Sprintf("%s uploadpipeline shutdown chunk %d", up.filepath, logicChunkIndex))
 	}
 }
diff --git a/weed/mount/page_writer/upload_pipeline_shutdown_test.go b/weed/mount/page_writer/upload_pipeline_shutdown_test.go
new file mode 100644
index 000000000..8549f5677
--- /dev/null
+++ b/weed/mount/page_writer/upload_pipeline_shutdown_test.go
@@ -0,0 +1,61 @@
+package page_writer
+
+import (
+	"io"
+	"testing"
+)
+
+// TestShutdown_ReleasesWritableReservation verifies that calling Shutdown
+// on a pipeline with dirty-but-unsealed writable chunks returns those
+// chunks' reservations to the accountant. Regression for the Destroy()
+// path (truncate / metadata invalidation) which never flushes before
+// dropping the pipeline, so writable chunks would otherwise leak their
+// write-budget slots permanently.
+func TestShutdown_ReleasesWritableReservation(t *testing.T) {
+	acc := NewWriteBufferAccountant(1024 * 1024)
+	saveFn := func(io.Reader, int64, int64, int64, func()) {}
+	up := NewUploadPipeline(nil, 64*1024, saveFn, 16, t.TempDir(), acc)
+
+	// A small partial write lands in writableChunks as a mem chunk and
+	// stays there (the chunk is not complete so maybeMoveToSealed is a
+	// no-op) — exactly the state Destroy() can hit.
+	buf := make([]byte, 128)
+	if _, err := up.SaveDataAt(buf, 0, true, 1); err != nil {
+		t.Fatalf("SaveDataAt: %v", err)
+	}
+	if got := acc.Used(); got != 64*1024 {
+		t.Fatalf("expected 64KiB reserved after one write, got %d", got)
+	}
+
+	up.Shutdown()
+
+	if got := acc.Used(); got != 0 {
+		t.Fatalf("expected 0 used after Shutdown, got %d (writable reservation leaked)", got)
+	}
+}
+
+// TestSealedChunk_FreeReferenceIsIdempotent verifies that FreeReference
+// releases the accountant slot exactly once even if it is invoked more
+// than once. This guards the path where Shutdown and the async uploader
+// goroutine both call FreeReference on the same sealed chunk: the second
+// call must be a no-op, not a double-release.
+func TestSealedChunk_FreeReferenceIsIdempotent(t *testing.T) {
+	acc := NewWriteBufferAccountant(1024 * 1024)
+	acc.Reserve(64 * 1024)
+
+	sc := &SealedChunk{
+		chunk:            NewMemChunk(0, 64*1024),
+		referenceCounter: 1,
+		accountant:       acc,
+		chunkSize:        64 * 1024,
+	}
+
+	sc.FreeReference("first")
+	if got := acc.Used(); got != 0 {
+		t.Fatalf("expected 0 used after first FreeReference, got %d", got)
+	}
+	sc.FreeReference("second")
+	if got := acc.Used(); got != 0 {
+		t.Fatalf("expected 0 used after second FreeReference, got %d (double release)", got)
+	}
+}
diff --git a/weed/mount/page_writer/upload_pipeline_test.go b/weed/mount/page_writer/upload_pipeline_test.go
index b84b61b71..82f6acf14 100644
--- a/weed/mount/page_writer/upload_pipeline_test.go
+++ b/weed/mount/page_writer/upload_pipeline_test.go
@@ -8,7 +8,7 @@ import (
 
 func TestUploadPipeline(t *testing.T) {
 
-	uploadPipeline := NewUploadPipeline(nil, 2*1024*1024, nil, 16, "")
+	uploadPipeline := NewUploadPipeline(nil, 2*1024*1024, nil, 16, "", nil)
 
 	writeRange(uploadPipeline, 0, 131072)
 	writeRange(uploadPipeline, 131072, 262144)
diff --git a/weed/mount/page_writer/write_buffer_accountant.go b/weed/mount/page_writer/write_buffer_accountant.go
new file mode 100644
index 000000000..5f594c06d
--- /dev/null
+++ b/weed/mount/page_writer/write_buffer_accountant.go
@@ -0,0 +1,65 @@
+package page_writer
+
+import (
+	"sync"
+)
+
+// WriteBufferAccountant enforces a global byte budget across all
+// UploadPipeline instances. Callers Reserve chunk-sized slots before
+// allocating a page chunk (in memory or on swap) and Release them when
+// the chunk is freed. Reserve blocks when the cap would be exceeded,
+// providing natural backpressure to the FUSE write path when volume
+// uploads stall (e.g. all assigned volumes are full) instead of letting
+// the swap file grow without bound.
+//
+// A nil receiver is treated as "unlimited" for backward compatibility.
+type WriteBufferAccountant struct {
+	mu   sync.Mutex
+	cond *sync.Cond
+	cap  int64 // 0 means unlimited
+	used int64
+}
+
+func NewWriteBufferAccountant(capBytes int64) *WriteBufferAccountant {
+	a := &WriteBufferAccountant{cap: capBytes}
+	a.cond = sync.NewCond(&a.mu)
+	return a
+}
+
+// Reserve blocks until n bytes can be accounted for under the cap.
+// It must not be called while holding any UploadPipeline lock, or the
+// uploader goroutines that eventually call Release will deadlock.
+func (a *WriteBufferAccountant) Reserve(n int64) {
+	if a == nil || a.cap <= 0 {
+		return
+	}
+	a.mu.Lock()
+	defer a.mu.Unlock()
+	for a.used+n > a.cap && a.used > 0 {
+		a.cond.Wait()
+	}
+	a.used += n
+}
+
+func (a *WriteBufferAccountant) Release(n int64) {
+	if a == nil || a.cap <= 0 {
+		return
+	}
+	a.mu.Lock()
+	a.used -= n
+	if a.used < 0 {
+		a.used = 0
+	}
+	a.cond.Broadcast()
+	a.mu.Unlock()
+}
+
+// Used returns the currently reserved byte count (for tests/metrics).
+func (a *WriteBufferAccountant) Used() int64 {
+	if a == nil || a.cap <= 0 {
+		return 0
+	}
+	a.mu.Lock()
+	defer a.mu.Unlock()
+	return a.used
+}
diff --git a/weed/mount/page_writer/write_buffer_accountant_test.go b/weed/mount/page_writer/write_buffer_accountant_test.go
new file mode 100644
index 000000000..f9dd8a9e1
--- /dev/null
+++ b/weed/mount/page_writer/write_buffer_accountant_test.go
@@ -0,0 +1,89 @@
+package page_writer
+
+import (
+	"sync/atomic"
+	"testing"
+	"time"
+)
+
+func TestWriteBufferAccountant_Unlimited(t *testing.T) {
+	a := NewWriteBufferAccountant(0)
+	a.Reserve(1 << 30)
+	a.Reserve(1 << 30)
+	if got := a.Used(); got != 0 {
+		t.Fatalf("unlimited accountant should not track usage, got %d", got)
+	}
+}
+
+func TestWriteBufferAccountant_Nil(t *testing.T) {
+	var a *WriteBufferAccountant
+	a.Reserve(100) // must not panic
+	a.Release(100)
+	if a.Used() != 0 {
+		t.Fatal("nil Used should return 0")
+	}
+}
+
+func TestWriteBufferAccountant_ReserveAndRelease(t *testing.T) {
+	a := NewWriteBufferAccountant(300)
+	a.Reserve(100)
+	a.Reserve(100)
+	if got := a.Used(); got != 200 {
+		t.Fatalf("expected 200 used, got %d", got)
+	}
+	a.Release(100)
+	if got := a.Used(); got != 100 {
+		t.Fatalf("expected 100 used after release, got %d", got)
+	}
+}
+
+func TestWriteBufferAccountant_BlocksWhenOverCap(t *testing.T) {
+	a := NewWriteBufferAccountant(100)
+	a.Reserve(100)
+
+	started := make(chan struct{})
+	var landed atomic.Bool
+	go func() {
+		close(started)
+		a.Reserve(50) // should block — used=100, cap=100
+		landed.Store(true)
+	}()
+
+	// Wait until the goroutine is about to enter Reserve. Once it does,
+	// Reserve cannot make progress until Release is called (cap is
+	// full), so landed is guaranteed to stay false until we release.
+	<-started
+	if landed.Load() {
+		t.Fatal("second reserve should be blocked while cap is full")
+	}
+
+	a.Release(100)
+
+	deadline := time.Now().Add(time.Second)
+	for !landed.Load() && time.Now().Before(deadline) {
+		time.Sleep(5 * time.Millisecond)
+	}
+	if !landed.Load() {
+		t.Fatal("second reserve should unblock after release")
+	}
+	if got := a.Used(); got != 50 {
+		t.Fatalf("expected 50 used after handoff, got %d", got)
+	}
+}
+
+func TestWriteBufferAccountant_AllowsOversizeWhenEmpty(t *testing.T) {
+	// A single request larger than cap must succeed when nothing is in
+	// flight, otherwise a single file handle with a large chunk size would
+	// deadlock forever.
+	a := NewWriteBufferAccountant(100)
+	done := make(chan struct{})
+	go func() {
+		a.Reserve(500)
+		close(done)
+	}()
+	select {
+	case <-done:
+	case <-time.After(time.Second):
+		t.Fatal("oversize reserve on empty accountant should not block")
+	}
+}
diff --git a/weed/mount/page_writer/write_buffer_cap_integration_test.go b/weed/mount/page_writer/write_buffer_cap_integration_test.go
new file mode 100644
index 000000000..2a44f4295
--- /dev/null
+++ b/weed/mount/page_writer/write_buffer_cap_integration_test.go
@@ -0,0 +1,169 @@
+package page_writer
+
+import (
+	"io"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/seaweedfs/seaweedfs/weed/util"
+)
+
+// TestWriteBufferCap_SharedAcrossPipelines exercises the full write-budget
+// plumbing end-to-end: one WriteBufferAccountant shared across several
+// UploadPipeline instances (modeling multiple open file handles in a
+// single mount), a deliberately stalled saveFn (modeling the reported
+// "volume server rejecting uploads" condition), and many concurrent
+// writers.  Without the cap this scenario is what filled /tmp to 1.8 TB;
+// with the cap, Used() must never exceed the configured budget and
+// writers must backpressure instead of allocating unboundedly.
+func TestWriteBufferCap_SharedAcrossPipelines(t *testing.T) {
+	const (
+		chunkSize    = 64 * 1024
+		capChunks    = 4 // global budget: 4 chunks across all pipelines
+		numPipelines = 3
+		numWriters   = 6
+		writesEach   = 8 // more than enough to exceed capChunks if unbounded
+	)
+	capBytes := int64(capChunks * chunkSize)
+	acc := NewWriteBufferAccountant(capBytes)
+
+	// Gate every upload so sealed chunks cannot drain until the test says
+	// so — this is what reproduces the "stalled volume server" scenario.
+	var inflight atomic.Int64
+	var maxInflight atomic.Int64
+	release := make(chan struct{})
+	saveFn := func(_ io.Reader, _, _, _ int64, cleanup func()) {
+		cur := inflight.Add(1)
+		for {
+			prev := maxInflight.Load()
+			if cur <= prev || maxInflight.CompareAndSwap(prev, cur) {
+				break
+			}
+		}
+		<-release
+		inflight.Add(-1)
+		if cleanup != nil {
+			cleanup()
+		}
+	}
+
+	// Sample Used() frequently on a background goroutine so we can catch
+	// any moment the accountant exceeds the cap.
+	var observedMax atomic.Int64
+	stopSampling := make(chan struct{})
+	var samplerWg sync.WaitGroup
+	samplerWg.Add(1)
+	go func() {
+		defer samplerWg.Done()
+		for {
+			select {
+			case <-stopSampling:
+				return
+			default:
+			}
+			u := acc.Used()
+			for {
+				prev := observedMax.Load()
+				if u <= prev || observedMax.CompareAndSwap(prev, u) {
+					break
+				}
+			}
+			time.Sleep(time.Millisecond)
+		}
+	}()
+
+	// Build N pipelines, each with its own uploader executor so chunks
+	// can actually enter the upload stage and hit the gated saveFn.
+	pipelines := make([]*UploadPipeline, numPipelines)
+	for i := range pipelines {
+		up := NewUploadPipeline(util.NewLimitedConcurrentExecutor(4), chunkSize, saveFn, 2, t.TempDir(), acc)
+		pipelines[i] = up
+	}
+
+	// Launch numWriters goroutines, round-robin across pipelines. Each
+	// writer writes writesEach full chunks, on distinct logical chunk
+	// indices so every write triggers a new reservation.
+	var writerWg sync.WaitGroup
+	writerStart := make(chan struct{})
+	writerWg.Add(numWriters)
+	for w := 0; w < numWriters; w++ {
+		w := w
+		go func() {
+			defer writerWg.Done()
+			<-writerStart
+			up := pipelines[w%numPipelines]
+			data := make([]byte, chunkSize)
+			for k := 0; k < writesEach; k++ {
+				// Offset each writer into a distinct region of the file
+				// so chunks don't collide on the same logicChunkIndex.
+				off := int64(w)*int64(writesEach)*chunkSize + int64(k)*chunkSize
+				if _, err := up.SaveDataAt(data, off, true, int64(k+1)); err != nil {
+					t.Errorf("writer %d: SaveDataAt: %v", w, err)
+					return
+				}
+			}
+		}()
+	}
+	close(writerStart)
+
+	// Give the writers time to pile up against the cap. Any unbounded
+	// allocation would show up as Used() > capBytes; the sampler catches
+	// that. Writers should be blocked on Reserve well before writesEach
+	// chunks are created per pipeline.
+	deadline := time.Now().Add(500 * time.Millisecond)
+	for time.Now().Before(deadline) {
+		if acc.Used() >= capBytes {
+			break
+		}
+		time.Sleep(2 * time.Millisecond)
+	}
+	if got := acc.Used(); got < capBytes/2 {
+		t.Fatalf("expected writers to fill the budget, used=%d cap=%d", got, capBytes)
+	}
+
+	// While the gate is closed and writers are blocked, Used() must stay
+	// at or below the cap. Sample explicitly for a short window to catch
+	// any overshoot.
+	for i := 0; i < 50; i++ {
+		if u := acc.Used(); u > capBytes {
+			t.Fatalf("Used()=%d exceeded cap=%d while writers were stalled", u, capBytes)
+		}
+		time.Sleep(2 * time.Millisecond)
+	}
+
+	// Open the gate: every sealed chunk completes, writers drain.
+	close(release)
+
+	// Wait for all writers to finish.
+	writerDone := make(chan struct{})
+	go func() {
+		writerWg.Wait()
+		close(writerDone)
+	}()
+	select {
+	case <-writerDone:
+	case <-time.After(10 * time.Second):
+		close(stopSampling)
+		samplerWg.Wait()
+		t.Fatalf("writers did not drain after release; used=%d", acc.Used())
+	}
+
+	// Shut down all pipelines and stop sampling.
+	for _, up := range pipelines {
+		up.Shutdown()
+	}
+	close(stopSampling)
+	samplerWg.Wait()
+
+	if got := observedMax.Load(); got > capBytes {
+		t.Fatalf("observed Used()=%d exceeded cap=%d", got, capBytes)
+	}
+	if got := acc.Used(); got != 0 {
+		t.Fatalf("expected 0 used after shutdown, got %d", got)
+	}
+
+	t.Logf("peak inflight uploads=%d, peak Used()=%d bytes (cap=%d)",
+		maxInflight.Load(), observedMax.Load(), capBytes)
+}
diff --git a/weed/mount/weedfs.go b/weed/mount/weedfs.go
index 42f3348e4..87460d594 100644
--- a/weed/mount/weedfs.go
+++ b/weed/mount/weedfs.go
@@ -17,6 +17,7 @@ import (
 	"github.com/seaweedfs/seaweedfs/weed/filer"
 	"github.com/seaweedfs/seaweedfs/weed/glog"
 	"github.com/seaweedfs/seaweedfs/weed/mount/meta_cache"
+	"github.com/seaweedfs/seaweedfs/weed/mount/page_writer"
 	"github.com/seaweedfs/seaweedfs/weed/pb"
 	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
 	"github.com/seaweedfs/seaweedfs/weed/pb/mount_pb"
@@ -49,6 +50,7 @@ type Option struct {
 	CacheDirForRead             string
 	CacheSizeMBForRead          int64
 	CacheDirForWrite            string
+	WriteBufferSizeMB           int64
 	CacheMetaTTlSec             int
 	DataCenter                  string
 	Umask                       os.FileMode
@@ -113,6 +115,7 @@ type WFS struct {
 	metaCache            *meta_cache.MetaCache
 	stats                statsCache
 	chunkCache           *chunk_cache.TieredChunkCache
+	writeBufferAccountant *page_writer.WriteBufferAccountant
 	signature            int32
 	concurrentWriters    *util.LimitedConcurrentExecutor
 	copyBufferPool       sync.Pool
@@ -246,6 +249,9 @@ func NewSeaweedFileSystem(option *Option) *WFS {
 	if option.CacheSizeMBForRead > 0 {
 		wfs.chunkCache = chunk_cache.NewTieredChunkCache(256, option.getUniqueCacheDirForRead(), option.CacheSizeMBForRead, 1024*1024)
 	}
+	if option.WriteBufferSizeMB > 0 {
+		wfs.writeBufferAccountant = page_writer.NewWriteBufferAccountant(option.WriteBufferSizeMB * 1024 * 1024)
+	}
 
 	wfs.metaCache = meta_cache.NewMetaCache(path.Join(option.getUniqueCacheDirForRead(), "meta"), option.UidGidMapper,
 		util.FullPath(option.FilerMountRootPath),