From 478b788d0590f52904f119f4bd2cd4bc5dd57518 Mon Sep 17 00:00:00 2001 From: Lan_zhijiang Date: Sat, 31 Oct 2026 14:45:04 +0800 Subject: [PATCH 1/3] =?UTF-8?q?fix(extensions):=20=E5=9B=BA=E5=AE=9A?= =?UTF-8?q?=E9=87=87=E9=9B=86=E5=BE=85=E5=8A=9E=E9=A1=BA=E5=BA=8F=E5=B9=B6?= =?UTF-8?q?=E9=9B=86=E4=B8=AD=E7=9B=AE=E5=BD=95=E7=9B=91=E8=A7=86=E7=94=9F?= =?UTF-8?q?=E5=91=BD=E5=91=A8=E6=9C=9F?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 在冻结事务内分配入队序号，重试与列表排序不再改变派发顺序 - 由每个目录监视器拥有事件合并、扫描、任务收束和目录访问释放 - 补充真实存储与界面流程验证，记录本机格式 v2 的开发数据边界 --- README.md | 2 +- .../CaptureTests/CaptureLifecycleTests.swift | 52 +++++++ Tests/CaptureTests/CaptureStoreTests.swift | 18 +++ .../LocalFilesInterfaceTests.swift | 34 ++++- extensions/capture/README.md | 2 + .../Sources/Capture/CaptureStore.swift | 41 +++-- extensions/local-files/README.md | 2 + .../LocalFilesMac/DirectoryMonitor.swift | 131 ++++++++++++++++ .../LocalFilesMac/LocalFilesModel.swift | 144 ++++-------------- 9 files changed, 297 insertions(+), 129 deletions(-) create mode 100644 extensions/local-files/Sources/LocalFilesMac/DirectoryMonitor.swift diff --git a/README.md b/README.md index b98383e..c60e3e7 100644 --- a/README.md +++ b/README.md @@ -92,4 +92,4 @@ Swift 测试覆盖冻结包重开、目标绑定、原子确认、原生构图 原创 Swift 代码使用 MIT，Cida 选区读取与采集协调改编保留 Apache-2.0，BetterShot 选区与冻结图组件保留 BSD-3-Clause。见 [开源许可与致谢](THIRD_PARTY_NOTICES.md)。构建脚本从本次解析的依赖收集完整许可，随开发包交付；菜单“开源许可与致谢…”可打开通知。 -数据库操作将租用、结果读取和事务限制在同一个可取消作用域，单次最多等待 30 秒；取消或超时关闭当前连接。在途 Capture 仍通过原提交恢复确认。各信息库独立续租、刷新目录和处理采集待办，同一信息库内的采集待办按原顺序处理。停用本地扫描后不再另起网络任务写回 Job 终态，未确认终态由共享 Job 的既有截止时间收束。 +数据库操作将租用、结果读取和事务限制在同一个可取消作用域，单次最多等待 30 秒；取消或超时关闭当前连接。在途 Capture 仍通过原提交恢复确认。各信息库独立续租、刷新目录和处理采集待办，同一信息库内，符合重试条件的采集待办按首次提交顺序处理，后续重试不改变次序。停用本地扫描后不再另起网络任务写回 Job 终态，未确认终态由共享 Job 的既有截止时间收束。 diff --git a/Tests/CaptureTests/CaptureLifecycleTests.swift b/Tests/CaptureTests/CaptureLifecycleTests.swift index ab565b9..ed7adb2 100644 --- a/Tests/CaptureTests/CaptureLifecycleTests.swift +++ b/Tests/CaptureTests/CaptureLifecycleTests.swift @@ -36,6 +36,58 @@ private actor PausedRecovery: CaptureTransport { } } +private actor RecordingSubmissions: CaptureTransport { + private(set) var submitted: [UUID] = [] + func recover(_ record: CaptureRecord, secret: String) async throws -> CaptureResult? { nil } + func submit(_ record: CaptureRecord, secret: String) async throws -> CaptureResult { + submitted.append(record.id) + return CaptureResult(submissionID: record.id, rootBlockID: 123, committedAt: Date()) + } +} + +@Test func captureQueueKeepsSubmissionOrderAfterRetryAndReopening() async throws { + let directory = FileManager.default.temporaryDirectory.appendingPathComponent(UUID().uuidString) + let service = "dev.inkcre.queue-order.\(UUID())" + defer { + SecItemDelete( + [kSecClass as String: kSecClassGenericPassword, kSecAttrService as String: service] + as CFDictionary) + try? FileManager.default.removeItem(at: directory) + } + let target = ConnectionTarget(host: "localhost", database: "inkcre", username: "test") + let credentials = CredentialStore(service: service) + let connections = ConnectionSettings(url: directory.appendingPathComponent("connection.json")) + try await connections.save(target) + try credentials.save("test-password", for: target.id) + let url = directory.appendingPathComponent("capture.sqlite") + let store = try CaptureStore(url: url) + var drafts: [CaptureRecord] = [] + for index in 0..<3 { + let draft = CaptureRecord( + payload: CapturePayload( + capturedAt: Date(timeIntervalSince1970: 1_780_000_000 + Double(index)), + materials: [Material(text: "第 \(index + 1) 份材料")])) + try await store.saveDraft(draft) + drafts.append(draft) + } + // Capture and draft creation order are independent from submission order. + var ids: [UUID] = [] + for draft in [drafts[2], drafts[0], drafts[1]] { + _ = try await store.freeze(draft, target: target) + ids.append(draft.id) + } + // Updating the middle record must affect its retry delay, never its position. + _ = try await store.markDispatch(ids[1]) + try await store.retry(ids[1]) + let reopened = try CaptureStore(url: url) + let transport = RecordingSubmissions() + let queue = CaptureQueue( + store: reopened, connections: connections, credentials: credentials, client: transport) + try await queue.process(force: true) + #expect(await transport.submitted == ids) + #expect(try await reopened.records().allSatisfy { $0.state == .confirmed }) +} + @Test func stoppingCaptureWaitsForRecoveryAndPreventsFurtherDispatch() async throws { let directory = FileManager.default.temporaryDirectory.appendingPathComponent(UUID().uuidString) let service = "dev.inkcre.lifecycle.\(UUID().uuidString)" diff --git a/Tests/CaptureTests/CaptureStoreTests.swift b/Tests/CaptureTests/CaptureStoreTests.swift index a3ad808..64ae6de 100644 --- a/Tests/CaptureTests/CaptureStoreTests.swift +++ b/Tests/CaptureTests/CaptureStoreTests.swift @@ -6,6 +6,24 @@ import Testing @testable import Capture struct CaptureStoreTests { + @Test(arguments: [1, 3]) + func incompatibleDevelopmentFormatsAreRejectedWithoutChangingTheFile(version: Int) throws { + let url = FileManager.default.temporaryDirectory.appendingPathComponent(UUID().uuidString) + defer { try? FileManager.default.removeItem(at: url) } + var connection: OpaquePointer? + #expect(sqlite3_open(url.path, &connection) == SQLITE_OK) + let database = try #require(connection) + let status = sqlite3_exec( + database, + "CREATE TABLE records(id TEXT PRIMARY KEY, body BLOB); PRAGMA user_version=\(version);", + nil, nil, nil) + sqlite3_close(database) + #expect(status == SQLITE_OK) + let original = try Data(contentsOf: url) + #expect(throws: PeerError.self) { try CaptureStore(url: url) } + #expect(try Data(contentsOf: url) == original) + } + @Test func deletedDraftIsAbsentAfterRestartAndLateAutosaveCannotRecreateIt() async throws { let directory = FileManager.default.temporaryDirectory.appendingPathComponent( UUID().uuidString) diff --git a/Tests/LocalFilesTests/LocalFilesInterfaceTests.swift b/Tests/LocalFilesTests/LocalFilesInterfaceTests.swift index 9d2ff1e..8d33613 100644 --- a/Tests/LocalFilesTests/LocalFilesInterfaceTests.swift +++ b/Tests/LocalFilesTests/LocalFilesInterfaceTests.swift @@ -123,16 +123,44 @@ struct LocalFilesInterfaceTests { try Data("# 自动监视后的正文".utf8).write(to: url, options: .atomic) try await until { model.selectedContent == "# 自动监视后的正文" } let source = try #require(model.source) - model.toggle(source) + let client = LocalFileClient( + target: target, password: try #require(env["INKCRE_FILES_TEST_PASSWORD"])) + let other = ConnectionTarget( + host: target.host, port: target.port, database: target.database, + username: target.username, tls: target.tls) + try context.credentials.save( + try #require(env["INKCRE_FILES_TEST_PASSWORD"]), for: other.id) + try await context.connections.save(other) + await model.connectionDidChange(other) + #expect(model.sources.isEmpty && model.records.isEmpty) + try Data("# 后台目录继续监视".utf8).write(to: url, options: .atomic) + try await until { + guard + let record = try? await client.records(source: source).first(where: { + $0.id == file.id + }) + else { return false } + return (try? await client.content(for: record)) == "# 后台目录继续监视" + } + #expect(model.target?.id == other.id && model.sources.isEmpty && model.records.isEmpty) + _ = try await context.connections.select(target.id) + await model.connectionDidChange(target) + try await until { model.records.contains { $0.id == file.id } } + model.selectedFile = file.id + try await until { model.selectedContent == "# 后台目录继续监视" } + model.toggle(try #require(model.source)) try await until { model.source?.config.enabled == false && !model.working } try Data("# 暂停时修改".utf8).write(to: url, options: .atomic) try await Task.sleep(for: .seconds(2)) - #expect(model.selectedContent == "# 自动监视后的正文") + #expect(model.selectedContent == "# 后台目录继续监视") try await snapshot(window, name: "files-paused") model.toggle(try #require(model.source)) try await until { model.selectedContent == "# 暂停时修改" && !model.working } await model.stop() try Data("# 停机时修改".utf8).write(to: url, options: .atomic) + try await Task.sleep(for: .seconds(2)) + #expect(try await client.content(for: file) == "# 暂停时修改") + #expect(!model.isScanning(source)) try model.start() try await until { model.selectedContent == "# 停机时修改" } try FileManager.default.removeItem(at: url) @@ -148,8 +176,6 @@ struct LocalFilesInterfaceTests { try await until { model.selectedContent != nil && !model.loadingContent } try await snapshot(window, name: "files-failure") let restoredSource = try #require(model.source) - let client = LocalFileClient( - target: target, password: try #require(env["INKCRE_FILES_TEST_PASSWORD"])) model.remove(restoredSource) try await until { model.sources.isEmpty && !model.working } #expect(try await client.records(source: restoredSource).count == 6) diff --git a/extensions/capture/README.md b/extensions/capture/README.md index e79e306..673e303 100644 --- a/extensions/capture/README.md +++ b/extensions/capture/README.md @@ -2,6 +2,8 @@ `inkcre/capture` 的 first-party 原生分发，独立 Swift Package。`Capture` 拥有材料、图、SQLite 待办、事务编排与恢复；`CaptureMac` 提供 macOS 选区／剪贴板／截图和原生采集界面。各平台遵守共享 Capture 合同，图结构不由宿主 Core 定义。 +待办在冻结材料和目标的同一 SQLite 事务内分配入队序号。同一信息库中，符合重试条件的待办按首次提交顺序处理；草稿创建时间、材料采集时间和后续重试都不会改变这个顺序。记录页仍按最近更新显示。序号只属于本机存储，不进入共享 Capture 合同。当前未发布格式为 SQLite `user_version=2`，遇到其他已标记版本会拒绝打开并保留原文件，不执行隐式迁移。 + Host 提供 PeerCore 的数据库、标准图／Storage、持久目录、凭据和扩展生命周期。精确分发身份来自 `Sources/Capture/Resources/extension.json`；随签名应用交付并独立启停，不依赖另一个 Peer 执行收入。 在仓库根执行 `swift test`；本包可通过 `swift build --package-path extensions/capture` 单独构建。Cida 选区读取与采集协调改编的 Apache-2.0 许可见 Legal/Cida；BetterShot 应用内选区与冻结图组件的 BSD-3-Clause 许可、固定版本和修改说明见 Legal/BetterShot。冻结帧只在截图会话内存中存在，确认后裁剪原图 PNG；停用关闭选区并丢弃会话，不调用系统截图工具栏。原创代码见 LICENSE。 diff --git a/extensions/capture/Sources/Capture/CaptureStore.swift b/extensions/capture/Sources/Capture/CaptureStore.swift index 7bbf7dc..285827d 100644 --- a/extensions/capture/Sources/Capture/CaptureStore.swift +++ b/extensions/capture/Sources/Capture/CaptureStore.swift @@ -58,27 +58,33 @@ public actor CaptureStore { guard sqlite3_step(statement) == SQLITE_ROW else { throw database.error() } return sqlite3_column_int(statement, 0) } - guard version == 0 || version == 1 else { - throw PeerError.storage("本地数据来自较新版本，请更新 InKCre 后再打开。") + guard version == 0 || version == 2 else { + throw PeerError.storage("本地数据格式与当前开发版不兼容，原文件已保留。") } try database.execute( """ PRAGMA journal_mode=WAL; PRAGMA synchronous=FULL; - CREATE TABLE IF NOT EXISTS records (id TEXT PRIMARY KEY, body BLOB NOT NULL); - PRAGMA user_version=1; + CREATE TABLE IF NOT EXISTS records ( + id TEXT PRIMARY KEY, body BLOB NOT NULL, queue_position INTEGER UNIQUE + ); + PRAGMA user_version=2; """) try FileManager.default.setAttributes([.posixPermissions: 0o600], ofItemAtPath: url.path) } public func records() throws -> [CaptureRecord] { - try records(query: "SELECT body FROM records") + try records(query: "SELECT body FROM records").sorted { $0.updatedAt > $1.updatedAt } } public func pendingRecords() throws -> [CaptureRecord] { try records( query: - "SELECT body FROM records WHERE json_extract(CAST(body AS TEXT), '$.state') IN ('pending','unknown')" + """ + SELECT body FROM records + WHERE json_extract(CAST(body AS TEXT), '$.state') IN ('pending','unknown') + ORDER BY queue_position + """ ) } @@ -91,7 +97,7 @@ public actor CaptureStore { guard result == SQLITE_ROW else { throw database.error() } records.append(try CaptureJSON.decode(CaptureRecord.self, from: data(statement))) } - return records.sorted { $0.updatedAt > $1.updatedAt } + return records } } @@ -135,6 +141,17 @@ public actor CaptureStore { frozen.message = nil frozen.updatedAt = Date() try save(frozen) + // Admission order is independent of capture time, UI edits and retry timestamps. + try database.statement( + """ + UPDATE records + SET queue_position=(SELECT COALESCE(MAX(queue_position),0)+1 FROM records) + WHERE id=? + """ + ) { statement in + sqlite3_bind_text(statement, 1, draft.id.uuidString, -1, transient) + guard sqlite3_step(statement) == SQLITE_DONE else { throw database.error() } + } try database.execute("COMMIT") return frozen } catch { @@ -219,7 +236,7 @@ public actor CaptureStore { value.message = nil value.payload = nil value.updatedAt = Date() - // One row replacement atomically keeps the receipt while discarding local material. + // One row update atomically keeps the receipt while discarding local material. try save(value) } @@ -252,8 +269,12 @@ public actor CaptureStore { private func save(_ record: CaptureRecord) throws { let encoded = try CaptureJSON.encode(record) - try database.statement("INSERT OR REPLACE INTO records(id,body) VALUES (?,?)") { - statement in + try database.statement( + """ + INSERT INTO records(id,body) VALUES (?,?) + ON CONFLICT(id) DO UPDATE SET body=excluded.body + """ + ) { statement in sqlite3_bind_text(statement, 1, record.id.uuidString, -1, transient) let result = encoded.withUnsafeBytes { bytes in sqlite3_bind_blob(statement, 2, bytes.baseAddress, Int32(bytes.count), transient) diff --git a/extensions/local-files/README.md b/extensions/local-files/README.md index 03b8b0a..3344e5e 100644 --- a/extensions/local-files/README.md +++ b/extensions/local-files/README.md @@ -4,6 +4,8 @@ Source 配置在 info-base，书签及原连接绑定在 Host 的 `extensions/inkcre-local-files/directories.json`，权限 0600。密码仅存系统 Keychain。先监视再扫描；通知合并为待核对状态，不保存事件日志。每 30 秒核对共享配置、待执行 Job 和失败扫描，正常监视不会每 30 秒重新读取所有文件。每轮从原连接的 Keychain 取凭据。停止等待当前文件事务和提取结束，再释放监视与目录访问。 +`LocalFilesModel` 拥有当前连接、目录绑定和前台读取；每个 `DirectoryMonitor` 独立拥有 FSEvents、扫描任务、变更合并、Job 收束和目录访问的释放。切换前台连接不停止其他已绑定目录，扫描结果只刷新所属连接的窗口数据。停止监视时，所有调用者都等待当前扫描结束，之后才释放目录访问。 + 列表与扫描只读取文件元数据和正文 ID；选中文件才读取该文件正文，修订变化后更新当前正文。未变化文件不进入逐文件写事务，完整扫描仍复核来源状态并处理失效文件。 读取端解释 `extensions.local_files.file.v1`／`file.v2` 与 `body`，不要注册不能执行的 Source 占位类。重连扫描当前目录，不引入原文件复制或正文 outbox。 diff --git a/extensions/local-files/Sources/LocalFilesMac/DirectoryMonitor.swift b/extensions/local-files/Sources/LocalFilesMac/DirectoryMonitor.swift new file mode 100644 index 0000000..7077c44 --- /dev/null +++ b/extensions/local-files/Sources/LocalFilesMac/DirectoryMonitor.swift @@ -0,0 +1,131 @@ +import Foundation +import LocalFiles +import PeerCore + +/// One directory owns its watcher, coalesced changes, scan task and scoped access together. +@MainActor +final class DirectoryMonitor { + let source: DirectorySource + private let target: ConnectionTarget + private let context: PeerHostContext + private let root: URL + private var scoped: Bool + private let onState: (String, Bool) -> Void + private let onSources: ([DirectorySource]) -> Void + private var watcher: DirectoryWatcher? + private var task: Task? + private var stopped = false + private(set) var needsScan = false + var isScanning: Bool { task != nil } + + init( + source: DirectorySource, target: ConnectionTarget, context: PeerHostContext, + root: URL, scoped: Bool, + onState: @escaping (String, Bool) -> Void, + onSources: @escaping ([DirectorySource]) -> Void + ) throws { + self.source = source + self.target = target + self.context = context + self.root = root + self.scoped = scoped + self.onState = onState + self.onSources = onSources + watcher = try DirectoryWatcher(path: root.path) { [weak self] in self?.schedule() } + } + + func stop() async { + stopped = true + watcher?.stop() + watcher = nil + task?.cancel() + // Every caller waits for the same scan before scoped access is released. + await task?.value + if scoped { + scoped = false + root.stopAccessingSecurityScopedResource() + } + } + + func schedule() { + guard !stopped, !Task.isCancelled else { return } + needsScan = true + guard task == nil else { return } + task = Task { + var status = "正在扫描…" + defer { + task = nil + if !stopped { onState(status, false) } + } + guard !stopped, !Task.isCancelled else { return } + onState(status, true) + while needsScan, !stopped, !Task.isCancelled { + needsScan = false + var job: SourceJob? + let client: LocalFileClient + do { + let connection = try await context.connections.resolve(target) + client = LocalFileClient( + target: connection, + password: try context.credentials.read(for: connection.id) + ) + } catch { + status = error.localizedDescription + needsScan = true + break + } + do { + let pending = try await client.pendingScans(source) + for candidate in pending { + if let claimed = try await client.claim(candidate, source: source) { + job = claimed + break + } + } + let state = try await DirectoryScanner.scan( + source: source, client: client, job: job + ) { [weak self] progress in + await self?.report(progress) + } + if let job { + try await client.close( + job, status: state.errors == 0 ? "finished" : "failed", state: state) + } + status = state.errors == 0 ? "正在监视" : "\(state.errors) 个文件需要处理" + let updated = try await client.sources() + if !stopped, !Task.isCancelled { onSources(updated) } + if !(try await client.pendingScans(source)).isEmpty { needsScan = true } + } catch { + status = + PeerDatabase.isCancellation(error) + ? "扫描已停止" : PeerDatabase.failure(error).localizedDescription + if !PeerDatabase.isCancellation(error), !stopped { + do { try await client.reportIssue(status, source: source) } catch { + status += "；状态暂未写回信息库" + } + } + if let job, !Task.isCancelled, !stopped { + let terminal = + Date() >= job.deadline + ? "timed_out" + : PeerDatabase.isCancellation(error) ? "aborted" : "failed" + do { + try await client.close( + job, status: terminal, state: DirectoryState(issue: status)) + } catch { + status += "；任务终态待信息库核对" + } + } + // Reconnection reconciles current files, without an outbox. + needsScan = true + break + } + } + } + } + + private func report(_ progress: ScanProgress) { + guard !stopped else { return } + onState("\(progress.completed)/\(progress.total) · \(progress.filename)", true) + } +} diff --git a/extensions/local-files/Sources/LocalFilesMac/LocalFilesModel.swift b/extensions/local-files/Sources/LocalFilesMac/LocalFilesModel.swift index 22f8be8..2472361 100644 --- a/extensions/local-files/Sources/LocalFilesMac/LocalFilesModel.swift +++ b/extensions/local-files/Sources/LocalFilesMac/LocalFilesModel.swift @@ -34,7 +34,7 @@ public final class LocalFilesModel: ObservableObject { private let context: PeerHostContext private var bindings: [DirectoryBinding] = [] - private var monitors: [String: Monitor] = [:] + private var monitors: [String: DirectoryMonitor] = [:] private var prepared: Set = [] private var timer: Task? private var recordsTask: Task? @@ -43,22 +43,6 @@ public final class LocalFilesModel: ObservableObject { private var actionTask: Task? private var directoryPanel: NSOpenPanel? - private final class Monitor { - let source: DirectorySource - let target: ConnectionTarget - let root: URL - let scoped: Bool - var watcher: DirectoryWatcher? - var task: Task? - var dirty = false - init(source: DirectorySource, target: ConnectionTarget, root: URL, scoped: Bool) { - self.source = source - self.target = target - self.root = root - self.scoped = scoped - } - } - public init(context: PeerHostContext) { self.context = context } private var bindingURL: URL { context.directory.appendingPathComponent("extensions/inkcre-local-files/directories.json") @@ -107,7 +91,11 @@ public final class LocalFilesModel: ObservableObject { timer?.cancel() recordsTask?.cancel() contentTask?.cancel() - for monitor in monitors.values { monitor.task?.cancel() } + await withTaskGroup(of: Void.self) { group in + for id in Array(monitors.keys) { + group.addTask { await self.stopMonitor(id) } + } + } await actionTask?.value await refreshTask?.value await timer?.value @@ -117,7 +105,6 @@ public final class LocalFilesModel: ObservableObject { contentTask?.cancel() await contentTask?.value loadingContent = false - for id in Array(monitors.keys) { await stopMonitor(id) } } public func connectionDidChange(_ connection: ConnectionTarget) async { @@ -230,9 +217,9 @@ public final class LocalFilesModel: ObservableObject { continue } } - if let monitor = monitors[binding.id], monitor.task == nil { + if let monitor = monitors[binding.id], !monitor.isScanning { let jobs = try await client.pendingScans(source) - if !jobs.isEmpty || monitor.dirty { schedule(binding.id) } + if !jobs.isEmpty || monitor.needsScan { monitor.schedule() } } } for source in found @@ -263,7 +250,6 @@ public final class LocalFilesModel: ObservableObject { == URL(fileURLWithPath: source.config.path).resolvingSymlinksInPath().path else { throw PeerError.needsAction("目录位置已改变，请重新选择目录。") } let scoped = root.startAccessingSecurityScopedResource() - let monitor = Monitor(source: source, target: target, root: root, scoped: scoped) do { if stale, let index = bindings.firstIndex(where: { $0.id == binding.id }) { bindings[index].bookmark = try root.bookmarkData( @@ -271,11 +257,21 @@ public final class LocalFilesModel: ObservableObject { relativeTo: nil) try saveBindings() } - monitor.watcher = try DirectoryWatcher(path: root.path) { [weak self] in - self?.schedule(binding.id) - } - monitors[binding.id] = monitor - schedule(binding.id) + let id = binding.id + let monitor = try DirectoryMonitor( + source: source, target: target, context: context, root: root, scoped: scoped, + onState: { [weak self] message, active in + guard let self else { return } + statuses[id] = message + if active { scanning.insert(id) } else { scanning.remove(id) } + }, + onSources: { [weak self] updated in + guard let self, running, self.target?.id == target.id else { return } + sources = updated + loadRecords() + }) + monitors[id] = monitor + monitor.schedule() } catch { if scoped { root.stopAccessingSecurityScopedResource() } throw error @@ -283,95 +279,15 @@ public final class LocalFilesModel: ObservableObject { } private func stopMonitor(_ id: String) async { - guard let monitor = monitors.removeValue(forKey: id) else { return } - monitor.watcher?.stop() - monitor.task?.cancel() - await monitor.task?.value - if monitor.scoped { monitor.root.stopAccessingSecurityScopedResource() } - scanning.remove(id) - } - - private func schedule(_ id: String) { - guard running, let monitor = monitors[id] else { return } - monitor.dirty = true - guard monitor.task == nil else { return } - monitor.task = Task { [weak self, weak monitor] in - guard let self, let monitor else { return } - defer { - monitor.task = nil - scanning.remove(id) - } - while monitor.dirty, running, !Task.isCancelled { - monitor.dirty = false - scanning.insert(id) - statuses[id] = "正在扫描…" - var job: SourceJob? - let client: LocalFileClient - do { client = try await self.client(monitor.target) } catch { - statuses[id] = error.localizedDescription - monitor.dirty = true - break - } - do { - let pending = try await client.pendingScans(monitor.source) - for candidate in pending { - if let claimed = try await client.claim(candidate, source: monitor.source) { - job = claimed - break - } - } - let state = try await DirectoryScanner.scan( - source: monitor.source, client: client, job: job - ) { [weak self] progress in - await self?.report(progress, id: id) - } - if let job { - try await client.close( - job, status: state.errors == 0 ? "finished" : "failed", state: state) - } - statuses[id] = state.errors == 0 ? "正在监视" : "\(state.errors) 个文件需要处理" - let updated = try await client.sources() - if target?.peerID == monitor.source.config.peer, running, !Task.isCancelled { - sources = updated - loadRecords() - } - if !(try await client.pendingScans(monitor.source)).isEmpty { - monitor.dirty = true - } - } catch { - let message = - PeerDatabase.isCancellation(error) - ? "扫描已停止" : PeerDatabase.failure(error).localizedDescription - statuses[id] = message - if !PeerDatabase.isCancellation(error), running { - do { try await client.reportIssue(message, source: monitor.source) } catch { - statuses[id] = "\(message)；状态暂未写回信息库" - } - } - if let job, !Task.isCancelled, running { - let status = - Date() >= job.deadline - ? "timed_out" - : PeerDatabase.isCancellation(error) ? "aborted" : "failed" - do { - try await client.close( - job, status: status, state: DirectoryState(issue: message)) - } catch { - statuses[id] = "\(message)；任务终态待信息库核对" - } - } - // Reconnection reconciles current files, without an outbox. - monitor.dirty = true - break - } - } + guard let monitor = monitors[id] else { return } + await monitor.stop() + // A concurrent stop must await this monitor too, without removing its replacement. + if monitors[id] === monitor { + monitors.removeValue(forKey: id) + scanning.remove(id) } } - private func report(_ progress: ScanProgress, id: String) { - statuses[id] = "\(progress.completed)/\(progress.total) · \(progress.filename)" - } - private func prepare(_ client: LocalFileClient) async throws { guard !prepared.contains(client.target.id) else { return } try await client.prepare() @@ -519,7 +435,7 @@ public final class LocalFilesModel: ObservableObject { public func rescan(_ source: DirectorySource) { perform { target in _ = try await self.client(target).submitScan(source) - self.schedule(self.key(source.id, target)) + self.monitors[self.key(source.id, target)]?.schedule() } } From ddcbd4255cbc7269856f7ef44b7ceb00992bcac3 Mon Sep 17 00:00:00 2001 From: Lan_zhijiang Date: Sat, 31 Oct 2026 19:25:38 +0800 Subject: [PATCH 2/3] =?UTF-8?q?ref(native):=20=E6=94=B6=E6=8B=A2=E7=9B=AE?= =?UTF-8?q?=E5=BD=95=E7=9B=91=E8=A7=86=E4=B8=8E=E5=B1=95=E7=A4=BA=E7=8A=B6?= =?UTF-8?q?=E6=80=81=E6=89=80=E6=9C=89=E6=9D=83=E5=B9=B6=E6=8E=A5=E5=85=A5?= =?UTF-8?q?=E7=A9=BA=E5=BA=93=E9=AA=8C=E6=94=B6?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .github/workflows/ci.yml | 31 +++++ .gitignore | 3 + Application.json | 1 + .../Sources/PeerCore/NativeExtension.swift | 21 ++++ .../Sources/PeerCore/PeerDatabase.swift | 12 +- README.md | 25 +++- Sources/InKCreApp/AppMain.swift | 20 +-- .../CaptureInteractionTests.swift | 67 ++++++++++ Tests/CaptureTests/LivePeerTests.swift | 5 +- .../DirectoryMonitorTests.swift | 107 ++++++++++++++++ .../Sources/Capture/CaptureClient.swift | 24 +--- .../Sources/Capture/CaptureStore.swift | 2 +- .../Sources/CaptureMac/CaptureExtension.swift | 27 +++-- extensions/local-files/README.md | 6 +- .../Sources/LocalFiles/Models.swift | 4 +- .../LocalFilesMac/DirectoryMonitor.swift | 67 +++++++--- .../LocalFilesMac/LocalFilesModel.swift | 61 ++++------ scripts/build-app.sh | 9 +- scripts/check-database.py | 114 ++++++++++++++++++ scripts/check.sh | 3 +- scripts/database-source-revision | 1 + 21 files changed, 502 insertions(+), 108 deletions(-) create mode 100644 Application.json create mode 100644 Tests/LocalFilesTests/DirectoryMonitorTests.swift create mode 100644 scripts/check-database.py create mode 100644 scripts/database-source-revision diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index a495954..3271dc6 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -21,3 +21,34 @@ jobs: - 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: 一次性数据库与原生主流程验收 + run: >- + python3 scripts/check-database.py --impl-py .verification/impl-py + --postgres-bin "$(brew --prefix postgresql@17)/bin" diff --git a/.gitignore b/.gitignore index a9c2649..877eb1a 100644 --- a/.gitignore +++ b/.gitignore @@ -7,3 +7,6 @@ xcuserdata/ .local-development-signing-identity **/Package.resolved !/Package.resolved + +# Disposable authority checkout used by database acceptance. +.verification/ diff --git a/Application.json b/Application.json new file mode 100644 index 0000000..de861b3 --- /dev/null +++ b/Application.json @@ -0,0 +1 @@ +{"version":"0.1.0","build":19} diff --git a/Packages/PeerCore/Sources/PeerCore/NativeExtension.swift b/Packages/PeerCore/Sources/PeerCore/NativeExtension.swift index 76c8f14..d706dba 100644 --- a/Packages/PeerCore/Sources/PeerCore/NativeExtension.swift +++ b/Packages/PeerCore/Sources/PeerCore/NativeExtension.swift @@ -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 diff --git a/Packages/PeerCore/Sources/PeerCore/PeerDatabase.swift b/Packages/PeerCore/Sources/PeerCore/PeerDatabase.swift index 0b9249b..7eac589 100644 --- a/Packages/PeerCore/Sources/PeerCore/PeerDatabase.swift +++ b/Packages/PeerCore/Sources/PeerCore/PeerDatabase.swift @@ -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() @@ -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( diff --git a/README.md b/README.md index c60e3e7..8d81ea6 100644 --- a/README.md +++ b/README.md @@ -20,7 +20,7 @@ open dist/InKCre.app 构建产物是 universal `.app` 和 `dist/InKCre-0.1.0-universal.dmg`。未指定开发签名身份时使用 ad-hoc 签名；两种开发包都不能当作已公证的发行版。将 `.app` 放在固定位置，并保持稳定的开发签名身份后再授权。 -首版尚未发布，应用版本当前为 `0.1.0`。`CFBundleShortVersionString` 保存应用版本，`CFBundleVersion` 单独记录构建序号；交付说明写作 `0.1.0（build 15）`，不能将构建序号称作 `v15`。首版发布前最多递增 patch 或 build，不提升 minor／major；一般开发构建只递增 build，正式版本准备遵循组织的 Release 流程。 +首版尚未发布，应用版本当前为 `0.1.0`，版本与 build 的构建输入统一在 `Application.json`。`CFBundleShortVersionString` 保存应用版本，`CFBundleVersion` 单独记录构建序号；交付说明按实际构建写作 `0.1.0（build 19）`，不能将构建序号称作 `v15`。首版发布前最多递增 patch 或 build，不提升 minor／major；一般开发构建只递增 build，正式版本准备遵循组织的 Release 流程。 需要反复开发和实测系统采集时，使用本人明确指定的 Apple Development 身份：`DEVELOPMENT_SIGNING_IDENTITY='Apple Development: …' bash scripts/build-app.sh`。也可将该身份的名称或证书 SHA-1 单独保存在 ignored `.local-development-signing-identity`（一行文本，不执行其中内容）。脚本不会自动选择证书或回退到另一个身份；正式发布仍使用独立的 `SIGNING_IDENTITY` 输入。ad-hoc 身份随代码版本变化，系统可能不再认可先前的隐私授权；固定路径和 bundle ID 本身不足以解决此问题。[Apple 对代码身份与授权的说明](https://developer.apple.com/documentation/technotes/tn3127-inside-code-signing-requirements)。 @@ -38,6 +38,29 @@ open dist/InKCre.app 手写想法使用单一主要输入区；采集原文与可选附注分别呈现，附注按需展开，已有附注默认可见，编辑区不嵌套在原文滚动区内。图片可用原生 Quick Look 查看原始尺寸，预览临时文件只对当前用户可读写，关闭预览或离开内容时移除。未收入记录可以展开查看完整原文、全部图片、来源和附注，无须导出 JSON 才能核对材料。上次收入状态与当前草稿保存状态分别显示，“查看记录”定位到对应条目，继续写下一条想法不会丢失入口。 +## 数据库与原生主流程验收 + +默认 `bash scripts/check.sh` 检查格式和不依赖数据库的行为；数据库用例未配置时会明确跳过。两种入口都显式串行执行测试，因为原生界面用例共享 NSApplication、菜单和焦点状态。`SWIFT_JOBS` 只调节编译并发，默认 2。 + +完整验收消费 impl-py 的权威初始化，不在本仓维护 schema。CI 使用 `scripts/database-source-revision` 固定该实现版本；本地可传入已完成 PDM 安装的 impl-py checkout，入口会显示实际 revision。准备 PostgreSQL 17 与 pgvector 后，从空库开始执行： + +```sh +brew install postgresql@17 pgvector +# impl-py 中先按其说明安装 Python 3.12 和锁定 PDM 依赖。 +python3 scripts/check-database.py --impl-py /path/to/impl-py \ + --postgres-bin "$(brew --prefix postgresql@17)/bin" +``` + +此入口仅在临时目录创建独立 PostgreSQL cluster、随机回环端口和随机凭据，调用权威 `scripts/database.py init --profile development`／`ready`，开启 Capture 和 Local Files 的全部数据库及隐藏原生窗口旅程，结束后停止并删除该临时 cluster。不会使用现有 development／preview 库，也不启动系统 PostgreSQL 服务。缺工具、迁移失败或验收失败均返回非零。 + +已有独占且完成权威迁移的一次性数据库时，可以消费权限为 0600 的 JSON descriptor： + +```sh +python3 scripts/check-database.py --descriptor /private/path/swift-test.json +``` + +descriptor 必须声明 `disposable: true`，包含 `host`（回环）、`port`、`database`、`username: "inkcre_core"` 和 `password`；调用者负责该库的独占性与销毁。凭据通过子进程环境传入，不 source dotenv，也不放入命令参数。脚本不修改 schema；`INKCRE_LAYOUT_EVIDENCE_DIR` 可指定已有目录保存合成界面的实际渲染图。这些检查不代表系统授权提示、真实跨应用选区、多屏实体环境或公证已通过。 + ## 连接信息库 连接设置使用 PostgreSQL 主机、端口、库名、独立运行角色及密码，可选 Web 地址。远端使用 TLS 并校验证书；只有回环开发地址允许关闭 TLS。密码存于 Keychain，角色拥有已准入数据库协议权限，不使用迁移管理员或 JWT 签名密钥。 diff --git a/Sources/InKCreApp/AppMain.swift b/Sources/InKCreApp/AppMain.swift index 8c30419..a5351c9 100644 --- a/Sources/InKCreApp/AppMain.swift +++ b/Sources/InKCreApp/AppMain.swift @@ -36,11 +36,7 @@ final class PeerHost: NSObject, NSApplicationDelegate, ObservableObject, NSUserI } @Published var columnVisibility = NavigationSplitViewVisibility.all @Published var page = Page.capture { - didSet { - capture.showingCapturePage = page == .capture || page == .records - if page == .capture { capture.selectPage(.capture) } - if page == .records { capture.selectPage(.queue) } - } + didSet { updateCapturePresentation() } } private(set) var mainWindow: PeerMainWindow? override convenience init() { @@ -90,7 +86,6 @@ final class PeerHost: NSObject, NSApplicationDelegate, ObservableObject, NSUserI makeMainWindow() do { try runtime.startIfEnabled() - capture.selectPage(page == .records ? .queue : .capture) } catch { show(error) } do { try filesRuntime.startIfEnabled() } catch { show(error) } showMain() @@ -115,7 +110,17 @@ final class PeerHost: NSObject, NSApplicationDelegate, ObservableObject, NSUserI func makeMainWindow() { guard mainWindow == nil else { return } mainWindow = PeerMainWindow(host: self) - capture.mainWindow = mainWindow + updateCapturePresentation() + } + + private func updateCapturePresentation() { + let capturePage: CapturePage? = + switch page { + case .capture: .capture + case .records: .queue + case .files, .peer: nil + } + capture.updateHostPresentation(page: capturePage, window: mainWindow) } @objc func showMain() { @@ -233,7 +238,6 @@ final class PeerHost: NSObject, NSApplicationDelegate, ObservableObject, NSUserI Task { objectWillChange.send() do { try await runtime.setEnabled(enable) } catch { show(error) } - capture.selectPage(page == .records ? .queue : .capture) manager?.refreshCaptureSettings(host: self) updateStatus() } diff --git a/Tests/CaptureMacTests/CaptureInteractionTests.swift b/Tests/CaptureMacTests/CaptureInteractionTests.swift index aade01b..2c5589c 100644 --- a/Tests/CaptureMacTests/CaptureInteractionTests.swift +++ b/Tests/CaptureMacTests/CaptureInteractionTests.swift @@ -9,6 +9,73 @@ import Testing @Suite(.serialized) @MainActor struct CaptureInteractionTests { init() { _ = NSApplication.shared } + @Test func hostPresentationCanArriveAfterStartAndChangeWindows() async throws { + let directory = FileManager.default.temporaryDirectory.appendingPathComponent( + UUID().uuidString) + defer { try? FileManager.default.removeItem(at: directory) } + let previousMenu = NSApp.mainMenu + let menu = NSMenu() + menu.addItem(NSMenuItem(title: "InKCre", action: nil, keyEquivalent: "")) + NSApp.mainMenu = menu + defer { NSApp.mainMenu = previousMenu } + let capture = CaptureExtension( + context: PeerHostContext( + identity: "dev.inkcre.presentation.\(UUID())", directory: directory)) + let first = NSWindow( + contentRect: NSRect(x: 0, y: 0, width: 400, height: 300), styleMask: [.titled], + backing: .buffered, defer: false) + let second = NSWindow( + contentRect: NSRect(x: 0, y: 0, width: 400, height: 300), styleMask: [.titled], + backing: .buffered, defer: false) + let sheet = NSWindow( + contentRect: NSRect(x: 0, y: 0, width: 400, height: 300), styleMask: [.titled], + backing: .buffered, defer: false) + defer { + first.orderOut(nil) + second.orderOut(nil) + sheet.orderOut(nil) + } + try capture.start() + do { + let commands = try #require(menu.items.first { $0.title == "采集" }?.submenu) + let records = try #require(commands.items.first { $0.title == "采集记录" }) + capture.updateHostPresentation(page: .capture, window: first) + // Swift Testing does not run the app event loop. Supply native windows directly + // to the same internal action policy used by menu and shortcut callbacks. + for _ in 0..<100 where !capture.canPerform(.paste, in: first) { + try await Task.sleep(for: .milliseconds(10)) + } + #expect(capture.canPerform(.paste, in: first)) + // The extension can navigate before the Host projects a new page. The real pane wins. + #expect( + NSApp.sendAction(try #require(records.action), to: records.target, from: records)) + #expect(!capture.canPerform(.paste, in: first)) + capture.updateHostPresentation(page: .queue, window: first) + #expect(!capture.canPerform(.paste, in: first)) + capture.updateHostPresentation(page: nil, window: first) + #expect(!capture.canPerform(.paste, in: first)) + capture.updateHostPresentation(page: .capture, window: second) + #expect(!capture.canPerform(.paste, in: first)) + #expect(capture.canPerform(.paste, in: second)) + second.beginSheet(sheet) { _ in } + #expect(!capture.canPerform(.paste, in: second)) + second.endSheet(sheet) + sheet.orderOut(nil) + try await capture.stop() + #expect(!capture.canPerform(.paste, in: first)) + // Disabled/restarted Capture consumes the retained Host input without resynchronization. + try capture.start() + for _ in 0..<100 where !capture.canPerform(.paste, in: second) { + try await Task.sleep(for: .milliseconds(10)) + } + #expect(capture.canPerform(.paste, in: second)) + try await capture.stop() + } catch { + try await capture.stop() + throw error + } + } + @Test func emptyCaptureShortcutDoesNotOpenConnectionOrStartSubmission() throws { let directory = FileManager.default.temporaryDirectory.appendingPathComponent( UUID().uuidString) diff --git a/Tests/CaptureTests/LivePeerTests.swift b/Tests/CaptureTests/LivePeerTests.swift index 8f625f0..8688517 100644 --- a/Tests/CaptureTests/LivePeerTests.swift +++ b/Tests/CaptureTests/LivePeerTests.swift @@ -13,7 +13,7 @@ func nativeTransactionCommitsOriginalImageOnceAndRecoversAfterRestart() async th database: try #require(env["INKCRE_CAPTURE_TEST_DATABASE"]), username: "inkcre_core", tls: .disable) let password = try #require(env["INKCRE_CAPTURE_TEST_PASSWORD"]) - let database = PeerDatabase() + let database = PeerDatabase(applicationVersion: "swift-acceptance") let client = CaptureClient(database: database) let pg = try await database.client(for: target, password: password) struct Fixture: Decodable { let payload: CapturePayload } @@ -33,6 +33,9 @@ func nativeTransactionCommitsOriginalImageOnceAndRecoversAfterRestart() async th let results = try await [a, b, c] let result = try #require(results.first) #expect(results.allSatisfy { $0 == result }) + let versions = try await pg.query( + "SELECT application_version FROM inkcre.peers WHERE id=\(target.peerID)") + #expect(try await versions.decode(String.self).first { _ in true } == "swift-acceptance") let rows = try await pg.query( """ WITH RECURSIVE captured(id) AS (SELECT \(result.rootBlockID)::integer UNION diff --git a/Tests/LocalFilesTests/DirectoryMonitorTests.swift b/Tests/LocalFilesTests/DirectoryMonitorTests.swift new file mode 100644 index 0000000..d691a2f --- /dev/null +++ b/Tests/LocalFilesTests/DirectoryMonitorTests.swift @@ -0,0 +1,107 @@ +import Foundation +import LocalFiles +import PeerCore +import Security +import Testing + +@testable import LocalFilesMac + +private actor ScanGate { + private(set) var count = 0 + private var waiting: [(Int, CheckedContinuation)] = [] + private var releases: [CheckedContinuation] = [] + + func scan() async { + count += 1 + let ready = waiting.filter { $0.0 <= count } + waiting.removeAll { $0.0 <= count } + for (_, continuation) in ready { continuation.resume() } + // Deliberately finish only when released: stopping must await in-flight extraction. + await withCheckedContinuation { releases.append($0) } + } + func waitForScan(_ number: Int) async { + if count < number { + await withCheckedContinuation { waiting.append((number, $0)) } + } + } + func finishScan() { releases.removeFirst().resume() } +} + +@Suite(.serialized) @MainActor +struct DirectoryMonitorTests { + @Test( + .enabled(if: ProcessInfo.processInfo.environment["INKCRE_FILES_TEST_DATABASE"] != nil), + .timeLimit(.minutes(1))) + func coalescedChangesAndConcurrentStopsWaitForTheSameScan() async throws { + let env = ProcessInfo.processInfo.environment + let root = FileManager.default.temporaryDirectory.appendingPathComponent(UUID().uuidString) + try FileManager.default.createDirectory(at: root, withIntermediateDirectories: true) + let context = PeerHostContext( + identity: "dev.inkcre.monitor-tests.\(UUID())", directory: root) + let target = ConnectionTarget( + host: "127.0.0.1", port: Int(env["INKCRE_FILES_TEST_PORT"] ?? "55483")!, + database: try #require(env["INKCRE_FILES_TEST_DATABASE"]), username: "inkcre_core", + tls: .disable) + let secret = try #require(env["INKCRE_FILES_TEST_PASSWORD"]) + try context.credentials.save(secret, for: target.id) + try await context.connections.save(target) + defer { + SecItemDelete( + [ + kSecClass as String: kSecClassGenericPassword, + kSecAttrService as String: "\(context.identity).connections", + ] as CFDictionary) + try? FileManager.default.removeItem(at: root) + } + let folder = root.appendingPathComponent("documents") + try FileManager.default.createDirectory(at: folder, withIntermediateDirectories: true) + let client = LocalFileClient(target: target, password: secret) + try await client.prepare() + let source = try await client.createDirectory(path: folder.path) + let gate = ScanGate() + var sourceUpdates = 0 + let monitor = try DirectoryMonitor( + source: source, target: target, context: context, + bookmark: folder.bookmarkData( + options: [.withSecurityScope], includingResourceValuesForKeys: nil, relativeTo: nil), + onBookmarkRenewed: { _ in }, onState: { _, _ in }, + onSources: { _ in sourceUpdates += 1 }, + scan: { _, _, _, _ in + await gate.scan() + return DirectoryState() + }) + // A remote Job wakes a clean monitor; the Model need not inspect pending jobs or dirty state. + _ = try await client.submitScan(source) + monitor.refresh() + await gate.waitForScan(1) + monitor.schedule() + monitor.schedule() + await gate.finishScan() + await gate.waitForScan(2) + #expect(sourceUpdates == 1) + var startedStops = 0 + var finishedStops = 0 + let first = Task { + startedStops += 1 + await monitor.stop() + finishedStops += 1 + } + let second = Task { + startedStops += 1 + await monitor.stop() + finishedStops += 1 + } + while startedStops != 2 { await Task.yield() } + #expect(finishedStops == 0) + monitor.schedule() + monitor.refresh() + await gate.finishScan() + await first.value + await second.value + #expect(finishedStops == 2) + #expect(await gate.count == 2) + #expect(sourceUpdates == 1) + await monitor.stop() + #expect(await gate.count == 2) + } +} diff --git a/extensions/capture/Sources/Capture/CaptureClient.swift b/extensions/capture/Sources/Capture/CaptureClient.swift index ebb97f2..a0184ff 100644 --- a/extensions/capture/Sources/Capture/CaptureClient.swift +++ b/extensions/capture/Sources/Capture/CaptureClient.swift @@ -8,28 +8,8 @@ public protocol CaptureTransport: Sendable { } public struct CaptureClient: CaptureTransport { - public static let descriptor: NativeExtensionDescriptor = { - let bundle: Bundle - if Bundle.main.bundleURL.pathExtension == "app" { - guard - let url = Bundle.main.url( - forResource: "InKCreCapture_Capture", withExtension: "bundle"), - let packaged = Bundle(url: url) - else { - preconditionFailure("Missing packaged Capture distribution") - } - bundle = packaged - } else { - bundle = .module - } - guard let url = bundle.url(forResource: "extension", withExtension: "json"), - let data = try? Data(contentsOf: url), - let descriptor = try? JSONDecoder().decode(NativeExtensionDescriptor.self, from: data) - else { - preconditionFailure("Invalid bundled Capture manifest") - } - return descriptor - }() + public static let descriptor = NativeExtensionDescriptor.bundled( + resourceBundle: "InKCreCapture_Capture", developmentBundle: .module) private let database: PeerDatabase public init(database: PeerDatabase = .shared) { self.database = database } diff --git a/extensions/capture/Sources/Capture/CaptureStore.swift b/extensions/capture/Sources/Capture/CaptureStore.swift index 285827d..ff88e26 100644 --- a/extensions/capture/Sources/Capture/CaptureStore.swift +++ b/extensions/capture/Sources/Capture/CaptureStore.swift @@ -59,7 +59,7 @@ public actor CaptureStore { return sqlite3_column_int(statement, 0) } guard version == 0 || version == 2 else { - throw PeerError.storage("本地数据格式与当前开发版不兼容，原文件已保留。") + throw PeerError.storage("本地数据格式 v\(version) 与当前开发版 v2 不兼容，原文件已保留；不会猜测旧待办顺序或自动导入。") } try database.execute( """ diff --git a/extensions/capture/Sources/CaptureMac/CaptureExtension.swift b/extensions/capture/Sources/CaptureMac/CaptureExtension.swift index 5d8235e..0f6818f 100644 --- a/extensions/capture/Sources/CaptureMac/CaptureExtension.swift +++ b/extensions/capture/Sources/CaptureMac/CaptureExtension.swift @@ -52,12 +52,22 @@ public final class CaptureExtension: NSObject, NativeExtension, NSWindowDelegate private var statusItem: NSStatusItem? private var shortcutObservation: AnyCancellable? private var panel: CapturePanel? - public weak var mainWindow: NSWindow? + private weak var mainWindow: NSWindow? public var presentMain: (() -> Void)? public var navigate: ((CapturePage) -> Void)? - public var showingCapturePage = true + private var mainPage: CapturePage? - public func selectPage(_ page: CapturePage) { model?.pane = page } + /// The Host supplies its current presentation in one operation, before or after starting. + public func updateHostPresentation(page: CapturePage?, window: NSWindow?) { + mainWindow = window + mainPage = page + if let page { model?.pane = page } + } + + private func isCaptureWindow(_ window: NSWindow?) -> Bool { + guard let window else { return false } + return window === panel || window === mainWindow + } public func pageView() -> AnyView { if let model { return AnyView(CapturePageView(model: model)) } return AnyView(ContentUnavailableView("采集扩展已停用", systemImage: "viewfinder")) @@ -98,6 +108,7 @@ public final class CaptureExtension: NSObject, NativeExtension, NSWindowDelegate directory: context.directory.appendingPathComponent("extensions/inkcre-capture"), identity: context.identity, connections: context.connections) self.model = model + if let mainPage { model.pane = mainPage } let panel = CapturePanel( contentRect: NSRect(x: 0, y: 0, width: 570, height: 640), styleMask: [.titled, .closable, .resizable, .nonactivatingPanel], @@ -121,8 +132,8 @@ public final class CaptureExtension: NSObject, NativeExtension, NSWindowDelegate self?.editConnection?(target, completion) } model.takeScreenshot = { [weak self] in self?.screenshot() } - model.shortcuts.isCaptureWindow = { [weak panel, weak mainWindow] window in - window === panel || window === mainWindow + model.shortcuts.isCaptureWindow = { [weak self] window in + self?.isCaptureWindow(window) == true } model.shortcuts.canPerform = { [weak self] action, window in self?.canPerform(action, in: window) == true @@ -201,14 +212,14 @@ public final class CaptureExtension: NSObject, NativeExtension, NSWindowDelegate } } - private func canPerform(_ action: ShortcutAction, in window: NSWindow?) -> Bool { + func canPerform(_ action: ShortcutAction, in window: NSWindow?) -> Bool { guard canChangeCaptureContext, let model, model.shortcuts.recording == nil else { return false } let quick = window === panel && model.editorOwner == .quick let editor = quick - || (showingCapturePage && window === mainWindow && model.pane == .capture + || (mainPage != nil && window === mainWindow && model.pane == .capture && model.editorOwner == .main) return model.canPerform(action, isEditor: editor, isQuickPanel: quick) } @@ -427,7 +438,7 @@ public final class CaptureExtension: NSObject, NativeExtension, NSWindowDelegate panel?.orderOut(nil) if let statusItem { NSStatusBar.system.removeStatusItem(statusItem) } statusItem = nil - if let commandItem { NSApp.mainMenu?.removeItem(commandItem) } + if let commandItem { commandItem.menu?.removeItem(commandItem) } commandItem = nil shortcutObservation = nil try await model.flush() diff --git a/extensions/local-files/README.md b/extensions/local-files/README.md index 3344e5e..2bc2ab5 100644 --- a/extensions/local-files/README.md +++ b/extensions/local-files/README.md @@ -4,15 +4,17 @@ Source 配置在 info-base，书签及原连接绑定在 Host 的 `extensions/inkcre-local-files/directories.json`，权限 0600。密码仅存系统 Keychain。先监视再扫描；通知合并为待核对状态，不保存事件日志。每 30 秒核对共享配置、待执行 Job 和失败扫描，正常监视不会每 30 秒重新读取所有文件。每轮从原连接的 Keychain 取凭据。停止等待当前文件事务和提取结束，再释放监视与目录访问。 -`LocalFilesModel` 拥有当前连接、目录绑定和前台读取；每个 `DirectoryMonitor` 独立拥有 FSEvents、扫描任务、变更合并、Job 收束和目录访问的释放。切换前台连接不停止其他已绑定目录，扫描结果只刷新所属连接的窗口数据。停止监视时，所有调用者都等待当前扫描结束，之后才释放目录访问。 +`LocalFilesModel` 拥有当前连接、目录绑定和前台读取；每个 `DirectoryMonitor` 独立拥有书签解析与更新、目录访问的获取／释放、FSEvents、扫描任务、变更合并、Job 轮询与收束。Model 仅核对绑定是否应存在并通知会话刷新，不读取调度内部状态。切换前台连接不停止其他已绑定目录，扫描结果只刷新所属连接的窗口数据。停止监视时，所有调用者都等待当前扫描结束，之后才释放目录访问。 列表与扫描只读取文件元数据和正文 ID；选中文件才读取该文件正文，修订变化后更新当前正文。未变化文件不进入逐文件写事务，完整扫描仍复核来源状态并处理失效文件。 读取端解释 `extensions.local_files.file.v1`／`file.v2` 与 `body`，不要注册不能执行的 Source 占位类。重连扫描当前目录，不引入原文件复制或正文 outbox。 -验证使用合成目录和一次性 PostgreSQL。`swift test` 包含 URI／六种格式提取；设置 `INKCRE_FILES_TEST_DATABASE`、`INKCRE_FILES_TEST_PORT`、`INKCRE_FILES_TEST_PASSWORD` 后执行真实事务、FSEvents、暂停／恢复、重启、删除保留与隐藏窗口主流程。测试运行角色是 `inkcre_core`，仅允许指定一次性数据库。`INKCRE_LAYOUT_EVIDENCE_DIR` 保存实际 AppKit 窗口的渲染图，不激活窗口。 +验证使用合成目录和一次性 PostgreSQL。`swift test` 包含 URI／六种格式提取；设置 `INKCRE_FILES_TEST_DATABASE`、`INKCRE_FILES_TEST_PORT`、`INKCRE_FILES_TEST_PASSWORD` 后执行真实事务、FSEvents、暂停／恢复、重启、删除保留与隐藏窗口主流程。测试运行角色是 `inkcre_core`，仅允许指定一次性数据库。从空库开始的准备与 CI 共用入口见根 README 的“数据库与原生主流程验收”。`INKCRE_LAYOUT_EVIDENCE_DIR` 保存实际 AppKit 窗口的渲染图，不激活窗口。 本分发写入 file.v2：HTML 保存提取后的结构，Markdown 保存原文，PDF／DOCX／Org／纯文沿原有文字范围。读取按正文的精确 resolver 选择纯文、HTML 或 GFM；WebKit 禁止来源脚本，净化及 CSP 阻止资源自动加载，已知资源保留说明和链接。没有可信网页地址时不把机器路径当 base。 成功重新提取才在原事务内升级旧根和正文的格式，保持 ID 与其他关系；失败／原件缺失保留旧成功内容和修订。`mac-native-content.v2` 触发原件未变时的必要重提取，分发版本选择沿用 Host 的既有保护。 + +分发身份仅定义在 `Sources/LocalFiles/Resources/extension.json`，运行时与应用包读取同一 manifest；版本变更不再另改 Swift 常量。 diff --git a/extensions/local-files/Sources/LocalFiles/Models.swift b/extensions/local-files/Sources/LocalFiles/Models.swift index 0cc3595..c8d5ecc 100644 --- a/extensions/local-files/Sources/LocalFiles/Models.swift +++ b/extensions/local-files/Sources/LocalFiles/Models.swift @@ -2,8 +2,8 @@ import Foundation import PeerCore public enum LocalFilesContract { - public static let descriptor = NativeExtensionDescriptor( - name: "inkcre/local-files", distribution: "swift_bundled", version: "0.2.0") + public static let descriptor = NativeExtensionDescriptor.bundled( + resourceBundle: "InKCreLocalFiles_LocalFiles", developmentBundle: .module) public static let sourceType = "extensions.local_files.directory.v1" public static let resolver = "extensions.local_files.file.v2" public static let legacyResolver = "extensions.local_files.file.v1" diff --git a/extensions/local-files/Sources/LocalFilesMac/DirectoryMonitor.swift b/extensions/local-files/Sources/LocalFilesMac/DirectoryMonitor.swift index 7077c44..822336e 100644 --- a/extensions/local-files/Sources/LocalFilesMac/DirectoryMonitor.swift +++ b/extensions/local-files/Sources/LocalFilesMac/DirectoryMonitor.swift @@ -15,23 +15,51 @@ final class DirectoryMonitor { private var watcher: DirectoryWatcher? private var task: Task? private var stopped = false - private(set) var needsScan = false - var isScanning: Bool { task != nil } + private var needsScan = false + typealias Scan = + @Sendable ( + DirectorySource, LocalFileClient, SourceJob?, + @escaping @Sendable (ScanProgress) async -> Void + ) async throws -> DirectoryState + private let scan: Scan init( source: DirectorySource, target: ConnectionTarget, context: PeerHostContext, - root: URL, scoped: Bool, + bookmark: Data, onBookmarkRenewed: (Data) throws -> Void, onState: @escaping (String, Bool) -> Void, - onSources: @escaping ([DirectorySource]) -> Void + onSources: @escaping ([DirectorySource]) -> Void, + scan: @escaping Scan = { source, client, job, progress in + try await DirectoryScanner.scan( + source: source, client: client, job: job, progress: progress) + } ) throws { self.source = source self.target = target self.context = context - self.root = root - self.scoped = scoped self.onState = onState self.onSources = onSources - watcher = try DirectoryWatcher(path: root.path) { [weak self] in self?.schedule() } + self.scan = scan + var stale = false + root = try URL( + resolvingBookmarkData: bookmark, options: [.withSecurityScope, .withoutUI], + relativeTo: nil, bookmarkDataIsStale: &stale) + guard + root.resolvingSymlinksInPath().path + == URL(fileURLWithPath: source.config.path).resolvingSymlinksInPath().path + else { throw PeerError.needsAction("目录位置已改变，请重新选择目录。") } + scoped = root.startAccessingSecurityScopedResource() + do { + if stale { + try onBookmarkRenewed( + root.bookmarkData( + options: [.withSecurityScope], includingResourceValuesForKeys: nil, + relativeTo: nil)) + } + watcher = try DirectoryWatcher(path: root.path) { [weak self] in self?.schedule() } + } catch { + if scoped { root.stopAccessingSecurityScopedResource() } + throw error + } } func stop() async { @@ -50,17 +78,20 @@ final class DirectoryMonitor { func schedule() { guard !stopped, !Task.isCancelled else { return } needsScan = true - guard task == nil else { return } + refresh() + } + + /// Poll shared work and retry failed reconciliation without exposing scheduling state. + func refresh() { + guard !stopped, !Task.isCancelled, task == nil else { return } task = Task { - var status = "正在扫描…" + var status = "" defer { task = nil - if !stopped { onState(status, false) } + if !stopped, !status.isEmpty { onState(status, false) } } guard !stopped, !Task.isCancelled else { return } - onState(status, true) - while needsScan, !stopped, !Task.isCancelled { - needsScan = false + while !stopped, !Task.isCancelled { var job: SourceJob? let client: LocalFileClient do { @@ -76,17 +107,20 @@ final class DirectoryMonitor { } do { let pending = try await client.pendingScans(source) + try Task.checkCancellation() + guard needsScan || !pending.isEmpty else { break } + needsScan = false + onState("正在扫描…", true) for candidate in pending { if let claimed = try await client.claim(candidate, source: source) { job = claimed break } } - let state = try await DirectoryScanner.scan( - source: source, client: client, job: job - ) { [weak self] progress in + let state = try await scan(source, client, job) { [weak self] progress in await self?.report(progress) } + try Task.checkCancellation() if let job { try await client.close( job, status: state.errors == 0 ? "finished" : "failed", state: state) @@ -94,7 +128,6 @@ final class DirectoryMonitor { status = state.errors == 0 ? "正在监视" : "\(state.errors) 个文件需要处理" let updated = try await client.sources() if !stopped, !Task.isCancelled { onSources(updated) } - if !(try await client.pendingScans(source)).isEmpty { needsScan = true } } catch { status = PeerDatabase.isCancellation(error) diff --git a/extensions/local-files/Sources/LocalFilesMac/LocalFilesModel.swift b/extensions/local-files/Sources/LocalFilesMac/LocalFilesModel.swift index 2472361..1eca175 100644 --- a/extensions/local-files/Sources/LocalFilesMac/LocalFilesModel.swift +++ b/extensions/local-files/Sources/LocalFilesMac/LocalFilesModel.swift @@ -217,10 +217,7 @@ public final class LocalFilesModel: ObservableObject { continue } } - if let monitor = monitors[binding.id], !monitor.isScanning { - let jobs = try await client.pendingScans(source) - if !jobs.isEmpty || monitor.needsScan { monitor.schedule() } - } + monitors[binding.id]?.refresh() } for source in found where !bindings.contains(where: { $0.id == key(source.id, connection) }) { @@ -241,41 +238,27 @@ public final class LocalFilesModel: ObservableObject { _ binding: DirectoryBinding, source: DirectorySource, target: ConnectionTarget ) throws { guard running, !Task.isCancelled else { throw CancellationError() } - var stale = false - let root = try URL( - resolvingBookmarkData: binding.bookmark, options: [.withSecurityScope, .withoutUI], - relativeTo: nil, bookmarkDataIsStale: &stale) - guard - root.resolvingSymlinksInPath().path - == URL(fileURLWithPath: source.config.path).resolvingSymlinksInPath().path - else { throw PeerError.needsAction("目录位置已改变，请重新选择目录。") } - let scoped = root.startAccessingSecurityScopedResource() - do { - if stale, let index = bindings.firstIndex(where: { $0.id == binding.id }) { - bindings[index].bookmark = try root.bookmarkData( - options: [.withSecurityScope], includingResourceValuesForKeys: nil, - relativeTo: nil) - try saveBindings() - } - let id = binding.id - let monitor = try DirectoryMonitor( - source: source, target: target, context: context, root: root, scoped: scoped, - onState: { [weak self] message, active in - guard let self else { return } - statuses[id] = message - if active { scanning.insert(id) } else { scanning.remove(id) } - }, - onSources: { [weak self] updated in - guard let self, running, self.target?.id == target.id else { return } - sources = updated - loadRecords() - }) - monitors[id] = monitor - monitor.schedule() - } catch { - if scoped { root.stopAccessingSecurityScopedResource() } - throw error - } + let id = binding.id + let monitor = try DirectoryMonitor( + source: source, target: target, context: context, bookmark: binding.bookmark, + onBookmarkRenewed: { bookmark in + if let index = bindings.firstIndex(where: { $0.id == id }) { + bindings[index].bookmark = bookmark + try saveBindings() + } + }, + onState: { [weak self] message, active in + guard let self else { return } + statuses[id] = message + if active { scanning.insert(id) } else { scanning.remove(id) } + }, + onSources: { [weak self] updated in + guard let self, running, self.target?.id == target.id else { return } + sources = updated + loadRecords() + }) + monitors[id] = monitor + monitor.schedule() } private func stopMonitor(_ id: String) async { diff --git a/scripts/build-app.sh b/scripts/build-app.sh index 2c2cb24..b7e30e5 100644 --- a/scripts/build-app.sh +++ b/scripts/build-app.sh @@ -8,7 +8,8 @@ development_identity="${DEVELOPMENT_SIGNING_IDENTITY:-}" if [[ -z "$development_identity" && -z "${SIGNING_IDENTITY:-}" && -f .local-development-signing-identity ]]; then development_identity="$(< .local-development-signing-identity)" fi -version="0.1.0" +version="$(python3 -c 'import json; print(json.load(open("Application.json"))["version"])')" +build="$(python3 -c 'import json; print(json.load(open("Application.json"))["build"])')" case "$bundle_id" in *[!a-zA-Z0-9.-]*|"") echo "Invalid BUNDLE_ID" >&2; exit 1;; esac if [[ -n "${SIGNING_IDENTITY:-}" && "$bundle_id" == dev.* ]]; then echo "Signed release builds require an explicit stable BUNDLE_ID outside dev.*." >&2 @@ -19,8 +20,8 @@ if [[ -n "${SIGNING_IDENTITY:-}" && -n "$development_identity" ]]; then exit 1 fi -swift build -c "$configuration" --arch arm64 --arch x86_64 -bin_path="$(swift build -c "$configuration" --arch arm64 --arch x86_64 --show-bin-path)" +swift build -c "$configuration" --arch arm64 --arch x86_64 --jobs "${SWIFT_JOBS:-2}" +bin_path="$(swift build -c "$configuration" --arch arm64 --arch x86_64 --jobs "${SWIFT_JOBS:-2}" --show-bin-path)" app="dist/InKCre.app" rm -rf "$app" mkdir -p "$app/Contents/MacOS" @@ -48,7 +49,7 @@ cat > "$app/Contents/Info.plist" <CFBundleIconFileAppIcon.icns CFBundlePackageTypeAPPL CFBundleShortVersionString$version -CFBundleVersion19 +CFBundleVersion$build LSMinimumSystemVersion15.0 NSHighResolutionCapable NSPrincipalClassNSApplication diff --git a/scripts/check-database.py b/scripts/check-database.py new file mode 100644 index 0000000..7283f12 --- /dev/null +++ b/scripts/check-database.py @@ -0,0 +1,114 @@ +"""Run Swift's database/native checks against an explicitly disposable, migrated database. + +Consume a restricted JSON descriptor, or own a temporary PostgreSQL cluster initialized by +impl-py's authoritative lifecycle. No schema is defined or copied by this script. +""" +import argparse +import json +import os +from pathlib import Path +import secrets +import shutil +import socket +import subprocess +import tempfile + +ROOT = Path(__file__).resolve().parents[1] + + +def test_environment(descriptor): + if descriptor.get("disposable") is not True: + raise ValueError("database descriptor must explicitly declare disposable: true") + if descriptor.get("host") not in {"localhost", "127.0.0.1"}: + raise ValueError("database access must use a loopback endpoint") + if descriptor.get("username") != "inkcre_core": + raise ValueError("tests require the admitted inkcre_core role, never an administrator") + port = descriptor.get("port") + if type(port) is not int or not 1 <= port <= 65535: + raise ValueError("descriptor port is invalid") + if not all(isinstance(descriptor.get(key), str) and descriptor[key] + for key in ("database", "password")): + raise ValueError("descriptor database and password are required") + env = os.environ.copy() + for prefix in ("INKCRE_CAPTURE_TEST", "INKCRE_FILES_TEST"): + env.update({f"{prefix}_DATABASE": descriptor["database"], + f"{prefix}_PORT": str(port), f"{prefix}_PASSWORD": descriptor["password"]}) + return env + + +def check(descriptor): + env = test_environment(descriptor) + print("Running Capture and Local Files database/native checks in the disposable database.", flush=True) + subprocess.run(["swift", "test", "--disable-automatic-resolution", "--no-parallel", "--jobs", + env.get("SWIFT_JOBS", "2")], cwd=ROOT, env=env, check=True) + + +def with_cluster(impl_py, postgres_bin): + impl_py = impl_py.resolve() + lifecycle = impl_py / "scripts/database.py" + if not lifecycle.is_file(): + raise ValueError("--impl-py must point to the authoritative impl-py checkout") + def pg(command): + executable = postgres_bin / command if postgres_bin else shutil.which(command) + if not executable or not Path(executable).is_file(): + raise ValueError(f"PostgreSQL 17 tools unavailable: {command}; see README database checks") + return str(executable) + tools = {name: pg(name) for name in ("initdb", "pg_ctl", "createdb")} + revision = subprocess.check_output(["git", "rev-parse", "HEAD"], cwd=impl_py, text=True).strip() + print(f"Initializing disposable database with impl-py {revision}", flush=True) + with tempfile.TemporaryDirectory(prefix="inkcre-swift-db-", dir="/tmp") as temporary: + root = Path(temporary) + data = root / "data" + password = secrets.token_urlsafe(36) + password_file = root / "password" + password_file.write_text(password) + password_file.chmod(0o600) + with socket.socket() as listener: + listener.bind(("127.0.0.1", 0)) + port = listener.getsockname()[1] + env = os.environ.copy() + env.update(PGPASSWORD=password, INKCRE_ENV_FILE="") + subprocess.run([tools["initdb"], "-D", str(data), "-U", "postgres", + "--auth-local=trust", "--auth-host=scram-sha-256", + "--pwfile", str(password_file)], env=env, check=True) + started = False + try: + subprocess.run([tools["pg_ctl"], "-D", str(data), "-l", str(root / "postgres.log"), + "-o", f"-h 127.0.0.1 -p {port} -k {root}", "-w", "start"], + env=env, check=True) + started = True + subprocess.run([tools["createdb"], "-h", "127.0.0.1", "-p", str(port), + "-U", "postgres", "swift_acceptance"], env=env, check=True) + env.update(MIGRATION_DATABASE_URL= + f"postgresql+psycopg://postgres:{password}@127.0.0.1:{port}/swift_acceptance", + CORE_DATABASE_PASSWORD=password, + POSTGREST_DATABASE_PASSWORD=secrets.token_urlsafe(36)) + subprocess.run(["pdm", "run", "python", str(lifecycle), "init", "--profile", "development"], + cwd=impl_py, env=env, check=True) + subprocess.run(["pdm", "run", "python", str(lifecycle), "ready", "--profile", "development"], + cwd=impl_py, env=env, check=True) + check(dict(disposable=True, host="127.0.0.1", port=port, + database="swift_acceptance", username="inkcre_core", password=password)) + finally: + if started: + subprocess.run([tools["pg_ctl"], "-D", str(data), "-m", "fast", "-w", "stop"], + check=True) + + +def main(): + parser = argparse.ArgumentParser(description=__doc__) + source = parser.add_mutually_exclusive_group(required=True) + source.add_argument("--descriptor", type=Path, help="0600 JSON describing an exclusively owned disposable database") + source.add_argument("--impl-py", type=Path, help="initialized PDM checkout owning migrations; creates a new temporary cluster") + parser.add_argument("--postgres-bin", type=Path, help="PostgreSQL 17 bin directory (pgvector must be installed)") + args = parser.parse_args() + if args.descriptor: + if args.descriptor.stat().st_mode & 0o077: + parser.error("database descriptor must be private (chmod 600)") + check(json.loads(args.descriptor.read_text())) + else: + with_cluster(args.impl_py, args.postgres_bin) + + +if __name__ == "__main__": + main() diff --git a/scripts/check.sh b/scripts/check.sh index 77056ae..cdf0a70 100755 --- a/scripts/check.sh +++ b/scripts/check.sh @@ -11,4 +11,5 @@ git ls-files --cached --others --exclude-standard -z -- \ 'Packages/**/*.swift' 'extensions/**/*.swift' \ ':!extensions/capture/Sources/CaptureMac/Vendor/**' \ | xargs -0 swift format lint --strict -swift test --disable-automatic-resolution +# AppKit suites share the process-wide application/menu/focus state. +swift test --disable-automatic-resolution --no-parallel --jobs "${SWIFT_JOBS:-2}" diff --git a/scripts/database-source-revision b/scripts/database-source-revision new file mode 100644 index 0000000..8327afb --- /dev/null +++ b/scripts/database-source-revision @@ -0,0 +1 @@ +9031f7e308c7f680ed8bda3b37c204e5bca62e3d From ba405968a38724c52100489b0b15d8d5b7f9ee1c Mon Sep 17 00:00:00 2001 From: Lan_zhijiang Date: Sat, 31 Oct 2026 19:55:51 +0800 Subject: [PATCH 3/3] =?UTF-8?q?fix(database):=20=E5=AE=8C=E6=88=90?= =?UTF-8?q?=E5=8F=96=E6=B6=88=E5=90=8E=E7=9A=84=E8=BF=9E=E6=8E=A5=E6=94=B6?= =?UTF-8?q?=E5=B0=BE=E5=B9=B6=E7=BA=A6=E6=9D=9F=E9=AA=8C=E6=94=B6=E8=BF=9B?= =?UTF-8?q?=E7=A8=8B=E5=AF=BF=E5=91=BD?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .github/workflows/ci.yml | 2 + .../PeerCore/BundledParticipation.swift | 8 +- .../PeerCore/Sources/PeerCore/InfoBase.swift | 26 ++-- .../Sources/PeerCore/PeerConnection.swift | 69 +++++++++++ .../Sources/PeerCore/PeerDatabase.swift | 25 ++-- .../Sources/PeerCore/SourceJobs.swift | 32 +++-- README.md | 6 +- Tests/CaptureTests/LivePeerTests.swift | 33 +++-- .../CaptureTests/PeerCancellationTests.swift | 117 ++++++++++++++++++ .../Sources/Capture/CaptureClient.swift | 14 ++- .../LocalFiles/FileReconciliation.swift | 15 ++- .../Sources/LocalFiles/LocalFileClient.swift | 39 ++++-- scripts/check-database.py | 51 +++++++- scripts/check.sh | 1 + 14 files changed, 377 insertions(+), 61 deletions(-) create mode 100644 Packages/PeerCore/Sources/PeerCore/PeerConnection.swift create mode 100644 Tests/CaptureTests/PeerCancellationTests.swift diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 3271dc6..efe5c18 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -15,6 +15,7 @@ jobs: check: name: impl-swift checks runs-on: macos-15 + timeout-minutes: 25 env: DEVELOPER_DIR: /Applications/Xcode_26.2.app/Contents/Developer steps: @@ -49,6 +50,7 @@ jobs: 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" diff --git a/Packages/PeerCore/Sources/PeerCore/BundledParticipation.swift b/Packages/PeerCore/Sources/PeerCore/BundledParticipation.swift index d230f56..92e2e93 100644 --- a/Packages/PeerCore/Sources/PeerCore/BundledParticipation.swift +++ b/Packages/PeerCore/Sources/PeerCore/BundledParticipation.swift @@ -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 选择的扩展版本不在当前应用中，请在扩展设置中对该信息库启用本机版本。") } diff --git a/Packages/PeerCore/Sources/PeerCore/InfoBase.swift b/Packages/PeerCore/Sources/PeerCore/InfoBase.swift index 4271261..df77a41 100644 --- a/Packages/PeerCore/Sources/PeerCore/InfoBase.swift +++ b/Packages/PeerCore/Sources/PeerCore/InfoBase.swift @@ -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))", @@ -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 { @@ -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 { diff --git a/Packages/PeerCore/Sources/PeerCore/PeerConnection.swift b/Packages/PeerCore/Sources/PeerCore/PeerConnection.swift new file mode 100644 index 0000000..c5aa347 --- /dev/null +++ b/Packages/PeerCore/Sources/PeerCore/PeerConnection.swift @@ -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( + 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 +} diff --git a/Packages/PeerCore/Sources/PeerCore/PeerDatabase.swift b/Packages/PeerCore/Sources/PeerCore/PeerDatabase.swift index 7eac589..2d98ebb 100644 --- a/Packages/PeerCore/Sources/PeerCore/PeerDatabase.swift +++ b/Packages/PeerCore/Sources/PeerCore/PeerDatabase.swift @@ -42,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() } @@ -85,16 +85,24 @@ public actor PeerDatabase { /// Keep row consumption and transactions inside this scope so cancellation closes their lease. public nonisolated func withConnection( 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: { @@ -119,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) } diff --git a/Packages/PeerCore/Sources/PeerCore/SourceJobs.swift b/Packages/PeerCore/Sources/PeerCore/SourceJobs.swift index d666073..1fbba9c 100644 --- a/Packages/PeerCore/Sources/PeerCore/SourceJobs.swift +++ b/Packages/PeerCore/Sources/PeerCore/SourceJobs.swift @@ -9,7 +9,7 @@ public struct SourceJob: Sendable { public enum SourceJobs { public static let type = "core.source.collect.v1" - public static func submit(source: Int, on connection: PostgresConnection) async throws -> Int { + public static func submit(source: Int, on connection: PeerConnection) async throws -> Int { let parameters = "{\"source\":\(source),\"config\":{}}" let rows = try await connection.query( """ @@ -17,11 +17,15 @@ public enum SourceJobs { SELECT id,\(parameters)::jsonb,default_timeout_seconds,'pending' FROM inkcre.job_types WHERE id=\(type) 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.needsAction("信息库尚未登记来源扫描命令。") } - public static func pending(source: Int, on client: PostgresConnection) async throws -> [Int] { + public static func pending(source: Int, on client: PeerConnection) async throws -> [Int] { try await client.query( "UPDATE inkcre.jobs SET status='timed_out' WHERE type=\(type) AND status='running' AND parameters->'source'=to_jsonb(\(source)::bigint) AND started_at + timeout_seconds * interval '1 second' < now()", logger: PeerDatabase.logger) @@ -32,12 +36,16 @@ public enum SourceJobs { "SELECT id FROM inkcre.jobs WHERE type=\(type) AND status='pending' AND parameters->'source'=to_jsonb(\(source)::bigint) ORDER BY id", logger: PeerDatabase.logger) var ids: [Int] = [] - for try await id in rows.decode(Int.self) { ids.append(id) } + for row in rows { + try Task.checkCancellation() + let id = try row.decode(Int.self) + ids.append(id) + } return ids } /// Eligibility is checked under the Source lock by the capability before claiming. - public static func claim(_ id: Int, source: Int, on connection: PostgresConnection) async throws + public static func claim(_ id: Int, source: Int, on connection: PeerConnection) async throws -> SourceJob? { let rows = try await connection.query( @@ -47,19 +55,25 @@ public enum SourceJobs { AND COALESCE(parameters->'config','{}'::jsonb)='{}'::jsonb RETURNING id,started_at + timeout_seconds * interval '1 second' """, logger: PeerDatabase.logger) - for try await (id, deadline) in rows.decode((Int, Date).self) { + for row in rows { + try Task.checkCancellation() + let (id, deadline) = try row.decode((Int, Date).self) + return SourceJob(id: id, deadline: deadline) } return nil } - public static func check(_ job: SourceJob, on client: PostgresConnection) async throws { + public static func check(_ job: SourceJob, on client: PeerConnection) async throws { try Task.checkCancellation() if Date() >= job.deadline { throw PeerError.needsAction("扫描已超时。") } let rows = try await client.query( "SELECT status::text,abort_requested FROM inkcre.jobs WHERE id=\(job.id)", logger: PeerDatabase.logger) - for try await (status, aborted) in rows.decode((String, Bool).self) { + for row in rows { + try Task.checkCancellation() + let (status, aborted) = try row.decode((String, Bool).self) + guard status == "running", !aborted else { throw CancellationError() } return } @@ -67,7 +81,7 @@ public enum SourceJobs { } public static func close( - _ job: SourceJob, status: String, state: String, on client: PostgresConnection + _ job: SourceJob, status: String, state: String, on client: PeerConnection ) async throws { guard ["finished", "failed", "timed_out", "aborted"].contains(status) else { throw PeerError.invalid("无效的任务终态。") diff --git a/README.md b/README.md index 8d81ea6..c4c6e4c 100644 --- a/README.md +++ b/README.md @@ -51,7 +51,7 @@ python3 scripts/check-database.py --impl-py /path/to/impl-py \ --postgres-bin "$(brew --prefix postgresql@17)/bin" ``` -此入口仅在临时目录创建独立 PostgreSQL cluster、随机回环端口和随机凭据，调用权威 `scripts/database.py init --profile development`／`ready`，开启 Capture 和 Local Files 的全部数据库及隐藏原生窗口旅程，结束后停止并删除该临时 cluster。不会使用现有 development／preview 库，也不启动系统 PostgreSQL 服务。缺工具、迁移失败或验收失败均返回非零。 +此入口仅在临时目录创建独立 PostgreSQL cluster、随机回环端口和随机凭据，调用权威 `scripts/database.py init --profile development`／`ready`，开启 Capture 和 Local Files 的全部数据库及隐藏原生窗口旅程，结束后停止并删除该临时 cluster。不会使用现有 development／preview 库，也不启动系统 PostgreSQL 服务。缺工具、迁移失败或验收失败均返回非零。测试输出即时刷新；Swift 测试进程最多运行 10 分钟，超时会终止其子进程及后代进程组、清理自有 cluster 并返回失败，不能视为验收通过。 已有独占且完成权威迁移的一次性数据库时，可以消费权限为 0600 的 JSON descriptor： @@ -61,6 +61,10 @@ python3 scripts/check-database.py --descriptor /private/path/swift-test.json descriptor 必须声明 `disposable: true`，包含 `host`（回环）、`port`、`database`、`username: "inkcre_core"` 和 `password`；调用者负责该库的独占性与销毁。凭据通过子进程环境传入，不 source dotenv，也不放入命令参数。脚本不修改 schema；`INKCRE_LAYOUT_EVIDENCE_DIR` 可指定已有目录保存合成界面的实际渲染图。这些检查不代表系统授权提示、真实跨应用选区、多屏实体环境或公证已通过。 +PeerCore 的 `withConnection` 借出限于作用域内使用的连接；`query` 明确收集 `[PostgresRow]` 后返回，调用方逐行解码并检查取消，不承诺流式背压。检查连接状态与发出查询必须留在同一次 NIO event loop 执行中，避免关闭后的驱动 promise 无人完成；不要绕回原始异步驱动查询。当前目录记录读取仍会同时持有查询行和最终记录数组，大结果的峰值内存不是原来的流式水平。 + +原生会话以运行角色设置 `client_connection_check_interval=1000`，在项目使用的 PostgreSQL 17/macOS 或 Linux 上，每秒检查执行中的查询是否已失去客户端；代价是长查询增加周期性检查。它检测内核已报告的断连，不是同步回滚确认；`statement_timeout=30000` 继续限制无法及时检测的服务器工作。具体平台与网络边界见 [PostgreSQL 17 连接配置](https://www.postgresql.org/docs/17/runtime-config-connection.html#GUC-CLIENT-CONNECTION-CHECK-INTERVAL)。 + ## 连接信息库 连接设置使用 PostgreSQL 主机、端口、库名、独立运行角色及密码，可选 Web 地址。远端使用 TLS 并校验证书；只有回环开发地址允许关闭 TLS。密码存于 Keychain，角色拥有已准入数据库协议权限，不使用迁移管理员或 JWT 签名密钥。 diff --git a/Tests/CaptureTests/LivePeerTests.swift b/Tests/CaptureTests/LivePeerTests.swift index 8688517..0231ebb 100644 --- a/Tests/CaptureTests/LivePeerTests.swift +++ b/Tests/CaptureTests/LivePeerTests.swift @@ -94,11 +94,16 @@ func nativeTransactionCommitsOriginalImageOnceAndRecoversAfterRestart() async th _ in true } await #expect(throws: (any Error).self) { - try await pg.withTransaction(logger: PeerDatabase.logger) { connection in - _ = try await InfoBase.persist( - blocks: [GraphBlock(id: -1, resolver: "core.text.v1", content: "must rollback")], - relations: [GraphRelation(from_: -1, to_: -999, content: "invalid")], blobs: [:], - on: connection) + try await database.withConnection(for: target, password: password) { connection in + try await connection.withTransaction(logger: PeerDatabase.logger) { connection in + _ = try await InfoBase.persist( + blocks: [ + GraphBlock(id: -1, resolver: "core.text.v1", content: "must rollback") + ], + relations: [GraphRelation(from_: -1, to_: -999, content: "invalid")], + blobs: [:], + on: connection) + } } } let after = try await pg.query( @@ -207,9 +212,8 @@ func cancellingAStalledQueryReleasesItsConnectionAndRollsBack() async throws { try await connection.query( "INSERT INTO inkcre.blocks(resolver,content) VALUES ('core.text.v1',\(marker))", logger: PeerDatabase.logger) - let rows = try await connection.query( + try await connection.query( PostgresQuery(unsafeSQL: sleepingQuery), logger: PeerDatabase.logger) - for try await _ in rows {} } } } @@ -229,14 +233,23 @@ func cancellingAStalledQueryReleasesItsConnectionAndRollsBack() async throws { do { try await blocked.value Issue.record("Cancelled transaction unexpectedly completed") - } catch {} + } catch { #expect(error is CancellationError) } #expect(start.duration(to: .now) < .seconds(2)) + var serverStillRunning = true + for _ in 0..<150 { + let rows = try await pg.query( + "SELECT count(*)::integer FROM pg_stat_activity WHERE datname=current_database() AND query=\(sleepingQuery)" + ) + for try await count in rows.decode(Int.self) { serverStillRunning = count > 0 } + if !serverStillRunning { break } + try await Task.sleep(for: .milliseconds(20)) + } + #expect(!serverStillRunning) let count = try await database.withConnection(for: target, password: password) { connection in let rows = try await connection.query( "SELECT count(*)::integer FROM inkcre.blocks WHERE content=\(marker)", logger: PeerDatabase.logger) - for try await count in rows.decode(Int.self) { return count } - return -1 + return try rows.first?.decode(Int.self) ?? -1 } #expect(count == 0) await database.shutdown() diff --git a/Tests/CaptureTests/PeerCancellationTests.swift b/Tests/CaptureTests/PeerCancellationTests.swift new file mode 100644 index 0000000..7e94b38 --- /dev/null +++ b/Tests/CaptureTests/PeerCancellationTests.swift @@ -0,0 +1,117 @@ +import Foundation +import PeerCore +import PostgresNIO +import Testing + +private actor QueryBoundary { + private var arrived = false + private var arrival: CheckedContinuation? + private var release: CheckedContinuation? + + func pause() async { + arrived = true + arrival?.resume() + arrival = nil + await withCheckedContinuation { release = $0 } + } + func waitUntilPaused() async { + if !arrived { await withCheckedContinuation { arrival = $0 } } + } + func resume() { + release?.resume() + release = nil + } +} + +@Suite(.enabled(if: ProcessInfo.processInfo.environment["INKCRE_CAPTURE_TEST_DATABASE"] != nil)) +struct PeerCancellationTests { + enum Interruption: Sendable, CaseIterable { case cancel, shutdown } + + @Test(arguments: Interruption.allCases) + func interruptionBetweenQueriesSettlesAndRollsBack(_ interruption: Interruption) async throws { + let (target, password) = try configuration() + let database = PeerDatabase() + let boundary = QueryBoundary() + let marker = "between-queries-\(UUID())" + let operation = Task { + try await database.withConnection(for: target, password: password) { connection in + try await connection.withTransaction { connection in + try await connection.query( + "INSERT INTO inkcre.blocks(resolver,content) VALUES ('core.text.v1',\(marker))" + ) + await boundary.pause() + // Cancellation attempts another body query; shutdown attempts COMMIT itself. + if interruption == .cancel { try await connection.query("SELECT 1") } + } + } + } + await boundary.waitUntilPaused() + let start = ContinuousClock.now + switch interruption { + case .cancel: operation.cancel() + case .shutdown: await database.shutdown() + } + await boundary.resume() + do { + try await operation.value + Issue.record("Interrupted transaction unexpectedly completed") + } catch { + if interruption == .cancel { + #expect(error is CancellationError) + } else { + if case .connection = PeerDatabase.failure(error) { + } else { + Issue.record("A closed connection must remain retryable") + } + if case .outcomeUnknown = PeerDatabase.failure(error, dispatched: true) { + } else { + Issue.record("A dispatched write must retain result verification") + } + } + } + #expect(start.duration(to: .now) < .seconds(2)) + let count = try await database.withConnection(for: target, password: password) { + connection in + try await connection.query( + "SELECT count(*)::integer FROM inkcre.blocks WHERE content=\(marker)" + ).first?.decode(Int.self) + } + #expect(count == 0) + await database.shutdown() + } + + @Test func ordinaryDatabaseFailureRollsBackAndKeepsItsError() async throws { + let (target, password) = try configuration() + let database = PeerDatabase() + let marker = "rollback-error-\(UUID())" + try await database.withConnection(for: target, password: password) { connection in + do { + try await connection.withTransaction { connection in + try await connection.query( + "INSERT INTO inkcre.blocks(resolver,content) VALUES ('core.text.v1',\(marker))" + ) + try await connection.query("SELECT 1/0") + } + Issue.record("Expected PostgreSQL division-by-zero error") + } catch let error as PSQLError { + #expect(error.serverInfo?[.sqlState] == "22012") + } + // Same borrowed connection remains usable and the earlier write did not survive. + let rows = try await connection.query( + "SELECT count(*)::integer FROM inkcre.blocks WHERE content=\(marker)") + #expect(try rows.first?.decode(Int.self) == 0) + } + await database.shutdown() + } + + private func configuration() throws -> (ConnectionTarget, String) { + let env = ProcessInfo.processInfo.environment + return ( + ConnectionTarget( + host: "127.0.0.1", port: Int(env["INKCRE_CAPTURE_TEST_PORT"] ?? "5432")!, + database: try #require(env["INKCRE_CAPTURE_TEST_DATABASE"]), + username: "inkcre_core", tls: .disable), + try #require(env["INKCRE_CAPTURE_TEST_PASSWORD"]) + ) + } +} diff --git a/extensions/capture/Sources/Capture/CaptureClient.swift b/extensions/capture/Sources/Capture/CaptureClient.swift index a0184ff..e5bd231 100644 --- a/extensions/capture/Sources/Capture/CaptureClient.swift +++ b/extensions/capture/Sources/Capture/CaptureClient.swift @@ -34,11 +34,14 @@ public struct CaptureClient: CaptureTransport { """ } - private static func result(from rows: PostgresRowSequence, id: UUID, digest: String) - async throws + private static func result(from rows: [PostgresRow], id: UUID, digest: String) + throws -> CaptureResult? { - for try await (saved, root, date) in rows.decode((String, Int, Date).self) { + for row in rows { + try Task.checkCancellation() + let (saved, root, date) = try row.decode((String, Int, Date).self) + guard saved == digest else { throw PeerError.needsAction("提交身份与已保存材料冲突，原材料已保留。") } return CaptureResult(submissionID: id, rootBlockID: root, committedAt: date) } @@ -77,7 +80,10 @@ public struct CaptureClient: CaptureTransport { let rows = try await connection.query( "SELECT created_at FROM inkcre.blocks WHERE id=\(root)", logger: PeerDatabase.logger) - for try await date in rows.decode(Date.self) { + for row in rows { + try Task.checkCancellation() + let date = try row.decode(Date.self) + return CaptureResult( submissionID: record.id, rootBlockID: root, committedAt: date) } diff --git a/extensions/local-files/Sources/LocalFiles/FileReconciliation.swift b/extensions/local-files/Sources/LocalFiles/FileReconciliation.swift index 70ea45b..04119df 100644 --- a/extensions/local-files/Sources/LocalFiles/FileReconciliation.swift +++ b/extensions/local-files/Sources/LocalFiles/FileReconciliation.swift @@ -98,7 +98,10 @@ extension LocalFileClient { let existing = try await connection.query( "SELECT b.id FROM inkcre.relations r JOIN inkcre.blocks b ON b.id=r.to_ WHERE r.from_=\(anchor) AND r.content='manages' AND b.resolver IN (\(LocalFilesContract.resolver),\(LocalFilesContract.legacyResolver)) AND b.content::jsonb->>'location'=\(metadata.location)", logger: PeerDatabase.logger) - for try await _ in existing.decode(Int.self) { + for row in existing { + try Task.checkCancellation() + let _ = try row.decode(Int.self) + throw PeerError.needsAction("资料已存在，正在重新核对目录。") } let root = try await InfoBase.createBlock( @@ -142,7 +145,7 @@ extension LocalFileClient { } } - private func currentRecord(_ id: Int, anchor: Int, on connection: PostgresConnection) + private func currentRecord(_ id: Int, anchor: Int, on connection: PeerConnection) async throws -> FileRecord { @@ -155,9 +158,11 @@ extension LocalFileClient { WHERE b.id=\(id) AND b.resolver IN (\(LocalFilesContract.resolver),\(LocalFilesContract.legacyResolver)) FOR UPDATE OF b """, logger: PeerDatabase.logger) var records: [FileRecord] = [] - for try await (resolver, content, body, bodyResolver, storage) in rows.decode( - (String, String, Int?, String?, Int?).self) - { + for row in rows { + try Task.checkCancellation() + let (resolver, content, body, bodyResolver, storage) = try row.decode( + (String, String, Int?, String?, Int?).self) + records.append( try storedFile( id: id, resolver: resolver, content: content, bodyID: body, diff --git a/extensions/local-files/Sources/LocalFiles/LocalFileClient.swift b/extensions/local-files/Sources/LocalFiles/LocalFileClient.swift index ca5dba0..c2a6487 100644 --- a/extensions/local-files/Sources/LocalFiles/LocalFileClient.swift +++ b/extensions/local-files/Sources/LocalFiles/LocalFileClient.swift @@ -51,9 +51,11 @@ public struct LocalFileClient: Sendable { "SELECT id,config::text,block,state::text FROM inkcre.sources WHERE type=\(LocalFilesContract.sourceType) AND config->>'peer'=\(target.peerID.uuidString.lowercased()) ORDER BY id", logger: PeerDatabase.logger) var sources: [DirectorySource] = [] - for try await (id, config, block, state) in rows.decode( - (Int, String, Int?, String).self) - { + for row in rows { + try Task.checkCancellation() + let (id, config, block, state) = try row.decode( + (Int, String, Int?, String).self) + sources.append(try decodeSource(id: id, config: config, block: block, state: state)) } return sources @@ -74,13 +76,19 @@ public struct LocalFileClient: Sendable { let existing = try await connection.query( "SELECT id FROM inkcre.sources WHERE type=\(LocalFilesContract.sourceType) AND config->>'peer'=\(target.peerID.uuidString.lowercased()) AND config->>'path'=\(path)", logger: PeerDatabase.logger) - for try await _ in existing.decode(Int.self) { + for row in existing { + try Task.checkCancellation() + let _ = try row.decode(Int.self) + throw PeerError.needsAction("这个目录已经添加。") } let rows = try await connection.query( "INSERT INTO inkcre.sources(type,nickname,config) VALUES (\(LocalFilesContract.sourceType),\(URL(fileURLWithPath: path).lastPathComponent),\(serialized)::jsonb) RETURNING id", logger: PeerDatabase.logger) - for try await id in rows.decode(Int.self) { + for row in rows { + try Task.checkCancellation() + let id = try row.decode(Int.self) + struct Anchor: Encodable { let id: Int let type: String @@ -139,9 +147,11 @@ public struct LocalFileClient: Sendable { """, logger: PeerDatabase.logger) var records: [FileRecord] = [] var seen: Set = [] - for try await (id, resolver, content, bodyID, bodyResolver, storage) in rows.decode( - (Int, String, String, Int?, String?, Int?).self) - { + for row in rows { + try Task.checkCancellation() + let (id, resolver, content, bodyID, bodyResolver, storage) = try row.decode( + (Int, String, String, Int?, String?, Int?).self) + guard seen.insert(id).inserted else { throw PeerError.invalid("文件资料存在重复正文关系。") } records.append( try storedFile( @@ -158,13 +168,17 @@ public struct LocalFileClient: Sendable { let rows = try await client.query( "SELECT content FROM inkcre.blocks WHERE id=\(body) AND resolver=\(record.bodyResolver) AND storage IS NULL", logger: PeerDatabase.logger) - for try await text in rows.decode(String.self) { return text } + for row in rows { + try Task.checkCancellation() + let text = try row.decode(String.self) + return text + } throw PeerError.needsAction("正文已移除，请刷新文件列表。") } } func locked( - _ id: Int, on connection: PostgresConnection, expected: DirectoryConfig? = nil, + _ id: Int, on connection: PeerConnection, expected: DirectoryConfig? = nil, requireEnabled: Bool = true ) async throws -> DirectorySource { try await PeerDatabase.requireSelection( @@ -172,7 +186,10 @@ public struct LocalFileClient: Sendable { let rows = try await connection.query( "SELECT id,config::text,block,state::text FROM inkcre.sources WHERE id=\(id) AND type=\(LocalFilesContract.sourceType) FOR UPDATE", logger: PeerDatabase.logger) - for try await (id, config, block, state) in rows.decode((Int, String, Int?, String).self) { + for row in rows { + try Task.checkCancellation() + let (id, config, block, state) = try row.decode((Int, String, Int?, String).self) + let source = try decodeSource(id: id, config: config, block: block, state: state) if let expected, source.config != expected || (requireEnabled && !source.config.enabled) { diff --git a/scripts/check-database.py b/scripts/check-database.py index 7283f12..a90f209 100644 --- a/scripts/check-database.py +++ b/scripts/check-database.py @@ -8,10 +8,12 @@ import os from pathlib import Path import secrets +import signal import shutil import socket import subprocess import tempfile +import time ROOT = Path(__file__).resolve().parents[1] @@ -30,17 +32,62 @@ def test_environment(descriptor): for key in ("database", "password")): raise ValueError("descriptor database and password are required") env = os.environ.copy() + env["NSUnbufferedIO"] = "YES" for prefix in ("INKCRE_CAPTURE_TEST", "INKCRE_FILES_TEST"): env.update({f"{prefix}_DATABASE": descriptor["database"], f"{prefix}_PORT": str(port), f"{prefix}_PASSWORD": descriptor["password"]}) return env +def run_tests(command, env, timeout=600): + process = subprocess.Popen(command, cwd=ROOT, env=env, start_new_session=True) + try: + status = process.wait(timeout=timeout) + except (subprocess.TimeoutExpired, KeyboardInterrupt): + print("Database/native checks did not finish; stopping the test process group.", flush=True) + # SwiftPM's testing helper creates its own process group. Snapshot only descendants + # of our child before stopping the parent, so orphaned helpers remain identifiable. + table = [tuple(map(int, row.split())) for row in subprocess.check_output( + ["ps", "-axo", "pid=,ppid=,pgid="], text=True).splitlines()] + owned = {process.pid} + while True: + children = {pid for pid, parent, _ in table if parent in owned} + if children <= owned: + break + owned.update(children) + groups = {process.pid} | {group for pid, _, group in table + if pid in owned and group in owned} + for sig in (signal.SIGTERM, signal.SIGKILL): + for group in groups: + try: + os.killpg(group, sig) + except ProcessLookupError: + pass + try: + process.wait(timeout=10) + except subprocess.TimeoutExpired: + pass + deadline = time.monotonic() + 5 + while groups: + for group in tuple(groups): + try: + os.killpg(group, 0) + except ProcessLookupError: + groups.remove(group) + if groups and time.monotonic() >= deadline: + raise RuntimeError("test process groups did not terminate after SIGKILL") + if groups: + time.sleep(0.05) + raise + if status: + raise subprocess.CalledProcessError(status, command) + + def check(descriptor): env = test_environment(descriptor) print("Running Capture and Local Files database/native checks in the disposable database.", flush=True) - subprocess.run(["swift", "test", "--disable-automatic-resolution", "--no-parallel", "--jobs", - env.get("SWIFT_JOBS", "2")], cwd=ROOT, env=env, check=True) + run_tests(["swift", "test", "--disable-automatic-resolution", "--no-parallel", "--jobs", + env.get("SWIFT_JOBS", "2")], env) def with_cluster(impl_py, postgres_bin): diff --git a/scripts/check.sh b/scripts/check.sh index cdf0a70..79dfbe5 100755 --- a/scripts/check.sh +++ b/scripts/check.sh @@ -11,5 +11,6 @@ git ls-files --cached --others --exclude-standard -z -- \ 'Packages/**/*.swift' 'extensions/**/*.swift' \ ':!extensions/capture/Sources/CaptureMac/Vendor/**' \ | xargs -0 swift format lint --strict +export NSUnbufferedIO=YES # AppKit suites share the process-wide application/menu/focus state. swift test --disable-automatic-resolution --no-parallel --jobs "${SWIFT_JOBS:-2}"