From bbc35ab4e39ba811e7ef0604292352a8458575ac Mon Sep 17 00:00:00 2001 From: Mark Gregory Date: Mon, 19 Oct 2026 11:10:42 +0100 Subject: [PATCH] connection updates --- DBSyn/DatabaseModel.swift | 5 +- DBSyn/MySQLProtocol.swift | 109 +++++++++++++++++++++++++++----------- 2 files changed, 82 insertions(+), 32 deletions(-) diff --git a/DBSyn/DatabaseModel.swift b/DBSyn/DatabaseModel.swift index 82a9c19..45195d2 100644 --- a/DBSyn/DatabaseModel.swift +++ b/DBSyn/DatabaseModel.swift @@ -683,6 +683,7 @@ final class DatabaseModel: ObservableObject { saveActiveQuery() if let oldID = selectedID { NativeMySQLRunner.disconnect(profileID: oldID) } result = QueryResult() + isRunning = false selectedID = id UserDefaults.standard.set(id.uuidString, forKey: "DBSyn.selectedConnectionID") selectedDatabase = profile.database @@ -887,7 +888,7 @@ final class DatabaseModel: ObservableObject { introspect(database: usedSchema) } } - guard selectedID == profile.id else { isRunning = false; return } + guard selectedID == profile.id else { return } if listingDatabases { databases = lastResult.rows.compactMap { $0.first } status = "Connected to \(profile.name)" @@ -914,7 +915,7 @@ final class DatabaseModel: ObservableObject { if !listingDatabases { queryDurationText = String(format: "%.3f s", Date().timeIntervalSince(queryStarted)) } - isRunning = false + if selectedID == profile.id { isRunning = false } } } diff --git a/DBSyn/MySQLProtocol.swift b/DBSyn/MySQLProtocol.swift index 1a3ece0..310a809 100644 --- a/DBSyn/MySQLProtocol.swift +++ b/DBSyn/MySQLProtocol.swift @@ -437,40 +437,78 @@ private final class PersistentMySQLSessions { let connection: MySQLWireConnection } - private let queue = DispatchQueue(label: "DBSyn.MySQLSessions") - private var connections: [UUID: Entry] = [:] + // Each profile has its own serial executor. An unreachable server cannot + // block requests to any other saved connection. + private final class SessionSlot { + let queue: DispatchQueue + var entry: Entry? + // Only accessed on queue except generation, which is lock protected. + private let generationLock = NSLock() + private var currentGeneration: UInt64 = 0 + + init(id: UUID) { + queue = DispatchQueue(label: "DBSyn.MySQLSession.\(id.uuidString)") + } + func generation() -> UInt64 { + generationLock.lock(); defer { generationLock.unlock() } + return currentGeneration + } + func invalidate() { + generationLock.lock(); currentGeneration &+= 1; generationLock.unlock() + } + } + + private let slotsLock = NSLock() + private var slots: [UUID: SessionSlot] = [:] + + private func slot(for id: UUID) -> SessionSlot { + slotsLock.lock(); defer { slotsLock.unlock() } + if let existing = slots[id] { return existing } + let created = SessionSlot(id: id) + slots[id] = created + return created + } func execute(profile: ConnectionProfile, database: String, password: String, sql: String) throws -> QueryResult { - try queue.sync { - // A changed connection profile or password requires a new session. - if let previous = connections[profile.id], - (previous.profile != profile || previous.password != password) { - connections.removeValue(forKey: profile.id) + let session = slot(for: profile.id) + let expectedGeneration = session.generation() + return try session.queue.sync { + guard session.generation() == expectedGeneration else { + throw MySQLWireError(message: "Connection request superseded") } - - if connections[profile.id] == nil { + if let previous = session.entry, + (previous.profile != profile || previous.password != password) { + session.entry = nil + DBSynSSHTunnels.shared.stop(profile.id) + } + if session.entry == nil { let endpoint = try DBSynSSHTunnels.shared.endpoint(for: profile) - let connection = try MySQLWireConnection(host: endpoint.0, port: endpoint.1, requireTLS: !profile.useSSHTunnel && profile.useTLS && !["127.0.0.1", "localhost", "::1"].contains(endpoint.0.lowercased())) + let connection = try MySQLWireConnection( + host: endpoint.0, port: endpoint.1, + requireTLS: !profile.useSSHTunnel && profile.useTLS && + !["127.0.0.1", "localhost", "::1"].contains(endpoint.0.lowercased()) + ) connection.markHost(endpoint.0) try connection.authenticate(user: profile.username, password: password, database: database) - connections[profile.id] = Entry(profile: profile, - password: password, - connection: connection) + guard session.generation() == expectedGeneration else { + throw MySQLWireError(message: "Connection request superseded") + } + session.entry = Entry(profile: profile, password: password, + connection: connection) } - - guard let entry = connections[profile.id] else { + guard let entry = session.entry else { throw MySQLWireError(message: "MySQL session was not created") } do { - // Crucially, no new authentication takes place here. USE, - // session variables and transactions survive between queries. - return try entry.connection.query(sql) + let result = try entry.connection.query(sql) + guard session.generation() == expectedGeneration else { + throw MySQLWireError(message: "Connection request superseded") + } + return result } catch { - // The wire protocol can be out of sync after a failed query. - // Discard it rather than silently issuing commands on a bad socket. - connections.removeValue(forKey: profile.id) + session.entry = nil throw error } } @@ -479,23 +517,36 @@ private final class PersistentMySQLSessions { func stream(profile: ConnectionProfile, database: String, password: String, sql: String, onRow: (String) throws -> Void) throws -> Int64 { - try queue.sync { - guard let entry = connections[profile.id], + let session = slot(for: profile.id) + let expectedGeneration = session.generation() + return try session.queue.sync { + guard session.generation() == expectedGeneration, + let entry = session.entry, entry.profile == profile, entry.password == password else { throw MySQLWireError(message: "Export session is no longer connected") } do { - return try entry.connection.streamFirstColumn(sql, onRow: onRow) + return try entry.connection.streamFirstColumn(sql, onRow: { row in + guard session.generation() == expectedGeneration else { + throw MySQLWireError(message: "Export cancelled: connection switched") + } + try onRow(row) + }) } catch { - connections.removeValue(forKey: profile.id) + session.entry = nil throw error } } } func disconnect(profileID: UUID) { - queue.sync { _ = connections.removeValue(forKey: profileID) } - DBSynSSHTunnels.shared.stop(profileID) + let session = slot(for: profileID) + // Invalidate immediately without waiting for a blocked network read. + session.invalidate() + session.queue.async { + session.entry = nil + DBSynSSHTunnels.shared.stop(profileID) + } } } @@ -526,8 +577,6 @@ enum NativeMySQLRunner { // Connection shutdown may wait behind a slow network operation on the // session's serial queue. Never make the SwiftUI main thread wait for it. static func disconnect(profileID: UUID) { - Task.detached(priority: .utility) { - PersistentMySQLSessions.shared.disconnect(profileID: profileID) - } + PersistentMySQLSessions.shared.disconnect(profileID: profileID) } }