diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 43951bcc..0a5795ed 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -123,6 +123,15 @@ jobs: working-directory: adcp/v3 run: go test -race -count=1 ./... + # Unit tests only (sqlmock-backed) — no Docker in this job. The + # Postgres-testcontainer suite is gated behind `-tags=integration` + # (see adcp/v3/signing/pgreplay/README.md) and runs locally / wherever + # Docker is available, matching registry/redisstore's and + # registry/glidestore's integration test convention. + - name: Test adcp/v3/signing/pgreplay module + working-directory: adcp/v3/signing/pgreplay + run: go test -race -count=1 ./... + - name: Test targeting module working-directory: targeting run: go test -race -count=1 ./... diff --git a/adcp/v3/signing/README.md b/adcp/v3/signing/README.md index d1f669b3..c864f12b 100644 --- a/adcp/v3/signing/README.md +++ b/adcp/v3/signing/README.md @@ -147,7 +147,11 @@ For custom deployments, implement the two-method `JWKSResolver` interface. ## Replay cache -`NewMemoryReplayStore(perKeyIDCap)` — LRU with TTL eviction and the profile's per-keyid entry cap (default 1,000,000). Implement the three-method `ReplayStore` interface to plug in Redis or another shared store for distributed deployments; the spec requires the step-13 insert to be atomic with a cap check in distributed setups to prevent cap drift. +`NewMemoryReplayStore(perKeyIDCap)` — LRU with TTL eviction and the profile's per-keyid entry cap (default 1,000,000). Fine for a single verifier process; a per-process cache cannot enforce RFC 9421 §11.1's replay-rejection MUST once a verifier runs as more than one process behind a load balancer, since a captured signature replayed against a sibling instance's cache is accepted. + +For multi-instance deployments, use [`adcp/v3/signing/pgreplay`](pgreplay/) — a Postgres-backed `ReplayStore` (own `go.mod`, zero third-party deps beyond `database/sql` in production code) that every verifier instance shares via one `adcp_replay_cache` table, closing that gap. See its package doc for the eager-connection-probe behavior and the test-vs-production wiring pattern. + +Implementing the three-method `ReplayStore` interface directly also plugs in Redis or another shared store; the spec requires the step-13 insert to be atomic with a cap check in distributed setups to prevent cap drift. ## Key generation diff --git a/adcp/v3/signing/pgreplay/README.md b/adcp/v3/signing/pgreplay/README.md new file mode 100644 index 00000000..35520513 --- /dev/null +++ b/adcp/v3/signing/pgreplay/README.md @@ -0,0 +1,112 @@ +# adcp/v3/signing/pgreplay + +Postgres-backed `adcp/v3/signing.ReplayStore` for multi-instance RFC 9421 verifier deployments. + +Closes [adcontextprotocol/adcp-go#105](https://github.com/adcontextprotocol/adcp-go/issues/105) and [#54](https://github.com/adcontextprotocol/adcp-go/issues/54). Cross-validated against the JS SDK's `PostgresReplayStore` ([adcp-client#1018](https://github.com/adcontextprotocol/adcp-client/pull/1018), in production at agenticadvertising.org) and Python's `adcp-client-python` (`src/adcp/signing/pg/replay_store.py`) — same `(keyid, scope, nonce)` schema, same atomic `INSERT ... ON CONFLICT DO NOTHING` idiom. + +**Module note:** both issues' text names the pre-v3 path (`adcp/signing/replay.go`). The root README's "Modules & versioning" section and `MIGRATING.md` say that module is frozen at v2.1.1, security-backports-only — `adcp/v3/signing` is the actively developed one, so this package lives there. `PostgresReplayStore` doesn't import either `signing` package (see below), so it's a structurally valid `ReplayStore` for the frozen `adcp/signing` too, byte-for-byte identical to `adcp/v3/signing.ReplayStore`'s method set as of this writing, if you haven't migrated to v3 yet. + +## Why a separate module + +`adcp/v3/signing.MemoryReplayStore` dedups `(keyid, nonce)` per process. That's sufficient for one verifier process; it can't deliver RFC 9421 §11.1's replay-rejection MUST once a verifier runs as ≥2 processes behind a load balancer — a captured signature replayed against a sibling instance whose in-memory cache hasn't seen the nonce is accepted. `PostgresReplayStore` gives every instance one shared table instead of N independent caches. + +This directory is its own Go module. Production code here imports only `database/sql` — no driver is linked in, so it doesn't strictly *need* isolation to stay dependency-free (`adcp/idempotency`'s own Postgres adapter lives inside the shared `adcp` module today, for exactly that reason: `database/sql` alone doesn't threaten the zero-dep guarantee). The module boundary exists because #54 asked for one explicitly, as a structural, compiler-enforced guarantee that this package can never accidentally grow a real dependency that leaks into core `signing`'s import graph. + +## Install + +```bash +cd your-project +go get github.com/adcontextprotocol/adcp-go/adcp/v3/signing/pgreplay +``` + +Bring your own driver (`github.com/jackc/pgx/v5/stdlib`, `github.com/lib/pq`, ...). + +## Usage + +```go +import ( + "database/sql" + + _ "github.com/jackc/pgx/v5/stdlib" + "github.com/adcontextprotocol/adcp-go/adcp/v3/signing" + "github.com/adcontextprotocol/adcp-go/adcp/v3/signing/pgreplay" +) + +db, err := sql.Open("pgx", os.Getenv("REPLAY_DATABASE_URL")) +if err != nil { + log.Fatal(err) +} + +// Apply once via your migration tooling: +// pgreplay.GetReplayStoreMigration() + +replay := pgreplay.NewPostgresReplayStore(db) // panics if db is unreachable + +mw := signing.Middleware(signing.MiddlewareOptions{ + Resolver: resolver, + Replay: replay, // satisfies signing.ReplayStore + // ... +}) +``` + +Run `pgreplay.SweepExpiredReplays(ctx, db)` on a cron / `pg_cron` job — this package doesn't schedule sweeping itself (out of scope per #105). + +## Test and local-dev environments + +`NewPostgresReplayStore` probes the connection eagerly and **panics** if it can't reach Postgres. That's deliberate — see the package doc comment for the production incident (adcp#3379) this is closing. **Do not construct a `PostgresReplayStore` in tests or local dev without a real reachable Postgres.** Use `signing.NewMemoryReplayStore(0)` there, gated behind an explicit production/staging environment check: + +```go +func replayStore() signing.ReplayStore { + if os.Getenv("ENVIRONMENT") != "production" && os.Getenv("ENVIRONMENT") != "staging" { + return signing.NewMemoryReplayStore(0) + } + db, err := sql.Open("pgx", os.Getenv("REPLAY_DATABASE_URL")) + if err != nil { + log.Fatalf("replay store: %v", err) + } + return pgreplay.NewPostgresReplayStore(db) +} +``` + +## Distinguishing "rejected" from "database is down" + +`ReplayStore.Insert` returns a single `bool`, so this package's `Insert` — like `MemoryReplayStore.Insert` — cannot tell a caller "cap rejected" apart from "couldn't reach the DB" through its return value alone ([#54](https://github.com/adcontextprotocol/adcp-go/issues/54) raised this). `PostgresReplayStore` fails closed (returns `false`, i.e. reject) in both cases — a database outage must not silently disable replay rejection. To keep visibility into *which* case happened without changing the shared interface: + +- `InsertContext(ctx, keyid, nonce, ttl) (bool, error)` — returns a non-nil error, wrapping `ErrConnDown`, only when the Postgres round trip itself failed. +- `LastInsertError()`, `LastSeenError()`, `LastHitCapError()` — the most recent round-trip error observed by each `ReplayStore`-interface method, for health checks / alerting. + +Widening `signing.ReplayStore.Insert` itself to `(bool, error)` was considered and deliberately left out of this PR — see the PR description for the reasoning (it's a public interface with implementers outside this repo; a signature change here can't be verified against them). It's a natural, separately-reviewable follow-up. + +## Schema + +```sql +CREATE TABLE adcp_replay_cache ( + keyid TEXT NOT NULL, + scope TEXT NOT NULL, + nonce TEXT NOT NULL, + expires_at TIMESTAMPTZ NOT NULL, + PRIMARY KEY (keyid, scope, nonce) +); +CREATE INDEX idx_adcp_replay_cache_expires_at ON adcp_replay_cache(expires_at); +CREATE INDEX idx_adcp_replay_cache_keyid_scope_active ON adcp_replay_cache(keyid, scope, expires_at); +``` + +`scope` lets one Postgres pool back more than one RFC 9421 tag profile (`adcp/request-signing/v1`, `adcp/webhook-signing/v1`) without nonce-namespace collisions — construct one store per profile via `pgreplay.WithScope(profile.Tag)`. Defaults to `"default"`. + +## Testing + +```bash +go test -race -count=1 ./... # unit tests (sqlmock; no Docker needed) +go test -race -tags=integration -count=1 -v ./... # + real Postgres via testcontainers-go; needs Docker +``` + +The integration suite proves, against a live `postgres:16-alpine` container: + +- **The actual race this package exists to close**: 50 concurrent `Insert` calls for the identical `(keyid, scope, nonce)` yield exactly one winner. +- The `Seen`-then-`Insert` verifier flow under concurrency still yields exactly one winner at the `Insert` step even when multiple callers pass the `Seen` pre-check. +- `HitCap` is enforced against rows `Insert` actually wrote. +- `SweepExpiredReplays` removes only expired rows, leaving live ones untouched. +- `NewPostgresReplayStore` panics against a real (closed) connection, not just a mocked one. +- Two stores on the same database but different `scope`s don't observe each other's nonces. + +Integration tests are skipped, not failed, when Docker isn't reachable (`t.Skipf`), matching `registry/redisstore`'s and `registry/glidestore`'s convention. They are not wired into `ci.yml`'s default `go test ./...` step — `-tags=integration` opts in explicitly, same as those two packages today. diff --git a/adcp/v3/signing/pgreplay/doc.go b/adcp/v3/signing/pgreplay/doc.go new file mode 100644 index 00000000..be328bbc --- /dev/null +++ b/adcp/v3/signing/pgreplay/doc.go @@ -0,0 +1,103 @@ +// Package pgreplay implements a Postgres-backed adcp/v3/signing.ReplayStore +// for multi-instance RFC 9421 verifier deployments. +// +// # Which module this targets +// +// This package lives under adcp/v3/signing, the actively developed module +// for AdCP 3.x (see the repo root README's "Modules & versioning" section +// and MIGRATING.md): the legacy adcp/signing module is frozen at v2.1.1 and +// receives security backports only, and a new distributed-store feature is +// not a security backport. Both #105 and #54's issue text name the pre-v3 +// path (adcp/signing/replay.go) — written before the v3 split, the same way +// #53 did (see adcp/v3/signing/signingtest's own doc comment for that +// precedent). PostgresReplayStore doesn't import either signing package +// (see "Module boundary" below), so it is structurally a drop-in +// implementation of adcp/signing.ReplayStore too, byte-for-byte identical to +// adcp/v3/signing.ReplayStore's method set as of this writing — nothing +// here prevents wiring it into the frozen module if you're not yet on v3. +// +// # Why this exists +// +// adcp/v3/signing.MemoryReplayStore dedups (keyid, nonce) pairs in a +// per-process map. That is sufficient for a single verifier process, but RFC +// 9421 §11.1 makes replay rejection a MUST, and a per-process cache cannot +// deliver it once a verifier runs as more than one process behind a load +// balancer: a captured signature replayed against a sibling instance whose +// cache hasn't seen the nonce is accepted. PostgresReplayStore closes that +// gap by giving every verifier instance one shared table +// (adcp_replay_cache) instead of N independent in-memory caches. +// +// See https://github.com/adcontextprotocol/adcp-go/issues/105 and +// https://github.com/adcontextprotocol/adcp-go/issues/54, and the reference +// implementations this package is cross-validated against: the JS SDK's +// PostgresReplayStore (adcp-client#1018, adopted in production at +// agenticadvertising.org) and Python's adcp-client-python +// (src/adcp/signing/pg/replay_store.py). All three share the +// (keyid, scope, nonce) primary key and the same atomic +// INSERT ... ON CONFLICT DO NOTHING idiom for the nonce race. +// +// # Module boundary +// +// This package is its own Go module (this directory has its own go.mod) +// rather than living inside adcp/v3/signing or the shared adcp/v3 module. +// Production code here imports only database/sql from the standard library +// — no Postgres driver is linked in; callers supply an already-open *sql.DB +// wired to whichever driver they prefer (pgx stdlib adapter, lib/pq, ...). +// That means a separate module isn't required to keep production code +// dependency-free — database/sql alone would do that even living inside +// adcp/v3/signing, the way adcp/idempotency's PgBackend does today (it lives +// in the shared adcp module with zero external deps in its non-test files). +// This package uses a separate module anyway because +// https://github.com/adcontextprotocol/adcp-go/issues/54 asked for one +// explicitly, as a structural, compiler-enforced guarantee that pgreplay can +// never accidentally grow a real dependency (a pgx-specific error type +// check, say) that leaks into core signing's import graph. Test-only +// dependencies (testcontainers-go, go-sqlmock, a driver for integration +// tests) are scoped to this module's go.mod and never reach a consumer that +// only imports the package, exactly as with adcp/idempotency's test-only +// go-sqlmock dependency today. +// +// # Test environments — read this before wiring PostgresReplayStore anywhere +// +// NewPostgresReplayStore probes the connection eagerly and panics with an +// actionable message if it cannot reach Postgres. That is deliberate: the +// JS rollout (adcp#3379) learned that when a PostgresReplayStore is +// constructed against a pool that doesn't exist in the current environment, +// every signed request fails closed identically to "the verifier is +// broken" — a debugging trap if it isn't caught at wire-up. Failing fast in +// the constructor turns that into a startup-time error instead of a +// runtime mystery. +// +// The consequence: do not construct a PostgresReplayStore in tests, local +// dev, or any environment without a real reachable Postgres. Use +// signing.NewMemoryReplayStore(0) there. A typical wiring pattern: +// +// func replayStore() signing.ReplayStore { +// if os.Getenv("ENVIRONMENT") != "production" && os.Getenv("ENVIRONMENT") != "staging" { +// return signing.NewMemoryReplayStore(0) +// } +// db, err := sql.Open("pgx", os.Getenv("REPLAY_DATABASE_URL")) +// if err != nil { +// log.Fatalf("replay store: %v", err) +// } +// return pgreplay.NewPostgresReplayStore(db) +// } +// +// Gate the Postgres path on an explicit production/staging check, not on +// "did REPLAY_DATABASE_URL happen to get set" — a misconfigured production +// deployment should fail loudly, not silently fall back to a per-process +// cache that reintroduces the multi-instance replay gap. This mirrors the +// gated fallback the JS adopter shipped (getReplayStore(), gated on +// NODE_ENV !== 'production'). +// +// # Fail-closed posture +// +// HitCap, Seen, and Insert all fail closed: any Postgres error (connection +// down, query timeout, context canceled) is treated as "reject this +// request," never as "allow it through." This matches RFC 9421 §11.1's MUST +// and this SDK's broader safety-over-availability posture — an outage in +// the replay store degrades to rejecting signed requests, not to accepting +// unverified replays. See the doc comments on Insert and LastInsertError for +// how a caller can distinguish a legitimate rejection from a database +// outage without changing the fail-closed behavior itself. +package pgreplay diff --git a/adcp/v3/signing/pgreplay/go.mod b/adcp/v3/signing/pgreplay/go.mod new file mode 100644 index 00000000..1deb3d00 --- /dev/null +++ b/adcp/v3/signing/pgreplay/go.mod @@ -0,0 +1,67 @@ +module github.com/adcontextprotocol/adcp-go/adcp/v3/signing/pgreplay + +go 1.26.2 + +require ( + github.com/DATA-DOG/go-sqlmock v1.5.2 + github.com/jackc/pgx/v5 v5.10.0 + github.com/stretchr/testify v1.11.1 + github.com/testcontainers/testcontainers-go v0.44.0 +) + +require ( + dario.cat/mergo v1.0.2 // indirect + github.com/Azure/go-ansiterm v0.0.0-20250102033503-faa5f7b0171c // indirect + github.com/Microsoft/go-winio v0.6.2 // indirect + github.com/cenkalti/backoff/v4 v4.3.0 // indirect + github.com/cespare/xxhash/v2 v2.3.0 // indirect + github.com/containerd/errdefs v1.0.0 // indirect + github.com/containerd/errdefs/pkg v0.3.0 // indirect + github.com/containerd/log v0.1.0 // indirect + github.com/containerd/platforms v0.2.1 // indirect + github.com/cpuguy83/dockercfg v0.3.2 // indirect + github.com/davecgh/go-spew v1.1.1 // indirect + github.com/distribution/reference v0.6.0 // indirect + github.com/docker/go-connections v0.7.0 // indirect + github.com/docker/go-units v0.5.0 // indirect + github.com/ebitengine/purego v0.10.1 // indirect + github.com/felixge/httpsnoop v1.1.0 // indirect + github.com/go-logr/logr v1.4.3 // indirect + github.com/go-logr/stdr v1.2.2 // indirect + github.com/go-ole/go-ole v1.3.0 // indirect + github.com/google/uuid v1.6.0 // indirect + github.com/jackc/pgpassfile v1.0.0 // indirect + github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 // indirect + github.com/jackc/puddle/v2 v2.2.2 // indirect + github.com/klauspost/compress v1.18.6 // indirect + github.com/lufia/plan9stats v0.0.0-20260330125221-c963978e514e // indirect + github.com/magiconair/properties v1.8.10 // indirect + github.com/moby/docker-image-spec v1.3.1 // indirect + github.com/moby/go-archive v0.2.0 // indirect + github.com/moby/moby/api v1.55.0 // indirect + github.com/moby/moby/client v0.5.0 // indirect + github.com/moby/patternmatcher v0.6.1 // indirect + github.com/moby/sys/sequential v0.7.0 // indirect + github.com/moby/sys/user v0.4.0 // indirect + github.com/moby/sys/userns v0.1.0 // indirect + github.com/moby/term v0.5.2 // indirect + github.com/opencontainers/go-digest v1.0.0 // indirect + github.com/opencontainers/image-spec v1.1.1 // indirect + github.com/pmezard/go-difflib v1.0.0 // indirect + github.com/power-devops/perfstat v0.0.0-20240221224432-82ca36839d55 // indirect + github.com/shirou/gopsutil/v4 v4.26.6 // indirect + github.com/sirupsen/logrus v1.9.4 // indirect + github.com/tklauser/go-sysconf v0.4.0 // indirect + github.com/tklauser/numcpus v0.12.0 // indirect + github.com/yusufpapurcu/wmi v1.2.4 // indirect + go.opentelemetry.io/auto/sdk v1.2.1 // indirect + go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.69.0 // indirect + go.opentelemetry.io/otel v1.44.0 // indirect + go.opentelemetry.io/otel/metric v1.44.0 // indirect + go.opentelemetry.io/otel/trace v1.44.0 // indirect + golang.org/x/crypto v0.54.0 // indirect + golang.org/x/sync v0.22.0 // indirect + golang.org/x/sys v0.47.0 // indirect + golang.org/x/text v0.40.0 // indirect + gopkg.in/yaml.v3 v3.0.1 // indirect +) diff --git a/adcp/v3/signing/pgreplay/go.sum b/adcp/v3/signing/pgreplay/go.sum new file mode 100644 index 00000000..eb07b97b --- /dev/null +++ b/adcp/v3/signing/pgreplay/go.sum @@ -0,0 +1,155 @@ +dario.cat/mergo v1.0.2 h1:85+piFYR1tMbRrLcDwR18y4UKJ3aH1Tbzi24VRW1TK8= +dario.cat/mergo v1.0.2/go.mod h1:E/hbnu0NxMFBjpMIE34DRGLWqDy0g5FuKDhCb31ngxA= +github.com/AdaLogics/go-fuzz-headers v0.0.0-20240806141605-e8a1dd7889d6 h1:He8afgbRMd7mFxO99hRNu+6tazq8nFF9lIwo9JFroBk= +github.com/AdaLogics/go-fuzz-headers v0.0.0-20240806141605-e8a1dd7889d6/go.mod h1:8o94RPi1/7XTJvwPpRSzSUedZrtlirdB3r9Z20bi2f8= +github.com/Azure/go-ansiterm v0.0.0-20250102033503-faa5f7b0171c h1:udKWzYgxTojEKWjV8V+WSxDXJ4NFATAsZjh8iIbsQIg= +github.com/Azure/go-ansiterm v0.0.0-20250102033503-faa5f7b0171c/go.mod h1:xomTg63KZ2rFqZQzSB4Vz2SUXa1BpHTVz9L5PTmPC4E= +github.com/DATA-DOG/go-sqlmock v1.5.2 h1:OcvFkGmslmlZibjAjaHm3L//6LiuBgolP7OputlJIzU= +github.com/DATA-DOG/go-sqlmock v1.5.2/go.mod h1:88MAG/4G7SMwSE3CeA0ZKzrT5CiOU3OJ+JlNzwDqpNU= +github.com/Microsoft/go-winio v0.6.2 h1:F2VQgta7ecxGYO8k3ZZz3RS8fVIXVxONVUPlNERoyfY= +github.com/Microsoft/go-winio v0.6.2/go.mod h1:yd8OoFMLzJbo9gZq8j5qaps8bJ9aShtEA8Ipt1oGCvU= +github.com/cenkalti/backoff/v4 v4.3.0 h1:MyRJ/UdXutAwSAT+s3wNd7MfTIcy71VQueUuFK343L8= +github.com/cenkalti/backoff/v4 v4.3.0/go.mod h1:Y3VNntkOUPxTVeUxJ/G5vcM//AlwfmyYozVcomhLiZE= +github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs= +github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs= +github.com/containerd/errdefs v1.0.0 h1:tg5yIfIlQIrxYtu9ajqY42W3lpS19XqdxRQeEwYG8PI= +github.com/containerd/errdefs v1.0.0/go.mod h1:+YBYIdtsnF4Iw6nWZhJcqGSg/dwvV7tyJ/kCkyJ2k+M= +github.com/containerd/errdefs/pkg v0.3.0 h1:9IKJ06FvyNlexW690DXuQNx2KA2cUJXx151Xdx3ZPPE= +github.com/containerd/errdefs/pkg v0.3.0/go.mod h1:NJw6s9HwNuRhnjJhM7pylWwMyAkmCQvQ4GpJHEqRLVk= +github.com/containerd/log v0.1.0 h1:TCJt7ioM2cr/tfR8GPbGf9/VRAX8D2B4PjzCpfX540I= +github.com/containerd/log v0.1.0/go.mod h1:VRRf09a7mHDIRezVKTRCrOq78v577GXq3bSa3EhrzVo= +github.com/containerd/platforms v0.2.1 h1:zvwtM3rz2YHPQsF2CHYM8+KtB5dvhISiXh5ZpSBQv6A= +github.com/containerd/platforms v0.2.1/go.mod h1:XHCb+2/hzowdiut9rkudds9bE5yJ7npe7dG/wG+uFPw= +github.com/cpuguy83/dockercfg v0.3.2 h1:DlJTyZGBDlXqUZ2Dk2Q3xHs/FtnooJJVaad2S9GKorA= +github.com/cpuguy83/dockercfg v0.3.2/go.mod h1:sugsbF4//dDlL/i+S+rtpIWp+5h0BHJHfjj5/jFyUJc= +github.com/creack/pty v1.1.24 h1:bJrF4RRfyJnbTJqzRLHzcGaZK1NeM5kTC9jGgovnR1s= +github.com/creack/pty v1.1.24/go.mod h1:08sCNb52WyoAwi2QDyzUCTgcvVFhUzewun7wtTfvcwE= +github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= +github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c= +github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= +github.com/distribution/reference v0.6.0 h1:0IXCQ5g4/QMHHkarYzh5l+u8T3t73zM5QvfrDyIgxBk= +github.com/distribution/reference v0.6.0/go.mod h1:BbU0aIcezP1/5jX/8MP0YiH4SdvB5Y4f/wlDRiLyi3E= +github.com/docker/go-connections v0.7.0 h1:6SsRfJddP22WMrCkj19x9WKjEDTB+ahsdiGYf0mN39c= +github.com/docker/go-connections v0.7.0/go.mod h1:no1qkHdjq7kLMGUXYAduOhYPSJxxvgWBh7ogVvptn3Q= +github.com/docker/go-units v0.5.0 h1:69rxXcBk27SvSaaxTtLh/8llcHD8vYHT7WSdRZ/jvr4= +github.com/docker/go-units v0.5.0/go.mod h1:fgPhTUdO+D/Jk86RDLlptpiXQzgHJF7gydDDbaIK4Dk= +github.com/ebitengine/purego v0.10.1 h1:dewVBCBT2GaMu1SrNTYxQhgQBethzfhiwvZiLGP/qyY= +github.com/ebitengine/purego v0.10.1/go.mod h1:iIjxzd6CiRiOG0UyXP+V1+jWqUXVjPKLAI0mRfJZTmQ= +github.com/felixge/httpsnoop v1.1.0 h1:3YtUj32ZZkqZtt3sZZsClsymw/QDuVfpNhoA31zeORc= +github.com/felixge/httpsnoop v1.1.0/go.mod h1:Zqxgdd+1Rkcz8euOqdr7lqgCRJztwr5hp9vDSi5UZCE= +github.com/go-logr/logr v1.2.2/go.mod h1:jdQByPbusPIv2/zmleS9BjJVeZ6kBagPoEUsqbVz/1A= +github.com/go-logr/logr v1.4.3 h1:CjnDlHq8ikf6E492q6eKboGOC0T8CDaOvkHCIg8idEI= +github.com/go-logr/logr v1.4.3/go.mod h1:9T104GzyrTigFIr8wt5mBrctHMim0Nb2HLGrmQ40KvY= +github.com/go-logr/stdr v1.2.2 h1:hSWxHoqTgW2S2qGc0LTAI563KZ5YKYRhT3MFKZMbjag= +github.com/go-logr/stdr v1.2.2/go.mod h1:mMo/vtBO5dYbehREoey6XUKy/eSumjCCveDpRre4VKE= +github.com/go-ole/go-ole v1.2.6/go.mod h1:pprOEPIfldk/42T2oK7lQ4v4JSDwmV0As9GaiUsvbm0= +github.com/go-ole/go-ole v1.3.0 h1:Dt6ye7+vXGIKZ7Xtk4s6/xVdGDQynvom7xCFEdWr6uE= +github.com/go-ole/go-ole v1.3.0/go.mod h1:5LS6F96DhAwUc7C+1HLexzMXY1xGRSryjyPPKW6zv78= +github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8= +github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX3N/iU= +github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0= +github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= +github.com/jackc/pgpassfile v1.0.0 h1:/6Hmqy13Ss2zCq62VdNG8tM1wchn8zjSGOBJ6icpsIM= +github.com/jackc/pgpassfile v1.0.0/go.mod h1:CEx0iS5ambNFdcRtxPj5JhEz+xB6uRky5eyVu/W2HEg= +github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 h1:iCEnooe7UlwOQYpKFhBabPMi4aNAfoODPEFNiAnClxo= +github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761/go.mod h1:5TJZWKEWniPve33vlWYSoGYefn3gLQRzjfDlhSJ9ZKM= +github.com/jackc/pgx/v5 v5.10.0 h1:VhSvgU2jSli8o3AqIEOTJr7rZwAEUVo4E4XhR94Zfr0= +github.com/jackc/pgx/v5 v5.10.0/go.mod h1:mal1tBGAFfLHvZzaYh77YS/eC6IX9OWbRV1QIIM0Jn4= +github.com/jackc/puddle/v2 v2.2.2 h1:PR8nw+E/1w0GLuRFSmiioY6UooMp6KJv0/61nB7icHo= +github.com/jackc/puddle/v2 v2.2.2/go.mod h1:vriiEXHvEE654aYKXXjOvZM39qJ0q+azkZFrfEOc3H4= +github.com/kisielk/sqlstruct v0.0.0-20201105191214-5f3e10d3ab46/go.mod h1:yyMNCyc/Ib3bDTKd379tNMpB/7/H5TjM2Y9QJ5THLbE= +github.com/klauspost/compress v1.18.6 h1:2jupLlAwFm95+YDR+NwD2MEfFO9d4z4Prjl1XXDjuao= +github.com/klauspost/compress v1.18.6/go.mod h1:cwPg85FWrGar70rWktvGQj8/hthj3wpl0PGDogxkrSQ= +github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE= +github.com/kr/pretty v0.3.1/go.mod h1:hoEshYVHaxMs3cyo3Yncou5ZscifuDolrwPKZanG3xk= +github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY= +github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE= +github.com/lufia/plan9stats v0.0.0-20260330125221-c963978e514e h1:Q6MvJtQK/iRcRtzAscm/zF23XxJlbECiGPyRicsX+Ak= +github.com/lufia/plan9stats v0.0.0-20260330125221-c963978e514e/go.mod h1:autxFIvghDt3jPTLoqZ9OZ7s9qTGNAWmYCjVFWPX/zg= +github.com/magiconair/properties v1.8.10 h1:s31yESBquKXCV9a/ScB3ESkOjUYYv+X0rg8SYxI99mE= +github.com/magiconair/properties v1.8.10/go.mod h1:Dhd985XPs7jluiymwWYZ0G4Z61jb3vdS329zhj2hYo0= +github.com/moby/docker-image-spec v1.3.1 h1:jMKff3w6PgbfSa69GfNg+zN/XLhfXJGnEx3Nl2EsFP0= +github.com/moby/docker-image-spec v1.3.1/go.mod h1:eKmb5VW8vQEh/BAr2yvVNvuiJuY6UIocYsFu/DxxRpo= +github.com/moby/go-archive v0.2.0 h1:zg5QDUM2mi0JIM9fdQZWC7U8+2ZfixfTYoHL7rWUcP8= +github.com/moby/go-archive v0.2.0/go.mod h1:mNeivT14o8xU+5q1YnNrkQVpK+dnNe/K6fHqnTg4qPU= +github.com/moby/moby/api v1.55.0 h1:2/sexvQyqIWS8pRSCFddBfpW2qE7vR7FCL+vN8pxwMc= +github.com/moby/moby/api v1.55.0/go.mod h1:+RQ6wluLwtYaTd1WnPLykIDPekkuyD/ROWQClE83pzs= +github.com/moby/moby/client v0.5.0 h1:5XhyPk2fuOWf6RlSFa3MkIIgDZkF25xToXW8Q/BH7cc= +github.com/moby/moby/client v0.5.0/go.mod h1:rcVpF8ncl9vo5gaIBdol6CnbEtSj1uxMvEV/UrykF/s= +github.com/moby/patternmatcher v0.6.1 h1:qlhtafmr6kgMIJjKJMDmMWq7WLkKIo23hsrpR3x084U= +github.com/moby/patternmatcher v0.6.1/go.mod h1:hDPoyOpDY7OrrMDLaYoY3hf52gNCR/YOUYxkhApJIxc= +github.com/moby/sys/sequential v0.7.0 h1:ASQNGNROJSuOO6LL6bPHbKvuZu6NU8P4ldPWk31zj/8= +github.com/moby/sys/sequential v0.7.0/go.mod h1:NfSTAp6V3fw4tmkD62PEcOKeZKquXT8VKCkf7aVR79o= +github.com/moby/sys/user v0.4.0 h1:jhcMKit7SA80hivmFJcbB1vqmw//wU61Zdui2eQXuMs= +github.com/moby/sys/user v0.4.0/go.mod h1:bG+tYYYJgaMtRKgEmuueC0hJEAZWwtIbZTB+85uoHjs= +github.com/moby/sys/userns v0.1.0 h1:tVLXkFOxVu9A64/yh59slHVv9ahO9UIev4JZusOLG/g= +github.com/moby/sys/userns v0.1.0/go.mod h1:IHUYgu/kao6N8YZlp9Cf444ySSvCmDlmzUcYfDHOl28= +github.com/moby/term v0.5.2 h1:6qk3FJAFDs6i/q3W/pQ97SX192qKfZgGjCQqfCJkgzQ= +github.com/moby/term v0.5.2/go.mod h1:d3djjFCrjnB+fl8NJux+EJzu0msscUP+f8it8hPkFLc= +github.com/opencontainers/go-digest v1.0.0 h1:apOUWs51W5PlhuyGyz9FCeeBIOUDA/6nW8Oi/yOhh5U= +github.com/opencontainers/go-digest v1.0.0/go.mod h1:0JzlMkj0TRzQZfJkVvzbP0HBR3IKzErnv2BNG4W4MAM= +github.com/opencontainers/image-spec v1.1.1 h1:y0fUlFfIZhPF1W537XOLg0/fcx6zcHCJwooC2xJA040= +github.com/opencontainers/image-spec v1.1.1/go.mod h1:qpqAh3Dmcf36wStyyWU+kCeDgrGnAve2nCC8+7h8Q0M= +github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= +github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= +github.com/power-devops/perfstat v0.0.0-20240221224432-82ca36839d55 h1:o4JXh1EVt9k/+g42oCprj/FisM4qX9L3sZB3upGN2ZU= +github.com/power-devops/perfstat v0.0.0-20240221224432-82ca36839d55/go.mod h1:OmDBASR4679mdNQnz2pUhc2G8CO2JrUAVFDRBDP/hJE= +github.com/rogpeppe/go-internal v1.14.1 h1:UQB4HGPB6osV0SQTLymcB4TgvyWu6ZyliaW0tI/otEQ= +github.com/rogpeppe/go-internal v1.14.1/go.mod h1:MaRKkUm5W0goXpeCfT7UZI6fk/L7L7so1lCWt35ZSgc= +github.com/shirou/gopsutil/v4 v4.26.6 h1:Mzr/npDtQC/xpeEuQKHZt8Zo9CmPvhTj8nkR8w5TLDs= +github.com/shirou/gopsutil/v4 v4.26.6/go.mod h1:LZ6ewCSkBqUpvSOf+LsTGnRinC6iaNUNMGBtDkJBaLQ= +github.com/sirupsen/logrus v1.9.4 h1:TsZE7l11zFCLZnZ+teH4Umoq5BhEIfIzfRDZ1Uzql2w= +github.com/sirupsen/logrus v1.9.4/go.mod h1:ftWc9WdOfJ0a92nsE2jF5u5ZwH8Bv2zdeOC42RjbV2g= +github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME= +github.com/stretchr/objx v0.5.3 h1:jmXUvGomnU1o3W/V5h2VEradbpJDwGrzugQQvL0POH4= +github.com/stretchr/objx v0.5.3/go.mod h1:rDQraq+vQZU7Fde9LOZLr8Tax6zZvy4kuNKF+QYS+U0= +github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI= +github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg= +github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U= +github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U= +github.com/testcontainers/testcontainers-go v0.44.0 h1:/Fwh6HY1mIikhnm9e7HwoxGycx0lzRAE0f5VQpjFxzI= +github.com/testcontainers/testcontainers-go v0.44.0/go.mod h1:IcnwQrYTO86xHXu5bvMaBH7ATlbS3Qn1M1QWW3c66rE= +github.com/tklauser/go-sysconf v0.4.0 h1:7H0uAN+7RkwWRaxhYXDLqa5V3LPrJeV8wmD9dRUgPQU= +github.com/tklauser/go-sysconf v0.4.0/go.mod h1:8mTNWyog7H+MpKijp4VmKJAd2bbYQ2zuUwkYRbUArPI= +github.com/tklauser/numcpus v0.12.0 h1:NR85qdvHA9pFse3x3weVZ0r0ST8R6l5RHbZrlRaqob4= +github.com/tklauser/numcpus v0.12.0/go.mod h1:ABHeXzJnr/qqwguhClkZKT1/8VABcYrsyUiUGobwWJg= +github.com/yusufpapurcu/wmi v1.2.4 h1:zFUKzehAFReQwLys1b/iSMl+JQGSCSjtVqQn9bBrPo0= +github.com/yusufpapurcu/wmi v1.2.4/go.mod h1:SBZ9tNy3G9/m5Oi98Zks0QjeHVDvuK0qfxQmPyzfmi0= +go.opentelemetry.io/auto/sdk v1.2.1 h1:jXsnJ4Lmnqd11kwkBV2LgLoFMZKizbCi5fNZ/ipaZ64= +go.opentelemetry.io/auto/sdk v1.2.1/go.mod h1:KRTj+aOaElaLi+wW1kO/DZRXwkF4C5xPbEe3ZiIhN7Y= +go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.69.0 h1:8tvICD4vSTOOsNrsI4Ljf6C+6UKvpTEH5XY3JMoyPoo= +go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.69.0/go.mod h1:z9+yiacE0IHRqM4qFfkbt/JYlmYXgss8GY/jXoNuPJI= +go.opentelemetry.io/otel v1.44.0 h1:JjwHmHpA4iZ3wBxluu2fbbE7j4kqlE8jXyAyPXH7HqU= +go.opentelemetry.io/otel v1.44.0/go.mod h1:BMgjTHL9WPRlRjL2oZCBTL4whCGtXch2H4BhOPIAyYc= +go.opentelemetry.io/otel/metric v1.44.0 h1:1w0gILTcHdr3YI+ixLyjemwrVnsMURbTZFrSYCdDdmc= +go.opentelemetry.io/otel/metric v1.44.0/go.mod h1:8O7hanEPBNgEMmybD3s2VBKcgWOCsA6tzHBPODAiquo= +go.opentelemetry.io/otel/sdk v1.44.0 h1:nHYwb9lK+fJPU/dnT6s7W7Z8itMWyqrnVfbheVYrZ58= +go.opentelemetry.io/otel/sdk v1.44.0/go.mod h1:Osuydd3Se74nqjAKxid74N5eC+jfEqfTegHRnq58oK0= +go.opentelemetry.io/otel/sdk/metric v1.44.0 h1:3LlKgI+VjbVsjNRFZJZAJ30WjXC5VkNRks6si09iEfI= +go.opentelemetry.io/otel/sdk/metric v1.44.0/go.mod h1:5B5pMARnXxKhltooO4xUuCBorl65a4EpnTalObqOigA= +go.opentelemetry.io/otel/trace v1.44.0 h1:jxF5CsGYCe74MCRx2X4g7WsY/VBKRqqpNvXlX/6gtIk= +go.opentelemetry.io/otel/trace v1.44.0/go.mod h1:oLl1jrMQAVo6v3GAggN+1VH9VIz9iUSvW53sW1Q8PIE= +golang.org/x/crypto v0.54.0 h1:YLIA59K4fiNzHzjnZt2tUJQjQtUWfWbeHBqKtk3eScw= +golang.org/x/crypto v0.54.0/go.mod h1:KWL8ny2AZdGR2cWmzeHrp2azQPGogOv+HeQaVEXC2dk= +golang.org/x/sync v0.22.0 h1:SZjpbeLmrCk4xhRSZFNZW5gFUeCeFgjekvI/+gfScek= +golang.org/x/sync v0.22.0/go.mod h1:9xrNwdLfx4jkKbNva9FpL6vEN7evnE43NNNJQ2LF3+0= +golang.org/x/sys v0.0.0-20190916202348-b4ddaad3f8a3/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= +golang.org/x/sys v0.0.0-20201204225414-ed752295db88/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= +golang.org/x/sys v0.0.0-20210616094352-59db8d763f22/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= +golang.org/x/sys v0.1.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= +golang.org/x/sys v0.47.0 h1:o7XGOvZQCADBQQ4Y7VNq2dRWQR7JmOUW8Kxx4ZsNgWs= +golang.org/x/sys v0.47.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw= +golang.org/x/term v0.45.0 h1:NwWyBmoJCbfTHpxrWoZ9C6/VxOf7ic219I8xZZFdrf0= +golang.org/x/term v0.45.0/go.mod h1:9aqxs0blBcrm/n0L9QW0aRVD+ktan8ssZromtqJC43w= +golang.org/x/text v0.40.0 h1:Ub2Z6/xjgF1WrYQz2nuITOEegKFtiIy+rieRJ5lHZKs= +golang.org/x/text v0.40.0/go.mod h1:hpnzDAfGV753zIKo+wk3u1bVKCGPbrnF7+7LBF/UHVY= +gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= +gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c h1:Hei/4ADfdWqJk1ZMxUNpqntNwaWcugrBjAiHlqqRiVk= +gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c/go.mod h1:JHkPIbrfpd72SG/EVd6muEfDQjcINNoR0C8j2r3qZ4Q= +gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= +gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA= +gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= +gotest.tools/v3 v3.5.2 h1:7koQfIKdy+I8UTetycgUqXWSDwpgv193Ka+qRsmBY8Q= +gotest.tools/v3 v3.5.2/go.mod h1:LtdLGcnqToBH83WByAAi/wiwSFCArdFIUV/xxN4pcjA= +pgregory.net/rapid v1.2.0 h1:keKAYRcjm+e1F0oAuU5F5+YPAWcyxNNRK2wud503Gnk= +pgregory.net/rapid v1.2.0/go.mod h1:PY5XlDGj0+V1FCq0o192FdRhpKHGTRIWBgqjDBTrq04= diff --git a/adcp/v3/signing/pgreplay/integration_test.go b/adcp/v3/signing/pgreplay/integration_test.go new file mode 100644 index 00000000..3a676488 --- /dev/null +++ b/adcp/v3/signing/pgreplay/integration_test.go @@ -0,0 +1,279 @@ +//go:build integration + +// Run with: go test -race -tags=integration -count=1 -v ./... +// +// Requires Docker (spins up a real postgres:16-alpine container via +// testcontainers-go, mirroring registry/redisstore's and +// registry/glidestore's integration_test.go pattern). Skipped, not failed, +// when Docker isn't reachable. +package pgreplay + +import ( + "context" + "database/sql" + "fmt" + "sync" + "sync/atomic" + "testing" + "time" + + _ "github.com/jackc/pgx/v5/stdlib" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "github.com/testcontainers/testcontainers-go" + "github.com/testcontainers/testcontainers-go/wait" +) + +// startPostgres16 spins up a Postgres 16 container, applies +// GetReplayStoreMigration, and returns a connected *sql.DB. Skipped when +// Docker isn't reachable. +func startPostgres16(t *testing.T) *sql.DB { + t.Helper() + ctx := context.Background() + + req := testcontainers.ContainerRequest{ + Image: "postgres:16-alpine", + ExposedPorts: []string{"5432/tcp"}, + Env: map[string]string{ + "POSTGRES_USER": "pgreplay", + "POSTGRES_PASSWORD": "pgreplay", + "POSTGRES_DB": "pgreplay_test", + }, + WaitingFor: wait.ForListeningPort("5432/tcp").WithStartupTimeout(60 * time.Second), + } + container, err := testcontainers.GenericContainer(ctx, testcontainers.GenericContainerRequest{ + ContainerRequest: req, + Started: true, + }) + if err != nil { + t.Skipf("Docker not available, skipping integration test: %v", err) + } + t.Cleanup(func() { _ = container.Terminate(context.Background()) }) + + host, err := container.Host(ctx) + require.NoError(t, err) + port, err := container.MappedPort(ctx, "5432/tcp") + require.NoError(t, err) + + dsn := fmt.Sprintf("postgres://pgreplay:pgreplay@%s:%s/pgreplay_test?sslmode=disable", host, port.Port()) + + db, err := sql.Open("pgx", dsn) + require.NoError(t, err) + t.Cleanup(func() { _ = db.Close() }) + + // wait.ForListeningPort proves the TCP port is open, not that Postgres + // has finished its own startup sequence (it briefly opens/closes the + // port during initdb). Retry the ping for a bit before giving up. + deadline := time.Now().Add(30 * time.Second) + for { + pingCtx, cancel := context.WithTimeout(ctx, 2*time.Second) + err = db.PingContext(pingCtx) + cancel() + if err == nil { + break + } + if time.Now().After(deadline) { + t.Fatalf("postgres container never became reachable: %v", err) + } + time.Sleep(250 * time.Millisecond) + } + + _, err = db.ExecContext(ctx, GetReplayStoreMigration()) + require.NoError(t, err, "applying GetReplayStoreMigration") + + return db +} + +// TestIntegration_ConcurrentInsertSameNonce_OnlyOneWins is the actual +// correctness proof adcontextprotocol/adcp-go#105 and #54 exist for: N +// concurrent Insert calls for the identical (keyid, scope, nonce) — modeling +// a captured signature replayed against every verifier instance in a pool +// at once — must yield exactly one winner. This is what an in-memory, +// per-process ReplayStore cannot guarantee across processes; it's what the +// (keyid, scope, nonce) primary key + ON CONFLICT DO NOTHING is for. +func TestIntegration_ConcurrentInsertSameNonce_OnlyOneWins(t *testing.T) { + db := startPostgres16(t) + store := NewPostgresReplayStore(db) + + const n = 50 + var wg sync.WaitGroup + var successes atomic.Int64 + start := make(chan struct{}) + + for i := 0; i < n; i++ { + wg.Add(1) + go func() { + defer wg.Done() + <-start // maximize the chance every goroutine races the same instant + if store.Insert("replayed-keyid", "replayed-nonce", time.Minute) { + successes.Add(1) + } + }() + } + close(start) + wg.Wait() + + assert.Equal(t, int64(1), successes.Load(), + "exactly one of %d concurrent Insert calls for the same (keyid, scope, nonce) must win; "+ + "more than one means the replay slipped through, fewer than one means a legitimate first request was rejected", n) + + // The row genuinely exists and Seen() now reports it, proving the winner + // actually persisted (not e.g. every call silently no-op'ing). + assert.True(t, store.Seen("replayed-keyid", "replayed-nonce")) +} + +// TestIntegration_ConcurrentInsertDistinctNonces_AllSucceed is the negative +// control for the test above: concurrency alone must not cause spurious +// rejections when the nonces actually differ. +func TestIntegration_ConcurrentInsertDistinctNonces_AllSucceed(t *testing.T) { + db := startPostgres16(t) + store := NewPostgresReplayStore(db) + + const n = 50 + var wg sync.WaitGroup + var successes atomic.Int64 + start := make(chan struct{}) + + for i := 0; i < n; i++ { + i := i + wg.Add(1) + go func() { + defer wg.Done() + <-start + if store.Insert("distinct-keyid", fmt.Sprintf("nonce-%d", i), time.Minute) { + successes.Add(1) + } + }() + } + close(start) + wg.Wait() + + assert.Equal(t, int64(n), successes.Load()) +} + +// TestIntegration_SeenThenInsertRace models the actual verifier flow (step +// 12 Seen, step 13 Insert) under concurrency: multiple goroutines each run +// the same Seen-then-Insert sequence a real signing.VerifyRequest call would. +// Seen() can race ahead of a concurrent Insert() and observe "not seen" for +// more than one caller — that's expected and matches +// adcp/v3/signing.ReplayStore's own doc comment (single Seen check is not +// required to be atomic with Insert). What must hold is Insert() itself: +// however many callers reach it for the same nonce, only one may return +// true. +func TestIntegration_SeenThenInsertRace(t *testing.T) { + db := startPostgres16(t) + store := NewPostgresReplayStore(db) + + const n = 50 + var wg sync.WaitGroup + var inserted atomic.Int64 + start := make(chan struct{}) + + for i := 0; i < n; i++ { + wg.Add(1) + go func() { + defer wg.Done() + <-start + if store.Seen("verify-flow-keyid", "verify-flow-nonce") { + return // this goroutine correctly detects the replay pre-insert + } + if store.Insert("verify-flow-keyid", "verify-flow-nonce", time.Minute) { + inserted.Add(1) + } + }() + } + close(start) + wg.Wait() + + assert.Equal(t, int64(1), inserted.Load(), "at most one goroutine may win the insert regardless of how many passed the Seen pre-check") +} + +// TestIntegration_HitCapEnforced proves HitCap observes rows Insert wrote, +// end to end against real Postgres (not sqlmock's SQL-shape approximation). +func TestIntegration_HitCapEnforced(t *testing.T) { + db := startPostgres16(t) + store := NewPostgresReplayStore(db, WithHitCapLimit(3), WithScope("hitcap-test")) + + for i := 0; i < 3; i++ { + ok := store.Insert("capped-keyid", fmt.Sprintf("nonce-%d", i), time.Minute) + require.True(t, ok, "insert %d should succeed before the cap is reached", i) + } + assert.True(t, store.HitCap("capped-keyid"), "cap of 3 should be reached after 3 inserts") + + ok := store.Insert("capped-keyid", "nonce-over-cap", time.Minute) + assert.False(t, ok, "insert past the cap must be rejected") +} + +// TestIntegration_SweepExpiredReplays_RemovesOnlyExpired seeds a mix of +// already-expired and still-live rows directly (bypassing Insert, which +// only ever writes future expiry) and proves the sweep removes exactly the +// expired ones. +func TestIntegration_SweepExpiredReplays_RemovesOnlyExpired(t *testing.T) { + db := startPostgres16(t) + ctx := context.Background() + + const scope = "sweep-test" + seed := func(nonce string, expiresAt time.Time) { + _, err := db.ExecContext(ctx, + `INSERT INTO adcp_replay_cache (keyid, scope, nonce, expires_at) VALUES ($1, $2, $3, $4)`, + "sweep-keyid", scope, nonce, expiresAt) + require.NoError(t, err) + } + + seed("expired-1", time.Now().Add(-time.Hour)) + seed("expired-2", time.Now().Add(-time.Minute)) + seed("live-1", time.Now().Add(time.Hour)) + seed("live-2", time.Now().Add(24*time.Hour)) + + n, err := SweepExpiredReplays(ctx, db) + require.NoError(t, err) + assert.Equal(t, 2, n, "sweep must remove exactly the two expired rows") + + var remaining int + require.NoError(t, db.QueryRowContext(ctx, + `SELECT count(*) FROM adcp_replay_cache WHERE scope = $1`, scope).Scan(&remaining)) + assert.Equal(t, 2, remaining, "the two live rows must survive the sweep") + + var remainingExpired int + require.NoError(t, db.QueryRowContext(ctx, + `SELECT count(*) FROM adcp_replay_cache WHERE scope = $1 AND expires_at <= now()`, scope).Scan(&remainingExpired)) + assert.Equal(t, 0, remainingExpired, "no expired row should survive the sweep") + + // A second sweep with nothing expired removes nothing. + n2, err := SweepExpiredReplays(ctx, db) + require.NoError(t, err) + assert.Equal(t, 0, n2) +} + +// TestIntegration_ConstructorPingUnreachable exercises the eager-probe +// gotcha against a real (terminated) container: once the container is gone, +// the pool can no longer dial it, and NewPostgresReplayStore must panic +// rather than hand back a store that will fail closed on every request +// without anyone having noticed at wire-up time. +func TestIntegration_ConstructorPingUnreachable(t *testing.T) { + db := startPostgres16(t) + require.NoError(t, db.Ping()) + + require.NoError(t, db.Close()) + + assert.Panics(t, func() { + NewPostgresReplayStore(db) + }) +} + +// TestIntegration_ScopeIsolation proves two stores sharing one database but +// different scopes (e.g. adcp/request-signing/v1 vs adcp/webhook-signing/v1 +// pointed at the same Postgres pool) do not see each other's nonces — the +// scenario WithScope's doc comment describes. +func TestIntegration_ScopeIsolation(t *testing.T) { + db := startPostgres16(t) + reqSigning := NewPostgresReplayStore(db, WithScope("adcp/request-signing/v1")) + webhookSigning := NewPostgresReplayStore(db, WithScope("adcp/webhook-signing/v1")) + + require.True(t, reqSigning.Insert("shared-keyid", "shared-nonce", time.Minute)) + + assert.False(t, webhookSigning.Seen("shared-keyid", "shared-nonce"), + "a different scope must not observe the other scope's nonce") + assert.True(t, webhookSigning.Insert("shared-keyid", "shared-nonce", time.Minute), + "the same (keyid, nonce) must be insertable again under a different scope") +} diff --git a/adcp/v3/signing/pgreplay/store.go b/adcp/v3/signing/pgreplay/store.go new file mode 100644 index 00000000..251d6d8c --- /dev/null +++ b/adcp/v3/signing/pgreplay/store.go @@ -0,0 +1,431 @@ +package pgreplay + +import ( + "context" + "database/sql" + "errors" + "fmt" + "log/slog" + "sync/atomic" + "time" +) + +// defaultKeyIDCap mirrors adcp/v3/signing.NewMemoryReplayStore's default of +// 1,000,000 entries per keyid (the spec recommendation). Kept as a local +// constant, not imported, per the module-boundary decision in doc.go — this +// package intentionally has no dependency on adcp/v3/signing. +const defaultKeyIDCap = 1_000_000 + +// defaultQueryTimeout bounds every round trip PostgresReplayStore makes +// (HitCap, Seen, Insert). ReplayStore's methods don't accept a context — the +// interface predates a distributed implementation — so PostgresReplayStore +// derives a bounded context internally per call. Override with +// WithQueryTimeout. +const defaultQueryTimeout = 3 * time.Second + +// defaultPingTimeout bounds the eager connection probe in +// NewPostgresReplayStore. +const defaultPingTimeout = 5 * time.Second + +// defaultScope is used when WithScope is not supplied. +const defaultScope = "default" + +// ErrConnDown wraps the error returned by InsertContext (and the one +// recorded by LastInsertError / LastSeenError / LastHitCapError) when a +// Postgres round trip itself failed, as opposed to the store correctly +// rejecting the request (nonce already present, or the per-keyid cap +// reached). Check errors.Is(err, ErrConnDown) to tell "the database is +// unreachable" apart from "the request was correctly rejected" — see the +// package doc and the Insert doc comment for why the ReplayStore-interface +// methods still fail closed (reject) in both cases. +var ErrConnDown = errors.New("pgreplay: database round trip failed") + +// replayCacheSchema is the DDL PostgresReplayStore expects. IF NOT EXISTS +// makes it safe to run on every deploy from a migration tool that doesn't +// track individual statements — mirrors adcp/idempotency.PostgresSchema's +// convention. +const replayCacheSchema = ` +CREATE TABLE IF NOT EXISTS adcp_replay_cache ( + keyid TEXT NOT NULL, + scope TEXT NOT NULL, + nonce TEXT NOT NULL, + expires_at TIMESTAMPTZ NOT NULL, + PRIMARY KEY (keyid, scope, nonce) +); + +CREATE INDEX IF NOT EXISTS idx_adcp_replay_cache_expires_at + ON adcp_replay_cache (expires_at); + +CREATE INDEX IF NOT EXISTS idx_adcp_replay_cache_keyid_scope_active + ON adcp_replay_cache (keyid, scope, expires_at); +` + +// GetReplayStoreMigration returns the DDL that creates adcp_replay_cache and +// its indexes, per adcontextprotocol/adcp-go#105's schema (also shipped as +// the JS SDK's adcp-client#1018 and Python's adcp-client-python replay +// store). Apply it once via your migration tooling before wiring a +// PostgresReplayStore. +func GetReplayStoreMigration() string { return replayCacheSchema } + +// PostgresOption configures a PostgresReplayStore. +type PostgresOption func(*PostgresReplayStore) + +// WithScope sets the scope column value this store reads and writes. +// Defaults to "default". +// +// Use it when one Postgres pool/table backs more than one RFC 9421 tag +// profile — for example adcp/request-signing/v1 +// (signing.MiddlewareOptions.Replay) and adcp/webhook-signing/v1 +// (webhook.VerificationOptions.Replay) both pointed at the same database. +// Construct one *PostgresReplayStore per profile with +// WithScope(profile.Tag) so their nonce namespaces stay isolated — this is +// exactly what the scope column in adcp_replay_cache is for, and matches how +// the JS/Python reference stores use it. +func WithScope(scope string) PostgresOption { + return func(s *PostgresReplayStore) { s.scope = scope } +} + +// WithHitCapLimit sets the per-(keyid, scope) entry cap enforced by HitCap +// and, on a best-effort basis, by Insert (see Insert's doc comment for the +// concurrency caveat). Defaults to 1,000,000, matching +// signing.NewMemoryReplayStore's default. A limit <= 0 passed here is +// treated as "use the default," not "unlimited" — there is no unlimited +// mode, matching MemoryReplayStore's behavior. +func WithHitCapLimit(n int) PostgresOption { + return func(s *PostgresReplayStore) { + if n > 0 { + s.hitCapLimit = n + } + } +} + +// WithQueryTimeout overrides the per-call timeout applied to HitCap, Seen, +// and Insert's Postgres round trips. Defaults to 3s. +func WithQueryTimeout(d time.Duration) PostgresOption { + return func(s *PostgresReplayStore) { + if d > 0 { + s.queryTimeout = d + } + } +} + +// WithLogger sets the logger used to record round-trip failures that Insert, +// Seen, and HitCap fail closed on (see the package doc's "Fail-closed +// posture" section). Defaults to slog.Default(). +func WithLogger(l *slog.Logger) PostgresOption { + return func(s *PostgresReplayStore) { + if l != nil { + s.logger = l + } + } +} + +// PostgresReplayStore is a Postgres-backed replay cache implementing the +// same HitCap/Seen/Insert method set as adcp/v3/signing.ReplayStore (see +// package doc for why this package does not import adcp/signing directly). +// All verifier instances sharing one underlying database observe the same +// cache. +// +// Safe for concurrent use — concurrency safety is delegated to Postgres +// itself (the (keyid, scope, nonce) primary key plus +// INSERT ... ON CONFLICT DO NOTHING), not to an in-process mutex. +type PostgresReplayStore struct { + db *sql.DB + scope string + hitCapLimit int + queryTimeout time.Duration + logger *slog.Logger + + lastInsertErr atomic.Pointer[error] + lastSeenErr atomic.Pointer[error] + lastHitCapErr atomic.Pointer[error] +} + +// NewPostgresReplayStore returns a PostgresReplayStore bound to db. +// +// It probes db with a PingContext (5s timeout, overridable is not exposed — +// construction is meant to happen once at process wire-up, not on a hot +// path) before returning, and panics with an actionable message if db is nil +// or the probe fails. This deliberately matches +// adcp/idempotency.New's fail-fast convention ("must not start in a state +// where cache writes silently fail") rather than returning an error: the +// suggested API in adcontextprotocol/adcp-go#105 has a single return value, +// and construction failure here is not a condition your verifier should try +// to route around by silently falling back to an in-memory store in +// production — see the package doc's "Test environments" section for the +// gotcha this is closing (adcp#3379) and the recommended gating pattern. +func NewPostgresReplayStore(db *sql.DB, opts ...PostgresOption) *PostgresReplayStore { + if db == nil { + panic("pgreplay: NewPostgresReplayStore: db is nil") + } + s := &PostgresReplayStore{ + db: db, + scope: defaultScope, + hitCapLimit: defaultKeyIDCap, + queryTimeout: defaultQueryTimeout, + logger: slog.Default(), + } + for _, opt := range opts { + opt(s) + } + + ctx, cancel := context.WithTimeout(context.Background(), defaultPingTimeout) + defer cancel() + if err := db.PingContext(ctx); err != nil { + panic(fmt.Sprintf( + "pgreplay: NewPostgresReplayStore: database unreachable (check the DSN, network access, and that adcp_replay_cache has been migrated via GetReplayStoreMigration): %v\n\n"+ + "If this is a test or local-dev environment without a real Postgres, do not construct a PostgresReplayStore here — use signing.NewMemoryReplayStore(0) instead. See the pgreplay package doc.", + err, + )) + } + return s +} + +// HitCap implements the HitCap/Seen/Insert method set adcp/v3/signing.ReplayStore +// expects: it returns true if the per-(keyid, scope) entry cap has been +// reached. +// +// Fails closed: a Postgres error is treated as "cap hit" (returns true) so +// the caller rejects the request rather than proceeding to an expensive +// crypto verify against a store that may not be able to record the result. +// The underlying error is recorded and retrievable via LastHitCapError, and +// logged at Error level. +func (s *PostgresReplayStore) HitCap(keyid string) bool { + ctx, cancel := context.WithTimeout(context.Background(), s.queryTimeout) + defer cancel() + + var count int + err := s.db.QueryRowContext(ctx, hitCapSQL, keyid, s.scope).Scan(&count) + if err != nil { + wrapped := fmt.Errorf("pgreplay: HitCap: %w: %w", ErrConnDown, err) + s.lastHitCapErr.Store(&wrapped) + s.logger.Error("pgreplay: HitCap round trip failed; failing closed", "keyid", keyid, "scope", s.scope, "error", err) + return true + } + s.lastHitCapErr.Store(nil) + return count >= s.hitCapLimit +} + +// Seen implements the HitCap/Seen/Insert method set adcp/v3/signing.ReplayStore +// expects: it returns true if the (keyid, nonce) pair is present (within +// this store's scope) and not yet expired. +// +// Fails closed: a Postgres error is treated as "seen" (returns true) — +// failing open here would mean a database outage silently disables replay +// rejection, which is exactly the failure mode RFC 9421 §11.1 exists to +// prevent. The underlying error is recorded and retrievable via +// LastSeenError, and logged at Error level. +func (s *PostgresReplayStore) Seen(keyid, nonce string) bool { + ctx, cancel := context.WithTimeout(context.Background(), s.queryTimeout) + defer cancel() + + var seen bool + err := s.db.QueryRowContext(ctx, seenSQL, keyid, s.scope, nonce).Scan(&seen) + if err != nil { + wrapped := fmt.Errorf("pgreplay: Seen: %w: %w", ErrConnDown, err) + s.lastSeenErr.Store(&wrapped) + s.logger.Error("pgreplay: Seen round trip failed; failing closed", "keyid", keyid, "scope", s.scope, "error", err) + return true + } + s.lastSeenErr.Store(nil) + return seen +} + +// Insert implements the HitCap/Seen/Insert method set adcp/v3/signing.ReplayStore +// expects: it atomically inserts (keyid, scope, nonce) with the given TTL +// and returns true only if this call performed the insert. +// +// # Atomicity +// +// The insert and cap check happen in one round trip: a single statement +// guarded by a WHERE clause on the live-row count for (keyid, scope) and +// ON CONFLICT (keyid, scope, nonce) DO NOTHING. The (keyid, scope, nonce) +// primary key is what makes the nonce-uniqueness half of this genuinely +// atomic under concurrent verifier instances — exactly the race +// adcontextprotocol/adcp-go#105 exists to close: two concurrent Insert calls +// for the same (keyid, scope, nonce), whether from the same process or two +// different verifier instances sharing this database, cannot both return +// true. The cap-check half is best-effort: under a concurrent burst that +// straddles the cap boundary, the WHERE-clause count can be stale between +// two transactions that both evaluate it before either commits, so a small +// amount of cap drift is possible. That mirrors the same caveat a naive +// counter-based Redis implementation would have; ReplayStore's own doc +// comment says distributed stores "SHOULD" (not MUST) make the cap check +// atomic with the insert. +// +// # Return value and failure modes +// +// Three distinct situations all return false, because ReplayStore.Insert's +// signature is bool-only: +// +// 1. The (keyid, scope, nonce) triple already exists (this is the actual +// replay case, or the losing side of the concurrent-same-nonce race). +// 2. The per-(keyid, scope) cap has been reached. +// 3. The Postgres round trip itself failed (connection down, timeout, +// context canceled). +// +// Case 3 is deliberately folded into the same "false" / reject outcome as +// cases 1 and 2 — Insert fails closed, the same posture as HitCap and Seen, +// so a database outage rejects signed requests rather than silently +// admitting unverified ones. The verifier surfaces all three as +// request_signature_rate_abuse, which is accurate for cases 1 and 2 and +// misleading for case 3. +// +// adcontextprotocol/adcp-go#54 raised exactly this: collapsing "cap +// rejected" and "couldn't reach the DB" loses operationally important +// information. This package resolves that by exposing InsertContext +// separately — it returns (bool, error) and lets a caller distinguish case 3 +// via errors.Is(err, ErrConnDown) — and by recording the same distinction +// via LastInsertError so an operator can build alerting/health checks around +// PostgresReplayStore without depending on the shared ReplayStore interface +// changing. Widening adcp/v3/signing.ReplayStore.Insert itself to +// (bool, error) was deliberately left out of this PR: it is a public +// exported interface with implementers outside this repo (anyone who wrote +// a custom ReplayStore, e.g. an existing Redis-backed one), and the +// compiler here can only verify call sites inside this module — it cannot +// verify or fix external implementations, which a signature change would +// silently break. It's a natural, separately-reviewable follow-up; see the +// PR description for the full reasoning. +func (s *PostgresReplayStore) Insert(keyid, nonce string, ttl time.Duration) bool { + ctx, cancel := context.WithTimeout(context.Background(), s.queryTimeout) + defer cancel() + + ok, err := s.InsertContext(ctx, keyid, nonce, ttl) + if err != nil { + s.lastInsertErr.Store(&err) + s.logger.Error("pgreplay: Insert round trip failed; failing closed (rejecting request per RFC 9421 §11.1)", "keyid", keyid, "scope", s.scope, "error", err) + return false + } + s.lastInsertErr.Store(nil) + return ok +} + +// InsertContext is the context-aware, error-distinguishing counterpart to +// Insert. It performs the same atomic insert-with-cap-check described on +// Insert, but returns (false, non-nil error) when the Postgres round trip +// itself failed instead of folding that into a bare false — check +// errors.Is(err, ErrConnDown). A nil error with ok=false means the insert +// was correctly rejected (replay or cap), not that anything failed. +// +// Prefer this method over Insert when wiring a custom ReplayStore adapter or +// building operational tooling (health checks, alerting) around +// PostgresReplayStore; use Insert (or the store as a whole, via the +// ReplayStore interface) when wiring signing.MiddlewareOptions.Replay or +// webhook.VerificationOptions.Replay, which require the bool-only shape. +func (s *PostgresReplayStore) InsertContext(ctx context.Context, keyid, nonce string, ttl time.Duration) (ok bool, err error) { + if ttl <= 0 { + return false, fmt.Errorf("pgreplay: InsertContext: ttl must be positive, got %s", ttl) + } + + // expires_at is computed by Postgres itself (now() + $4 seconds), not + // from the app's clock: HitCap/Seen/insertSQL's own cap check all + // compare expires_at against Postgres's now(), so an expiry minted on a + // different clock skews the actual replay window by however far the + // app and database clocks have drifted apart — shortening it under + // skew where the app clock lags, which is exactly the failure mode + // RFC 9421 §11.1's replay window exists to prevent. + var inserted int + err = s.db.QueryRowContext(ctx, insertSQL, keyid, s.scope, nonce, ttl.Seconds(), s.hitCapLimit).Scan(&inserted) + if errors.Is(err, sql.ErrNoRows) { + // WHERE clause guard (cap) or ON CONFLICT DO NOTHING (nonce already + // present) suppressed the insert. A legitimate rejection, not a + // round-trip failure. + return false, nil + } + if err != nil { + return false, fmt.Errorf("pgreplay: InsertContext: %w: %w", ErrConnDown, err) + } + return inserted == 1, nil +} + +// LastInsertError returns the error from the most recent Insert call whose +// Postgres round trip failed (case 3 in Insert's doc comment), or nil if the +// most recent call did not fail that way. Intended for building +// health-check / alerting on top of a wired PostgresReplayStore without +// changing production request handling — the ReplayStore interface itself +// keeps failing closed regardless of what this reports. Safe for concurrent +// use. +func (s *PostgresReplayStore) LastInsertError() error { + if p := s.lastInsertErr.Load(); p != nil { + return *p + } + return nil +} + +// LastSeenError is LastInsertError's counterpart for Seen. Safe for +// concurrent use. +func (s *PostgresReplayStore) LastSeenError() error { + if p := s.lastSeenErr.Load(); p != nil { + return *p + } + return nil +} + +// LastHitCapError is LastInsertError's counterpart for HitCap. Safe for +// concurrent use. +func (s *PostgresReplayStore) LastHitCapError() error { + if p := s.lastHitCapErr.Load(); p != nil { + return *p + } + return nil +} + +// SweepExpiredReplays deletes every adcp_replay_cache row whose expires_at +// has passed and returns the number of rows removed. Postgres has no native +// per-row TTL, so callers are expected to run this periodically (adopter +// cron or pg_cron — scheduling is explicitly out of scope for this package, +// see adcontextprotocol/adcp-go#105's "Out of scope" section). +// +// Safe to run concurrently with HitCap/Seen/Insert and with itself; it only +// ever removes rows already past expiry, so it cannot race a legitimate +// dedup check (an expired row is, by definition, one a WHERE expires_at > +// now() clause has already stopped counting). +func SweepExpiredReplays(ctx context.Context, db *sql.DB) (int, error) { + res, err := db.ExecContext(ctx, sweepSQL) + if err != nil { + return 0, fmt.Errorf("pgreplay: SweepExpiredReplays: %w", err) + } + n, err := res.RowsAffected() + if err != nil { + return 0, fmt.Errorf("pgreplay: SweepExpiredReplays: rows affected: %w", err) + } + return int(n), nil +} + +const hitCapSQL = ` +SELECT count(*) FROM adcp_replay_cache +WHERE keyid = $1 AND scope = $2 AND expires_at > now() +` + +const seenSQL = ` +SELECT EXISTS ( + SELECT 1 FROM adcp_replay_cache + WHERE keyid = $1 AND scope = $2 AND nonce = $3 AND expires_at > now() +) +` + +// insertSQL performs the cap check and the insert in one round trip. The +// WHERE clause on the CTE's count guards the cap (best-effort under +// concurrency, see Insert's doc comment); ON CONFLICT DO NOTHING on the +// (keyid, scope, nonce) primary key is what makes the nonce-uniqueness half +// genuinely atomic. QueryRowContext + Scan(&inserted) distinguishes +// "inserted" (one row, inserted=1) from "suppressed by either guard" +// (sql.ErrNoRows). +// insertSQL takes ttl (seconds, $4) rather than a precomputed expires_at +// timestamp so expiry is minted from Postgres's own now() — the same clock +// hitCapSQL/seenSQL and this statement's own cap check compare expires_at +// against — instead of from app-clock time.Now(), which would skew the +// actual replay window by the app/DB clock drift (see InsertContext). +const insertSQL = ` +WITH capped AS ( + SELECT count(*) AS n FROM adcp_replay_cache + WHERE keyid = $1 AND scope = $2 AND expires_at > now() +) +INSERT INTO adcp_replay_cache (keyid, scope, nonce, expires_at) +SELECT $1, $2, $3, now() + ($4 * interval '1 second') +WHERE (SELECT n FROM capped) < $5 +ON CONFLICT (keyid, scope, nonce) DO NOTHING +RETURNING 1 +` + +const sweepSQL = `DELETE FROM adcp_replay_cache WHERE expires_at <= now()` diff --git a/adcp/v3/signing/pgreplay/store_test.go b/adcp/v3/signing/pgreplay/store_test.go new file mode 100644 index 00000000..28a50cc6 --- /dev/null +++ b/adcp/v3/signing/pgreplay/store_test.go @@ -0,0 +1,336 @@ +package pgreplay + +import ( + "context" + "database/sql" + "errors" + "regexp" + "testing" + "time" + + sqlmock "github.com/DATA-DOG/go-sqlmock" + _ "github.com/jackc/pgx/v5/stdlib" // registers the "pgx" database/sql driver + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// newMock returns a PostgresReplayStore wired to a sqlmock driver, bypassing +// the constructor's eager ping (sqlmock's default ExpectPing behavior treats +// an unset expectation as a no-op success, so NewPostgresReplayStore's probe +// passes without an explicit ExpectPing call — set MonitorPingsOption(true) +// when a test needs to assert on the ping itself). +func newMock(t *testing.T) (*PostgresReplayStore, sqlmock.Sqlmock, func()) { + t.Helper() + db, mock, err := sqlmock.New( + sqlmock.QueryMatcherOption(sqlmock.QueryMatcherRegexp), + sqlmock.MonitorPingsOption(true), + ) + require.NoError(t, err) + mock.ExpectPing() + s := NewPostgresReplayStore(db) + return s, mock, func() { + assert.NoError(t, mock.ExpectationsWereMet()) + mock.ExpectClose() + assert.NoError(t, db.Close()) + } +} + +var ( + hitCapRegexp = regexp.MustCompile(`SELECT count\(\*\) FROM adcp_replay_cache`) + seenRegexp = regexp.MustCompile(`SELECT EXISTS`) + insertRegexp = regexp.MustCompile(`INSERT INTO adcp_replay_cache`) +) + +// ---- constructor ---- + +func TestNewPostgresReplayStore_NilDB(t *testing.T) { + assert.PanicsWithValue(t, "pgreplay: NewPostgresReplayStore: db is nil", func() { + NewPostgresReplayStore(nil) + }) +} + +func TestNewPostgresReplayStore_PingFails(t *testing.T) { + db, mock, err := sqlmock.New(sqlmock.MonitorPingsOption(true)) + require.NoError(t, err) + defer func() { _ = db.Close() }() + + mock.ExpectPing().WillReturnError(errors.New("connection refused")) + + assert.Panics(t, func() { + NewPostgresReplayStore(db) + }, "constructor must panic (fail loudly) rather than return a store bound to an unreachable database") +} + +// TestNewPostgresReplayStore_UnreachableRealDial exercises the eager-probe +// gotcha end to end without a container: dialing a TCP address nothing +// listens on fails fast, so this proves the constructor actually surfaces a +// clear, actionable panic message against a real driver failure mode (not +// just a mocked one). Uses pgx's stdlib adapter, already a module +// dependency for the integration tests. +func TestNewPostgresReplayStore_UnreachableRealDial(t *testing.T) { + // Port 1 is a reserved/unassigned TCP port; nothing listens there. + db, err := sql.Open("pgx", "postgres://user:pass@127.0.0.1:1/db?connect_timeout=1&sslmode=disable") + require.NoError(t, err, "sql.Open must not itself dial — only PingContext should") + defer func() { _ = db.Close() }() + + defer func() { + r := recover() + require.NotNil(t, r, "constructor must panic against an unreachable database") + msg, ok := r.(string) + require.True(t, ok) + assert.Contains(t, msg, "database unreachable") + assert.Contains(t, msg, "NewMemoryReplayStore", "panic message should point test/dev callers at the in-memory fallback") + }() + NewPostgresReplayStore(db) +} + +// ---- Insert ---- + +func TestInsert_Success(t *testing.T) { + s, mock, done := newMock(t) + defer done() + + mock.ExpectQuery(insertRegexp.String()). + WithArgs("key1", "default", "nonce1", sqlmock.AnyArg(), defaultKeyIDCap). + WillReturnRows(sqlmock.NewRows([]string{"?column?"}).AddRow(1)) + + ok := s.Insert("key1", "nonce1", time.Minute) + assert.True(t, ok) + assert.NoError(t, s.LastInsertError()) +} + +func TestInsert_AlreadyPresent_ReturnsFalseNoError(t *testing.T) { + s, mock, done := newMock(t) + defer done() + + mock.ExpectQuery(insertRegexp.String()). + WithArgs("key1", "default", "nonce1", sqlmock.AnyArg(), defaultKeyIDCap). + WillReturnRows(sqlmock.NewRows([]string{"?column?"})) // no rows => suppressed by ON CONFLICT or cap guard + + ok := s.Insert("key1", "nonce1", time.Minute) + assert.False(t, ok) + assert.NoError(t, s.LastInsertError(), "a legitimate rejection (replay/cap) is not a round-trip error") +} + +func TestInsert_DBError_FailsClosedAndRecordsError(t *testing.T) { + s, mock, done := newMock(t) + defer done() + + mock.ExpectQuery(insertRegexp.String()). + WithArgs("key1", "default", "nonce1", sqlmock.AnyArg(), defaultKeyIDCap). + WillReturnError(errors.New("connection reset by peer")) + + ok := s.Insert("key1", "nonce1", time.Minute) + assert.False(t, ok, "Insert must fail closed on a DB error") + + err := s.LastInsertError() + require.Error(t, err) + assert.True(t, errors.Is(err, ErrConnDown), "LastInsertError must be distinguishable via errors.Is(err, ErrConnDown) per adcp-go#54") +} + +// TestInsert_ExpiryComputedByPostgresNotAppClock locks in the fix for the +// clock-skew bug: InsertContext must pass ttl as a seconds value for +// Postgres's own now() + interval arithmetic to compute expires_at, not a +// timestamp precomputed from the app's clock. HitCap/Seen/insertSQL's own +// cap check all compare expires_at against Postgres's now(); an expiry +// minted on any other clock would skew the actual replay window by however +// far the app and database clocks have drifted. +func TestInsert_ExpiryComputedByPostgresNotAppClock(t *testing.T) { + s, mock, done := newMock(t) + defer done() + + mock.ExpectQuery(insertRegexp.String()). + WithArgs("key1", "default", "nonce1", (90 * time.Second).Seconds(), defaultKeyIDCap). + WillReturnRows(sqlmock.NewRows([]string{"?column?"}).AddRow(1)) + + ok := s.Insert("key1", "nonce1", 90*time.Second) + assert.True(t, ok) +} + +func TestInsertContext_DistinguishesRejectionFromDBError(t *testing.T) { + s, mock, done := newMock(t) + defer done() + + mock.ExpectQuery(insertRegexp.String()). + WithArgs("k", "default", "n", sqlmock.AnyArg(), defaultKeyIDCap). + WillReturnError(errors.New("timeout")) + + ok, err := s.InsertContext(context.Background(), "k", "n", time.Minute) + assert.False(t, ok) + require.Error(t, err) + assert.True(t, errors.Is(err, ErrConnDown)) +} + +func TestInsertContext_NonPositiveTTLRejected(t *testing.T) { + s, _, done := newMock(t) + defer done() + + ok, err := s.InsertContext(context.Background(), "k", "n", 0) + assert.False(t, ok) + assert.Error(t, err) +} + +// ---- Seen ---- + +func TestSeen_True(t *testing.T) { + s, mock, done := newMock(t) + defer done() + + mock.ExpectQuery(seenRegexp.String()). + WithArgs("key1", "default", "nonce1"). + WillReturnRows(sqlmock.NewRows([]string{"exists"}).AddRow(true)) + + assert.True(t, s.Seen("key1", "nonce1")) +} + +func TestSeen_False(t *testing.T) { + s, mock, done := newMock(t) + defer done() + + mock.ExpectQuery(seenRegexp.String()). + WithArgs("key1", "default", "nonce-not-seen"). + WillReturnRows(sqlmock.NewRows([]string{"exists"}).AddRow(false)) + + assert.False(t, s.Seen("key1", "nonce-not-seen")) +} + +func TestSeen_DBError_FailsClosed(t *testing.T) { + s, mock, done := newMock(t) + defer done() + + mock.ExpectQuery(seenRegexp.String()). + WithArgs("key1", "default", "nonce1"). + WillReturnError(errors.New("no connection")) + + assert.True(t, s.Seen("key1", "nonce1"), "Seen must fail closed (treat as already-seen) on a DB error") + err := s.LastSeenError() + require.Error(t, err) + assert.True(t, errors.Is(err, ErrConnDown)) +} + +// ---- HitCap ---- + +func TestHitCap_False(t *testing.T) { + s, mock, done := newMock(t) + defer done() + + mock.ExpectQuery(hitCapRegexp.String()). + WithArgs("key1", "default"). + WillReturnRows(sqlmock.NewRows([]string{"count"}).AddRow(3)) + + assert.False(t, s.HitCap("key1")) +} + +func TestHitCap_True(t *testing.T) { + s, mock, done := newMock(t) + defer done() + + mock.ExpectQuery(hitCapRegexp.String()). + WithArgs("key1", "default"). + WillReturnRows(sqlmock.NewRows([]string{"count"}).AddRow(defaultKeyIDCap)) + + assert.True(t, s.HitCap("key1")) +} + +func TestHitCap_DBError_FailsClosed(t *testing.T) { + s, mock, done := newMock(t) + defer done() + + mock.ExpectQuery(hitCapRegexp.String()). + WithArgs("key1", "default"). + WillReturnError(errors.New("db is down")) + + assert.True(t, s.HitCap("key1"), "HitCap must fail closed (treat as capped) on a DB error") + err := s.LastHitCapError() + require.Error(t, err) + assert.True(t, errors.Is(err, ErrConnDown)) +} + +func TestHitCap_RespectsWithHitCapLimit(t *testing.T) { + db, mock, err := sqlmock.New(sqlmock.QueryMatcherOption(sqlmock.QueryMatcherRegexp), sqlmock.MonitorPingsOption(true)) + require.NoError(t, err) + defer func() { _ = db.Close() }() + mock.ExpectPing() + s := NewPostgresReplayStore(db, WithHitCapLimit(5)) + + mock.ExpectQuery(hitCapRegexp.String()). + WithArgs("key1", "default"). + WillReturnRows(sqlmock.NewRows([]string{"count"}).AddRow(5)) + assert.True(t, s.HitCap("key1")) + require.NoError(t, mock.ExpectationsWereMet()) +} + +// ---- options ---- + +func TestWithScope_ChangesScopeColumnValue(t *testing.T) { + db, mock, err := sqlmock.New(sqlmock.QueryMatcherOption(sqlmock.QueryMatcherRegexp), sqlmock.MonitorPingsOption(true)) + require.NoError(t, err) + defer func() { _ = db.Close() }() + mock.ExpectPing() + s := NewPostgresReplayStore(db, WithScope("adcp/webhook-signing/v1")) + + mock.ExpectQuery(seenRegexp.String()). + WithArgs("key1", "adcp/webhook-signing/v1", "nonce1"). + WillReturnRows(sqlmock.NewRows([]string{"exists"}).AddRow(false)) + s.Seen("key1", "nonce1") + require.NoError(t, mock.ExpectationsWereMet()) +} + +func TestWithHitCapLimit_NonPositiveIgnored(t *testing.T) { + db, mock, err := sqlmock.New(sqlmock.MonitorPingsOption(true)) + require.NoError(t, err) + defer func() { _ = db.Close() }() + mock.ExpectPing() + s := NewPostgresReplayStore(db, WithHitCapLimit(-1), WithHitCapLimit(0)) + assert.Equal(t, defaultKeyIDCap, s.hitCapLimit) +} + +// ---- migration / sweep SQL shape ---- + +func TestGetReplayStoreMigration_ContainsExpectedSchema(t *testing.T) { + ddl := GetReplayStoreMigration() + assert.Contains(t, ddl, "CREATE TABLE IF NOT EXISTS adcp_replay_cache") + assert.Contains(t, ddl, "PRIMARY KEY (keyid, scope, nonce)") + assert.Contains(t, ddl, "idx_adcp_replay_cache_expires_at") + assert.Contains(t, ddl, "idx_adcp_replay_cache_keyid_scope_active") +} + +func TestSweepExpiredReplays_DeletesAndReturnsCount(t *testing.T) { + db, mock, err := sqlmock.New(sqlmock.QueryMatcherOption(sqlmock.QueryMatcherRegexp)) + require.NoError(t, err) + defer func() { _ = db.Close() }() + + mock.ExpectExec(`DELETE FROM adcp_replay_cache WHERE expires_at`). + WillReturnResult(sqlmock.NewResult(0, 7)) + + n, err := SweepExpiredReplays(context.Background(), db) + require.NoError(t, err) + assert.Equal(t, 7, n) + require.NoError(t, mock.ExpectationsWereMet()) +} + +func TestSweepExpiredReplays_DBError(t *testing.T) { + db, mock, err := sqlmock.New(sqlmock.QueryMatcherOption(sqlmock.QueryMatcherRegexp)) + require.NoError(t, err) + defer func() { _ = db.Close() }() + + mock.ExpectExec(`DELETE FROM adcp_replay_cache WHERE expires_at`). + WillReturnError(errors.New("deadlock")) + + _, err = SweepExpiredReplays(context.Background(), db) + assert.Error(t, err) +} + +// ---- interface shape (compile-time-ish check without importing adcp/signing) ---- + +// replayStoreShape mirrors adcp/v3/signing.ReplayStore's method set. This +// package deliberately does not import adcp/v3/signing (see doc.go), so this +// local interface is how we assert PostgresReplayStore stays +// structurally assignable to it without adding that dependency. +type replayStoreShape interface { + HitCap(keyid string) bool + Seen(keyid, nonce string) bool + Insert(keyid, nonce string, ttl time.Duration) bool +} + +var _ replayStoreShape = (*PostgresReplayStore)(nil) diff --git a/go.work.example b/go.work.example index f0f50b3f..e466ba6b 100644 --- a/go.work.example +++ b/go.work.example @@ -6,6 +6,7 @@ use ( . ./adcp ./adcp/v3 + ./adcp/v3/signing/pgreplay ./bench ./bench/context-perf ./bench/identity-perf