feat(ios): show an open transaction on every tab, with Commit and Roll Back

apple/Packages/TableProCore/Sources/TableProDatabase/ConnectionManager.swift

@@ -16,6 +16,7 @@ public final class ConnectionManager: @unchecked Sendable {
     private var blockingTeardowns: Set<UUID> = []
     private var attemptGenerations: [UUID: Int] = [:]
     private var lastTeardownId = 0
+    private var sessionEndObservers: [@Sendable (UUID, any DatabaseDriver) -> Void] = []
 
     public init(
         driverFactory: DriverFactory,
@@ -213,9 +214,10 @@ public final class ConnectionManager: @unchecked Sendable {
     /// `retireTeardown` cannot clear an entry that is not installed yet.
     private func claimTeardown(for connectionId: UUID) -> ConnectionTeardown? {
         lock.lock()
-        defer { lock.unlock() }
         guard let session = sessions.removeValue(forKey: connectionId) else {
-            return teardowns[connectionId]
+            let running = teardowns[connectionId]
+            lock.unlock()
+            return running
         }
         lastTeardownId += 1
         let teardown = ConnectionTeardown(id: lastTeardownId)
@@ -232,9 +234,21 @@ public final class ConnectionManager: @unchecked Sendable {
         if session.driver.holdsSuspensionBlockingResource {
             blockingTeardowns.insert(connectionId)
         }
+        let observers = sessionEndObservers
+        lock.unlock()
+        // After the unlock: an observer may call back into the manager, and this lock is not recursive.
+        for observer in observers {
+            observer(connectionId, session.driver)
+        }
         return teardown
     }
 
+    public func observeSessionEnds(_ observer: @escaping @Sendable (UUID, any DatabaseDriver) -> Void) {
+        lock.lock()
+        defer { lock.unlock() }
+        sessionEndObservers.append(observer)
+    }
+
     /// Runs before `ConnectionTeardown.finish()`, so a waiter it wakes already reads the cleared state.
     private func retireTeardown(_ teardownId: Int, for connectionId: UUID) {
         lock.lock()

apple/Packages/TableProCore/Sources/TableProDatabase/DriverTransactionState.swift

@@ -4,5 +4,7 @@ public enum DriverTransactionState: Sendable, Equatable {
     case idle
     case explicitTransaction
     case implicitTransaction
+    // A statement failed inside the transaction: the server refuses everything until a ROLLBACK, and a COMMIT rolls back.
+    case abortedTransaction
     case unknown
 }

apple/Packages/TableProCore/Sources/TableProDatabase/WriteTransactionPolicy.swift

@@ -8,9 +8,10 @@ public enum WriteTransactionPolicy {
     ) -> Bool {
         guard supportsTransactions else { return false }
         switch state {
-        case .idle, .implicitTransaction:
+        case .idle:
             return true
-        case .explicitTransaction:
+        // MySQL's START TRANSACTION commits the work an autocommit-off session holds, so the write joins it.
+        case .explicitTransaction, .implicitTransaction, .abortedTransaction:
             return false
         case .unknown:
             return statementCount > 1

apple/Packages/TableProCore/Tests/TableProDatabaseTests/ConnectionManagerTests.swift

@@ -570,4 +570,72 @@ struct ConnectionManagerTests {
 
         #expect(ssh.tunnelledConnectionIds == [connection.id])
     }
+
+    @Test("A replaced or disconnected session reaches the observer with the driver it ended")
+    func sessionEndsReachTheObserver() async throws {
+        let factory = MockDriverFactory()
+        let manager = ConnectionManager(driverFactory: factory, secureStore: MockSecureStore())
+        let connection = DatabaseConnection(name: "Test", type: DatabaseType(rawValue: "mock"))
+        let ended = EndedSessions()
+        manager.observeSessionEnds { id, driver in
+            ended.record(id, driver, stillListed: false)
+        }
+
+        let first = MockDatabaseDriver()
+        factory.drivers["mock"] = first
+        _ = try await manager.connect(connection)
+        #expect(ended.records.isEmpty)
+
+        let second = MockDatabaseDriver()
+        factory.drivers["mock"] = second
+        _ = try await manager.connect(connection)
+        await manager.disconnect(connection.id)
+
+        let records = ended.records
+        let endedIds = records.map { $0.id }
+        #expect(endedIds == [connection.id, connection.id])
+        #expect(records.first?.driver === first)
+        #expect(records.last?.driver === second)
+    }
+
+    @Test("An observer that reads the manager back runs after the session is gone, outside the lock")
+    func observerCanReadTheManager() async throws {
+        let factory = MockDriverFactory()
+        factory.drivers["mock"] = MockDatabaseDriver()
+        let manager = ConnectionManager(driverFactory: factory, secureStore: MockSecureStore())
+        let connection = DatabaseConnection(name: "Test", type: DatabaseType(rawValue: "mock"))
+        let ended = EndedSessions()
+        manager.observeSessionEnds { id, driver in
+            ended.record(id, driver, stillListed: manager.session(for: id) != nil)
+        }
+
+        _ = try await manager.connect(connection)
+        await manager.disconnect(connection.id)
+
+        let stillListed = ended.records.map { $0.stillListed }
+        #expect(stillListed == [false])
+    }
+}
+
+private final class EndedSessions: @unchecked Sendable {
+    struct Record {
+        let id: UUID
+        let driver: any DatabaseDriver
+        let stillListed: Bool
+    }
+
+    private let lock = NSLock()
+    private var recorded: [Record] = []
+
+    var records: [Record] {
+        lock.lock()
+        defer { lock.unlock() }
+        return recorded
+    }
+
+    func record(_ id: UUID, _ driver: any DatabaseDriver, stillListed: Bool) {
+        lock.lock()
+        defer { lock.unlock() }
+        recorded.append(Record(id: id, driver: driver, stillListed: stillListed))
+    }
 }

apple/Packages/TableProCore/Tests/TableProDatabaseTests/WriteTransactionTests.swift

@@ -111,7 +111,7 @@ private final class ProtocolDefaultDriver: DatabaseDriver, @unchecked Sendable {
 struct WriteTransactionPolicyTests {
     @Test("A driver without transactions never opens one")
     func withoutTransactions() {
-        for state in [DriverTransactionState.idle, .explicitTransaction, .implicitTransaction, .unknown] {
+        for state in [DriverTransactionState.idle, .explicitTransaction, .implicitTransaction, .abortedTransaction, .unknown] {
             for count in 1...2 {
                 #expect(
                     WriteTransactionPolicy.opensTransaction(
@@ -128,17 +128,31 @@ struct WriteTransactionPolicyTests {
         #expect(WriteTransactionPolicy.opensTransaction(supportsTransactions: true, state: .idle, statementCount: 2))
     }
 
-    @Test("A transaction the server opened because autocommit is off still wraps and commits")
-    func implicitTransactionWraps() {
+    @Test("A transaction the server opened because autocommit is off is joined, since a BEGIN would commit it")
+    func implicitTransactionIsJoined() {
         #expect(
             WriteTransactionPolicy.opensTransaction(
                 supportsTransactions: true, state: .implicitTransaction, statementCount: 1
-            )
+            ) == false
         )
         #expect(
             WriteTransactionPolicy.opensTransaction(
                 supportsTransactions: true, state: .implicitTransaction, statementCount: 2
-            )
+            ) == false
+        )
+    }
+
+    @Test("A transaction a failed statement aborted is joined, so the server's refusal reaches the caller")
+    func abortedTransactionIsJoined() {
+        #expect(
+            WriteTransactionPolicy.opensTransaction(
+                supportsTransactions: true, state: .abortedTransaction, statementCount: 1
+            ) == false
+        )
+        #expect(
+            WriteTransactionPolicy.opensTransaction(
+                supportsTransactions: true, state: .abortedTransaction, statementCount: 2
+            ) == false
         )
     }
 
@@ -192,14 +206,27 @@ struct DatabaseDriverWriteTests {
         #expect(driver.events == [.execute("UPDATE t SET v = 1")])
     }
 
-    @Test("A session with autocommit off is wrapped and committed")
-    func implicitTransactionWraps() async throws {
+    @Test("A session with autocommit off runs the batch inside the work it holds, with no BEGIN or COMMIT")
+    func implicitTransactionIsJoined() async throws {
         let driver = RecordingWriteDriver()
         driver.scriptedState = .implicitTransaction
 
-        _ = try await driver.executeWrite(["UPDATE t SET v = 1"])
+        _ = try await driver.executeWrite(["INSERT a", "INSERT b"])
 
-        #expect(driver.events == [.begin(.readWrite), .execute("UPDATE t SET v = 1"), .commit])
+        #expect(driver.events == [.execute("INSERT a"), .execute("INSERT b")])
+    }
+
+    @Test("A session in an aborted transaction sends the statement bare and leaves the rollback to the user")
+    func abortedTransactionIsJoined() async throws {
+        let driver = RecordingWriteDriver()
+        driver.scriptedState = .abortedTransaction
+        driver.failingStatement = "UPDATE t SET v = 1"
+
+        await #expect(throws: WriteFailure.self) {
+            _ = try await driver.executeWrite(["UPDATE t SET v = 1"])
+        }
+
+        #expect(driver.events == [.execute("UPDATE t SET v = 1")])
     }
 
     @Test("An unknown state leaves a single statement bare and wraps a batch")

apple/Packages/TableProOracle/Sources/TableProOracleCore/OracleCoreConnection.swift

@@ -579,14 +579,21 @@ public final class OracleCoreConnection: @unchecked Sendable {
         }
     }
 
-    private func recordSuccess(of role: OracleTransactionRole, on session: Int) {
-        state.withLock { $0.transaction.statementSucceeded(role, on: session) }
+    private func recordSuccess(of role: OracleTransactionRole, on session: Int) async {
+        var answer: Bool?
+        if role == .dataDefinition, holdsTransaction {
+            answer = await serverHoldsTransaction()
+        }
+        let serverHoldsTransaction = answer
+        state.withLock {
+            $0.transaction.statementSucceeded(role, on: session, serverHoldsTransaction: serverHoldsTransaction)
+        }
     }
 
     /// Called under the query gate, so the answer about the transaction comes from the connection the refused
-    /// `COMMIT` or `ROLLBACK` ran on.
+    /// statement ran on.
     private func recordFailure(of role: OracleTransactionRole) async {
-        guard role == .endsTransaction, holdsTransaction else { return }
+        guard role.mayEndTransactionWhenRefused, holdsTransaction else { return }
         let serverHoldsTransaction = await serverHoldsTransaction()
         state.withLock { $0.transaction.statementFailed(role, serverHoldsTransaction: serverHoldsTransaction) }
     }
@@ -765,7 +772,7 @@ public final class OracleCoreConnection: @unchecked Sendable {
                 let zones = await sessionZones(on: connection)
                 return try await collectRows(statement, options: admitted.options, zones: zones, on: connection)
             }
-            recordSuccess(of: role, on: admitted.session)
+            await recordSuccess(of: role, on: admitted.session)
             forgetSessionZonesIfChanged(by: query)
             await queryGate.release()
             return result
@@ -872,7 +879,7 @@ public final class OracleCoreConnection: @unchecked Sendable {
                 let zones = await sessionZones(on: connection)
                 try await streamRows(query, options: admitted.options, zones: zones, on: connection, continuation: continuation)
             }
-            recordSuccess(of: role, on: admitted.session)
+            await recordSuccess(of: role, on: admitted.session)
             forgetSessionZonesIfChanged(by: query)
             await queryGate.release()
             continuation.finish()

apple/Packages/TableProOracle/Sources/TableProOracleCore/OracleSessionTransaction.swift

@@ -8,7 +8,7 @@ import Foundation
 /// driver's commit flag and the server commits it in the same round trip.
 ///
 /// A transaction opens with ``open()``, which is what the app's own writes call, or with a statement that opens one
-/// (``OracleTransactionRole/opensTransaction``). It ends with a `COMMIT` or `ROLLBACK`, whoever sends it.
+/// (``OracleTransactionRole/opensTransaction``). It ends with a `COMMIT`, a `ROLLBACK` or DDL, whoever sends it.
 ///
 /// The transaction belongs to the connection its first statement ran on. A connection that closes takes the
 /// transaction with it, so a statement that finds the transaction's connection replaced fails with
@@ -31,34 +31,40 @@ struct OracleSessionTransaction: Equatable, Sendable {
             return false
         case .opensTransaction, .endsTransaction:
             guard isOpen else { return false }
-        case .other:
+        case .dataDefinition, .other:
             guard isOpen else { return true }
         }
         try bind(to: session)
         return false
     }
 
-    /// A statement that opens a transaction opens it here only once Oracle has accepted it, and a `COMMIT` or
-    /// `ROLLBACK` ends it.
-    mutating func statementSucceeded(_ role: OracleTransactionRole, on session: Int) {
+    // DDL commits around itself, but DDL on a private temporary table can leave the transaction open: the server decides.
+    mutating func statementSucceeded(
+        _ role: OracleTransactionRole,
+        on session: Int,
+        serverHoldsTransaction: Bool? = nil
+    ) {
         switch role {
         case .opensTransaction:
             isOpen = true
             self.session = session
         case .endsTransaction:
             close()
+        case .dataDefinition:
+            if serverHoldsTransaction == false { close() }
         case .query, .other:
             break
         }
     }
 
-    /// A `COMMIT` or `ROLLBACK` the server refused ends the transaction only when the server no longer holds one: a
-    /// malformed one leaves it open, and a commit a deferred constraint rolled back (ORA-02091) ends it.
+    /// A `COMMIT`, `ROLLBACK` or DDL statement the server refused ends the transaction only when the server no longer
+    /// holds one: a malformed one leaves it open, a commit a deferred constraint rolled back (ORA-02091) ends it, and
+    /// so does DDL that failed after the commit Oracle runs before it.
     ///
     /// `serverHoldsTransaction` is nil when there was no live connection to ask. The transaction is then left as it
     /// is, and the next write finds out whether its connection survived.
     mutating func statementFailed(_ role: OracleTransactionRole, serverHoldsTransaction: Bool?) {
-        guard role == .endsTransaction, serverHoldsTransaction == false else { return }
+        guard role.mayEndTransactionWhenRefused, serverHoldsTransaction == false else { return }
         close()
     }
 

apple/Packages/TableProOracle/Sources/TableProOracleCore/OracleTransactionRole.swift

@@ -15,8 +15,12 @@ enum OracleTransactionRole: Equatable, Sendable {
     /// `COMMIT`, or a `ROLLBACK` that is not to a savepoint.
     case endsTransaction
 
-    /// Everything else: DML, DDL, PL/SQL, `CALL`, `ROLLBACK TO`, and the `FORCE` forms of `COMMIT` and `ROLLBACK`,
-    /// which settle an in-doubt distributed transaction rather than the session's own.
+    // Oracle commits before and after DDL (SQL Reference, COMMIT). ALTER SESSION and ALTER SYSTEM are
+    // session and system control, which commit nothing, and the schema switch sends ALTER SESSION.
+    case dataDefinition
+
+    /// Everything else: DML, PL/SQL, `CALL`, `ROLLBACK TO`, session and system control, and the `FORCE` forms of
+    /// `COMMIT` and `ROLLBACK`, which settle an in-doubt distributed transaction rather than the session's own.
     case other
 
     /// How much of a statement is read for its first words. A dump can hold a statement millions of characters long,
@@ -39,11 +43,24 @@ enum OracleTransactionRole: Equatable, Sendable {
         case "ROLLBACK":
             let word = Self.skippingWork(&reader)
             self = word == "TO" || word == "FORCE" ? .other : .endsTransaction
+        case "CREATE", "DROP", "TRUNCATE", "RENAME", "GRANT", "REVOKE", "COMMENT", "AUDIT", "NOAUDIT", "ANALYZE",
+             "PURGE", "FLASHBACK":
+            self = .dataDefinition
+        case "ALTER":
+            let word = reader.nextWord()
+            self = word == "SESSION" || word == "SYSTEM" ? .other : .dataDefinition
+        case "ASSOCIATE", "DISASSOCIATE":
+            self = reader.nextWord() == "STATISTICS" ? .dataDefinition : .other
         default:
             self = .other
         }
     }
 
+    // A refused COMMIT, ROLLBACK or DDL statement may still have ended the transaction, so only the server can say.
+    var mayEndTransactionWhenRefused: Bool {
+        self == .endsTransaction || self == .dataDefinition
+    }
+
     private static func skippingWork(_ reader: inout HeaderReader) -> String? {
         let word = reader.nextWord()
         guard word == "WORK" else { return word }