feat(messaging): a deferred announcement survives a broker outage - the durable publish

components/api/api-messaging/src/main/java/org/eclipse/dirigible/components/api/messaging/DurableMessagePublisher.java

@@ -0,0 +1,35 @@
+/*
+ * 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.api.messaging;
+
+/**
+ * Publishes a message that survives a broker outage: the message is recorded durably before the
+ * broker sees it, and whatever the broker refuses is retried until it is taken, so delivery is
+ * at-least-once instead of fire-and-forget.
+ *
+ * <p>
+ * This is the contract for an announcement that is deliberately DECOUPLED from the write it is
+ * about — deferred past a synchronous chain's commit, or ordered after several transactions — where
+ * the transactional outbox's write-attached recording cannot apply. The implementation lives with
+ * the outbox (the event-store layer provides it); callers reach it through the SDK
+ * {@code Producer.sendToTopicDurable}, never directly.
+ */
+public interface DurableMessagePublisher {
+
+    /**
+     * Records the message durably and hands it to the broker; what the broker refuses is retried until
+     * delivered. Never throws for a broker problem — the caller's work has already committed and must
+     * not be failed for an announcement the retry machinery owns.
+     *
+     * @param topic the topic to publish on
+     * @param payload the message body
+     */
+    void publishToTopic(String topic, String payload);
+}

components/api/api-modules-java/src/main/java/org/eclipse/dirigible/sdk/messaging/Producer.java

@@ -9,7 +9,9 @@
  */
 package org.eclipse.dirigible.sdk.messaging;
 
+import org.eclipse.dirigible.components.api.messaging.DurableMessagePublisher;
 import org.eclipse.dirigible.components.api.messaging.MessagingFacade;
+import org.eclipse.dirigible.components.base.spring.BeanProvider;
 
 /**
  * Sends a message into the embedded ActiveMQ broker. {@link #sendToQueue(String, String)} delivers
@@ -34,4 +36,21 @@ public static void sendToQueue(String queue, String message) {
     public static void sendToTopic(String topic, String message) {
         MessagingFacade.sendToTopic(topic, message);
     }
+
+    /**
+     * Publishes to a topic with at-least-once delivery: the message is recorded durably before the
+     * broker sees it, and whatever the broker refuses is retried until it is taken. Use this for an
+     * announcement whose loss would silently break a downstream reaction — a status transition a
+     * posting or a create-from is keyed on — when the announcement is deliberately decoupled from the
+     * write it is about (deferred past a workflow chain's commit, or ordered after several writes). An
+     * announcement that belongs to ONE write should ride that write instead: hand the topic to the
+     * repository's write methods, which record it in the same transaction.
+     *
+     * @param topic the topic to publish on
+     * @param message the message body
+     */
+    public static void sendToTopicDurable(String topic, String message) {
+        BeanProvider.getBean(DurableMessagePublisher.class)
+                    .publishToTopic(topic, message);
+    }
 }

components/data/data-store-java/src/main/java/org/eclipse/dirigible/components/data/store/java/outbox/EventOutbox.java

@@ -43,7 +43,7 @@
  * "sent" is only known once the row is gone.
  */
 @Component
-public class EventOutbox {
+public class EventOutbox implements org.eclipse.dirigible.components.api.messaging.DurableMessagePublisher {
 
     private static final Logger LOGGER = LoggerFactory.getLogger(EventOutbox.class);
 
@@ -89,6 +89,34 @@ public Batch record(Session session, List<DomainEvent> events) {
         return new Batch(this, pending);
     }
 
+    /**
+     * The durable publish for an announcement DECOUPLED from the write it is about — deferred past a
+     * synchronous chain's commit, or ordered after several transactions — where {@link #record} has no
+     * transaction to join. The entry is recorded in its own short transaction and handed to the broker
+     * immediately; whatever the broker refuses stays for the relay, exactly as a write-attached event
+     * would. If even the recording fails (the outbox table unreachable), the message is handed to the
+     * broker directly — the caller's work has already committed and an at-most-once attempt beats
+     * failing a caller whose announcement machinery is down.
+     *
+     * @param topic the topic to publish on
+     * @param payload the message body
+     */
+    @Override
+    public void publishToTopic(String topic, String payload) {
+        PendingEvent event = new PendingEvent(UUID.randomUUID()
+                                                  .toString(),
+                topic, payload, 0);
+        try {
+            prepare();
+            store.insert(event, nextAttemptAt());
+        } catch (RuntimeException | SQLException ex) {
+            LOGGER.error("Failed to record a durable publish on topic [{}]; attempting the direct publish instead.", topic, ex);
+            MessagingFacade.sendToTopic(topic, payload);
+            return;
+        }
+        deliver(event);
+    }
+
     /**
      * Makes sure the current tenant's outbox table exists. Called before the write transaction opens so
      * that the table's DDL never runs inside it.

components/data/data-store-java/src/main/java/org/eclipse/dirigible/components/data/store/java/outbox/EventOutboxStore.java

@@ -124,6 +124,21 @@ void insert(Connection connection, PendingEvent event, Instant nextAttemptAt) th
         }
     }
 
+    /**
+     * Records one event on a connection of its own — the standalone variant for an announcement that is
+     * deliberately decoupled from the write it is about (a deferred publish after a synchronous chain's
+     * commit), where there is no enclosing transaction to join.
+     *
+     * @param event the event to record
+     * @param nextAttemptAt when the relay may first take the entry over
+     * @throws SQLException if the insert fails
+     */
+    void insert(PendingEvent event, Instant nextAttemptAt) throws SQLException {
+        try (Connection connection = connection()) {
+            insert(connection, event, nextAttemptAt);
+        }
+    }
+
     /**
      * @return true when the current tenant has an outbox table at all — a tenant that never wrote an
      *         entity has none, and the relay must not fail on it

components/engine/engine-java/CLAUDE.md

@@ -201,7 +201,7 @@ the default user-data datasource, not SystemDB.
 **A write and the event announcing it commit together — the transactional outbox.** A repository that publishes an entity event does not commit the row and then call the broker: it hands the topic to the write itself (`save(entity, topic)`, `update(entity, topic[, extraEvents])`, `updateProperties(id, values, topic)`, `delete(entity, topic)`, `deleteById(id, topic)`), which records the event in the tenant's `DIRIGIBLE_EVENT_OUTBOX` **on the write's own connection, inside its transaction**, and only then — after the commit — hands it to the broker in-process. Two failures die with this: an event lost for good because the broker was briefly down while its row committed anyway (nothing retried it), and a `500` raised to a REST caller whose write had actually succeeded, inviting a retry that duplicated the record (issue #6816). What the in-process dispatch cannot deliver simply stays in the table, and `EventOutboxRelayJob` retries it per tenant every `DIRIGIBLE_EVENT_OUTBOX_RELAY_INTERVAL_SECONDS` (30) for entries idle longer than `DIRIGIBLE_EVENT_OUTBOX_RELAY_GRACE_SECONDS` (60). Consequences to keep in mind:
 
 - **Delivery is at-least-once, not exactly-once.** "Sent" is only known once the entry is gone, so an entry published just before the node died is published again. Handlers must tolerate a repeat — which the generated glue already does, since it recomputes from the store rather than accumulating.
-- **`Producer.sendToTopic` in hand-written client code is still a bare publish** with none of this. It is the raw messaging API; the outbox is reached only by giving a *write* its topic. Announce an entity change through its repository, not by publishing next to it.
+- **`Producer.sendToTopic` in hand-written client code is still a bare publish** with none of this. It is the raw messaging API; the outbox is reached only by giving a *write* its topic. Announce an entity change through its repository, not by publishing next to it. For an announcement that is deliberately DECOUPLED from any single write - deferred past a workflow chain's commit, or ordered after several transactions - use **`Producer.sendToTopicDurable`**: the message is recorded in the outbox in its own short transaction and the relay retries whatever the broker refuses, so an outage delays it instead of losing it (at-least-once; the generated deferred publishes - setField, Writer, Numbering, step events, the create-from completion announce - all use it).
 - **The event's payload is the row as the transaction left it** — read back on the write's own connection for the targeted path, never a re-read afterwards that a concurrent write could have moved on. On a `multilingual: true` entity that means the untranslated row: an event carries canonical data, not the writer's `Accept-Language`.
 - **A repository that overrides targeted writes must override the event-carrying form.** `updateProperties(id, values, topic)` is where a generated repository hangs its declarative checks, stored label and document resum; the plain two-argument form delegates to it. The base two-argument form deliberately does NOT re-dispatch, because `recalculate` reaches it through `super` precisely to bypass those semantics.
 - **No outbox, no write.** If the entry cannot be recorded the transaction fails, which is the whole contract: a row whose event was never recorded is exactly the state this replaces. `JavaEventOutboxIT` covers both halves — an ordinary create reaching its listener, and an entry only the relay can deliver.

components/template/template-application-events-java/src/main/resources/META-INF/dirigible/template-application-events-java/events/Generate.java.template

@@ -207,7 +207,7 @@ public class ${className}Generate {
         // Reload so the payload carries the committed row (the flipped status plus any write the target
         // creation performed on the source), not the stale pre-generation snapshot.
         source = sourceRepository.findById(sourceId);
-        org.eclipse.dirigible.sdk.messaging.Producer.sendToTopic(
+        org.eclipse.dirigible.sdk.messaging.Producer.sendToTopicDurable(
                 "${fromProjectName}-${fromPerspective}-${fromEntity}-transitioned", Json.stringify(source));
 #end
         return saved;

components/template/template-application-events-java/src/main/resources/META-INF/dirigible/template-application-events-java/events/Numbering.java.template

@@ -55,7 +55,7 @@ public class ${entity}NumberStamp implements JavaDelegate {
         ${entity}Entity stamped = repository.findById(id);
         if (stamped != null) {
             String payload = Json.stringify(stamped);
-            Process.executeAfterCommit(() -> Producer.sendToTopic("${projectName}-${perspective}-${entity}-updated", payload));
+            Process.executeAfterCommit(() -> Producer.sendToTopicDurable("${projectName}-${perspective}-${entity}-updated", payload));
         }
     }
 

components/template/template-application-events-java/src/main/resources/META-INF/dirigible/template-application-events-java/events/SetField.java.template

@@ -58,7 +58,7 @@ public class ${className} implements JavaDelegate {
             // of racing them (an auto-posted entry used to catch the create-time UUID placeholder).
             String transitioned = Json.stringify(entity);
             Process.executeAfterCommit(
-                    () -> Producer.sendToTopic("${projectName}-${perspective}-${entity}-transitioned", transitioned));
+                    () -> Producer.sendToTopicDurable("${projectName}-${perspective}-${entity}-transitioned", transitioned));
         }
     }
 }

components/template/template-application-events-java/src/main/resources/META-INF/dirigible/template-application-events-java/events/StepEvent.java.template

@@ -42,6 +42,6 @@ public class ${className} implements JavaDelegate {
             return;
         }
         String payload = Json.stringify(entity);
-        Process.executeAfterCommit(() -> Producer.sendToTopic("${projectName}-${perspective}-${entity}${topicSuffix}", payload));
+        Process.executeAfterCommit(() -> Producer.sendToTopicDurable("${projectName}-${perspective}-${entity}${topicSuffix}", payload));
     }
 }

components/template/template-application-events-java/src/main/resources/META-INF/dirigible/template-application-events-java/events/Writer.java.template

@@ -78,7 +78,7 @@ public class ${className} implements JavaDelegate {
         ${entity}Entity entity = repository.findById(id);
         if (entity != null) {
             String payload = Json.stringify(entity);
-            Process.executeAfterCommit(() -> Producer.sendToTopic("${projectName}-${perspective}-${entity}-updated", payload));
+            Process.executeAfterCommit(() -> Producer.sendToTopicDurable("${projectName}-${perspective}-${entity}-updated", payload));
         }
     }
 }

tests/tests-integrations/src/main/java/org/eclipse/dirigible/integration/tests/api/IntentEngineIT.java

@@ -1084,7 +1084,7 @@ void set_field_glue_sets_entity_status_on_approve_reject_branches() {
         // and the async consumer re-loads the source on receive - it must observe those writes.
         assertTrue(
                 activate.contains("Process.executeAfterCommit(")
-                        && activate.contains("Producer.sendToTopic(\"" + PROJECT + "-Member-Member-transitioned\", transitioned)"),
+                        && activate.contains("Producer.sendToTopicDurable(\"" + PROJECT + "-Member-Member-transitioned\", transitioned)"),
                 "the setter should publish the -transitioned topic after the BPMN chain commits");
     }
 
@@ -1313,7 +1313,7 @@ void the_status_channel_is_bindable_by_notifications_and_waits() {
                 "an onTransition wait must bind the -transitioned topic, got: " + wait);
         // The setter on the same entity is the publisher the two now hear.
         String setter = contentOf("gen/events/fines/IdentifyAttribute.java");
-        assertTrue(setter.contains("Producer.sendToTopic(\"" + PROJECT + "-Fine-Fine-transitioned\", transitioned)"),
+        assertTrue(setter.contains("Producer.sendToTopicDurable(\"" + PROJECT + "-Fine-Fine-transitioned\", transitioned)"),
                 "the setter must publish the very topic the notification and the wait subscribe to");
     }
 
@@ -3008,7 +3008,7 @@ void editable_task_form_fields_are_coerced_to_their_java_type_on_write_back() {
         // edits only reached anything by accident - when an unrelated setter on the same task happened
         // to sweep them into its own reload. Deferred, because a consumer re-loads on receive and would
         // otherwise race the rest of the BPMN chain.
-        assertTrue(writer.contains("Producer.sendToTopic(\"" + PROJECT + "-SalesOrder-SalesOrder-updated\", payload)"),
+        assertTrue(writer.contains("Producer.sendToTopicDurable(\"" + PROJECT + "-SalesOrder-SalesOrder-updated\", payload)"),
                 "the writer must publish the entity's -updated topic, got: " + writer);
         assertTrue(writer.contains("Process.executeAfterCommit("), "the publish must be deferred to after the BPMN chain commits");
         int write = writer.indexOf("repository.updateProperties(id, values)");
@@ -3040,7 +3040,7 @@ void numbering_stamp_publishes_the_stamped_document_number() {
 
         String stamp = contentOf("gen/events/orders/SalesInvoiceNumberStamp.java");
         assertTrue(stamp.contains("DocumentNumbers.next(\"Sales Invoice\")"), "the stamp must allocate from the declared series");
-        assertTrue(stamp.contains("Producer.sendToTopic(\"" + PROJECT + "-SalesInvoice-SalesInvoice-updated\", payload)"),
+        assertTrue(stamp.contains("Producer.sendToTopicDurable(\"" + PROJECT + "-SalesInvoice-SalesInvoice-updated\", payload)"),
                 "the stamp must publish the entity's -updated topic - the raw perspective, not the sanitized Java one, got: " + stamp);
         assertTrue(stamp.contains("Process.executeAfterCommit("), "the publish must be deferred to after the BPMN chain commits");
         int write = stamp.indexOf("repository.updateProperty(id, \"Number\", number)");