From e7307457ce141152732c0ded2bb392e613bc492a Mon Sep 17 00:00:00 2001
From: Jacob Gelman <3182119+ladvoc@users.noreply.github.com>
Date: Wed, 2 Sep 2026 15:02:10 -0700
Subject: [PATCH 1/9] Handle RequestResponse signal messages

---
 engine.go                   | 11 +++++++++++
 signalling/interfaces.go    |  1 +
 signalling/signalhandler.go |  3 +++
 3 files changed, 15 insertions(+)

diff --git a/engine.go b/engine.go
index 7f994658..4f388422 100644
--- a/engine.go
+++ b/engine.go
@@ -1677,6 +1677,17 @@ func (e *RTCEngine) OnMediaSectionsRequirement(mediaSectionsRequirement *livekit
 	e.engineHandler.OnMediaSectionsRequirement(mediaSectionsRequirement)
 }
 
+func (e *RTCEngine) OnRequestResponse(res *livekit.RequestResponse) {
+	if res.GetReason() != livekit.RequestResponse_OK {
+		e.log.Warnw(
+			"signal request failed", nil,
+			"requestID", res.GetRequestId(),
+			"reason", res.GetReason(),
+			"message", res.GetMessage(),
+		)
+	}
+}
+
 // ------------------------------------
 
 func setConfiguration(pcTransport *PCTransport, configuration webrtc.Configuration) {
diff --git a/signalling/interfaces.go b/signalling/interfaces.go
index 526f7d0b..ec57eb95 100644
--- a/signalling/interfaces.go
+++ b/signalling/interfaces.go
@@ -181,4 +181,5 @@ type SignalProcessor interface {
 	OnSubscribedQualityUpdate(subscribedQualityUpdate *livekit.SubscribedQualityUpdate)
 	OnSubscribedAudioCodecUpdate(subscribedAudioCodecUpdate *livekit.SubscribedAudioCodecUpdate)
 	OnMediaSectionsRequirement(mediaSectionsRequirement *livekit.MediaSectionsRequirement)
+	OnRequestResponse(requestResponse *livekit.RequestResponse)
 }
diff --git a/signalling/signalhandler.go b/signalling/signalhandler.go
index 74bd26e0..638b85a2 100644
--- a/signalling/signalhandler.go
+++ b/signalling/signalhandler.go
@@ -121,6 +121,9 @@ func (s *signalhandler) HandleMessage(msg proto.Message) error {
 
 	case *livekit.SignalResponse_MediaSectionsRequirement:
 		s.params.Processor.OnMediaSectionsRequirement(payload.MediaSectionsRequirement)
+
+	case *livekit.SignalResponse_RequestResponse:
+		s.params.Processor.OnRequestResponse(payload.RequestResponse)
 	}
 
 	return nil

From d17a9e95d6f6a7e1ca2ff4f68dbbd7bd8020a2c0 Mon Sep 17 00:00:00 2001
From: Jacob Gelman <3182119+ladvoc@users.noreply.github.com>
Date: Wed, 2 Sep 2026 15:02:33 -0700
Subject: [PATCH 2/9] Track pending signal requests by request id

---
 engine.go               |  11 ++++
 requestresponse.go      | 120 ++++++++++++++++++++++++++++++++++++++++
 requestresponse_test.go | 109 ++++++++++++++++++++++++++++++++++++
 3 files changed, 240 insertions(+)
 create mode 100644 requestresponse.go
 create mode 100644 requestresponse_test.go

diff --git a/engine.go b/engine.go
index 4f388422..f63467c4 100644
--- a/engine.go
+++ b/engine.go
@@ -130,6 +130,11 @@ type RTCEngine struct {
 	trackPublishedListenersLock sync.Mutex
 	trackPublishedListeners     map[string]chan *livekit.TrackPublishedResponse
 
+	// signal requests waiting for their RequestResponse, keyed by request id
+	requestIDCounter    atomic.Uint32
+	pendingRequestsLock sync.Mutex
+	pendingRequests     map[uint32]chan *livekit.RequestResponse
+
 	subscriberPrimary bool
 	hasPublish        atomic.Bool
 	closed            atomic.Bool
@@ -154,6 +159,7 @@ func NewRTCEngine(
 		engineHandler:            engineHandler,
 		cbGetLocalParticipantSID: getLocalParticipantSID,
 		trackPublishedListeners:  make(map[string]chan *livekit.TrackPublishedResponse),
+		pendingRequests:          make(map[uint32]chan *livekit.RequestResponse),
 		reliableMsgSeq:           1,
 		connectionManager:        newConnectionManager(regionProvider),
 	}
@@ -359,6 +365,7 @@ func (e *RTCEngine) Close() {
 	}
 
 	e.connectionManager.setClosed()
+	e.abortPendingRequests()
 
 	e.pclock.Lock()
 	e.pendingPublisherOffer = webrtc.SessionDescription{}
@@ -1678,6 +1685,10 @@ func (e *RTCEngine) OnMediaSectionsRequirement(mediaSectionsRequirement *livekit
 }
 
 func (e *RTCEngine) OnRequestResponse(res *livekit.RequestResponse) {
+	if e.deliverRequestResponse(res) {
+		return
+	}
+
 	if res.GetReason() != livekit.RequestResponse_OK {
 		e.log.Warnw(
 			"signal request failed", nil,
diff --git a/requestresponse.go b/requestresponse.go
new file mode 100644
index 00000000..ada54ed5
--- /dev/null
+++ b/requestresponse.go
@@ -0,0 +1,120 @@
+// Copyright 2026 LiveKit, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package lksdk
+
+import (
+	"context"
+
+	"github.com/livekit/protocol/livekit"
+)
+
+// pendingRequest reserves the RequestResponse for a signal request that carries a request id.
+//
+// Create it with newPendingRequest before sending the request, put ID() in the request's
+// request_id field, then Await the response. Once Await returns the reservation is gone and
+// a late response is treated as uncorrelated.
+type pendingRequest struct {
+	engine *RTCEngine
+	id     uint32
+	ch     chan *livekit.RequestResponse
+}
+
+func (e *RTCEngine) newPendingRequest() *pendingRequest {
+	p := &pendingRequest{
+		engine: e,
+		id:     e.nextRequestID(),
+		ch:     make(chan *livekit.RequestResponse, 1),
+	}
+
+	e.pendingRequestsLock.Lock()
+	if e.closed.Load() {
+		// nothing will answer, fail Await right away
+		close(p.ch)
+	} else {
+		e.pendingRequests[p.id] = p.ch
+	}
+	e.pendingRequestsLock.Unlock()
+
+	return p
+}
+
+// nextRequestID returns an id for correlating a signal request with its RequestResponse.
+// 0 is never returned, it marks a request that is not correlated.
+func (e *RTCEngine) nextRequestID() uint32 {
+	for {
+		if id := e.requestIDCounter.Inc(); id != 0 {
+			return id
+		}
+	}
+}
+
+// ID is the value to send in the request's request_id field.
+func (p *pendingRequest) ID() uint32 {
+	return p.id
+}
+
+// Await blocks until the response arrives, ctx is done, or the engine is closed.
+func (p *pendingRequest) Await(ctx context.Context) (*livekit.RequestResponse, error) {
+	defer p.engine.removePendingRequest(p.id)
+
+	select {
+	case res, ok := <-p.ch:
+		if !ok {
+			return nil, ErrAborted
+		}
+		return res, nil
+
+	case <-ctx.Done():
+		return nil, ctx.Err()
+	}
+}
+
+func (e *RTCEngine) removePendingRequest(requestID uint32) {
+	e.pendingRequestsLock.Lock()
+	delete(e.pendingRequests, requestID)
+	e.pendingRequestsLock.Unlock()
+}
+
+// deliverRequestResponse hands res to the request waiting for its id.
+// Returns false if no request is waiting for it.
+func (e *RTCEngine) deliverRequestResponse(res *livekit.RequestResponse) bool {
+	if res.GetRequestId() == 0 {
+		return false
+	}
+
+	e.pendingRequestsLock.Lock()
+	ch, ok := e.pendingRequests[res.RequestId]
+	delete(e.pendingRequests, res.RequestId)
+	e.pendingRequestsLock.Unlock()
+	if !ok {
+		return false
+	}
+
+	// buffered, and removed from the map above so this is the only send
+	ch <- res
+	return true
+}
+
+// abortPendingRequests fails every request still waiting for a response.
+func (e *RTCEngine) abortPendingRequests() {
+	e.pendingRequestsLock.Lock()
+	pending := e.pendingRequests
+	e.pendingRequests = make(map[uint32]chan *livekit.RequestResponse)
+	e.pendingRequestsLock.Unlock()
+
+	for _, ch := range pending {
+		close(ch)
+	}
+}
diff --git a/requestresponse_test.go b/requestresponse_test.go
new file mode 100644
index 00000000..b805cf3e
--- /dev/null
+++ b/requestresponse_test.go
@@ -0,0 +1,109 @@
+// Copyright 2026 LiveKit, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package lksdk
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/livekit/protocol/livekit"
+	"github.com/stretchr/testify/require"
+)
+
+func newTestEngine(t *testing.T) *RTCEngine {
+	engine := NewRoom(&RoomCallback{}).engine
+	t.Cleanup(engine.Close)
+	return engine
+}
+
+func (e *RTCEngine) pendingRequestCount() int {
+	e.pendingRequestsLock.Lock()
+	defer e.pendingRequestsLock.Unlock()
+	return len(e.pendingRequests)
+}
+
+func TestPendingRequestDelivered(t *testing.T) {
+	engine := newTestEngine(t)
+
+	pending := engine.newPendingRequest()
+	require.NotZero(t, pending.ID())
+	require.Equal(t, 1, engine.pendingRequestCount())
+
+	go engine.OnRequestResponse(&livekit.RequestResponse{
+		RequestId: pending.ID(),
+		Reason:    livekit.RequestResponse_NOT_ALLOWED,
+		Message:   "no permission",
+	})
+
+	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
+	defer cancel()
+	res, err := pending.Await(ctx)
+	require.NoError(t, err)
+	require.Equal(t, pending.ID(), res.RequestId)
+	require.Equal(t, livekit.RequestResponse_NOT_ALLOWED, res.Reason)
+	require.Equal(t, "no permission", res.Message)
+	require.Zero(t, engine.pendingRequestCount())
+}
+
+func TestPendingRequestContextDone(t *testing.T) {
+	engine := newTestEngine(t)
+
+	pending := engine.newPendingRequest()
+	ctx, cancel := context.WithCancel(context.Background())
+	cancel()
+	_, err := pending.Await(ctx)
+	require.ErrorIs(t, err, context.Canceled)
+	require.Zero(t, engine.pendingRequestCount())
+
+	// a response arriving after the caller gave up is no longer correlated
+	require.False(t, engine.deliverRequestResponse(&livekit.RequestResponse{RequestId: pending.ID()}))
+}
+
+func TestPendingRequestAbortedOnClose(t *testing.T) {
+	engine := newTestEngine(t)
+
+	pending := engine.newPendingRequest()
+	engine.Close()
+	_, err := pending.Await(context.Background())
+	require.ErrorIs(t, err, ErrAborted)
+
+	// requests created after close fail immediately
+	_, err = engine.newPendingRequest().Await(context.Background())
+	require.ErrorIs(t, err, ErrAborted)
+	require.Zero(t, engine.pendingRequestCount())
+}
+
+func TestUncorrelatedRequestResponseIgnored(t *testing.T) {
+	engine := newTestEngine(t)
+
+	// request id 0 is never matched to a pending request
+	require.False(t, engine.deliverRequestResponse(&livekit.RequestResponse{Reason: livekit.RequestResponse_NOT_ALLOWED}))
+	// neither is an id nobody is waiting on
+	require.False(t, engine.deliverRequestResponse(&livekit.RequestResponse{RequestId: 12345}))
+}
+
+func TestRequestIDsUnique(t *testing.T) {
+	engine := newTestEngine(t)
+
+	seen := make(map[uint32]struct{})
+	for range 1000 {
+		id := engine.nextRequestID()
+		require.NotZero(t, id)
+		_, dup := seen[id]
+		require.False(t, dup)
+		seen[id] = struct{}{}
+	}
+}

From 4b03238b10ce98a5b2a3a2b1e3abe078d5d721a0 Mon Sep 17 00:00:00 2001
From: Jacob Gelman <3182119+ladvoc@users.noreply.github.com>
Date: Wed, 2 Sep 2026 15:03:28 -0700
Subject: [PATCH 3/9] Forward unmatched request responses to the engine handler

---
 engine.go               | 10 ++--------
 requestresponse_test.go | 32 ++++++++++++++++++++++++++++++++
 room.go                 | 12 ++++++++++++
 3 files changed, 46 insertions(+), 8 deletions(-)

diff --git a/engine.go b/engine.go
index f63467c4..a94f23a1 100644
--- a/engine.go
+++ b/engine.go
@@ -77,6 +77,7 @@ type engineHandler interface {
 	OnSubscribedQualityUpdate(subscribedQualityUpdate *livekit.SubscribedQualityUpdate)
 	OnSubscribedAudioCodecUpdate(subscribedAudioCodecUpdate *livekit.SubscribedAudioCodecUpdate)
 	OnMediaSectionsRequirement(mediaSectionsRequirement *livekit.MediaSectionsRequirement)
+	OnRequestResponse(requestResponse *livekit.RequestResponse)
 }
 
 // -------------------------------------------
@@ -1689,14 +1690,7 @@ func (e *RTCEngine) OnRequestResponse(res *livekit.RequestResponse) {
 		return
 	}
 
-	if res.GetReason() != livekit.RequestResponse_OK {
-		e.log.Warnw(
-			"signal request failed", nil,
-			"requestID", res.GetRequestId(),
-			"reason", res.GetReason(),
-			"message", res.GetMessage(),
-		)
-	}
+	e.engineHandler.OnRequestResponse(res)
 }
 
 // ------------------------------------
diff --git a/requestresponse_test.go b/requestresponse_test.go
index b805cf3e..7c563783 100644
--- a/requestresponse_test.go
+++ b/requestresponse_test.go
@@ -107,3 +107,35 @@ func TestRequestIDsUnique(t *testing.T) {
 		seen[id] = struct{}{}
 	}
 }
+
+type requestResponseRecorder struct {
+	*Room
+	responses chan *livekit.RequestResponse
+}
+
+func (r *requestResponseRecorder) OnRequestResponse(res *livekit.RequestResponse) {
+	r.responses <- res
+}
+
+func TestRequestResponseForwardedToHandler(t *testing.T) {
+	recorder := &requestResponseRecorder{
+		Room:      NewRoom(&RoomCallback{}),
+		responses: make(chan *livekit.RequestResponse, 1),
+	}
+	t.Cleanup(recorder.Room.engine.Close)
+	engine := NewRTCEngine(false, recorder, func() string { return "" }, newRegionURLProvider())
+	t.Cleanup(engine.Close)
+
+	// no request id: nothing to correlate, handed to the handler
+	uncorrelated := &livekit.RequestResponse{Reason: livekit.RequestResponse_NOT_ALLOWED}
+	engine.OnRequestResponse(uncorrelated)
+	require.Same(t, uncorrelated, <-recorder.responses)
+
+	// a waiting request consumes its response, the handler is not involved
+	pending := engine.newPendingRequest()
+	engine.OnRequestResponse(&livekit.RequestResponse{RequestId: pending.ID()})
+	res, err := pending.Await(context.Background())
+	require.NoError(t, err)
+	require.Equal(t, pending.ID(), res.RequestId)
+	require.Empty(t, recorder.responses)
+}
diff --git a/room.go b/room.go
index bfa3039f..075f17f0 100644
--- a/room.go
+++ b/room.go
@@ -1399,6 +1399,18 @@ func (r *Room) OnMediaSectionsRequirement(mediaSectionsRequirement *livekit.Medi
 	publisher.Negotiate()
 }
 
+// OnRequestResponse handles responses to signal requests that no caller is waiting on.
+func (r *Room) OnRequestResponse(res *livekit.RequestResponse) {
+	if res.GetReason() != livekit.RequestResponse_OK {
+		r.log.Warnw(
+			"signal request failed", nil,
+			"requestID", res.GetRequestId(),
+			"reason", res.GetReason(),
+			"message", res.GetMessage(),
+		)
+	}
+}
+
 func (r *Room) OnStreamHeader(streamHeader *livekit.DataStream_Header, participantIdentity string) {
 	switch header := streamHeader.ContentHeader.(type) {
 	case *livekit.DataStream_Header_TextHeader:

From a024206d9039140cbb345baa350eb7b68b974326 Mon Sep 17 00:00:00 2001
From: Jacob Gelman <3182119+ladvoc@users.noreply.github.com>
Date: Wed, 2 Sep 2026 15:41:11 -0700
Subject: [PATCH 4/9] Reserve request ids under the pending request lock

---
 engine.go               |  2 +-
 requestresponse.go      | 16 +++++++++------
 requestresponse_test.go | 45 ++++++++++-------------------------------
 3 files changed, 22 insertions(+), 41 deletions(-)

diff --git a/engine.go b/engine.go
index a94f23a1..d8cfd663 100644
--- a/engine.go
+++ b/engine.go
@@ -132,8 +132,8 @@ type RTCEngine struct {
 	trackPublishedListeners     map[string]chan *livekit.TrackPublishedResponse
 
 	// signal requests waiting for their RequestResponse, keyed by request id
-	requestIDCounter    atomic.Uint32
 	pendingRequestsLock sync.Mutex
+	nextRequestID       uint32 // guarded by pendingRequestsLock
 	pendingRequests     map[uint32]chan *livekit.RequestResponse
 
 	subscriberPrimary bool
diff --git a/requestresponse.go b/requestresponse.go
index ada54ed5..d56d8de8 100644
--- a/requestresponse.go
+++ b/requestresponse.go
@@ -34,11 +34,11 @@ type pendingRequest struct {
 func (e *RTCEngine) newPendingRequest() *pendingRequest {
 	p := &pendingRequest{
 		engine: e,
-		id:     e.nextRequestID(),
 		ch:     make(chan *livekit.RequestResponse, 1),
 	}
 
 	e.pendingRequestsLock.Lock()
+	p.id = e.allocateRequestIDLocked()
 	if e.closed.Load() {
 		// nothing will answer, fail Await right away
 		close(p.ch)
@@ -50,12 +50,16 @@ func (e *RTCEngine) newPendingRequest() *pendingRequest {
 	return p
 }
 
-// nextRequestID returns an id for correlating a signal request with its RequestResponse.
-// 0 is never returned, it marks a request that is not correlated.
-func (e *RTCEngine) nextRequestID() uint32 {
+// allocateRequestIDLocked returns the next request id not in use by a pending request, skipping
+// zero. Wrapping around is harmless: an id is reused only once its previous request is done.
+func (e *RTCEngine) allocateRequestIDLocked() uint32 {
 	for {
-		if id := e.requestIDCounter.Inc(); id != 0 {
-			return id
+		e.nextRequestID++
+		if e.nextRequestID == 0 {
+			continue
+		}
+		if _, inUse := e.pendingRequests[e.nextRequestID]; !inUse {
+			return e.nextRequestID
 		}
 	}
 }
diff --git a/requestresponse_test.go b/requestresponse_test.go
index 7c563783..b1a22870 100644
--- a/requestresponse_test.go
+++ b/requestresponse_test.go
@@ -16,6 +16,7 @@ package lksdk
 
 import (
 	"context"
+	"math"
 	"testing"
 	"time"
 
@@ -95,47 +96,23 @@ func TestUncorrelatedRequestResponseIgnored(t *testing.T) {
 	require.False(t, engine.deliverRequestResponse(&livekit.RequestResponse{RequestId: 12345}))
 }
 
-func TestRequestIDsUnique(t *testing.T) {
+func TestRequestIDAllocation(t *testing.T) {
 	engine := newTestEngine(t)
 
+	// ids are unique while their requests are pending
 	seen := make(map[uint32]struct{})
-	for range 1000 {
-		id := engine.nextRequestID()
+	for range 100 {
+		id := engine.newPendingRequest().ID()
 		require.NotZero(t, id)
 		_, dup := seen[id]
 		require.False(t, dup)
 		seen[id] = struct{}{}
 	}
-}
-
-type requestResponseRecorder struct {
-	*Room
-	responses chan *livekit.RequestResponse
-}
-
-func (r *requestResponseRecorder) OnRequestResponse(res *livekit.RequestResponse) {
-	r.responses <- res
-}
-
-func TestRequestResponseForwardedToHandler(t *testing.T) {
-	recorder := &requestResponseRecorder{
-		Room:      NewRoom(&RoomCallback{}),
-		responses: make(chan *livekit.RequestResponse, 1),
-	}
-	t.Cleanup(recorder.Room.engine.Close)
-	engine := NewRTCEngine(false, recorder, func() string { return "" }, newRegionURLProvider())
-	t.Cleanup(engine.Close)
-
-	// no request id: nothing to correlate, handed to the handler
-	uncorrelated := &livekit.RequestResponse{Reason: livekit.RequestResponse_NOT_ALLOWED}
-	engine.OnRequestResponse(uncorrelated)
-	require.Same(t, uncorrelated, <-recorder.responses)
 
-	// a waiting request consumes its response, the handler is not involved
-	pending := engine.newPendingRequest()
-	engine.OnRequestResponse(&livekit.RequestResponse{RequestId: pending.ID()})
-	res, err := pending.Await(context.Background())
-	require.NoError(t, err)
-	require.Equal(t, pending.ID(), res.RequestId)
-	require.Empty(t, recorder.responses)
+	// the counter wraps around without handing out zero or an id that is still in use
+	engine.pendingRequestsLock.Lock()
+	engine.nextRequestID = math.MaxUint32 - 1
+	engine.pendingRequestsLock.Unlock()
+	require.Equal(t, uint32(math.MaxUint32), engine.newPendingRequest().ID())
+	require.Equal(t, uint32(101), engine.newPendingRequest().ID())
 }

From c7644f768ba1125e1c368575aac5a6a22eb7ea00 Mon Sep 17 00:00:00 2001
From: Jacob Gelman <3182119+ladvoc@users.noreply.github.com>
Date: Wed, 2 Sep 2026 15:41:21 -0700
Subject: [PATCH 5/9] Deliver typed responses to pending requests

---
 engine.go               |  8 ++--
 requestresponse.go      | 68 +++++++++++++++++++++++--------
 requestresponse_test.go | 90 +++++++++++++++++++++++++++++++++++------
 3 files changed, 133 insertions(+), 33 deletions(-)

diff --git a/engine.go b/engine.go
index d8cfd663..85c45a0e 100644
--- a/engine.go
+++ b/engine.go
@@ -131,10 +131,10 @@ type RTCEngine struct {
 	trackPublishedListenersLock sync.Mutex
 	trackPublishedListeners     map[string]chan *livekit.TrackPublishedResponse
 
-	// signal requests waiting for their RequestResponse, keyed by request id
+	// signal requests waiting for a response, keyed by request id
 	pendingRequestsLock sync.Mutex
 	nextRequestID       uint32 // guarded by pendingRequestsLock
-	pendingRequests     map[uint32]chan *livekit.RequestResponse
+	pendingRequests     map[uint32]chan proto.Message
 
 	subscriberPrimary bool
 	hasPublish        atomic.Bool
@@ -160,7 +160,7 @@ func NewRTCEngine(
 		engineHandler:            engineHandler,
 		cbGetLocalParticipantSID: getLocalParticipantSID,
 		trackPublishedListeners:  make(map[string]chan *livekit.TrackPublishedResponse),
-		pendingRequests:          make(map[uint32]chan *livekit.RequestResponse),
+		pendingRequests:          make(map[uint32]chan proto.Message),
 		reliableMsgSeq:           1,
 		connectionManager:        newConnectionManager(regionProvider),
 	}
@@ -1686,7 +1686,7 @@ func (e *RTCEngine) OnMediaSectionsRequirement(mediaSectionsRequirement *livekit
 }
 
 func (e *RTCEngine) OnRequestResponse(res *livekit.RequestResponse) {
-	if e.deliverRequestResponse(res) {
+	if e.deliverResponse(res.GetRequestId(), res) {
 		return
 	}
 
diff --git a/requestresponse.go b/requestresponse.go
index d56d8de8..af999dba 100644
--- a/requestresponse.go
+++ b/requestresponse.go
@@ -16,25 +16,35 @@ package lksdk
 
 import (
 	"context"
+	"fmt"
 
 	"github.com/livekit/protocol/livekit"
+	"google.golang.org/protobuf/proto"
 )
 
-// pendingRequest reserves the RequestResponse for a signal request that carries a request id.
-//
-// Create it with newPendingRequest before sending the request, put ID() in the request's
-// request_id field, then Await the response. Once Await returns the reservation is gone and
-// a late response is treated as uncorrelated.
+// SignalRequestError is returned when the server rejects a signal request.
+type SignalRequestError struct {
+	RequestID uint32
+	Reason    livekit.RequestResponse_Reason
+	Message   string
+}
+
+func (e *SignalRequestError) Error() string {
+	return fmt.Sprintf("signal request rejected (%s): %s", e.Reason, e.Message)
+}
+
+// pendingRequest reserves the response to a signal request that carries a request id: create it
+// before sending the request, put ID() in the request_id field, then Await the response.
 type pendingRequest struct {
 	engine *RTCEngine
 	id     uint32
-	ch     chan *livekit.RequestResponse
+	ch     chan proto.Message
 }
 
 func (e *RTCEngine) newPendingRequest() *pendingRequest {
 	p := &pendingRequest{
 		engine: e,
-		ch:     make(chan *livekit.RequestResponse, 1),
+		ch:     make(chan proto.Message, 1),
 	}
 
 	e.pendingRequestsLock.Lock()
@@ -64,13 +74,16 @@ func (e *RTCEngine) allocateRequestIDLocked() uint32 {
 	}
 }
 
-// ID is the value to send in the request's request_id field.
 func (p *pendingRequest) ID() uint32 {
 	return p.id
 }
 
-// Await blocks until the response arrives, ctx is done, or the engine is closed.
-func (p *pendingRequest) Await(ctx context.Context) (*livekit.RequestResponse, error) {
+// Await blocks until a response arrives, ctx is done, or the engine is closed.
+//
+// A RequestResponse with a reason other than OK is returned as a *SignalRequestError. Any
+// other message, including an OK RequestResponse, is returned as is; see awaitResponse to
+// also assert its type.
+func (p *pendingRequest) Await(ctx context.Context) (proto.Message, error) {
 	defer p.engine.removePendingRequest(p.id)
 
 	select {
@@ -78,6 +91,13 @@ func (p *pendingRequest) Await(ctx context.Context) (*livekit.RequestResponse, e
 		if !ok {
 			return nil, ErrAborted
 		}
+		if rr, isRequestResponse := res.(*livekit.RequestResponse); isRequestResponse && rr.GetReason() != livekit.RequestResponse_OK {
+			return nil, &SignalRequestError{
+				RequestID: rr.GetRequestId(),
+				Reason:    rr.GetReason(),
+				Message:   rr.GetMessage(),
+			}
+		}
 		return res, nil
 
 	case <-ctx.Done():
@@ -85,22 +105,36 @@ func (p *pendingRequest) Await(ctx context.Context) (*livekit.RequestResponse, e
 	}
 }
 
+// awaitResponse is Await for requests whose success is reported by a dedicated message type.
+func awaitResponse[T proto.Message](ctx context.Context, p *pendingRequest) (T, error) {
+	var zero T
+
+	res, err := p.Await(ctx)
+	if err != nil {
+		return zero, err
+	}
+	typed, ok := res.(T)
+	if !ok {
+		return zero, fmt.Errorf("unexpected response %T to signal request %d", res, p.id)
+	}
+	return typed, nil
+}
+
 func (e *RTCEngine) removePendingRequest(requestID uint32) {
 	e.pendingRequestsLock.Lock()
 	delete(e.pendingRequests, requestID)
 	e.pendingRequestsLock.Unlock()
 }
 
-// deliverRequestResponse hands res to the request waiting for its id.
-// Returns false if no request is waiting for it.
-func (e *RTCEngine) deliverRequestResponse(res *livekit.RequestResponse) bool {
-	if res.GetRequestId() == 0 {
+// deliverResponse hands res to the request waiting for requestID, if any.
+func (e *RTCEngine) deliverResponse(requestID uint32, res proto.Message) bool {
+	if requestID == 0 {
 		return false
 	}
 
 	e.pendingRequestsLock.Lock()
-	ch, ok := e.pendingRequests[res.RequestId]
-	delete(e.pendingRequests, res.RequestId)
+	ch, ok := e.pendingRequests[requestID]
+	delete(e.pendingRequests, requestID)
 	e.pendingRequestsLock.Unlock()
 	if !ok {
 		return false
@@ -115,7 +149,7 @@ func (e *RTCEngine) deliverRequestResponse(res *livekit.RequestResponse) bool {
 func (e *RTCEngine) abortPendingRequests() {
 	e.pendingRequestsLock.Lock()
 	pending := e.pendingRequests
-	e.pendingRequests = make(map[uint32]chan *livekit.RequestResponse)
+	e.pendingRequests = make(map[uint32]chan proto.Message)
 	e.pendingRequestsLock.Unlock()
 
 	for _, ch := range pending {
diff --git a/requestresponse_test.go b/requestresponse_test.go
index b1a22870..345b6cf2 100644
--- a/requestresponse_test.go
+++ b/requestresponse_test.go
@@ -43,6 +43,20 @@ func TestPendingRequestDelivered(t *testing.T) {
 	require.NotZero(t, pending.ID())
 	require.Equal(t, 1, engine.pendingRequestCount())
 
+	go engine.OnRequestResponse(&livekit.RequestResponse{RequestId: pending.ID()})
+
+	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
+	defer cancel()
+	res, err := pending.Await(ctx)
+	require.NoError(t, err)
+	require.Equal(t, pending.ID(), res.(*livekit.RequestResponse).RequestId)
+	require.Zero(t, engine.pendingRequestCount())
+}
+
+func TestPendingRequestRejected(t *testing.T) {
+	engine := newTestEngine(t)
+
+	pending := engine.newPendingRequest()
 	go engine.OnRequestResponse(&livekit.RequestResponse{
 		RequestId: pending.ID(),
 		Reason:    livekit.RequestResponse_NOT_ALLOWED,
@@ -51,12 +65,37 @@ func TestPendingRequestDelivered(t *testing.T) {
 
 	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
 	defer cancel()
-	res, err := pending.Await(ctx)
+	_, err := pending.Await(ctx)
+	var rejected *SignalRequestError
+	require.ErrorAs(t, err, &rejected)
+	require.Equal(t, pending.ID(), rejected.RequestID)
+	require.Equal(t, livekit.RequestResponse_NOT_ALLOWED, rejected.Reason)
+	require.Equal(t, "no permission", rejected.Message)
+}
+
+func TestPendingRequestTypedResponse(t *testing.T) {
+	engine := newTestEngine(t)
+	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
+	defer cancel()
+
+	pending := engine.newPendingRequest()
+	go engine.deliverResponse(pending.ID(), &livekit.StoreDataBlobResponse{RequestId: pending.ID()})
+	res, err := awaitResponse[*livekit.StoreDataBlobResponse](ctx, pending)
 	require.NoError(t, err)
 	require.Equal(t, pending.ID(), res.RequestId)
-	require.Equal(t, livekit.RequestResponse_NOT_ALLOWED, res.Reason)
-	require.Equal(t, "no permission", res.Message)
-	require.Zero(t, engine.pendingRequestCount())
+
+	pending = engine.newPendingRequest()
+	go engine.deliverResponse(pending.ID(), &livekit.RequestResponse{RequestId: pending.ID(), Reason: livekit.RequestResponse_NOT_FOUND})
+	_, err = awaitResponse[*livekit.GetDataBlobResponse](ctx, pending)
+	var rejected *SignalRequestError
+	require.ErrorAs(t, err, &rejected)
+	require.Equal(t, livekit.RequestResponse_NOT_FOUND, rejected.Reason)
+
+	// a response of the wrong type is an error, not a panic
+	pending = engine.newPendingRequest()
+	go engine.deliverResponse(pending.ID(), &livekit.RequestResponse{RequestId: pending.ID()})
+	_, err = awaitResponse[*livekit.GetDataBlobResponse](ctx, pending)
+	require.Error(t, err)
 }
 
 func TestPendingRequestContextDone(t *testing.T) {
@@ -69,8 +108,7 @@ func TestPendingRequestContextDone(t *testing.T) {
 	require.ErrorIs(t, err, context.Canceled)
 	require.Zero(t, engine.pendingRequestCount())
 
-	// a response arriving after the caller gave up is no longer correlated
-	require.False(t, engine.deliverRequestResponse(&livekit.RequestResponse{RequestId: pending.ID()}))
+	require.False(t, engine.deliverResponse(pending.ID(), &livekit.RequestResponse{RequestId: pending.ID()}))
 }
 
 func TestPendingRequestAbortedOnClose(t *testing.T) {
@@ -81,25 +119,22 @@ func TestPendingRequestAbortedOnClose(t *testing.T) {
 	_, err := pending.Await(context.Background())
 	require.ErrorIs(t, err, ErrAborted)
 
-	// requests created after close fail immediately
 	_, err = engine.newPendingRequest().Await(context.Background())
 	require.ErrorIs(t, err, ErrAborted)
 	require.Zero(t, engine.pendingRequestCount())
 }
 
-func TestUncorrelatedRequestResponseIgnored(t *testing.T) {
+func TestUncorrelatedResponseIgnored(t *testing.T) {
 	engine := newTestEngine(t)
 
 	// request id 0 is never matched to a pending request
-	require.False(t, engine.deliverRequestResponse(&livekit.RequestResponse{Reason: livekit.RequestResponse_NOT_ALLOWED}))
-	// neither is an id nobody is waiting on
-	require.False(t, engine.deliverRequestResponse(&livekit.RequestResponse{RequestId: 12345}))
+	require.False(t, engine.deliverResponse(0, &livekit.RequestResponse{Reason: livekit.RequestResponse_NOT_ALLOWED}))
+	require.False(t, engine.deliverResponse(12345, &livekit.RequestResponse{RequestId: 12345}))
 }
 
 func TestRequestIDAllocation(t *testing.T) {
 	engine := newTestEngine(t)
 
-	// ids are unique while their requests are pending
 	seen := make(map[uint32]struct{})
 	for range 100 {
 		id := engine.newPendingRequest().ID()
@@ -116,3 +151,34 @@ func TestRequestIDAllocation(t *testing.T) {
 	require.Equal(t, uint32(math.MaxUint32), engine.newPendingRequest().ID())
 	require.Equal(t, uint32(101), engine.newPendingRequest().ID())
 }
+
+type requestResponseRecorder struct {
+	*Room
+	responses chan *livekit.RequestResponse
+}
+
+func (r *requestResponseRecorder) OnRequestResponse(res *livekit.RequestResponse) {
+	r.responses <- res
+}
+
+func TestRequestResponseForwardedToHandler(t *testing.T) {
+	recorder := &requestResponseRecorder{
+		Room:      NewRoom(&RoomCallback{}),
+		responses: make(chan *livekit.RequestResponse, 1),
+	}
+	t.Cleanup(recorder.Room.engine.Close)
+	engine := NewRTCEngine(false, recorder, func() string { return "" }, newRegionURLProvider())
+	t.Cleanup(engine.Close)
+
+	// no request id: nothing to correlate, handed to the handler
+	uncorrelated := &livekit.RequestResponse{Reason: livekit.RequestResponse_NOT_ALLOWED}
+	engine.OnRequestResponse(uncorrelated)
+	require.Same(t, uncorrelated, <-recorder.responses)
+
+	// a waiting request consumes its response, the handler is not involved
+	pending := engine.newPendingRequest()
+	engine.OnRequestResponse(&livekit.RequestResponse{RequestId: pending.ID()})
+	_, err := pending.Await(context.Background())
+	require.NoError(t, err)
+	require.Empty(t, recorder.responses)
+}

From 6cdebaed64091bf64337afb32dd0cd42ab343bcd Mon Sep 17 00:00:00 2001
From: Jacob Gelman <3182119+ladvoc@users.noreply.github.com>
Date: Wed, 2 Sep 2026 15:54:34 -0700
Subject: [PATCH 6/9] Add default timeout helper for signal requests

---
 requestresponse.go      | 18 +++++++++++++-----
 requestresponse_test.go | 26 ++++++++++++++++++++++++++
 2 files changed, 39 insertions(+), 5 deletions(-)

diff --git a/requestresponse.go b/requestresponse.go
index af999dba..b90dba00 100644
--- a/requestresponse.go
+++ b/requestresponse.go
@@ -17,6 +17,7 @@ package lksdk
 import (
 	"context"
 	"fmt"
+	"time"
 
 	"github.com/livekit/protocol/livekit"
 	"google.golang.org/protobuf/proto"
@@ -78,11 +79,9 @@ func (p *pendingRequest) ID() uint32 {
 	return p.id
 }
 
-// Await blocks until a response arrives, ctx is done, or the engine is closed.
-//
-// A RequestResponse with a reason other than OK is returned as a *SignalRequestError. Any
-// other message, including an OK RequestResponse, is returned as is; see awaitResponse to
-// also assert its type.
+// Await blocks until a response arrives, ctx is done, or the engine is closed. It has no timeout
+// of its own, see withDefaultTimeout. A RequestResponse with a reason other than OK is returned
+// as a *SignalRequestError.
 func (p *pendingRequest) Await(ctx context.Context) (proto.Message, error) {
 	defer p.engine.removePendingRequest(p.id)
 
@@ -120,6 +119,15 @@ func awaitResponse[T proto.Message](ctx context.Context, p *pendingRequest) (T,
 	return typed, nil
 }
 
+// withDefaultTimeout applies deadline d to ctx when it has none. A deadline the caller set is
+// never shortened. The returned cancel must be called.
+func withDefaultTimeout(ctx context.Context, d time.Duration) (context.Context, context.CancelFunc) {
+	if _, ok := ctx.Deadline(); ok {
+		return ctx, func() {}
+	}
+	return context.WithTimeout(ctx, d)
+}
+
 func (e *RTCEngine) removePendingRequest(requestID uint32) {
 	e.pendingRequestsLock.Lock()
 	delete(e.pendingRequests, requestID)
diff --git a/requestresponse_test.go b/requestresponse_test.go
index 345b6cf2..5135f85e 100644
--- a/requestresponse_test.go
+++ b/requestresponse_test.go
@@ -111,6 +111,32 @@ func TestPendingRequestContextDone(t *testing.T) {
 	require.False(t, engine.deliverResponse(pending.ID(), &livekit.RequestResponse{RequestId: pending.ID()}))
 }
 
+func TestWithDefaultTimeout(t *testing.T) {
+	ctx, cancel := withDefaultTimeout(context.Background(), time.Minute)
+	defer cancel()
+	deadline, ok := ctx.Deadline()
+	require.True(t, ok)
+	require.WithinDuration(t, time.Now().Add(time.Minute), deadline, 5*time.Second)
+
+	// a caller's own deadline is left alone, even when it is longer than the default
+	parent, cancelParent := context.WithTimeout(context.Background(), time.Hour)
+	defer cancelParent()
+	parentDeadline, _ := parent.Deadline()
+	ctx, cancel = withDefaultTimeout(parent, time.Minute)
+	defer cancel()
+	deadline, ok = ctx.Deadline()
+	require.True(t, ok)
+	require.Equal(t, parentDeadline, deadline)
+
+	engine := newTestEngine(t)
+	pending := engine.newPendingRequest()
+	ctx, cancel = withDefaultTimeout(context.Background(), 20*time.Millisecond)
+	defer cancel()
+	_, err := pending.Await(ctx)
+	require.ErrorIs(t, err, context.DeadlineExceeded)
+	require.Zero(t, engine.pendingRequestCount())
+}
+
 func TestPendingRequestAbortedOnClose(t *testing.T) {
 	engine := newTestEngine(t)
 

From 73089252cdd0d22fcb4bbc15a12a9067a1ebc2a1 Mon Sep 17 00:00:00 2001
From: Jacob Gelman <3182119+ladvoc@users.noreply.github.com>
Date: Wed, 2 Sep 2026 15:42:03 -0700
Subject: [PATCH 7/9] Abort pending requests on full reconnect

---
 engine.go | 2 ++
 1 file changed, 2 insertions(+)

diff --git a/engine.go b/engine.go
index 85c45a0e..8f395268 100644
--- a/engine.go
+++ b/engine.go
@@ -1051,6 +1051,8 @@ func (e *RTCEngine) cleanupConnection() {
 }
 
 func (e *RTCEngine) restartConnection() error {
+	// the new session will never answer requests made on the old one
+	e.abortPendingRequests()
 	if err := e.join(nil, nil); err != nil {
 		return err
 	}

From 5837d59c6811baeafe3972f9d50f3f9d97d429fc Mon Sep 17 00:00:00 2001
From: Jacob Gelman <3182119+ladvoc@users.noreply.github.com>
Date: Fri, 4 Sep 2026 15:50:17 -0700
Subject: [PATCH 8/9] Move non-struct methods to the end

---
 requestresponse.go | 48 +++++++++++++++++++++++-----------------------
 1 file changed, 24 insertions(+), 24 deletions(-)

diff --git a/requestresponse.go b/requestresponse.go
index b90dba00..04b7007e 100644
--- a/requestresponse.go
+++ b/requestresponse.go
@@ -104,30 +104,6 @@ func (p *pendingRequest) Await(ctx context.Context) (proto.Message, error) {
 	}
 }
 
-// awaitResponse is Await for requests whose success is reported by a dedicated message type.
-func awaitResponse[T proto.Message](ctx context.Context, p *pendingRequest) (T, error) {
-	var zero T
-
-	res, err := p.Await(ctx)
-	if err != nil {
-		return zero, err
-	}
-	typed, ok := res.(T)
-	if !ok {
-		return zero, fmt.Errorf("unexpected response %T to signal request %d", res, p.id)
-	}
-	return typed, nil
-}
-
-// withDefaultTimeout applies deadline d to ctx when it has none. A deadline the caller set is
-// never shortened. The returned cancel must be called.
-func withDefaultTimeout(ctx context.Context, d time.Duration) (context.Context, context.CancelFunc) {
-	if _, ok := ctx.Deadline(); ok {
-		return ctx, func() {}
-	}
-	return context.WithTimeout(ctx, d)
-}
-
 func (e *RTCEngine) removePendingRequest(requestID uint32) {
 	e.pendingRequestsLock.Lock()
 	delete(e.pendingRequests, requestID)
@@ -164,3 +140,27 @@ func (e *RTCEngine) abortPendingRequests() {
 		close(ch)
 	}
 }
+
+// awaitResponse is Await for requests whose success is reported by a dedicated message type.
+func awaitResponse[T proto.Message](ctx context.Context, p *pendingRequest) (T, error) {
+	var zero T
+
+	res, err := p.Await(ctx)
+	if err != nil {
+		return zero, err
+	}
+	typed, ok := res.(T)
+	if !ok {
+		return zero, fmt.Errorf("unexpected response %T to signal request %d", res, p.id)
+	}
+	return typed, nil
+}
+
+// withDefaultTimeout applies deadline d to ctx when it has none. A deadline the caller set is
+// never shortened. The returned cancel must be called.
+func withDefaultTimeout(ctx context.Context, d time.Duration) (context.Context, context.CancelFunc) {
+	if _, ok := ctx.Deadline(); ok {
+		return ctx, func() {}
+	}
+	return context.WithTimeout(ctx, d)
+}

From 816142d3d9728cee2437691a20a10e3f3e0ee808 Mon Sep 17 00:00:00 2001
From: Jacob Gelman <3182119+ladvoc@users.noreply.github.com>
Date: Fri, 4 Sep 2026 15:55:28 -0700
Subject: [PATCH 9/9] Keep engine methods together

---
 engine.go          | 70 ++++++++++++++++++++++++++++++++++++++++++++
 requestresponse.go | 73 ++--------------------------------------------
 2 files changed, 72 insertions(+), 71 deletions(-)

diff --git a/engine.go b/engine.go
index 8f395268..c013d411 100644
--- a/engine.go
+++ b/engine.go
@@ -763,6 +763,76 @@ func (e *RTCEngine) UnregisterTrackPublishedListener(cid string) {
 	e.trackPublishedListenersLock.Unlock()
 }
 
+func (e *RTCEngine) newPendingRequest() *pendingRequest {
+	p := &pendingRequest{
+		engine: e,
+		ch:     make(chan proto.Message, 1),
+	}
+
+	e.pendingRequestsLock.Lock()
+	p.id = e.allocateRequestIDLocked()
+	if e.closed.Load() {
+		// nothing will answer, fail Await right away
+		close(p.ch)
+	} else {
+		e.pendingRequests[p.id] = p.ch
+	}
+	e.pendingRequestsLock.Unlock()
+
+	return p
+}
+
+// allocateRequestIDLocked returns the next request id not in use by a pending request, skipping
+// zero. Wrapping around is harmless: an id is reused only once its previous request is done.
+func (e *RTCEngine) allocateRequestIDLocked() uint32 {
+	for {
+		e.nextRequestID++
+		if e.nextRequestID == 0 {
+			continue
+		}
+		if _, inUse := e.pendingRequests[e.nextRequestID]; !inUse {
+			return e.nextRequestID
+		}
+	}
+}
+
+func (e *RTCEngine) removePendingRequest(requestID uint32) {
+	e.pendingRequestsLock.Lock()
+	delete(e.pendingRequests, requestID)
+	e.pendingRequestsLock.Unlock()
+}
+
+// deliverResponse hands res to the request waiting for requestID, if any.
+func (e *RTCEngine) deliverResponse(requestID uint32, res proto.Message) bool {
+	if requestID == 0 {
+		return false
+	}
+
+	e.pendingRequestsLock.Lock()
+	ch, ok := e.pendingRequests[requestID]
+	delete(e.pendingRequests, requestID)
+	e.pendingRequestsLock.Unlock()
+	if !ok {
+		return false
+	}
+
+	// buffered, and removed from the map above so this is the only send
+	ch <- res
+	return true
+}
+
+// abortPendingRequests fails every request still waiting for a response.
+func (e *RTCEngine) abortPendingRequests() {
+	e.pendingRequestsLock.Lock()
+	pending := e.pendingRequests
+	e.pendingRequests = make(map[uint32]chan proto.Message)
+	e.pendingRequestsLock.Unlock()
+
+	for _, ch := range pending {
+		close(ch)
+	}
+}
+
 func (e *RTCEngine) handleDataPacket(msg webrtc.DataChannelMessage) {
 	packet, err := e.readDataPacket(msg)
 	if err != nil {
diff --git a/requestresponse.go b/requestresponse.go
index 04b7007e..e1e29983 100644
--- a/requestresponse.go
+++ b/requestresponse.go
@@ -35,46 +35,14 @@ func (e *SignalRequestError) Error() string {
 }
 
 // pendingRequest reserves the response to a signal request that carries a request id: create it
-// before sending the request, put ID() in the request_id field, then Await the response.
+// before sending the request, put ID() in the request_id field, then Await the response. The
+// engine owns the registry these are tracked in, see (*RTCEngine).newPendingRequest.
 type pendingRequest struct {
 	engine *RTCEngine
 	id     uint32
 	ch     chan proto.Message
 }
 
-func (e *RTCEngine) newPendingRequest() *pendingRequest {
-	p := &pendingRequest{
-		engine: e,
-		ch:     make(chan proto.Message, 1),
-	}
-
-	e.pendingRequestsLock.Lock()
-	p.id = e.allocateRequestIDLocked()
-	if e.closed.Load() {
-		// nothing will answer, fail Await right away
-		close(p.ch)
-	} else {
-		e.pendingRequests[p.id] = p.ch
-	}
-	e.pendingRequestsLock.Unlock()
-
-	return p
-}
-
-// allocateRequestIDLocked returns the next request id not in use by a pending request, skipping
-// zero. Wrapping around is harmless: an id is reused only once its previous request is done.
-func (e *RTCEngine) allocateRequestIDLocked() uint32 {
-	for {
-		e.nextRequestID++
-		if e.nextRequestID == 0 {
-			continue
-		}
-		if _, inUse := e.pendingRequests[e.nextRequestID]; !inUse {
-			return e.nextRequestID
-		}
-	}
-}
-
 func (p *pendingRequest) ID() uint32 {
 	return p.id
 }
@@ -104,43 +72,6 @@ func (p *pendingRequest) Await(ctx context.Context) (proto.Message, error) {
 	}
 }
 
-func (e *RTCEngine) removePendingRequest(requestID uint32) {
-	e.pendingRequestsLock.Lock()
-	delete(e.pendingRequests, requestID)
-	e.pendingRequestsLock.Unlock()
-}
-
-// deliverResponse hands res to the request waiting for requestID, if any.
-func (e *RTCEngine) deliverResponse(requestID uint32, res proto.Message) bool {
-	if requestID == 0 {
-		return false
-	}
-
-	e.pendingRequestsLock.Lock()
-	ch, ok := e.pendingRequests[requestID]
-	delete(e.pendingRequests, requestID)
-	e.pendingRequestsLock.Unlock()
-	if !ok {
-		return false
-	}
-
-	// buffered, and removed from the map above so this is the only send
-	ch <- res
-	return true
-}
-
-// abortPendingRequests fails every request still waiting for a response.
-func (e *RTCEngine) abortPendingRequests() {
-	e.pendingRequestsLock.Lock()
-	pending := e.pendingRequests
-	e.pendingRequests = make(map[uint32]chan proto.Message)
-	e.pendingRequestsLock.Unlock()
-
-	for _, ch := range pending {
-		close(ch)
-	}
-}
-
 // awaitResponse is Await for requests whose success is reported by a dedicated message type.
 func awaitResponse[T proto.Message](ctx context.Context, p *pendingRequest) (T, error) {
 	var zero T