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

@@ -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,