fix(datagrid): stop every second refresh from cancelling its own query (#2021)

CHANGELOG.md

@@ -7,6 +7,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Fixed
+
+- Refreshing a table no longer fails with "Query cancelled" on every second click. The refresh was sending the database a cancel for a row count that had already finished, and that cancel then aborted the reload it had just started. Covers PostgreSQL, Redshift, CockroachDB, MySQL, MariaDB, and Redis. (#2021)
+- Stopping a query no longer shows it as a failed query with a red error message. A query you stopped is no longer recorded as a failure in query history either.
+- Stopping a MySQL or MariaDB write, or a Redis command, no longer cancels the next statement you run on that connection.
+
 ## [0.63.0] - 2026-08-05
 
 ### Added

Plugins/MySQLDriverPlugin/MariaDBPluginConnection.swift

@@ -67,7 +67,7 @@
 func mysqlTypeToString(_ fieldPtr: UnsafePointer<MYSQL_FIELD>) -> String {
     let field = fieldPtr.pointee
     let flags = UInt(field.flags)
     let length = field.length
 
     // MariaDB extended metadata: detect JSON stored as LONGTEXT.
     // `MARIADB_CONST_STRING` is length-prefixed (not null-terminated), so we must read
@@ -162,10 +162,10 @@
     private let queryTimeoutSeconds: Int
 
     private let stateLock = NSLock()
+    private let cancellationGate = PluginQueryCancellationGate()
     private var _isConnected: Bool = false
     private var _isShuttingDown: Bool = false
     private var _cachedServerVersion: String?
-    private var _isCancelled: Bool = false
 
     var isConnected: Bool {
         stateLock.lock()
@@ -379,9 +379,7 @@
     // MARK: - Query Cancellation
 
     func cancelCurrentQuery() {
-        stateLock.lock()
-        _isCancelled = true
-        stateLock.unlock()
+        guard cancellationGate.cancel() != nil else { return }
 
         guard let mysql = mysql else { return }
         killQueryOnServer(threadId: mysql_thread_id(mysql))
@@ -420,14 +418,6 @@
         mysql_close(killConn)
     }
 
-    private func consumeCancellation() -> Bool {
-        stateLock.lock()
-        defer { stateLock.unlock() }
-        guard _isCancelled else { return false }
-        _isCancelled = false
-        return true
-    }
-
     private static func isExpectedInterruption(errno: UInt32, wasTruncated: Bool) -> Bool {
         wasTruncated && errno == UInt32(ER_QUERY_INTERRUPTED)
     }
@@ -462,6 +452,9 @@
             throw MariaDBPluginError.notConnected
         }
 
+        let generation = cancellationGate.beginQuery()
+        defer { cancellationGate.endQuery(generation) }
+
         let queryStatus = query.withCString { queryPtr in
             mysql_real_query(mysql, queryPtr, UInt(query.utf8.count))
         }
@@ -528,7 +521,7 @@
         var truncated = false
 
         while let rowPtr = mysql_fetch_row(resultPtr) {
-            if consumeCancellation() {
+            if cancellationGate.isCancelled(generation) {
                 while mysql_fetch_row(resultPtr) != nil {}
                 mysql_free_result(resultPtr)
                 throw CancellationError()
@@ -573,7 +566,7 @@
             while mysql_fetch_row(resultPtr) != nil {}
         }
 
-        if consumeCancellation() {
+        if cancellationGate.isCancelled(generation) {
             mysql_free_result(resultPtr)
             throw CancellationError()
         }
@@ -677,7 +670,8 @@
         columnTypes: [UInt32],
         columnTypeNames: [String],
         columnIsBinary: [Bool],
-        rowCap: Int? = nil
+        rowCap: Int? = nil,
+        generation: Int
     ) throws -> (rows: [[PluginCellValue]], isTruncated: Bool) {
         let numFields = columns.count
         var resultBinds: [MYSQL_BIND] = Array(repeating: MYSQL_BIND(), count: numFields)
@@ -722,7 +716,7 @@
                 throw getStmtError(stmt)
             }
 
-            if consumeCancellation() {
+            if cancellationGate.isCancelled(generation) {
                 throw CancellationError()
             }
 
@@ -788,6 +782,9 @@
             throw MariaDBPluginError.notConnected
         }
 
+        let generation = cancellationGate.beginQuery()
+        defer { cancellationGate.endQuery(generation) }
+
         guard let stmt = mysql_stmt_init(mysql) else {
             throw MariaDBPluginError(code: 0, message: "Failed to initialize prepared statement", sqlState: nil)
         }
@@ -878,7 +875,7 @@
         let fetchResult = try fetchResultSet(
             from: stmt, metadata: metadata,
             columns: columns, columnTypes: columnTypes, columnTypeNames: columnTypeNames,
-            columnIsBinary: columnIsBinary, rowCap: rowCap
+            columnIsBinary: columnIsBinary, rowCap: rowCap, generation: generation
         )
 
         return MariaDBPluginQueryResult(

Plugins/PostgreSQLDriverPlugin/LibPQPluginConnection.swift

@@ -102,11 +102,11 @@
     private let suppressServerSideCancel: Bool
 
     private let stateLock = NSLock()
+    private let cancellationGate = PluginQueryCancellationGate()
     private var _isConnected: Bool = false
     private var _isShuttingDown: Bool = false
     private var _cachedServerVersion: String?
     private var _cachedServerVersionNumber: Int32 = 0
-    private var _isCancelled: Bool = false
     private var _isConnectCancelled: Bool = false
     private var _postgisOidMap: [UInt32: String] = [:]
 
@@ -163,9 +163,9 @@
     // MARK: - Connection Management
 
     func connect() async throws {
         stateLock.lock()
         _isConnectCancelled = false
         stateLock.unlock()
 
         try await withTaskCancellationHandler {
             try await pluginDispatchAsyncCancellable(
@@ -356,8 +356,9 @@
     // MARK: - Query Cancellation
 
     func cancelCurrentQuery() {
+        guard cancellationGate.cancel() != nil else { return }
+
         stateLock.lock()
-        _isCancelled = true
         let currentConn = conn
         stateLock.unlock()
 
@@ -416,6 +417,9 @@
             throw LibPQPluginError.notConnected
         }
 
+        let generation = cancellationGate.beginQuery()
+        defer { cancellationGate.endQuery(generation) }
+
         let localQuery = String(query)
         let result: OpaquePointer? = localQuery.withCString { queryPtr in
             PQexec(conn, queryPtr)
@@ -444,11 +448,12 @@
 
         case PGRES_TUPLES_OK:
             defer { PQclear(result) }
-            return try fetchResults(from: result)
+            return try fetchResults(from: result, generation: generation)
 
         default:
             let error = getResultError(from: result)
             PQclear(result)
+            if cancellationGate.isCancelled(generation) { throw CancellationError() }
             throw error
         }
     }
@@ -462,6 +467,9 @@
             throw LibPQPluginError.notConnected
         }
 
+        let generation = cancellationGate.beginQuery()
+        defer { cancellationGate.endQuery(generation) }
+
         var paramValues: [UnsafePointer<CChar>?] = []
         var paramLengths: [Int32] = []
         var paramFormats: [Int32] = []
@@ -547,11 +555,12 @@
 
         case PGRES_TUPLES_OK:
             defer { PQclear(result) }
-            return try fetchResults(from: result)
+            return try fetchResults(from: result, generation: generation)
 
         default:
             let error = getResultError(from: result)
             PQclear(result)
+            if cancellationGate.isCancelled(generation) { throw CancellationError() }
             throw error
         }
     }
@@ -609,6 +618,9 @@
                     return
                 }
 
+                let generation = cancellationGate.beginQuery()
+                defer { cancellationGate.endQuery(generation) }
+
                 while let res = PQgetResult(conn) { PQclear(res) }
 
                 let sendOk = queryToRun.withCString { queryPtr in
@@ -675,31 +687,12 @@
                         row.reserveCapacity(numFields)
 
                         for colIndex in 0..<numFields {
-                            if PQgetisnull(result, 0, Int32(colIndex)) == 1 {
-                                row.append(.null)
-                            } else if let valuePtr = PQgetvalue(result, 0, Int32(colIndex)) {
-                                let length = Int(PQgetlength(result, 0, Int32(colIndex)))
-                                let bufferPtr = UnsafeRawBufferPointer(start: valuePtr, count: length)
-                                let oid = columnOids[colIndex]
-
-                                if oid == 17 {
-                                    let text = String(bytes: bufferPtr, encoding: .utf8) ?? ""
-                                    if let data = LibPQByteaDecoder.decode(text) {
-                                        row.append(.bytes(data))
-                                    } else {
-                                        row.append(.text(text))
-                                    }
-                                } else if oid == 16 {
-                                    let str = String(bytes: bufferPtr, encoding: .utf8) ?? ""
-                                    row.append(.text(str == "t" ? "true" : "false"))
-                                } else if let str = String(bytes: bufferPtr, encoding: .utf8) {
-                                    row.append(.text(str))
-                                } else {
-                                    row.append(.text(String(bytes: bufferPtr, encoding: .isoLatin1) ?? ""))
-                                }
-                            } else {
-                                row.append(.null)
-                            }
+                            row.append(Self.decodeCell(
+                                from: result,
+                                row: 0,
+                                column: Int32(colIndex),
+                                oid: columnOids[colIndex]
+                            ))
                         }
 
                         PQclear(result)
@@ -733,6 +726,10 @@
                         streamState.lock.lock()
                         streamState.drained = true
                         streamState.lock.unlock()
+                        if cancellationGate.isCancelled(generation) {
+                            continuation.finish(throwing: CancellationError())
+                            return
+                        }
                         continuation.finish(throwing: error)
                         return
                     }
@@ -752,13 +749,14 @@
 
     // MARK: - Result Parsing
 
-    private func fetchResults(from result: OpaquePointer) throws -> LibPQPluginQueryResult {
+    private func fetchResults(from result: OpaquePointer, generation: Int) throws -> LibPQPluginQueryResult {
         let metadata = readColumnMetadata(from: result)
         let parsed = try parseRows(
             from: result,
             columns: metadata.columns,
             columnOids: metadata.columnOids,
-            columnTypeNames: metadata.columnTypeNames
+            columnTypeNames: metadata.columnTypeNames,
+            generation: generation
         )
 
         let oidMap = postgisOidMap
@@ -882,11 +880,41 @@
         return converted
     }
 
+    private static func decodeCell(
+        from result: OpaquePointer,
+        row: Int32,
+        column: Int32,
+        oid: UInt32
+    ) -> PluginCellValue {
+        guard PQgetisnull(result, row, column) != 1,
+              let valuePtr = PQgetvalue(result, row, column) else {
+            return .null
+        }
+
+        let length = Int(PQgetlength(result, row, column))
+        let bufferPtr = UnsafeRawBufferPointer(start: valuePtr, count: length)
+
+        if oid == 17 {
+            let text = String(bytes: bufferPtr, encoding: .utf8) ?? ""
+            guard let data = LibPQByteaDecoder.decode(text) else { return .text(text) }
+            return .bytes(data)
+        }
+
+        if oid == 16 {
+            let str = String(bytes: bufferPtr, encoding: .utf8) ?? ""
+            return .text(str == "t" ? "true" : "false")
+        }
+
+        if let str = String(bytes: bufferPtr, encoding: .utf8) { return .text(str) }
+        return .text(String(bytes: bufferPtr, encoding: .isoLatin1) ?? "")
+    }
+
     private func parseRows(
         from result: OpaquePointer,
         columns: [String],
         columnOids: [UInt32],
-        columnTypeNames: [String]
+        columnTypeNames: [String],
+        generation: Int
     ) throws -> LibPQPluginQueryResult {
         let numFields = columns.count
         let numRows = Int(PQntuples(result))
@@ -899,43 +927,20 @@
         rows.reserveCapacity(effectiveRowCount)
 
         for rowIndex in 0..<effectiveRowCount {
-            stateLock.lock()
-            let shouldCancel = _isCancelled
-            if shouldCancel { _isCancelled = false }
-            stateLock.unlock()
-            if shouldCancel {
-                throw LibPQPluginError(message: "Query cancelled", sqlState: nil, detail: nil)
+            if cancellationGate.isCancelled(generation) {
+                throw CancellationError()
             }
 
             var row: [PluginCellValue] = []
             row.reserveCapacity(numFields)
 
             for colIndex in 0..<numFields {
-                if PQgetisnull(result, Int32(rowIndex), Int32(colIndex)) == 1 {
-                    row.append(.null)
-                } else if let valuePtr = PQgetvalue(result, Int32(rowIndex), Int32(colIndex)) {
-                    let length = Int(PQgetlength(result, Int32(rowIndex), Int32(colIndex)))
-                    let bufferPtr = UnsafeRawBufferPointer(start: valuePtr, count: length)
-                    let oid = columnOids[colIndex]
-
-                    if oid == 17 {
-                        let text = String(bytes: bufferPtr, encoding: .utf8) ?? ""
-                        if let data = LibPQByteaDecoder.decode(text) {
-                            row.append(.bytes(data))
-                        } else {
-                            row.append(.text(text))
-                        }
-                    } else if oid == 16 {
-                        let str = String(bytes: bufferPtr, encoding: .utf8) ?? ""
-                        row.append(.text(str == "t" ? "true" : "false"))
-                    } else if let str = String(bytes: bufferPtr, encoding: .utf8) {
-                        row.append(.text(str))
-                    } else {
-                        row.append(.text(String(bytes: bufferPtr, encoding: .isoLatin1) ?? ""))
-                    }
-                } else {
-                    row.append(.null)
-                }
+                row.append(Self.decodeCell(
+                    from: result,
+                    row: Int32(rowIndex),
+                    column: Int32(colIndex),
+                    oid: columnOids[colIndex]
+                ))
             }
             rows.append(row)
         }