synchronizers: a transient start failure is FAILED and retried, never FATAL (#7248)

.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).

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<TopologyWrapper<? extends Artefact>> depleter = new TopologicalDepleter<>();
+        List<Artefact> failed = artefacts.values()
+                                         .stream()
+                                         .filter(a -> ArtefactLifecycle.FAILED.equals(a.getLifecycle()))
+                                         .collect(Collectors.toList());
+        List<TopologyWrapper<? extends Artefact>> wrappers = TopologyFactory.wrap(failed, synchronizers);
+        logger.info("Retrying [{}] FAILED artefacts: [{}]", wrappers.size(), wrappers);
+        for (Synchronizer<? extends Artefact, ?> synchronizer : synchronizers) {
+            Set<TopologyWrapper<? extends Artefact>> own = wrappers.stream()
+                                                                   .filter(w -> w.getSynchronizer()
+                                                                                 .equals(synchronizer))
+                                                                   .collect(Collectors.toSet());
+            if (own.isEmpty()) {
+                continue;
+            }
+            try {
+                Set<TopologyWrapper<? extends Artefact>> 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<? extends Artefact, ?> 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);

components/data/data-structures/src/main/java/org/eclipse/dirigible/components/data/structures/synchronizer/SchemasSynchronizer.java

@@ -548,13 +548,9 @@ protected boolean completeImpl(TopologyWrapper<Schema> 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;

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