diff --git a/common/src/main/kotlin/gg/grounds/config/internal/sync/BootstrapCoordinator.kt b/common/src/main/kotlin/gg/grounds/config/internal/sync/BootstrapCoordinator.kt index ffec174..f46c63a 100644 --- a/common/src/main/kotlin/gg/grounds/config/internal/sync/BootstrapCoordinator.kt +++ b/common/src/main/kotlin/gg/grounds/config/internal/sync/BootstrapCoordinator.kt @@ -36,6 +36,7 @@ internal class BootstrapCoordinator( syncDefault(client, scope, binding) null } catch (error: Exception) { + refreshScheduler.requireDefaultSync(scope) error } refreshScheduler.subscribeToChanges(listener, scope) diff --git a/common/src/main/kotlin/gg/grounds/config/internal/sync/ConfigScopeSynchronizer.kt b/common/src/main/kotlin/gg/grounds/config/internal/sync/ConfigScopeSynchronizer.kt index a8a9be8..b1d452d 100644 --- a/common/src/main/kotlin/gg/grounds/config/internal/sync/ConfigScopeSynchronizer.kt +++ b/common/src/main/kotlin/gg/grounds/config/internal/sync/ConfigScopeSynchronizer.kt @@ -58,6 +58,7 @@ internal class ConfigScopeSynchronizer( private val refreshScheduler = RefreshScheduler( logger = logger, + objectMapper = objectMapper, snapshotApplier = snapshotApplier, refreshExecutorFactory = refreshExecutorFactory, refreshWorkerExecutorFactory = refreshWorkerExecutorFactory, diff --git a/common/src/main/kotlin/gg/grounds/config/internal/sync/RefreshScheduler.kt b/common/src/main/kotlin/gg/grounds/config/internal/sync/RefreshScheduler.kt index 6759741..d898fe3 100644 --- a/common/src/main/kotlin/gg/grounds/config/internal/sync/RefreshScheduler.kt +++ b/common/src/main/kotlin/gg/grounds/config/internal/sync/RefreshScheduler.kt @@ -1,6 +1,8 @@ package gg.grounds.config.internal.sync +import gg.grounds.config.client.ConfigDefaultData import gg.grounds.config.client.SnapshotResult +import gg.grounds.config.internal.binding.ConfigBinding import gg.grounds.config.internal.scope.AppEnvScope import gg.grounds.config.nats.ConfigChangeListener import java.util.concurrent.ConcurrentHashMap @@ -9,9 +11,11 @@ import java.util.concurrent.ScheduledExecutorService import java.util.concurrent.ScheduledFuture import java.util.concurrent.TimeUnit import org.slf4j.Logger +import tools.jackson.databind.ObjectMapper internal class RefreshScheduler( private val logger: Logger, + private val objectMapper: ObjectMapper, private val snapshotApplier: SnapshotApplier, private val refreshExecutorFactory: () -> ScheduledExecutorService, private val refreshWorkerExecutorFactory: () -> ExecutorService, @@ -20,6 +24,7 @@ internal class RefreshScheduler( private val withLifecycleReadLock: ((() -> Unit) -> Unit), ) { private val trackedScopes = ConcurrentHashMap.newKeySet() + private val scopesRequiringDefaultSync = ConcurrentHashMap.newKeySet() private var refreshExecutor: ScheduledExecutorService? = null private var refreshFuture: ScheduledFuture<*>? = null private var refreshWorkerExecutor: ExecutorService? = null @@ -42,6 +47,7 @@ internal class RefreshScheduler( refreshWorkerExecutor?.shutdownNow() refreshWorkerExecutor = null trackedScopes.clear() + scopesRequiringDefaultSync.clear() } fun close() { @@ -53,6 +59,10 @@ internal class RefreshScheduler( trackedScopes.add(scope) } + fun requireDefaultSync(scope: AppEnvScope) { + scopesRequiringDefaultSync.add(scope) + } + fun subscribeToChanges(listener: ConfigChangeListener, scope: AppEnvScope) { if (!scope.markSubscriptionStarted()) { return @@ -91,6 +101,10 @@ internal class RefreshScheduler( forceFullSnapshot: Boolean, ) { val requiresFullSnapshot = forceFullSnapshot || scope.hasUninitializedBindings() + if (scopesRequiringDefaultSync.contains(scope)) { + syncUninitializedDefaults(client, scope) + scopesRequiringDefaultSync.remove(scope) + } val response = if (requiresFullSnapshot) { executeRetryableCall( @@ -116,6 +130,34 @@ internal class RefreshScheduler( applyResponse(scope, response) } + private fun syncUninitializedDefaults( + client: gg.grounds.config.client.ConfigSyncClient, + scope: AppEnvScope, + ) { + val defaults = + scope.bindingsSnapshot().values.filterNot(ConfigBinding<*>::initialized).map { binding + -> + ConfigDefaultData( + namespace = binding.definition.namespace, + configKey = binding.definition.key, + defaultContentJson = + objectMapper.writeValueAsString(binding.definition.defaultValue), + ) + } + if (defaults.isEmpty()) { + return + } + executeRetryableCall( + logger = logger, + scope = scope, + operation = "sync_defaults", + maxAttempts = REFRESH_MAX_ATTEMPTS, + sleepMillis = sleepMillis, + ) { + client.syncDefaults(scope.app, scope.env, defaults) + } + } + private fun applyResponse(scope: AppEnvScope, response: SnapshotResult) { if (response.changed) { snapshotApplier.applySnapshot(scope, response.version, response.documents) diff --git a/common/src/test/kotlin/gg/grounds/config/internal/sync/ConfigScopeSynchronizerTest.kt b/common/src/test/kotlin/gg/grounds/config/internal/sync/ConfigScopeSynchronizerTest.kt index e8a8103..3cae16a 100644 --- a/common/src/test/kotlin/gg/grounds/config/internal/sync/ConfigScopeSynchronizerTest.kt +++ b/common/src/test/kotlin/gg/grounds/config/internal/sync/ConfigScopeSynchronizerTest.kt @@ -23,6 +23,7 @@ import java.util.concurrent.Executors import java.util.concurrent.Future import java.util.concurrent.ScheduledExecutorService import java.util.concurrent.TimeUnit +import java.util.concurrent.atomic.AtomicBoolean import java.util.concurrent.atomic.AtomicInteger import kotlin.io.path.createTempDirectory import kotlin.test.Test @@ -439,6 +440,67 @@ class ConfigScopeSynchronizerTest { } } + @Test + fun `periodic refresh seeds an uninitialized degraded binding after the service recovers`() { + val serviceAvailable = AtomicBoolean(false) + val refreshCompleted = CountDownLatch(1) + var onChangeReceived: (() -> Unit)? = null + val client = + RecordingConfigSyncClient( + syncDefaultsHandler = { + if (!serviceAvailable.get()) { + throw ConfigServiceException("service unavailable", 503) + } + SyncDefaultsResult(version = 1, createdKeys = emptyList()) + }, + getSnapshotHandler = { + check(syncDefaultCalls >= 4) { "defaults were not re-synced" } + refreshCompleted.countDown() + defaultSnapshotResponse(version = 1, value = "recovered") + }, + ) + val executors = mutableListOf() + val synchronizer = + ConfigScopeSynchronizer( + logger = LoggerFactory.getLogger("ConfigScopeSynchronizerRecoveryTest"), + configClientFactory = { client }, + natsListenerFactory = { + RecordingConfigChangeListener { callback -> onChangeReceived = callback } + }, + refreshExecutorFactory = { + Executors.newSingleThreadScheduledExecutor().also { executor -> + executors += executor + } + }, + sleepMillis = {}, + ) + val scope = AppEnvScope(app = "test-app", env = "dev") + val binding = ConfigBinding(TestStringConfig) + scope.putBindingIfAbsent( + ConfigKey(TestStringConfig.namespace, TestStringConfig.key), + binding, + ) + + try { + synchronizer.start("dns:///config", "nats://localhost:4222") + assertEquals( + ConfigRegistrationStatus.NOT_READY, + synchronizer.bootstrap(scope, binding, ConfigStartupMode.DEGRADED).status, + ) + + serviceAvailable.set(true) + checkNotNull(onChangeReceived).invoke() + + assertTrue(refreshCompleted.await(1, TimeUnit.SECONDS)) + assertTrue(binding.initialized()) + assertEquals("recovered", binding.get()) + assertEquals(4, client.syncDefaultCalls) + } finally { + synchronizer.close() + executors.forEach { executor -> executor.shutdownNow() } + } + } + @Test fun `bootstrap leaves binding uninitialized when snapshot is missing document`() { val client =