From 2adacdaa8b7494103e0a5e5989147a5c5124e423 Mon Sep 17 00:00:00 2001 From: delchev Date: Thu, 10 Sep 2026 16:47:27 +0300 Subject: [PATCH] synchronizers: a transient start failure is FAILED and retried, never FATAL (#7248) A .listener whose subscription the embedded broker refused while it was still taking its store lease was retried once in the same pass, promoted to FATAL on the next pass and stripped from synchronization for good, so every message on that topic was silently lost until the file's bytes changed. The same FAILED-then-FATAL promotion sat in JobSynchronizer, CamelSynchronizer (within the very first pass, via the in-pass retry) and SchemasSynchronizer (an unresolved data source). All four now record a failed start as FAILED with its cause, return false so the in-pass retry has it, and retry it on later passes, healing to CREATED on success; the gate includes FAILED because the multitenant wrapper resets only the lifecycle between tenants. ListenersManager throws for a missing handler instead of returning normally as CREATED and running. The processor ran its phases only on a pass carrying a NEW or MODIFIED artefact, so an idle instance never retried anything; it now gives every FAILED artefact one START attempt on such a pass and, every DIRIGIBLE_SYNCHRONIZER_FAILED_RETRY_INTERVAL_SECONDS (30), on an idle instance, without touching isSynchronizationNeeded, which the test framework waits on. A repeat of the same error logs at DEBUG. Verified: ListenerArtefactSubscriptionRetryIT (new, fails without the processor change), the new *SynchronizerRetryTest unit tests, the unit suites of the six touched modules, formatter:validate with the cache wiped, and the release javadoc build on commons-config and core-initializers. Co-Authored-By: Claude Fable 5.1 --- .claude/docs/synchronizer-model.md | 3 +- .../SynchronizationProcessor.java | 121 +++++++++-- .../synchronizer/SchemasSynchronizer.java | 10 +- .../SchemasSynchronizerRetryTest.java | 110 ++++++++++ .../camel/synchronizer/CamelSynchronizer.java | 11 +- .../CamelSynchronizerRetryTest.java | 144 +++++++++++++ .../jobs/synchronizer/JobSynchronizer.java | 29 ++- .../JobSynchronizerRetryTest.java | 167 +++++++++++++++ .../listeners/service/ListenersManager.java | 7 +- .../synchronizer/ListenerSynchronizer.java | 27 ++- .../service/ListenersManagerTest.java | 5 +- .../ListenerSynchronizerRetryTest.java | 181 ++++++++++++++++ .../commons/config/DirigibleConfig.java | 7 + .../ListenerArtefactSubscriptionRetryIT.java | 195 ++++++++++++++++++ 14 files changed, 962 insertions(+), 55 deletions(-) create mode 100644 components/data/data-structures/src/test/java/org/eclipse/dirigible/components/data/structures/synchronizer/SchemasSynchronizerRetryTest.java create mode 100644 components/engine/engine-camel/src/test/java/org/eclipse/dirigible/components/engine/camel/synchronizer/CamelSynchronizerRetryTest.java create mode 100644 components/engine/engine-jobs/src/test/java/org/eclipse/dirigible/components/jobs/synchronizer/JobSynchronizerRetryTest.java create mode 100644 components/engine/engine-listeners/src/test/java/org/eclipse/dirigible/components/listeners/synchronizer/ListenerSynchronizerRetryTest.java create mode 100644 tests/tests-integrations/src/main/java/org/eclipse/dirigible/integration/tests/api/ListenerArtefactSubscriptionRetryIT.java diff --git a/.claude/docs/synchronizer-model.md b/.claude/docs/synchronizer-model.md index 67b8fc06b3e..d3c5ff67011 100644 --- a/.claude/docs/synchronizer-model.md +++ b/.claude/docs/synchronizer-model.md @@ -12,5 +12,6 @@ Existing synchronizer implementations (grep `extends BaseSynchronizer` / `extend JS/TS user code is **not** synchronized — it is loaded on demand by `engine-javascript` (`JavascriptEndpoint` at `/services/js/...`, `/public/js/...`) via `DirigibleJavascriptCodeRunner` backed by Graalium/GraalVM polyglot. The `api-*` Java modules under `components/api/` register the JS-callable APIs (`@dirigible/db`, `@dirigible/http`, etc.) into the GraalJS context — pre-built TS/JS bundles for those APIs live in `components/api/api-modules-javascript/src/main/resources/META-INF/dirigible/modules/`. - **A pass only runs when something says the registry changed — and the file-system watcher is shallow.** `SynchronizationProcessor.processSynchronizers()` returns immediately unless `SynchronizationWatcher.isModified()`, and that watcher registers **`/registry/public` itself, non-recursively**: a write several folders deep produces no event on Linux at all. A publish is fine — `SynchronizationWatcherPublisherHandler.afterPublish` forces a pass — but anything else that writes the registry must do the same, and must additionally bracket the write with `RegistryMutationTracker` so a pass looking into a half-applied copy does not reap artefacts whose sources are still arriving. `RecursiveFolderWatcher` (`DIRIGIBLE_REGISTRY_EXTERNAL_FOLDER`) did neither, so a boot that lost the race against the copy installed a partial client-Java generation that was never rebuilt — controllers answering 404 for the life of the process with all their sources on disk (#7192). `RegistryMutationTracker`'s javadoc states the rule: bracket the write, do not enumerate callers. + +**A failure of a collaborator records `FAILED` and is retried; `FATAL` is not a synchronizer's answer to it** (#7248). A `.listener` whose subscription the embedded broker refused while it was still taking its store lease, a `.job` the scheduler could not schedule yet, a `.camel` route missing a bean, a `.schema` naming a `.datasource` published on a later pass - each of these is transient, and each of the four synchronizers used to promote the second failure to `FATAL`, after which `SynchronizationProcessor.parseDefinitions` strips the artefact from every later pass until the file's bytes change (a topic subscription lost to a boot race stayed lost, with every message discarded). The shape now, in all four: a failed start registers `FAILED` **with its cause** (never `CREATED` while nothing runs) and returns `false`, which hands the artefact to the in-pass cross-retry loop (`DIRIGIBLE_SYNCHRONIZER_CROSS_RETRY_COUNT` x `_INTERVAL_MILLIS`, so a permanently failing artefact costs that loop every pass - the same cost a `FAILED` view carries since #6942); the `START` phase retries it and heals it to `CREATED` on success. **A pass runs its phases only when it carries a `NEW`/`MODIFIED` artefact** (`isSynchronizationNeeded` answers "the registry changed", and the test framework waits on that answer for a quiet period - do not widen it), so the processor additionally gives every `FAILED` artefact ONE `START` attempt per pass on an idle instance every `DIRIGIBLE_SYNCHRONIZER_FAILED_RETRY_INTERVAL_SECONDS` (30) via a private gate (`isFailedRetryDue` / `retryFailed`) - no cross-retry loop, and a repeat of the same error logs at DEBUG rather than stack-tracing at ERROR on every attempt. Two multitenant details: `MultitenantBaseSynchronizer` completes one shared artefact once per tenant and resets only the lifecycle between tenants, so a retry gate must include `FAILED` and not rely on `running` alone (the first tenant to subscribe flips it and would skip the rest), and the managers' idempotence per tenant (`ListenersManager.LISTENERS.containsKey`, `JobsManager.scheduleJob`'s `checkExists`) is what makes the repeated attempt safe. `ListenersManager.startListener` throws for a missing handler rather than returning normally - a normal return read as CREATED and running with nothing subscribed. Artefacts already persisted as `FATAL` by an earlier version stay stripped until republished; `case BROKEN:` in `parseDefinitions` is the same philosophy one level down (a definition that failed to parse is re-parsed every pass). diff --git a/components/core/core-initializers/src/main/java/org/eclipse/dirigible/components/initializers/synchronizer/SynchronizationProcessor.java b/components/core/core-initializers/src/main/java/org/eclipse/dirigible/components/initializers/synchronizer/SynchronizationProcessor.java index b5503bc2b31..8a74b839567 100644 --- a/components/core/core-initializers/src/main/java/org/eclipse/dirigible/components/initializers/synchronizer/SynchronizationProcessor.java +++ b/components/core/core-initializers/src/main/java/org/eclipse/dirigible/components/initializers/synchronizer/SynchronizationProcessor.java @@ -44,7 +44,9 @@ import java.nio.file.attribute.BasicFileAttributes; import java.text.ParseException; import java.util.*; +import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicLong; import java.util.stream.Collectors; /** @@ -91,6 +93,12 @@ public class SynchronizationProcessor implements SynchronizationWalkerCallback, /** The processing. */ private final AtomicBoolean processing; + /** + * When the FAILED artefacts left by the last pass are due one more START attempt (#7248), or -1 + * when the last pass left none. JVM-local on purpose: the boot pass processes everything anyway. + */ + private final AtomicLong failedRetryDueAt = new AtomicLong(-1); + /** * Instantiates a new synchronization processor. * @@ -196,7 +204,7 @@ public void forceProcessSynchronizers() { */ public void processSynchronizers() { - if (!isSynchronizationNeeded()) { + if (!isSynchronizationNeeded() && !isFailedRetryDue()) { logger.debug("Skipping synchronization since it is not needed..."); return; } @@ -236,6 +244,7 @@ public void processSynchronizers() { int countNew = 0; int countModified = 0; + int countFailed = 0; for (Artefact artefact : artefacts.values()) { if (ArtefactLifecycle.NEW.equals(artefact.getLifecycle())) { logger.debug("Processing a new artefact: {}", artefact.getKey()); @@ -245,6 +254,9 @@ public void processSynchronizers() { logger.debug("Processing a modified artefact: {}", artefact.getKey()); countModified++; } + if (ArtefactLifecycle.FAILED.equals(artefact.getLifecycle())) { + countFailed++; + } } logger.debug("Loading of [{}] definitions done. [{}] artefacts found in total. [{}] new and [{}] modified.", definitions.size(), @@ -394,6 +406,8 @@ public void processSynchronizers() { logger.trace("Processing of artefacts done."); + } else if (countFailed > 0) { + retryFailed(); } logger.trace("Cleaning up removed artefacts..."); @@ -447,25 +461,23 @@ public void processSynchronizers() { logger.info("Processing synchronizers completed!"); } finally { - if (logger.isDebugEnabled()) { - int countCreated = 0; - int countUpdated = 0; - int countFailed = 0; - int countFatal = 0; - for (Artefact artefact : artefacts.values()) { - if (ArtefactLifecycle.CREATED.equals(artefact.getLifecycle())) - countCreated++; - if (ArtefactLifecycle.UPDATED.equals(artefact.getLifecycle())) - countUpdated++; - if (ArtefactLifecycle.FAILED.equals(artefact.getLifecycle())) - countFailed++; - if (ArtefactLifecycle.FATAL.equals(artefact.getLifecycle())) - countFatal++; - } - logger.debug( - "Processing synchronizers done. {} artefacts processed in total. {} ({}/{}) successful, {} failed and {} fatal.", - artefacts.size(), countCreated + countUpdated, countCreated, countUpdated, countFailed, countFatal); + int countCreated = 0; + int countUpdated = 0; + int countFailed = 0; + int countFatal = 0; + for (Artefact artefact : artefacts.values()) { + if (ArtefactLifecycle.CREATED.equals(artefact.getLifecycle())) + countCreated++; + if (ArtefactLifecycle.UPDATED.equals(artefact.getLifecycle())) + countUpdated++; + if (ArtefactLifecycle.FAILED.equals(artefact.getLifecycle())) + countFailed++; + if (ArtefactLifecycle.FATAL.equals(artefact.getLifecycle())) + countFatal++; } + logger.debug("Processing synchronizers done. {} artefacts processed in total. {} ({}/{}) successful, {} failed and {} fatal.", + artefacts.size(), countCreated + countUpdated, countCreated, countUpdated, countFailed, countFatal); + scheduleFailedRetry(countFailed); // clear maps definitions.clear(); artefacts.clear(); @@ -502,6 +514,70 @@ public boolean isSynchronizationNeeded() { return true; } + /** + * Whether the FAILED artefacts the last pass left are due one more START attempt (#7248). Kept out + * of {@link #isSynchronizationNeeded()} on purpose: that answer means "the registry changed", and + * the test framework waits on it for a quiet period. + * + * @return true when a retry pass should run now + */ + private boolean isFailedRetryDue() { + long dueAt = failedRetryDueAt.get(); + return dueAt >= 0 && System.currentTimeMillis() >= dueAt && initialized.get() && prepared.get() && !processing.get(); + } + + /** + * Arms the next FAILED retry after a pass, or disarms it when the pass left nothing FAILED. + * + * @param countFailed the artefacts the pass left FAILED + */ + private void scheduleFailedRetry(int countFailed) { + if (countFailed == 0) { + failedRetryDueAt.set(-1); + return; + } + long intervalMillis = TimeUnit.SECONDS.toMillis(DirigibleConfig.SYNCHRONIZER_FAILED_RETRY_INTERVAL_SECONDS.getIntValue()); + failedRetryDueAt.set(System.currentTimeMillis() + intervalMillis); + } + + /** + * A pass with nothing new or modified gives every FAILED artefact one START attempt (#7248). A + * start refused by a collaborator that was not ready yet - the embedded broker still taking its + * store lease, the scheduler's store, a handler published later - heals only by being attempted + * again, and the phases otherwise run only on a pass carrying a change, so an idle instance never + * retried it. One attempt per artefact per pass and no cross-retry loop: a permanently failing + * artefact costs one refused call per pass, not the loop. The synchronizer records the outcome + * itself, so the error stays the cause rather than the undepleted rewrite, which is what lets a + * repeat log quietly. + */ + private void retryFailed() { + TopologicalDepleter> depleter = new TopologicalDepleter<>(); + List failed = artefacts.values() + .stream() + .filter(a -> ArtefactLifecycle.FAILED.equals(a.getLifecycle())) + .collect(Collectors.toList()); + List> wrappers = TopologyFactory.wrap(failed, synchronizers); + logger.info("Retrying [{}] FAILED artefacts: [{}]", wrappers.size(), wrappers); + for (Synchronizer synchronizer : synchronizers) { + Set> own = wrappers.stream() + .filter(w -> w.getSynchronizer() + .equals(synchronizer)) + .collect(Collectors.toSet()); + if (own.isEmpty()) { + continue; + } + try { + Set> left = depleter.deplete(own, ArtefactPhase.START); + if (!left.isEmpty()) { + logger.warn("[{}] artefacts are still FAILED after the retry: [{}]", left.size(), left); + } + } catch (Exception e) { + logger.error("Error occurred while retrying FAILED artefacts of [{}]", synchronizer, e); + addError(e.getMessage()); + } + } + } + public boolean isSynchronizationRunning() { return processing.get(); } @@ -953,6 +1029,13 @@ public void registerState(Synchronizer synchronizer, Arte case FAILED: case FATAL: + if (message != null && message.equals(artefact.getError())) { + // The same failure again - a FAILED artefact is retried every pass (#7248), and a + // permanently refused one must not stack-trace at ERROR on each of them. + logger.debug("Processing of artefact with key [{}], location [{}] for lifecycle [{}] failed again with [{}]", + artefact.getKey(), artefact.getLocation(), lifecycle, message, cause); + break; + } logger.error( "Processing of artefact with key [{}], location [{}] for lifecycle [{}] has failed, synchronizer [{}], error [{}], message [{}]", artefact.getKey(), artefact.getLocation(), lifecycle, synchronizer, artefact.getError(), message, cause); diff --git a/components/data/data-structures/src/main/java/org/eclipse/dirigible/components/data/structures/synchronizer/SchemasSynchronizer.java b/components/data/data-structures/src/main/java/org/eclipse/dirigible/components/data/structures/synchronizer/SchemasSynchronizer.java index 586165f5c95..d0a9729d0dd 100644 --- a/components/data/data-structures/src/main/java/org/eclipse/dirigible/components/data/structures/synchronizer/SchemasSynchronizer.java +++ b/components/data/data-structures/src/main/java/org/eclipse/dirigible/components/data/structures/synchronizer/SchemasSynchronizer.java @@ -548,13 +548,9 @@ protected boolean completeImpl(TopologyWrapper wrapper, ArtefactPhase fl logger.error(e.getMessage(), e); } - if (dataSource == null) { - if (ArtefactLifecycle.FAILED.equals(schema.getLifecycle())) { - callback.addError(e.getMessage()); - callback.registerState(this, wrapper, ArtefactLifecycle.FATAL, e); - return true; - } - } + // FAILED and retried, never FATAL (#7248): the data source a schema names may be an + // artefact of a project published on a later pass, and FATAL stripped the schema from + // every later pass until its file's bytes changed. callback.addError(e.getMessage()); callback.registerState(this, wrapper, ArtefactLifecycle.FAILED, e); return false; diff --git a/components/data/data-structures/src/test/java/org/eclipse/dirigible/components/data/structures/synchronizer/SchemasSynchronizerRetryTest.java b/components/data/data-structures/src/test/java/org/eclipse/dirigible/components/data/structures/synchronizer/SchemasSynchronizerRetryTest.java new file mode 100644 index 00000000000..f29b0bcaa03 --- /dev/null +++ b/components/data/data-structures/src/test/java/org/eclipse/dirigible/components/data/structures/synchronizer/SchemasSynchronizerRetryTest.java @@ -0,0 +1,110 @@ +/* + * Copyright (c) 2010-2026 Eclipse Dirigible contributors + * + * All rights reserved. This program and the accompanying materials are made available under the + * terms of the Eclipse Public License v2.0 which accompanies this distribution, and is available at + * http://www.eclipse.org/legal/epl-v20.html + * + * SPDX-FileCopyrightText: Eclipse Dirigible contributors SPDX-License-Identifier: EPL-2.0 + */ +package org.eclipse.dirigible.components.data.structures.synchronizer; + +import org.eclipse.dirigible.components.base.artefact.ArtefactLifecycle; +import org.eclipse.dirigible.components.base.artefact.ArtefactPhase; +import org.eclipse.dirigible.components.base.artefact.topology.TopologyWrapper; +import org.eclipse.dirigible.components.base.synchronizer.SynchronizerCallback; +import org.eclipse.dirigible.components.data.sources.manager.DataSourcesManager; +import org.eclipse.dirigible.components.data.structures.domain.Schema; +import org.eclipse.dirigible.components.data.structures.service.SchemaService; +import org.eclipse.dirigible.components.data.structures.service.TableService; +import org.eclipse.dirigible.components.data.structures.service.ViewService; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +import java.util.HashMap; +import java.util.Set; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.doAnswer; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +/** + * A schema whose data source cannot be resolved yet is recorded FAILED and retried on every pass - + * never promoted to FATAL on the second attempt (#7248). The data source may be an artefact of a + * project published on a later pass. + */ +class SchemasSynchronizerRetryTest { + + /** The data source the schema names and nothing has published yet. */ + private static final String LATER_DATA_SOURCE = "LaterDB"; + + /** The synchronizer under test. */ + private SchemasSynchronizer synchronizer; + + /** The data sources, refusing the one the schema names. */ + private DataSourcesManager dataSourcesManager; + + /** + * Wires the synchronizer over a data-sources double that knows no such data source and a callback + * that writes the registered state onto the artefact, as the synchronization processor does. + */ + @BeforeEach + void setUp() { + SchemaService schemaService = mock(SchemaService.class); + when(schemaService.save(any())).thenAnswer(invocation -> invocation.getArgument(0)); + dataSourcesManager = mock(DataSourcesManager.class); + when(dataSourcesManager.getDataSource(LATER_DATA_SOURCE)).thenThrow( + new IllegalArgumentException("DataSource [" + LATER_DATA_SOURCE + "] not found")); + synchronizer = + new SchemasSynchronizer(schemaService, dataSourcesManager, mock(TableService.class), mock(ViewService.class), "DefaultDB"); + + SynchronizerCallback callback = mock(SynchronizerCallback.class); + doAnswer(invocation -> { + synchronizer.setStatus(artefact(invocation.getArgument(1)), invocation.getArgument(2), ""); + return null; + }).when(callback) + .registerState(any(), any(TopologyWrapper.class), any(ArtefactLifecycle.class)); + doAnswer(invocation -> { + Throwable cause = invocation.getArgument(3); + synchronizer.setStatus(artefact(invocation.getArgument(1)), invocation.getArgument(2), cause.getMessage()); + return null; + }).when(callback) + .registerState(any(), any(TopologyWrapper.class), any(ArtefactLifecycle.class), any(Throwable.class)); + synchronizer.setCallback(callback); + } + + /** + * The first refusal reads FAILED; the second used to read FATAL, after which the processor stripped + * the schema from every later pass. Both now stay FAILED and not completed, so the pass retries. + */ + @Test + void aSchemaWhoseDataSourceIsNotThereYetStaysFailedAndIsRetried() { + Schema schema = new Schema("/later/later.schema", "later", "", Set.of()); + schema.setDataSource(LATER_DATA_SOURCE); + schema.setLifecycle(ArtefactLifecycle.NEW); + TopologyWrapper wrapper = new TopologyWrapper<>(schema, new HashMap<>(), synchronizer); + + assertFalse(synchronizer.completeImpl(wrapper, ArtefactPhase.CREATE), + "An unresolved data source is not completed - the pass retries it"); + assertEquals(ArtefactLifecycle.FAILED, schema.getLifecycle()); + + assertFalse(synchronizer.completeImpl(wrapper, ArtefactPhase.CREATE), "The retry is refused the same way"); + assertEquals(ArtefactLifecycle.FAILED, schema.getLifecycle(), "A data source published later must never leave the schema FATAL"); + verify(dataSourcesManager, times(2)).getDataSource(LATER_DATA_SOURCE); + } + + /** + * Unwraps the artefact of a wrapper handed to the callback. + * + * @param wrapper the wrapper + * @return the schema + */ + private static Schema artefact(Object wrapper) { + return ((TopologyWrapper) wrapper).getArtefact() instanceof Schema schema ? schema : null; + } +} diff --git a/components/engine/engine-camel/src/main/java/org/eclipse/dirigible/components/engine/camel/synchronizer/CamelSynchronizer.java b/components/engine/engine-camel/src/main/java/org/eclipse/dirigible/components/engine/camel/synchronizer/CamelSynchronizer.java index 22d115cbdf8..fef39b50557 100644 --- a/components/engine/engine-camel/src/main/java/org/eclipse/dirigible/components/engine/camel/synchronizer/CamelSynchronizer.java +++ b/components/engine/engine-camel/src/main/java/org/eclipse/dirigible/components/engine/camel/synchronizer/CamelSynchronizer.java @@ -180,12 +180,11 @@ protected boolean completeImpl(TopologyWrapper wrapper, ArtefactPhase flo } break; case START: { - if (ArtefactLifecycle.FAILED.equals(camel.getLifecycle())) { - String message = "Cannot start a Route in a failing state: " + camel.getKey(); - callback.addError(message); - callback.registerState(this, wrapper, ArtefactLifecycle.FATAL, message); - return true; - } + // A FAILED route is retried, never promoted to FATAL (#7248). The outer catch + // records a failed create as FAILED and hands it to the in-pass retry, whose + // START phase then reached this guard - so a route refused for a transient + // reason went FATAL within the very pass that first saw it, and was stripped + // from every later one until the file's bytes changed. addToProcessor(camel); callback.registerState(this, wrapper, ArtefactLifecycle.CREATED); } diff --git a/components/engine/engine-camel/src/test/java/org/eclipse/dirigible/components/engine/camel/synchronizer/CamelSynchronizerRetryTest.java b/components/engine/engine-camel/src/test/java/org/eclipse/dirigible/components/engine/camel/synchronizer/CamelSynchronizerRetryTest.java new file mode 100644 index 00000000000..d6c20fed3bf --- /dev/null +++ b/components/engine/engine-camel/src/test/java/org/eclipse/dirigible/components/engine/camel/synchronizer/CamelSynchronizerRetryTest.java @@ -0,0 +1,144 @@ +/* + * Copyright (c) 2010-2026 Eclipse Dirigible contributors + * + * All rights reserved. This program and the accompanying materials are made available under the + * terms of the Eclipse Public License v2.0 which accompanies this distribution, and is available at + * http://www.eclipse.org/legal/epl-v20.html + * + * SPDX-FileCopyrightText: Eclipse Dirigible contributors SPDX-License-Identifier: EPL-2.0 + */ +package org.eclipse.dirigible.components.engine.camel.synchronizer; + +import org.eclipse.dirigible.components.base.artefact.ArtefactLifecycle; +import org.eclipse.dirigible.components.base.artefact.ArtefactPhase; +import org.eclipse.dirigible.components.base.artefact.topology.TopologyWrapper; +import org.eclipse.dirigible.components.base.synchronizer.SynchronizerCallback; +import org.eclipse.dirigible.components.engine.camel.domain.Camel; +import org.eclipse.dirigible.components.engine.camel.processor.CamelProcessor; +import org.eclipse.dirigible.components.engine.camel.service.CamelService; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +import java.util.HashMap; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.doAnswer; +import static org.mockito.Mockito.doNothing; +import static org.mockito.Mockito.doThrow; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +/** + * A route the Camel context refuses is recorded FAILED and retried on every pass until it starts - + * never promoted to FATAL (#7248). + */ +class CamelSynchronizerRetryTest { + + /** What the context answers while a collaborator the route needs is not there yet. */ + private static final String ROUTE_REFUSAL = "No bean could be found in the registry for: laterDataSource"; + + /** The synchronizer under test. */ + private CamelSynchronizer synchronizer; + + /** The processor whose route registration the tests make fail or succeed. */ + private CamelProcessor camelProcessor; + + /** + * Wires the synchronizer over a processor double and a callback that writes the registered state + * onto the artefact, as the synchronization processor does. + */ + @BeforeEach + void setUp() { + CamelService camelService = mock(CamelService.class); + when(camelService.save(any())).thenAnswer(invocation -> invocation.getArgument(0)); + camelProcessor = mock(CamelProcessor.class); + synchronizer = new CamelSynchronizer(camelService, camelProcessor); + + SynchronizerCallback callback = mock(SynchronizerCallback.class); + doAnswer(invocation -> { + synchronizer.setStatus(artefact(invocation.getArgument(1)), invocation.getArgument(2), ""); + return null; + }).when(callback) + .registerState(any(), any(TopologyWrapper.class), any(ArtefactLifecycle.class)); + doAnswer(invocation -> { + Throwable cause = invocation.getArgument(3); + synchronizer.setStatus(artefact(invocation.getArgument(1)), invocation.getArgument(2), cause.getMessage()); + return null; + }).when(callback) + .registerState(any(), any(TopologyWrapper.class), any(ArtefactLifecycle.class), any(Throwable.class)); + doAnswer(invocation -> { + synchronizer.setStatus(artefact(invocation.getArgument(1)), invocation.getArgument(2), invocation.getArgument(3)); + return null; + }).when(callback) + .registerState(any(), any(TopologyWrapper.class), any(ArtefactLifecycle.class), any(String.class)); + synchronizer.setCallback(callback); + } + + /** + * A refused create is FAILED and handed to the in-pass retry, whose START phase used to promote it + * to FATAL within the very same pass; it now retries the route, and it stays FAILED while refused. + */ + @Test + void aRefusedRouteStaysFailedAndIsRetriedNotPromotedToFatal() { + Camel camel = camel(ArtefactLifecycle.NEW); + TopologyWrapper wrapper = wrapper(camel); + doThrow(new IllegalArgumentException(ROUTE_REFUSAL)).when(camelProcessor) + .onCreateOrUpdate(camel); + + assertFalse(synchronizer.completeImpl(wrapper, ArtefactPhase.CREATE), "A refused create is not completed - the pass retries it"); + assertEquals(ArtefactLifecycle.FAILED, camel.getLifecycle()); + assertEquals(ROUTE_REFUSAL, camel.getError()); + + assertFalse(synchronizer.completeImpl(wrapper, ArtefactPhase.START), "A refused start is not completed either"); + assertEquals(ArtefactLifecycle.FAILED, camel.getLifecycle(), "A transiently refused route must never read FATAL"); + verify(camelProcessor, times(2)).onCreateOrUpdate(camel); + } + + /** + * Once the context accepts, the very same FAILED artefact starts, reads CREATED and carries no + * stale error. + */ + @Test + void aFailedRouteHealsOnceTheContextAccepts() { + Camel camel = camel(ArtefactLifecycle.FAILED); + camel.setError(ROUTE_REFUSAL); + TopologyWrapper wrapper = wrapper(camel); + doNothing().when(camelProcessor) + .onCreateOrUpdate(camel); + + assertTrue(synchronizer.completeImpl(wrapper, ArtefactPhase.START), "A start the context accepted completes"); + assertEquals(ArtefactLifecycle.CREATED, camel.getLifecycle(), "The healed route must read CREATED"); + assertEquals("", camel.getError(), "The healed route must carry no stale error"); + verify(camelProcessor).onCreateOrUpdate(camel); + } + + private static Camel camel(ArtefactLifecycle lifecycle) { + Camel camel = new Camel(); + camel.setLocation("/retry/probe.camel"); + camel.setName("probe.camel"); + camel.setType(Camel.ARTEFACT_TYPE); + camel.updateKey(); + camel.setContent(""); + camel.setLifecycle(lifecycle); + return camel; + } + + private TopologyWrapper wrapper(Camel camel) { + return new TopologyWrapper<>(camel, new HashMap<>(), synchronizer); + } + + /** + * Unwraps the artefact of a wrapper handed to the callback. + * + * @param wrapper the wrapper + * @return the route + */ + private static Camel artefact(Object wrapper) { + return ((TopologyWrapper) wrapper).getArtefact() instanceof Camel camel ? camel : null; + } +} diff --git a/components/engine/engine-jobs/src/main/java/org/eclipse/dirigible/components/jobs/synchronizer/JobSynchronizer.java b/components/engine/engine-jobs/src/main/java/org/eclipse/dirigible/components/jobs/synchronizer/JobSynchronizer.java index 07c0b689f59..e087d74f272 100644 --- a/components/engine/engine-jobs/src/main/java/org/eclipse/dirigible/components/jobs/synchronizer/JobSynchronizer.java +++ b/components/engine/engine-jobs/src/main/java/org/eclipse/dirigible/components/jobs/synchronizer/JobSynchronizer.java @@ -233,11 +233,12 @@ protected boolean completeImpl(TopologyWrapper wrapper, ArtefactPhase flow) getService().save(job); callback.registerState(this, wrapper, ArtefactLifecycle.CREATED); } catch (Exception e) { - if (logger.isErrorEnabled()) { - logger.error(e.getMessage(), e); - } + // FAILED, not CREATED: nothing was scheduled, and only FAILED keeps the job + // eligible for the START phase of this pass and of every later one (#7248). + // Returning false hands it to the in-pass topological retry as well. callback.addError(e.getMessage()); - callback.registerState(this, wrapper, ArtefactLifecycle.CREATED); + callback.registerState(this, wrapper, ArtefactLifecycle.FAILED, e); + return false; } } break; @@ -275,19 +276,25 @@ protected boolean completeImpl(TopologyWrapper wrapper, ArtefactPhase flow) } break; case START: - if (ArtefactLifecycle.FAILED.equals(job.getLifecycle())) { - String message = "Cannot start a Job in a failing state: " + job.getKey(); - callback.addError(message); - callback.registerState(this, wrapper, ArtefactLifecycle.FATAL, message); - return true; - } - if (job.getRunning() == null || !job.getRunning()) { + // A FAILED job is retried, never promoted to FATAL (#7248): scheduling fails for + // reasons that heal later - the scheduler's store not reachable yet at boot - and + // FATAL stripped the artefact from every later pass, so the job stayed unscheduled + // until the file's bytes changed. Gated on FAILED as well as on running: the artefact + // completes once per tenant on one shared object, and only the lifecycle is reset + // between tenants - so the first tenant to schedule flips running and would otherwise + // skip every tenant after it. + if (ArtefactLifecycle.FAILED.equals(job.getLifecycle()) || job.getRunning() == null || !job.getRunning()) { try { jobsManager.scheduleJob(job); job.setRunning(true); getService().save(job); + if (ArtefactLifecycle.FAILED.equals(job.getLifecycle())) { + callback.registerState(this, wrapper, ArtefactLifecycle.CREATED); + } } catch (Exception e) { + callback.addError(e.getMessage()); callback.registerState(this, wrapper, ArtefactLifecycle.FAILED, e); + return false; } } break; diff --git a/components/engine/engine-jobs/src/test/java/org/eclipse/dirigible/components/jobs/synchronizer/JobSynchronizerRetryTest.java b/components/engine/engine-jobs/src/test/java/org/eclipse/dirigible/components/jobs/synchronizer/JobSynchronizerRetryTest.java new file mode 100644 index 00000000000..7c70cb4f544 --- /dev/null +++ b/components/engine/engine-jobs/src/test/java/org/eclipse/dirigible/components/jobs/synchronizer/JobSynchronizerRetryTest.java @@ -0,0 +1,167 @@ +/* + * Copyright (c) 2010-2026 Eclipse Dirigible contributors + * + * All rights reserved. This program and the accompanying materials are made available under the + * terms of the Eclipse Public License v2.0 which accompanies this distribution, and is available at + * http://www.eclipse.org/legal/epl-v20.html + * + * SPDX-FileCopyrightText: Eclipse Dirigible contributors SPDX-License-Identifier: EPL-2.0 + */ +package org.eclipse.dirigible.components.jobs.synchronizer; + +import org.eclipse.dirigible.components.base.artefact.ArtefactLifecycle; +import org.eclipse.dirigible.components.base.artefact.ArtefactPhase; +import org.eclipse.dirigible.components.base.artefact.topology.TopologyWrapper; +import org.eclipse.dirigible.components.base.synchronizer.SynchronizerCallback; +import org.eclipse.dirigible.components.jobs.domain.Job; +import org.eclipse.dirigible.components.jobs.manager.JobsManager; +import org.eclipse.dirigible.components.jobs.service.JobEmailService; +import org.eclipse.dirigible.components.jobs.service.JobLogService; +import org.eclipse.dirigible.components.jobs.service.JobService; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +import java.util.HashMap; +import java.util.Set; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNotEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.doAnswer; +import static org.mockito.Mockito.doNothing; +import static org.mockito.Mockito.doThrow; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +/** + * A job whose scheduling fails is recorded FAILED and retried on every pass until it schedules - + * never CREATED while nothing runs, never promoted to FATAL (#7248). + */ +class JobSynchronizerRetryTest { + + /** What the scheduler answers while its store is not reachable yet. */ + private static final String SCHEDULER_REFUSAL = "Scheduler store not available"; + + /** The synchronizer under test. */ + private JobSynchronizer synchronizer; + + /** The manager whose scheduling the tests make fail or succeed. */ + private JobsManager jobsManager; + + /** + * Wires the synchronizer over a manager double and a callback that writes the registered state onto + * the artefact, as the synchronization processor does. + */ + @BeforeEach + void setUp() { + JobService jobService = mock(JobService.class); + when(jobService.save(any())).thenAnswer(invocation -> invocation.getArgument(0)); + jobsManager = mock(JobsManager.class); + synchronizer = new JobSynchronizer(jobService, jobsManager, mock(JobEmailService.class), mock(JobLogService.class)); + + SynchronizerCallback callback = mock(SynchronizerCallback.class); + doAnswer(invocation -> { + synchronizer.setStatus(artefact(invocation.getArgument(1)), invocation.getArgument(2), ""); + return null; + }).when(callback) + .registerState(any(), any(TopologyWrapper.class), any(ArtefactLifecycle.class)); + doAnswer(invocation -> { + Throwable cause = invocation.getArgument(3); + synchronizer.setStatus(artefact(invocation.getArgument(1)), invocation.getArgument(2), cause.getMessage()); + return null; + }).when(callback) + .registerState(any(), any(TopologyWrapper.class), any(ArtefactLifecycle.class), any(Throwable.class)); + doAnswer(invocation -> { + synchronizer.setStatus(artefact(invocation.getArgument(1)), invocation.getArgument(2), invocation.getArgument(3)); + return null; + }).when(callback) + .registerState(any(), any(TopologyWrapper.class), any(ArtefactLifecycle.class), any(String.class)); + synchronizer.setCallback(callback); + } + + /** + * The first pass: a failed scheduling is FAILED with its cause and not completed - not CREATED with + * nothing scheduled. + * + * @throws Exception from the manager double + */ + @Test + void aFailedSchedulingIsRecordedFailedNotCreated() throws Exception { + Job job = job(ArtefactLifecycle.NEW); + TopologyWrapper wrapper = wrapper(job); + doThrow(new IllegalStateException(SCHEDULER_REFUSAL)).when(jobsManager) + .scheduleJob(job); + + assertFalse(synchronizer.completeImpl(wrapper, ArtefactPhase.CREATE), "A failed scheduling is not completed - the pass retries it"); + assertEquals(ArtefactLifecycle.FAILED, job.getLifecycle(), "A job that was never scheduled must read FAILED, not CREATED"); + assertEquals(SCHEDULER_REFUSAL, job.getError(), "The recorded error must be the scheduler's refusal"); + assertNotEquals(Boolean.TRUE, job.getRunning(), "Nothing was scheduled, so nothing may read as running"); + } + + /** + * The passes after: a FAILED job the scheduler still refuses stays FAILED and keeps being retried, + * where the second attempt used to register FATAL. + * + * @throws Exception from the manager double + */ + @Test + void aFailedJobTheSchedulerStillRefusesStaysFailedAndIsRetried() throws Exception { + Job job = job(ArtefactLifecycle.FAILED); + TopologyWrapper wrapper = wrapper(job); + doThrow(new IllegalStateException(SCHEDULER_REFUSAL)).when(jobsManager) + .scheduleJob(job); + + assertFalse(synchronizer.completeImpl(wrapper, ArtefactPhase.START)); + assertFalse(synchronizer.completeImpl(wrapper, ArtefactPhase.START)); + + assertEquals(ArtefactLifecycle.FAILED, job.getLifecycle(), "A transiently refused job must never read FATAL"); + assertNotEquals(Boolean.TRUE, job.getRunning(), "Nothing was scheduled, so nothing may read as running"); + verify(jobsManager, times(2)).scheduleJob(job); + } + + /** + * Once the scheduler accepts, the very same FAILED artefact schedules, reads CREATED and carries no + * stale error. + * + * @throws Exception from the manager double + */ + @Test + void aFailedJobHealsOnceTheSchedulerAccepts() throws Exception { + Job job = job(ArtefactLifecycle.FAILED); + job.setError(SCHEDULER_REFUSAL); + TopologyWrapper wrapper = wrapper(job); + doNothing().when(jobsManager) + .scheduleJob(job); + + assertTrue(synchronizer.completeImpl(wrapper, ArtefactPhase.START), "A scheduling the scheduler accepted completes"); + assertEquals(ArtefactLifecycle.CREATED, job.getLifecycle(), "The healed job must read CREATED"); + assertEquals("", job.getError(), "The healed job must carry no stale error"); + assertEquals(Boolean.TRUE, job.getRunning(), "The healed job is running"); + } + + private static Job job(ArtefactLifecycle lifecycle) { + Job job = new Job("/retry/probe.job", "probe", "", Set.of(), "retry", "org.eclipse.dirigible.components.jobs.handler.JobHandler", + "0 0 * * * ?", "retry/handler.js", "javascript", false, true, null, null, null); + job.setLifecycle(lifecycle); + job.setRunning(false); + return job; + } + + private TopologyWrapper wrapper(Job job) { + return new TopologyWrapper<>(job, new HashMap<>(), synchronizer); + } + + /** + * Unwraps the artefact of a wrapper handed to the callback. + * + * @param wrapper the wrapper + * @return the job + */ + private static Job artefact(Object wrapper) { + return ((TopologyWrapper) wrapper).getArtefact() instanceof Job job ? job : null; + } +} diff --git a/components/engine/engine-listeners/src/main/java/org/eclipse/dirigible/components/listeners/service/ListenersManager.java b/components/engine/engine-listeners/src/main/java/org/eclipse/dirigible/components/listeners/service/ListenersManager.java index 202f9e95ffa..444e63cb865 100644 --- a/components/engine/engine-listeners/src/main/java/org/eclipse/dirigible/components/listeners/service/ListenersManager.java +++ b/components/engine/engine-listeners/src/main/java/org/eclipse/dirigible/components/listeners/service/ListenersManager.java @@ -77,8 +77,11 @@ public void startListener(org.eclipse.dirigible.components.listeners.domain.List } if (isMissingHandler(listenerDescriptor)) { - LOGGER.error("Listener {} cannot be started, because the handler does not exist!", listenerDescriptor); - return; + // Thrown, not logged and swallowed: a normal return had the synchronizer record the + // artefact as CREATED and running with nothing subscribed (#7248). The handler may + // simply be published on a later pass, and FAILED is what keeps the listener retried. + throw new IllegalStateException("Listener " + listenerDescriptor + " cannot be started, because the handler [" + + listenerDescriptor.getHandlerPath() + "] does not exist"); } ListenerManager listenerManager = messageListenerManagerFactory.create(listenerDescriptor); listenerManager.startListener(); diff --git a/components/engine/engine-listeners/src/main/java/org/eclipse/dirigible/components/listeners/synchronizer/ListenerSynchronizer.java b/components/engine/engine-listeners/src/main/java/org/eclipse/dirigible/components/listeners/synchronizer/ListenerSynchronizer.java index 063c70f793d..a6d3d78b4b3 100644 --- a/components/engine/engine-listeners/src/main/java/org/eclipse/dirigible/components/listeners/synchronizer/ListenerSynchronizer.java +++ b/components/engine/engine-listeners/src/main/java/org/eclipse/dirigible/components/listeners/synchronizer/ListenerSynchronizer.java @@ -196,8 +196,12 @@ protected boolean completeImpl(TopologyWrapper wrapper, ArtefactPhase getService().save(listener); callback.registerState(this, wrapper, ArtefactLifecycle.CREATED); } catch (Exception e) { + // FAILED, not CREATED: nothing started, and only FAILED keeps the listener + // eligible for the START phase of this pass and of every later one (#7248). + // Returning false hands it to the in-pass topological retry as well. callback.addError(e.getMessage()); - callback.registerState(this, wrapper, ArtefactLifecycle.CREATED, e); + callback.registerState(this, wrapper, ArtefactLifecycle.FAILED, e); + return false; } } break; @@ -235,20 +239,27 @@ protected boolean completeImpl(TopologyWrapper wrapper, ArtefactPhase } break; case START: - if (ArtefactLifecycle.FAILED.equals(listener.getLifecycle())) { - String message = "Cannot start a Listener in a failing state: " + listener.getKey(); - callback.addError(message); - callback.registerState(this, wrapper, ArtefactLifecycle.FATAL, message); - return true; - } - if (listener.getRunning() == null || !listener.getRunning()) { + // A FAILED listener is retried, never promoted to FATAL (#7248). A subscription fails + // for reasons that heal later - the embedded broker still taking its store lease at + // boot, the handler published on a later pass - and FATAL stripped the artefact from + // every later pass, so a topic subscription lost to a boot race stayed lost, with + // every message to that destination discarded, until the file's bytes changed. + // Gated on FAILED as well as on running: the artefact completes once per tenant on one + // shared object, and only the lifecycle is reset between tenants - so the first tenant + // to subscribe flips running and would otherwise skip every tenant after it. The + // manager is idempotent per tenant, so a tenant already subscribed is a no-op. + if (ArtefactLifecycle.FAILED.equals(listener.getLifecycle()) || listener.getRunning() == null || !listener.getRunning()) { try { listenersManager.startListener(listener); listener.setRunning(true); getService().save(listener); + if (ArtefactLifecycle.FAILED.equals(listener.getLifecycle())) { + callback.registerState(this, wrapper, ArtefactLifecycle.CREATED); + } } catch (Exception e) { callback.addError(e.getMessage()); callback.registerState(this, wrapper, ArtefactLifecycle.FAILED, e); + return false; } } break; diff --git a/components/engine/engine-listeners/src/test/java/org/eclipse/dirigible/components/listeners/service/ListenersManagerTest.java b/components/engine/engine-listeners/src/test/java/org/eclipse/dirigible/components/listeners/service/ListenersManagerTest.java index 34a95ad6ea1..8ed2504ed1d 100644 --- a/components/engine/engine-listeners/src/test/java/org/eclipse/dirigible/components/listeners/service/ListenersManagerTest.java +++ b/components/engine/engine-listeners/src/test/java/org/eclipse/dirigible/components/listeners/service/ListenersManagerTest.java @@ -19,6 +19,7 @@ import org.mockito.Mock; import org.mockito.junit.jupiter.MockitoExtension; +import static org.junit.jupiter.api.Assertions.assertThrows; import static org.mockito.Mockito.*; /** @@ -121,7 +122,9 @@ void testStartListenerOnMissingListener() { resource); when(resource.exists()).thenReturn(false); - listenersManager.startListener(listenerEntity); + // A missing handler is a failure the synchronizer must see, not a normal return that reads + // as CREATED and running with nothing subscribed (#7248). + assertThrows(IllegalStateException.class, () -> listenersManager.startListener(listenerEntity)); verifyNoInteractions(messageListenerManagerFactory); } diff --git a/components/engine/engine-listeners/src/test/java/org/eclipse/dirigible/components/listeners/synchronizer/ListenerSynchronizerRetryTest.java b/components/engine/engine-listeners/src/test/java/org/eclipse/dirigible/components/listeners/synchronizer/ListenerSynchronizerRetryTest.java new file mode 100644 index 00000000000..1efac68e517 --- /dev/null +++ b/components/engine/engine-listeners/src/test/java/org/eclipse/dirigible/components/listeners/synchronizer/ListenerSynchronizerRetryTest.java @@ -0,0 +1,181 @@ +/* + * Copyright (c) 2010-2026 Eclipse Dirigible contributors + * + * All rights reserved. This program and the accompanying materials are made available under the + * terms of the Eclipse Public License v2.0 which accompanies this distribution, and is available at + * http://www.eclipse.org/legal/epl-v20.html + * + * SPDX-FileCopyrightText: Eclipse Dirigible contributors SPDX-License-Identifier: EPL-2.0 + */ +package org.eclipse.dirigible.components.listeners.synchronizer; + +import org.eclipse.dirigible.components.base.artefact.ArtefactLifecycle; +import org.eclipse.dirigible.components.base.artefact.ArtefactPhase; +import org.eclipse.dirigible.components.base.artefact.topology.TopologyWrapper; +import org.eclipse.dirigible.components.base.synchronizer.SynchronizerCallback; +import org.eclipse.dirigible.components.listeners.domain.Listener; +import org.eclipse.dirigible.components.listeners.domain.ListenerKind; +import org.eclipse.dirigible.components.listeners.service.ListenerService; +import org.eclipse.dirigible.components.listeners.service.ListenersManager; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.springframework.test.util.ReflectionTestUtils; + +import java.util.HashMap; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNotEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.doAnswer; +import static org.mockito.Mockito.doNothing; +import static org.mockito.Mockito.doThrow; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +/** + * A listener whose subscription the broker refuses is recorded FAILED and retried on every pass + * until it subscribes - never CREATED while nothing runs, never promoted to FATAL (#7248). + */ +class ListenerSynchronizerRetryTest { + + /** What the broker answers while it is not accepting connections yet. */ + private static final String BROKER_REFUSAL = "Failed to start listener for [my-topic]"; + + /** The synchronizer under test. */ + private ListenerSynchronizer synchronizer; + + /** The manager whose start the tests make fail or succeed. */ + private ListenersManager listenersManager; + + /** The callback, persisting state the way the synchronization processor does. */ + private SynchronizerCallback callback; + + /** + * Wires the synchronizer over a manager double and a callback that writes the registered state onto + * the artefact, as the synchronization processor does. + */ + @BeforeEach + void setUp() { + ListenerService listenerService = mock(ListenerService.class); + when(listenerService.save(any())).thenAnswer(invocation -> invocation.getArgument(0)); + listenersManager = mock(ListenersManager.class); + + synchronizer = new ListenerSynchronizer(); + ReflectionTestUtils.setField(synchronizer, "listenerService", listenerService); + ReflectionTestUtils.setField(synchronizer, "listenersManager", listenersManager); + + callback = mock(SynchronizerCallback.class); + doAnswer(invocation -> { + synchronizer.setStatus(artefact(invocation.getArgument(1)), invocation.getArgument(2), ""); + return null; + }).when(callback) + .registerState(any(), any(TopologyWrapper.class), any(ArtefactLifecycle.class)); + doAnswer(invocation -> { + Throwable cause = invocation.getArgument(3); + synchronizer.setStatus(artefact(invocation.getArgument(1)), invocation.getArgument(2), cause.getMessage()); + return null; + }).when(callback) + .registerState(any(), any(TopologyWrapper.class), any(ArtefactLifecycle.class), any(Throwable.class)); + doAnswer(invocation -> { + synchronizer.setStatus(artefact(invocation.getArgument(1)), invocation.getArgument(2), invocation.getArgument(3)); + return null; + }).when(callback) + .registerState(any(), any(TopologyWrapper.class), any(ArtefactLifecycle.class), any(String.class)); + synchronizer.setCallback(callback); + } + + /** + * The first pass: a refused start is FAILED with its cause and not completed - not CREATED with + * nothing running, which is how the Registry read a listener that never subscribed. + */ + @Test + void aRefusedStartIsRecordedFailedNotCreated() { + Listener listener = listener(ArtefactLifecycle.NEW); + TopologyWrapper wrapper = wrapper(listener); + doThrow(new IllegalStateException(BROKER_REFUSAL)).when(listenersManager) + .startListener(listener); + + assertFalse(synchronizer.completeImpl(wrapper, ArtefactPhase.CREATE), "A refused start is not completed - the pass retries it"); + assertEquals(ArtefactLifecycle.FAILED, listener.getLifecycle(), "A listener that never subscribed must read FAILED, not CREATED"); + assertEquals(BROKER_REFUSAL, listener.getError(), "The recorded error must be the broker's refusal"); + assertNotEquals(Boolean.TRUE, listener.getRunning(), "Nothing started, so nothing may read as running"); + } + + /** + * The passes after: a FAILED listener the broker still refuses stays FAILED and keeps being + * retried. Before the fix the second attempt registered FATAL, after which the processor stripped + * the artefact from every later pass. + */ + @Test + void aFailedListenerTheBrokerStillRefusesStaysFailedAndIsRetried() { + Listener listener = listener(ArtefactLifecycle.FAILED); + TopologyWrapper wrapper = wrapper(listener); + doThrow(new IllegalStateException(BROKER_REFUSAL)).when(listenersManager) + .startListener(listener); + + assertFalse(synchronizer.completeImpl(wrapper, ArtefactPhase.START), "A start the broker refused is not completed"); + assertFalse(synchronizer.completeImpl(wrapper, ArtefactPhase.START), "The next attempt is refused the same way"); + + assertEquals(ArtefactLifecycle.FAILED, listener.getLifecycle(), "A transiently refused listener must never read FATAL"); + assertNotEquals(Boolean.TRUE, listener.getRunning(), "Nothing started, so nothing may read as running"); + verify(listenersManager, times(2)).startListener(listener); + } + + /** + * Once the broker accepts, the very same FAILED artefact subscribes, reads CREATED and carries no + * stale error. + */ + @Test + void aFailedListenerHealsOnceTheBrokerAccepts() { + Listener listener = listener(ArtefactLifecycle.FAILED); + listener.setError(BROKER_REFUSAL); + TopologyWrapper wrapper = wrapper(listener); + doNothing().when(listenersManager) + .startListener(listener); + + assertTrue(synchronizer.completeImpl(wrapper, ArtefactPhase.START), "A start the broker accepted completes"); + assertEquals(ArtefactLifecycle.CREATED, listener.getLifecycle(), "The healed listener must read CREATED"); + assertEquals("", listener.getError(), "The healed listener must carry no stale error"); + assertEquals(Boolean.TRUE, listener.getRunning(), "The healed listener is running"); + } + + /** + * A listener that is already running is left alone by the START phase - the retry is for what is + * not running, not a re-subscription on every pass. + */ + @Test + void aRunningListenerIsNotStartedAgain() { + Listener listener = listener(ArtefactLifecycle.CREATED); + listener.setRunning(true); + TopologyWrapper wrapper = wrapper(listener); + + assertTrue(synchronizer.completeImpl(wrapper, ArtefactPhase.START)); + verify(listenersManager, times(0)).startListener(any()); + assertEquals(ArtefactLifecycle.CREATED, listener.getLifecycle()); + } + + private static Listener listener(ArtefactLifecycle lifecycle) { + Listener listener = new Listener("/retry/probe.listener", "my-topic", "", "retry/handler.js", ListenerKind.TOPIC); + listener.setLifecycle(lifecycle); + listener.setRunning(false); + return listener; + } + + private TopologyWrapper wrapper(Listener listener) { + return new TopologyWrapper<>(listener, new HashMap<>(), synchronizer); + } + + /** + * Unwraps the artefact of a wrapper handed to the callback. + * + * @param wrapper the wrapper + * @return the listener + */ + private static Listener artefact(Object wrapper) { + return ((TopologyWrapper) wrapper).getArtefact() instanceof Listener listener ? listener : null; + } +} diff --git a/modules/commons/commons-config/src/main/java/org/eclipse/dirigible/commons/config/DirigibleConfig.java b/modules/commons/commons-config/src/main/java/org/eclipse/dirigible/commons/config/DirigibleConfig.java index a3509073cf7..765a552e980 100644 --- a/modules/commons/commons-config/src/main/java/org/eclipse/dirigible/commons/config/DirigibleConfig.java +++ b/modules/commons/commons-config/src/main/java/org/eclipse/dirigible/commons/config/DirigibleConfig.java @@ -62,6 +62,13 @@ public enum DirigibleConfig { SYNCHRONIZER_CROSS_RETRY_INTERVAL_MILLIS("DIRIGIBLE_SYNCHRONIZER_CROSS_RETRY_INTERVAL_MILLIS", "10000"), // + /** + * How often an idle instance gives its FAILED artefacts one more START attempt (#7248). A pass + * otherwise runs only on a registry change, so a listener the broker refused at boot was never + * retried until someone published something else. + */ + SYNCHRONIZER_FAILED_RETRY_INTERVAL_SECONDS("DIRIGIBLE_SYNCHRONIZER_FAILED_RETRY_INTERVAL_SECONDS", "30"), // + /** Bridge the platform readiness onto Spring's ApplicationAvailability (#6448). */ READINESS_AVAILABILITY_BRIDGE_ENABLED("DIRIGIBLE_READINESS_AVAILABILITY_BRIDGE_ENABLED", Boolean.FALSE.toString()), // diff --git a/tests/tests-integrations/src/main/java/org/eclipse/dirigible/integration/tests/api/ListenerArtefactSubscriptionRetryIT.java b/tests/tests-integrations/src/main/java/org/eclipse/dirigible/integration/tests/api/ListenerArtefactSubscriptionRetryIT.java new file mode 100644 index 00000000000..74d3343dfb2 --- /dev/null +++ b/tests/tests-integrations/src/main/java/org/eclipse/dirigible/integration/tests/api/ListenerArtefactSubscriptionRetryIT.java @@ -0,0 +1,195 @@ +/* + * Copyright (c) 2010-2026 Eclipse Dirigible contributors + * + * All rights reserved. This program and the accompanying materials are made available under the + * terms of the Eclipse Public License v2.0 which accompanies this distribution, and is available at + * http://www.eclipse.org/legal/epl-v20.html + * + * SPDX-FileCopyrightText: Eclipse Dirigible contributors SPDX-License-Identifier: EPL-2.0 + */ +package org.eclipse.dirigible.integration.tests.api; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotEquals; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.doAnswer; + +import java.nio.charset.StandardCharsets; +import java.util.List; +import java.util.concurrent.TimeUnit; + +import org.awaitility.Awaitility; +import org.eclipse.dirigible.commons.config.DirigibleConfig; +import org.eclipse.dirigible.components.api.messaging.MessagingFacade; +import org.eclipse.dirigible.components.base.artefact.ArtefactLifecycle; +import org.eclipse.dirigible.components.initializers.synchronizer.SynchronizationProcessor; +import org.eclipse.dirigible.components.listeners.config.ActiveMQConnectionArtifactsFactory; +import org.eclipse.dirigible.components.listeners.domain.Listener; +import org.eclipse.dirigible.components.listeners.service.ListenerService; +import org.eclipse.dirigible.integration.tests.api.java.messaging.MessagesHolder; +import org.eclipse.dirigible.repository.api.IRepository; +import org.eclipse.dirigible.repository.api.IRepositoryStructure; +import org.eclipse.dirigible.tests.base.IntegrationTest; +import org.junit.jupiter.api.AfterAll; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.test.annotation.DirtiesContext; +import org.springframework.test.context.bean.override.mockito.MockitoSpyBean; +import org.springframework.test.util.ReflectionTestUtils; + +import jakarta.jms.ExceptionListener; + +/** + * A {@code .listener} artefact whose subscription the broker refuses at startup is retried on every + * synchronization pass until it subscribes (#7248). Before the fix the second pass promoted it to + * FATAL and the third stripped it from synchronization for good, so every message published to its + * topic afterwards was silently lost. + */ +@DirtiesContext(classMode = DirtiesContext.ClassMode.AFTER_CLASS) +class ListenerArtefactSubscriptionRetryIT extends IntegrationTest { + + private static final String PROJECT = "listener-artefact-retry-it"; + private static final String LISTENER_LOCATION = "/" + PROJECT + "/probe.listener"; + private static final String HANDLER_PATH = PROJECT + "/handler.js"; + private static final String TOPIC = PROJECT + "-topic"; + + /** + * Keeps a pass with a refused listener short: one in-pass retry a second later, not ten a 10 s + * apart. + */ + private static final String CROSS_RETRY_COUNT = "1"; + private static final String CROSS_RETRY_INTERVAL_MILLIS = "1000"; + + /** + * The idle instance's own retry cadence, shortened so the heal arrives within the test's patience. + */ + private static final String FAILED_RETRY_INTERVAL_SECONDS = "2"; + + private static final int HEAL_TIMEOUT_SECONDS = 90; + private static final int MESSAGE_TIMEOUT_SECONDS = 60; + + private static String previousCrossRetryCount; + private static String previousCrossRetryInterval; + private static String previousFailedRetryInterval; + + private volatile boolean refuseProbeSubscription = true; + + @MockitoSpyBean + private ActiveMQConnectionArtifactsFactory connectionArtifactsFactory; + + @Autowired + private IRepository repository; + + @Autowired + private SynchronizationProcessor synchronizationProcessor; + + @Autowired + private ListenerService listenerService; + + @BeforeAll + static void shortenTheInPassRetry() { + previousCrossRetryCount = DirigibleConfig.SYNCHRONIZER_CROSS_RETRY_COUNT.getStringValue(); + previousCrossRetryInterval = DirigibleConfig.SYNCHRONIZER_CROSS_RETRY_INTERVAL_MILLIS.getStringValue(); + previousFailedRetryInterval = DirigibleConfig.SYNCHRONIZER_FAILED_RETRY_INTERVAL_SECONDS.getStringValue(); + DirigibleConfig.SYNCHRONIZER_CROSS_RETRY_COUNT.setStringValue(CROSS_RETRY_COUNT); + DirigibleConfig.SYNCHRONIZER_CROSS_RETRY_INTERVAL_MILLIS.setStringValue(CROSS_RETRY_INTERVAL_MILLIS); + DirigibleConfig.SYNCHRONIZER_FAILED_RETRY_INTERVAL_SECONDS.setStringValue(FAILED_RETRY_INTERVAL_SECONDS); + } + + @AfterAll + static void restoreTheInPassRetry() { + DirigibleConfig.SYNCHRONIZER_CROSS_RETRY_COUNT.setStringValue(previousCrossRetryCount); + DirigibleConfig.SYNCHRONIZER_CROSS_RETRY_INTERVAL_MILLIS.setStringValue(previousCrossRetryInterval); + DirigibleConfig.SYNCHRONIZER_FAILED_RETRY_INTERVAL_SECONDS.setStringValue(previousFailedRetryInterval); + } + + @BeforeEach + void armTheBrokerRefusal() { + MessagesHolder.clearLatestReceivedMessage(); + doAnswer(invocation -> { + // The artefact path opens its connection with the listener's own exception handler, which + // carries the handler path - the one thing that identifies this probe. Everything else, + // the platform's shared connection included, must behave normally. + if (refuseProbeSubscription && HANDLER_PATH.equals(handlerPathOf(invocation.getArgument(0)))) { + throw new IllegalStateException("Failed to create connection to ActiveMQ"); + } + return invocation.callRealMethod(); + }).when(connectionArtifactsFactory) + .createConnection(any(), any()); + } + + @Test + void aSubscriptionTheBrokerRefusedIsRetriedOnEveryPassUntilItConnects() { + deployProbeListener(); + + // Pass 1: the refusal happened (without this the test would pass on a spy that intercepted + // nothing), and it reads FAILED - not CREATED with nothing subscribed. + synchronizationProcessor.forceProcessSynchronizers(); + Listener afterFirstPass = probeListener(); + assertEquals(ArtefactLifecycle.FAILED, afterFirstPass.getLifecycle(), "a refused subscription must read FAILED"); + assertNotEquals(Boolean.TRUE, afterFirstPass.getRunning(), "nothing subscribed, so nothing may read as running"); + + // Pass 2: still refused. This is the pass that used to register FATAL. + synchronizationProcessor.forceProcessSynchronizers(); + assertEquals(ArtefactLifecycle.FAILED, probeListener().getLifecycle(), "a transiently refused listener must never go FATAL"); + + refuseProbeSubscription = false; + + // The broker accepts. Nothing is published and nothing is forced from here on: only the idle + // instance's own FAILED retry can subscribe the listener now. Before the fix the artefact was + // stripped from every later pass, and a pass without a registry change ran no phase at all. + Awaitility.await() + .pollInterval(1, TimeUnit.SECONDS) + .atMost(HEAL_TIMEOUT_SECONDS, TimeUnit.SECONDS) + .until(() -> ArtefactLifecycle.CREATED.equals(probeListener().getLifecycle())); + assertEquals(Boolean.TRUE, probeListener().getRunning(), "the retried listener must be running"); + + String message = "ping-" + System.currentTimeMillis(); + MessagingFacade.sendToTopic(TOPIC, message); + Awaitility.await() + .atMost(MESSAGE_TIMEOUT_SECONDS, TimeUnit.SECONDS) + .until(() -> message.equals(MessagesHolder.getLatestReceivedMessage())); + } + + private Listener probeListener() { + List listeners = listenerService.findByLocation(LISTENER_LOCATION); + assertEquals(1, listeners.size(), "exactly one artefact must be recorded for " + LISTENER_LOCATION); + return listeners.get(0); + } + + private static String handlerPathOf(ExceptionListener exceptionListener) { + if (exceptionListener == null || !"ListenerExceptionHandler".equals(exceptionListener.getClass() + .getSimpleName())) { + return null; + } + return (String) ReflectionTestUtils.getField(exceptionListener, "handlerPath"); + } + + private void deployProbeListener() { + String handler = """ + const MessagesHolder = Java.type("org.eclipse.dirigible.integration.tests.api.java.messaging.MessagesHolder"); + + export function onMessage(message) { + MessagesHolder.setLatestReceivedMessage(message); + } + + export function onError(error) { + MessagesHolder.setLatestReceivedError(error); + } + """; + String listener = """ + { + "name": "%s", + "kind": "T", + "handler": "%s", + "description": "retried until the broker accepts" + } + """.formatted(TOPIC, HANDLER_PATH); + repository.createResource(IRepositoryStructure.PATH_REGISTRY_PUBLIC + "/" + HANDLER_PATH, handler.getBytes(StandardCharsets.UTF_8), + false, "text/javascript", true); + repository.createResource(IRepositoryStructure.PATH_REGISTRY_PUBLIC + LISTENER_LOCATION, listener.getBytes(StandardCharsets.UTF_8), + false, "application/json", true); + } +}