Second draft-20 review: backlog and the six high-severity fixes

pkg/moqt/session/example_test.go

@@ -228,9 +228,9 @@ func ExampleRequestMux() {
 		_ = r.RejectError(moqt.RequestNotSupported, "unsupported request type")
 	})
 
-	// Run returns when ctx is cancelled or AcceptRequest fails. A session-fatal
-	// error (e.g. *session.ErrDuplicateRequestID) should be escalated by closing
-	// the session with the mapped code.
+	// Run returns when ctx is cancelled or AcceptRequest fails; a protocol
+	// violation (e.g. *session.ErrDuplicateRequestID) has already closed the
+	// session with the mapped code.
 	_ = mux.Run(ctx, server)
 }
 

pkg/moqt/session/helpers_test.go

@@ -166,19 +166,24 @@ func requireRefusedOpen(t *testing.T, want string, open func(context.Context) (*
 
 // requireClosedProtocolViolation waits for sess to close and checks the code.
 func requireClosedProtocolViolation(t *testing.T, sess *session.Session) {
+	t.Helper()
+	requireClosedCode(t, sess, moqt.SessionProtocolViolation)
+}
+
+// requireClosedCode checks that sess closes itself with want.
+func requireClosedCode(t *testing.T, sess *session.Session, want moqt.SessionErrorCode) {
 	t.Helper()
 	select {
 	case <-sess.Done():
 	case <-time.After(2 * time.Second):
-		t.Fatal("session stayed open; want PROTOCOL_VIOLATION close")
+		t.Fatalf("session stayed open; want close with %#x", uint64(want))
 	}
 	closed, ok := errors.AsType[*session.ClosedError](sess.Err())
 	if !ok {
 		t.Fatalf("Err() = %v, want a *session.ClosedError", sess.Err())
 	}
-	if closed.Code != moqt.SessionProtocolViolation {
-		t.Errorf("closed with code %#x, want PROTOCOL_VIOLATION (%#x)",
-			uint64(closed.Code), uint64(moqt.SessionProtocolViolation))
+	if closed.Code != want {
+		t.Errorf("closed with code %#x, want %#x", uint64(closed.Code), uint64(want))
 	}
 }
 

pkg/moqt/session/malformed_test.go

@@ -66,24 +66,43 @@ func TestMalformedOpenerClosesSession(t *testing.T) {
 	}
 }
 
-// TestTruncatedOpenerResetsOnlyStream: a stream ending before its first frame
-// is complete is the peer giving up, not a Length mismatch.
+// TestTruncatedOpenerResetsOnlyStream: a stream ending or reset before its
+// first frame is complete is the peer giving up on that request (§3.3.2,
+// §3.3.3), not a Length mismatch: AcceptRequest moves on to the next request.
 func TestTruncatedOpenerResetsOnlyStream(t *testing.T) {
 	t.Parallel()
-	_, server, cliConn, _ := openPairWithConns(t)
-	stream, err := cliConn.OpenStream()
-	if err != nil {
-		t.Fatalf("OpenStream: %v", err)
-	}
-	go func() {
-		// Type SUBSCRIBE, Length 16, then only two body bytes and a FIN.
-		_, _ = stream.Write([]byte{byte(message.TypeSubscribe), 0x00, 0x10, 0x00, 0x00})
-		_ = stream.Close()
-	}()
-	if _, err := server.AcceptRequest(t.Context()); err == nil {
-		t.Fatal("AcceptRequest accepted a truncated opener")
+	for _, tc := range []struct {
+		name string
+		end  func(session.Stream)
+	}{
+		{"FIN", func(s session.Stream) { _ = s.Close() }},
+		{"reset", func(s session.Stream) { s.CancelWrite(0) }},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			t.Parallel()
+			client, server, cliConn, _ := openPairWithConns(t)
+			stream, err := cliConn.OpenStream()
+			if err != nil {
+				t.Fatalf("OpenStream: %v", err)
+			}
+			go func() {
+				// Type SUBSCRIBE, Length 16, then only two body bytes.
+				_, _ = stream.Write([]byte{byte(message.TypeSubscribe), 0x00, 0x10, 0x00, 0x00})
+				tc.end(stream)
+				_, _ = session.OpenRequestForTest(client, &message.Subscribe{
+					RequestID: 0, Namespace: videoNS, Name: []byte("next"),
+				})
+			}()
+			req, err := server.AcceptRequest(t.Context())
+			if err != nil {
+				t.Fatalf("AcceptRequest: %v, want the request after the truncated one", err)
+			}
+			if name := string(req.First.(*message.Subscribe).Name); name != "next" {
+				t.Fatalf("accepted %q, want the request after the truncated one", name)
+			}
+			requireStaysOpen(t, server, 100*time.Millisecond)
+		})
 	}
-	requireStaysOpen(t, server, 100*time.Millisecond)
 }
 
 func TestMalformedResponseClosesSession(t *testing.T) {

pkg/moqt/session/request.go

@@ -15,9 +15,10 @@ import (
 	"github.com/floatdrop/moq-go/pkg/moqt/track"
 )
 
-// ErrRequestIDParityViolation is returned by AcceptRequest when the peer sends
-// a Request ID whose parity does not match the expected value per §10.1.
-// The caller MUST close the session with SessionInvalidRequestID.
+// ErrRequestIDParityViolation is returned by [Session.CheckPeerRequestID] when
+// the peer sends a Request ID whose parity does not match the expected value
+// per §10.1. [Session.AcceptRequest] has already closed the session with
+// INVALID_REQUEST_ID; another caller of CheckPeerRequestID MUST.
 type ErrRequestIDParityViolation struct {
 	RequestID    uint64
 	ExpectedEven bool // true = expected even (peer is client), false = expected odd (peer is server)
@@ -40,7 +41,8 @@ func (e *ErrRequestIDParityViolation) Error() string {
 // Request ID" MUST close the session with INVALID_REQUEST_ID). Cross-stream
 // delivery reordering is tolerated — an ID below the high-water mark counts
 // as a duplicate only once every unseen ID it could have been is accounted
-// for. The caller MUST close the session with SessionInvalidRequestID.
+// for. [Session.AcceptRequest] has already closed the session with
+// INVALID_REQUEST_ID; another caller of CheckPeerRequestID MUST.
 type ErrDuplicateRequestID struct {
 	RequestID uint64
 	MaxSeen   uint64
@@ -239,8 +241,11 @@ type Request struct {
 // malformed (§10, wrapping [message.ErrMalformedMessage]), closes the session
 // with PROTOCOL_VIOLATION; the error is *ErrUnexpectedRequestOpener,
 // *ErrUnexpectedRequestUpdate, ErrUnexpectedPublishStateNotify or the parse
-// error. A stream that ends before its first message is complete only resets
-// that stream.
+// error. A Request ID violation (§10.1) closes it with INVALID_REQUEST_ID and
+// returns *ErrRequestIDParityViolation or *ErrDuplicateRequestID, and a token
+// cache fault (§10.2.2) with the *TokenCacheError's Code. A stream that ends
+// or is reset before its first message is complete fails only that request
+// (§3.3.2, §3.3.3): it is reset and AcceptRequest moves on.
 func (s *Session) AcceptRequest(ctx context.Context) (*Request, error) {
 	for {
 		stream, err := s.conn.AcceptStream(ctx)
@@ -252,16 +257,24 @@ func (s *Session) AcceptRequest(ctx context.Context) (*Request, error) {
 		// accept loop past cancellation.
 		msg, err := s.readResponse(ctx, stream)
 		if err != nil {
-			resetStream(stream)
 			if ctx.Err() != nil {
+				resetStream(stream)
 				return nil, ctx.Err()
 			}
 			// §3.3: readResponse already closed the session; this only shapes
 			// the error.
 			if typ, ok := errors.AsType[message.ErrUnknownType](err); ok {
+				resetStream(stream)
 				return nil, s.closeProtocolViolation(&ErrUnexpectedRequestOpener{Type: message.Type(typ)})
 			}
-			return nil, fmt.Errorf("moqt/session: parse request first message: %w", err)
+			if errors.Is(err, message.ErrMalformedMessage) {
+				resetStream(stream)
+				return nil, fmt.Errorf("moqt/session: parse request first message: %w", err)
+			}
+			// §3.3.2, §3.3.3: the peer ended or reset the stream first, which
+			// fails that request only. A closed session fails AcceptStream.
+			cancelRequest(stream)
+			continue
 		}
 
 		// §10.9, §3.3: REQUEST_UPDATE never opens a stream.
@@ -295,16 +308,20 @@ func (s *Session) AcceptRequest(ctx context.Context) (*Request, error) {
 		if m, ok := msg.(message.WithRequestID); ok {
 			if err := s.CheckPeerRequestID(m.GetRequestID()); err != nil {
 				resetStream(stream)
+				_ = s.Close(moqt.SessionInvalidRequestID, err.Error())
 				return nil, err
 			}
 		}
 
 		// §10.2.2: REGISTER tokens commit before any rejection, so the alias
 		// persists even if the request fails. A *TokenCacheError is
-		// session-fatal; the caller closes the session with its Code.
+		// session-fatal.
 		tokens, err := s.processRequestTokens(msg)
 		if err != nil {
 			resetStream(stream)
+			if tce, ok := errors.AsType[*TokenCacheError](err); ok {
+				_ = s.Close(tce.Code, tce.Error())
+			}
 			return nil, err
 		}
 
@@ -350,8 +367,7 @@ func evictLowestGapsLocked(gaps map[uint64]struct{}, n int) {
 // for those).
 //
 // Two violations are session-fatal per §10.1, and the caller MUST close the
-// session with [moqt.SessionInvalidRequestID] (AcceptRequest instead returns
-// the error to its caller, which owns that decision):
+// session with [moqt.SessionInvalidRequestID] (AcceptRequest does so itself):
 //
 //   - wrong parity for the sender (*ErrRequestIDParityViolation);
 //   - a duplicate ID (*ErrDuplicateRequestID).

pkg/moqt/session/request_test.go

@@ -237,6 +237,7 @@ func TestAcceptRequestDuplicateID(t *testing.T) {
 	if dupErr.RequestID != 0 {
 		t.Errorf("ErrDuplicateRequestID.RequestID = %d, want 0", dupErr.RequestID)
 	}
+	requireClosedCode(t, server, moqt.SessionInvalidRequestID)
 }
 
 // TestAcceptRequestOutOfOrderID verifies §10.1's receiver rules under
@@ -330,6 +331,7 @@ func TestAcceptRequestOutOfOrderID(t *testing.T) {
 	if dupErr.MaxSeen != 4 {
 		t.Errorf("ErrDuplicateRequestID.MaxSeen = %d, want 4", dupErr.MaxSeen)
 	}
+	requireClosedCode(t, server, moqt.SessionInvalidRequestID)
 }
 
 // TestAcceptRequestMonotonicHappyPath verifies that multiple requests with
@@ -428,6 +430,7 @@ func TestAcceptRequestParityViolation_ServerReceivesOddID(t *testing.T) {
 	if !parityErr.ExpectedEven {
 		t.Errorf("ErrRequestIDParityViolation.ExpectedEven = false, want true (server expects even IDs from client)")
 	}
+	requireClosedCode(t, server, moqt.SessionInvalidRequestID)
 }
 
 // TestAcceptRequestParityViolation_ClientReceivesEvenID verifies that when the
@@ -476,6 +479,7 @@ func TestAcceptRequestParityViolation_ClientReceivesEvenID(t *testing.T) {
 	if parityErr.ExpectedEven {
 		t.Errorf("ErrRequestIDParityViolation.ExpectedEven = true, want false (client expects odd IDs from server)")
 	}
+	requireClosedCode(t, client, moqt.SessionInvalidRequestID)
 }
 
 // TestAcceptRequestParityHappyPath verifies that correct-parity IDs are

pkg/moqt/session/requestmux.go

@@ -86,14 +86,9 @@ func (m *RequestMux) OnUnknown(f func(*Request)) {
 // until ctx is cancelled or [Session.AcceptRequest] returns an error, which Run
 // returns.
 //
-// Some AcceptRequest errors are session-fatal protocol violations — a §10.1
-// Request-ID parity/monotonicity violation (*ErrRequestIDParityViolation /
-// *ErrDuplicateRequestID) or a token-cache fault (*TokenCacheError) — that the
-// caller MUST escalate by closing the session with the mapped code (see
-// [Session.AcceptRequest]). Run surfaces the error unchanged so the caller can
-// inspect it with errors.As and Close accordingly. A request stream opened by
-// anything but a request message (§3.3, *ErrUnexpectedRequestOpener and
-// friends) arrives with the session already closed.
+// Run surfaces an AcceptRequest error unchanged. Any but ctx's means the
+// session has ended: a protocol violation (§3.3, §10, §10.1, §10.2.2) arrives
+// already closed with the mapped code (see [Session.AcceptRequest]).
 //
 // Dispatch is synchronous: a handler runs to completion before Run accepts the
 // next request, mirroring a hand-written accept loop and [Demux.Run]. A handler

pkg/moqt/session/token_verify.go

@@ -65,8 +65,9 @@ func (f TokenVerifierFunc) VerifyToken(ctx context.Context, sess *Session, tok R
 // inbound request's AUTHORIZATION_TOKEN parameters fails at the cache layer
 // (§10.2.2). These are session-level faults: a malformed token, a duplicate
 // REGISTER alias, a cache overflow, or a USE_ALIAS / DELETE referencing an
-// unknown alias. Code is the SESSION_ERROR the caller should close the
-// session with.
+// unknown alias. Code is the SESSION_ERROR the session is closed with:
+// AcceptRequest has already done so, and a caller of
+// [Session.ProcessFollowupTokens] MUST.
 type TokenCacheError struct {
 	// Code is the §10.2.2 SESSION_ERROR code to terminate the session with.
 	Code moqt.SessionErrorCode
@@ -212,10 +213,16 @@ func (s *Session) applyToken(t *message.Token) (tok ResolvedToken, ok bool, err
 // can call it from their own request loop. It is safe to call with a req whose
 // Tokens slice is empty.
 func (s *Session) VerifyRequestTokens(ctx context.Context, req *Request) error {
-	if s.tokenVerifier == nil || len(req.Tokens) == 0 {
+	return s.VerifyTokens(ctx, req.Tokens)
+}
+
+// VerifyTokens is [Session.VerifyRequestTokens] for tokens resolved by
+// [Session.ProcessFollowupTokens], such as a REQUEST_UPDATE's (§10.2.2).
+func (s *Session) VerifyTokens(ctx context.Context, toks []ResolvedToken) error {
+	if s.tokenVerifier == nil || len(toks) == 0 {
 		return nil
 	}
-	for _, tok := range req.Tokens {
+	for _, tok := range toks {
 		if err := s.tokenVerifier.VerifyToken(ctx, s, tok); err != nil {
 			if denied, ok := errors.AsType[*TokenDeniedError](err); ok {
 				return denied

pkg/moqt/session/token_verify_test.go

@@ -137,6 +137,7 @@ func TestAcceptRequestDuplicateAliasIsSessionError(t *testing.T) {
 	if tce.Code != moqt.SessionDuplicateAuthTokenAlias {
 		t.Errorf("Code = 0x%X, want SessionDuplicateAuthTokenAlias", uint64(tce.Code))
 	}
+	requireClosedCode(t, server, moqt.SessionDuplicateAuthTokenAlias)
 }
 
 // TestAcceptRequestUnknownAliasIsSessionError verifies that USE_ALIAS for an
@@ -159,6 +160,7 @@ func TestAcceptRequestUnknownAliasIsSessionError(t *testing.T) {
 	if tce.Code != moqt.SessionUnknownAuthTokenAlias {
 		t.Errorf("Code = 0x%X, want SessionUnknownAuthTokenAlias", uint64(tce.Code))
 	}
+	requireClosedCode(t, server, moqt.SessionUnknownAuthTokenAlias)
 }
 
 // TestAcceptRequestRegisterPersistsWhenAliasingProhibited verifies that with
@@ -184,6 +186,7 @@ func TestAcceptRequestRegisterProhibitedIsOverflow(t *testing.T) {
 	if tce.Code != moqt.SessionAuthTokenCacheOverflow {
 		t.Errorf("Code = 0x%X, want SessionAuthTokenCacheOverflow", uint64(tce.Code))
 	}
+	requireClosedCode(t, server, moqt.SessionAuthTokenCacheOverflow)
 }
 
 // TestVerifyRequestTokensAllow verifies that a verifier returning nil

pkg/relay/auth.go

@@ -15,6 +15,13 @@ import (
 // mutation; a non-nil return causes the relay to reply REQUEST_ERROR with
 // the [DeniedError]'s mapped code (see [DeniedError.RequestErrorCode]).
 //
+// A REQUEST_UPDATE that changes a SUBSCRIBE_NAMESPACE's or SUBSCRIBE_TRACKS's
+// TRACK_NAMESPACE_PREFIX is authorized again (§10.19, §10.20): the method
+// receives the subscription as updated: the new prefix, and the
+// AUTHORIZATION_TOKENs of the latest request or update that carried any (a
+// DELETE authorizes nothing). A denial refuses the update and ends the
+// subscription (§10.9.1).
+//
 // The interface is split per request type for two reasons:
 //
 //   - It lets a policy reject categories of request without having to
@@ -130,9 +137,9 @@ func ReasonForAuthorizerError(err error) string {
 //
 // Production deployments SHOULD replace this with a token- or
 // session-attestation-aware implementation via [Config.Authorizer]. The relay
-// only invokes the authorizer once per request before any state mutation, so
-// the cost of policy evaluation is bounded by the request rate rather than
-// the object rate.
+// invokes the authorizer once per request, and per prefix update, before any
+// state mutation, so the cost of policy evaluation is bounded by the request
+// rate rather than the object rate.
 type AllowAllAuthorizer struct{}
 
 var _ Authorizer = AllowAllAuthorizer{}