Memory queue: cancel in-progress writes on queue closed, not producer closed

CHANGELOG.next.asciidoc

@@ -95,6 +95,7 @@ fields added to events containing the Beats version. {pull}37553[37553]
 - Update github.com/lestrrat-go/jwx dependency. {pull}37799[37799]
 - [threatintel] MISP pagination fixes {pull}37898[37898]
 - Fix file handle leak when handling errors in filestream {pull}37973[37973]
+- Fix a race condition that could crash Filebeat with a "negative WaitGroup counter" error {pull}38094[38094]
 
 *Heartbeat*
 

libbeat/publisher/queue/memqueue/produce.go

@@ -36,9 +36,10 @@ type ackProducer struct {
 }
 
 type openState struct {
- log *logp.Logger
- done chan struct{}
- events chan pushRequest
+ log *logp.Logger
+ done chan struct{}
+ queueDone <-chan struct{}
+ events chan pushRequest
 }
 
 // producerID stores the order of events within a single producer, so multiple
@@ -58,9 +59,10 @@ type ackHandler func(count int)
 
 func newProducer(b *broker, cb ackHandler, dropCB func(interface{}), dropOnCancel bool) queue.Producer {
  openState := openState{
- log: b.logger,
- done: make(chan struct{}),
- events: b.pushChan,
+ log: b.logger,
+ done: make(chan struct{}),
+ queueDone: b.ctx.Done(),
+ events: b.pushChan,
  }
 
  if cb != nil {
@@ -143,27 +145,40 @@ func (st *openState) Close() {
 func (st *openState) publish(req pushRequest) (queue.EntryID, bool) {
  select {
  case st.events <- req:
- // If the output is blocked and the queue is full, `req` is written
- // to `st.events`, however the queue never writes back to `req.resp`,
- // which effectively blocks for ever. So we also need to select on the
- // done channel to ensure we don't miss the shutdown signal.
+ // The events channel is buffered, which means we may successfully
+ // write to it even if the queue is shutting down. To avoid blocking
+ // forever during shutdown, we also have to wait on the queue's
+ // shutdown channel.
  select {
  case resp := <-req.resp:
  return resp, true
- case <-st.done:
+ case <-st.queueDone:
  st.events = nil
  return 0, false
  }
  case <-st.done:
  st.events = nil
  return 0, false
+ case <-st.queueDone:
+ st.events = nil
+ return 0, false
  }
 }
 
 func (st *openState) tryPublish(req pushRequest) (queue.EntryID, bool) {
  select {
  case st.events <- req:
- return <-req.resp, true
+ // The events channel is buffered, which means we may successfully
+ // write to it even if the queue is shutting down. To avoid blocking
+ // forever during shutdown, we also have to wait on the queue's
+ // shutdown channel.
+ select {
+ case resp := <-req.resp:
+ return resp, true
+ case <-st.queueDone:
+ st.events = nil
+ return 0, false
+ }
  case <-st.done:
  st.events = nil
  return 0, false

libbeat/publisher/queue/memqueue/queue_test.go

@@ -77,17 +77,17 @@ func TestProduceConsumer(t *testing.T) {
  t.Run("flush", testWith(makeTestQueue(bufferSize, batchSize/2, 100*time.Millisecond)))
 }
 
-// TestProducerDoesNotBlockWhenCancelled ensures the producer Publish
-// does not block indefinitely.
+// TestProducerDoesNotBlockWhenQueueClosed ensures the producer Publish
+// does not block indefinitely during queue shutdown.
 //
-// Once we get a producer `p` from the queue we want to ensure
+// Once we get a producer `p` from the queue `q` we want to ensure
 // that if p.Publish is called and blocks it will unblock once
-// p.Cancel is called.
+// `q.Close` is called.
 //
 // For this test we start a queue with size 2 and try to add more
-// than 2 events to it, p.Publish will block, once we call p.Cancel,
+// than 2 events to it, p.Publish will block, once we call q.Close,
 // we ensure the 3rd event was not successfully published.
-func TestProducerDoesNotBlockWhenCancelled(t *testing.T) {
+func TestProducerDoesNotBlockWhenQueueClosed(t *testing.T) {
  q := NewQueue(nil, nil,
  Settings{
  Events: 2, // Queue size
@@ -138,8 +138,8 @@ func TestProducerDoesNotBlockWhenCancelled(t *testing.T) {
  time.Millisecond,
  "the first two events were not successfully published")
 
- // Cancel the producer, this should unblock its Publish method
- p.Cancel()
+ // Close the queue, this should unblock the pending Publish call
+ q.Close()
 
  require.Eventually(
  t,