diff --git a/pkg/rsqueue/impls/database/queue.go b/pkg/rsqueue/impls/database/queue.go index 23b1de2..e21220e 100644 --- a/pkg/rsqueue/impls/database/queue.go +++ b/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 diff --git a/pkg/rsqueue/impls/database/queue_test.go b/pkg/rsqueue/impls/database/queue_test.go index 6853e76..24bb4ce 100644 --- a/pkg/rsqueue/impls/database/queue_test.go +++ b/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)