feat: Update suspendTransaction to use stack as SSOT and cleanup

Author: bog-walk

exposed-r2dbc/api/exposed-r2dbc.api

@@ -1072,7 +1072,6 @@ public final class org/jetbrains/exposed/v1/r2dbc/transactions/TransactionsKt {
 	public static synthetic fun inTopLevelSuspendTransaction$default (Lorg/jetbrains/exposed/v1/r2dbc/R2dbcDatabase;Lio/r2dbc/spi/IsolationLevel;Ljava/lang/Boolean;Lorg/jetbrains/exposed/v1/r2dbc/R2dbcTransaction;Lkotlin/jvm/functions/Function2;Lkotlin/coroutines/Continuation;ILjava/lang/Object;)Ljava/lang/Object;
 	public static final fun suspendTransaction (Lorg/jetbrains/exposed/v1/r2dbc/R2dbcDatabase;Lio/r2dbc/spi/IsolationLevel;Ljava/lang/Boolean;Lkotlin/jvm/functions/Function2;Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
 	public static synthetic fun suspendTransaction$default (Lorg/jetbrains/exposed/v1/r2dbc/R2dbcDatabase;Lio/r2dbc/spi/IsolationLevel;Ljava/lang/Boolean;Lkotlin/jvm/functions/Function2;Lkotlin/coroutines/Continuation;ILjava/lang/Object;)Ljava/lang/Object;
-	public static final fun viewThreadStack ()Ljava/lang/String;
 }
 
 public abstract class org/jetbrains/exposed/v1/r2dbc/vendors/DatabaseDialectMetadata {

exposed-r2dbc/src/main/kotlin/org/jetbrains/exposed/v1/r2dbc/statements/R2dbcConnectionImpl.kt

@@ -48,7 +48,14 @@ class R2dbcConnectionImpl(
 
     /**
      * Retrieves a publisher that provides only the single current underlying database [Connection] instance in use.
-     * If none is active, it awaits a value from the [connection] publisher and internally stores the result.
+     *
+     * If none is active, it awaits a value from the [connection] publisher and internally stores the result. Retrieval
+     * of a new value will invoke `Connection.beginTransaction()`, so this method should preferably be called when an active
+     * transaction is underway, in order to retrieve the accurate connection object instance.
+     *
+     * If this method is invoked intentionally to trigger the start of a transaction, then [setTransactionDefinition]
+     * should ideally be manually called first with the appropriate [TransactionDefinition], as well as any other
+     * [Connection] configuration methods.
      */
     override suspend fun activeConnection(): Publisher<out Connection> {
         // retrieves localConnection if not null, otherwise awaits value from connection publisher

exposed-r2dbc/src/main/kotlin/org/jetbrains/exposed/v1/r2dbc/transactions/R2dbcTransactionManager.kt

@@ -1,7 +1,6 @@
 package org.jetbrains.exposed.v1.r2dbc.transactions
 
 import io.r2dbc.spi.IsolationLevel
-import kotlinx.coroutines.currentCoroutineContext
 import org.jetbrains.exposed.v1.core.InternalApi
 import org.jetbrains.exposed.v1.core.Transaction
 import org.jetbrains.exposed.v1.core.transactions.ThreadLocalTransactionsStack
@@ -65,23 +64,23 @@ internal fun R2dbcTransactionManager.createTransactionContext(transaction: Trans
 }
 
 /**
- * Returns the current R2DBC transaction from the coroutine context, or null if none exists.
+ * Returns the current R2DBC transaction from the stack, or null if none exists.
  *
- * This method performs type checking to ensure the transaction in the context is actually
- * an [R2dbcTransaction]. If a non-R2DBC transaction is found in the context, an error is thrown
+ * This method performs type checking to ensure the transaction in the stack is actually
+ * an [R2dbcTransaction]. If a non-R2DBC transaction is found in the stack, an error is thrown
  * to prevent type confusion between JDBC and R2DBC transactions.
  *
- * @return The current [R2dbcTransaction] from the coroutine context, or null if no transaction exists
- * @throws [IllegalStateException] If the transaction in the context is not an [R2dbcTransaction]
+ * @return The current [R2dbcTransaction] from the stack, or null if no transaction exists
+ * @throws [IllegalStateException] If the transaction in the stack is not an [R2dbcTransaction]
  */
-internal suspend fun R2dbcTransactionManager.getCurrentContextTransaction(): R2dbcTransaction? {
+internal fun R2dbcTransactionManager.getCurrentStackTransaction(): R2dbcTransaction? {
     @OptIn(InternalApi::class)
-    val transaction = currentCoroutineContext()[contextKey]?.transaction
+    val transaction = ThreadLocalTransactionsStack.getTransactionOrNull(db)
     return when {
         transaction == null -> null
         transaction is R2dbcTransaction -> transaction
         else -> error(
-            "Expected R2dbcTransaction in coroutine context but found ${transaction::class.simpleName}. " +
+            "Expected R2dbcTransaction in stack but found ${transaction::class.simpleName}. " +
                 "This may indicate mixing JDBC and R2DBC transactions incorrectly."
         )
     }

exposed-r2dbc/src/main/kotlin/org/jetbrains/exposed/v1/r2dbc/transactions/Transactions.kt

@@ -113,9 +113,7 @@ suspend fun <T> suspendTransaction(
     statement: suspend R2dbcTransaction.() -> T
 ): T {
     val databaseToUse = resolveR2dbcDatabaseOrThrow(db)
-    val outer = databaseToUse.transactionManager.getCurrentContextTransaction()
-//    @OptIn(InternalApi::class)
-//    val outer = ThreadLocalTransactionsStack.getTransactionOrNull(databaseToUse) as? R2dbcTransaction
+    val outer = databaseToUse.transactionManager.getCurrentStackTransaction()
 
     return if (outer != null) {
         val transaction = outer.transactionManager.newTransaction(
@@ -287,11 +285,3 @@ internal suspend fun closeStatementsAndConnection(transaction: R2dbcTransaction)
         exposedLogger.warn("Transaction close failed: ${it.message}. Statement: $currentStatement", it)
     }
 }
-
-@OptIn(InternalApi::class)
-fun viewThreadStack(): String {
-    val currentThread = Thread.currentThread().name
-    val currentTrx = ThreadLocalTransactionsStack.getTransactionOrNull()?.transactionId ?: "NOT IN TRX"
-    val allTrx = ThreadLocalTransactionsStack.threadTransactions()?.map { it.transactionId } ?: listOf("EMPTY STACK")
-    return "\n\tTHREAD --> $currentThread\n\tTRX --> $currentTrx\n\tSTACK --> $allTrx"
-}

spring7-reactive-transaction/src/main/kotlin/org/jetbrains/exposed/v1/spring7/reactive/transaction/SpringReactiveTransactionManager.kt

@@ -18,7 +18,6 @@ import org.jetbrains.exposed.v1.r2dbc.R2dbcDatabaseConfig
 import org.jetbrains.exposed.v1.r2dbc.R2dbcTransaction
 import org.jetbrains.exposed.v1.r2dbc.transactions.currentOrNull
 import org.jetbrains.exposed.v1.r2dbc.transactions.transactionManager
-import org.jetbrains.exposed.v1.r2dbc.transactions.viewThreadStack
 import org.reactivestreams.Publisher
 import org.springframework.r2dbc.UncategorizedR2dbcException
 import org.springframework.r2dbc.connection.ConnectionHolder
@@ -52,8 +51,6 @@ class SpringReactiveTransactionManager(
     override fun doGetTransaction(
         synchronizationManager: TransactionSynchronizationManager
     ): Any {
-        synchronizationManager.printEverything(::doGetTransaction.name)
-
         return ExposedTransactionObject(
             database = database,
         ).apply {
@@ -68,9 +65,7 @@ class SpringReactiveTransactionManager(
         return Mono.defer {
             val trxObject = transaction as ExposedTransactionObject
 
-            synchronizationManager.printEverything(::doSuspend.name)
-
-            synchronizationManager.getResourceOrNull() ?: error("No transaction to suspend")
+            synchronizationManager.getResourceOrNull() ?: error("No synchronized transaction to suspend")
 
             Mono
                 .justOrEmpty(
@@ -81,8 +76,6 @@ class SpringReactiveTransactionManager(
 
                     @OptIn(InternalApi::class)
                     ThreadLocalTransactionsStack.popTransaction()
-
-                    synchronizationManager.printEverything(::doSuspend.name)
                 }
         }
     }
@@ -100,17 +93,13 @@ class SpringReactiveTransactionManager(
             @OptIn(InternalApi::class)
             ThreadLocalTransactionsStack.pushTransaction(suspendedObject.transaction)
 
-            synchronizationManager.printEverything(::doResume.name)
-
             Mono.empty()
         }
     }
 
     override fun isExistingTransaction(transaction: Any): Boolean {
         val trxObject = transaction as ExposedTransactionObject
 
-        println("In existingTransaction with ${trxObject.getCurrentTransaction()?.transactionId}")
-
         return trxObject.getCurrentTransaction() != null
     }
 
@@ -129,7 +118,6 @@ class SpringReactiveTransactionManager(
             } else {
                 null
             }
-            synchronizationManager.printEverything(::doBegin.name)
 
             val newTransaction = trxObject.database.transactionManager.newTransaction(
                 isolation = definition.isolationLevel.resolveIsolationLevel(),
@@ -174,11 +162,11 @@ class SpringReactiveTransactionManager(
                     if (trxObject.isNewConnectionHolder) {
                         // otherwise a PROPAGATION_NESTED transaction would incorrectly have the context of its outer
                         // transaction used when doCommit() or doRollback() is invoked
+                        // https://youtrack.jetbrains.com/issue/EXPOSED-996/Reassess-support-for-Spring-PROPAGATIONNESTED
                         synchronizationManager.unbindResourceIfPossible(connectionFactory) as? ExposedHolderObject
 
                         synchronizationManager.bindResource(connectionFactory, trxObject.connectionHolder!!)
                     }
-                    synchronizationManager.printEverything("${::doBegin.name} [inner]")
                 }
                 .doOnError { ex ->
                     trxObject.connectionHolder = null
@@ -199,14 +187,11 @@ class SpringReactiveTransactionManager(
         val trxObject = status.transaction as ExposedTransactionObject
 
         synchronizationManager.getResourceOrNull() ?: error("No synchronized transaction to commit")
-        synchronizationManager.printEverything(::doCommit.name)
 
         // force new coroutine to start in current thread so that doCleanupOnCompletion can access correct stack
         return mono(Dispatchers.Unconfined) {
             trxObject.commit()
 
-            synchronizationManager.printEverything("${::doCommit.name} [inner]")
-
             null
         }
     }
@@ -218,14 +203,11 @@ class SpringReactiveTransactionManager(
         val trxObject = status.transaction as ExposedTransactionObject
 
         synchronizationManager.getResourceOrNull() ?: error("No synchronized transaction to rollback")
-        synchronizationManager.printEverything(::doRollback.name)
 
         // force new coroutine to start in current thread so that doCleanupOnCompletion can access correct stack
         return mono(Dispatchers.Unconfined) {
             trxObject.rollback()
 
-            synchronizationManager.printEverything("${::doRollback.name} [inner]")
-
             null
         }
     }
@@ -244,13 +226,12 @@ class SpringReactiveTransactionManager(
             @OptIn(InternalApi::class)
             ThreadLocalTransactionsStack.popTransaction()
 
-            synchronizationManager.printEverything(::doCleanupAfterCompletion.name)
-
             if (trxObject.isNewConnectionHolder) {
                 val previous = synchronizationManager.unbindResource(connectionFactory) as ExposedHolderObject
 
                 // otherwise a PROPAGATION_NESTED transaction would incorrectly have the context of its
                 // now closed inner transaction used when doCommit() or doRollback() is later invoked
+                // https://youtrack.jetbrains.com/issue/EXPOSED-996/Reassess-support-for-Spring-PROPAGATIONNESTED
                 completedTransaction?.outerTransaction?.let { outer ->
                     synchronizationManager.bindResource(connectionFactory, ExposedHolderObject(previous.connection, outer))
                 }
@@ -260,16 +241,13 @@ class SpringReactiveTransactionManager(
             mono(Dispatchers.Unconfined) {
                 completedTransaction?.close()
 
-                synchronizationManager.printEverything(::doCleanupAfterCompletion.name)
-
                 null
             }
                 .doOnEach {
                     if (trxObject.isNewConnectionHolder) {
                         trxObject.connectionHolder?.released()
                     }
                     trxObject.connectionHolder?.clear()
-                    println("In ${::doCleanupAfterCompletion.name}: Releasing & clearing CH")
                 }
         }
     }
@@ -290,8 +268,6 @@ class SpringReactiveTransactionManager(
         return Mono.fromRunnable {
             val trxObject = status.transaction as ExposedTransactionObject
 
-            synchronizationManager.printEverything(::doSetRollbackOnly.name)
-
             if (status.isDebug) {
                 exposedLogger.debug("Exposed transaction [${status.transactionName}] set rollback-only")
             }
@@ -363,11 +339,6 @@ class SpringReactiveTransactionManager(
     private fun TransactionSynchronizationManager.getResourceOrNull(): R2dbcTransaction? {
         return this.getResourceHolderOrNull()?.transaction
     }
-
-    private fun TransactionSynchronizationManager.printEverything(methodName: String) {
-        val resource = this.getResourceOrNull()?.transactionId ?: "NO SPRING TRX"
-        println("In $methodName...${viewThreadStack()}\n\tSPRING --> $resource")
-    }
 }
 
 private var R2dbcTransaction.isRollback: Boolean by transactionScope { false }

spring7-reactive-transaction/src/test/kotlin/org/jetbrains/exposed/v1/spring7/reactive/transaction/ExposedTransactionManagerTest.kt

@@ -7,7 +7,6 @@ import org.jetbrains.exposed.v1.r2dbc.SchemaUtils
 import org.jetbrains.exposed.v1.r2dbc.insert
 import org.jetbrains.exposed.v1.r2dbc.selectAll
 import org.jetbrains.exposed.v1.r2dbc.transactions.suspendTransaction
-import org.jetbrains.exposed.v1.r2dbc.transactions.viewThreadStack
 import org.junit.jupiter.api.AfterEach
 import org.junit.jupiter.api.Assertions.assertEquals
 import org.junit.jupiter.api.BeforeEach
@@ -80,29 +79,20 @@ open class ExposedTransactionManagerTest : SpringReactiveTransactionTestBase() {
         }
     }
 
-    // TODO - This (& only this test) fails because of line 114 in suspendTransaction();
-    // If the line is reverted to original, it passes -> ThreadLocalTransactionsStack.getTransactionOrNull(databaseToUse)
-    @RepeatedTest(1)
+    @RepeatedTest(5)
     @Commit
 //    @Transactional // see [runTestWithMockTransactional]
     open fun testConnectionCombineWithExposedTransaction2() = runTestWithMockTransactional {
-        println("Starting TEST...\n${viewThreadStack()}")
         val rnd = Random().nextInt().toString()
         T1.insert {
             it[c1] = rnd
         }
         assertEquals(rnd, T1.selectAll().single()[T1.c1])
 
-        println("About to enter nested...")
         suspendTransaction {
-            println("Starting NESTED...\n${viewThreadStack()}")
             T1.insertRandom()
             assertEquals(2, T1.selectAll().count())
-            println("NESTED = ${T1.selectAll().count()}")
-            println("Finishing NESTED...")
         }
-        println("TEST = ${T1.selectAll().count()}")
-        println("Finishing TEST...\n${viewThreadStack()}")
     }
 
     /**

spring7-reactive-transaction/src/test/kotlin/org/jetbrains/exposed/v1/spring7/reactive/transaction/SpringMultiContainerTransactionTest.kt

@@ -113,6 +113,34 @@ open class SpringMultiContainerTransactionTest {
         Assertions.assertEquals(1, orders.findAllWithExposedTrxBlock().size)
         Assertions.assertEquals(2, payments.findAllWithExposedTrxBlock().size)
     }
+
+    @Test
+    open fun test9() = runTest {
+        kotlin.runCatching {
+            orders.suspendTransaction {
+                orders.createWithExposedTrxBlock()
+                payments.createWithExposedTrxBlock()
+                throw SpringTransactionTestException()
+            }
+        }
+        Assertions.assertEquals(0, orders.findAllWithExposedTrxBlock().size)
+        Assertions.assertEquals(1, payments.findAllWithExposedTrxBlock().size)
+    }
+
+    @Test
+    open fun test10() = runTest {
+        kotlin.runCatching {
+            orders.suspendTransaction {
+                orders.createWithExposedTrxBlock()
+                payments.databaseTemplate {
+                    payments.createWithExposedTrxBlock()
+                    throw SpringTransactionTestException()
+                }
+            }
+        }
+        Assertions.assertEquals(0, orders.findAllWithExposedTrxBlock().size)
+        Assertions.assertEquals(0, payments.findAllWithExposedTrxBlock().size)
+    }
 }
 
 @Configuration

spring7-transaction/src/test/kotlin/org/jetbrains/exposed/v1/spring7/transaction/JdbcExposedTransactionManagerTest.kt

@@ -17,7 +17,7 @@ import org.springframework.transaction.IllegalTransactionStateException
 import org.springframework.transaction.TransactionDefinition
 import org.springframework.transaction.annotation.Isolation
 import org.springframework.transaction.annotation.Transactional
-import java.util.*
+import java.util.Random
 import javax.sql.DataSource
 import kotlin.test.assertFailsWith
 

spring7-transaction/src/test/kotlin/org/jetbrains/exposed/v1/spring7/transaction/SpringTransactionRollbackTest.kt

@@ -275,8 +275,10 @@ class SpringTransactionRollbackTest {
         assertTrue { entities.none { it.name.startsWith("No ") || it.name.startsWith("New ") } }
     }
 
-    // Left out because rollback involving Spring (partial) NESTED is not actually supported by Exposed;
-    // this test would fail with "Transaction manager does not allow nested transactions by default" if catch block removed
+    // Left out because rollback involving Spring (partial) NESTED is not actually fully supported by Exposed;
+    // this test would fail with "Transaction manager does not allow nested transactions by default" if catch block removed,
+    // not the artificial RuntimeException as expected
+    // https://youtrack.jetbrains.com/issue/EXPOSED-996/Reassess-support-for-Spring-PROPAGATIONNESTED
     @Tag(MISSING_R2DBC_TEST)
     @Test
     fun `nested should rollback innerTx without affecting outerTx`() {