|
1 | 1 | package io.runcycles.protocol.api; |
2 | 2 |
|
| 3 | +import io.micrometer.core.instrument.Counter; |
| 4 | +import io.micrometer.core.instrument.MeterRegistry; |
3 | 5 | import org.junit.jupiter.api.DisplayName; |
4 | 6 | import org.junit.jupiter.api.Nested; |
5 | 7 | import org.junit.jupiter.api.Test; |
| 8 | +import org.springframework.beans.factory.annotation.Autowired; |
6 | 9 | import org.springframework.http.ResponseEntity; |
7 | 10 | import redis.clients.jedis.Jedis; |
8 | 11 |
|
9 | 12 | import java.util.HashMap; |
| 13 | +import java.util.HashSet; |
10 | 14 | import java.util.Map; |
| 15 | +import java.util.Set; |
11 | 16 | import java.util.UUID; |
| 17 | +import java.util.concurrent.ConcurrentLinkedQueue; |
| 18 | +import java.util.concurrent.CountDownLatch; |
| 19 | +import java.util.concurrent.ExecutorService; |
| 20 | +import java.util.concurrent.Executors; |
| 21 | +import java.util.concurrent.TimeUnit; |
| 22 | +import java.util.concurrent.atomic.AtomicInteger; |
12 | 23 |
|
13 | 24 | import static org.assertj.core.api.Assertions.assertThat; |
14 | 25 |
|
|
34 | 45 | @DisplayName("Idempotency cache expiry") |
35 | 46 | class IdempotencyCacheExpiryIntegrationTest extends BaseIntegrationTest { |
36 | 47 |
|
| 48 | + @Autowired |
| 49 | + private MeterRegistry meterRegistry; |
| 50 | + |
| 51 | + /** Sum of counts across every counter whose tags include the given filters. */ |
| 52 | + private double counterCount(String name, String... kvs) { |
| 53 | + var search = meterRegistry.find(name); |
| 54 | + for (int i = 0; i + 1 < kvs.length; i += 2) { |
| 55 | + search = search.tag(kvs[i], kvs[i + 1]); |
| 56 | + } |
| 57 | + return search.counters().stream().mapToDouble(Counter::count).sum(); |
| 58 | + } |
| 59 | + |
37 | 60 | @Nested |
38 | 61 | @DisplayName("(A) reserve-cache expiry") |
39 | 62 | class ReserveCacheExpiry { |
@@ -128,4 +151,119 @@ void retryAfterCommittedIdempotencyKeyScrubbedReturnsFinalized() { |
128 | 151 | assertThat(retry.getBody().get("error")).isEqualTo("RESERVATION_FINALIZED"); |
129 | 152 | } |
130 | 153 | } |
| 154 | + |
| 155 | + @Nested |
| 156 | + @DisplayName("(C) thundering-herd retry") |
| 157 | + class ThunderingHerd { |
| 158 | + |
| 159 | + /** |
| 160 | + * The ops-realistic pattern when an idempotency cache has expired: a client |
| 161 | + * that originally timed out fires N concurrent retries with the same key. |
| 162 | + * All N arrive at the server within microseconds of each other, all N see |
| 163 | + * the idempotency cache gone, all N race through reserve.lua. |
| 164 | + * |
| 165 | + * Correctness depends on Redis's single-threaded Lua execution serialising |
| 166 | + * the calls: the first one to grab the script slot creates the reservation |
| 167 | + * and writes the idempotency cache; the rest see the cache and return the |
| 168 | + * same id. Without this test, that guarantee is a documented assumption |
| 169 | + * only — a future refactor (idempotency logic moved out of Lua into Java, |
| 170 | + * for instance) could silently violate it. |
| 171 | + */ |
| 172 | + @Test |
| 173 | + @DisplayName("N concurrent retries after cache expiry produce exactly one reservation") |
| 174 | + void concurrentRetriesAfterExpiryProduceOneReservation() throws Exception { |
| 175 | + String idempotencyKey = UUID.randomUUID().toString(); |
| 176 | + int concurrency = 10; |
| 177 | + |
| 178 | + // Prime the cache with a first reserve, then nuke it to simulate post-TTL. |
| 179 | + Map<String, Object> body = reservationBody(TENANT_A, 1_000); |
| 180 | + body.put("idempotency_key", idempotencyKey); |
| 181 | + ResponseEntity<Map> first = post("/v1/reservations", API_KEY_SECRET_A, body); |
| 182 | + assertThat(first.getStatusCode().value()).isEqualTo(200); |
| 183 | + String firstId = (String) first.getBody().get("reservation_id"); |
| 184 | + |
| 185 | + try (Jedis jedis = jedisPool.getResource()) { |
| 186 | + String idemKey = "idem:" + TENANT_A + ":reserve:" + idempotencyKey; |
| 187 | + jedis.del(idemKey); |
| 188 | + jedis.del(idemKey + ":hash"); |
| 189 | + } |
| 190 | + |
| 191 | + double okBefore = counterCount("cycles.reservations.reserve", |
| 192 | + "tenant", TENANT_A, "decision", "ALLOW", "reason", "OK"); |
| 193 | + double replayBefore = counterCount("cycles.reservations.reserve", |
| 194 | + "tenant", TENANT_A, "decision", "ALLOW", "reason", "IDEMPOTENT_REPLAY"); |
| 195 | + |
| 196 | + // Fire N retries through the server simultaneously via a CountDownLatch. |
| 197 | + ExecutorService exec = Executors.newFixedThreadPool(concurrency); |
| 198 | + CountDownLatch start = new CountDownLatch(1); |
| 199 | + ConcurrentLinkedQueue<String> returnedIds = new ConcurrentLinkedQueue<>(); |
| 200 | + ConcurrentLinkedQueue<Integer> statuses = new ConcurrentLinkedQueue<>(); |
| 201 | + AtomicInteger errors = new AtomicInteger(); |
| 202 | + CountDownLatch done = new CountDownLatch(concurrency); |
| 203 | + |
| 204 | + for (int i = 0; i < concurrency; i++) { |
| 205 | + exec.submit(() -> { |
| 206 | + try { |
| 207 | + start.await(); |
| 208 | + ResponseEntity<Map> resp = post("/v1/reservations", API_KEY_SECRET_A, body); |
| 209 | + statuses.add(resp.getStatusCode().value()); |
| 210 | + if (resp.getStatusCode().is2xxSuccessful()) { |
| 211 | + String id = (String) resp.getBody().get("reservation_id"); |
| 212 | + if (id != null) returnedIds.add(id); |
| 213 | + } |
| 214 | + } catch (Exception e) { |
| 215 | + errors.incrementAndGet(); |
| 216 | + } finally { |
| 217 | + done.countDown(); |
| 218 | + } |
| 219 | + }); |
| 220 | + } |
| 221 | + start.countDown(); |
| 222 | + assertThat(done.await(30, TimeUnit.SECONDS)).isTrue(); |
| 223 | + exec.shutdown(); |
| 224 | + |
| 225 | + // I1: exactly one distinct reservation id returned across all N retries. |
| 226 | + // The first call re-creates a NEW reservation (cache was cleared); subsequent |
| 227 | + // retries land on the newly-written idempotency cache and replay it. |
| 228 | + Set<String> distinct = new HashSet<>(returnedIds); |
| 229 | + assertThat(distinct) |
| 230 | + .as("N=%d retries produced multiple reservations: %s", concurrency, distinct) |
| 231 | + .hasSize(1); |
| 232 | + String winningId = distinct.iterator().next(); |
| 233 | + assertThat(winningId) |
| 234 | + .as("new reservation id should differ from pre-expiry id") |
| 235 | + .isNotEqualTo(firstId); |
| 236 | + |
| 237 | + // I2: every request got a clean 200. |
| 238 | + assertThat(new HashSet<>(statuses)).containsExactly(200); |
| 239 | + assertThat(errors.get()).as("no HTTP errors expected").isZero(); |
| 240 | + |
| 241 | + // I3: exactly one Redis reservation hash exists for this id (no duplicates). |
| 242 | + try (Jedis jedis = jedisPool.getResource()) { |
| 243 | + Map<String, String> hash = jedis.hgetAll("reservation:res_" + winningId); |
| 244 | + assertThat(hash).isNotEmpty(); |
| 245 | + assertThat(hash.get("idempotency_key")).isEqualTo(idempotencyKey); |
| 246 | + } |
| 247 | + |
| 248 | + // I4: metric tags reflect reality. Exactly one reserve.lua call produced a |
| 249 | + // real reservation (reason=OK); the remaining N-1 took the idempotent-replay |
| 250 | + // branch (reason=IDEMPOTENT_REPLAY). A wrong-tag regression (e.g. marking |
| 251 | + // all N as OK, or all N as REPLAY) would surface here. |
| 252 | + double okDelta = counterCount("cycles.reservations.reserve", |
| 253 | + "tenant", TENANT_A, "decision", "ALLOW", "reason", "OK") - okBefore; |
| 254 | + double replayDelta = counterCount("cycles.reservations.reserve", |
| 255 | + "tenant", TENANT_A, "decision", "ALLOW", "reason", "IDEMPOTENT_REPLAY") - replayBefore; |
| 256 | + |
| 257 | + assertThat(okDelta + replayDelta) |
| 258 | + .as("total reserve counter increments must equal concurrency (%d); ok=%.0f replay=%.0f", |
| 259 | + concurrency, okDelta, replayDelta) |
| 260 | + .isEqualTo((double) concurrency); |
| 261 | + assertThat(okDelta) |
| 262 | + .as("exactly one retry re-created the reservation (OK); rest replayed") |
| 263 | + .isEqualTo(1.0); |
| 264 | + assertThat(replayDelta) |
| 265 | + .as("remaining %d retries must have tagged IDEMPOTENT_REPLAY", concurrency - 1) |
| 266 | + .isEqualTo((double) (concurrency - 1)); |
| 267 | + } |
| 268 | + } |
131 | 269 | } |
0 commit comments