diff --git a/CHANGELOG.md b/CHANGELOG.md index 17c96c9..f8c512f 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/README.md b/README.md index 8580663..eae72c6 100644 --- a/README.md +++ b/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 diff --git a/easyssf-receiver-jdbc/AGENTS.md b/easyssf-receiver-jdbc/AGENTS.md index 2ea46bb..1a071ff 100644 --- a/easyssf-receiver-jdbc/AGENTS.md +++ b/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__.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. 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 ae7ec66..857cb5a 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 @@ -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,10 +15,17 @@ * again after a restart. * *

- * 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. + * + *

+ * 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 */ @@ -25,16 +33,30 @@ public class JdbcSsfJtiDedupStore implements SsfJtiDedupStore, JdbcSsfExpiringSt 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 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) { } } diff --git a/easyssf-receiver-jdbc/src/main/java/org/easyssf/receiver/jdbc/JdbcSsfSchema.java b/easyssf-receiver-jdbc/src/main/java/org/easyssf/receiver/jdbc/JdbcSsfSchema.java index b4a3066..1e1e397 100644 --- a/easyssf-receiver-jdbc/src/main/java/org/easyssf/receiver/jdbc/JdbcSsfSchema.java +++ b/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 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 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 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 createStatements, + List 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 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); } } diff --git a/easyssf-receiver-jdbc/src/main/resources/org/easyssf/receiver/jdbc/migration/V0_1_0__processed_sets_and_revocations.sql b/easyssf-receiver-jdbc/src/main/resources/org/easyssf/receiver/jdbc/migration/V0_1_0__processed_sets_and_revocations.sql new file mode 100644 index 0000000..d47e8f8 --- /dev/null +++ b/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); diff --git a/easyssf-receiver-jdbc/src/main/resources/org/easyssf/receiver/jdbc/migration/V0_3_0__dedup_state_and_poll_acks.sql b/easyssf-receiver-jdbc/src/main/resources/org/easyssf/receiver/jdbc/migration/V0_3_0__dedup_state_and_poll_acks.sql new file mode 100644 index 0000000..02d02c6 --- /dev/null +++ b/easyssf-receiver-jdbc/src/main/resources/org/easyssf/receiver/jdbc/migration/V0_3_0__dedup_state_and_poll_acks.sql @@ -0,0 +1,14 @@ +-- easyssf 0.3.0: the state of a processed SET (IN_PROGRESS while its handlers run), and the acknowledgements a +-- polling receiver owes its transmitter. + +ALTER TABLE EASYSSF_PROCESSED_SET ADD STATE VARCHAR(16) DEFAULT 'PROCESSED' NOT NULL; + +CREATE TABLE EASYSSF_POLL_ACK ( + ISSUER VARCHAR(255) NOT NULL, + JTI VARCHAR(255) NOT NULL, + ERROR_CODE VARCHAR(64), + ERROR_DESCRIPTION VARCHAR(1024), + RECORDED_AT BIGINT NOT NULL, + CONSTRAINT EASYSSF_POLL_ACK_PK PRIMARY KEY (ISSUER, JTI) +); +CREATE INDEX EASYSSF_POLL_ACK_IX1 ON EASYSSF_POLL_ACK (RECORDED_AT); diff --git a/easyssf-receiver-jdbc/src/main/resources/org/easyssf/receiver/jdbc/schema.sql b/easyssf-receiver-jdbc/src/main/resources/org/easyssf/receiver/jdbc/schema.sql index 1696710..c5022bf 100644 --- a/easyssf-receiver-jdbc/src/main/resources/org/easyssf/receiver/jdbc/schema.sql +++ b/easyssf-receiver-jdbc/src/main/resources/org/easyssf/receiver/jdbc/schema.sql @@ -1,10 +1,14 @@ --- Tables of the JDBC stores of the easyssf receiver. +-- Tables of the JDBC stores of the easyssf receiver, as this release expects them (a fresh installation). -- Timestamps are milliseconds since the epoch. On Oracle use NUMBER(19) instead of BIGINT. +-- 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, see JdbcSsfJtiDedupStore +-- SETs that were processed or are being processed (STATE 'PROCESSED' / 'IN_PROGRESS', PROCESSED_AT the time of +-- the last state change), 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, CONSTRAINT EASYSSF_PROCESSED_SET_PK PRIMARY KEY (ISSUER, JTI) ); 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 4a51731..6d80730 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 @@ -2,9 +2,12 @@ import java.net.URI; import java.nio.charset.StandardCharsets; +import java.time.Duration; import java.util.ArrayList; import java.util.Base64; import java.util.List; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; import java.util.regex.Matcher; import java.util.regex.Pattern; @@ -131,6 +134,49 @@ void setDeliveredAgainLaterIsNotHandledTwiceByAnyInstance() { assertThat(this.transmitter.acknowledgedSets()).containsExactly(jti, jti); } + @Test + void setBeingHandledByOneInstanceIsNeitherHandledNorAcknowledgedByAnother() throws Exception { + Receiver a = receiver(); + Receiver b = receiver(); + String jti = queue("session-1"); + CountDownLatch aIsHandling = new CountDownLatch(1); + CountDownLatch bHasPolled = new CountDownLatch(1); + // a's handler blocks until b has polled, then fails + a.afterHandling = () -> { + aIsHandling.countDown(); + try { + assertThat(bHasPolled.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 + } + }); + assertThat(aIsHandling.await(10, TimeUnit.SECONDS)).isTrue(); + + // b receives the SET while a is still on it: in progress, neither handled nor + // acknowledged + assertThat(b.poller.pollNow()).isEqualTo(1); + assertThat(b.handled).isEmpty(); + assertThat(transmitter.acknowledgedSets()).isEmpty(); + bHasPolled.countDown(); + aPolls.join(Duration.ofSeconds(10)); + assertThat(transmitter.acknowledgedSets()).isEmpty(); + + // the redelivery is handled, by whichever instance polls + assertThat(b.poller.pollNow()).isEqualTo(1); + assertThat(b.handled).containsExactly(jti); + assertThat(transmitter.acknowledgedSets()).containsExactly(jti); + } + private Receiver receiver() { Receiver receiver = new Receiver(); SsfHttpClient http = new JdkSsfHttpClient(); 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 2d0ac82..b4c5947 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 @@ -13,6 +13,7 @@ import org.easyssf.core.event.SsfEventToken; import org.easyssf.receiver.poll.SsfPendingAck; +import org.easyssf.receiver.set.SsfJtiDedupStore; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; @@ -85,6 +86,64 @@ void schemaFileMatchesTheStatementsThatCreateTheTables() throws Exception { assertThat(statements).isEqualTo(expected.stream().map(AbstractJdbcSsfStoresTests::normalize).toList()); } + @Test + 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")) { + statements(script).forEach(fresh::execute); + } + // every table has every column of this release: nothing to create or upgrade + JdbcSsfSchema.prepareTable(fresh, JdbcSsfSchema.processedSetTable(PREFIX), + JdbcSsfSchema.createProcessedSetTable(PREFIX), JdbcSsfSchema.processedSetUpgrades(PREFIX), false, null); + JdbcSsfSchema.prepareTable(fresh, JdbcSsfSchema.revocationTable(PREFIX), + 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); + JdbcSsfPollAckStore acks = new JdbcSsfPollAckStore(fresh, PREFIX); + acks.record(ISSUER, SsfPendingAck.ack("jti-1")); + assertThat(acks.size(ISSUER)).isEqualTo(1); + new JdbcSsfTokenRevocationStore(fresh, PREFIX, Duration.ofMinutes(1)).revokeSession(ISSUER, "s"); + } + + @Test + void missingColumnsAreAddedWhereTablesMayBeCreatedAndReportedOtherwise() { + this.jdbc.execute("DROP TABLE EASYSSF_PROCESSED_SET"); + statements("migration/V0_1_0__processed_sets_and_revocations.sql").stream() + .filter((statement) -> statement.contains("EASYSSF_PROCESSED_SET")) + .forEach(this.jdbc::execute); + List upgrades = JdbcSsfSchema.processedSetUpgrades(PREFIX); + + assertThatIllegalStateException() + .isThrownBy(() -> JdbcSsfSchema.prepareTable(this.jdbc, JdbcSsfSchema.processedSetTable(PREFIX), + JdbcSsfSchema.createProcessedSetTable(PREFIX), upgrades, false, "or else")) + .withMessageContaining("[STATE]") + .withMessageContaining("ALTER TABLE EASYSSF_PROCESSED_SET ADD STATE") + .withMessageContaining("migration/") + .withMessageContaining("or else"); + + JdbcSsfSchema.prepareTable(this.jdbc, JdbcSsfSchema.processedSetTable(PREFIX), + JdbcSsfSchema.createProcessedSetTable(PREFIX), upgrades, true, null); + // repeatable + JdbcSsfSchema.prepareTable(this.jdbc, JdbcSsfSchema.processedSetTable(PREFIX), + JdbcSsfSchema.createProcessedSetTable(PREFIX), upgrades, true, null); + assertThat(dedupStore().claim(set("jti-1"))).isEqualTo(SsfJtiDedupStore.Claim.NEW); + } + + private List statements(String resource) { + try { + String script = new String(getClass().getResourceAsStream(resource).readAllBytes(), StandardCharsets.UTF_8); + return Arrays.stream(script.replaceAll("(?m)^--.*$", "").split(";")) + .map(AbstractJdbcSsfStoresTests::normalize) + .filter((statement) -> !statement.isEmpty()) + .toList(); + } + catch (java.io.IOException ex) { + throw new IllegalStateException(ex); + } + } + private static String normalize(String statement) { return statement.replaceAll("\\s+", " ").trim(); } @@ -128,6 +187,64 @@ void processedSetsAreSharedBetweenInstances() { assertThat(dedupStore().seenBefore(set("jti-1"))).isTrue(); } + @Test + void claimTellsWhetherAnotherInstanceIsOnTheSet() { + JdbcSsfJtiDedupStore a = dedupStore(); + JdbcSsfJtiDedupStore b = dedupStore(); + assertThat(a.claim(set("jti-1"))).isEqualTo(SsfJtiDedupStore.Claim.NEW); + // 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); + // 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); + assertThat(this.jdbc.query("SELECT STATE FROM EASYSSF_PROCESSED_SET WHERE JTI = ?", (row) -> row.getString(1), + "jti-1")) + .containsExactly("PROCESSED"); + } + + @Test + void abandonedClaimIsTakenOverAfterTheLease() { + JdbcSsfJtiDedupStore crashed = dedupStore(); + JdbcSsfJtiDedupStore survivor = dedupStore(); + crashed.setLease(Duration.ofSeconds(30)); + survivor.setLease(Duration.ofSeconds(30)); + assertThat(crashed.claim(set("jti-1"))).isEqualTo(SsfJtiDedupStore.Claim.NEW); + this.clock.advance(Duration.ofSeconds(29)); + assertThat(survivor.claim(set("jti-1"))).isEqualTo(SsfJtiDedupStore.Claim.IN_PROGRESS); + this.clock.advance(Duration.ofSeconds(1)); + assertThat(survivor.claim(set("jti-1"))).isEqualTo(SsfJtiDedupStore.Claim.NEW); + // 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); + } + + @Test + void processedAfterThePurgeRecordsTheSetAgain() { + JdbcSsfJtiDedupStore store = dedupStore(); + 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); + } + + @Test + void tableWithoutTheStateColumnIsRejectedWithTheMigration() { + 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"); + 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); + } + @Test void forgottenSetIsNotSeenBefore() { JdbcSsfJtiDedupStore store = dedupStore(); @@ -139,11 +256,12 @@ void forgottenSetIsNotSeenBefore() { @Test void processedSetsAreForgottenAfterTheRetentionTime() { JdbcSsfJtiDedupStore store = dedupStore(); - store.seenBefore(set("jti-1")); + store.claim(set("jti-1")); + store.processed(set("jti-1")); this.clock.advance(Duration.ofHours(23)); - assertThat(store.seenBefore(set("jti-1"))).isTrue(); + assertThat(store.claim(set("jti-1"))).isEqualTo(SsfJtiDedupStore.Claim.PROCESSED); this.clock.advance(Duration.ofHours(2)); - assertThat(store.seenBefore(set("jti-1"))).isFalse(); + assertThat(store.claim(set("jti-1"))).isEqualTo(SsfJtiDedupStore.Claim.NEW); assertThat(count("EASYSSF_PROCESSED_SET")).isEqualTo(1); } diff --git a/easyssf-receiver-spring-boot-autoconfigure/src/main/java/org/easyssf/receiver/spring/boot/SsfReceiverProperties.java b/easyssf-receiver-spring-boot-autoconfigure/src/main/java/org/easyssf/receiver/spring/boot/SsfReceiverProperties.java index 7241be8..a7df615 100644 --- a/easyssf-receiver-spring-boot-autoconfigure/src/main/java/org/easyssf/receiver/spring/boot/SsfReceiverProperties.java +++ b/easyssf-receiver-spring-boot-autoconfigure/src/main/java/org/easyssf/receiver/spring/boot/SsfReceiverProperties.java @@ -410,6 +410,14 @@ public static class Dedup { */ private Duration retention = Duration.ofDays(7); + /** + * How long a SET counts as being handled by the instance that claimed it. Another + * instance that receives the SET in that time leaves it for a redelivery; after + * it, the claim counts as abandoned by a crashed instance and the SET is handled + * again. Longer than the longest event handler. + */ + private Duration lease = Duration.ofSeconds(60); + public boolean isEnabled() { return this.enabled; } @@ -434,6 +442,14 @@ public void setRetention(Duration retention) { this.retention = retention; } + public Duration getLease() { + return this.lease; + } + + public void setLease(Duration lease) { + this.lease = lease; + } + } /** diff --git a/easyssf-receiver-spring-boot-autoconfigure/src/main/java/org/easyssf/receiver/spring/boot/autoconfigure/SsfReceiverAutoConfiguration.java b/easyssf-receiver-spring-boot-autoconfigure/src/main/java/org/easyssf/receiver/spring/boot/autoconfigure/SsfReceiverAutoConfiguration.java index 44fe97b..cd48918 100644 --- a/easyssf-receiver-spring-boot-autoconfigure/src/main/java/org/easyssf/receiver/spring/boot/autoconfigure/SsfReceiverAutoConfiguration.java +++ b/easyssf-receiver-spring-boot-autoconfigure/src/main/java/org/easyssf/receiver/spring/boot/autoconfigure/SsfReceiverAutoConfiguration.java @@ -181,7 +181,9 @@ SsfPoller ssfPoller(SsfReceiverProperties properties, SsfHttpClient httpClient, @ConditionalOnMissingBean(SsfJtiDedupStore.class) @ConditionalOnBooleanProperty(name = "easyssf.receiver.dedup.enabled", matchIfMissing = true) InMemorySsfJtiDedupStore ssfJtiDedupStore(SsfReceiverProperties properties) { - return new InMemorySsfJtiDedupStore(properties.getDedup().getCapacity()); + InMemorySsfJtiDedupStore store = new InMemorySsfJtiDedupStore(properties.getDedup().getCapacity()); + store.setLease(properties.getDedup().getLease()); + return store; } /** diff --git a/easyssf-receiver-spring-boot-autoconfigure/src/main/java/org/easyssf/receiver/spring/boot/autoconfigure/SsfReceiverJdbcAutoConfiguration.java b/easyssf-receiver-spring-boot-autoconfigure/src/main/java/org/easyssf/receiver/spring/boot/autoconfigure/SsfReceiverJdbcAutoConfiguration.java index fd4d802..c8cf6ed 100644 --- a/easyssf-receiver-spring-boot-autoconfigure/src/main/java/org/easyssf/receiver/spring/boot/autoconfigure/SsfReceiverJdbcAutoConfiguration.java +++ b/easyssf-receiver-spring-boot-autoconfigure/src/main/java/org/easyssf/receiver/spring/boot/autoconfigure/SsfReceiverJdbcAutoConfiguration.java @@ -64,9 +64,10 @@ JdbcSsfJtiDedupStore jdbcSsfJtiDedupStore(JdbcOperations jdbc, ObjectProvider this.jdbc.update("INSERT INTO EASYSSF_PROCESSED_SET (ISSUER, JTI, PROCESSED_AT) VALUES (?, ?, ?)", - "i", "j", System.currentTimeMillis())); + this.jdbc.update( + "INSERT INTO EASYSSF_PROCESSED_SET (ISSUER, JTI, STATE, PROCESSED_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', ?)", + "i", "j", System.currentTimeMillis())); JdbcSsfJtiDedupStore store = new JdbcSsfJtiDedupStore(this.jdbc, prefix); store.setRetention(Duration.ofDays(1)); assertThat(this.jdbc.query("SELECT COUNT(*) FROM EASYSSF_PROCESSED_SET", (row) -> row.getInt(1))) diff --git a/easyssf-receiver-spring-boot-starter/README.md b/easyssf-receiver-spring-boot-starter/README.md index 69c8ea9..2134a5f 100644 --- a/easyssf-receiver-spring-boot-starter/README.md +++ b/easyssf-receiver-spring-boot-starter/README.md @@ -311,18 +311,31 @@ the state is kept in its database instead, without further configuration: | Table | Store | Content | |---|---|---| -| `EASYSSF_PROCESSED_SET` | `JdbcSsfJtiDedupStore` | the SETs that were processed, forgotten after `easyssf.receiver.dedup.retention` (7 days) | +| `EASYSSF_PROCESSED_SET` | `JdbcSsfJtiDedupStore` | the SETs that were processed or are being processed (`STATE`), forgotten after `easyssf.receiver.dedup.retention` (7 days) | | `EASYSSF_REVOCATION` | `JdbcSsfTokenRevocationStore` | revoked sessions and subjects, per issuer, removed after `easyssf.receiver.resource-server.revocation-ttl` | | `EASYSSF_POLL_ACK` | `JdbcSsfPollAckStore` | acknowledgements and error reports a polling receiver owes its transmitter until a poll request carried them, forgotten after `easyssf.receiver.jdbc.ack-retention` (7 days) | -- **Tables**: in an embedded database (H2, HSQLDB, Derby) the tables are created on startup. For any - other database create them with your migration tool, the statements are in - [`schema.sql`](../easyssf-receiver-jdbc/src/main/resources/org/easyssf/receiver/jdbc/schema.sql) - (`classpath:org/easyssf/receiver/jdbc/schema.sql`), or set - `easyssf.receiver.jdbc.initialize-schema=always`. If a table is missing the application fails on - startup and says so, rather than on the first event. The stores use plain SQL (`VARCHAR`, +- **Tables**: in an embedded database (H2, HSQLDB, Derby) the tables are created on startup, and + columns a newer release added to an existing table as well. For any other database create them with + your migration tool: the + [`migration/`](../easyssf-receiver-jdbc/src/main/resources/org/easyssf/receiver/jdbc/migration) + scripts (`classpath:org/easyssf/receiver/jdbc/migration/`, one per release that changed the schema, + named for Flyway, plain SQL for Liquibase and others) keep an installation current from release to + release, [`schema.sql`](../easyssf-receiver-jdbc/src/main/resources/org/easyssf/receiver/jdbc/schema.sql) + is the current schema for a fresh one. Or set `easyssf.receiver.jdbc.initialize-schema=always` to have + the receiver create and upgrade the tables itself. If a table or a column is missing the application + fails on startup with the statements to run, rather than on the first event. The stores use plain SQL (`VARCHAR`, `BIGINT`, no vendor syntax; a result limit is applied by the JDBC driver) and are tested on H2, PostgreSQL and MySQL. +- **Several instances**: a SET is claimed before its handlers run and marked processed afterwards. An + instance that receives a SET another one is handling leaves it for the transmitter's redelivery, + neither handling nor acknowledging it, so a handler failure on the first instance is not masked by + an acknowledgement of the second. A claim older than `easyssf.receiver.dedup.lease` (60 seconds) + counts as abandoned by a crashed instance and the SET is handled again, which is one more reason for + handlers to be idempotent. 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. - **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. @@ -377,7 +390,7 @@ If the application has a Micrometer `MeterRegistry` (for example with | Meter | Type | Tags | |---|---|---| -| `easyssf.receiver.sets` | counter | `transmitter` (the name of the transmitter, `default` for the one at `easyssf.receiver.*`), `delivery` (`push`, `poll`), `outcome` (`handled`, `duplicate`, `invalid`, `unauthenticated`, `unavailable`, `failed`) | +| `easyssf.receiver.sets` | counter | `transmitter` (the name of the transmitter, `default` for the one at `easyssf.receiver.*`), `delivery` (`push`, `poll`), `outcome` (`handled`, `duplicate`, `invalid`, `unauthenticated`, `unavailable`, `failed`, `in_progress`) | | `easyssf.receiver.events` | counter | `transmitter`, `delivery`, `event` (for example `CaepSessionRevoked`) | | `easyssf.receiver.poll` | timer | `transmitter`, `outcome` (`success`, `failure`) | | `easyssf.receiver.poll.pending-acks` | gauge | `transmitter`; acknowledgements and error reports waiting for the next poll request, see [POLL delivery](#poll-delivery) | @@ -452,6 +465,7 @@ name, with its own status. Switch it off with `management.health.easyssf.enabled | `easyssf.receiver.dedup.enabled` | `true` | | | `easyssf.receiver.dedup.capacity` | `10000` | Size of the in-memory `jti` store. | | `easyssf.receiver.dedup.retention` | `7d` | How long the JDBC store remembers a processed SET. | +| `easyssf.receiver.dedup.lease` | `60s` | How long a SET counts as being handled by the instance that claimed it; longer than the longest handler. | | `easyssf.receiver.jdbc.enabled` | `true` | Keep state in the database if the application has a `JdbcTemplate`. | | `easyssf.receiver.jdbc.initialize-schema` | `embedded` | When to create missing tables: `embedded`, `always` or `never`. | | `easyssf.receiver.jdbc.table-prefix` | `EASYSSF_` | Prefix of the table names. | 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 new file mode 100644 index 0000000..ec15896 --- /dev/null +++ b/easyssf-receiver/src/main/java/org/easyssf/receiver/event/SsfSetInProgressException.java @@ -0,0 +1,15 @@ +package org.easyssf.receiver.event; + +/** + * 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 + * {@link SsfEventHandlingException} it leaves the SET unacknowledged, so that the + * transmitter delivers it again, by which time the other instance has finished with it. + */ +public class SsfSetInProgressException extends SsfEventHandlingException { + + public SsfSetInProgressException(String message) { + super(message, null); + } + +} diff --git a/easyssf-receiver/src/main/java/org/easyssf/receiver/metrics/SsfReceiverMetrics.java b/easyssf-receiver/src/main/java/org/easyssf/receiver/metrics/SsfReceiverMetrics.java index ed544ef..bfefb2e 100644 --- a/easyssf-receiver/src/main/java/org/easyssf/receiver/metrics/SsfReceiverMetrics.java +++ b/easyssf-receiver/src/main/java/org/easyssf/receiver/metrics/SsfReceiverMetrics.java @@ -32,6 +32,12 @@ enum SetOutcome { */ DUPLICATE, + /** + * The SET is being handled by another instance and has been left for a + * redelivery. + */ + IN_PROGRESS, + /** * The SET was rejected because it is invalid. */ diff --git a/easyssf-receiver/src/main/java/org/easyssf/receiver/poll/SsfPoller.java b/easyssf-receiver/src/main/java/org/easyssf/receiver/poll/SsfPoller.java index 42f2f50..7d266ca 100644 --- a/easyssf-receiver/src/main/java/org/easyssf/receiver/poll/SsfPoller.java +++ b/easyssf-receiver/src/main/java/org/easyssf/receiver/poll/SsfPoller.java @@ -14,6 +14,7 @@ import org.easyssf.core.SsfDeliveryMethod; import org.easyssf.core.support.SsfAssert; import org.easyssf.receiver.event.SsfEventHandlingException; +import org.easyssf.receiver.event.SsfSetInProgressException; import org.easyssf.receiver.http.SsfHttpClient; import org.easyssf.receiver.http.SsfHttpRequest; import org.easyssf.receiver.http.SsfHttpResponse; @@ -466,6 +467,10 @@ private void process(String key, String jti, Object encodedSet) { logger.warn("Rejecting polled SET " + jti + ": " + ex.getMessage()); this.ackStore.record(key, SsfPendingAck.error(jti, ex.getErrorCode(), ex.getMessage())); } + catch (SsfSetInProgressException ex) { + // another instance is on it, the transmitter delivers the SET again + logger.debug("Not acknowledging polled SET " + jti + ": " + ex.getMessage()); + } catch (SsfTransmitterUnavailableException | SsfEventHandlingException ex) { // not acknowledged, the transmitter delivers the SET again logger.warn("Could not process polled SET " + jti + ": " + ex.getMessage()); 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 16d1780..24ec843 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 @@ -1,37 +1,88 @@ package org.easyssf.receiver.set; +import java.time.Clock; +import java.time.Duration; +import java.time.Instant; import java.util.LinkedHashMap; import java.util.Map; import java.util.concurrent.locks.ReentrantLock; import org.easyssf.core.event.SsfEventToken; +import org.easyssf.core.support.SsfAssert; /** * {@link SsfJtiDedupStore} that remembers a bounded number of SETs in memory, evicting - * the oldest first. Its content is lost on restart and not shared between instances. + * the oldest first. Its content is lost on restart and not shared between instances, so + * the {@link Claim#IN_PROGRESS in progress} state only matters when the same SET arrives + * twice while a handler of this instance still runs. */ public class InMemorySsfJtiDedupStore implements SsfJtiDedupStore { - private final Map seen; + private final Map entries; /** a lock rather than synchronized, which pins virtual threads before JDK 24 */ private final ReentrantLock lock = new ReentrantLock(); + private Duration lease = Duration.ofSeconds(60); + + private Clock clock = Clock.systemUTC(); + public InMemorySsfJtiDedupStore(int capacity) { int maxEntries = Math.max(1, capacity); - this.seen = new LinkedHashMap<>() { + this.entries = new LinkedHashMap<>() { @Override - protected boolean removeEldestEntry(Map.Entry eldest) { + protected boolean removeEldestEntry(Map.Entry eldest) { return size() > maxEntries; } }; } + /** + * Sets how long a claim holds before the SET counts as abandoned by a crashed or + * stuck handler 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 Claim claim(SsfEventToken eventToken) { + String key = SsfJtiDedupStore.key(eventToken); + Instant now = this.clock.instant(); + this.lock.lock(); + try { + Known entry = this.entries.get(key); + if (entry == null) { + this.entries.put(key, new Known(Claim.IN_PROGRESS, now)); + return Claim.NEW; + } + if (entry.state == Claim.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; + } + return Claim.IN_PROGRESS; + } + finally { + this.lock.unlock(); + } + } + @Override - public boolean seenBefore(SsfEventToken eventToken) { + public void processed(SsfEventToken eventToken) { this.lock.lock(); try { - return this.seen.put(SsfJtiDedupStore.key(eventToken), Boolean.TRUE) != null; + this.entries.put(SsfJtiDedupStore.key(eventToken), new Known(Claim.PROCESSED, this.clock.instant())); } finally { this.lock.unlock(); @@ -42,7 +93,7 @@ public boolean seenBefore(SsfEventToken eventToken) { public void forget(SsfEventToken eventToken) { this.lock.lock(); try { - this.seen.remove(SsfJtiDedupStore.key(eventToken)); + this.entries.remove(SsfJtiDedupStore.key(eventToken)); } finally { this.lock.unlock(); @@ -52,11 +103,14 @@ public void forget(SsfEventToken eventToken) { public int size() { this.lock.lock(); try { - return this.seen.size(); + return this.entries.size(); } finally { this.lock.unlock(); } } + private record Known(Claim state, Instant since) { + } + } 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 237448b..a65be3b 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 @@ -5,14 +5,52 @@ /** * Remembers which SETs were already processed, so that a SET delivered more than once is * handled only once. Implement it to back it with a shared or durable store. + * + *

+ * A SET is {@link #claim claimed} before its handlers run and marked {@link #processed + * processed} once they succeeded. In between it is in progress: another instance + * of the application that receives the same SET then neither handles nor acknowledges it, + * and leaves it to the transmitter's redelivery. An instance that crashes mid-processing + * 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. */ public interface SsfJtiDedupStore { /** - * Atomically checks whether the given SET was seen before and records it. - * @return {@code true} if the SET was already recorded + * What a store knows about a SET when it is claimed. */ - boolean seenBefore(SsfEventToken eventToken); + enum Claim { + + /** + * Not seen before, or abandoned by a crashed instance: the caller handles the + * SET. + */ + NEW, + + /** + * Another instance is handling the SET right now: neither handle nor acknowledge + * it. + */ + IN_PROGRESS, + + /** + * The SET was handled before: acknowledge it, do not handle it again. + */ + PROCESSED + + } + + /** + * Atomically records that the caller is about to handle the SET, unless it was + * handled before or is being handled by another instance. + */ + Claim claim(SsfEventToken eventToken); + + /** + * Records that the handlers of a claimed SET succeeded. + */ + void processed(SsfEventToken eventToken); /** * Forgets the given SET again, because handling it failed and it is expected to be @@ -20,6 +58,15 @@ public interface SsfJtiDedupStore { */ void forget(SsfEventToken eventToken); + /** + * Claims the SET and tells whether it was known before, as processed or in progress. + * @deprecated use {@link #claim(SsfEventToken)}, which distinguishes the two + */ + @Deprecated(since = "0.3.0") + default boolean seenBefore(SsfEventToken eventToken) { + return claim(eventToken) != Claim.NEW; + } + /** * The key identifying a SET. A {@code jti} is only unique per issuer (RFC 8417, * section 2.2). 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 8b1f77b..7deb260 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 @@ -13,6 +13,7 @@ import org.easyssf.receiver.event.SsfEventContext; import org.easyssf.receiver.event.SsfEventHandler; import org.easyssf.receiver.event.SsfEventHandlingException; +import org.easyssf.receiver.event.SsfSetInProgressException; import org.easyssf.receiver.metrics.SsfReceiverMetrics; import org.easyssf.receiver.metrics.SsfReceiverMetrics.SetOutcome; import org.easyssf.receiver.stream.SsfStreamVerification; @@ -158,10 +159,23 @@ public Outcome process(String encodedSet, SsfDeliveryMethod deliveryMethod) { throw ex; } issuer = eventToken.iss(); - if (this.dedupStore != null && this.dedupStore.seenBefore(eventToken)) { - logger.debug("Skipping SET " + eventToken.jti() + ", it was processed before"); - this.metrics.setReceived(issuer, deliveryMethod, SetOutcome.DUPLICATE); - return Outcome.DUPLICATE; + if (this.dedupStore != null) { + switch (this.dedupStore.claim(eventToken)) { + case PROCESSED -> { + logger.debug("Skipping SET " + eventToken.jti() + ", it was processed before"); + this.metrics.setReceived(issuer, deliveryMethod, SetOutcome.DUPLICATE); + return Outcome.DUPLICATE; + } + case IN_PROGRESS -> { + logger.debug( + "Leaving SET " + eventToken.jti() + " for a redelivery, another instance is handling it"); + this.metrics.setReceived(issuer, deliveryMethod, SetOutcome.IN_PROGRESS); + throw new SsfSetInProgressException( + "SET " + eventToken.jti() + " is being handled by another instance"); + } + case NEW -> { + } + } } if (logger.isDebugEnabled()) { logger.debug("Received SET " + eventToken.jti() + " (" + deliveryMethod + ") with events " @@ -191,6 +205,9 @@ public Outcome process(String encodedSet, SsfDeliveryMethod deliveryMethod) { 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); + } this.metrics.setReceived(issuer, deliveryMethod, SetOutcome.HANDLED); String transmitter = issuer; eventContext.eventTypes() 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 new file mode 100644 index 0000000..c48b4ae --- /dev/null +++ b/easyssf-receiver/src/test/java/org/easyssf/receiver/set/InMemorySsfJtiDedupStoreTests.java @@ -0,0 +1,96 @@ +package org.easyssf.receiver.set; + +import java.time.Clock; +import java.time.Duration; +import java.time.Instant; +import java.time.ZoneOffset; +import java.util.List; +import java.util.Map; + +import org.easyssf.core.event.SsfEventToken; +import org.easyssf.receiver.set.SsfJtiDedupStore.Claim; +import org.junit.jupiter.api.Test; + +import static org.assertj.core.api.Assertions.assertThat; + +class InMemorySsfJtiDedupStoreTests { + + private static final Instant NOW = Instant.parse("2026-10-05T10:00:00Z"); + + private final MutableClock clock = new MutableClock(); + + private final InMemorySsfJtiDedupStore store = new InMemorySsfJtiDedupStore(10); + + InMemorySsfJtiDedupStoreTests() { + this.store.setClock(this.clock); + } + + @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); + // 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); + } + + @Test + void abandonedClaimIsTakenOverAfterTheLease() { + this.store.setLease(Duration.ofSeconds(30)); + assertThat(this.store.claim(set("jti-1"))).isEqualTo(Claim.NEW); + this.clock.advance(Duration.ofSeconds(29)); + assertThat(this.store.claim(set("jti-1"))).isEqualTo(Claim.IN_PROGRESS); + 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); + // processed rows do not expire with the lease + this.store.processed(set("jti-1")); + this.clock.advance(Duration.ofHours(1)); + assertThat(this.store.claim(set("jti-1"))).isEqualTo(Claim.PROCESSED); + } + + @Test + @SuppressWarnings("deprecation") + void seenBeforeClaimsAndTellsWhetherTheSetWasKnown() { + assertThat(this.store.seenBefore(set("jti-1"))).isFalse(); + assertThat(this.store.seenBefore(set("jti-1"))).isTrue(); + } + + private static SsfEventToken set(String jti) { + return set("https://idp.example", jti); + } + + private static SsfEventToken set(String issuer, String jti) { + return new SsfEventToken(jti, issuer, NOW, List.of(), Map.of(), null, null, Map.of()); + } + + private static final class MutableClock extends Clock { + + private Instant now = NOW; + + void advance(Duration duration) { + this.now = this.now.plus(duration); + } + + @Override + public ZoneOffset getZone() { + return ZoneOffset.UTC; + } + + @Override + public Clock withZone(java.time.ZoneId zone) { + return this; + } + + @Override + public Instant instant() { + return this.now; + } + + } + +} 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 dd19c92..394b9c4 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 @@ -10,6 +10,7 @@ import org.easyssf.core.event.SsfEventTypes; import org.easyssf.receiver.event.SsfEventHandler; import org.easyssf.receiver.event.SsfEventHandlingException; +import org.easyssf.receiver.event.SsfSetInProgressException; import org.easyssf.receiver.set.SsfSetProcessor.Outcome; import org.junit.jupiter.api.Test; @@ -20,6 +21,8 @@ class SsfSetProcessorTests { private final List handled = new ArrayList<>(); + private final List failures = new ArrayList<>(); + private final SsfSetVerifier verifier = (encodedSet) -> new SsfEventToken(encodedSet, "https://idp.example", Instant.now(), List.of(), Map.of(SsfEventTypes.CAEP_SESSION_REVOKED, Map.of()), null, null, Map.of()); @@ -66,6 +69,47 @@ void failingHandlerDoesNotKeepOthersFromHandlingAndAllowsRedelivery() { assertThat(this.handled).containsExactly("jti-1", "jti-1"); } + @Test + void setBeingHandledElsewhereIsLeftForARedelivery() { + SsfJtiDedupStore inProgress = new SsfJtiDedupStore() { + @Override + public Claim claim(org.easyssf.core.event.SsfEventToken eventToken) { + return Claim.IN_PROGRESS; + } + + @Override + public void processed(org.easyssf.core.event.SsfEventToken eventToken) { + } + + @Override + public void forget(org.easyssf.core.event.SsfEventToken eventToken) { + } + }; + SsfSetProcessor processor = new SsfSetProcessor(this.verifier, inProgress, List.of(this.recordingHandler)); + assertThatExceptionOfType(SsfSetInProgressException.class).isThrownBy(() -> processor.process("jti-1")) + .withMessageContaining("another instance"); + assertThat(this.handled).isEmpty(); + } + + @Test + void handledSetIsMarkedProcessedAndFailedOneForgotten() { + InMemorySsfJtiDedupStore store = new InMemorySsfJtiDedupStore(10); + this.failures.add(new IllegalStateException("store is down")); + SsfEventHandler failingOnce = (context) -> { + if (!this.failures.isEmpty()) { + throw this.failures.remove(0); + } + this.handled.add(context.eventToken().jti()); + }; + 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")); + assertThat(processor.process("jti-1")).isEqualTo(Outcome.HANDLED); + assertThat(store.claim(this.verifier.verify("jti-1"))).isEqualTo(SsfJtiDedupStore.Claim.PROCESSED); + } + @Test void dedupStoreEvictsOldestEntryWhenFull() { InMemorySsfJtiDedupStore store = new InMemorySsfJtiDedupStore(2);