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 <[email protected]> Claude-Session: https://claude.ai/code/session_017ic4b5yg8grY17JsVAJviV
This commit is contained in:
Claude Sonnet 5.5
@@ -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.
|
||||
@@ -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
|
||||
@@ -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
|
||||
@@ -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
|
||||
@@ -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
|
||||
@@ -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
|
||||
@@ -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
|
||||
@@ -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
|
||||
@@ -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
|
||||
@@ -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
|
||||
@@ -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
|
||||
@@ -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
|
||||
@@ -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
|
||||
@@ -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
|
||||
@@ -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
|
||||
@@ -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
|
||||
@@ -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
|
||||
@@ -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
|
||||
@@ -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
|
||||
@@ -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)
|
||||
@@ -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
|
||||
@@ -0,0 +1,66 @@
|
||||
<?xml version="1.0" encoding="UTF-8"?>
|
||||
<project xmlns="http://maven.apache.org/POM/4.0.0"
|
||||
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
|
||||
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
|
||||
<modelVersion>4.0.0</modelVersion>
|
||||
|
||||
<parent>
|
||||
<groupId>org.springframework.boot</groupId>
|
||||
<artifactId>spring-boot-starter-parent</artifactId>
|
||||
<version>4.1.2</version>
|
||||
<relativePath/>
|
||||
</parent>
|
||||
|
||||
<groupId>com.ankurm</groupId>
|
||||
<artifactId>idempotency</artifactId>
|
||||
<version>1.0.0</version>
|
||||
<name>idempotency</name>
|
||||
<description>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</description>
|
||||
|
||||
<properties>
|
||||
<java.version>25</java.version>
|
||||
</properties>
|
||||
|
||||
<dependencies>
|
||||
<dependency>
|
||||
<groupId>org.springframework.boot</groupId>
|
||||
<artifactId>spring-boot-starter-webmvc</artifactId>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.springframework.boot</groupId>
|
||||
<artifactId>spring-boot-starter-data-redis</artifactId>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.springframework.boot</groupId>
|
||||
<artifactId>spring-boot-starter-jdbc</artifactId>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.postgresql</groupId>
|
||||
<artifactId>postgresql</artifactId>
|
||||
<scope>runtime</scope>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.springframework.boot</groupId>
|
||||
<artifactId>spring-boot-starter-test</artifactId>
|
||||
<scope>test</scope>
|
||||
</dependency>
|
||||
</dependencies>
|
||||
|
||||
<build>
|
||||
<plugins>
|
||||
<plugin>
|
||||
<groupId>org.springframework.boot</groupId>
|
||||
<artifactId>spring-boot-maven-plugin</artifactId>
|
||||
</plugin>
|
||||
<plugin>
|
||||
<groupId>org.apache.maven.plugins</groupId>
|
||||
<artifactId>maven-surefire-plugin</artifactId>
|
||||
<configuration>
|
||||
<!-- pgjdbc sends the JVM's zone to Postgres as a startup parameter; a legacy zone id such as
|
||||
Asia/Calcutta makes Postgres 16 refuse the connection. See scripts/timezone-trap.sh. -->
|
||||
<argLine>-Duser.timezone=UTC</argLine>
|
||||
</configuration>
|
||||
</plugin>
|
||||
</plugins>
|
||||
</build>
|
||||
</project>
|
||||
Executable
+24
@@ -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
|
||||
Executable
+28
@@ -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
|
||||
@@ -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));
|
||||
}
|
||||
}
|
||||
@@ -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;
|
||||
}
|
||||
}
|
||||
@@ -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);
|
||||
}
|
||||
}
|
||||
@@ -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 <idempotency.path>}:
|
||||
* <ul>
|
||||
* <li>no {@code Idempotency-Key} header: 400</li>
|
||||
* <li>same key, different request: 422</li>
|
||||
* <li>same key while the first request is still running: 409 with {@code Retry-After}</li>
|
||||
* <li>same key after the first finished: the stored response, with {@code Idempotent-Replayed: true}</li>
|
||||
* </ul>
|
||||
* 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 + "}");
|
||||
}
|
||||
}
|
||||
@@ -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 }
|
||||
}
|
||||
@@ -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.
|
||||
*
|
||||
* <p>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);
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,3 @@
|
||||
package com.ankurm.idempotency;
|
||||
|
||||
public record Payment(long id, String customer, long amount, String currency) { }
|
||||
@@ -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<Payment> 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()));
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,3 @@
|
||||
package com.ankurm.idempotency;
|
||||
|
||||
public record PaymentRequest(String customer, long amount, String currency) { }
|
||||
@@ -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.
|
||||
*
|
||||
|
||||
* <p>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.
|
||||
*
|
||||
* <p>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> 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;
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -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.
|
||||
*
|
||||
* <pre>
|
||||
* IN_PROGRESS\n<fingerprint>\n<owner uuid> lives inProgressTtl
|
||||
* DONE\n<fingerprint>\n<status>\n<content type>\n<body> lives ttl
|
||||
* </pre>
|
||||
*
|
||||
* 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.
|
||||
*
|
||||
* <p>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<Long> 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<Long> 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);
|
||||
}
|
||||
}
|
||||
@@ -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;
|
||||
}
|
||||
}
|
||||
@@ -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
|
||||
@@ -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);
|
||||
@@ -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
|
||||
}
|
||||
}
|
||||
@@ -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<String> 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();
|
||||
}
|
||||
}
|
||||
@@ -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;
|
||||
}
|
||||
}
|
||||
@@ -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.
|
||||
*
|
||||
* <p>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<String> 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<Resp> responses = new ArrayList<>();
|
||||
try (ExecutorService pool = Executors.newVirtualThreadPerTaskExecutor()) {
|
||||
List<Future<Resp>> 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<Resp> 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<Resp> a = pool.submit(() -> post("k-8", BODY));
|
||||
Thread.sleep(2500);
|
||||
Future<Resp> 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);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -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<String> 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);
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user