package worker
import "context"
func (p *Pool) Submit(job Job) error {
// Reject fast if draining, otherwise enqueue with backpressure.
select {
case <-p.ctx.Done():
return ErrDraining
default:
}
select {
case p.jobs <- job:
return nil
case <-p.ctx.Done():
return ErrDraining
}
}
func (p *Pool) Shutdown(ctx context.Context) error {
p.cancel() // stop accepting new submissions
p.closeOnce.Do(func() {
close(p.jobs) // let ranging workers finish the buffer then exit
})
done := make(chan struct{})
go func() {
p.wg.Wait()
close(done)
}()
select {
case <-done:
return nil
case <-ctx.Done():
return ErrDrainTimeout
}
}
package main
import (
"context"
"errors"
"log"
"os"
"os/signal"
"syscall"
"time"
"example.com/app/worker"
)
func main() {
sigCtx, stop := signal.NotifyContext(context.Background(),
syscall.SIGINT, syscall.SIGTERM)
defer stop()
pool := worker.New(4, 128)
go produce(sigCtx, pool)
<-sigCtx.Done()
log.Println("shutdown signal received, draining in-flight jobs...")
drainCtx, cancel := context.WithTimeout(context.Background(), 25*time.Second)
defer cancel()
if err := pool.Shutdown(drainCtx); err != nil {
if errors.Is(err, worker.ErrDrainTimeout) {
log.Println("grace period exceeded, forcing exit")
os.Exit(1)
}
log.Printf("shutdown error: %v", err)
}
log.Println("all jobs drained, exiting")
}
func produce(ctx context.Context, pool *worker.Pool) {
for i := 0; ; i++ {
select {
case <-ctx.Done():
return
default:
}
n := i
if err := pool.Submit(func(context.Context) {
time.Sleep(200 * time.Millisecond)
log.Printf("processed job %d", n)
}); err != nil {
log.Printf("submit refused: %v", err)
return
}
}
}
package worker
import (
"context"
"errors"
"log"
"sync"
)
var (
ErrDraining = errors.New("pool is draining, new jobs rejected")
ErrDrainTimeout = errors.New("timed out waiting for in-flight jobs")
)
type Job func(ctx context.Context)
type Pool struct {
jobs chan Job
wg sync.WaitGroup
ctx context.Context
cancel context.CancelFunc
closeOnce sync.Once
}
func New(workers, buffer int) *Pool {
ctx, cancel := context.WithCancel(context.Background())
p := &Pool{
jobs: make(chan Job, buffer),
ctx: ctx,
cancel: cancel,
}
for i := 0; i < workers; i++ {
p.wg.Add(1)
go p.worker(i)
}
return p
}
func (p *Pool) worker(id int) {
defer p.wg.Done()
// Range keeps pulling buffered jobs until the channel is closed and empty,
// even after the context has been cancelled.
for job := range p.jobs {
job(context.Background())
}
log.Printf("worker %d exited cleanly", id)
}
This snippet shows how a Go worker pool drains in-flight jobs before the process exits, so that a SIGTERM from an orchestrator like Kubernetes doesn't kill work that is halfway done. The core idea is to separate two concerns: stopping the intake of new work, and waiting for already-started work to finish. Confusing these two is the most common cause of dropped jobs during a deploy.
In pool.go, the Pool owns a buffered jobs channel, a sync.WaitGroup that tracks running workers, and a context.Context that is cancelled on shutdown. Each worker in worker loops on two things: a case reading from jobs and a case on p.ctx.Done(). Crucially, when the context is cancelled the worker does not bail immediately — it keeps pulling from jobs until the channel is drained and closed, using the classic for job := range fallthrough after cancellation. This guarantees that anything already buffered still runs.
Submit respects backpressure and refuses new work once draining begins: it selects on p.ctx.Done() and returns ErrDraining so callers get a clear signal instead of blocking forever on a channel that will never be read. Shutdown performs the ordered dance — it cancels the context to reject new submissions, closes the jobs channel so ranging workers terminate once empty, then wg.Wait()s bounded by the caller's ctx. If that outer context expires first, Shutdown returns ErrDrainTimeout, which lets the caller decide between waiting longer or forcing exit.
main.go wires this to real signals with signal.NotifyContext, giving a SIGINT/SIGTERM-aware context. After the signal fires it builds a separate timeout context for the drain, so the grace period is explicit and independent of when the signal arrived. This mirrors how terminationGracePeriodSeconds works in practice.
The main trade-off is that a slow job can consume the entire grace window; the timeout path exists precisely for that. A subtle pitfall avoided here is closing jobs from a producer while workers still range over it — the code closes exactly once from Shutdown and never sends afterward, which is the only safe ordering. This pattern is what a developer reaches for whenever a task must complete atomically or not at all under deploy churn.
Related snips
package api
import (
"net/http"
"runtime/debug"
)Expose build metadata for debugging deploys
export interface RetryOptions {
retries: number;
baseMs: number;
maxMs: number;
signal?: AbortSignal;
onRetry?: (attempt: number, delay: number, err: unknown) => void;Exponential backoff with jitter for retries
class Product(models.Model):
name = models.CharField(max_length=200)
slug = models.SlugField(blank=True)
price = models.DecimalField(max_digits=10, decimal_places=2)
cost = models.DecimalField(max_digits=10, decimal_places=2)
margin = models.DecimalField(max_digits=5, decimal_places=2, blank=True)Django model signals vs overriding save
package dbutil
import (
"context"
"github.com/jackc/pgconn"Retry Postgres serialization failures with bounded attempts
require "csv"
class PeopleCsvStream
include Enumerable
HEADERS = %w[id full_name email signed_up_at plan].freezeResilient CSV Export as a Streamed Response
use crossbeam::channel::unbounded;
use std::thread;
fn main() {
let (tx, rx) = unbounded();Crossbeam for advanced concurrent data structures
Share this code
Here's the card — post it anywhere.
