fix: distinguish post-commit audit delivery failures

Author: Philip Z

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

@@ -577,8 +577,9 @@ public void saveGraph(UserContext context, Entity entity) {
                 BaseEntity target = realEntities.get(key);
                 target.__internalHydrateDynamicFields(values, target.__internalLoadState().withDynamicSelection(values.selectedCodes()));
             });
-            completeLedgerPlan(context, completed);
+            completeLedgerPlan(completed);
             entityMutationLedger.clearCurrentChangeSet();
+            deliverCommittedAuditEvents(context, completed);
             telemetryScope.success();
         } catch (RuntimeException | Error error) {
             telemetryScope.failure(error);
@@ -1102,18 +1103,33 @@ 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());
+            } 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,7 +1427,7 @@ private List<MutationResult> mutateBatchWithTelemetry(
         }
     }
 
-    private void emitAuditEvent(
+    private int emitAuditEvent(
             UserContext context,
             Entity entity,
             MutationAuditKind kind,
@@ -1457,18 +1473,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;

teaql-sqlite/src/test/java/io/teaql/sqlite/GraphTraceSqliteTest.java

@@ -55,6 +55,7 @@ static final class Fixture {
         final List<MutationResult> results = new CopyOnWriteArrayList<>();
         boolean queryLogging = true;
         boolean mutationLogging = true;
+        boolean failStandardAudit;
         final List<Integer> itemInsertBatchSizes = new CopyOnWriteArrayList<>();
         final List<Integer> itemUpdateBatchSizes = new CopyOnWriteArrayList<>();
         final List<Integer> itemDeleteBatchSizes = new CopyOnWriteArrayList<>();
@@ -162,7 +163,15 @@ static final class Fixture {
             var ids = new AtomicLong(1000);
             var builder = TeaQLRuntime.builder().metadata(metadata).dataService("sqlite", provider)
                     .idGenerationService((caller, entity) -> ids.getAndIncrement())
-                    .logSink((caller, entry) -> sql.add(entry));
+                    .logSink(new RuntimeLogSink() {
+                        @Override public void writeExecutionLog(UserContext caller, ExecutionMetadata entry) {
+                            sql.add(entry);
+                        }
+
+                        @Override public void writeAuditEvent(UserContext caller, RawAuditEvent event) {
+                            if (failStandardAudit) throw new IllegalStateException("standard audit sink failure");
+                        }
+                    });
             if (committedRecorder != null) builder.committedMutationRecorder(committedRecorder);
             var runtime = builder.build();
             context = new DefaultUserContext(runtime) {
@@ -688,6 +697,67 @@ private static CheckedSave checkedSave(Fixture fixture, GraphEntity entity, Stri
         assertTrue("no post-commit audit on rolled back save", fixture.audit.isEmpty());
     }
 
+    @Test public void postCommitAuditFailureReportsCommittedOutcomeWithoutReplayingLedger() throws Exception {
+        var fixture = new Fixture(true);
+        fixture.context.putAttribute(AppAuditEventSink.class.getName(),
+                (AppAuditEventSink) (caller, event) -> {
+                    throw new IllegalStateException("diagnostic audit sink failure");
+                });
+        var order = fixture.create("CustomerOrder", 199, "committed before audit delivery");
+
+        var failure = assertThrows(AuditDeliveryAfterCommitException.class,
+                () -> order.auditAs("diagnose post-commit audit failure").save(fixture.context));
+        assertTrue(failure.getMessage().contains("AUDIT_DELIVERY_AFTER_COMMIT"));
+        assertEquals(1, failure.failureCount());
+        assertEquals(1, fixture.driver.queryForList(
+                "SELECT id FROM customer_order_data WHERE id = ?", new Object[] {199L}).size());
+        assertEquals(1, fixture.committedRecorder.eventStore()
+                .outboxAfter("java-graph-source", 0, 10).size());
+
+        fixture.context.putAttribute(AppAuditEventSink.class.getName(),
+                (AppAuditEventSink) (caller, event) -> fixture.audit.add(event));
+        order.auditAs("retry after audit delivery failure").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();
+        fixture.context.putAttribute(AppAuditEventSink.class.getName(),
+                (AppAuditEventSink) (caller, event) -> {
+                    attempts.incrementAndGet();
+                    throw new IllegalStateException("diagnostic audit sink failure");
+                });
+        var order = fixture.create("CustomerOrder", 198, "root audit fails");
+        var item = fixture.create("OrderItem", 298, "child audit fails");
+        order.__internalSet("children", List.of(item));
+
+        var failure = assertThrows(AuditDeliveryAfterCommitException.class,
+                () -> order.auditAs("commit graph before audit delivery").save(fixture.context));
+        assertEquals(2, failure.failureCount());
+        assertEquals(2, attempts.get());
+        assertEquals(1, fixture.committedRecorder.eventStore()
+                .outboxAfter("java-graph-source", 0, 10).size());
+        assertEquals(1, fixture.driver.queryForList(
+                "SELECT id FROM customer_order_data WHERE id = ?", new Object[] {198L}).size());
+        assertEquals(1, fixture.driver.queryForList(
+                "SELECT id FROM order_item_data WHERE id = ?", new Object[] {298L}).size());
+    }
+
+    @Test public void standardAuditSinkFailureDoesNotSuppressApplicationAuditDelivery() throws Exception {
+        var fixture = new Fixture(true);
+        fixture.failStandardAudit = true;
+        var order = fixture.create("CustomerOrder", 197, "both audit sinks are attempted");
+
+        var failure = assertThrows(AuditDeliveryAfterCommitException.class,
+                () -> order.auditAs("commit before standard audit delivery").save(fixture.context));
+        assertEquals(1, failure.failureCount());
+        assertEquals(1, fixture.audit.size());
+        assertEquals(1, fixture.committedRecorder.eventStore()
+                .outboxAfter("java-graph-source", 0, 10).size());
+    }
+
     @Test public void realSqliteProviderFailureKeepsAttemptedLineageWithoutCommittedAudit() throws Exception {
         var fixture = new Fixture();
         fixture.driver.execute("CREATE UNIQUE INDEX payment_name_unique ON payment_data(name)");