[exporterhelper] boundedMemoryQueue may send on closed channel

[atingchen]

Describe the bug
When the gorutine prepares to submit new item to the queue, q.stopped.Load() returns true. The gorutine has not finished sending items to the channel.

func (q *boundedMemoryQueue) Produce(item Request) bool {
	if q.stopped.Load() {
		return false
	}

	if q.size.Load() >= q.capacity {
		return false
	}

	q.size.Add(1)
	select {
	case q.items <- item:
		return true
	default:
		// should not happen, as overflows should have been captured earlier
		q.size.Add(^uint32(0))
		return false
	}
}

At the same time, service shutdown and invokes stop() to close the channel. The gorutine may send item to the closed channel.

func (q *boundedMemoryQueue) Stop() {
	q.stopped.Store(true) // disable producer
	close(q.items)
	q.stopWG.Wait()
}

What did you expect to see?
thread-safe

What did you see instead?

What version did you use?
v0.73.0

[yutingcaicyt]

The collector will stop in topological order so that upstream components are stopped before downstream components. This ensures that each component has a chance to drain to its consumer before the consumer is stopped. Therefore, under normal circumstances, before the exporter is closed, the receiver and processor have guaranteed to complete the data transmission and close. When the exporter is closing, no data will enter the exporter.

[atingchen]

The collector will stop in topological order so that upstream components are stopped before downstream components. This ensures that each component has a chance to drain to its consumer before the consumer is stopped. Therefore, under normal circumstances, before the exporter is closed, the receiver and processor have guaranteed to complete the data transmission and close. When the exporter is closing, no data will enter the exporter.

Thanks for your reply @yutingcaicyt. boundedMemoryQueue will be used as a retry queue. When the upstream component is closed, data may still re-enter the queue due to the retry strategy.

When queuedRetrySender calls shutdown, it will first close retryStopCh, and then close the retry queue.
OnTemporaryFailure is called between these two steps, which has a small chance of putting the data back into the retry queue.

		select {
		case <-req.Context().Done():
			return fmt.Errorf("Request is cancelled or timed out %w", err)
		case <-rs.stopCh:
			return rs.onTemporaryFailure(rs.logger, req, fmt.Errorf("interrupted due to shutdown %w", err))
		case <-time.After(backoffDelay):
		}

[dmitryax]

The re-enqueuing is only applicable to the persistent queue, which doesn't have a closed channel. Memory queue doesn't get data re-enqueued

[atoulme]

I believe this issue to be fixed as we have changed the implementation of queues quite a bit, as @dmitryax points above. The code is completely different now. Previously, a test was flaky because of this problem, but it no longer is.

Closing as resolved. Feel free to comment or create new issues to follow up.