From 0d8342d94697cb1537b484712d194a64f1117ce8 Mon Sep 17 00:00:00 2001 From: Thomas Darimont Date: Mon, 5 Oct 2026 22:33:06 +0200 Subject: [PATCH 1/3] Keep poll acknowledgements in a store and support long polling An SsfPollAckStore holds what the poller owes the transmitter until a request carried it, in memory by default or in the new EASYSSF_POLL_ACK table, so a SET handled before a restart is acknowledged afterwards instead of delivered again; stop() flushes. setLongPolling(hold) keeps one request outstanding (RFC 8936, section 2.5), with a timeout per request and a polling thread instead of the scheduler; the Spring Boot starter picks the poll client in SsfTransmitterFactory.pollHttpClient. The in-memory stores lock with ReentrantLock, which does not pin virtual threads. Result limits go through the JDBC driver, and the stores are tested on MySQL as well as H2 and PostgreSQL. SsfPoller.setThreadFactory lets a framework create the polling thread Closes #23 Signed-off-by: Thomas Darimont --- .github/workflows/ci.yml | 4 +- AGENTS.md | 2 +- CHANGELOG.md | 24 ++ CONTRIBUTING.md | 6 +- README.md | 4 +- easyssf-receiver-jdbc/AGENTS.md | 14 +- easyssf-receiver-jdbc/pom.xml | 10 + .../jdbc/DataSourceSsfJdbcOperations.java | 6 + .../receiver/jdbc/JdbcSsfPollAckStore.java | 157 +++++++++ .../easyssf/receiver/jdbc/JdbcSsfSchema.java | 26 ++ .../receiver/jdbc/SsfJdbcOperations.java | 18 + .../org/easyssf/receiver/jdbc/schema.sql | 12 + .../jdbc/AbstractJdbcSsfStoresTests.java | 76 ++++ .../jdbc/MySqlJdbcSsfStoresTests.java | 51 +++ .../jdbc/PostgresJdbcSsfStoresTests.java | 2 +- .../spring/boot/SsfReceiverProperties.java | 29 ++ .../spring/boot/SsfTransmitterFactory.java | 81 ++++- .../spring/boot/SsfTransmitterProperties.java | 31 ++ .../SsfReceiverAutoConfiguration.java | 13 +- .../SsfReceiverJdbcAutoConfiguration.java | 23 ++ .../health/SsfReceiverHealthIndicator.java | 1 + .../jdbc/JdbcTemplateSsfJdbcOperations.java | 17 + .../SsfReceiverAutoConfigurationTests.java | 44 +++ .../JdbcTemplateSsfJdbcOperationsTests.java | 17 + .../README.md | 35 +- .../receiver/http/JdkSsfHttpClient.java | 2 +- .../easyssf/receiver/http/SsfHttpRequest.java | 24 +- .../poll/InMemorySsfPollAckStore.java | 90 +++++ .../easyssf/receiver/poll/SsfPendingAck.java | 44 +++ .../receiver/poll/SsfPollAckStore.java | 48 +++ .../org/easyssf/receiver/poll/SsfPoller.java | 331 +++++++++++++++--- .../set/InMemorySsfJtiDedupStore.java | 34 +- .../poll/InMemorySsfPollAckStoreTests.java | 57 +++ .../easyssf/receiver/poll/SsfPollerTests.java | 154 +++++++- .../org/easyssf/test/TestTransmitter.java | 121 +++++-- 35 files changed, 1489 insertions(+), 119 deletions(-) create mode 100644 easyssf-receiver-jdbc/src/main/java/org/easyssf/receiver/jdbc/JdbcSsfPollAckStore.java create mode 100644 easyssf-receiver-jdbc/src/test/java/org/easyssf/receiver/jdbc/MySqlJdbcSsfStoresTests.java create mode 100644 easyssf-receiver/src/main/java/org/easyssf/receiver/poll/InMemorySsfPollAckStore.java create mode 100644 easyssf-receiver/src/main/java/org/easyssf/receiver/poll/SsfPendingAck.java create mode 100644 easyssf-receiver/src/main/java/org/easyssf/receiver/poll/SsfPollAckStore.java create mode 100644 easyssf-receiver/src/test/java/org/easyssf/receiver/poll/InMemorySsfPollAckStoreTests.java diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 3b08ab3..d131409 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -200,7 +200,7 @@ jobs: if-no-files-found: ignore retention-days: 14 - # The JDBC stores against PostgreSQL in Docker, see the database-tests profile of easyssf-receiver-jdbc. + # The JDBC stores against PostgreSQL and MySQL in Docker, see the database-tests profile of easyssf-receiver-jdbc. database: needs: build runs-on: ubuntu-latest @@ -219,7 +219,7 @@ jobs: name: easyssf-jars path: ~/.m2/repository/org/easyssf/ - - name: Test the stores on PostgreSQL + - name: Test the stores on PostgreSQL and MySQL run: ./mvnw -pl easyssf-receiver-jdbc -Pdatabase-tests verify - name: Upload test reports diff --git a/AGENTS.md b/AGENTS.md index 63b720a..d94fae5 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -65,7 +65,7 @@ pins a released version. Consequences for a change here: - `-o` (offline) is fine for everything but dependency updates; `-q` keeps the output short, then check `target/surefire-reports/*.xml` for `failures="0"` and `errors="0"`. - Two test groups are excluded from the normal build and need Docker: the `database` tag of - `easyssf-receiver-jdbc` (PostgreSQL, `-Pdatabase-tests`) and the `conformance` tag of the + `easyssf-receiver-jdbc` (PostgreSQL and MySQL, `-Pdatabase-tests`) and the `conformance` tag of the conformance tests (`-Pconformance`, minutes per plan). Run them only when a change touches the stores or the conformance harness, and say so. - Spotless applies the Spring Java Format conventions (`etc/eclipse-formatter.prefs`): 4 spaces, diff --git a/CHANGELOG.md b/CHANGELOG.md index adf6fe7..4571a87 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,30 @@ The notable changes of every release. The section of a version is the text of it ## [Unreleased] +### Added + +- POLL delivery keeps the acknowledgements and error reports it owes the transmitter in an + `SsfPollAckStore` until a poll request has carried them: in memory by default, or durably with + `JdbcSsfPollAckStore` (table `EASYSSF_POLL_ACK`, auto-configured with the other JDBC stores, + `easyssf.receiver.jdbc.ack-retention`), so that a SET handled right before a restart is + acknowledged afterwards instead of being delivered again. `SsfPoller.stop()` sends the pending + acknowledgements with a last request; the health details show them as `pendingAcks`. +- Long polling (RFC 8936, section 2.5): `SsfPoller.setLongPolling(hold)` keeps one request + outstanding that the transmitter holds until SETs are available; `easyssf.receiver.poll.long-polling` + and `poll.long-polling-hold` in the Spring Boot starter, off by default. `SsfHttpRequest` carries an + optional timeout per request. `TestTransmitter` holds long polls for `setLongPollHold`. + `SsfPoller.setThreadFactory` lets a framework create the polling thread, or use a virtual one. + +### Changed + +- 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 + lock with `ReentrantLock` instead of `synchronized`, which pins virtual threads before JDK 24. +- `SsfJdbcOperations.query(sql, maxRows, mapper, args)` limits a result through the JDBC driver, so + the stores need no dialect-specific SQL; the JDBC stores are now tested on MySQL as well as on H2 + and PostgreSQL. + ## [0.2.0] - 2026-10-04 SCIM Events (RFC 9967) for the receiver: the event types, the `scim` subject, a typed payload and a diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index fc2b938..1986112 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -56,8 +56,8 @@ to submit the change under the project's license. The CI checks it for every com `target/central-publishing/central-bundle.zip` for `_remote.repositories`. The settings step of the workflows that scopes the Central credentials (`repositoryOrigins`, needed by 3.10) stays: Maven 3.9 only warns about the unknown tag. -- **Database tests**: the JDBC stores are tested on H2 by the normal build and on PostgreSQL in a - Testcontainers container by the tests tagged `database`, which the normal build excludes: +- **Database tests**: the JDBC stores are tested on H2 by the normal build and on PostgreSQL and + MySQL in Testcontainers containers by the tests tagged `database`, which the normal build excludes: `./mvnw -pl easyssf-receiver-jdbc -Pdatabase-tests verify` runs them (Docker required). CI runs them in the `database` job. - **Test logging**: `easyssf-receiver/src/test/resources/logback-test.xml` keeps the tests of the plain @@ -83,7 +83,7 @@ GitHub Actions, see [`.github/workflows`](.github/workflows): 3. Behind `build`: `spring` (the Spring Boot modules on Java 21), `jdk` (every module on Java 25, and on the newest JDK as a job that does not fail the build; it passes `-Dbytebuddy.experimental=true` in case Mockito's Byte Buddy does not know the class file - version yet), and `database` (the stores of `easyssf-receiver-jdbc` on PostgreSQL). + version yet), and `database` (the stores of `easyssf-receiver-jdbc` on PostgreSQL and MySQL). 4. `status` is green when every job that had to run succeeded; skipped jobs count as green. It is the one check to require in the branch protection of `main`, the others come and go with the change. diff --git a/README.md b/README.md index c8a41dd..8580663 100644 --- a/README.md +++ b/README.md @@ -291,6 +291,8 @@ SsfPushResponse response = pushHandler.handle(authorizationHeader, requestBody); // POLL: fetch SETs from the transmitter instead SsfPoller poller = new SsfPoller(httpClient, tokenProvider, () -> pollEndpoint, processor); +poller.setAckStore(ackStore); // optional: JdbcSsfPollAckStore keeps acknowledgements across restarts +poller.setLongPolling(Duration.ofSeconds(30)); // optional: hold requests instead of polling every interval poller.start(); ``` @@ -372,8 +374,6 @@ streams a receiver registered. The subject helpers are the static methods of `Ss ## Not included (yet) - A transmitter. `easyssf-core` is written to be shared with one. -- Long polling, and a durable store for acknowledgements that were not sent yet (a SET whose - acknowledgement is lost is delivered again and skipped as a duplicate). ## Build diff --git a/easyssf-receiver-jdbc/AGENTS.md b/easyssf-receiver-jdbc/AGENTS.md index fd04a8a..4b8ae90 100644 --- a/easyssf-receiver-jdbc/AGENTS.md +++ b/easyssf-receiver-jdbc/AGENTS.md @@ -3,12 +3,14 @@ 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 and - PostgreSQL unchanged. Prefer standard SQL to dialect features. -- Tests run on H2 in the normal build. `PostgresJdbcSsfStoresTests` is tagged `database`, runs in - Testcontainers and is excluded by default: - `./mvnw -pl easyssf-receiver-jdbc -Pdatabase-tests verify` (Docker). Run it when the SQL or the - schema changes. Both share `AbstractJdbcSsfStoresTests`; add store tests there. +- 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 + 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` + are tagged `database`, run in Testcontainers and are excluded by default: + `./mvnw -pl easyssf-receiver-jdbc -Pdatabase-tests verify` (Docker). Run them when the SQL or the + schema changes. All share `AbstractJdbcSsfStoresTests`; add store tests there. - 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/pom.xml b/easyssf-receiver-jdbc/pom.xml index 51687b1..b5ad03d 100644 --- a/easyssf-receiver-jdbc/pom.xml +++ b/easyssf-receiver-jdbc/pom.xml @@ -61,6 +61,16 @@ postgresql test + + org.testcontainers + testcontainers-mysql + test + + + com.mysql + mysql-connector-j + test + diff --git a/easyssf-receiver-jdbc/src/main/java/org/easyssf/receiver/jdbc/DataSourceSsfJdbcOperations.java b/easyssf-receiver-jdbc/src/main/java/org/easyssf/receiver/jdbc/DataSourceSsfJdbcOperations.java index fe64cef..b391101 100644 --- a/easyssf-receiver-jdbc/src/main/java/org/easyssf/receiver/jdbc/DataSourceSsfJdbcOperations.java +++ b/easyssf-receiver-jdbc/src/main/java/org/easyssf/receiver/jdbc/DataSourceSsfJdbcOperations.java @@ -50,8 +50,14 @@ public int update(String sql, Object... args) { @Override public List query(String sql, SsfJdbcRowMapper mapper, Object... args) { + return query(sql, 0, mapper, args); + } + + @Override + public List query(String sql, int maxRows, SsfJdbcRowMapper mapper, Object... args) { try (Connection connection = this.dataSource.getConnection(); PreparedStatement statement = connection.prepareStatement(sql)) { + statement.setMaxRows(Math.max(0, maxRows)); bind(statement, args); try (ResultSet resultSet = statement.executeQuery()) { List rows = new ArrayList<>(); diff --git a/easyssf-receiver-jdbc/src/main/java/org/easyssf/receiver/jdbc/JdbcSsfPollAckStore.java b/easyssf-receiver-jdbc/src/main/java/org/easyssf/receiver/jdbc/JdbcSsfPollAckStore.java new file mode 100644 index 0000000..92410c7 --- /dev/null +++ b/easyssf-receiver-jdbc/src/main/java/org/easyssf/receiver/jdbc/JdbcSsfPollAckStore.java @@ -0,0 +1,157 @@ +package org.easyssf.receiver.jdbc; + +import java.time.Clock; +import java.time.Duration; +import java.time.Instant; +import java.util.Collection; +import java.util.List; + +import org.easyssf.core.support.SsfAssert; +import org.easyssf.receiver.poll.SsfPendingAck; +import org.easyssf.receiver.poll.SsfPollAckStore; + +/** + * {@link SsfPollAckStore} that keeps the acknowledgements a poller owes its transmitter + * in a database table, so that they survive a restart: a SET handled right before the + * application stopped is acknowledged with the first poll after it started again, instead + * of being delivered once more. + * + *

+ * Entries a transmitter never accepts are forgotten after the retention time (7 days by + * default): 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 JdbcSsfPollAckStore implements SsfPollAckStore, JdbcSsfExpiringStore { + + private static final Duration CLEANUP_INTERVAL = Duration.ofMinutes(1); + + private static final int MAX_DESCRIPTION_LENGTH = 1024; + + /** + * How many entries one DELETE removes by default; databases limit the number of + * parameters of a statement. + */ + public static final int DEFAULT_DELETE_BATCH_SIZE = 100; + + private final SsfJdbcOperations jdbc; + + private final String insert; + + private final String select; + + private final String deletePrefix; + + private final String count; + + private final String deleteExpired; + + private Duration retention = Duration.ofDays(7); + + private int deleteBatchSize = DEFAULT_DELETE_BATCH_SIZE; + + private Clock clock = Clock.systemUTC(); + + private volatile Instant nextCleanup = Instant.MIN; + + /** + * @param jdbc used to access the database + * @param tablePrefix the prefix of the table name, see + * {@link JdbcSsfSchema#DEFAULT_TABLE_PREFIX} + */ + public JdbcSsfPollAckStore(SsfJdbcOperations jdbc, String tablePrefix) { + SsfAssert.notNull(jdbc, "jdbc must not be null"); + String table = JdbcSsfSchema.pollAckTable(tablePrefix); + this.jdbc = jdbc; + this.insert = "INSERT INTO " + table + + " (ISSUER, JTI, ERROR_CODE, ERROR_DESCRIPTION, RECORDED_AT) VALUES (?, ?, ?, ?, ?)"; + // the limit is applied by the JDBC driver (Statement.setMaxRows), the SQL for it + // differs + // between databases + this.select = "SELECT JTI, ERROR_CODE, ERROR_DESCRIPTION FROM " + table + + " WHERE ISSUER = ? ORDER BY RECORDED_AT, JTI"; + this.deletePrefix = "DELETE FROM " + table + " WHERE ISSUER = ? AND JTI IN ("; + this.count = "SELECT COUNT(*) FROM " + table + " WHERE ISSUER = ?"; + this.deleteExpired = "DELETE FROM " + table + " WHERE RECORDED_AT < ?"; + } + + /** + * Sets how long an entry the transmitter does not accept is kept. + */ + public void setRetention(Duration retention) { + SsfAssert.isTrue(retention != null && retention.isPositive(), "retention must be positive"); + this.retention = retention; + } + + /** + * Sets how many entries one DELETE statement removes ({@code JTI IN (...)}), 100 by + * default. Databases limit the number of parameters of a statement (SQL Server to + * 2100, Oracle to 1000 list members); a smaller batch means more statements per poll. + */ + public void setDeleteBatchSize(int deleteBatchSize) { + SsfAssert.isTrue(deleteBatchSize > 0, "deleteBatchSize must be positive"); + this.deleteBatchSize = deleteBatchSize; + } + + public void setClock(Clock clock) { + SsfAssert.notNull(clock, "clock must not be null"); + this.clock = clock; + } + + @Override + public void record(String issuer, SsfPendingAck ack) { + Instant now = this.clock.instant(); + removeExpired(now); + String description = ack.errorDescription(); + if (description != null && description.length() > MAX_DESCRIPTION_LENGTH) { + description = description.substring(0, MAX_DESCRIPTION_LENGTH); + } + try { + this.jdbc.update(this.insert, issuer, ack.jti(), ack.errorCode(), description, now.toEpochMilli()); + } + catch (SsfJdbcDuplicateKeyException ex) { + // recorded already, for example a SET delivered twice before it was + // acknowledged + } + } + + @Override + public List pending(String issuer, int limit) { + return this.jdbc.query(this.select, limit, + (row) -> new SsfPendingAck(row.getString(1), row.getString(2), row.getString(3)), issuer); + } + + @Override + public void remove(String issuer, Collection jtis) { + List remaining = List.copyOf(jtis); + for (int from = 0; from < remaining.size(); from += this.deleteBatchSize) { + List chunk = remaining.subList(from, Math.min(from + this.deleteBatchSize, remaining.size())); + Object[] args = new Object[chunk.size() + 1]; + args[0] = issuer; + for (int i = 0; i < chunk.size(); i++) { + args[i + 1] = chunk.get(i); + } + this.jdbc.update(this.deletePrefix + "?, ".repeat(chunk.size() - 1) + "?)", args); + } + } + + @Override + public int size(String issuer) { + return this.jdbc.query(this.count, (row) -> row.getInt(1), issuer).get(0); + } + + @Override + public int purgeExpired() { + Instant now = this.clock.instant(); + this.nextCleanup = now.plus(CLEANUP_INTERVAL); + return this.jdbc.update(this.deleteExpired, now.minus(this.retention).toEpochMilli()); + } + + private void removeExpired(Instant now) { + if (now.isAfter(this.nextCleanup)) { + purgeExpired(); + } + } + +} 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 11d7914..b4a3066 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 @@ -39,6 +39,13 @@ public static String revocationTable(String tablePrefix) { return validPrefix(tablePrefix) + "REVOCATION"; } + /** + * The name of the table of the {@link JdbcSsfPollAckStore}. + */ + public static String pollAckTable(String tablePrefix) { + return validPrefix(tablePrefix) + "POLL_ACK"; + } + private static String validPrefix(String tablePrefix) { SsfAssert.isTrue(tablePrefix != null && tablePrefix.matches("[A-Za-z0-9_.]*"), "The table prefix may only consist of letters, digits, '_' and '.'"); @@ -77,6 +84,25 @@ ID VARCHAR(255) NOT NULL, )""".formatted(table, name), "CREATE INDEX %s_IX1 ON %s (EXPIRES_AT)".formatted(name, table)); } + /** + * The statements that create the table of the {@link JdbcSsfPollAckStore}. + */ + public static List createPollAckTable(String tablePrefix) { + String table = pollAckTable(tablePrefix); + String name = table.replace('.', '_'); + String tableDdl = """ + CREATE TABLE %s ( + ISSUER VARCHAR(255) NOT NULL, + JTI VARCHAR(255) NOT NULL, + ERROR_CODE VARCHAR(64), + ERROR_DESCRIPTION VARCHAR(1024), + RECORDED_AT BIGINT NOT NULL, + CONSTRAINT %s_PK PRIMARY KEY (ISSUER, JTI) + )""".formatted(table, name); + String indexDdl = "CREATE INDEX %s_IX1 ON %s (RECORDED_AT)".formatted(name, table); + return List.of(tableDdl, indexDdl); + } + /** * Makes sure a table exists. * @param jdbc runs the statements diff --git a/easyssf-receiver-jdbc/src/main/java/org/easyssf/receiver/jdbc/SsfJdbcOperations.java b/easyssf-receiver-jdbc/src/main/java/org/easyssf/receiver/jdbc/SsfJdbcOperations.java index 73d7b1b..1e08bf7 100644 --- a/easyssf-receiver-jdbc/src/main/java/org/easyssf/receiver/jdbc/SsfJdbcOperations.java +++ b/easyssf-receiver-jdbc/src/main/java/org/easyssf/receiver/jdbc/SsfJdbcOperations.java @@ -29,6 +29,24 @@ public interface SsfJdbcOperations { */ List query(String sql, SsfJdbcRowMapper mapper, Object... args); + /** + * Runs a query and returns at most {@code maxRows} rows. The limit is applied with + * {@code Statement.setMaxRows}, which every JDBC driver supports, rather than with + * the SQL for it, which differs between databases ({@code FETCH FIRST}, + * {@code LIMIT}, {@code TOP}). This default runs the query without a limit and cuts + * the result; implementations should override it. + * @param sql the query with {@code ?} placeholders + * @param maxRows the most rows to return, positive + * @param mapper maps each row of the result + * @param args the values of the placeholders + * @return the mapped rows + * @throws SsfJdbcException if the query fails + */ + default List query(String sql, int maxRows, SsfJdbcRowMapper mapper, Object... args) { + List rows = query(sql, mapper, args); + return (rows.size() > maxRows) ? List.copyOf(rows.subList(0, maxRows)) : rows; + } + /** * Runs a statement that returns nothing, for example DDL. * @throws SsfJdbcException if the statement fails 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 fa7ca3e..1696710 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 @@ -20,3 +20,15 @@ CREATE TABLE EASYSSF_REVOCATION ( CONSTRAINT EASYSSF_REVOCATION_PK PRIMARY KEY (KIND, ISSUER, ID) ); CREATE INDEX EASYSSF_REVOCATION_IX1 ON EASYSSF_REVOCATION (EXPIRES_AT); + +-- acknowledgements and error reports a poller owes its transmitter until a poll request carried them, +-- see JdbcSsfPollAckStore (ERROR_CODE NULL: acknowledgement) +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/test/java/org/easyssf/receiver/jdbc/AbstractJdbcSsfStoresTests.java b/easyssf-receiver-jdbc/src/test/java/org/easyssf/receiver/jdbc/AbstractJdbcSsfStoresTests.java index 89505f0..2d0ac82 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 @@ -12,6 +12,7 @@ import java.util.Map; import org.easyssf.core.event.SsfEventToken; +import org.easyssf.receiver.poll.SsfPendingAck; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; @@ -47,6 +48,15 @@ void createTables() { JdbcSsfSchema.createProcessedSetTable(PREFIX), true, null); JdbcSsfSchema.prepareTable(this.jdbc, JdbcSsfSchema.revocationTable(PREFIX), JdbcSsfSchema.createRevocationTable(PREFIX), true, null); + JdbcSsfSchema.prepareTable(this.jdbc, JdbcSsfSchema.pollAckTable(PREFIX), + JdbcSsfSchema.createPollAckTable(PREFIX), true, null); + } + + private JdbcSsfPollAckStore ackStore() { + JdbcSsfPollAckStore store = new JdbcSsfPollAckStore(this.jdbc, PREFIX); + store.setRetention(Duration.ofDays(1)); + store.setClock(this.clock); + return store; } private JdbcSsfJtiDedupStore dedupStore() { @@ -71,6 +81,7 @@ void schemaFileMatchesTheStatementsThatCreateTheTables() throws Exception { .toList(); List expected = new ArrayList<>(JdbcSsfSchema.createProcessedSetTable(PREFIX)); expected.addAll(JdbcSsfSchema.createRevocationTable(PREFIX)); + expected.addAll(JdbcSsfSchema.createPollAckTable(PREFIX)); assertThat(statements).isEqualTo(expected.stream().map(AbstractJdbcSsfStoresTests::normalize).toList()); } @@ -201,6 +212,71 @@ void revocationsExpireAndAreRemoved() { .containsExactly("session-2"); } + @Test + void acknowledgementsWaitUntilRemovedAndAreSharedBetweenInstances() { + JdbcSsfPollAckStore store = ackStore(); + store.record(ISSUER, SsfPendingAck.ack("jti-1")); + this.clock.advance(Duration.ofSeconds(1)); + store.record(ISSUER, SsfPendingAck.error("jti-2", "invalid_request", "not a SET")); + store.record(ISSUER, SsfPendingAck.ack("jti-2")); + store.record("https://other.example", SsfPendingAck.ack("jti-1")); + + assertThat(ackStore().size(ISSUER)).isEqualTo(2); + List pending = ackStore().pending(ISSUER, 10); + assertThat(pending).extracting(SsfPendingAck::jti).containsExactly("jti-1", "jti-2"); + assertThat(pending.get(0).isError()).isFalse(); + assertThat(pending.get(1).errorCode()).isEqualTo("invalid_request"); + assertThat(pending.get(1).errorDescription()).isEqualTo("not a SET"); + assertThat(ackStore().pending(ISSUER, 1)).extracting(SsfPendingAck::jti).containsExactly("jti-1"); + + ackStore().remove(ISSUER, List.of("jti-1", "jti-unknown")); + assertThat(ackStore().pending(ISSUER, 10)).extracting(SsfPendingAck::jti).containsExactly("jti-2"); + assertThat(ackStore().size("https://other.example")).isEqualTo(1); + } + + @Test + void manyAcknowledgementsAreListedWithALimitAndRemovedInChunks() { + JdbcSsfPollAckStore store = ackStore(); + store.setDeleteBatchSize(50); + for (int i = 0; i < 250; i++) { + store.record(ISSUER, SsfPendingAck.ack("jti-%03d".formatted(i))); + this.clock.advance(Duration.ofMillis(1)); + } + assertThat(store.size(ISSUER)).isEqualTo(250); + List pending = store.pending(ISSUER, 120); + assertThat(pending).hasSize(120); + assertThat(pending.get(0).jti()).isEqualTo("jti-000"); + assertThat(pending.get(119).jti()).isEqualTo("jti-119"); + + store.remove(ISSUER, pending.stream().map(SsfPendingAck::jti).toList()); + assertThat(store.size(ISSUER)).isEqualTo(130); + assertThat(store.pending(ISSUER, 1).get(0).jti()).isEqualTo("jti-120"); + store.remove(ISSUER, List.of()); + assertThat(store.size(ISSUER)).isEqualTo(130); + } + + @Test + void longErrorDescriptionsAreCut() { + JdbcSsfPollAckStore store = ackStore(); + store.record(ISSUER, SsfPendingAck.error("jti-1", "invalid_request", "x".repeat(5000))); + assertThat(store.pending(ISSUER, 1).get(0).errorDescription()).hasSize(1024); + } + + @Test + void acknowledgementsAreForgottenAfterTheRetentionTime() { + JdbcSsfPollAckStore store = ackStore(); + store.record(ISSUER, SsfPendingAck.ack("old")); + this.clock.advance(Duration.ofHours(12)); + store.record(ISSUER, SsfPendingAck.ack("recent")); + this.clock.advance(Duration.ofHours(12).plusSeconds(1)); + assertThat(store.purgeExpired()).isEqualTo(1); + assertThat(store.pending(ISSUER, 10)).extracting(SsfPendingAck::jti).containsExactly("recent"); + // writing purges as well, at most once a minute + this.clock.advance(Duration.ofDays(1)); + store.record(ISSUER, SsfPendingAck.ack("newer")); + assertThat(store.pending(ISSUER, 10)).extracting(SsfPendingAck::jti).containsExactly("newer"); + } + @Test void purgeExpiredDeletesOnlyTheExpiredRows() { // processed SETs are retained for a day in this test diff --git a/easyssf-receiver-jdbc/src/test/java/org/easyssf/receiver/jdbc/MySqlJdbcSsfStoresTests.java b/easyssf-receiver-jdbc/src/test/java/org/easyssf/receiver/jdbc/MySqlJdbcSsfStoresTests.java new file mode 100644 index 0000000..6032b7a --- /dev/null +++ b/easyssf-receiver-jdbc/src/test/java/org/easyssf/receiver/jdbc/MySqlJdbcSsfStoresTests.java @@ -0,0 +1,51 @@ +package org.easyssf.receiver.jdbc; + +import java.util.List; + +import org.junit.jupiter.api.AfterAll; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.Tag; +import org.testcontainers.mysql.MySQLContainer; + +import com.mysql.cj.jdbc.MysqlDataSource; + +/** + * The stores on MySQL, in a Testcontainers container shared by the tests of the class. + * Tagged {@code database}: the normal build excludes it, {@code -Pdatabase-tests} runs it + * (Docker required). + */ +@Tag("database") +class MySqlJdbcSsfStoresTests extends AbstractJdbcSsfStoresTests { + + private static final String IMAGE = "mysql:8.4"; + + private static final List TABLES = List.of("EASYSSF_PROCESSED_SET", "EASYSSF_REVOCATION", + "APP_SSF_PROCESSED_SET", "OTHER_REVOCATION", "EASYSSF_POLL_ACK"); + + private static MySQLContainer mysql; + + @BeforeAll + static void startDatabase() { + mysql = new MySQLContainer(IMAGE); + mysql.start(); + } + + @AfterAll + static void stopDatabase() { + mysql.stop(); + } + + @Override + protected SsfJdbcOperations emptyDatabase() { + MysqlDataSource dataSource = new MysqlDataSource(); + dataSource.setUrl(mysql.getJdbcUrl()); + dataSource.setUser(mysql.getUsername()); + dataSource.setPassword(mysql.getPassword()); + SsfJdbcOperations jdbc = new DataSourceSsfJdbcOperations(dataSource); + for (String table : TABLES) { + jdbc.execute("DROP TABLE IF EXISTS " + table); + } + return jdbc; + } + +} diff --git a/easyssf-receiver-jdbc/src/test/java/org/easyssf/receiver/jdbc/PostgresJdbcSsfStoresTests.java b/easyssf-receiver-jdbc/src/test/java/org/easyssf/receiver/jdbc/PostgresJdbcSsfStoresTests.java index 206540b..6d0ccbf 100644 --- a/easyssf-receiver-jdbc/src/test/java/org/easyssf/receiver/jdbc/PostgresJdbcSsfStoresTests.java +++ b/easyssf-receiver-jdbc/src/test/java/org/easyssf/receiver/jdbc/PostgresJdbcSsfStoresTests.java @@ -19,7 +19,7 @@ class PostgresJdbcSsfStoresTests extends AbstractJdbcSsfStoresTests { private static final String IMAGE = "postgres:18.4"; private static final List TABLES = List.of("EASYSSF_PROCESSED_SET", "EASYSSF_REVOCATION", - "APP_SSF_PROCESSED_SET", "OTHER_REVOCATION"); + "APP_SSF_PROCESSED_SET", "OTHER_REVOCATION", "EASYSSF_POLL_ACK"); private static PostgreSQLContainer postgres; 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 3971718..7241be8 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 @@ -7,6 +7,7 @@ import java.util.Map; import org.easyssf.core.event.SsfSubject; +import org.easyssf.receiver.jdbc.JdbcSsfPollAckStore; import org.easyssf.receiver.session.SsfSubjectClaimsMatcher; import org.easyssf.receiver.transmitter.SsfTransmitter; import org.springframework.boot.context.properties.ConfigurationProperties; @@ -261,6 +262,18 @@ public static class Jdbc { */ private Duration cleanupInterval = Duration.ofMinutes(15); + /** + * How long the JDBC store keeps an acknowledgement a polling receiver owes its + * transmitter if no poll request manages to deliver it. + */ + private Duration ackRetention = Duration.ofDays(7); + + /** + * How many acknowledgements one DELETE of the JDBC store removes at once; + * databases limit the parameters of a statement. + */ + private int ackDeleteBatchSize = JdbcSsfPollAckStore.DEFAULT_DELETE_BATCH_SIZE; + public boolean isEnabled() { return this.enabled; } @@ -293,6 +306,22 @@ public void setCleanupInterval(Duration cleanupInterval) { this.cleanupInterval = cleanupInterval; } + public Duration getAckRetention() { + return this.ackRetention; + } + + public void setAckRetention(Duration ackRetention) { + this.ackRetention = ackRetention; + } + + public int getAckDeleteBatchSize() { + return this.ackDeleteBatchSize; + } + + public void setAckDeleteBatchSize(int ackDeleteBatchSize) { + this.ackDeleteBatchSize = ackDeleteBatchSize; + } + public enum InitializeSchema { EMBEDDED, ALWAYS, NEVER diff --git a/easyssf-receiver-spring-boot-autoconfigure/src/main/java/org/easyssf/receiver/spring/boot/SsfTransmitterFactory.java b/easyssf-receiver-spring-boot-autoconfigure/src/main/java/org/easyssf/receiver/spring/boot/SsfTransmitterFactory.java index 768509a..c8b3d54 100644 --- a/easyssf-receiver-spring-boot-autoconfigure/src/main/java/org/easyssf/receiver/spring/boot/SsfTransmitterFactory.java +++ b/easyssf-receiver-spring-boot-autoconfigure/src/main/java/org/easyssf/receiver/spring/boot/SsfTransmitterFactory.java @@ -1,6 +1,7 @@ package org.easyssf.receiver.spring.boot; import java.net.URI; +import java.time.Duration; import java.util.List; import java.util.function.Supplier; import java.util.stream.Stream; @@ -9,8 +10,10 @@ import org.apache.commons.logging.LogFactory; import org.easyssf.core.SsfDeliveryMethod; import org.easyssf.core.stream.SsfStreamConfiguration; +import org.easyssf.receiver.http.JdkSsfHttpClient; import org.easyssf.receiver.http.SsfHttpClient; import org.easyssf.receiver.metrics.SsfReceiverMetrics; +import org.easyssf.receiver.poll.SsfPollAckStore; import org.easyssf.receiver.poll.SsfPoller; import org.easyssf.receiver.set.NimbusSsfSetVerifier; import org.easyssf.receiver.set.SsfSetProcessor; @@ -180,12 +183,62 @@ public static SsfStreamRegistrar streamRegistrar(String name, SsfTransmitterProp public static SsfPoller poller(String name, SsfTransmitterProperties properties, SsfHttpClient httpClient, SsfTransmitterTokenProvider tokenProvider, SsfSetProcessor processor, SsfReceiverStream receiverStream, SsfReceiverMetrics metrics) { + return poller(name, properties, httpClient, tokenProvider, processor, receiverStream, metrics, null); + } + + /** + * The HTTP client the poller of a transmitter calls the transmitter with: the given + * client, unless long polling is on and the client is not a {@link JdkSsfHttpClient}. + * A long poll waits longer than the read timeout of the application's + * {@code RestClient} allows, and the {@code RestClient} has no timeout per request, + * so the poller then gets a JDK client of its own, with the connect timeout and user + * agent of the {@code http} properties and a read timeout that covers the hold time. + * The auto-configuration uses it; an application that builds its own poller bean + * passes whatever client it wants to {@link #poller}. + * @param httpClient the client of the application + * @param http the {@code easyssf.receiver.http.*} properties, {@code null} for their + * defaults + */ + public static SsfHttpClient pollHttpClient(String name, SsfTransmitterProperties properties, + SsfHttpClient httpClient, SsfReceiverProperties.Http http) { + SsfTransmitterProperties.Poll poll = properties.getPoll(); + if (properties.getDeliveryMethod() != SsfDeliveryMethod.POLL || !poll.isLongPolling() + || httpClient instanceof JdkSsfHttpClient) { + return httpClient; + } + Duration connectTimeout = (http != null && http.getConnectTimeout() != null) ? http.getConnectTimeout() + : SsfReceiverProperties.Http.DEFAULT_TIMEOUT; + JdkSsfHttpClient pollClient = new JdkSsfHttpClient(connectTimeout, poll.getLongPollingHold().plusSeconds(15)); + if (http != null) { + pollClient.setUserAgent(http.getUserAgent()); + } + logger.info(prefix(name) + "poll.long-polling is on, the poller calls the transmitter with the JDK HTTP " + + "client rather than the RestClient of the application"); + return pollClient; + } + + /** + * @param httpClient the client the poller calls the transmitter with, see + * {@link #pollHttpClient} for the one the auto-configuration chooses + * @param ackStore keeps the acknowledgements the poller owes the transmitter, + * {@code null} for the in-memory default + * @return the poller, {@code null} with PUSH delivery + */ + public static SsfPoller poller(String name, SsfTransmitterProperties properties, SsfHttpClient httpClient, + SsfTransmitterTokenProvider tokenProvider, SsfSetProcessor processor, SsfReceiverStream receiverStream, + SsfReceiverMetrics metrics, SsfPollAckStore ackStore) { if (properties.getDeliveryMethod() != SsfDeliveryMethod.POLL) { return null; } SsfTransmitterProperties.Poll poll = properties.getPoll(); SsfPoller poller = new SsfPoller(httpClient, tokenProvider, pollEndpoint(name, properties, receiverStream), processor); + if (ackStore != null) { + poller.setAckStore(ackStore); + } + if (poll.isLongPolling()) { + poller.setLongPolling(poll.getLongPollingHold()); + } poller.setTransmitter(properties.getTransmitterIssuer()); poller.setMetrics((metrics != null) ? metrics : SsfReceiverMetrics.NOOP); poller.setInterval(poll.getInterval()); @@ -201,6 +254,30 @@ public static SsfPoller poller(String name, SsfTransmitterProperties properties, */ public static SsfTransmitter.Builder builder(String name, SsfTransmitterProperties properties, SsfHttpClient httpClient, SsfSetProcessor processor, SsfReceiverMetrics metrics) { + return builder(name, properties, httpClient, processor, metrics, null); + } + + /** + * A builder with every part of the transmitter built from its properties. + * @param ackStore keeps the acknowledgements a poller owes the transmitter, + * {@code null} for the in-memory default + */ + public static SsfTransmitter.Builder builder(String name, SsfTransmitterProperties properties, + SsfHttpClient httpClient, SsfSetProcessor processor, SsfReceiverMetrics metrics, SsfPollAckStore ackStore) { + return builder(name, properties, httpClient, processor, metrics, ackStore, null); + } + + /** + * A builder with every part of the transmitter built from its properties. + * @param ackStore keeps the acknowledgements a poller owes the transmitter, + * {@code null} for the in-memory default + * @param http the {@code easyssf.receiver.http.*} properties, for the + * {@link #pollHttpClient poll client} of a long polling transmitter; {@code null} for + * their defaults + */ + public static SsfTransmitter.Builder builder(String name, SsfTransmitterProperties properties, + SsfHttpClient httpClient, SsfSetProcessor processor, SsfReceiverMetrics metrics, SsfPollAckStore ackStore, + SsfReceiverProperties.Http http) { SsfReceiverStream receiverStream = new SsfReceiverStream(); SsfTransmitterMetadataResolver metadataResolver = metadataResolver(name, properties, httpClient); SsfTransmitterTokenProvider tokenProvider = tokenProvider(name, properties, httpClient); @@ -213,8 +290,8 @@ public static SsfTransmitter.Builder builder(String name, SsfTransmitterProperti .receiverStream(receiverStream) .streamVerification(streamVerification(receiverStream)) .streamRegistrar(streamRegistrar(name, properties, streamClient, receiverStream)) - .poller(poller(name, properties, httpClient, tokenProvider, processor, receiverStream, metrics), - properties.getPoll().isAutoStartup()) + .poller(poller(name, properties, pollHttpClient(name, properties, httpClient, http), tokenProvider, + processor, receiverStream, metrics, ackStore), properties.getPoll().isAutoStartup()) .pushAuthorizationHeader(properties.getPush().getExpectedAuthHeader()); } diff --git a/easyssf-receiver-spring-boot-autoconfigure/src/main/java/org/easyssf/receiver/spring/boot/SsfTransmitterProperties.java b/easyssf-receiver-spring-boot-autoconfigure/src/main/java/org/easyssf/receiver/spring/boot/SsfTransmitterProperties.java index 75a3a38..5e68dc9 100644 --- a/easyssf-receiver-spring-boot-autoconfigure/src/main/java/org/easyssf/receiver/spring/boot/SsfTransmitterProperties.java +++ b/easyssf-receiver-spring-boot-autoconfigure/src/main/java/org/easyssf/receiver/spring/boot/SsfTransmitterProperties.java @@ -363,6 +363,21 @@ public static class Poll { */ private int maxEvents = 100; + /** + * Whether to long poll (RFC 8936): the transmitter holds a request until SETs are + * available, and the next request goes out as soon as a response was handled. + * Otherwise the transmitter is asked to answer at once and polled every + * 'interval'. Requires a transmitter that holds requests; one that answers at + * once is polled every 'interval' nevertheless. + */ + private boolean longPolling = false; + + /** + * How long the transmitter holds a long poll request, part of the agreement with + * it. The request waits that long plus a margin for the response. + */ + private Duration longPollingHold = Duration.ofSeconds(30); + private final RateLimit rateLimit = new RateLimit(); public URI getEndpointUrl() { @@ -405,6 +420,22 @@ public void setMaxEvents(int maxEvents) { this.maxEvents = maxEvents; } + public boolean isLongPolling() { + return this.longPolling; + } + + public void setLongPolling(boolean longPolling) { + this.longPolling = longPolling; + } + + public Duration getLongPollingHold() { + return this.longPollingHold; + } + + public void setLongPollingHold(Duration longPollingHold) { + this.longPollingHold = longPollingHold; + } + public RateLimit getRateLimit() { return this.rateLimit; } 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 28c52f2..44fe97b 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 @@ -12,6 +12,7 @@ import org.easyssf.receiver.http.JdkSsfHttpClient; import org.easyssf.receiver.http.SsfHttpClient; import org.easyssf.receiver.metrics.SsfReceiverMetrics; +import org.easyssf.receiver.poll.SsfPollAckStore; import org.easyssf.receiver.poll.SsfPoller; import org.easyssf.receiver.set.InMemorySsfJtiDedupStore; import org.easyssf.receiver.set.NimbusSsfSetVerifier; @@ -167,9 +168,11 @@ SsfStreamRegistrar ssfStreamRegistrar(SsfReceiverProperties properties, SsfStrea @ConditionalOnProperty(name = "easyssf.receiver.delivery-method", havingValue = "poll") SsfPoller ssfPoller(SsfReceiverProperties properties, SsfHttpClient httpClient, SsfTransmitterTokenProvider tokenProvider, SsfSetProcessor processor, SsfReceiverStream receiverStream, - ObjectProvider metrics) { - return SsfTransmitterFactory.poller(SsfTransmitter.DEFAULT_NAME, properties, httpClient, tokenProvider, - processor, receiverStream, metrics.getIfAvailable()); + ObjectProvider metrics, ObjectProvider ackStore) { + SsfHttpClient pollClient = SsfTransmitterFactory.pollHttpClient(SsfTransmitter.DEFAULT_NAME, properties, + httpClient, properties.getHttp()); + return SsfTransmitterFactory.poller(SsfTransmitter.DEFAULT_NAME, properties, pollClient, tokenProvider, + processor, receiverStream, metrics.getIfAvailable(), ackStore.getIfAvailable()); } // ---- shared by all transmitters @@ -213,7 +216,7 @@ SsfTransmitters ssfTransmitters(SsfReceiverProperties properties, SsfHttpClient ObjectProvider tokenProvider, ObjectProvider streamClient, ObjectProvider receiverStream, ObjectProvider streamVerification, ObjectProvider streamRegistrar, ObjectProvider poller, - ObjectProvider customizers) { + ObjectProvider customizers, ObjectProvider ackStore) { Map configured; try { configured = properties.getConfiguredTransmitters(); @@ -245,7 +248,7 @@ SsfTransmitters ssfTransmitters(SsfReceiverProperties properties, SsfHttpClient } else { builder = SsfTransmitterFactory.builder(name, transmitterProperties, httpClient, processor, - metrics.getIfAvailable()); + metrics.getIfAvailable(), ackStore.getIfAvailable(), properties.getHttp()); } customizers.orderedStream().forEach((customizer) -> customizer.customize(builder)); transmitters.add(builder.build()); 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 9d2fab5..fd4d802 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 @@ -6,9 +6,11 @@ import org.apache.commons.logging.LogFactory; import org.easyssf.receiver.jdbc.JdbcSsfExpiringStore; import org.easyssf.receiver.jdbc.JdbcSsfJtiDedupStore; +import org.easyssf.receiver.jdbc.JdbcSsfPollAckStore; import org.easyssf.receiver.jdbc.JdbcSsfSchema; import org.easyssf.receiver.jdbc.JdbcSsfTokenRevocationStore; import org.easyssf.receiver.jdbc.SsfJdbcOperations; +import org.easyssf.receiver.poll.SsfPollAckStore; import org.easyssf.receiver.revocation.SsfTokenRevocationStore; import org.easyssf.receiver.set.SsfJtiDedupStore; import org.easyssf.receiver.spring.boot.SsfReceiverProperties; @@ -87,6 +89,27 @@ JdbcSsfTokenRevocationStore jdbcSsfTokenRevocationStore(JdbcOperations jdbc, Obj properties.getResourceServer().getRevocationTtl()); } + /** + * Keeps the acknowledgements a polling receiver owes its transmitter, so that a SET + * handled right before a restart is acknowledged afterwards instead of delivered + * again. + */ + @Bean + @ConditionalOnMissingBean(SsfPollAckStore.class) + JdbcSsfPollAckStore jdbcSsfPollAckStore(JdbcOperations jdbc, ObjectProvider dataSource, + SsfReceiverProperties properties) { + String tablePrefix = properties.getJdbc().getTablePrefix(); + String table = JdbcSsfSchema.pollAckTable(tablePrefix); + SsfJdbcOperations operations = new JdbcTemplateSsfJdbcOperations(jdbc); + JdbcSsfSchema.prepareTable(operations, table, JdbcSsfSchema.createPollAckTable(tablePrefix), + createTables(properties, dataSource), SCHEMA_HINT); + JdbcSsfPollAckStore store = new JdbcSsfPollAckStore(operations, tablePrefix); + store.setRetention(properties.getJdbc().getAckRetention()); + store.setDeleteBatchSize(properties.getJdbc().getAckDeleteBatchSize()); + logger.info("Pending poll acknowledgements are kept in the table " + table); + return store; + } + /** * Purges the expired rows of the stores every * {@code easyssf.receiver.jdbc.cleanup-interval}. diff --git a/easyssf-receiver-spring-boot-autoconfigure/src/main/java/org/easyssf/receiver/spring/boot/health/SsfReceiverHealthIndicator.java b/easyssf-receiver-spring-boot-autoconfigure/src/main/java/org/easyssf/receiver/spring/boot/health/SsfReceiverHealthIndicator.java index 00179d6..5cc3e78 100644 --- a/easyssf-receiver-spring-boot-autoconfigure/src/main/java/org/easyssf/receiver/spring/boot/health/SsfReceiverHealthIndicator.java +++ b/easyssf-receiver-spring-boot-autoconfigure/src/main/java/org/easyssf/receiver/spring/boot/health/SsfReceiverHealthIndicator.java @@ -95,6 +95,7 @@ else if (state != SsfStreamRegistrar.State.REGISTERED) { details.put("lastPoll", poller.getLastPollAt()); details.put("lastSuccessfulPoll", lastSuccess); details.put("pollError", error); + details.put("pendingAcks", poller.getPendingAckCount()); Instant pausedUntil = poller.getPausedUntil(); if (pausedUntil != null && pausedUntil.isAfter(Instant.now())) { details.put("pausedUntil", pausedUntil); diff --git a/easyssf-receiver-spring-boot-autoconfigure/src/main/java/org/easyssf/receiver/spring/boot/jdbc/JdbcTemplateSsfJdbcOperations.java b/easyssf-receiver-spring-boot-autoconfigure/src/main/java/org/easyssf/receiver/spring/boot/jdbc/JdbcTemplateSsfJdbcOperations.java index 8f09381..b1e7e30 100644 --- a/easyssf-receiver-spring-boot-autoconfigure/src/main/java/org/easyssf/receiver/spring/boot/jdbc/JdbcTemplateSsfJdbcOperations.java +++ b/easyssf-receiver-spring-boot-autoconfigure/src/main/java/org/easyssf/receiver/spring/boot/jdbc/JdbcTemplateSsfJdbcOperations.java @@ -1,5 +1,6 @@ package org.easyssf.receiver.spring.boot.jdbc; +import java.sql.PreparedStatement; import java.util.List; import org.easyssf.receiver.jdbc.SsfJdbcDuplicateKeyException; @@ -9,6 +10,7 @@ import org.springframework.dao.DataAccessException; import org.springframework.dao.DuplicateKeyException; import org.springframework.jdbc.BadSqlGrammarException; +import org.springframework.jdbc.core.ArgumentPreparedStatementSetter; import org.springframework.jdbc.core.JdbcOperations; import org.springframework.util.Assert; @@ -49,6 +51,21 @@ public List query(String sql, SsfJdbcRowMapper mapper, Object... args) } } + @Override + public List query(String sql, int maxRows, SsfJdbcRowMapper mapper, Object... args) { + try { + return this.jdbc.query((connection) -> { + PreparedStatement statement = connection.prepareStatement(sql); + statement.setMaxRows(Math.max(0, maxRows)); + new ArgumentPreparedStatementSetter(args).setValues(statement); + return statement; + }, (row, index) -> mapper.map(row)); + } + catch (DataAccessException ex) { + throw new SsfJdbcException("Could not run " + sql + ": " + ex.getMessage(), null, ex); + } + } + @Override public void execute(String sql) { try { diff --git a/easyssf-receiver-spring-boot-autoconfigure/src/test/java/org/easyssf/receiver/spring/boot/autoconfigure/SsfReceiverAutoConfigurationTests.java b/easyssf-receiver-spring-boot-autoconfigure/src/test/java/org/easyssf/receiver/spring/boot/autoconfigure/SsfReceiverAutoConfigurationTests.java index 9e8b4e0..8481e58 100644 --- a/easyssf-receiver-spring-boot-autoconfigure/src/test/java/org/easyssf/receiver/spring/boot/autoconfigure/SsfReceiverAutoConfigurationTests.java +++ b/easyssf-receiver-spring-boot-autoconfigure/src/test/java/org/easyssf/receiver/spring/boot/autoconfigure/SsfReceiverAutoConfigurationTests.java @@ -4,15 +4,18 @@ import java.util.List; import java.util.concurrent.CopyOnWriteArrayList; +import org.easyssf.core.SsfDeliveryMethod; import org.easyssf.core.event.SsfEventTypes; import org.easyssf.receiver.http.JdkSsfHttpClient; import org.easyssf.receiver.http.SsfHttpClient; import org.easyssf.receiver.http.SsfHttpRequest; import org.easyssf.receiver.jdbc.JdbcSsfJtiDedupStore; +import org.easyssf.receiver.jdbc.JdbcSsfPollAckStore; import org.easyssf.receiver.jdbc.JdbcSsfStoreCleanup; import org.easyssf.receiver.jdbc.JdbcSsfTokenRevocationStore; import org.easyssf.receiver.metrics.MicrometerSsfReceiverMetrics; import org.easyssf.receiver.metrics.SsfReceiverMetrics; +import org.easyssf.receiver.poll.SsfPollAckStore; import org.easyssf.receiver.poll.SsfPoller; import org.easyssf.receiver.revocation.InMemorySsfTokenRevocationStore; import org.easyssf.receiver.revocation.SsfTokenRevocationEventHandler; @@ -23,7 +26,10 @@ import org.easyssf.receiver.set.SsfJtiDedupStore; import org.easyssf.receiver.set.SsfSetProcessor; import org.easyssf.receiver.set.SsfSetVerifier; +import org.easyssf.receiver.spring.boot.SsfReceiverProperties; import org.easyssf.receiver.spring.boot.SsfTransmitterCustomizer; +import org.easyssf.receiver.spring.boot.SsfTransmitterFactory; +import org.easyssf.receiver.spring.boot.SsfTransmitterProperties; import org.easyssf.receiver.spring.boot.http.RestClientSsfHttpClient; import org.easyssf.receiver.spring.boot.resourceserver.SsfRevokedTokenValidator; import org.easyssf.receiver.spring.boot.web.SsfPushEndpoint; @@ -90,6 +96,43 @@ class SsfReceiverAutoConfigurationTests { AutoConfigurations.of(DispatcherServletAutoConfiguration.class, WebMvcAutoConfiguration.class, SecurityAutoConfiguration.class, ServletWebSecurityAutoConfiguration.class)); + @Test + void longPollingIsConfiguredOnThePoller() { + this.webContextRunner + .withPropertyValues("easyssf.receiver.delivery-method=poll", + "easyssf.receiver.poll.endpoint-url=https://idp.example/realms/test/poll", + "easyssf.receiver.poll.long-polling=true", "easyssf.receiver.poll.long-polling-hold=20s") + .run((context) -> { + assertThat(context).hasNotFailed().hasSingleBean(SsfPoller.class); + assertThat(context.getBean(SsfPoller.class).isLongPolling()).isTrue(); + }); + this.webContextRunner + .withPropertyValues("easyssf.receiver.delivery-method=poll", + "easyssf.receiver.poll.endpoint-url=https://idp.example/realms/test/poll") + .run((context) -> assertThat(context.getBean(SsfPoller.class).isLongPolling()).isFalse()); + } + + @Test + void longPollingGetsAJdkClientUnlessTheApplicationClientIsOne() { + SsfTransmitterProperties properties = new SsfTransmitterProperties(); + properties.setTransmitterIssuer("https://idp.example/realms/test"); + properties.setDeliveryMethod(SsfDeliveryMethod.POLL); + SsfHttpClient restClient = (request) -> null; + JdkSsfHttpClient jdkClient = new JdkSsfHttpClient(); + SsfReceiverProperties.Http http = new SsfReceiverProperties.Http(); + http.setUserAgent("my-receiver/1.0"); + + // short polling: the client of the application, whatever it is + assertThat(SsfTransmitterFactory.pollHttpClient("default", properties, restClient, http)).isSameAs(restClient); + properties.getPoll().setLongPolling(true); + assertThat(SsfTransmitterFactory.pollHttpClient("default", properties, jdkClient, http)).isSameAs(jdkClient); + SsfHttpClient pollClient = SsfTransmitterFactory.pollHttpClient("default", properties, restClient, http); + assertThat(pollClient).isInstanceOf(JdkSsfHttpClient.class).isNotSameAs(restClient); + assertThat(((JdkSsfHttpClient) pollClient).getUserAgent()).isEqualTo("my-receiver/1.0"); + properties.setDeliveryMethod(SsfDeliveryMethod.PUSH); + assertThat(SsfTransmitterFactory.pollHttpClient("default", properties, restClient, http)).isSameAs(restClient); + } + @Test void namedTransmittersWorkWithoutTheDefaultOne() { this.namedOnlyContextRunner @@ -337,6 +380,7 @@ void stateIsKeptInMemoryWithoutJdbcTemplate() { void stateIsKeptInTheDatabaseWhenApplicationHasJdbcTemplate() { this.webContextRunner.withConfiguration(JDBC).run((context) -> { assertThat(context).hasSingleBean(SsfJtiDedupStore.class).hasSingleBean(JdbcSsfJtiDedupStore.class); + assertThat(context).hasSingleBean(SsfPollAckStore.class).hasSingleBean(JdbcSsfPollAckStore.class); assertThat(context).hasSingleBean(SsfTokenRevocationStore.class) .hasSingleBean(JdbcSsfTokenRevocationStore.class); // the tables are created in an embedded database diff --git a/easyssf-receiver-spring-boot-autoconfigure/src/test/java/org/easyssf/receiver/spring/boot/jdbc/JdbcTemplateSsfJdbcOperationsTests.java b/easyssf-receiver-spring-boot-autoconfigure/src/test/java/org/easyssf/receiver/spring/boot/jdbc/JdbcTemplateSsfJdbcOperationsTests.java index b9872fb..f68ce63 100644 --- a/easyssf-receiver-spring-boot-autoconfigure/src/test/java/org/easyssf/receiver/spring/boot/jdbc/JdbcTemplateSsfJdbcOperationsTests.java +++ b/easyssf-receiver-spring-boot-autoconfigure/src/test/java/org/easyssf/receiver/spring/boot/jdbc/JdbcTemplateSsfJdbcOperationsTests.java @@ -42,4 +42,21 @@ void mapsDuplicateKeysAndMissingTables() { assertThat(store.purgeExpired()).isZero(); } + @Test + void limitsAQueryThroughTheDriver() { + String prefix = JdbcSsfSchema.DEFAULT_TABLE_PREFIX; + JdbcSsfSchema.prepareTable(this.jdbc, JdbcSsfSchema.pollAckTable(prefix), + JdbcSsfSchema.createPollAckTable(prefix), true, null); + for (int i = 0; i < 3; i++) { + this.jdbc.update("INSERT INTO EASYSSF_POLL_ACK (ISSUER, JTI, RECORDED_AT) VALUES (?, ?, ?)", "i", + "jti-" + i, (long) i); + } + assertThat(this.jdbc.query("SELECT JTI FROM EASYSSF_POLL_ACK WHERE ISSUER = ? ORDER BY RECORDED_AT", 2, + (row) -> row.getString(1), "i")) + .containsExactly("jti-0", "jti-1"); + assertThat(this.jdbc.query("SELECT JTI FROM EASYSSF_POLL_ACK WHERE ISSUER = ? ORDER BY RECORDED_AT", + (row) -> row.getString(1), "i")) + .hasSize(3); + } + } diff --git a/easyssf-receiver-spring-boot-starter/README.md b/easyssf-receiver-spring-boot-starter/README.md index 4f7a91d..d8817a5 100644 --- a/easyssf-receiver-spring-boot-starter/README.md +++ b/easyssf-receiver-spring-boot-starter/README.md @@ -190,7 +190,24 @@ easyssf: `easyssf.receiver.poll.rate-limit.max-backoff` (5 minutes). A `429` without the header pauses for `easyssf.receiver.poll.rate-limit.fallback-backoff`, if set. - With `easyssf.receiver.poll.auto-startup=false` nothing is polled until you call `SsfPoller.pollNow()`. -- Long polling is not used, requests always ask the transmitter to return immediately. +- **Acknowledgements** ride on the next poll request (RFC 8936, section 2.4) and wait in an + `SsfPollAckStore` until a request has carried them. In memory by default; with a `JdbcTemplate` they + are kept in the table `EASYSSF_POLL_ACK` (see [Keeping state in the database](#keeping-state-in-the-database)), + so that a SET handled right before a restart is acknowledged afterwards instead of being delivered + again and skipped as a duplicate. `stop()` sends the pending acknowledgements with a last request. + The health details show them as `pendingAcks`. +- **Long polling**: by default every request asks the transmitter to answer at once + (`returnImmediately: true`) and the transmitter is polled every `poll.interval`. With + `easyssf.receiver.poll.long-polling=true` the poller keeps one request outstanding instead: the + transmitter holds it until SETs are available or its hold time elapses (section 2.5), and the next + request goes out as soon as the response was handled. Set `poll.long-polling-hold` to the hold time + agreed with the transmitter (30 seconds by default); the request waits that long plus a margin. A + transmitter that answers an empty long poll at once is polled every `poll.interval` nevertheless. + Whether Keycloak holds poll requests has not been verified; its default remains short polling. + With long polling on, the poller calls the transmitter with the JDK HTTP client even if the + application has a `RestClient`, because the `RestClient` has no timeout per request + (`SsfTransmitterFactory.pollHttpClient`); a poller bean of your own, built with + `SsfTransmitterFactory.poller(...)`, uses whatever client you pass. ## Stream management @@ -296,6 +313,7 @@ the state is kept in its database instead, without further configuration: |---|---|---| | `EASYSSF_PROCESSED_SET` | `JdbcSsfJtiDedupStore` | the SETs that were processed, 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 @@ -303,21 +321,22 @@ the state is kept in its database instead, without further configuration: (`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`, - `BIGINT`, no vendor syntax) and are tested on H2 and PostgreSQL. + `BIGINT`, no vendor syntax; a result limit is applied by the JDBC driver) and are tested on H2, + PostgreSQL and MySQL. - **Opting out**: `easyssf.receiver.jdbc.enabled=false` keeps the state in memory although the - application has a database. Your own `SsfJtiDedupStore` or `SsfTokenRevocationStore` bean takes - precedence in any case. + application has a database. Your own `SsfJtiDedupStore`, `SsfTokenRevocationStore` or + `SsfPollAckStore` bean takes precedence in any case. - **Cleanup**: expired rows are deleted when a store is written to (the SET table at most once a minute) and every `easyssf.receiver.jdbc.cleanup-interval` (15 minutes) by `JdbcSsfStoreCleanup` on a thread of its own; `easyssf.receiver.jdbc.cleanup-interval=0` turns the periodic cleanup off. - Both stores offer `purgeExpired()` to do it from your own scheduler. + The stores offer `purgeExpired()` to do it from your own scheduler. - **Cost**: a resource server checks every access token with one query on the primary key of `EASYSSF_REVOCATION`. - The stores live in `easyssf-receiver-jdbc` and have no Spring dependency: `SsfJdbcOperations` runs their SQL, over a `DataSource` (`DataSourceSsfJdbcOperations`) or, in Spring Boot, the `JdbcTemplate` of the application, so its transactions and exception translation apply. - The stores use plain SQL (timestamps are stored as milliseconds since the epoch) and were tested - with H2 and PostgreSQL. The revocation table is only used by resource servers. Sessions of an + with H2, PostgreSQL and MySQL. The revocation table is only used by resource servers. Sessions of an OIDC client are not affected, see `SsfSessionTerminator`. ## Calling the transmitter @@ -414,6 +433,8 @@ name, with its own status. Switch it off with `management.health.easyssf.enabled | `easyssf.receiver.poll.interval` | `30s` | | | `easyssf.receiver.poll.initial-delay` | `1s` | | | `easyssf.receiver.poll.max-events` | `100` | SETs per request. | +| `easyssf.receiver.poll.long-polling` | `false` | Keep one request outstanding that the transmitter holds until SETs are available, instead of polling every `interval`. | +| `easyssf.receiver.poll.long-polling-hold` | `30s` | How long the transmitter holds a long poll request, agreed with it; the request waits that long plus a margin. | | `easyssf.receiver.poll.rate-limit.fallback-backoff` | | Pause after a `429` without `Retry-After`; unset, the next poll comes at the regular interval. | | `easyssf.receiver.poll.rate-limit.max-backoff` | `5m` | Longest pause a `Retry-After` header or the fallback can cause. | | `easyssf.receiver.metrics.enabled` | `true` | | @@ -434,6 +455,8 @@ name, with its own status. Switch it off with `management.health.easyssf.enabled | `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. | | `easyssf.receiver.jdbc.cleanup-interval` | `15m` | How often expired rows are purged; `0` turns it off. | +| `easyssf.receiver.jdbc.ack-retention` | `7d` | How long the JDBC store keeps a poll acknowledgement no request managed to deliver. | +| `easyssf.receiver.jdbc.ack-delete-batch-size` | `100` | How many delivered acknowledgements one `DELETE` of the JDBC store removes at once. | | `easyssf.receiver.resource-server.enabled` | `true` | | | `easyssf.receiver.resource-server.event-types` | `CaepSessionRevoked` | Events that revoke access tokens. | | `easyssf.receiver.resource-server.revocation-ttl` | `1h` | | diff --git a/easyssf-receiver/src/main/java/org/easyssf/receiver/http/JdkSsfHttpClient.java b/easyssf-receiver/src/main/java/org/easyssf/receiver/http/JdkSsfHttpClient.java index 4bae5fa..13f8e8d 100644 --- a/easyssf-receiver/src/main/java/org/easyssf/receiver/http/JdkSsfHttpClient.java +++ b/easyssf-receiver/src/main/java/org/easyssf/receiver/http/JdkSsfHttpClient.java @@ -64,7 +64,7 @@ public SsfHttpResponse execute(SsfHttpRequest request) throws IOException { HttpRequest.BodyPublisher body = (request.body() != null) ? HttpRequest.BodyPublishers.ofString(request.body()) : HttpRequest.BodyPublishers.noBody(); HttpRequest.Builder builder = HttpRequest.newBuilder(request.uri()) - .timeout(this.readTimeout) + .timeout((request.timeout() != null) ? request.timeout() : this.readTimeout) .method(request.method(), body); request.headers().forEach(builder::header); if (this.userAgent != null && request.headers().keySet().stream().noneMatch(USER_AGENT::equalsIgnoreCase)) { diff --git a/easyssf-receiver/src/main/java/org/easyssf/receiver/http/SsfHttpRequest.java b/easyssf-receiver/src/main/java/org/easyssf/receiver/http/SsfHttpRequest.java index a99237e..ef2fbd5 100644 --- a/easyssf-receiver/src/main/java/org/easyssf/receiver/http/SsfHttpRequest.java +++ b/easyssf-receiver/src/main/java/org/easyssf/receiver/http/SsfHttpRequest.java @@ -1,6 +1,7 @@ package org.easyssf.receiver.http; import java.net.URI; +import java.time.Duration; import java.util.LinkedHashMap; import java.util.Map; @@ -11,11 +12,17 @@ * @param uri the target of the request * @param headers the request headers * @param body the request body, {@code null} for none + * @param timeout how long to wait for the response, {@code null} for the timeout of the + * client; a long poll (RFC 8936) waits longer than other calls */ -public record SsfHttpRequest(String method, URI uri, Map headers, String body) { +public record SsfHttpRequest(String method, URI uri, Map headers, String body, Duration timeout) { + + public SsfHttpRequest(String method, URI uri, Map headers, String body) { + this(method, uri, headers, body, null); + } public static SsfHttpRequest of(String method, URI uri) { - return new SsfHttpRequest(method, uri, Map.of(), null); + return new SsfHttpRequest(method, uri, Map.of(), null, null); } public static SsfHttpRequest get(URI uri) { @@ -25,7 +32,15 @@ public static SsfHttpRequest get(URI uri) { public SsfHttpRequest withHeader(String name, String value) { Map headers = new LinkedHashMap<>(this.headers); headers.put(name, value); - return new SsfHttpRequest(this.method, this.uri, headers, this.body); + return new SsfHttpRequest(this.method, this.uri, headers, this.body, this.timeout); + } + + /** + * @param timeout how long to wait for the response of this request, instead of the + * timeout of the client + */ + public SsfHttpRequest withTimeout(Duration timeout) { + return new SsfHttpRequest(this.method, this.uri, this.headers, this.body, timeout); } /** @@ -37,7 +52,8 @@ public SsfHttpRequest withBearerToken(String accessToken) { } public SsfHttpRequest withBody(String contentType, String body) { - return new SsfHttpRequest(this.method, this.uri, this.headers, body).withHeader("Content-Type", contentType); + return new SsfHttpRequest(this.method, this.uri, this.headers, body, this.timeout).withHeader("Content-Type", + contentType); } public SsfHttpRequest withJsonBody(String json) { diff --git a/easyssf-receiver/src/main/java/org/easyssf/receiver/poll/InMemorySsfPollAckStore.java b/easyssf-receiver/src/main/java/org/easyssf/receiver/poll/InMemorySsfPollAckStore.java new file mode 100644 index 0000000..818c478 --- /dev/null +++ b/easyssf-receiver/src/main/java/org/easyssf/receiver/poll/InMemorySsfPollAckStore.java @@ -0,0 +1,90 @@ +package org.easyssf.receiver.poll; + +import java.util.ArrayList; +import java.util.Collection; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.locks.ReentrantLock; + +/** + * {@link SsfPollAckStore} that keeps the pending acknowledgements in memory, in the order + * they were recorded. Lost when the application stops, the default of the poller. + */ +public class InMemorySsfPollAckStore implements SsfPollAckStore { + + private final Map pending = new ConcurrentHashMap<>(); + + @Override + public void record(String issuer, SsfPendingAck ack) { + Acks acks = acks(issuer); + acks.lock.lock(); + try { + acks.byJti.putIfAbsent(ack.jti(), ack); + } + finally { + acks.lock.unlock(); + } + } + + @Override + public List pending(String issuer, int limit) { + Acks acks = acks(issuer); + acks.lock.lock(); + try { + List result = new ArrayList<>(Math.min(limit, acks.byJti.size())); + for (SsfPendingAck ack : acks.byJti.values()) { + if (result.size() >= limit) { + break; + } + result.add(ack); + } + return result; + } + finally { + acks.lock.unlock(); + } + } + + @Override + public void remove(String issuer, Collection jtis) { + Acks acks = acks(issuer); + acks.lock.lock(); + try { + jtis.forEach(acks.byJti::remove); + } + finally { + acks.lock.unlock(); + } + } + + @Override + public int size(String issuer) { + Acks acks = acks(issuer); + acks.lock.lock(); + try { + return acks.byJti.size(); + } + finally { + acks.lock.unlock(); + } + } + + private Acks acks(String issuer) { + return this.pending.computeIfAbsent(String.valueOf(issuer), (key) -> new Acks()); + } + + /** + * The pending acknowledgements of one transmitter. A lock rather than + * {@code synchronized}, which pins virtual threads before JDK 24. + */ + private static final class Acks { + + private final ReentrantLock lock = new ReentrantLock(); + + private final Map byJti = new LinkedHashMap<>(); + + } + +} diff --git a/easyssf-receiver/src/main/java/org/easyssf/receiver/poll/SsfPendingAck.java b/easyssf-receiver/src/main/java/org/easyssf/receiver/poll/SsfPendingAck.java new file mode 100644 index 0000000..d12db32 --- /dev/null +++ b/easyssf-receiver/src/main/java/org/easyssf/receiver/poll/SsfPendingAck.java @@ -0,0 +1,44 @@ +package org.easyssf.receiver.poll; + +import org.easyssf.core.support.SsfAssert; + +/** + * What the poller owes the transmitter for a SET it fetched (RFC 8936, section 2.2): an + * acknowledgement that the SET was received and handled, or an error that it was + * rejected, which goes into the {@code setErrs} member of the next poll request. + * + * @param jti the identifier of the SET + * @param errorCode the error code of a rejected SET ({@code invalid_request}, ...), + * {@code null} for an acknowledgement + * @param errorDescription a description of the error, may be {@code null} + */ +public record SsfPendingAck(String jti, String errorCode, String errorDescription) { + + public SsfPendingAck { + SsfAssert.notNull(jti, "jti must not be null"); + SsfAssert.isTrue(errorCode != null || errorDescription == null, "an acknowledgement has no error description"); + } + + /** + * The acknowledgement of a handled SET. + */ + public static SsfPendingAck ack(String jti) { + return new SsfPendingAck(jti, null, null); + } + + /** + * The report of a rejected SET. + */ + public static SsfPendingAck error(String jti, String errorCode, String errorDescription) { + SsfAssert.notNull(errorCode, "errorCode must not be null"); + return new SsfPendingAck(jti, errorCode, errorDescription); + } + + /** + * Whether this reports an error rather than acknowledging the SET. + */ + public boolean isError() { + return this.errorCode != null; + } + +} diff --git a/easyssf-receiver/src/main/java/org/easyssf/receiver/poll/SsfPollAckStore.java b/easyssf-receiver/src/main/java/org/easyssf/receiver/poll/SsfPollAckStore.java new file mode 100644 index 0000000..e710654 --- /dev/null +++ b/easyssf-receiver/src/main/java/org/easyssf/receiver/poll/SsfPollAckStore.java @@ -0,0 +1,48 @@ +package org.easyssf.receiver.poll; + +import java.util.Collection; +import java.util.List; + +/** + * Keeps the acknowledgements and error reports the poller owes a transmitter until a poll + * request has carried them (RFC 8936, section 2.4: they ride on the next request). The + * {@link InMemorySsfPollAckStore} loses them when the application stops, so a SET handled + * but not yet acknowledged is delivered again after a restart and skipped as a duplicate. + * A durable implementation, such as the JDBC store of {@code easyssf-receiver-jdbc}, + * acknowledges it with the first poll after the restart instead. + * + *

+ * Entries are kept per transmitter, identified by its issuer, because a {@code jti} is + * only unique per issuer and a store may serve several transmitters. Implementations have + * to be safe for concurrent use. + */ +public interface SsfPollAckStore { + + /** + * Records an acknowledgement or error report. Recording the same {@code jti} again + * does nothing. + * @param issuer the issuer of the transmitter the SET came from + * @param ack what to tell the transmitter + */ + void record(String issuer, SsfPendingAck ack); + + /** + * The acknowledgements and error reports waiting for the transmitter, oldest first. + * @param issuer the issuer of the transmitter + * @param limit the maximum number to return + */ + List pending(String issuer, int limit); + + /** + * Removes entries the transmitter has received. + * @param issuer the issuer of the transmitter + * @param jtis the identifiers of the SETs whose entries were delivered + */ + void remove(String issuer, Collection jtis); + + /** + * @return the number of entries waiting for the transmitter + */ + int size(String issuer); + +} 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 11fd522..4b201c3 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 @@ -7,9 +7,7 @@ import java.util.LinkedHashMap; import java.util.List; import java.util.Map; -import java.util.concurrent.Executors; -import java.util.concurrent.ScheduledExecutorService; -import java.util.concurrent.TimeUnit; +import java.util.concurrent.ThreadFactory; import java.util.concurrent.locks.ReentrantLock; import java.util.function.Supplier; @@ -35,7 +33,18 @@ *

* A SET is acknowledged once it was handled. An invalid SET is reported as an error to * the transmitter. A SET that could not be handled is neither acknowledged nor reported, - * the transmitter delivers it again with a later poll. + * the transmitter delivers it again with a later poll. Acknowledgements and error reports + * ride on the next poll request (section 2.4) and wait in the {@link SsfPollAckStore} + * until that request succeeded; with the in-memory store of the default they are lost + * when the application stops, with a durable store they survive a restart. + * + *

+ * By default the poller polls at a fixed {@link #setInterval(Duration) interval} and asks + * the transmitter to answer right away ({@code returnImmediately: true}, short polling). + * With {@link #setLongPolling(Duration) long polling} it keeps one request outstanding + * instead: the transmitter holds the request until SETs are available or its hold time + * elapses (section 2.5), and the poller sends the next request as soon as it has handled + * the response. A backlog is fetched with immediate requests in both modes. */ public class SsfPoller { @@ -45,6 +54,19 @@ public class SsfPoller { private static final int MAX_REQUESTS_PER_POLL = 100; + /** + * How much longer than the hold time of the transmitter a long poll waits for the + * response. + */ + private static final Duration LONG_POLL_MARGIN = Duration.ofSeconds(10); + + /** + * A long poll that comes back empty faster than this was not held: the transmitter + * does not support long polling, and the poller waits the interval before the next + * request instead of asking again at once. + */ + private static final Duration MIN_HOLD = Duration.ofSeconds(1); + private final SsfHttpClient httpClient; private final SsfTransmitterTokenProvider tokenProvider; @@ -53,9 +75,9 @@ public class SsfPoller { private final SsfSetProcessor processor; - private final List pendingAcks = new ArrayList<>(); + private SsfPollAckStore ackStore = new InMemorySsfPollAckStore(); - private final Map pendingErrors = new LinkedHashMap<>(); + private int maxAckBatch = 1000; private SsfReceiverMetrics metrics = SsfReceiverMetrics.NOOP; @@ -63,6 +85,12 @@ public class SsfPoller { private Duration initialDelay = Duration.ZERO; + private Duration longPollingHold; + + private boolean flushOnStop = true; + + private ThreadFactory threadFactory = Thread.ofPlatform().name("ssf-poller").daemon().factory(); + private int maxEvents = 100; private String transmitter; @@ -79,7 +107,9 @@ public class SsfPoller { private volatile String lastPollError; - private volatile ScheduledExecutorService scheduler; + private volatile Thread thread; + + private volatile boolean running; private final ReentrantLock polling = new ReentrantLock(); @@ -109,6 +139,28 @@ public void setMetrics(SsfReceiverMetrics metrics) { this.metrics = metrics; } + /** + * @param ackStore keeps the acknowledgements until a poll request carried them; an + * {@link InMemorySsfPollAckStore} by default + */ + public void setAckStore(SsfPollAckStore ackStore) { + SsfAssert.notNull(ackStore, "ackStore must not be null"); + this.ackStore = ackStore; + } + + /** + * @param maxAckBatch the most acknowledgements and error reports one request carries, + * the rest goes with the next one; 1000 by default + */ + public void setMaxAckBatch(int maxAckBatch) { + SsfAssert.isTrue(maxAckBatch > 0, "maxAckBatch must be positive"); + this.maxAckBatch = maxAckBatch; + } + + /** + * Sets the time between two polls. With long polling, the pause after a failed + * request or after a transmitter answered an empty long poll at once. + */ public void setInterval(Duration interval) { SsfAssert.isTrue(interval != null && interval.isPositive(), "interval must be positive"); this.interval = interval; @@ -120,10 +172,48 @@ public void setInitialDelay(Duration initialDelay) { } /** - * Sets the maximum number of SETs fetched with one request. + * Switches to long polling (RFC 8936, section 2.5): the first request of a poll asks + * the transmitter to hold it until SETs are available + * ({@code returnImmediately: false}), and a running poller sends the next request as + * soon as a response was handled. + * @param hold how long the transmitter holds a request, part of the agreement with it + * (section 2.2); the request waits that long plus a margin for the response. + * {@code null} returns to short polling. */ + public void setLongPolling(Duration hold) { + SsfAssert.isTrue(hold == null || hold.isPositive(), "hold must be positive"); + this.longPollingHold = hold; + } + + public boolean isLongPolling() { + return this.longPollingHold != null; + } + + /** + * @param flushOnStop whether {@link #stop()} sends the pending acknowledgements with + * a last request, so that the transmitter does not deliver the SETs again after a + * restart; {@code true} by default + */ + public void setFlushOnStop(boolean flushOnStop) { + this.flushOnStop = flushOnStop; + } + + /** + * Sets what creates the thread {@link #start()} polls on, for a framework that + * manages its threads, names them per transmitter or prefers virtual threads + * ({@code Thread.ofVirtual().name("ssf-poller").factory()} works: the loop only + * blocks in the HTTP call and in sleeps). By default a platform daemon thread named + * {@code ssf-poller}. {@link #stop()} interrupts the thread, so the factory must not + * hand out threads that swallow interrupts. + */ + public void setThreadFactory(ThreadFactory threadFactory) { + SsfAssert.notNull(threadFactory, "threadFactory must not be null"); + this.threadFactory = threadFactory; + } + /** * @param transmitter the issuer of the transmitter that is polled, for the metrics + * and as the key of its acknowledgements in the {@link SsfPollAckStore} */ public void setTransmitter(String transmitter) { this.transmitter = transmitter; @@ -149,26 +239,31 @@ public void setMaxPause(Duration maxPause) { this.maxPause = maxPause; } + /** + * Sets the maximum number of SETs fetched with one request. + */ public void setMaxEvents(int maxEvents) { SsfAssert.isTrue(maxEvents > 0, "maxEvents must be positive"); this.maxEvents = maxEvents; } /** - * Starts polling the transmitter periodically. Without it, SETs are only fetched by - * Calling it again while running has no effect. calling {@link #pollNow()}. + * Starts polling the transmitter on a thread of its own (see + * {@link #setThreadFactory(ThreadFactory)}): periodically, or with one request always + * outstanding when long polling. Without it, SETs are only fetched by calling + * {@link #pollNow()}. Calling it again while running has no effect. */ public void start() { this.lifecycle.lock(); try { - if (this.scheduler != null) { + if (this.thread != null) { return; } - ScheduledExecutorService scheduler = Executors - .newSingleThreadScheduledExecutor(Thread.ofPlatform().name("ssf-poller").daemon().factory()); - scheduler.scheduleWithFixedDelay(this::pollQuietly, this.initialDelay.toMillis(), this.interval.toMillis(), - TimeUnit.MILLISECONDS); - this.scheduler = scheduler; + this.running = true; + Thread thread = this.threadFactory.newThread(this::run); + SsfAssert.notNull(thread, "the thread factory returned no thread"); + this.thread = thread; + thread.start(); } finally { this.lifecycle.unlock(); @@ -176,15 +271,27 @@ public void start() { } /** - * Stops polling the transmitter periodically. + * Stops polling: ends the polling thread, interrupting an outstanding long poll, and + * sends the pending acknowledgements with a last request unless + * {@link #setFlushOnStop(boolean)} says otherwise. */ public void stop() { this.lifecycle.lock(); try { - ScheduledExecutorService scheduler = this.scheduler; - this.scheduler = null; - if (scheduler != null) { - scheduler.shutdownNow(); + Thread thread = this.thread; + this.thread = null; + this.running = false; + if (thread != null) { + thread.interrupt(); + try { + thread.join(Duration.ofSeconds(5)); + } + catch (InterruptedException ex) { + Thread.currentThread().interrupt(); + } + if (this.flushOnStop) { + flushAcks(); + } } } finally { @@ -193,7 +300,7 @@ public void stop() { } public boolean isRunning() { - return this.scheduler != null; + return this.thread != null; } /** @@ -225,20 +332,69 @@ public Instant getPausedUntil() { return this.pausedUntil; } - private void pollQuietly() { + /** + * @return the number of acknowledgements and error reports waiting for the next poll + * request + */ + public int getPendingAckCount() { + String key = ackKey(this.endpoint.get()); + return (key != null) ? this.ackStore.size(key) : 0; + } + + private void run() { try { - pollNow(); + Thread.sleep(this.initialDelay); + while (this.running) { + Duration pause = this.interval; + long started = System.nanoTime(); + try { + PollResult result = this.polling.tryLock() ? pollLocked() : PollResult.SKIPPED; + if (result.skipped()) { + pause = pauseWhileSkipped(); + } + else if (isLongPolling()) { + Duration elapsed = Duration.ofNanos(System.nanoTime() - started); + boolean held = result.fetched() > 0 || elapsed.compareTo(MIN_HOLD) >= 0; + pause = held ? Duration.ZERO : this.interval; + } + } + catch (RuntimeException ex) { + if (!this.running) { + return; + } + logger.warn("Could not poll the SSF transmitter: " + ex.getMessage()); + logger.debug("Cause of the failed poll", ex); + } + if (this.running && pause.isPositive()) { + Thread.sleep(pause); + } + } } - catch (RuntimeException ex) { - logger.warn("Could not poll the SSF transmitter: " + ex.getMessage()); - logger.debug("Cause of the failed poll", ex); + catch (InterruptedException ex) { + // stop() ends the thread } } + private PollResult pollLocked() { + try { + return poll(); + } + finally { + this.polling.unlock(); + } + } + + private Duration pauseWhileSkipped() { + Instant pausedUntil = this.pausedUntil; + Duration untilResume = Duration.between(Instant.now(), pausedUntil); + return (untilResume.isPositive() && untilResume.compareTo(this.interval) > 0) ? untilResume : this.interval; + } + /** * Fetches and processes the SETs that are available at the transmitter and - * acknowledges them. One poll runs at a time: a call while another is in progress - * returns right away. + * acknowledges them, with the request answered immediately whether or not long + * polling is on. One poll runs at a time: a call while another is in progress returns + * right away. * @return the number of SETs fetched, {@code 0} as well if the poll endpoint is not * known yet, the transmitter asked to slow down or a poll is in progress */ @@ -248,43 +404,51 @@ public int pollNow() { return 0; } try { - return poll(); + return poll(false).fetched(); } finally { this.polling.unlock(); } } - private int poll() { + private PollResult poll() { + return poll(isLongPolling()); + } + + private PollResult poll(boolean hold) { URI endpoint = this.endpoint.get(); if (endpoint == null) { logger.debug("Not polling, the poll endpoint of the SSF stream is not known yet"); - return 0; + return PollResult.SKIPPED; } if (Instant.now().isBefore(this.pausedUntil)) { logger.debug("Not polling, the SSF transmitter asked to wait until " + this.pausedUntil); - return 0; + return PollResult.SKIPPED; } this.lastPollAt = Instant.now(); + String key = ackKey(endpoint); try { int fetched = 0; boolean moreAvailable = true; for (int request = 0; moreAvailable && request < MAX_REQUESTS_PER_POLL; request++) { - Map response = poll(endpoint, this.maxEvents); + // only the first request of a poll is held, a backlog is fetched right + // away + Map response = poll(endpoint, key, this.maxEvents, !(hold && request == 0)); Map sets = sets(response); - sets.forEach(this::process); + sets.forEach((jti, set) -> process(key, jti, set)); fetched += sets.size(); moreAvailable = Boolean.TRUE.equals(response.get("moreAvailable")) && !sets.isEmpty(); } - if (!this.pendingAcks.isEmpty() || !this.pendingErrors.isEmpty()) { - // acknowledge right away instead of with the next poll - Map sets = sets(poll(endpoint, 0)); - sets.forEach(this::process); + if (!hold && this.ackStore.size(key) > 0) { + // acknowledge right away instead of with the next poll; a long poll sends + // its next request at once anyway + Map sets = sets(poll(endpoint, key, 0, true)); + sets.forEach((jti, set) -> process(key, jti, set)); fetched += sets.size(); } this.lastSuccessfulPollAt = Instant.now(); this.lastPollError = null; - return fetched; + return new PollResult(fetched, false); } catch (RuntimeException ex) { this.lastPollError = ex.getMessage(); @@ -292,17 +456,14 @@ private int poll() { } } - private void process(String jti, Object encodedSet) { + private void process(String key, String jti, Object encodedSet) { try { this.processor.process(String.valueOf(encodedSet), SsfDeliveryMethod.POLL); - this.pendingAcks.add(jti); + this.ackStore.record(key, SsfPendingAck.ack(jti)); } catch (SsfSetVerificationException ex) { logger.warn("Rejecting polled SET " + jti + ": " + ex.getMessage()); - Map error = new LinkedHashMap<>(); - error.put("err", ex.getErrorCode()); - error.put("description", ex.getMessage()); - this.pendingErrors.put(jti, error); + this.ackStore.record(key, SsfPendingAck.error(jti, ex.getErrorCode(), ex.getMessage())); } catch (SsfTransmitterUnavailableException | SsfEventHandlingException ex) { // not acknowledged, the transmitter delivers the SET again @@ -310,12 +471,26 @@ private void process(String jti, Object encodedSet) { } } - private Map poll(URI endpoint, int maxEvents) { + private Map poll(URI endpoint, String key, int maxEvents, boolean returnImmediately) { + List pending = this.ackStore.pending(key, this.maxAckBatch); + List acks = new ArrayList<>(); + Map errors = new LinkedHashMap<>(); + for (SsfPendingAck ack : pending) { + if (ack.isError()) { + Map error = new LinkedHashMap<>(); + error.put("err", ack.errorCode()); + if (ack.errorDescription() != null) { + error.put("description", ack.errorDescription()); + } + errors.put(ack.jti(), error); + } + else { + acks.add(ack.jti()); + } + } Map request = new LinkedHashMap<>(); request.put("maxEvents", maxEvents); - request.put("returnImmediately", true); - List acks = List.copyOf(this.pendingAcks); - Map errors = Map.copyOf(this.pendingErrors); + request.put("returnImmediately", returnImmediately); if (!acks.isEmpty()) { request.put("ack", acks); } @@ -326,11 +501,11 @@ private Map poll(URI endpoint, int maxEvents) { SsfHttpResponse httpResponse; Map response; try { - httpResponse = send(endpoint, request); + httpResponse = send(endpoint, request, returnImmediately); if (httpResponse.status() == 401) { // the access token expired or was revoked: try once more with a new one this.tokenProvider.invalidate(); - httpResponse = send(endpoint, request); + httpResponse = send(endpoint, request, returnImmediately); } String body = httpResponse.body(); response = (httpResponse.isSuccessful() && body != null && !body.isBlank()) ? JSONObjectUtils.parse(body) @@ -346,17 +521,53 @@ private Map poll(URI endpoint, int maxEvents) { throw new IllegalStateException( "Poll request to " + endpoint + " failed with status " + httpResponse.status()); } - this.pendingAcks.removeAll(acks); - errors.keySet().forEach(this.pendingErrors::remove); + if (!pending.isEmpty()) { + this.ackStore.remove(key, pending.stream().map(SsfPendingAck::jti).toList()); + } this.metrics.pollCompleted(this.transmitter, Duration.ofNanos(System.nanoTime() - started), true); return response; } - private SsfHttpResponse send(URI endpoint, Map request) throws java.io.IOException { - return this.httpClient.execute(SsfHttpRequest.of("POST", endpoint) + private SsfHttpResponse send(URI endpoint, Map request, boolean returnImmediately) + throws java.io.IOException { + SsfHttpRequest httpRequest = SsfHttpRequest.of("POST", endpoint) .withHeader("Accept", "application/json") .withBearerToken(this.tokenProvider.getAccessToken()) - .withJsonBody(JSONObjectUtils.toJSONString(request))); + .withJsonBody(JSONObjectUtils.toJSONString(request)); + if (!returnImmediately && this.longPollingHold != null) { + httpRequest = httpRequest.withTimeout(this.longPollingHold.plus(LONG_POLL_MARGIN)); + } + return this.httpClient.execute(httpRequest); + } + + /** + * Sends the pending acknowledgements with a request that asks for no SETs, best + * effort. + */ + private void flushAcks() { + URI endpoint = this.endpoint.get(); + String key = ackKey(endpoint); + if (key == null || this.ackStore.size(key) == 0 || !this.polling.tryLock()) { + return; + } + try { + poll(endpoint, key, 0, true); + logger.debug("Acknowledged the pending SETs before stopping"); + } + catch (RuntimeException ex) { + logger.debug("Could not acknowledge the pending SETs before stopping, they are sent with the next poll: " + + ex.getMessage()); + } + finally { + this.polling.unlock(); + } + } + + private String ackKey(URI endpoint) { + if (this.transmitter != null) { + return this.transmitter; + } + return (endpoint != null) ? endpoint.toString() : null; } private void pauseIfAsked(SsfHttpResponse response) { @@ -384,4 +595,10 @@ private static Map sets(Map response) { return (response.get("sets") instanceof Map sets) ? (Map) sets : Map.of(); } + private record PollResult(int fetched, boolean skipped) { + + static final PollResult SKIPPED = new PollResult(0, true); + + } + } 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 b66288a..16d1780 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 @@ -2,6 +2,7 @@ import java.util.LinkedHashMap; import java.util.Map; +import java.util.concurrent.locks.ReentrantLock; import org.easyssf.core.event.SsfEventToken; @@ -13,6 +14,9 @@ public class InMemorySsfJtiDedupStore implements SsfJtiDedupStore { private final Map seen; + /** a lock rather than synchronized, which pins virtual threads before JDK 24 */ + private final ReentrantLock lock = new ReentrantLock(); + public InMemorySsfJtiDedupStore(int capacity) { int maxEntries = Math.max(1, capacity); this.seen = new LinkedHashMap<>() { @@ -24,17 +28,35 @@ protected boolean removeEldestEntry(Map.Entry eldest) { } @Override - public synchronized boolean seenBefore(SsfEventToken eventToken) { - return this.seen.put(SsfJtiDedupStore.key(eventToken), Boolean.TRUE) != null; + public boolean seenBefore(SsfEventToken eventToken) { + this.lock.lock(); + try { + return this.seen.put(SsfJtiDedupStore.key(eventToken), Boolean.TRUE) != null; + } + finally { + this.lock.unlock(); + } } @Override - public synchronized void forget(SsfEventToken eventToken) { - this.seen.remove(SsfJtiDedupStore.key(eventToken)); + public void forget(SsfEventToken eventToken) { + this.lock.lock(); + try { + this.seen.remove(SsfJtiDedupStore.key(eventToken)); + } + finally { + this.lock.unlock(); + } } - public synchronized int size() { - return this.seen.size(); + public int size() { + this.lock.lock(); + try { + return this.seen.size(); + } + finally { + this.lock.unlock(); + } } } diff --git a/easyssf-receiver/src/test/java/org/easyssf/receiver/poll/InMemorySsfPollAckStoreTests.java b/easyssf-receiver/src/test/java/org/easyssf/receiver/poll/InMemorySsfPollAckStoreTests.java new file mode 100644 index 0000000..5a4f095 --- /dev/null +++ b/easyssf-receiver/src/test/java/org/easyssf/receiver/poll/InMemorySsfPollAckStoreTests.java @@ -0,0 +1,57 @@ +package org.easyssf.receiver.poll; + +import java.util.List; + +import org.junit.jupiter.api.Test; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatIllegalArgumentException; + +class InMemorySsfPollAckStoreTests { + + private static final String ISSUER = "https://idp.example"; + + private final InMemorySsfPollAckStore store = new InMemorySsfPollAckStore(); + + @Test + void keepsAcknowledgementsAndErrorsInOrderPerIssuer() { + this.store.record(ISSUER, SsfPendingAck.ack("jti-1")); + this.store.record(ISSUER, SsfPendingAck.error("jti-2", "invalid_request", "not a SET")); + this.store.record(ISSUER, SsfPendingAck.ack("jti-3")); + this.store.record("https://other.example", SsfPendingAck.ack("jti-1")); + + assertThat(this.store.size(ISSUER)).isEqualTo(3); + assertThat(this.store.pending(ISSUER, 10)).extracting(SsfPendingAck::jti) + .containsExactly("jti-1", "jti-2", "jti-3"); + assertThat(this.store.pending(ISSUER, 2)).extracting(SsfPendingAck::jti).containsExactly("jti-1", "jti-2"); + assertThat(this.store.pending(ISSUER, 10).get(1).isError()).isTrue(); + assertThat(this.store.pending(ISSUER, 10).get(1).errorDescription()).isEqualTo("not a SET"); + assertThat(this.store.size("https://other.example")).isEqualTo(1); + assertThat(this.store.size("https://unknown.example")).isZero(); + } + + @Test + void recordingTheSameSetAgainChangesNothing() { + this.store.record(ISSUER, SsfPendingAck.ack("jti-1")); + this.store.record(ISSUER, SsfPendingAck.error("jti-1", "invalid_request", null)); + assertThat(this.store.pending(ISSUER, 10)).singleElement() + .satisfies((ack) -> assertThat(ack.isError()).isFalse()); + } + + @Test + void removesWhatTheTransmitterReceived() { + this.store.record(ISSUER, SsfPendingAck.ack("jti-1")); + this.store.record(ISSUER, SsfPendingAck.ack("jti-2")); + this.store.remove(ISSUER, List.of("jti-1", "jti-unknown")); + assertThat(this.store.pending(ISSUER, 10)).extracting(SsfPendingAck::jti).containsExactly("jti-2"); + } + + @Test + void pendingAckIsValidated() { + assertThatIllegalArgumentException().isThrownBy(() -> SsfPendingAck.error("jti", null, "x")); + assertThatIllegalArgumentException() + .isThrownBy(() -> new SsfPendingAck("jti", null, "description without code")); + assertThat(SsfPendingAck.ack("jti").isError()).isFalse(); + } + +} diff --git a/easyssf-receiver/src/test/java/org/easyssf/receiver/poll/SsfPollerTests.java b/easyssf-receiver/src/test/java/org/easyssf/receiver/poll/SsfPollerTests.java index 36dbe6c..d03d88f 100644 --- a/easyssf-receiver/src/test/java/org/easyssf/receiver/poll/SsfPollerTests.java +++ b/easyssf-receiver/src/test/java/org/easyssf/receiver/poll/SsfPollerTests.java @@ -34,6 +34,10 @@ class SsfPollerTests { private final List failures = new ArrayList<>(); + /** runs after a SET was handled, before it is acknowledged */ + private Runnable afterHandling = () -> { + }; + private final SimpleMeterRegistry meterRegistry = new SimpleMeterRegistry(); private SsfPoller poller; @@ -41,6 +45,7 @@ class SsfPollerTests { @BeforeEach void setUp() { transmitter.reset(); + transmitter.setAvailable(true); this.poller = poller(() -> URI.create(transmitter.pollUri()), TestTransmitter.ACCESS_TOKEN); } @@ -59,6 +64,7 @@ private SsfPoller poller(java.util.function.Supplier endpoint, SsfTransmitt throw this.failures.remove(0); } this.handled.add(eventContext.eventToken().jti()); + this.afterHandling.run(); }; SsfHttpClient http = new JdkSsfHttpClient(); NimbusSsfSetVerifier verifier = new NimbusSsfSetVerifier(transmitter.issuer(), transmitter::jwksUri, http); @@ -84,14 +90,43 @@ void startIsIdempotentAndStopEndsThePollingThread() { } private static List pollerThreads() { + return pollerThreads("ssf-poller"); + } + + private static List pollerThreads(String name) { return Thread.getAllStackTraces() .keySet() .stream() - .filter((thread) -> thread.getName().equals("ssf-poller")) + .filter((thread) -> thread.getName().equals(name)) .filter(Thread::isAlive) .toList(); } + @Test + void pollsOnTheThreadOfTheGivenFactory() { + this.poller.setThreadFactory(Thread.ofPlatform().name("ssf-poller-idp").daemon().factory()); + this.poller.setInitialDelay(Duration.ofHours(1)); + this.poller.start(); + assertThat(pollerThreads("ssf-poller-idp")).hasSize(1); + assertThat(pollerThreads()).isEmpty(); + this.poller.stop(); + await().atMost(Duration.ofSeconds(5)) + .untilAsserted(() -> assertThat(pollerThreads("ssf-poller-idp")).isEmpty()); + } + + @Test + void pollsOnAVirtualThreadIfTheFactoryMakesOne() { + transmitter.setLongPollHold(Duration.ofSeconds(5)); + this.poller.setLongPolling(Duration.ofSeconds(5)); + this.poller.setThreadFactory(Thread.ofVirtual().name("ssf-poller-virtual").factory()); + this.poller.start(); + await().atMost(Duration.ofSeconds(5)).until(() -> transmitter.pollRequests() >= 1); + String jti = queue("session-1"); + await().atMost(Duration.ofSeconds(3)).untilAsserted(() -> assertThat(this.handled).containsExactly(jti)); + this.poller.stop(); + assertThat(this.poller.isRunning()).isFalse(); + } + @Test void handlesAndAcknowledgesAvailableSets() { String first = queue("session-1"); @@ -179,6 +214,123 @@ void reportsFailedPollRequest() { .isEqualTo(1); } + @Test + void acknowledgementsWaitInTheStoreUntilARequestCarriedThem() { + InMemorySsfPollAckStore store = new InMemorySsfPollAckStore(); + this.poller.setAckStore(store); + this.poller.setTransmitter(transmitter.issuer()); + String jti = queue("session-1"); + transmitter.queueSet("broken", "not-a-set"); + // the transmitter goes away between handling the SETs and the request that + // acknowledges them + this.afterHandling = () -> transmitter.setAvailable(false); + + assertThatIllegalStateException().isThrownBy(this.poller::pollNow).withMessageContaining("503"); + + assertThat(this.handled).containsExactly(jti); + assertThat(transmitter.acknowledgedSets()).isEmpty(); + assertThat(store.pending(transmitter.issuer(), 10)).extracting(SsfPendingAck::jti) + .containsExactly(jti, "broken"); + assertThat(this.poller.getPendingAckCount()).isEqualTo(2); + + transmitter.setAvailable(true); + this.afterHandling = () -> { + }; + this.poller.pollNow(); + assertThat(transmitter.acknowledgedSets()).containsExactly(jti); + assertThat(transmitter.reportedErrors()).containsOnlyKeys("broken"); + assertThat(store.size(transmitter.issuer())).isZero(); + } + + @Test + void acknowledgementsAreSentInBatches() { + this.poller.setMaxAckBatch(2); + List queued = List.of(queue("session-1"), queue("session-2"), queue("session-3")); + + assertThat(this.poller.pollNow()).isEqualTo(3); + // the request after handling carried two, the third waits for the next poll + assertThat(transmitter.acknowledgedSets()).containsExactly(queued.get(0), queued.get(1)); + assertThat(this.poller.getPendingAckCount()).isEqualTo(1); + + this.poller.pollNow(); + assertThat(transmitter.acknowledgedSets()).containsExactlyElementsOf(queued); + assertThat(this.poller.getPendingAckCount()).isZero(); + } + + @Test + void stopFlushesThePendingAcknowledgements() { + this.poller.setInitialDelay(Duration.ofHours(1)); + String jti = queue("session-1"); + this.afterHandling = () -> transmitter.setAvailable(false); + assertThatIllegalStateException().isThrownBy(this.poller::pollNow); + assertThat(this.poller.getPendingAckCount()).isEqualTo(1); + transmitter.setAvailable(true); + + this.poller.start(); + this.poller.stop(); + + assertThat(transmitter.acknowledgedSets()).containsExactly(jti); + assertThat(this.poller.getPendingAckCount()).isZero(); + } + + @Test + void longPollIsHeldUntilASetArrives() throws Exception { + transmitter.setLongPollHold(Duration.ofSeconds(5)); + this.poller.setLongPolling(Duration.ofSeconds(5)); + this.poller.setInterval(Duration.ofSeconds(30)); + this.poller.start(); + await().atMost(Duration.ofSeconds(5)).until(() -> transmitter.pollRequests() >= 1); + assertThat(transmitter.lastPollRequest()).containsEntry("returnImmediately", false); + assertThat(this.handled).isEmpty(); + + long queuedAt = System.nanoTime(); + String jti = queue("session-1"); + await().atMost(Duration.ofSeconds(3)).untilAsserted(() -> assertThat(this.handled).containsExactly(jti)); + // handled while the request was held, long before the hold time elapsed + assertThat(Duration.ofNanos(System.nanoTime() - queuedAt)).isLessThan(Duration.ofSeconds(3)); + // the next request went out right away and carries the acknowledgement + await().atMost(Duration.ofSeconds(3)) + .untilAsserted(() -> assertThat(transmitter.acknowledgedSets()).containsExactly(jti)); + assertThat(transmitter.lastPollRequest()).containsEntry("returnImmediately", false); + this.poller.stop(); + } + + @Test + void longPollAgainstATransmitterThatAnswersAtOnceWaitsTheInterval() { + // the test transmitter does not hold by default + this.poller.setLongPolling(Duration.ofSeconds(5)); + this.poller.setInterval(Duration.ofMinutes(5)); + this.poller.start(); + await().atMost(Duration.ofSeconds(5)).until(() -> transmitter.pollRequests() >= 1); + await().during(Duration.ofSeconds(1)) + .atMost(Duration.ofSeconds(2)) + .untilAsserted(() -> assertThat(transmitter.pollRequests()).isEqualTo(1)); + this.poller.stop(); + } + + @Test + void stopInterruptsAHeldLongPoll() { + transmitter.setLongPollHold(Duration.ofSeconds(30)); + this.poller.setLongPolling(Duration.ofSeconds(30)); + this.poller.start(); + await().atMost(Duration.ofSeconds(5)).until(() -> transmitter.pollRequests() >= 1); + long stoppedAt = System.nanoTime(); + this.poller.stop(); + assertThat(Duration.ofNanos(System.nanoTime() - stoppedAt)).isLessThan(Duration.ofSeconds(5)); + assertThat(this.poller.isRunning()).isFalse(); + await().atMost(Duration.ofSeconds(5)).untilAsserted(() -> assertThat(pollerThreads()).isEmpty()); + } + + @Test + void pollNowAsksForAnImmediateAnswerEvenWhenLongPolling() { + transmitter.setLongPollHold(Duration.ofSeconds(30)); + this.poller.setLongPolling(Duration.ofSeconds(30)); + long started = System.nanoTime(); + assertThat(this.poller.pollNow()).isZero(); + assertThat(Duration.ofNanos(System.nanoTime() - started)).isLessThan(Duration.ofSeconds(5)); + assertThat(transmitter.lastPollRequest()).containsEntry("returnImmediately", true); + } + private String queue(String sessionId) { var claims = transmitter.setClaims("CaepSessionRevoked", opaque(sessionId)).build(); transmitter.queueSet(transmitter.signSet(claims)); diff --git a/easyssf-test/src/main/java/org/easyssf/test/TestTransmitter.java b/easyssf-test/src/main/java/org/easyssf/test/TestTransmitter.java index 4f56c18..6e604d8 100644 --- a/easyssf-test/src/main/java/org/easyssf/test/TestTransmitter.java +++ b/easyssf-test/src/main/java/org/easyssf/test/TestTransmitter.java @@ -4,9 +4,9 @@ import java.net.InetAddress; import java.net.InetSocketAddress; import java.nio.charset.StandardCharsets; +import java.time.Duration; import java.time.Instant; import java.util.Base64; -import java.util.Collections; import java.util.Date; import java.util.LinkedHashMap; import java.util.List; @@ -14,7 +14,10 @@ import java.util.UUID; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.CopyOnWriteArrayList; +import java.util.concurrent.Executors; import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.locks.Condition; +import java.util.concurrent.locks.ReentrantLock; import org.easyssf.core.event.SsfEventTypes; @@ -52,7 +55,15 @@ public final class TestTransmitter implements AutoCloseable { private final List> streams = new CopyOnWriteArrayList<>(); - private final Map queuedSets = Collections.synchronizedMap(new LinkedHashMap<>()); + /** + * guarded by queueLock; a lock rather than synchronized, which pins virtual threads + * before JDK 24 + */ + private final Map queuedSets = new LinkedHashMap<>(); + + private final ReentrantLock queueLock = new ReentrantLock(); + + private final Condition setQueued = this.queueLock.newCondition(); private final List acknowledgedSets = new CopyOnWriteArrayList<>(); @@ -66,6 +77,12 @@ public final class TestTransmitter implements AutoCloseable { private final AtomicInteger tokenRequests = new AtomicInteger(); + private final AtomicInteger pollRequests = new AtomicInteger(); + + private volatile Map lastPollRequest = Map.of(); + + private volatile Duration longPollHold = Duration.ZERO; + private volatile boolean available = true; private volatile String validAccessToken = ACCESS_TOKEN; @@ -88,6 +105,10 @@ public TestTransmitter(MetadataLocation metadataLocation) { try { this.key = generateKey(2048, "test-key"); this.server = HttpServer.create(new InetSocketAddress(InetAddress.getLoopbackAddress(), 0), 0); + // one thread per request, so that a held poll request does not block the + // others + this.server.setExecutor( + Executors.newThreadPerTaskExecutor(Thread.ofVirtual().name("test-transmitter-", 0).factory())); this.server.createContext("/", (exchange) -> { try { handle(exchange); @@ -212,29 +233,44 @@ private void handleStreams(HttpExchange exchange, String method, String streamId } @SuppressWarnings("unchecked") - private Map poll(Map request) { - if (request.get("ack") instanceof List acks) { - acks.forEach((jti) -> { - if (this.queuedSets.remove(jti) != null) { - this.acknowledgedSets.add((String) jti); + private Map poll(Map request) throws InterruptedException { + this.pollRequests.incrementAndGet(); + this.lastPollRequest = Map.copyOf(request); + this.queueLock.lock(); + try { + if (request.get("ack") instanceof List acks) { + acks.forEach((jti) -> { + if (this.queuedSets.remove(jti) != null) { + this.acknowledgedSets.add((String) jti); + } + }); + } + if (request.get("setErrs") instanceof Map errors) { + ((Map) errors).forEach((jti, error) -> { + this.queuedSets.remove(jti); + this.reportedErrors.put(jti, error); + }); + } + int maxEvents = ((Number) request.getOrDefault("maxEvents", 10)).intValue(); + // a long poll (returnImmediately false or absent, RFC 8936 section 2.5) is + // held until a + // SET is queued or the hold time elapses + if (!Boolean.TRUE.equals(request.get("returnImmediately")) && maxEvents > 0) { + long remaining = this.longPollHold.toNanos(); + while (this.queuedSets.isEmpty() && remaining > 0) { + remaining = this.setQueued.awaitNanos(remaining); } - }); - } - if (request.get("setErrs") instanceof Map errors) { - ((Map) errors).forEach((jti, error) -> { - this.queuedSets.remove(jti); - this.reportedErrors.put(jti, error); - }); - } - int maxEvents = ((Number) request.getOrDefault("maxEvents", 10)).intValue(); - Map sets = new LinkedHashMap<>(); - synchronized (this.queuedSets) { + } + Map sets = new LinkedHashMap<>(); this.queuedSets.entrySet() .stream() .limit(maxEvents) .forEach((set) -> sets.put(set.getKey(), set.getValue())); + return Map.of("sets", sets, "moreAvailable", this.queuedSets.size() > sets.size()); + } + finally { + this.queueLock.unlock(); } - return Map.of("sets", sets, "moreAvailable", this.queuedSets.size() > sets.size()); } @SuppressWarnings("unchecked") @@ -330,7 +366,7 @@ public List> streams() { */ public void queueSet(String set) { try { - this.queuedSets.put(SignedJWT.parse(set).getJWTClaimsSet().getJWTID(), set); + queueSet(SignedJWT.parse(set).getJWTClaimsSet().getJWTID(), set); } catch (java.text.ParseException ex) { throw new IllegalArgumentException(ex); @@ -338,7 +374,38 @@ public void queueSet(String set) { } public void queueSet(String jti, String set) { - this.queuedSets.put(jti, set); + this.queueLock.lock(); + try { + this.queuedSets.put(jti, set); + // wakes up a held poll request + this.setQueued.signalAll(); + } + finally { + this.queueLock.unlock(); + } + } + + /** + * How long a poll request without {@code returnImmediately: true} is held while no + * SET is queued. Zero by default: the transmitter then answers at once, as if it did + * not support long polling. + */ + public void setLongPollHold(Duration longPollHold) { + this.longPollHold = (longPollHold != null) ? longPollHold : Duration.ZERO; + } + + /** + * @return the number of poll requests received + */ + public int pollRequests() { + return this.pollRequests.get(); + } + + /** + * @return the body of the last poll request, empty if none was received + */ + public Map lastPollRequest() { + return this.lastPollRequest; } public List acknowledgedSets() { @@ -366,7 +433,17 @@ public List> removedSubjects() { */ public void reset() { this.streams.clear(); - this.queuedSets.clear(); + this.queueLock.lock(); + try { + this.queuedSets.clear(); + this.setQueued.signalAll(); + } + finally { + this.queueLock.unlock(); + } + this.pollRequests.set(0); + this.lastPollRequest = Map.of(); + this.longPollHold = Duration.ZERO; this.acknowledgedSets.clear(); this.reportedErrors.clear(); this.verificationRequests.clear(); From eea81f33ad98c91fb1636a2e91c3e0a3bae9222b Mon Sep 17 00:00:00 2001 From: Thomas Darimont Date: Mon, 5 Oct 2026 22:52:26 +0200 Subject: [PATCH 2/3] Test two receivers sharing the JDBC stores Two pollers on one TestTransmitter stream with a shared dedup and acknowledgement store, on H2, PostgreSQL and MySQL: a SET is handled once, and the acknowledgement or error report one instance could not deliver leaves with the next poll of the other. Signed-off-by: Thomas Darimont --- easyssf-receiver-jdbc/AGENTS.md | 3 + easyssf-receiver-jdbc/pom.xml | 6 + .../AbstractJdbcSharedReceiversTests.java | 183 ++++++++++++++++++ .../jdbc/JdbcSharedReceiversTests.java | 19 ++ .../jdbc/MySqlJdbcSharedReceiversTests.java | 51 +++++ .../PostgresJdbcSharedReceiversTests.java | 50 +++++ .../org/easyssf/test/TestTransmitter.java | 13 ++ 7 files changed, 325 insertions(+) create mode 100644 easyssf-receiver-jdbc/src/test/java/org/easyssf/receiver/jdbc/AbstractJdbcSharedReceiversTests.java create mode 100644 easyssf-receiver-jdbc/src/test/java/org/easyssf/receiver/jdbc/JdbcSharedReceiversTests.java create mode 100644 easyssf-receiver-jdbc/src/test/java/org/easyssf/receiver/jdbc/MySqlJdbcSharedReceiversTests.java create mode 100644 easyssf-receiver-jdbc/src/test/java/org/easyssf/receiver/jdbc/PostgresJdbcSharedReceiversTests.java diff --git a/easyssf-receiver-jdbc/AGENTS.md b/easyssf-receiver-jdbc/AGENTS.md index 4b8ae90..2ea46bb 100644 --- a/easyssf-receiver-jdbc/AGENTS.md +++ b/easyssf-receiver-jdbc/AGENTS.md @@ -11,6 +11,9 @@ The JDBC stores of the receiver in plain SQL, over the small `SsfJdbcOperations` are tagged `database`, run in Testcontainers and are excluded by default: `./mvnw -pl easyssf-receiver-jdbc -Pdatabase-tests verify` (Docker). Run them when the SQL or the 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. - 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/pom.xml b/easyssf-receiver-jdbc/pom.xml index b5ad03d..6facec4 100644 --- a/easyssf-receiver-jdbc/pom.xml +++ b/easyssf-receiver-jdbc/pom.xml @@ -51,6 +51,12 @@ test + + + org.easyssf + easyssf-test + test + org.testcontainers testcontainers-postgresql 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 new file mode 100644 index 0000000..4a51731 --- /dev/null +++ b/easyssf-receiver-jdbc/src/test/java/org/easyssf/receiver/jdbc/AbstractJdbcSharedReceiversTests.java @@ -0,0 +1,183 @@ +package org.easyssf.receiver.jdbc; + +import java.net.URI; +import java.nio.charset.StandardCharsets; +import java.util.ArrayList; +import java.util.Base64; +import java.util.List; +import java.util.regex.Matcher; +import java.util.regex.Pattern; + +import org.easyssf.core.event.SsfSubjectIdentifiers; +import org.easyssf.receiver.event.SsfEventHandler; +import org.easyssf.receiver.http.JdkSsfHttpClient; +import org.easyssf.receiver.http.SsfHttpClient; +import org.easyssf.receiver.poll.SsfPoller; +import org.easyssf.receiver.set.NimbusSsfSetVerifier; +import org.easyssf.receiver.set.SsfSetProcessor; +import org.easyssf.test.TestTransmitter; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatIllegalStateException; + +/** + * Several receivers (instances of one application) polling the same stream and sharing + * the JDBC stores: a SET is handled once, and an acknowledgement one instance could not + * deliver leaves with the next poll of another. + */ +abstract class AbstractJdbcSharedReceiversTests { + + /** + * One transmitter per test: a static one would be shared by the subclasses for the + * different databases and closed by the first of them to finish. + */ + private TestTransmitter transmitter; + + private static final String PREFIX = JdbcSsfSchema.DEFAULT_TABLE_PREFIX; + + private JdbcSsfJtiDedupStore dedupStore; + + private JdbcSsfPollAckStore ackStore; + + /** + * Returns operations on an empty database: none of the tables of the tests exist. + */ + protected abstract SsfJdbcOperations emptyDatabase(); + + @BeforeEach + void setUp() { + this.transmitter = new TestTransmitter(); + SsfJdbcOperations jdbc = emptyDatabase(); + JdbcSsfSchema.prepareTable(jdbc, JdbcSsfSchema.processedSetTable(PREFIX), + JdbcSsfSchema.createProcessedSetTable(PREFIX), true, null); + JdbcSsfSchema.prepareTable(jdbc, JdbcSsfSchema.pollAckTable(PREFIX), JdbcSsfSchema.createPollAckTable(PREFIX), + true, null); + this.dedupStore = new JdbcSsfJtiDedupStore(jdbc, PREFIX); + this.ackStore = new JdbcSsfPollAckStore(jdbc, PREFIX); + } + + @AfterEach + void stopTransmitter() { + this.transmitter.close(); + } + + @Test + void setIsHandledOnceAndAcknowledgedByWhicheverInstancePollsNext() { + Receiver a = receiver(); + Receiver b = receiver(); + String jti = queue("session-1"); + // a handles the SET, then loses the transmitter before the acknowledgement leaves + a.afterHandling = () -> this.transmitter.setAvailable(false); + + assertThatIllegalStateException().isThrownBy(a.poller::pollNow); + + assertThat(a.handled).containsExactly(jti); + assertThat(this.transmitter.acknowledgedSets()).isEmpty(); + // the acknowledgement waits in the shared store, visible to both instances + assertThat(a.poller.getPendingAckCount()).isEqualTo(1); + assertThat(b.poller.getPendingAckCount()).isEqualTo(1); + + // b polls: its request carries a's acknowledgement from the shared store, so the + // transmitter + // releases the SET before answering and b does not even see it + this.transmitter.setAvailable(true); + assertThat(b.poller.pollNow()).isZero(); + + assertThat(b.handled).isEmpty(); + assertThat(this.transmitter.acknowledgedSets()).containsExactly(jti); + assertThat(a.poller.getPendingAckCount()).isZero(); + assertThat(b.poller.getPendingAckCount()).isZero(); + assertThat(this.transmitter.queuedSetCount()).isZero(); + } + + @Test + void errorReportOfOneInstanceLeavesWithTheNextPollOfAnother() { + Receiver a = receiver(); + Receiver b = receiver(); + String jti = queue("session-1"); + this.transmitter.queueSet("broken", "not-a-set"); + a.afterHandling = () -> this.transmitter.setAvailable(false); + + assertThatIllegalStateException().isThrownBy(a.poller::pollNow); + assertThat(this.transmitter.reportedErrors()).isEmpty(); + + this.transmitter.setAvailable(true); + b.poller.pollNow(); + + assertThat(this.transmitter.acknowledgedSets()).containsExactly(jti); + assertThat(this.transmitter.reportedErrors()).containsOnlyKeys("broken"); + assertThat(b.handled).isEmpty(); + } + + @Test + void setDeliveredAgainLaterIsNotHandledTwiceByAnyInstance() { + Receiver a = receiver(); + Receiver b = receiver(); + String set = this.transmitter.set("CaepSessionRevoked", SsfSubjectIdentifiers.opaque("session-1")); + String jti = jti(set); + this.transmitter.queueSet(set); + assertThat(a.poller.pollNow()).isEqualTo(1); + assertThat(this.transmitter.acknowledgedSets()).containsExactly(jti); + + // the transmitter delivers the same SET once more, to the other instance + this.transmitter.queueSet(jti, set); + assertThat(b.poller.pollNow()).isEqualTo(1); + + assertThat(a.handled).containsExactly(jti); + assertThat(b.handled).isEmpty(); + assertThat(this.transmitter.acknowledgedSets()).containsExactly(jti, jti); + } + + private Receiver receiver() { + Receiver receiver = new Receiver(); + SsfHttpClient http = new JdkSsfHttpClient(); + NimbusSsfSetVerifier verifier = new NimbusSsfSetVerifier(this.transmitter.issuer(), transmitter::jwksUri, http); + SsfEventHandler handler = (eventContext) -> { + receiver.handled.add(eventContext.eventToken().jti()); + receiver.afterHandling.run(); + }; + SsfSetProcessor processor = new SsfSetProcessor(verifier, this.dedupStore, List.of(handler)); + receiver.poller = new SsfPoller(http, () -> TestTransmitter.ACCESS_TOKEN, + () -> URI.create(this.transmitter.pollUri()), processor); + receiver.poller.setAckStore(this.ackStore); + receiver.poller.setTransmitter(this.transmitter.issuer()); + receiver.poller.setFlushOnStop(false); + return receiver; + } + + private String queue(String sessionId) { + String set = this.transmitter.set("CaepSessionRevoked", SsfSubjectIdentifiers.opaque(sessionId)); + this.transmitter.queueSet(set); + return jti(set); + } + + /** + * The {@code jti} of a SET, read from its payload without a JOSE library: this module + * does not read Nimbus on the module path. + */ + private static String jti(String encodedSet) { + String payload = new String(Base64.getUrlDecoder().decode(encodedSet.split("\\.")[1]), StandardCharsets.UTF_8); + Matcher jti = Pattern.compile("\"jti\"\\s*:\\s*\"([^\"]+)\"").matcher(payload); + assertThat(jti.find()).isTrue(); + return jti.group(1); + } + + /** + * One instance of the application: its poller and what its handler saw. + */ + private static final class Receiver { + + private SsfPoller poller; + + private final List handled = new ArrayList<>(); + + /** runs after a SET was handled, before it is acknowledged */ + private Runnable afterHandling = () -> { + }; + + } + +} diff --git a/easyssf-receiver-jdbc/src/test/java/org/easyssf/receiver/jdbc/JdbcSharedReceiversTests.java b/easyssf-receiver-jdbc/src/test/java/org/easyssf/receiver/jdbc/JdbcSharedReceiversTests.java new file mode 100644 index 0000000..fa3d82d --- /dev/null +++ b/easyssf-receiver-jdbc/src/test/java/org/easyssf/receiver/jdbc/JdbcSharedReceiversTests.java @@ -0,0 +1,19 @@ +package org.easyssf.receiver.jdbc; + +import java.util.UUID; + +import org.h2.jdbcx.JdbcDataSource; + +/** + * Several receivers on one H2 database, a new in-memory database for every test. + */ +class JdbcSharedReceiversTests extends AbstractJdbcSharedReceiversTests { + + @Override + protected SsfJdbcOperations emptyDatabase() { + JdbcDataSource dataSource = new JdbcDataSource(); + dataSource.setURL("jdbc:h2:mem:" + UUID.randomUUID() + ";DB_CLOSE_DELAY=-1"); + return new DataSourceSsfJdbcOperations(dataSource); + } + +} diff --git a/easyssf-receiver-jdbc/src/test/java/org/easyssf/receiver/jdbc/MySqlJdbcSharedReceiversTests.java b/easyssf-receiver-jdbc/src/test/java/org/easyssf/receiver/jdbc/MySqlJdbcSharedReceiversTests.java new file mode 100644 index 0000000..924100c --- /dev/null +++ b/easyssf-receiver-jdbc/src/test/java/org/easyssf/receiver/jdbc/MySqlJdbcSharedReceiversTests.java @@ -0,0 +1,51 @@ +package org.easyssf.receiver.jdbc; + +import java.util.List; + +import org.junit.jupiter.api.AfterAll; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.Tag; +import org.testcontainers.mysql.MySQLContainer; + +import com.mysql.cj.jdbc.MysqlDataSource; + +/** + * Several receivers on one MySQL, in a Testcontainers container shared by the tests of + * the class. Tagged {@code database}: the normal build excludes it, + * {@code -Pdatabase-tests} runs it (Docker required). + */ +@Tag("database") +class MySqlJdbcSharedReceiversTests extends AbstractJdbcSharedReceiversTests { + + private static final String IMAGE = "mysql:8.4"; + + private static final List TABLES = List.of("EASYSSF_PROCESSED_SET", "EASYSSF_REVOCATION", + "APP_SSF_PROCESSED_SET", "OTHER_REVOCATION", "EASYSSF_POLL_ACK"); + + private static MySQLContainer mysql; + + @BeforeAll + static void startDatabase() { + mysql = new MySQLContainer(IMAGE); + mysql.start(); + } + + @AfterAll + static void stopDatabase() { + mysql.stop(); + } + + @Override + protected SsfJdbcOperations emptyDatabase() { + MysqlDataSource dataSource = new MysqlDataSource(); + dataSource.setUrl(mysql.getJdbcUrl()); + dataSource.setUser(mysql.getUsername()); + dataSource.setPassword(mysql.getPassword()); + SsfJdbcOperations jdbc = new DataSourceSsfJdbcOperations(dataSource); + for (String table : TABLES) { + jdbc.execute("DROP TABLE IF EXISTS " + table); + } + return jdbc; + } + +} diff --git a/easyssf-receiver-jdbc/src/test/java/org/easyssf/receiver/jdbc/PostgresJdbcSharedReceiversTests.java b/easyssf-receiver-jdbc/src/test/java/org/easyssf/receiver/jdbc/PostgresJdbcSharedReceiversTests.java new file mode 100644 index 0000000..775b694 --- /dev/null +++ b/easyssf-receiver-jdbc/src/test/java/org/easyssf/receiver/jdbc/PostgresJdbcSharedReceiversTests.java @@ -0,0 +1,50 @@ +package org.easyssf.receiver.jdbc; + +import java.util.List; + +import org.junit.jupiter.api.AfterAll; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.Tag; +import org.postgresql.ds.PGSimpleDataSource; +import org.testcontainers.postgresql.PostgreSQLContainer; + +/** + * Several receivers on one PostgreSQL, in a Testcontainers container shared by the tests + * of the class. Tagged {@code database}: the normal build excludes it, + * {@code -Pdatabase-tests} runs it (Docker required). + */ +@Tag("database") +class PostgresJdbcSharedReceiversTests extends AbstractJdbcSharedReceiversTests { + + private static final String IMAGE = "postgres:18.4"; + + private static final List TABLES = List.of("EASYSSF_PROCESSED_SET", "EASYSSF_REVOCATION", + "APP_SSF_PROCESSED_SET", "OTHER_REVOCATION", "EASYSSF_POLL_ACK"); + + private static PostgreSQLContainer postgres; + + @BeforeAll + static void startDatabase() { + postgres = new PostgreSQLContainer(IMAGE); + postgres.start(); + } + + @AfterAll + static void stopDatabase() { + postgres.stop(); + } + + @Override + protected SsfJdbcOperations emptyDatabase() { + PGSimpleDataSource dataSource = new PGSimpleDataSource(); + dataSource.setUrl(postgres.getJdbcUrl()); + dataSource.setUser(postgres.getUsername()); + dataSource.setPassword(postgres.getPassword()); + SsfJdbcOperations jdbc = new DataSourceSsfJdbcOperations(dataSource); + for (String table : TABLES) { + jdbc.execute("DROP TABLE IF EXISTS " + table); + } + return jdbc; + } + +} diff --git a/easyssf-test/src/main/java/org/easyssf/test/TestTransmitter.java b/easyssf-test/src/main/java/org/easyssf/test/TestTransmitter.java index 6e604d8..a095049 100644 --- a/easyssf-test/src/main/java/org/easyssf/test/TestTransmitter.java +++ b/easyssf-test/src/main/java/org/easyssf/test/TestTransmitter.java @@ -394,6 +394,19 @@ public void setLongPollHold(Duration longPollHold) { this.longPollHold = (longPollHold != null) ? longPollHold : Duration.ZERO; } + /** + * @return the number of SETs queued and not yet acknowledged + */ + public int queuedSetCount() { + this.queueLock.lock(); + try { + return this.queuedSets.size(); + } + finally { + this.queueLock.unlock(); + } + } + /** * @return the number of poll requests received */ From d1037cba963faad778d76e55644b5cc1d2bdeca0 Mon Sep 17 00:00:00 2001 From: Thomas Darimont Date: Mon, 5 Oct 2026 23:05:08 +0200 Subject: [PATCH 3/3] Count the pending poll acknowledgements in a gauge easyssf.receiver.poll.pending-acks, tagged with the transmitter, registered when a poller starts through the new SsfReceiverMetrics.pollerStarted; the health details already showed the count. Signed-off-by: Thomas Darimont --- CHANGELOG.md | 3 ++- easyssf-receiver-spring-boot-starter/README.md | 3 ++- .../metrics/MicrometerSsfReceiverMetrics.java | 11 +++++++++++ .../receiver/metrics/SsfReceiverMetrics.java | 10 ++++++++++ .../org/easyssf/receiver/poll/SsfPoller.java | 1 + .../easyssf/receiver/poll/SsfPollerTests.java | 17 +++++++++++++++++ 6 files changed, 43 insertions(+), 2 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 4571a87..17c96c9 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -13,7 +13,8 @@ The notable changes of every release. The section of a version is the text of it `JdbcSsfPollAckStore` (table `EASYSSF_POLL_ACK`, auto-configured with the other JDBC stores, `easyssf.receiver.jdbc.ack-retention`), so that a SET handled right before a restart is acknowledged afterwards instead of being delivered again. `SsfPoller.stop()` sends the pending - acknowledgements with a last request; the health details show them as `pendingAcks`. + acknowledgements with a last request; the health details show them as `pendingAcks` and the + gauge `easyssf.receiver.poll.pending-acks` counts them. - Long polling (RFC 8936, section 2.5): `SsfPoller.setLongPolling(hold)` keeps one request outstanding that the transmitter holds until SETs are available; `easyssf.receiver.poll.long-polling` and `poll.long-polling-hold` in the Spring Boot starter, off by default. `SsfHttpRequest` carries an diff --git a/easyssf-receiver-spring-boot-starter/README.md b/easyssf-receiver-spring-boot-starter/README.md index d8817a5..69c8ea9 100644 --- a/easyssf-receiver-spring-boot-starter/README.md +++ b/easyssf-receiver-spring-boot-starter/README.md @@ -195,7 +195,7 @@ easyssf: are kept in the table `EASYSSF_POLL_ACK` (see [Keeping state in the database](#keeping-state-in-the-database)), so that a SET handled right before a restart is acknowledged afterwards instead of being delivered again and skipped as a duplicate. `stop()` sends the pending acknowledgements with a last request. - The health details show them as `pendingAcks`. + The health details show them as `pendingAcks`, the gauge `easyssf.receiver.poll.pending-acks` too. - **Long polling**: by default every request asks the transmitter to answer at once (`returnImmediately: true`) and the transmitter is polled every `poll.interval`. With `easyssf.receiver.poll.long-polling=true` the poller keeps one request outstanding instead: the @@ -380,6 +380,7 @@ If the application has a Micrometer `MeterRegistry` (for example with | `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.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) | Switch it off with `easyssf.receiver.metrics.enabled=false`, or provide your own `SsfReceiverMetrics` bean. diff --git a/easyssf-receiver/src/main/java/org/easyssf/receiver/metrics/MicrometerSsfReceiverMetrics.java b/easyssf-receiver/src/main/java/org/easyssf/receiver/metrics/MicrometerSsfReceiverMetrics.java index a17e16d..7bfcea9 100644 --- a/easyssf-receiver/src/main/java/org/easyssf/receiver/metrics/MicrometerSsfReceiverMetrics.java +++ b/easyssf-receiver/src/main/java/org/easyssf/receiver/metrics/MicrometerSsfReceiverMetrics.java @@ -3,12 +3,14 @@ import java.time.Duration; import java.util.Locale; import java.util.function.Function; +import java.util.function.IntSupplier; import org.easyssf.core.SsfDeliveryMethod; import org.easyssf.core.event.SsfEventTypes; import org.easyssf.core.support.SsfAssert; import io.micrometer.core.instrument.Counter; +import io.micrometer.core.instrument.Gauge; import io.micrometer.core.instrument.MeterRegistry; import io.micrometer.core.instrument.Timer; @@ -73,6 +75,15 @@ public void pollCompleted(Duration duration, boolean success) { pollCompleted(null, duration, success); } + @Override + public void pollerStarted(String transmitter, IntSupplier pendingAcks) { + Gauge.builder("easyssf.receiver.poll.pending-acks", pendingAcks, IntSupplier::getAsInt) + .description( + "Acknowledgements and error reports a poller owes its transmitter, waiting for the next poll request") + .tag("transmitter", transmitter(transmitter)) + .register(this.meterRegistry); + } + @Override public void pollCompleted(String transmitter, Duration duration, boolean success) { Timer.builder("easyssf.receiver.poll") 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 dde2332..ed544ef 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 @@ -1,6 +1,7 @@ package org.easyssf.receiver.metrics; import java.time.Duration; +import java.util.function.IntSupplier; import org.easyssf.core.SsfDeliveryMethod; @@ -101,4 +102,13 @@ default void pollCompleted(String transmitter, Duration duration, boolean succes pollCompleted(duration, success); } + /** + * A poller started and tells how to read the number of acknowledgements and error + * reports it still owes its transmitter, for a gauge. + * @param transmitter the issuer of the transmitter, {@code null} if unknown + * @param pendingAcks reads the current number + */ + default void pollerStarted(String transmitter, IntSupplier pendingAcks) { + } + } 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 4b201c3..42f2f50 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 @@ -260,6 +260,7 @@ public void start() { return; } this.running = true; + this.metrics.pollerStarted(this.transmitter, this::getPendingAckCount); Thread thread = this.threadFactory.newThread(this::run); SsfAssert.notNull(thread, "the thread factory returned no thread"); this.thread = thread; diff --git a/easyssf-receiver/src/test/java/org/easyssf/receiver/poll/SsfPollerTests.java b/easyssf-receiver/src/test/java/org/easyssf/receiver/poll/SsfPollerTests.java index d03d88f..ff857e4 100644 --- a/easyssf-receiver/src/test/java/org/easyssf/receiver/poll/SsfPollerTests.java +++ b/easyssf-receiver/src/test/java/org/easyssf/receiver/poll/SsfPollerTests.java @@ -257,6 +257,23 @@ void acknowledgementsAreSentInBatches() { assertThat(this.poller.getPendingAckCount()).isZero(); } + @Test + void pendingAcknowledgementsAreAGauge() { + this.poller.setTransmitter(transmitter.issuer()); + this.poller.setInitialDelay(Duration.ofHours(1)); + this.poller.start(); + this.poller.stop(); + assertThat(this.meterRegistry.get("easyssf.receiver.poll.pending-acks") + .tag("transmitter", transmitter.issuer()) + .gauge() + .value()).isZero(); + String jti = queue("session-1"); + this.afterHandling = () -> transmitter.setAvailable(false); + assertThatIllegalStateException().isThrownBy(this.poller::pollNow); + assertThat(this.handled).containsExactly(jti); + assertThat(this.meterRegistry.get("easyssf.receiver.poll.pending-acks").gauge().value()).isEqualTo(1); + } + @Test void stopFlushesThePendingAcknowledgements() { this.poller.setInitialDelay(Duration.ofHours(1));