Keep poll acknowledgements in a store and support long polling

.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

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,

CHANGELOG.md

@@ -6,6 +6,31 @@ 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` 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
+  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

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.

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
 

easyssf-receiver-jdbc/AGENTS.md

@@ -3,12 +3,17 @@
 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.
+  `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.

easyssf-receiver-jdbc/pom.xml

@@ -51,6 +51,12 @@
             <scope>test</scope>
         </dependency>
         <!-- the database tests, see the database-tests profile -->
+        <!-- TestTransmitter, for the tests of several receivers on one database -->
+        <dependency>
+            <groupId>org.easyssf</groupId>
+            <artifactId>easyssf-test</artifactId>
+            <scope>test</scope>
+        </dependency>
         <dependency>
             <groupId>org.testcontainers</groupId>
             <artifactId>testcontainers-postgresql</artifactId>
@@ -61,6 +67,16 @@
             <artifactId>postgresql</artifactId>
             <scope>test</scope>
         </dependency>
+        <dependency>
+            <groupId>org.testcontainers</groupId>
+            <artifactId>testcontainers-mysql</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>com.mysql</groupId>
+            <artifactId>mysql-connector-j</artifactId>
+            <scope>test</scope>
+        </dependency>
     </dependencies>
 
     <build>

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 <T> List<T> query(String sql, SsfJdbcRowMapper<T> mapper, Object... args) {
+        return query(sql, 0, mapper, args);
+    }
+
+    @Override
+    public <T> List<T> query(String sql, int maxRows, SsfJdbcRowMapper<T> 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<T> rows = new ArrayList<>();

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.
+ *
+ * <p>
+ * 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<SsfPendingAck> 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<String> jtis) {
+        List<String> remaining = List.copyOf(jtis);
+        for (int from = 0; from < remaining.size(); from += this.deleteBatchSize) {
+            List<String> 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();
+        }
+    }
+
+}

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<String> 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

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

@@ -29,6 +29,24 @@ public interface SsfJdbcOperations {
      */
     <T> List<T> query(String sql, SsfJdbcRowMapper<T> 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 <T> List<T> query(String sql, int maxRows, SsfJdbcRowMapper<T> mapper, Object... args) {
+        List<T> 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