fix(collector): ignore shutdown closed signal in stop

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,

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）。

macos/EngramCollectorCore/CollectorRuntime.swift

@@ -101,6 +101,20 @@ public actor CollectorRuntime {
     private var blockedStreaks: [Int: Int] = [:]
     private var blockedRoots = Set<Int>()
     private var reportedBlockedRoots = Set<Int>()
+    /// 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 }
         }

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<Value>: @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<Void, Never>?
+    private var opened = false
+
+    func wait() async {
+        await withCheckedContinuation { (continuation: CheckedContinuation<Void, Never>) 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"