From 33713cf74ab05fb41c452374b66bc19fe7cffeac Mon Sep 17 00:00:00 2001 From: reviewer Date: Sat, 10 Oct 2026 08:47:53 +0800 Subject: [PATCH] fix(collector): ignore shutdown closed signal in stop stop set stopping and then joined the live capture and upload loops. A turn that observed stopping threw CollectorRuntimeError.closed, and when that beat CancellationError, stop reported a producer failure. swift-unit hit this on 2026-10-08 in the replay-loss and start-loop tests. Background turns now throw CancellationError once stopping is set. stop also treats an already-thrown closed error from the joined loop or cycle as the same shutdown signal. runOnce after stop still throws closed. invalidConfiguration and reconciliationRequired still fail the join. --- CHANGELOG.md | 46 +++++++++++ MEMO.md | 2 + .../CollectorRuntime.swift | 41 ++++++++-- .../CollectorRuntimeTests.swift | 77 +++++++++++++++++++ 4 files changed, 159 insertions(+), 7 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 96d716aa..30727b5a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -3,6 +3,52 @@ All notable changes to this project will be documented in this file. Format based on [Keep a Changelog](https://keepachangelog.com/). +## Collector stop no longer fails on its own shutdown signal (2026-10-10) + +**Failure.** Three `swift-unit` jobs on 2026-10-08 failed with +`caught error: "closed"` and passed on rerun: + +- job `113128319766` (PR #457 attempt 1): + `CollectorRuntimeTests.testReplayLossDuringStartKeepsRuntimeAndOtherRootsRunning_repro` + at `CollectorRuntimeTests.swift:649` +- job `113147173237` (PR #458 attempt 1): + `CollectorRuntimeTests.testStartLoopPublishesAndStopJoinsBeforeReopen` + at `CollectorRuntimeTests.swift:2098` +- job `113152487959` (PR #458 attempt 2): the replay-loss test again, same line + +`CollectorRuntime.stop` sets `stopping` and then joins the live capture and +upload loops. A turn that observes `stopping` used to throw +`CollectorRuntimeError.closed`. When that error won the race against +`CancellationError`, `stop` treated it as a producer failure and the test +failed. `runOnce` after a finished `stop` is still supposed to throw +`.closed`. `stop` still rethrows `invalidConfiguration` and +`reconciliationRequired`. + +**Fix.** `macos/EngramCollectorCore/CollectorRuntime.swift`: background +capture and upload turns throw `CancellationError` once `stopping` is set. +`stop` also treats `.closed` from the joined loop or cycle as the same +shutdown signal, so an in-flight turn that already threw `.closed` cannot +fail the join. `macos/EngramServiceCoreTests/CollectorRuntimeTests.swift` +adds `testStopDuringCaptureTurnStartDoesNotSurfaceClosed_repro`. The repro +holds the capture turn, sets `stopping` before cancel, then calls `stop`. +The first version of that test passed in 0.112s because cancel completed +the loop before the guard ran. The revised repro failed with +`caught error: "closed"` at `CollectorRuntimeTests.swift:2144` before this +fix. + +**Verification.** From `macos/`: + +```bash +xcodebuild test -project Engram.xcodeproj -scheme EngramServiceCore \ + -destination 'platform=macOS' CODE_SIGNING_ALLOWED=NO \ + -only-testing:EngramServiceCoreTests/CollectorRuntimeTests +``` + +`CollectorRuntimeTests`: 82 tests, 0 failures, 31.6s (`/tmp/engram-flake-class.log`). +That includes the new repro, the two CI cases, the post-stop `runOnce` +`.closed` check, and the invalid-settings stop failure. Full `swift-unit` +was not rerun. + ## Web identity sign-in merged (#457) and deployed to the HQ receiver (2026-10-08) Owner authorization "GO" was given in-session for the chain commit, PR, diff --git a/MEMO.md b/MEMO.md index ba2daae4..0516ff83 100644 --- a/MEMO.md +++ b/MEMO.md @@ -4,6 +4,8 @@ ### 2026-10-08 +- [修复] 采集器 `stop` 不再把自身停机信号 `CollectorRuntimeError.closed` 当成失败。2026-10-08 有 3 个 `swift-unit` 因此失败后重跑才过。详见 `CHANGELOG.md`。 +- [验证] 本地 `CollectorRuntimeTests` 82 项、0 失败（含新复现 `testStopDuringCaptureTurnStartDoesNotSurfaceClosed_repro`）。完整 `swift-unit` 由该修复的 PR 跑。 - [合并] PR #457（Web 以 Tailscale 身份登录，第一阶段）以普通合并进 main（`1e3b5c34`）；`swift-unit` 因采集器启动时序偶发失败一次，重跑通过。 - [部署] HQ receiver 切到包 `remote-server-main-1e3b5c34-20261008`（verify-only 通过，本地回环身份模式冒烟通过）；plist 已备份，环境变量加 AUTH/VIEWERS/EDITORS、删两项口令；04:07:31 UTC 重启，1.5 秒内健康 200，新 PID 98581。回滚：恢复备份 plist 并 bootstrap，旧包保留。详见 `CHANGELOG.md`。 - [验证] 从 MacBook 与 HQ 经 8443 验证：`POST {}` 204 并发 cookie、状态返回 `zzbhlx@gmail.com` 且可写、会话读取 200、伪造头被 serve 替换、口令体 400。现在打开 Web 只需在 tailnet 设备访问该地址，无需口令。摄取未受影响（arrivals 41,700）。 diff --git a/macos/EngramCollectorCore/CollectorRuntime.swift b/macos/EngramCollectorCore/CollectorRuntime.swift index 95ebff66..326644f7 100644 --- a/macos/EngramCollectorCore/CollectorRuntime.swift +++ b/macos/EngramCollectorCore/CollectorRuntime.swift @@ -101,6 +101,20 @@ public actor CollectorRuntime { private var blockedStreaks: [Int: Int] = [:] private var blockedRoots = Set() private var reportedBlockedRoots = Set() + /// Awaited immediately before a background capture turn checks `stopping`. + /// Tests hold this until `stop` has entered; production leaves it nil. + private var testCaptureTurnBarrier: (@Sendable () async -> Void)? + + func setTestCaptureTurnBarrier(_ barrier: (@Sendable () async -> Void)?) { + testCaptureTurnBarrier = barrier + } + + /// Marks shutdown without cancelling the loop, so a turn already inside the + /// runtime can observe `stopping` and fail the loop with `.closed` before + /// `stop` joins it. + func testBeginStopping() { + stopping = true + } /// Production uses the native FSEvents stream; tests inject a fake stream. typealias EventStreamFactory = @Sendable (CollectorEventStreamRequest, CollectorEventIngressBudget) @@ -272,9 +286,14 @@ public actor CollectorRuntime { private func runCaptureTurn( now: Int64, historyWaiting: [Int: Bool] ) async throws -> (hints: CollectorCaptureWorkHints, historyWaiting: [Int: Bool]) { - guard !stopping, !closed, let owner, let worker else { throw CollectorRuntimeError.closed } - guard now >= 0 else { throw CollectorRuntimeError.invalidConfiguration } + if let barrier = testCaptureTurnBarrier { await barrier() } try Task.checkCancellation() + // `stopping` means this turn lost the race with stop. Cancellation keeps + // shutdown out of the producer-failure path; `.closed` stays the answer + // for a caller that uses the runtime after stop. + if stopping { throw CancellationError() } + guard !closed, let owner, let worker else { throw CollectorRuntimeError.closed } + guard now >= 0 else { throw CollectorRuntimeError.invalidConfiguration } let inventory = try advanceInventory(owner: owner) let publication = try await worker.captureOnce(now: now, captureRootIDs: inventory.availableRoots) var waiting = historyWaiting @@ -297,8 +316,9 @@ public actor CollectorRuntime { } private func runUploadOnce(replicaID: String, now: Int64) async throws { - guard !stopping, !closed, let uploader else { throw CollectorRuntimeError.closed } try Task.checkCancellation() + if stopping { throw CancellationError() } + guard !closed, let uploader else { throw CollectorRuntimeError.closed } _ = try await uploader.uploadOnce(replicaID: replicaID, now: now) } @@ -388,6 +408,15 @@ public actor CollectorRuntime { } onCancel: { joining.cancel() } } + /// Cancellation and `.closed` from a turn that observed `stopping` are this + /// stop's shutdown signal. Configuration and reconciliation failures still + /// fail the join. A later `runOnce` on this instance still throws `.closed`. + private static func isShutdownSignal(_ error: Error) -> Bool { + if error is CancellationError { return true } + if case CollectorRuntimeError.closed = error { return true } + return false + } + public func stop() async throws { if closed { return } if let cleanup { try await cleanup.value; return } @@ -401,11 +430,9 @@ public actor CollectorRuntime { let joining = Task { [self] in var producerFailure: Error? do { _ = try await joiningLoop?.value } - catch is CancellationError {} - catch { producerFailure = error } + catch { if !Self.isShutdownSignal(error) { producerFailure = error } } do { _ = try await joiningCycle?.value } - catch is CancellationError {} - catch { if producerFailure == nil { producerFailure = error } } + catch { if producerFailure == nil, !Self.isShutdownSignal(error) { producerFailure = error } } try finishStop() if let producerFailure { throw producerFailure } } diff --git a/macos/EngramServiceCoreTests/CollectorRuntimeTests.swift b/macos/EngramServiceCoreTests/CollectorRuntimeTests.swift index 967b34d4..aabbcab1 100644 --- a/macos/EngramServiceCoreTests/CollectorRuntimeTests.swift +++ b/macos/EngramServiceCoreTests/CollectorRuntimeTests.swift @@ -2098,6 +2098,53 @@ final class CollectorRuntimeTests: XCTestCase { } catch { await replicas.stop(); throw error } } + // Repro for swift-unit on 2026-10-08 (jobs 113128319766, 113147173237, + // 113152487959): stop sets `stopping` while a capture turn is already + // inside the runtime, the turn guard throws `.closed`, and stop reports + // that shutdown signal as its own failure. `testReplayLossDuringStartKeepsRuntimeAndOtherRootsRunning_repro` + // and `testStartLoopPublishesAndStopJoinsBeforeReopen` failed this way. + func testStopDuringCaptureTurnStartDoesNotSurfaceClosed_repro() async throws { + let fixture = try RuntimeFixture() + defer { fixture.remove() } + let replicas = try await RuntimeReplicas.start(parent: fixture.base) + let gate = RuntimeGate() + let entered = RuntimeLocked(false) + var runtime: Runtime? + do { + try fixture.writeTranscript("turn in flight at stop") + try fixture.writeSettings(fixture.document(replicas: replicas)) + let active = try XCTUnwrap(Runtime.open(settingsURL: fixture.settings, secretLoader: fixture.secret)) + runtime = active + await active.setTestCaptureTurnBarrier { + entered.update { $0 = true } + await gate.wait() + } + try await active.start() + let enteredDeadline = Date().addingTimeInterval(5) + while !entered.value { + guard Date() < enteredDeadline else { throw RuntimeFixture.Failure.deadline } + try await Task.sleep(for: .milliseconds(10)) + } + // Set the flag before cancelling. The turn then fails the loop with + // `.closed`, and stop must not report that as its own failure. + await active.testBeginStopping() + gate.open() + try await active.stop() + do { + _ = try await active.runOnce(now: 1) + XCTFail("closed runtime accepted work") + } catch { XCTAssertEqual(error as? RuntimeError, .closed) } + let reopened = try XCTUnwrap(Runtime.open(settingsURL: fixture.settings, secretLoader: fixture.secret)) + try await reopened.stop() + await replicas.stop() + } catch { + gate.open() + try? await runtime?.stop() + await replicas.stop() + throw error + } + } + func testBootstrapIsBoundedAndPrivacyExclusionPreventsPublication() async throws { let fixture = try RuntimeFixture() defer { fixture.remove() } @@ -4085,6 +4132,36 @@ private final class RuntimeLocked: @unchecked Sendable { func update(_ operation: (inout Value) -> Void) { lock.withLock { operation(&stored) } } } +/// Continuation gate that ignores task cancellation, so a runtime turn can stay +/// inside `stop`'s join until the test releases it. +private final class RuntimeGate: @unchecked Sendable { + private let lock = NSLock() + private var continuation: CheckedContinuation? + private var opened = false + + func wait() async { + await withCheckedContinuation { (continuation: CheckedContinuation) in + self.lock.lock() + if self.opened { + self.lock.unlock() + continuation.resume() + return + } + self.continuation = continuation + self.lock.unlock() + } + } + + func open() { + self.lock.lock() + self.opened = true + let continuation = self.continuation + self.continuation = nil + self.lock.unlock() + continuation?.resume() + } +} + final class RuntimeFixture: @unchecked Sendable { enum Failure: Error { case unsafeFixture, deadline } static let machineID = "A0000000-1111-2222-3333-444444444444"