Fence the dedup claim with a token

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.

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.
  *
  * <p>
  * 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<Row> rows = this.jdbc.query(this.select, (row) -> new Row(row.getString(1), row.getLong(2)),
-                    eventToken.iss(), eventToken.jti());
+            List<Row> 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<Row> select(SsfEventToken eventToken) {
+        return this.jdbc.query(this.select, (row) -> new Row(row.getString(1), row.getLong(2)), eventToken.iss(),
+                eventToken.jti());
     }
 
     @Override

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<SsfJtiDedupStore.Claim> 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<SsfJtiDedupStore.Claim> 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();
         }
     }