[skip ci] Integrate approved post-commit reconciliation into audit fix

Author: Philip Z

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));

teaql-core/src/main/java/io/teaql/core/EntityMutationLedger.java

@@ -16,6 +16,7 @@ public class EntityMutationLedger {
     private final Set<EntityKey> newKeys = new TreeSet<>();
     private final Map<EntityKey, List<TraceNode>> traceChains = new TreeMap<>();
     private final Map<EntityKey, Long> 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<String, Object> 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();

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();

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");
+    }
+}

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<PendingMutati
             try {
                 failures += emitAuditEvent(
                         context, mutation.target(), mutation.auditKind(), mutation.changedValues(),
-                        mutation.governance(), mutation.intent(), mutation.traceChain(), mutation.redactions());
+                        mutation.governance(), mutation.intent(), mutation.traceChain(), mutation.redactions(),
+                        mutation.result().persistedEntity().getVersion());
             } catch (RuntimeException failure) {
                 // The provider transaction has already committed. Continue delivery
                 // for the other graph members and never expose a retryable save error.
@@ -1433,7 +1452,7 @@ private int emitAuditEvent(
             MutationAuditKind kind,
             Map<String, Object> changedValues,
             MutationGovernanceSnapshot governance, MutationIntent intent, List<TraceNode> traceChain,
-            SqlIntentRedactions redactions) {
+            SqlIntentRedactions redactions, Long committedVersion) {
         List<AuditFieldChange> changes = new ArrayList<>();
         if (changedValues != null) {
             for (Map.Entry<String, Object> 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);
 

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<String, Object> 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<ExecutionMetadata> sql = new CopyOnWriteArrayList<>();
         final List<SafeAuditEvent> audit = new CopyOnWriteArrayList<>();
+        final List<RawAuditEvent> rawAudit = new CopyOnWriteArrayList<>();
         final List<EntityPersistenceMutation> commands = new CopyOnWriteArrayList<>();
         final List<MutationResult> 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();