thomasdarimont · Oct 19, 2026 · Oct 19, 2026 · Oct 19, 2026 · Oct 20, 2026 · Oct 20, 2026
@@ -39,3 +39,11 @@ Configure the accepted credential types (`vct`) in the new *Default Credential T
 A DCQL query can still be configured for requests that mappers cannot express, such as claim sets.
 It takes precedence over the inferred query and is now validated when the identity provider is saved.
 Only a single credential in the `dc+sd-jwt` format can be requested.
+
+= Transactional outbox
+
+== Outbox shared by multiple consumers
+
+The generic transactional outbox that backs the push delivery of the experimental SSF transmitter can now be shared by several consumers, such as an outbound SCIM provisioning extension.
+Every consumer's drainer holds its own cluster lock, deliveries run outside of database transactions under a per-entry lease, different destinations are delivered in parallel with a per-destination batch cap, and a generic Micrometer listener publishes `keycloak.outbox.*` meters.
+The `OUTBOX_ENTRY` timestamps are converted to epoch milliseconds; see the upgrading guide.
@@ -371,6 +371,16 @@ Review and update their template overrides and CSS to match the new structure.
 // ------------------------ Notable changes ------------------------ //
 == Notable changes
 
+=== Outbox entry timestamps stored as epoch milliseconds
+// https://github.com/keycloak/keycloak/issues/54137
+
+The `OUTBOX_ENTRY` table, which backs the push delivery of the experimental SSF transmitter, now stores its `CREATED_AT`, `NEXT_ATTEMPT_AT` and `DELIVERED_AT` columns as `BIGINT` epoch milliseconds, like every other {project_name} table, instead of `TIMESTAMP`.
+`TIMESTAMP` has whole-second precision on MySQL and MariaDB, which made the delivery order of entries created within the same second undefined.
+Two nullable columns, `CLAIM_TOKEN` and `CLAIMED_UNTIL`, are added for the lease a drainer holds on an entry while delivering it.
+
+The migration converts existing rows and makes every queued entry due immediately, so all pending push deliveries are attempted on the first drainer tick after the upgrade.
+The table exists in every installation since {project_name} 26.7, so the migration runs for everyone; on an empty table it completes instantly.
+
 Notable changes may include internal behavior changes that prevent common misconfigurations, bugs that are fixed, or changes to simplify running {project_name}.
 It also lists significant changes to internal APIs.
 

@@ -0,0 +1,88 @@
+package org.keycloak.connections.jpa.updater.liquibase.custom;
+
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.sql.Timestamp;
+import java.util.Calendar;
+import java.util.TimeZone;
+
+import liquibase.exception.CustomChangeException;
+import liquibase.snapshot.SnapshotGeneratorFactory;
+import liquibase.statement.core.RawParameterizedSqlStatement;
+import liquibase.structure.core.Table;
+
+/**
+ * Back-fills the BIGINT epoch-millisecond columns of {@code OUTBOX_ENTRY}
+ * from the TIMESTAMP columns they replace.
+ *
+ * <p>The old columns were written by Hibernate's UTC timestamp binding,
+ * so they are read back with a UTC calendar. Deployments whose database
+ * session time zone differed when the rows were written may see a
+ * constant offset; to make that harmless, the next attempt of every
+ * queued row is clamped to "now" so no row ends up scheduled hours into
+ * the future. The outbox is at-least-once, so an early retry is safe.
+ *
+ * <p>Skipped when {@code OUTBOX_ENTRY} does not exist in the live
+ * database: with the manual migration strategy the SQL of a pre-26.7
+ * database is only exported, so the table that the 26.7.0 changeset
+ * creates is not there to read from (and has no rows to convert).
+ */
+public class JpaUpdate27_0_0_OutboxTimestampsToEpochMillis extends CustomKeycloakTask {
+
+    private static final Calendar UTC = Calendar.getInstance(TimeZone.getTimeZone("UTC"));
+
+    @Override
+    protected boolean isApplicable() throws CustomChangeException {
+        return super.isApplicable() && tableExists("OUTBOX_ENTRY");
+    }
+
+    private boolean tableExists(String tableName) throws CustomChangeException {
+        try {
+            String correctedTableName = database.correctObjectName(tableName, Table.class);
+            return SnapshotGeneratorFactory.getInstance().has(new Table().setName(correctedTableName), database);
+        } catch (Exception e) {
+            throw new CustomChangeException(getTaskId() + ": failed to check for existence of " + tableName, e);
+        }
+    }
+
+    @Override
+    protected void generateStatementsImpl() throws CustomChangeException {
+        String table = getTableName("OUTBOX_ENTRY");
+        long now = System.currentTimeMillis();
+        String select = "SELECT ID, STATUS, CREATED_AT, NEXT_ATTEMPT_AT, DELIVERED_AT FROM " + table;
+        String update = "UPDATE " + table
+                + " SET CREATED_AT_MS = ?, NEXT_ATTEMPT_AT_MS = ?, DELIVERED_AT_MS = ? WHERE ID = ?";
+        int rows = 0;
+        try (PreparedStatement ps = connection.prepareStatement(select);
+             ResultSet rs = ps.executeQuery()) {
+            while (rs.next()) {
+                String id = rs.getString(1);
+                String status = rs.getString(2);
+                Long createdAt = millis(rs.getTimestamp(3, UTC));
+                    Long nextAttemptAt = millis(rs.getTimestamp(4, UTC));
+                Long deliveredAt = millis(rs.getTimestamp(5, UTC));
+                if (createdAt == null) {
+                    createdAt = now;
+                }
+                boolean queued = "PENDING".equals(status) || "HELD".equals(status);
+                if (nextAttemptAt == null || (queued && nextAttemptAt > now)) {
+                    nextAttemptAt = now;
+                }
+                statements.add(new RawParameterizedSqlStatement(update, createdAt, nextAttemptAt, deliveredAt, id));
+                rows++;
+            }
+        } catch (Exception e) {
+            throw new CustomChangeException(getTaskId() + ": failed to read OUTBOX_ENTRY timestamps", e);
+        }
+        confirmationMessage.append("Converted timestamps of ").append(rows).append(" OUTBOX_ENTRY row(s) to epoch milliseconds");
+    }
+
+    private static Long millis(Timestamp ts) {
+        return ts == null ? null : ts.getTime();
+    }
+
+    @Override
+    protected String getTaskId() {
+        return "Convert OUTBOX_ENTRY timestamps to epoch milliseconds";
+    }
+}
@@ -0,0 +1,86 @@
+package org.keycloak.events.outbox;
+
+import java.time.Instant;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.function.Consumer;
+
+import org.keycloak.models.KeycloakSession;
+
+import org.jboss.logging.Logger;
+
+/**
+ * Fan-out {@link OutboxDrainerListener}; see
+ * {@link OutboxDrainerListener#composite}.
+ */
+final class CompositeOutboxDrainerListener implements OutboxDrainerListener {
+
+    private static final Logger log = Logger.getLogger(CompositeOutboxDrainerListener.class);
+
+    private final List<OutboxDrainerListener> delegates;
+
+    private CompositeOutboxDrainerListener(List<OutboxDrainerListener> delegates) {
+        this.delegates = List.copyOf(delegates);
+    }
+
+    static OutboxDrainerListener of(OutboxDrainerListener... listeners) {
+        List<OutboxDrainerListener> kept = new ArrayList<>();
+        if (listeners != null) {
+            for (OutboxDrainerListener l : listeners) {
+                if (l == null || l == NOOP) {
+                    continue;
+                }
+                if (l instanceof CompositeOutboxDrainerListener c) {
+                    kept.addAll(c.delegates);
+                } else {
+                    kept.add(l);
+                }
+            }
+        }
+        return switch (kept.size()) {
+            case 0 -> NOOP;
+            case 1 -> kept.get(0);
+            default -> new CompositeOutboxDrainerListener(kept);
+        };
+    }
+
+    private void each(String callback, Consumer<OutboxDrainerListener> call) {
+        for (OutboxDrainerListener delegate : delegates) {
+            try {
+                call.accept(delegate);
+            } catch (RuntimeException e) {
+                log.warnf(e, "Outbox drainer listener %s threw in %s — skipping it", delegate.getClass().getName(), callback);
+            }
+        }
+    }
+
+    @Override
+    public void onTickStart(KeycloakSession session, String entryKind) {
+        each("onTickStart", l -> l.onTickStart(session, entryKind));
+    }
+
+    @Override
+    public void onTickEnd(KeycloakSession session, OutboxDrainerTickSummary summary) {
+        each("onTickEnd", l -> l.onTickEnd(session, summary));
+    }
+
+    @Override
+    public void onDelivered(KeycloakSession session, OutboxEntry row) {
+        each("onDelivered", l -> l.onDelivered(session, row));
+    }
+
+    @Override
+    public void onRetryScheduled(KeycloakSession session, OutboxEntry row, Instant nextAttemptAt, String reason) {
+        each("onRetryScheduled", l -> l.onRetryScheduled(session, row, nextAttemptAt, reason));
+    }
+
+    @Override
+    public void onDeferred(KeycloakSession session, OutboxEntry row, Instant notBefore, String reason) {
+        each("onDeferred", l -> l.onDeferred(session, row, notBefore, reason));
+    }
+
+    @Override
+    public void onDeadLetter(KeycloakSession session, OutboxEntry row, DeadLetterCause cause, String reason) {
+        each("onDeadLetter", l -> l.onDeadLetter(session, row, cause, reason));
+    }
+}
@@ -39,16 +39,98 @@
  * legitimate backoff aren't prematurely promoted, and shorter than
  * {@code deadLetterRetention} so promoted rows retain a meaningful
  * forensic window before the dead-letter purge deletes them.
+ *
+ * <p>{@code claimLease} is how long a tick may hold a claimed row
+ * before other ticks treat it as abandoned and re-claim it; it must
+ * cover the tick budget plus the slowest single delivery. A
+ * delivery that outlives its lease is still recorded if no other
+ * tick re-claimed the row in the meantime, but may be duplicated
+ * otherwise (at-least-once). {@code tickBudget} bounds how long one
+ * tick keeps delivering; claimed rows it does not get to are handed
+ * back for the next tick. {@code null} means unbounded.
+ *
+ * <p>{@code perOwnerBatchSize} switches the claim to owner-fair mode:
+ * instead of the oldest {@code batchSize} due rows regardless of
+ * owner, the tick visits the owners with due rows oldest-first and
+ * takes at most this many rows from each per round, going round
+ * again until the batch is full or nothing is due. One slow or
+ * flooding destination then gets its share of a batch rather than
+ * all of it. {@code null} keeps the plain oldest-first claim.
+ *
+ * <p>{@code deliveryConcurrency} is the number of delivery workers a
+ * tick may use. Rows are grouped by owner; one owner's rows are always
+ * delivered in order by one worker, different owners in parallel.
+ * {@code null} or 1 delivers on the tick thread. Workers come from the
+ * {@code outbox-delivery-<kind>} executor of the {@code ExecutorsProvider}.
  */
 public record OutboxConfig(
         String entryKind,
         int batchSize,
         OutboxBackoff backoff,
         Duration deadLetterRetention,
         Duration deliveredRetention,
-        Duration pendingMaxAge) {
+        Duration pendingMaxAge,
+        Duration claimLease,
+        Duration tickBudget,
+        Integer perOwnerBatchSize,
+        Integer deliveryConcurrency) {
+
+    public static final Duration DEFAULT_CLAIM_LEASE = Duration.ofMinutes(5);
+
+    /** Default lease, unbounded tick, plain oldest-first claim. */
+    public OutboxConfig(String entryKind,
+                        int batchSize,
+                        OutboxBackoff backoff,
+                        Duration deadLetterRetention,
+                        Duration deliveredRetention,
+                        Duration pendingMaxAge) {
+        this(entryKind, batchSize, backoff, deadLetterRetention, deliveredRetention, pendingMaxAge,
+                DEFAULT_CLAIM_LEASE, null, null, null);
+    }
+
+    /** Plain oldest-first claim. */
+    public OutboxConfig(String entryKind,
+                        int batchSize,
+                        OutboxBackoff backoff,
+                        Duration deadLetterRetention,
+                        Duration deliveredRetention,
+                        Duration pendingMaxAge,
+                        Duration claimLease,
+                        Duration tickBudget) {
+        this(entryKind, batchSize, backoff, deadLetterRetention, deliveredRetention, pendingMaxAge,
+                claimLease, tickBudget, null, null);
+    }
+
+    /** Serial delivery on the tick thread. */
+    public OutboxConfig(String entryKind,
+                        int batchSize,
+                        OutboxBackoff backoff,
+                        Duration deadLetterRetention,
+                        Duration deliveredRetention,
+                        Duration pendingMaxAge,
+                        Duration claimLease,
+                        Duration tickBudget,
+                        Integer perOwnerBatchSize) {
+        this(entryKind, batchSize, backoff, deadLetterRetention, deliveredRetention, pendingMaxAge,
+                claimLease, tickBudget, perOwnerBatchSize, null);
+    }
 
     public OutboxConfig {
+        if (perOwnerBatchSize != null && perOwnerBatchSize <= 0) {
+            throw new IllegalArgumentException("perOwnerBatchSize must be positive or null, got " + perOwnerBatchSize);
+        }
+        if (deliveryConcurrency != null && deliveryConcurrency <= 0) {
+            throw new IllegalArgumentException("deliveryConcurrency must be positive or null, got " + deliveryConcurrency);
+        }
+        if (claimLease == null || claimLease.isZero() || claimLease.isNegative()) {
+            throw new IllegalArgumentException("claimLease must be positive, got " + claimLease);
+        }
+        if (tickBudget != null && (tickBudget.isZero() || tickBudget.isNegative())) {
+            throw new IllegalArgumentException("tickBudget must be positive or null, got " + tickBudget);
+        }
+        if (tickBudget != null && tickBudget.compareTo(claimLease) >= 0) {
+            throw new IllegalArgumentException("tickBudget " + tickBudget + " must be shorter than claimLease " + claimLease);
+        }
         if (entryKind == null || entryKind.isBlank()) {
             throw new IllegalArgumentException("entryKind must not be blank");
         }

@@ -0,0 +1,30 @@
+package org.keycloak.events.outbox;
+
+import java.util.Objects;
+
+/**
+ * Phase two of a delivery, produced by
+ * {@link OutboxDeliveryHandler#prepare(org.keycloak.models.KeycloakSession, OutboxEntry)}
+ * and executed by the drainer <em>outside any transaction</em>, on a
+ * thread with no bound {@code KeycloakSession}. Everything the
+ * delivery needs (destination URL, credentials, HTTP client, payload)
+ * must have been captured during {@code prepare}.
+ *
+ * <p>Return {@link #settled(OutboxDeliveryResult)} from {@code prepare}
+ * when the outcome is already known there (destination gone, paused,
+ * backed off) and no call has to be made.
+ */
+@FunctionalInterface
+public interface OutboxDelivery {
+
+    /**
+     * Performs the delivery. A thrown {@link RuntimeException} is
+     * treated as {@link OutboxDeliveryResult#retry(String)}.
+     */
+    OutboxDeliveryResult execute();
+
+    static OutboxDelivery settled(OutboxDeliveryResult result) {
+        Objects.requireNonNull(result, "result");
+        return () -> result;
+    }
+}