diff --git a/CHANGELOG.md b/CHANGELOG.md index 488ea80..24fc1c0 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -25,7 +25,8 @@ The notable changes of every release. The section of a version is the text of it instance that receives it meanwhile neither handles nor acknowledges it (`SsfSetInProgressException`, metrics outcome `in_progress`), so a handler failure on the first instance is not masked. A claim older than the lease (`easyssf.receiver.dedup.lease`, 60 seconds) counts as abandoned and the SET is - handled again. + handled again. The claim carries a token that fences it: an instance whose handlers outlived the + lease can neither complete nor forget the claim of the instance that took the SET over. - `SsfEventToken.idempotencyKey()` and `SsfEventContext.idempotencyKey()`, the issuer and `jti` of a SET, as the key for the side effects of a handler: processing is at least once, which the @@ -46,8 +47,9 @@ The notable changes of every release. The section of a version is the text of it ### Changed -- `SsfJtiDedupStore` has `claim`, `processed` and `forget`; `seenBefore` is a deprecated default - method. The table `EASYSSF_PROCESSED_SET` gained the column `STATE`; existing tables need +- `SsfJtiDedupStore` has `claim`, which returns a `Claim` with the `State` of the SET and the token of + the claim granted, and `processed` and `forget`, which take that claim; `seenBefore` is a deprecated + default method. The table `EASYSSF_PROCESSED_SET` gained the column `STATE`; existing tables need `ALTER TABLE EASYSSF_PROCESSED_SET ADD STATE VARCHAR(16) DEFAULT 'PROCESSED' NOT NULL` (the migration script `V0_3_0__dedup_state_and_poll_acks.sql`), applied on startup where tables may be created and named in the error otherwise. diff --git a/easyssf-receiver-jdbc/src/main/java/org/easyssf/receiver/jdbc/JdbcSsfJtiDedupStore.java b/easyssf-receiver-jdbc/src/main/java/org/easyssf/receiver/jdbc/JdbcSsfJtiDedupStore.java index e077bcd..271adee 100644 --- a/easyssf-receiver-jdbc/src/main/java/org/easyssf/receiver/jdbc/JdbcSsfJtiDedupStore.java +++ b/easyssf-receiver-jdbc/src/main/java/org/easyssf/receiver/jdbc/JdbcSsfJtiDedupStore.java @@ -22,7 +22,10 @@ * The handlers' success turns the state into {@code PROCESSED}. {@code PROCESSED_AT} is * the time of the last state change, so an {@code IN_PROGRESS} row older than the lease * (60 seconds by default) counts as abandoned by a crashed instance and the next claim - * takes it over. + * takes it over. That {@code PROCESSED_AT} is also the token of the claim: the success or + * failure of the handlers only changes the row while it still carries the token they were + * granted, so an instance that outlived the lease leaves the claim of the taker-over + * alone. * *

* SETs are forgotten after the retention time: the store deletes them when it is written @@ -78,9 +81,13 @@ public JdbcSsfJtiDedupStore(SsfJdbcOperations jdbc, String tablePrefix) { this.select = "SELECT STATE, PROCESSED_AT FROM " + this.table + " WHERE ISSUER = ? AND JTI = ?"; this.takeOver = "UPDATE " + this.table + " SET PROCESSED_AT = ? WHERE ISSUER = ? AND JTI = ? AND STATE = '" + IN_PROGRESS + "' AND PROCESSED_AT = ?"; + // processed and forget are fenced by the claim's token, the PROCESSED_AT of the + // claim this.markProcessed = "UPDATE " + this.table + " SET STATE = '" + PROCESSED - + "', PROCESSED_AT = ? WHERE ISSUER = ? AND JTI = ?"; - this.delete = "DELETE FROM " + this.table + " WHERE ISSUER = ? AND JTI = ?"; + + "', PROCESSED_AT = ? WHERE ISSUER = ? AND JTI = ? AND STATE = '" + IN_PROGRESS + + "' AND PROCESSED_AT = ?"; + this.delete = "DELETE FROM " + this.table + " WHERE ISSUER = ? AND JTI = ? AND STATE = '" + IN_PROGRESS + + "' AND PROCESSED_AT = ?"; this.deleteExpired = "DELETE FROM " + this.table + " WHERE PROCESSED_AT < ?"; requireStateColumn(); } @@ -128,50 +135,62 @@ public void setClock(Clock clock) { public Claim claim(SsfEventToken eventToken) { Instant now = this.clock.instant(); removeExpired(now); + long token = now.toEpochMilli(); try { - this.jdbc.update(this.insert, eventToken.iss(), eventToken.jti(), IN_PROGRESS, now.toEpochMilli()); - return Claim.NEW; + this.jdbc.update(this.insert, eventToken.iss(), eventToken.jti(), IN_PROGRESS, token); + return Claim.granted(token); } catch (SsfJdbcDuplicateKeyException ex) { - List rows = this.jdbc.query(this.select, (row) -> new Row(row.getString(1), row.getLong(2)), - eventToken.iss(), eventToken.jti()); + List rows = select(eventToken); if (rows.isEmpty()) { // deleted in between: the redelivery will be claimed - return Claim.IN_PROGRESS; + return Claim.inProgress(); } Row row = rows.get(0); if (PROCESSED.equals(row.state())) { - return Claim.PROCESSED; + return Claim.processed(); } Instant since = Instant.ofEpochMilli(row.since()); if (!since.plus(this.lease).isAfter(now)) { - // abandoned: take it over, unless another instance did just now - int taken = this.jdbc.update(this.takeOver, now.toEpochMilli(), eventToken.iss(), eventToken.jti(), - row.since()); - return (taken == 1) ? Claim.NEW : Claim.IN_PROGRESS; + // abandoned: take it over, unless another instance did just now. The new + // PROCESSED_AT is the token of the new claim; the old holder's token no + // longer matches, so it can neither complete nor forget this claim. + int taken = this.jdbc.update(this.takeOver, token, eventToken.iss(), eventToken.jti(), row.since()); + return (taken == 1) ? Claim.granted(token) : Claim.inProgress(); } - return Claim.IN_PROGRESS; + return Claim.inProgress(); } } @Override - public void processed(SsfEventToken eventToken) { + public void processed(SsfEventToken eventToken, Claim claim) { + SsfAssert.notNull(claim, "claim must not be null"); long now = this.clock.instant().toEpochMilli(); - int updated = this.jdbc.update(this.markProcessed, now, eventToken.iss(), eventToken.jti()); - if (updated == 0) { - // the claim was purged in between + int updated = this.jdbc.update(this.markProcessed, now, eventToken.iss(), eventToken.jti(), claim.token()); + if (updated == 0 && select(eventToken).isEmpty()) { + // the claim was purged in between: record the success anyway. A duplicate + // means + // another instance claimed the SET since, which then owns it. try { this.jdbc.update(this.insert, eventToken.iss(), eventToken.jti(), PROCESSED, now); } catch (SsfJdbcDuplicateKeyException ex) { - this.jdbc.update(this.markProcessed, now, eventToken.iss(), eventToken.jti()); + // taken over meanwhile } } + // otherwise the claim was taken over after the lease: the SET belongs to the new + // holder } @Override - public void forget(SsfEventToken eventToken) { - this.jdbc.update(this.delete, eventToken.iss(), eventToken.jti()); + public void forget(SsfEventToken eventToken, Claim claim) { + SsfAssert.notNull(claim, "claim must not be null"); + this.jdbc.update(this.delete, eventToken.iss(), eventToken.jti(), claim.token()); + } + + private List select(SsfEventToken eventToken) { + return this.jdbc.query(this.select, (row) -> new Row(row.getString(1), row.getLong(2)), eventToken.iss(), + eventToken.jti()); } @Override diff --git a/easyssf-receiver-jdbc/src/test/java/org/easyssf/receiver/jdbc/AbstractJdbcSharedReceiversTests.java b/easyssf-receiver-jdbc/src/test/java/org/easyssf/receiver/jdbc/AbstractJdbcSharedReceiversTests.java index 7c8b383..56b5628 100644 --- a/easyssf-receiver-jdbc/src/test/java/org/easyssf/receiver/jdbc/AbstractJdbcSharedReceiversTests.java +++ b/easyssf-receiver-jdbc/src/test/java/org/easyssf/receiver/jdbc/AbstractJdbcSharedReceiversTests.java @@ -202,13 +202,13 @@ public Claim claim(SsfEventToken eventToken) { } @Override - public void processed(SsfEventToken eventToken) { + public void processed(SsfEventToken eventToken, Claim claim) { throw new IllegalStateException("the instance died"); } @Override - public void forget(SsfEventToken eventToken) { - AbstractJdbcSharedReceiversTests.this.dedupStore.forget(eventToken); + public void forget(SsfEventToken eventToken, Claim claim) { + AbstractJdbcSharedReceiversTests.this.dedupStore.forget(eventToken, claim); } }); Receiver b = receiver(dedupStore()); @@ -235,12 +235,54 @@ public void forget(SsfEventToken eventToken) { assertThat(a.keys).containsExactlyElementsOf(b.keys); } + @Test + void lateFailureOfTheFirstInstanceDoesNotForgetTheSetTheSecondHandledAfterTheLease() throws Exception { + Receiver a = receiver(); + Receiver b = receiver(); + String jti = queue("session-1"); + CountDownLatch aIsHandling = new CountDownLatch(1); + CountDownLatch bHasHandled = new CountDownLatch(1); + // a's handler outlives the lease, then fails + a.afterHandling = () -> { + aIsHandling.countDown(); + try { + assertThat(bHasHandled.await(10, TimeUnit.SECONDS)).isTrue(); + } + catch (InterruptedException ex) { + Thread.currentThread().interrupt(); + } + throw new IllegalStateException("store is down"); + }; + Thread aPolls = Thread.ofVirtual().start(() -> { + try { + a.poller.pollNow(); + } + catch (RuntimeException ex) { + // the failed handler leaves the SET unacknowledged by a + } + }); + assertThat(aIsHandling.await(10, TimeUnit.SECONDS)).isTrue(); + + // the lease expires while a is still on the SET: b takes it over and handles it + this.clock.advance(Duration.ofSeconds(61)); + assertThat(b.poller.pollNow()).isEqualTo(1); + assertThat(b.handled).containsExactly(jti); + assertThat(this.transmitter.acknowledgedSets()).containsExactly(jti); + assertThat(state(jti)).isEqualTo("PROCESSED"); + + // a's late failure must not forget b's success + bHasHandled.countDown(); + aPolls.join(Duration.ofSeconds(10)); + assertThat(state(jti)).isEqualTo("PROCESSED"); + assertThat(this.dedupStore.claim(token(jti))).isEqualTo(SsfJtiDedupStore.Claim.processed()); + } + @Test void setClaimedByAnInstanceThatDiedBeforeItsHandlerRanIsHandledAfterTheLease() { Receiver b = receiver(dedupStore()); String jti = queue("session-1"); // the dead instance's claim, left behind in the shared table - assertThat(this.dedupStore.claim(token(jti))).isEqualTo(SsfJtiDedupStore.Claim.NEW); + assertThat(this.dedupStore.claim(token(jti)).isNew()).isTrue(); assertThat(b.poller.pollNow()).isEqualTo(1); assertThat(b.handled).isEmpty(); @@ -255,14 +297,15 @@ void setClaimedByAnInstanceThatDiedBeforeItsHandlerRanIsHandledAfterTheLease() { @Test void concurrentClaimsOfOneSetYieldExactlyOneNew() throws Exception { String jti = "jti-concurrent"; - assertThat(claimConcurrently(jti, 8)).containsExactly(SsfJtiDedupStore.Claim.NEW); + assertThat(claimConcurrently(jti, 8)).hasSize(1); assertThat(state(jti)).isEqualTo("IN_PROGRESS"); // the lease expired: exactly one of the concurrent take-overs wins this.clock.advance(Duration.ofSeconds(61)); - assertThat(claimConcurrently(jti, 8)).containsExactly(SsfJtiDedupStore.Claim.NEW); + List takenOver = claimConcurrently(jti, 8); + assertThat(takenOver).hasSize(1); - this.dedupStore.processed(token(jti)); + this.dedupStore.processed(token(jti), takenOver.get(0)); assertThat(claimConcurrently(jti, 8)).isEmpty(); assertThat(state(jti)).isEqualTo("PROCESSED"); } @@ -289,7 +332,7 @@ private List claimConcurrently(String jti, int threads) results.add(claim.get(30, TimeUnit.SECONDS)); } assertThat(results).doesNotContain((SsfJtiDedupStore.Claim) null); - return results.stream().filter((claim) -> claim == SsfJtiDedupStore.Claim.NEW).toList(); + return results.stream().filter(SsfJtiDedupStore.Claim::isNew).toList(); } } diff --git a/easyssf-receiver-jdbc/src/test/java/org/easyssf/receiver/jdbc/AbstractJdbcSsfStoresTests.java b/easyssf-receiver-jdbc/src/test/java/org/easyssf/receiver/jdbc/AbstractJdbcSsfStoresTests.java index b4c5947..245e816 100644 --- a/easyssf-receiver-jdbc/src/test/java/org/easyssf/receiver/jdbc/AbstractJdbcSsfStoresTests.java +++ b/easyssf-receiver-jdbc/src/test/java/org/easyssf/receiver/jdbc/AbstractJdbcSsfStoresTests.java @@ -100,7 +100,7 @@ void migrationScriptsLeadToTheCurrentSchema() throws Exception { JdbcSsfSchema.createRevocationTable(PREFIX), false, null); JdbcSsfSchema.prepareTable(fresh, JdbcSsfSchema.pollAckTable(PREFIX), JdbcSsfSchema.createPollAckTable(PREFIX), false, null); - assertThat(new JdbcSsfJtiDedupStore(fresh, PREFIX).claim(set("jti-1"))).isEqualTo(SsfJtiDedupStore.Claim.NEW); + assertThat(new JdbcSsfJtiDedupStore(fresh, PREFIX).claim(set("jti-1")).isNew()).isTrue(); JdbcSsfPollAckStore acks = new JdbcSsfPollAckStore(fresh, PREFIX); acks.record(ISSUER, SsfPendingAck.ack("jti-1")); assertThat(acks.size(ISSUER)).isEqualTo(1); @@ -128,7 +128,7 @@ void missingColumnsAreAddedWhereTablesMayBeCreatedAndReportedOtherwise() { // repeatable JdbcSsfSchema.prepareTable(this.jdbc, JdbcSsfSchema.processedSetTable(PREFIX), JdbcSsfSchema.createProcessedSetTable(PREFIX), upgrades, true, null); - assertThat(dedupStore().claim(set("jti-1"))).isEqualTo(SsfJtiDedupStore.Claim.NEW); + assertThat(dedupStore().claim(set("jti-1")).isNew()).isTrue(); } private List statements(String resource) { @@ -191,16 +191,18 @@ void processedSetsAreSharedBetweenInstances() { void claimTellsWhetherAnotherInstanceIsOnTheSet() { JdbcSsfJtiDedupStore a = dedupStore(); JdbcSsfJtiDedupStore b = dedupStore(); - assertThat(a.claim(set("jti-1"))).isEqualTo(SsfJtiDedupStore.Claim.NEW); + SsfJtiDedupStore.Claim first = a.claim(set("jti-1")); + assertThat(first.isNew()).isTrue(); // handled right now by a: b leaves it alone - assertThat(b.claim(set("jti-1"))).isEqualTo(SsfJtiDedupStore.Claim.IN_PROGRESS); - assertThat(a.claim(set("jti-1"))).isEqualTo(SsfJtiDedupStore.Claim.IN_PROGRESS); - a.processed(set("jti-1")); - assertThat(b.claim(set("jti-1"))).isEqualTo(SsfJtiDedupStore.Claim.PROCESSED); + assertThat(b.claim(set("jti-1"))).isEqualTo(SsfJtiDedupStore.Claim.inProgress()); + assertThat(a.claim(set("jti-1"))).isEqualTo(SsfJtiDedupStore.Claim.inProgress()); + a.processed(set("jti-1"), first); + assertThat(b.claim(set("jti-1"))).isEqualTo(SsfJtiDedupStore.Claim.processed()); // a failed handler forgets the claim, the redelivery is handled - assertThat(a.claim(set("jti-2"))).isEqualTo(SsfJtiDedupStore.Claim.NEW); - a.forget(set("jti-2")); - assertThat(b.claim(set("jti-2"))).isEqualTo(SsfJtiDedupStore.Claim.NEW); + SsfJtiDedupStore.Claim second = a.claim(set("jti-2")); + assertThat(second.isNew()).isTrue(); + a.forget(set("jti-2"), second); + assertThat(b.claim(set("jti-2")).isNew()).isTrue(); assertThat(this.jdbc.query("SELECT STATE FROM EASYSSF_PROCESSED_SET WHERE JTI = ?", (row) -> row.getString(1), "jti-1")) .containsExactly("PROCESSED"); @@ -212,24 +214,51 @@ void abandonedClaimIsTakenOverAfterTheLease() { JdbcSsfJtiDedupStore survivor = dedupStore(); crashed.setLease(Duration.ofSeconds(30)); survivor.setLease(Duration.ofSeconds(30)); - assertThat(crashed.claim(set("jti-1"))).isEqualTo(SsfJtiDedupStore.Claim.NEW); + SsfJtiDedupStore.Claim abandoned = crashed.claim(set("jti-1")); + assertThat(abandoned.isNew()).isTrue(); this.clock.advance(Duration.ofSeconds(29)); - assertThat(survivor.claim(set("jti-1"))).isEqualTo(SsfJtiDedupStore.Claim.IN_PROGRESS); + assertThat(survivor.claim(set("jti-1"))).isEqualTo(SsfJtiDedupStore.Claim.inProgress()); this.clock.advance(Duration.ofSeconds(1)); - assertThat(survivor.claim(set("jti-1"))).isEqualTo(SsfJtiDedupStore.Claim.NEW); + SsfJtiDedupStore.Claim takenOver = survivor.claim(set("jti-1")); + assertThat(takenOver.isNew()).isTrue(); + assertThat(takenOver.token()).isNotEqualTo(abandoned.token()); // the take-over renewed the claim - assertThat(crashed.claim(set("jti-1"))).isEqualTo(SsfJtiDedupStore.Claim.IN_PROGRESS); - survivor.processed(set("jti-1")); - assertThat(crashed.claim(set("jti-1"))).isEqualTo(SsfJtiDedupStore.Claim.PROCESSED); + assertThat(crashed.claim(set("jti-1"))).isEqualTo(SsfJtiDedupStore.Claim.inProgress()); + survivor.processed(set("jti-1"), takenOver); + assertThat(crashed.claim(set("jti-1"))).isEqualTo(SsfJtiDedupStore.Claim.processed()); + } + + @Test + void lateInstanceCanNeitherCompleteNorForgetAClaimTakenOverFromIt() { + JdbcSsfJtiDedupStore late = dedupStore(); + JdbcSsfJtiDedupStore current = dedupStore(); + SsfJtiDedupStore.Claim lateClaim = late.claim(set("jti-1")); + this.clock.advance(Duration.ofSeconds(60)); + SsfJtiDedupStore.Claim currentClaim = current.claim(set("jti-1")); + assertThat(currentClaim.isNew()).isTrue(); + // the late instance's handlers succeed after the lease: the SET belongs to the + // current holder, whose claim stays as it is + late.processed(set("jti-1"), lateClaim); + assertThat(state("jti-1")).isEqualTo("IN_PROGRESS"); + assertThat(late.claim(set("jti-1"))).isEqualTo(SsfJtiDedupStore.Claim.inProgress()); + // the late instance's handlers fail: the current claim is not deleted + late.forget(set("jti-1"), lateClaim); + assertThat(state("jti-1")).isEqualTo("IN_PROGRESS"); + current.processed(set("jti-1"), currentClaim); + assertThat(state("jti-1")).isEqualTo("PROCESSED"); + // and a processed SET is not forgotten by a stale failure either + late.forget(set("jti-1"), lateClaim); + current.forget(set("jti-1"), currentClaim); + assertThat(state("jti-1")).isEqualTo("PROCESSED"); } @Test void processedAfterThePurgeRecordsTheSetAgain() { JdbcSsfJtiDedupStore store = dedupStore(); - store.claim(set("jti-1")); + SsfJtiDedupStore.Claim claim = store.claim(set("jti-1")); this.jdbc.update("DELETE FROM EASYSSF_PROCESSED_SET"); - store.processed(set("jti-1")); - assertThat(store.claim(set("jti-1"))).isEqualTo(SsfJtiDedupStore.Claim.PROCESSED); + store.processed(set("jti-1"), claim); + assertThat(store.claim(set("jti-1"))).isEqualTo(SsfJtiDedupStore.Claim.processed()); } @Test @@ -241,27 +270,25 @@ void tableWithoutTheStateColumnIsRejectedWithTheMigration() { assertThatIllegalStateException().isThrownBy(() -> new JdbcSsfJtiDedupStore(this.jdbc, PREFIX)) .withMessageContaining("ALTER TABLE EASYSSF_PROCESSED_SET ADD STATE"); this.jdbc.execute("ALTER TABLE EASYSSF_PROCESSED_SET ADD STATE VARCHAR(16) DEFAULT 'PROCESSED' NOT NULL"); - assertThat(new JdbcSsfJtiDedupStore(this.jdbc, PREFIX).claim(set("jti-1"))) - .isEqualTo(SsfJtiDedupStore.Claim.NEW); + assertThat(new JdbcSsfJtiDedupStore(this.jdbc, PREFIX).claim(set("jti-1")).isNew()).isTrue(); } @Test void forgottenSetIsNotSeenBefore() { JdbcSsfJtiDedupStore store = dedupStore(); - store.seenBefore(set("jti-1")); - store.forget(set("jti-1")); + SsfJtiDedupStore.Claim claim = store.claim(set("jti-1")); + store.forget(set("jti-1"), claim); assertThat(store.seenBefore(set("jti-1"))).isFalse(); } @Test void processedSetsAreForgottenAfterTheRetentionTime() { JdbcSsfJtiDedupStore store = dedupStore(); - store.claim(set("jti-1")); - store.processed(set("jti-1")); + store.processed(set("jti-1"), store.claim(set("jti-1"))); this.clock.advance(Duration.ofHours(23)); - assertThat(store.claim(set("jti-1"))).isEqualTo(SsfJtiDedupStore.Claim.PROCESSED); + assertThat(store.claim(set("jti-1"))).isEqualTo(SsfJtiDedupStore.Claim.processed()); this.clock.advance(Duration.ofHours(2)); - assertThat(store.claim(set("jti-1"))).isEqualTo(SsfJtiDedupStore.Claim.NEW); + assertThat(store.claim(set("jti-1")).isNew()).isTrue(); assertThat(count("EASYSSF_PROCESSED_SET")).isEqualTo(1); } @@ -447,6 +474,11 @@ private int count(String table) { return this.jdbc.query("SELECT COUNT(*) FROM " + table, (row) -> row.getInt(1)).get(0); } + private String state(String jti) { + return this.jdbc.query("SELECT STATE FROM EASYSSF_PROCESSED_SET WHERE JTI = ?", (row) -> row.getString(1), jti) + .get(0); + } + private static SsfEventToken set(String jti) { return set("https://idp.example", jti); } diff --git a/easyssf-receiver/src/main/java/org/easyssf/receiver/event/SsfSetInProgressException.java b/easyssf-receiver/src/main/java/org/easyssf/receiver/event/SsfSetInProgressException.java index ec15896..dc6e8cc 100644 --- a/easyssf-receiver/src/main/java/org/easyssf/receiver/event/SsfSetInProgressException.java +++ b/easyssf-receiver/src/main/java/org/easyssf/receiver/event/SsfSetInProgressException.java @@ -2,7 +2,7 @@ /** * Thrown when a SET is not handled because another instance of the application is - * handling it right now (see {@code SsfJtiDedupStore.Claim#IN_PROGRESS}). Like any + * handling it right now (see {@code SsfJtiDedupStore.State#IN_PROGRESS}). Like any * {@link SsfEventHandlingException} it leaves the SET unacknowledged, so that the * transmitter delivers it again, by which time the other instance has finished with it. */ diff --git a/easyssf-receiver/src/main/java/org/easyssf/receiver/set/InMemorySsfJtiDedupStore.java b/easyssf-receiver/src/main/java/org/easyssf/receiver/set/InMemorySsfJtiDedupStore.java index 24ec843..aca24fe 100644 --- a/easyssf-receiver/src/main/java/org/easyssf/receiver/set/InMemorySsfJtiDedupStore.java +++ b/easyssf-receiver/src/main/java/org/easyssf/receiver/set/InMemorySsfJtiDedupStore.java @@ -5,6 +5,7 @@ import java.time.Instant; import java.util.LinkedHashMap; import java.util.Map; +import java.util.concurrent.atomic.AtomicLong; import java.util.concurrent.locks.ReentrantLock; import org.easyssf.core.event.SsfEventToken; @@ -27,6 +28,8 @@ public class InMemorySsfJtiDedupStore implements SsfJtiDedupStore { private Clock clock = Clock.systemUTC(); + private final AtomicLong nextToken = new AtomicLong(); + public InMemorySsfJtiDedupStore(int capacity) { int maxEntries = Math.max(1, capacity); this.entries = new LinkedHashMap<>() { @@ -60,29 +63,39 @@ public Claim claim(SsfEventToken eventToken) { try { Known entry = this.entries.get(key); if (entry == null) { - this.entries.put(key, new Known(Claim.IN_PROGRESS, now)); - return Claim.NEW; + return grant(key, now); } - if (entry.state == Claim.PROCESSED) { - return Claim.PROCESSED; + if (entry.state == State.PROCESSED) { + return Claim.processed(); } if (!entry.since.plus(this.lease).isAfter(now)) { - // abandoned: taken over - this.entries.put(key, new Known(Claim.IN_PROGRESS, now)); - return Claim.NEW; + // abandoned: taken over, with a token of its own + return grant(key, now); } - return Claim.IN_PROGRESS; + return Claim.inProgress(); } finally { this.lock.unlock(); } } + private Claim grant(String key, Instant now) { + long token = this.nextToken.incrementAndGet(); + this.entries.put(key, new Known(State.IN_PROGRESS, now, token)); + return Claim.granted(token); + } + @Override - public void processed(SsfEventToken eventToken) { + public void processed(SsfEventToken eventToken, Claim claim) { + SsfAssert.notNull(claim, "claim must not be null"); + String key = SsfJtiDedupStore.key(eventToken); this.lock.lock(); try { - this.entries.put(SsfJtiDedupStore.key(eventToken), new Known(Claim.PROCESSED, this.clock.instant())); + Known entry = this.entries.get(key); + // the claim may have been evicted meanwhile: record the success anyway + if (entry == null || holds(entry, claim)) { + this.entries.put(key, new Known(State.PROCESSED, this.clock.instant(), claim.token())); + } } finally { this.lock.unlock(); @@ -90,16 +103,28 @@ public void processed(SsfEventToken eventToken) { } @Override - public void forget(SsfEventToken eventToken) { + public void forget(SsfEventToken eventToken, Claim claim) { + SsfAssert.notNull(claim, "claim must not be null"); + String key = SsfJtiDedupStore.key(eventToken); this.lock.lock(); try { - this.entries.remove(SsfJtiDedupStore.key(eventToken)); + Known entry = this.entries.get(key); + if (entry != null && holds(entry, claim)) { + this.entries.remove(key); + } } finally { this.lock.unlock(); } } + /** + * Whether the entry is the claim given: in progress and not taken over since. + */ + private static boolean holds(Known entry, Claim claim) { + return entry.state == State.IN_PROGRESS && entry.token == claim.token(); + } + public int size() { this.lock.lock(); try { @@ -110,7 +135,7 @@ public int size() { } } - private record Known(Claim state, Instant since) { + private record Known(State state, Instant since, long token) { } } diff --git a/easyssf-receiver/src/main/java/org/easyssf/receiver/set/SsfJtiDedupStore.java b/easyssf-receiver/src/main/java/org/easyssf/receiver/set/SsfJtiDedupStore.java index e428863..d6b6ef1 100644 --- a/easyssf-receiver/src/main/java/org/easyssf/receiver/set/SsfJtiDedupStore.java +++ b/easyssf-receiver/src/main/java/org/easyssf/receiver/set/SsfJtiDedupStore.java @@ -1,6 +1,7 @@ package org.easyssf.receiver.set; import org.easyssf.core.event.SsfEventToken; +import org.easyssf.core.support.SsfAssert; /** * Remembers which SETs were processed, so that a redelivery of a handled SET is skipped @@ -17,13 +18,19 @@ * leaves the claim behind, so a claim older than the store's lease counts as abandoned * and the next instance takes the SET over; handlers have to be idempotent for that * reason as well. + * + *

+ * A granted claim carries a {@link Claim#token() token} that fences it: + * {@link #processed} and {@link #forget} only act on the claim with that token. An + * instance whose handlers outlived the lease therefore neither completes nor deletes the + * claim of the instance that took the SET over; the SET belongs to the new holder. */ public interface SsfJtiDedupStore { /** * What a store knows about a SET when it is claimed. */ - enum Claim { + enum State { /** * Not seen before, or abandoned by a crashed instance: the caller handles the @@ -44,22 +51,76 @@ enum Claim { } + /** + * The outcome of {@link #claim}: the {@link State} of the SET and, for a + * {@link State#NEW new} claim, the token that identifies this claim against a later + * take-over of the same SET. + * + * @param state what the store knew about the SET + * @param token the token of the claim granted, meaningless unless the state is + * {@link State#NEW}; a value the store chooses, for example the time of the claim + */ + record Claim(State state, long token) { + + private static final Claim IN_PROGRESS = new Claim(State.IN_PROGRESS, 0); + + private static final Claim PROCESSED = new Claim(State.PROCESSED, 0); + + public Claim { + SsfAssert.notNull(state, "state must not be null"); + } + + /** + * @return a claim granted to the caller, who handles the SET + */ + public static Claim granted(long token) { + return new Claim(State.NEW, token); + } + + /** + * @return the SET is being handled by another instance + */ + public static Claim inProgress() { + return IN_PROGRESS; + } + + /** + * @return the SET was handled before + */ + public static Claim processed() { + return PROCESSED; + } + + /** + * @return whether the claim was granted: the caller handles the SET + */ + public boolean isNew() { + return this.state == State.NEW; + } + + } + /** * Atomically records that the caller is about to handle the SET, unless it was * handled before or is being handled by another instance. + * @return the state of the SET and, if it is new, the token of the claim */ Claim claim(SsfEventToken eventToken); /** - * Records that the handlers of a claimed SET succeeded. + * Records that the handlers of a claimed SET succeeded. Does nothing if the claim was + * taken over meanwhile, because the handlers outlived the lease: the SET then belongs + * to the instance that took it over. + * @param claim the claim {@link #claim} granted for the SET */ - void processed(SsfEventToken eventToken); + void processed(SsfEventToken eventToken, Claim claim); /** * Forgets the given SET again, because handling it failed and it is expected to be - * delivered once more. + * delivered once more. Does nothing if the claim was taken over meanwhile. + * @param claim the claim {@link #claim} granted for the SET */ - void forget(SsfEventToken eventToken); + void forget(SsfEventToken eventToken, Claim claim); /** * Claims the SET and tells whether it was known before, as processed or in progress. @@ -67,7 +128,7 @@ enum Claim { */ @Deprecated(since = "0.3.0") default boolean seenBefore(SsfEventToken eventToken) { - return claim(eventToken) != Claim.NEW; + return !claim(eventToken).isNew(); } /** diff --git a/easyssf-receiver/src/main/java/org/easyssf/receiver/set/SsfSetProcessor.java b/easyssf-receiver/src/main/java/org/easyssf/receiver/set/SsfSetProcessor.java index 7deb260..6efd10a 100644 --- a/easyssf-receiver/src/main/java/org/easyssf/receiver/set/SsfSetProcessor.java +++ b/easyssf-receiver/src/main/java/org/easyssf/receiver/set/SsfSetProcessor.java @@ -159,8 +159,10 @@ public Outcome process(String encodedSet, SsfDeliveryMethod deliveryMethod) { throw ex; } issuer = eventToken.iss(); + SsfJtiDedupStore.Claim claim = null; if (this.dedupStore != null) { - switch (this.dedupStore.claim(eventToken)) { + claim = this.dedupStore.claim(eventToken); + switch (claim.state()) { case PROCESSED -> { logger.debug("Skipping SET " + eventToken.jti() + ", it was processed before"); this.metrics.setReceived(issuer, deliveryMethod, SetOutcome.DUPLICATE); @@ -199,14 +201,14 @@ public Outcome process(String encodedSet, SsfDeliveryMethod deliveryMethod) { } } if (failure != null) { - if (this.dedupStore != null) { - this.dedupStore.forget(eventToken); + if (claim != null) { + this.dedupStore.forget(eventToken, claim); } this.metrics.setReceived(issuer, deliveryMethod, SetOutcome.FAILED); throw new SsfEventHandlingException("Could not handle SET " + eventToken.jti(), failure); } - if (this.dedupStore != null) { - this.dedupStore.processed(eventToken); + if (claim != null) { + this.dedupStore.processed(eventToken, claim); } this.metrics.setReceived(issuer, deliveryMethod, SetOutcome.HANDLED); String transmitter = issuer; diff --git a/easyssf-receiver/src/test/java/org/easyssf/receiver/set/InMemorySsfJtiDedupStoreTests.java b/easyssf-receiver/src/test/java/org/easyssf/receiver/set/InMemorySsfJtiDedupStoreTests.java index c48b4ae..be2e2bb 100644 --- a/easyssf-receiver/src/test/java/org/easyssf/receiver/set/InMemorySsfJtiDedupStoreTests.java +++ b/easyssf-receiver/src/test/java/org/easyssf/receiver/set/InMemorySsfJtiDedupStoreTests.java @@ -9,6 +9,7 @@ import org.easyssf.core.event.SsfEventToken; import org.easyssf.receiver.set.SsfJtiDedupStore.Claim; +import org.easyssf.receiver.set.SsfJtiDedupStore.State; import org.junit.jupiter.api.Test; import static org.assertj.core.api.Assertions.assertThat; @@ -27,30 +28,64 @@ class InMemorySsfJtiDedupStoreTests { @Test void claimInProgressProcessedForget() { - assertThat(this.store.claim(set("jti-1"))).isEqualTo(Claim.NEW); - assertThat(this.store.claim(set("jti-1"))).isEqualTo(Claim.IN_PROGRESS); - this.store.processed(set("jti-1")); - assertThat(this.store.claim(set("jti-1"))).isEqualTo(Claim.PROCESSED); - this.store.forget(set("jti-1")); - assertThat(this.store.claim(set("jti-1"))).isEqualTo(Claim.NEW); + Claim claim = this.store.claim(set("jti-1")); + assertThat(claim.isNew()).isTrue(); + assertThat(this.store.claim(set("jti-1"))).isEqualTo(Claim.inProgress()); + this.store.processed(set("jti-1"), claim); + assertThat(this.store.claim(set("jti-1"))).isEqualTo(Claim.processed()); + // a processed SET is not forgotten, the claim is over + this.store.forget(set("jti-1"), claim); + assertThat(this.store.claim(set("jti-1"))).isEqualTo(Claim.processed()); + Claim second = this.store.claim(set("jti-2")); + this.store.forget(set("jti-2"), second); + assertThat(this.store.claim(set("jti-2")).state()).isEqualTo(State.NEW); // a jti is only unique per issuer - assertThat(this.store.claim(set("https://other.example", "jti-1"))).isEqualTo(Claim.NEW); - assertThat(this.store.size()).isEqualTo(2); + assertThat(this.store.claim(set("https://other.example", "jti-1")).state()).isEqualTo(State.NEW); + assertThat(this.store.size()).isEqualTo(3); } @Test void abandonedClaimIsTakenOverAfterTheLease() { this.store.setLease(Duration.ofSeconds(30)); - assertThat(this.store.claim(set("jti-1"))).isEqualTo(Claim.NEW); + Claim first = this.store.claim(set("jti-1")); + assertThat(first.isNew()).isTrue(); this.clock.advance(Duration.ofSeconds(29)); - assertThat(this.store.claim(set("jti-1"))).isEqualTo(Claim.IN_PROGRESS); + assertThat(this.store.claim(set("jti-1"))).isEqualTo(Claim.inProgress()); this.clock.advance(Duration.ofSeconds(1)); - assertThat(this.store.claim(set("jti-1"))).isEqualTo(Claim.NEW); - assertThat(this.store.claim(set("jti-1"))).isEqualTo(Claim.IN_PROGRESS); + Claim second = this.store.claim(set("jti-1")); + assertThat(second.isNew()).isTrue(); + assertThat(second.token()).isNotEqualTo(first.token()); + assertThat(this.store.claim(set("jti-1"))).isEqualTo(Claim.inProgress()); // processed rows do not expire with the lease - this.store.processed(set("jti-1")); + this.store.processed(set("jti-1"), second); this.clock.advance(Duration.ofHours(1)); - assertThat(this.store.claim(set("jti-1"))).isEqualTo(Claim.PROCESSED); + assertThat(this.store.claim(set("jti-1"))).isEqualTo(Claim.processed()); + } + + @Test + void lateHolderCanNeitherCompleteNorForgetTheClaimTakenOverFromIt() { + this.store.setLease(Duration.ofSeconds(30)); + Claim late = this.store.claim(set("jti-1")); + this.clock.advance(Duration.ofSeconds(30)); + Claim current = this.store.claim(set("jti-1")); + assertThat(current.isNew()).isTrue(); + // the late holder's handlers succeed: the SET still belongs to the current holder + this.store.processed(set("jti-1"), late); + assertThat(this.store.claim(set("jti-1"))).isEqualTo(Claim.inProgress()); + // the late holder's handlers fail: the current claim stays + this.store.forget(set("jti-1"), late); + assertThat(this.store.claim(set("jti-1"))).isEqualTo(Claim.inProgress()); + this.store.processed(set("jti-1"), current); + assertThat(this.store.claim(set("jti-1"))).isEqualTo(Claim.processed()); + } + + @Test + void successOfAnEvictedClaimIsRecorded() { + InMemorySsfJtiDedupStore small = new InMemorySsfJtiDedupStore(1); + Claim claim = small.claim(set("jti-1")); + small.claim(set("jti-2")); + small.processed(set("jti-1"), claim); + assertThat(small.claim(set("jti-1"))).isEqualTo(Claim.processed()); } @Test diff --git a/easyssf-receiver/src/test/java/org/easyssf/receiver/set/SsfSetProcessorTests.java b/easyssf-receiver/src/test/java/org/easyssf/receiver/set/SsfSetProcessorTests.java index 394b9c4..7be7e6c 100644 --- a/easyssf-receiver/src/test/java/org/easyssf/receiver/set/SsfSetProcessorTests.java +++ b/easyssf-receiver/src/test/java/org/easyssf/receiver/set/SsfSetProcessorTests.java @@ -74,15 +74,15 @@ void setBeingHandledElsewhereIsLeftForARedelivery() { SsfJtiDedupStore inProgress = new SsfJtiDedupStore() { @Override public Claim claim(org.easyssf.core.event.SsfEventToken eventToken) { - return Claim.IN_PROGRESS; + return Claim.inProgress(); } @Override - public void processed(org.easyssf.core.event.SsfEventToken eventToken) { + public void processed(org.easyssf.core.event.SsfEventToken eventToken, Claim claim) { } @Override - public void forget(org.easyssf.core.event.SsfEventToken eventToken) { + public void forget(org.easyssf.core.event.SsfEventToken eventToken, Claim claim) { } }; SsfSetProcessor processor = new SsfSetProcessor(this.verifier, inProgress, List.of(this.recordingHandler)); @@ -104,10 +104,11 @@ void handledSetIsMarkedProcessedAndFailedOneForgotten() { SsfSetProcessor processor = new SsfSetProcessor(this.verifier, store, List.of(failingOnce)); assertThatExceptionOfType(SsfEventHandlingException.class).isThrownBy(() -> processor.process("jti-1")); // forgotten: the redelivery is new, not in progress - assertThat(store.claim(this.verifier.verify("jti-1"))).isEqualTo(SsfJtiDedupStore.Claim.NEW); - store.forget(this.verifier.verify("jti-1")); + SsfJtiDedupStore.Claim claim = store.claim(this.verifier.verify("jti-1")); + assertThat(claim.isNew()).isTrue(); + store.forget(this.verifier.verify("jti-1"), claim); assertThat(processor.process("jti-1")).isEqualTo(Outcome.HANDLED); - assertThat(store.claim(this.verifier.verify("jti-1"))).isEqualTo(SsfJtiDedupStore.Claim.PROCESSED); + assertThat(store.claim(this.verifier.verify("jti-1"))).isEqualTo(SsfJtiDedupStore.Claim.processed()); } @Test