angd-dev/data-raft
1
0
Fork 0
Code Pull Requests Releases Activity

Merge branch 'feature/lazy-connection-setup' into develop

Browse Source
This commit is contained in:
Oleksii Zghurskyi
2025-08-21 15:03:02 +03:00
parent f4fefb3391 d26587cfc3
commit 5566f6f6ba
8 changed files with 129 additions and 173 deletions
Download Patch File Download Diff File Expand all files Collapse all files

214
Sources/DataRaft/Classes/DatabaseService.swift
View File

@@ -1,16 +1,20 @@
import Foundation
import DataLiteCore
import DataLiteC
import DataLiteCore
/// Base service for working with a database.
///
/// `DatabaseService` provides a unified interface for performing operations
/// using a database connection, with built-in support for transactions,
/// reconnection, and optional encryption key management.
/// `DatabaseService` provides a unified interface for performing operations using a database
/// connection, with built-in support for transactions, reconnection, and optional encryption
/// key management.
///
/// The service ensures thread-safe execution by serializing access to the
/// connection through an internal queue. This enables building modular and safe
/// data access layers without duplicating low-level logic.
/// The service ensures thread-safe execution by serializing access to the connection through
/// an internal queue. This enables building modular and safe data access layers without
/// duplicating low-level logic.
///
/// The connection is established lazily on first use (e.g., within `perform`), not during
/// initialization. If a key provider is set, the key is applied as part of establishing or
/// restoring the connection.
///
/// Below is an example of creating a service for managing notes:
///
@@ -18,9 +22,7 @@ import DataLiteC
/// final class NoteService: DatabaseService {
/// func insertNote(_ text: String) throws {
/// try perform { connection in
/// let stmt = try connection.prepare(
/// sql: "INSERT INTO notes (text) VALUES (?)"
/// )
/// let stmt = try connection.prepare(sql: "INSERT INTO notes (text) VALUES (?)")
/// try stmt.bind(text, at: 0)
/// try stmt.step()
/// }
@@ -50,43 +52,39 @@ import DataLiteC
///
/// ## Error Handling
///
/// All operations are executed on an internal serial queue, ensuring thread safety.
/// If an encryption error (`SQLITE_NOTADB`) is detected, the service may reopen the
/// connection and retry the transactional block exactly once. If the error occurs again,
/// it is propagated without further retries.
/// All operations are executed on an internal serial queue, ensuring thread safety. If a
/// decryption error (`SQLITE_NOTADB`) is detected, the service may reopen the connection and
/// retry the transactional block exactly once. If the error occurs again, it is propagated
/// without further retries.
///
/// ## Encryption Key Management
///
/// If a ``keyProvider`` is set, the service uses it to obtain and apply an encryption
/// key when creating or restoring a connection. If an error occurs while obtaining
/// or applying the key, the provider is notified through
/// If a ``keyProvider`` is set, the service uses it to obtain and apply an encryption key when
/// establishing or restoring the connection. If an error occurs while obtaining or applying the
/// key, the provider is notified through
/// ``DatabaseServiceKeyProvider/databaseService(_:didReceive:)``.
///
/// ## Reconnection
///
/// Automatic reconnection is available only during transactional blocks executed with
/// ``perform(in:closure:)``. If a decryption error (`SQLITE_NOTADB`) occurs during
/// a transaction and the provider allows reconnection, the service obtains a new key,
/// creates a new connection, and retries the block once. If the second attempt fails
/// or reconnection is disallowed, the error is propagated without further retries.
/// ``perform(in:closure:)``. If a decryption error (`SQLITE_NOTADB`) occurs during a
/// transaction and the provider allows reconnection, the service obtains a new key, creates a
/// new connection, and retries the block once. If the second attempt fails or reconnection is
/// disallowed, the error is propagated without further retries.
///
/// ## Topics
///
/// ### Initializers
///
/// - ``init(provider:keyProvider:queue:)``
/// - ``init(connection:keyProvider:queue:)``
/// - ``ConnectionProvider``
/// - ``ConnectionConfig``
/// - ``init(provider:config:keyProvider:queue:)``
/// - ``init(connection:config:keyProvider:queue:)``
///
/// ### Key Management
///
/// - ``DatabaseServiceKeyProvider``
/// - ``keyProvider``
/// - ``applyKeyProvider()``
///
/// ### Connection Management
///
/// - ``ConnectionProvider``
/// - ``reconnect()``
///
/// ### Database Operations
///
@@ -94,6 +92,8 @@ import DataLiteC
/// - ``perform(_:)``
/// - ``perform(in:closure:)``
open class DatabaseService: DatabaseServiceProtocol, @unchecked Sendable {
// MARK: - Types
/// A closure that creates a new database connection.
///
/// `ConnectionProvider` is used for deferred connection creation.
@@ -104,146 +104,140 @@ open class DatabaseService: DatabaseServiceProtocol, @unchecked Sendable {
/// - Throws: An error if the connection cannot be created or configured.
public typealias ConnectionProvider = () throws -> Connection
/// A closure used to configure a newly created connection.
///
/// Called after the connection is established (and after key application if present).
/// Can be used to set PRAGMA options or perform other initialization logic.
///
/// - Parameter connection: The newly created connection.
/// - Throws: Any error if configuration fails.
public typealias ConnectionConfig = (Connection) throws -> Void
// MARK: - Properties
private let provider: ConnectionProvider
private let config: ConnectionConfig?
private let queue: DispatchQueue
private let queueKey = DispatchSpecificKey<Void>()
private var connection: Connection
private var cachedConnection: Connection?
private var connection: Connection {
get throws {
guard let cachedConnection else {
let connection = try connect()
cachedConnection = connection
return connection
}
return cachedConnection
}
}
/// Encryption key provider.
///
/// Used to obtain and apply a key when creating or restoring a connection.
/// Used to obtain and apply a key when establishing or restoring a connection. The key is
/// requested on first access to the connection and on reconnection if needed.
public weak var keyProvider: DatabaseServiceKeyProvider?
// MARK: - Inits
/// Creates a new database service.
///
/// Calls `provider` to create the initial connection and configures
/// the internal serial queue for thread-safe access to the database.
/// Configures the internal serial queue for thread-safe access to the database.
/// The connection is **not** created during initialization. It is established
/// lazily on first use (for example, inside `perform`).
///
/// The internal queue is always created with QoS `.utility`. If the `queue`
/// parameter is provided, it is used as the target queue for the internal one.
///
/// If a `keyProvider` is set, the encryption key is applied immediately
/// after the initial connection is created.
/// If a `keyProvider` is set, the encryption key will be applied when the
/// connection is established or restored.
///
/// - Parameters:
/// - provider: A closure that returns a new connection.
/// - config: An optional configuration closure called after the connection
/// is created (and after key application if present).
/// - keyProvider: An optional encryption key provider.
/// - queue: An optional target queue for the internal one.
/// - Throws: An error if the connection cannot be created or configured.
public init(
provider: @escaping ConnectionProvider,
config: ConnectionConfig? = nil,
keyProvider: DatabaseServiceKeyProvider? = nil,
queue: DispatchQueue? = nil
) throws {
) {
self.provider = provider
self.config = config
self.keyProvider = keyProvider
self.connection = try provider()
self.queue = .init(for: Self.self, qos: .utility)
self.queue.setSpecific(key: queueKey, value: ())
if let queue = queue {
self.queue.setTarget(queue: queue)
}
if self.keyProvider != nil {
try applyKey(to: self.connection)
}
}
/// Creates a new database service.
///
/// The connection is created lazily on first use. If a `keyProvider` is set,
/// the key will be applied when the connection is established.
///
/// - Parameters:
/// - provider: An expression that creates a new connection.
/// - config: An optional configuration closure called after the connection
/// is created (and after key application if present).
/// - keyProvider: An optional encryption key provider.
/// - queue: An optional target queue for the internal one.
/// - Throws: An error if the connection cannot be created or configured.
public convenience init(
connection provider: @escaping @autoclosure ConnectionProvider,
config: ConnectionConfig? = nil,
keyProvider: DatabaseServiceKeyProvider? = nil,
queue: DispatchQueue? = nil
) throws {
try self.init(provider: provider, keyProvider: keyProvider, queue: queue)
) {
self.init(
provider: provider,
config: config,
keyProvider: keyProvider,
queue: queue
)
}
// MARK: - Methods
/// Applies the encryption key from `keyProvider` to the current connection.
/// Executes a closure with the current connection.
///
/// The method executes synchronously on the internal queue. If the key provider
/// is missing, the method does nothing. If the key has already been successfully
/// applied, subsequent calls have no effect. To apply a new key, use ``reconnect()``.
///
/// If an error occurs while obtaining or applying the key, it is thrown further
/// and also reported to the provider via
/// ``DatabaseServiceKeyProvider/databaseService(_:didReceive:)``.
///
/// - Throws: An error while obtaining or applying the key.
final public func applyKeyProvider() throws {
try withConnection { connection in
try applyKey(to: connection)
}
}
/// Establishes a new database connection.
///
/// Creates a new `Connection` using the stored connection provider and,
/// if a ``keyProvider`` is set, applies the encryption key. The new connection
/// replaces the previous one only if it is successfully created and configured.
///
/// If an error occurs while obtaining or applying the key, it is thrown further
/// and also reported to the provider via
/// ``DatabaseServiceKeyProvider/databaseService(_:didReceive:)``.
///
/// Executed synchronously on the internal queue, ensuring thread safety.
///
/// - Throws: An error if the connection cannot be created or the key cannot
/// be obtained/applied.
final public func reconnect() throws {
try withConnection { _ in
let connection = try provider()
try applyKey(to: connection)
self.connection = connection
}
}
/// Executes a closure with the active connection.
///
/// Runs the `closure` on the internal serial queue, ensuring
/// thread-safe access to the `Connection`.
/// Ensures thread-safe access by running the closure on the internal serial queue.
/// The connection is created lazily if needed.
///
/// - Parameter closure: A closure that takes the active connection.
/// - Returns: The value returned by the closure.
/// - Throws: Any error thrown by the closure.
final public func perform<T>(_ closure: Perform<T>) rethrows -> T {
/// - Throws: An error if the connection cannot be created or if the closure throws.
final public func perform<T>(_ closure: Perform<T>) throws -> T {
try withConnection(closure)
}
/// Executes a closure inside a transaction if the connection is in autocommit mode.
///
/// If the connection is in autocommit mode, starts a new transaction of the
/// specified type, executes the closure, and commits changes on success.
/// If the closure throws an error, the transaction is rolled back.
/// If the connection is in autocommit mode, starts a new transaction of the specified
/// type, executes the closure, and commits changes on success. If the closure throws
/// an error, the transaction is rolled back.
///
/// If the closure throws `Connection.Error` with code `SQLITE_NOTADB`
/// and reconnection is allowed, the service attempts to reconnect and retries
/// the transaction block once.
/// If the closure throws `Connection.Error` with code `SQLITE_NOTADB` and reconnection
/// is allowed, the service attempts to create a new connection, reapply the key, and
/// retries the transaction block once. If the second attempt fails or reconnection
/// is disallowed, the error is propagated without further retries.
///
/// If a transaction is already active (connection not in autocommit mode),
/// the closure is executed directly without starting a new transaction.
/// If a transaction is already active (connection not in autocommit mode), the closure
/// is executed directly without starting a new transaction.
///
/// - Parameters:
/// - transaction: The type of transaction to start.
/// - closure: A closure that takes the active connection and returns a result.
/// - Returns: The value returned by the closure.
/// - Throws: Any error thrown by the closure, transaction management, or
/// reconnection logic.
/// - Throws: Errors from connection creation, key application, configuration,
/// transaction management, or from the closure itself.
/// - Important: The closure may be executed more than once. Ensure it is idempotent.
final public func perform<T>(
in transaction: TransactionType,
closure: Perform<T>
) rethrows -> T {
) throws -> T {
try withConnection { connection in
if connection.isAutocommit {
do {
@@ -286,21 +280,31 @@ private extension DatabaseService {
keyProvider?.databaseService(shouldReconnect: self) ?? false
}
func withConnection<T>(_ closure: Perform<T>) rethrows -> T {
func withConnection<T>(_ closure: Perform<T>) throws -> T {
switch DispatchQueue.getSpecific(key: queueKey) {
case .none: try queue.asyncAndWait { try closure(connection) }
case .some: try closure(connection)
}
}
func reconnect() throws {
cachedConnection = try connect()
}
func connect() throws -> Connection {
let connection = try provider()
try applyKey(to: connection)
try config?(connection)
return connection
}
func applyKey(to connection: Connection) throws {
guard let keyProvider = keyProvider else { return }
do {
if let key = try keyProvider.databaseService(keyFor: self) {
let sql = "SELECT count(*) FROM sqlite_master"
try connection.apply(key)
try connection.execute(raw: sql)
}
let key = try keyProvider.databaseService(keyFor: self)
let sql = "SELECT count(*) FROM sqlite_master"
try connection.apply(key)
try connection.execute(raw: sql)
} catch {
keyProvider.databaseService(self, didReceive: error)
throw error

10
Sources/DataRaft/Classes/MigrationService.swift
View File

@@ -74,16 +74,6 @@ public final class MigrationService<
pthread_mutex_destroy(&mutex)
}
/// Applies settings to the active database connection.
public func applyKeyProvider() throws {
try service.applyKeyProvider()
}
/// Recreates the database connection.
public func reconnect() throws {
try service.reconnect()
}
/// Registers a new migration, ensuring version and script URL uniqueness.
///
/// - Parameter migration: The migration to register.

8
Sources/DataRaft/Classes/RowDatabaseService.swift
View File

@@ -78,8 +78,8 @@ open class RowDatabaseService:
encoder: RowEncoder = RowEncoder(),
decoder: RowDecoder = RowDecoder(),
queue: DispatchQueue? = nil
) throws {
try self.init(
) {
self.init(
provider: provider,
encoder: encoder,
decoder: decoder,
@@ -104,10 +104,10 @@ open class RowDatabaseService:
encoder: RowEncoder = RowEncoder(),
decoder: RowDecoder = RowDecoder(),
queue: DispatchQueue? = nil
) throws {
) {
self.encoder = encoder
self.decoder = decoder
try super.init(
super.init(
provider: provider,
queue: queue
)

8
Sources/DataRaft/Protocols/DatabaseServiceKeyProvider.swift
View File

@@ -25,13 +25,13 @@ import DataLiteCore
public protocol DatabaseServiceKeyProvider: AnyObject, Sendable {
/// Returns the encryption key for the specified database service.
///
/// May return `nil` if the encryption key is currently unavailable or if the database
/// does not require encryption.
/// This method must either return a valid encryption key or throw an error if
/// the key cannot be retrieved.
///
/// - Parameter service: The service requesting the key.
/// - Returns: The encryption key or `nil`.
/// - Returns: The encryption key.
/// - Throws: An error if the key cannot be retrieved.
func databaseService(keyFor service: DatabaseServiceProtocol) throws -> Connection.Key?
func databaseService(keyFor service: DatabaseServiceProtocol) throws -> Connection.Key
/// Indicates whether the service should attempt to reconnect if applying the key fails.
///

27
Sources/DataRaft/Protocols/DatabaseServiceProtocol.swift
View File

@@ -17,11 +17,6 @@ import DataLiteCore
///
/// - ``DatabaseServiceKeyProvider``
/// - ``keyProvider``
/// - ``applyKeyProvider()``
///
/// ### Connection Management
///
/// - ``reconnect()``
///
/// ### Database Operations
///
@@ -47,24 +42,6 @@ public protocol DatabaseServiceProtocol: AnyObject, Sendable {
/// encountered while applying a key.
var keyProvider: DatabaseServiceKeyProvider? { get set }
/// Applies the encryption key from the current provider.
///
/// Calls the configured ``keyProvider`` to obtain a key and applies
/// it to the active connection. If the key is unavailable or an
/// error occurs while applying it, the method throws.
///
/// - Throws: An error if the key cannot be retrieved or applied.
func applyKeyProvider() throws
/// Reopens the database connection.
///
/// Creates a new connection using the provider and applies the
/// encryption key if ``keyProvider`` is set. Typically used when
/// the previous connection has become invalid.
///
/// - Throws: An error if the new connection cannot be created or the key cannot be applied.
func reconnect() throws
/// Executes the given closure with an active connection.
///
/// The closure receives the connection and may perform any
@@ -73,7 +50,7 @@ public protocol DatabaseServiceProtocol: AnyObject, Sendable {
/// - Parameter closure: The closure that accepts a connection.
/// - Returns: The value returned by the closure.
/// - Throws: An error if one occurs during closure execution.
func perform<T>(_ closure: Perform<T>) rethrows -> T
func perform<T>(_ closure: Perform<T>) throws -> T
/// Executes the given closure within a transaction.
///
@@ -86,5 +63,5 @@ public protocol DatabaseServiceProtocol: AnyObject, Sendable {
/// - closure: The closure that accepts a connection.
/// - Returns: The value returned by the closure.
/// - Throws: An error if one occurs during closure execution.
func perform<T>(in transaction: TransactionType, closure: Perform<T>) rethrows -> T
func perform<T>(in transaction: TransactionType, closure: Perform<T>) throws -> T
}

10
Sources/DataRaft/Protocols/MigrationServiceProtocol.swift
View File

@@ -15,16 +15,6 @@ public protocol MigrationServiceProtocol: AnyObject, Sendable {
/// Encryption key provider for the database service.
var keyProvider: DatabaseServiceKeyProvider? { get set }
/// Applies an encryption key to the current database connection.
///
/// - Throws: Any error that occurs while retrieving or applying the key.
func applyKeyProvider() throws
/// Recreates the database connection and reapplies the encryption key if available.
///
/// - Throws: Any error that occurs while creating the connection or applying the key.
func reconnect() throws
/// Registers a migration to be executed by the service.
///
/// - Parameter migration: The migration to register.

23
Tests/DataRaftTests/Classes/DatabaseServiceTests.swift
View File

@@ -29,7 +29,7 @@ class DatabaseServiceTests: DatabaseServiceKeyProvider, @unchecked Sendable {
.appendingPathComponent(UUID().uuidString)
.appendingPathExtension("sqlite")
let service = try DatabaseService(provider: {
let service = DatabaseService(provider: {
try Connection(
path: fileURL.path,
options: [.create, .readwrite]
@@ -40,7 +40,6 @@ class DatabaseServiceTests: DatabaseServiceKeyProvider, @unchecked Sendable {
self.service = service
self.service.keyProvider = self
try self.service.applyKeyProvider()
try self.service.perform { connection in
try connection.execute(sql: """
CREATE TABLE IF NOT EXISTS Item (
@@ -55,7 +54,7 @@ class DatabaseServiceTests: DatabaseServiceKeyProvider, @unchecked Sendable {
try? FileManager.default.removeItem(at: fileURL)
}
func databaseService(keyFor service: any DatabaseServiceProtocol) throws -> Connection.Key? {
func databaseService(keyFor service: any DatabaseServiceProtocol) throws -> Connection.Key {
currentKey
}
@@ -184,16 +183,12 @@ extension DatabaseServiceTests {
try stmt.step()
}
})
currentKey = keyTwo
try service.reconnect()
try service.perform { connection in
let stmt = try connection.prepare(
sql: "SELECT COUNT(*) FROM Item",
options: []
)
try stmt.step()
#expect(connection.isAutocommit)
#expect(stmt.columnValue(at: 0) == 0)
}
let stmt = try connection.prepare(
sql: "SELECT COUNT(*) FROM Item",
options: []
)
try stmt.step()
#expect(connection.isAutocommit)
#expect(stmt.columnValue(at: 0) == 0)
}
}

2
Tests/DataRaftTests/Classes/MigrationServiceTests.swift
View File

@@ -12,7 +12,7 @@ import DataLiteCore
init() throws {
let connection = try Connection(location: .inMemory, options: .readwrite)
self.connection = connection
self.migrationService = .init(service: try .init(connection: connection), storage: .init())
self.migrationService = .init(service: .init(connection: connection), storage: .init())
}
@Test func addMigration() throws {