Rename PROCESSED_AT to STATE_CHANGED_AT

CHANGELOG.md

@@ -6,6 +6,14 @@ The notable changes of every release. The section of a version is the text of it
 
 ## [Unreleased]
 
+### Changed
+
+- The column `PROCESSED_AT` of `EASYSSF_PROCESSED_SET` is now `STATE_CHANGED_AT`: since 0.3.0 it has
+  recorded the time of the claim as well as that of the completion, and it fences the claim. Existing
+  tables need `ALTER TABLE EASYSSF_PROCESSED_SET RENAME COLUMN PROCESSED_AT TO STATE_CHANGED_AT` (the
+  migration script `V0_4_0__state_changed_at.sql`; `RENAME COLUMN` works on H2, PostgreSQL and MySQL 8),
+  applied on startup where tables may be created and named in the error otherwise.
+
 ## [0.3.0] - 2026-10-06
 
 POLL delivery that survives restarts and runs on several instances: acknowledgements wait in a

easyssf-receiver-jdbc/AGENTS.md

@@ -5,9 +5,10 @@ The JDBC stores of the receiver in plain SQL, over the small `SsfJdbcOperations`
 
 - 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
+  ships a migration script `migration/V<release>__<topic>.sql` (Flyway naming) and, for a new or
+  renamed column, an entry in the table's `JdbcSsfSchema.*Upgrades` list so that `prepareTable` adds
+  it where it may create tables (the entry names the column that is missing afterwards and the
+  statement that makes it exist, an `ADD` or a `RENAME COLUMN`). 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`.

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

@@ -19,13 +19,13 @@
  * <p>
  * 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. 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.
+ * The handlers' success turns the state into {@code PROCESSED}. {@code STATE_CHANGED_AT}
+ * (called {@code PROCESSED_AT} before 0.4.0) 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. That
+ * {@code STATE_CHANGED_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
@@ -77,40 +77,40 @@ public JdbcSsfJtiDedupStore(SsfJdbcOperations jdbc, String tablePrefix) {
         SsfAssert.notNull(jdbc, "jdbc must not be null");
         this.table = JdbcSsfSchema.processedSetTable(tablePrefix);
         this.jdbc = jdbc;
-        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 = ?";
-        // processed and forget are fenced by the claim's token, the PROCESSED_AT of the
+        this.insert = "INSERT INTO " + this.table + " (ISSUER, JTI, STATE, STATE_CHANGED_AT) VALUES (?, ?, ?, ?)";
+        this.select = "SELECT STATE, STATE_CHANGED_AT FROM " + this.table + " WHERE ISSUER = ? AND JTI = ?";
+        this.takeOver = "UPDATE " + this.table + " SET STATE_CHANGED_AT = ? WHERE ISSUER = ? AND JTI = ? AND STATE = '"
+                + IN_PROGRESS + "' AND STATE_CHANGED_AT = ?";
+        // processed and forget are fenced by the claim's token, the STATE_CHANGED_AT of
+        // the
         // claim
         this.markProcessed = "UPDATE " + this.table + " SET STATE = '" + PROCESSED
-                + "', PROCESSED_AT = ? WHERE ISSUER = ? AND JTI = ? AND STATE = '" + IN_PROGRESS
-                + "' AND PROCESSED_AT = ?";
+                + "', STATE_CHANGED_AT = ? WHERE ISSUER = ? AND JTI = ? AND STATE = '" + IN_PROGRESS
+                + "' AND STATE_CHANGED_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();
+                + "' AND STATE_CHANGED_AT = ?";
+        this.deleteExpired = "DELETE FROM " + this.table + " WHERE STATE_CHANGED_AT < ?";
+        requireCurrentColumns(tablePrefix);
     }
 
-    private void requireStateColumn() {
-        try {
-            this.jdbc.query("SELECT STATE FROM " + this.table + " WHERE 1 = 0", (row) -> row.getString(1));
+    /**
+     * Fails with the statements to run if the table predates this release: {@code STATE}
+     * came with 0.3.0, {@code STATE_CHANGED_AT} with 0.4.0.
+     */
+    private void requireCurrentColumns(String tablePrefix) {
+        if (!this.jdbc.tableExists(this.table)) {
+            return; // the first access reports the missing table
         }
-        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;
+        List<JdbcSsfSchema.Upgrade> missing = JdbcSsfSchema.missingUpgrades(this.jdbc, this.table,
+                JdbcSsfSchema.processedSetUpgrades(tablePrefix));
+        if (!missing.isEmpty()) {
+            throw new IllegalStateException("The table " + this.table + " of the easyssf receiver predates this "
+                    + "release and lacks the column(s) " + missing.stream().map(JdbcSsfSchema.Upgrade::column).toList()
+                    + ". Run " + missing.stream().map(JdbcSsfSchema.Upgrade::statement).toList()
+                    + " or the migration scripts in " + JdbcSsfSchema.MIGRATION_LOCATION);
         }
     }
 
-    /**
-     * Sets how long a processed SET is remembered. It has to cover the time a transmitter
-     * keeps trying to deliver a SET.
-     */
     public void setRetention(Duration retention) {
         SsfAssert.isTrue(retention != null && retention.isPositive(), "retention must be positive");
         this.retention = retention;
@@ -153,7 +153,8 @@ public Claim claim(SsfEventToken eventToken) {
             Instant since = Instant.ofEpochMilli(row.since());
             if (!since.plus(this.lease).isAfter(now)) {
                 // 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
+                // STATE_CHANGED_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();

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

@@ -80,19 +80,23 @@ public static List<String> createProcessedSetTable(String tablePrefix) {
                     ISSUER VARCHAR(255) NOT NULL,
                     JTI VARCHAR(255) NOT NULL,
                     STATE VARCHAR(16) NOT NULL,
-                    PROCESSED_AT BIGINT NOT NULL,
+                    STATE_CHANGED_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));
+                )""".formatted(table, name), "CREATE INDEX %s_IX1 ON %s (STATE_CHANGED_AT)".formatted(name, table));
     }
 
     /**
      * The columns later releases added to the table of the {@link JdbcSsfJtiDedupStore}:
-     * {@code STATE} in 0.3.0.
+     * {@code STATE} in 0.3.0, and {@code STATE_CHANGED_AT} in 0.4.0, which is
+     * {@code PROCESSED_AT} renamed ({@code RENAME COLUMN}, which H2, PostgreSQL and MySQL
+     * 8 support) because it records the claim as well as the completion.
      */
     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"));
+        return List.of(
+                new Upgrade("STATE", "ALTER TABLE " + table + " ADD STATE VARCHAR(16) DEFAULT 'PROCESSED' NOT NULL"),
+                new Upgrade("STATE_CHANGED_AT",
+                        "ALTER TABLE " + table + " RENAME COLUMN PROCESSED_AT TO STATE_CHANGED_AT"));
     }
 
     /**
@@ -170,9 +174,7 @@ public static void prepareTable(SsfJdbcOperations jdbc, String table, List<Strin
             createStatements.forEach(jdbc::execute);
             return;
         }
-        List<Upgrade> missing = upgrades.stream()
-            .filter((upgrade) -> !columnExists(jdbc, table, upgrade.column()))
-            .toList();
+        List<Upgrade> missing = missingUpgrades(jdbc, table, upgrades);
         if (missing.isEmpty()) {
             return;
         }
@@ -195,6 +197,13 @@ public static void prepareTable(SsfJdbcOperations jdbc, String table, List<Strin
         }
     }
 
+    /**
+     * @return the upgrades whose column the existing table lacks
+     */
+    public static List<Upgrade> missingUpgrades(SsfJdbcOperations jdbc, String table, List<Upgrade> upgrades) {
+        return upgrades.stream().filter((upgrade) -> !columnExists(jdbc, table, upgrade.column())).toList();
+    }
+
     private static boolean columnExists(SsfJdbcOperations jdbc, String table, String column) {
         try {
             jdbc.query("SELECT " + column + " FROM " + table + " WHERE 1 = 0", (row) -> row.getString(1));

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

@@ -0,0 +1,5 @@
+-- easyssf 0.4.0: PROCESSED_AT of EASYSSF_PROCESSED_SET becomes STATE_CHANGED_AT. The column has recorded the time
+-- of the claim as well as that of the completion since 0.3.0, and it fences the claim; the name now says so.
+-- RENAME COLUMN is supported by H2, PostgreSQL and MySQL 8; the index EASYSSF_PROCESSED_SET_IX1 follows the column.
+
+ALTER TABLE EASYSSF_PROCESSED_SET RENAME COLUMN PROCESSED_AT TO STATE_CHANGED_AT;

easyssf-receiver-jdbc/src/main/resources/org/easyssf/receiver/jdbc/schema.sql

@@ -3,16 +3,16 @@
 -- Installations that migrate from release to release use the scripts in migration/ instead, one per release
 -- that changed the schema, named for Flyway.
 
--- SETs that were processed or are being processed (STATE 'PROCESSED' / 'IN_PROGRESS', PROCESSED_AT the time of
--- the last state change), see JdbcSsfJtiDedupStore.
+-- SETs that were processed or are being processed (STATE 'PROCESSED' / 'IN_PROGRESS', STATE_CHANGED_AT the time
+-- of the last state change, which also fences the claim), see JdbcSsfJtiDedupStore.
 CREATE TABLE EASYSSF_PROCESSED_SET (
     ISSUER VARCHAR(255) NOT NULL,
     JTI VARCHAR(255) NOT NULL,
     STATE VARCHAR(16) NOT NULL,
-    PROCESSED_AT BIGINT NOT NULL,
+    STATE_CHANGED_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 INDEX EASYSSF_PROCESSED_SET_IX1 ON EASYSSF_PROCESSED_SET (STATE_CHANGED_AT);
 
 -- revoked sessions (KIND 'SESSION') and subjects (KIND 'SUBJECT') of an issuer, see JdbcSsfTokenRevocationStore
 CREATE TABLE EASYSSF_REVOCATION (

easyssf-receiver-jdbc/src/test/java/org/easyssf/receiver/jdbc/AbstractJdbcSsfStoresTests.java

@@ -90,7 +90,7 @@ void schemaFileMatchesTheStatementsThatCreateTheTables() throws Exception {
     void migrationScriptsLeadToTheCurrentSchema() throws Exception {
         SsfJdbcOperations fresh = emptyDatabase();
         for (String script : List.of("migration/V0_1_0__processed_sets_and_revocations.sql",
-                "migration/V0_3_0__dedup_state_and_poll_acks.sql")) {
+                "migration/V0_3_0__dedup_state_and_poll_acks.sql", "migration/V0_4_0__state_changed_at.sql")) {
             statements(script).forEach(fresh::execute);
         }
         // every table has every column of this release: nothing to create or upgrade
@@ -118,8 +118,9 @@ void missingColumnsAreAddedWhereTablesMayBeCreatedAndReportedOtherwise() {
         assertThatIllegalStateException()
             .isThrownBy(() -> JdbcSsfSchema.prepareTable(this.jdbc, JdbcSsfSchema.processedSetTable(PREFIX),
                     JdbcSsfSchema.createProcessedSetTable(PREFIX), upgrades, false, "or else"))
-            .withMessageContaining("[STATE]")
+            .withMessageContaining("[STATE, STATE_CHANGED_AT]")
             .withMessageContaining("ALTER TABLE EASYSSF_PROCESSED_SET ADD STATE")
+            .withMessageContaining("ALTER TABLE EASYSSF_PROCESSED_SET RENAME COLUMN PROCESSED_AT TO STATE_CHANGED_AT")
             .withMessageContaining("migration/")
             .withMessageContaining("or else");
 
@@ -129,6 +130,34 @@ void missingColumnsAreAddedWhereTablesMayBeCreatedAndReportedOtherwise() {
         JdbcSsfSchema.prepareTable(this.jdbc, JdbcSsfSchema.processedSetTable(PREFIX),
                 JdbcSsfSchema.createProcessedSetTable(PREFIX), upgrades, true, null);
         assertThat(dedupStore().claim(set("jti-1")).isNew()).isTrue();
+        // the index moved with the column: the retention purge still uses it
+        assertThat(this.jdbc.query("SELECT STATE_CHANGED_AT FROM EASYSSF_PROCESSED_SET", (row) -> row.getLong(1)))
+            .hasSize(1);
+    }
+
+    @Test
+    void tableOf0dot3dot0IsUpgradedByRenamingTheColumn() {
+        this.jdbc.execute("DROP TABLE EASYSSF_PROCESSED_SET");
+        for (String script : List.of("migration/V0_1_0__processed_sets_and_revocations.sql",
+                "migration/V0_3_0__dedup_state_and_poll_acks.sql")) {
+            statements(script).stream()
+                .filter((statement) -> statement.contains("EASYSSF_PROCESSED_SET"))
+                .forEach(this.jdbc::execute);
+        }
+        // a row of the old release keeps its timestamp under the new name (and within the
+        // retention)
+        long processedAt = NOW.minusSeconds(60).toEpochMilli();
+        this.jdbc.update("INSERT INTO EASYSSF_PROCESSED_SET (ISSUER, JTI, STATE, PROCESSED_AT) VALUES (?, ?, ?, ?)",
+                ISSUER, "old", "PROCESSED", processedAt);
+        assertThatIllegalStateException().isThrownBy(() -> new JdbcSsfJtiDedupStore(this.jdbc, PREFIX))
+            .withMessageContaining("[STATE_CHANGED_AT]")
+            .withMessageContaining("RENAME COLUMN PROCESSED_AT TO STATE_CHANGED_AT");
+        JdbcSsfSchema.prepareTable(this.jdbc, JdbcSsfSchema.processedSetTable(PREFIX),
+                JdbcSsfSchema.createProcessedSetTable(PREFIX), JdbcSsfSchema.processedSetUpgrades(PREFIX), true, null);
+        assertThat(this.jdbc.query("SELECT STATE_CHANGED_AT FROM EASYSSF_PROCESSED_SET WHERE JTI = ?",
+                (row) -> row.getLong(1), "old"))
+            .containsExactly(processedAt);
+        assertThat(dedupStore().claim(set("old"))).isEqualTo(SsfJtiDedupStore.Claim.processed());
     }
 
     private List<String> statements(String resource) {
@@ -262,14 +291,19 @@ void processedAfterThePurgeRecordsTheSetAgain() {
     }
 
     @Test
-    void tableWithoutTheStateColumnIsRejectedWithTheMigration() {
+    void tableOfAnOldReleaseIsRejectedWithTheStatementsToRun() {
         this.jdbc.execute("DROP TABLE EASYSSF_PROCESSED_SET");
         this.jdbc
             .execute("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))");
         assertThatIllegalStateException().isThrownBy(() -> new JdbcSsfJtiDedupStore(this.jdbc, PREFIX))
-            .withMessageContaining("ALTER TABLE EASYSSF_PROCESSED_SET ADD STATE");
+            .withMessageContaining("ALTER TABLE EASYSSF_PROCESSED_SET ADD STATE")
+            .withMessageContaining("RENAME COLUMN PROCESSED_AT TO STATE_CHANGED_AT")
+            .withMessageContaining("migration/");
         this.jdbc.execute("ALTER TABLE EASYSSF_PROCESSED_SET ADD STATE VARCHAR(16) DEFAULT 'PROCESSED' NOT NULL");
+        assertThatIllegalStateException().isThrownBy(() -> new JdbcSsfJtiDedupStore(this.jdbc, PREFIX))
+            .withMessageContaining("[STATE_CHANGED_AT]");
+        this.jdbc.execute("ALTER TABLE EASYSSF_PROCESSED_SET RENAME COLUMN PROCESSED_AT TO STATE_CHANGED_AT");
         assertThat(new JdbcSsfJtiDedupStore(this.jdbc, PREFIX).claim(set("jti-1")).isNew()).isTrue();
     }
 

easyssf-receiver-spring-boot-autoconfigure/src/test/java/org/easyssf/receiver/spring/boot/jdbc/JdbcTemplateSsfJdbcOperationsTests.java

@@ -31,10 +31,10 @@ void mapsDuplicateKeysAndMissingTables() {
                 JdbcSsfSchema.createProcessedSetTable(prefix), true, null);
         assertThat(this.jdbc.tableExists(JdbcSsfSchema.processedSetTable(prefix))).isTrue();
         this.jdbc.update(
-                "INSERT INTO EASYSSF_PROCESSED_SET (ISSUER, JTI, STATE, PROCESSED_AT) VALUES (?, ?, 'PROCESSED', ?)",
+                "INSERT INTO EASYSSF_PROCESSED_SET (ISSUER, JTI, STATE, STATE_CHANGED_AT) VALUES (?, ?, 'PROCESSED', ?)",
                 "i", "j", System.currentTimeMillis());
         assertThatExceptionOfType(SsfJdbcDuplicateKeyException.class).isThrownBy(() -> this.jdbc.update(
-                "INSERT INTO EASYSSF_PROCESSED_SET (ISSUER, JTI, STATE, PROCESSED_AT) VALUES (?, ?, 'PROCESSED', ?)",
+                "INSERT INTO EASYSSF_PROCESSED_SET (ISSUER, JTI, STATE, STATE_CHANGED_AT) VALUES (?, ?, 'PROCESSED', ?)",
                 "i", "j", System.currentTimeMillis()));
         JdbcSsfJtiDedupStore store = new JdbcSsfJtiDedupStore(this.jdbc, prefix);
         store.setRetention(Duration.ofDays(1));

easyssf-receiver-spring-boot-starter/README.md

@@ -338,9 +338,10 @@ the state is kept in its database instead, without further configuration:
   at-least-once processing with concurrent deliveries suppressed, not exactly once: handlers stay
   idempotent and key their side effects by `eventContext.idempotencyKey()`. Set the lease longer than
   your longest handler.
-- **Upgrading from 0.1.0 or 0.2.0**: run `migration/V0_3_0__dedup_state_and_poll_acks.sql`, which adds
-  the column `STATE` to `EASYSSF_PROCESSED_SET` and creates `EASYSSF_POLL_ACK`; with
-  `initialize-schema` `embedded` or `always` the receiver does it on startup.
+- **Upgrading**: from 0.1.0 or 0.2.0 run `migration/V0_3_0__dedup_state_and_poll_acks.sql`, which adds
+  the column `STATE` to `EASYSSF_PROCESSED_SET` and creates `EASYSSF_POLL_ACK`; from 0.3.0 or earlier run
+  `migration/V0_4_0__state_changed_at.sql`, which renames `PROCESSED_AT` to `STATE_CHANGED_AT`. With
+  `initialize-schema` `embedded` or `always` the receiver does both on startup.
 - **Opting out**: `easyssf.receiver.jdbc.enabled=false` keeps the state in memory although the
   application has a database. Your own `SsfJtiDedupStore`, `SsfTokenRevocationStore` or
   `SsfPollAckStore` bean takes precedence in any case.