fix(rsqueue): buffer PollAddress errCh to prevent goroutine leak on abandoned poll

pkg/rsqueue/impls/database/queue.go

@@ -246,7 +246,13 @@ func (q *DatabaseQueue) IsAddressInQueue(ctx context.Context, address string) (b
 }
 
 func (q *DatabaseQueue) PollAddress(ctx context.Context, address string) <-chan error {
-	errCh := make(chan error)
+	// Buffered by 1 so the polling goroutine's single error send below can never
+	// block, even if the caller has stopped receiving (e.g. its request context
+	// was canceled and it returned). An unbuffered channel here leaked one
+	// goroutine per abandoned poll, parked forever on the send
+	// (rstudio/package-manager#19008). The goroutine sends at most one value
+	// before closing, so a single slot is sufficient.
+	errCh := make(chan error, 1)
 
 	go func() {
 		var done, ticked bool

pkg/rsqueue/impls/database/queue_test.go

@@ -478,6 +478,48 @@ func (s *QueueSuite) TestPollErr(c *check.C) {
 	c.Assert(s.store.polled, check.Equals, 1)
 }
 
+// TestPollErrAbandonedReceiver reproduces the goroutine leak from
+// rstudio/package-manager#19008: when the caller stops receiving from the
+// channel returned by PollAddress (e.g. its request context is canceled), the
+// poll goroutine must still be able to complete and exit rather than parking
+// forever on a send that no one will receive. Here we call PollAddress and
+// never read from errCh; leaktest.Check then asserts the poll goroutine has
+// exited. With an unbuffered errCh the goroutine blocks on `errCh <- err` and
+// this fails; with a buffered errCh it sends, closes, and returns.
+func (s *QueueSuite) TestPollErrAbandonedReceiver(c *check.C) {
+	// NOTE: this uses the correct leaktest form — `defer leaktest.Check(c)()`
+	// (Check returns a checker func that must be invoked at return). The other
+	// tests in this file call `defer leaktest.Check(c)` without the trailing
+	// (), which discards the checker and never actually asserts; fixing those
+	// is out of scope here.
+	defer leaktest.Check(c)()
+
+	s.store.pollErr = errors.New("horrible error")
+	q := &DatabaseQueue{
+		store:               s.store,
+		addressPollInterval: time.Millisecond * 5,
+		subscribe:           make(chan broadcaster.Subscription),
+		unsubscribe:         make(chan (<-chan listener.Notification)),
+		wrapper:             &fakeWrapper{},
+	}
+
+	queueMsgs := make(chan listener.Notification)
+	workMsgs := make(chan listener.Notification)
+	chunkMsgs := make(chan listener.Notification)
+	defer close(queueMsgs)
+	defer close(workMsgs)
+	defer close(chunkMsgs)
+
+	stopper := make(chan bool)
+	defer func() { stopper <- true }()
+	go q.broadcast(stopper, queueMsgs, workMsgs, chunkMsgs)
+
+	// Start polling but never receive from the returned channel, simulating a
+	// caller that gave up (context canceled). The poll goroutine hits the error
+	// path and must not leak.
+	_ = q.PollAddress(context.Background(), "something")
+}
+
 func (s *QueueSuite) TestPollLockErr(c *check.C) {
 	defer leaktest.Check(c)