From ee033651dc667ae6999189c6d0360b8877576594 Mon Sep 17 00:00:00 2001 From: delchev Date: Wed, 9 Sep 2026 14:39:21 +0300 Subject: [PATCH] intent: the flat per-item post is one transaction, so a refused row leaves nothing (#7179) Posts.java.template's per-item mode (posts: with forEach:) had the same multi-write shape PR #7167 fixed in the document-shaped posting and no unit of work: one save per derived row, one transaction each. A row the target repository refused - a validation, a constraint, a required column the derived row leaves null - left the rows before it durable. That is worse here than in the posting, because this mode's idempotency guard is coarser: it asks whether ANY row back-references the source, so the partial set read as a finished post and every redelivery afterwards was a no-op. The missing rows were never written by anything. The half-post was PERMANENT. The rows one source event derives are now mapped in memory first and written inside one UnitOfWork.run(...): all of them commit or none does, so a failed tick leaves nothing and the redelivery writes the whole set. That also makes the existing guard exact rather than needing to be replaced - with atomicity the back-reference is present for a whole post and absent for no post, and there is no partial set left for it to misread. The single-row mode keeps its one save: one repository call is already one transaction. Verification is a new IntentPostsAtomicityIT, the runtime half this mode never had - it publishes the app (which compiles the generated handler) and drives the sequence the issue describes: a goods issue whose second line carries no quantity derives a movement the target refuses, the ledger must hold NOTHING, the line is then repaired and the event redelivered, and the ledger must hold the WHOLE post. Against the unfixed template that run ends with exactly one durable row and stays there. IntentEngineIT covers the emission by POSITION (derivation before the unit, the sole save site inside it, no unit in the single-row mode) - the flat mode had no rendering coverage at all until now. Fixes #7179 Co-Authored-By: Claude Opus 5 --- .../events/Posts.java.template | 26 +- .../integration/tests/api/IntentEngineIT.java | 98 +++++++ .../tests/api/IntentPostsAtomicityIT.java | 272 ++++++++++++++++++ 3 files changed, 393 insertions(+), 3 deletions(-) create mode 100644 tests/tests-integrations/src/main/java/org/eclipse/dirigible/integration/tests/api/IntentPostsAtomicityIT.java 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 5b678273ae9..e069b6c0791 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 @@ -3333,6 +3333,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(); + } +}