diff --git a/teaql-core/src/main/java/io/teaql/core/BaseEntity.java b/teaql-core/src/main/java/io/teaql/core/BaseEntity.java index d2fa2f4f..f682479e 100644 --- a/teaql-core/src/main/java/io/teaql/core/BaseEntity.java +++ b/teaql-core/src/main/java/io/teaql/core/BaseEntity.java @@ -427,6 +427,7 @@ public EntityMutationLedger getEntityMutationLedger() { public boolean __internalHasMutationLedger() { return entityMutationLedger != null; } public void setEntityMutationLedger(EntityMutationLedger entityMutationLedger) { + if (this.entityMutationLedger != null) this.entityMutationLedger.requireSaveable(); this.entityMutationLedger = entityMutationLedger; if (entityMutationLedger != null && id != null && newItem()) { entityMutationLedger.markAsNew(new EntityKey(typeName(), id)); diff --git a/teaql-core/src/main/java/io/teaql/core/EntityMutationLedger.java b/teaql-core/src/main/java/io/teaql/core/EntityMutationLedger.java index d1cdd4cd..18bb83af 100644 --- a/teaql-core/src/main/java/io/teaql/core/EntityMutationLedger.java +++ b/teaql-core/src/main/java/io/teaql/core/EntityMutationLedger.java @@ -16,6 +16,7 @@ public class EntityMutationLedger { private final Set newKeys = new TreeSet<>(); private final Map> traceChains = new TreeMap<>(); private final Map originalVersions = new TreeMap<>(); + private boolean committedStateRequiresReload; // --- Change Set Stack --- @@ -41,6 +42,19 @@ public void clearCurrentChangeSet() { traceChains.clear(); } + /** A provider write completed, but the in-memory graph could not be reconciled. */ + public void markCommittedStateRequiresReload() { + committedStateRequiresReload = true; + } + + /** Prevents a stale graph from replaying a write whose outcome is already committed. */ + public void requireSaveable() { + if (committedStateRequiresReload) { + throw new TeaQLRuntimeException( + "COMMITTED_STATE_RELOAD_REQUIRED: reload the entity graph before another save"); + } + } + public void set(EntityKey key, String field, Object value) { changeSets.set(key, field, value); } @@ -141,12 +155,14 @@ public Long getOriginalVersion(EntityKey key) { * for a read-only entity, which must retain its private ledger ownership. */ public boolean mergeEntityFrom(EntityMutationLedger other, EntityKey key) { + requireSaveable(); if (other == null || other == this) return false; Map fields = other.currentChangeSet().changes().get(key); var dynamic = other.currentChangeSet().dynamicChanges().get(key); boolean pending = (fields != null && !fields.isEmpty()) || other.newKeys.contains(key) || other.deletedKeys.contains(key) || other.recoveredKeys.contains(key) || (dynamic != null && !dynamic.isEmpty()); if (!pending) return false; + other.requireSaveable(); if (fields != null) fields.forEach((field, value) -> set(key, field, value)); if (dynamic != null) dynamic.values().forEach(mutation -> setDynamic(key, mutation)); if (other.deletedKeys.contains(key)) markAsDelete(key); @@ -162,7 +178,9 @@ public boolean mergeEntityFrom(EntityMutationLedger other, EntityKey key) { * Used when saving an entity graph (e.g., Order + OrderItems). */ public void mergeFrom(EntityMutationLedger other) { + requireSaveable(); if (other == null) return; + other.requireSaveable(); // Merge change sets EntityChangeSet otherChangeSet = other.currentChangeSet(); diff --git a/teaql-core/src/test/java/io/teaql/core/EntityMutationLedgerTest.java b/teaql-core/src/test/java/io/teaql/core/EntityMutationLedgerTest.java index d95112d9..9f5476c3 100644 --- a/teaql-core/src/test/java/io/teaql/core/EntityMutationLedgerTest.java +++ b/teaql-core/src/test/java/io/teaql/core/EntityMutationLedgerTest.java @@ -13,6 +13,23 @@ public class EntityMutationLedgerTest { private static final EntityKey ORDER = new EntityKey("Order", 1L); private static final EntityKey OTHER_ORDER = new EntityKey("Order", 2L); + @Test + public void committedReconciliationFailureCannotBeClearedOrMergedIntoAnotherLedger() { + EntityMutationLedger failed = new EntityMutationLedger(); + failed.markAsNew(ORDER); + failed.markCommittedStateRequiresReload(); + assertThrows(TeaQLRuntimeException.class, failed::requireSaveable); + + EntityMutationLedger independent = new EntityMutationLedger(); + assertThrows(TeaQLRuntimeException.class, + () -> independent.mergeEntityFrom(failed, ORDER)); + assertThrows(TeaQLRuntimeException.class, () -> independent.mergeFrom(failed)); + + failed.clearCurrentChangeSet(); + assertThrows(TeaQLRuntimeException.class, failed::requireSaveable); + independent.requireSaveable(); + } + @Test public void markAsNewRecordsOnlyRequestedKeyAndIsIdempotent() { EntityMutationLedger root = new EntityMutationLedger(); diff --git a/teaql-runtime/src/main/java/io/teaql/runtime/CommittedStateReconciliationException.java b/teaql-runtime/src/main/java/io/teaql/runtime/CommittedStateReconciliationException.java new file mode 100644 index 00000000..39f57f21 --- /dev/null +++ b/teaql-runtime/src/main/java/io/teaql/runtime/CommittedStateReconciliationException.java @@ -0,0 +1,11 @@ +package io.teaql.runtime; + +import io.teaql.core.TeaQLRuntimeException; + +/** The provider write completed, but the in-memory graph requires a fresh load. */ +public final class CommittedStateReconciliationException extends TeaQLRuntimeException { + public CommittedStateReconciliationException() { + super("COMMITTED_STATE_RELOAD_REQUIRED: provider write completed but entity reconciliation failed; " + + "reload the graph and do not retry this instance"); + } +} diff --git a/teaql-runtime/src/main/java/io/teaql/runtime/TeaQLRuntime.java b/teaql-runtime/src/main/java/io/teaql/runtime/TeaQLRuntime.java index c8b5ee91..39d15d19 100644 --- a/teaql-runtime/src/main/java/io/teaql/runtime/TeaQLRuntime.java +++ b/teaql-runtime/src/main/java/io/teaql/runtime/TeaQLRuntime.java @@ -457,9 +457,10 @@ public void saveGraph(UserContext context, Entity entity) { "teaql.mutation.kind", "save"))); try { MutationIntent intent = MutationIntent.of(entity.getComment()); - checkAndFix(context, entity); // Get entity's own EntityMutationLedger EntityMutationLedger entityMutationLedger = ((BaseEntity) entity).getEntityMutationLedger(); + entityMutationLedger.requireSaveable(); + checkAndFix(context, entity); // Allocate identifiers before roots are merged. New children commonly // receive their field updates while their id is still null, so those @@ -573,13 +574,30 @@ public void saveGraph(UserContext context, Entity entity) { Collections.newSetFromMap(new IdentityHashMap<>())); throw failure; } - dynamicReadbacks.forEach((key, values) -> { - BaseEntity target = realEntities.get(key); - target.__internalHydrateDynamicFields(values, target.__internalLoadState().withDynamicSelection(values.selectedCodes())); - }); - completeLedgerPlan(completed); - entityMutationLedger.clearCurrentChangeSet(); - deliverCommittedAuditEvents(context, completed); + CommittedStateReconciliationException reconciliationOutcome = null; + try { + dynamicReadbacks.forEach((key, values) -> { + BaseEntity target = realEntities.get(key); + target.__internalHydrateDynamicFields(values, target.__internalLoadState().withDynamicSelection(values.selectedCodes())); + }); + completeLedgerPlan(completed); + entityMutationLedger.clearCurrentChangeSet(); + } catch (RuntimeException reconciliationFailure) { + // The provider has returned from its transaction. The graph may be + // partially reconciled, so neither its pending ledger nor its + // entity instances are safe to save again without a fresh load. + entityMutationLedger.markCommittedStateRequiresReload(); + reconciliationOutcome = new CommittedStateReconciliationException(); + } + try { + // The provider has already committed. A failed in-memory readback + // must not silently suppress the independent audit delivery. + deliverCommittedAuditEvents(context, completed); + } catch (AuditDeliveryAfterCommitException auditFailure) { + if (reconciliationOutcome == null) throw auditFailure; + reconciliationOutcome.addSuppressed(auditFailure); + } + if (reconciliationOutcome != null) throw reconciliationOutcome; telemetryScope.success(); } catch (RuntimeException | Error error) { telemetryScope.failure(error); @@ -1116,7 +1134,8 @@ private void deliverCommittedAuditEvents(UserContext context, List changedValues, MutationGovernanceSnapshot governance, MutationIntent intent, List traceChain, - SqlIntentRedactions redactions) { + SqlIntentRedactions redactions, Long committedVersion) { List changes = new ArrayList<>(); if (changedValues != null) { for (Map.Entry entry : changedValues.entrySet()) { @@ -1463,7 +1482,7 @@ private int emitAuditEvent( context.getAttribute(GeneratedSchemaBootstrap.AUDIT_ACTOR_ATTRIBUTE, String.class), context.getAttribute(GeneratedSchemaBootstrap.AUDIT_CATEGORY_ATTRIBUTE, String.class), intent.auditReason(), - entity.getVersion(), + committedVersion, java.time.Instant.now(), governance); diff --git a/teaql-sqlite/src/test/java/io/teaql/sqlite/GraphTraceSqliteTest.java b/teaql-sqlite/src/test/java/io/teaql/sqlite/GraphTraceSqliteTest.java index c9c833e8..e8b366eb 100644 --- a/teaql-sqlite/src/test/java/io/teaql/sqlite/GraphTraceSqliteTest.java +++ b/teaql-sqlite/src/test/java/io/teaql/sqlite/GraphTraceSqliteTest.java @@ -30,6 +30,7 @@ public class GraphTraceSqliteTest { public static final class GraphEntity extends BaseEntity { private final String type; private final Map values = new HashMap<>(); + boolean failPostCommitReconcile; public GraphEntity(String type) { this.type = type; } @Override public String typeName() { return type; } @Override public Object __internalGet(String field) { @@ -40,6 +41,9 @@ public static final class GraphEntity extends BaseEntity { ? values.get(field) : super.__internalGet(field); } @Override public void __internalSet(String field, Object value) { + if (failPostCommitReconcile && field.equals("version")) { + throw new IllegalStateException("diagnostic reconciliation failure"); + } if (field.equals("name") || field.equals("memo") || field.equals("children") || field.equals("business_counter") || field.equals("planned_date") || field.equals("active") || field.equals("ratio") || field.equals("optional_note") @@ -51,6 +55,7 @@ public static final class GraphEntity extends BaseEntity { static final class Fixture { final List sql = new CopyOnWriteArrayList<>(); final List audit = new CopyOnWriteArrayList<>(); + final List rawAudit = new CopyOnWriteArrayList<>(); final List commands = new CopyOnWriteArrayList<>(); final List results = new CopyOnWriteArrayList<>(); boolean queryLogging = true; @@ -170,6 +175,7 @@ static final class Fixture { @Override public void writeAuditEvent(UserContext caller, RawAuditEvent event) { if (failStandardAudit) throw new IllegalStateException("standard audit sink failure"); + rawAudit.add(event); } }); if (committedRecorder != null) builder.committedMutationRecorder(committedRecorder); @@ -184,7 +190,7 @@ static final class Fixture { } void clear() { - sql.clear(); audit.clear(); commands.clear(); results.clear(); itemInsertBatchSizes.clear(); + sql.clear(); audit.clear(); rawAudit.clear(); commands.clear(); results.clear(); itemInsertBatchSizes.clear(); itemUpdateBatchSizes.clear(); itemDeleteBatchSizes.clear(); itemRecoverBatchSizes.clear(); } @@ -721,6 +727,93 @@ private static CheckedSave checkedSave(Fixture fixture, GraphEntity entity, Stri .outboxAfter("java-graph-source", 0, 10).size()); } + @Test public void postCommitReconciliationFailureMustReportCommittedOutcome() throws Exception { + var fixture = new Fixture(true); + var order = fixture.create("CustomerOrder", 196, "reconcile after commit"); + order.failPostCommitReconcile = true; + + var failure = assertThrows(CommittedStateReconciliationException.class, + () -> order.auditAs("diagnose committed reconciliation failure").save(fixture.context)); + assertEquals(1, fixture.driver.queryForList( + "SELECT id FROM customer_order_data WHERE id = ?", new Object[] {196L}).size()); + assertEquals(1, fixture.committedRecorder.eventStore() + .outboxAfter("java-graph-source", 0, 10).size()); + assertEquals("committed mutations still require application audit delivery", 1, fixture.audit.size()); + assertEquals(1, fixture.rawAudit.size()); + assertEquals(Long.valueOf(1), fixture.rawAudit.get(0).resultingVersion()); + assertTrue("committed outcome must be explicit to prevent blind retry", + failure.getMessage().contains("provider write completed")); + order.failPostCommitReconcile = false; + var retryFailure = assertThrows(TeaQLRuntimeException.class, + () -> order.auditAs("retry committed reconciliation failure").save(fixture.context)); + assertTrue(retryFailure.getMessage().contains("COMMITTED_STATE_RELOAD_REQUIRED")); + assertThrows(TeaQLRuntimeException.class, + () -> order.setEntityMutationLedger(new EntityMutationLedger())); + assertEquals(1, fixture.committedRecorder.eventStore() + .outboxAfter("java-graph-source", 0, 10).size()); + assertEquals(1, fixture.audit.size()); + var independent = fixture.create("Payment", 195, "unrelated graph may still save"); + independent.auditAs("save independent graph after failure").save(fixture.context); + assertEquals(2, fixture.committedRecorder.eventStore() + .outboxAfter("java-graph-source", 0, 10).size()); + } + + @Test public void postCommitChildReconciliationFailurePoisonsOnlyItsSaveGraph() throws Exception { + var fixture = new Fixture(true); + var order = fixture.create("CustomerOrder", 194, "root already committed"); + var item = fixture.create("OrderItem", 294, "child needs reload"); + item.failPostCommitReconcile = true; + order.__internalSet("children", List.of(item)); + + assertThrows(CommittedStateReconciliationException.class, + () -> order.auditAs("commit graph before child reconciliation").save(fixture.context)); + assertEquals(1, fixture.driver.queryForList( + "SELECT id FROM customer_order_data WHERE id = ?", new Object[] {194L}).size()); + assertEquals(1, fixture.driver.queryForList( + "SELECT id FROM order_item_data WHERE id = ?", new Object[] {294L}).size()); + assertEquals(1, fixture.committedRecorder.eventStore() + .outboxAfter("java-graph-source", 0, 10).size()); + + item.failPostCommitReconcile = false; + assertThrows(TeaQLRuntimeException.class, + () -> item.auditAs("unsafe child retry").save(fixture.context)); + assertThrows(TeaQLRuntimeException.class, + () -> order.auditAs("unsafe root retry").save(fixture.context)); + assertEquals(1, fixture.committedRecorder.eventStore() + .outboxAfter("java-graph-source", 0, 10).size()); + assertEquals("both committed graph members must still be audited", 2, fixture.audit.size()); + assertEquals(2, fixture.rawAudit.size()); + } + + @Test public void postCommitReconciliationAndAuditFailureRetainBothOutcomes() throws Exception { + var fixture = new Fixture(true); + var attempts = new java.util.concurrent.atomic.AtomicInteger(); + fixture.context.putAttribute(AppAuditEventSink.class.getName(), + (AppAuditEventSink) (caller, event) -> { + attempts.incrementAndGet(); + throw new IllegalStateException("diagnostic audit sink failure"); + }); + var order = fixture.create("CustomerOrder", 193, "committed despite two post-commit failures"); + order.failPostCommitReconcile = true; + + var failure = assertThrows(CommittedStateReconciliationException.class, + () -> order.auditAs("commit before reconciliation and audit failures").save(fixture.context)); + assertEquals(1, attempts.get()); + assertEquals(1, failure.getSuppressed().length); + assertTrue(failure.getSuppressed()[0] instanceof AuditDeliveryAfterCommitException); + assertEquals(1, fixture.rawAudit.size()); + assertEquals(Long.valueOf(1), fixture.rawAudit.get(0).resultingVersion()); + assertEquals(1, fixture.driver.queryForList( + "SELECT id FROM customer_order_data WHERE id = ?", new Object[] {193L}).size()); + assertEquals(1, fixture.committedRecorder.eventStore() + .outboxAfter("java-graph-source", 0, 10).size()); + order.failPostCommitReconcile = false; + assertThrows(TeaQLRuntimeException.class, + () -> order.auditAs("reject retry after two post-commit failures").save(fixture.context)); + assertEquals(1, fixture.committedRecorder.eventStore() + .outboxAfter("java-graph-source", 0, 10).size()); + } + @Test public void postCommitAuditFailureStillAttemptsEveryGraphMember() throws Exception { var fixture = new Fixture(true); var attempts = new java.util.concurrent.atomic.AtomicInteger();