This is an automated email from the ASF dual-hosted git repository.
sergehuber pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/unomi.git
The following commit(s) were added to refs/heads/master by this push:
new 39d6c640b UNOMI-1002: Stop serializing public requests in
compatibility mode (#915)
39d6c640b is described below
commit 39d6c640bd2730072270dc8f6ab3f29b1f99743d
Author: Serge Huber <[email protected]>
AuthorDate: Tue Oct 13 07:41:53 2026 +0200
UNOMI-1002: Stop serializing public requests in compatibility mode (#915)
Co-authored-by: Claude Opus 5.5 <[email protected]>
---
.../apache/unomi/api/tenants/TenantService.java | 19 +-
.../asciidoc/migrations/v2-compatibility-mode.adoc | 3 +
.../AuthenticationFilterBlankPasswordTest.java | 24 ++-
.../services/impl/tenants/TenantServiceImpl.java | 76 ++++++-
.../impl/tenants/TenantServiceImplTest.java | 228 +++++++++++++++++++++
5 files changed, 340 insertions(+), 10 deletions(-)
diff --git a/api/src/main/java/org/apache/unomi/api/tenants/TenantService.java
b/api/src/main/java/org/apache/unomi/api/tenants/TenantService.java
index dff0128b2..136e80e17 100644
--- a/api/src/main/java/org/apache/unomi/api/tenants/TenantService.java
+++ b/api/src/main/java/org/apache/unomi/api/tenants/TenantService.java
@@ -55,17 +55,14 @@ public interface TenantService {
*/
ApiKeyCreationResult generateApiKey(String tenantId, Long validityPeriod);
- /**
- * Retrieves a tenant by its ID.
- *
- * @param tenantId the ID of the tenant to retrieve
- * @return the Tenant object if found, null otherwise
- */
/**
* Returns the tenant with the given identifier, and creates it when it
does not exist yet.
* <p>
- * Two callers that race on the same identifier both get the tenant, and
only one creation
- * happens.
+ * Two callers on the same node that race on the same identifier both get
the tenant, and only
+ * one creation happens.
+ * <p>
+ * The result may be cached for a short time and shared between callers:
do not modify it, and
+ * use {@link #getTenant(String)} when the current state of the tenant
matters.
*
* @param tenantId the identifier of the tenant
* @param properties the properties to set when the tenant has to be
created, may be {@code null}
@@ -73,6 +70,12 @@ public interface TenantService {
*/
Tenant getOrCreateTenant(String tenantId, Map<String, Object> properties);
+ /**
+ * Retrieves a tenant by its ID.
+ *
+ * @param tenantId the ID of the tenant to retrieve
+ * @return the Tenant object if found, null otherwise
+ */
Tenant getTenant(String tenantId);
/**
diff --git a/manual/src/main/asciidoc/migrations/v2-compatibility-mode.adoc
b/manual/src/main/asciidoc/migrations/v2-compatibility-mode.adoc
index dfded3512..ea53c442a 100644
--- a/manual/src/main/asciidoc/migrations/v2-compatibility-mode.adoc
+++ b/manual/src/main/asciidoc/migrations/v2-compatibility-mode.adoc
@@ -29,6 +29,9 @@ When single-tenant compatibility mode is enabled:
- **Protected events** (like `login`, `updateProperties`) require IP +
X-Unomi-Peer (like V2)
- **Private endpoints** require system administrator authentication (like V2)
- **Tenant `default`** is used for every operation. The mode cannot be pointed
at another tenant.
+ The tenant is created on a public request if it is missing. Each node then
keeps it in memory for
+ about 30 seconds, so public requests do not wait on each other or re-read it
from the database
+ every time. A change made on another node is picked up after about that time.
- **No authentication** is required for non-protected events (like V2)
This allows V2 clients to work with Unomi V3 immediately after migration,
giving you time to gradually update client applications to use the new V3
authentication model.
diff --git
a/rest/src/test/java/org/apache/unomi/rest/authentication/AuthenticationFilterBlankPasswordTest.java
b/rest/src/test/java/org/apache/unomi/rest/authentication/AuthenticationFilterBlankPasswordTest.java
index 1b4baac8e..fa707202a 100644
---
a/rest/src/test/java/org/apache/unomi/rest/authentication/AuthenticationFilterBlankPasswordTest.java
+++
b/rest/src/test/java/org/apache/unomi/rest/authentication/AuthenticationFilterBlankPasswordTest.java
@@ -59,6 +59,7 @@ class AuthenticationFilterBlankPasswordTest {
private RestAuthenticationConfig restAuthenticationConfig;
private TenantService tenantService;
+ private SecurityService securityService;
private JAASAuthenticationFilter jaasAuthenticationFilter;
private AuthenticationFilter filter;
@@ -66,6 +67,7 @@ class AuthenticationFilterBlankPasswordTest {
void setUp() {
restAuthenticationConfig = mock(RestAuthenticationConfig.class);
tenantService = mock(TenantService.class);
+ securityService = mock(SecurityService.class);
// Stubbed so the tests below can assert the credential never reached
JAAS. Every refusal
// path in the filter answers 401, so the status alone cannot tell
"refused for a blank
// password" apart from "JAAS rejected it" — only this can.
@@ -73,7 +75,7 @@ class AuthenticationFilterBlankPasswordTest {
filter = new AuthenticationFilter(
restAuthenticationConfig,
tenantService,
- mock(SecurityService.class),
+ securityService,
mock(ExecutionContextManager.class),
jaasAuthenticationFilter);
}
@@ -216,6 +218,26 @@ class AuthenticationFilterBlankPasswordTest {
verify(tenantService).getOrCreateTenant(eq("default"), any());
}
+ /**
+ * Compatibility mode always uses tenant {@code default}. A client must
not be able to steer
+ * which tenant is created via request headers (UNOMI-1002).
+ */
+ @Test
+ void
filterAlwaysUsesDefaultTenantOnPublicPathInSingleTenantCompatibilityMode()
throws IOException {
+
when(restAuthenticationConfig.isSingleTenantCompatibilityModeEnabled()).thenReturn(true);
+ when(restAuthenticationConfig.getPublicPathPatterns())
+ .thenReturn(Collections.singletonList(Pattern.compile("POST
context\\.json")));
+ ContainerRequestContext requestContext = request("context.json", null);
+
when(requestContext.getHeaderString("X-Unomi-Tenant-Id")).thenReturn("attacker-tenant");
+
when(requestContext.getHeaderString("X-Unomi-Api-Key")).thenReturn("some-key");
+
+ filter.filter(requestContext);
+
+ verify(tenantService).getOrCreateTenant(eq("default"), any());
+ verify(tenantService,
never()).getOrCreateTenant(eq("attacker-tenant"), any());
+ verify(securityService).createSubject(eq("default"), eq(false));
+ }
+
private void assertUnauthorizedWithoutReachingJaas(ContainerRequestContext
requestContext) throws IOException {
ArgumentCaptor<Response> aborted =
ArgumentCaptor.forClass(Response.class);
verify(requestContext).abortWith(aborted.capture());
diff --git
a/services/src/main/java/org/apache/unomi/services/impl/tenants/TenantServiceImpl.java
b/services/src/main/java/org/apache/unomi/services/impl/tenants/TenantServiceImpl.java
index c2c6e474d..cc2951703 100644
---
a/services/src/main/java/org/apache/unomi/services/impl/tenants/TenantServiceImpl.java
+++
b/services/src/main/java/org/apache/unomi/services/impl/tenants/TenantServiceImpl.java
@@ -29,7 +29,11 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.*;
+import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.locks.ReentrantLock;
+import java.util.function.LongSupplier;
/**
* Default implementation of {@link TenantService}.
@@ -40,6 +44,23 @@ public class TenantServiceImpl implements TenantService {
private static final String TENANT_ID_PATTERN =
"^[a-zA-Z0-9][a-zA-Z0-9-_]*[a-zA-Z0-9]$";
private final List<TenantLifecycleListener> lifecycleListeners = new
CopyOnWriteArrayList<>();
+ /**
+ * How long {@link #getOrCreateTenant} trusts a cached tenant before it
reads persistence again.
+ * The cache is local to this node, so a change made on another node, or
directly in the search
+ * engine, goes unnoticed here for about this long, and longer while a
reload is still running.
+ */
+ static final long GET_OR_CREATE_CACHE_TTL_NANOS =
TimeUnit.SECONDS.toNanos(30);
+
+ /**
+ * Hot-path cache for {@link #getOrCreateTenant}. Compatibility-mode
public requests call that
+ * method on every context/eventcollector hit; without a cache they
serialized on a method lock
+ * and re-read the tenant from persistence each time (UNOMI-1002).
+ */
+ private final ConcurrentHashMap<String, CachedTenant> getOrCreateCache =
new ConcurrentHashMap<>();
+ /** Serializes load/create so concurrent first requests on this node
create the tenant once. */
+ private final ReentrantLock getOrCreateLock = new ReentrantLock();
+ /** Replaced in tests to make cache entries expire without waiting. */
+ LongSupplier nanoClock = System::nanoTime;
private PersistenceService persistenceService;
private ExecutionContextManager executionContextManager;
private SecretHashService secretHashService;
@@ -114,7 +135,35 @@ public class TenantServiceImpl implements TenantService {
}
@Override
- public synchronized Tenant getOrCreateTenant(String tenantId, Map<String,
Object> properties) {
+ public Tenant getOrCreateTenant(String tenantId, Map<String, Object>
properties) {
+ if (tenantId == null) {
+ throw new IllegalArgumentException("Tenant ID cannot be null or
empty");
+ }
+ CachedTenant cached = getOrCreateCache.get(tenantId);
+ if (isFresh(cached)) {
+ return cached.tenant;
+ }
+ if (cached == null) {
+ getOrCreateLock.lock();
+ } else if (!getOrCreateLock.tryLock()) {
+ // Another thread is already refreshing. Keep answering from the
expired entry rather
+ // than queue every request behind that read.
+ return cached.tenant;
+ }
+ try {
+ cached = getOrCreateCache.get(tenantId);
+ if (isFresh(cached)) {
+ return cached.tenant;
+ }
+ Tenant tenant = loadOrCreateTenant(tenantId, properties);
+ getOrCreateCache.put(tenantId, new CachedTenant(tenant,
nanoClock.getAsLong()));
+ return tenant;
+ } finally {
+ getOrCreateLock.unlock();
+ }
+ }
+
+ private Tenant loadOrCreateTenant(String tenantId, Map<String, Object>
properties) {
Tenant tenant = getTenant(tenantId);
if (tenant != null) {
return tenant;
@@ -132,6 +181,26 @@ public class TenantServiceImpl implements TenantService {
}
}
+ private boolean isFresh(CachedTenant cached) {
+ return cached != null && nanoClock.getAsLong() - cached.loadedAtNanos
< GET_OR_CREATE_CACHE_TTL_NANOS;
+ }
+
+ private void invalidateGetOrCreateCache(String tenantId) {
+ if (tenantId != null) {
+ getOrCreateCache.remove(tenantId);
+ }
+ }
+
+ private static final class CachedTenant {
+ private final Tenant tenant;
+ private final long loadedAtNanos;
+
+ private CachedTenant(Tenant tenant, long loadedAtNanos) {
+ this.tenant = tenant;
+ this.loadedAtNanos = loadedAtNanos;
+ }
+ }
+
@Override
public Tenant createTenant(String requestedId, Map<String, Object>
properties) {
validateTenantId(requestedId);
@@ -198,6 +267,7 @@ public class TenantServiceImpl implements TenantService {
tenant.getApiKeys().removeIf(existingKey ->
existingKey.getKeyType() == keyType);
tenant.getApiKeys().add(apiKey);
persistenceService.save(tenant);
+ invalidateGetOrCreateCache(tenantId);
}
return new ApiKeyCreationResult(apiKey, plainTextKey);
@@ -217,6 +287,9 @@ public class TenantServiceImpl implements TenantService {
@Override
public void saveTenant(Tenant tenant) {
executionContextManager.executeAsSystem(() ->
persistenceService.save(tenant));
+ if (tenant != null) {
+ invalidateGetOrCreateCache(tenant.getItemId());
+ }
}
@Override
@@ -236,6 +309,7 @@ public class TenantServiceImpl implements TenantService {
}
persistenceService.remove(tenantId, Tenant.class);
});
+ invalidateGetOrCreateCache(tenantId);
}
@Override
diff --git
a/services/src/test/java/org/apache/unomi/services/impl/tenants/TenantServiceImplTest.java
b/services/src/test/java/org/apache/unomi/services/impl/tenants/TenantServiceImplTest.java
index aad74bebf..12ec5b3b0 100644
---
a/services/src/test/java/org/apache/unomi/services/impl/tenants/TenantServiceImplTest.java
+++
b/services/src/test/java/org/apache/unomi/services/impl/tenants/TenantServiceImplTest.java
@@ -29,20 +29,38 @@ import org.mockito.junit.jupiter.MockitoExtension;
import org.mockito.junit.jupiter.MockitoSettings;
import org.mockito.quality.Strictness;
+import java.time.Duration;
import java.util.ArrayList;
import java.util.Collections;
+import java.util.HashMap;
import java.util.List;
+import java.util.Map;
import java.util.Objects;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.concurrent.locks.LockSupport;
+import java.util.function.Consumer;
import java.util.function.Supplier;
import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotSame;
import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertSame;
import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTimeoutPreemptively;
+import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
@@ -61,9 +79,12 @@ public class TenantServiceImplTest {
private TenantServiceImpl tenantService;
+ private final AtomicLong clock = new AtomicLong();
+
@BeforeEach
public void setUp() {
tenantService = new TenantServiceImpl();
+ tenantService.nanoClock = clock::get;
tenantService.setPersistenceService(persistenceService);
tenantService.setExecutionContextManager(executionContextManager);
tenantService.setSecretHashService(secretHashService);
@@ -82,6 +103,213 @@ public class TenantServiceImplTest {
// without depending on the real hash implementation.
when(secretHashService.verify(anyString(), anyString()))
.thenAnswer(invocation ->
Objects.equals(invocation.getArgument(0), invocation.getArgument(1)));
+ when(secretHashService.hash(anyString())).thenAnswer(invocation ->
invocation.getArgument(0));
+ }
+
+ @Test
+ public void getOrCreateTenant_readsPersistenceOnceAfterCacheWarm() {
+ Tenant existing = new Tenant();
+ existing.setItemId("default");
+ when(persistenceService.load("default",
Tenant.class)).thenReturn(existing);
+
+ assertSame(existing, tenantService.getOrCreateTenant("default",
Collections.emptyMap()));
+ assertSame(existing, tenantService.getOrCreateTenant("default",
Collections.emptyMap()));
+ assertSame(existing, tenantService.getOrCreateTenant("default",
Collections.emptyMap()));
+
+ verify(persistenceService, times(1)).load("default", Tenant.class);
+ }
+
+ @Test
+ public void getOrCreateTenant_reloadsAfterSave() {
+ Tenant first = new Tenant();
+ first.setItemId("default");
+ Map<String, Object> firstProps = new HashMap<>();
+ firstProps.put("name", "first");
+ first.setProperties(firstProps);
+
+ Tenant second = new Tenant();
+ second.setItemId("default");
+ Map<String, Object> secondProps = new HashMap<>();
+ secondProps.put("name", "second");
+ second.setProperties(secondProps);
+
+ when(persistenceService.load("default",
Tenant.class)).thenReturn(first, second);
+
+ assertEquals("first", tenantService.getOrCreateTenant("default",
Collections.emptyMap())
+ .getProperties().get("name"));
+
+ tenantService.saveTenant(second);
+ assertEquals("second", tenantService.getOrCreateTenant("default",
Collections.emptyMap())
+ .getProperties().get("name"));
+
+ verify(persistenceService, times(2)).load("default", Tenant.class);
+ }
+
+ @Test
+ public void getOrCreateTenant_reloadsAfterDelete() {
+ Tenant existing = defaultTenant();
+ AtomicReference<Tenant> store = inMemoryTenantStore(existing, tenant
-> { });
+
+ assertSame(existing, getOrCreateDefaultTenant());
+
+ tenantService.deleteTenant("default");
+ assertNull(store.get());
+
+ Tenant afterDelete = getOrCreateDefaultTenant();
+ assertNotSame(existing, afterDelete);
+ assertSame(store.get(), afterDelete);
+ }
+
+ @Test
+ public void getOrCreateTenant_reloadsAfterApiKeyGeneration() {
+ inMemoryTenantStore(defaultTenant(), tenant -> { });
+
+ getOrCreateDefaultTenant();
+ tenantService.generateApiKeyWithType("default",
ApiKey.ApiKeyType.PUBLIC, null);
+ getOrCreateDefaultTenant();
+
+ // One read per getOrCreateTenant call, plus the one
generateApiKeyWithType does itself.
+ verify(persistenceService, times(3)).load("default", Tenant.class);
+ }
+
+ /**
+ * Nothing tells this node about a delete done on another one, so the
entry has to expire by
+ * itself for the tenant to be created again.
+ */
+ @Test
+ public void
getOrCreateTenant_recreatesTenantDeletedOnAnotherNodeOnceEntryExpires() {
+ Tenant existing = defaultTenant();
+ AtomicReference<Tenant> store = inMemoryTenantStore(existing, tenant
-> { });
+ assertSame(existing, getOrCreateDefaultTenant());
+
+ store.set(null);
+ assertSame(existing, getOrCreateDefaultTenant());
+
+ clock.addAndGet(TenantServiceImpl.GET_OR_CREATE_CACHE_TTL_NANOS);
+ Tenant recreated = getOrCreateDefaultTenant();
+ assertNotSame(existing, recreated);
+ assertSame(store.get(), recreated);
+ }
+
+ @Test
+ public void
getOrCreateTenant_answersFromExpiredEntryWhileAnotherThreadRefreshes() throws
Exception {
+ Tenant existing = defaultTenant();
+ CountDownLatch refreshStarted = new CountDownLatch(1);
+ CountDownLatch releaseRefresh = new CountDownLatch(1);
+ AtomicInteger loads = new AtomicInteger();
+ when(persistenceService.load("default",
Tenant.class)).thenAnswer(invocation -> {
+ if (loads.incrementAndGet() > 1) {
+ refreshStarted.countDown();
+ releaseRefresh.await();
+ }
+ return existing;
+ });
+ getOrCreateDefaultTenant();
+ clock.addAndGet(TenantServiceImpl.GET_OR_CREATE_CACHE_TTL_NANOS);
+
+ ExecutorService pool = Executors.newSingleThreadExecutor();
+ try {
+ Future<Tenant> refresh =
pool.submit(this::getOrCreateDefaultTenant);
+ assertTrue(refreshStarted.await(10, TimeUnit.SECONDS));
+
+ assertSame(existing,
assertTimeoutPreemptively(Duration.ofSeconds(10),
this::getOrCreateDefaultTenant));
+
+ releaseRefresh.countDown();
+ assertSame(existing, refresh.get(10, TimeUnit.SECONDS));
+ } finally {
+ releaseRefresh.countDown();
+ pool.shutdownNow();
+ }
+ }
+
+ @Test
+ public void getOrCreateTenant_concurrentFirstRequestsCreateOnce() throws
Exception {
+ AtomicInteger creations = new AtomicInteger();
+ inMemoryTenantStore(null, tenant -> {
+ // Only createTenant's own save carries no API key yet. Hold it
open so that callers
+ // which are not kept out by the lock would all get past the
existence check.
+ if (tenant.getApiKeys() == null || tenant.getApiKeys().isEmpty()) {
+ creations.incrementAndGet();
+ LockSupport.parkNanos(TimeUnit.MILLISECONDS.toNanos(50));
+ }
+ });
+
+ int threads = 20;
+ ExecutorService pool = Executors.newFixedThreadPool(threads);
+ try {
+ CountDownLatch start = new CountDownLatch(1);
+ List<Future<Tenant>> futures = new ArrayList<>();
+ for (int i = 0; i < threads; i++) {
+ futures.add(pool.submit(() -> {
+ start.await();
+ return getOrCreateDefaultTenant();
+ }));
+ }
+ start.countDown();
+
+ for (Future<Tenant> future : futures) {
+ assertEquals("default", future.get(10,
TimeUnit.SECONDS).getItemId());
+ }
+ } finally {
+ pool.shutdownNow();
+ }
+ assertEquals(1, creations.get(), "Exactly one tenant must be created
under concurrent first requests");
+ }
+
+ @Test
+ public void
getOrCreateTenant_returnsTenantCreatedByAnotherNodeDuringCreation() {
+ Tenant createdElsewhere = defaultTenant();
+ // Missing on the first read, there by the time createTenant checks
for a duplicate.
+ when(persistenceService.load("default",
Tenant.class)).thenReturn(null, createdElsewhere);
+
+ assertSame(createdElsewhere, getOrCreateDefaultTenant());
+
+ verify(persistenceService, never()).save(any(Tenant.class));
+ }
+
+ @Test
+ public void getOrCreateTenant_doesNotCacheFailure() {
+ Tenant existing = defaultTenant();
+ when(persistenceService.load("default", Tenant.class))
+ .thenThrow(new IllegalStateException("search engine
unavailable"))
+ .thenReturn(existing);
+
+ assertThrows(IllegalStateException.class,
this::getOrCreateDefaultTenant);
+ // Retried from another thread, which a lock left held by the failed
call would block.
+ assertSame(existing, assertTimeoutPreemptively(Duration.ofSeconds(10),
this::getOrCreateDefaultTenant));
+ }
+
+ @Test
+ public void getOrCreateTenant_rejectsNullTenantId() {
+ assertThrows(IllegalArgumentException.class,
+ () -> tenantService.getOrCreateTenant(null,
Collections.emptyMap()));
+ }
+
+ private static Tenant defaultTenant() {
+ Tenant tenant = new Tenant();
+ tenant.setItemId("default");
+ return tenant;
+ }
+
+ private Tenant getOrCreateDefaultTenant() {
+ return tenantService.getOrCreateTenant("default",
Collections.singletonMap("name", "default"));
+ }
+
+ /** Backs tenant {@code default} with a single in-memory slot; {@code
onSave} sees each tenant before it is stored. */
+ private AtomicReference<Tenant> inMemoryTenantStore(Tenant initial,
Consumer<Tenant> onSave) {
+ AtomicReference<Tenant> store = new AtomicReference<>(initial);
+ when(persistenceService.load(eq("default"),
eq(Tenant.class))).thenAnswer(invocation -> store.get());
+ when(persistenceService.save(any(Tenant.class))).thenAnswer(invocation
-> {
+ Tenant tenant = invocation.getArgument(0);
+ onSave.accept(tenant);
+ store.set(tenant);
+ return true;
+ });
+ doAnswer(invocation -> {
+ store.set(null);
+ return null;
+ }).when(persistenceService).remove("default", Tenant.class);
+ return store;
}
@Test
