From cad56146e942b8daab89213495dba1dac0728146 Mon Sep 17 00:00:00 2001 From: delchev Date: Thu, 20 Aug 2026 12:31:46 +0300 Subject: [PATCH] feat(messaging): a deferred announcement survives a broker outage - the durable publish (#6867) The write-attached outbox (#6854) covers an event that belongs to ONE write. Five generated announcements are deliberately DECOUPLED from any single write and stayed bare Producer.sendToTopic calls - each a live instance of #6816's loss class, where a briefly-down broker swallows the notice and every downstream reaction keyed on it silently never runs: - SetField, Writer and Numbering defer their "-transitioned"/"-updated" publish past the synchronous Flowable chain's commit, so consumers observe the whole chain's writes rather than a mid-chain snapshot - attaching the event to the write would regress exactly that; - a StepEvent moment has no entity write of its own at all; - the create-from's completion announce is ordered after the target and its items exist, across several transactions. They now publish through Producer.sendToTopicDurable: the message is recorded in the tenant's event outbox in its own short transaction and handed to the broker immediately - a healthy system publishes exactly as promptly as before - and whatever the broker refuses stays for EventOutboxRelayJob, the same at-least-once contract the write-attached events have. Ordering is untouched: the deferral points stay where they are; only the loss window closes. If even the recording fails, the message falls back to the direct publish - the caller's work has already committed and an at-most-once attempt beats none. The plumbing: DurableMessagePublisher is a small SPI in api-messaging (the module both sides already see - the SDK cannot depend on the outbox's module and the outbox already depends on messaging), EventOutbox implements it with an own-connection insert on the store, and the SDK Producer resolves it through BeanProvider. The javadoc states the rule: an announcement that belongs to one write rides that write; this is for the decoupled ones. Emission is pinned by the existing IntentEngineIT anchors, updated to the durable call; runtime by IntentEmissionCoverageIT (the workflow setters, step moments, stamped numbers and completion announces all flow through the new path) and JavaEventOutboxIT. Co-Authored-By: Claude Fable 5 --- .../messaging/DurableMessagePublisher.java | 35 +++++++++++++++++++ .../dirigible/sdk/messaging/Producer.java | 19 ++++++++++ .../data/store/java/outbox/EventOutbox.java | 30 +++++++++++++++- .../store/java/outbox/EventOutboxStore.java | 15 ++++++++ components/engine/engine-java/CLAUDE.md | 2 +- .../events/Generate.java.template | 2 +- .../events/Numbering.java.template | 2 +- .../events/SetField.java.template | 2 +- .../events/StepEvent.java.template | 2 +- .../events/Writer.java.template | 2 +- .../integration/tests/api/IntentEngineIT.java | 8 ++--- 11 files changed, 108 insertions(+), 11 deletions(-) create mode 100644 components/api/api-messaging/src/main/java/org/eclipse/dirigible/components/api/messaging/DurableMessagePublisher.java diff --git a/components/api/api-messaging/src/main/java/org/eclipse/dirigible/components/api/messaging/DurableMessagePublisher.java b/components/api/api-messaging/src/main/java/org/eclipse/dirigible/components/api/messaging/DurableMessagePublisher.java new file mode 100644 index 00000000000..5cec93ef323 --- /dev/null +++ b/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. + * + *

+ * 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); +} diff --git a/components/api/api-modules-java/src/main/java/org/eclipse/dirigible/sdk/messaging/Producer.java b/components/api/api-modules-java/src/main/java/org/eclipse/dirigible/sdk/messaging/Producer.java index 218adfb81d3..e66e82b1c00 100644 --- a/components/api/api-modules-java/src/main/java/org/eclipse/dirigible/sdk/messaging/Producer.java +++ b/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); + } } diff --git a/components/data/data-store-java/src/main/java/org/eclipse/dirigible/components/data/store/java/outbox/EventOutbox.java b/components/data/data-store-java/src/main/java/org/eclipse/dirigible/components/data/store/java/outbox/EventOutbox.java index 72d0b1b837f..6c66a9c59c9 100644 --- a/components/data/data-store-java/src/main/java/org/eclipse/dirigible/components/data/store/java/outbox/EventOutbox.java +++ b/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 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. diff --git a/components/data/data-store-java/src/main/java/org/eclipse/dirigible/components/data/store/java/outbox/EventOutboxStore.java b/components/data/data-store-java/src/main/java/org/eclipse/dirigible/components/data/store/java/outbox/EventOutboxStore.java index ed0a088b0ad..d6547f02b4d 100644 --- a/components/data/data-store-java/src/main/java/org/eclipse/dirigible/components/data/store/java/outbox/EventOutboxStore.java +++ b/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 diff --git a/components/engine/engine-java/CLAUDE.md b/components/engine/engine-java/CLAUDE.md index 4aa0c5cf64c..108f4563a05 100644 --- a/components/engine/engine-java/CLAUDE.md +++ b/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. diff --git a/components/template/template-application-events-java/src/main/resources/META-INF/dirigible/template-application-events-java/events/Generate.java.template b/components/template/template-application-events-java/src/main/resources/META-INF/dirigible/template-application-events-java/events/Generate.java.template index e7e15748fbb..7ef84765d46 100644 --- a/components/template/template-application-events-java/src/main/resources/META-INF/dirigible/template-application-events-java/events/Generate.java.template +++ b/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; diff --git a/components/template/template-application-events-java/src/main/resources/META-INF/dirigible/template-application-events-java/events/Numbering.java.template b/components/template/template-application-events-java/src/main/resources/META-INF/dirigible/template-application-events-java/events/Numbering.java.template index 67ec52dfcc3..2b2a49bd098 100644 --- a/components/template/template-application-events-java/src/main/resources/META-INF/dirigible/template-application-events-java/events/Numbering.java.template +++ b/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)); } } diff --git a/components/template/template-application-events-java/src/main/resources/META-INF/dirigible/template-application-events-java/events/SetField.java.template b/components/template/template-application-events-java/src/main/resources/META-INF/dirigible/template-application-events-java/events/SetField.java.template index ca06ca67807..313308b868b 100644 --- a/components/template/template-application-events-java/src/main/resources/META-INF/dirigible/template-application-events-java/events/SetField.java.template +++ b/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)); } } } diff --git a/components/template/template-application-events-java/src/main/resources/META-INF/dirigible/template-application-events-java/events/StepEvent.java.template b/components/template/template-application-events-java/src/main/resources/META-INF/dirigible/template-application-events-java/events/StepEvent.java.template index 0de93ca216b..a681d6f938d 100644 --- a/components/template/template-application-events-java/src/main/resources/META-INF/dirigible/template-application-events-java/events/StepEvent.java.template +++ b/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)); } } diff --git a/components/template/template-application-events-java/src/main/resources/META-INF/dirigible/template-application-events-java/events/Writer.java.template b/components/template/template-application-events-java/src/main/resources/META-INF/dirigible/template-application-events-java/events/Writer.java.template index 2a089f0d12e..f963945327a 100644 --- a/components/template/template-application-events-java/src/main/resources/META-INF/dirigible/template-application-events-java/events/Writer.java.template +++ b/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)); } } } diff --git a/tests/tests-integrations/src/main/java/org/eclipse/dirigible/integration/tests/api/IntentEngineIT.java b/tests/tests-integrations/src/main/java/org/eclipse/dirigible/integration/tests/api/IntentEngineIT.java index fc47ae00838..3a1fe65de16 100644 --- a/tests/tests-integrations/src/main/java/org/eclipse/dirigible/integration/tests/api/IntentEngineIT.java +++ b/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)");