diff --git a/components/template/template-application-events-java/src/main/resources/META-INF/dirigible/template-application-events-java/events/Posts.java.template b/components/template/template-application-events-java/src/main/resources/META-INF/dirigible/template-application-events-java/events/Posts.java.template index d971012e4b6..8a4f6ff704e 100644 --- a/components/template/template-application-events-java/src/main/resources/META-INF/dirigible/template-application-events-java/events/Posts.java.template +++ b/components/template/template-application-events-java/src/main/resources/META-INF/dirigible/template-application-events-java/events/Posts.java.template @@ -18,7 +18,10 @@ import org.eclipse.dirigible.sdk.utils.Json; * - binds the source's #if($isCreate)create (bare entity)#{else}-transitioned#end topic and RE-LOADS the source by * id (the payload is as-of the event and may lack later-step data); * - idempotent by ${backRef}: if ${into} rows already back-reference this source, it is a no-op; - * - all writes go through the generated ${into} repository (so its own create logic fires). + * - all writes go through the generated ${into} repository (so its own create logic fires), and the + * rows one source event derives are ONE transaction, so the back-reference either finds the whole + * post or finds nothing at all - a tick that fails part-way leaves nothing behind to be mistaken + * for a finished post (dirigible #7179). */ @Component("${javaGenFolderName}_${className}Post") public class ${className}Post implements MessageHandler { @@ -55,7 +58,10 @@ public class ${className}Post implements MessageHandler { gen.${javaGenFolderName}.data.${targetJavaPerspective}.${into}Repository targetRepository = new gen.${javaGenFolderName}.data.${targetJavaPerspective}.${into}Repository(); #if($backRef != "") - // Idempotency: a row already back-referencing this source means the post ran - no-op. + // Idempotency: a row already back-referencing this source means the post ran - no-op. Mere + // EXISTENCE is an exact test only because the rows below are written as one transaction: a + // failed tick commits none of them, so the back-reference is present for a whole post and + // absent for no post - there is no partial set for this guard to read as finished (#7179). if (!targetRepository.findAll(Criteria.create().eq("${backRef}", source.${sourceKeyField})).isEmpty()) { return; } @@ -63,6 +69,8 @@ public class ${className}Post implements MessageHandler { #if($perItem) gen.${javaGenFolderName}.data.${itemsJavaPerspective}.${itemsEntity}Repository itemsRepository = new gen.${javaGenFolderName}.data.${itemsJavaPerspective}.${itemsEntity}Repository(); + // Derive every row first - reads and in-memory mapping only, nothing written yet. + java.util.List rows = new java.util.ArrayList<>(); for (gen.${javaGenFolderName}.data.${itemsJavaPerspective}.${itemsEntity}Entity item : itemsRepository.findAll(Criteria.create().eq("${itemsFk}", source.${sourceKeyField}))) { gen.${javaGenFolderName}.data.${targetJavaPerspective}.${into}Entity row = @@ -73,8 +81,19 @@ public class ${className}Post implements MessageHandler { #if($backRef != "") row.${backRef} = source.${sourceKeyField}; #end - targetRepository.save(row); + rows.add(row); } + // ONE transaction for every row this source event derives. Written one save per transaction, a + // row the repository refused (a validation, a constraint) left the earlier rows durable - and + // since the guard above reads the back-reference as mere existence, every redelivery afterwards + // was a no-op and the missing rows were never written: the half-post was PERMANENT + // (dirigible #7179). All of them commit or none does, so the redelivery finds nothing and + // writes the whole set. + org.eclipse.dirigible.components.data.store.java.repository.UnitOfWork.run(() -> { + for (gen.${javaGenFolderName}.data.${targetJavaPerspective}.${into}Entity row : rows) { + targetRepository.save(row); + } + }); #else gen.${javaGenFolderName}.data.${targetJavaPerspective}.${into}Entity row = new gen.${javaGenFolderName}.data.${targetJavaPerspective}.${into}Entity(); @@ -84,6 +103,7 @@ public class ${className}Post implements MessageHandler { #if($backRef != "") row.${backRef} = source.${sourceKeyField}; #end + // One row, one repository call - a transaction on its own, so no unit of work is needed here. targetRepository.save(row); #end } 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 d0f2c255ff3..e5e4864896b 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 @@ -3348,6 +3348,104 @@ void postings_generates_the_idempotent_resumable_handler() { assertFalse(posting.contains("was NOT rewritten"), "a target with no status lifecycle is always rewritable"); } + @Test + void posts_writes_every_row_of_one_source_event_in_one_transaction() { + // #7179: the FLAT per-item mode (posts:, no header document) had the same multi-write shape as + // the posting rewrite and no unit of work - one save per row, one transaction each. A row the + // repository refused left the rows before it durable, and the guard here is coarser than the + // posting's: it asks whether ANY row back-references this source, so the partial set read as a + // finished post and no redelivery ever wrote the rest. The half-post was PERMANENT. The rows + // are derived first and written together, so the guard sees a whole post or nothing. + String yaml = """ + name: poststest + entities: + - name: GoodsIssueStatus + kind: setting + fields: + - { name: id, type: integer, primaryKey: true, generated: true } + - { name: name, type: string, required: true, length: 100 } + - name: Product + fields: + - { name: id, type: integer, primaryKey: true, generated: true } + - { name: name, type: string, required: true, length: 100 } + - name: GoodsIssue + fields: + - { name: id, type: integer, primaryKey: true, generated: true } + - { name: number, type: string, length: 40 } + relations: + - { name: Status, kind: manyToOne, to: GoodsIssueStatus, function: EntityStatus, init: 1 } + - name: GoodsIssueItem + fields: + - { name: id, type: integer, primaryKey: true, generated: true } + - { name: quantity, type: decimal, precision: 18, scale: 3 } + relations: + - { name: GoodsIssue, kind: manyToOne, to: GoodsIssue, composition: true, required: true } + - { name: Product, kind: manyToOne, to: Product } + - name: StockMovement + fields: + - { name: id, type: integer, primaryKey: true, generated: true } + - { name: quantity, type: decimal, precision: 18, scale: 3 } + relations: + - { name: Product, kind: manyToOne, to: Product } + - { name: GoodsIssue, kind: manyToOne, to: GoodsIssue } + - name: StockNote + fields: + - { name: id, type: integer, primaryKey: true, generated: true } + - { name: note, type: string, length: 100 } + relations: + - { name: GoodsIssue, kind: manyToOne, to: GoodsIssue } + posts: + - name: goodsIssueLedger + forEntity: GoodsIssue + event: 2 + forEach: items + into: StockMovement + idempotentBy: GoodsIssue + set: + Product: item.Product + Quantity: "-item.Quantity" + - name: goodsIssueNote + forEntity: GoodsIssue + event: create + into: StockNote + idempotentBy: GoodsIssue + set: + Note: source.Number + """; + writeIntent(yaml); + restAssuredExecutor.execute(() -> given().when() + .post(GENERATE_URL) + .then() + .statusCode(200)); + + String glue = contentOf("poststest.glue"); + assertTrue(glue.contains("\"posts\""), "the .glue should carry the posts collection"); + assertTrue(glue.contains("GoodsIssueLedger"), "the post className should be carried in the glue"); + + generateFromModel("template-application-events-java/template/template.js", "poststest.glue"); + String post = codeOf("gen/events/poststest/GoodsIssueLedgerPost.java"); + assertTrue(post.contains("implements MessageHandler"), "the post is a self-describing message handler"); + assertTrue(post.contains("-transitioned"), "a status-triggered post listens on the source's -transitioned channel"); + assertTrue(post.contains("Criteria.create().eq(\"GoodsIssue\", source.Id)"), "the guard asks the back-reference on the target"); + // Asserted by POSITION, since a save left inside the derivation loop would still "mention + // UnitOfWork": every row is mapped in memory first, and the ONE save site sits inside the block. + int derived = post.indexOf("rows.add(row)"); + int unitOfWork = post.indexOf("UnitOfWork.run(() -> {"); + int save = post.indexOf("targetRepository.save(row)"); + assertTrue(derived > 0, "the rows must be derived into a list before anything is written"); + assertTrue(unitOfWork > derived, "the unit of work must open after the derivation, not around the reads"); + assertTrue(save > unitOfWork, "every row must be saved inside the unit of work"); + assertEquals(save, post.lastIndexOf("targetRepository.save(row)"), + "there must be exactly ONE save site - a second one outside the block would write rows unprotected"); + assertFalse(post.contains("${"), "the post template must render every placeholder"); + + // The single-row mode (no forEach) writes one row through one repository call - a transaction on + // its own, so it needs no unit of work and must not pretend to open one. + String single = codeOf("gen/events/poststest/GoodsIssueNotePost.java"); + assertTrue(single.contains("targetRepository.save(row)"), "the single-row post writes its one row"); + assertFalse(single.contains("UnitOfWork"), "one repository call is already one transaction"); + } + @Test void a_post_is_not_rewritten_once_the_created_document_has_left_the_status_it_was_created_in() { // #7071: an amended source (rejected, edited, re-issued) raises the SAME moment again, and the diff --git a/tests/tests-integrations/src/main/java/org/eclipse/dirigible/integration/tests/api/IntentPostsAtomicityIT.java b/tests/tests-integrations/src/main/java/org/eclipse/dirigible/integration/tests/api/IntentPostsAtomicityIT.java new file mode 100644 index 00000000000..8875e8d1a13 --- /dev/null +++ b/tests/tests-integrations/src/main/java/org/eclipse/dirigible/integration/tests/api/IntentPostsAtomicityIT.java @@ -0,0 +1,272 @@ +/* + * Copyright (c) 2010-2026 Eclipse Dirigible contributors + * + * All rights reserved. This program and the accompanying materials are made available under the + * terms of the Eclipse Public License v2.0 which accompanies this distribution, and is available at + * http://www.eclipse.org/legal/epl-v20.html + * + * SPDX-FileCopyrightText: Eclipse Dirigible contributors SPDX-License-Identifier: EPL-2.0 + */ +package org.eclipse.dirigible.integration.tests.api; + +import static io.restassured.RestAssured.given; +import static org.hamcrest.Matchers.equalTo; +import static org.hamcrest.Matchers.greaterThanOrEqualTo; +import static org.junit.jupiter.api.Assertions.assertEquals; + +import java.nio.charset.StandardCharsets; +import java.util.List; +import java.util.Map; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; + +import org.eclipse.dirigible.components.initializers.synchronizer.SynchronizationProcessor; +import org.eclipse.dirigible.repository.api.IRepository; +import org.eclipse.dirigible.repository.api.IRepositoryStructure; +import org.eclipse.dirigible.repository.api.IResource; +import org.eclipse.dirigible.tests.base.IntegrationTest; +import org.eclipse.dirigible.tests.framework.restassured.RestAssuredExecutor; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.Tag; +import org.junit.jupiter.api.Test; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.test.annotation.DirtiesContext; + +/** + * The flat per-item post ({@code posts:} with {@code forEach:}) writes the rows one source event + * derives as ONE transaction - at runtime, through the published application. + * + *

+ * Each row used to be saved in its own transaction, so a row the target repository refused (here a + * required column the derived row leaves null) left the rows before it durable. This mode's + * idempotency guard is coarser than the posting's - it asks whether ANY row back-references the + * source - so that partial set read as a finished post and no redelivery ever wrote the rest: the + * half-post was PERMANENT (issue #7179). The scenario below is exactly that sequence: a refused + * row, the cause repaired, the event redelivered - and the ledger must end up carrying the whole + * post. + */ +@DirtiesContext(classMode = DirtiesContext.ClassMode.AFTER_CLASS) +@Tag("slow") +class IntentPostsAtomicityIT extends IntegrationTest { + + private static final String WORKSPACE = "workspace"; + private static final String PROJECT = "postsatomic"; + private static final String PROJECT_PATH = IRepositoryStructure.PATH_USERS + "/admin/" + WORKSPACE + "/" + PROJECT; + private static final String API = "/services/java/" + PROJECT + "/gen/" + PROJECT + "/api"; + private static final String TRANSITION = "/services/java/" + PROJECT + "/gen/events/" + PROJECT + "/"; + private static final long TIMEOUT_SECONDS = 90; + + /** + * A goods issue whose lines post one stock movement each. {@code StockMovement.quantity} is + * REQUIRED, which is how a derived row gets refused: an item with no quantity derives a movement + * the target repository will not accept. The sibling {@code posts:} rule into {@code StockNote} is + * the observable proof that the event was delivered and consumed, so "no movement rows" can be + * asserted at a point where the handler has demonstrably run. + */ + private static final String INTENT_YAML = """ + name: postsatomic + description: flat per-item post fixture - a refused row leaves nothing behind + + entities: + - name: GoodsIssueStatus + kind: setting + fields: + - { name: id, type: integer, primaryKey: true, generated: true } + - { name: name, type: string, required: true, length: 100 } + + - name: GoodsIssue + fields: + - { name: id, type: integer, primaryKey: true, generated: true } + - { name: number, type: string, length: 40 } + relations: + - { name: Status, kind: manyToOne, to: GoodsIssueStatus, function: EntityStatus, init: 1 } + + - name: GoodsIssueItem + fields: + - { name: id, type: integer, primaryKey: true, generated: true } + - { name: quantity, type: decimal, precision: 18, scale: 2 } + relations: + - { name: GoodsIssue, kind: manyToOne, to: GoodsIssue, composition: true, required: true } + + - name: StockMovement + fields: + - { name: id, type: integer, primaryKey: true, generated: true } + - { name: quantity, type: decimal, precision: 18, scale: 2, required: true } + relations: + - { name: GoodsIssue, kind: manyToOne, to: GoodsIssue } + + - name: StockNote + fields: + - { name: id, type: integer, primaryKey: true, generated: true } + - { name: note, type: string, length: 40 } + relations: + - { name: GoodsIssue, kind: manyToOne, to: GoodsIssue } + + transitions: + - { name: PostGoodsIssue, forEntity: GoodsIssue, from: [1], setStatus: 2, label: Post, icon: check } + - { name: ReopenGoodsIssue, forEntity: GoodsIssue, from: [2], setStatus: 3, label: Reopen, icon: undo } + - { name: RepostGoodsIssue, forEntity: GoodsIssue, from: [3], setStatus: 2, label: Repost, icon: check } + + posts: + - name: goodsIssueLedger + forEntity: GoodsIssue + event: 2 + forEach: items + into: StockMovement + idempotentBy: GoodsIssue + set: + Quantity: item.Quantity + - name: goodsIssueNote + forEntity: GoodsIssue + event: 2 + into: StockNote + idempotentBy: GoodsIssue + set: + Note: source.Number + + seeds: + - name: goods-issue-statuses + entity: GoodsIssueStatus + rows: + - { id: 1, name: Draft } + - { id: 2, name: Posted } + - { id: 3, name: Reopened } + """; + + @Autowired + private IRepository repository; + + @Autowired + private RestAssuredExecutor restAssuredExecutor; + + @Autowired + private SynchronizationProcessor synchronizationProcessor; + + @Test + void a_refused_row_leaves_no_post_and_the_redelivery_writes_the_whole_one() { + generateProject(); + publishProject(); + synchronizationProcessor.forceProcessSynchronizers(); + + // A goods issue with two lines, the second one missing its quantity - so the movement derived + // from it is a row the target repository must refuse. + int issue = create("/goodsissue/GoodsIssueController", "{\"Number\":\"GI-1\"}"); + create("/goodsissue/GoodsIssueItemController", "{\"GoodsIssue\":" + issue + ",\"Quantity\":5}"); + int blank = create("/goodsissue/GoodsIssueItemController", "{\"GoodsIssue\":" + issue + "}"); + + // Post it: the event is raised, both handlers hear it, and the ledger's second row is refused. + transition("PostGoodsIssue", issue); + // The sibling post's row is the delivery receipt - once it exists, the event has been consumed. + awaitCount("/stocknote/StockNoteController", issue, 1); + + // ...and the ledger carries NOTHING. One row per transaction, this was a durable single row - + // which the guard below then reads as a finished post. + assertCount("/stockmovement/StockMovementController", issue, 0); + + // Repair the line and redeliver the event (reopen, post again). With a partial set on file the + // guard answers "already posted" and this is where the missing row was lost for good. + update("/goodsissue/GoodsIssueItemController/" + blank, "{\"Id\":" + blank + ",\"GoodsIssue\":" + issue + ",\"Quantity\":3}"); + transition("ReopenGoodsIssue", issue); + transition("RepostGoodsIssue", issue); + + // The whole post, exactly once: one movement per line, and no duplicate of the row that had + // succeeded on the failed tick. + awaitCount("/stockmovement/StockMovementController", issue, 2); + restAssuredExecutor.execute(() -> given().when() + .get(API + "/stockmovement/StockMovementController") + .then() + .statusCode(200) + .body("findAll { it.GoodsIssue == " + issue + " }.Quantity.sum()", equalTo(8.0))); + } + + private void transition(String name, int id) { + restAssuredExecutor.execute(() -> given().contentType("application/json") + .body("{\"id\":" + id + "}") + .when() + .post(TRANSITION + name + "Transition/run") + .then() + .statusCode(200)); + } + + /** The row count against one goods issue, retried until it holds (the post is asynchronous). */ + private void awaitCount(String controller, int issue, int expected) { + restAssuredExecutor.execute(() -> assertCount(controller, issue, expected), TIMEOUT_SECONDS); + } + + private void assertCount(String controller, int issue, int expected) { + restAssuredExecutor.execute(() -> given().when() + .get(API + controller) + .then() + .statusCode(200) + .body("findAll { it.GoodsIssue == " + issue + " }.size()", equalTo(expected))); + } + + private int create(String controller, String body) { + AtomicInteger id = new AtomicInteger(); + restAssuredExecutor.execute(() -> id.set(given().contentType("application/json") + .body(body) + .when() + .post(API + controller) + .then() + .statusCode(200) + .extract() + .path("Id")), + TIMEOUT_SECONDS); + return id.get(); + } + + private void update(String controller, String body) { + restAssuredExecutor.execute(() -> given().contentType("application/json") + .body(body) + .when() + .put(API + controller) + .then() + .statusCode(200)); + } + + private void generateProject() { + writeIntent(); + AtomicReference>> plan = new AtomicReference<>(); + restAssuredExecutor.execute(() -> plan.set(given().when() + .post("/services/ide/intent/generate?workspace=" + WORKSPACE + "&project=" + + PROJECT + "&path=app.intent") + .then() + .statusCode(200) + .extract() + .jsonPath() + .getList("codeGenerations"))); + for (Map codeGeneration : plan.get()) { + assertEquals(Boolean.TRUE, codeGeneration.get("generated"), + "generating code from " + codeGeneration.get("path") + " failed: " + codeGeneration.get("error")); + } + } + + private void publishProject() { + restAssuredExecutor.execute(() -> given().when() + .post("/services/ide/publisher/" + WORKSPACE + "/" + PROJECT + "/") + .then() + .statusCode(200)); + } + + private void writeIntent() { + String path = PROJECT_PATH + "/app.intent"; + IResource existing = repository.getResource(path); + if (existing.exists()) { + existing.setContent(INTENT_YAML.getBytes(StandardCharsets.UTF_8)); + } else { + repository.createResource(path, INTENT_YAML.getBytes(StandardCharsets.UTF_8)); + } + } + + @AfterEach + void cleanup() { + restAssuredExecutor.execute(() -> given().when() + .delete("/services/ide/publisher/" + WORKSPACE + "/" + PROJECT) + .then() + .statusCode(greaterThanOrEqualTo(200))); + if (repository.hasCollection(PROJECT_PATH)) { + repository.removeCollection(PROJECT_PATH); + } + synchronizationProcessor.forceProcessSynchronizers(); + } +}