From 6a36e848582f5c7abe48b51c481ba726ed4482b7 Mon Sep 17 00:00:00 2001 From: Ankur Mhatre Date: Sun, 8 Nov 2026 18:22:05 +0000 Subject: [PATCH] Add idempotency module: Idempotency-Key for POST APIs on Spring Boot 4.1.2 with Redis and Postgres stores One servlet filter, two interchangeable stores, 18 tests (9 scenarios x 2 stores) against a real redis-server 7.0.15 and PostgreSQL 16.15, and 20 captured transcripts under output/: replay, 422 on a changed body, concurrent duplicates, expiry, the crash window and a handler slower than the lock. Co-Authored-By: Claude Sonnet 5.5 Claude-Session: https://claude.ai/code/session_017ic4b5yg8grY17JsVAJviV --- README.md | 1 + idempotency/README.md | 83 ++++ idempotency/output/00-timezone-trap.txt | 14 + idempotency/output/01-replay-postgres.txt | 5 + idempotency/output/01-replay-redis.txt | 5 + idempotency/output/02-mismatch-postgres.txt | 6 + idempotency/output/02-mismatch-redis.txt | 6 + .../output/03-missing-key-postgres.txt | 5 + idempotency/output/03-missing-key-redis.txt | 5 + ...04a-timeout-retry-unprotected-postgres.txt | 4 + .../04a-timeout-retry-unprotected-redis.txt | 4 + .../04b-timeout-retry-keyed-postgres.txt | 5 + .../output/04b-timeout-retry-keyed-redis.txt | 5 + idempotency/output/05-concurrent-postgres.txt | 6 + idempotency/output/05-concurrent-redis.txt | 6 + idempotency/output/06-expiry-postgres.txt | 5 + idempotency/output/06-expiry-redis.txt | 5 + .../output/07-crash-window-postgres.txt | 7 + idempotency/output/07-crash-window-redis.txt | 7 + .../output/08-slow-handler-postgres.txt | 6 + idempotency/output/08-slow-handler-redis.txt | 6 + idempotency/output/09-test-summary.txt | 3 + idempotency/pom.xml | 66 ++++ idempotency/scripts/run-all.sh | 24 ++ idempotency/scripts/timezone-trap.sh | 28 ++ .../ankurm/idempotency/CachedBodyRequest.java | 41 ++ .../com/ankurm/idempotency/FaultInjector.java | 22 ++ .../idempotency/IdempotencyApplication.java | 14 + .../ankurm/idempotency/IdempotencyFilter.java | 117 ++++++ .../idempotency/IdempotencyProperties.java | 17 + .../ankurm/idempotency/IdempotencyStore.java | 39 ++ .../java/com/ankurm/idempotency/Payment.java | 3 + .../ankurm/idempotency/PaymentController.java | 44 +++ .../ankurm/idempotency/PaymentRequest.java | 3 + .../idempotency/PostgresIdempotencyStore.java | 114 ++++++ .../idempotency/RedisIdempotencyStore.java | 111 ++++++ .../ankurm/idempotency/SimulatedCrash.java | 24 ++ .../src/main/resources/application.yml | 18 + idempotency/src/main/resources/schema.sql | 20 + .../ankurm/idempotency/PostgresStoreTest.java | 25 ++ .../com/ankurm/idempotency/RedisServer.java | 92 +++++ .../ankurm/idempotency/RedisStoreTest.java | 34 ++ .../ankurm/idempotency/StoreContractTest.java | 362 ++++++++++++++++++ .../com/ankurm/idempotency/Transcript.java | 42 ++ 44 files changed, 1459 insertions(+) create mode 100644 idempotency/README.md create mode 100644 idempotency/output/00-timezone-trap.txt create mode 100644 idempotency/output/01-replay-postgres.txt create mode 100644 idempotency/output/01-replay-redis.txt create mode 100644 idempotency/output/02-mismatch-postgres.txt create mode 100644 idempotency/output/02-mismatch-redis.txt create mode 100644 idempotency/output/03-missing-key-postgres.txt create mode 100644 idempotency/output/03-missing-key-redis.txt create mode 100644 idempotency/output/04a-timeout-retry-unprotected-postgres.txt create mode 100644 idempotency/output/04a-timeout-retry-unprotected-redis.txt create mode 100644 idempotency/output/04b-timeout-retry-keyed-postgres.txt create mode 100644 idempotency/output/04b-timeout-retry-keyed-redis.txt create mode 100644 idempotency/output/05-concurrent-postgres.txt create mode 100644 idempotency/output/05-concurrent-redis.txt create mode 100644 idempotency/output/06-expiry-postgres.txt create mode 100644 idempotency/output/06-expiry-redis.txt create mode 100644 idempotency/output/07-crash-window-postgres.txt create mode 100644 idempotency/output/07-crash-window-redis.txt create mode 100644 idempotency/output/08-slow-handler-postgres.txt create mode 100644 idempotency/output/08-slow-handler-redis.txt create mode 100644 idempotency/output/09-test-summary.txt create mode 100644 idempotency/pom.xml create mode 100755 idempotency/scripts/run-all.sh create mode 100755 idempotency/scripts/timezone-trap.sh create mode 100644 idempotency/src/main/java/com/ankurm/idempotency/CachedBodyRequest.java create mode 100644 idempotency/src/main/java/com/ankurm/idempotency/FaultInjector.java create mode 100644 idempotency/src/main/java/com/ankurm/idempotency/IdempotencyApplication.java create mode 100644 idempotency/src/main/java/com/ankurm/idempotency/IdempotencyFilter.java create mode 100644 idempotency/src/main/java/com/ankurm/idempotency/IdempotencyProperties.java create mode 100644 idempotency/src/main/java/com/ankurm/idempotency/IdempotencyStore.java create mode 100644 idempotency/src/main/java/com/ankurm/idempotency/Payment.java create mode 100644 idempotency/src/main/java/com/ankurm/idempotency/PaymentController.java create mode 100644 idempotency/src/main/java/com/ankurm/idempotency/PaymentRequest.java create mode 100644 idempotency/src/main/java/com/ankurm/idempotency/PostgresIdempotencyStore.java create mode 100644 idempotency/src/main/java/com/ankurm/idempotency/RedisIdempotencyStore.java create mode 100644 idempotency/src/main/java/com/ankurm/idempotency/SimulatedCrash.java create mode 100644 idempotency/src/main/resources/application.yml create mode 100644 idempotency/src/main/resources/schema.sql create mode 100644 idempotency/src/test/java/com/ankurm/idempotency/PostgresStoreTest.java create mode 100644 idempotency/src/test/java/com/ankurm/idempotency/RedisServer.java create mode 100644 idempotency/src/test/java/com/ankurm/idempotency/RedisStoreTest.java create mode 100644 idempotency/src/test/java/com/ankurm/idempotency/StoreContractTest.java create mode 100644 idempotency/src/test/java/com/ankurm/idempotency/Transcript.java diff --git a/README.md b/README.md index a06f8ef..a91cb1d 100644 --- a/README.md +++ b/README.md @@ -9,6 +9,7 @@ files. | Directory | Article | What it demonstrates | |---|---|---| +| [`idempotency/`](idempotency) | [Idempotency Keys for POST APIs in Spring Boot (Stripe-Style) with Redis and Postgres](https://ankurm.com/idempotency-keys-spring-boot-redis-postgres/) | one servlet filter and two interchangeable stores: a retried POST after a client timeout creating two payments without a key and one with; 400 / 422 / 409 and replay with `Idempotent-Replayed`; ten simultaneous duplicates (nine `409` on Redis, nine replays on Postgres because the second `INSERT` waits on the first's uncommitted row); a crash after the payment commits leaving Redis with a stuck lock and then a duplicate payment, and Postgres with neither; a handler slower than the Redis lock finishing with nothing stored; the legacy `Asia/Calcutta` zone id that PostgreSQL 16 refuses at connect time | | [`http-service-clients/`](http-service-clients) | [Spring Boot 4 HTTP Service Clients: @HttpExchange Interfaces with RestClient (Complete Guide)](https://ankurm.com/spring-boot-4-http-service-clients/) | `@HttpExchange` interfaces registered with `@ImportHttpServices` and configured under `spring.http.serviceclient..*` (the keys the first version of the post gave are silently ignored), run against a local stub upstream: `defaultStatusHandler` with ProblemDetail decoding, `NotFoundRestClientAdapterDecorator` for `Optional` on a 404, Framework 7 `@Retryable` on the interface vs `RetryTemplate` vs an interceptor vs Resilience4j (call counts side by side, and a retried POST created twice), bearer, API-key and OAuth2 client-credentials auth (`@ClientRegistrationId` ignored without its group configurer), `MockRestServiceServer`, and the `@RestClientTest` slice where a group client is not bound to the mock | | [`pdf-invoice/`](pdf-invoice) | [Generate PDF Invoices in Spring Boot: OpenPDF vs iText vs Flying Saucer (HTML to PDF)](https://ankurm.com/spring-boot-generate-pdf-invoice-openpdf-itext-flying-saucer/) | one invoice rendered by OpenPDF 3.0.5, iText Core 9.8.0 (AGPL 3 or commercial) and Flying Saucer 10.7.0 from a Thymeleaf template, with embedded Noto Sans Devanagari: none of the three shapes Marathi/Hindi conjuncts by default (checked by rendering to PNG against a libraqm reference), OpenPDF's `GlyphLayoutManager` does but breaks text extraction until `ActualText` is on, OpenPDF's static `LayoutProcessor` is JVM-global and breaks other writers, Apache FOP on the classpath did not help, streamed vs buffered failure (`200` + curl exit 18 vs `500`), font subsetting, indicative per-invoice timings | | [`filter-interceptor-aop/`](filter-interceptor-aop) | [Filter vs HandlerInterceptor vs AOP in Spring Boot 4: Which Runs When](https://ankurm.com/spring-boot-filter-vs-handlerinterceptor-vs-aop-which-runs-when/) | the real order of a servlet filter, a `HandlerInterceptor` and an `@Aspect` printed from a running Tomcat and from MockMvc, the callbacks that still fire (and the ones that do not) after an exception, a 404, a `preHandle` that returns false and an async request, `afterCompletion` seeing `ex=null` after `@ControllerAdvice`, a header added in `postHandle` silently lost, a request body read twice, `ContentCachingResponseWrapper` returning an empty 200, Spring Security 7's chain sitting at order -100, a filter bean running twice, the Framework 7 path-matching removals read with `javap`, and the three self-invocation-style cases an aspect never sees | diff --git a/idempotency/README.md b/idempotency/README.md new file mode 100644 index 0000000..f2c444e --- /dev/null +++ b/idempotency/README.md @@ -0,0 +1,83 @@ +# idempotency + +Companion code for **Idempotency Keys for POST APIs in Spring Boot (Stripe-Style) with Redis and Postgres** on +[ankurm.com](https://ankurm.com/idempotency-keys-spring-boot-redis-postgres/). + +One servlet filter puts the `Idempotency-Key` rules in front of `POST /payments`, and two interchangeable +stores sit behind it: Redis (a `SET NX PX` lock plus a stored response) and Postgres (one row per key, claimed +with `INSERT ... ON CONFLICT`, inside the same transaction as the payment). The test suite runs every scenario +against both stores and writes what it saw to `output/`. + +| Piece | Version | Checked against | +|---|---|---| +| Spring Boot | 4.1.2 | `maven-metadata.xml` for `spring-boot-starter-parent` (4.2.0-RC2 exists but is not GA) | +| JDK | 25 | Temurin 25.0.5 | +| Redis | 7.0.15 (`redis-server` binary) | `redis-server --version` | +| PostgreSQL | 16.15 | `SHOW server_version` (transcript 00) | +| IETF draft | `draft-ietf-httpapi-idempotency-key-header-08`, 22 Oct 2026, still an active Internet-Draft | datatracker page | + +## Run it + +```bash +export JAVA_HOME=/path/to/jdk-25 +createdb idem # a role idem/idem owning database idem, or set IDEM_DB_URL, IDEM_DB_USER, IDEM_DB_PASSWORD +mvn -B test # 18 tests (9 scenarios x 2 stores), transcripts 01-08 written to output/ +./scripts/run-all.sh # the above plus transcript 00 and the test summary +``` + +`redis-server` and `redis-cli` must be on the `PATH`; the tests start a throwaway server on a free port. The +tests truncate `payments` and `idempotency_keys` in the database they are pointed at. Docker was not available +where this module was written, so there are no Testcontainers; the Redis helper and the plain PostgreSQL +connection are the only things a `GenericContainer` would replace. + +To run the application: `mvn spring-boot:run` (store chosen by `idempotency.store=redis|postgres`). + +**Time zone trap.** The tests pin `-Duser.timezone=UTC`. A JVM whose zone is the legacy id `Asia/Calcutta` cannot +connect to this PostgreSQL at all, because pgjdbc sends the zone as a startup parameter and the server has no such +alias. If you run the application from a machine with `TZ=Asia/Calcutta`, start it with +`-Duser.timezone=Asia/Kolkata`. See [`scripts/timezone-trap.sh`](scripts/timezone-trap.sh) and +[`output/00-timezone-trap.txt`](output/00-timezone-trap.txt). + +## What is where + +| File | What it is | +|---|---| +| [`IdempotencyFilter`](src/main/java/com/ankurm/idempotency/IdempotencyFilter.java) | reads the header, fingerprints the request, asks the store, writes 400 / 409 / 422 or the (replayed) response | +| [`IdempotencyStore`](src/main/java/com/ankurm/idempotency/IdempotencyStore.java) | the one interface both stores implement | +| [`RedisIdempotencyStore`](src/main/java/com/ankurm/idempotency/RedisIdempotencyStore.java) | `SET NX PX` lock, compare-and-set Lua scripts to finish or release it | +| [`PostgresIdempotencyStore`](src/main/java/com/ankurm/idempotency/PostgresIdempotencyStore.java) | claim by `INSERT ... ON CONFLICT DO UPDATE ... WHERE expired`, whole request in one transaction | +| [`CachedBodyRequest`](src/main/java/com/ankurm/idempotency/CachedBodyRequest.java) | lets the body be read twice (fingerprint, then controller) | +| [`PaymentController`](src/main/java/com/ankurm/idempotency/PaymentController.java) | the operation that must not run twice; `/payments/unprotected` is the same code with no filter | +| [`FaultInjector`](src/main/java/com/ankurm/idempotency/FaultInjector.java), [`SimulatedCrash`](src/main/java/com/ankurm/idempotency/SimulatedCrash.java) | make the slow-handler and crash-window failures reachable; delete both before shipping | +| [`schema.sql`](src/main/resources/schema.sql) | `payments` and `idempotency_keys` | +| [`StoreContractTest`](src/test/java/com/ankurm/idempotency/StoreContractTest.java) | all nine scenarios, per-store assertions where the stores differ | + +## Rules the filter applies + +| Situation | Response | +|---|---| +| no `Idempotency-Key`, blank, or longer than 255 characters | `400` | +| key seen, request body (or method/path) differs | `422` | +| key seen, first request still running (Redis) | `409` with `Retry-After: 1` | +| key seen, first request still running (Postgres) | waits, then replays the stored response | +| key seen, first request finished | the stored response, plus `Idempotent-Replayed: true` | +| handler answered 5xx | nothing stored; the key is released so the client can retry | + +## Captured output + +| File | Scenario | +|---|---| +| [`00-timezone-trap.txt`](output/00-timezone-trap.txt) | `psql` refusing `Asia/Calcutta` and accepting `Asia/Kolkata` | +| `01-replay-{redis,postgres}.txt` | a retry gets the first response back; one payment row | +| `02-mismatch-*.txt` | same key, changed amount: 422; same JSON with extra spaces: also 422 | +| `03-missing-key-*.txt` | missing key and the 255 / 256 character boundary | +| `04a-timeout-retry-unprotected-*.txt` | a client timeout and a retry with no key: two payments | +| `04b-timeout-retry-keyed-*.txt` | the same story with a key: one payment (409 on Redis, a wait and a replay on Postgres) | +| `05-concurrent-*.txt` | ten simultaneous duplicates: one payment, nine 409s on Redis, nine replays on Postgres | +| `06-expiry-*.txt` | the key is forgotten after the TTL and the same key pays again | +| `07-crash-window-*.txt` | a crash after the payment commits: Redis ends up with two payments, Postgres with one | +| `08-slow-handler-*.txt` | a handler slower than the Redis lock: two payments and nothing stored; Postgres replays | +| `09-test-summary.txt` | surefire's own counts | + +Transcript lines containing times or dates (the 500 body in `07`) change between runs; everything else is +deterministic. diff --git a/idempotency/output/00-timezone-trap.txt b/idempotency/output/00-timezone-trap.txt new file mode 100644 index 0000000..e65284d --- /dev/null +++ b/idempotency/output/00-timezone-trap.txt @@ -0,0 +1,14 @@ +$ psql -c "SHOW server_version" +16.15 (Ubuntu 16.15-0ubuntu0.24.04.1) + +$ psql -c "SET timezone='Asia/Calcutta'" +ERROR: invalid value for parameter "TimeZone": "Asia/Calcutta" + +$ psql -c "SET timezone='Asia/Kolkata'" +SET + +$ ls /usr/share/zoneinfo/Asia | grep -E 'Calcutta|Kolkata' +Kolkata + +$ dpkg -l tzdata | tail -1 +tzdata 2026e-0ubuntu0.24.04.1 diff --git a/idempotency/output/01-replay-postgres.txt b/idempotency/output/01-replay-postgres.txt new file mode 100644 index 0000000..9487e68 --- /dev/null +++ b/idempotency/output/01-replay-postgres.txt @@ -0,0 +1,5 @@ +store: postgres +POST /payments key=k-1 -> HTTP 201 replayed=- retry-after=- {"id":1,"customer":"cus_42","amount":1999,"currency":"INR"} +POST /payments key=k-1 (retry) -> HTTP 201 replayed=true retry-after=- {"id":1,"customer":"cus_42","amount":1999,"currency":"INR"} +payments in the table: 1 +what the store holds: row status COMPLETED, response_status 201 diff --git a/idempotency/output/01-replay-redis.txt b/idempotency/output/01-replay-redis.txt new file mode 100644 index 0000000..361427c --- /dev/null +++ b/idempotency/output/01-replay-redis.txt @@ -0,0 +1,5 @@ +store: redis +POST /payments key=k-1 -> HTTP 201 replayed=- retry-after=- {"id":1,"customer":"cus_42","amount":1999,"currency":"INR"} +POST /payments key=k-1 (retry) -> HTTP 201 replayed=true retry-after=- {"id":1,"customer":"cus_42","amount":1999,"currency":"INR"} +payments in the table: 1 +what the store holds: string beginning "DONE", expiring: true diff --git a/idempotency/output/02-mismatch-postgres.txt b/idempotency/output/02-mismatch-postgres.txt new file mode 100644 index 0000000..9082113 --- /dev/null +++ b/idempotency/output/02-mismatch-postgres.txt @@ -0,0 +1,6 @@ +store: postgres +key=k-2 amount 1999 -> HTTP 201 replayed=- retry-after=- {"id":1,"customer":"cus_42","amount":1999,"currency":"INR"} +key=k-2 amount 2999 -> HTTP 422 replayed=- retry-after=- {"title":"Idempotency-Key was already used with a different request","status":422} +key=k-2 same JSON, extra spaces -> HTTP 422 replayed=- retry-after=- {"title":"Idempotency-Key was already used with a different request","status":422} +key=k-3 amount 2999 -> HTTP 201 replayed=- retry-after=- {"id":2,"customer":"cus_42","amount":2999,"currency":"INR"} +payments in the table: 2 diff --git a/idempotency/output/02-mismatch-redis.txt b/idempotency/output/02-mismatch-redis.txt new file mode 100644 index 0000000..04107c0 --- /dev/null +++ b/idempotency/output/02-mismatch-redis.txt @@ -0,0 +1,6 @@ +store: redis +key=k-2 amount 1999 -> HTTP 201 replayed=- retry-after=- {"id":1,"customer":"cus_42","amount":1999,"currency":"INR"} +key=k-2 amount 2999 -> HTTP 422 replayed=- retry-after=- {"title":"Idempotency-Key was already used with a different request","status":422} +key=k-2 same JSON, extra spaces -> HTTP 422 replayed=- retry-after=- {"title":"Idempotency-Key was already used with a different request","status":422} +key=k-3 amount 2999 -> HTTP 201 replayed=- retry-after=- {"id":2,"customer":"cus_42","amount":2999,"currency":"INR"} +payments in the table: 2 diff --git a/idempotency/output/03-missing-key-postgres.txt b/idempotency/output/03-missing-key-postgres.txt new file mode 100644 index 0000000..05d76cc --- /dev/null +++ b/idempotency/output/03-missing-key-postgres.txt @@ -0,0 +1,5 @@ +store: postgres +no Idempotency-Key header -> HTTP 400 replayed=- retry-after=- {"title":"Idempotency-Key header is required for this operation","status":400} +key of 256 characters -> HTTP 400 replayed=- retry-after=- {"title":"Idempotency-Key is longer than 255 characters","status":400} +key of 255 characters -> HTTP 201 replayed=- retry-after=- {"id":1,"customer":"cus_42","amount":1999,"currency":"INR"} +payments in the table: 1 diff --git a/idempotency/output/03-missing-key-redis.txt b/idempotency/output/03-missing-key-redis.txt new file mode 100644 index 0000000..a320c96 --- /dev/null +++ b/idempotency/output/03-missing-key-redis.txt @@ -0,0 +1,5 @@ +store: redis +no Idempotency-Key header -> HTTP 400 replayed=- retry-after=- {"title":"Idempotency-Key header is required for this operation","status":400} +key of 256 characters -> HTTP 400 replayed=- retry-after=- {"title":"Idempotency-Key is longer than 255 characters","status":400} +key of 255 characters -> HTTP 201 replayed=- retry-after=- {"id":1,"customer":"cus_42","amount":1999,"currency":"INR"} +payments in the table: 1 diff --git a/idempotency/output/04a-timeout-retry-unprotected-postgres.txt b/idempotency/output/04a-timeout-retry-unprotected-postgres.txt new file mode 100644 index 0000000..2bfec6e --- /dev/null +++ b/idempotency/output/04a-timeout-retry-unprotected-postgres.txt @@ -0,0 +1,4 @@ +store: postgres (not used: this endpoint has no filter) +attempt 1: client gave up after 300 ms (the server is still working) +attempt 2 (the retry) -> HTTP 201 replayed=- retry-after=- {"id":2,"customer":"cus_42","amount":1999,"currency":"INR"} +payments in the table: 2 diff --git a/idempotency/output/04a-timeout-retry-unprotected-redis.txt b/idempotency/output/04a-timeout-retry-unprotected-redis.txt new file mode 100644 index 0000000..5ab9c41 --- /dev/null +++ b/idempotency/output/04a-timeout-retry-unprotected-redis.txt @@ -0,0 +1,4 @@ +store: redis (not used: this endpoint has no filter) +attempt 1: client gave up after 300 ms (the server is still working) +attempt 2 (the retry) -> HTTP 201 replayed=- retry-after=- {"id":2,"customer":"cus_42","amount":1999,"currency":"INR"} +payments in the table: 2 diff --git a/idempotency/output/04b-timeout-retry-keyed-postgres.txt b/idempotency/output/04b-timeout-retry-keyed-postgres.txt new file mode 100644 index 0000000..f29dac6 --- /dev/null +++ b/idempotency/output/04b-timeout-retry-keyed-postgres.txt @@ -0,0 +1,5 @@ +store: postgres +attempt 1: client gave up after 300 ms (the server is still working) +attempt 2, 300 ms later -> HTTP 201 replayed=true retry-after=- {"id":1,"customer":"cus_42","amount":1999,"currency":"INR"} +attempt 3, after the first finished -> HTTP 201 replayed=true retry-after=- {"id":1,"customer":"cus_42","amount":1999,"currency":"INR"} +payments in the table: 1 diff --git a/idempotency/output/04b-timeout-retry-keyed-redis.txt b/idempotency/output/04b-timeout-retry-keyed-redis.txt new file mode 100644 index 0000000..acbc201 --- /dev/null +++ b/idempotency/output/04b-timeout-retry-keyed-redis.txt @@ -0,0 +1,5 @@ +store: redis +attempt 1: client gave up after 300 ms (the server is still working) +attempt 2, 300 ms later -> HTTP 409 replayed=- retry-after=1 {"title":"A request is outstanding for this Idempotency-Key","status":409} +attempt 3, after the first finished -> HTTP 201 replayed=true retry-after=- {"id":1,"customer":"cus_42","amount":1999,"currency":"INR"} +payments in the table: 1 diff --git a/idempotency/output/05-concurrent-postgres.txt b/idempotency/output/05-concurrent-postgres.txt new file mode 100644 index 0000000..8e1f438 --- /dev/null +++ b/idempotency/output/05-concurrent-postgres.txt @@ -0,0 +1,6 @@ +store: postgres +10 requests, same key, sent together, 500 ms of work each +201 created now : 1 +201 replayed : 9 +409 in progress : 0 +payments in the table: 1 diff --git a/idempotency/output/05-concurrent-redis.txt b/idempotency/output/05-concurrent-redis.txt new file mode 100644 index 0000000..561f484 --- /dev/null +++ b/idempotency/output/05-concurrent-redis.txt @@ -0,0 +1,6 @@ +store: redis +10 requests, same key, sent together, 500 ms of work each +201 created now : 1 +201 replayed : 0 +409 in progress : 9 +payments in the table: 1 diff --git a/idempotency/output/06-expiry-postgres.txt b/idempotency/output/06-expiry-postgres.txt new file mode 100644 index 0000000..22a64c7 --- /dev/null +++ b/idempotency/output/06-expiry-postgres.txt @@ -0,0 +1,5 @@ +store: postgres, ttl: 2 s +first request, key=k-6 -> HTTP 201 replayed=- retry-after=- {"id":1,"customer":"cus_42","amount":1999,"currency":"INR"} +immediately after, key=k-6 -> HTTP 201 replayed=true retry-after=- {"id":1,"customer":"cus_42","amount":1999,"currency":"INR"} +after sleeping 2.5 s, key=k-6 -> HTTP 201 replayed=- retry-after=- {"id":2,"customer":"cus_42","amount":1999,"currency":"INR"} +payments in the table: 2 diff --git a/idempotency/output/06-expiry-redis.txt b/idempotency/output/06-expiry-redis.txt new file mode 100644 index 0000000..b961c02 --- /dev/null +++ b/idempotency/output/06-expiry-redis.txt @@ -0,0 +1,5 @@ +store: redis, ttl: 2 s +first request, key=k-6 -> HTTP 201 replayed=- retry-after=- {"id":1,"customer":"cus_42","amount":1999,"currency":"INR"} +immediately after, key=k-6 -> HTTP 201 replayed=true retry-after=- {"id":1,"customer":"cus_42","amount":1999,"currency":"INR"} +after sleeping 2.5 s, key=k-6 -> HTTP 201 replayed=- retry-after=- {"id":2,"customer":"cus_42","amount":1999,"currency":"INR"} +payments in the table: 2 diff --git a/idempotency/output/07-crash-window-postgres.txt b/idempotency/output/07-crash-window-postgres.txt new file mode 100644 index 0000000..32a2af8 --- /dev/null +++ b/idempotency/output/07-crash-window-postgres.txt @@ -0,0 +1,7 @@ +store: postgres, in-progress ttl: 2 s +attempt 1 (process dies after the write) -> HTTP 500 replayed=- retry-after=- {"timestamp":"2026-11-08T18:16:36.198Z","status":500,"error":"Internal Server Error","path":"/payments"} +payments in the table: 0 +attempt 2, immediately -> HTTP 201 replayed=- retry-after=- {"id":2,"customer":"cus_42","amount":1999,"currency":"INR"} +payments in the table: 1 +attempt 3 -> HTTP 201 replayed=true retry-after=- {"id":2,"customer":"cus_42","amount":1999,"currency":"INR"} +payments in the table: 1 diff --git a/idempotency/output/07-crash-window-redis.txt b/idempotency/output/07-crash-window-redis.txt new file mode 100644 index 0000000..79f4381 --- /dev/null +++ b/idempotency/output/07-crash-window-redis.txt @@ -0,0 +1,7 @@ +store: redis, in-progress ttl: 2 s +attempt 1 (process dies after the write) -> HTTP 500 replayed=- retry-after=- {"timestamp":"2026-11-08T18:16:50.214Z","status":500,"error":"Internal Server Error","path":"/payments"} +payments in the table: 1 +attempt 2, immediately -> HTTP 409 replayed=- retry-after=1 {"title":"A request is outstanding for this Idempotency-Key","status":409} +payments in the table: 1 +attempt 3, after the lock expired -> HTTP 201 replayed=- retry-after=- {"id":2,"customer":"cus_42","amount":1999,"currency":"INR"} +payments in the table: 2 diff --git a/idempotency/output/08-slow-handler-postgres.txt b/idempotency/output/08-slow-handler-postgres.txt new file mode 100644 index 0000000..16610bf --- /dev/null +++ b/idempotency/output/08-slow-handler-postgres.txt @@ -0,0 +1,6 @@ +store: postgres, in-progress ttl: 2 s, handler takes 3.2 s +request A (t=0 s) -> HTTP 201 replayed=- retry-after=- {"id":1,"customer":"cus_42","amount":1999,"currency":"INR"} +request B (t=2.5 s, same key) -> HTTP 201 replayed=true retry-after=- {"id":1,"customer":"cus_42","amount":1999,"currency":"INR"} +payments in the table: 1 +requests that finished after their lock expired: n/a (a row lock under an open transaction cannot expire) +what the store holds afterwards: row status COMPLETED, response_status 201 diff --git a/idempotency/output/08-slow-handler-redis.txt b/idempotency/output/08-slow-handler-redis.txt new file mode 100644 index 0000000..1e55f8f --- /dev/null +++ b/idempotency/output/08-slow-handler-redis.txt @@ -0,0 +1,6 @@ +store: redis, in-progress ttl: 2 s, handler takes 3.2 s +request A (t=0 s) -> HTTP 201 replayed=- retry-after=- {"id":1,"customer":"cus_42","amount":1999,"currency":"INR"} +request B (t=2.5 s, same key) -> HTTP 201 replayed=- retry-after=- {"id":2,"customer":"cus_42","amount":1999,"currency":"INR"} +payments in the table: 2 +requests that finished after their lock expired: 2 +what the store holds afterwards: nothing (EXISTS idem:k-8 = 0) diff --git a/idempotency/output/09-test-summary.txt b/idempotency/output/09-test-summary.txt new file mode 100644 index 0000000..b9d6afc --- /dev/null +++ b/idempotency/output/09-test-summary.txt @@ -0,0 +1,3 @@ +Tests run: 9, Failures: 0, Errors: 0, Skipped: 0, Time elapsed: 15.69 s -- in com.ankurm.idempotency.PostgresStoreTest +Tests run: 9, Failures: 0, Errors: 0, Skipped: 0, Time elapsed: 16.31 s -- in com.ankurm.idempotency.RedisStoreTest +Tests run: 18, Failures: 0, Errors: 0, Skipped: 0 diff --git a/idempotency/pom.xml b/idempotency/pom.xml new file mode 100644 index 0000000..7b93a0d --- /dev/null +++ b/idempotency/pom.xml @@ -0,0 +1,66 @@ + + + 4.0.0 + + + org.springframework.boot + spring-boot-starter-parent + 4.1.2 + + + + com.ankurm + idempotency + 1.0.0 + idempotency + Idempotency-Key for POST APIs on Spring Boot 4.1: a Redis store and a Postgres store behind one servlet filter, replays, 422 on a changed body, concurrent duplicates, expiry and the crash window + + + 25 + + + + + org.springframework.boot + spring-boot-starter-webmvc + + + org.springframework.boot + spring-boot-starter-data-redis + + + org.springframework.boot + spring-boot-starter-jdbc + + + org.postgresql + postgresql + runtime + + + org.springframework.boot + spring-boot-starter-test + test + + + + + + + org.springframework.boot + spring-boot-maven-plugin + + + org.apache.maven.plugins + maven-surefire-plugin + + + -Duser.timezone=UTC + + + + + diff --git a/idempotency/scripts/run-all.sh b/idempotency/scripts/run-all.sh new file mode 100755 index 0000000..aaed1d1 --- /dev/null +++ b/idempotency/scripts/run-all.sh @@ -0,0 +1,24 @@ +#!/usr/bin/env bash +# Regenerates every file under output/. +# +# ./scripts/run-all.sh +# +# Needs JDK 25, Maven 3.9, redis-server and redis-cli (7.x) on the PATH, and a PostgreSQL 14+ that +# the application can reach (IDEM_DB_URL / IDEM_DB_USER / IDEM_DB_PASSWORD, default +# jdbc:postgresql://localhost:5432/idem as idem/idem). The tests start a throwaway redis-server on a +# free port; they do truncate the payments and idempotency_keys tables in that database. +# Transcripts 01-08 come out of the test suite, so each line in them is an assertion that fails the +# build when it stops being true. Transcript 00 is psql, run by scripts/timezone-trap.sh. +set -euo pipefail +cd "$(dirname "$0")/.." +mkdir -p output + +echo "== time zone trap (transcript 00)" +./scripts/timezone-trap.sh >/dev/null || true + +echo "== test suite (transcripts 01-08, two stores)" +mvn -B test | tee /tmp/idempotency-mvn.log | grep -E '^\[(INFO|WARNING|ERROR)\] Tests run:' | sed 's/^\[[A-Z]*\] //' > output/09-test-summary.txt + +echo +echo "output:" +ls -1 output diff --git a/idempotency/scripts/timezone-trap.sh b/idempotency/scripts/timezone-trap.sh new file mode 100755 index 0000000..760e90e --- /dev/null +++ b/idempotency/scripts/timezone-trap.sh @@ -0,0 +1,28 @@ +#!/usr/bin/env bash +# Shows why the tests pin -Duser.timezone=UTC. pgjdbc sends the JVM's default zone to Postgres as +# the TimeZone startup parameter. A JVM whose zone is the legacy id Asia/Calcutta then fails with +# FATAL: invalid value for parameter "TimeZone": "Asia/Calcutta" +# on a Postgres whose tzdata lacks the legacy alias. Needs psql and a database reachable with +# IDEM_DB_* (defaults match application.yml). Writes output/00-timezone-trap.txt. +set -uo pipefail +cd "$(dirname "$0")/.." +mkdir -p output +export PGPASSWORD="${IDEM_DB_PASSWORD:-idem}" +P=(psql -h "${IDEM_DB_HOST:-localhost}" -U "${IDEM_DB_USER:-idem}" "${IDEM_DB_NAME:-idem}") +{ + echo "\$ psql -c \"SHOW server_version\"" + "${P[@]}" -t -A -c "SHOW server_version" 2>&1 + echo + echo "\$ psql -c \"SET timezone='Asia/Calcutta'\"" + "${P[@]}" -c "SET timezone='Asia/Calcutta'" 2>&1 + echo + echo "\$ psql -c \"SET timezone='Asia/Kolkata'\"" + "${P[@]}" -c "SET timezone='Asia/Kolkata'" 2>&1 + echo + echo "\$ ls /usr/share/zoneinfo/Asia | grep -E 'Calcutta|Kolkata'" + ls /usr/share/zoneinfo/Asia | grep -E 'Calcutta|Kolkata' + echo + echo "\$ dpkg -l tzdata | tail -1" + dpkg -l tzdata | tail -1 | awk '{print $2, $3}' +} > output/00-timezone-trap.txt +cat output/00-timezone-trap.txt diff --git a/idempotency/src/main/java/com/ankurm/idempotency/CachedBodyRequest.java b/idempotency/src/main/java/com/ankurm/idempotency/CachedBodyRequest.java new file mode 100644 index 0000000..f409f73 --- /dev/null +++ b/idempotency/src/main/java/com/ankurm/idempotency/CachedBodyRequest.java @@ -0,0 +1,41 @@ +package com.ankurm.idempotency; + +import java.io.BufferedReader; +import java.io.ByteArrayInputStream; +import java.io.InputStreamReader; +import java.nio.charset.StandardCharsets; + +import jakarta.servlet.ReadListener; +import jakarta.servlet.ServletInputStream; +import jakarta.servlet.http.HttpServletRequest; +import jakarta.servlet.http.HttpServletRequestWrapper; + +/** + * The filter has to read the body to fingerprint it, and a servlet request body can be read only + * once. This hands the same bytes to the controller afterwards. + */ +final class CachedBodyRequest extends HttpServletRequestWrapper { + + private final byte[] body; + + CachedBodyRequest(HttpServletRequest request, byte[] body) { + super(request); + this.body = body; + } + + @Override + public ServletInputStream getInputStream() { + ByteArrayInputStream in = new ByteArrayInputStream(body); + return new ServletInputStream() { + @Override public int read() { return in.read(); } + @Override public boolean isFinished() { return in.available() == 0; } + @Override public boolean isReady() { return true; } + @Override public void setReadListener(ReadListener listener) { throw new UnsupportedOperationException(); } + }; + } + + @Override + public BufferedReader getReader() { + return new BufferedReader(new InputStreamReader(getInputStream(), StandardCharsets.UTF_8)); + } +} diff --git a/idempotency/src/main/java/com/ankurm/idempotency/FaultInjector.java b/idempotency/src/main/java/com/ankurm/idempotency/FaultInjector.java new file mode 100644 index 0000000..be9c033 --- /dev/null +++ b/idempotency/src/main/java/com/ankurm/idempotency/FaultInjector.java @@ -0,0 +1,22 @@ +package com.ankurm.idempotency; + +import org.springframework.stereotype.Component; + +/** + * Two switches the tests flip to make the failure windows reachable. Delete this class and the + * two lines in {@link PaymentController} that read it before shipping anything like this. + */ +@Component +public class FaultInjector { + + /** Sleep this long before writing the payment, so concurrent duplicates overlap. */ + public volatile long slowMillis = 0; + + /** Throw {@link SimulatedCrash} right after the payment row is written. */ + public volatile boolean crashAfterBusiness = false; + + public void reset() { + slowMillis = 0; + crashAfterBusiness = false; + } +} diff --git a/idempotency/src/main/java/com/ankurm/idempotency/IdempotencyApplication.java b/idempotency/src/main/java/com/ankurm/idempotency/IdempotencyApplication.java new file mode 100644 index 0000000..4d0042d --- /dev/null +++ b/idempotency/src/main/java/com/ankurm/idempotency/IdempotencyApplication.java @@ -0,0 +1,14 @@ +package com.ankurm.idempotency; + +import org.springframework.boot.SpringApplication; +import org.springframework.boot.autoconfigure.SpringBootApplication; +import org.springframework.boot.context.properties.ConfigurationPropertiesScan; + +@SpringBootApplication +@ConfigurationPropertiesScan +public class IdempotencyApplication { + + public static void main(String[] args) { + SpringApplication.run(IdempotencyApplication.class, args); + } +} diff --git a/idempotency/src/main/java/com/ankurm/idempotency/IdempotencyFilter.java b/idempotency/src/main/java/com/ankurm/idempotency/IdempotencyFilter.java new file mode 100644 index 0000000..7c951ac --- /dev/null +++ b/idempotency/src/main/java/com/ankurm/idempotency/IdempotencyFilter.java @@ -0,0 +1,117 @@ +package com.ankurm.idempotency; + +import java.io.IOException; +import java.nio.charset.StandardCharsets; +import java.security.MessageDigest; +import java.security.NoSuchAlgorithmException; +import java.util.HexFormat; + +import jakarta.servlet.FilterChain; +import jakarta.servlet.ServletException; +import jakarta.servlet.http.HttpServletRequest; +import jakarta.servlet.http.HttpServletResponse; + +import org.springframework.stereotype.Component; +import org.springframework.web.filter.OncePerRequestFilter; +import org.springframework.web.util.ContentCachingResponseWrapper; + +/** + * Puts the idempotency rules in front of {@code POST }: + *
    + *
  • no {@code Idempotency-Key} header: 400
    • + *
  • same key, different request: 422
    • + *
  • same key while the first request is still running: 409 with {@code Retry-After}
    • + *
  • same key after the first finished: the stored response, with {@code Idempotent-Replayed: true}
    • + *
+ * The status codes follow the IETF draft; see the post for what the draft says and does not say. + */ +@Component +class IdempotencyFilter extends OncePerRequestFilter { + + static final String HEADER = "Idempotency-Key"; + static final String REPLAYED = "Idempotent-Replayed"; + + private final IdempotencyStore store; + private final IdempotencyProperties props; + + IdempotencyFilter(IdempotencyStore store, IdempotencyProperties props) { + this.store = store; + this.props = props; + } + + @Override + protected boolean shouldNotFilter(HttpServletRequest request) { + return !("POST".equals(request.getMethod()) && props.path().equals(request.getRequestURI())); + } + + @Override + protected void doFilterInternal(HttpServletRequest request, HttpServletResponse response, FilterChain chain) + throws ServletException, IOException { + String key = request.getHeader(HEADER); + if (key == null || key.isBlank()) { + problem(response, 400, "Idempotency-Key header is required for this operation"); + return; + } + if (key.length() > 255) { + problem(response, 400, "Idempotency-Key is longer than 255 characters"); + return; + } + + byte[] body = request.getInputStream().readAllBytes(); + String fingerprint = fingerprint(request, body); + + IdempotencyStore.Outcome outcome; + try { + outcome = store.handle(key, fingerprint, () -> { + ContentCachingResponseWrapper captured = new ContentCachingResponseWrapper(response); + chain.doFilter(new CachedBodyRequest(request, body), captured); + return new IdempotencyStore.Captured( + captured.getStatus(), captured.getContentType(), captured.getContentAsByteArray()); + }); + } catch (RuntimeException | IOException | ServletException e) { + throw e; + } catch (Exception e) { + throw new ServletException(e); + } + + switch (outcome.kind()) { + case PROCESSED -> write(response, outcome.response(), false); + case REPLAYED -> write(response, outcome.response(), true); + case IN_PROGRESS -> { + response.setHeader("Retry-After", "1"); + problem(response, 409, "A request is outstanding for this Idempotency-Key"); + } + case MISMATCH -> problem(response, 422, "Idempotency-Key was already used with a different request"); + } + } + + /** SHA-256 over method, path and the exact body bytes. Whitespace and key order count. */ + static String fingerprint(HttpServletRequest request, byte[] body) { + try { + MessageDigest sha = MessageDigest.getInstance("SHA-256"); + sha.update((request.getMethod() + " " + request.getRequestURI() + "\n").getBytes(StandardCharsets.UTF_8)); + sha.update(body); + return HexFormat.of().formatHex(sha.digest()); + } catch (NoSuchAlgorithmException e) { + throw new IllegalStateException(e); + } + } + + private static void write(HttpServletResponse response, IdempotencyStore.Captured captured, boolean replayed) + throws IOException { + response.setStatus(captured.status()); + if (captured.contentType() != null) { + response.setContentType(captured.contentType()); + } + if (replayed) { + response.setHeader(REPLAYED, "true"); + } + response.getOutputStream().write(captured.body()); + } + + private static void problem(HttpServletResponse response, int status, String title) throws IOException { + response.setStatus(status); + response.setContentType("application/problem+json"); + response.getWriter().write("{\"title\":\"" + title + "\",\"status\":" + status + "}"); + } +} diff --git a/idempotency/src/main/java/com/ankurm/idempotency/IdempotencyProperties.java b/idempotency/src/main/java/com/ankurm/idempotency/IdempotencyProperties.java new file mode 100644 index 0000000..baf9795 --- /dev/null +++ b/idempotency/src/main/java/com/ankurm/idempotency/IdempotencyProperties.java @@ -0,0 +1,17 @@ +package com.ankurm.idempotency; + +import java.time.Duration; + +import org.springframework.boot.context.properties.ConfigurationProperties; +import org.springframework.boot.context.properties.bind.DefaultValue; + +/** Everything configurable about the idempotency layer, from {@code idempotency.*}. */ +@ConfigurationProperties("idempotency") +public record IdempotencyProperties( + @DefaultValue("redis") Store store, + @DefaultValue("24h") Duration ttl, + @DefaultValue("30s") Duration inProgressTtl, + @DefaultValue("/payments") String path) { + + public enum Store { REDIS, POSTGRES } +} diff --git a/idempotency/src/main/java/com/ankurm/idempotency/IdempotencyStore.java b/idempotency/src/main/java/com/ankurm/idempotency/IdempotencyStore.java new file mode 100644 index 0000000..eddc069 --- /dev/null +++ b/idempotency/src/main/java/com/ankurm/idempotency/IdempotencyStore.java @@ -0,0 +1,39 @@ +package com.ankurm.idempotency; + +/** + * The one decision the filter delegates: given a key and a fingerprint of the request, either run + * the real handler (once) or answer from what an earlier request left behind. + * + *

Redis and Postgres implement it very differently, and that difference is the point of the + * module: see {@link RedisIdempotencyStore} and {@link PostgresIdempotencyStore}. + */ +public interface IdempotencyStore { + + Outcome handle(String key, String fingerprint, Handler handler) throws Exception; + + /** Runs the rest of the filter chain and captures what it wrote. */ + @FunctionalInterface + interface Handler { + Captured run() throws Exception; + } + + /** A response, held in memory so it can be stored and replayed byte for byte. */ + record Captured(int status, String contentType, byte[] body) { } + + enum Kind { + /** The handler ran now. */ + PROCESSED, + /** An earlier request finished; this is its stored response. */ + REPLAYED, + /** An earlier request with this key has not finished yet (Redis only). */ + IN_PROGRESS, + /** The key exists but belongs to a different request body. */ + MISMATCH + } + + record Outcome(Kind kind, Captured response) { + static Outcome of(Kind kind) { + return new Outcome(kind, null); + } + } +} diff --git a/idempotency/src/main/java/com/ankurm/idempotency/Payment.java b/idempotency/src/main/java/com/ankurm/idempotency/Payment.java new file mode 100644 index 0000000..cce3c79 --- /dev/null +++ b/idempotency/src/main/java/com/ankurm/idempotency/Payment.java @@ -0,0 +1,3 @@ +package com.ankurm.idempotency; + +public record Payment(long id, String customer, long amount, String currency) { } diff --git a/idempotency/src/main/java/com/ankurm/idempotency/PaymentController.java b/idempotency/src/main/java/com/ankurm/idempotency/PaymentController.java new file mode 100644 index 0000000..33fb322 --- /dev/null +++ b/idempotency/src/main/java/com/ankurm/idempotency/PaymentController.java @@ -0,0 +1,44 @@ +package com.ankurm.idempotency; + +import org.springframework.http.ResponseEntity; +import org.springframework.jdbc.core.simple.JdbcClient; +import org.springframework.web.bind.annotation.PostMapping; +import org.springframework.web.bind.annotation.RequestBody; +import org.springframework.web.bind.annotation.RestController; + +/** + * The business operation that must not run twice. {@code /payments} is protected by + * {@link IdempotencyFilter}; {@code /payments/unprotected} is the same code with nothing in front + * of it, so a test can show what a retry does without the filter. + */ +@RestController +class PaymentController { + + private final JdbcClient jdbc; + private final FaultInjector faults; + + PaymentController(JdbcClient jdbc, FaultInjector faults) { + this.jdbc = jdbc; + this.faults = faults; + } + + @PostMapping({"/payments", "/payments/unprotected"}) + ResponseEntity create(@RequestBody PaymentRequest request) throws InterruptedException { + if (faults.slowMillis > 0) { + Thread.sleep(faults.slowMillis); + } + long id = jdbc.sql(""" + INSERT INTO payments (customer, amount, currency) + VALUES (:customer, :amount, :currency) RETURNING id""") + .param("customer", request.customer()) + .param("amount", request.amount()) + .param("currency", request.currency()) + .query(Long.class) + .single(); + if (faults.crashAfterBusiness) { + throw new SimulatedCrash(); + } + return ResponseEntity.status(201) + .body(new Payment(id, request.customer(), request.amount(), request.currency())); + } +} diff --git a/idempotency/src/main/java/com/ankurm/idempotency/PaymentRequest.java b/idempotency/src/main/java/com/ankurm/idempotency/PaymentRequest.java new file mode 100644 index 0000000..e41cf07 --- /dev/null +++ b/idempotency/src/main/java/com/ankurm/idempotency/PaymentRequest.java @@ -0,0 +1,3 @@ +package com.ankurm.idempotency; + +public record PaymentRequest(String customer, long amount, String currency) { } diff --git a/idempotency/src/main/java/com/ankurm/idempotency/PostgresIdempotencyStore.java b/idempotency/src/main/java/com/ankurm/idempotency/PostgresIdempotencyStore.java new file mode 100644 index 0000000..21f41fb --- /dev/null +++ b/idempotency/src/main/java/com/ankurm/idempotency/PostgresIdempotencyStore.java @@ -0,0 +1,114 @@ +package com.ankurm.idempotency; + +import java.nio.charset.StandardCharsets; +import java.util.Optional; + +import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; +import org.springframework.jdbc.core.simple.JdbcClient; +import org.springframework.stereotype.Component; +import org.springframework.transaction.PlatformTransactionManager; +import org.springframework.transaction.support.TransactionTemplate; + +/** + * Keeps one row per key and runs the whole request inside one database transaction, so the + * business write and the idempotency record commit together or not at all. + * + + *

Expiry is measured with {@code clock_timestamp()}, not {@code now()}: {@code now()} is the time the + * transaction started, so a request slower than the TTL would write a key that is already expired. + * + *

The claim is a single {@code INSERT ... ON CONFLICT DO UPDATE ... WHERE expires_at < now()}. + * A live conflicting row makes it affect zero rows; an expired one is taken over. When two + * requests race, the second one's insert waits on the first one's uncommitted row, which is why + * this store never answers 409: a concurrent duplicate waits and then replays. + */ +@Component +@ConditionalOnProperty(name = "idempotency.store", havingValue = "postgres") +class PostgresIdempotencyStore implements IdempotencyStore { + + private final JdbcClient jdbc; + private final TransactionTemplate tx; + private final IdempotencyProperties props; + + PostgresIdempotencyStore(JdbcClient jdbc, PlatformTransactionManager txManager, IdempotencyProperties props) { + this.jdbc = jdbc; + this.tx = new TransactionTemplate(txManager); + this.props = props; + } + + @Override + public Outcome handle(String key, String fingerprint, Handler handler) throws Exception { + try { + return tx.execute(status -> { + int claimed = jdbc.sql(""" + INSERT INTO idempotency_keys (idem_key, fingerprint, status, expires_at) + VALUES (:key, :fp, 'IN_PROGRESS', clock_timestamp() + (:ttlMs * interval '1 millisecond')) + ON CONFLICT (idem_key) DO UPDATE + SET fingerprint = EXCLUDED.fingerprint, status = 'IN_PROGRESS', + response_status = NULL, content_type = NULL, response_body = NULL, + created_at = clock_timestamp(), expires_at = EXCLUDED.expires_at + WHERE idempotency_keys.expires_at < clock_timestamp()""") + .param("key", key) + .param("fp", fingerprint) + .param("ttlMs", props.ttl().toMillis()) + .update(); + + if (claimed == 0) { + return existing(key, fingerprint); + } + + Captured captured; + try { + captured = handler.run(); + } catch (Exception e) { + throw new Failure(e); + } + if (captured.status() >= 500) { + status.setRollbackOnly(); // the payment and the key both disappear + return new Outcome(Kind.PROCESSED, captured); + } + jdbc.sql(""" + UPDATE idempotency_keys + SET status = 'COMPLETED', response_status = :s, + content_type = :ct, response_body = :body, + expires_at = clock_timestamp() + (:ttlMs * interval '1 millisecond') + WHERE idem_key = :key""") + .param("s", captured.status()) + .param("ct", captured.contentType()) + .param("body", new String(captured.body(), StandardCharsets.UTF_8)) + .param("ttlMs", props.ttl().toMillis()) + .param("key", key) + .update(); + return new Outcome(Kind.PROCESSED, captured); + }); + } catch (Failure f) { + throw f.cause; + } + } + + private Outcome existing(String key, String fingerprint) { + record Row(String fingerprint, Integer status, String contentType, String body) { } + Optional row = jdbc.sql(""" + SELECT fingerprint, response_status, content_type, response_body + FROM idempotency_keys WHERE idem_key = :key""") + .param("key", key) + .query((rs, n) -> new Row(rs.getString(1), (Integer) rs.getObject(2), rs.getString(3), rs.getString(4))) + .optional(); + Row r = row.orElseThrow(() -> new IllegalStateException("key vanished between claim and read")); + if (!r.fingerprint().equals(fingerprint)) { + return Outcome.of(Kind.MISMATCH); + } + return new Outcome(Kind.REPLAYED, new Captured(r.status(), r.contentType(), + r.body().getBytes(StandardCharsets.UTF_8))); + } + + /** Carries a checked exception out of the transaction callback without losing the rollback. */ + private static final class Failure extends RuntimeException { + final Exception cause; + + Failure(Exception cause) { + super(cause); + this.cause = cause; + } + } +} diff --git a/idempotency/src/main/java/com/ankurm/idempotency/RedisIdempotencyStore.java b/idempotency/src/main/java/com/ankurm/idempotency/RedisIdempotencyStore.java new file mode 100644 index 0000000..7c9810a --- /dev/null +++ b/idempotency/src/main/java/com/ankurm/idempotency/RedisIdempotencyStore.java @@ -0,0 +1,111 @@ +package com.ankurm.idempotency; + +import java.nio.charset.StandardCharsets; +import java.util.List; +import java.util.UUID; +import java.util.concurrent.atomic.AtomicInteger; + +import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; +import org.springframework.data.redis.core.StringRedisTemplate; +import org.springframework.data.redis.core.script.DefaultRedisScript; +import org.springframework.stereotype.Component; + +/** + * Keeps one Redis string per key. + * + * 

+ * IN_PROGRESS\n<fingerprint>\n<owner uuid>                         lives inProgressTtl
+ * DONE\n<fingerprint>\n<status>\n<content type>\n<body>           lives ttl
+ *
+ * + * The first request wins with {@code SET key value NX PX ttl}. Finishing and releasing are + * compare-and-set Lua scripts, so a request whose lock already expired cannot overwrite or delete + * the lock a newer request now holds. + * + *

What Redis cannot do is write the payment and the key together. Anything that happens between + * the payment commit and the {@code DONE} write is the crash window the tests measure. + */ +@Component +@ConditionalOnProperty(name = "idempotency.store", havingValue = "redis", matchIfMissing = true) +class RedisIdempotencyStore implements IdempotencyStore { + + private static final String PREFIX = "idem:"; + + private static final DefaultRedisScript CAS_SET = new DefaultRedisScript<>(""" + if redis.call('get', KEYS[1]) == ARGV[1] then + redis.call('set', KEYS[1], ARGV[2], 'PX', ARGV[3]) + return 1 + end + return 0""", Long.class); + + private static final DefaultRedisScript CAS_DELETE = new DefaultRedisScript<>(""" + if redis.call('get', KEYS[1]) == ARGV[1] then + return redis.call('del', KEYS[1]) + end + return 0""", Long.class); + + private final StringRedisTemplate redis; + private final IdempotencyProperties props; + + /** Times a request finished after its lock had already expired. Read by the tests. */ + final AtomicInteger lostLocks = new AtomicInteger(); + + RedisIdempotencyStore(StringRedisTemplate redis, IdempotencyProperties props) { + this.redis = redis; + this.props = props; + } + + @Override + public Outcome handle(String key, String fingerprint, Handler handler) throws Exception { + String redisKey = PREFIX + key; + String lock = "IN_PROGRESS\n" + fingerprint + "\n" + UUID.randomUUID(); + + for (int attempt = 0; attempt < 3; attempt++) { + if (Boolean.TRUE.equals(redis.opsForValue().setIfAbsent(redisKey, lock, props.inProgressTtl()))) { + return runAndRecord(redisKey, lock, handler); + } + String existing = redis.opsForValue().get(redisKey); + if (existing == null) { + continue; // it expired between the two calls; try to take it + } + String[] parts = existing.split("\n", 5); + if (!parts[1].equals(fingerprint)) { + return Outcome.of(Kind.MISMATCH); + } + if (parts[0].equals("IN_PROGRESS")) { + return Outcome.of(Kind.IN_PROGRESS); + } + return new Outcome(Kind.REPLAYED, new Captured( + Integer.parseInt(parts[2]), + parts[3].isEmpty() ? null : parts[3], + parts[4].getBytes(StandardCharsets.UTF_8))); + } + return Outcome.of(Kind.IN_PROGRESS); + } + + private Outcome runAndRecord(String redisKey, String lock, Handler handler) throws Exception { + Captured captured; + try { + captured = handler.run(); + } catch (Exception e) { + if (!SimulatedCrash.isCrash(e)) { + redis.execute(CAS_DELETE, List.of(redisKey), lock); // a failed request may be retried + } + throw e; // a crashed process releases nothing: the lock stays until it expires + } + if (captured.status() >= 500) { + redis.execute(CAS_DELETE, List.of(redisKey), lock); + return new Outcome(Kind.PROCESSED, captured); + } + String fingerprint = lock.split("\n", 3)[1]; + String done = "DONE\n" + fingerprint + "\n" + captured.status() + "\n" + + (captured.contentType() == null ? "" : captured.contentType()) + "\n" + + new String(captured.body(), StandardCharsets.UTF_8); + Long stored = redis.execute(CAS_SET, List.of(redisKey), lock, done, + Long.toString(props.ttl().toMillis())); + if (stored == null || stored == 0L) { + lostLocks.incrementAndGet(); // the lock expired and someone else owns the key now + } + return new Outcome(Kind.PROCESSED, captured); + } +} diff --git a/idempotency/src/main/java/com/ankurm/idempotency/SimulatedCrash.java b/idempotency/src/main/java/com/ankurm/idempotency/SimulatedCrash.java new file mode 100644 index 0000000..2b388ee --- /dev/null +++ b/idempotency/src/main/java/com/ankurm/idempotency/SimulatedCrash.java @@ -0,0 +1,24 @@ +package com.ankurm.idempotency; + +/** + * Stands in for the process dying at an awkward moment. It is thrown after the payment row has + * been written and before the idempotency layer has recorded the response. A Redis store must not + * tidy up after it (a dead process tidies up nothing); a Postgres store rolls everything back + * because the payment and the key were written in one transaction. + */ +public class SimulatedCrash extends RuntimeException { + + public SimulatedCrash() { + super("simulated crash after the business write"); + } + + /** Spring MVC wraps controller exceptions, so look through the cause chain. */ + public static boolean isCrash(Throwable t) { + for (Throwable c = t; c != null; c = c.getCause()) { + if (c instanceof SimulatedCrash) { + return true; + } + } + return false; + } +} diff --git a/idempotency/src/main/resources/application.yml b/idempotency/src/main/resources/application.yml new file mode 100644 index 0000000..adc4055 --- /dev/null +++ b/idempotency/src/main/resources/application.yml @@ -0,0 +1,18 @@ +spring: + datasource: + url: ${IDEM_DB_URL:jdbc:postgresql://localhost:5432/idem} + username: ${IDEM_DB_USER:idem} + password: ${IDEM_DB_PASSWORD:idem} + sql: + init: + mode: always + data: + redis: + host: localhost + port: 6379 + +idempotency: + store: redis # redis | postgres + ttl: 24h # how long a finished response is replayed (Stripe prunes keys after 24 hours) + in-progress-ttl: 30s # how long a Redis lock for an unfinished request lives + path: /payments diff --git a/idempotency/src/main/resources/schema.sql b/idempotency/src/main/resources/schema.sql new file mode 100644 index 0000000..9235707 --- /dev/null +++ b/idempotency/src/main/resources/schema.sql @@ -0,0 +1,20 @@ +CREATE TABLE IF NOT EXISTS payments ( + id BIGSERIAL PRIMARY KEY, + customer TEXT NOT NULL, + amount BIGINT NOT NULL, + currency CHAR(3) NOT NULL, + created_at TIMESTAMPTZ NOT NULL DEFAULT now() +); + +-- Used only when idempotency.store=postgres. +CREATE TABLE IF NOT EXISTS idempotency_keys ( + idem_key TEXT PRIMARY KEY, + fingerprint CHAR(64) NOT NULL, + status TEXT NOT NULL, + response_status INT, + content_type TEXT, + response_body TEXT, + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + expires_at TIMESTAMPTZ NOT NULL +); +CREATE INDEX IF NOT EXISTS idempotency_keys_expires_at ON idempotency_keys (expires_at); diff --git a/idempotency/src/test/java/com/ankurm/idempotency/PostgresStoreTest.java b/idempotency/src/test/java/com/ankurm/idempotency/PostgresStoreTest.java new file mode 100644 index 0000000..4561618 --- /dev/null +++ b/idempotency/src/test/java/com/ankurm/idempotency/PostgresStoreTest.java @@ -0,0 +1,25 @@ +package com.ankurm.idempotency; + +import org.springframework.boot.test.context.SpringBootTest; + +@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT, properties = { + "idempotency.store=postgres", "idempotency.ttl=2s", "idempotency.in-progress-ttl=2s"}) +class PostgresStoreTest extends StoreContractTest { + + @Override + String storeName() { + return "postgres"; + } + + @Override + String inspect(String key) { + return jdbc.sql("SELECT 'row status ' || status || ', response_status ' || response_status " + + "FROM idempotency_keys WHERE idem_key = :key") + .param("key", key).query(String.class).single(); + } + + @Override + int lostLocks() { + return 0; // a Postgres row lock cannot expire under a running transaction + } +} diff --git a/idempotency/src/test/java/com/ankurm/idempotency/RedisServer.java b/idempotency/src/test/java/com/ankurm/idempotency/RedisServer.java new file mode 100644 index 0000000..01e981c --- /dev/null +++ b/idempotency/src/test/java/com/ankurm/idempotency/RedisServer.java @@ -0,0 +1,92 @@ +package com.ankurm.idempotency; + +import java.io.IOException; +import java.net.InetSocketAddress; +import java.net.ServerSocket; +import java.net.Socket; +import java.nio.charset.StandardCharsets; +import java.util.List; + +/** + * A throwaway {@code redis-server} on a free port. Docker was not available where the module was + * written, so the tests start the real Redis binary themselves. Nothing here is mocked. + */ +public final class RedisServer implements AutoCloseable { + + private static RedisServer shared; + + private final Process process; + private final int port; + + private RedisServer(Process process, int port) { + this.process = process; + this.port = port; + } + + /** One server per test JVM, stopped by a shutdown hook. */ + public static synchronized RedisServer shared() { + if (shared == null) { + shared = start(); + Runtime.getRuntime().addShutdownHook(new Thread(shared::close)); + } + return shared; + } + + public static RedisServer start() { + try { + int port; + try (ServerSocket s = new ServerSocket(0)) { + port = s.getLocalPort(); + } + Process p = new ProcessBuilder("redis-server", "--port", Integer.toString(port), + "--save", "", "--appendonly", "no", "--bind", "127.0.0.1") + .redirectOutput(ProcessBuilder.Redirect.DISCARD) + .redirectError(ProcessBuilder.Redirect.DISCARD) + .start(); + RedisServer server = new RedisServer(p, port); + server.awaitReady(); + return server; + } catch (IOException | InterruptedException e) { + throw new IllegalStateException("could not start redis-server (is it on the PATH?)", e); + } + } + + public int port() { + return port; + } + + private void awaitReady() throws InterruptedException { + for (int i = 0; i < 100; i++) { + try (Socket s = new Socket()) { + s.connect(new InetSocketAddress("127.0.0.1", port), 100); + return; + } catch (IOException ignored) { + Thread.sleep(50); + } + } + throw new IllegalStateException("redis-server did not start on port " + port); + } + + /** Runs redis-cli against this server and returns trimmed stdout. */ + public String cli(String... args) { + try { + List cmd = new java.util.ArrayList<>(List.of("redis-cli", "-p", Integer.toString(port))); + cmd.addAll(List.of(args)); + Process p = new ProcessBuilder(cmd).redirectErrorStream(true).start(); + String out = new String(p.getInputStream().readAllBytes(), StandardCharsets.UTF_8).trim(); + p.waitFor(); + return out; + } catch (IOException | InterruptedException e) { + throw new IllegalStateException(e); + } + } + + public void flush() { + cli("FLUSHALL"); + } + + @Override + public void close() { + process.destroy(); + } +} diff --git a/idempotency/src/test/java/com/ankurm/idempotency/RedisStoreTest.java b/idempotency/src/test/java/com/ankurm/idempotency/RedisStoreTest.java new file mode 100644 index 0000000..a3dec81 --- /dev/null +++ b/idempotency/src/test/java/com/ankurm/idempotency/RedisStoreTest.java @@ -0,0 +1,34 @@ +package com.ankurm.idempotency; + +import org.junit.jupiter.api.BeforeEach; + +import org.springframework.boot.test.context.SpringBootTest; + +@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT, properties = { + "idempotency.store=redis", "idempotency.ttl=2s", "idempotency.in-progress-ttl=2s"}) +class RedisStoreTest extends StoreContractTest { + + @Override + String storeName() { + return "redis"; + } + + @Override + String inspect(String key) { + String value = REDIS.cli("GET", "idem:" + key); + long ttl = Long.parseLong(REDIS.cli("PTTL", "idem:" + key)); + return "string beginning \"" + value.lines().findFirst().orElse("") + "\", expiring: " + (ttl > 0 && ttl <= 2000); + } + + private int baseline; + + @BeforeEach + void rememberLostLocks() { + baseline = ((RedisIdempotencyStore) store).lostLocks.get(); + } + + @Override + int lostLocks() { + return ((RedisIdempotencyStore) store).lostLocks.get() - baseline; + } +} diff --git a/idempotency/src/test/java/com/ankurm/idempotency/StoreContractTest.java b/idempotency/src/test/java/com/ankurm/idempotency/StoreContractTest.java new file mode 100644 index 0000000..832aa3a --- /dev/null +++ b/idempotency/src/test/java/com/ankurm/idempotency/StoreContractTest.java @@ -0,0 +1,362 @@ +package com.ankurm.idempotency; + +import java.net.URI; +import java.net.http.HttpClient; +import java.net.http.HttpRequest; +import java.net.http.HttpResponse; +import java.net.http.HttpTimeoutException; +import java.time.Duration; +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; + +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.beans.factory.annotation.Value; +import org.springframework.jdbc.core.simple.JdbcClient; +import org.springframework.test.context.DynamicPropertyRegistry; +import org.springframework.test.context.DynamicPropertySource; + +import static org.assertj.core.api.Assertions.assertThat; + +/** + * Every scenario the post talks about, run against whichever store the subclass selects. The test + * prints what it sees into {@code output/NN-scenario-store.txt} and asserts the same facts, so a + * line in a transcript is something the build checks. + * + *

Where the two stores legitimately behave differently, the assertion is per store and the + * difference is the finding. + */ +abstract class StoreContractTest { + + static final RedisServer REDIS = RedisServer.shared(); + + static final String BODY = "{\"customer\":\"cus_42\",\"amount\":1999,\"currency\":\"INR\"}"; + static final String BODY_OTHER_AMOUNT = "{\"customer\":\"cus_42\",\"amount\":2999,\"currency\":\"INR\"}"; + static final String BODY_EXTRA_SPACES = "{\"customer\": \"cus_42\", \"amount\": 1999, \"currency\": \"INR\"}"; + + private static final HttpClient HTTP = HttpClient.newBuilder().version(HttpClient.Version.HTTP_1_1).build(); + + @Value("${local.server.port}") + int port; + + @Autowired + JdbcClient jdbc; + + @Autowired + FaultInjector faults; + + @Autowired + IdempotencyStore store; + + @DynamicPropertySource + static void redis(DynamicPropertyRegistry registry) { + registry.add("spring.data.redis.port", REDIS::port); + } + + abstract String storeName(); + + /** One deterministic line describing what the store holds for this key. */ + abstract String inspect(String key); + + abstract int lostLocks(); + + @BeforeEach + void reset() { + jdbc.sql("TRUNCATE payments, idempotency_keys RESTART IDENTITY").update(); + REDIS.flush(); + faults.reset(); + } + + // ------------------------------------------------------------------ helpers + + record Resp(int status, String replayed, String retryAfter, String body) { } + + Resp post(String path, String key, String json, Duration timeout) throws Exception { + HttpRequest.Builder b = HttpRequest.newBuilder(URI.create("http://localhost:" + port + path)) + .timeout(timeout) + .header("Content-Type", "application/json") + .POST(HttpRequest.BodyPublishers.ofString(json)); + if (key != null) { + b.header("Idempotency-Key", key); + } + HttpResponse r = HTTP.send(b.build(), HttpResponse.BodyHandlers.ofString()); + return new Resp(r.statusCode(), + r.headers().firstValue("Idempotent-Replayed").orElse(null), + r.headers().firstValue("Retry-After").orElse(null), + r.body()); + } + + Resp post(String key, String json) throws Exception { + return post("/payments", key, json, Duration.ofSeconds(10)); + } + + long payments() { + return jdbc.sql("SELECT count(*) FROM payments").query(Long.class).single(); + } + + Transcript transcript(String scenario) { + return Transcript.open(scenario + "-" + storeName() + ".txt"); + } + + static String show(String label, Resp r) { + return String.format("%-34s -> HTTP %d replayed=%-4s retry-after=%-4s %s", + label, r.status(), r.replayed() == null ? "-" : r.replayed(), + r.retryAfter() == null ? "-" : r.retryAfter(), r.body()); + } + + // ------------------------------------------------------------------ scenarios + + @Test + void replayReturnsTheFirstResponse() throws Exception { + try (Transcript t = transcript("01-replay")) { + t.line("store: %s", storeName()); + Resp first = post("k-1", BODY); + t.line(show("POST /payments key=k-1", first)); + Resp retry = post("k-1", BODY); + t.line(show("POST /payments key=k-1 (retry)", retry)); + t.line("payments in the table: %d", payments()); + t.line("what the store holds: %s", inspect("k-1")); + + assertThat(first.status()).isEqualTo(201); + assertThat(first.replayed()).isNull(); + assertThat(first.body()).contains("\"id\":1"); + assertThat(retry.status()).isEqualTo(201); + assertThat(retry.replayed()).isEqualTo("true"); + assertThat(retry.body()).isEqualTo(first.body()); + assertThat(payments()).isEqualTo(1); + } + } + + @Test + void sameKeyDifferentBodyIs422() throws Exception { + try (Transcript t = transcript("02-mismatch")) { + t.line("store: %s", storeName()); + Resp first = post("k-2", BODY); + t.line(show("key=k-2 amount 1999", first)); + Resp changed = post("k-2", BODY_OTHER_AMOUNT); + t.line(show("key=k-2 amount 2999", changed)); + Resp spaces = post("k-2", BODY_EXTRA_SPACES); + t.line(show("key=k-2 same JSON, extra spaces", spaces)); + Resp fresh = post("k-3", BODY_OTHER_AMOUNT); + t.line(show("key=k-3 amount 2999", fresh)); + t.line("payments in the table: %d", payments()); + + assertThat(first.status()).isEqualTo(201); + assertThat(changed.status()).isEqualTo(422); + assertThat(spaces.status()).isEqualTo(422); + assertThat(fresh.status()).isEqualTo(201); + assertThat(payments()).isEqualTo(2); + } + } + + @Test + void missingOrOversizedKeyIs400() throws Exception { + try (Transcript t = transcript("03-missing-key")) { + t.line("store: %s", storeName()); + Resp none = post("/payments", null, BODY, Duration.ofSeconds(10)); + t.line(show("no Idempotency-Key header", none)); + Resp tooLong = post("x".repeat(256), BODY); + t.line(show("key of 256 characters", tooLong)); + Resp ok = post("x".repeat(255), BODY); + t.line(show("key of 255 characters", ok)); + t.line("payments in the table: %d", payments()); + + assertThat(none.status()).isEqualTo(400); + assertThat(tooLong.status()).isEqualTo(400); + assertThat(ok.status()).isEqualTo(201); + assertThat(payments()).isEqualTo(1); + } + } + + @Test + void aTimeoutRetryWithoutAKeyPaysTwice() throws Exception { + try (Transcript t = transcript("04a-timeout-retry-unprotected")) { + t.line("store: %s (not used: this endpoint has no filter)", storeName()); + faults.slowMillis = 800; + try { + post("/payments/unprotected", null, BODY, Duration.ofMillis(300)); + t.line("attempt 1: answered (unexpected)"); + } catch (HttpTimeoutException e) { + t.line("attempt 1: client gave up after 300 ms (the server is still working)"); + } + Resp retry = post("/payments/unprotected", null, BODY, Duration.ofSeconds(5)); + t.line(show("attempt 2 (the retry)", retry)); + Thread.sleep(1200); + t.line("payments in the table: %d", payments()); + + assertThat(retry.status()).isEqualTo(201); + assertThat(payments()).isEqualTo(2); + } + } + + @Test + void aTimeoutRetryWithAKeyPaysOnce() throws Exception { + try (Transcript t = transcript("04b-timeout-retry-keyed")) { + t.line("store: %s", storeName()); + faults.slowMillis = 800; + try { + post("/payments", "k-4", BODY, Duration.ofMillis(300)); + t.line("attempt 1: answered (unexpected)"); + } catch (HttpTimeoutException e) { + t.line("attempt 1: client gave up after 300 ms (the server is still working)"); + } + Resp retry = post("/payments", "k-4", BODY, Duration.ofSeconds(5)); + t.line(show("attempt 2, 300 ms later", retry)); + Thread.sleep(1200); + Resp late = post("k-4", BODY); + t.line(show("attempt 3, after the first finished", late)); + t.line("payments in the table: %d", payments()); + + if (storeName().equals("redis")) { + assertThat(retry.status()).isEqualTo(409); + assertThat(retry.retryAfter()).isEqualTo("1"); + } else { + assertThat(retry.status()).isEqualTo(201); + assertThat(retry.replayed()).isEqualTo("true"); + } + assertThat(late.status()).isEqualTo(201); + assertThat(late.replayed()).isEqualTo("true"); + assertThat(payments()).isEqualTo(1); + } + } + + @Test + void tenConcurrentDuplicatesPayOnce() throws Exception { + try (Transcript t = transcript("05-concurrent")) { + t.line("store: %s", storeName()); + faults.slowMillis = 500; + int n = 10; + CountDownLatch start = new CountDownLatch(1); + List responses = new ArrayList<>(); + try (ExecutorService pool = Executors.newVirtualThreadPerTaskExecutor()) { + List> futures = new ArrayList<>(); + for (int i = 0; i < n; i++) { + futures.add(pool.submit(() -> { + start.await(); + return post("k-5", BODY); + })); + } + start.countDown(); + for (Future f : futures) { + responses.add(f.get()); + } + } + long created = responses.stream().filter(r -> r.status() == 201 && r.replayed() == null).count(); + long replayed = responses.stream().filter(r -> r.status() == 201 && r.replayed() != null).count(); + long conflicts = responses.stream().filter(r -> r.status() == 409).count(); + t.line("%d requests, same key, sent together, 500 ms of work each", n); + t.line("201 created now : %d", created); + t.line("201 replayed : %d", replayed); + t.line("409 in progress : %d", conflicts); + t.line("payments in the table: %d", payments()); + + assertThat(created).isEqualTo(1); + assertThat(payments()).isEqualTo(1); + if (storeName().equals("redis")) { + assertThat(conflicts).isEqualTo(n - 1); + } else { + assertThat(conflicts).isZero(); + assertThat(replayed).isEqualTo(n - 1); + } + } + } + + @Test + void aKeyIsForgottenAfterItsTtl() throws Exception { + try (Transcript t = transcript("06-expiry")) { + t.line("store: %s, ttl: 2 s", storeName()); + Resp first = post("k-6", BODY); + t.line(show("first request, key=k-6", first)); + Resp soon = post("k-6", BODY); + t.line(show("immediately after, key=k-6", soon)); + Thread.sleep(2500); + Resp later = post("k-6", BODY); + t.line(show("after sleeping 2.5 s, key=k-6", later)); + t.line("payments in the table: %d", payments()); + + assertThat(soon.replayed()).isEqualTo("true"); + assertThat(later.status()).isEqualTo(201); + assertThat(later.replayed()).isNull(); + assertThat(later.body()).contains("\"id\":2"); + assertThat(payments()).isEqualTo(2); + } + } + + @Test + void theCrashWindow() throws Exception { + try (Transcript t = transcript("07-crash-window")) { + t.line("store: %s, in-progress ttl: 2 s", storeName()); + faults.crashAfterBusiness = true; + Resp crashed = post("k-7", BODY); + t.line(show("attempt 1 (process dies after the write)", crashed)); + t.line("payments in the table: %d", payments()); + faults.crashAfterBusiness = false; + + Resp retry = post("k-7", BODY); + t.line(show("attempt 2, immediately", retry)); + t.line("payments in the table: %d", payments()); + + Resp last; + if (storeName().equals("redis")) { + Thread.sleep(2300); + last = post("k-7", BODY); + t.line(show("attempt 3, after the lock expired", last)); + t.line("payments in the table: %d", payments()); + assertThat(crashed.status()).isEqualTo(500); + assertThat(retry.status()).isEqualTo(409); + assertThat(last.status()).isEqualTo(201); + assertThat(last.replayed()).isNull(); + assertThat(payments()).isEqualTo(2); + } else { + last = post("k-7", BODY); + t.line(show("attempt 3", last)); + t.line("payments in the table: %d", payments()); + assertThat(crashed.status()).isEqualTo(500); + assertThat(retry.status()).isEqualTo(201); + assertThat(retry.replayed()).isNull(); + assertThat(last.replayed()).isEqualTo("true"); + assertThat(payments()).isEqualTo(1); + } + } + } + + @Test + void aHandlerSlowerThanTheLockTtl() throws Exception { + try (Transcript t = transcript("08-slow-handler")) { + t.line("store: %s, in-progress ttl: 2 s, handler takes 3.2 s", storeName()); + faults.slowMillis = 3200; + try (ExecutorService pool = Executors.newVirtualThreadPerTaskExecutor()) { + Future a = pool.submit(() -> post("k-8", BODY)); + Thread.sleep(2500); + Future b = pool.submit(() -> post("k-8", BODY)); + Resp ra = a.get(); + Resp rb = b.get(); + t.line(show("request A (t=0 s)", ra)); + t.line(show("request B (t=2.5 s, same key)", rb)); + t.line("payments in the table: %d", payments()); + t.line("requests that finished after their lock expired: %s", + storeName().equals("redis") ? String.valueOf(lostLocks()) : "n/a (a row lock under an open transaction cannot expire)"); + t.line("what the store holds afterwards: %s", + storeName().equals("redis") ? "nothing (EXISTS idem:k-8 = " + REDIS.cli("EXISTS", "idem:k-8") + ")" : inspect("k-8")); + + assertThat(ra.status()).isEqualTo(201); + assertThat(rb.status()).isEqualTo(201); + if (storeName().equals("redis")) { + assertThat(rb.replayed()).isNull(); + assertThat(payments()).isEqualTo(2); + assertThat(lostLocks()).isEqualTo(2); // B's own lock expired too, before B finished + assertThat(REDIS.cli("EXISTS", "idem:k-8")).isEqualTo("0"); + } else { + assertThat(rb.replayed()).isEqualTo("true"); + assertThat(payments()).isEqualTo(1); + } + } + } + } +} diff --git a/idempotency/src/test/java/com/ankurm/idempotency/Transcript.java b/idempotency/src/test/java/com/ankurm/idempotency/Transcript.java new file mode 100644 index 0000000..4566a21 --- /dev/null +++ b/idempotency/src/test/java/com/ankurm/idempotency/Transcript.java @@ -0,0 +1,42 @@ +package com.ankurm.idempotency; + +import java.io.IOException; +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.ArrayList; +import java.util.List; + +/** + * Collects the lines a test prints and writes them to {@code output/NN-name.txt}. The same test + * asserts the same numbers, so a line in a transcript is something the build checks. + */ +public final class Transcript implements AutoCloseable { + + private final Path file; + private final List lines = new ArrayList<>(); + + private Transcript(Path file) { + this.file = file; + } + + public static Transcript open(String name) { + return new Transcript(Path.of("output", name)); + } + + public synchronized Transcript line(String format, Object... args) { + String text = args.length == 0 ? format : String.format(format, args); + lines.add(text); + System.out.println(text); + return this; + } + + @Override + public synchronized void close() { + try { + Files.createDirectories(file.getParent()); + Files.write(file, lines); + } catch (IOException e) { + throw new IllegalStateException(e); + } + } +}