Multi-tenant aware DB sequence id generation (replaces #2305) (#3805)

## Problem

SequenceIdGenerator captured a single DataSource at deploy time and held one shared pre-fetch buffer. Under multi-tenancy this is wrong:

 - TenantMode.DB/DB_WITH_MASTER — there is no bootstrap DataSource, so BeanDescriptorManager passed null; sequence allocation couldn't resolve the current tenant's database.
 - TenantMode.SCHEMA/CATALOG — a single shared buffer let one tenant's pre-fetched ids be handed to another (cross-tenant bleed), and pre-fetch used a connection that wasn't scoped to the requesting tenant.

(Supersedes the per-datasource delegator approach in #2305, which leaked via a WeakHashMap whose values strongly referenced the keys, only handled DB mode, and re-resolved the tenant on the background thread.)

## Fix:

Make SequenceIdGenerator itself tenant aware, keeping all platform modules untouched.

 - New TenantConnectionSource (ebean-api, additive): optional interface a DataSource may implement — currentTenantId() + connectionForTenant(tenantId).
 - SequenceIdGenerator: the shared idList/lock/loading flag become a per-tenant TenantBuffer keyed by tenantId in a ConcurrentHashMap. Connections are obtained per tenant. Background pre-fetch captures the tenant at submit time (the executor thread has no tenant in scope) and fetches by explicit tenantId — fixing a latent ThreadLocal-propagation bug.
 - SequenceDataSource (ebean-core): adapts DataSourceSupplier to TenantConnectionSource; routes to the tenant DB (DB mode) or sets schema/catalog (SCHEMA/CATALOG).
 - Wiring: InternalConfiguration exposes the DataSourceSupplier; BeanDescriptorManager wraps it only for dynamic-datasource tenant modes.

## Performance (single-tenant unaffected)

 - A cached single buffer field short-circuits the ConcurrentHashMap for the non-tenant key.
 - NONE/PARTITION pass the plain DataSource (not wrapped), so tenantSource == null and the hot path is just a couple of cheap branches — equivalent to the original.

## Compatibility

 - Platform constructor signature (be, ds, seqName, allocationSize) unchanged — no changes to the 9 platform modules.
 - One protected-method signature changed: getMoreIds(int) → getMoreIds(Object tenantKey, int). Rarely overridden (subclasses override getSql/readIds), but a source-incompat for any external custom platform that did.

## Tests

 - TenantSequenceTest — DB-per-tenant: tenant 1 → 1,2,3; tenant 2 independently → 1.
 - SequenceBatchIdGeneratorTest adapted to the per-tenant buffer.
 - Existing sequence + multitenancy suites pass.

## Potential Follow-ups (not in this PR)

 - Add a SCHEMA-mode test.
 - Optional removeTenant(tenantId) hook if unbounded tenant churn is a concern (buffers hold only Longs + a lock, no DataSource, so no real leak).
 - SimpleSequenceIdGenerator (non-batching) left as-is — already uses the txn connection.

Co-authored-by: robin.bygrave <robin.bygrave@eroad.com>

ebean-api/src/main/java/io/ebean/config/dbplatform/SequenceIdGenerator.java

@@ -15,6 +15,8 @@ import java.util.Collections;
 import java.util.List;
 import java.util.NavigableSet;
 import java.util.TreeSet;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.locks.ReentrantLock;
 
@@ -23,18 +25,36 @@ import static java.lang.System.Logger.Level.ERROR;
 
 /**
  * Database sequence based IdGenerator.
+ * <p>
+ * Maintains a separate buffer of pre-fetched id values per tenant when the supplied
+ * DataSource implements {@link TenantConnectionSource}. For the common single-tenant
+ * case a single buffer is used (keyed by {@link #SINGLE}).
  */
 public abstract class SequenceIdGenerator implements PlatformIdGenerator {
 
   protected static final System.Logger log = AppLog.getLogger("io.ebean.SEQ");
 
-  private final ReentrantLock lock = new ReentrantLock();
+  /**
+   * Buffer key used when there is no current tenant (single-tenant or no tenant in scope).
+   */
+  private static final Object SINGLE = new Object();
+
   protected final String seqName;
   protected final DataSource dataSource;
   protected final BackgroundExecutor backgroundExecutor;
-  protected final NavigableSet<Long> idList = new TreeSet<>();
   protected final int allocationSize;
-  protected AtomicBoolean currentlyBackgroundLoading = new AtomicBoolean(false);
+  private final TenantConnectionSource tenantSource;
+  private final TenantBuffer single = new TenantBuffer();
+  private final ConcurrentMap<Object, TenantBuffer> buffers = new ConcurrentHashMap<>();
+
+  /**
+   * Per-tenant pre-fetched id buffer with its own lock and background-loading flag.
+   */
+  private static final class TenantBuffer {
+    final ReentrantLock lock = new ReentrantLock();
+    final NavigableSet<Long> idList = new TreeSet<>();
+    final AtomicBoolean currentlyBackgroundLoading = new AtomicBoolean(false);
+  }
 
   /**
    * Construct given a dataSource and sql to return the next sequence value.
@@ -44,6 +64,7 @@ public abstract class SequenceIdGenerator implements PlatformIdGenerator {
     this.dataSource = ds;
     this.seqName = seqName;
     this.allocationSize = allocationSize;
+    this.tenantSource = (ds instanceof TenantConnectionSource) ? (TenantConnectionSource) ds : null;
   }
 
   public abstract String getSql(int batchSize);
@@ -64,6 +85,24 @@ public abstract class SequenceIdGenerator implements PlatformIdGenerator {
     return true;
   }
 
+  private Object currentTenantKey() {
+    if (tenantSource != null) {
+      Object tenantId = tenantSource.currentTenantId();
+      if (tenantId != null) {
+        return tenantId;
+      }
+    }
+    return SINGLE;
+  }
+
+  private TenantBuffer buffer(Object tenantKey) {
+    if (tenantKey == SINGLE) {
+      // common single-tenant path - avoid the concurrent map lookup
+      return single;
+    }
+    return buffers.computeIfAbsent(tenantKey, k -> new TenantBuffer());
+  }
+
   /**
    * If allocateSize is large load some sequences in a background thread.
    * <p>
@@ -78,23 +117,22 @@ public abstract class SequenceIdGenerator implements PlatformIdGenerator {
 
   /**
    * Return the next Id.
-   * <p>
-   * If a Transaction has been passed in use the Connection from it.
-   * </p>
    */
   @Override
   public Object nextId(Transaction t) {
-    lock.lock();
+    Object tenantKey = currentTenantKey();
+    TenantBuffer buffer = buffer(tenantKey);
+    buffer.lock.lock();
     try {
-      int size = idList.size();
+      int size = buffer.idList.size();
       if (size > 0) {
         maybeLoadMoreInBackground(size);
       } else {
-        loadMore(allocationSize);
+        loadMore(tenantKey, buffer, allocationSize);
       }
-      return idList.pollFirst();
+      return buffer.idList.pollFirst();
     } finally {
-      lock.unlock();
+      buffer.lock.unlock();
     }
   }
 
@@ -106,29 +144,36 @@ public abstract class SequenceIdGenerator implements PlatformIdGenerator {
     }
   }
 
-  private void loadMore(int requestSize) {
-    List<Long> newIds = getMoreIds(requestSize);
-    lock.lock();
+  private void loadMore(Object tenantKey, TenantBuffer buffer, int requestSize) {
+    List<Long> newIds = getMoreIds(tenantKey, requestSize);
+    buffer.lock.lock();
     try {
-      idList.addAll(newIds);
+      buffer.idList.addAll(newIds);
     } finally {
-      lock.unlock();
+      buffer.lock.unlock();
     }
   }
 
   /**
    * Load another batch of Id's using a background thread.
+   * <p>
+   * The tenant is captured here (submit time) as the current tenant is not in scope
+   * on the background executor thread.
    */
   protected void loadInBackground(final int requestSize) {
-    if (currentlyBackgroundLoading.get()) {
+    final Object tenantKey = currentTenantKey();
+    final TenantBuffer buffer = buffer(tenantKey);
+    if (!buffer.currentlyBackgroundLoading.compareAndSet(false, true)) {
       // skip as already background loading
       log.log(DEBUG, "... skip background sequence load (another load in progress)");
       return;
     }
-    currentlyBackgroundLoading.set(true);
     backgroundExecutor.execute(() -> {
-      loadMore(requestSize);
-      currentlyBackgroundLoading.set(false);
+      try {
+        loadMore(tenantKey, buffer, requestSize);
+      } finally {
+        buffer.currentlyBackgroundLoading.set(false);
+      }
     });
   }
 
@@ -140,7 +185,7 @@ public abstract class SequenceIdGenerator implements PlatformIdGenerator {
   /**
    * Get more Id's by executing a query and reading the Id's returned.
    */
-  protected List<Long> getMoreIds(int requestSize) {
+  protected List<Long> getMoreIds(Object tenantKey, int requestSize) {
 
     String sql = getSql(requestSize);
 
@@ -148,7 +193,7 @@ public abstract class SequenceIdGenerator implements PlatformIdGenerator {
     PreparedStatement statement = null;
     ResultSet resultSet = null;
     try {
-      connection = dataSource.getConnection();
+      connection = connectionFor(tenantKey);
 
       statement = connection.prepareStatement(sql);
       resultSet = statement.executeQuery();
@@ -174,6 +219,17 @@ public abstract class SequenceIdGenerator implements PlatformIdGenerator {
     }
   }
 
+  /**
+   * Return a connection for the given tenant. For multi-tenant this is routed to the
+   * tenant database/schema/catalog; otherwise the plain DataSource connection is used.
+   */
+  private Connection connectionFor(Object tenantKey) throws SQLException {
+    if (tenantSource != null && tenantKey != SINGLE) {
+      return tenantSource.connectionForTenant(tenantKey);
+    }
+    return dataSource.getConnection();
+  }
+
   /**
    * Close the JDBC resources.
    */

ebean-api/src/main/java/io/ebean/config/dbplatform/TenantConnectionSource.java

@@ -0,0 +1,29 @@
+package io.ebean.config.dbplatform;
+
+import java.sql.Connection;
+import java.sql.SQLException;
+
+/**
+ * Optionally implemented by the DataSource passed to a {@link SequenceIdGenerator}
+ * to make sequence id allocation multi-tenant aware.
+ * <p>
+ * When the DataSource implements this interface the sequence generator maintains
+ * a separate id buffer per tenant and obtains connections that are routed to the
+ * correct tenant database (TenantMode.DB) or schema/catalog (TenantMode.SCHEMA / CATALOG).
+ * <p>
+ * The {@link #connectionForTenant(Object)} method takes an explicit tenantId so that
+ * background pre-fetch (which runs on a separate thread without the current tenant
+ * in scope) can fetch sequence values for the tenant captured at submit time.
+ */
+public interface TenantConnectionSource {
+
+  /**
+   * Return the current tenant id, or null when there is no current tenant scope.
+   */
+  Object currentTenantId();
+
+  /**
+   * Return a connection routed to the given tenant (its database, schema or catalog).
+   */
+  Connection connectionForTenant(Object tenantId) throws SQLException;
+}

ebean-core/src/main/java/io/ebeaninternal/server/core/InternalConfiguration.java

@@ -81,6 +81,7 @@ public final class InternalConfiguration {
   private final Binder binder;
   private final DeployCreateProperties deployCreateProperties;
   private final DeployUtil deployUtil;
+  private final DataSourceSupplier dataSourceSupplier;
   private final BeanDescriptorManager beanDescriptorManager;
   private final CQueryEngine cQueryEngine;
   private final ClusterManager clusterManager;
@@ -124,6 +125,7 @@ public final class InternalConfiguration {
 
     final InternalConfigXmlMap xmlMap = initExternalMapping();
     this.dtoBeanManager = new DtoBeanManager(typeManager, xmlMap.readDtoMapping());
+    this.dataSourceSupplier = createDataSourceSupplier();
     this.beanDescriptorManager = new BeanDescriptorManager(this);
     Map<String, String> asOfTableMapping = beanDescriptorManager.deploy(xmlMap.xmlDeployment());
     Map<String, String> draftTableMap = beanDescriptorManager.draftTableMap();
@@ -391,7 +393,7 @@ public final class InternalConfiguration {
 
     TransactionManagerOptions options =
       new TransactionManagerOptions(server, notifyL2CacheInForeground, config, scopeManager, clusterManager, backgroundExecutor,
-        indexUpdateProcessor, beanDescriptorManager, dataSource(), profileHandler(), logManager,
+        indexUpdateProcessor, beanDescriptorManager, dataSourceSupplier, profileHandler(), logManager,
         tableModState, cacheNotify);
 
     if (config.isDocStoreOnly()) {
@@ -409,9 +411,16 @@ public final class InternalConfiguration {
   }
 
   /**
-   * Return the DataSource supplier based on the tenancy mode.
+   * Return the DataSource supplier (multi-tenant aware) based on the tenancy mode.
    */
-  private DataSourceSupplier dataSource() {
+  public DataSourceSupplier getDataSourceSupplier() {
+    return dataSourceSupplier;
+  }
+
+  /**
+   * Create the DataSource supplier based on the tenancy mode.
+   */
+  private DataSourceSupplier createDataSourceSupplier() {
     switch (config.getTenantMode()) {
       case DB:
       case DB_WITH_MASTER:

ebean-core/src/main/java/io/ebeaninternal/server/deploy/BeanDescriptorManager.java

@@ -46,6 +46,8 @@ import io.ebeanservice.docstore.api.DocStoreFactory;
 import jakarta.persistence.MappedSuperclass;
 import jakarta.persistence.PersistenceException;
 import jakarta.persistence.Transient;
+import io.ebeaninternal.server.transaction.DataSourceSupplier;
+import io.ebeaninternal.server.transaction.SequenceDataSource;
 import javax.sql.DataSource;
 import java.io.Serializable;
 import java.lang.reflect.Field;
@@ -96,7 +98,7 @@ public final class BeanDescriptorManager implements BeanDescriptorMap, SpiBeanTy
   private final Map<String, List<BeanDescriptor<?>>> tableToDescMap = new HashMap<>();
   private final Map<String, List<BeanDescriptor<?>>> tableToViewDescMap = new HashMap<>();
   private final DbIdentity dbIdentity;
-  private final DataSource dataSource;
+  private final DataSourceSupplier dataSourceSupplier;
   private final DatabasePlatform databasePlatform;
   private final SpiCacheManager cacheManager;
   private final BackgroundExecutor backgroundExecutor;
@@ -132,7 +134,7 @@ public final class BeanDescriptorManager implements BeanDescriptorMap, SpiBeanTy
     this.cacheManager = config.getCacheManager();
     this.docStoreFactory = config.getDocStoreFactory();
     this.backgroundExecutor = config.getBackgroundExecutor();
-    this.dataSource = this.config.getDataSource();
+    this.dataSourceSupplier = config.getDataSourceSupplier();
     this.encryptKeyManager = this.config.getEncryptKeyManager();
     this.databasePlatform = this.config.getDatabasePlatform();
     this.multiValueBind = config.getMultiValueBind();
@@ -1272,7 +1274,10 @@ public final class BeanDescriptorManager implements BeanDescriptorMap, SpiBeanTy
   }
 
   private PlatformIdGenerator createSequenceIdGenerator(String seqName, int stepSize) {
-    return databasePlatform.createSequenceIdGenerator(backgroundExecutor, dataSource, stepSize, seqName);
+    DataSource ds = config.getTenantMode().isDynamicDataSource()
+      ? new SequenceDataSource(dataSourceSupplier)
+      : dataSourceSupplier.dataSource();
+    return databasePlatform.createSequenceIdGenerator(backgroundExecutor, ds, stepSize, seqName);
   }
 
   private void setAccessors(DeployBeanDescriptor<?> deploy) {

ebean-core/src/main/java/io/ebeaninternal/server/transaction/SequenceDataSource.java

@@ -0,0 +1,82 @@
+package io.ebeaninternal.server.transaction;
+
+import io.ebean.config.dbplatform.TenantConnectionSource;
+
+import javax.sql.DataSource;
+import java.io.PrintWriter;
+import java.sql.Connection;
+import java.sql.SQLException;
+import java.sql.SQLFeatureNotSupportedException;
+import java.util.logging.Logger;
+
+/**
+ * DataSource used for sequence id generation that is multi-tenant aware.
+ * <p>
+ * Delegates to the current tenant DataSource via the {@link DataSourceSupplier} and
+ * additionally implements {@link TenantConnectionSource} so the sequence generator can
+ * maintain a separate id buffer per tenant and obtain connections routed to a specific
+ * tenant (needed for background pre-fetch where the current tenant is not in scope).
+ */
+public final class SequenceDataSource implements DataSource, TenantConnectionSource {
+
+  private final DataSourceSupplier supplier;
+
+  public SequenceDataSource(DataSourceSupplier supplier) {
+    this.supplier = supplier;
+  }
+
+  @Override
+  public Object currentTenantId() {
+    return supplier.currentTenantId();
+  }
+
+  @Override
+  public Connection connectionForTenant(Object tenantId) throws SQLException {
+    return supplier.connection(tenantId);
+  }
+
+  @Override
+  public Connection getConnection() throws SQLException {
+    return supplier.connection(supplier.currentTenantId());
+  }
+
+  @Override
+  public Connection getConnection(String username, String password) throws SQLException {
+    return supplier.dataSource().getConnection(username, password);
+  }
+
+  @Override
+  public <T> T unwrap(Class<T> iface) throws SQLException {
+    return supplier.dataSource().unwrap(iface);
+  }
+
+  @Override
+  public boolean isWrapperFor(Class<?> iface) throws SQLException {
+    return supplier.dataSource().isWrapperFor(iface);
+  }
+
+  @Override
+  public PrintWriter getLogWriter() throws SQLException {
+    return supplier.dataSource().getLogWriter();
+  }
+
+  @Override
+  public void setLogWriter(PrintWriter out) throws SQLException {
+    supplier.dataSource().setLogWriter(out);
+  }
+
+  @Override
+  public void setLoginTimeout(int seconds) throws SQLException {
+    supplier.dataSource().setLoginTimeout(seconds);
+  }
+
+  @Override
+  public int getLoginTimeout() throws SQLException {
+    return supplier.dataSource().getLoginTimeout();
+  }
+
+  @Override
+  public Logger getParentLogger() throws SQLFeatureNotSupportedException {
+    throw new SQLFeatureNotSupportedException();
+  }
+}

ebean-test/src/test/java/io/ebean/xtest/config/dbplatform/SequenceBatchIdGeneratorTest.java

@@ -16,11 +16,6 @@ public class SequenceBatchIdGeneratorTest {
 
     TD generator = new TD();
 
-    // simulate out of order adding of sequence ids
-    generator.add(asList(1L, 2L));
-    generator.add(asList(5L, 6L));
-    generator.add(asList(3L, 4L));
-
     Assertions.assertThat(generator.nextId(null)).isEqualTo(1L);
     Assertions.assertThat(generator.nextId(null)).isEqualTo(2L);
     Assertions.assertThat(generator.nextId(null)).isEqualTo(3L);
@@ -35,10 +30,6 @@ public class SequenceBatchIdGeneratorTest {
       super(null, null, null, 10);
     }
 
-    void add(List<Long> ids) {
-      idList.addAll(ids);
-    }
-
     @Override
     public String getSql(int batchSize) {
       return "not used";
@@ -51,13 +42,14 @@ public class SequenceBatchIdGeneratorTest {
     }
 
     @Override
-    protected List<Long> getMoreIds(int requestSize) {
-      return null;
+    protected List<Long> getMoreIds(Object tenantKey, int requestSize) {
+      // simulate out of order ids returned from the database (verifies sorted polling)
+      return asList(1L, 2L, 5L, 6L, 3L, 4L);
     }
 
     @Override
     protected void loadInBackground(int requestSize) {
-      // do nothing
+      // do nothing - avoid background reload re-introducing already polled ids
     }
   }
 }

ebean-test/src/test/java/org/tests/idkeys/TenantSequenceTest.java

@@ -0,0 +1,88 @@
+package org.tests.idkeys;
+
+import io.ebean.Database;
+import io.ebean.DatabaseFactory;
+import io.ebean.config.DatabaseConfig;
+import io.ebean.config.TenantDataSourceProvider;
+import io.ebean.config.TenantMode;
+import io.ebean.platform.h2.H2Platform;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Test;
+import org.multitenant.partition.UserContext;
+import org.tests.idkeys.db.GenKeySeqA;
+
+import javax.sql.DataSource;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * Sequence id generation must be tenant aware - each tenant (DB per tenant) has its
+ * own sequence so allocation does not bleed across tenants.
+ */
+class TenantSequenceTest {
+
+  @AfterEach
+  void after() {
+    UserContext.reset();
+  }
+
+  @Test
+  void dbPerTenant_sequencesAreIndependent() {
+    Database db = setup();
+    try {
+      UserContext.set("u1", "1");
+      assertThat(db.nextId(GenKeySeqA.class)).isEqualTo(1L);
+      assertThat(db.nextId(GenKeySeqA.class)).isEqualTo(2L);
+
+      // tenant 2 uses its own database/sequence - starts fresh at 1
+      UserContext.set("u2", "2");
+      assertThat(db.nextId(GenKeySeqA.class)).isEqualTo(1L);
+
+      // tenant 1 continues from where it left off
+      UserContext.set("u1", "1");
+      assertThat(db.nextId(GenKeySeqA.class)).isEqualTo(3L);
+    } finally {
+      db.shutdown();
+    }
+  }
+
+  private Database setup() {
+    DatabaseConfig config = new DatabaseConfig();
+    config.setName("tenantSeq");
+    config.setRegister(false);
+    config.setDefaultServer(false);
+    config.setDatabasePlatform(new H2Platform());
+    config.setTenantMode(TenantMode.DB);
+    config.setCurrentTenantProvider(() -> UserContext.get().getTenantId());
+    config.setTenantDataSourceProvider(new TenantDataSourceProvider() {
+      final Map<Object, DataSource> map = new ConcurrentHashMap<>();
+
+      @Override
+      public DataSource dataSource(Object tenantId) {
+        if (tenantId == null) {
+          tenantId = "1";
+        }
+        return map.computeIfAbsent(tenantId, this::create);
+      }
+
+      private DataSource create(Object tenantId) {
+        DatabaseConfig c = new DatabaseConfig();
+        c.setName("tenantSeq-" + tenantId);
+        c.setRegister(false);
+        c.setDefaultServer(false);
+        c.setDdlGenerate(true);
+        c.setDdlRun(true);
+        c.setDdlExtra(false);
+        c.getDataSourceConfig().setUrl("jdbc:h2:mem:tenantSeq-" + tenantId + ";DB_CLOSE_DELAY=-1");
+        c.getDataSourceConfig().setUsername("sa");
+        c.getDataSourceConfig().setPassword("");
+        c.getClasses().add(GenKeySeqA.class);
+        return DatabaseFactory.create(c).dataSource();
+      }
+    });
+    config.getClasses().add(GenKeySeqA.class);
+    return DatabaseFactory.create(config);
+  }
+}