Skip to content
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
33 changes: 33 additions & 0 deletions .github/workflows/ci.yml
Original file line number Diff line number Diff line change
Expand Up @@ -15,9 +15,42 @@ jobs:
check:
name: impl-swift checks
runs-on: macos-15
timeout-minutes: 25
env:
DEVELOPER_DIR: /Applications/Xcode_26.2.app/Contents/Developer
steps:
- uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
- name: 检查格式、lint 与 SwiftPM 测试
run: bash scripts/check.sh

- name: 读取权威数据库实现基线
id: database-source
run: echo "revision=$(cat scripts/database-source-revision)" >> "$GITHUB_OUTPUT"
- name: 获取权威迁移实现
uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
with:
repository: InKCre/impl-py
ref: ${{ steps.database-source.outputs.revision }}
path: .verification/impl-py
- name: 准备 Python 与 PDM
uses: pdm-project/setup-pdm@973541a5febeafcfdadf8a51211435be6ecfd90f # v4.5
with:
python-version: '3.12'
version: 2.28.0
update-python: false
- name: 安装权威迁移依赖
working-directory: .verification/impl-py
run: |
pdm use -f 3.12
pdm install --prod --frozen-lockfile
- name: 准备原生 PostgreSQL 工具
run: |
# Hosted runners are disposable; keep any preinstalled libpq links from hiding server assets.
brew unlink libpq || true
brew install postgresql@17 pgvector
brew link --overwrite --force postgresql@17
- name: 一次性数据库与原生主流程验收
timeout-minutes: 12
run: >-
python3 scripts/check-database.py --impl-py .verification/impl-py
--postgres-bin "$(brew --prefix postgresql@17)/bin"
3 changes: 3 additions & 0 deletions .gitignore
Original file line number Diff line number Diff line change
Expand Up @@ -7,3 +7,6 @@ xcuserdata/
.local-development-signing-identity
**/Package.resolved
!/Package.resolved

# Disposable authority checkout used by database acceptance.
.verification/
1 change: 1 addition & 0 deletions Application.json
Original file line number Diff line number Diff line change
@@ -0,0 +1 @@
{"version":"0.1.0","build":19}
8 changes: 5 additions & 3 deletions Packages/PeerCore/Sources/PeerCore/BundledParticipation.swift
Original file line number Diff line number Diff line change
Expand Up @@ -36,15 +36,17 @@ extension PeerDatabase {
}

public static func requireSelection(
_ descriptor: NativeExtensionDescriptor, peerID: UUID, on connection: PostgresConnection
_ descriptor: NativeExtensionDescriptor, peerID: UUID, on connection: PeerConnection
) async throws {
let rows = try await connection.query(
"""
SELECT distribution,version,enabled FROM inkcre.extension_peer_participations
WHERE name=\(descriptor.name) AND peer_id=\(peerID) FOR SHARE
""", logger: Self.logger)
for try await (distribution, version, enabled) in rows.decode((String?, String, Bool).self)
{
for row in rows {
try Task.checkCancellation()
let (distribution, version, enabled) = try row.decode((String?, String, Bool).self)

guard distribution == descriptor.distribution, version == descriptor.version else {
throw PeerError.needsAction("此 Peer 选择的扩展版本不在当前应用中,请在扩展设置中对该信息库启用本机版本。")
}
Expand Down
26 changes: 19 additions & 7 deletions Packages/PeerCore/Sources/PeerCore/InfoBase.swift
Original file line number Diff line number Diff line change
Expand Up @@ -27,29 +27,37 @@ public struct GraphRelation: Sendable, Encodable {
/// Uses the caller's transaction; storage pointers and byte mechanics belong to the Host.
public enum InfoBase {
public static func createBlock(
resolver: String, content: String, on connection: PostgresConnection
resolver: String, content: String, on connection: PeerConnection
) async throws -> Int {
let rows = try await connection.query(
"INSERT INTO inkcre.blocks(resolver,content) VALUES (\(resolver),\(content)) RETURNING id",
logger: PeerDatabase.logger)
for try await id in rows.decode(Int.self) { return id }
for row in rows {
try Task.checkCancellation()
let id = try row.decode(Int.self)
return id
}
throw PeerError.connection("信息库未返回资料标识。")
}

public static func updateBlock(
_ id: Int, content: String, resolver: String? = nil, on connection: PostgresConnection
_ id: Int, content: String, resolver: String? = nil, on connection: PeerConnection
)
async throws
{
let rows = try await connection.query(
"UPDATE inkcre.blocks SET content=\(content),resolver=COALESCE(\(resolver),resolver) WHERE id=\(id) RETURNING id",
logger: PeerDatabase.logger)
for try await _ in rows.decode(Int.self) { return }
for row in rows {
try Task.checkCancellation()
let _ = try row.decode(Int.self)
return
}
throw PeerError.needsAction("资料已被移除。")
}

public static func relate(
_ from: Int, to: Int, content: String, on connection: PostgresConnection
_ from: Int, to: Int, content: String, on connection: PeerConnection
) async throws {
try await connection.query(
"INSERT INTO inkcre.relations(from_,to_,content) VALUES (\(from),\(to),\(content))",
Expand All @@ -58,7 +66,7 @@ public enum InfoBase {

public static func persist(
blocks: [GraphBlock], relations: [GraphRelation],
blobs: [Int: Data], on connection: PostgresConnection
blobs: [Int: Data], on connection: PeerConnection
) async throws -> [Int: Int] {
var ids: [Int: Int] = [:]
for block in blocks {
Expand All @@ -76,7 +84,11 @@ public enum InfoBase {
"""
INSERT INTO inkcre.blocks(resolver,content,storage) VALUES (\(block.resolver),\(content),\(storage)) RETURNING id
""", logger: PeerDatabase.logger)
for try await id in rows.decode(Int.self) { ids[block.id] = id }
for row in rows {
try Task.checkCancellation()
let id = try row.decode(Int.self)
ids[block.id] = id
}
}
for relation in relations {
guard let from = ids[relation.from_], let to = ids[relation.to_] else {
Expand Down
21 changes: 21 additions & 0 deletions Packages/PeerCore/Sources/PeerCore/NativeExtension.swift
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,27 @@ public struct NativeExtensionDescriptor: Codable, Sendable, Equatable {
self.distribution = distribution
self.version = version
}

/// SwiftPM and the packaged app read the same shipped manifest.
public static func bundled(
resourceBundle: String, developmentBundle: @autoclosure () -> Bundle
) -> Self {
let bundle: Bundle
if Bundle.main.bundleURL.pathExtension == "app" {
guard let url = Bundle.main.url(forResource: resourceBundle, withExtension: "bundle"),
let packaged = Bundle(url: url)
else { preconditionFailure("Missing packaged extension: \(resourceBundle)") }
bundle = packaged
} else {
bundle = developmentBundle()
}
guard let url = bundle.url(forResource: "extension", withExtension: "json"),
let data = try? Data(contentsOf: url),
let descriptor = try? JSONDecoder().decode(Self.self, from: data)
else { preconditionFailure("Invalid extension manifest: \(resourceBundle)") }
return descriptor
}

}

@MainActor
Expand Down
69 changes: 69 additions & 0 deletions Packages/PeerCore/Sources/PeerCore/PeerConnection.swift
Original file line number Diff line number Diff line change
@@ -0,0 +1,69 @@
import Logging
import PostgresNIO

/// A connection borrowed inside PeerDatabase.withConnection. Do not retain it beyond that scope.
public struct PeerConnection: Sendable {
let connection: PostgresConnection

/// Collect a query before returning. Current callers materialize their results; this is not a
/// streaming API. Check and write share the event loop so close cannot strand a driver promise.
@discardableResult
public func query(_ query: PostgresQuery, logger: Logger = PeerDatabase.logger) async throws
-> [PostgresRow]
{
try Task.checkCancellation()
let result: PostgresQueryResult = try await connection.eventLoop.flatSubmit {
guard !connection.isClosed else {
return connection.eventLoop.makeFailedFuture(
ClosedConnection())
}
return connection.query(query, logger: logger)
}.get()
try Task.checkCancellation()
return result.rows
}

func close() async {
connection.close().whenFailure { _ in }
// close() may return an already-succeeded future as soon as isActive becomes false.
// closeFuture waits for handler cleanup; the pool observes the same future.
try? await connection.closeFuture.get()
}

private struct ClosedConnection: Error {}

public func withTransaction<Result: Sendable>(
logger: Logger = PeerDatabase.logger,
_ operation: @Sendable (PeerConnection) async throws -> Result
) async throws -> Result {
try await query("BEGIN", logger: logger)
let result: Result
do {
result = try await operation(self)
} catch {
// Cancellation closes the socket in the enclosing lease and PostgreSQL rolls back.
// Never issue another command on that connection after cancellation.
if Task.isCancelled { throw error }
do {
try await query("ROLLBACK", logger: logger)
} catch let rollbackError {
await close()
throw PeerTransactionError(operationError: error, rollbackError: rollbackError)
}
throw error
}
do {
try await query("COMMIT", logger: logger)
} catch {
// A failed/uncertain commit must not return an open transaction to the pool.
await close()
throw error
}
return result
}
}

public struct PeerTransactionError: Error {
public let operationError: any Error
public let rollbackError: any Error
}
37 changes: 26 additions & 11 deletions Packages/PeerCore/Sources/PeerCore/PeerDatabase.swift
Original file line number Diff line number Diff line change
Expand Up @@ -14,7 +14,14 @@ public actor PeerDatabase {
}
private var pools: [UUID: Pool] = [:]

public init() {}
private let applicationVersion: String
public init(
applicationVersion: String = Bundle.main.object(
forInfoDictionaryKey: "CFBundleShortVersionString")
as? String ?? "development"
) {
self.applicationVersion = applicationVersion
}

public func client(for target: ConnectionTarget, password: String) throws -> PostgresClient {
try target.validate()
Expand All @@ -35,9 +42,9 @@ public actor PeerDatabase {
configuration.options.minimumConnections = 0
configuration.options.maximumConnections = 2
configuration.options.connectTimeout = .seconds(10)
// Bound server work too when a disconnected client cannot deliver cancellation.
// Detect a closed client during long queries; bound server work if a lost network hides it.
configuration.options.additionalStartupParameters = [
("options", "-c statement_timeout=30000")
("options", "-c statement_timeout=30000 -c client_connection_check_interval=1000")
]
let client = PostgresClient(configuration: configuration, backgroundLogger: Self.logger)
let task = Task { await client.run() }
Expand All @@ -59,12 +66,13 @@ public actor PeerDatabase {
let client = try client(for: target, password: password)
if pools[target.id]?.registered == true { return }
let name = Host.current().localizedName ?? "InKCre Mac"
let applicationVersion = self.applicationVersion
try await withConnection(for: target, password: password) { connection in
try await connection.withTransaction(logger: Self.logger) { connection in
try await connection.query(
"""
INSERT INTO inkcre.peers (id,name,application_version,capabilities,config_schema,config,labels)
VALUES (\(target.peerID),\(name),'0.1.0','[]','{}','{}',ARRAY['macOS'])
VALUES (\(target.peerID),\(name),\(applicationVersion),'[]','{}','{}',ARRAY['macOS'])
ON CONFLICT (id) DO UPDATE SET application_version=EXCLUDED.application_version
""", logger: Self.logger)
try await connection.query(
Expand All @@ -77,16 +85,24 @@ public actor PeerDatabase {
/// Keep row consumption and transactions inside this scope so cancellation closes their lease.
public nonisolated func withConnection<Result: Sendable>(
for target: ConnectionTarget, password: String,
_ operation: @escaping @Sendable (PostgresConnection) async throws -> Result
_ operation: @escaping @Sendable (PeerConnection) async throws -> Result
) async throws -> Result {
let client = try await client(for: target, password: password)
return try await withThrowingTaskGroup(of: Result.self) { group in
group.addTask {
try await client.withConnection { connection in
try await withTaskCancellationHandler {
try Task.checkCancellation()
do { return try await operation(connection) } catch {
if Task.isCancelled { throw CancellationError() }
let borrowed = PeerConnection(connection: connection)
return try await withTaskCancellationHandler {
do {
try Task.checkCancellation()
let result = try await operation(borrowed)
try Task.checkCancellation()
return result
} catch {
if Task.isCancelled {
await borrowed.close()
throw CancellationError()
}
throw error
}
} onCancel: {
Expand All @@ -111,10 +127,9 @@ public actor PeerDatabase {
do {
try await self.withConnection(for: pool.target, password: pool.password) {
connection in
let rows = try await connection.query(
try await connection.query(
"SELECT inkcre.renew_peer_lease(\(pool.target.peerID),120)",
logger: Self.logger)
for try await _ in rows {}
}
return (id, nil)
} catch { return (id, Self.failure(error).localizedDescription) }
Expand Down
Loading
Loading