From b95a5c3fe5ea1004995da2aa70ae49f8dc7a2ee3 Mon Sep 17 00:00:00 2001
From: Eli Mallon <eli@iame.li>
Date: Fri, 9 Oct 2026 22:51:54 -0700
Subject: [PATCH] moq: release a finished subscription's read side

A completed subscription FINs its send side; a subscriber that never
closes its own side would otherwise hold the stream slot for the life of
the session. The read side is now cancelled too, through one helper
shared by every completion path.
---
 pkg/moq/publish.go | 32 +++++++++++++++++---------------
 1 file changed, 17 insertions(+), 15 deletions(-)

diff --git a/pkg/moq/publish.go b/pkg/moq/publish.go
index 838f214e3..00411625b 100644
--- a/pkg/moq/publish.go
+++ b/pkg/moq/publish.go
@@ -74,11 +74,7 @@ func (s *Session) serveSubscribe(br *bufio.Reader, st bidiStream) {
 			if errors.Is(err, io.EOF) {
 				// Every group below next has been sent, so END is also
 				// the moment the stream can be finished (draft §5.1.2).
-				if werr := writeAll(st, msg(nil).varint(next).frame(subscribeEnd)); werr != nil {
-					reset(st, errInternal)
-					return
-				}
-				_ = st.Close()
+				finishSubscribe(st, next)
 				return
 			}
 			if ctx.Err() != nil {
@@ -98,11 +94,7 @@ func (s *Session) serveSubscribe(br *bufio.Reader, st bidiStream) {
 		// it does not want (draft §7.9); a track that has already passed
 		// it owes nothing, not even an OK.
 		if groupEnd != 0 && g.Sequence >= groupEnd {
-			if werr := writeAll(st, msg(nil).varint(groupEnd).frame(subscribeEnd)); werr != nil {
-				reset(st, errInternal)
-				return
-			}
-			_ = st.Close()
+			finishSubscribe(st, groupEnd)
 			return
 		}
 		if first {
@@ -123,16 +115,26 @@ func (s *Session) serveSubscribe(br *bufio.Reader, st bidiStream) {
 			return
 		}
 		if groupEnd != 0 && next >= groupEnd {
-			if werr := writeAll(st, msg(nil).varint(groupEnd).frame(subscribeEnd)); werr != nil {
-				reset(st, errInternal)
-				return
-			}
-			_ = st.Close()
+			finishSubscribe(st, groupEnd)
 			return
 		}
 	}
 }
 
+// finishSubscribe completes a subscription cleanly: SUBSCRIBE_END naming
+// the first sequence never delivered, a FIN, and the read side released.
+// A conforming subscriber closes its side on seeing ours (draft §4.3), but
+// the stream slot must not depend on it: the only thing it could still
+// send is a SUBSCRIBE_UPDATE to a subscription that is over.
+func finishSubscribe(st bidiStream, end uint64) {
+	if err := writeAll(st, msg(nil).varint(end).frame(subscribeEnd)); err != nil {
+		reset(st, errInternal)
+		return
+	}
+	_ = st.Close()
+	st.CancelRead(errCancelled)
+}
+
 // sendGroup ships one group on a fresh unidirectional stream: STREAM_TYPE,
 // GROUP, then each FRAME as a zigzag timestamp delta, a length, and the
 // payload. The payload is written as-is rather than copied into the header.
-- 
2.54.0