Keep a Concurrent Go Pipeline from Getting Stuck

Table of Contents

1. Challenge

   Step 1: Meet the stuck telemetry pipeline

   The telemetry pipeline keeps freezing

   Hundreds of delivery drones stream battery, GPS, and altitude readings into a Go ingestion service. During the last fleet-wide firmware rollout, ops watched the service hang mid-shift: goroutines piled up with nothing coordinating them, workers blocked forever on unbuffered channels the moment one reading looked malformed, a shared stats map got corrupted under concurrent writes, and a single unresponsive drone connection was enough to stall the entire ingest.

   The repository you've inherited compiles and looks reasonable at a glance, but every one of those failure modes is baked into the current code: a sync.WaitGroup that never tracks any workers, bidirectional channels used carelessly, an unprotected map[string]int touched from multiple goroutines, and no way to cancel or time out a stuck reading. You will fix this stage by stage, in the same order ops hit the failures. ### The four stages you'll rebuild

   1. worker.go—launch a fixed pool of worker goroutines and use a sync.WaitGroup so the pipeline reports completion only when every worker has truly finished.
   2. channels.go—send readings through a send-only channel, add a buffered error channel, and report errors without blocking so a failing worker never has to wait for someone to listen.
   3. dispatcher.go—close the producer-owned input channel once dispatch is done, then drain results and errors with select and range so nothing blocks forever.
   4. pipeline.go—guard the shared stats map with a sync.Mutex and confirm the fix with Go's race detector. Cancel the run on the first error, add a per-reading deadline, and wire the deadline into worker.go so a hung drone releases its worker instead of freezing the fleet.

   Each step builds directly on the previous one: by Step 6, you'll run the whole pipeline and see every stage work together.

2. Challenge

   Step 2: Build the foundational worker pool

   Coordinating a worker pool

   worker.go defines the Reading and Result types, a worker function that reads from a jobs channel and writes to a results channel, and two functions you'll complete: StartWorkers, which launches a fixed-size worker pool, and CloseWhenDone, which handles shutdown after the workers finish. StartWorkers also accepts a context.Context that you'll pass to each worker. You'll use this context in Step 5 to cancel a run and release blocked workers.

   Currently, StartWorkers is a stub that launches no goroutines, so the pipeline has no way to track when the worker pool has finished. As a result, callers may return too early and miss results or block indefinitely while waiting for completion.

   You can use sync.WaitGroup to coordinate the workers. Call Add once per worker before launching its goroutine, have each worker call Done when it finishes, and use Wait to block until every worker has completed. ### Shutting the pool down without a race

   Starting the pool correctly is only half the problem—something still has to wait for it to finish and close the shared results channel exactly once, at exactly the right moment. Close it too early and a worker panics trying to send on a closed channel; forget to close it and any range results loop downstream blocks forever. The WaitGroup you just wired up is exactly the signal that tells you it's safe. ### Step summary

   • sync.WaitGroup tracks the pool: Add before each go worker(...), Done when a worker exits, Wait to block until all are done.
   • StartWorkers launches n workers and returns the WaitGroup.
   • CloseWhenDone waits for the pool and only then closes results, so no worker ever sends on a closed channel and downstream readers know when to stop.
3. Challenge

   Step 3: Route readings through directional, non-blocking channels

   Making channel direction explicit

   channels.go is where readings enter the pipeline. Using a plain chan Reading everywhere means any function that accesses the channel could accidentally receive when it should send, or vice versa. The compiler cannot catch this mistake. Declaring a parameter as chan<- Reading (send-only) or <-chan Reading (receive-only) turns these mistakes into compile-time errors instead of runtime deadlocks.

   SendReadings acts as the producer and should only send readings. ### Giving errors somewhere to land

   An unbuffered chan error forces the sender to wait for a receiver that might be busy elsewhere—exactly the deadlock ops saw during the firmware rollout, where one bad reading froze a worker mid-flight. A buffered channel, make(chan error, size), holds up to size pending errors so a send can complete immediately even if nothing has read the previous ones yet. ### Sending without ever blocking

   A buffer only helps until it fills up. select with a send case and a default case gives you a send that never blocks: if the channel can accept the value right now, it does; otherwise the default branch runs instead and the worker moves on. That's the shape you need for a worker reporting a malformed reading—it should never stall the rest of the pool just because the error channel happens to be full. ### Step summary

   • chan<- Reading and <-chan Reading make a channel's direction a compile-time guarantee.
   • SendReadings only sends, in order, and leaves closing to the dispatcher.
   • NewErrorChannel returns a buffered channel, so a worker can report a failure without waiting for a receiver.
   • ReportError uses select with a default case, so a full buffer drops the error instead of stalling the worker.
4. Challenge

   Step 4: Close producer channels and drain results cleanly

   Closing the channel you own

   dispatcher.go owns the producer side of the pipeline. A range loop over a channel only ever stops when that channel is closed—until then it blocks waiting for one more value, forever if nothing more is coming. That means whichever function sends the last reading must also be the one that closes the channel, right after the last send, and only once. ### Draining results with range

   Once jobs closes, workers finish and CloseWhenDone closes results behind them. for r := range results is the natural way to consume a channel until it closes: no fixed count to guess, no risk of missing a late value, and the loop exits exactly when the channel does—not before, not after. ### Merging two channels with select

   Results and errors arrive on two separate channels, and neither one is guaranteed to close before the other. A select with one receive case per channel lets you handle whichever has a value ready without picking favorites, and checking the second return value from a receive (v, ok := <-ch) tells you when a channel has closed so you can stop selecting on it and eventually exit the loop once both are done. ### Step summary

   • RunDispatch owns the send side of jobs: it sends every reading, then closes the channel exactly once.
   • CollectResults uses for r := range results to drain until results is closed.
   • Aggregate uses one select over results and errs, checks the ok value to detect a closed channel, and sets that channel to nil, so it finishes only when both are done and neither channel can block the other.
5. Challenge

   Step 5: Guard shared stats and cancel stuck workers

   Protecting shared state with a mutex

   pipeline.go contains a Stats struct with a map[string]int that counts readings per device. Multiple worker goroutines update this map concurrently. Concurrent map writes in Go don't just risk incorrect counts—they can cause a runtime failure. A sync.Mutex prevents concurrent access by using Lock before accessing the map and Unlock when finished (ideally with defer, which runs when the function returns), ensuring only one goroutine accesses the map at a time. ## Reading safely and proving it with -race

   Locking on write isn't enough if reads go unguarded—a caller inspecting stats while a worker writes is just as much a race as two writers colliding.

   Snapshot needs the same lock, plus a copy of the map rather than the live one, so a caller can look at the data without holding the lock or seeing writes that happen after the snapshot was taken.

   Go's race detector is how you confirm this holds under real concurrent load, not just on paper. The -race flag builds the test binary with instrumentation to detect data races. For example, the following command runs the tests for Tasks 5.1 and 5.2 with the race detector enabled:

   go test . -race -run 'TestTask5_1|TestTask5_2'
   

   If a goroutine accesses the map without holding the lock while another goroutine performs a conflicting access, the output can contain a WARNING: DATA RACE report identifying the conflicting lines, and the test fails. A successful run ends with an ok line and no reported data races. The lab's check for Task 5.2 runs a stress test with many concurrent Increment and Snapshot callers under -race, so an unprotected map access can cause the check to fail. ### Canceling the whole run on failure

   Right now an error in one stage doesn't stop the others—goroutines keep running work that no longer matters. context.WithCancel gives you a child context plus a cancel function; pass the context into StartWorkers so every worker can notice it, and call cancel() the moment an error arrives on errs. The collection loop selects on ctx.Done() too, so after cancel() it returns instead of draining every remaining result.

   A canceled worker returns even when it's halfway through a send nobody will receive, and buffering jobs to hold every reading means the dispatcher can always finish sending and close the channel. Without both, returning early just leaves those goroutines parked forever. The check for this task covers it from both sides: a run with one malformed reading followed by a thousand valid ones must stop long before it collects them all (so removing cancel() fails), and no goroutine may still be running once RunWithContext returns. ### Timing out a single silent drone

   Cancellation handles a global stop signal; a per-reading deadline handles the narrower case of a read that never completes, such as a drone that has gone quiet. context.WithTimeout derives a context that cancels itself after a duration; select between the channel receive and ctx.Done() means whichever happens first wins, so a hung read returns ctx.Err() instead of blocking that worker—and the fleet—indefinitely.

   A deadline only helps if the workers actually use it. Once ReadWithDeadline works, you'll call it from worker in worker.go with the package's readTimeout, so a worker waiting on a jobs channel that never sends or closes reports a deadline error and exits instead of waiting forever. A closed channel is a normal end of work, not a timeout, so ReadWithDeadline returns io.EOF for it and the worker can tell the two apart. ### Step summary

   • Stats guards its map with a sync.Mutex: Increment and Snapshot lock first and unlock with defer, and Snapshot returns a copy.
   • go test . -race proves it: a stress test with concurrent writers and readers passes with no DATA RACE report.
   • RunWithContext derives a cancelable context, passes it to the workers, and calls cancel() as soon as an error arrives, so blocked workers are released.
   • ReadWithDeadline bounds each read with context.WithTimeout, and workers call it, so a silent channel releases its worker with a deadline error instead of blocking it forever.
6. Challenge

   Step 6: Run the application

   Every task is complete, and you can check each one using its Validate button. What you haven't done yet is run the entire application and see how the pieces you implemented separately work together as a single pipeline.

   Start the application

   In the Terminal, run:

   go run .
   

   Expected result: The terminal displays two lines. The first shows the total number of processed readings and errors, and the second shows the reading count for each drone.

   processed 3 readings, 0 errors
   stats: map[drone-1:1 drone-2:1 drone-3:1]
   

   The output reflects the functionality you implemented in the previous steps. Compare it with the expected behavior from each task to see how the complete pipeline works.

   The readings reached the workers because RunDispatch called SendReadings to put them on the send-only jobs channel, demonstrating Task 3.1, and then closed jobs, demonstrating Task 4.1. Once jobs is empty and closed, each worker's call to ReadWithDeadline returns io.EOF. The worker then returns and calls Done, allowing the WaitGroup to finish.

   The first line, processed 3 readings, 0 errors, comes from the PipelineResult that RunWithContext returns for the three readings in main.go. StartWorkers launched four workers, calling wg.Add(1) before each go worker(...) so the WaitGroup count was correct from the start, demonstrating Task 2.1. CloseWhenDone then waited on the WaitGroup and closed the results channel after the last worker finished. This allows the collection loop to stop instead of waiting for the five-second timeout in main.go, demonstrating Task 2.2.

   The collection loop in RunWithContext merges the results and errs channels using select and sets each channel variable to nil once its channel closes. This follows the same drain-until-closed pattern you implemented in Tasks 4.2 and 4.3. The loop finishes after both channels close, ensuring all three readings are counted.

   The 0 errors output means the error path remained quiet, as expected. Every reading has a DeviceID, so no worker called ReportError on the buffered channel created by NewErrorChannel, demonstrating Tasks 3.2 and 3.3. Because no error triggered early cancellation, the pipeline completed normally. The deferred cancel() still runs when RunWithContext returns, releasing the context's resources. This demonstrates the behavior implemented in Task 5.3. No worker reached its deadline either: ReadWithDeadline detected the closed jobs channel and returned io.EOF instead of a timeout error, demonstrating Task 5.4.

   The second line, stats: map[drone-1:1 drone-2:1 drone-3:1], shows one reading counted for each drone. RunWithContext called Increment once per result, and Increment locked mu before updating the map, demonstrating Task 5.1. Snapshot then acquired the same mutex to create an independent copy of the map, so the statistics returned to main.go cannot be changed by subsequent updates to the original map, demonstrating Task 5.2. Go formats maps with keys in sorted order, which is why the drones appear in this order.

   Across the lab, you built these controls:

   • Track every worker with a WaitGroup and close the results channel after the last worker finishes (Tasks 2.1 and 2.2).
   • Send readings through a send-only channel, buffer the error channel, and report errors without blocking (Tasks 3.1 to 3.3).
   • Close jobs from the producer and drain results and errors until both channels close (Tasks 4.1 to 4.3).
   • Guard the shared stats map with a sync.Mutex and return an independent copy from Snapshot (Tasks 5.1 and 5.2).
   • Cancel the run when an error occurs and apply a deadline to each read (Tasks 5.3 and 5.4). ### Run the checks yourself

   The lab ran each task's test when you clicked Validate, but they are ordinary tests and you can run them whenever you like. In the Terminal, run the whole suite:

   go test . -count=1 -timeout 60s
   

   Expected result: the output ends with a line that starts with ok.

   Or run one task's test at a time, here Task 2.1:

   bash runTest.sh TestTask2_1
   

   Each failure names the task it belongs to, so the output tells you which step to go back to rather than which line to stare at. One thing worth trying before you finish: intentionally introduce an error and observe how the corresponding check fails, then restore the original code. For example, remove the wg.Wait() line from CloseWhenDone, run bash runTest.sh TestTask2_2, and review the failure message to identify which task failed. Understanding what a failed test looks like is just as valuable as knowing what a passing test looks like.

   When you're done, the code you wrote remains available in the editor for you to review.