diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 200f6a2..82049c9 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -67,6 +67,20 @@ jobs: - run: go test ./internal/admin/ -v -count=1 -race env: TOLLGATE_TEST_POSTGRES: postgres://tollgate:tollgate@localhost:5432/tollgate?sslmode=disable + # The outbox's claim is that a usage window and the message reporting it + # commit together, and that a redelivery lands in an inbox row that is + # already there. Both are claims about two SEPARATE databases, so the + # suite needs a second one and skips itself without it - which meant it + # ran nowhere. This creates the consumer database in the same service + # container and points the suite at both. + - name: consumer database for the outbox suite + run: | + PGPASSWORD=tollgate psql -h localhost -U tollgate -d tollgate \ + -v ON_ERROR_STOP=1 -c 'CREATE DATABASE billing' + - run: go test ./internal/outbox/ -v -count=1 -race + env: + TOLLGATE_TEST_POSTGRES: postgres://tollgate:tollgate@localhost:5432/tollgate?sslmode=disable + TOLLGATE_TEST_BILLING: postgres://tollgate:tollgate@localhost:5432/billing?sslmode=disable # A NetworkPolicy whose selector matches nothing is not a weaker policy, it is # no policy, and Kubernetes accepts it without complaint. The chart shipped diff --git a/cmd/tollgate-outbox/main.go b/cmd/tollgate-outbox/main.go new file mode 100644 index 0000000..8e4f929 --- /dev/null +++ b/cmd/tollgate-outbox/main.go @@ -0,0 +1,301 @@ +// Command tollgate-outbox runs the two halves of the transactional outbox as +// separate processes, which is what makes the crash tests mean anything: the +// relay can be killed without taking the consumer with it, and the consumer's +// database is genuinely a different database from the gateway's. +// +// tollgate-outbox migrate-sink -db $BILLING_URL +// tollgate-outbox sink -db $BILLING_URL -addr :9411 +// tollgate-outbox seal -db $DATABASE_URL -tenant acme -window ... -requests 120 +// tollgate-outbox relay -db $DATABASE_URL -sink http://localhost:9411 +// tollgate-outbox reconcile -db $DATABASE_URL -sink http://localhost:9411 +// tollgate-outbox status -db $DATABASE_URL +// +// -crash names a point at which the process kills itself with SIGKILL. It is +// for the crash tests and nothing else; unset, the hook is nil. +package main + +import ( + "context" + "encoding/json" + "errors" + "flag" + "fmt" + "net/http" + "os" + "os/signal" + "syscall" + "time" + + "github.com/jackc/pgx/v5/pgxpool" + + "github.com/lgoyal6/tollgate/internal/outbox" +) + +func main() { + if len(os.Args) < 2 { + usage() + } + ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) + defer stop() + + var err error + switch os.Args[1] { + case "migrate-sink": + err = migrateSink(ctx, os.Args[2:]) + case "sink": + err = runSink(ctx, os.Args[2:]) + case "seal": + err = seal(ctx, os.Args[2:]) + case "relay": + err = relay(ctx, os.Args[2:]) + case "reconcile": + err = reconcile(ctx, os.Args[2:]) + case "status": + err = status(ctx, os.Args[2:]) + default: + usage() + } + if err != nil { + fmt.Fprintln(os.Stderr, "error:", err) + os.Exit(1) + } +} + +func usage() { + fmt.Fprintln(os.Stderr, "usage: tollgate-outbox [flags]") + os.Exit(2) +} + +func connect(ctx context.Context, url string) (*pgxpool.Pool, error) { + if url == "" { + return nil, errors.New("-db is required") + } + pool, err := pgxpool.New(ctx, url) + if err != nil { + return nil, err + } + if err := pool.Ping(ctx); err != nil { + pool.Close() + return nil, err + } + return pool, nil +} + +// killer returns a Crasher that really kills this process at the named point. +// +// SIGKILL, not panic and not os.Exit: a panic runs deferred functions and an +// exit flushes buffers, and either would let the process tidy up in a way a +// power cut or an OOM kill never would. SIGKILL cannot be caught, blocked or +// ignored, so what survives is only what Postgres already committed. +func killer(point string) outbox.Crasher { + if point == "" { + return nil + } + want := outbox.CrashPoint(point) + return func(p outbox.CrashPoint) { + if p != want { + return + } + fmt.Fprintf(os.Stderr, "crash point %s reached; SIGKILL to pid %d\n", p, os.Getpid()) + _ = os.Stderr.Sync() + _ = syscall.Kill(os.Getpid(), syscall.SIGKILL) + select {} // unreachable + } +} + +func migrateSink(ctx context.Context, args []string) error { + fs := flag.NewFlagSet("migrate-sink", flag.ExitOnError) + db := fs.String("db", os.Getenv("BILLING_URL"), "consumer database URL") + _ = fs.Parse(args) + pool, err := connect(ctx, *db) + if err != nil { + return err + } + defer pool.Close() + if err := outbox.MigrateSink(ctx, pool); err != nil { + return err + } + fmt.Println("sink schema ready") + return nil +} + +func runSink(ctx context.Context, args []string) error { + fs := flag.NewFlagSet("sink", flag.ExitOnError) + db := fs.String("db", os.Getenv("BILLING_URL"), "consumer database URL") + addr := fs.String("addr", ":9411", "listen address") + dedupe := fs.Bool("dedupe", true, "apply inbox deduplication (false is the negative control)") + _ = fs.Parse(args) + pool, err := connect(ctx, *db) + if err != nil { + return err + } + defer pool.Close() + sink := &outbox.BillingSink{Pool: pool, Dedupe: *dedupe} + srv := &http.Server{Addr: *addr, Handler: sink.Handler(), ReadHeaderTimeout: 5 * time.Second} + go func() { + <-ctx.Done() + sctx, cancel := context.WithTimeout(context.Background(), 3*time.Second) + defer cancel() + _ = srv.Shutdown(sctx) + }() + fmt.Printf("billing sink listening on %s (dedupe=%v)\n", *addr, *dedupe) + if err := srv.ListenAndServe(); !errors.Is(err, http.ErrServerClosed) { + return err + } + return nil +} + +func seal(ctx context.Context, args []string) error { + fs := flag.NewFlagSet("seal", flag.ExitOnError) + db := fs.String("db", os.Getenv("DATABASE_URL"), "gateway database URL") + tenant := fs.String("tenant", "", "tenant id") + window := fs.String("window", "", "window start, RFC3339") + requests := fs.Int64("requests", 0, "requests in the window") + limited := fs.Int64("limited", 0, "rate-limited requests in the window") + mode := fs.String("mode", "outbox", "outbox | dual-write (dual-write is the negative control)") + sinkURL := fs.String("sink", "", "sink base URL, dual-write mode only") + crash := fs.String("crash", "", "crash point: after_commit") + _ = fs.Parse(args) + + start, err := time.Parse(time.RFC3339, *window) + if err != nil { + return fmt.Errorf("parsing -window: %w", err) + } + pool, err := connect(ctx, *db) + if err != nil { + return err + } + defer pool.Close() + + w := outbox.Window{ + TenantID: *tenant, WindowStart: start, WindowEnd: start.Add(time.Minute), + Requests: *requests, Admitted: *requests - *limited, Limited: *limited, + } + switch *mode { + case "outbox": + n, err := outbox.SealWindows(ctx, pool, []outbox.Window{w}, killer(*crash)) + if err != nil { + return err + } + fmt.Printf("sealed %d window(s); key=%s\n", n, outbox.UsageKey(*tenant, start)) + case "dual-write": + if *sinkURL == "" { + return errors.New("-sink is required in dual-write mode") + } + if err := outbox.SealWindowsDualWrite(ctx, pool, []outbox.Window{w}, outbox.HTTPSink{BaseURL: *sinkURL}, killer(*crash)); err != nil { + return err + } + fmt.Printf("dual-write completed; key=%s\n", outbox.UsageKey(*tenant, start)) + default: + return fmt.Errorf("unknown -mode %q", *mode) + } + return nil +} + +func newRelay(pool *pgxpool.Pool, sinkURL string, noLookup bool, lease time.Duration, crash string) *outbox.Relay { + host, _ := os.Hostname() + return &outbox.Relay{ + Pool: pool, + Sink: outbox.HTTPSink{BaseURL: sinkURL, LookupDisabled: noLookup}, + Owner: fmt.Sprintf("%s/%d", host, os.Getpid()), + Lease: lease, + Batch: 32, + Crash: killer(crash), + } +} + +func relay(ctx context.Context, args []string) error { + fs := flag.NewFlagSet("relay", flag.ExitOnError) + db := fs.String("db", os.Getenv("DATABASE_URL"), "gateway database URL") + sinkURL := fs.String("sink", "", "sink base URL") + once := fs.Bool("once", true, "run a single pass and exit") + interval := fs.Duration("interval", 2*time.Second, "poll interval when not -once") + lease := fs.Duration("lease", 30*time.Second, "attempt lease") + crash := fs.String("crash", "", "crash point: before_send | after_send") + _ = fs.Parse(args) + if *sinkURL == "" { + return errors.New("-sink is required") + } + pool, err := connect(ctx, *db) + if err != nil { + return err + } + defer pool.Close() + r := newRelay(pool, *sinkURL, false, *lease, *crash) + + run := func() error { + res, err := r.Once(ctx) + if err != nil { + return err + } + b, _ := json.Marshal(res) + fmt.Printf("pass %s\n", b) + return nil + } + if *once { + return run() + } + tick := time.NewTicker(*interval) + defer tick.Stop() + for { + if err := run(); err != nil { + return err + } + select { + case <-ctx.Done(): + return nil + case <-tick.C: + } + } +} + +func reconcile(ctx context.Context, args []string) error { + fs := flag.NewFlagSet("reconcile", flag.ExitOnError) + db := fs.String("db", os.Getenv("DATABASE_URL"), "gateway database URL") + sinkURL := fs.String("sink", "", "sink base URL") + noLookup := fs.Bool("no-lookup", false, "model a downstream that cannot be asked") + expire := fs.Bool("expire", true, "expire dead leases into UNKNOWN before reconciling") + lease := fs.Duration("lease", 30*time.Second, "attempt lease") + _ = fs.Parse(args) + if *sinkURL == "" { + return errors.New("-sink is required") + } + pool, err := connect(ctx, *db) + if err != nil { + return err + } + defer pool.Close() + r := newRelay(pool, *sinkURL, *noLookup, *lease, "") + if *expire { + n, err := r.ExpireLeases(ctx) + if err != nil { + return err + } + fmt.Printf("expired %d lease(s) into UNKNOWN\n", n) + } + res, err := r.Reconcile(ctx) + if err != nil { + return err + } + b, _ := json.Marshal(res) + fmt.Printf("reconcile %s\n", b) + return nil +} + +func status(ctx context.Context, args []string) error { + fs := flag.NewFlagSet("status", flag.ExitOnError) + db := fs.String("db", os.Getenv("DATABASE_URL"), "gateway database URL") + _ = fs.Parse(args) + pool, err := connect(ctx, *db) + if err != nil { + return err + } + defer pool.Close() + counts, err := outbox.Counts(ctx, pool) + if err != nil { + return err + } + b, _ := json.Marshal(counts) + fmt.Printf("outbox %s\n", b) + return nil +} diff --git a/internal/config/config.go b/internal/config/config.go index 3d445a8..6069548 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -50,6 +50,12 @@ type Config struct { // giving load balancers time to remove this replica before Shutdown. DrainDelay time.Duration + // UsageSealInterval is how often a usage window is closed into the ledger + // and queued for the billing sink. Zero, the default, keeps the sealer off: + // a deployment with no downstream to bill should not accumulate messages + // nobody will ever relay. + UsageSealInterval time.Duration + // HedgingEnabled is the global gate; a route must also opt in. HedgingEnabled bool // MaxBodyBuffer caps how much of a request body is buffered to make it @@ -172,6 +178,9 @@ func Load() (Config, error) { if cfg.DrainDelay, err = getDuration("DRAIN_DELAY", 0); err != nil { return Config{}, err } + if cfg.UsageSealInterval, err = getDuration("USAGE_SEAL_INTERVAL", 0); err != nil { + return Config{}, err + } if cfg.MaxBodyBuffer, err = getInt64("MAX_BODY_BUFFER_BYTES", 1<<20); err != nil { return Config{}, err } diff --git a/internal/gateway/gateway.go b/internal/gateway/gateway.go index 3d4f768..061dc89 100644 --- a/internal/gateway/gateway.go +++ b/internal/gateway/gateway.go @@ -336,6 +336,14 @@ func (g *Gateway) Run(ctx context.Context) error { errCh <- fmt.Errorf("main listener: %w", err) } }() + // Sealing is the gateway's half of the outbox: it closes a usage window and + // queues the charge in one transaction. Delivering that charge is a + // separate process (cmd/tollgate-outbox relay), which is what lets either + // side be killed without losing or double-billing a window. + if g.cfg.UsageSealInterval > 0 && g.store != nil { + go g.runUsageSealer(ctx, g.cfg.UsageSealInterval) + g.logger.Info("usage sealer started", "interval", g.cfg.UsageSealInterval) + } go func() { g.logger.Info("admin listening", "addr", g.cfg.AdminAddr) if err := adminSrv.ListenAndServe(); !errors.Is(err, http.ErrServerClosed) { diff --git a/internal/gateway/usage_seal.go b/internal/gateway/usage_seal.go new file mode 100644 index 0000000..7ed31ad --- /dev/null +++ b/internal/gateway/usage_seal.go @@ -0,0 +1,86 @@ +package gateway + +import ( + "context" + "time" + + "github.com/lgoyal6/tollgate/internal/admin" + "github.com/lgoyal6/tollgate/internal/outbox" + "github.com/lgoyal6/tollgate/internal/reqctx" +) + +// sealUsageWindows closes one usage window and writes it to the ledger. +// +// The counters come from this replica's own Prometheus registry, which is where +// they already were; what changes is that a closed window now leaves a durable +// row and a durable message instead of only a gauge that a restart resets. The +// request path is untouched: nothing here runs per request. +// +// Deltas against the previous snapshot, because Prometheus counters are +// cumulative and a window is the difference. A counter that went backwards +// means the registry was reset, which for a single process means it restarted; +// that window is skipped rather than recorded as a negative charge. +func (g *Gateway) sealUsageWindows(ctx context.Context, previous map[string]admin.TenantCounters, start, end time.Time) (map[string]admin.TenantCounters, error) { + current := usageFromMetrics{g.metrics}.TenantUsage() + var windows []outbox.Window + for tenant, now := range current { + // Health probes and rejected keys all land under one metrics label that + // is not an account. Sealing it would queue a charge every interval, + // forever, addressed to a tenant no billing system has; a Kubernetes + // readiness probe alone would keep the outbox permanently non-empty. + if tenant == reqctx.UnauthenticatedTenant { + continue + } + was := previous[tenant] + requests := int64(now.Requests - was.Requests) + if requests <= 0 { + continue + } + windows = append(windows, outbox.Window{ + TenantID: tenant, WindowStart: start, WindowEnd: end, + Requests: requests, + Admitted: max(int64(now.Admitted-was.Admitted), 0), + Limited: max(int64(now.Limited-was.Limited), 0), + ServerErr: max(int64(now.ServerErr-was.ServerErr), 0), + }) + } + if len(windows) == 0 { + return current, nil + } + sealed, err := outbox.SealWindows(ctx, g.store.Pool, windows, nil) + if err != nil { + // The snapshot is deliberately NOT advanced on failure, so the next tick + // bills the whole span rather than losing the traffic in between. + return previous, err + } + g.logger.Debug("usage window sealed", "tenants", sealed, "window_start", start) + return current, nil +} + +// runUsageSealer seals a window every interval until the context is cancelled. +// +// It is a goroutine in the gateway rather than a cron job because the numbers +// live in this process's registry and nowhere else; a separate sealer would +// have nothing to read. Delivery is somebody else's process: cmd/tollgate-outbox +// relay drains what this queues, and can be killed and restarted at will +// precisely because the queue is in Postgres. +func (g *Gateway) runUsageSealer(ctx context.Context, interval time.Duration) { + tick := time.NewTicker(interval) + defer tick.Stop() + previous := map[string]admin.TenantCounters{} + start := time.Now().UTC().Truncate(interval) + for { + select { + case <-ctx.Done(): + return + case now := <-tick.C: + end := now.UTC() + next, err := g.sealUsageWindows(ctx, previous, start, end) + if err != nil { + g.logger.Error("sealing usage window", "err", err) + continue + } + previous, start = next, end + } + } +} diff --git a/internal/outbox/outbox.go b/internal/outbox/outbox.go new file mode 100644 index 0000000..9fcb9bd --- /dev/null +++ b/internal/outbox/outbox.go @@ -0,0 +1,93 @@ +// Package outbox carries a side effect and the state change it belongs to +// across a process crash. +// +// The problem it solves is the dual write: commit a row, then make an HTTP +// call. A process killed between the two has produced a state change nobody +// downstream will ever hear about, and no amount of retrying inside the dead +// process helps, because the process is gone. Writing the message into the same +// transaction as the state change removes that window: after the commit, the +// intent to deliver is as durable as the fact it reports. +// +// What this package does NOT do is make delivery happen once. Delivery is +// retried, so the network sees the message more than once. Landing once is the +// consumer's job, and Sink implements it: the effect and the inbox row commit +// together, so a redelivered key applies nothing. See RECORD_c13_outbox.md for +// the precise guarantee and the assumptions it rests on. +package outbox + +import ( + "context" + "encoding/json" + "fmt" + "time" + + "github.com/jackc/pgx/v5" +) + +// State values as they appear in the outbox.state column. +const ( + StatePending = "PENDING" + StateInflight = "INFLIGHT" + StateUnknown = "UNKNOWN" + StateDelivered = "DELIVERED" + StateFailed = "FAILED" +) + +// Message is one side effect waiting to be delivered. +type Message struct { + ID int64 + IdempotencyKey string + Topic string + Payload json.RawMessage + Attempts int +} + +// Enqueue writes a message inside the caller's transaction. +// +// The caller MUST be inside the transaction that performs the state change. +// Passing a pool here instead of a transaction would reintroduce the exact dual +// write this package exists to remove, which is why the signature takes a +// pgx.Tx and not a *pgxpool.Pool. +// +// A repeated key is dropped rather than treated as an error: sealing the same +// usage window twice is a legitimate retry of the producer, and the second +// attempt must not enqueue a second charge. +func Enqueue(ctx context.Context, tx pgx.Tx, key, topic string, payload any) error { + body, err := json.Marshal(payload) + if err != nil { + return fmt.Errorf("encoding outbox payload for %s: %w", key, err) + } + if _, err := tx.Exec(ctx, ` + INSERT INTO outbox (idempotency_key, topic, payload) + VALUES ($1, $2, $3) + ON CONFLICT (idempotency_key) DO NOTHING`, + key, topic, body); err != nil { + return fmt.Errorf("enqueueing %s: %w", key, err) + } + return nil +} + +// UsageEvent is the payload for the usage.window.sealed topic. +type UsageEvent struct { + TenantID string `json:"tenant_id"` + WindowStart time.Time `json:"window_start"` + WindowEnd time.Time `json:"window_end"` + Requests int64 `json:"requests"` + Admitted int64 `json:"admitted"` + Limited int64 `json:"limited"` + ServerErr int64 `json:"server_err"` +} + +// TopicUsageSealed is the only topic today. It exists as a constant because the +// consumer switches on it and a typo would deliver into nothing. +const TopicUsageSealed = "usage.window.sealed" + +// UsageKey is the idempotency key for a sealed window. +// +// It is a function of WHAT happened - this tenant, this window - and never of +// WHEN it was attempted or by whom. A key that varied per attempt would make +// every retry look like a new charge to the consumer, which is the failure this +// whole mechanism is built to avoid. +func UsageKey(tenantID string, windowStart time.Time) string { + return fmt.Sprintf("usage:%s:%d", tenantID, windowStart.UTC().Unix()) +} diff --git a/internal/outbox/outbox_test.go b/internal/outbox/outbox_test.go new file mode 100644 index 0000000..7c38060 --- /dev/null +++ b/internal/outbox/outbox_test.go @@ -0,0 +1,322 @@ +package outbox_test + +import ( + "context" + "encoding/json" + "net/http/httptest" + "os" + "testing" + "time" + + "github.com/jackc/pgx/v5/pgxpool" + + "github.com/lgoyal6/tollgate/internal/outbox" + "github.com/lgoyal6/tollgate/migrations" +) + +// These need two real databases, because "the effect and the inbox row commit +// together" is a claim about a transaction and there is nothing to test without +// one. The crash half of C13 is scripts/outbox-crash.sh, which kills real +// processes; a test binary cannot SIGKILL itself and keep asserting. +// +// TOLLGATE_TEST_POSTGRES=postgres://.../tollgate \ +// TOLLGATE_TEST_BILLING=postgres://.../billing go test ./internal/outbox/ +func dbs(t *testing.T) (producer, consumer *pgxpool.Pool) { + t.Helper() + pURL, cURL := os.Getenv("TOLLGATE_TEST_POSTGRES"), os.Getenv("TOLLGATE_TEST_BILLING") + if pURL == "" || cURL == "" { + t.Skip("set TOLLGATE_TEST_POSTGRES and TOLLGATE_TEST_BILLING") + } + ctx := context.Background() + open := func(url string) *pgxpool.Pool { + pool, err := pgxpool.New(ctx, url) + if err != nil { + t.Fatalf("connecting to %s: %v", url, err) + } + if err := pool.Ping(ctx); err != nil { + t.Fatalf("pinging %s: %v", url, err) + } + t.Cleanup(pool.Close) + return pool + } + producer, consumer = open(pURL), open(cURL) + body, err := migrations.FS.ReadFile("004_outbox.sql") + if err != nil { + t.Fatal(err) + } + if _, err := producer.Exec(ctx, string(body)); err != nil { + t.Fatalf("applying 004_outbox.sql: %v", err) + } + if err := outbox.MigrateSink(ctx, consumer); err != nil { + t.Fatal(err) + } + if _, err := producer.Exec(ctx, `TRUNCATE outbox, usage_ledger`); err != nil { + t.Fatal(err) + } + if _, err := consumer.Exec(ctx, `TRUNCATE billing_charges, inbox CASCADE`); err != nil { + t.Fatal(err) + } + return producer, consumer +} + +func window(n int) outbox.Window { + start := time.Unix(1_700_000_000, 0).UTC().Add(time.Duration(n) * time.Minute) + return outbox.Window{ + TenantID: "acme", WindowStart: start, WindowEnd: start.Add(time.Minute), + Requests: int64(100 * n), Admitted: int64(100 * n), Limited: 0, + } +} + +func charges(t *testing.T, consumer *pgxpool.Pool, key string) int { + t.Helper() + var n int + if err := consumer.QueryRow(context.Background(), + `SELECT count(*) FROM billing_charges WHERE idempotency_key = $1`, key).Scan(&n); err != nil { + t.Fatal(err) + } + return n +} + +func state(t *testing.T, producer *pgxpool.Pool, key string) (string, string) { + t.Helper() + var s, res string + if err := producer.QueryRow(context.Background(), + `SELECT state, coalesce(resolution, '') FROM outbox WHERE idempotency_key = $1`, key).Scan(&s, &res); err != nil { + t.Fatal(err) + } + return s, res +} + +func harness(t *testing.T) (context.Context, *pgxpool.Pool, *pgxpool.Pool, *outbox.Relay, *httptest.Server) { + t.Helper() + ctx := context.Background() + producer, consumer := dbs(t) + sink := &outbox.BillingSink{Pool: consumer, Dedupe: true} + srv := httptest.NewServer(sink.Handler()) + t.Cleanup(srv.Close) + relay := &outbox.Relay{ + Pool: producer, Sink: outbox.HTTPSink{BaseURL: srv.URL}, Owner: t.Name(), + Lease: time.Minute, Batch: 16, Backoff: func(int) time.Duration { return 0 }, + } + return ctx, producer, consumer, relay, srv +} + +// The ledger row and the message that reports it are one write. Rolling the +// transaction back must leave neither behind. +func TestSealIsAtomic(t *testing.T) { + ctx, producer, _, _, _ := harness(t) + w := window(1) + + tx, err := producer.Begin(ctx) + if err != nil { + t.Fatal(err) + } + if _, err := tx.Exec(ctx, ` + INSERT INTO usage_ledger (tenant_id, window_start, window_end, requests) + VALUES ($1,$2,$3,$4)`, w.TenantID, w.WindowStart, w.WindowEnd, w.Requests); err != nil { + t.Fatal(err) + } + if err := outbox.Enqueue(ctx, tx, outbox.UsageKey(w.TenantID, w.WindowStart), outbox.TopicUsageSealed, outbox.UsageEvent{TenantID: w.TenantID}); err != nil { + t.Fatal(err) + } + if err := tx.Rollback(ctx); err != nil { + t.Fatal(err) + } + + var ledger, msgs int + if err := producer.QueryRow(ctx, `SELECT count(*) FROM usage_ledger`).Scan(&ledger); err != nil { + t.Fatal(err) + } + if err := producer.QueryRow(ctx, `SELECT count(*) FROM outbox`).Scan(&msgs); err != nil { + t.Fatal(err) + } + if ledger != 0 || msgs != 0 { + t.Fatalf("rollback left %d ledger rows and %d messages; both must be 0", ledger, msgs) + } +} + +// Replay is the property the crash tests depend on: however many times the same +// message is delivered, the consumer charges once. +func TestReplayDoesNotDuplicateTheEffect(t *testing.T) { + ctx, producer, consumer, relay, _ := harness(t) + w := window(2) + key := outbox.UsageKey(w.TenantID, w.WindowStart) + if _, err := outbox.SealWindows(ctx, producer, []outbox.Window{w}, nil); err != nil { + t.Fatal(err) + } + for i := range 5 { + if _, err := relay.Once(ctx); err != nil { + t.Fatalf("pass %d: %v", i, err) + } + if _, err := producer.Exec(ctx, ` + UPDATE outbox SET state='PENDING', next_attempt_at=now(), receipt=NULL, delivered_at=NULL + WHERE idempotency_key=$1 AND state='DELIVERED'`, key); err != nil { + t.Fatal(err) + } + } + if got := charges(t, consumer, key); got != 1 { + t.Fatalf("5 deliveries produced %d charges, want 1", got) + } + var deliveries int + if err := consumer.QueryRow(ctx, `SELECT deliveries FROM inbox WHERE idempotency_key=$1`, key).Scan(&deliveries); err != nil { + t.Fatal(err) + } + // If this were 1 the test would be passing because nothing was redelivered, + // not because the inbox suppressed anything. + if deliveries != 5 { + t.Fatalf("the consumer saw %d deliveries; the replay did not actually happen", deliveries) + } +} + +// Sealing the same window again must not queue a second charge, whatever the +// caller believes about the first attempt. +func TestResealIsIdempotent(t *testing.T) { + ctx, producer, consumer, relay, _ := harness(t) + w := window(3) + key := outbox.UsageKey(w.TenantID, w.WindowStart) + for range 3 { + if _, err := outbox.SealWindows(ctx, producer, []outbox.Window{w}, nil); err != nil { + t.Fatal(err) + } + } + var msgs int + if err := producer.QueryRow(ctx, `SELECT count(*) FROM outbox WHERE idempotency_key=$1`, key).Scan(&msgs); err != nil { + t.Fatal(err) + } + if msgs != 1 { + t.Fatalf("3 seals produced %d messages, want 1", msgs) + } + if _, err := relay.Once(ctx); err != nil { + t.Fatal(err) + } + if got := charges(t, consumer, key); got != 1 { + t.Fatalf("charges = %d, want 1", got) + } +} + +// An attempt whose process vanished becomes UNKNOWN and stays there until +// something asks the consumer. It must never be silently resolved either way. +func TestExpiredLeaseBecomesUnknownNotPending(t *testing.T) { + ctx, producer, _, relay, _ := harness(t) + w := window(4) + key := outbox.UsageKey(w.TenantID, w.WindowStart) + if _, err := outbox.SealWindows(ctx, producer, []outbox.Window{w}, nil); err != nil { + t.Fatal(err) + } + // Stand in for the claim a killed relay committed before it died. + if _, err := producer.Exec(ctx, ` + UPDATE outbox SET state='INFLIGHT', attempts=1, lease_owner='ghost', + lease_expires_at = now() - interval '1 second' WHERE idempotency_key=$1`, key); err != nil { + t.Fatal(err) + } + n, err := relay.ExpireLeases(ctx) + if err != nil { + t.Fatal(err) + } + if n != 1 { + t.Fatalf("expired %d leases, want 1", n) + } + if s, _ := state(t, producer, key); s != outbox.StateUnknown { + t.Fatalf("state = %s, want UNKNOWN; PENDING would be asserting the consumer did not get it", s) + } +} + +// The two ambiguous crashes leave identical rows in the producer's database. +// Only the consumer can tell them apart, and reconcile must reach the opposite +// conclusion in each case. +func TestReconcileSettlesBothDirections(t *testing.T) { + ctx, producer, consumer, relay, _ := harness(t) + + // (a) the consumer never got it. + wa := window(5) + ka := outbox.UsageKey(wa.TenantID, wa.WindowStart) + // (b) the consumer got it and the acknowledgement was lost. + wb := window(6) + kb := outbox.UsageKey(wb.TenantID, wb.WindowStart) + if _, err := outbox.SealWindows(ctx, producer, []outbox.Window{wa, wb}, nil); err != nil { + t.Fatal(err) + } + sink := &outbox.BillingSink{Pool: consumer, Dedupe: true} + payload, _ := json.Marshal(outbox.UsageEvent{TenantID: wb.TenantID, WindowStart: wb.WindowStart, WindowEnd: wb.WindowEnd, Requests: wb.Requests}) + if _, err := sink.Apply(ctx, kb, outbox.TopicUsageSealed, payload); err != nil { + t.Fatal(err) + } + if _, err := producer.Exec(ctx, ` + UPDATE outbox SET state='UNKNOWN', attempts=1 WHERE idempotency_key = ANY($1)`, + []string{ka, kb}); err != nil { + t.Fatal(err) + } + + res, err := relay.Reconcile(ctx) + if err != nil { + t.Fatal(err) + } + if res.Examined != 2 || res.Absent != 1 || res.Confirmed != 1 || res.Unresolved != 0 { + t.Fatalf("reconcile = %+v, want 2 examined / 1 absent / 1 confirmed / 0 unresolved", res) + } + if s, r := state(t, producer, ka); s != outbox.StatePending || r != "absent_at_consumer_after_crash" { + t.Fatalf("never-received message settled as %s/%s", s, r) + } + if s, r := state(t, producer, kb); s != outbox.StateDelivered || r != "confirmed_by_consumer_after_crash" { + t.Fatalf("already-applied message settled as %s/%s", s, r) + } + + if _, err := relay.Once(ctx); err != nil { + t.Fatal(err) + } + if got := charges(t, consumer, ka); got != 1 { + t.Fatalf("charges for the redelivered message = %d, want 1", got) + } + if got := charges(t, consumer, kb); got != 1 { + t.Fatalf("charges for the confirmed message = %d, want 1", got) + } +} + +// A downstream that cannot answer leaves the row unresolved. Reconciliation is +// allowed to fail; it is not allowed to guess. +func TestReconcileLeavesUnknownWhenTheSinkCannotAnswer(t *testing.T) { + ctx, producer, _, relay, srv := harness(t) + w := window(7) + key := outbox.UsageKey(w.TenantID, w.WindowStart) + if _, err := outbox.SealWindows(ctx, producer, []outbox.Window{w}, nil); err != nil { + t.Fatal(err) + } + if _, err := producer.Exec(ctx, ` + UPDATE outbox SET state='UNKNOWN', attempts=1 WHERE idempotency_key=$1`, key); err != nil { + t.Fatal(err) + } + relay.Sink = outbox.HTTPSink{BaseURL: srv.URL, LookupDisabled: true} + + res, err := relay.Reconcile(ctx) + if err != nil { + t.Fatal(err) + } + if res.Unresolved != 1 || res.Confirmed != 0 || res.Absent != 0 { + t.Fatalf("reconcile = %+v, want 1 unresolved and nothing settled", res) + } + if s, _ := state(t, producer, key); s != outbox.StateUnknown { + t.Fatalf("state = %s, want UNKNOWN", s) + } + var lastErr string + if err := producer.QueryRow(ctx, `SELECT coalesce(last_error,'') FROM outbox WHERE idempotency_key=$1`, key).Scan(&lastErr); err != nil { + t.Fatal(err) + } + if lastErr == "" { + t.Fatal("an unresolved row must say why") + } +} + +// The key is a function of the state change. If it ever becomes a function of +// the attempt, every retry looks like a new charge and the whole mechanism is +// decorative. +func TestIdempotencyKeyIsStableAcrossAttempts(t *testing.T) { + start := time.Unix(1_700_000_000, 0) + first := outbox.UsageKey("acme", start) + second := outbox.UsageKey("acme", start.In(time.FixedZone("elsewhere", 3600))) + if first != second { + t.Fatalf("the same window produced two keys: %q and %q", first, second) + } + if outbox.UsageKey("acme", start) == outbox.UsageKey("acme", start.Add(time.Minute)) { + t.Fatal("two different windows produced the same key") + } +} diff --git a/internal/outbox/relay.go b/internal/outbox/relay.go new file mode 100644 index 0000000..0492c3d --- /dev/null +++ b/internal/outbox/relay.go @@ -0,0 +1,317 @@ +package outbox + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "time" + + "github.com/jackc/pgx/v5/pgxpool" +) + +// Sink is the downstream. Deliver applies the effect; Lookup answers whether a +// key was already applied. +// +// Lookup is what makes an ambiguous outcome resolvable. Without it a message +// whose attempting process died can only be redelivered and hoped about; with +// it the producer can ask the one party that actually knows. +type Sink interface { + Deliver(ctx context.Context, m Message) (receipt string, err error) + Lookup(ctx context.Context, idempotencyKey string) (receipt string, found bool, err error) +} + +// ErrNoLookup is what a sink returns from Lookup when it cannot answer. A +// message whose sink cannot answer stays UNKNOWN; nothing guesses on its +// behalf. +var ErrNoLookup = errors.New("sink does not support lookup by idempotency key") + +// Relay moves messages out of the outbox. One Relay is one process's worth of +// delivery; several may run at once, and the lease is what keeps them off each +// other's rows. +type Relay struct { + Pool *pgxpool.Pool + Sink Sink + Owner string + // Lease is how long a claimed row is left alone before another pass decides + // its owner is gone. Too short and a slow but living delivery is declared + // ambiguous; too long and recovery after a crash waits. + Lease time.Duration + Batch int + Backoff func(attempt int) time.Duration + Crash Crasher + Now func() time.Time +} + +func (r *Relay) now() time.Time { + if r.Now != nil { + return r.Now() + } + return time.Now() +} + +func (r *Relay) backoff(attempt int) time.Duration { + if r.Backoff != nil { + return r.Backoff(attempt) + } + d := time.Duration(1<= max_attempts THEN 'FAILED' ELSE 'PENDING' END, + next_attempt_at = $2, + last_error = $3, + lease_owner = NULL, lease_expires_at = NULL, + updated_at = now() + WHERE id = $1 + RETURNING state`, m.ID, r.now().Add(r.backoff(m.Attempts)), cause.Error()).Scan(&state) + if err != nil { + return false, fmt.Errorf("recording failure of %d: %w", m.ID, err) + } + return state == StateFailed, nil +} + +// ReconcileResult reports how the ambiguous rows were settled. +type ReconcileResult struct { + Examined int + // Confirmed: the consumer holds the key, so the effect landed and the + // producer simply never heard it. + Confirmed int + // Absent: the consumer does not hold the key, so redelivery is safe and the + // row goes back to PENDING. + Absent int + // Unresolved: the consumer could not be asked. The row stays UNKNOWN. This + // number is the one worth alerting on. + Unresolved int +} + +// Reconcile settles UNKNOWN rows by asking the consumer whether it has the key. +// +// Neither outcome is assumed. A consumer that cannot be reached leaves the row +// where it is, and the row stays visible as unresolved for as long as that is +// true. The alternatives - assume delivered, or assume not delivered - are a +// silently lost charge and a silently duplicated one respectively, and the +// negative controls in the record show both happening. +func (r *Relay) Reconcile(ctx context.Context) (ReconcileResult, error) { + var res ReconcileResult + rows, err := r.Pool.Query(ctx, ` + SELECT id, idempotency_key FROM outbox WHERE state = 'UNKNOWN' ORDER BY id`) + if err != nil { + return res, fmt.Errorf("listing unknown rows: %w", err) + } + type pending struct { + id int64 + key string + } + var todo []pending + for rows.Next() { + var p pending + if err := rows.Scan(&p.id, &p.key); err != nil { + rows.Close() + return res, err + } + todo = append(todo, p) + } + rows.Close() + if err := rows.Err(); err != nil { + return res, err + } + + for _, p := range todo { + res.Examined++ + receipt, found, err := r.Sink.Lookup(ctx, p.key) + if err != nil { + res.Unresolved++ + if _, uerr := r.Pool.Exec(ctx, ` + UPDATE outbox SET last_error = $2, updated_at = now() WHERE id = $1`, + p.id, "unresolved: "+err.Error()); uerr != nil { + return res, uerr + } + continue + } + if found { + if err := r.recordDelivered(ctx, p.id, receipt, "confirmed_by_consumer_after_crash"); err != nil { + return res, err + } + res.Confirmed++ + continue + } + if _, err := r.Pool.Exec(ctx, ` + UPDATE outbox + SET state = 'PENDING', next_attempt_at = now(), + resolution = 'absent_at_consumer_after_crash', + last_error = NULL, updated_at = now() + WHERE id = $1`, p.id); err != nil { + return res, err + } + res.Absent++ + } + return res, nil +} + +// Counts reports the outbox by state, for the harness and for an operator. +func Counts(ctx context.Context, pool *pgxpool.Pool) (map[string]int, error) { + rows, err := pool.Query(ctx, `SELECT state, count(*) FROM outbox GROUP BY state`) + if err != nil { + return nil, err + } + defer rows.Close() + out := map[string]int{} + for rows.Next() { + var s string + var n int + if err := rows.Scan(&s, &n); err != nil { + return nil, err + } + out[s] = n + } + return out, rows.Err() +} + +func mustJSON(v any) json.RawMessage { + b, err := json.Marshal(v) + if err != nil { + panic(err) + } + return b +} diff --git a/internal/outbox/seal.go b/internal/outbox/seal.go new file mode 100644 index 0000000..fcc46f5 --- /dev/null +++ b/internal/outbox/seal.go @@ -0,0 +1,135 @@ +package outbox + +import ( + "context" + "fmt" + "time" + + "github.com/jackc/pgx/v5/pgxpool" +) + +// Window is one tenant's counters for one closed interval. +type Window struct { + TenantID string + WindowStart time.Time + WindowEnd time.Time + Requests int64 + Admitted int64 + Limited int64 + ServerErr int64 +} + +// CrashPoint names a place the process may be killed, for the crash tests. +// Nothing in the delivery path branches on it beyond calling the hook, so a nil +// hook - the production case - costs a nil check. +type CrashPoint string + +const ( + // CrashAfterCommit fires the instant the sealing transaction commits, so + // the process dies owning a committed ledger row and a committed message it + // has not begun to deliver. + CrashAfterCommit CrashPoint = "after_commit" + // CrashBeforeSend fires after the attempt is durably recorded and before + // the request leaves. The database cannot tell afterwards whether the + // request left; that is the point. + CrashBeforeSend CrashPoint = "before_send" + // CrashAfterSend fires after the consumer has answered and before the + // answer is recorded. The consumer has the effect; the producer does not + // know it. + CrashAfterSend CrashPoint = "after_send" +) + +// Crasher is called at a named point. Tests install one that really kills the +// process; production installs none. +type Crasher func(CrashPoint) + +func (c Crasher) at(p CrashPoint) { + if c != nil { + c(p) + } +} + +// SealWindows writes the ledger rows and the messages that report them in one +// transaction. +// +// This is the atomic part of "atomic side effects and delivery". Either the +// window is sealed and the charge is queued, or neither. There is no ordering +// of two writes to get wrong because there is only one write. +// +// Re-sealing the same window is a no-op on both tables, so the caller may retry +// freely; that is what makes a crash before the commit harmless. +func SealWindows(ctx context.Context, pool *pgxpool.Pool, windows []Window, crash Crasher) (int, error) { + if len(windows) == 0 { + return 0, nil + } + tx, err := pool.Begin(ctx) + if err != nil { + return 0, fmt.Errorf("beginning seal tx: %w", err) + } + defer tx.Rollback(ctx) //nolint:errcheck + + sealed := 0 + for _, w := range windows { + tag, err := tx.Exec(ctx, ` + INSERT INTO usage_ledger + (tenant_id, window_start, window_end, requests, admitted, limited, server_err) + VALUES ($1, $2, $3, $4, $5, $6, $7) + ON CONFLICT (tenant_id, window_start) DO NOTHING`, + w.TenantID, w.WindowStart, w.WindowEnd, w.Requests, w.Admitted, w.Limited, w.ServerErr) + if err != nil { + return 0, fmt.Errorf("sealing %s at %s: %w", w.TenantID, w.WindowStart, err) + } + if tag.RowsAffected() == 0 { + // Already sealed. The message that reports this window was written + // by the same transaction that wrote the row, so if the row is here + // the message was enqueued; re-enqueueing would be asking to be + // charged twice for a window already accounted for. + continue + } + if err := Enqueue(ctx, tx, UsageKey(w.TenantID, w.WindowStart), TopicUsageSealed, UsageEvent{ + TenantID: w.TenantID, + WindowStart: w.WindowStart, + WindowEnd: w.WindowEnd, + Requests: w.Requests, + Admitted: w.Admitted, + Limited: w.Limited, + ServerErr: w.ServerErr, + }); err != nil { + return 0, err + } + sealed++ + } + if err := tx.Commit(ctx); err != nil { + return 0, fmt.Errorf("committing seal: %w", err) + } + crash.at(CrashAfterCommit) + return sealed, nil +} + +// SealWindowsDualWrite is the version without an outbox, kept as the negative +// control for the crash tests: the ledger row commits, then the delivery is +// attempted separately. A process killed between the two has lost the charge +// permanently, and no restart recovers it because nothing recorded that it was +// owed. It is exported so the harness can run it; nothing in the gateway calls +// it. +func SealWindowsDualWrite(ctx context.Context, pool *pgxpool.Pool, windows []Window, sink Sink, crash Crasher) error { + for _, w := range windows { + if _, err := pool.Exec(ctx, ` + INSERT INTO usage_ledger + (tenant_id, window_start, window_end, requests, admitted, limited, server_err) + VALUES ($1, $2, $3, $4, $5, $6, $7) + ON CONFLICT (tenant_id, window_start) DO NOTHING`, + w.TenantID, w.WindowStart, w.WindowEnd, w.Requests, w.Admitted, w.Limited, w.ServerErr); err != nil { + return fmt.Errorf("dual-write ledger insert: %w", err) + } + crash.at(CrashAfterCommit) + if _, err := sink.Deliver(ctx, Message{ + IdempotencyKey: UsageKey(w.TenantID, w.WindowStart), + Topic: TopicUsageSealed, + Payload: mustJSON(UsageEvent{TenantID: w.TenantID, WindowStart: w.WindowStart, WindowEnd: w.WindowEnd, Requests: w.Requests, Admitted: w.Admitted, Limited: w.Limited, ServerErr: w.ServerErr}), + }); err != nil { + return fmt.Errorf("dual-write delivery: %w", err) + } + } + return nil +} diff --git a/internal/outbox/sink.go b/internal/outbox/sink.go new file mode 100644 index 0000000..d3da299 --- /dev/null +++ b/internal/outbox/sink.go @@ -0,0 +1,250 @@ +package outbox + +import ( + "bytes" + "context" + "crypto/rand" + "encoding/hex" + "encoding/json" + "errors" + "fmt" + "net/http" + "time" + + _ "embed" + + "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgxpool" +) + +//go:embed sink_schema.sql +var sinkSchema string + +// BillingSink is the consumer: it charges a tenant for a sealed usage window, +// and it is the party that decides how many times that charge lands. +// +// Dedupe is a switch because the negative control needs it off. With it off the +// sink is an ordinary at-least-once consumer and a redelivered message charges +// twice, which is what the crash tests show before turning it on. Nothing in +// the gateway ever constructs one with Dedupe false. +type BillingSink struct { + Pool *pgxpool.Pool + Dedupe bool +} + +// MigrateSink creates the consumer's own schema in the consumer's own database. +func MigrateSink(ctx context.Context, pool *pgxpool.Pool) error { + if _, err := pool.Exec(ctx, sinkSchema); err != nil { + return fmt.Errorf("creating sink schema: %w", err) + } + return nil +} + +// ApplyResult reports what the consumer did with one delivery. +type ApplyResult struct { + Receipt string `json:"receipt"` + Duplicate bool `json:"duplicate"` +} + +// Apply charges for a usage window, at most once per idempotency key. +// +// The inbox row and the charge are written in ONE transaction of THIS database. +// That is the entire dedup guarantee: if the charge is visible then so is the +// key that suppresses the next delivery of it, and if the key is visible then +// the charge it stands for was written by the same commit. Splitting them - a +// "have I seen this?" SELECT, then an INSERT - would leave a window where two +// concurrent deliveries both see nothing and both charge. +// +// billing_charges deliberately carries NO unique constraint on +// idempotency_key. If it did, Postgres would be doing the deduplication and +// this inbox would be decoration; the counts in the crash tests would prove a +// property of the schema rather than of the mechanism. +func (s *BillingSink) Apply(ctx context.Context, key, topic string, payload json.RawMessage) (ApplyResult, error) { + var ev UsageEvent + if err := json.Unmarshal(payload, &ev); err != nil { + return ApplyResult{}, fmt.Errorf("decoding %s: %w", topic, err) + } + if topic != TopicUsageSealed { + return ApplyResult{}, fmt.Errorf("unknown topic %q", topic) + } + + tx, err := s.Pool.Begin(ctx) + if err != nil { + return ApplyResult{}, err + } + defer tx.Rollback(ctx) //nolint:errcheck + + receipt := "rcpt_" + randomHex(12) + if s.Dedupe { + var existing string + err = tx.QueryRow(ctx, ` + INSERT INTO inbox (idempotency_key, topic, payload, receipt) + VALUES ($1, $2, $3, $4) + ON CONFLICT (idempotency_key) DO NOTHING + RETURNING receipt`, key, topic, payload, receipt).Scan(&existing) + if errors.Is(err, pgx.ErrNoRows) { + // The key is already here, so the charge it stands for is already + // here too, by the same commit. Report the original receipt and + // charge nothing. The counter says how loud the relay was being. + var prior string + if err := tx.QueryRow(ctx, ` + UPDATE inbox SET deliveries = deliveries + 1 + WHERE idempotency_key = $1 RETURNING receipt`, key).Scan(&prior); err != nil { + return ApplyResult{}, fmt.Errorf("reading prior receipt for %s: %w", key, err) + } + if err := tx.Commit(ctx); err != nil { + return ApplyResult{}, err + } + return ApplyResult{Receipt: prior, Duplicate: true}, nil + } + if err != nil { + return ApplyResult{}, fmt.Errorf("recording inbox row for %s: %w", key, err) + } + } else { + // Control mode: keep a record of the delivery but let it through. + if _, err := tx.Exec(ctx, ` + INSERT INTO inbox (idempotency_key, topic, payload, receipt) + VALUES ($1, $2, $3, $4) + ON CONFLICT (idempotency_key) DO UPDATE SET deliveries = inbox.deliveries + 1`, + key, topic, payload, receipt); err != nil { + return ApplyResult{}, err + } + } + + if _, err := tx.Exec(ctx, ` + INSERT INTO billing_charges (idempotency_key, tenant_id, window_start, requests, amount_cents) + VALUES ($1, $2, $3, $4, $5)`, + key, ev.TenantID, ev.WindowStart, ev.Requests, ev.Requests); err != nil { + return ApplyResult{}, fmt.Errorf("charging %s: %w", ev.TenantID, err) + } + if err := tx.Commit(ctx); err != nil { + return ApplyResult{}, err + } + return ApplyResult{Receipt: receipt, Duplicate: false}, nil +} + +// Receipt answers the reconciliation question: do you have this key? +func (s *BillingSink) Receipt(ctx context.Context, key string) (string, bool, error) { + var receipt string + err := s.Pool.QueryRow(ctx, `SELECT receipt FROM inbox WHERE idempotency_key = $1`, key).Scan(&receipt) + if errors.Is(err, pgx.ErrNoRows) { + return "", false, nil + } + if err != nil { + return "", false, err + } + return receipt, true, nil +} + +// Handler is the consumer's HTTP surface: one endpoint to deliver into, one to +// reconcile against. +func (s *BillingSink) Handler() http.Handler { + mux := http.NewServeMux() + mux.HandleFunc("POST /v1/charges", func(w http.ResponseWriter, req *http.Request) { + key := req.Header.Get("Idempotency-Key") + topic := req.Header.Get("X-Outbox-Topic") + if key == "" || topic == "" { + http.Error(w, "Idempotency-Key and X-Outbox-Topic are required", http.StatusBadRequest) + return + } + var body json.RawMessage + if err := json.NewDecoder(req.Body).Decode(&body); err != nil { + http.Error(w, err.Error(), http.StatusBadRequest) + return + } + res, err := s.Apply(req.Context(), key, topic, body) + if err != nil { + http.Error(w, err.Error(), http.StatusInternalServerError) + return + } + w.Header().Set("Content-Type", "application/json") + _ = json.NewEncoder(w).Encode(res) + }) + mux.HandleFunc("GET /v1/charges/{key}", func(w http.ResponseWriter, req *http.Request) { + receipt, found, err := s.Receipt(req.Context(), req.PathValue("key")) + if err != nil { + http.Error(w, err.Error(), http.StatusInternalServerError) + return + } + if !found { + http.Error(w, "not found", http.StatusNotFound) + return + } + w.Header().Set("Content-Type", "application/json") + _ = json.NewEncoder(w).Encode(ApplyResult{Receipt: receipt, Duplicate: true}) + }) + return mux +} + +// HTTPSink is the producer's view of a remote consumer. +type HTTPSink struct { + BaseURL string + Client *http.Client + // LookupDisabled models a downstream that offers no way to ask whether it + // holds a key. Reconciliation then cannot settle anything and says so. + LookupDisabled bool +} + +func (h HTTPSink) client() *http.Client { + if h.Client != nil { + return h.Client + } + return &http.Client{Timeout: 10 * time.Second} +} + +func (h HTTPSink) Deliver(ctx context.Context, m Message) (string, error) { + req, err := http.NewRequestWithContext(ctx, http.MethodPost, h.BaseURL+"/v1/charges", bytes.NewReader(m.Payload)) + if err != nil { + return "", err + } + req.Header.Set("Idempotency-Key", m.IdempotencyKey) + req.Header.Set("X-Outbox-Topic", m.Topic) + req.Header.Set("Content-Type", "application/json") + resp, err := h.client().Do(req) + if err != nil { + return "", err + } + defer resp.Body.Close() + if resp.StatusCode/100 != 2 { + return "", fmt.Errorf("sink returned %s", resp.Status) + } + var out ApplyResult + if err := json.NewDecoder(resp.Body).Decode(&out); err != nil { + return "", err + } + return out.Receipt, nil +} + +func (h HTTPSink) Lookup(ctx context.Context, key string) (string, bool, error) { + if h.LookupDisabled { + return "", false, ErrNoLookup + } + req, err := http.NewRequestWithContext(ctx, http.MethodGet, h.BaseURL+"/v1/charges/"+key, nil) + if err != nil { + return "", false, err + } + resp, err := h.client().Do(req) + if err != nil { + return "", false, err + } + defer resp.Body.Close() + if resp.StatusCode == http.StatusNotFound { + return "", false, nil + } + if resp.StatusCode/100 != 2 { + return "", false, fmt.Errorf("sink lookup returned %s", resp.Status) + } + var out ApplyResult + if err := json.NewDecoder(resp.Body).Decode(&out); err != nil { + return "", false, err + } + return out.Receipt, true, nil +} + +func randomHex(n int) string { + b := make([]byte, n) + if _, err := rand.Read(b); err != nil { + panic(err) + } + return hex.EncodeToString(b) +} diff --git a/internal/outbox/sink_schema.sql b/internal/outbox/sink_schema.sql new file mode 100644 index 0000000..34c4620 --- /dev/null +++ b/internal/outbox/sink_schema.sql @@ -0,0 +1,34 @@ +-- Consumer side. This lives in the SINK's own database, not the gateway's, so +-- that "the effect and the inbox row commit together" is a claim about one +-- transaction in one database and not a trick played across two. + +BEGIN; + +-- The inbox is the deduplication record. A key present here means this sink has +-- already applied the effect for that message, whatever the producer believes. +CREATE TABLE IF NOT EXISTS inbox ( + idempotency_key TEXT PRIMARY KEY, + topic TEXT NOT NULL, + payload JSONB NOT NULL, + receipt TEXT NOT NULL, + received_at TIMESTAMPTZ NOT NULL DEFAULT now(), + -- Every redelivery bumps this. It is the measure of how much duplicate + -- traffic the at-least-once relay actually produced. + deliveries INT NOT NULL DEFAULT 1 +); + +-- The effect. Exactly this table is what "landed once" is counted from. +CREATE TABLE IF NOT EXISTS billing_charges ( + id BIGSERIAL PRIMARY KEY, + idempotency_key TEXT NOT NULL REFERENCES inbox(idempotency_key), + tenant_id TEXT NOT NULL, + window_start TIMESTAMPTZ NOT NULL, + requests BIGINT NOT NULL, + amount_cents BIGINT NOT NULL, + charged_at TIMESTAMPTZ NOT NULL DEFAULT now() +); + +CREATE INDEX IF NOT EXISTS billing_charges_tenant_idx + ON billing_charges (tenant_id, window_start); + +COMMIT; diff --git a/internal/reqctx/reqctx.go b/internal/reqctx/reqctx.go index 970f1b6..0391768 100644 --- a/internal/reqctx/reqctx.go +++ b/internal/reqctx/reqctx.go @@ -36,10 +36,15 @@ func (i *Info) RouteLabel() string { return i.RoutePrefix } -// TenantLabel returns the tenant for metrics, or "unauthenticated". +// UnauthenticatedTenant is the label carried by requests that never resolved to +// a tenant: health probes, unmatched paths, rejected keys. It is a metrics +// bucket and not an account, so anything that bills a tenant has to exclude it. +const UnauthenticatedTenant = "unauthenticated" + +// TenantLabel returns the tenant for metrics, or UnauthenticatedTenant. func (i *Info) TenantLabel() string { if i.TenantID == "" { - return "unauthenticated" + return UnauthenticatedTenant } return i.TenantID } diff --git a/migrations/004_outbox.sql b/migrations/004_outbox.sql new file mode 100644 index 0000000..c070297 --- /dev/null +++ b/migrations/004_outbox.sql @@ -0,0 +1,72 @@ +-- Producer side of the transactional outbox. +-- +-- The gateway's usage numbers lived only in this process's Prometheus registry, +-- so "what did tenant X burn between 10:00 and 10:01" had no durable answer and +-- nothing downstream could be billed for it. Sealing a window writes the ledger +-- row and the message that reports it in ONE transaction; a relay delivers the +-- message afterwards. The point is that there is no moment where the ledger row +-- exists and the intent to report it does not. + +BEGIN; + +-- One row per tenant per closed window. The primary key is what makes sealing +-- the same window twice a no-op rather than a double count. +CREATE TABLE IF NOT EXISTS usage_ledger ( + tenant_id TEXT NOT NULL, + window_start TIMESTAMPTZ NOT NULL, + window_end TIMESTAMPTZ NOT NULL, + requests BIGINT NOT NULL DEFAULT 0 CHECK (requests >= 0), + admitted BIGINT NOT NULL DEFAULT 0 CHECK (admitted >= 0), + limited BIGINT NOT NULL DEFAULT 0 CHECK (limited >= 0), + server_err BIGINT NOT NULL DEFAULT 0 CHECK (server_err >= 0), + sealed_at TIMESTAMPTZ NOT NULL DEFAULT now(), + PRIMARY KEY (tenant_id, window_start), + CHECK (window_end > window_start) +); + +CREATE TABLE IF NOT EXISTS outbox ( + id BIGSERIAL PRIMARY KEY, + -- Derived from the state change, never from the attempt, so every retry of + -- the same fact carries the same key. This is the whole basis of the + -- consumer's ability to deduplicate. + idempotency_key TEXT NOT NULL UNIQUE, + topic TEXT NOT NULL, + payload JSONB NOT NULL, + -- PENDING : never attempted, or an attempt is known to have failed. + -- INFLIGHT : an attempt was durably recorded and has not reported back. + -- UNKNOWN : the attempting process died. Whether the consumer received + -- it is not knowable from this database, and nothing here will + -- guess. Reconciliation asks the consumer. + -- DELIVERED : the consumer acknowledged, and its receipt is stored. + -- FAILED : attempts exhausted. The effect has NOT landed; this is a + -- deliberate stop, not a success. + state TEXT NOT NULL DEFAULT 'PENDING' + CHECK (state IN ('PENDING','INFLIGHT','UNKNOWN','DELIVERED','FAILED')), + attempts INT NOT NULL DEFAULT 0 CHECK (attempts >= 0), + max_attempts INT NOT NULL DEFAULT 10 CHECK (max_attempts > 0), + next_attempt_at TIMESTAMPTZ NOT NULL DEFAULT now(), + -- A lease, not a lock: a process that is SIGKILLed cannot release anything, + -- so the lease has to expire on its own or the row is stranded forever. + lease_owner TEXT, + lease_expires_at TIMESTAMPTZ, + last_attempt_at TIMESTAMPTZ, + last_error TEXT, + -- How an UNKNOWN row was settled, kept so the settlement is auditable + -- rather than inferred from the state alone. + resolution TEXT, + receipt TEXT, + delivered_at TIMESTAMPTZ, + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + updated_at TIMESTAMPTZ NOT NULL DEFAULT now(), + -- A delivered row without a receipt would be a claim with no evidence. + CHECK (state <> 'DELIVERED' OR receipt IS NOT NULL) +); + +CREATE INDEX IF NOT EXISTS outbox_due_idx + ON outbox (next_attempt_at) WHERE state = 'PENDING'; +CREATE INDEX IF NOT EXISTS outbox_lease_idx + ON outbox (lease_expires_at) WHERE state = 'INFLIGHT'; +CREATE INDEX IF NOT EXISTS outbox_unresolved_idx + ON outbox (id) WHERE state = 'UNKNOWN'; + +COMMIT; diff --git a/scripts/outbox-crash.sh b/scripts/outbox-crash.sh new file mode 100755 index 0000000..2f089f7 --- /dev/null +++ b/scripts/outbox-crash.sh @@ -0,0 +1,204 @@ +#!/usr/bin/env bash +# Kills the relay for real, between the transaction that commits a usage window +# and the delivery that reports it, and counts what the consumer ended up with. +# +# Nothing here mocks a failure. Every "crash" below is syscall.Kill(getpid(), +# SIGKILL) inside the process under test, so no deferred function runs, no +# buffer flushes, and no connection is closed politely. What survives is only +# what Postgres had already committed. +# +# Needs two databases. scripts/outbox-crash.sh expects them to exist and to be +# migrated; see RECORD_c13_outbox.md for the two commands that make them. +set -uo pipefail + +DB="${DATABASE_URL:?set DATABASE_URL to the gateway database}" +BILLING="${BILLING_URL:?set BILLING_URL to the consumer database}" +SINK_ADDR="${SINK_ADDR:-127.0.0.1:9411}" +SINK_URL="http://${SINK_ADDR}" +# Built here rather than assumed, so a run can never exercise a binary older +# than the source it is meant to be testing. +BIN="${BIN:-}" +PGC="${PGCONTAINER:-c13-pg-tollgate}" +PGUSER="${PGUSER:-tollgate}" + +if [ -z "$BIN" ]; then + BIN="$(mktemp -d)/tollgate-outbox" + go build -o "$BIN" ./cmd/tollgate-outbox || exit 1 + echo "built $BIN sha256 $(shasum -a 256 "$BIN" | cut -d' ' -f1)" +fi + +pass=0; fail=0 +sink_pid="" + +q() { docker exec "$PGC" psql -qtAX -U "$PGUSER" -d tollgate -c "$1"; } +qb() { docker exec "$PGC" psql -qtAX -U "$PGUSER" -d billing -c "$1"; } + +start_sink() { # $1 = dedupe true|false + stop_sink + $BIN sink -db "$BILLING" -addr "$SINK_ADDR" -dedupe="$1" >/tmp/c13-sink.log 2>&1 & + sink_pid=$! + for _ in $(seq 1 50); do + curl -fsS -o /dev/null "${SINK_URL}/v1/charges/probe" 2>/dev/null && break + curl -sS -o /dev/null -w '%{http_code}' "${SINK_URL}/v1/charges/probe" 2>/dev/null | grep -q 404 && break + sleep 0.1 + done +} +stop_sink() { [ -n "$sink_pid" ] && kill "$sink_pid" 2>/dev/null; wait "$sink_pid" 2>/dev/null; sink_pid=""; } +trap stop_sink EXIT + +reset() { + q "TRUNCATE outbox, usage_ledger" >/dev/null + qb "TRUNCATE billing_charges, inbox CASCADE" >/dev/null +} + +# Runs a command, reports its exit status, and says plainly whether the kernel +# killed it. 137 is 128+SIGKILL and is what a real kill looks like from here. +run_and_report() { + local label="$1"; shift + "$@" >/tmp/c13-cmd.log 2>&1 + local rc=$? + printf ' %-28s exit=%s%s\n' "$label" "$rc" \ + "$( [ $rc -eq 137 ] && echo ' (SIGKILL: 128+9)' )" + sed 's/^/ | /' /tmp/c13-cmd.log + return $rc +} + +check() { # name expected actual + if [ "$2" = "$3" ]; then printf ' PASS %s: %s\n' "$1" "$3"; pass=$((pass+1)); + else printf ' FAIL %s: expected %s, got %s\n' "$1" "$2" "$3"; fail=$((fail+1)); fi +} + +charges() { qb "SELECT count(*) FROM billing_charges WHERE idempotency_key='$1'"; } +obstate() { q "SELECT state FROM outbox WHERE idempotency_key='$1'"; } +obres() { q "SELECT coalesce(resolution,'-') FROM outbox WHERE idempotency_key='$1'"; } +obatt() { q "SELECT attempts FROM outbox WHERE idempotency_key='$1'"; } +deliv() { qb "SELECT coalesce(max(deliveries)::text,'0') FROM inbox WHERE idempotency_key='$1'"; } + +W() { date -u -r $((1750000000 + $1 * 60)) +%Y-%m-%dT%H:%M:%SZ; } + +banner() { echo; echo "=================================================================="; echo "$1"; echo "=================================================================="; } + +# ---------------------------------------------------------------- CONTROL 1 -- +# No outbox. The ledger commits, then delivery is attempted separately. Killing +# the process between them is the bug the outbox exists to remove; if this does +# NOT lose the charge, the crash is not landing where it claims to. +banner "CONTROL 1 dual write, no outbox: SIGKILL between commit and delivery" +reset; start_sink true +KEY="usage:acme:$((1750000000 + 60))" +run_and_report "seal -mode=dual-write" $BIN seal -db "$DB" -tenant acme -window "$(W 1)" -requests 120 -mode dual-write -sink "$SINK_URL" -crash after_commit +check "ledger row committed" 1 "$(q "SELECT count(*) FROM usage_ledger WHERE tenant_id='acme'")" +check "outbox rows" 0 "$(q "SELECT count(*) FROM outbox")" +echo " restarting: nothing recorded that a charge was owed, so there is nothing to relay" +run_and_report "relay (recovery attempt)" $BIN relay -db "$DB" -sink "$SINK_URL" -once +check "charges after recovery" 0 "$(charges "$KEY")" +echo " -> the side effect is lost permanently. This is the control." + +# ------------------------------------------------------------------ TEST 1 --- +banner "TEST 1 outbox: SIGKILL right after the sealing transaction commits" +reset +KEY="usage:acme:$((1750000000 + 120))" +run_and_report "seal -crash=after_commit" $BIN seal -db "$DB" -tenant acme -window "$(W 2)" -requests 200 -mode outbox -crash after_commit +check "ledger row committed" 1 "$(q "SELECT count(*) FROM usage_ledger WHERE window_start='$(W 2)'")" +check "outbox row committed" 1 "$(q "SELECT count(*) FROM outbox WHERE idempotency_key='$KEY'")" +check "outbox state" PENDING "$(obstate "$KEY")" +check "charges before relay" 0 "$(charges "$KEY")" +run_and_report "relay (restart)" $BIN relay -db "$DB" -sink "$SINK_URL" -once +check "outbox state" DELIVERED "$(obstate "$KEY")" +check "charges" 1 "$(charges "$KEY")" + +# ------------------------------------------------------------------ TEST 2 --- +# The relay recorded that attempt 1 began, then died before the request left. +# From the gateway database alone this is indistinguishable from TEST 3. +banner "TEST 2 SIGKILL after the attempt is recorded, before the request is sent" +KEY="usage:acme:$((1750000000 + 180))" +run_and_report "seal" $BIN seal -db "$DB" -tenant acme -window "$(W 3)" -requests 300 +run_and_report "relay -crash=before_send" $BIN relay -db "$DB" -sink "$SINK_URL" -once -lease 1ms -crash before_send +check "outbox state" INFLIGHT "$(obstate "$KEY")" +check "attempts recorded" 1 "$(obatt "$KEY")" +check "charges" 0 "$(charges "$KEY")" +echo " the row says attempt 1 began and nothing else. That is all that is true." +run_and_report "reconcile (lease=0)" $BIN reconcile -db "$DB" -sink "$SINK_URL" -lease 0s +check "outbox state after reconcile" PENDING "$(obstate "$KEY")" +check "resolution" absent_at_consumer_after_crash "$(obres "$KEY")" +run_and_report "relay (redeliver)" $BIN relay -db "$DB" -sink "$SINK_URL" -once +check "outbox state" DELIVERED "$(obstate "$KEY")" +check "charges" 1 "$(charges "$KEY")" + +# ------------------------------------------------------------------ TEST 3 --- +# The genuinely ambiguous one: the consumer has the effect, the producer does +# not know it. Same INFLIGHT row as TEST 2, opposite truth. +banner "TEST 3 SIGKILL after the consumer applied, before the producer recorded it" +KEY="usage:acme:$((1750000000 + 240))" +run_and_report "seal" $BIN seal -db "$DB" -tenant acme -window "$(W 4)" -requests 400 +run_and_report "relay -crash=after_send" $BIN relay -db "$DB" -sink "$SINK_URL" -once -lease 1ms -crash after_send +check "outbox state" INFLIGHT "$(obstate "$KEY")" +check "charges (consumer already has it)" 1 "$(charges "$KEY")" +echo " the producer's row is byte-for-byte the TEST 2 shape, and the truth is the opposite." +run_and_report "reconcile (lease=0)" $BIN reconcile -db "$DB" -sink "$SINK_URL" -lease 0s +check "outbox state" DELIVERED "$(obstate "$KEY")" +check "resolution" confirmed_by_consumer_after_crash "$(obres "$KEY")" +check "charges (still one)" 1 "$(charges "$KEY")" + +# ------------------------------------------------------------------ TEST 4 --- +banner "TEST 4 replay: force every delivered row back to PENDING and rerun the relay" +before=$(qb "SELECT count(*) FROM billing_charges") +q "UPDATE outbox SET state='PENDING', next_attempt_at=now(), receipt=NULL, delivered_at=NULL WHERE state='DELIVERED'" >/dev/null +run_and_report "relay pass 1" $BIN relay -db "$DB" -sink "$SINK_URL" -once +q "UPDATE outbox SET state='PENDING', next_attempt_at=now(), receipt=NULL, delivered_at=NULL WHERE state='DELIVERED'" >/dev/null +run_and_report "relay pass 2" $BIN relay -db "$DB" -sink "$SINK_URL" -once +q "UPDATE outbox SET state='PENDING', next_attempt_at=now(), receipt=NULL, delivered_at=NULL WHERE state='DELIVERED'" >/dev/null +run_and_report "relay pass 3" $BIN relay -db "$DB" -sink "$SINK_URL" -once +after=$(qb "SELECT count(*) FROM billing_charges") +check "charges unchanged by 3 replays" "$before" "$after" +check "deliveries counted at the consumer (at-least-once is real)" 4 "$(deliv "$KEY")" + +# ---------------------------------------------------------------- CONTROL 2 -- +# Turn the inbox off and replay the same messages. If the count does not double, +# the dedupe being measured is coming from somewhere other than the inbox. +banner "CONTROL 2 inbox off: the same replay duplicates every charge" +start_sink false +before=$(qb "SELECT count(*) FROM billing_charges") +q "UPDATE outbox SET state='PENDING', next_attempt_at=now(), receipt=NULL, delivered_at=NULL WHERE state='DELIVERED'" >/dev/null +run_and_report "relay (dedupe=off)" $BIN relay -db "$DB" -sink "$SINK_URL" -once +after=$(qb "SELECT count(*) FROM billing_charges") +check "charges doubled" "$((before * 2))" "$after" +echo " -> the inbox, not the schema and not the relay, is what makes the effect land once." +start_sink true + +# ---------------------------------------------------------------- CONTROL 3 -- +# A downstream that cannot be asked. Reconciliation must leave the row alone. +banner "CONTROL 3 a consumer with no lookup: the row stays UNKNOWN, unresolved" +reset +KEY="usage:acme:$((1750000000 + 300))" +run_and_report "seal" $BIN seal -db "$DB" -tenant acme -window "$(W 5)" -requests 500 +run_and_report "relay -crash=before_send" $BIN relay -db "$DB" -sink "$SINK_URL" -once -lease 1ms -crash before_send +run_and_report "reconcile -no-lookup" $BIN reconcile -db "$DB" -sink "$SINK_URL" -lease 0s -no-lookup +check "state stays UNKNOWN" UNKNOWN "$(obstate "$KEY")" +check "charges still zero" 0 "$(charges "$KEY")" +echo " last_error: $(q "SELECT last_error FROM outbox WHERE idempotency_key='$KEY'")" +echo " -> nothing was guessed in either direction." + +# ---------------------------------------------------------------- CONTROL 4 -- +# What the two silent resolutions actually cost, shown rather than argued. +banner "CONTROL 4 the two ways of guessing, and what each one costs" +echo " (a) assume delivered: mark the UNKNOWN row DELIVERED without asking" +q "UPDATE outbox SET state='DELIVERED', receipt='assumed', resolution='ASSUMED_DELIVERED' WHERE idempotency_key='$KEY'" >/dev/null +check "outbox says delivered" DELIVERED "$(obstate "$KEY")" +check "charges the consumer actually holds" 0 "$(charges "$KEY")" +echo " -> a charge the ledger says was billed and nobody was billed for. Silent loss." +echo " (b) assume not delivered, with the inbox off: redeliver TEST 3's key" +start_sink false +K3="usage:acme:$((1750000000 + 240))" +qb "TRUNCATE billing_charges, inbox CASCADE" >/dev/null +q "TRUNCATE outbox, usage_ledger" >/dev/null +run_and_report "seal" $BIN seal -db "$DB" -tenant acme -window "$(W 4)" -requests 400 +run_and_report "relay -crash=after_send" $BIN relay -db "$DB" -sink "$SINK_URL" -once -lease 1ms -crash after_send +c1=$(charges "$K3") +q "UPDATE outbox SET state='PENDING', next_attempt_at=now() WHERE idempotency_key='$K3'" >/dev/null +run_and_report "relay (assumed not delivered)" $BIN relay -db "$DB" -sink "$SINK_URL" -once +check "charges after guessing wrong" 2 "$(charges "$K3")" +echo " -> the tenant is billed twice. Silent duplication." +start_sink true + +banner "RESULT pass=$pass fail=$fail" +[ "$fail" -eq 0 ]