diff --git a/components/data/data-store-java/src/main/java/org/eclipse/dirigible/components/data/store/java/repository/JavaRepository.java b/components/data/data-store-java/src/main/java/org/eclipse/dirigible/components/data/store/java/repository/JavaRepository.java index 3aa685ae35e..e0fce567cf0 100644 --- a/components/data/data-store-java/src/main/java/org/eclipse/dirigible/components/data/store/java/repository/JavaRepository.java +++ b/components/data/data-store-java/src/main/java/org/eclipse/dirigible/components/data/store/java/repository/JavaRepository.java @@ -80,6 +80,20 @@ public T save(T entity, String eventTopic) { return store().save(entity, eventTopic); } + /** + * Insert a new entity instance, publishing it on the given topic plus any further events the write + * emits about other rows — e.g. a create-from announcing its source's completed transition only + * once the document that transition was about exists. All of them share the insert's transaction. + * + * @param entity the entity to insert + * @param eventTopic the topic to publish the saved entity on; {@code null} publishes nothing + * @param additionalEvents further events to record with the same write + * @return the saved entity (with any generated identifier populated) + */ + public T save(T entity, String eventTopic, List additionalEvents) { + return store().save(entity, eventTopic, additionalEvents); + } + /** * Update an existing entity instance. * @@ -174,6 +188,23 @@ public int updateProperties(Object id, Map values, String eventT return store().updateProperties(entityClass, id, values, eventTopic); } + /** + * Targeted multi-column write publishing the resulting row on the given topic plus any further + * events the write emits about other rows — an aggregate's {@code "-rekeyed"} notice about the + * tuple the row just left. All of them share the mutation's transaction and are recorded only when + * the row actually existed to be written. + * + * @param id the primary-key value + * @param values the properties to set (plain identifiers) with their new values + * @param eventTopic the topic to publish the resulting row on; {@code null} publishes nothing + * @param additionalEvents further events to record with the same write + * @return the number of updated rows ({@code 0} when the id does not exist or {@code values} is + * empty) + */ + public int updateProperties(Object id, Map values, String eventTopic, List additionalEvents) { + return store().updateProperties(entityClass, id, values, eventTopic, additionalEvents); + } + /** * Look up an entity by primary key. An absent id is an ordinary outcome — a dangling foreign key an * event handler should skip, a path parameter a controller should answer {@code 404} for — so it diff --git a/components/data/data-store-java/src/main/java/org/eclipse/dirigible/components/data/store/java/store/JavaEntityStore.java b/components/data/data-store-java/src/main/java/org/eclipse/dirigible/components/data/store/java/store/JavaEntityStore.java index 45fa007d306..74d9a3b6357 100644 --- a/components/data/data-store-java/src/main/java/org/eclipse/dirigible/components/data/store/java/store/JavaEntityStore.java +++ b/components/data/data-store-java/src/main/java/org/eclipse/dirigible/components/data/store/java/store/JavaEntityStore.java @@ -91,10 +91,25 @@ public T save(T entity) { * @return the same entity (with any generated identifier populated) */ public T save(T entity, String eventTopic) { + return save(entity, eventTopic, List.of()); + } + + /** + * Insert a new entity, publishing it on the given topic plus any further events the write emits + * about other rows — e.g. a create-from announcing its source's completed transition only once the + * document that transition was about exists. All of them share the insert's transaction. + * + * @param the entity type + * @param entity the entity to insert + * @param eventTopic the topic to publish the saved entity on; {@code null} publishes nothing + * @param additionalEvents further events to record with the same write + * @return the same entity (with any generated identifier populated) + */ + public T save(T entity, String eventTopic, List additionalEvents) { RegisteredEntity meta = resolve(entity.getClass()); applyCreateAudit(entity, meta); Map data = EntityBeanMapper.toMap(entity, meta); - prepareOutbox(eventTopic != null); + prepareOutbox(eventTopic != null || !additionalEvents.isEmpty()); try (Session session = entityManager.getSessionFactory() .openSession()) { @@ -112,7 +127,7 @@ public T save(T entity, String eventTopic) { // the row was actually inserted with. writeId(entity, meta, generatedId); } - events = outbox.record(session, eventsOf(eventTopic, entity, List.of())); + events = outbox.record(session, eventsOf(eventTopic, entity, additionalEvents)); tx.commit(); } catch (RuntimeException ex) { rollback(tx, ex); @@ -258,6 +273,26 @@ public int updateProperties(Class type, Object id, Map va * empty) */ public int updateProperties(Class type, Object id, Map values, String eventTopic) { + return updateProperties(type, id, values, eventTopic, List.of()); + } + + /** + * Targeted multi-column write publishing the resulting row on the given topic plus any further + * events the write emits about other rows — an aggregate's {@code "-rekeyed"} notice about the + * tuple the row just left. All of them share the mutation's transaction and are recorded only when + * the row actually existed to be written. + * + * @param the entity type + * @param type the entity class + * @param id the primary-key value + * @param values the properties to set (plain identifiers) with their new values + * @param eventTopic the topic to publish the resulting row on; {@code null} publishes nothing + * @param additionalEvents further events to record with the same write + * @return the number of updated rows ({@code 0} when the id does not exist or {@code values} is + * empty) + */ + public int updateProperties(Class type, Object id, Map values, String eventTopic, + List additionalEvents) { if (values == null || values.isEmpty()) { return 0; } @@ -278,7 +313,7 @@ public int updateProperties(Class type, Object id, Map va RegisteredEntity meta = resolve(type); String idProperty = meta.idField() .getName(); - prepareOutbox(eventTopic != null); + prepareOutbox(eventTopic != null || !additionalEvents.isEmpty()); try (Session session = entityManager.getSessionFactory() .openSession()) { Transaction tx = session.beginTransaction(); @@ -293,8 +328,8 @@ public int updateProperties(Class type, Object id, Map va } updated = query.setParameter("id", id) .executeUpdate(); - events = outbox.record(session, eventTopic == null || updated == 0 ? List.of() - : eventsOf(eventTopic, readInTransaction(session, type, meta, id), List.of())); + events = outbox.record(session, updated == 0 ? List.of() + : eventsOf(eventTopic, eventTopic == null ? null : readInTransaction(session, type, meta, id), additionalEvents)); tx.commit(); } catch (RuntimeException ex) { rollback(tx, ex); diff --git a/components/template/template-application-dao-java/src/main/resources/META-INF/dirigible/template-application-dao-java/data/Repository.java.template b/components/template/template-application-dao-java/src/main/resources/META-INF/dirigible/template-application-dao-java/data/Repository.java.template index 77fdd71ba97..9a8e297ac7c 100644 --- a/components/template/template-application-dao-java/src/main/resources/META-INF/dirigible/template-application-dao-java/data/Repository.java.template +++ b/components/template/template-application-dao-java/src/main/resources/META-INF/dirigible/template-application-dao-java/data/Repository.java.template @@ -134,12 +134,9 @@ import org.eclipse.dirigible.components.data.store.java.repository.JavaRepositor import org.eclipse.dirigible.sdk.component.Component; import org.eclipse.dirigible.sdk.component.Repository; #if($groupingKeys && $groupingKeys.size() > 0) -## The rekey notices are the ONLY events this repository still names: the update event itself is -## recorded in the outbox by the base write. The update path records its notice there too (a -## DomainEvent riding the write's transaction); the targeted path publishes directly, which is why -## Producer is still needed here. +## The rekey notices are the ONLY events this repository still names; both paths record them in the +## outbox as DomainEvents riding the write's own transaction, so no bare publish remains here. import org.eclipse.dirigible.components.data.store.java.repository.DomainEvent; -import org.eclipse.dirigible.sdk.messaging.Producer; import org.eclipse.dirigible.sdk.utils.Json; #end #if(($documentChecks && $documentChecks.size() > 0) || $rollupGuard || ($guardChecks && $guardChecks.size() > 0) || $lifecycleEdges) @@ -442,9 +439,9 @@ public class ${name}Repository extends JavaRepository<${name}Entity> { *

* Still event-free for the reactions: the only event it can publish is {@code "-rekeyed"}, and only * when the write MOVED the row between groups - a topic nothing but the generated aggregate / - * roll-up handlers subscribes to. Those two notices are published directly rather than recorded in - * the outbox: the targeted write carries at most one outbox topic, and this path needs two bodies - * (the row as it stood, and the row as written). + * roll-up handlers subscribes to. Both notices are recorded in the outbox with the write itself, + * so the mutation and the recompute signals commit together - a broker outage delays the + * recompute instead of losing it. #end */ @Override @@ -531,18 +528,39 @@ public class ${name}Repository extends JavaRepository<${name}Entity> { #end } #end +#if($groupingKeys && $groupingKeys.size() > 0) + // BOTH sides of the move, on the one topic only the aggregate / roll-up handlers listen to - + // so no other reaction sees a second event, which is why this path can signal them without + // re-publishing "-updated". The previous row names the group the row LEFT (recomputed without + // it, its total drops instead of staying stale); the written row names the one it moved INTO, + // which on the targeted path gets no event of its own at all. Handed to the base write so both + // notices commit with the mutation itself - a broker outage delays the recompute instead of + // losing it - and recorded only when the row actually existed to be written. + java.util.List rekeyEvents = groupingMoved + ? java.util.List.of(new DomainEvent("${projectName}-${perspectiveName}-${name}-rekeyed", groupingPrevious), + new DomainEvent("${projectName}-${perspectiveName}-${name}-rekeyed", Json.stringify(entity))) + : java.util.List.of(); +#end #if($history == "true") // Change history for the targeted write: it is how the system writes (a roll-up total, a // workflow write-back, a stamped number), so the trail attributes it to SYSTEM - a person // reading the record must be able to tell those apart from an edit somebody made. ${name}Entity historyBefore = super.findById(id); +#if($groupingKeys && $groupingKeys.size() > 0) + int updatedCount = super.updateProperties(id, values, eventTopic, rekeyEvents); +#else int updatedCount = super.updateProperties(id, values, eventTopic); +#end if (updatedCount > 0) { History.recordUpdate(HISTORY_TABLE, id, History.SYSTEM, historyBefore, super.findById(id), HISTORY_PROPERTIES); } +#else +#if($groupingKeys && $groupingKeys.size() > 0) + int updatedCount = super.updateProperties(id, values, eventTopic, rekeyEvents); #else int updatedCount = super.updateProperties(id, values, eventTopic); #end +#end #if($documentItem) if (updatedCount > 0 && (values.containsKey("${documentItem.fkProperty}")#foreach($f in $documentItem.fields) || values.containsKey("${f.field}")#end)) { // A line written by a TARGETED primitive - a workflow setField, a glue updateProperty / @@ -563,17 +581,6 @@ public class ${name}Repository extends JavaRepository<${name}Entity> { new ${documentItem.parentEntity}Repository().recalculate(documentBefore); } } -#end -#if($groupingKeys && $groupingKeys.size() > 0) - if (updatedCount > 0 && groupingMoved) { - // BOTH sides of the move, on the one topic only the aggregate / roll-up handlers listen to - - // so no other reaction sees a second event, which is why this path can signal them without - // re-publishing "-updated". The previous row names the group the row LEFT (recomputed without - // it, its total drops instead of staying stale); the written row names the one it moved INTO, - // which on the targeted path gets no event of its own at all. - Producer.sendToTopic("${projectName}-${perspectiveName}-${name}-rekeyed", groupingPrevious); - Producer.sendToTopic("${projectName}-${perspectiveName}-${name}-rekeyed", Json.stringify(entity)); - } #end return updatedCount; } diff --git a/components/template/template-application-events-java/src/main/resources/META-INF/dirigible/template-application-events-java/events/Resolve.java.template b/components/template/template-application-events-java/src/main/resources/META-INF/dirigible/template-application-events-java/events/Resolve.java.template index 2cf2e7f0d9d..c23391b1f6f 100644 --- a/components/template/template-application-events-java/src/main/resources/META-INF/dirigible/template-application-events-java/events/Resolve.java.template +++ b/components/template/template-application-events-java/src/main/resources/META-INF/dirigible/template-application-events-java/events/Resolve.java.template @@ -126,8 +126,9 @@ public class ${className}Resolve implements MessageHandler { * longer destroy the identification, and the audit trace is the LAST thing a failure can take. * * Nothing else of the record is touched (both writes are targeted), so no "-updated" re-fires and a - * concurrent write to any other column cannot be reverted. A targeted write also publishes NO event, - * so a status that IS written announces itself - see the "-transitioned" publish below. + * concurrent write to any other column cannot be reverted. A plain targeted write publishes NO + * event, so the routing write carries the "-transitioned" topic itself - the flip and its + * announcement commit together through the outbox. */ private static void stamp(Object id, String outcome, Integer resolved#if($writesStatus == "true"), Integer status#end) { ${entity}Repository repository = new ${entity}Repository(); @@ -144,21 +145,16 @@ public class ${className}Resolve implements MessageHandler { #if($writesStatus == "true") if (status != null) { try { - repository.updateProperty(id, "${statusProperty}", status); // The status this lookup just routed the record to IS a transition, and the constructs // that react to one - `generates:` and `postings:` bound to `event: { onTransition: ... }` - // - listen on "-transitioned". The targeted write above publishes nothing at all, so - // without this the AUTOMATIC path silently did nothing while the manual one (a - // `transitions:` button, which does publish) worked: exactly the wrong way round, and with - // no log line to show it. Inside the try on purpose - a status the record could not take - // is not a transition, so a rejected move must announce nothing. Reload so the payload - // carries the committed row rather than the pre-write snapshot, the same shape the - // transition controller and the create-from completion hook publish. - ${entity}Entity transitioned = repository.findById(id); - if (transitioned != null) { - org.eclipse.dirigible.sdk.messaging.Producer.sendToTopic("${projectName}-${perspective}-${entity}-transitioned", - Json.stringify(transitioned)); - } + // - listen on "-transitioned". A plain targeted write publishes nothing at all, so the + // topic rides the routing write into the outbox: the flip and its announcement commit + // together, the payload is the row exactly as the statement left it, and a broker outage + // delays the announcement instead of losing it. Inside the try on purpose - a status the + // record could not take is not a transition, so a rejected move must announce nothing, + // and a rejected write records no event either. + repository.updateProperties(id, java.util.Map.of("${statusProperty}", status), + "${projectName}-${perspective}-${entity}-transitioned"); } catch (org.eclipse.dirigible.sdk.db.ValidationException rejected) { // The record cannot take this status where it currently stands - an unmodeled lifecycle // move, or a gate the status would trip. Retrying cannot help (nothing about the record diff --git a/components/template/template-application-events-java/src/main/resources/META-INF/dirigible/template-application-events-java/events/Transition.java.template b/components/template/template-application-events-java/src/main/resources/META-INF/dirigible/template-application-events-java/events/Transition.java.template index 321453b3732..fa301b7a4ab 100644 --- a/components/template/template-application-events-java/src/main/resources/META-INF/dirigible/template-application-events-java/events/Transition.java.template +++ b/components/template/template-application-events-java/src/main/resources/META-INF/dirigible/template-application-events-java/events/Transition.java.template @@ -20,10 +20,11 @@ import org.eclipse.dirigible.sdk.utils.Json; * Transition ${name}: the guarded on-demand status flip on a ${entity} (loaded by the id in the * posted body). The record must currently be in one of the allowed statuses [${fromStatuses}]#if($guardExpr != "") * and satisfy the guard [${guardText}]#end; otherwise 409 and the record stays untouched. On success - * ONLY the status column is written - through the targeted updateProperty primitive, a + * ONLY the status column is written - through the targeted updateProperties primitive, a * workflow-style system write (no "-updated" re-fire; a full-row update would merge the stale - * pre-check snapshot and revert concurrent writes) - and the "-transitioned" topic is published so - * posting glue / integrations observe the transition. + * pre-check snapshot and revert concurrent writes) - and the "-transitioned" notice rides that write + * into the outbox, so the flip and its announcement commit together and posting glue / integrations + * can never miss a transition whose row is durable. * * Generated from the intent transitions block - do not edit; it is re-generated with the application. * Entity access goes ONLY through the generated repositories. @@ -69,11 +70,14 @@ public class ${className}Transition { return "{\"error\": \"${name} requires ${guardText}\"}"; } #end - repository.updateProperty(req.id, "${statusProperty}", ${setStatus}); - // Reload so the "-transitioned" payload carries the committed row, not the pre-check snapshot. + // The status write and its "-transitioned" notice commit together: the topic rides the + // targeted write into the outbox, so a broker outage delays the announcement instead of losing + // it, and the event payload is the row exactly as the statement left it - never a re-read a + // concurrent write could have moved on. + repository.updateProperties(req.id, java.util.Map.of("${statusProperty}", ${setStatus}), + "${projectName}-${perspective}-${entity}-transitioned"); + // Reload so the response (and the notify block) carries the committed row, not the pre-check snapshot. source = repository.findById(req.id); - org.eclipse.dirigible.sdk.messaging.Producer.sendToTopic( - "${projectName}-${perspective}-${entity}-transitioned", Json.stringify(source)); #if($notify == "true") sendNotification(source); #end diff --git a/tests/tests-integrations/src/main/java/org/eclipse/dirigible/integration/tests/api/IntentEmissionCoverageIT.java b/tests/tests-integrations/src/main/java/org/eclipse/dirigible/integration/tests/api/IntentEmissionCoverageIT.java index 2e570ad3341..1cfefec9fe1 100644 --- a/tests/tests-integrations/src/main/java/org/eclipse/dirigible/integration/tests/api/IntentEmissionCoverageIT.java +++ b/tests/tests-integrations/src/main/java/org/eclipse/dirigible/integration/tests/api/IntentEmissionCoverageIT.java @@ -1766,10 +1766,14 @@ private void assertEmission() { && ledgerRepository.contains("!java.util.Objects.equals(groupingPreviousPerson, entity.Person)"), "the targeted write must compare every grouping key before and after the write: " + ledgerRepository); assertTrue( - ledgerRepository.contains("if (updatedCount > 0 && groupingMoved)") + ledgerRepository.contains("java.util.List rekeyEvents = groupingMoved") + && ledgerRepository.contains("super.updateProperties(id, values, eventTopic, rekeyEvents)") && ledgerRepository.contains("-rekeyed\", groupingPrevious)") && ledgerRepository.contains("-rekeyed\", Json.stringify(entity))"), - "a targeted write that moved a grouping key must publish BOTH the previous and the written row"); + "a targeted write that moved a grouping key must record BOTH the previous and the written row" + + " with the write itself (the outbox), never as a bare publish beside it"); + assertFalse(ledgerRepository.contains("Producer.sendToTopic"), + "a generated repository must announce every event through its writes - no bare publish may remain"); // Fix 2: the same move on a ROLL-UP child. Its parent FK is a grouping column too, so the child's // DAO tracks it and a roll-up handler binds "-rekeyed" - without it the parent a child was moved @@ -2280,8 +2284,12 @@ private void assertEmission() { assertTrue(transition.contains("currentStatus == 1"), "transitions must emit the allowed-statuses guard"); assertTrue(transition.contains("Calc.eval(\"Paid\", source, 6)"), "the when guard must emit a Calc comparison"); assertTrue(transition.contains("Response.setStatus(409)"), "a failed guard must surface as 409"); - assertTrue(transition.contains("updateProperty"), "the status flip must be the targeted single-column write"); - assertTrue(transition.contains("-transitioned"), "the flip must publish the -transitioned topic"); + assertTrue(transition.contains("repository.updateProperties(req.id, java.util.Map.of("), + "the status flip must be the targeted write, touching only the status column"); + assertTrue(transition.contains("-transitioned\");"), + "the -transitioned notice must ride the targeted write into the outbox, so flip and announcement commit together"); + assertFalse(transition.contains("Producer.sendToTopic"), + "the transition must not publish beside its write - a broker outage would lose the announcement"); String transitionExtension = contentOf("CancelEntry-transition-action.extension"); assertTrue(transitionExtension.contains("-custom-action"), "the transition button must contribute to the app's custom-action extension point"); @@ -2542,22 +2550,24 @@ private void assertEmission() { assertTrue(contentOf("gen/events/emission/ShipmentFlowSettleCompleted.java").contains("implements JavaDelegate"), "a create-from asking for a step moment must get that moment's emitter, even as its only consumer"); - // resolves (#6712): the lookup persists its outcome with a TARGETED write, which publishes no - // event at all - so when that write routes the record by status it has to announce the - // transition itself. Without this the automatic path wrote the status and told nobody, while a + // resolves (#6712): the lookup persists its outcome with a TARGETED write, and when that write + // routes the record by status it announces the transition by handing the "-transitioned" topic + // to the routing write itself - flip and announcement commit together through the outbox. + // Without an announcement the automatic path wrote the status and told nobody, while a // transitions: button on the same entity worked: the primary path silently dead, the fallback // fine. The consumers are bound to "-transitioned" (see the create-from above), so that is the - // channel it must publish on. + // channel the write must carry. String resolve = contentOf("gen/events/emission/AssignInspectorResolve.java"); assertTrue(resolve.contains("updateProperties("), "the lookup must persist its outcome as one targeted write"); - assertTrue(resolve.contains("-Patrol-transitioned"), - "a resolve that routes the record by status must publish the record's -transitioned topic, " + assertTrue(resolve.contains("-Patrol-transitioned\");"), + "a resolve that routes the record by status must carry the record's -transitioned topic on the routing write, " + "or nothing bound to onTransition can ever observe an automatic resolution"); - assertTrue(resolve.contains("Producer.sendToTopic"), "the resolve must publish through the messaging producer"); + assertFalse(resolve.contains("Producer.sendToTopic"), + "the resolve must not publish beside its writes - a broker outage would lose the announcement"); // Guarded on a status having been written: a lookup that only filled the relation (or found // nothing) transitioned nothing, and must not announce one. - assertTrue(resolve.indexOf("if (status != null)") < resolve.indexOf("Producer.sendToTopic"), - "the publish must sit under the status guard, so a lookup that wrote no status announces no transition"); + assertTrue(resolve.indexOf("if (status != null)") < resolve.indexOf("-Patrol-transitioned\");"), + "the announcing write must sit under the status guard, so a lookup that wrote no status announces no transition"); String reportOnEvent = contentOf("gen/events/emission/ReportFromPatrolGenerateOnEvent.java"); assertTrue(reportOnEvent.contains("-Patrol-transitioned"), "the create-from driven by the lookup must listen on the very topic the lookup publishes"); 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..507d6438e91 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 @@ -980,10 +980,15 @@ void resolve_writes_the_result_before_the_routing_status() { generateFromModel("template-application-events-java/template/template.js", "fines.glue"); String lookup = codeOf("gen/events/fines/IdentifyDriverResolve.java"); - // The result goes out on its own, and the status is NOT in that batch. + // The result goes out on its own, and the status is NOT in that batch. The routing write hands + // its "-transitioned" topic to the write itself, so the flip and its announcement commit + // together through the outbox - never a bare publish beside the write. int result = onlyIndexOf(lookup, "repository.updateProperties(id, values)"); - int routing = onlyIndexOf(lookup, "repository.updateProperty(id, \"Status\", status)"); + int routing = onlyIndexOf(lookup, "repository.updateProperties(id, java.util.Map.of(\"Status\", status)"); assertTrue(result < routing, "the resolved relation and the trace must be persisted BEFORE the routing status is attempted"); + assertTrue(lookup.contains("-Fine-Fine-transitioned\");"), "the routing write must carry the -transitioned topic into the outbox"); + assertFalse(lookup.contains("Producer.sendToTopic"), + "the lookup must not publish beside its writes - a broker outage would lose the announcement"); assertFalse(lookup.contains("values.put(\"Status\", status)"), "the status must NOT ride in the same map - a rejected move would take the relation and the trace with it"); diff --git a/tests/tests-integrations/src/main/java/org/eclipse/dirigible/integration/tests/api/JavaEventOutboxIT.java b/tests/tests-integrations/src/main/java/org/eclipse/dirigible/integration/tests/api/JavaEventOutboxIT.java index 055a9dc39b0..4377ec37c24 100644 --- a/tests/tests-integrations/src/main/java/org/eclipse/dirigible/integration/tests/api/JavaEventOutboxIT.java +++ b/tests/tests-integrations/src/main/java/org/eclipse/dirigible/integration/tests/api/JavaEventOutboxIT.java @@ -20,6 +20,8 @@ import java.sql.Timestamp; import java.time.Duration; import java.time.Instant; +import java.util.HashSet; +import java.util.Set; import java.util.UUID; import java.util.concurrent.TimeUnit; @@ -107,6 +109,37 @@ void a_write_publishes_through_the_outbox_and_a_stranded_entry_is_relayed() thro .atMost(TIMEOUT_SECONDS, TimeUnit.SECONDS) .until(() -> countOutboxEntries() == 0); + // A targeted write's EXTRA events (the shape of the generated DAO's "-rekeyed" pair) ride the + // same outbox: one mutation, two notices - the written row and the caller's own message - both + // delivered and both cleared. Without the event-carrying overload these were bare publishes a + // broker outage would swallow. + drainEchoQueue(); + restAssuredExecutor.execute(() -> given().when() + .get(CONTROLLER + "/retarget") + .then() + .statusCode(200)); + Set notices = new HashSet<>(); + Awaitility.await() + .pollInterval(1, TimeUnit.SECONDS) + .atMost(TIMEOUT_SECONDS, TimeUnit.SECONDS) + .until(() -> { + String echo = receiveEcho(); + if (echo != null) { + notices.add(echo); + } + return notices.size() >= 2; + }); + assertTrue(notices.stream() + .anyMatch(notice -> notice.contains("moved")), + "the targeted write must record the written row: " + notices); + assertTrue(notices.stream() + .anyMatch(notice -> notice.contains("previous")), + "the targeted write must record the caller's extra event with the same write: " + notices); + Awaitility.await() + .pollInterval(1, TimeUnit.SECONDS) + .atMost(TIMEOUT_SECONDS, TimeUnit.SECONDS) + .until(() -> countOutboxEntries() == 0); + // Now the failure the issue is about: an entry the broker refused, left behind by a write that // nonetheless succeeded. Nothing else will publish it - only the relay can. The queue is drained // first so the echo that arrives can only be this one. diff --git a/tests/tests-integrations/src/main/resources/JavaEventOutboxIT/things/OutboxThingController.java b/tests/tests-integrations/src/main/resources/JavaEventOutboxIT/things/OutboxThingController.java index 93c83c41eda..9962ac01b86 100644 --- a/tests/tests-integrations/src/main/resources/JavaEventOutboxIT/things/OutboxThingController.java +++ b/tests/tests-integrations/src/main/resources/JavaEventOutboxIT/things/OutboxThingController.java @@ -27,4 +27,10 @@ public String seed() { thing.name = "seeded"; return String.valueOf(things.save(thing).id); } + + @Get("/retarget") + public String retarget() { + OutboxThing thing = things.findAll().get(0); + return String.valueOf(things.move(thing.id)); + } } diff --git a/tests/tests-integrations/src/main/resources/JavaEventOutboxIT/things/OutboxThingRepository.java b/tests/tests-integrations/src/main/resources/JavaEventOutboxIT/things/OutboxThingRepository.java index aeab661deeb..abbfae9f0dd 100644 --- a/tests/tests-integrations/src/main/resources/JavaEventOutboxIT/things/OutboxThingRepository.java +++ b/tests/tests-integrations/src/main/resources/JavaEventOutboxIT/things/OutboxThingRepository.java @@ -9,6 +9,7 @@ */ package things; +import org.eclipse.dirigible.components.data.store.java.repository.DomainEvent; import org.eclipse.dirigible.components.data.store.java.repository.JavaRepository; import org.eclipse.dirigible.sdk.component.Repository; @@ -29,4 +30,13 @@ public OutboxThingRepository() { public OutboxThing save(OutboxThing thing) { return super.save(thing, CREATED_TOPIC); } + + /** + * A targeted write carrying an extra event beside the row's own — the shape of the generated + * DAO's "-rekeyed" pair: one mutation, two notices, all committed together. + */ + public int move(Object id) { + return super.updateProperties(id, java.util.Map.of("name", "moved"), CREATED_TOPIC, + java.util.List.of(new DomainEvent(CREATED_TOPIC, "{\"name\":\"previous\"}"))); + } }