diff --git a/CHANGELOG.md b/CHANGELOG.md index 6508d77..c157cc0 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -31,6 +31,8 @@ All notable changes to this project are documented here. The format follows Keep - None for the health migration; both Test connection record actions, including on-demand write tests, are retained. ### Changed +- Multi-node deployments: the runtime caches are registered with QQQ `ClusterCaches`. Shared (cleared on every node): the bucket rows' presign limits (`qbit-object-storage.bucket-row-limits`, on a bucket row insert or update), the enabled header rules (`qbit-object-storage.header-rules`, on a header rule insert, update or delete), both through the new `ConfigRowChangeCustomizer`, and the capability statuses (`qbit-object-storage.capabilities`: a probe result recorded on one node clears that capability on the others, which reload it from the capability row; before, other nodes kept the old status until restart). The 15 s TTLs remain the safety net. Node-local, with reasons: the alias and connection row-id maps, the S3 client map, the process-permission map, the provider defaults, the running-scan guard and the per-pod presign rate limiter (`StorageBucket.PRESIGN_COUNTS`; per-node rate limits are accepted for now). `ObjectStorageClusterCachesTest` audits the package for unregistered static caches and covers remote and local invalidation. +- Builds against QQQ `4.1.0-SNAPSHOT` (`qqq.snapshot.version`) for `ClusterCaches` until the next QQQ RC ships; a release must pin that RC. - Read the qbit metadata version from the Maven-filtered artifact version so release metadata matches the build and help bundle. - Rechecked the released RC.8 JARs: core still has no S3 session-token/provider hook or file-reference interpreter. Retained both compatibility shims and moved their review deadline to RC.9 under issue #10. - Raised the QQQ BOM baseline to released `4.1.0-RC.8` for the central health API. diff --git a/src/main/java/com/kingsrook/qbits/objectstorage/api/StorageBucket.java b/src/main/java/com/kingsrook/qbits/objectstorage/api/StorageBucket.java index c3e31ee..99b8b00 100644 --- a/src/main/java/com/kingsrook/qbits/objectstorage/api/StorageBucket.java +++ b/src/main/java/com/kingsrook/qbits/objectstorage/api/StorageBucket.java @@ -46,6 +46,7 @@ import com.kingsrook.qbits.objectstorage.internal.PostPolicySigner; import com.kingsrook.qbits.objectstorage.internal.ProtectionRules; import com.kingsrook.qbits.objectstorage.internal.RateLimiter; import com.kingsrook.qqq.backend.core.actions.permissions.CorePermissionEnforcer; +import com.kingsrook.qqq.backend.core.cluster.ClusterCaches; import com.kingsrook.qqq.backend.core.context.QContext; import com.kingsrook.qqq.backend.core.logging.QLogger; import com.kingsrook.qqq.backend.core.utils.StringUtils; @@ -96,7 +97,8 @@ public final class StorageBucket { private static final QLogger LOG = QLogger.getLogger(StorageBucket.class); - private static final Map PRESIGN_COUNTS = new ConcurrentHashMap<>(); + private static final Map PRESIGN_COUNTS = ClusterCaches.nodeLocal(new ConcurrentHashMap<>(), + "per-pod presign rate limiter; per-node rate limits are accepted for now (cluster HA spec, owner decision 4)"); private final StorageBucketConfig bucket; private final ConnectionRuntime connection; diff --git a/src/main/java/com/kingsrook/qbits/objectstorage/index/ObjectIndex.java b/src/main/java/com/kingsrook/qbits/objectstorage/index/ObjectIndex.java index 1dd9076..9fc05e5 100644 --- a/src/main/java/com/kingsrook/qbits/objectstorage/index/ObjectIndex.java +++ b/src/main/java/com/kingsrook/qbits/objectstorage/index/ObjectIndex.java @@ -43,6 +43,7 @@ import com.kingsrook.qbits.objectstorage.internal.ConfigSync; import com.kingsrook.qbits.objectstorage.internal.Db; import com.kingsrook.qbits.objectstorage.internal.ObjectStorageRuntime; import com.kingsrook.qbits.objectstorage.model.ObjectStorageTables; +import com.kingsrook.qqq.backend.core.cluster.ClusterCaches; import com.kingsrook.qqq.backend.core.exceptions.QException; import com.kingsrook.qqq.backend.core.logging.QLogger; import com.kingsrook.qqq.backend.core.model.actions.tables.query.QCriteriaOperator; @@ -78,7 +79,8 @@ public final class ObjectIndex public static final int PAGE_SIZE = 1000; - private static final Set RUNNING = java.util.concurrent.ConcurrentHashMap.newKeySet(); + private static final Set RUNNING = ClusterCaches.nodeLocal(java.util.concurrent.ConcurrentHashMap.newKeySet(), + "the buckets this node is scanning now (an in-flight guard, not cached data)"); /** Test seam: simulate a crash after this many pages of a scan (so resume can be tested). */ static volatile Integer failAfterPages = null; diff --git a/src/main/java/com/kingsrook/qbits/objectstorage/internal/CapabilityStore.java b/src/main/java/com/kingsrook/qbits/objectstorage/internal/CapabilityStore.java index 10fbe9c..b59b520 100644 --- a/src/main/java/com/kingsrook/qbits/objectstorage/internal/CapabilityStore.java +++ b/src/main/java/com/kingsrook/qbits/objectstorage/internal/CapabilityStore.java @@ -30,6 +30,7 @@ import com.kingsrook.qbits.objectstorage.api.Capability; import com.kingsrook.qbits.objectstorage.api.CapabilityStatus; import com.kingsrook.qbits.objectstorage.config.StorageBucketConfig; import com.kingsrook.qbits.objectstorage.model.ObjectStorageTables; +import com.kingsrook.qqq.backend.core.cluster.ClusterCaches; import com.kingsrook.qqq.backend.core.logging.QLogger; import com.kingsrook.qqq.backend.core.model.data.QRecord; @@ -47,7 +48,11 @@ public final class CapabilityStore public static final Set CONNECTION_SCOPE = EnumSet.of(Capability.LIST_V2, Capability.PRESIGNED_GET, Capability.PRESIGNED_PUT, Capability.PRESIGNED_POST, Capability.USAGE_ADMIN_API, Capability.QUOTA_ADMIN_API); - private static final Map CACHE = new ConcurrentHashMap<>(); + /** Cluster cache name of the capability statuses (a probe result recorded on one node clears that key on every node). */ + public static final String CACHE_NAME = "qbit-object-storage.capabilities"; + + private static final Map CACHE = ClusterCaches.shared(CACHE_NAME, new ConcurrentHashMap<>(), + CapabilityStore::clearCache, CapabilityStore::clearKey); private static final Set LOADED = ConcurrentHashMap.newKeySet(); @@ -62,7 +67,7 @@ public final class CapabilityStore /******************************************************************************* - ** Drop everything cached. + ** Drop everything cached on this node. *******************************************************************************/ public static void clearCache() { @@ -72,6 +77,28 @@ public final class CapabilityStore + /******************************************************************************* + ** Drop one cached status on this node; the next read reloads the missing + ** statuses from the capability rows (cached ones are kept). + *******************************************************************************/ + static void clearKey(String key) + { + CACHE.remove(key); + LOADED.clear(); + } + + + + /******************************************************************************* + ** Whether a status is cached on this node (tests). + *******************************************************************************/ + static boolean isCached(StorageBucketConfig bucket, Capability capability) + { + return (CACHE.containsKey(scopeKey(bucket, capability))); + } + + + /******************************************************************************* ** The status of a capability for a bucket alias. *******************************************************************************/ @@ -112,10 +139,24 @@ public final class CapabilityStore /******************************************************************************* ** Record one probe result (cache and, when the tables are there, the row). + ** The row is written first, then the key is cleared on every node (they + ** reload it from the row), then cached here. *******************************************************************************/ public static void record(StorageBucketConfig bucket, Capability capability, CapabilityStatus status, String detail) { - CACHE.put(scopeKey(bucket, capability), status); + String key = scopeKey(bucket, capability); + persist(bucket, capability, status, detail); + ClusterCaches.invalidateKey(CACHE_NAME, key); + CACHE.put(key, status); + } + + + + /******************************************************************************* + ** Write one probe result to its capability row; never throws. + *******************************************************************************/ + private static void persist(StorageBucketConfig bucket, Capability capability, CapabilityStatus status, String detail) + { if(!Db.available(ObjectStorageTables.BUCKET_CAPABILITY)) { return; diff --git a/src/main/java/com/kingsrook/qbits/objectstorage/internal/ConfigRowChangeCustomizer.java b/src/main/java/com/kingsrook/qbits/objectstorage/internal/ConfigRowChangeCustomizer.java new file mode 100644 index 0000000..b003484 --- /dev/null +++ b/src/main/java/com/kingsrook/qbits/objectstorage/internal/ConfigRowChangeCustomizer.java @@ -0,0 +1,95 @@ +/* + * QQQ - Low-code Application Framework for Engineers. + * Copyright (C) 2021-2025. Kingsrook, LLC + * 651 N Broad St Ste 205 # 6917 | Middletown DE 19709 | United States + * contact@kingsrook.com + * https://github.com/Kingsrook/ + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package com.kingsrook.qbits.objectstorage.internal; + + +import java.util.List; +import java.util.Optional; +import com.kingsrook.qbits.objectstorage.model.ObjectStorageTables; +import com.kingsrook.qqq.backend.core.actions.customizers.TableCustomizerInterface; +import com.kingsrook.qqq.backend.core.cluster.ClusterCaches; +import com.kingsrook.qqq.backend.core.exceptions.QException; +import com.kingsrook.qqq.backend.core.model.actions.tables.delete.DeleteInput; +import com.kingsrook.qqq.backend.core.model.actions.tables.insert.InsertInput; +import com.kingsrook.qqq.backend.core.model.actions.tables.update.UpdateInput; +import com.kingsrook.qqq.backend.core.model.data.QRecord; + + +/******************************************************************************* + ** Post insert, update and delete of the editable config rows that the runtime + ** caches: a bucket row (presign limits, ConfigSync) or a header rule + ** (DefaultObjectHeaderPolicy). Clears that cache on every node, so an admin's + ** edit takes effect everywhere at once; the short TTLs remain the safety net. + *******************************************************************************/ +public class ConfigRowChangeCustomizer implements TableCustomizerInterface +{ + + /******************************************************************************* + ** {@inheritDoc} + *******************************************************************************/ + @Override + public List postInsert(InsertInput insertInput, List records) throws QException + { + invalidate(insertInput.getTableName()); + return (records); + } + + + + /******************************************************************************* + ** {@inheritDoc} + *******************************************************************************/ + @Override + public List postUpdate(UpdateInput updateInput, List records, Optional> oldRecordList) throws QException + { + invalidate(updateInput.getTableName()); + return (records); + } + + + + /******************************************************************************* + ** {@inheritDoc} + *******************************************************************************/ + @Override + public List postDelete(DeleteInput deleteInput, List records) throws QException + { + invalidate(deleteInput.getTableName()); + return (records); + } + + + + /******************************************************************************* + ** Clear the cache that holds the table's rows, on every node. + *******************************************************************************/ + static void invalidate(String tableName) + { + if(ObjectStorageTables.BUCKET.equals(tableName)) + { + ClusterCaches.invalidate(ConfigSync.ROW_LIMITS_CACHE_NAME); + } + else if(ObjectStorageTables.HEADER_RULE.equals(tableName)) + { + ClusterCaches.invalidate(DefaultObjectHeaderPolicy.CACHE_NAME); + } + } +} diff --git a/src/main/java/com/kingsrook/qbits/objectstorage/internal/ConfigSync.java b/src/main/java/com/kingsrook/qbits/objectstorage/internal/ConfigSync.java index 425c83c..8c9323c 100644 --- a/src/main/java/com/kingsrook/qbits/objectstorage/internal/ConfigSync.java +++ b/src/main/java/com/kingsrook/qbits/objectstorage/internal/ConfigSync.java @@ -30,6 +30,7 @@ import com.kingsrook.qbits.objectstorage.config.CredentialsMode; import com.kingsrook.qbits.objectstorage.config.StorageBucketConfig; import com.kingsrook.qbits.objectstorage.config.StorageConnectionConfig; import com.kingsrook.qbits.objectstorage.model.ObjectStorageTables; +import com.kingsrook.qqq.backend.core.cluster.ClusterCaches; import com.kingsrook.qqq.backend.core.exceptions.QException; import com.kingsrook.qqq.backend.core.logging.QLogger; import com.kingsrook.qqq.backend.core.model.data.QRecord; @@ -48,10 +49,16 @@ public final class ConfigSync { private static final QLogger LOG = QLogger.getLogger(ConfigSync.class); - private static final Map BUCKET_IDS = new ConcurrentHashMap<>(); - private static final Map CONNECTION_IDS = new ConcurrentHashMap<>(); + /** Cluster cache name of the bucket rows' presign limits (cleared on every node when a bucket row changes). */ + public static final String ROW_LIMITS_CACHE_NAME = "qbit-object-storage.bucket-row-limits"; - private static final Map ROW_LIMITS = new ConcurrentHashMap<>(); + private static final Map BUCKET_IDS = ClusterCaches.nodeLocal(new ConcurrentHashMap<>(), + "bucket alias to row id, written by the config sync; bucket rows cannot be inserted or deleted through the UI or API, so the ids are the same on every node"); + private static final Map CONNECTION_IDS = ClusterCaches.nodeLocal(new ConcurrentHashMap<>(), + "connection name to row id, written by the config sync; connection rows cannot be inserted or deleted through the UI or API, so the ids are the same on every node"); + + private static final Map ROW_LIMITS = ClusterCaches.shared(ROW_LIMITS_CACHE_NAME, new ConcurrentHashMap<>(), + ConfigSync::clearRowLimits, alias -> ConfigSync.ROW_LIMITS.remove(alias)); private static final long ROW_LIMIT_CACHE_MILLIS = 15_000L; private static volatile boolean synced = false; @@ -313,7 +320,8 @@ public final class ConfigSync /******************************************************************************* - ** Forget the cached row limits (after a row edit in a test; they also expire). + ** Forget the cached row limits on this node (a bucket row edit clears them on + ** every node through ConfigRowChangeCustomizer; they also expire). *******************************************************************************/ public static void clearRowLimits() { diff --git a/src/main/java/com/kingsrook/qbits/objectstorage/internal/DefaultObjectHeaderPolicy.java b/src/main/java/com/kingsrook/qbits/objectstorage/internal/DefaultObjectHeaderPolicy.java index 6a20a7f..a93de88 100644 --- a/src/main/java/com/kingsrook/qbits/objectstorage/internal/DefaultObjectHeaderPolicy.java +++ b/src/main/java/com/kingsrook/qbits/objectstorage/internal/DefaultObjectHeaderPolicy.java @@ -23,12 +23,14 @@ package com.kingsrook.qbits.objectstorage.internal; import java.util.Comparator; import java.util.List; +import java.util.concurrent.atomic.AtomicReference; import java.util.regex.Pattern; import com.kingsrook.qbits.objectstorage.api.ObjectHeaderPolicy; import com.kingsrook.qbits.objectstorage.api.ObjectHeaders; import com.kingsrook.qbits.objectstorage.config.BucketRole; import com.kingsrook.qbits.objectstorage.config.StorageBucketConfig; import com.kingsrook.qbits.objectstorage.model.ObjectStorageTables; +import com.kingsrook.qqq.backend.core.cluster.ClusterCaches; import com.kingsrook.qqq.backend.core.logging.QLogger; import com.kingsrook.qqq.backend.core.model.data.QRecord; import com.kingsrook.qqq.backend.core.utils.StringUtils; @@ -44,19 +46,31 @@ public class DefaultObjectHeaderPolicy implements ObjectHeaderPolicy { private static final QLogger LOG = QLogger.getLogger(DefaultObjectHeaderPolicy.class); + /** Cluster cache name of the enabled header rules (cleared on every node when a rule changes). */ + public static final String CACHE_NAME = "qbit-object-storage.header-rules"; + private static final long CACHE_MILLIS = 15_000L; - private static volatile List cachedRules; - private static volatile long cachedAt; + private static final AtomicReference CACHED = ClusterCaches.shared(CACHE_NAME, new AtomicReference<>(), DefaultObjectHeaderPolicy::clearCache, null); /******************************************************************************* - ** Drop cached rules (after the rules change). + ** The enabled rules and when they were read. + *******************************************************************************/ + private record CachedRules(List rules, long readAt) + { + } + + + + /******************************************************************************* + ** Drop the cached rules on this node (a rule edit clears them on every node + ** through ConfigRowChangeCustomizer). *******************************************************************************/ public static void clearCache() { - cachedRules = null; + CACHED.set(null); } @@ -119,11 +133,12 @@ public class DefaultObjectHeaderPolicy implements ObjectHeaderPolicy *******************************************************************************/ private static List rules() { - List rules = cachedRules; - if(rules != null && System.currentTimeMillis() - cachedAt < CACHE_MILLIS) + CachedRules cached = CACHED.get(); + if(cached != null && System.currentTimeMillis() - cached.readAt() < CACHE_MILLIS) { - return (rules); + return (cached.rules()); } + List rules; if(!Db.available(ObjectStorageTables.HEADER_RULE)) { return (List.of()); @@ -133,8 +148,7 @@ public class DefaultObjectHeaderPolicy implements ObjectHeaderPolicy rules = Db.asSystem(() -> Db.query(ObjectStorageTables.HEADER_RULE, "enabled", true)); rules.sort(Comparator.comparing((QRecord r) -> r.getValueInteger("priority") == null ? 100 : r.getValueInteger("priority")) .thenComparing(r -> r.getValueInteger("id") == null ? 0 : r.getValueInteger("id"))); - cachedRules = rules; - cachedAt = System.currentTimeMillis(); + CACHED.set(new CachedRules(rules, System.currentTimeMillis())); return (rules); } catch(Exception e) diff --git a/src/main/java/com/kingsrook/qbits/objectstorage/internal/ObjectStorageRuntime.java b/src/main/java/com/kingsrook/qbits/objectstorage/internal/ObjectStorageRuntime.java index 56ec9bd..afb3354 100644 --- a/src/main/java/com/kingsrook/qbits/objectstorage/internal/ObjectStorageRuntime.java +++ b/src/main/java/com/kingsrook/qbits/objectstorage/internal/ObjectStorageRuntime.java @@ -33,6 +33,7 @@ import com.kingsrook.qbits.objectstorage.api.StorageAlertSink; import com.kingsrook.qbits.objectstorage.api.StorageChangeListener; import com.kingsrook.qbits.objectstorage.config.StorageBucketConfig; import com.kingsrook.qbits.objectstorage.config.StorageConnectionConfig; +import com.kingsrook.qqq.backend.core.cluster.ClusterCaches; /******************************************************************************* @@ -43,9 +44,11 @@ import com.kingsrook.qbits.objectstorage.config.StorageConnectionConfig; *******************************************************************************/ public final class ObjectStorageRuntime { - private static final Map CONNECTIONS = new ConcurrentHashMap<>(); + private static final Map CONNECTIONS = ClusterCaches.nodeLocal(new ConcurrentHashMap<>(), + "this node's S3 clients, built from the code config at first use; row overrides apply when a client is built, the same on every node"); private static final List LISTENERS = new CopyOnWriteArrayList<>(); - private static final Map PROCESS_PERMISSIONS = new ConcurrentHashMap<>(); + private static final Map PROCESS_PERMISSIONS = ClusterCaches.nodeLocal(new ConcurrentHashMap<>(), + "built from the qbit config at startup, the same on every node"); private static volatile ObjectStorageQBitConfig config; private static volatile ObjectHeaderPolicy headerPolicy; diff --git a/src/main/java/com/kingsrook/qbits/objectstorage/internal/ProviderDefaults.java b/src/main/java/com/kingsrook/qbits/objectstorage/internal/ProviderDefaults.java index 71506f5..8da26fb 100644 --- a/src/main/java/com/kingsrook/qbits/objectstorage/internal/ProviderDefaults.java +++ b/src/main/java/com/kingsrook/qbits/objectstorage/internal/ProviderDefaults.java @@ -26,6 +26,7 @@ import java.util.Map; import com.kingsrook.qbits.objectstorage.api.Capability; import com.kingsrook.qbits.objectstorage.api.CapabilityStatus; import com.kingsrook.qbits.objectstorage.config.ProviderType; +import com.kingsrook.qqq.backend.core.cluster.ClusterCaches; import static com.kingsrook.qbits.objectstorage.api.CapabilityStatus.PARTIAL; import static com.kingsrook.qbits.objectstorage.api.CapabilityStatus.SUPPORTED; import static com.kingsrook.qbits.objectstorage.api.CapabilityStatus.UNKNOWN; @@ -39,7 +40,8 @@ import static com.kingsrook.qbits.objectstorage.api.CapabilityStatus.UNSUPPORTED *******************************************************************************/ public final class ProviderDefaults { - private static final Map> TABLE = new EnumMap<>(ProviderType.class); + private static final Map> TABLE = ClusterCaches.nodeLocal(new EnumMap<>(ProviderType.class), + "provider defaults from code, the same on every node"); static { diff --git a/src/main/java/com/kingsrook/qbits/objectstorage/model/ObjectStorageTables.java b/src/main/java/com/kingsrook/qbits/objectstorage/model/ObjectStorageTables.java index 98a8d50..407354a 100644 --- a/src/main/java/com/kingsrook/qbits/objectstorage/model/ObjectStorageTables.java +++ b/src/main/java/com/kingsrook/qbits/objectstorage/model/ObjectStorageTables.java @@ -25,6 +25,7 @@ import java.util.ArrayList; import java.util.List; import java.util.Map; import com.kingsrook.qbits.objectstorage.ObjectStorageQBitConfig; +import com.kingsrook.qbits.objectstorage.internal.ConfigRowChangeCustomizer; import com.kingsrook.qbits.objectstorage.permissions.ObjectStoragePermissionChecker; import com.kingsrook.qqq.backend.core.actions.customizers.TableCustomizers; import com.kingsrook.qqq.backend.core.actions.dashboard.widgets.ChildRecordListRenderer; @@ -564,7 +565,9 @@ public final class ObjectStorageTables .withField(ro("orphanedCount", QFieldType.LONG)) .withField(ro("orphanedBytes", QFieldType.LONG)) .withField(ro("missingCount", QFieldType.LONG)) - .withField(ro("lastReferenceScanDate", QFieldType.DATE_TIME)), KEY_TYPE_BUCKET, "id", false)); + .withField(ro("lastReferenceScanDate", QFieldType.DATE_TIME)), KEY_TYPE_BUCKET, "id", false) + .withCustomizer(TableCustomizers.POST_INSERT_RECORD, new QCodeReference(ConfigRowChangeCustomizer.class)) + .withCustomizer(TableCustomizers.POST_UPDATE_RECORD, new QCodeReference(ConfigRowChangeCustomizer.class))); } @@ -701,7 +704,10 @@ public final class ObjectStorageTables .withField(str("cacheControl", 300)) .withField(str("contentDisposition", 300)) .withField(new QFieldMetaData("priority", QFieldType.INTEGER).withDefaultValue(100)) - .withField(new QFieldMetaData("enabled", QFieldType.BOOLEAN).withDefaultValue(true)), KEY_TYPE_BUCKET, "bucketId", true)); + .withField(new QFieldMetaData("enabled", QFieldType.BOOLEAN).withDefaultValue(true)), KEY_TYPE_BUCKET, "bucketId", true) + .withCustomizer(TableCustomizers.POST_INSERT_RECORD, new QCodeReference(ConfigRowChangeCustomizer.class)) + .withCustomizer(TableCustomizers.POST_UPDATE_RECORD, new QCodeReference(ConfigRowChangeCustomizer.class)) + .withCustomizer(TableCustomizers.POST_DELETE_RECORD, new QCodeReference(ConfigRowChangeCustomizer.class))); } diff --git a/src/test/java/com/kingsrook/qbits/objectstorage/internal/ObjectStorageClusterCachesTest.java b/src/test/java/com/kingsrook/qbits/objectstorage/internal/ObjectStorageClusterCachesTest.java new file mode 100644 index 0000000..b5b0593 --- /dev/null +++ b/src/test/java/com/kingsrook/qbits/objectstorage/internal/ObjectStorageClusterCachesTest.java @@ -0,0 +1,240 @@ +/* + * QQQ - Low-code Application Framework for Engineers. + * Copyright (C) 2021-2026. Kingsrook, LLC + * 651 N Broad St Ste 205 # 6917 | Middletown DE 19709 | United States + * contact@kingsrook.com + * https://github.com/Kingsrook/ + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package com.kingsrook.qbits.objectstorage.internal; + + +import java.util.EnumSet; +import java.util.List; +import java.util.concurrent.CopyOnWriteArrayList; +import java.util.function.Consumer; +import com.kingsrook.qbits.objectstorage.ObjectStorageQBitConfig; +import com.kingsrook.qbits.objectstorage.api.Capability; +import com.kingsrook.qbits.objectstorage.api.CapabilityStatus; +import com.kingsrook.qbits.objectstorage.config.BucketRole; +import com.kingsrook.qbits.objectstorage.config.ProviderType; +import com.kingsrook.qbits.objectstorage.config.StorageBucketConfig; +import com.kingsrook.qbits.objectstorage.config.StorageConnectionConfig; +import com.kingsrook.qbits.objectstorage.model.ObjectStorageTables; +import com.kingsrook.qbits.objectstorage.testsupport.StorageTestSupport; +import com.kingsrook.qqq.backend.core.cluster.ClusterBus; +import com.kingsrook.qqq.backend.core.cluster.ClusterBusInterface; +import com.kingsrook.qqq.backend.core.cluster.ClusterBusMetaData; +import com.kingsrook.qqq.backend.core.cluster.ClusterCacheAudit; +import com.kingsrook.qqq.backend.core.cluster.ClusterEvent; +import com.kingsrook.qqq.backend.core.cluster.QNodeRoles; +import com.kingsrook.qqq.backend.core.context.QContext; +import com.kingsrook.qqq.backend.core.model.data.QRecord; +import com.kingsrook.qqq.backend.core.model.metadata.code.QCodeReference; +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; + + +/******************************************************************************* + ** The Object Storage caches are correct on several nodes: every static cache + ** is registered with ClusterCaches, a probe result recorded on one node clears + ** that capability on the others, and an edited bucket row or header rule + ** clears its cache on every node. + *******************************************************************************/ +class ObjectStorageClusterCachesTest +{ + /** the "wire": events every node's transport sees */ + static final List> WIRE = new CopyOnWriteArrayList<>(); + + + + /******************************************************************************* + ** A transport over the static WIRE, so a "remote" publish reaches us. + *******************************************************************************/ + public static class WireBus implements ClusterBusInterface + { + /******************************************************************************* + ** + *******************************************************************************/ + @Override + public void publish(ClusterEvent event) + { + WIRE.forEach(listener -> listener.accept(event)); + } + + + + /******************************************************************************* + ** + *******************************************************************************/ + @Override + public AutoCloseable subscribe(Consumer listener) + { + WIRE.add(listener); + return (() -> WIRE.remove(listener)); + } + } + + + + /******************************************************************************* + ** A synced instance whose buckets never touch a real store, on the wire as + ** node-a. + *******************************************************************************/ + @BeforeEach + void setUp() throws Exception + { + System.setProperty("os.unit.ak", "ak"); + System.setProperty("os.unit.sk", "sk"); + StorageTestSupport.setUp(new ObjectStorageQBitConfig().withBackendName(StorageTestSupport.MEMORY).withBucketScoping(false).withOpenWhenAnonymous(true).withAdminApp(true) + .withAllowedSecretRefPatterns(List.of("prop.os.unit.*")) + .withConnection(new StorageConnectionConfig("main").withProviderType(ProviderType.MINIO).withEndpoint("http://localhost:1").withRegion("us-east-1") + .withAccessKeyRef("${prop.os.unit.ak}").withSecretKeyRef("${prop.os.unit.sk}")) + .withBucket(new StorageBucketConfig("media").withConnection("main").withBucketName("media-bucket").withRole(BucketRole.PRIVATE).withIndexed(false)), + instance -> instance.withSupplementalMetaData(new ClusterBusMetaData().withBus(new QCodeReference(WireBus.class)))); + QNodeRoles.set(new QNodeRoles.Settings(EnumSet.allOf(QNodeRoles.Role.class), true, true, true, null, "node-a")); + ClusterBus.connect(QContext.getQInstance()); + } + + + + /******************************************************************************* + ** + *******************************************************************************/ + @AfterEach + void tearDown() + { + ClusterBus.close(QContext.getQInstance()); + WIRE.clear(); + QNodeRoles.set(null); + StorageTestSupport.tearDown(); + } + + + + /******************************************************************************* + ** Every static Memoization in the QBit is registered (shared or node-local). + *******************************************************************************/ + @Test + void testNoUnregisteredStaticCaches() throws Exception + { + assertThat(ClusterCacheAudit.unregisteredStaticCaches("com.kingsrook.qbits.objectstorage")).isEmpty(); + } + + + + /******************************************************************************* + ** Another node's probe result clears the capability here; the next read + ** reloads it from the capability row. + *******************************************************************************/ + @Test + void testRemoteProbeResultClearsCapability() throws Exception + { + StorageBucketConfig bucket = ObjectStorageRuntime.bucketConfig("media"); + CapabilityStore.record(bucket, Capability.VERSIONING, CapabilityStatus.UNSUPPORTED, "probe here"); + assertThat(CapabilityStore.isCached(bucket, Capability.VERSIONING)).isTrue(); + + //////////////////////////////////////////////////////////// + // node-b probes again and writes SUPPORTED to the row // + //////////////////////////////////////////////////////////// + Integer bucketId = ConfigSync.bucketId("media"); + Db.asSystem(() -> + { + QRecord row = Db.first(ObjectStorageTables.BUCKET_CAPABILITY, "bucketId", bucketId, "capability", Capability.VERSIONING.name()); + Db.update(ObjectStorageTables.BUCKET_CAPABILITY, List.of(new QRecord().withValue("id", row.getValue("id")).withValue("status", CapabilityStatus.SUPPORTED.name()))); + return (null); + }); + ClusterEvent remote = new ClusterEvent(ClusterEvent.TYPE_MEMO_INVALIDATE, CapabilityStore.CACHE_NAME, "b:media:" + Capability.VERSIONING); + remote.setOriginNodeId("node-b"); + WIRE.forEach(listener -> listener.accept(remote)); + + assertThat(CapabilityStore.isCached(bucket, Capability.VERSIONING)).isFalse(); + assertThat(CapabilityStore.status("media", Capability.VERSIONING)).isEqualTo(CapabilityStatus.SUPPORTED); + } + + + + /******************************************************************************* + ** A probe result recorded here tells the other nodes, and stays cached here. + *******************************************************************************/ + @Test + void testProbeResultPublishesInvalidation() + { + List sent = sentEvents(); + StorageBucketConfig bucket = ObjectStorageRuntime.bucketConfig("media"); + CapabilityStore.record(bucket, Capability.VERSIONING, CapabilityStatus.SUPPORTED, "probe here"); + + assertThat(CapabilityStore.status("media", Capability.VERSIONING)).isEqualTo(CapabilityStatus.SUPPORTED); + assertThat(sent).anySatisfy(event -> assertInvalidation(event, CapabilityStore.CACHE_NAME, "b:media:VERSIONING")); + } + + + + /******************************************************************************* + ** An edited bucket row takes effect at once and tells the other nodes. + *******************************************************************************/ + @Test + void testBucketRowEditInvalidatesRowLimitsEverywhere() throws Exception + { + StorageBucketConfig bucket = ObjectStorageRuntime.bucketConfig("media"); + assertThat(ConfigSync.effectiveMaxPresignGetSeconds(bucket)).isEqualTo(3600); + + List sent = sentEvents(); + Db.update(ObjectStorageTables.BUCKET, List.of(new QRecord().withValue("id", ConfigSync.bucketId("media")).withValue("maxPresignGetSeconds", 60))); + + assertThat(ConfigSync.effectiveMaxPresignGetSeconds(bucket)).as("no wait for the 15 s cache").isEqualTo(60); + assertThat(sent).anySatisfy(event -> assertInvalidation(event, ConfigSync.ROW_LIMITS_CACHE_NAME, null)); + } + + + + /******************************************************************************* + ** An edited header rule tells the other nodes. + *******************************************************************************/ + @Test + void testHeaderRuleEditInvalidatesRulesEverywhere() throws Exception + { + List sent = sentEvents(); + StorageTestSupport.insert(ObjectStorageTables.HEADER_RULE, new QRecord().withValue("name", "all").withValue("keyGlob", "**").withValue("cacheControl", "no-store")); + assertThat(sent).anySatisfy(event -> assertInvalidation(event, DefaultObjectHeaderPolicy.CACHE_NAME, null)); + } + + + + /******************************************************************************* + ** Collect what this node puts on the wire. + *******************************************************************************/ + private static List sentEvents() + { + List sent = new CopyOnWriteArrayList<>(); + WIRE.add(sent::add); + return (sent); + } + + + + /******************************************************************************* + ** A memo invalidation from node-a for a cache (and key). + *******************************************************************************/ + private static void assertInvalidation(ClusterEvent event, String name, String key) + { + assertThat(event.getType()).isEqualTo(ClusterEvent.TYPE_MEMO_INVALIDATE); + assertThat(event.getName()).isEqualTo(name); + assertThat(event.getKey()).isEqualTo(key); + assertThat(event.getOriginNodeId()).isEqualTo("node-a"); + } +}