What causes "pipeline queue is draining" in owasp-amass/amass?

Stop producer goroutines (signal via context cancellation or done channel) and wait for them before calling Drain

pipeline queue is draining

Error message

pipeline queue is draining

What it means

PipelineQueue.Append returns this error once the queue has entered draining mode (after Drain was called). During drain, the queue is being flushed for session shutdown and refuses all new EventDataElement appends. The library throws it to protect the draining invariant — events appended during drain would be lost or race with the flush.

Source

Thrown at engine/types/registry.go:64

	draining bool
	drainCh  chan struct{}
	q        queue.Queue
}

func NewPipelineQueue() *PipelineQueue {
	return &PipelineQueue{
		q:       queue.NewQueue(),
		drainCh: make(chan struct{}, 1),
	}
}

func (pq *PipelineQueue) Len() int {
	return pq.q.Len()
}

func (pq *PipelineQueue) Append(data *EventDataElement) error {
	if pq.draining {
		return errors.New("pipeline queue is draining")
	}
	pq.q.Append(data)
	return nil
}

func (pq *PipelineQueue) Drain() {
	if pq.draining {
		return
	}
	pq.draining = true
	close(pq.drainCh)
}

// Next implements the pipeline InputSource interface.
func (pq *PipelineQueue) Next(ctx context.Context) bool {
	if pq.q.Len() > 0 {
		return true
	}

View on GitHub (pinned to 79299dce87)

Solutions

Stop producer goroutines (signal via context cancellation or done channel) and wait for them before calling Drain
Handle the error from Append by dropping or re-routing the event when draining is expected
Buffer events during drain and flush them after the queue finishes, if the events must not be lost
Check a queue draining flag (e.g. expose/inspect state) before appending in hot producers

Example fix

// before
go producer(pq)
pq.Drain()
// after
done := make(chan struct{})
go func() { defer close(done); producer(pq) }()
<-done
pq.Drain()

Defensive patterns

Strategy: try-catch

Try / catch

if err := pq.Append(evt); err != nil {
    if err.Error() == "pipeline queue is draining" {
        // producers must stop: session is shutting down
        return
    }
    return err
}

Prevention

Cancel producer goroutines and wait for them (WaitGroup/context) before calling Drain
Do not call Append after Drain in shutdown code paths
Route late-arriving events to a fallback buffer if they must be preserved
Centralize Drain in one owner goroutine so appends cannot race it

When it happens

Trigger: Calling pq.Append after pq.Drain() was invoked on the same PipelineQueue, typically when a producer goroutine is still emitting events while the engine session is shutting down.

Common situations: Race between session teardown (Drain) and long-running pipelines still producing events; failing to wait for producer goroutines before draining; calling Drain twice and appending afterwards.

Understand the failure class

Background: "Invalid state transition" errors: "status must be X, actually Y", "already rejected/charging/uninstalled", "cannot ... while running" — what they mean when a library rejects your call — this error's family across 31 libraries.

AI-assisted analysis of owasp-amass/amass@79299dce87 (2026-09-06). Data as JSON: /api/errors/f02f715fc6477a43. Report an issue: GitHub.