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); + } +}