Bound pending cancel notification queue

Change-Id: I1e6195bae6969b5e07b532c5f042244530eb9a9e
Signed-off-by: Thomas Kosiewski <tk@coder.com>

Author: ThomasK33

connection.go

@@ -114,6 +114,7 @@ func (c *Connection) loggerOrDefault() *slog.Logger {
 const (
 	maxCanonicalJSONRPCIDKeyLen   = 4096
 	maxCanonicalJSONRPCIDAbsExp10 = 4096
+	maxPendingCancelRequests      = 1024
 )
 
 var (
@@ -666,10 +667,20 @@ func (c *Connection) sendCancelRequest(idKey string) {
 	default:
 	}
 
+	queueFull := false
 	c.mu.Lock()
-	c.pendingCancelRequest = append(c.pendingCancelRequest, idKey)
+	if len(c.pendingCancelRequest) >= maxPendingCancelRequests {
+		queueFull = true
+	} else {
+		c.pendingCancelRequest = append(c.pendingCancelRequest, idKey)
+	}
 	c.mu.Unlock()
 
+	if queueFull {
+		c.loggerOrDefault().Debug("dropping $/cancel_request due to full queue", "queue_len", maxPendingCancelRequests)
+		return
+	}
+
 	select {
 	case c.cancelRequestSignal <- struct{}{}:
 	default:

connection_cancel_test.go

@@ -586,6 +586,39 @@ func TestConnectionOutboundCancelRequest_DoesNotBlockWhenPeerStopsReading(t *tes
 	}
 }
 
+func TestConnectionSendCancelRequest_BoundsPendingQueue(t *testing.T) {
+	baseCtx, baseCancel := context.WithCancelCause(context.Background())
+	defer baseCancel(nil)
+
+	c := &Connection{
+		pending:             make(map[string]*pendingResponse),
+		inflight:            make(map[string]context.CancelCauseFunc),
+		cancelRequestSignal: make(chan struct{}, 1),
+		ctx:                 baseCtx,
+		cancel:              baseCancel,
+	}
+
+	for i := 0; i < maxPendingCancelRequests+128; i++ {
+		c.sendCancelRequest(fmt.Sprintf("%d", i))
+	}
+
+	c.mu.Lock()
+	defer c.mu.Unlock()
+
+	if len(c.pendingCancelRequest) != maxPendingCancelRequests {
+		t.Fatalf("expected pending cancel queue length %d, got %d", maxPendingCancelRequests, len(c.pendingCancelRequest))
+	}
+
+	if got := c.pendingCancelRequest[0]; got != "0" {
+		t.Fatalf("expected queue to retain earliest id when full, got first id %q", got)
+	}
+
+	expectedLast := fmt.Sprintf("%d", maxPendingCancelRequests-1)
+	if got := c.pendingCancelRequest[len(c.pendingCancelRequest)-1]; got != expectedLast {
+		t.Fatalf("expected queue to drop ids beyond capacity, got last id %q want %q", got, expectedLast)
+	}
+}
+
 func TestConnectionWaitForResponse_PeerDisconnectWinsOverDerivedContextCancel(t *testing.T) {
 	const iterations = 64