[rusq]

Problem

When conversation processing returned on the first fatal result or callback error, its producers and workers could remain blocked:

• channel and thread workers could stay blocked sending results;
• the input loop could stay blocked routing channel or thread requests;
• the thread channel could race with its close;
• the API channel generator could remain blocked while runWorkers waited for it;
• the user collector and processor shutdown could remain pending behind that chain.

A SIGQUIT stack trace from a production hang showed this exact pattern: runWorkers waiting for the generator, the generator blocked emitting channels, and conversation workers blocked on result sends after the consumer had returned.

In addition, after the fix, the original error that led to Hang-up was discovered —

18:19:16 ERROR 006 (Application Error): error in subroutine 
                     │   conversations on stage worker: error streaming 
                     │   conversations: Channel channel XXXXX: encode: 
                     │   insertchunk: insert: insertchunk: payload: insert all 
                     │   CHANNEL_USER (ID=XXXXX): constraint failed: UNIQUE 
                     │   constraint failed: CHANNEL_USER.CHANNEL_ID, 
                     │   CHANNEL_USER.USER_ID, CHANNEL_USER.CHUNK_ID (1555).

to address it a channel users slices deduplication was introduced during streaming.

Solution

• Give Conversations an internal cancellable context and retain the first fatal result or callback error.
• Cancel immediately on failure, then drain results and join every internal worker before returning.
• Make result, channel-request, and thread-request sends cancellation-aware.
• Close threadsC only after both the input loop and channel worker have stopped.
• Preserve nonfatal channel_not_found and not_in_channel handling.
• Coordinate runWorkers with a shared generator/conversation context and concurrently drain generator errors, removing the fixed error-channel capacity.
• Cancel and join the ConversationsCB item producer deterministically.
• Add regression tests for queued fatal work, callback failures, external cancellation, nonfatal errors, blocked sends, producer joining, and generator shutdown.
• Deduplicate channel user IDs and keep related conversation constants local to their use.

Validation

• go test -race ./stream ./internal/chunk/control
• Repeated focused race tests
• go test ./... -run '^$'
• git diff --check

[rusq]

Thanks go to @volker-fr for the report and helping with solving the issue.

pr720-stack.txt

[Copilot]

Pull request overview

This PR hardens the Slack conversation streaming pipeline and the chunk control runner against shutdown deadlocks by making producer/worker sends cancellation-aware, ensuring all internal goroutines are joined/drained before returning, and adding regression coverage around cancellation and generator coordination.

Changes:

• Refactors the conversations pipeline to cancel on first fatal error/callback failure while still draining results and joining all producers/workers before returning.
• Makes result, channel-request, and thread-request sends cancellation-aware; adjusts thread channel close ownership to avoid close/send races.
• Coordinates runWorkers with a cancelable generator context and concurrent generator-error draining; deduplicates channel user IDs during streaming; adds targeted regression tests.

Reviewed changes

Copilot reviewed 7 out of 7 changed files in this pull request and generated 2 comments.

Show a summary per file

File	Description
stream/stream.go	Moves pipeline buffer-size constants out of the main stream file.
stream/stream_workers.go	Makes result sends cancellation-aware via sendResult to prevent worker/producers blocking after early consumer return.
stream/stream_test.go	Updates processLink tests for new context-aware signature and adds a cancellation unblock regression case.
stream/conversation.go	Refactors pipeline orchestration to cancel/drain/join deterministically; adds cancellation-aware request routing and channel-user dedupe helper.
stream/conversation_test.go	Adds regression tests covering fatal cancellation, callback failure shutdown, external cancellation, nonfatal handling, blocked send unblocking, and user dedupe.
internal/chunk/control/runner.go	Updates conversation worker to cancel/join the API generator deterministically and drain generator errors concurrently.
internal/chunk/control/runner_test.go	Adds regression test ensuring generator is canceled/unblocked when conversations fail.

---

💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.