Frontend Labs
⋯
Read as one page Markdown source ↗

11. Part I capstone: Orbit under Compose, fully working

Now complete the three services. These are the final versions; Part IV adds telemetry with a few lines in main.go.

internal/orders/handler.go: idempotent order creation with an outbox-style publish.

package orders

import (
	"context"
	"encoding/json"
	"errors"
	"log/slog"
	"net/http"

	"github.com/google/uuid"
	"github.com/jackc/pgx/v5"
	"github.com/jackc/pgx/v5/pgxpool"
	"github.com/redis/go-redis/v9"
)

const Stream = "orders.created"

type Handler struct {
	DB  *pgxpool.Pool
	RDB *redis.Client
}

type createReq struct {
	ProductID int64 `json:"product_id"`
	Quantity  int   `json:"quantity"`
}

func (h *Handler) Create(w http.ResponseWriter, r *http.Request) {
	key := r.Header.Get("Idempotency-Key")
	if key == "" {
		http.Error(w, "Idempotency-Key header required", http.StatusBadRequest)
		return
	}
	var req createReq
	if err := json.NewDecoder(http.MaxBytesReader(w, r.Body, 1<<16)).Decode(&req); err != nil || req.Quantity <= 0 {
		http.Error(w, "invalid body", http.StatusBadRequest)
		return
	}

	id, created, err := h.insert(r.Context(), key, req)
	if err != nil {
		slog.ErrorContext(r.Context(), "create order failed", "err", err)
		http.Error(w, "internal error", http.StatusInternalServerError)
		return
	}
	if created {
		// Publish after commit. If this fails, a reconciler (or an outbox table) re-publishes
		// pending orders; the worker is idempotent, so duplicates are harmless.
		if err := h.RDB.XAdd(r.Context(), &redis.XAddArgs{
			Stream: Stream, MaxLen: 100_000, Approx: true,
			Values: map[string]any{"order_id": id},
		}).Err(); err != nil {
			slog.WarnContext(r.Context(), "publish failed; will be re-driven", "order_id", id, "err", err)
		}
	}
	w.Header().Set("Content-Type", "application/json")
	if created {
		w.WriteHeader(http.StatusCreated)
	}
	_ = json.NewEncoder(w).Encode(map[string]string{"id": id, "status": "pending"})
}

// insert returns the existing order for a repeated idempotency key instead of creating a duplicate.
func (h *Handler) insert(ctx context.Context, key string, req createReq) (string, bool, error) {
	id := uuid.NewString()
	err := h.DB.QueryRow(ctx,
		`INSERT INTO orders (id, idempotency_key, product_id, quantity)
		 VALUES ($1, $2, $3, $4)
		 ON CONFLICT (idempotency_key) DO NOTHING
		 RETURNING id`, id, key, req.ProductID, req.Quantity).Scan(&id)
	if errors.Is(err, pgx.ErrNoRows) { // conflict: already exists
		err = h.DB.QueryRow(ctx, `SELECT id FROM orders WHERE idempotency_key=$1`, key).Scan(&id)
		return id, false, err
	}
	return id, err == nil, err
}

internal/worker/worker.go: a consumer group with acknowledgements, retry via pending-entry reclaim, and graceful stop.

package worker

import (
	"context"
	"errors"
	"log/slog"
	"math/rand/v2"
	"strings"
	"time"

	"github.com/jackc/pgx/v5/pgxpool"
	"github.com/redis/go-redis/v9"
)

type Worker struct {
	DB       *pgxpool.Pool
	RDB      *redis.Client
	Stream   string
	Group    string
	Consumer string // pod name: unique per replica
}

func (w *Worker) Run(ctx context.Context) error {
	// Create the group once; BUSYGROUP means it already exists.
	if err := w.RDB.XGroupCreateMkStream(ctx, w.Stream, w.Group, "0").Err(); err != nil &&
		!strings.HasPrefix(err.Error(), "BUSYGROUP") {
		return err
	}
	for ctx.Err() == nil {
		// Reclaim messages another (crashed) consumer held for > 60s: at-least-once delivery.
		claimed, _, _ := w.RDB.XAutoClaim(ctx, &redis.XAutoClaimArgs{
			Stream: w.Stream, Group: w.Group, Consumer: w.Consumer,
			MinIdle: time.Minute, Start: "0-0", Count: 10,
		}).Result()
		for _, m := range claimed {
			w.handle(ctx, m)
		}

		res, err := w.RDB.XReadGroup(ctx, &redis.XReadGroupArgs{
			Group: w.Group, Consumer: w.Consumer, Streams: []string{w.Stream, ">"},
			Count: 10, Block: 5 * time.Second,
		}).Result()
		if errors.Is(err, redis.Nil) || errors.Is(err, context.Canceled) {
			continue
		}
		if err != nil {
			slog.Error("xreadgroup", "err", err)
			time.Sleep(time.Second)
			continue
		}
		for _, s := range res {
			for _, m := range s.Messages {
				w.handle(ctx, m)
			}
		}
	}
	return nil
}

func (w *Worker) handle(ctx context.Context, m redis.XMessage) {
	orderID, _ := m.Values["order_id"].(string)
	// Use a context that survives SIGTERM so an in-flight message can finish.
	hctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), 10*time.Second)
	defer cancel()

	time.Sleep(time.Duration(50+rand.IntN(150)) * time.Millisecond) // "call the payment provider"

	// Idempotent state transition: only pending -> paid.
	if _, err := w.DB.Exec(hctx,
		`UPDATE orders SET status='paid', updated_at=now() WHERE id=$1 AND status='pending'`, orderID); err != nil {
		slog.Error("process order failed; will be retried", "order_id", orderID, "err", err)
		return // not ACKed -> reclaimed later by XAUTOCLAIM
	}
	w.RDB.XAck(hctx, w.Stream, w.Group, m.ID)
	slog.Info("order paid", "order_id", orderID)
}

Production notes embedded in this code (each one maps to a later chapter):

  • Idempotency keys make client retries safe. Retries happen constantly in Kubernetes: rolling updates, pod evictions, LB timeouts.
  • At-least-once + idempotent consumer is the realistic delivery contract. Exactly-once is a property you build, not one you get from the queue.
  • Consumer = pod name (from the Downward API, §16.4), so reclaimed messages can be traced to the pod that dropped them.
  • A message that fails forever would loop forever. In production, check the delivery count (XPENDING) and move a message to a dead-letter stream after N attempts. Left as an exercise; the pattern is in §43.

The main.go files wire it together. cmd/orders/main.go (the worker's main is the same shape: it builds a worker.Worker and calls Run(ctx) under signal.NotifyContext instead of platform.Run):

package main

import (
	"context"
	"log/slog"
	"net/http"
	"os"
	"time"

	"github.com/jackc/pgx/v5/pgxpool"
	"github.com/redis/go-redis/v9"

	"github.com/acme/orbit/internal/orders"
	"github.com/acme/orbit/internal/platform"
)

func main() {
	slog.SetDefault(slog.New(slog.NewJSONHandler(os.Stdout, nil)))
	ctx := context.Background()

	pcfg, err := pgxpool.ParseConfig(platform.MustEnv("DATABASE_URL"))
	if err != nil {
		slog.Error("bad DATABASE_URL", "err", err)
		os.Exit(1)
	}
	pcfg.MaxConns = int32(platform.EnvInt("DB_MAX_CONNS", 10)) // x replicas must stay < Postgres max_connections
	db, err := pgxpool.NewWithConfig(ctx, pcfg)                 // lazy: does not fail if the DB is down at boot
	if err != nil {
		slog.Error("db pool", "err", err)
		os.Exit(1)
	}
	defer db.Close()
	rdb := redis.NewClient(&redis.Options{Addr: platform.MustEnv("REDIS_ADDR")})
	defer rdb.Close()

	health := platform.NewHealth(
		platform.Check{Name: "postgres", Fn: db.Ping},
		platform.Check{Name: "redis", Fn: func(ctx context.Context) error { return rdb.Ping(ctx).Err() }},
	)
	h := &orders.Handler{DB: db, RDB: rdb}

	mux := http.NewServeMux()
	mux.HandleFunc("GET /livez", health.Livez)
	mux.HandleFunc("GET /readyz", health.Readyz)
	mux.HandleFunc("POST /orders", h.Create)

	if err := platform.Run(platform.ServerConfig{
		Addr:            ":" + platform.Env("PORT", "8080"),
		DrainDelay:      platform.EnvDuration("DRAIN_DELAY", 5*time.Second),
		ShutdownTimeout: platform.EnvDuration("SHUTDOWN_TIMEOUT", 20*time.Second),
	}, mux, health); err != nil {
		slog.Error("server failed", "err", err)
		os.Exit(1)
	}
}

Note the boot behavior: the process starts even if Postgres is down (the pool connects lazily), stays alive, and reports not ready until the dependency is reachable. That is exactly what Kubernetes probes expect. A service that crashes on boot when its DB is unavailable produces crash-loop storms during every database failover.

Run it:

docker compose --profile obs up -d --build --wait
curl -s localhost:8081/products | jq
curl -s -XPOST localhost:8082/orders -H 'Idempotency-Key: demo-1' \
     -H 'Content-Type: application/json' -d '{"product_id":1,"quantity":2}' | jq
docker compose exec db psql -U orbit -c 'select id,status from orders'   # pending -> paid within ~1s

Part I checkpoint: a 10 MB distroless, non-root, read-only image per service; a reproducible local stack with healthchecks and migrations; and you can explain what happens on docker stop.

⚠️ Caveat (verify): the Go code in §4.1 and §11 was written for this guide and has not been compiled. Assemble the repo and run go mod tidy && go build ./... && go vet ./... before trusting it. Known spots to check: the return values of go-redis v9 XAutoClaim, and the uuid dependency (go get github.com/google/uuid).

Generated from real-life-k8s-guide.md · 8 Oct 2026 16:36