[skip ci] Integrate approved post-commit audit and reconciliation fixes

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/AuditDeliveryAfterCommitException.java

@@ -0,0 +1,19 @@
+package io.teaql.runtime;
+
+import io.teaql.core.TeaQLRuntimeException;
+
+/** The graph committed, but one or more post-commit audit deliveries failed. */
+public final class AuditDeliveryAfterCommitException extends TeaQLRuntimeException {
+    private final int failureCount;
+
+    public AuditDeliveryAfterCommitException(int failureCount) {
+        super("AUDIT_DELIVERY_AFTER_COMMIT: transaction committed but audit delivery failed; "
+                + "do not retry the mutation blindly");
+        if (failureCount < 1) throw new IllegalArgumentException("failureCount must be positive");
+        this.failureCount = failureCount;
+    }
+
+    public int failureCount() {
+        return failureCount;
+    }
+}

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,12 +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(context, completed);
-            entityMutationLedger.clearCurrentChangeSet();
+            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);
@@ -1102,18 +1121,34 @@ private List<PendingMutation> executeLedgerPlan(
         return completed;
     }
 
-    private void completeLedgerPlan(UserContext context, List<PendingMutation> completed) {
+    private void completeLedgerPlan(List<PendingMutation> completed) {
         for (PendingMutation mutation : completed) {
             applyPersistedEntity(mutation.descriptor(), mutation.target(), mutation.result());
-            emitAuditEvent(
-                    context, mutation.target(), mutation.auditKind(), mutation.changedValues(),
-                    mutation.governance(), mutation.intent(), mutation.traceChain(), mutation.redactions());
             mutation.target().clearUpdatedProperties();
-            // A committed delete supersedes any staged extension edits too.
-            mutation.target().__internalClearDynamicMutations();
         }
     }
 
+    private void deliverCommittedAuditEvents(UserContext context, List<PendingMutation> completed) {
+        int failures = 0;
+        for (PendingMutation mutation : completed) {
+            try {
+                failures += emitAuditEvent(
+                        context, mutation.target(), mutation.auditKind(), mutation.changedValues(),
+                        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.
+                failures++;
+            } finally {
+                // Preserve dynamic original values until the audit projection has
+                // observed them; then clear staged extension edits even on failure.
+                mutation.target().__internalClearDynamicMutations();
+            }
+        }
+        if (failures != 0) throw new AuditDeliveryAfterCommitException(failures);
+    }
+
     private Map<String, Object> snapshotChanges(Map<String, Object> changes) {
         return Collections.unmodifiableMap(new LinkedHashMap<>(changes));
     }
@@ -1411,13 +1446,13 @@ private List<MutationResult> mutateBatchWithTelemetry(
         }
     }
 
-    private void emitAuditEvent(
+    private int emitAuditEvent(
             UserContext context,
             Entity entity,
             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()) {
@@ -1447,7 +1482,7 @@ private void 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);
 
@@ -1457,18 +1492,29 @@ private void emitAuditEvent(
                         "teaql.mutation.kind", kind.name().toLowerCase(Locale.ROOT),
                         "teaql.audit.changed_field_count", changes.size())));
         try {
+            int failures = 0;
+            // The server-owned and application sinks are independent delivery
+            // attempts. One failure must not suppress the other after commit.
+            if (logSink != null) {
+                try {
+                    logSink.writeAuditEvent(context,
+                            LogPrivacy.audit(rawEvent, LogPrivacy.plaintextEnabled(), redactions));
+                } catch (RuntimeException failure) {
+                    failures++;
+                }
+            }
 
-        // The standard sink is server-owned by TeaQLRuntime and cannot be replaced by
-        // dynamic input or an application capability registered on UserContext.
-        if (logSink != null) {
-            logSink.writeAuditEvent(context, LogPrivacy.audit(rawEvent, LogPrivacy.plaintextEnabled(), redactions));
-        }
-
-        AppAuditEventSink appSink = context.capability(AppAuditEventSink.class);
-        if (appSink != null) {
-            appSink.onAuditEvent(context, buildSafeAuditEvent(rawEvent, redactions));
-        }
-        telemetryScope.success();
+            AppAuditEventSink appSink = context.capability(AppAuditEventSink.class);
+            if (appSink != null) {
+                try {
+                    appSink.onAuditEvent(context, buildSafeAuditEvent(rawEvent, redactions));
+                } catch (RuntimeException failure) {
+                    failures++;
+                }
+            }
+            if (failures == 0) telemetryScope.success();
+            else telemetryScope.failure(new AuditDeliveryAfterCommitException(failures));
+            return failures;
         } catch (RuntimeException | Error error) {
             telemetryScope.failure(error);
             throw error;