diff --git a/weed/command/filer.go b/weed/command/filer.go
index 14e6f9051..317e960b8 100644
--- a/weed/command/filer.go
+++ b/weed/command/filer.go
@@ -574,7 +574,7 @@ func (fo *FilerOptions) startFiler() {
 		}
 		httpS := newHttpServer(defaultHandler, tlsConfig)
 		httpServers = append(httpServers, httpS)
-		shutdown := newFilerShutdown(fs.LeaveLockRing, stopGrpcServer, fs.Shutdown, httpServers...)
+		shutdown := newFilerShutdown(fs.LeaveLockRing, fs.StopSubscriptions, stopGrpcServer, fs.Shutdown, httpServers...)
 
 		grace.OnInterrupt(shutdown)
 
@@ -603,7 +603,7 @@ func (fo *FilerOptions) startFiler() {
 		}
 		httpS := newHttpServer(defaultHandler, nil)
 		httpServers = append(httpServers, httpS)
-		shutdown := newFilerShutdown(fs.LeaveLockRing, stopGrpcServer, fs.Shutdown, httpServers...)
+		shutdown := newFilerShutdown(fs.LeaveLockRing, fs.StopSubscriptions, stopGrpcServer, fs.Shutdown, httpServers...)
 
 		grace.OnInterrupt(shutdown)
 
@@ -625,10 +625,13 @@ func (fo *FilerOptions) startFiler() {
 // newFilerShutdown joins shutdown callers while gRPC and HTTP drain concurrently.
 // The filer leaves the lock ring first: peers and S3 gateways route keys to it
 // until the ring changes, and would hit refused connections once gRPC stops.
-func newFilerShutdown(leaveLockRing, stopGrpc, shutdownFiler func(), httpServers ...*http.Server) func() {
+// Then it ends its metadata subscriptions, so gRPC GracefulStop does not wait
+// out its timeout on streams whose subscribers sit in this same process.
+func newFilerShutdown(leaveLockRing, stopSubscriptions, stopGrpc, shutdownFiler func(), httpServers ...*http.Server) func() {
 	drain := newGracefulShutdown(stopGrpc, shutdownFiler, httpServers...)
 	return sync.OnceFunc(func() {
 		leaveLockRing()
+		stopSubscriptions()
 		drain()
 	})
 }
diff --git a/weed/command/filer_shutdown_test.go b/weed/command/filer_shutdown_test.go
index 60f32c7ad..775f6ee84 100644
--- a/weed/command/filer_shutdown_test.go
+++ b/weed/command/filer_shutdown_test.go
@@ -56,7 +56,7 @@ func TestFilerShutdownJoinsServeDuringParallelDrain(t *testing.T) {
 	grpcStopped := make(chan struct{})
 	filerClosed := make(chan struct{})
 	var closes atomic.Int32
-	shutdown := newFilerShutdown(func() {}, func() {
+	shutdown := newFilerShutdown(func() {}, func() {}, func() {
 		close(grpcStarted)
 		<-grpcRelease
 		close(grpcStopped)
@@ -138,7 +138,7 @@ func TestFilerShutdownWaitsForHTTPAfterGrpcStops(t *testing.T) {
 	server.Config.RegisterOnShutdown(func() { close(httpClosing) })
 	grpcStopped := make(chan struct{})
 	filerClosed := make(chan struct{})
-	shutdown := newFilerShutdown(func() {}, func() { close(grpcStopped) }, func() { close(filerClosed) }, server.Config)
+	shutdown := newFilerShutdown(func() {}, func() {}, func() { close(grpcStopped) }, func() { close(filerClosed) }, server.Config)
 	joined := make(chan struct{})
 	go func() { shutdown(); close(joined) }()
 	<-httpClosing
@@ -172,7 +172,7 @@ func TestFilerShutdownLeavesLockRingBeforeDraining(t *testing.T) {
 	shutdown := newFilerShutdown(func() {
 		close(leaving)
 		<-releaseLeave
-	}, func() { close(grpcStopping) }, func() {}, server.Config)
+	}, func() {}, func() { close(grpcStopping) }, func() {}, server.Config)
 	joined := make(chan struct{})
 	go func() { shutdown(); close(joined) }()
 
@@ -196,3 +196,31 @@ func TestFilerShutdownLeavesLockRingBeforeDraining(t *testing.T) {
 		t.Error("gRPC was not stopped after leaving the lock ring")
 	}
 }
+
+// The filer's own MetaAggregator and an in-process S3 gateway keep metadata
+// subscription streams open until the filer is shut down, which happens after
+// gRPC GracefulStop - so the subscriptions have to end before it, or every
+// shutdown sits out the full graceful-stop timeout.
+func TestFilerShutdownStopsSubscriptionsBeforeGrpc(t *testing.T) {
+	var order []string
+	var mu sync.Mutex
+	record := func(step string) func() {
+		return func() {
+			mu.Lock()
+			defer mu.Unlock()
+			order = append(order, step)
+		}
+	}
+	shutdown := newFilerShutdown(record("leave"), record("subscriptions"), record("grpc"), record("filer"))
+	shutdown()
+
+	want := []string{"leave", "subscriptions", "grpc", "filer"}
+	if len(order) != len(want) {
+		t.Fatalf("shutdown steps %v, want %v", order, want)
+	}
+	for i := range want {
+		if order[i] != want[i] {
+			t.Fatalf("shutdown steps %v, want %v", order, want)
+		}
+	}
+}
diff --git a/weed/server/filer_grpc_server_sub_meta.go b/weed/server/filer_grpc_server_sub_meta.go
index a40380633..387024a7d 100644
--- a/weed/server/filer_grpc_server_sub_meta.go
+++ b/weed/server/filer_grpc_server_sub_meta.go
@@ -11,6 +11,8 @@ import (
 
 	"github.com/seaweedfs/seaweedfs/weed/stats"
 
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
 	"google.golang.org/protobuf/proto"
 
 	"github.com/seaweedfs/seaweedfs/weed/filer"
@@ -109,14 +111,16 @@ const (
 // current time (backlog catch-up), multiple events are packed into a single
 // stream.Send() using the Events field. Otherwise events are sent one-by-one.
 type pipelinedSender struct {
+	ctx      context.Context
 	sendCh   chan *filer_pb.SubscribeMetadataResponse
 	errCh    chan error
 	done     chan struct{}
 	canBatch bool // true only if client set ClientSupportsBatching
 }
 
-func newPipelinedSender(stream metadataStreamSender, bufSize int, clientSupportsBatching bool) *pipelinedSender {
+func newPipelinedSender(ctx context.Context, stream metadataStreamSender, bufSize int, clientSupportsBatching bool) *pipelinedSender {
 	s := &pipelinedSender{
+		ctx:      ctx,
 		sendCh:   make(chan *filer_pb.SubscribeMetadataResponse, bufSize),
 		errCh:    make(chan error, 1),
 		done:     make(chan struct{}),
@@ -128,6 +132,21 @@ func newPipelinedSender(stream metadataStreamSender, bufSize int, clientSupports
 
 func (s *pipelinedSender) sendLoop(stream metadataStreamSender) {
 	defer close(s.done)
+	// No Send may start after the subscription context ends: the handler is
+	// returning, and gRPC tears the stream down behind it. A Send already in
+	// flight is released by that teardown (the stream context unblocks it).
+	send := func(msg *filer_pb.SubscribeMetadataResponse) bool {
+		if s.ctx.Err() != nil {
+			return false
+		}
+		if err := stream.Send(msg); err != nil {
+			if s.ctx.Err() == nil {
+				s.reportErr(err)
+			}
+			return false
+		}
+		return true
+	}
 	for msg := range s.sendCh {
 		// LogFileRefs messages are unbatchable: the client recognizes them by
 		// the top-level field and skips the rest of the response, so a refs
@@ -142,8 +161,7 @@ func (s *pipelinedSender) sendLoop(stream metadataStreamSender) {
 
 		if !shouldBatch {
 			// Real-time: send immediately for low latency
-			if err := stream.Send(msg); err != nil {
-				s.reportErr(err)
+			if !send(msg) {
 				return
 			}
 			continue
@@ -181,18 +199,14 @@ func (s *pipelinedSender) sendLoop(stream metadataStreamSender) {
 			toSend = batch[0]
 			toSend.Events = batch[1:]
 		}
-		if err := stream.Send(toSend); err != nil {
-			s.reportErr(err)
+		if !send(toSend) {
 			return
 		}
 		if toSend.Events != nil {
 			toSend.Events = nil
 		}
-		if trailingSolo != nil {
-			if err := stream.Send(trailingSolo); err != nil {
-				s.reportErr(err)
-				return
-			}
+		if trailingSolo != nil && !send(trailingSolo) {
+			return
 		}
 	}
 }
@@ -207,11 +221,16 @@ func (s *pipelinedSender) reportErr(err error) {
 }
 
 func (s *pipelinedSender) Send(msg *filer_pb.SubscribeMetadataResponse) error {
+	if err := s.ctx.Err(); err != nil {
+		return err
+	}
 	select {
 	case s.sendCh <- msg:
 		return nil
 	case err := <-s.errCh:
 		return err
+	case <-s.ctx.Done():
+		return s.ctx.Err()
 	case <-s.done:
 		// Sender goroutine exited (stream error or shutdown).
 		select {
@@ -225,7 +244,12 @@ func (s *pipelinedSender) Send(msg *filer_pb.SubscribeMetadataResponse) error {
 
 func (s *pipelinedSender) Close() error {
 	close(s.sendCh)
-	<-s.done
+	// A sendLoop stuck in stream.Send only unblocks once the handler's return
+	// ends the stream, so stop waiting when the subscription ends.
+	select {
+	case <-s.done:
+	case <-s.ctx.Done():
+	}
 	select {
 	case err := <-s.errCh:
 		return err
@@ -602,7 +626,7 @@ func (p *gapPass) park(ctx context.Context, cursor *log_buffer.MessagePosition,
 	return gapContinue
 }
 
-func (fs *FilerServer) SubscribeMetadata(req *filer_pb.SubscribeMetadataRequest, stream filer_pb.SeaweedFiler_SubscribeMetadataServer) error {
+func (fs *FilerServer) SubscribeMetadata(req *filer_pb.SubscribeMetadataRequest, stream filer_pb.SeaweedFiler_SubscribeMetadataServer) (err error) {
 	// A filer that has not learned remote peers yet serves the local log and
 	// upgrades when the first one appears. RemotePeerArrivedChan takes the
 	// arrival channel under the same lock as the peer check, so a peer
@@ -616,7 +640,9 @@ func (fs *FilerServer) SubscribeMetadata(req *filer_pb.SubscribeMetadataRequest,
 		return fs.subscribeLocalMetadata(req, stream, nil)
 	}
 
-	ctx := stream.Context()
+	ctx, cancelSubscription := fs.subscriptionContext(stream.Context())
+	defer cancelSubscription()
+	defer func() { err = fs.endOfSubscription(stream.Context(), err) }()
 	peerAddress := findClientAddress(ctx, 0)
 
 	isReplacing, alreadyKnown, clientName := fs.addClient("", req.ClientName, peerAddress, req.PathPrefix, req.ClientId, req.ClientEpoch)
@@ -643,7 +669,7 @@ func (fs *FilerServer) SubscribeMetadata(req *filer_pb.SubscribeMetadataRequest,
 	// had already arrived and been delivered).
 	diskAnchorTsNs := req.SinceNs
 
-	sender := newPipelinedSender(stream, 1024, req.ClientSupportsBatching)
+	sender := newPipelinedSender(ctx, stream, 1024, req.ClientSupportsBatching)
 	defer sender.Close()
 
 	// Register for instant notification when new data arrives in the aggregated log buffer.
@@ -669,7 +695,7 @@ func (fs *FilerServer) SubscribeMetadata(req *filer_pb.SubscribeMetadataRequest,
 	// written from this single goroutine, so no synchronization is needed.
 	var lastSeenTsNs int64
 	var lastHeartbeatNs int64
-	baseEachLogEntryFn := eachLogEntryFn(req, sender, eachEventNotificationFn, &unsyncedEvents)
+	baseEachLogEntryFn := eachLogEntryFn(ctx, req, sender, eachEventNotificationFn, &unsyncedEvents)
 	// heldAtTsNs remembers the entry a read was held at (for the log line);
 	// diskHeldAtTsNs is the same marker for the disk pass alone: a pending
 	// disk hold keeps the pass re-reading until the entry is served.
@@ -1004,6 +1030,59 @@ func (fs *FilerServer) SubscribeMetadata(req *filer_pb.SubscribeMetadataRequest,
 
 }
 
+// StopSubscriptions ends every metadata subscription and any started later.
+// Shutdown calls it before gRPC GracefulStop, which waits on open streams;
+// subscribers living in this process - the filer's own MetaAggregator and an
+// in-process S3 gateway - are otherwise torn down only by the later
+// fs.Shutdown, so every stop waited out the graceful-stop timeout. Ended
+// subscribers get codes.Unavailable (see endOfSubscription) and reconnect
+// with their usual retry.
+//
+// A subscriber blocked in stream.Send is released once its handler returns:
+// the sender's Send and Close honor the subscription context, so the handler
+// exits and gRPC tears the stream down.
+func (fs *FilerServer) StopSubscriptions() {
+	if fs.stopSubscriptions != nil {
+		fs.stopSubscriptions()
+	}
+}
+
+// subscriptionContext ends with the stream or with StopSubscriptions,
+// whichever comes first. It keeps the stream's values (peer address).
+func (fs *FilerServer) subscriptionContext(stream context.Context) (context.Context, context.CancelFunc) {
+	ctx, cancel := context.WithCancel(stream)
+	if fs.subscriptionsStopped == nil {
+		return ctx, cancel
+	}
+	stop := context.AfterFunc(fs.subscriptionsStopped, cancel)
+	return ctx, func() {
+		stop()
+		cancel()
+	}
+}
+
+// endOfSubscription turns the end of a subscription that StopSubscriptions
+// cut short into codes.Unavailable. Followers read a clean end as "caught up,
+// done": the client returns nil on io.EOF, and util.RetryUntil - which the S3
+// gateway and mount follow with - stops on nil. A context.Canceled from the
+// subscription context is the stop signal surfacing, not a real error, and
+// reads as non-retryable to transient-error classifiers. Unavailable is what
+// a dropped connection looks like, so followers reconnect with their usual
+// retry. Handler errors pass through, and so does the end of a stream the
+// client closed.
+func (fs *FilerServer) endOfSubscription(stream context.Context, err error) error {
+	if stream.Err() != nil {
+		return err
+	}
+	if fs.subscriptionsStopped == nil || fs.subscriptionsStopped.Err() == nil {
+		return err
+	}
+	if err == nil || errors.Is(err, context.Canceled) {
+		return status.Error(codes.Unavailable, "filer is shutting down")
+	}
+	return err
+}
+
 func (fs *FilerServer) SubscribeLocalMetadata(req *filer_pb.SubscribeMetadataRequest, stream filer_pb.SeaweedFiler_SubscribeLocalMetadataServer) error {
 	return fs.subscribeLocalMetadata(req, stream, nil)
 }
@@ -1012,9 +1091,11 @@ func (fs *FilerServer) SubscribeLocalMetadata(req *filer_pb.SubscribeMetadataReq
 // aggregation streams pass upgradeOnRemotePeer == nil; the SubscribeMetadata
 // delegation passes the aggregator's arrival channel so the stream ends
 // when a remote peer appears and the client reconnects to the aggregated path.
-func (fs *FilerServer) subscribeLocalMetadata(req *filer_pb.SubscribeMetadataRequest, stream metadataLocalStream, upgradeOnRemotePeer <-chan struct{}) error {
+func (fs *FilerServer) subscribeLocalMetadata(req *filer_pb.SubscribeMetadataRequest, stream metadataLocalStream, upgradeOnRemotePeer <-chan struct{}) (err error) {
 
-	ctx := stream.Context()
+	ctx, cancelSubscription := fs.subscriptionContext(stream.Context())
+	defer cancelSubscription()
+	defer func() { err = fs.endOfSubscription(stream.Context(), err) }()
 	peerAddress := findClientAddress(ctx, 0)
 
 	// use negative client id to differentiate from addClient()/deleteClient() used in SubscribeMetadata()
@@ -1033,7 +1114,7 @@ func (fs *FilerServer) subscribeLocalMetadata(req *filer_pb.SubscribeMetadataReq
 	lastReadTime := log_buffer.NewMessagePosition(req.SinceNs, gapResumeCursorOffset)
 	glog.V(0).Infof(" + %v local subscribe %s from %+v clientId:%d", clientName, req.PathPrefix, lastReadTime, req.ClientId)
 
-	sender := newPipelinedSender(stream, 1024, req.ClientSupportsBatching)
+	sender := newPipelinedSender(ctx, stream, 1024, req.ClientSupportsBatching)
 	defer sender.Close()
 
 	// Bounded gap waits use the buffer's subscriber notification plus a retry
@@ -1059,7 +1140,7 @@ func (fs *FilerServer) subscribeLocalMetadata(req *filer_pb.SubscribeMetadataReq
 	var lastSeenTsNs int64
 	var lastHeartbeatNs int64
 	var lastFlushReportNs int64
-	baseEachLogEntryFn := eachLogEntryFn(req, sender, eachEventNotificationFn, &unsyncedEvents)
+	baseEachLogEntryFn := eachLogEntryFn(ctx, req, sender, eachEventNotificationFn, &unsyncedEvents)
 	eachLogEntryFn := func(logEntry *filer_pb.LogEntry) (bool, error) {
 		if upgradeOnRemotePeer != nil {
 			select {
@@ -1224,12 +1305,17 @@ func (fs *FilerServer) subscribeLocalMetadata(req *filer_pb.SubscribeMetadataReq
 
 }
 
-func eachLogEntryFn(req *filer_pb.SubscribeMetadataRequest, sender metadataStreamSender, eachEventNotificationFn func(dirPath string, eventNotification *filer_pb.EventNotification, tsNs int64) error, filtered *int64) log_buffer.EachLogEntryFuncType {
+func eachLogEntryFn(ctx context.Context, req *filer_pb.SubscribeMetadataRequest, sender metadataStreamSender, eachEventNotificationFn func(dirPath string, eventNotification *filer_pb.EventNotification, tsNs int64) error, filtered *int64) log_buffer.EachLogEntryFuncType {
 	// A shallow scan of the path fields skips unmarshaling chunk-heavy events
 	// this subscriber would filter out anyway; scan surprises fall back to the
 	// full decode. Only a delivery resets the shared unsynced-events counter.
 	prefilter := req.PathPrefix != "" || len(req.PathPrefixes) > 0 || len(req.Directories) > 0
 	return func(logEntry *filer_pb.LogEntry) (bool, error) {
+		// A cancelled context ends the pass here: neither loop checks it
+		// per entry, and the send path alone cannot cover filtered entries.
+		if ctx.Err() != nil {
+			return true, nil
+		}
 		if prefilter {
 			if skeleton, ok := filer_pb.ScanMetadataEventSkeleton(logEntry.Data); ok &&
 				!filer_pb.MetadataEventMatchesSubscription(skeleton, req.PathPrefix, req.PathPrefixes, req.Directories) {
@@ -1370,7 +1456,7 @@ func (fs *FilerServer) chunkDiskPass(ctx context.Context, sender metadataStreamS
 	if len(refs) == 0 {
 		return startPos.Time.UnixNano(), false, nil
 	}
-	if err := fs.sendRefsBatched(sender, refs, upgradeOnRemotePeer); err != nil {
+	if err := fs.sendRefsBatched(ctx, sender, refs, upgradeOnRemotePeer); err != nil {
 		return 0, false, err
 	}
 	if upgradeOnRemotePeer != nil {
@@ -1430,9 +1516,12 @@ func (fs *FilerServer) chunkDiskPass(ctx context.Context, sender metadataStreamS
 // sendRefsBatched sends refs through the pipelined sender, which keeps them
 // out of Events batches; gRPC allows one sending goroutine per stream and the
 // sender's goroutine is it.
-func (fs *FilerServer) sendRefsBatched(sender metadataStreamSender, refs []*filer_pb.LogFileChunkRef, upgradeOnRemotePeer <-chan struct{}) error {
+func (fs *FilerServer) sendRefsBatched(ctx context.Context, sender metadataStreamSender, refs []*filer_pb.LogFileChunkRef, upgradeOnRemotePeer <-chan struct{}) error {
 	const maxRefsPerMessage = 64
 	for i := 0; i < len(refs); i += maxRefsPerMessage {
+		if err := ctx.Err(); err != nil {
+			return err
+		}
 		if upgradeOnRemotePeer != nil {
 			select {
 			case <-upgradeOnRemotePeer:
diff --git a/weed/server/filer_grpc_server_sub_meta_prefilter_test.go b/weed/server/filer_grpc_server_sub_meta_prefilter_test.go
index 10652bf6b..5630a9179 100644
--- a/weed/server/filer_grpc_server_sub_meta_prefilter_test.go
+++ b/weed/server/filer_grpc_server_sub_meta_prefilter_test.go
@@ -1,6 +1,7 @@
 package weed_server
 
 import (
+	"context"
 	"testing"
 
 	"google.golang.org/protobuf/proto"
@@ -37,7 +38,7 @@ func TestEachLogEntryFnPrefilterSkipsDecode(t *testing.T) {
 	sender := &recordingSender{}
 	var decoded int
 	var unsyncedEvents int64
-	fn := eachLogEntryFn(req, sender, func(dirPath string, eventNotification *filer_pb.EventNotification, tsNs int64) error {
+	fn := eachLogEntryFn(context.Background(), req, sender, func(dirPath string, eventNotification *filer_pb.EventNotification, tsNs int64) error {
 		decoded++
 		unsyncedEvents = 0 // emulate a delivery, like the notification fn after a send
 		return nil
@@ -83,7 +84,7 @@ func TestEachLogEntryFnNoFilterDecodesEverything(t *testing.T) {
 	req := &filer_pb.SubscribeMetadataRequest{}
 	var decoded int
 	var unsyncedEvents int64
-	fn := eachLogEntryFn(req, &recordingSender{}, func(dirPath string, eventNotification *filer_pb.EventNotification, tsNs int64) error {
+	fn := eachLogEntryFn(context.Background(), req, &recordingSender{}, func(dirPath string, eventNotification *filer_pb.EventNotification, tsNs int64) error {
 		decoded++
 		return nil
 	}, &unsyncedEvents)
diff --git a/weed/server/filer_grpc_server_sub_meta_test.go b/weed/server/filer_grpc_server_sub_meta_test.go
index 072cf3cbf..a0a8162a2 100644
--- a/weed/server/filer_grpc_server_sub_meta_test.go
+++ b/weed/server/filer_grpc_server_sub_meta_test.go
@@ -1,6 +1,8 @@
 package weed_server
 
 import (
+	"context"
+
 	"fmt"
 	"sync"
 	"sync/atomic"
@@ -131,7 +133,7 @@ func TestPipelinedSenderThroughput(t *testing.T) {
 	var batchedRate float64
 	t.Run("pipelined_batched_send", func(t *testing.T) {
 		stream := &slowStream{sendDelay: sendDelay}
-		sender := newPipelinedSender(stream, 1024, true)
+		sender := newPipelinedSender(context.Background(), stream, 1024, true)
 
 		start := time.Now()
 		for _, file := range files {
@@ -217,7 +219,7 @@ func TestBatchingAdaptive(t *testing.T) {
 
 	t.Run("old_events_are_batched", func(t *testing.T) {
 		stream := &slowStream{sendDelay: 10 * time.Microsecond}
-		sender := newPipelinedSender(stream, 1024, true)
+		sender := newPipelinedSender(context.Background(), stream, 1024, true)
 
 		// Push all events at once (no read delay) so the sender can batch aggressively
 		for _, ev := range makeOldEvents(numEvents) {
@@ -237,7 +239,7 @@ func TestBatchingAdaptive(t *testing.T) {
 
 	t.Run("recent_events_sent_individually", func(t *testing.T) {
 		stream := &slowStream{sendDelay: 10 * time.Microsecond}
-		sender := newPipelinedSender(stream, 1024, true)
+		sender := newPipelinedSender(context.Background(), stream, 1024, true)
 
 		for _, ev := range makeRecentEvents(numEvents) {
 			sender.Send(ev)
@@ -281,7 +283,7 @@ func TestPipelinedSenderErrorPropagation(t *testing.T) {
 	t.Run("send_returns_error", func(t *testing.T) {
 		// Stream fails after 5 successful sends
 		stream := &errorStreamImpl{failAfter: 5, err: sendErr}
-		sender := newPipelinedSender(stream, 4, true)
+		sender := newPipelinedSender(context.Background(), stream, 4, true)
 
 		var lastErr error
 		for i := 0; i < 100; i++ {
@@ -303,7 +305,7 @@ func TestPipelinedSenderErrorPropagation(t *testing.T) {
 		// since Send may have already returned before the sender goroutine
 		// processes the message.
 		stream := &errorStreamImpl{failAfter: 0, err: sendErr}
-		sender := newPipelinedSender(stream, 1024, true)
+		sender := newPipelinedSender(context.Background(), stream, 1024, true)
 
 		ev := makeOldEvents(1)[0]
 		sender.Send(ev)
@@ -352,7 +354,7 @@ func TestPipelinedSingleVsParallelStreams(t *testing.T) {
 	// simulatePipeline: read files with I/O delay, push events, send via pipelinedSender
 	simulatePipeline := func(files []logFile) (eventsSent, sends int64, elapsed time.Duration, err error) {
 		stream := &slowStream{sendDelay: sendDelay}
-		sender := newPipelinedSender(stream, 1024, true)
+		sender := newPipelinedSender(context.Background(), stream, 1024, true)
 
 		start := time.Now()
 	outer:
diff --git a/weed/server/filer_grpc_server_sub_meta_watermark_test.go b/weed/server/filer_grpc_server_sub_meta_watermark_test.go
index b482d089a..6b76a3da2 100644
--- a/weed/server/filer_grpc_server_sub_meta_watermark_test.go
+++ b/weed/server/filer_grpc_server_sub_meta_watermark_test.go
@@ -1,6 +1,8 @@
 package weed_server
 
 import (
+	"context"
+
 	"errors"
 	"sync"
 	"testing"
@@ -55,7 +57,7 @@ func (s *gatedRecordingStream) Send(msg *filer_pb.SubscribeMetadataResponse) err
 // reads past it.
 func TestPipelinedSenderControlMessagesNeverNested(t *testing.T) {
 	stream := &gatedRecordingStream{gate: make(chan struct{})}
-	sender := newPipelinedSender(stream, 16, true)
+	sender := newPipelinedSender(context.Background(), stream, 16, true)
 
 	oldTs := time.Now().Add(-time.Hour).UnixNano()
 	if err := sender.Send(makeEvent("/d", "e1", oldTs)); err != nil {
diff --git a/weed/server/filer_pipelined_sender_test.go b/weed/server/filer_pipelined_sender_test.go
index b50455501..aeada1b67 100644
--- a/weed/server/filer_pipelined_sender_test.go
+++ b/weed/server/filer_pipelined_sender_test.go
@@ -1,6 +1,9 @@
 package weed_server
 
 import (
+	"context"
+	"sync/atomic"
+
 	"fmt"
 	"sync"
 	"testing"
@@ -46,7 +49,7 @@ func (s *recordingStream) snapshot() []*filer_pb.SubscribeMetadataResponse {
 // event. Everything must arrive, in order, whatever the batcher does.
 func TestPipelinedSenderRefsNeverBatched(t *testing.T) {
 	stream := &recordingStream{slow: 2 * time.Millisecond} // let the queue back up so batching engages
-	sender := newPipelinedSender(stream, 64, true)
+	sender := newPipelinedSender(context.Background(), stream, 64, true)
 
 	oldTs := time.Now().Add(-time.Hour).UnixNano() // far behind: the batch heuristic fires
 	var wantOrder []string
@@ -102,3 +105,89 @@ func TestPipelinedSenderRefsNeverBatched(t *testing.T) {
 		t.Fatalf("delivery order/count changed:\n got %v\nwant %v", gotOrder, wantOrder)
 	}
 }
+
+type blockedStream struct {
+	release chan struct{}
+	entered chan struct{}
+	sends   atomic.Int32
+}
+
+func (s *blockedStream) Send(*filer_pb.SubscribeMetadataResponse) error {
+	s.sends.Add(1)
+	select {
+	case s.entered <- struct{}{}:
+	default:
+	}
+	<-s.release
+	return nil
+}
+
+// A subscriber that stops reading wedges stream.Send. The subscription
+// context must release Send and Close, so the handler can return and end the
+// stream during filer shutdown.
+func TestPipelinedSenderUnblocksOnCancel(t *testing.T) {
+	stream := &blockedStream{release: make(chan struct{})}
+	defer close(stream.release)
+	ctx, cancel := context.WithCancel(context.Background())
+	sender := newPipelinedSender(ctx, stream, 1, false)
+
+	msg := &filer_pb.SubscribeMetadataResponse{EventNotification: &filer_pb.EventNotification{}}
+	if err := sender.Send(msg); err != nil { // consumed by the blocked sendLoop
+		t.Fatal(err)
+	}
+	if err := sender.Send(msg); err != nil { // fills the one-deep queue
+		t.Fatal(err)
+	}
+
+	cancel()
+
+	done := make(chan error, 1)
+	go func() { done <- sender.Send(msg) }()
+	select {
+	case err := <-done:
+		if err != context.Canceled {
+			t.Fatalf("Send returned %v, want context.Canceled", err)
+		}
+	case <-time.After(2 * time.Second):
+		t.Fatal("Send still blocked after cancel")
+	}
+	if err := sender.Close(); err != nil {
+		t.Fatalf("Close returned %v", err)
+	}
+}
+
+// Once the subscription context ends, sendLoop must not start another
+// stream.Send: the handler returns and gRPC ends the stream behind it, so a
+// send started then would race the stream's teardown. The one Send already
+// in flight is released by that teardown.
+func TestPipelinedSenderStopsSendingOnCancel(t *testing.T) {
+	stream := &blockedStream{release: make(chan struct{}), entered: make(chan struct{}, 1)}
+	ctx, cancel := context.WithCancel(context.Background())
+	sender := newPipelinedSender(ctx, stream, 4, false)
+
+	msg := &filer_pb.SubscribeMetadataResponse{EventNotification: &filer_pb.EventNotification{}}
+	if err := sender.Send(msg); err != nil {
+		t.Fatal(err)
+	}
+	<-stream.entered         // sendLoop is inside stream.Send
+	for i := 0; i < 2; i++ { // queued behind the blocked send
+		if err := sender.Send(msg); err != nil {
+			t.Fatal(err)
+		}
+	}
+
+	cancel()
+	close(stream.release)
+
+	if err := sender.Close(); err != nil {
+		t.Fatalf("Close returned %v", err)
+	}
+	select {
+	case <-sender.done:
+	case <-time.After(2 * time.Second):
+		t.Fatal("sendLoop did not exit after cancel")
+	}
+	if n := stream.sends.Load(); n != 1 {
+		t.Fatalf("stream.Send ran %d times, want 1; a new send started after cancel", n)
+	}
+}
diff --git a/weed/server/filer_server.go b/weed/server/filer_server.go
index 49c99a87a..a823bfd5b 100644
--- a/weed/server/filer_server.go
+++ b/weed/server/filer_server.go
@@ -170,6 +170,12 @@ type FilerServer struct {
 	posixLocks *posixlock.Manager
 	// posixLockSweeperStop stops the lease-reaping sweeper goroutine on Shutdown.
 	posixLockSweeperStop chan struct{}
+
+	// subscriptionsStopped is cancelled by StopSubscriptions when the filer
+	// starts shutting down; every metadata subscription derives its context
+	// from it as well as from its stream.
+	subscriptionsStopped context.Context
+	stopSubscriptions    context.CancelFunc
 	// posixLockReadyAt is the unix-nanos when this filer began serving POSIX
 	// locks. For posixLockWarmup after it, the owner defers would-be grants while
 	// mounts re-assert, so a (re)started owner does not double-grant from empty
@@ -220,6 +226,7 @@ func NewFilerServer(defaultMux, readonlyMux *http.ServeMux, option *FilerOption)
 		entryLockTable:        util.NewLockTable[util.FullPath](),
 		posixLocks:            posixlock.NewManager(),
 	}
+	fs.subscriptionsStopped, fs.stopSubscriptions = context.WithCancel(context.Background())
 	fs.startPosixLockSweeper()
 	fs.mountPeerRegistry = filer.NewMountPeerRegistry()
 	go fs.runMountPeerRegistrySweeper()
diff --git a/weed/server/filer_server_subscription_stop_test.go b/weed/server/filer_server_subscription_stop_test.go
new file mode 100644
index 000000000..7bf97ddcf
--- /dev/null
+++ b/weed/server/filer_server_subscription_stop_test.go
@@ -0,0 +1,150 @@
+package weed_server
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"testing"
+	"time"
+
+	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
+)
+
+func newStoppableFilerServer() *FilerServer {
+	fs := &FilerServer{}
+	fs.subscriptionsStopped, fs.stopSubscriptions = context.WithCancel(context.Background())
+	return fs
+}
+
+func TestStopSubscriptionsEndsOpenSubscriptions(t *testing.T) {
+	fs := newStoppableFilerServer()
+	ctx, cancel := fs.subscriptionContext(context.Background())
+	defer cancel()
+
+	select {
+	case <-ctx.Done():
+		t.Fatal("subscription ended before StopSubscriptions")
+	default:
+	}
+	fs.StopSubscriptions()
+	select {
+	case <-ctx.Done():
+	case <-time.After(5 * time.Second):
+		t.Fatal("subscription still open after StopSubscriptions")
+	}
+}
+
+func TestSubscriptionStartedAfterStopEndsAtOnce(t *testing.T) {
+	fs := newStoppableFilerServer()
+	fs.StopSubscriptions()
+	ctx, cancel := fs.subscriptionContext(context.Background())
+	defer cancel()
+	select {
+	case <-ctx.Done():
+	case <-time.After(5 * time.Second):
+		t.Fatal("a subscription started during shutdown stayed open")
+	}
+}
+
+func TestSubscriptionStillEndsWithItsStream(t *testing.T) {
+	fs := newStoppableFilerServer()
+	stream, closeStream := context.WithCancel(context.WithValue(context.Background(), struct{}{}, "peer"))
+	ctx, cancel := fs.subscriptionContext(stream)
+	defer cancel()
+	if ctx.Value(struct{}{}) != "peer" {
+		t.Error("subscription context lost the stream's values")
+	}
+	closeStream()
+	select {
+	case <-ctx.Done():
+	case <-time.After(5 * time.Second):
+		t.Fatal("subscription outlived its stream")
+	}
+}
+
+func TestSubscriptionContextWithoutStopSignal(t *testing.T) {
+	fs := &FilerServer{}
+	fs.StopSubscriptions()
+	ctx, cancel := fs.subscriptionContext(context.Background())
+	select {
+	case <-ctx.Done():
+		t.Fatal("subscription ended although nothing stopped it")
+	default:
+	}
+	cancel()
+	<-ctx.Done()
+}
+
+// A follower reads a clean end as "done" and stops for good (util.RetryUntil
+// returns on nil), so a subscription that StopSubscriptions ended must not
+// end cleanly.
+func TestStoppedSubscriptionEndsUnavailable(t *testing.T) {
+	fs := newStoppableFilerServer()
+	fs.StopSubscriptions()
+	err := fs.endOfSubscription(context.Background(), nil)
+	if status.Code(err) != codes.Unavailable {
+		t.Fatalf("stopped subscription ended with %v, want Unavailable", err)
+	}
+}
+
+func TestSubscriptionEndsCleanlyWithoutStop(t *testing.T) {
+	fs := newStoppableFilerServer()
+	if err := fs.endOfSubscription(context.Background(), nil); err != nil {
+		t.Fatalf("subscription without shutdown ended with %v, want nil", err)
+	}
+	if err := (&FilerServer{}).endOfSubscription(context.Background(), nil); err != nil {
+		t.Fatalf("server without stop signal ended with %v, want nil", err)
+	}
+}
+
+func TestSubscriptionErrorPassesThroughStop(t *testing.T) {
+	fs := newStoppableFilerServer()
+	fs.StopSubscriptions()
+	want := errors.New("reading from persisted logs")
+	if err := fs.endOfSubscription(context.Background(), want); !errors.Is(err, want) {
+		t.Fatalf("got %v, want the handler's own error", err)
+	}
+}
+
+func TestClosedStreamEndsCleanlyDuringStop(t *testing.T) {
+	fs := newStoppableFilerServer()
+	fs.StopSubscriptions()
+	stream, closeStream := context.WithCancel(context.Background())
+	closeStream()
+	if err := fs.endOfSubscription(stream, nil); err != nil {
+		t.Fatalf("stream the client closed ended with %v, want nil", err)
+	}
+}
+
+// A read pass surfaces StopSubscriptions as a wrapped context.Canceled; on the
+// wire that must still read as Unavailable so followers retry.
+func TestCanceledReadEndsUnavailableDuringStop(t *testing.T) {
+	fs := newStoppableFilerServer()
+	fs.StopSubscriptions()
+	canceled := fmt.Errorf("reading from persisted logs: %w", context.Canceled)
+	if err := fs.endOfSubscription(context.Background(), canceled); status.Code(err) != codes.Unavailable {
+		t.Fatalf("stopped read ended with %v, want Unavailable", err)
+	}
+}
+
+func TestCanceledReadPassesThroughWithoutStop(t *testing.T) {
+	fs := newStoppableFilerServer()
+	canceled := fmt.Errorf("reading from persisted logs: %w", context.Canceled)
+	if err := fs.endOfSubscription(context.Background(), canceled); !errors.Is(err, context.Canceled) {
+		t.Fatalf("got %v, want the handler's own error", err)
+	}
+}
+
+// Neither read loop checks ctx per entry; the entry callback does.
+func TestEachLogEntryFnEndsOnCancel(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	cancel()
+	var filtered int64
+	fn := eachLogEntryFn(ctx, &filer_pb.SubscribeMetadataRequest{}, nil, nil, &filtered)
+	done, err := fn(&filer_pb.LogEntry{})
+	if !done || err != nil {
+		t.Fatalf("cancelled subscription returned done=%v err=%v, want a clean end", done, err)
+	}
+}