From 4577185a5ddeb4f52e2f2d9aa25852b4dd4e26b7 Mon Sep 17 00:00:00 2001 From: Philip Z Date: Tue, 20 Oct 2026 01:43:36 +0800 Subject: [PATCH 1/4] fix: distinguish unknown JDBC transaction outcomes --- .../io/teaql/core/EntityMutationLedger.java | 12 +- .../TransactionOutcomeUnknownException.java | 9 ++ .../teaql/provider/jdbc/JdbcSqlExecutor.java | 78 ++++++--- .../java/io/teaql/runtime/TeaQLRuntime.java | 5 + .../io/teaql/sqlite/GraphTraceSqliteTest.java | 149 +++++++++++++++++- 5 files changed, 227 insertions(+), 26 deletions(-) create mode 100644 teaql-core/src/main/java/io/teaql/core/TransactionOutcomeUnknownException.java 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 18bb83af..84b49f8f 100644 --- a/teaql-core/src/main/java/io/teaql/core/EntityMutationLedger.java +++ b/teaql-core/src/main/java/io/teaql/core/EntityMutationLedger.java @@ -17,6 +17,7 @@ public class EntityMutationLedger { private final Map> traceChains = new TreeMap<>(); private final Map originalVersions = new TreeMap<>(); private boolean committedStateRequiresReload; + private boolean transactionOutcomeUnknownRequiresReload; // --- Change Set Stack --- @@ -47,8 +48,17 @@ public void markCommittedStateRequiresReload() { committedStateRequiresReload = true; } - /** Prevents a stale graph from replaying a write whose outcome is already committed. */ + /** A provider could not confirm commit or rollback; this graph must not be replayed. */ + public void markTransactionOutcomeUnknownRequiresReload() { + transactionOutcomeUnknownRequiresReload = true; + } + + /** Prevents replay of a graph whose persistence outcome needs a fresh read. */ public void requireSaveable() { + if (transactionOutcomeUnknownRequiresReload) { + throw new TeaQLRuntimeException( + "TRANSACTION_OUTCOME_UNKNOWN_RELOAD_REQUIRED: reload the entity graph before another save"); + } if (committedStateRequiresReload) { throw new TeaQLRuntimeException( "COMMITTED_STATE_RELOAD_REQUIRED: reload the entity graph before another save"); diff --git a/teaql-core/src/main/java/io/teaql/core/TransactionOutcomeUnknownException.java b/teaql-core/src/main/java/io/teaql/core/TransactionOutcomeUnknownException.java new file mode 100644 index 00000000..1a3ed9b4 --- /dev/null +++ b/teaql-core/src/main/java/io/teaql/core/TransactionOutcomeUnknownException.java @@ -0,0 +1,9 @@ +package io.teaql.core; + +/** A provider could not confirm whether a transaction committed; reload before retrying. */ +public final class TransactionOutcomeUnknownException extends TeaQLRuntimeException { + public TransactionOutcomeUnknownException() { + super("TRANSACTION_OUTCOME_UNKNOWN: the provider could not confirm the transaction result; " + + "reload the affected graph and committed events before any retry"); + } +} diff --git a/teaql-provider-jdbc/src/main/java/io/teaql/provider/jdbc/JdbcSqlExecutor.java b/teaql-provider-jdbc/src/main/java/io/teaql/provider/jdbc/JdbcSqlExecutor.java index 59016536..7362696a 100644 --- a/teaql-provider-jdbc/src/main/java/io/teaql/provider/jdbc/JdbcSqlExecutor.java +++ b/teaql-provider-jdbc/src/main/java/io/teaql/provider/jdbc/JdbcSqlExecutor.java @@ -1,5 +1,6 @@ package io.teaql.provider.jdbc; +import io.teaql.core.TransactionOutcomeUnknownException; import io.teaql.dataservice.sql.SqlExecutionAdapter; import io.teaql.dataservice.sql.SqlRowMapper; @@ -437,37 +438,67 @@ public void executeInTransaction(Runnable action) { throw new RuntimeException("JDBC transaction failed", failure); } boolean committed = false; + boolean outcomeUnknown = false; + boolean rollbackUnconfirmed = false; try { boolean autoCommit = connection.getAutoCommit(); connection.setAutoCommit(false); transactionConnection.set(connection); transactionRollbackOnly.set(false); try { - action.run(); - if (Boolean.TRUE.equals(transactionRollbackOnly.get())) { - throw new IllegalStateException( - "JDBC transaction is rollback-only after a failed operation"); + try { + action.run(); + if (Boolean.TRUE.equals(transactionRollbackOnly.get())) { + throw new IllegalStateException( + "JDBC transaction is rollback-only after a failed operation"); + } + } catch (RuntimeException | Error failure) { + try { + connection.rollback(); + } catch (SQLException | RuntimeException rollbackFailure) { + outcomeUnknown = true; + rollbackUnconfirmed = true; + warnTransactionCleanup("rollback after failed action", false, true); + if (failure instanceof Error) { + // Preserve a JVM-level failure, but do not implicitly + // commit when restoring autoCommit below. + failure.addSuppressed(new IllegalStateException("JDBC rollback failed")); + throw failure; + } + throw new TransactionOutcomeUnknownException(); + } + throw failure; } - connection.commit(); - committed = true; - } catch (RuntimeException | Error failure) { try { - connection.rollback(); - } catch (SQLException | RuntimeException rollbackFailure) { - // Keep the original failure. The JDBC message may contain - // sensitive connection details and is not safe to expose. - failure.addSuppressed(new IllegalStateException("JDBC rollback failed")); + connection.commit(); + committed = true; + } catch (SQLException | RuntimeException commitFailure) { + // The database may have committed before the driver lost the + // acknowledgement. Conversely, restoring autoCommit without a + // rollback can commit a transaction the driver never attempted. + outcomeUnknown = true; + warnTransactionCleanup("commit confirmation", false, true); + try { + connection.rollback(); + } catch (SQLException | RuntimeException rollbackFailure) { + rollbackUnconfirmed = true; + warnTransactionCleanup("rollback after unconfirmed commit", false, true); + } + throw new TransactionOutcomeUnknownException(); } - throw failure; } finally { transactionConnection.remove(); transactionRollbackOnly.remove(); - try { - connection.setAutoCommit(autoCommit); - } catch (SQLException | RuntimeException cleanupFailure) { - // A successful commit is already durable. A cleanup error - // must not make the caller replay that business mutation. - warnTransactionCleanup("autoCommit restoration", committed); + // A confirmed rollback leaves no pending write to commit. If + // rollback itself failed, changing autoCommit could commit it. + if (!rollbackUnconfirmed) { + try { + connection.setAutoCommit(autoCommit); + } catch (SQLException | RuntimeException cleanupFailure) { + // A successful commit is already durable. A cleanup error + // must not make the caller replay that business mutation. + warnTransactionCleanup("autoCommit restoration", committed, outcomeUnknown); + } } } } catch (SQLException e) { @@ -476,16 +507,15 @@ public void executeInTransaction(Runnable action) { try { connection.close(); } catch (SQLException | RuntimeException cleanupFailure) { - warnTransactionCleanup("connection close", committed); + warnTransactionCleanup("connection close", committed, outcomeUnknown); } } } - private static void warnTransactionCleanup(String operation, boolean committed) { + private static void warnTransactionCleanup(String operation, boolean committed, boolean outcomeUnknown) { try { - LOG.warning(committed - ? "JDBC transaction committed; " + operation + " failed" - : "JDBC transaction unsuccessful; " + operation + " failed"); + String state = committed ? "committed" : outcomeUnknown ? "outcome unknown" : "unsuccessful"; + LOG.warning("JDBC transaction " + state + "; " + operation + " failed"); } catch (RuntimeException ignored) { // A user-installed logging handler must not turn a committed write // back into an apparently retryable transaction failure. 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 39d15d19..ba385451 100644 --- a/teaql-runtime/src/main/java/io/teaql/runtime/TeaQLRuntime.java +++ b/teaql-runtime/src/main/java/io/teaql/runtime/TeaQLRuntime.java @@ -566,6 +566,11 @@ public void saveGraph(UserContext context, Entity entity) { completed = executeLedgerPlan( context, entityMutationLedger, mutationExecutor, realEntities, governance, intent, traceScopes, graphScope); } + } catch (TransactionOutcomeUnknownException failure) { + // The JDBC commit acknowledgement can fail after durable commit. + // Restoring the old version would make this graph appear retryable. + entityMutationLedger.markTransactionOutcomeUnknownRequiresReload(); + throw failure; } catch (RuntimeException | Error failure) { restoreGraphPersistenceState( entity, 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 e8b366eb..eef3cc9d 100644 --- a/teaql-sqlite/src/test/java/io/teaql/sqlite/GraphTraceSqliteTest.java +++ b/teaql-sqlite/src/test/java/io/teaql/sqlite/GraphTraceSqliteTest.java @@ -10,13 +10,24 @@ import com.fasterxml.jackson.databind.ObjectMapper; import io.teaql.provider.jdbc.JdbcSqlExecutor; import io.teaql.runtime.*; +import java.lang.reflect.InvocationTargetException; +import java.lang.reflect.Method; +import java.lang.reflect.Proxy; import java.nio.file.Files; +import java.sql.Connection; +import java.sql.SQLException; import java.util.*; import java.util.concurrent.CopyOnWriteArrayList; import java.util.concurrent.CountDownLatch; import java.util.concurrent.Executors; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicLong; +import java.util.function.Function; +import java.util.logging.Handler; +import java.util.logging.LogRecord; +import java.util.logging.Logger; +import javax.sql.DataSource; import io.teaql.core.checker.Checker; import io.teaql.core.checker.CheckException; import io.teaql.core.checker.FixEvidence; @@ -75,10 +86,15 @@ static final class Fixture { Fixture() throws Exception { this(false); } Fixture(boolean captureCommittedMutations) throws Exception { + this(captureCommittedMutations, source -> source); + } + + Fixture(boolean captureCommittedMutations, + Function transactionSource) throws Exception { var ds = new SQLiteDataSource(); dataSource = ds; ds.setUrl("jdbc:sqlite:" + Files.createTempFile("teaql-graph-trace-", ".db")); - driver = new JdbcSqlExecutor(ds) { + driver = new JdbcSqlExecutor(transactionSource.apply(ds)) { @Override public int[] batchUpdate(String text, List rows) { if (text.startsWith("INSERT INTO order_item_data")) { itemInsertBatchSizes.add(rows.size()); @@ -703,6 +719,137 @@ private static CheckedSave checkedSave(Fixture fixture, GraphEntity entity, Stri assertTrue("no post-commit audit on rolled back save", fixture.audit.isEmpty()); } + @Test public void lostCommitAcknowledgementAfterDurableWriteRequiresReload() throws Exception { + assertCommitAcknowledgementUnknown(true); + } + + @Test public void failedCommitBeforeWriteRequiresTheSameConservativeReload() throws Exception { + assertCommitAcknowledgementUnknown(false); + } + + @Test public void failedRollbackMustNotImplicitlyCommitOnAutoCommitRestoration() throws Exception { + var armed = new AtomicBoolean(); + var injected = new AtomicBoolean(); + var restoredAutoCommit = new AtomicBoolean(); + var fixture = new Fixture(true, + source -> rollbackFault(source, armed, injected, restoredAutoCommit)); + fixture.driver.execute("DROP TABLE teaql_event_outbox"); + armed.set(true); + var order = fixture.create("CustomerOrder", 187, "rollback failed after Outbox error"); + + var failure = assertThrows(RuntimeException.class, + () -> order.auditAs("probe uncertain rollback outcome").save(fixture.context)); + assertTrue("the rollback failure must be injected", injected.get()); + assertFalse("restoring autoCommit after failed rollback can commit pending writes", + restoredAutoCommit.get()); + assertEquals(0, fixture.driver.queryForList( + "SELECT id FROM customer_order_data WHERE id = ?", new Object[] {187L}).size()); + assertTrue(failure.getMessage().contains("TRANSACTION_OUTCOME_UNKNOWN")); + var retryFailure = assertThrows(TeaQLRuntimeException.class, + () -> order.auditAs("reject replay after failed rollback").save(fixture.context)); + assertTrue(retryFailure.getMessage().contains("TRANSACTION_OUTCOME_UNKNOWN_RELOAD_REQUIRED")); + } + + private static void assertCommitAcknowledgementUnknown(boolean afterActualCommit) throws Exception { + var armed = new AtomicBoolean(); + var injected = new AtomicBoolean(); + var fixture = new Fixture(true, + source -> commitAcknowledgementFault(source, armed, injected, afterActualCommit)); + armed.set(true); + var order = fixture.create("CustomerOrder", afterActualCommit ? 189 : 188, + "commit acknowledgement may be lost"); + var item = fixture.create("OrderItem", afterActualCommit ? 289 : 288, + "same uncertain transaction"); + order.__internalSet("children", List.of(item)); + + var warnings = new ArrayList(); + var logger = Logger.getLogger(JdbcSqlExecutor.class.getName()); + Handler capture = new Handler() { + @Override public void publish(LogRecord record) { warnings.add(record.getMessage()); } + @Override public void flush() { } + @Override public void close() { } + }; + logger.addHandler(capture); + RuntimeException failure; + try { + failure = assertThrows(RuntimeException.class, + () -> order.auditAs("probe uncertain commit outcome").save(fixture.context)); + } finally { + logger.removeHandler(capture); + } + assertTrue("fault must reach the actual commit boundary", injected.get()); + assertEquals(afterActualCommit ? 1 : 0, fixture.driver.queryForList( + "SELECT id FROM customer_order_data WHERE id = ?", new Object[] {order.getId()}).size()); + assertEquals(afterActualCommit ? 1 : 0, fixture.driver.queryForList( + "SELECT id FROM order_item_data WHERE id = ?", new Object[] {item.getId()}).size()); + assertEquals(afterActualCommit ? 1 : 0, fixture.committedRecorder.eventStore() + .outboxAfter("java-graph-source", 0, 10).size()); + assertTrue("the outcome must not be reported as a known rollback", + failure.getMessage().contains("TRANSACTION_OUTCOME_UNKNOWN")); + assertFalse("the public message must not contain the JDBC diagnostic", + failure.getMessage().contains("COMMIT-ACK-SECRET-CANARY")); + assertTrue(warnings.stream().anyMatch(message -> message.contains("outcome unknown"))); + assertTrue("the operational warning must not contain the JDBC diagnostic", + warnings.stream().noneMatch(message -> message.contains("COMMIT-ACK-SECRET-CANARY"))); + var retryFailure = assertThrows(TeaQLRuntimeException.class, + () -> order.auditAs("reject replay before reload").save(fixture.context)); + assertTrue(retryFailure.getMessage().contains("TRANSACTION_OUTCOME_UNKNOWN_RELOAD_REQUIRED")); + var childRetryFailure = assertThrows(TeaQLRuntimeException.class, + () -> item.auditAs("reject child replay before reload").save(fixture.context)); + assertTrue(childRetryFailure.getMessage().contains("TRANSACTION_OUTCOME_UNKNOWN_RELOAD_REQUIRED")); + assertThrows(TeaQLRuntimeException.class, + () -> order.setEntityMutationLedger(new EntityMutationLedger())); + } + + private static DataSource commitAcknowledgementFault(SQLiteDataSource source, + AtomicBoolean armed, AtomicBoolean injected, boolean afterActualCommit) { + return (DataSource) Proxy.newProxyInstance(DataSource.class.getClassLoader(), + new Class[] {DataSource.class}, (proxy, method, args) -> { + Object result = invoke(source, method, args); + if (!method.getName().equals("getConnection")) return result; + Connection actual = (Connection) result; + return Proxy.newProxyInstance(Connection.class.getClassLoader(), + new Class[] {Connection.class}, (connectionProxy, operation, parameters) -> { + if (operation.getName().equals("commit") && armed.get() + && injected.compareAndSet(false, true)) { + if (afterActualCommit) invoke(actual, operation, parameters); + throw new SQLException("COMMIT-ACK-SECRET-CANARY"); + } + return invoke(actual, operation, parameters); + }); + }); + } + + private static DataSource rollbackFault(SQLiteDataSource source, + AtomicBoolean armed, AtomicBoolean injected, AtomicBoolean restoredAutoCommit) { + return (DataSource) Proxy.newProxyInstance(DataSource.class.getClassLoader(), + new Class[] {DataSource.class}, (proxy, method, args) -> { + Object result = invoke(source, method, args); + if (!method.getName().equals("getConnection")) return result; + Connection actual = (Connection) result; + return Proxy.newProxyInstance(Connection.class.getClassLoader(), + new Class[] {Connection.class}, (connectionProxy, operation, parameters) -> { + if (operation.getName().equals("rollback") && armed.get() + && injected.compareAndSet(false, true)) { + throw new SQLException("ROLLBACK-SECRET-CANARY"); + } + if (operation.getName().equals("setAutoCommit") && injected.get() + && Boolean.TRUE.equals(parameters[0])) { + restoredAutoCommit.set(true); + } + return invoke(actual, operation, parameters); + }); + }); + } + + private static Object invoke(Object target, Method method, Object[] args) throws Throwable { + try { + return method.invoke(target, args); + } catch (InvocationTargetException failure) { + throw failure.getCause(); + } + } + @Test public void postCommitAuditFailureReportsCommittedOutcomeWithoutReplayingLedger() throws Exception { var fixture = new Fixture(true); fixture.context.putAttribute(AppAuditEventSink.class.getName(), From dee5bbe6b857d264f86206ba91917b81dfa5bd4e Mon Sep 17 00:00:00 2001 From: Philip Z Date: Tue, 20 Oct 2026 02:54:56 +0800 Subject: [PATCH 2/4] fix: fence graph after unconfirmed rollback of JVM Error --- .../teaql/provider/jdbc/JdbcSqlExecutor.java | 5 +-- .../java/io/teaql/runtime/TeaQLRuntime.java | 17 +++++++--- .../io/teaql/sqlite/GraphTraceSqliteTest.java | 31 +++++++++++++++++++ 3 files changed, 46 insertions(+), 7 deletions(-) diff --git a/teaql-provider-jdbc/src/main/java/io/teaql/provider/jdbc/JdbcSqlExecutor.java b/teaql-provider-jdbc/src/main/java/io/teaql/provider/jdbc/JdbcSqlExecutor.java index 7362696a..e3fc676f 100644 --- a/teaql-provider-jdbc/src/main/java/io/teaql/provider/jdbc/JdbcSqlExecutor.java +++ b/teaql-provider-jdbc/src/main/java/io/teaql/provider/jdbc/JdbcSqlExecutor.java @@ -461,8 +461,9 @@ public void executeInTransaction(Runnable action) { warnTransactionCleanup("rollback after failed action", false, true); if (failure instanceof Error) { // Preserve a JVM-level failure, but do not implicitly - // commit when restoring autoCommit below. - failure.addSuppressed(new IllegalStateException("JDBC rollback failed")); + // commit when restoring autoCommit below. The sanitized + // marker also prevents the caller's graph from replaying. + failure.addSuppressed(new TransactionOutcomeUnknownException()); throw failure; } throw new TransactionOutcomeUnknownException(); 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 ba385451..184f6d01 100644 --- a/teaql-runtime/src/main/java/io/teaql/runtime/TeaQLRuntime.java +++ b/teaql-runtime/src/main/java/io/teaql/runtime/TeaQLRuntime.java @@ -572,11 +572,18 @@ public void saveGraph(UserContext context, Entity entity) { entityMutationLedger.markTransactionOutcomeUnknownRequiresReload(); throw failure; } catch (RuntimeException | Error failure) { - restoreGraphPersistenceState( - entity, - entityMutationLedger, - persistenceStates, - Collections.newSetFromMap(new IdentityHashMap<>())); + if (Arrays.stream(failure.getSuppressed()) + .anyMatch(TransactionOutcomeUnknownException.class::isInstance)) { + // Preserve the original JVM Error while preventing replay after + // an unconfirmed rollback of its partially written graph. + entityMutationLedger.markTransactionOutcomeUnknownRequiresReload(); + } else { + restoreGraphPersistenceState( + entity, + entityMutationLedger, + persistenceStates, + Collections.newSetFromMap(new IdentityHashMap<>())); + } throw failure; } CommittedStateReconciliationException reconciliationOutcome = null; 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 eef3cc9d..bcf79025 100644 --- a/teaql-sqlite/src/test/java/io/teaql/sqlite/GraphTraceSqliteTest.java +++ b/teaql-sqlite/src/test/java/io/teaql/sqlite/GraphTraceSqliteTest.java @@ -81,6 +81,7 @@ static final class Fixture { final DefaultUserContext context; final SqliteCommittedMutationRecorder committedRecorder; volatile boolean failReadback; + volatile boolean errorReadback; volatile boolean unknownUpdateCounts; Fixture() throws Exception { this(false); } @@ -114,6 +115,9 @@ static final class Fixture { return counts; } @Override public List> queryForList(String text, Object[] args) { + if (errorReadback && text.startsWith("SELECT * FROM") && text.contains("customer_order_data")) { + throw new AssertionError("synthetic Error after business write"); + } if (failReadback && text.startsWith("SELECT * FROM") && text.contains("customer_order_data")) { // A real driver failure after a successful write, inside the same transaction. execute("DROP TABLE customer_order_data"); @@ -750,6 +754,33 @@ private static CheckedSave checkedSave(Fixture fixture, GraphEntity entity, Stri assertTrue(retryFailure.getMessage().contains("TRANSACTION_OUTCOME_UNKNOWN_RELOAD_REQUIRED")); } + @Test public void failedRollbackAfterErrorRequiresReloadBeforeRetry() throws Exception { + var armed = new AtomicBoolean(); + var injected = new AtomicBoolean(); + var restoredAutoCommit = new AtomicBoolean(); + var fixture = new Fixture(false, + source -> rollbackFault(source, armed, injected, restoredAutoCommit)); + var order = fixture.create("CustomerOrder", 190, "error before rollback"); + fixture.errorReadback = true; + armed.set(true); + + var failure = assertThrows(AssertionError.class, + () -> order.auditAs("probe rollback failure after Error").save(fixture.context)); + assertTrue("the original Error must carry a sanitized unknown-outcome marker", + Arrays.stream(failure.getSuppressed()) + .anyMatch(TransactionOutcomeUnknownException.class::isInstance)); + assertTrue("rollback fault must be injected", injected.get()); + assertFalse("failed rollback must not restore autoCommit", restoredAutoCommit.get()); + fixture.errorReadback = false; + var retryFailure = assertThrows(TeaQLRuntimeException.class, + () -> order.auditAs("reject retry after unknown Error outcome").save(fixture.context)); + assertTrue(retryFailure.getMessage().contains("TRANSACTION_OUTCOME_UNKNOWN_RELOAD_REQUIRED")); + var independent = fixture.create("Payment", 191, "independent graph after Error"); + independent.auditAs("save independent graph after unknown Error outcome") + .save(fixture.context); + assertEquals(Long.valueOf(1), independent.getVersion()); + } + private static void assertCommitAcknowledgementUnknown(boolean afterActualCommit) throws Exception { var armed = new AtomicBoolean(); var injected = new AtomicBoolean(); From dd32106ddc0ee8a23ed5f60641fb9df6cfa0f7b0 Mon Sep 17 00:00:00 2001 From: Philip Z Date: Tue, 20 Oct 2026 03:12:38 +0800 Subject: [PATCH 3/4] fix: fence unknown rollback without suppressed Error metadata --- .../TransactionOutcomeUnknownException.java | 18 ++++++++-- .../teaql/provider/jdbc/JdbcSqlExecutor.java | 9 +++-- .../java/io/teaql/runtime/TeaQLRuntime.java | 18 ++++------ .../io/teaql/sqlite/GraphTraceSqliteTest.java | 35 +++++++++++++++++-- 4 files changed, 58 insertions(+), 22 deletions(-) diff --git a/teaql-core/src/main/java/io/teaql/core/TransactionOutcomeUnknownException.java b/teaql-core/src/main/java/io/teaql/core/TransactionOutcomeUnknownException.java index 1a3ed9b4..f6ecceac 100644 --- a/teaql-core/src/main/java/io/teaql/core/TransactionOutcomeUnknownException.java +++ b/teaql-core/src/main/java/io/teaql/core/TransactionOutcomeUnknownException.java @@ -2,8 +2,22 @@ /** A provider could not confirm whether a transaction committed; reload before retrying. */ public final class TransactionOutcomeUnknownException extends TeaQLRuntimeException { + private static final String MESSAGE = "TRANSACTION_OUTCOME_UNKNOWN: the provider could not confirm the transaction result; " + + "reload the affected graph and committed events before any retry"; + private final Error originalError; + public TransactionOutcomeUnknownException() { - super("TRANSACTION_OUTCOME_UNKNOWN: the provider could not confirm the transaction result; " - + "reload the affected graph and committed events before any retry"); + this(null); + } + + /** Retain a business Error without exposing it as a diagnostic cause. */ + public TransactionOutcomeUnknownException(Error originalError) { + super(MESSAGE); + this.originalError = originalError; + } + + /** The graph runtime rethrows this Error only after fencing the mutation ledger. */ + public Error originalError() { + return originalError; } } diff --git a/teaql-provider-jdbc/src/main/java/io/teaql/provider/jdbc/JdbcSqlExecutor.java b/teaql-provider-jdbc/src/main/java/io/teaql/provider/jdbc/JdbcSqlExecutor.java index e3fc676f..c7d47231 100644 --- a/teaql-provider-jdbc/src/main/java/io/teaql/provider/jdbc/JdbcSqlExecutor.java +++ b/teaql-provider-jdbc/src/main/java/io/teaql/provider/jdbc/JdbcSqlExecutor.java @@ -460,11 +460,10 @@ public void executeInTransaction(Runnable action) { rollbackUnconfirmed = true; warnTransactionCleanup("rollback after failed action", false, true); if (failure instanceof Error) { - // Preserve a JVM-level failure, but do not implicitly - // commit when restoring autoCommit below. The sanitized - // marker also prevents the caller's graph from replaying. - failure.addSuppressed(new TransactionOutcomeUnknownException()); - throw failure; + // The graph runtime must fence the ledger before + // rethrowing the original Error. A suppressed marker + // is unreliable because Error may disable suppression. + throw new TransactionOutcomeUnknownException((Error) failure); } throw new TransactionOutcomeUnknownException(); } 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 184f6d01..02d6504d 100644 --- a/teaql-runtime/src/main/java/io/teaql/runtime/TeaQLRuntime.java +++ b/teaql-runtime/src/main/java/io/teaql/runtime/TeaQLRuntime.java @@ -570,20 +570,14 @@ public void saveGraph(UserContext context, Entity entity) { // The JDBC commit acknowledgement can fail after durable commit. // Restoring the old version would make this graph appear retryable. entityMutationLedger.markTransactionOutcomeUnknownRequiresReload(); + if (failure.originalError() != null) throw failure.originalError(); throw failure; } catch (RuntimeException | Error failure) { - if (Arrays.stream(failure.getSuppressed()) - .anyMatch(TransactionOutcomeUnknownException.class::isInstance)) { - // Preserve the original JVM Error while preventing replay after - // an unconfirmed rollback of its partially written graph. - entityMutationLedger.markTransactionOutcomeUnknownRequiresReload(); - } else { - restoreGraphPersistenceState( - entity, - entityMutationLedger, - persistenceStates, - Collections.newSetFromMap(new IdentityHashMap<>())); - } + restoreGraphPersistenceState( + entity, + entityMutationLedger, + persistenceStates, + Collections.newSetFromMap(new IdentityHashMap<>())); throw failure; } CommittedStateReconciliationException reconciliationOutcome = null; 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 bcf79025..91302db4 100644 --- a/teaql-sqlite/src/test/java/io/teaql/sqlite/GraphTraceSqliteTest.java +++ b/teaql-sqlite/src/test/java/io/teaql/sqlite/GraphTraceSqliteTest.java @@ -82,6 +82,7 @@ static final class Fixture { final SqliteCommittedMutationRecorder committedRecorder; volatile boolean failReadback; volatile boolean errorReadback; + volatile boolean errorReadbackWithoutSuppression; volatile boolean unknownUpdateCounts; Fixture() throws Exception { this(false); } @@ -115,6 +116,10 @@ static final class Fixture { return counts; } @Override public List> queryForList(String text, Object[] args) { + if (errorReadbackWithoutSuppression && text.startsWith("SELECT * FROM") + && text.contains("customer_order_data")) { + throw new NoSuppressionError(); + } if (errorReadback && text.startsWith("SELECT * FROM") && text.contains("customer_order_data")) { throw new AssertionError("synthetic Error after business write"); } @@ -223,6 +228,12 @@ GraphEntity create(String type, long id, String name) { } } + private static final class NoSuppressionError extends Error { + private NoSuppressionError() { + super("synthetic Error without suppressed diagnostics", null, false, false); + } + } + private static void await(CountDownLatch latch) { try { assertTrue("overlapping execution must reach its checkpoint", latch.await(10, TimeUnit.SECONDS)); @@ -766,9 +777,7 @@ private static CheckedSave checkedSave(Fixture fixture, GraphEntity entity, Stri var failure = assertThrows(AssertionError.class, () -> order.auditAs("probe rollback failure after Error").save(fixture.context)); - assertTrue("the original Error must carry a sanitized unknown-outcome marker", - Arrays.stream(failure.getSuppressed()) - .anyMatch(TransactionOutcomeUnknownException.class::isInstance)); + assertEquals("synthetic Error after business write", failure.getMessage()); assertTrue("rollback fault must be injected", injected.get()); assertFalse("failed rollback must not restore autoCommit", restoredAutoCommit.get()); fixture.errorReadback = false; @@ -781,6 +790,26 @@ private static CheckedSave checkedSave(Fixture fixture, GraphEntity entity, Stri assertEquals(Long.valueOf(1), independent.getVersion()); } + @Test public void failedRollbackAfterUnsuppressibleErrorRequiresReloadBeforeRetry() throws Exception { + var armed = new AtomicBoolean(); + var injected = new AtomicBoolean(); + var restoredAutoCommit = new AtomicBoolean(); + var fixture = new Fixture(false, + source -> rollbackFault(source, armed, injected, restoredAutoCommit)); + var order = fixture.create("CustomerOrder", 192, "unsuppressible error before rollback"); + fixture.errorReadbackWithoutSuppression = true; + armed.set(true); + + assertThrows(NoSuppressionError.class, + () -> order.auditAs("probe unsuppressible Error rollback failure").save(fixture.context)); + assertTrue("rollback fault must be injected", injected.get()); + assertFalse("failed rollback must not restore autoCommit", restoredAutoCommit.get()); + fixture.errorReadbackWithoutSuppression = false; + var retryFailure = assertThrows(TeaQLRuntimeException.class, + () -> order.auditAs("reject retry after unsuppressible Error").save(fixture.context)); + assertTrue(retryFailure.getMessage().contains("TRANSACTION_OUTCOME_UNKNOWN_RELOAD_REQUIRED")); + } + private static void assertCommitAcknowledgementUnknown(boolean afterActualCommit) throws Exception { var armed = new AtomicBoolean(); var injected = new AtomicBoolean(); From edbae2bf1286876ecbf84f6cb2d2db1105df5de6 Mon Sep 17 00:00:00 2001 From: Philip Z Date: Tue, 20 Oct 2026 03:29:51 +0800 Subject: [PATCH 4/4] fix: fence JDBC Error outcomes and preserve committed cleanup --- .../teaql/provider/jdbc/JdbcSqlExecutor.java | 22 ++-- .../io/teaql/sqlite/GraphTraceSqliteTest.java | 113 ++++++++++++++++++ .../JdbcCommittedCleanupSqliteTest.java | 37 +++++- 3 files changed, 159 insertions(+), 13 deletions(-) diff --git a/teaql-provider-jdbc/src/main/java/io/teaql/provider/jdbc/JdbcSqlExecutor.java b/teaql-provider-jdbc/src/main/java/io/teaql/provider/jdbc/JdbcSqlExecutor.java index c7d47231..e945a2c7 100644 --- a/teaql-provider-jdbc/src/main/java/io/teaql/provider/jdbc/JdbcSqlExecutor.java +++ b/teaql-provider-jdbc/src/main/java/io/teaql/provider/jdbc/JdbcSqlExecutor.java @@ -455,7 +455,7 @@ public void executeInTransaction(Runnable action) { } catch (RuntimeException | Error failure) { try { connection.rollback(); - } catch (SQLException | RuntimeException rollbackFailure) { + } catch (SQLException | RuntimeException | Error rollbackFailure) { outcomeUnknown = true; rollbackUnconfirmed = true; warnTransactionCleanup("rollback after failed action", false, true); @@ -465,6 +465,9 @@ public void executeInTransaction(Runnable action) { // is unreliable because Error may disable suppression. throw new TransactionOutcomeUnknownException((Error) failure); } + if (rollbackFailure instanceof Error) { + throw new TransactionOutcomeUnknownException((Error) rollbackFailure); + } throw new TransactionOutcomeUnknownException(); } throw failure; @@ -472,7 +475,7 @@ public void executeInTransaction(Runnable action) { try { connection.commit(); committed = true; - } catch (SQLException | RuntimeException commitFailure) { + } catch (SQLException | RuntimeException | Error commitFailure) { // The database may have committed before the driver lost the // acknowledgement. Conversely, restoring autoCommit without a // rollback can commit a transaction the driver never attempted. @@ -480,11 +483,13 @@ public void executeInTransaction(Runnable action) { warnTransactionCleanup("commit confirmation", false, true); try { connection.rollback(); - } catch (SQLException | RuntimeException rollbackFailure) { + } catch (SQLException | RuntimeException | Error rollbackFailure) { rollbackUnconfirmed = true; warnTransactionCleanup("rollback after unconfirmed commit", false, true); } - throw new TransactionOutcomeUnknownException(); + throw commitFailure instanceof Error + ? new TransactionOutcomeUnknownException((Error) commitFailure) + : new TransactionOutcomeUnknownException(); } } finally { transactionConnection.remove(); @@ -494,7 +499,7 @@ public void executeInTransaction(Runnable action) { if (!rollbackUnconfirmed) { try { connection.setAutoCommit(autoCommit); - } catch (SQLException | RuntimeException cleanupFailure) { + } catch (SQLException | RuntimeException | Error cleanupFailure) { // A successful commit is already durable. A cleanup error // must not make the caller replay that business mutation. warnTransactionCleanup("autoCommit restoration", committed, outcomeUnknown); @@ -506,7 +511,7 @@ public void executeInTransaction(Runnable action) { } finally { try { connection.close(); - } catch (SQLException | RuntimeException cleanupFailure) { + } catch (SQLException | RuntimeException | Error cleanupFailure) { warnTransactionCleanup("connection close", committed, outcomeUnknown); } } @@ -516,9 +521,10 @@ private static void warnTransactionCleanup(String operation, boolean committed, try { String state = committed ? "committed" : outcomeUnknown ? "outcome unknown" : "unsuccessful"; LOG.warning("JDBC transaction " + state + "; " + operation + " failed"); - } catch (RuntimeException ignored) { + } catch (RuntimeException | Error ignored) { // A user-installed logging handler must not turn a committed write - // back into an apparently retryable transaction failure. + // back into an apparently retryable transaction failure, even if it + // throws an Error with suppressed diagnostics disabled. } } 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 91302db4..ad981399 100644 --- a/teaql-sqlite/src/test/java/io/teaql/sqlite/GraphTraceSqliteTest.java +++ b/teaql-sqlite/src/test/java/io/teaql/sqlite/GraphTraceSqliteTest.java @@ -742,6 +742,49 @@ private static CheckedSave checkedSave(Fixture fixture, GraphEntity entity, Stri assertCommitAcknowledgementUnknown(false); } + @Test public void commitErrorAfterDurableWriteRequiresReload() throws Exception { + var armed = new AtomicBoolean(); + var injected = new AtomicBoolean(); + var fixture = new Fixture(true, + source -> commitErrorAfterWrite(source, armed, injected)); + var order = fixture.create("CustomerOrder", 193, "commit Error after durable write"); + armed.set(true); + + assertThrows(NoSuppressionError.class, + () -> order.auditAs("probe uncertain commit Error").save(fixture.context)); + assertTrue("the commit Error must be injected", injected.get()); + 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()); + var retryFailure = assertThrows(TeaQLRuntimeException.class, + () -> order.auditAs("reject replay after commit Error").save(fixture.context)); + assertTrue(retryFailure.getMessage().contains("TRANSACTION_OUTCOME_UNKNOWN_RELOAD_REQUIRED")); + } + + @Test public void commitFailureAndRollbackErrorMustNotRestoreAutoCommit() throws Exception { + var armed = new AtomicBoolean(); + var rollbackInjected = new AtomicBoolean(); + var restoredAutoCommit = new AtomicBoolean(); + var fixture = new Fixture(true, source -> commitRollbackErrorFault( + source, armed, rollbackInjected, restoredAutoCommit)); + var order = fixture.create("CustomerOrder", 195, "commit and rollback uncertain"); + armed.set(true); + + var failure = assertThrows(TransactionOutcomeUnknownException.class, + () -> order.auditAs("probe commit and rollback Error").save(fixture.context)); + assertTrue(failure.getMessage().contains("TRANSACTION_OUTCOME_UNKNOWN")); + assertTrue("rollback Error must be injected", rollbackInjected.get()); + assertFalse("unconfirmed rollback must not restore autoCommit", restoredAutoCommit.get()); + assertTrue("failed commit must not materialize the business row", fixture.driver.queryForList( + "SELECT id FROM customer_order_data WHERE id = ?", new Object[] {195L}).isEmpty()); + assertTrue("failed commit must not materialize Outbox", fixture.committedRecorder.eventStore() + .outboxAfter("java-graph-source", 0, 10).isEmpty()); + var retryFailure = assertThrows(TeaQLRuntimeException.class, + () -> order.auditAs("reject replay after commit and rollback Error").save(fixture.context)); + assertTrue(retryFailure.getMessage().contains("TRANSACTION_OUTCOME_UNKNOWN_RELOAD_REQUIRED")); + } + @Test public void failedRollbackMustNotImplicitlyCommitOnAutoCommitRestoration() throws Exception { var armed = new AtomicBoolean(); var injected = new AtomicBoolean(); @@ -765,6 +808,25 @@ private static CheckedSave checkedSave(Fixture fixture, GraphEntity entity, Stri assertTrue(retryFailure.getMessage().contains("TRANSACTION_OUTCOME_UNKNOWN_RELOAD_REQUIRED")); } + @Test public void rollbackErrorMustNotRestoreAutoCommitOrPermitReplay() throws Exception { + var armed = new AtomicBoolean(); + var injected = new AtomicBoolean(); + var restoredAutoCommit = new AtomicBoolean(); + var fixture = new Fixture(true, + source -> rollbackFault(source, armed, injected, restoredAutoCommit, true)); + fixture.driver.execute("DROP TABLE teaql_event_outbox"); + var order = fixture.create("CustomerOrder", 194, "rollback Error after Outbox failure"); + armed.set(true); + + assertThrows(NoSuppressionError.class, + () -> order.auditAs("probe rollback Error").save(fixture.context)); + assertTrue("the rollback Error must be injected", injected.get()); + assertFalse("unconfirmed rollback must not restore autoCommit", restoredAutoCommit.get()); + var retryFailure = assertThrows(TeaQLRuntimeException.class, + () -> order.auditAs("reject replay after rollback Error").save(fixture.context)); + assertTrue(retryFailure.getMessage().contains("TRANSACTION_OUTCOME_UNKNOWN_RELOAD_REQUIRED")); + } + @Test public void failedRollbackAfterErrorRequiresReloadBeforeRetry() throws Exception { var armed = new AtomicBoolean(); var injected = new AtomicBoolean(); @@ -880,8 +942,58 @@ private static DataSource commitAcknowledgementFault(SQLiteDataSource source, }); } + private static DataSource commitErrorAfterWrite(SQLiteDataSource source, + AtomicBoolean armed, AtomicBoolean injected) { + return (DataSource) Proxy.newProxyInstance(DataSource.class.getClassLoader(), + new Class[] {DataSource.class}, (proxy, method, args) -> { + Object result = invoke(source, method, args); + if (!method.getName().equals("getConnection")) return result; + Connection actual = (Connection) result; + return Proxy.newProxyInstance(Connection.class.getClassLoader(), + new Class[] {Connection.class}, (connectionProxy, operation, parameters) -> { + if (operation.getName().equals("commit") && armed.get() + && injected.compareAndSet(false, true)) { + invoke(actual, operation, parameters); + throw new NoSuppressionError(); + } + return invoke(actual, operation, parameters); + }); + }); + } + + private static DataSource commitRollbackErrorFault(SQLiteDataSource source, + AtomicBoolean armed, AtomicBoolean rollbackInjected, AtomicBoolean restoredAutoCommit) { + return (DataSource) Proxy.newProxyInstance(DataSource.class.getClassLoader(), + new Class[] {DataSource.class}, (proxy, method, args) -> { + Object result = invoke(source, method, args); + if (!method.getName().equals("getConnection")) return result; + Connection actual = (Connection) result; + return Proxy.newProxyInstance(Connection.class.getClassLoader(), + new Class[] {Connection.class}, (connectionProxy, operation, parameters) -> { + if (operation.getName().equals("commit") && armed.get()) { + throw new SQLException("COMMIT-ACK-SECRET-CANARY"); + } + if (operation.getName().equals("rollback") && armed.get() + && rollbackInjected.compareAndSet(false, true)) { + throw new NoSuppressionError(); + } + if (operation.getName().equals("setAutoCommit") + && rollbackInjected.get() && Boolean.TRUE.equals(parameters[0])) { + restoredAutoCommit.set(true); + } + return invoke(actual, operation, parameters); + }); + }); + } + private static DataSource rollbackFault(SQLiteDataSource source, AtomicBoolean armed, AtomicBoolean injected, AtomicBoolean restoredAutoCommit) { + return rollbackFault(source, armed, injected, restoredAutoCommit, false); + } + + private static DataSource rollbackFault(SQLiteDataSource source, + AtomicBoolean armed, AtomicBoolean injected, AtomicBoolean restoredAutoCommit, + boolean throwError) { return (DataSource) Proxy.newProxyInstance(DataSource.class.getClassLoader(), new Class[] {DataSource.class}, (proxy, method, args) -> { Object result = invoke(source, method, args); @@ -891,6 +1003,7 @@ private static DataSource rollbackFault(SQLiteDataSource source, new Class[] {Connection.class}, (connectionProxy, operation, parameters) -> { if (operation.getName().equals("rollback") && armed.get() && injected.compareAndSet(false, true)) { + if (throwError) throw new NoSuppressionError(); throw new SQLException("ROLLBACK-SECRET-CANARY"); } if (operation.getName().equals("setAutoCommit") && injected.get() diff --git a/teaql-sqlite/src/test/java/io/teaql/sqlite/JdbcCommittedCleanupSqliteTest.java b/teaql-sqlite/src/test/java/io/teaql/sqlite/JdbcCommittedCleanupSqliteTest.java index 14e88db0..6c471370 100644 --- a/teaql-sqlite/src/test/java/io/teaql/sqlite/JdbcCommittedCleanupSqliteTest.java +++ b/teaql-sqlite/src/test/java/io/teaql/sqlite/JdbcCommittedCleanupSqliteTest.java @@ -23,7 +23,13 @@ /** A committed JDBC write must not look like a rolled-back, retryable save. */ public class JdbcCommittedCleanupSqliteTest { private static final String DIAGNOSTIC_CANARY = "POST-COMMIT-CLEANUP-SECRET-CANARY"; - private enum Fault { AUTO_COMMIT_RESTORE, CLOSE } + private enum Fault { AUTO_COMMIT_RESTORE, CLOSE, AUTO_COMMIT_RESTORE_ERROR, CLOSE_ERROR } + + private static final class NoSuppressionError extends Error { + private NoSuppressionError() { + super(DIAGNOSTIC_CANARY, null, false, false); + } + } @Test public void failedAutoCommitRestoreAfterCommitKeepsCommittedOutcome() throws Exception { verifyCommittedCleanupFailure(Fault.AUTO_COMMIT_RESTORE); @@ -37,12 +43,29 @@ private enum Fault { AUTO_COMMIT_RESTORE, CLOSE } verifyCommittedCleanupFailure(Fault.AUTO_COMMIT_RESTORE, true); } + @Test public void autoCommitRestoreErrorAfterCommitKeepsCommittedOutcome() throws Exception { + verifyCommittedCleanupFailure(Fault.AUTO_COMMIT_RESTORE_ERROR); + } + + @Test public void closeErrorAfterCommitKeepsCommittedOutcome() throws Exception { + verifyCommittedCleanupFailure(Fault.CLOSE_ERROR); + } + + @Test public void warningHandlerErrorAfterCommitKeepsCommittedOutcome() throws Exception { + verifyCommittedCleanupFailure(Fault.AUTO_COMMIT_RESTORE, false, true); + } + private static void verifyCommittedCleanupFailure(Fault fault) throws Exception { verifyCommittedCleanupFailure(fault, false); } private static void verifyCommittedCleanupFailure(Fault fault, boolean warningHandlerFails) throws Exception { + verifyCommittedCleanupFailure(fault, warningHandlerFails, false); + } + + private static void verifyCommittedCleanupFailure(Fault fault, boolean warningHandlerFails, + boolean warningHandlerThrowsError) throws Exception { var backing = new SQLiteDataSource(); backing.setUrl("jdbc:sqlite:" + Files.createTempFile("teaql-committed-cleanup-", ".db")); try (var connection = backing.getConnection(); var statement = connection.createStatement()) { @@ -59,13 +82,16 @@ private static void verifyCommittedCleanupFailure(Fault fault, boolean warningHa new Class[] {Connection.class}, (connectionProxy, operation, parameters) -> { if (operation.getName().equals("setAutoCommit") && Boolean.TRUE.equals(parameters[0]) && committed.get() - && fault == Fault.AUTO_COMMIT_RESTORE) { + && (fault == Fault.AUTO_COMMIT_RESTORE + || fault == Fault.AUTO_COMMIT_RESTORE_ERROR)) { + if (fault == Fault.AUTO_COMMIT_RESTORE_ERROR) throw new NoSuppressionError(); throw new SQLException(DIAGNOSTIC_CANARY); } Object result = invoke(actual, operation, parameters); if (operation.getName().equals("commit")) committed.set(true); if (operation.getName().equals("close") && committed.get() - && fault == Fault.CLOSE) { + && (fault == Fault.CLOSE || fault == Fault.CLOSE_ERROR)) { + if (fault == Fault.CLOSE_ERROR) throw new NoSuppressionError(); throw new SQLException(DIAGNOSTIC_CANARY); } return result; @@ -77,6 +103,7 @@ private static void verifyCommittedCleanupFailure(Fault fault, boolean warningHa Handler capture = new Handler() { @Override public void publish(LogRecord record) { warnings.add(record.getMessage()); + if (warningHandlerThrowsError) throw new NoSuppressionError(); if (warningHandlerFails) throw new IllegalStateException(DIAGNOSTIC_CANARY); } @Override public void flush() { } @@ -84,11 +111,11 @@ private static void verifyCommittedCleanupFailure(Fault fault, boolean warningHa }; logger.addHandler(capture); - RuntimeException reported = null; + Throwable reported = null; try { executor.executeInTransaction(() -> executor.update( "INSERT INTO committed_cleanup_probe (id) VALUES (?)", new Object[] {1L})); - } catch (RuntimeException failure) { + } catch (RuntimeException | Error failure) { reported = failure; } finally { logger.removeHandler(capture);