附录：随书实验完整代码

package main

import (
    "context"
    "errors"
    "fmt"
    "log/slog"
    "net"
    "net/http"
    _ "net/http/pprof"
    "os"
    "os/signal"
    "strconv"
    "syscall"
    "time"

    "example.com/go-service-engineering/lab/internal/service"
)

func env(key, fallback string) string {
    if value := os.Getenv(key); value != "" {
        return value
    }
    return fallback
}

func main() {
    logger := slog.New(slog.NewJSONHandler(os.Stdout, nil))
    ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
    defer stop()
    if err := run(ctx, logger); err != nil {
        logger.Error("service stopped", "error", err)
        os.Exit(1)
    }
}

func run(ctx context.Context, logger *slog.Logger) error {
    cfg := service.DefaultConfig()
    cfg.Logger = logger
    var err error
    if cfg.Workers, err = strconv.Atoi(env("WORKERS", "2")); err != nil {
        return fmt.Errorf("WORKERS: %w", err)
    }
    if cfg.QueueCapacity, err = strconv.Atoi(env("QUEUE_CAPACITY", "8")); err != nil {
        return fmt.Errorf("QUEUE_CAPACITY: %w", err)
    }
    if cfg.TaskTimeout, err = time.ParseDuration(env("TASK_TIMEOUT", "5s")); err != nil {
        return fmt.Errorf("TASK_TIMEOUT: %w", err)
    }
    if cfg.StreamWriteTimeout, err = time.ParseDuration(env("STREAM_WRITE_TIMEOUT", "250ms")); err != nil {
        return fmt.Errorf("STREAM_WRITE_TIMEOUT: %w", err)
    }
    grace, err := time.ParseDuration(env("SHUTDOWN_TIMEOUT", "5s"))
    if err != nil || grace <= 0 {
        return errors.New("SHUTDOWN_TIMEOUT must be a positive duration")
    }
    app, err := service.New(cfg)
    if err != nil {
        return err
    }
    defer app.Fetcher.Close()
    defer func() {
        cleanup, cancel := context.WithCancel(context.Background())
        cancel()
        _ = app.Pool.Shutdown(cleanup)
    }()
    listener, err := net.Listen("tcp", env("LISTEN_ADDR", "127.0.0.1:8080"))
    if err != nil {
        return err
    }
    defer listener.Close()
    adminListener, err := net.Listen("tcp", env("ADMIN_ADDR", "127.0.0.1:6060"))
    if err != nil {
        return err
    }
    defer adminListener.Close()
    server := &http.Server{Handler: app.Handler(), ReadHeaderTimeout: 2 * time.Second, ReadTimeout: 5 * time.Second, IdleTimeout: 30 * time.Second, MaxHeaderBytes: 16 << 10}
    admin := &http.Server{Handler: http.DefaultServeMux, ReadHeaderTimeout: 2 * time.Second, IdleTimeout: 30 * time.Second}
    defer server.Close()
    defer admin.Close()
    serveErrors := make(chan error, 2)
    go func() { serveErrors <- server.Serve(listener) }()
    go func() { serveErrors <- admin.Serve(adminListener) }()
    logger.Info("listening", "address", listener.Addr().String(), "admin", adminListener.Addr().String())
    select {
    case <-ctx.Done():
    case err := <-serveErrors:
        if !errors.Is(err, http.ErrServerClosed) {
            return err
        }
    }
    logger.Info("shutdown started")
    shutdown, cancel := context.WithTimeout(context.Background(), grace)
    defer cancel()
    // Shutdown first stops admission at the listener and waits for request
    // handlers. Only then close task admission and drain accepted background work.
    httpErr := server.Shutdown(shutdown)
    if httpErr != nil {
        _ = server.Close()
    }
    _ = admin.Close()
    poolErr := app.Pool.Shutdown(shutdown)
    // Closing sockets cancels request contexts, but Server.Close does not wait
    // for their handlers. Explicitly wait for cooperative cleanup as well.
    cleanup, cleanupCancel := context.WithTimeout(context.Background(), time.Second)
    defer cleanupCancel()
    requestErr := app.DrainRequests(cleanup)
    select {
    case <-app.Pool.Done():
    case <-cleanup.Done():
        return errors.New("workers ignored cancellation for one second")
    }
    logger.Info("shutdown complete", "stats", app.Pool.Stats(), "http_error", httpErr, "pool_error", poolErr, "request_error", requestErr)
    return errors.Join(httpErr, poolErr, requestErr)
}

package main

import (
    "bufio"
    "bytes"
    "encoding/json"
    "fmt"
    "io"
    "net/http"
    "os"
    "os/exec"
    "strings"
    "sync"
    "syscall"
    "testing"
    "time"
)

// The helper is a real child process running the production main and signal
// handler. It also runs under the race runtime when invoked by go test -race.
func TestProcessHelper(t *testing.T) {
    if os.Getenv("GO_SERVICE_PROCESS_HELPER") != "1" {
        return
    }
    main()
    os.Exit(0)
}

type process struct {
    cmd    *exec.Cmd
    events chan map[string]any
    done   chan error
    mu     sync.Mutex
    logs   bytes.Buffer
}

func startProcess(t *testing.T, grace string) (*process, string) {
    t.Helper()
    cmd := exec.Command(os.Args[0], "-test.run=^TestProcessHelper$")
    cmd.Env = append(os.Environ(), "GO_SERVICE_PROCESS_HELPER=1", "LISTEN_ADDR=127.0.0.1:0", "ADMIN_ADDR=127.0.0.1:0", "WORKERS=1", "QUEUE_CAPACITY=2", "TASK_TIMEOUT=10s", "SHUTDOWN_TIMEOUT="+grace)
    stdout, err := cmd.StdoutPipe()
    if err != nil {
        t.Fatal(err)
    }
    cmd.Stderr = cmd.Stdout
    p := &process{cmd: cmd, events: make(chan map[string]any, 128), done: make(chan error, 1)}
    if err := cmd.Start(); err != nil {
        t.Fatal(err)
    }
    go func() {
        scanner := bufio.NewScanner(stdout)
        for scanner.Scan() {
            p.mu.Lock()
            _, _ = fmt.Fprintln(&p.logs, scanner.Text())
            p.mu.Unlock()
            var event map[string]any
            if json.Unmarshal(scanner.Bytes(), &event) == nil {
                p.events <- event
            }
        }
        p.done <- cmd.Wait()
    }()
    t.Cleanup(func() { _ = cmd.Process.Kill() })
    event := p.waitEvent(t, "listening")
    return p, "http://" + event["address"].(string)
}

func (p *process) output() string { p.mu.Lock(); defer p.mu.Unlock(); return p.logs.String() }

func (p *process) waitEvent(t *testing.T, message string) map[string]any {
    t.Helper()
    timer := time.NewTimer(5 * time.Second)
    defer timer.Stop()
    for {
        select {
        case event := <-p.events:
            if event["msg"] == message {
                return event
            }
        case <-timer.C:
            t.Fatalf("missing event %q; logs:\n%s", message, p.output())
            return nil
        }
    }
}

func postTask(t *testing.T, url string) {
    t.Helper()
    client := &http.Client{Timeout: 2 * time.Second}
    resp, err := client.Post(url, "", nil)
    if err != nil {
        t.Fatal(err)
    }
    defer resp.Body.Close()
    _, _ = io.Copy(io.Discard, resp.Body)
    if resp.StatusCode != http.StatusAccepted {
        t.Fatalf("task status: %d", resp.StatusCode)
    }
}

func TestSIGTERMDrainsRequestAndAcceptedTasks(t *testing.T) {
    if testing.Short() {
        t.Skip("real process integration test")
    }
    p, url := startProcess(t, "3s")
    postTask(t, url+"/tasks?delay=350ms")
    p.waitEvent(t, "task started")
    postTask(t, url+"/tasks?delay=150ms") // queued, also must drain
    requestDone := make(chan error, 1)
    go func() {
        client := &http.Client{Timeout: 3 * time.Second}
        resp, err := client.Get(url + "/work?delay=150ms")
        if err == nil {
            _, err = io.Copy(io.Discard, resp.Body)
            _ = resp.Body.Close()
            if resp.StatusCode != http.StatusOK {
                err = fmt.Errorf("request status %d", resp.StatusCode)
            }
        }
        requestDone <- err
    }()
    p.waitEvent(t, "request started")
    if err := p.cmd.Process.Signal(syscall.SIGTERM); err != nil {
        t.Fatal(err)
    }
    p.waitEvent(t, "shutdown started")
    select {
    case err := <-requestDone:
        if err != nil {
            t.Fatalf("in-flight request lost: %v", err)
        }
    case <-time.After(4 * time.Second):
        t.Fatal("request did not finish")
    }
    event := p.waitEvent(t, "shutdown complete")
    stats := event["stats"].(map[string]any)
    if stats["accepted"] != float64(2) || stats["completed"] != float64(2) || stats["failed"] != float64(0) || stats["active"] != float64(0) {
        t.Fatalf("task boundary violated: %v", stats)
    }
    select {
    case err := <-p.done:
        if err != nil {
            t.Fatalf("process: %v\n%s", err, p.output())
        }
    case <-time.After(4 * time.Second):
        t.Fatal("process did not exit")
    }
    // Listener must be gone after shutdown, not merely return a health error.
    client := &http.Client{Timeout: 200 * time.Millisecond}
    if resp, err := client.Get(url + "/healthz"); err == nil {
        resp.Body.Close()
        t.Fatal("listener still accepts after exit")
    }
    logs := p.output()
    if strings.Index(logs, "request finished") > strings.Index(logs, "shutdown complete") {
        t.Fatal("shutdown returned before handler completion")
    }
}

func TestSIGTERMDeadlineCancelsBlockedTask(t *testing.T) {
    if testing.Short() {
        t.Skip("real process integration test")
    }
    p, url := startProcess(t, "80ms")
    postTask(t, url+"/tasks?delay=5s")
    p.waitEvent(t, "task started")
    requestDone := make(chan error, 1)
    go func() {
        client := &http.Client{Timeout: 3 * time.Second}
        resp, err := client.Get(url + "/work?delay=5s")
        if err == nil {
            _, err = io.Copy(io.Discard, resp.Body)
            _ = resp.Body.Close()
        }
        requestDone <- err
    }()
    p.waitEvent(t, "request started")
    if err := p.cmd.Process.Signal(syscall.SIGTERM); err != nil {
        t.Fatal(err)
    }
    event := p.waitEvent(t, "shutdown complete")
    stats := event["stats"].(map[string]any)
    if stats["cancelled"] != float64(1) || stats["completed"] != float64(1) || stats["active"] != float64(0) {
        t.Fatalf("cancelled task did not release: %v", stats)
    }
    select {
    case err := <-requestDone:
        if err == nil {
            t.Fatal("deadline-expired request unexpectedly succeeded")
        }
    case <-time.After(3 * time.Second):
        t.Fatal("closed request did not return")
    }
    logs := p.output()
    if !strings.Contains(logs, `"msg":"request finished","path":"/work","error":"context canceled"`) {
        t.Fatalf("request cleanup was not observed before shutdown completion:\n%s", logs)
    }
    select {
    case err := <-p.done:
        if err == nil {
            t.Fatal("deadline must produce nonzero exit status")
        }
    case <-time.After(4 * time.Second):
        t.Fatal("process did not exit after cancellation")
    }
}

package service

import (
    "context"
    "encoding/json"
    "errors"
    "fmt"
    "log/slog"
    "net/http"
    "net/url"
    "strconv"
    "sync"
    "time"
)

type Config struct {
    Workers            int
    QueueCapacity      int
    TaskTimeout        time.Duration
    StreamWriteTimeout time.Duration
    Logger             *slog.Logger
}

func DefaultConfig() Config {
    return Config{Workers: 2, QueueCapacity: 8, TaskTimeout: 5 * time.Second, StreamWriteTimeout: 250 * time.Millisecond}
}

type App struct {
    Pool           *Pool
    Fetcher        *Fetcher
    cfg            Config
    log            *slog.Logger
    requestMu      sync.Mutex
    requestsClosed bool
    requests       sync.WaitGroup
}

func New(cfg Config) (*App, error) {
    if cfg.StreamWriteTimeout <= 0 {
        return nil, fmt.Errorf("stream write timeout must be positive")
    }
    if cfg.Logger == nil {
        cfg.Logger = slog.Default()
    }
    pool, err := NewPool(cfg.Workers, cfg.QueueCapacity, cfg.TaskTimeout, cfg.Logger)
    if err != nil {
        return nil, err
    }
    return &App{Pool: pool, Fetcher: NewFetcher(2), cfg: cfg, log: cfg.Logger}, nil
}

func (a *App) Handler() http.Handler {
    mux := http.NewServeMux()
    mux.HandleFunc("POST /tasks", a.tasks)
    mux.HandleFunc("GET /work", a.work)
    mux.HandleFunc("GET /stream", a.stream)
    mux.HandleFunc("GET /stats", func(w http.ResponseWriter, r *http.Request) { writeJSON(w, http.StatusOK, a.Pool.Stats()) })
    mux.HandleFunc("GET /healthz", func(w http.ResponseWriter, r *http.Request) { w.WriteHeader(http.StatusOK) })
    return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
        a.requestMu.Lock()
        if a.requestsClosed {
            a.requestMu.Unlock()
            http.Error(w, "requests draining", http.StatusServiceUnavailable)
            return
        }
        a.requests.Add(1)
        a.requestMu.Unlock()
        defer a.requests.Done()
        mux.ServeHTTP(w, r)
    })
}

// DrainRequests gates new handlers and waits for existing handlers to return.
// Call it after stopping the listener. The gate prevents Add racing with Wait
// when the previous request count is zero. It does not itself cancel requests.
func (a *App) DrainRequests(ctx context.Context) error {
    a.requestMu.Lock()
    a.requestsClosed = true
    a.requestMu.Unlock()
    done := make(chan struct{})
    go func() { a.requests.Wait(); close(done) }()
    select {
    case <-done:
        return nil
    case <-ctx.Done():
        return ctx.Err()
    }
}

func duration(r *http.Request) (time.Duration, error) {
    s := r.URL.Query().Get("delay")
    if s == "" {
        return 100 * time.Millisecond, nil
    }
    d, err := time.ParseDuration(s)
    if err != nil || d < 0 || d > 30*time.Second {
        return 0, errors.New("delay must be between 0 and 30s")
    }
    return d, nil
}

func wait(ctx context.Context, d time.Duration) error {
    timer := time.NewTimer(d)
    defer timer.Stop()
    select {
    case <-timer.C:
        return nil
    case <-ctx.Done():
        return ctx.Err()
    }
}

func (a *App) tasks(w http.ResponseWriter, r *http.Request) {
    d, err := duration(r)
    if err != nil {
        http.Error(w, err.Error(), http.StatusBadRequest)
        return
    }
    target := r.URL.Query().Get("url")
    if target != "" {
        u, parseErr := url.Parse(target)
        if parseErr != nil || (u.Scheme != "http" && u.Scheme != "https") || u.Host == "" {
            http.Error(w, "url must be an absolute http(s) URL", http.StatusBadRequest)
            return
        }
    }
    // Capture values only. Do not retain r, ResponseWriter, or r.Context().
    id, err := a.Pool.Submit(r.Context(), func(ctx context.Context) error {
        if target != "" {
            _, err := a.Fetcher.Fetch(ctx, target)
            return err
        }
        return wait(ctx, d)
    })
    if err != nil {
        status := http.StatusServiceUnavailable
        if errors.Is(err, context.Canceled) {
            status = http.StatusRequestTimeout
        }
        w.Header().Set("Retry-After", "1")
        http.Error(w, err.Error(), status)
        return
    }
    writeJSON(w, http.StatusAccepted, map[string]uint64{"id": id})
}

func (a *App) work(w http.ResponseWriter, r *http.Request) {
    d, err := duration(r)
    if err != nil {
        http.Error(w, err.Error(), http.StatusBadRequest)
        return
    }
    a.log.Info("request started", "path", "/work")
    err = wait(r.Context(), d)
    a.log.Info("request finished", "path", "/work", "error", err)
    if err != nil {
        return
    }
    writeJSON(w, http.StatusOK, map[string]string{"status": "done"})
}

func (a *App) stream(w http.ResponseWriter, r *http.Request) {
    n := int64(64 << 20)
    if raw := r.URL.Query().Get("bytes"); raw != "" {
        value, err := strconv.ParseInt(raw, 10, 64)
        if err != nil || value < 1 || value > 256<<20 {
            http.Error(w, "bytes must be 1..268435456", http.StatusBadRequest)
            return
        }
        n = value
    }
    controller := http.NewResponseController(w)
    // One absolute deadline bounds the entire stream, including blocked writes.
    if err := controller.SetWriteDeadline(time.Now().Add(a.cfg.StreamWriteTimeout)); err != nil {
        http.Error(w, "write deadlines unavailable", http.StatusInternalServerError)
        return
    }
    // Keep the deadline active through net/http's final buffered flush. This
    // HTTP/1.1 demo closes streaming connections rather than reusing one with a
    // stale deadline. Other endpoints retain ordinary keep-alive behavior.
    w.Header().Set("Connection", "close")
    w.Header().Set("Content-Type", "application/octet-stream")
    chunk := make([]byte, 32<<10)
    var sent int64
    for sent < n {
        if err := r.Context().Err(); err != nil {
            a.log.Info("stream finished", "bytes", sent, "error", err)
            return
        }
        size := min(int64(len(chunk)), n-sent)
        written, err := w.Write(chunk[:size])
        sent += int64(written)
        if err != nil {
            a.log.Info("stream finished", "bytes", sent, "error", err)
            return
        }
    }
    a.log.Info("stream finished", "bytes", sent)
}

func writeJSON(w http.ResponseWriter, status int, value any) {
    w.Header().Set("Content-Type", "application/json")
    w.WriteHeader(status)
    _ = json.NewEncoder(w).Encode(value)
}

package service

import (
    "context"
    "errors"
    "fmt"
    "io"
    "log/slog"
    "net"
    "net/http"
    "net/http/httptest"
    "strings"
    "sync"
    "testing"
    "time"
)

func app(t *testing.T, cfg Config) *App {
    t.Helper()
    a, err := New(cfg)
    if err != nil {
        t.Fatal(err)
    }
    t.Cleanup(func() {
        ctx, cancel := context.WithTimeout(context.Background(), time.Second)
        defer cancel()
        _ = a.Pool.Shutdown(ctx)
        receive(t, a.Pool.Done())
        a.Fetcher.Close()
    })
    return a
}

func eventually(t *testing.T, condition func() bool) {
    t.Helper()
    deadline := time.NewTimer(3 * time.Second)
    defer deadline.Stop()
    tick := time.NewTicker(time.Millisecond)
    defer tick.Stop()
    for !condition() {
        select {
        case <-tick.C:
        case <-deadline.C:
            t.Fatal("condition did not become true")
        }
    }
}

func TestHTTPOverloadAndTaskOwnership(t *testing.T) {
    cfg := DefaultConfig()
    cfg.Workers, cfg.QueueCapacity, cfg.Logger = 1, 1, quiet()
    a := app(t, cfg)
    srv := httptest.NewServer(a.Handler())
    defer srv.Close()
    ctx, cancel := context.WithCancel(context.Background())
    req, _ := http.NewRequestWithContext(ctx, http.MethodPost, srv.URL+"/tasks?delay=150ms", nil)
    resp, err := srv.Client().Do(req)
    if err != nil {
        t.Fatal(err)
    }
    _, _ = io.Copy(io.Discard, resp.Body)
    _ = resp.Body.Close()
    if resp.StatusCode != http.StatusAccepted {
        t.Fatalf("first task: %d", resp.StatusCode)
    }
    cancel() // Accepted task is owned by the pool, not this now-ended request.
    eventually(t, func() bool { return a.Pool.Stats().Active == 1 })
    for _, want := range []int{http.StatusAccepted, http.StatusServiceUnavailable} {
        resp, err := srv.Client().Post(srv.URL+"/tasks?delay=150ms", "", nil)
        if err != nil {
            t.Fatal(err)
        }
        _, _ = io.Copy(io.Discard, resp.Body)
        _ = resp.Body.Close()
        if resp.StatusCode != want {
            t.Fatalf("got %d, want %d", resp.StatusCode, want)
        }
        if want == http.StatusServiceUnavailable && resp.Header.Get("Retry-After") != "1" {
            t.Fatal("missing retry hint")
        }
    }
    a.Pool.Close()
    receive(t, a.Pool.Done())
    if s := a.Pool.Stats(); s.Completed != 2 || s.Failed != 0 {
        t.Fatalf("request cancellation leaked into task: %+v", s)
    }
}

type lockedBuffer struct {
    mu   sync.Mutex
    text strings.Builder
}

func (b *lockedBuffer) Write(p []byte) (int, error) {
    b.mu.Lock()
    defer b.mu.Unlock()
    return b.text.Write(p)
}
func (b *lockedBuffer) String() string { b.mu.Lock(); defer b.mu.Unlock(); return b.text.String() }

func TestSlowStreamDeadlineReleasesHandler(t *testing.T) {
    var logs lockedBuffer
    cfg := DefaultConfig()
    cfg.StreamWriteTimeout = 80 * time.Millisecond
    cfg.Logger = slog.New(slog.NewTextHandler(&logs, nil))
    a := app(t, cfg)
    finished := make(chan struct{})
    srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
        a.Handler().ServeHTTP(w, r)
        close(finished)
    }))
    defer srv.Close()
    conn, err := net.DialTimeout("tcp", srv.Listener.Addr().String(), time.Second)
    if err != nil {
        t.Fatal(err)
    }
    defer conn.Close()
    if tcp, ok := conn.(*net.TCPConn); ok {
        _ = tcp.SetReadBuffer(1024)
    }
    // A real peer sends its request and never reads the 256 MiB response.
    if _, err := fmt.Fprint(conn, "GET /stream?bytes=268435456 HTTP/1.1\r\nHost: localhost\r\n\r\n"); err != nil {
        t.Fatal(err)
    }
    receive(t, finished)
    if text := logs.String(); !strings.Contains(text, "stream finished") || !strings.Contains(text, "i/o timeout") {
        t.Fatalf("stream did not hit its write deadline: %s", text)
    }
}

func TestHTTPInvalidInput(t *testing.T) {
    cfg := DefaultConfig()
    cfg.Logger = quiet()
    a := app(t, cfg)
    for _, path := range []string{"/tasks?delay=-1s", "/tasks?url=file:///etc/passwd", "/stream?bytes=999999999"} {
        method := http.MethodPost
        if strings.HasPrefix(path, "/stream") {
            method = http.MethodGet
        }
        rr := httptest.NewRecorder()
        a.Handler().ServeHTTP(rr, httptest.NewRequest(method, path, nil))
        if rr.Code != http.StatusBadRequest {
            t.Errorf("%s: %d", path, rr.Code)
        }
    }
}

func TestDrainRequestsWaitsForHandlerAndGatesAdmission(t *testing.T) {
    var logs lockedBuffer
    cfg := DefaultConfig()
    cfg.Logger = slog.New(slog.NewTextHandler(&logs, nil))
    a := app(t, cfg)
    ctx, cancel := context.WithCancel(context.Background())
    defer cancel()
    finished := make(chan struct{})
    go func() {
        a.Handler().ServeHTTP(httptest.NewRecorder(), httptest.NewRequest(http.MethodGet, "/work?delay=5s", nil).WithContext(ctx))
        close(finished)
    }()
    eventually(t, func() bool { return strings.Contains(logs.String(), "request started") })
    short, stop := context.WithTimeout(context.Background(), 20*time.Millisecond)
    defer stop()
    if err := a.DrainRequests(short); !errors.Is(err, context.DeadlineExceeded) {
        t.Fatalf("drain returned while handler still owned resources: %v", err)
    }
    cancel()
    receive(t, finished)
    long, end := context.WithTimeout(context.Background(), time.Second)
    defer end()
    if err := a.DrainRequests(long); err != nil {
        t.Fatal(err)
    }
    rr := httptest.NewRecorder()
    a.Handler().ServeHTTP(rr, httptest.NewRequest(http.MethodGet, "/healthz", nil))
    if rr.Code != http.StatusServiceUnavailable {
        t.Fatalf("new handler admitted during drain: %d", rr.Code)
    }
}

package service

import (
    "context"
    "fmt"
    "io"
    "net"
    "net/http"
    "time"
)

// Fetcher reuses one transport. Every response body has exactly one owner,
// including error-status bodies; reading EOF permits HTTP/1.1 connection reuse.
type Fetcher struct {
    Client    *http.Client
    Transport *http.Transport
}

func NewFetcher(maxConns int) *Fetcher {
    tr := &http.Transport{
        Proxy:                 http.ProxyFromEnvironment,
        DialContext:           (&net.Dialer{Timeout: 2 * time.Second, KeepAlive: 30 * time.Second}).DialContext,
        MaxConnsPerHost:       maxConns,
        MaxIdleConns:          32,
        MaxIdleConnsPerHost:   maxConns,
        IdleConnTimeout:       30 * time.Second,
        TLSHandshakeTimeout:   2 * time.Second,
        ResponseHeaderTimeout: 2 * time.Second,
        ForceAttemptHTTP2:     true,
    }
    return &Fetcher{Client: &http.Client{Transport: tr}, Transport: tr}
}

// Fetch limits response bytes. An oversized response is closed before EOF,
// intentionally sacrificing reuse rather than consuming unbounded bandwidth.
func (f *Fetcher) Fetch(ctx context.Context, url string) (int64, error) {
    req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil)
    if err != nil {
        return 0, err
    }
    resp, err := f.Client.Do(req)
    if err != nil {
        return 0, err
    }
    defer resp.Body.Close()
    const maxBody = 1 << 20
    n, err := io.Copy(io.Discard, io.LimitReader(resp.Body, maxBody+1))
    if err != nil {
        return n, err
    }
    if n > maxBody {
        return n, fmt.Errorf("response exceeds %d bytes", maxBody)
    }
    if resp.StatusCode < 200 || resp.StatusCode > 299 {
        return n, fmt.Errorf("upstream status %d", resp.StatusCode)
    }
    return n, nil
}

func (f *Fetcher) Close() { f.Transport.CloseIdleConnections() }

package service

import (
    "context"
    "errors"
    "io"
    "net"
    "net/http"
    "net/http/httptest"
    "sync/atomic"
    "testing"
    "time"
)

func TestConnectionExhaustionAndRecovery(t *testing.T) {
    released := make(chan struct{})
    var okRequests atomic.Int64
    srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
        if r.URL.Path == "/hold" {
            w.WriteHeader(http.StatusOK)
            w.(http.Flusher).Flush()
            <-r.Context().Done()
            close(released)
            return
        }
        okRequests.Add(1)
        _, _ = io.WriteString(w, "recovered")
    }))
    defer srv.Close()
    f := NewFetcher(1)
    defer f.Close()
    resp, err := f.Client.Get(srv.URL + "/hold")
    if err != nil {
        t.Fatal(err)
    }
    defer resp.Body.Close()
    ctx, cancel := context.WithTimeout(context.Background(), 50*time.Millisecond)
    defer cancel()
    if _, err := f.Fetch(ctx, srv.URL+"/ok"); !errors.Is(err, context.DeadlineExceeded) {
        t.Fatalf("waiting for occupied connection: %v", err)
    }
    if got := okRequests.Load(); got != 0 {
        t.Fatalf("request bypassed connection bound: %d", got)
    }
    _ = resp.Body.Close()
    receive(t, released)
    recovery, stop := context.WithTimeout(context.Background(), time.Second)
    defer stop()
    if n, err := f.Fetch(recovery, srv.URL+"/ok"); err != nil || n != 9 {
        t.Fatalf("recovery: bytes=%d error=%v", n, err)
    }
}

func TestFetcherReusesConnectionIncludingErrorStatus(t *testing.T) {
    var connections atomic.Int64
    srv := httptest.NewUnstartedServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
        if r.URL.Path == "/error" {
            w.WriteHeader(http.StatusBadGateway)
        }
        _, _ = io.WriteString(w, "small body")
    }))
    srv.Config.ConnState = func(_ net.Conn, state http.ConnState) {
        if state == http.StateNew {
            connections.Add(1)
        }
    }
    srv.Start()
    defer srv.Close()
    f := NewFetcher(1)
    defer f.Close()
    ctx, cancel := context.WithTimeout(context.Background(), time.Second)
    defer cancel()
    if _, err := f.Fetch(ctx, srv.URL+"/error"); err == nil {
        t.Fatal("expected status error")
    }
    for range 3 {
        if _, err := f.Fetch(ctx, srv.URL); err != nil {
            t.Fatal(err)
        }
    }
    if n := connections.Load(); n != 1 {
        t.Fatalf("body was not drained/reused: connections=%d", n)
    }
}

func TestFetcherBlockedBodyCancellationReleasesConnection(t *testing.T) {
    released := make(chan struct{})
    srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
        if r.URL.Path == "/blocked" {
            w.WriteHeader(http.StatusOK)
            w.(http.Flusher).Flush()
            <-r.Context().Done()
            close(released)
            return
        }
        _, _ = io.WriteString(w, "ok")
    }))
    defer srv.Close()
    f := NewFetcher(1)
    defer f.Close()
    ctx, cancel := context.WithTimeout(context.Background(), 50*time.Millisecond)
    defer cancel()
    if _, err := f.Fetch(ctx, srv.URL+"/blocked"); !errors.Is(err, context.DeadlineExceeded) {
        t.Fatalf("blocked body: %v", err)
    }
    receive(t, released)
    recovery, stop := context.WithTimeout(context.Background(), time.Second)
    defer stop()
    if _, err := f.Fetch(recovery, srv.URL); err != nil {
        t.Fatalf("connection slot not recovered: %v", err)
    }
}

// Package service demonstrates bounded concurrency and explicit task ownership.
package service

import (
    "context"
    "errors"
    "fmt"
    "log/slog"
    "sync"
    "sync/atomic"
    "time"
)

var (
    ErrFull   = errors.New("task queue full")
    ErrClosed = errors.New("task pool closed")
)

type Work func(context.Context) error

type task struct {
    id   uint64
    work Work
}

type Stats struct {
    Accepted  uint64 `json:"accepted"`
    Started   uint64 `json:"started"`
    Completed uint64 `json:"completed"`
    Failed    uint64 `json:"failed"`
    Cancelled uint64 `json:"cancelled"`
    Active    int64  `json:"active"`
    Queued    int    `json:"queued"`
}

// Pool owns every accepted task. A caller's context controls admission only;
// after admission, only the pool lifetime and per-task deadline cancel the task.
type Pool struct {
    mu        sync.Mutex
    closed    bool
    queue     chan task
    ctx       context.Context
    cancel    context.CancelFunc
    done      chan struct{}
    timeout   time.Duration
    log       *slog.Logger
    next      atomic.Uint64
    accepted  atomic.Uint64
    started   atomic.Uint64
    completed atomic.Uint64
    failed    atomic.Uint64
    cancelled atomic.Uint64
    active    atomic.Int64
}

func NewPool(workers, capacity int, timeout time.Duration, logger *slog.Logger) (*Pool, error) {
    if workers < 1 || capacity < 1 || timeout <= 0 {
        return nil, fmt.Errorf("workers, capacity, and timeout must be positive")
    }
    if logger == nil {
        logger = slog.Default()
    }
    ctx, cancel := context.WithCancel(context.Background())
    p := &Pool{queue: make(chan task, capacity), ctx: ctx, cancel: cancel, done: make(chan struct{}), timeout: timeout, log: logger}
    var wg sync.WaitGroup
    wg.Add(workers)
    for range workers {
        go func() {
            defer wg.Done()
            for t := range p.queue {
                p.run(t)
            }
        }()
    }
    go func() { wg.Wait(); cancel(); close(p.done) }()
    return p, nil
}

func (p *Pool) run(t task) {
    ctx, cancel := context.WithTimeout(p.ctx, p.timeout)
    defer cancel()
    p.active.Add(1)
    p.started.Add(1)
    p.log.Info("task started", "id", t.id)
    var err error
    if err = ctx.Err(); err == nil {
        err = t.work(ctx)
    }
    if err != nil {
        p.failed.Add(1)
        if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) {
            p.cancelled.Add(1)
        }
    }
    p.completed.Add(1)
    p.active.Add(-1)
    p.log.Info("task finished", "id", t.id, "error", err)
}

// Submit never waits for queue space. The queue bound excludes running tasks.
func (p *Pool) Submit(ctx context.Context, work Work) (uint64, error) {
    if work == nil {
        return 0, errors.New("nil work")
    }
    p.mu.Lock()
    defer p.mu.Unlock()
    if err := ctx.Err(); err != nil {
        return 0, err
    }
    if p.closed {
        return 0, ErrClosed
    }
    id := p.next.Add(1)
    select {
    case p.queue <- task{id: id, work: work}:
        p.accepted.Add(1)
        p.log.Info("task accepted", "id", id)
        return id, nil
    default:
        return 0, ErrFull
    }
}

// Close stops admission and lets workers drain the queue. It is idempotent.
func (p *Pool) Close() {
    p.mu.Lock()
    defer p.mu.Unlock()
    if !p.closed {
        p.closed = true
        close(p.queue)
    }
}

// Shutdown drains until the deadline, then signals cancellation. It cannot
// force a Work function that ignores context to stop. Done reports actual exit.
func (p *Pool) Shutdown(ctx context.Context) error {
    p.Close()
    select {
    case <-p.done:
        return nil
    case <-ctx.Done():
        p.cancel()
        return ctx.Err()
    }
}

func (p *Pool) Done() <-chan struct{} { return p.done }

func (p *Pool) Stats() Stats {
    return Stats{Accepted: p.accepted.Load(), Started: p.started.Load(), Completed: p.completed.Load(), Failed: p.failed.Load(), Cancelled: p.cancelled.Load(), Active: p.active.Load(), Queued: len(p.queue)}
}

package service

import (
    "context"
    "errors"
    "io"
    "log/slog"
    "runtime"
    "sync"
    "testing"
    "time"
)

func quiet() *slog.Logger { return slog.New(slog.NewTextHandler(io.Discard, nil)) }

func receive(t *testing.T, ch <-chan struct{}) {
    t.Helper()
    select {
    case <-ch:
    case <-time.After(3 * time.Second):
        t.Fatal("timed out waiting for completion")
    }
}

func pool(t *testing.T, workers, capacity int, timeout time.Duration) *Pool {
    t.Helper()
    p, err := NewPool(workers, capacity, timeout, quiet())
    if err != nil {
        t.Fatal(err)
    }
    t.Cleanup(func() {
        ctx, cancel := context.WithTimeout(context.Background(), time.Second)
        defer cancel()
        _ = p.Shutdown(ctx)
        receive(t, p.Done())
    })
    return p
}

func TestPoolQueueBound(t *testing.T) {
    p := pool(t, 1, 2, time.Second)
    started, release := make(chan struct{}), make(chan struct{})
    _, err := p.Submit(context.Background(), func(ctx context.Context) error {
        close(started)
        select {
        case <-release:
            return nil
        case <-ctx.Done():
            return ctx.Err()
        }
    })
    if err != nil {
        t.Fatal(err)
    }
    receive(t, started)
    for range 2 {
        if _, err := p.Submit(context.Background(), func(context.Context) error { return nil }); err != nil {
            t.Fatal(err)
        }
    }
    if _, err := p.Submit(context.Background(), func(context.Context) error { return nil }); !errors.Is(err, ErrFull) {
        t.Fatalf("got %v, want ErrFull", err)
    }
    if s := p.Stats(); s.Queued != 2 || s.Active != 1 || s.Accepted != 3 {
        t.Fatalf("unexpected stats: %+v", s)
    }
    close(release)
    p.Close()
    receive(t, p.Done())
    if s := p.Stats(); s.Completed != 3 || s.Active != 0 || s.Queued != 0 {
        t.Fatalf("not drained: %+v", s)
    }
}

func TestAcceptedTaskOutlivesCaller(t *testing.T) {
    p := pool(t, 1, 1, time.Second)
    caller, cancel := context.WithCancel(context.Background())
    started, release, finished := make(chan struct{}), make(chan struct{}), make(chan struct{})
    _, err := p.Submit(caller, func(ctx context.Context) error {
        close(started)
        select {
        case <-release:
            close(finished)
            return nil
        case <-ctx.Done():
            return ctx.Err()
        }
    })
    if err != nil {
        t.Fatal(err)
    }
    receive(t, started)
    cancel()
    close(release)
    receive(t, finished)
    if _, err := p.Submit(caller, func(context.Context) error { return nil }); !errors.Is(err, context.Canceled) {
        t.Fatalf("cancelled admission: %v", err)
    }
}

func TestPoolBlockedCancellationReleases(t *testing.T) {
    p := pool(t, 1, 2, 5*time.Second)
    started, released := make(chan struct{}), make(chan struct{})
    _, err := p.Submit(context.Background(), func(ctx context.Context) error { close(started); <-ctx.Done(); close(released); return ctx.Err() })
    if err != nil {
        t.Fatal(err)
    }
    receive(t, started)
    // A queued function must never execute after forced cancellation.
    if _, err := p.Submit(context.Background(), func(context.Context) error { t.Error("cancelled queued work executed"); return nil }); err != nil {
        t.Fatal(err)
    }
    ctx, cancel := context.WithTimeout(context.Background(), 20*time.Millisecond)
    defer cancel()
    if err := p.Shutdown(ctx); !errors.Is(err, context.DeadlineExceeded) {
        t.Fatalf("shutdown: %v", err)
    }
    receive(t, released)
    receive(t, p.Done())
    if s := p.Stats(); s.Completed != 2 || s.Cancelled != 2 || s.Active != 0 {
        t.Fatalf("resources not released: %+v", s)
    }
}

func TestPoolTaskDeadline(t *testing.T) {
    p := pool(t, 1, 1, 20*time.Millisecond)
    _, err := p.Submit(context.Background(), func(ctx context.Context) error { <-ctx.Done(); return ctx.Err() })
    if err != nil {
        t.Fatal(err)
    }
    p.Close()
    receive(t, p.Done())
    if s := p.Stats(); s.Cancelled != 1 {
        t.Fatalf("deadline not observed: %+v", s)
    }
}

func TestPoolConcurrentCloseAndSubmit(t *testing.T) {
    p := pool(t, 4, 8, time.Second)
    var wg sync.WaitGroup
    for range 16 {
        wg.Add(1)
        go func() {
            defer wg.Done()
            for range 100 {
                _, err := p.Submit(context.Background(), func(context.Context) error { return nil })
                if err != nil && !errors.Is(err, ErrFull) && !errors.Is(err, ErrClosed) {
                    t.Errorf("submit: %v", err)
                }
            }
        }()
    }
    p.Close()
    wg.Wait()
    receive(t, p.Done())
    if s := p.Stats(); s.Completed != s.Accepted {
        t.Fatalf("accepted tasks lost: %+v", s)
    }
}

func BenchmarkPoolSubmit(b *testing.B) {
    p, _ := NewPool(4, 128, time.Second, quiet())
    defer func() { p.Close(); <-p.Done() }()
    b.ReportAllocs()
    for b.Loop() {
        for {
            _, err := p.Submit(context.Background(), func(context.Context) error { return nil })
            if err == nil {
                break
            }
            if !errors.Is(err, ErrFull) {
                b.Fatal(err)
            }
            runtime.Gosched()
        }
    }
}