{"record":{"id":"f02f715fc6477a43","repo":"owasp-amass/amass","slug":"pipeline-queue-is-draining","errorCode":null,"errorMessage":"pipeline queue is draining","messagePattern":"pipeline queue is draining","errorType":"error_code","errorClass":null,"httpStatus":null,"severity":"warning","filePath":"engine/types/registry.go","lineNumber":64,"sourceCode":"\tdraining bool\n\tdrainCh  chan struct{}\n\tq        queue.Queue\n}\n\nfunc NewPipelineQueue() *PipelineQueue {\n\treturn &PipelineQueue{\n\t\tq:       queue.NewQueue(),\n\t\tdrainCh: make(chan struct{}, 1),\n\t}\n}\n\nfunc (pq *PipelineQueue) Len() int {\n\treturn pq.q.Len()\n}\n\nfunc (pq *PipelineQueue) Append(data *EventDataElement) error {\n\tif pq.draining {\n\t\treturn errors.New(\"pipeline queue is draining\")\n\t}\n\tpq.q.Append(data)\n\treturn nil\n}\n\nfunc (pq *PipelineQueue) Drain() {\n\tif pq.draining {\n\t\treturn\n\t}\n\tpq.draining = true\n\tclose(pq.drainCh)\n}\n\n// Next implements the pipeline InputSource interface.\nfunc (pq *PipelineQueue) Next(ctx context.Context) bool {\n\tif pq.q.Len() > 0 {\n\t\treturn true\n\t}","sourceCodeStart":46,"sourceCodeEnd":82,"githubUrl":"https://github.com/owasp-amass/amass/blob/79299dce87b0085db0f2f4ef3e9c52cccb49f514/engine/types/registry.go#L46-L82","documentation":"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.","triggerScenarios":"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.","commonSituations":"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.","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"],"exampleFix":"// before\ngo producer(pq)\npq.Drain()\n// after\ndone := make(chan struct{})\ngo func() { defer close(done); producer(pq) }()\n<-done\npq.Drain()","handlingStrategy":"try-catch","validationCode":null,"typeGuard":null,"tryCatchPattern":"if err := pq.Append(evt); err != nil {\n    if err.Error() == \"pipeline queue is draining\" {\n        // producers must stop: session is shutting down\n        return\n    }\n    return err\n}","preventionTips":["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"],"tags":["concurrency","lifecycle","shutdown"],"backgroundTag":"invalid-state-transition","analyzedSha":"79299dce87b0085db0f2f4ef3e9c52cccb49f514","analyzedAt":"2026-09-06T08:22:48.198Z","contentChangedAt":"2026-09-06T08:22:48.198Z","schemaVersion":2},"datasetVersion":"2026-09-14T00:17:10.932Z"}