Gh 25 robust dedup

CHANGELOG.md

@@ -21,8 +21,24 @@ The notable changes of every release. The section of a version is the text of it
   optional timeout per request. `TestTransmitter` holds long polls for `setLongPollHold`.
   `SsfPoller.setThreadFactory` lets a framework create the polling thread, or use a virtual one.
 
+- A SET is claimed in the `SsfJtiDedupStore` while its handlers run (`Claim.IN_PROGRESS`): another
+  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.
+
+- Schema migration: `easyssf-receiver-jdbc` ships versioned migration scripts in
+  `org/easyssf/receiver/jdbc/migration/` (Flyway naming, plain SQL), and `JdbcSsfSchema.prepareTable`
+  adds the columns a release introduced to an existing table where it may create tables
+  (`initialize-schema` `embedded` or `always`); otherwise it names the statements to run.
+
 ### 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
+  `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.
 - The schema of `easyssf-receiver-jdbc` has a third table, `EASYSSF_POLL_ACK`; create it with the
   statements of `schema.sql` where the tables are not created on startup.
 - `SsfPoller` polls on a thread of its own instead of a scheduled executor, and the in-memory stores

README.md

@@ -158,7 +158,9 @@ class StepUpHandler implements SsfEventHandler {
 ```
 
 Handlers must be idempotent. If a handler throws, the SET is not acknowledged and the transmitter is
-expected to deliver it again, in which case all handlers run again.
+expected to deliver it again, in which case all handlers run again, possibly on another instance of
+the application: with a shared dedup store, a SET is claimed while its handlers run, another instance
+leaves it alone meanwhile, and a claim a crashed instance left behind is taken over after a lease.
 
 `SsfSubject` is the `sub_id` of the SET: a simple subject identifier (RFC 9493: `iss_sub`, `email`,
 `opaque`, `account`, `phone_number`, `did`, `uri`, `aliases`; RFC 9967: `scim`) or a complex subject whose members

easyssf-receiver-jdbc/AGENTS.md

@@ -3,8 +3,12 @@
 The JDBC stores of the receiver in plain SQL, over the small `SsfJdbcOperations` interface that a
 `DataSource` or a framework's template implements.
 
-- The schema is `src/main/resources/org/easyssf/receiver/jdbc/schema.sql`; it has to run on H2,
-  PostgreSQL and MySQL unchanged. Prefer standard SQL to dialect features; a result limit goes
+- The schema is `src/main/resources/org/easyssf/receiver/jdbc/schema.sql`, the current state for a
+  fresh installation; it has to run on H2, PostgreSQL and MySQL unchanged. Every schema change also
+  ships a migration script `migration/V<release>__<topic>.sql` (Flyway naming) and, for a new column,
+  an entry in the table's `JdbcSsfSchema.*Upgrades` list so that `prepareTable` adds it where it may
+  create tables. Keep changes additive with defaults; `migrationScriptsLeadToTheCurrentSchema` checks
+  that the scripts end at the current schema. Prefer standard SQL to dialect features; a result limit goes
   through `SsfJdbcOperations.query(sql, maxRows, ...)` (the driver's `setMaxRows`), not through
   `FETCH FIRST` or `LIMIT`.
 - Tests run on H2 in the normal build. `PostgresJdbcSsfStoresTests` and `MySqlJdbcSsfStoresTests`
@@ -13,7 +17,8 @@ The JDBC stores of the receiver in plain SQL, over the small `SsfJdbcOperations`
   schema changes. All share `AbstractJdbcSsfStoresTests`; add store tests there.
   `AbstractJdbcSharedReceiversTests` (same three databases) runs two receivers against one
   `TestTransmitter` on shared stores: a SET handled once, acknowledgements handed over between
-  instances. The dedup race of `scratch/plans/dedup-in-progress-state.md` is not covered on purpose.
+  instances, and the race where one instance acknowledges a SET another is still handling (covered
+  by the in-progress state of the dedup store, see `scratch/plans/dedup-in-progress-state.md`).
 - The module is an explicit Java module; test compilation adds `java.naming` because H2's data
   source implements a `javax.naming` interface (see `pom.xml`). Keep that if you touch the
   surefire or compiler configuration.

easyssf-receiver-jdbc/src/main/java/org/easyssf/receiver/jdbc/JdbcSsfJtiDedupStore.java

@@ -3,6 +3,7 @@
 import java.time.Clock;
 import java.time.Duration;
 import java.time.Instant;
+import java.util.List;
 
 import org.easyssf.core.event.SsfEventToken;
 import org.easyssf.core.support.SsfAssert;
@@ -14,27 +15,48 @@
  * again after a restart.
  *
  * <p>
- * Whether a SET was seen before is decided by inserting it: the primary key of the table
- * makes that atomic. SETs are forgotten after the retention time: the store deletes them
- * when it is written to, at most once a minute, and {@link #purgeExpired()} does so on
- * demand, see {@link JdbcSsfStoreCleanup}.
+ * A claim inserts the row with the state {@code IN_PROGRESS}: the primary key of the
+ * table makes that atomic, and a second instance that meets the row leaves the SET alone.
+ * 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.
+ *
+ * <p>
+ * SETs are forgotten after the retention time: the store deletes them when it is written
+ * to, at most once a minute, and {@link #purgeExpired()} does so on demand, see
+ * {@link JdbcSsfStoreCleanup}.
  *
  * @see JdbcSsfSchema
  */
 public class JdbcSsfJtiDedupStore implements SsfJtiDedupStore, JdbcSsfExpiringStore {
 
     private static final Duration CLEANUP_INTERVAL = Duration.ofMinutes(1);
 
+    private static final String IN_PROGRESS = "IN_PROGRESS";
+
+    private static final String PROCESSED = "PROCESSED";
+
     private final SsfJdbcOperations jdbc;
 
+    private final String table;
+
     private final String insert;
 
+    private final String select;
+
+    private final String takeOver;
+
+    private final String markProcessed;
+
     private final String delete;
 
     private final String deleteExpired;
 
     private Duration retention = Duration.ofDays(7);
 
+    private Duration lease = Duration.ofSeconds(60);
+
     private Clock clock = Clock.systemUTC();
 
     private volatile Instant nextCleanup = Instant.MIN;
@@ -43,14 +65,37 @@ public class JdbcSsfJtiDedupStore implements SsfJtiDedupStore, JdbcSsfExpiringSt
      * @param jdbc used to access the database
      * @param tablePrefix the prefix of the table name, see
      * {@link JdbcSsfSchema#DEFAULT_TABLE_PREFIX}
+     * @throws IllegalStateException if the table exists without the {@code STATE} column
+     * of easyssf 0.3.0, with the statement that adds it
      */
     public JdbcSsfJtiDedupStore(SsfJdbcOperations jdbc, String tablePrefix) {
         SsfAssert.notNull(jdbc, "jdbc must not be null");
-        String table = JdbcSsfSchema.processedSetTable(tablePrefix);
+        this.table = JdbcSsfSchema.processedSetTable(tablePrefix);
         this.jdbc = jdbc;
-        this.insert = "INSERT INTO " + table + " (ISSUER, JTI, PROCESSED_AT) VALUES (?, ?, ?)";
-        this.delete = "DELETE FROM " + table + " WHERE ISSUER = ? AND JTI = ?";
-        this.deleteExpired = "DELETE FROM " + table + " WHERE PROCESSED_AT < ?";
+        this.insert = "INSERT INTO " + this.table + " (ISSUER, JTI, STATE, PROCESSED_AT) VALUES (?, ?, ?, ?)";
+        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 = ?";
+        this.markProcessed = "UPDATE " + this.table + " SET STATE = '" + PROCESSED
+                + "', PROCESSED_AT = ? WHERE ISSUER = ? AND JTI = ?";
+        this.delete = "DELETE FROM " + this.table + " WHERE ISSUER = ? AND JTI = ?";
+        this.deleteExpired = "DELETE FROM " + this.table + " WHERE PROCESSED_AT < ?";
+        requireStateColumn();
+    }
+
+    private void requireStateColumn() {
+        try {
+            this.jdbc.query("SELECT STATE FROM " + this.table + " WHERE 1 = 0", (row) -> row.getString(1));
+        }
+        catch (SsfJdbcException ex) {
+            if (this.jdbc.tableExists(this.table)) {
+                throw new IllegalStateException("The table " + this.table + " of the easyssf receiver predates "
+                        + "easyssf 0.3.0 and lacks the STATE column. Add it with: ALTER TABLE " + this.table
+                        + " ADD STATE VARCHAR(16) DEFAULT '" + PROCESSED + "' NOT NULL, or run the migration "
+                        + "scripts in " + JdbcSsfSchema.MIGRATION_LOCATION, ex);
+            }
+            throw ex;
+        }
     }
 
     /**
@@ -62,21 +107,63 @@ public void setRetention(Duration retention) {
         this.retention = retention;
     }
 
+    /**
+     * Sets how long a claim holds before the SET counts as abandoned by a crashed
+     * instance and is handled again; 60 seconds by default, longer than the longest
+     * handler.
+     */
+    public void setLease(Duration lease) {
+        SsfAssert.isTrue(lease != null && lease.isPositive(), "lease must be positive");
+        this.lease = lease;
+    }
+
     public void setClock(Clock clock) {
         SsfAssert.notNull(clock, "clock must not be null");
         this.clock = clock;
     }
 
     @Override
-    public boolean seenBefore(SsfEventToken eventToken) {
+    public Claim claim(SsfEventToken eventToken) {
         Instant now = this.clock.instant();
         removeExpired(now);
         try {
-            this.jdbc.update(this.insert, eventToken.iss(), eventToken.jti(), now.toEpochMilli());
-            return false;
+            this.jdbc.update(this.insert, eventToken.iss(), eventToken.jti(), IN_PROGRESS, now.toEpochMilli());
+            return Claim.NEW;
         }
         catch (SsfJdbcDuplicateKeyException ex) {
-            return true;
+            List<Row> rows = this.jdbc.query(this.select, (row) -> new Row(row.getString(1), row.getLong(2)),
+                    eventToken.iss(), eventToken.jti());
+            if (rows.isEmpty()) {
+                // deleted in between: the redelivery will be claimed
+                return Claim.IN_PROGRESS;
+            }
+            Row row = rows.get(0);
+            if (PROCESSED.equals(row.state())) {
+                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;
+            }
+            return Claim.IN_PROGRESS;
+        }
+    }
+
+    @Override
+    public void processed(SsfEventToken eventToken) {
+        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
+            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());
+            }
         }
     }
 
@@ -93,10 +180,12 @@ public int purgeExpired() {
     }
 
     private void removeExpired(Instant now) {
-        if (now.isBefore(this.nextCleanup)) {
-            return;
+        if (now.isAfter(this.nextCleanup)) {
+            purgeExpired();
         }
-        purgeExpired();
+    }
+
+    private record Row(String state, long since) {
     }
 
 }

easyssf-receiver-jdbc/src/main/java/org/easyssf/receiver/jdbc/JdbcSsfSchema.java

@@ -22,6 +22,23 @@ public final class JdbcSsfSchema {
      */
     public static final String SCHEMA_LOCATION = "classpath:org/easyssf/receiver/jdbc/schema.sql";
 
+    /**
+     * Where the versioned migration scripts are, one per release that changed the schema,
+     * named for Flyway ({@code V0_1_0__...sql}) and plain SQL for any other tool.
+     */
+    public static final String MIGRATION_LOCATION = "classpath:org/easyssf/receiver/jdbc/migration/";
+
+    /**
+     * A column a current table has and the statement that adds it to a table of an older
+     * release.
+     *
+     * @param column the name of the column
+     * @param statement the {@code ALTER TABLE} that adds it, with a default for the rows
+     * that exist
+     */
+    public record Upgrade(String column, String statement) {
+    }
+
     private JdbcSsfSchema() {
     }
 
@@ -62,11 +79,22 @@ public static List<String> createProcessedSetTable(String tablePrefix) {
                 CREATE TABLE %s (
                     ISSUER VARCHAR(255) NOT NULL,
                     JTI VARCHAR(255) NOT NULL,
+                    STATE VARCHAR(16) NOT NULL,
                     PROCESSED_AT BIGINT NOT NULL,
                     CONSTRAINT %s_PK PRIMARY KEY (ISSUER, JTI)
                 )""".formatted(table, name), "CREATE INDEX %s_IX1 ON %s (PROCESSED_AT)".formatted(name, table));
     }
 
+    /**
+     * The columns later releases added to the table of the {@link JdbcSsfJtiDedupStore}:
+     * {@code STATE} in 0.3.0.
+     */
+    public static List<Upgrade> processedSetUpgrades(String tablePrefix) {
+        String table = processedSetTable(tablePrefix);
+        return List
+            .of(new Upgrade("STATE", "ALTER TABLE " + table + " ADD STATE VARCHAR(16) DEFAULT 'PROCESSED' NOT NULL"));
+    }
+
     /**
      * The statements that create the table of the {@link JdbcSsfTokenRevocationStore}.
      */
@@ -115,15 +143,66 @@ ERROR_DESCRIPTION VARCHAR(1024),
      */
     public static void prepareTable(SsfJdbcOperations jdbc, String table, List<String> createStatements, boolean create,
             String hint) {
-        if (jdbc.tableExists(table)) {
+        prepareTable(jdbc, table, createStatements, List.of(), create, hint);
+    }
+
+    /**
+     * Makes sure a table exists and has the columns of this release.
+     * @param jdbc runs the statements
+     * @param table the name of the table
+     * @param createStatements the statements that create it
+     * @param upgrades the columns older releases lack and the statements that add them
+     * @param create whether to create a missing table, and to add missing columns to an
+     * existing one
+     * @param hint what else the application could do, appended to the message of the
+     * exception, may be {@code null}
+     * @throws IllegalStateException if the table does not exist or lacks a column and is
+     * not to be changed; the message carries the statements to run
+     */
+    public static void prepareTable(SsfJdbcOperations jdbc, String table, List<String> createStatements,
+            List<Upgrade> upgrades, boolean create, String hint) {
+        String orElse = (hint != null && !hint.isBlank()) ? ", " + hint : "";
+        if (!jdbc.tableExists(table)) {
+            if (!create) {
+                throw new IllegalStateException("The table " + table + " of the easyssf receiver does not exist. "
+                        + "Create it with the statements in " + SCHEMA_LOCATION + orElse + ".");
+            }
+            createStatements.forEach(jdbc::execute);
+            return;
+        }
+        List<Upgrade> missing = upgrades.stream()
+            .filter((upgrade) -> !columnExists(jdbc, table, upgrade.column()))
+            .toList();
+        if (missing.isEmpty()) {
             return;
         }
         if (!create) {
-            throw new IllegalStateException("The table " + table + " of the easyssf receiver does not exist. "
-                    + "Create it with the statements in " + SCHEMA_LOCATION
-                    + ((hint != null && !hint.isBlank()) ? ", " + hint : "") + ".");
+            throw new IllegalStateException("The table " + table + " of the easyssf receiver predates this release "
+                    + "and lacks the column(s) " + missing.stream().map(Upgrade::column).toList() + ". Run "
+                    + missing.stream().map(Upgrade::statement).toList() + " or the migration scripts in "
+                    + MIGRATION_LOCATION + orElse + ".");
+        }
+        for (Upgrade upgrade : missing) {
+            try {
+                jdbc.execute(upgrade.statement());
+            }
+            catch (SsfJdbcException ex) {
+                // another instance starting at the same time may have added the column
+                if (!columnExists(jdbc, table, upgrade.column())) {
+                    throw ex;
+                }
+            }
+        }
+    }
+
+    private static boolean columnExists(SsfJdbcOperations jdbc, String table, String column) {
+        try {
+            jdbc.query("SELECT " + column + " FROM " + table + " WHERE 1 = 0", (row) -> row.getString(1));
+            return true;
+        }
+        catch (SsfJdbcException ex) {
+            return false;
         }
-        createStatements.forEach(jdbc::execute);
     }
 
 }

easyssf-receiver-jdbc/src/main/resources/org/easyssf/receiver/jdbc/migration/V0_1_0__processed_sets_and_revocations.sql

@@ -0,0 +1,20 @@
+-- easyssf 0.1.0: the tables of the JDBC stores, with the default table prefix.
+-- Timestamps are milliseconds since the epoch. On Oracle use NUMBER(19) instead of BIGINT.
+
+CREATE TABLE EASYSSF_PROCESSED_SET (
+    ISSUER VARCHAR(255) NOT NULL,
+    JTI VARCHAR(255) NOT NULL,
+    PROCESSED_AT BIGINT NOT NULL,
+    CONSTRAINT EASYSSF_PROCESSED_SET_PK PRIMARY KEY (ISSUER, JTI)
+);
+CREATE INDEX EASYSSF_PROCESSED_SET_IX1 ON EASYSSF_PROCESSED_SET (PROCESSED_AT);
+
+CREATE TABLE EASYSSF_REVOCATION (
+    KIND VARCHAR(16) NOT NULL,
+    ISSUER VARCHAR(255) NOT NULL,
+    ID VARCHAR(255) NOT NULL,
+    REVOKED_AT BIGINT NOT NULL,
+    EXPIRES_AT BIGINT NOT NULL,
+    CONSTRAINT EASYSSF_REVOCATION_PK PRIMARY KEY (KIND, ISSUER, ID)
+);
+CREATE INDEX EASYSSF_REVOCATION_IX1 ON EASYSSF_REVOCATION (EXPIRES_AT);