diff --git a/Docs/Testing.md b/Docs/Testing.md index 0be4d6a3..fe4520f2 100644 --- a/Docs/Testing.md +++ b/Docs/Testing.md @@ -128,6 +128,29 @@ await clock.advance(by: .seconds(1)) await expect(model.secondsElapsed == 1) ``` +#### Telling SwiftModel where a custom clock suspends + +`expect`, `settle` and `waitUntil` all have to answer one question: *is the model done reacting?* Work that is suspended waiting for input that can only arrive from outside — a clock deadline, an external event source — is **not** reacting, and a wait must not sit around for it. SwiftModel marks its own suspension points automatically, so in almost every case there is nothing to do: + +- `node.forEach(anySequence) { … }` parks around its own `next()`. That covers *any* `AsyncSequence`, including `swift-async-algorithms` operators like `debounce` and `throttle` — **no adoption at all**. +- `node.event(…)`, `Observed`, and modification streams park at the source, so a hand-written `for await` over them behaves identically to `forEach`. + +The one shape SwiftModel cannot see is a **bare suspension that is not an `AsyncSequence`** — canonically a custom clock's `sleep`. Wrap that clock's own `sleep` in `withModelParked`: + +```swift +extension MyClock { + public func sleep(until deadline: Instant, tolerance: Duration?) async throws { + try await withModelParked { // ← the only change + try await self.nonAdjustedSleep(until: deadline, tolerance: tolerance) + } + } +} +``` + +**The adoption point is the clock implementation, not its call sites.** A clock protocol that funnels every caller through one `sleep` method needs one wrap, and every `clock.sleep` in every model is covered — models themselves are untouched. `withModelParked` finds the work unit through a task-local, so a suspension inside a child task the clock spawned still marks the right unit; called outside any model task it is a plain passthrough. + +Not adopting is safe, merely less precise: an unmarked suspension counts as running work, so quiescence for that task falls back to observing the scheduler (exactly what SwiftModel did before this existed). If a wait in that test later times out, the diagnostic names the task and its call site so you can find the clock to wrap. + ### Refactor-resilient tests SwiftModel tests assert **final state**, not the sequence of actions or effects that produced it. There is no action enum to enumerate and no `send`/`receive` script to keep in sync — you call a method and assert the outcome: diff --git a/Sources/SwiftModel/Internal/AnyContext.swift b/Sources/SwiftModel/Internal/AnyContext.swift index 2783b92c..7bb7206d 100644 --- a/Sources/SwiftModel/Internal/AnyContext.swift +++ b/Sources/SwiftModel/Internal/AnyContext.swift @@ -169,15 +169,20 @@ class AnyContext: @unchecked Sendable { /// catch (see `_modificationCount` below for the same argument). @exclusivity(unchecked) private var modeLifeTime: ModelLifetime = .anchored - private var eventContinuationsStore: [Int: AsyncStream.Continuation]? - private var eventContinuations: [Int: AsyncStream.Continuation] { + /// `_ParkedYield` rather than a bare `AsyncStream.Continuation`: it is the + /// same continuation, plus the eager unpark of whichever work unit is + /// suspended in this stream's `next()` (semantic quiescence, tier 1 — see + /// `AsyncSequenceExtensions.swift`). `yield` / `finish` / `onTermination` + /// are spelled identically, so every use site below is unchanged. + private var eventContinuationsStore: [Int: _ParkedYield]? + private var eventContinuations: [Int: _ParkedYield] { _read { yield eventContinuationsStore ?? [:] } _modify { if eventContinuationsStore != nil { yield &eventContinuationsStore! if eventContinuationsStore!.isEmpty { eventContinuationsStore = nil } } else { - var temp: [Int: AsyncStream.Continuation] = [:] + var temp: [Int: _ParkedYield] = [:] yield &temp if !temp.isEmpty { eventContinuationsStore = temp } } @@ -854,6 +859,72 @@ class AnyContext: @unchecked Sendable { return snapshot.contains { $0.hasPendingStartTask } } + // MARK: - Semantic quiescence (computed, NOT used for verdicts) + // + // The alternative answer to "is the model done reacting?" described in + // `Docs/test-quiescence-redesign.md`: instead of observing the scheduler + // (executor idle / queues idle / pending-start), ask the registry which + // framework-owned work is RUNNING. Everything the framework spawns is a + // registered `TaskCancellable` (one `ModelWorkUnit` each); a unit suspended + // at a suspension point SwiftModel owns is parked and does not block + // quiescence. + // + // Currently only compared against the existing answer inside + // `_driveToStableFixpoint` (dual-run instrumentation). Nothing depends on + // it. + + /// True if any context in this subtree has a running work unit. + var hasRunningWorkUnit: Bool { + // Same lock-protected snapshot pattern as `activeTasks`. + let (selfRunning, snapshot) = lock { (cancellationsStore?.hasRunningWorkUnit ?? false, allChildren) } + if selfRunning { return true } + return snapshot.contains { $0.hasRunningWorkUnit } + } + + /// `(registered, parked)` work-unit counts across this subtree — see + /// `Cancellations.workUnitCensus`. Diagnostic only. + var workUnitCensus: _WorkUnitCensus { + let (selfCensus, snapshot) = lock { (cancellationsStore?.workUnitCensus ?? _WorkUnitCensus(), allChildren) } + return snapshot.reduce(into: selfCensus) { acc, child in + let c = child.workUnitCensus + acc.registered += c.registered + acc.parked += c.parked + acc.parkGeneration &+= c.parkGeneration + } + } + + /// The running work units in this subtree, for diagnostics. + var runningWorkUnits: [(modelName: String, name: String, fileAndLine: FileAndLine)] { + let (selfUnits, snapshot) = lock { (cancellationsStore?.runningWorkUnits ?? [], allChildren) } + return snapshot.reduce(into: selfUnits) { $0.append(contentsOf: $1.runningWorkUnits) } + } + + /// The semantic quiescence answer for this subtree: no registered work unit + /// is running, and both call queues are idle. + /// + /// Design §4 folds the queues into the same counter — a background + /// (`Observed` / memoize recompute) or main-registrar (`@ObservedModel` + /// notification) queue item *is* a running unit. They are not + /// `TaskCancellable`s, so this prototype keeps them as the two extra + /// predicates they already are rather than re-plumbing `CallQueue`; the + /// resulting answer is the same. + var semanticQuiescence: Bool { + !hasRunningWorkUnit && backgroundCall.isIdle && mainCallQueue.isIdle + } + + /// The semantic verdict for this subtree — running-unit count plus the + /// subset that looks undeclared. Backs the combined rule in + /// `TestAccess._driveToStableFixpoint`; the queues are deliberately NOT + /// folded in here, because the rule only consults this once the existing + /// answer (which already requires both queues idle) has said "done". + func semanticVerdict(nowNs: UInt64, quietNs: UInt64) -> _SemanticVerdict { + // Same lock-protected snapshot pattern as `activeTasks`. + let (selfVerdict, snapshot) = lock { + (cancellationsStore?.semanticVerdict(nowNs: nowNs, quietNs: quietNs) ?? _SemanticVerdict(), allChildren) + } + return snapshot.reduce(into: selfVerdict) { $0.merge($1.semanticVerdict(nowNs: nowNs, quietNs: quietNs)) } + } + /// Returns the main registrar if the main channel has been created (lazy), or nil /// otherwise. `_main` is lock-published, so the read takes the hierarchy lock too. @available(macOS 14.0, iOS 17.0, watchOS 10.0, tvOS 17.0, *) @@ -1374,7 +1445,15 @@ class AnyContext: @unchecked Sendable { guard !isDestructed else { return .finished } - let (stream, cont) = AsyncStream.makeStream() + // Park-marked at the SOURCE (design §5 / tier 1): a consumer + // suspended in this stream's `next()` reads parked, and every + // resume of that wait — `yield`, `finish`, or the consumer's task + // being cancelled (which the stdlib routes through + // `onTermination` before it resumes `next()` with nil) — unparks it + // eagerly. The `node.event(…)` overloads filter DOWNSTREAM of this, + // which is exactly why the mark is here and not on the filtered + // stream: see `ModelNode+Events.swift`. + let (stream, cont) = _makeParkedStream(of: EventInfo.self) let key = generateKey() cont.onTermination = { [weak self] _ in diff --git a/Sources/SwiftModel/Internal/AsyncSequenceExtensions.swift b/Sources/SwiftModel/Internal/AsyncSequenceExtensions.swift index 42e2410f..bcbebc83 100644 --- a/Sources/SwiftModel/Internal/AsyncSequenceExtensions.swift +++ b/Sources/SwiftModel/Internal/AsyncSequenceExtensions.swift @@ -1,3 +1,5 @@ +import Foundation + // MARK: - Internal replacement for swift-async-algorithms // // eraseToStream() is NOT defined here — ConcurrencyExtras (re-exported by Dependencies) already @@ -21,7 +23,17 @@ private final class _DedupBox: @unchecked Sendable { var iter = iterator var previous: Element? = nil _next = { - while let value = try? await iter.next() { + // Park around the upstream wait (semantic quiescence, hook 1). This + // closure runs on the CONSUMER's task — a `node.forEach` body — so + // when it is driven through `forEach` the park is nested inside + // `forEach`'s own park and the counter (not a Bool) keeps the unit + // parked until both scopes exit. Marking it here too means a + // hand-written `for await` over one of these streams parks as well. + // The park scope is per UPSTREAM element (it is inside the `while` + // condition), so a duplicate that gets swallowed re-parks — the + // property `_ParkSource` has to arrange explicitly for the filtered + // sources below. + while let value = await _withCurrentWorkUnitParked({ try? await iter.next() }) { if value != previous { previous = value; return value } } return nil @@ -38,3 +50,268 @@ extension AsyncSequence where Element: Equatable & Sendable { return AsyncStream { await box.next() } } } + +// MARK: - Tier 1: source-side parking +// +// `Docs/test-quiescence-redesign.md` §5: **the park mark belongs to the input +// source, not to the consuming loop.** `node.forEach` parks around its own +// `next()` (hook 1), but a user can write the loop by hand — +// +// node.task { for await value in Observed { … } } // no forEach +// +// — and if `forEach` were the only marking site that task would read as +// *running* forever, blocking quiescence for as long as it lives. SwiftModel +// owns every one of these sources (it produces the `AsyncStream` and holds the +// continuation), so the mark goes where the framework can guarantee it: the +// iterator's wait for the next yield. +// +// ## Two things the first cut got wrong, and why the source now owns both ends +// +// The first cut wrapped the *consumer-facing* sequence +// (`someChain._eraseToParkedWaitStream()`) and unparked lazily, from the +// resumed task's `defer`. Measuring it (design §5 property 2) found both halves +// of that wrong: +// +// 1. **Lazy is late.** Between the continuation being resumed and the resumed +// task's `defer` running, the unit still reads parked while model-writing +// work is already inbound. 62 % of the surviving "new says quiescent, old +// says busy" checks were an `AsyncStream.next()` cancellation resume +// (`_Storage.finish()` called from the cancel handler) and 20 % a +// `Continuation.yield`. The cancellation half has no activity signal behind +// it at all, so §5's containment argument does not cover it, and the body +// it resumes goes on to run model-writing `defer`s. +// `ModelWorkUnit.noteResumeInFlight()` is the fix: the *resumer* unparks, +// before it resumes. +// +// 2. **The outer position is the wrong position for an eager unpark.** +// `node.event(ofType:)` is `context.events().compactMap { … }`: an event +// that fails the filter still resumes the consumer, which filters it and +// loops back into the *same* open park scope. An eager unpark at the outer +// position would therefore leave the unit reading *running* until the next +// event that happens to pass the filter — potentially never. So the park +// moved inwards, to the one scope that is entered and left exactly once per +// upstream element: the iterator over the raw `AsyncStream` that the +// framework's own continuation feeds. Filtering downstream of it re-parks +// by construction. +// +// `_ParkSource` is the rendezvous between the two ends, and it also closes a +// third window the outer position hid: a value yielded while the consumer was +// *running* sits in the stream's buffer, so the consumer's next `park()` would +// mark parked for a delivery that is already queued (and whose resume, being a +// buffered fast-path, is an executor hop away rather than a suspension away). +// `_ParkSource` counts undelivered yields and refuses to park while any are +// outstanding. +// +// Only the wait is parked — never the consumer's body. Running the user's +// closure is real model work. +// +// ## Why `AsyncStream(unfolding:)` and not a custom iterator +// +// The obvious alternative is to give `Observed` its own `AsyncIterator` that +// parks around the upstream `next()`. Two reasons not to: +// +// * **Public API.** `Observed.makeAsyncIterator()` returns +// `AsyncStream.Iterator`, and `node.event(…)` / +// `observeModifications()` return `AsyncStream` outright. Unfolding keeps +// every one of those types exactly as it was; a custom iterator would change +// the `AsyncIterator` associated type. +// * **SE-0431 (`next(isolation:)`).** Because the consumer still iterates a +// plain `AsyncStream`, it gets the *stdlib's* `next(isolation:)`, so `for +// await`'s desugaring is untouched and we never have to decide whether to +// witness an availability-gated requirement (`next(isolation:)` is +// SwiftStdlib 6.0; this library deploys to macOS 11). A hand-written +// iterator implementing only `next()` would silently downgrade every +// `for await` over these streams to the non-isolated path. +// +// The residual cost is real and worth knowing: the produce closure is +// `@Sendable` and non-isolated, so a consumer that iterates one of these streams +// **from an actor** hops off it once per element to run `produce`. Model task +// bodies are non-isolated, so the common path is unaffected — and every +// `node.event(…)` stream already paid exactly this, because `eraseToStream()` +// is itself an unfolding wrapper. + +/// The producer↔consumer rendezvous for one SwiftModel-owned `AsyncStream`. +/// +/// Held by both ends of a stream built with `_makeParkedStream`: the +/// `_ParkedYield` the framework yields through, and the `_ParkedWaitBox` the +/// consumer drives. Everything it does happens under one `NSLock`, which is +/// what makes "park" and "resume in flight" mutually exclusive rather than +/// racing — a producer can only force a unit it can still see parked, and a +/// consumer can only park when no delivery is outstanding. Without that, an +/// eager unpark could land just after the consumer re-parked and strand the +/// unit reading *running*. +/// +/// Lock order is `_ParkSource` → `ModelWorkUnit`; nothing takes them the other +/// way round, and `AsyncStream.Continuation.yield` — which can resume a +/// continuation — is always called *outside* this lock. +final class _ParkSource: @unchecked Sendable { + private let lock = NSLock() + /// Yields handed to the stream's buffer that the consumer has not taken out + /// yet. While this is non-zero the consumer refuses to park: the value is + /// already queued, so no wait may read the unit as quiescent. + private var undelivered = 0 + /// The unit currently suspended in this stream's `next()`, if any. + private var waitingUnit: ModelWorkUnit? + private var isTerminated = false + + // MARK: Producer side + + /// Call IMMEDIATELY BEFORE `AsyncStream.Continuation.yield(_:)`. + func willYield() { + lock.withLock { + undelivered += 1 + resumeWaiterLocked() + } + } + + /// Call before anything that resumes a suspended `next()` with `nil`: + /// `finish()`, and the stream's `onTermination` handler — which the stdlib + /// invokes *before* resuming on cancellation (`AsyncStream._Storage.cancel`: + /// "handler must be invoked before yielding nil for termination"). That is + /// the cancellation half of the window, and the half with no activity + /// signal behind it. + func willTerminate() { + lock.withLock { + isTerminated = true + resumeWaiterLocked() + } + } + + private func resumeWaiterLocked() { + guard let unit = waitingUnit else { return } + waitingUnit = nil + unit.noteResumeInFlight() + } + + // MARK: Consumer side + + /// About to `await` the next element. Parks the calling work unit unless a + /// delivery is already outstanding (or the stream is over), and returns the + /// ticket the caller must release when the await returns. + /// + /// Refusing to park is not enough on its own: `node.forEach` wraps its own + /// `park()` around this whole call (hook 1, for foreign sequences), so a + /// unit that declines to park here would still read parked from that OUTER + /// scope while the buffered value it is about to take out is delivered. + /// `noteResumeInFlight()` is what breaks out of an enclosing scope — it + /// invalidates every open ticket on the unit, including `forEach`'s, and the + /// unit stays running until `forEach`'s *next* loop parks it again. + func beginWait() -> _ParkTicket? { + guard let unit = ModelWorkUnit.current else { return nil } + return lock.withLock { () -> _ParkTicket? in + guard undelivered == 0, !isTerminated else { + unit.noteResumeInFlight() + return nil + } + let ticket = unit.park() + waitingUnit = unit + return ticket + } + } + + /// The await returned. `delivered` is `false` for the terminal `nil`. + func endWait(delivered: Bool) { + lock.withLock { + waitingUnit = nil + if delivered, undelivered > 0 { undelivered -= 1 } + } + } +} + +/// The continuation SwiftModel's own stream sources yield through: an +/// `AsyncStream.Continuation` that tells its `_ParkSource` about every resume +/// *before* performing it. +/// +/// Deliberately mirrors `AsyncStream.Continuation`'s member names, so adopting +/// it at a source site is a change to the stream's construction only — the body +/// that yields, finishes and installs `onTermination` needs no edit. +struct _ParkedYield: Sendable { + fileprivate let base: AsyncStream.Continuation + fileprivate let source: _ParkSource + + func yield(_ value: Element) { + source.willYield() + base.yield(value) + } + + func finish() { + source.willTerminate() + base.finish() + } + + /// Composed rather than assigned: `_makeParkedStream` installs the park + /// hook first, and a source site setting its own handler must not drop it. + /// (That hook is what makes a *cancelled* consumer unpark eagerly, which no + /// source site knows to do.) + var onTermination: (@Sendable (AsyncStream.Continuation.Termination) -> Void)? { + get { base.onTermination } + nonmutating set { + let source = self.source + base.onTermination = { termination in + source.willTerminate() + newValue?(termination) + } + } + } +} + +/// Drives the raw stream's iterator, parking the calling work unit around the +/// wait. +/// +/// `@unchecked Sendable` on the same terms as `_DedupBox`: `AsyncStream`'s +/// unfolding iterator serialises calls to `next()`, so only one call is ever in +/// flight and the captured iterator is never accessed concurrently. +private final class _ParkedWaitBox: @unchecked Sendable { + private let _next: () async -> Element? + + init(_ iterator: AsyncStream.Iterator, source: _ParkSource) { + var iter = iterator + _next = { + let ticket = source.beginWait() + var delivered = false + defer { + source.endWait(delivered: delivered) + // A no-op whenever the producer already unparked eagerly; see + // `ModelWorkUnit`'s epoch discussion. + ticket?.release() + } + let value = await iter.next() + delivered = value != nil + return value + } + } + + func next() async -> Element? { await _next() } +} + +/// Builds a SwiftModel-owned `AsyncStream` whose consumer's wait is marked +/// parked and whose every resume is marked eagerly. The drop-in replacement for +/// `AsyncStream { cont in … }` at any site where the framework hands async +/// input to model code. +func _makeParkedStream( + of elementType: Element.Type = Element.self, + _ build: (_ParkedYield) -> Void +) -> AsyncStream { + let source = _ParkSource() + let raw = AsyncStream { cont in + // Installed before `build` so a site that never sets `onTermination` + // still unparks eagerly on cancellation; `_ParkedYield`'s setter + // composes rather than replaces. + cont.onTermination = { _ in source.willTerminate() } + build(_ParkedYield(base: cont, source: source)) + } + let box = _ParkedWaitBox(raw.makeAsyncIterator(), source: source) + return AsyncStream { await box.next() } +} + +/// `AsyncStream.makeStream()` for a park-marked stream — for sources that store +/// the continuation rather than closing over it (`AnyContext.events()`). +func _makeParkedStream( + of elementType: Element.Type +) -> (stream: AsyncStream, continuation: _ParkedYield) { + let source = _ParkSource() + let (raw, cont) = AsyncStream.makeStream() + cont.onTermination = { _ in source.willTerminate() } + let box = _ParkedWaitBox(raw.makeAsyncIterator(), source: source) + return (AsyncStream { await box.next() }, _ParkedYield(base: cont, source: source)) +} diff --git a/Sources/SwiftModel/Internal/Cancellables.swift b/Sources/SwiftModel/Internal/Cancellables.swift index 38226bec..64e69121 100644 --- a/Sources/SwiftModel/Internal/Cancellables.swift +++ b/Sources/SwiftModel/Internal/Cancellables.swift @@ -69,13 +69,18 @@ final class TaskCancellable: Cancellable, InternalCancellable, @unchecked Sendab /// racing a `forEach` task registration). Passing the box in removes the /// window rather than defending it; the observable value is unchanged /// (`false` until the body runs). - let _hasStartedRunningBox: LockIsolated - var hasStartedRunning: Bool { _hasStartedRunningBox.value } + /// + /// The box is a `ModelWorkUnit`, which also carries this task's + /// running/parked state for semantic quiescence. Same object, same + /// publication rules — no extra allocation over the `LockIsolated` + /// it replaced. + let workUnit: ModelWorkUnit + var hasStartedRunning: Bool { workUnit.hasStartedRunning } - init(modelName: String, taskName: String, fileAndLine: FileAndLine, context: AnyContext, hasStartedRunningBox: LockIsolated, task: @escaping @Sendable (@escaping @Sendable () -> Void) -> Task) { + init(modelName: String, taskName: String, fileAndLine: FileAndLine, context: AnyContext, workUnit: ModelWorkUnit, task: @escaping @Sendable (@escaping @Sendable () -> Void) -> Task) { // Assigned before `cancellations.register(self)` below publishes this - // instance to any settle thread — see `_hasStartedRunningBox`. - self._hasStartedRunningBox = hasStartedRunningBox + // instance to any settle thread — see `workUnit`. + self.workUnit = workUnit // Resolve the registry ONCE, before `lock` is taken. `AnyContext.cancellations` // acquires the per-context hierarchy lock (H); this instance's `lock` is T. // Evaluating `context.cancellations` *inside* `lock { }` — as the capture-list @@ -109,15 +114,29 @@ final class TaskCancellable: Cancellable, InternalCancellable, @unchecked Sendab cancellations.register(self) - lock { + let bodyWillNeverRun: Bool = lock { guard !self.hasBeenCancelled else { // Task was cancelled before init reached the task-creation point; // the underlying Task stays nil and never runs. - return + return true } self.task = task { [weak cancellations] in _ = cancellations?.unregister(id) + // Semantic quiescence: the body cannot run again, so the work + // unit retires here and ONLY here — see + // `Cancellations.liveWorkUnits`. + cancellations?.retireWorkUnit(id) } + return false + } + // Nothing will ever call `onDone` on this path, so the live work unit + // has to be retired explicitly or it would pin the model as + // non-quiescent forever. Done OUTSIDE `lock` (T): every other path into + // `Cancellations` (C) runs with T released, and adding a T→C edge here + // would be the only one in the file going that way — see the H→T + // ordering note in `init` above. + if bodyWillNeverRun { + cancellations.retireWorkUnit(id) } } @@ -152,41 +171,74 @@ extension TaskCancellable { convenience init(modelName: String, taskName: String, fileAndLine: FileAndLine, context: AnyContext, isDetached: Bool, priority: TaskPriority?, @_inheritActorContext @_implicitSelfCapture operation: @escaping @Sendable () async throws -> Void, `catch`: (@Sendable (Error) -> Void)?) { // Constructed BEFORE self.init so the factory closure can capture it. - // Stored on `self` AFTER self.init completes — see `_hasStartedRunningBox`. - let hasStartedRunningBox = LockIsolated(false) + // Stored on `self` AFTER self.init completes — see `workUnit`. + let workUnit = ModelWorkUnit() - self.init(modelName: modelName, taskName: taskName, fileAndLine: fileAndLine, context: context, hasStartedRunningBox: hasStartedRunningBox) { onDone in + self.init(modelName: modelName, taskName: taskName, fileAndLine: fileAndLine, context: context, workUnit: workUnit) { onDone in let contexts = AnyCancellable.contexts let operation = { @Sendable in - do { - // Use context.capturedDependencies directly (not withDependencies(from: context)) - // so the task inherits exactly the context's dep overrides. withDependencies(from:) - // would merge against DependencyValues._current, potentially losing overrides. - try await DependencyValues.$_current.withValue(context.capturedDependencies) { - try await ModelAccess.$isInModelTaskContext.withValue(true) { - try await AnyCancellable.$inheritedContexts.withValue(contexts) { - try await AnyCancellable.$contexts.withValue([]) { - defer { onDone() } - - guard !Task.isCancelled, !context.isDestructed else { return } - - // Signal that the body has now actually started executing — - // see `ModelAccess.taskBodyStarted` and - // `TaskCancellable._hasStartedRunningBox`. Setting the box - // BEFORE notifying the access avoids a window where settle - // could re-check `hasPendingStartTask`, see this task still - // not-started, and re-arm pointlessly. - hasStartedRunningBox.setValue(true) - ModelAccess.current?.taskBodyStarted() - - try await operation() + // UNREGISTER LAST — the work unit must outlive the whole task + // body, `catch` handler included. + // + // `onDone()` unregisters this `TaskCancellable` from + // `Cancellations`, which is what makes the unit disappear from + // `hasRunningWorkUnit`. It used to sit in a `defer` INSIDE the + // innermost task-local scope, i.e. it fired before the + // `catch` block below ran — and `catch` is user code + // (`node.task(catch:)`, `node.onChange(catch:)`) that routinely + // writes model state: + // + // node.task { try await mayFail() } catch: { error in + // self.error = "\(error)" // <- model write + // } + // + // Under the scheduler-observing answer that window was covered + // for free (the executor job is still outstanding), but under + // semantic quiescence the registry would already read + // "quiescent" while a model write was still to come — a + // premature pass. Hoisting the `defer` to the closure's + // outermost scope closes it: the unit is unregistered only once + // nothing in the body can run again. + // + // Unconditional either way, so `underlyingTask.value` (which + // `_forEachImpl` uses to serialise bodies) still resolves after + // it, exactly as its doc-comment states. + defer { onDone() } + + // Publish this body's work unit to the task-local so + // `withModelParked` — called from anywhere inside the body, + // including child tasks and library code the body calls into, + // and including the `catch` handler — finds the right unit to + // mark parked. Set once per body; see `ModelWorkUnit`. + await ModelWorkUnit.$current.withValue(workUnit) { + do { + // Use context.capturedDependencies directly (not withDependencies(from: context)) + // so the task inherits exactly the context's dep overrides. withDependencies(from:) + // would merge against DependencyValues._current, potentially losing overrides. + try await DependencyValues.$_current.withValue(context.capturedDependencies) { + try await ModelAccess.$isInModelTaskContext.withValue(true) { + try await AnyCancellable.$inheritedContexts.withValue(contexts) { + try await AnyCancellable.$contexts.withValue([]) { + guard !Task.isCancelled, !context.isDestructed else { return } + + // Signal that the body has now actually started executing — + // see `ModelAccess.taskBodyStarted` and + // `TaskCancellable.workUnit`. Setting the box + // BEFORE notifying the access avoids a window where settle + // could re-check `hasPendingStartTask`, see this task still + // not-started, and re-arm pointlessly. + workUnit.markBodyStarted() + ModelAccess.current?.taskBodyStarted() + + try await operation() + } } } } + } catch { + if Task.isCancelled || error is CancellationError { return } + `catch`?(error) } - } catch { - if Task.isCancelled || error is CancellationError { return } - `catch`?(error) } } diff --git a/Sources/SwiftModel/Internal/Cancellations.swift b/Sources/SwiftModel/Internal/Cancellations.swift index 5aa7aec3..79eb88db 100644 --- a/Sources/SwiftModel/Internal/Cancellations.swift +++ b/Sources/SwiftModel/Internal/Cancellations.swift @@ -1,11 +1,99 @@ import Foundation +/// Diagnostic snapshot of a subtree's work units. `parkGeneration` is the sum +/// of every live unit's park/unpark transition count — it changes iff some unit +/// parked or unparked, which is design §5's "no flicker between two +/// observations" check. +struct _WorkUnitCensus { + var registered = 0 + var parked = 0 + var parkGeneration: UInt64 = 0 +} + +/// One running work unit, named for a diagnostic. +struct _RunningWorkUnitInfo: Sendable, Hashable { + let modelName: String + let name: String + let fileAndLine: FileAndLine + + /// `taskName` already defaults to `"function @ file:line"`, so only append + /// the site when it isn't already there. + var description: String { + let site = fileAndLine.description + return name.hasSuffix(site) ? "\(modelName).\(name)" : "\(modelName).\(name) @ \(site)" + } +} + +/// The semantic answer for one subtree, in the form the combined rule needs. +/// +/// `Docs/test-quiescence-redesign.md` §3 says quiescent ⇔ no registered work is +/// running. That is `isQuiescent`. `undeclared` carries the fallback: the +/// running units the framework cannot see inside (never parked, and silent for +/// a whole grace window), which is the only shape where deferring to the +/// scheduler-observing answer is right — see +/// `TestAccess._driveToStableFixpoint`. +struct _SemanticVerdict { + var running = 0 + var undeclared: [_RunningWorkUnitInfo] = [] + + /// The new answer: no registered unit is running. + var isQuiescent: Bool { running == 0 } + + /// At least one unit is running and EVERY one of them looks undeclared. + /// One mid-flight unit anywhere in the subtree defeats it — the whole point + /// of the conjunction is that a resumption in flight must still be waited + /// for even when an unadopted clock is sleeping beside it. + var allRunningLookUndeclared: Bool { running > 0 && undeclared.count == running } + + mutating func merge(_ other: _SemanticVerdict) { + running += other.running + undeclared.append(contentsOf: other.undeclared) + } +} + +/// One entry in `Cancellations.liveWorkUnits` — a `TaskCancellable`'s work +/// unit plus the identity the diagnostics need. +struct _LiveWorkUnit { + let id: Int + let modelName: String + let taskName: String + let fileAndLine: FileAndLine + let unit: ModelWorkUnit +} + final class Cancellations: @unchecked Sendable { fileprivate let lock = NSLock() fileprivate var registered: [Int: InternalCancellable] = [:] fileprivate var keyed: [CancellableKey: [Int]] = [:] private var _sealed = false + /// SEMANTIC QUIESCENCE — the set of task bodies that are still ABLE TO RUN. + /// + /// This is deliberately NOT `registered`. `registered` is the + /// *cancellation* registry, and cancelling drops an entry **before** the + /// task has unwound: `cancel(_:)` goes through `unregister`, and + /// `cancelAll()` empties the dictionary and only then calls `onCancel()`. + /// A cancelled body keeps running through its `defer`s afterwards, and + /// those routinely write model state — + /// + /// node.task { + /// defer { playerController = nil; marker = "cleared" } // <- writes + /// ... + /// } + /// + /// — so an answer derived from `registered` reads "quiescent" during every + /// teardown, `task(id:)` replacement and `cancelPrevious` swap while model + /// writes are still to come. That is the same premature-pass shape as the + /// `catch`-handler window (see `TaskCancellable`'s `defer { onDone() }`), + /// and it is much more common. + /// + /// Entries are inserted at registration and removed by exactly one thing: + /// the task body's outermost `defer`, via `retireWorkUnit(_:)`. So the unit + /// outlives cancellation and is retired only when the body genuinely cannot + /// run again. (The one path where the body never runs — cancelled before + /// `TaskCancellable.init` creates the `Task` — retires explicitly there.) + private var liveWorkUnits: [Int: _LiveWorkUnit] = [:] + deinit { cancelAll() } @@ -31,8 +119,20 @@ final class Cancellations: @unchecked Sendable { func register(_ c: InternalCancellable) { let shouldImmediatelyCancel: Bool = lock { + // Sealed: no `Task` is ever created for this cancellable, so nothing + // will call `retireWorkUnit`. Registering a live unit here would + // pin the model as permanently non-quiescent. if _sealed { return true } registered[c.id] = c + if let task = c as? TaskCancellable { + liveWorkUnits[c.id] = _LiveWorkUnit( + id: task.id, + modelName: task.modelName, + taskName: task.taskName, + fileAndLine: task.fileAndLine, + unit: task.workUnit + ) + } for key in AnyCancellable.contexts { keyed[key, default: []].append(c.id) } @@ -43,6 +143,13 @@ final class Cancellations: @unchecked Sendable { } } + /// Drops the live work unit for `id`. Called from the task body's outermost + /// `defer` (and from the never-started path in `TaskCancellable.init`) — + /// see `liveWorkUnits`. + func retireWorkUnit(_ id: Int) { + lock { _ = liveWorkUnits.removeValue(forKey: id) } + } + func unregister(_ id: Int) -> InternalCancellable? { lock { let cancellable = registered.removeValue(forKey: id) @@ -97,6 +204,71 @@ final class Cancellations: @unchecked Sendable { } } + /// SEMANTIC QUIESCENCE (computed, not yet used for verdicts). + /// + /// True if any registered work unit in this registry is **running** — i.e. + /// executing, or suspended somewhere SwiftModel did not put it. Work parked + /// at a suspension the framework owns (`forEach`'s `next()`, anything inside + /// `withModelParked`) does not count. See `ModelWorkUnit`. + var hasRunningWorkUnit: Bool { + lock { + liveWorkUnits.values.contains { $0.unit.isRunning } + } + } + + /// `(registered task-unit count, parked count)` for this registry. + /// Diagnostic only: it is what separates the two shapes of "new says + /// quiescent, old says busy". A snapshot with **parked > 0** is a work unit + /// whose continuation may already have been resumed while its `unpark()` + /// has not run yet (design §5's park→resume window); a snapshot with + /// **registered == 0** is an executor job that owns no unit at all, which is + /// where an unregistered-work hole would hide. + var workUnitCensus: _WorkUnitCensus { + lock { + var census = _WorkUnitCensus() + for entry in liveWorkUnits.values { + census.registered += 1 + if !entry.unit.isRunning { census.parked += 1 } + census.parkGeneration &+= entry.unit.parkGeneration + } + return census + } + } + + /// The running work units, for the disagreement trace / future backstop + /// message. Sorted by registration order for stable output. + var runningWorkUnits: [(modelName: String, name: String, fileAndLine: FileAndLine)] { + lock { + liveWorkUnits.values + .filter { $0.unit.isRunning } + .sorted { $0.id < $1.id } + .map { (modelName: $0.modelName, name: $0.taskName, fileAndLine: $0.fileAndLine) } + } + } + + /// SEMANTIC QUIESCENCE — the verdict this registry contributes to the + /// combined rule. One pass, classifying each live unit exactly once so the + /// running count and the undeclared list are a consistent snapshot. + func semanticVerdict(nowNs: UInt64, quietNs: UInt64) -> _SemanticVerdict { + lock { + var verdict = _SemanticVerdict() + for entry in liveWorkUnits.values.sorted(by: { $0.id < $1.id }) { + switch entry.unit.classify(nowNs: nowNs, quietNs: quietNs) { + case .parked: + continue + case .midFlight: + verdict.running += 1 + case .looksUndeclared: + verdict.running += 1 + verdict.undeclared.append( + _RunningWorkUnitInfo(modelName: entry.modelName, name: entry.taskName, fileAndLine: entry.fileAndLine) + ) + } + } + return verdict + } + } + func cancelAll() { lock { defer { diff --git a/Sources/SwiftModel/Internal/Context.swift b/Sources/SwiftModel/Internal/Context.swift index b1b94f1c..03ba0168 100644 --- a/Sources/SwiftModel/Internal/Context.swift +++ b/Sources/SwiftModel/Internal/Context.swift @@ -400,6 +400,13 @@ final class Context: AnyContext, @unchecked Sendable { } let generation = referenceGeneration + // SEMANTIC QUIESCENCE — excluded housekeeping, category (d). + // Justification: this is memory reclamation on a user-configured TTL, + // not a model reaction. It writes nothing observable (it only releases + // already-destructed state, generation-guarded), so no test can be + // waiting for it; counting it as a running work unit would make every + // `settle()` after a teardown wait out the whole `lastSeenTimeToLive`. + // See the audit in `ModelWorkUnit.swift`. Task { try? await Task.sleep(nanoseconds: 1_000_000*UInt64(lastSeenTimeToLive*1000)) // Same fix as the non-TTL path: hold AnyContext.lock while swapping state so that diff --git a/Sources/SwiftModel/Internal/ModelWorkUnit.swift b/Sources/SwiftModel/Internal/ModelWorkUnit.swift new file mode 100644 index 00000000..e5769e1f --- /dev/null +++ b/Sources/SwiftModel/Internal/ModelWorkUnit.swift @@ -0,0 +1,365 @@ +import Foundation + +/// How one running work unit looks to the combined quiescence rule — see +/// `ModelWorkUnit.classify(nowNs:quietNs:)`. +enum _WorkUnitRunState: Sendable, Equatable { + case parked + case midFlight + case looksUndeclared +} + +/// The running/parked state of one unit of framework-owned async work. +/// +/// Every `TaskCancellable` — i.e. every `node.task` / `node.forEach` / +/// `node.onChange` body, including the inner per-element bodies `forEach` +/// spawns — owns exactly one `ModelWorkUnit`. The unit exists from the +/// `TaskCancellable`'s registration in `Cancellations` until its body returns +/// and the `defer` unregisters it, so "registered unit" and "unfinished work" +/// are the same set by construction. +/// +/// ## Why a counter and not a `Bool` +/// +/// The state we care about is *running vs parked*, which reads like a `Bool`. +/// It is stored as a signed **counter of non-parked activities** instead: +/// +/// * it starts at `1` (running), +/// * `park()` decrements, `unpark()` increments, +/// * the unit is **running** iff the count is `> 0`. +/// +/// This makes nesting compose. `withModelParked` can nest — a clock that wraps +/// its own `sleep` (design §6b hook 3) called from inside a `forEach` that +/// already parked around its `next()` (hook 1), or simply a `withModelParked` +/// inside a `withModelParked`. With a `Bool` the inner unpark would wrongly +/// mark the unit running while the outer park is still in scope; with a counter +/// `1 → 0 → -1 → 0` stays parked until the outermost scope exits. +/// +/// **Known limitation (documented, not fixed here).** The counter is shared by +/// every task that inherits the task-local — including child tasks a library +/// spawns inside the unit. If two such children park concurrently the count +/// reaches `-1`, and the first to unpark leaves it at `0`, i.e. still parked +/// even though one child is running again. Design §6b calls this out as the +/// reason hook 1 (`forEach`'s single `next()`) is preferred over hooks that can +/// fan out. Nothing in SwiftModel itself produces that shape today. +/// +/// ## Eager unpark: `noteResumeInFlight()` and the park epoch +/// +/// The counter alone is a *lazy* mark: the unit leaves "parked" only when the +/// resumed task next gets a CPU slot and runs the parking scope's `defer`. +/// Between a continuation being **resumed** and that slot, the unit still reads +/// parked while model-writing work is already inbound — the +/// park→resume-in-flight window. Design §5 assumed that window was benign (the +/// yield is itself an activity signal) and that a park-generation double-check +/// would catch it; measurement killed both claims. The `AsyncStream` +/// cancellation path (`_Storage.finish()` from `next()`'s cancel handler — 62 % +/// of the observed window) emits no activity signal at all and resumes a body +/// that then runs model-writing `defer`s, and the generation never changed +/// across a check. +/// +/// So the unpark is made **eager**: whoever resumes the continuation calls +/// `noteResumeInFlight()` *before* resuming, and the resumed side's `defer` +/// becomes a no-op. That requires the release to be idempotent, which is what +/// the **epoch** is for: `park()` hands back a `_ParkTicket` stamped with the +/// epoch current at park time, `noteResumeInFlight()` bumps the epoch, and a +/// ticket whose epoch is stale releases nothing. One force therefore +/// invalidates *every* open park scope on the unit — which is exactly right, +/// because a resumed task is running regardless of how many nested `park()` +/// scopes (`forEach`'s own `next()` wrapped around a stream that also parks) it +/// is unwinding through. It stays running until it *voluntarily parks again*, +/// which is the property a lazy, per-scope unpark cannot provide. +/// +/// ## Why it starts running rather than at body entry +/// +/// The sketch says "start 1 when the body begins". A unit is created (and +/// registered) before the cooperative pool schedules its body, and design §7 +/// wants `hasPendingStartTask` to fall out of the same rule — "a task that has +/// not started is simply *running*". So the counter starts at `1` at +/// construction; the not-yet-started window is therefore running, and the +/// semantic answer needs no separate pending-start predicate. +/// +/// ## The async-work audit (design §4 / §4a) — what is NOT a work unit +/// +/// Semantic quiescence is only sound if every framework path that can still +/// write to a model owns a registered unit. Everything else must be in one of +/// two explicitly-justified buckets. This is the completed inventory; keep it +/// in sync when adding a `Task` to `Sources/`. +/// +/// **(a) Test-drive machinery — not model work by construction.** None of it +/// runs on the per-test drain executor (`_DrainTestExecutor` is applied *only* +/// to `TaskCancellable` bodies, in the `convenience init` below), so it is +/// invisible to the scheduler-observing answer too: +/// * `TestExecutorDrive._startExecutorDrive`'s driver `Task` and its +/// `_gtsSleep`s — the thing doing the asking; counting it would make every +/// wait its own reason to keep waiting. +/// * `GlobalTickScheduler` deadline entries — a scheduled wake, not work +/// (design §4). +/// * `ModelTestingTrait`'s `group.addTask` timeout/watchdog arms. +/// +/// **(d) Excluded housekeeping — framework-spawned work no test may wait for.** +/// Every entry here is a deliberate blind spot, so each carries its +/// justification at the spawn site as well: +/// * `Context.swift`'s last-seen TTL `Task` — memory reclamation on a +/// user-configured TTL, never a model reaction. Counting it would make +/// `settle()` wait out the whole TTL (design §4a). +/// * `ObservedModel.swift`'s first-activation priming `Task` — a SwiftUI +/// `objectWillChange.send()`; touches no model state. +/// +/// **Counted, but not as `TaskCancellable`s.** `CallQueue`'s background-drain +/// and main-registrar pumps are real model work and *are* part of the answer — +/// `AnyContext.semanticQuiescence` folds in `backgroundCall.isIdle` and +/// `mainCallQueue.isIdle` (design §4 counts a queue item as one running unit; +/// the prototype keeps them as the two predicates they already were rather than +/// re-plumbing `CallQueue`, which gives the same answer). +final class ModelWorkUnit: @unchecked Sendable { + /// The work unit owning the current task, if any. Set once per + /// `TaskCancellable` body (see `TaskCancellable`'s convenience init), so it + /// propagates into every child task the body — or a library the body calls + /// into — spawns. `nil` outside model-owned async work, which is what makes + /// `withModelParked` a no-op passthrough there. + @TaskLocal static var current: ModelWorkUnit? + + private let lock = NSLock() + private var _activityCount: Int = 1 + private var _hasStartedRunning = false + private var _parkGeneration: UInt64 = 0 + /// Bumped by `noteResumeInFlight()` only. A `_ParkTicket` minted under an + /// older epoch releases nothing — see the type doc. + private var _epoch: UInt64 = 0 + /// `true` once this unit has parked at least once, i.e. it has suspended at + /// a point SwiftModel owns (`forEach`'s `next()`, a source-side park, a + /// `withModelParked` scope). Never reset — the question the fallback asks is + /// "has the framework EVER been told where this unit suspends", not "is it + /// parked now". See `classify(nowNs:quietNs:)`. + private var _hasEverParked = false + /// Monotonic timestamp of the last observable transition of this unit: + /// creation, body start, and every park / unpark / resume-in-flight. It is + /// the per-unit analogue of the drive's global activity stamp, and the only + /// evidence available for a unit that is running but silent. + private var _lastActivityNs: UInt64 = _drainMonotonicNs() + /// Nesting depth of `_withUserBodyRegion` — the stretches of this unit's + /// lifetime that are executing the USER's closure rather than framework + /// code. See `classify(nowNs:quietNs:)` for why the park history has to be + /// scoped to a region rather than the unit. + private var _userBodyDepth = 0 + + /// Bumped on every park/unpark transition. Design §5 property 2: a + /// quiescence answer is only trustworthy if it holds across two observations + /// with **no park-generation change between them**, because between marking + /// parked and actually suspending — and, more importantly, between a + /// continuation being resumed and the resumed task running its `unpark()` — + /// the unit reads "parked" while it is about to run. + /// + /// Currently reported by the dual-run trace only; nothing consumes it for a + /// verdict. It is what a verdict switch would have to gate on. + var parkGeneration: UInt64 { + lock.withLock { _parkGeneration } + } + + /// `true` while this unit is not parked — see the type doc for the counter + /// rule. A unit that has been created but whose body has not run yet is + /// running. + var isRunning: Bool { + lock.withLock { _activityCount > 0 } + } + + /// `true` once the wrapped Task's body has begun executing. Backs the + /// existing `hasPendingStartTask` verdict path, which is untouched by the + /// semantic-quiescence work. + var hasStartedRunning: Bool { + lock.withLock { _hasStartedRunning } + } + + func markBodyStarted() { + lock.withLock { + _hasStartedRunning = true + _lastActivityNs = _drainMonotonicNs() + } + } + + /// Classifies this unit for the combined quiescence rule — see + /// `AnyContext.semanticVerdict(nowNs:quietNs:)`. + /// + /// * **parked** — suspended where the framework put it; never blocks. + /// * **midFlight** — running, and *either* it has parked before (so it is + /// a unit whose suspensions we can see, currently between two of them: + /// a resumption, a yield hop, a starved job) *or* it changed state + /// within `quietNs` (it is visibly still moving). Waiting for it is the + /// correctness the semantic answer exists to provide. + /// * **looksUndeclared** — running, has *never* parked, and has produced + /// no transition for a whole `quietNs`. That is a compute loop or a + /// suspension the framework was never told about (an unadopted clock, + /// a bare `Task.sleep`, a foreign `await`). The framework cannot see + /// inside it, so it defers to the scheduler-observing answer. + /// + /// ## Why the park history is scoped to a region, not to the unit + /// + /// One work unit can span two code regions with completely different + /// visibility. `node.forEach`'s outer loop parks around its own `next()` + /// (hook 1) and then runs the USER's per-element closure **in the same + /// unit** — the body is deliberately outside the park, because running the + /// closure is real model work. So a unit whose loop has parked can be + /// sitting inside user code that suspends somewhere the framework knows + /// nothing about: + /// + /// node.forEach(stream) { value in + /// try await someUnadoptedClock.sleep(for: .seconds(1)) // invisible + /// } + /// + /// Attributing the *loop's* park to the *body* would say "this unit is one + /// we can see inside" about a stretch we demonstrably cannot, and the wait + /// would hang where it returns today (verified: `settle()` 22 ms → the trait + /// cap). `_withUserBodyRegion` marks those stretches, and inside one the + /// park history is ignored — only the quiet window decides. + /// + /// This does NOT give back the mid-flight guarantee the fallback exists to + /// protect, because the window it protects lies OUTSIDE the region: a + /// producer's eager unpark, the scheduling gap before the resumed consumer + /// returns from `next()`, and the return into the loop are all framework + /// stretches, where `_hasEverParked` still applies in full. A starved + /// resumption is therefore still waited for; only user code that has gone + /// quiet for a whole grace window inside its own body is allowed to fall + /// back — which is exactly the set the framework has no claim to know + /// about. (Residual: a hand-written `for await` loop, where SwiftModel does + /// not own the loop and so cannot mark the body. Adoption — + /// `withModelParked` at the clock — closes that one.) + func classify(nowNs: UInt64, quietNs: UInt64) -> _WorkUnitRunState { + lock.withLock { + guard _activityCount > 0 else { return .parked } + guard !_hasEverParked || _userBodyDepth > 0 else { return .midFlight } + guard nowNs >= _lastActivityNs, nowNs &- _lastActivityNs >= quietNs else { return .midFlight } + return .looksUndeclared + } + } + + /// Enters a user-body region — see `classify(nowNs:quietNs:)`. Entering is + /// itself a transition, so the body gets a full quiet window before it can + /// look undeclared. + fileprivate func beginUserBody() { + lock.withLock { + _userBodyDepth += 1 + _lastActivityNs = _drainMonotonicNs() + } + } + + fileprivate func endUserBody() { + lock.withLock { + _userBodyDepth -= 1 + _lastActivityNs = _drainMonotonicNs() + } + } + + /// Marks the unit parked (one level). Balanced by exactly one + /// `_ParkTicket.release()`, which the parking scope runs from a `defer` — + /// and which is a no-op if the unit was force-resumed in the meantime. + func park() -> _ParkTicket { + lock.withLock { + _activityCount -= 1 + _parkGeneration &+= 1 + _hasEverParked = true + _lastActivityNs = _drainMonotonicNs() + return _ParkTicket(unit: self, epoch: _epoch) + } + } + + /// EAGER UNPARK. Called by whoever is about to **resume** the continuation + /// this unit is parked on: the producer immediately before + /// `AsyncStream.Continuation.yield`, and the stream's `onTermination` + /// handler, which the stdlib documents as running before `next()` is + /// resumed with `nil` ("handler must be invoked before yielding nil for + /// termination", `AsyncStream._Storage.cancel`). + /// + /// Makes the unit running *now* and invalidates every open park scope, so + /// the resumed task's own `defer` adds nothing and the unit stays running + /// until it parks again of its own accord. + func noteResumeInFlight() { + lock.withLock { + if _activityCount < 1 { _activityCount = 1 } + _epoch &+= 1 + _parkGeneration &+= 1 + _lastActivityNs = _drainMonotonicNs() + } + } + + fileprivate func release(epoch: UInt64) { + lock.withLock { + // Stale: a `noteResumeInFlight()` already took this unit out of the + // park, so the scope's `defer` has nothing left to undo. + guard epoch == _epoch else { return } + _activityCount += 1 + _parkGeneration &+= 1 + _lastActivityNs = _drainMonotonicNs() + } + } +} + +/// One open park scope, handed out by `ModelWorkUnit.park()`. +/// +/// A value type on purpose: `park()` sits on the per-element path of every +/// SwiftModel-owned stream, so the ticket must not allocate. Idempotence +/// against an eager unpark comes from the epoch stamp, not from a per-ticket +/// flag. +struct _ParkTicket { + fileprivate let unit: ModelWorkUnit + fileprivate let epoch: UInt64 + + /// Ends the scope. A no-op if `ModelWorkUnit.noteResumeInFlight()` ran + /// while the scope was open — which is what makes it safe to call from a + /// `defer` that runs after an eager unpark. + func release() { + unit.release(epoch: epoch) + } +} + +/// Parks the *current* work unit for the duration of `body`, if there is one. +/// +/// Internal spelling of `withModelParked`, used by the framework's own hooks +/// (`forEach` parking around its `next()`); identical semantics, no public +/// surface. +/// +/// ## The cancellation handler +/// +/// This is the generic half of the eager unpark, and the only half available +/// for a **foreign** suspension — `node.forEach` over a third-party +/// `AsyncSequence` (design §6b hook 1), or a clock that adopted +/// `withModelParked` (hook 3). SwiftModel produces neither, so it cannot mark +/// the resume the way `_ParkSource` does for its own streams. But it *does* +/// own the cancellation: `Cancellations.cancelAll` → `TaskCancellable.onCancel` +/// → `Task.cancel()` runs this handler synchronously, before the cancelled task +/// is resumed — and cancellation is the resume that matters most, because it is +/// the one with no activity signal behind it and the one that goes straight +/// into model-writing `defer`s. +/// +/// The residual is a foreign source *delivering a value*: `swift-clocks`' +/// `ImmediateClock` timer resuming its consumer is not a cancellation and not +/// our continuation, so the unit reads parked until the resumed task runs. That +/// is inherent to hook 1 — the price of parking any `AsyncSequence` with no +/// adoption at all — and design §6b hook 2 (`parkedInModelTasks()`) is the +/// opt-in that would close it. +/// Runs `body` as a USER-CODE stretch of the current work unit. +/// +/// `node.forEach` / `node.onChange` run the user's per-element closure inside +/// the *same* unit as the loop that parked around `next()`. The framework knows +/// where its own loop suspends and nothing at all about where the closure does, +/// so the park history must not be carried into it — see +/// `ModelWorkUnit.classify(nowNs:quietNs:)`. A no-op outside a model task, and +/// a no-op for the `cancelPrevious` branches, whose bodies already get their +/// own `TaskCancellable` (and therefore their own unit, which has never +/// parked). +@inline(__always) +func _withUserBodyRegion(_ body: () async throws -> T) async rethrows -> T { + guard let unit = ModelWorkUnit.current else { return try await body() } + unit.beginUserBody() + defer { unit.endUserBody() } + return try await body() +} + +@inline(__always) +func _withCurrentWorkUnitParked(_ body: () async throws -> T) async rethrows -> T { + guard let unit = ModelWorkUnit.current else { return try await body() } + let ticket = unit.park() + defer { ticket.release() } + return try await withTaskCancellationHandler { + try await body() + } onCancel: { + unit.noteResumeInFlight() + } +} diff --git a/Sources/SwiftModel/Internal/QuiescenceComparison.swift b/Sources/SwiftModel/Internal/QuiescenceComparison.swift new file mode 100644 index 00000000..dd37a915 --- /dev/null +++ b/Sources/SwiftModel/Internal/QuiescenceComparison.swift @@ -0,0 +1,247 @@ +import Foundation + +// MARK: - Dual-run instrumentation for semantic quiescence +// +// Step 2 of `Docs/test-quiescence-redesign.md` §10: compute the SEMANTIC +// quiescence answer (`AnyContext.semanticQuiescence`) alongside the existing +// scheduler-observing answer on every check inside `_driveToStableFixpoint`, +// and record every disagreement. **The existing answer still decides every +// verdict** — nothing here feeds back into a wait. The point is to build the +// disagreement inventory over a whole suite run before anything depends on the +// new rule. +// +// Enable the per-disagreement log with `SWIFT_MODEL_QUIESCENCE_TRACE=1`; it is +// written to /tmp/swift-model-quiescence-trace.log (same pattern as the GTS +// trace), with a summary appended at process exit. The in-memory tally is +// always maintained — it is two integers behind a lock, only touched on a +// drive check, and it is what the unit tests assert against. + +/// Opt-in, *expensive* second-level diagnostic: symbolicate the stack of +/// whatever made each drain-executor job runnable, so a "new says quiescent, +/// old says busy — `existingBusy=[executor]`" disagreement can be attributed to +/// a concrete source instead of guessed at. Separate from +/// `SWIFT_MODEL_QUIESCENCE_TRACE` because it symbolicates on EVERY enqueue and +/// slows the suite by roughly an order of magnitude; use it on a filter, never +/// on a full run. +let _quiescenceJobTraceEnabled: Bool = + ProcessInfo.processInfo.environment["SWIFT_MODEL_QUIESCENCE_JOB_TRACE"] == "1" + +/// The current call stack, trimmed to the frames that identify the *source* of +/// an enqueue (the executor/dispatch plumbing at the top is noise). +func _quiescenceEnqueueStack() -> String { + #if os(WASI) + return "" + #else + return Thread.callStackSymbols.dropFirst(2).prefix(14) + .map { symbol -> String in + // `N Image 0xADDR mangled + off` → keep the mangled name only; + // addresses differ per run and defeat aggregation. + let parts = symbol.split(separator: " ", omittingEmptySubsequences: true) + return parts.count > 3 ? parts[3...].joined(separator: " ") : symbol + } + .joined(separator: " ← ") + #endif +} + +/// One disagreement direction. +enum _QuiescenceDisagreement: String, Sendable { + /// Existing (executor/queue) answer said quiescent; semantic said work is + /// still running. The interesting direction: the old answer would let a + /// wait conclude while a registered unit is still running. + case oldQuiescentNewNot + /// Semantic answer said quiescent; existing answer said busy. Usually + /// executor/queue churn that owns no registered work unit. + case newQuiescentOldNot +} + +struct _QuiescenceTally: Sendable, Equatable { + var checks = 0 + var agreements = 0 + var oldQuiescentNewNot = 0 + var newQuiescentOldNot = 0 + /// How many waits concluded via the combined rule's UNDECLARED-WORK + /// fallback: the existing answer said "done", the semantic answer said work + /// was still running, and every running unit looked undeclared (never + /// parked, silent for a grace window) so the old answer was deferred to. + /// Every one of these is a site where `withModelParked` adoption would turn + /// a guess into knowledge. + var undeclaredFallbacks = 0 +} + +enum _QuiescenceComparison { + /// The name of the test whose scope we are in, set by `ModelTestingTrait`. + @TaskLocal static var testTag: String? + + static let isTracing: Bool = ProcessInfo.processInfo.environment["SWIFT_MODEL_QUIESCENCE_TRACE"] == "1" + + private static let lock = NSLock() + nonisolated(unsafe) private static var _tally = _QuiescenceTally() + /// Per-test disagreement counts, for the exit summary. + nonisolated(unsafe) private static var _byTest: [String: (old: Int, new: Int, checks: Int)] = [:] + /// `site → (occurrences, tests that hit it)` for the undeclared-work + /// fallback. This is the adoption worklist: every entry is a suspension + /// SwiftModel had to guess about. + nonisolated(unsafe) private static var _undeclaredSites: [String: (count: Int, tests: Set)] = [:] + + static var tally: _QuiescenceTally { lock.withLock { _tally } } + + static func resetTally() { + lock.withLock { + _tally = _QuiescenceTally() + _byTest = [:] + _undeclaredSites = [:] + } + } + + /// One wait concluded through the combined rule's undeclared-work fallback. + /// `units` is every running unit at that instant (all of which looked + /// undeclared — that is the precondition for the fallback). + static func recordUndeclaredFallback(_ units: [_RunningWorkUnitInfo]) { + let tag = testTag ?? "" + lock.withLock { + _tally.undeclaredFallbacks += 1 + for unit in units { + var entry = _undeclaredSites[unit.description] ?? (count: 0, tests: []) + entry.count += 1 + entry.tests.insert(tag) + _undeclaredSites[unit.description] = entry + } + } + guard isTracing else { return } + _quiescenceTrace( + "test=\"\(tag)\" undeclaredFallback running=[\(units.map(\.description).sorted().joined(separator: ", "))] count=\(units.count)" + ) + } + + /// `(site, occurrences, tests)` for the undeclared-work fallback, most + /// frequent first. + static func undeclaredFallbackSites() -> [(site: String, count: Int, tests: [String])] { + lock.withLock { _undeclaredSites } + .map { (site: $0.key, count: $0.value.count, tests: $0.value.tests.sorted()) } + .sorted { $0.count == $1.count ? $0.site < $1.site : $0.count > $1.count } + } + + /// Records one check. `runningUnits` and `existingBusyReason` are only + /// evaluated when a disagreement is being traced. + static func record( + existingIsQuiescent: Bool, + semanticIsQuiescent: Bool, + runningUnits: @autoclosure () -> [(modelName: String, name: String, fileAndLine: FileAndLine)], + existingBusyReason: @autoclosure () -> String = "", + existingBusyJobStacks: @autoclosure () -> [String] = [], + workUnitCensus: @autoclosure () -> _WorkUnitCensus = _WorkUnitCensus(), + parkGenerationChangedSinceLastCheck: Bool? = nil + ) { + let tag = testTag ?? "" + let disagreement: _QuiescenceDisagreement? + switch (existingIsQuiescent, semanticIsQuiescent) { + case (true, false): disagreement = .oldQuiescentNewNot + case (false, true): disagreement = .newQuiescentOldNot + default: disagreement = nil + } + + lock.withLock { + _tally.checks += 1 + var entry = _byTest[tag] ?? (old: 0, new: 0, checks: 0) + entry.checks += 1 + switch disagreement { + case .none: _tally.agreements += 1 + case .oldQuiescentNewNot: _tally.oldQuiescentNewNot += 1; entry.old += 1 + case .newQuiescentOldNot: _tally.newQuiescentOldNot += 1; entry.new += 1 + } + _byTest[tag] = entry + } + + guard isTracing, let disagreement else { return } + var line = "test=\"\(tag)\" \(disagreement.rawValue) existing=\(existingIsQuiescent ? "quiescent" : "busy") semantic=\(semanticIsQuiescent ? "quiescent" : "busy")" + switch disagreement { + case .oldQuiescentNewNot: + let units = runningUnits() + if units.isEmpty { + line += " running=[queue]" // no registered unit; a call queue was busy + } else { + // Cap the list: a wide hierarchy can have dozens of identical + // units and the interesting information is the call site. + let described = units.map { unit -> String in + let site = unit.fileAndLine.description + // `taskName` defaults to "function @ file:line" — don't repeat the site. + return unit.name.hasSuffix(site) ? "\(unit.modelName).\(unit.name)" : "\(unit.modelName).\(unit.name) @ \(site)" + } + var shown = Array(Set(described)).sorted().prefix(5).joined(separator: ", ") + if units.count > 5 { shown += ", +\(units.count - 5) more" } + line += " running=[\(shown)] count=\(units.count)" + } + case .newQuiescentOldNot: + let census = workUnitCensus() + line += " existingBusy=[\(existingBusyReason())] units=(registered: \(census.registered), parked: \(census.parked))" + if let changed = parkGenerationChangedSinceLastCheck { + line += " parkGenChanged=\(changed)" + } + if _quiescenceJobTraceEnabled { + for stack in existingBusyJobStacks() { + line += "\n job: \(stack)" + } + } + } + _quiescenceTrace(line) + } + + static func summaryLines() -> [String] { + let (tally, byTest) = lock.withLock { (_tally, _byTest) } + var lines: [String] = [] + lines.append("=== SEMANTIC QUIESCENCE DISAGREEMENT SUMMARY ===") + lines.append("checks=\(tally.checks) agree=\(tally.agreements) oldQuiescentNewNot=\(tally.oldQuiescentNewNot) newQuiescentOldNot=\(tally.newQuiescentOldNot) undeclaredFallbacks=\(tally.undeclaredFallbacks)") + let sites = undeclaredFallbackSites() + if !sites.isEmpty { + lines.append("--- undeclared-work fallback sites (adoption worklist) ---") + for site in sites { + lines.append(" \(site.site): \(site.count)× in \(site.tests.count) test(s): \(site.tests.prefix(4).joined(separator: ", "))") + } + } + let interesting = byTest.filter { $0.value.old > 0 || $0.value.new > 0 } + .sorted { ($0.value.old + $0.value.new) > ($1.value.old + $1.value.new) } + for (test, counts) in interesting { + lines.append(" \(test): checks=\(counts.checks) oldQuiescentNewNot=\(counts.old) newQuiescentOldNot=\(counts.new)") + } + return lines + } +} + +// MARK: - Trace file + +private let _quiescenceTraceFile: FileHandle? = { + guard _QuiescenceComparison.isTracing else { return nil } + let path = "/tmp/swift-model-quiescence-trace.log" + try? FileManager.default.removeItem(atPath: path) + _ = FileManager.default.createFile(atPath: path, contents: nil) + let handle = try? FileHandle(forWritingTo: URL(fileURLWithPath: path)) + // The exit summary is a convenience over the per-check lines (which carry + // the same information). `atexit` is not part of the WASI surface this + // library is compile-checked against, and the trace is a macOS/Linux + // developer tool, so it is simply skipped there. + #if !os(WASI) + atexit(_quiescenceDumpSummaryAtExit) + #endif + return handle +}() + +private let _quiescenceTraceLock = NSLock() + +func _quiescenceTrace(_ msg: @autoclosure () -> String) { + guard _QuiescenceComparison.isTracing, let fh = _quiescenceTraceFile else { return } + let line = msg() + "\n" + _quiescenceTraceLock.withLock { + try? fh.write(contentsOf: Data(line.utf8)) + } +} + +/// `atexit` handler — must capture nothing (`@convention(c)`). +#if !os(WASI) +private func _quiescenceDumpSummaryAtExit() { + guard let fh = _quiescenceTraceFile else { return } + let text = _QuiescenceComparison.summaryLines().joined(separator: "\n") + "\n" + _quiescenceTraceLock.withLock { + try? fh.write(contentsOf: Data(text.utf8)) + } +} +#endif diff --git a/Sources/SwiftModel/Internal/TestAccess.swift b/Sources/SwiftModel/Internal/TestAccess.swift index ac50cdb9..ea01d1a5 100644 --- a/Sources/SwiftModel/Internal/TestAccess.swift +++ b/Sources/SwiftModel/Internal/TestAccess.swift @@ -542,6 +542,42 @@ final class TestAccess: ModelAccess, @unchecked Sendable { _fireStatsLock.withLock { _fireStats } } + // MARK: - Undeclared-work fallback diagnostic + + /// Every work unit that has, at least once during this test, been the + /// reason `_driveToStableFixpoint`'s combined rule fell back to the + /// scheduler-observing answer — i.e. it was running, had never parked, and + /// had been silent for a whole grace window. + /// + /// The user-facing value is that the *site* can be found: a clock that has + /// not adopted `withModelParked` is a place where SwiftModel's quiescence + /// answer is guessing rather than knowing. It is a diagnostic, never a + /// failure — falling back is the documented, correct behaviour for work the + /// framework cannot see inside. Surfaced only when some other wait in the + /// same test later times out (`settleDiagnostics()`), and — for every + /// occurrence — under `SWIFT_MODEL_QUIESCENCE_TRACE=1`. + /// + /// Shares `_fireStatsLock` for the same reason the fire stats have their + /// own lock: it must never contend the main access lock. + nonisolated(unsafe) private var _undeclaredFallbackSites: [_RunningWorkUnitInfo: Int] = [:] + + func _noteUndeclaredWorkFallback(_ units: [_RunningWorkUnitInfo]) { + _fireStatsLock.withLock { + for unit in units { _undeclaredFallbackSites[unit, default: 0] += 1 } + } + // Process-wide tally + per-occurrence trace line (item 4: how often the + // fallback actually fires across a suite, and where). + _QuiescenceComparison.recordUndeclaredFallback(units) + } + + /// `(site description, times it was the fallback's reason)`, most frequent + /// first. Empty when the fallback never fired in this test. + func _undeclaredWorkFallbackSites() -> [(site: String, count: Int)] { + _fireStatsLock.withLock { _undeclaredFallbackSites } + .map { (site: $0.key.description, count: $0.value) } + .sorted { $0.count == $1.count ? $0.site < $1.site : $0.count > $1.count } + } + // Erased executor-drain hooks (overrides must be in the class body, not an // extension). Implementations live in TestExecutorDrive.swift. override var hasTestExecutorErased: Bool { _isExecutorDriveActive } diff --git a/Sources/SwiftModel/Internal/TestExecutorDrive.swift b/Sources/SwiftModel/Internal/TestExecutorDrive.swift index f5766599..764d37fe 100644 --- a/Sources/SwiftModel/Internal/TestExecutorDrive.swift +++ b/Sources/SwiftModel/Internal/TestExecutorDrive.swift @@ -139,6 +139,15 @@ private let _sharedDrainQueue = DispatchQueue(label: "swift-model.test-drain.sha final class _DrainTestExecutor: TaskExecutor, @unchecked Sendable { private let lock = NSLock() private var outstanding = 0 + /// Per-outstanding-job enqueue backtraces, kept ONLY when + /// `SWIFT_MODEL_QUIESCENCE_JOB_TRACE=1`. This is the instrument that turned + /// the semantic-quiescence "new says quiescent, old says busy" bucket from a + /// guess into a classification: at a disagreement it names, per still-ready + /// job, the stack of whatever made that job runnable. Disabled it costs one + /// already-loaded `Bool` per enqueue; enabled it symbolicates a stack per + /// enqueue and is far too slow for anything but a targeted diagnostic run. + private var _jobStacks: [UInt64: String] = [:] + private var _nextJobId: UInt64 = 0 /// Birth time — the floor for `activityNs` so a test that hasn't yet /// enqueued any executor work reads "active as of now", not the epoch. (The /// raw timestamps below start at 0; `monotonicNs` is a large uptime value, @@ -201,9 +210,23 @@ final class _DrainTestExecutor: TaskExecutor, @unchecked Sendable { } } + /// The enqueue stacks of every job that is still ready/running, newest + /// first. Empty unless `SWIFT_MODEL_QUIESCENCE_JOB_TRACE=1`. + var outstandingJobStacks: [String] { + lock.withLock { _jobStacks.sorted { $0.key > $1.key }.map(\.value) } + } + func enqueue(_ job: consuming ExecutorJob) { let unowned = UnownedJob(job) - lock.withLock { outstanding += 1; _lastEnqueueNs = _drainMonotonicNs() } + let stack = _quiescenceJobTraceEnabled ? _quiescenceEnqueueStack() : nil + let jobId: UInt64 = lock.withLock { + outstanding += 1 + _lastEnqueueNs = _drainMonotonicNs() + guard let stack else { return 0 } + _nextJobId += 1 + _jobStacks[_nextJobId] = stack + return _nextJobId + } Self._globalOutstanding.wrappingAdd(1, ordering: .relaxed) Self._globalLastActivityNs.store(_drainMonotonicNs(), ordering: .relaxed) _sharedDrainQueue.async { @@ -211,6 +234,7 @@ final class _DrainTestExecutor: TaskExecutor, @unchecked Sendable { let toFire: [@Sendable () -> Void] = self.lock.withLock { self.outstanding -= 1 self._lastCompletionNs = _drainMonotonicNs() + if jobId != 0 { self._jobStacks.removeValue(forKey: jobId) } guard self.outstanding == 0 else { return [] } let fns = self.idleWaiters.map(\.fire) self.idleWaiters.removeAll() @@ -342,6 +366,11 @@ extension TestAccess { // re-check cadence (not the ceiling) so a runaway that never lets // the executor go idle is still inspected periodically. let fireBaseline = runawayBound != nil ? _reactiveFireStats() : [:] + // Dual-run instrumentation only (design §5 property 2): the + // park-generation observed at the PREVIOUS check, so a disagreement + // can say whether some unit parked or unparked in between. `nil` on + // the first iteration. + var previousParkGeneration: UInt64? = nil while !Task.isCancelled { let now = _drainMonotonicNs() if now >= hangDeadlineNs { return .gaveUp } @@ -353,6 +382,32 @@ extension TestAccess { if !bg.isIdle { await bg.waitForCurrentItems(deadline: checkDeadline) } if !main.isIdle { await main.waitForCurrentItems(deadline: checkDeadline) } let idleNow = exec.isExecutorIdle && bg.isIdle && main.isIdle && !self.context.hasPendingStartTask + // One census read per iteration (cheap: a locked walk of the + // live-unit dictionaries). Feeds the trace only. + let census = _QuiescenceComparison.isTracing ? self.context.workUnitCensus : _WorkUnitCensus() + + // DUAL-RUN INSTRUMENTATION (step 2 of the semantic-quiescence + // plan — `Docs/test-quiescence-redesign.md` §10). Compute the + // semantic answer beside the existing one and record every + // disagreement. `idleNow` — NOT the semantic answer — still + // decides this and every other verdict; this call has no effect + // on control flow. + _QuiescenceComparison.record( + existingIsQuiescent: idleNow, + semanticIsQuiescent: self.context.semanticQuiescence, + runningUnits: self.context.runningWorkUnits, + existingBusyReason: [ + exec.isExecutorIdle ? nil : "executor", + bg.isIdle ? nil : "bg", + main.isIdle ? nil : "main", + self.context.hasPendingStartTask ? "pendingStart" : nil, + ].compactMap { $0 }.joined(separator: "+"), + existingBusyJobStacks: exec.outstandingJobStacks, + workUnitCensus: census, + parkGenerationChangedSinceLastCheck: previousParkGeneration.map { $0 != census.parkGeneration } + ) + previousParkGeneration = census.parkGeneration + if idleNow { // Debounce against COMPLETIONS too, not just writes and // enqueues (`exec.activityNs` when idle = max(birth, @@ -373,7 +428,60 @@ extension TestAccess { let lastActivity = max(self._lastActivityNsLocked, exec.activityNs) let sinceActivity = _drainMonotonicNs() &- lastActivity if sinceActivity >= graceNs { - return .reached // idle, and no activity of any kind for a full grace window + // THE COMBINED RULE (`Docs/test-quiescence-redesign.md` + // §3/§6, adoption increment). The existing + // scheduler-observing answer has just said "done"; the + // SEMANTIC answer now gets a veto: + // + // quiescent := oldSaysDone + // && (newSaysDone || runningWorkLooksUndeclared) + // + // `oldSaysDone` stays a REQUIRED conjunct, which is what + // makes this increment safe on its own: the residual + // window where a foreign source delivers a value we + // cannot observe (design §6b hook 1's known price) + // could only ever make the semantic answer say "done" + // early, and an early "done" it cannot act on. So the + // only verdicts that change are ones where the old + // answer passed and the new one says work is still + // running — i.e. we now WAIT where we used to conclude. + // + // `runningWorkLooksUndeclared` is the fallback, and the + // whole reason the design is adoptable incrementally: + // work SwiftModel cannot see inside (an unadopted + // clock's sleep, a bare `Task.sleep`, a compute loop) is + // correctly reported as *running*, and believing that + // would hang waits that return today. A running unit + // that has NEVER parked and has produced no transition + // for a full settle grace is exactly that shape, and + // only there do we defer to the old answer — + // reproducing today's behaviour bit for bit. + // + // A running unit that HAS parked before, or that + // transitioned within the grace, is mid-flight (a yield + // hop, a resumption already in flight, a starved job). + // The fallback must not swallow those: waiting for them + // is the correctness fix this design exists for, so a + // single mid-flight unit anywhere in the subtree keeps + // the wait open (`allRunningLookUndeclared` is an + // all-quantifier, not an any-quantifier). + let verdict = self.context.semanticVerdict( + nowNs: _drainMonotonicNs(), + quietNs: Self._settleGraceNs + ) + if verdict.isQuiescent { + return .reached // both answers agree: done + } + if verdict.allRunningLookUndeclared { + self._noteUndeclaredWorkFallback(verdict.undeclared) + return .reached + } + // Semantic answer says real framework-owned work is + // still mid-flight. Keep waiting — no wall clock is + // added here; the loop is bounded by the same + // `hangDeadlineNs` as every other path. + await _gtsSleep(Self._settleGraceNs, hangDeadlineNs: hangDeadlineNs) + continue } // Idle but recent activity — wait out the remainder of the // grace (non-starvable), then re-check; a resuming task will diff --git a/Sources/SwiftModel/Internal/TestWaitSupport.swift b/Sources/SwiftModel/Internal/TestWaitSupport.swift index c83c6392..f9ea4012 100644 --- a/Sources/SwiftModel/Internal/TestWaitSupport.swift +++ b/Sources/SwiftModel/Internal/TestWaitSupport.swift @@ -202,13 +202,40 @@ extension TestAccess { lines.append(" \(info.modelName): \"\(taskName)\" @ \(fl.fileID):\(fl.line)") } } - let listing = lines.joined(separator: "\n") + var listing = lines.joined(separator: "\n") + if let undeclared = _undeclaredWorkDiagnosticLine() { + listing += (listing.isEmpty ? "" : "\n") + undeclared + } if let runaway = _runawayDiagnosticLine() { return runaway + "\n" + listing } return listing } + /// If some earlier wait in this test concluded through the combined + /// quiescence rule's UNDECLARED-WORK fallback, name the sites. + /// + /// This is not the cause of the timeout — the fallback lets a wait finish, + /// it never stalls one. It is here because a timeout is the moment someone + /// is actually reading this output, and an undeclared suspension is the one + /// thing about this test's timing that SwiftModel had to guess at. A clock + /// that adopts `withModelParked` moves from "guessed" to "known". Returns + /// nil (and so changes no existing message) when the fallback never fired. + private func _undeclaredWorkDiagnosticLine() -> String? { + let sites = _undeclaredWorkFallbackSites() + guard !sites.isEmpty else { return nil } + let named = sites.prefix(5).map { " • \($0.site) (\($0.count)×)" }.joined(separator: "\n") + return """ + ℹ️ undeclared async work: SwiftModel could not see where the following task(s) suspend, + so quiescence for them fell back to observing the scheduler: + \(named) + Each has never parked at a suspension point the framework owns. If one of them + sleeps on a clock (or any other bare suspension), wrap that source's own `sleep` + in `withModelParked { }` — see Docs/Testing.md. `node.forEach` over any + AsyncSequence already parks with no adoption. + """ + } + /// If exactly the failure shape "a reactive body that never stops firing" /// is present, name it. The runaway is the call site that (a) fired far more /// than a one-shot would and (b) was *still firing* at the timeout — i.e. the diff --git a/Sources/SwiftModel/Model+Changes.swift b/Sources/SwiftModel/Model+Changes.swift index 8c2ede60..970b78ca 100644 --- a/Sources/SwiftModel/Model+Changes.swift +++ b/Sources/SwiftModel/Model+Changes.swift @@ -64,7 +64,14 @@ public extension Model { ) -> AsyncStream<()> { guard let context = enforcedContext() else { return .finished } - return AsyncStream { cont in + // `_makeParkedStream` adds the tier-1 park mark (design §5): a consumer + // suspended waiting for the next modification is parked, so a + // hand-written `for await _ in observeModifications() { … }` inside a + // `node.task` reads as parked rather than running forever — and every + // resume of that wait (`cont.yield`, `cont.finish`, the consumer being + // cancelled) unparks it EAGERLY, at the resume rather than at the + // resumed task's next CPU slot. See `AsyncSequenceExtensions.swift`. + return _makeParkedStream { cont in #if DEBUG // Capture label and printer once at setup time, not on every emission. let debugState: (label: String, printer: PrinterBox)? = debug.flatMap { d in @@ -361,7 +368,15 @@ public extension ModelNode { extension Observed { init(access: @Sendable @escaping () -> Element, initial: Bool = true, isSame: (@Sendable (Element, Element) -> Bool)?, coalesceUpdates: Bool = false, debug: DebugOptions? = nil) { - stream = AsyncStream { cont in + // `_makeParkedStream` is the tier-1 park mark (design §5). `Observed` is + // the single most common hand-written loop in the suite — + // `node.task { for await v in Observed { … } }` — and without a + // source-side mark every one of those tasks reads as *running* for its + // whole lifetime. Marking the source (rather than `forEach`) makes the + // sugar and the hand-written loop behave identically; yielding through + // the wrapper makes each resume of the consumer's wait unpark it + // eagerly. See `AsyncSequenceExtensions.swift`. + stream = _makeParkedStream { cont in // Detect whether accessed models use ObservationRegistrar. // If any accessed model was created with .disableObservationRegistrar, the // withObservationTracking path won't fire (the model's access(path:from:) is a no-op). diff --git a/Sources/SwiftModel/ModelNode+Events.swift b/Sources/SwiftModel/ModelNode+Events.swift index 00aae83e..bcb89957 100644 --- a/Sources/SwiftModel/ModelNode+Events.swift +++ b/Sources/SwiftModel/ModelNode+Events.swift @@ -42,6 +42,13 @@ public extension ModelNode { } } +// The tier-1 park mark for events lives in `AnyContext.events()`, NOT here. +// These APIs are `events().compactMap { … }`, and a park scope wrapped around +// the filter would stay open across events that fail it — so an eager unpark at +// the yield would strand the consumer's work unit reading *running* until the +// next event that happens to match. Parking the raw source instead means the +// scope is entered and left exactly once per event, and the filter re-parks by +// construction. See `AsyncSequenceExtensions.swift`. public extension ModelNode { /// Returns a stream of all events sent from this model or any of its descendants, typed as `Any`. /// diff --git a/Sources/SwiftModel/ModelNode+Reactive.swift b/Sources/SwiftModel/ModelNode+Reactive.swift index 8e5a7031..a77ccaea 100644 --- a/Sources/SwiftModel/ModelNode+Reactive.swift +++ b/Sources/SwiftModel/ModelNode+Reactive.swift @@ -291,7 +291,13 @@ public extension ModelNode { guard cancelPrevious else { let fireFL = FileAndLine(fileID: fileID, filePath: filePath, line: line, column: column) return task(name, function: function, isDetached: isDetached, priority: priority, fileID: fileID, filePath: filePath, line: line, column: column) { - for await newValue in observed { + // Hook 1 (semantic quiescence, `Docs/test-quiescence-redesign.md` §6b): + // the loop is written out rather than using `for await` so the + // wait for the next element — a suspension SwiftModel owns — + // marks this work unit PARKED. The body is deliberately outside + // the park: running the user's closure is real work. + var iterator = observed.makeAsyncIterator() + while let newValue = await _withCurrentWorkUnitParked({ await iterator.next() }) { guard !Task.isCancelled, !context.isDestructed else { return } // Count this delivery for the settle-timeout runaway trigger and // diagnostic (no-op outside tests). See ModelAccess.reactiveBodyFired. @@ -305,7 +311,12 @@ public extension ModelNode { } do { - try await operation(oldValue, newValue) + // The user's closure runs in THIS unit, which has + // parked around `next()`. Mark the stretch so the + // quiescence rule does not credit the loop's park + // history to code it cannot see inside — see + // `_withUserBodyRegion`. + try await _withUserBodyRegion { try await operation(oldValue, newValue) } } catch { onError(error) } @@ -330,7 +341,10 @@ public extension ModelNode { // scheduled. See `_forEachImpl` for the full rationale. var previousInner: TaskCancellable? = nil - for await newValue in observed { + // Hook 1 — park around the element wait only; see the non- + // cancelPrevious branch above. + var iterator = observed.makeAsyncIterator() + while let newValue = await _withCurrentWorkUnitParked({ await iterator.next() }) { guard !Task.isCancelled, !context.isDestructed else { break } // Count this delivery for the settle-timeout runaway trigger and // diagnostic (no-op outside tests). See ModelAccess.reactiveBodyFired. @@ -404,14 +418,23 @@ public extension ModelNode { guard cancelPrevious else { let fireFL = FileAndLine(fileID: fileID, filePath: filePath, line: line, column: column) return task(name, function: function, isDetached: isDetached, priority: priority, fileID: fileID, filePath: filePath, line: line, column: column, operation: { - for try await value in sequence { + // Hook 1 (semantic quiescence, `Docs/test-quiescence-redesign.md` §6b): + // park around `next()` only. This covers ANY `AsyncSequence` — + // including third-party operators such as + // swift-async-algorithms' `debounce`/`throttle` — because the + // consumer is suspended on this one await no matter how many + // tasks the operator runs internally, with no adoption by the + // sequence's author. + var iterator = sequence.makeAsyncIterator() + while let value = try await _withCurrentWorkUnitParked({ try await iterator.next() }) { guard !Task.isCancelled, !context.isDestructed else { return } // Count this delivery for the settle-timeout runaway diagnostic // (no-op outside tests). See ModelAccess.reactiveBodyFired. ModelAccess.current?.reactiveBodyFired(fireFL) do { - try await operation(value) + // See the `onChange` branch above / `_withUserBodyRegion`. + try await _withUserBodyRegion { try await operation(value) } } catch { if abortIfOperationThrows { throw error @@ -447,7 +470,10 @@ public extension ModelNode { // closure deadlocked whenever cancellation arrived before the body was scheduled. var previousInner: TaskCancellable? = nil - for try await value in sequence { + // Hook 1 — park around the element wait only; see the non- + // cancelPrevious branch above. + var iterator = sequence.makeAsyncIterator() + while let value = try await _withCurrentWorkUnitParked({ try await iterator.next() }) { // Count this delivery for the settle-timeout runaway diagnostic // (no-op outside tests). See ModelAccess.reactiveBodyFired. ModelAccess.current?.reactiveBodyFired(fileAndLine) diff --git a/Sources/SwiftModel/SwiftUI/ObservedModel.swift b/Sources/SwiftModel/SwiftUI/ObservedModel.swift index c3b1fbd6..c7937e35 100644 --- a/Sources/SwiftModel/SwiftUI/ObservedModel.swift +++ b/Sources/SwiftModel/SwiftUI/ObservedModel.swift @@ -599,6 +599,14 @@ internal final class ViewAccess: ModelAccess, ObservableObject, @unchecked Senda return firstActivation } if needsPriming { + // SEMANTIC QUIESCENCE — excluded housekeeping, category (d). + // Justification: this is a SwiftUI view invalidation + // (`objectWillChange`), reached only from `attachDebug` on the first + // render that requests debug output. It touches no model state, so + // no model assertion can depend on it. (Design §4a suggests routing + // it through `mainCallQueue` so it would be counted for free; that + // is a SwiftUI-timing change and is deliberately left out of the + // quiescence work.) See the audit in `ModelWorkUnit.swift`. Task { @MainActor [weak self] in self?.objectWillChange.send() } diff --git a/Sources/SwiftModel/Testing/ModelTestingTrait.swift b/Sources/SwiftModel/Testing/ModelTestingTrait.swift index bcc9a9f4..b29fe687 100644 --- a/Sources/SwiftModel/Testing/ModelTestingTrait.swift +++ b/Sources/SwiftModel/Testing/ModelTestingTrait.swift @@ -565,6 +565,9 @@ extension ModelTestingTrait: TestScoping, TestTrait, SuiteTrait { activityProbe = nil #endif try await withoutActuallyEscaping(function) { escapingFunction in + // Names this test in the semantic-quiescence disagreement trace + // (`SWIFT_MODEL_QUIESCENCE_TRACE=1`). Instrumentation only. + try await _QuiescenceComparison.$testTag.withValue(testTag) { try await _withTestTimeout(seconds: ModelTestingTraitOptions.testWallClockSeconds, testTag: testTag, activityProbe: activityProbe) { try await _TestExecutorBox.$current.withValue(execBox) { try await _ModelTestingLocals.$scope.withValue(pending) { @@ -577,6 +580,7 @@ extension ModelTestingTrait: TestScoping, TestTrait, SuiteTrait { } } } + } } } } diff --git a/Sources/SwiftModel/WithModelParked.swift b/Sources/SwiftModel/WithModelParked.swift new file mode 100644 index 00000000..d5675499 --- /dev/null +++ b/Sources/SwiftModel/WithModelParked.swift @@ -0,0 +1,38 @@ +/// Marks the calling model task **parked** for the duration of `body`. +/// +/// SwiftModel's test-wait verbs (`expect`, `settle`, `waitUntil`) need to know +/// when a model is done reacting. Work that is suspended waiting for input that +/// can only arrive from outside — a clock deadline, an external event source — +/// is not "reacting"; work that is executing, or suspended somewhere the +/// framework cannot see, is. SwiftModel marks its own suspension points +/// automatically (notably `node.forEach`, which parks around its `next()`, so +/// any `AsyncSequence` — including `swift-async-algorithms` `debounce` / +/// `throttle` — parks with no adoption at all). +/// +/// Use this primitive when you hand a model a *bare* suspension that is not an +/// `AsyncSequence`: the canonical case is a clock's own `sleep` implementation. +/// The adoption point is the source's implementation, not its call sites — a +/// clock protocol that funnels every caller through one `sleep` method needs +/// one wrap, and every `clock.sleep` in every model is covered: +/// +/// ```swift +/// extension MyClock { +/// public func sleep(until date: Date) async throws { +/// try await withModelParked { +/// try await self.nonAdjustedSleep(until: date) +/// } +/// } +/// } +/// ``` +/// +/// The current work unit is found through a task-local set when a model task +/// body starts, so a suspension inside a child task the library spawned still +/// marks the right unit. Outside any model task this is a plain passthrough — +/// it never traps and never has an effect. +/// +/// - Note: Not marking something is safe, merely less precise: unmarked +/// suspensions count as running work, so a wait verb waits for them and can +/// report them by name rather than being silently wrong. +public func withModelParked(_ body: () async throws -> T) async rethrows -> T { + try await _withCurrentWorkUnitParked(body) +} diff --git a/Tests/SwiftModelTests/ClockTests.swift b/Tests/SwiftModelTests/ClockTests.swift index 0e90f556..d9dc2d04 100644 --- a/Tests/SwiftModelTests/ClockTests.swift +++ b/Tests/SwiftModelTests/ClockTests.swift @@ -38,7 +38,7 @@ struct ClockTests { @Test func testClockStepByStep() async { let clock = TestClock() let model = TimerModel().withAnchor { - $0.continuousClock = clock + $0.continuousClock = ParkedClock(clock) } // The `forEach(clock.timer(...))` consumer subscribes to the clock // lazily, on its first `next()`. Settling here (and between steps) @@ -68,7 +68,7 @@ struct ClockTests { @available(iOS 16, macOS 13, tvOS 16, watchOS 9, *) @Test(.modelTesting(.removing(.state))) func testImmediateClock() async { let model = TimerModel().withAnchor { - $0.continuousClock = ImmediateClock() + $0.continuousClock = ParkedClock(ImmediateClock()) } // ImmediateClock drives the timer as fast as the model processes it; // we just need to let it settle before asserting. diff --git a/Tests/SwiftModelTests/OnChangeTests.swift b/Tests/SwiftModelTests/OnChangeTests.swift index 251060d3..6a143daf 100644 --- a/Tests/SwiftModelTests/OnChangeTests.swift +++ b/Tests/SwiftModelTests/OnChangeTests.swift @@ -108,7 +108,7 @@ struct OnChangeTests { @Test func testOnChangeCancelPreviousDiscardsStalework() async { let clock = TestClock() let model = OnChangeCancelPreviousModel().withAnchor { - $0.continuousClock = clock + $0.continuousClock = ParkedClock(clock) } // initial: false — no initial emission await settle {} diff --git a/Tests/SwiftModelTests/SemanticQuiescenceTests.swift b/Tests/SwiftModelTests/SemanticQuiescenceTests.swift new file mode 100644 index 00000000..d713a4b6 --- /dev/null +++ b/Tests/SwiftModelTests/SemanticQuiescenceTests.swift @@ -0,0 +1,1048 @@ +import Testing +import Foundation +import Dependencies +import ConcurrencyExtras +@testable import SwiftModel + +// Tests for the SEMANTIC quiescence accounting (`Docs/test-quiescence-redesign.md` +// §3–§6b): every `TaskCancellable` is a work unit that is either RUNNING (its +// body is executing, or suspended somewhere SwiftModel did not put it) or +// PARKED (suspended at a suspension point the framework owns, or inside +// `withModelParked`). +// +// Nothing here exercises a verdict: `AnyContext.semanticQuiescence` is computed +// alongside the existing executor/queue answer and compared, but the existing +// answer still decides every `expect`/`settle`/`waitUntil`. These tests assert +// the accounting directly, which is the point of §8 — a semantic invariant is +// deterministic and needs no loaded machine to validate. + +// MARK: - Test fixtures + +/// A clock SwiftModel knows nothing about: it conforms to no protocol the +/// framework can see (not `_Concurrency.Clock`, not anything of ours), suspends +/// on a raw continuation, and only resumes when the test calls `advance()`. +/// This is the shape design §6c calls "waiting for the test to act". +final class ForeignClock: @unchecked Sendable { + private let lock = NSLock() + private var continuations: [CheckedContinuation] = [] + private var isCancelled = false + + /// Number of sleepers currently suspended. + var sleeperCount: Int { lock.withLock { continuations.count } } + + func sleep() async { + await withTaskCancellationHandler { + await withCheckedContinuation { (continuation: CheckedContinuation) in + let resumeImmediately: Bool = lock.withLock { + if isCancelled { return true } + continuations.append(continuation) + return false + } + if resumeImmediately { continuation.resume() } + } + } onCancel: { + // Keep test teardown honest: a cancelled task must be able to finish. + self.lock.withLock { self.isCancelled = true } + self.advance() + } + } + + /// Releases every current sleeper. + func advance() { + let pending: [CheckedContinuation] = lock.withLock { + defer { continuations = [] } + return continuations + } + for continuation in pending { continuation.resume() } + } +} + +/// Shared out-of-band control surface for the models below: stop flags, an +/// externally-fed stream, and a gate the `forEach` body can be held inside. +final class QuiescenceControl: @unchecked Sendable { + let stop = LockIsolated(false) + let bodyEntered = LockIsolated(false) + /// Opt-in: when true, the consumer body parks at the gate after each delivery, so + /// "the unit is running" is a durable state a test can observe rather than a + /// transient one it has to catch. Counts deliveries so a test can wait for one. + let holdBody = LockIsolated(false) + let bodyIterations = LockIsolated(0) + + private let lock = NSLock() + private var streamContinuation: AsyncStream.Continuation? + private var gateContinuations: [CheckedContinuation] = [] + private var gateIsOpen = false + + let stream: AsyncStream + + init() { + var continuation: AsyncStream.Continuation! + stream = AsyncStream { continuation = $0 } + streamContinuation = continuation + } + + func yieldValue(_ value: Int) { + lock.withLock { streamContinuation }?.yield(value) + } + + /// Suspends until `openGate()` — used to hold a `forEach` body inside the + /// body (i.e. RUNNING), deterministically. + func waitAtGate() async { + await withTaskCancellationHandler { + await withCheckedContinuation { (continuation: CheckedContinuation) in + let resumeImmediately: Bool = lock.withLock { + if gateIsOpen { return true } + gateContinuations.append(continuation) + return false + } + if resumeImmediately { continuation.resume() } + } + } onCancel: { + self.openGate() + } + } + + func openGate() { + let pending: [CheckedContinuation] = lock.withLock { + gateIsOpen = true + defer { gateContinuations = [] } + return gateContinuations + } + for continuation in pending { continuation.resume() } + } +} + +extension DependencyValues { + var foreignClock: ForeignClock { + get { self[ForeignClockKey.self] } + set { self[ForeignClockKey.self] = newValue } + } + + var quiescenceControl: QuiescenceControl { + get { self[QuiescenceControlKey.self] } + set { self[QuiescenceControlKey.self] = newValue } + } + + enum ForeignClockKey: DependencyKey { + static let liveValue = ForeignClock() + } + + enum QuiescenceControlKey: DependencyKey { + static let liveValue = QuiescenceControl() + } +} + +// MARK: - Models + +/// Sleeps on the foreign clock inside `withModelParked` — design §6b hook 3. +@Model private struct ParkedClockSleeper { + var didFinish = false + + func onActivate() { + node.task { + await withModelParked { + await node.foreignClock.sleep() + } + didFinish = true + } + } +} + +/// The same sleep with no wrap — the unmarked (tier 3) case. +@Model private struct UnmarkedClockSleeper { + var didFinish = false + + func onActivate() { + node.task { + await node.foreignClock.sleep() + didFinish = true + } + } +} + +/// The PR #70 blind spot: a task that only ever `Task.yield()`s. The framework +/// did not park it, so it must count as RUNNING throughout. +@Model private struct YieldingLooper { + func onActivate() { + node.task { + let control = node.quiescenceControl + control.bodyEntered.setValue(true) + while !Task.isCancelled && !control.stop.value { + await Task.yield() + } + } + } +} + +/// A compute loop with no suspension point at all — running forever, by design +/// (§6a: user error, to be reported rather than accommodated). +@Model private struct SpinningLooper { + func onActivate() { + node.task { + let control = node.quiescenceControl + control.bodyEntered.setValue(true) + var sink = 0 + while !Task.isCancelled && !control.stop.value { + sink &+= 1 + } + _ = sink + } + } +} + +/// `node.forEach` over an externally-fed `AsyncStream` — design §6b hook 1. +@Model private struct StreamConsumer { + var received = 0 + + func onActivate() { + node.forEach(node.quiescenceControl.stream) { value in + node.quiescenceControl.bodyEntered.setValue(true) + // Hold the body open so "running while the body executes" is + // observable without a race. + await node.quiescenceControl.waitAtGate() + received = value + } + } +} + +// MARK: - Tests + +@Suite(.modelTesting(exhaustivity: .off)) +struct SemanticQuiescenceTests { + /// A task parked on a FOREIGN clock inside `withModelParked` does not block + /// quiescence: the semantic answer is quiescent while it sleeps. + @Test func foreignClockSleepInsideWithModelParkedIsParked() async throws { + let clock = ForeignClock() + let model = ParkedClockSleeper().withAnchor { + $0.foreignClock = clock + } + let context = model.anyContext! + + try await waitUntil(clock.sleeperCount == 1) + // The registered unit exists (the task has not returned) … + #expect(context.activeTasks.flatMap(\.tasks).count == 1) + // … but it is parked, so the model is semantically quiescent. + #expect(context.hasRunningWorkUnit == false) + #expect(context.runningWorkUnits.isEmpty) + #expect(context.semanticQuiescence == true) + + // Unparks and completes when the test acts. + clock.advance() + await expect(model.didFinish) + } + + /// The same sleep WITHOUT the wrap is running work — SwiftModel cannot see + /// the suspension, so the honest answer is "not quiescent" (§6 tier 3). + @Test func foreignClockSleepWithoutWithModelParkedIsRunning() async throws { + let clock = ForeignClock() + let model = UnmarkedClockSleeper().withAnchor { + $0.foreignClock = clock + } + let context = model.anyContext! + + try await waitUntil(clock.sleeperCount == 1) + #expect(context.hasRunningWorkUnit == true) + #expect(context.semanticQuiescence == false) + // The backstop message material: the unit names itself. + let running = context.runningWorkUnits + #expect(running.count == 1) + #expect(running.first?.modelName == "UnmarkedClockSleeper") + + clock.advance() + await expect(model.didFinish) + // Once the body returns the unit is finished and unregistered. + try await waitUntil(model.anyContext?.hasRunningWorkUnit == false) + } + + /// The blind spot that caused PR #70: a task looping on `Task.yield()` is + /// suspended constantly, but SwiftModel did not park it, so it is RUNNING + /// throughout — sampled repeatedly, never once quiescent. + @Test func yieldLoopIsRunningThroughout() async throws { + let control = QuiescenceControl() + let model = YieldingLooper().withAnchor { + $0.quiescenceControl = control + } + let context = model.anyContext! + + try await waitUntil(control.bodyEntered.value) + var sawQuiescent = false + for _ in 0..<50 { + if !context.hasRunningWorkUnit { sawQuiescent = true; break } + try? await Task.sleep(nanoseconds: 2_000_000) + } + #expect(sawQuiescent == false) + + control.stop.setValue(true) + try await waitUntil(model.anyContext?.hasRunningWorkUnit == false) + } + + /// A compute loop with no await never parks — running for as long as it + /// runs. Bounded assertion: sample for a fixed number of iterations, then + /// let the loop finish rather than hanging the suite. + @Test func computeLoopWithNoAwaitIsRunningForever() async throws { + let control = QuiescenceControl() + let model = SpinningLooper().withAnchor { + $0.quiescenceControl = control + } + let context = model.anyContext! + + try await waitUntil(control.bodyEntered.value) + var sawQuiescent = false + for _ in 0..<20 { + if !context.hasRunningWorkUnit { sawQuiescent = true; break } + try? await Task.sleep(nanoseconds: 2_000_000) + } + #expect(sawQuiescent == false) + + control.stop.setValue(true) + try await waitUntil(model.anyContext?.hasRunningWorkUnit == false) + } + + /// Hook 1: `node.forEach` parks around its own `next()`, so a consumer with + /// nothing buffered is parked — and becomes running the moment a value is + /// delivered and its body starts. + @Test func forEachIsParkedWhileWaitingAndRunningWhileDelivering() async throws { + let control = QuiescenceControl() + let model = StreamConsumer().withAnchor { + $0.quiescenceControl = control + } + let context = model.anyContext! + + // Nothing buffered: the consumer is parked inside `next()`. + try await waitUntil(context.activeTasks.flatMap(\.tasks).count == 1) + try await waitUntil(context.hasRunningWorkUnit == false) + #expect(context.semanticQuiescence == true) + + // A yielded value wakes the consumer; its body is held at the gate, so + // the unit is unambiguously running. + control.yieldValue(7) + try await waitUntil(control.bodyEntered.value) + #expect(context.hasRunningWorkUnit == true) + + control.openGate() + await expect(model.received == 7) + // Back to parked once the body returns and the loop awaits `next()`. + try await waitUntil(model.anyContext?.hasRunningWorkUnit == false) + } + + /// Outside any model task there is no work unit; `withModelParked` is a + /// plain passthrough — value and errors both propagate, no trap. + @Test func withModelParkedOutsideAModelTaskIsAPassthrough() async { + #expect(ModelWorkUnit.current == nil) + + let value = await withModelParked { 42 } + #expect(value == 42) + + struct Boom: Error {} + await #expect(throws: Boom.self) { + try await withModelParked { throw Boom() } + } + } + + /// Nesting: the counter (not a `Bool`) keeps the unit parked until the + /// OUTERMOST `withModelParked` exits. + @Test func nestedWithModelParkedStaysParkedUntilOutermostExits() async { + let unit = ModelWorkUnit() + #expect(unit.isRunning == true) + await ModelWorkUnit.$current.withValue(unit) { + await withModelParked { + #expect(unit.isRunning == false) + await withModelParked { + #expect(unit.isRunning == false) + } + #expect(unit.isRunning == false) // outer park still in scope + } + } + #expect(unit.isRunning == true) + } + + // MARK: - The combined rule's fallback classification + // + // quiescent := oldSaysDone AND (newSaysDone OR runningWorkLooksUndeclared) + // + // These assert the third term directly. `looksUndeclared` is the ONLY shape + // that may defer to the scheduler-observing answer, so the two failure modes + // to guard are (a) it misses a genuinely undeclared sleep and a wait that + // returns today starts hanging, and (b) it swallows a mid-flight unit and + // the correctness fix this design exists for is silently undone. + + /// A brand-new unit has never parked and has just been created, so it is + /// MID-FLIGHT against any non-zero quiet window (it may simply not have had + /// its first CPU slot yet — design §7's `hasPendingStartTask` case) and + /// LOOKS UNDECLARED only once it has been silent for the whole window. + @Test func classifyNeverParkedUnitNeedsAFullQuietWindow() { + let unit = ModelWorkUnit() + let now = _drainMonotonicNs() + #expect(unit.classify(nowNs: now, quietNs: 1_000_000_000) == .midFlight) + #expect(unit.classify(nowNs: now, quietNs: 0) == .looksUndeclared) + } + + /// A parked unit is never either: it does not block quiescence and it is + /// not a fallback reason. + @Test func classifyParkedUnitIsParked() { + let unit = ModelWorkUnit() + let ticket = unit.park() + #expect(unit.classify(nowNs: _drainMonotonicNs(), quietNs: 0) == .parked) + ticket.release() + } + + /// THE ANTI-SWALLOW GUARANTEE. Once a unit has parked even once, SwiftModel + /// knows where it suspends — so any later window in which it reads *running* + /// is a resumption, a yield hop or a starved job, i.e. mid-flight. It can + /// never look undeclared again, no matter how long it stays quiet. + @Test func classifyUnitThatHasParkedIsNeverUndeclared() { + let unit = ModelWorkUnit() + let ticket = unit.park() + ticket.release() + #expect(unit.isRunning == true) + // Quiet window of zero — the most permissive possible — still midFlight. + #expect(unit.classify(nowNs: _drainMonotonicNs(), quietNs: 0) == .midFlight) + // And a far-future "now", i.e. arbitrarily long silence. + #expect(unit.classify(nowNs: _drainMonotonicNs() &+ 60_000_000_000, quietNs: 0) == .midFlight) + } + + /// The eager unpark is a transition too, so a unit resumed by + /// `noteResumeInFlight()` (a producer about to yield, a cancellation) is + /// mid-flight even on the never-parked path — the resume itself is recent + /// activity. + @Test func classifyResumeInFlightCountsAsActivity() { + let unit = ModelWorkUnit() + unit.noteResumeInFlight() + #expect(unit.classify(nowNs: _drainMonotonicNs(), quietNs: 1_000_000_000) == .midFlight) + } + + /// End to end: an unmarked foreign sleep is what the fallback is FOR. The + /// subtree verdict says work is running, and says every running unit looks + /// undeclared — which is what lets the combined rule conclude. + @Test func unmarkedForeignSleepIsTheFallbackShape() async throws { + let clock = ForeignClock() + let model = UnmarkedClockSleeper().withAnchor { + $0.foreignClock = clock + } + let context = model.anyContext! + + try await waitUntil(clock.sleeperCount == 1) + // `quietNs: 0` removes the wall clock from the assertion entirely: the + // classification is structural (never parked), not a timing race. + let verdict = context.semanticVerdict(nowNs: _drainMonotonicNs(), quietNs: 0) + #expect(verdict.isQuiescent == false) + #expect(verdict.running == 1) + #expect(verdict.allRunningLookUndeclared == true) + #expect(verdict.undeclared.first?.modelName == "UnmarkedClockSleeper") + + clock.advance() + await expect(model.didFinish) + } + + /// The same sleep with `withModelParked` disappears from the verdict + /// altogether — the adoption path (design §6b hook 3): no running unit, so + /// no fallback is needed and the new answer decides on its own. + @Test func markedForeignSleepNeedsNoFallback() async throws { + let clock = ForeignClock() + let model = ParkedClockSleeper().withAnchor { + $0.foreignClock = clock + } + let context = model.anyContext! + + try await waitUntil(clock.sleeperCount == 1) + let verdict = context.semanticVerdict(nowNs: _drainMonotonicNs(), quietNs: 0) + #expect(verdict.isQuiescent == true) + #expect(verdict.allRunningLookUndeclared == false) + + clock.advance() + await expect(model.didFinish) + } + + /// A `forEach` body runs in the SAME unit as the loop that parked around + /// `next()`, so the loop's park history must not be credited to it: the + /// framework knows where its own `next()` suspends and nothing about where + /// the user's closure does. Inside the body the quiet window is the whole + /// discriminator — which means a body that has just started is mid-flight, + /// and a body that has gone quiet (here, suspended at a gate SwiftModel + /// cannot see) is honestly reported as undeclared. Without the region mark + /// this hangs where it returns today; see `_withUserBodyRegion`. + @Test func forEachBodyIsClassifiedByItsOwnQuietWindowNotTheLoopsPark() async throws { + let control = QuiescenceControl() + let model = StreamConsumer().withAnchor { + $0.quiescenceControl = control + } + let context = model.anyContext! + + try await waitUntil(context.hasRunningWorkUnit == false) + + control.yieldValue(7) + try await waitUntil(control.bodyEntered.value) + + // Just entered the body: recent activity, so still mid-flight. + let fresh = context.semanticVerdict(nowNs: _drainMonotonicNs(), quietNs: 60_000_000_000) + #expect(fresh.running == 1) + #expect(fresh.allRunningLookUndeclared == false) + + // Quiet for a whole window while suspended at a gate we cannot see: + // the honest answer is "undeclared", and the fallback may conclude. + let quiet = context.semanticVerdict(nowNs: _drainMonotonicNs() &+ 60_000_000_000, quietNs: 0) + #expect(quiet.running == 1) + #expect(quiet.allRunningLookUndeclared == true) + + control.openGate() + await expect(model.received == 7) + } + + /// A mixed subtree: one undeclared sleeper beside one mid-flight consumer. + /// `allRunningLookUndeclared` is an ALL-quantifier, so the mid-flight unit + /// vetoes the fallback for the whole subtree — the wait keeps waiting. + /// + /// The mid-flight unit here is a HAND-WRITTEN `for await` loop: SwiftModel + /// owns the stream (so the wait parks) but not the loop, so there is no + /// user-body region and the unit's park history applies to its body too. + /// That is the design's known residual, and it is what makes it a stable + /// fixture for "running, has parked, arbitrarily quiet, still mid-flight". + @Test func oneMidFlightUnitVetoesTheFallbackForTheWholeSubtree() async throws { + let clock = ForeignClock() + let control = QuiescenceControl() + let parent = MixedFallbackParent().withAnchor { + $0.foreignClock = clock + $0.quiescenceControl = control + } + let context = parent.anyContext! + + try await waitUntil(clock.sleeperCount == 1) + try await waitUntil(context.runningWorkUnits.count == 1) // just the sleeper + #expect(context.semanticVerdict(nowNs: _drainMonotonicNs(), quietNs: 0).allRunningLookUndeclared == true) + + // Wake the child consumer and hold its body: now two units are running, + // and one of them has parked before. + parent.consumer.trigger = 1 + try await waitUntil(control.bodyEntered.value) + let verdict = context.semanticVerdict(nowNs: _drainMonotonicNs() &+ 60_000_000_000, quietNs: 0) + #expect(verdict.running == 2) + #expect(verdict.undeclared.count == 1) + #expect(verdict.allRunningLookUndeclared == false) + + control.openGate() + clock.advance() + await expect { + parent.didFinish + parent.consumer.received == 1 + } + } + + /// The region is what separates the two, asserted on a bare unit: the same + /// already-parked unit is mid-flight in a framework stretch and classified + /// by its quiet window inside a user-body stretch. + @Test func theUserBodyRegionIsWhatDropsTheParkHistory() async { + let unit = ModelWorkUnit() + unit.park().release() + #expect(unit.classify(nowNs: _drainMonotonicNs() &+ 60_000_000_000, quietNs: 0) == .midFlight) + + await ModelWorkUnit.$current.withValue(unit) { + await _withUserBodyRegion { + #expect(unit.classify(nowNs: _drainMonotonicNs() &+ 60_000_000_000, quietNs: 0) == .looksUndeclared) + // …but still mid-flight while the region is fresh. + #expect(unit.classify(nowNs: _drainMonotonicNs(), quietNs: 60_000_000_000) == .midFlight) + } + } + // Back in framework code, the park history applies again. + #expect(unit.classify(nowNs: _drainMonotonicNs() &+ 60_000_000_000, quietNs: 0) == .midFlight) + } +} + +/// Parent holding both fallback shapes at once — see +/// `oneMidFlightUnitVetoesTheFallbackForTheWholeSubtree`. +@Model private struct MixedFallbackParent { + var didFinish = false + var consumer = ObservedLoopConsumer() + + func onActivate() { + node.task { + await node.foreignClock.sleep() // undeclared: never parks + didFinish = true + } + } +} + +// MARK: - Tier 1: source-side parking (design §5) +// +// The park mark belongs to the INPUT SOURCE, not to `node.forEach`'s loop, +// because a user can write the loop by hand. Every model below iterates a +// SwiftModel-produced stream with a raw `for await` inside a plain `node.task` +// — no `forEach` anywhere — and must behave exactly like the `forEach` case +// above: parked while waiting for the next element, running while the body +// runs. + +private struct QuiescenceTestError: Error, Equatable {} + +/// Hand-written `for await` over `Observed` — by far the most common shape in +/// the suite, and the one that produced 23 of the 33 decisive disagreements in +/// the first dual-run inventory. +@Model private struct ObservedLoopConsumer { + var trigger = 0 + var received = 0 + + func onActivate() { + node.task { + for await value in Observed(initial: false, removeDuplicates: false, { trigger }) { + node.quiescenceControl.bodyEntered.setValue(true) + await node.quiescenceControl.waitAtGate() + received = value + } + } + } +} + +/// Hand-written `for await` over an event stream. +@Model private struct EventLoopConsumer { + enum Event: Equatable, Sendable { case ping } + + var received = 0 + + func onActivate() { + node.task { + for await _ in node.event(of: Event.ping) { + node.quiescenceControl.bodyEntered.setValue(true) + await node.quiescenceControl.waitAtGate() + received += 1 + } + } + } + + func ping() { node.send(.ping) } +} + +/// Hand-written `for await` over `observeModifications()`. The body writes only +/// out-of-band state — a model write here would re-trigger the stream and spin. +@Model private struct ModificationLoopConsumer { + var trigger = 0 + + func onActivate() { + node.task { + let control = node.quiescenceControl + for await _ in observeModifications() { + control.bodyEntered.setValue(true) + await control.waitAtGate() + control.stop.setValue(true) + } + } + } +} + +/// `node.task(catch:)` whose `catch` handler writes model state — the epilogue +/// window (a task still executing after its work unit was unregistered). +@Model private struct ThrowingCatcher { + var caught = "" + /// What the registry said about this very task *while its `catch` handler + /// was running*. Must be `true`: the unit has to outlive the epilogue. + var registryStillSawRunningWork: Bool? = nil + + func onActivate() { + node.task { + throw QuiescenceTestError() + } catch: { _ in + registryStillSawRunningWork = anyContext?.hasRunningWorkUnit + caught = "caught" + } + } +} + +@Suite(.modelTesting(exhaustivity: .off)) +struct SemanticQuiescenceTier1Tests { + /// A raw `for await` over `Observed` parks while waiting, exactly as + /// `node.forEach` does — no adoption, no `forEach`. + @Test func handWrittenObservedLoopParksWhileWaiting() async throws { + let control = QuiescenceControl() + let model = ObservedLoopConsumer().withAnchor { + $0.quiescenceControl = control + } + let context = model.anyContext! + + // The task is registered and has not returned … + try await waitUntil(context.activeTasks.flatMap(\.tasks).count == 1) + // … but with nothing to deliver it is PARKED, so the model is + // semantically quiescent. + try await waitUntil(context.hasRunningWorkUnit == false) + #expect(context.semanticQuiescence == true) + + // A write produces a value; the body is held at the gate, so the unit is + // unambiguously running while it is being delivered. + model.trigger = 7 + try await waitUntil(control.bodyEntered.value) + #expect(context.hasRunningWorkUnit == true) + + control.openGate() + await expect(model.received == 7) + // Parked again once the body returns to the wait. + try await waitUntil(model.anyContext?.hasRunningWorkUnit == false) + } + + /// The same for a hand-written loop over `node.event(of:)`. + @Test func handWrittenEventLoopParksWhileWaiting() async throws { + let control = QuiescenceControl() + let model = EventLoopConsumer().withAnchor { + $0.quiescenceControl = control + } + let context = model.anyContext! + + try await waitUntil(context.activeTasks.flatMap(\.tasks).count == 1) + try await waitUntil(context.hasRunningWorkUnit == false) + #expect(context.semanticQuiescence == true) + + model.ping() + try await waitUntil(control.bodyEntered.value) + #expect(context.hasRunningWorkUnit == true) + + control.openGate() + await expect(model.received == 1) + try await waitUntil(model.anyContext?.hasRunningWorkUnit == false) + } + + /// And for `observeModifications()`. + @Test func handWrittenObserveModificationsLoopParksWhileWaiting() async throws { + let control = QuiescenceControl() + let model = ModificationLoopConsumer().withAnchor { + $0.quiescenceControl = control + } + let context = model.anyContext! + + try await waitUntil(context.activeTasks.flatMap(\.tasks).count == 1) + try await waitUntil(context.hasRunningWorkUnit == false) + + model.trigger = 1 + try await waitUntil(control.bodyEntered.value) + #expect(context.hasRunningWorkUnit == true) + + control.openGate() + try await waitUntil(control.stop.value) + try await waitUntil(model.anyContext?.hasRunningWorkUnit == false) + } + + /// The epilogue window: a `catch` handler is user code that writes model + /// state, and it runs AFTER the body threw. The work unit must still be + /// registered and running at that point — otherwise the registry reads + /// "quiescent" with a model write still to come, and a wait could pass + /// prematurely. + @Test func catchHandlerRunsBeforeTheWorkUnitIsUnregistered() async throws { + let model = ThrowingCatcher().withAnchor() + + await expect(model.caught == "caught") + #expect(model.registryStillSawRunningWork == true) + // …and the unit is gone once the epilogue is over. + try await waitUntil(model.anyContext?.hasRunningWorkUnit == false) + } +} + + +/// A cancelled task keeps running through its `defer` — and that `defer` writes +/// model state. This is the *other* epilogue window, and by far the common one: +/// `Cancellations` drops a task from its cancellation registry BEFORE the body +/// unwinds (`cancelAll()` empties the dictionary and only then calls +/// `onCancel()`), so an answer read off `registered` goes "quiescent" during +/// every teardown / `task(id:)` replacement / `cancelPrevious` swap. +@Model private struct CancelledDeferWriter { + var marker = "live" + + func onActivate() { + node.task { + let control = node.quiescenceControl + defer { + // Observed from the unwind of a *cancelled* body, which is + // exactly where the registry used to say "nothing running". + control.stop.setValue(anyContext?.hasRunningWorkUnit == true) + marker = "cleared" + } + control.bodyEntered.setValue(true) + await control.waitAtGate() + } + } + +} + +@Suite(.modelTesting(exhaustivity: .off)) +struct SemanticQuiescenceCancellationEpilogueTests { + /// The work unit must still be visible — and running — while a cancelled + /// body runs its `defer`, because that `defer` can write to the model. + @Test func cancelledBodyStillOwnsItsWorkUnitWhileUnwinding() async throws { + let control = QuiescenceControl() + let model = CancelledDeferWriter().withAnchor { + $0.quiescenceControl = control + } + let context = model.anyContext! + + // Wait for the body to be INSIDE its `defer`'s scope: a task cancelled + // before its first slot returns at the pre-body `guard !Task.isCancelled` + // without ever entering the user closure, and then there is no epilogue + // to test. + try await waitUntil(control.bodyEntered.value) + // (The gate is a raw continuation — tier 3, unmarked — so the body reads + // as running while it waits. That is not what this test is about.) + + // Tear the anchor's tasks down, then release the body so it unwinds + // through its `defer`. + context.cancellations.cancelAll() + control.openGate() + + try await waitUntil(model.marker == "cleared") + #expect(control.stop.value == true) + // …and the unit retires once the body has genuinely finished. + try await waitUntil(model.anyContext?.hasRunningWorkUnit == false) + } +} + +// MARK: - Eager unpark: the park→resume-in-flight window +// +// A lazy unpark — the resumed task's own `defer` — leaves the unit reading +// PARKED from the moment its continuation is resumed until the cooperative pool +// gives it a slot. Model-writing work is inbound throughout that window, so a +// wait that consulted the semantic answer there would pass early. It was the +// single largest surviving disagreement in the dual-run inventory, and design +// §5's containment argument does not cover the half of it that matters: an +// `AsyncStream` consumer resumed by CANCELLATION emits no activity signal, and +// the body it resumes goes on to run model-writing `defer`s. +// +// The fix is for the RESUMER to unpark, before it resumes +// (`ModelWorkUnit.noteResumeInFlight()`), with the resumed side's `defer` +// becoming a no-op via the park epoch. The tests below assert both halves +// synchronously — every `#expect` after a `send` / `cancelAll` runs on the +// producing thread, before the resumed task can possibly have had a slot. + +/// Parks in a SwiftModel-owned event stream, and writes model state from the +/// `defer` that cancellation unwinds through: the exact shape the eager unpark +/// exists for. +@Model private struct ParkedEventWaiter { + enum Event: Equatable, Sendable { case matching, other } + + var marker = "live" + var received = 0 + + func onActivate() { + node.task { + let control = node.quiescenceControl + defer { + // A cancelled body writes model state on its way out. If the + // unit read parked while this was still to come, a wait could + // conclude before it ran. + marker = "cleared" + } + control.bodyEntered.setValue(true) + for await _ in node.event(of: Event.matching) { + received += 1 + control.bodyIterations.withValue { $0 += 1 } + if control.holdBody.value { await control.waitAtGate() } + } + } + } + + func sendMatching() { node.send(.matching) } + func sendOther() { node.send(.other) } +} + +@Suite(.modelTesting(exhaustivity: .off)) +struct SemanticQuiescenceEagerUnparkTests { + /// A yield unparks the consumer **at the yield**, not at the resumed task's + /// next slot. `node.send` is synchronous, so the assertion below runs on the + /// sending thread with the consumer's resume still in flight. + @Test func yieldUnparksTheConsumerBeforeItIsResumed() async throws { + let control = QuiescenceControl() + let model = ParkedEventWaiter().withAnchor { + $0.quiescenceControl = control + } + let context = model.anyContext! + + try await waitUntil(control.bodyEntered.value) + // Reach the parked state through the documented wait verb rather than racing + // the consumer's first `next()`; the assertion below is about the STATE. + await settle() + try await waitUntil(context.hasRunningWorkUnit == false) + + // Hold the body so "running" is durable: without the eager unpark the unit + // would still read PARKED here, because the lazy `defer { unpark() }` cannot + // run while the body is stopped at the gate. That is what makes this a real + // test of the property rather than a race against the consumer finishing. + control.holdBody.setValue(true) + model.sendMatching() + try await waitUntil(control.bodyIterations.value == 1) + #expect(context.hasRunningWorkUnit == true) + + control.openGate() + await expect(model.received == 1) + try await waitUntil(model.anyContext?.hasRunningWorkUnit == false) + } + + /// The half with no activity signal behind it: cancelling a consumer parked + /// in `next()` resumes it with `nil` through `AsyncStream`'s `onTermination` + /// — which the stdlib invokes *before* the resume — and the body then runs a + /// model-writing `defer`. The unit must read running from the instant + /// `cancelAll()` returns. + @Test func cancellationUnparksTheConsumerBeforeItIsResumed() async throws { + let control = QuiescenceControl() + let model = ParkedEventWaiter().withAnchor { + $0.quiescenceControl = control + } + let context = model.anyContext! + + try await waitUntil(control.bodyEntered.value) + // Reach the parked state through the documented wait verb rather than racing + // the consumer's first `next()`; the assertion below is about the STATE. + await settle() + try await waitUntil(context.hasRunningWorkUnit == false) + + context.cancellations.cancelAll() + #expect(context.hasRunningWorkUnit == true) + + try await waitUntil(model.marker == "cleared") + try await waitUntil(model.anyContext?.hasRunningWorkUnit == false) + } + + /// The reason the park mark sits on the RAW source rather than on the + /// filtered stream `node.event(of:)` hands back. A non-matching event + /// resumes the consumer just the same — so it unparks eagerly — and the + /// consumer then drops it and goes back to waiting. With the mark wrapped + /// around the filter, that second wait would be inside the *same* open park + /// scope, and the unit would read running until the next matching event. + @Test func anEventThatFailsTheFilterUnparksAndThenReParks() async throws { + let control = QuiescenceControl() + let model = ParkedEventWaiter().withAnchor { + $0.quiescenceControl = control + } + let context = model.anyContext! + + try await waitUntil(control.bodyEntered.value) + // Reach the parked state through the documented wait verb rather than racing + // the consumer's first `next()`; the assertion below is about the STATE. + await settle() + try await waitUntil(context.hasRunningWorkUnit == false) + + model.sendOther() + // The durable property, not the transient one: the unit returns to parked + // WITHOUT the body ever running. Asserting the intermediate `running` state + // here is unobservable — the eager unpark, the filter and the re-park can all + // complete between the send and any check. This still fails on the placement + // the test exists to reject: marking around the filter would leave the unit + // running until the next MATCHING event, so the wait below would never resolve. + await settle() + try await waitUntil(context.hasRunningWorkUnit == false) + #expect(model.received == 0) + + // Still live afterwards. + model.sendMatching() + await expect(model.received == 1) + } + + /// The generic half, for a source SwiftModel does NOT own: `node.forEach` + /// over a foreign `AsyncStream` parks through hook 1, and its *delivery* is + /// not ours to mark — but its **cancellation** is, because SwiftModel is the + /// one cancelling the task. `_withCurrentWorkUnitParked`'s cancellation + /// handler runs synchronously inside `Task.cancel()`, before the parked task + /// is resumed into its `defer`s. + @Test func cancellingAForeignSequenceConsumerUnparksEagerly() async throws { + let control = QuiescenceControl() + let model = StreamConsumer().withAnchor { + $0.quiescenceControl = control + } + let context = model.anyContext! + + try await waitUntil(context.activeTasks.flatMap(\.tasks).count == 1) + // Reach the parked state through the documented wait verb rather than racing + // the consumer's first `next()`; the assertion below is about the STATE. + await settle() + try await waitUntil(context.hasRunningWorkUnit == false) + + context.cancellations.cancelAll() + // The durable property: cancellation must not leave the unit PARKED, so the + // consumer finishes unwinding and the subtree goes quiet. The intermediate + // `running` state is not reliably observable here — the cancelled task can + // complete and unregister between `cancelAll()` returning and any check. The + // eager-unpark-before-resume property is pinned deterministically for the YIELD + // path by `yieldUnparksTheConsumerBeforeItIsResumed`, which holds the body at a + // gate so the lazy unpark cannot run. For the cancellation path there is no + // equivalent hold — the resumed body only unwinds `defer`s — so that half rests + // on the job-stack measurement instead (the AsyncStream cancellation-resume + // family went from 160/256 samples to 0), which is stated in the PR rather than + // asserted here. + try await waitUntil(model.anyContext?.hasRunningWorkUnit == false) + } + + /// A stale ticket releases nothing, so an eagerly-unparked unit is not + /// re-parked by the `defer`s unwinding behind it — however many park scopes + /// were open — and a fresh park still works afterwards. + @Test func aTicketStaleAfterAnEagerUnparkReleasesNothing() { + let unit = ModelWorkUnit() + let outer = unit.park() + let inner = unit.park() + #expect(unit.isRunning == false) + + unit.noteResumeInFlight() + #expect(unit.isRunning == true) + + inner.release() + outer.release() + #expect(unit.isRunning == true) + + let next = unit.park() + #expect(unit.isRunning == false) + next.release() + #expect(unit.isRunning == true) + } + + /// `_ParkSource`, directly: park, eager unpark on yield, no-op release, + /// re-park once the delivery has been taken out. + @Test func parkSourceUnparksOnYieldAndReParksAfterDelivery() { + let source = _ParkSource() + let unit = ModelWorkUnit() + ModelWorkUnit.$current.withValue(unit) { + let ticket = source.beginWait() + #expect(ticket != nil) + #expect(unit.isRunning == false) + + source.willYield() + #expect(unit.isRunning == true) + + ticket?.release() + #expect(unit.isRunning == true) + source.endWait(delivered: true) + + let next = source.beginWait() + #expect(next != nil) + #expect(unit.isRunning == false) + next?.release() + } + } + + /// A value yielded while the consumer was *running* sits in the stream's + /// buffer, so the next wait must not park — and refusing to park is not + /// enough on its own, because `node.forEach` wraps its own park around the + /// whole call. `beginWait` therefore breaks out of the enclosing scope too. + @Test func parkSourceRefusesToParkUnderAnEnclosingScopeWhenAValueIsBuffered() { + let source = _ParkSource() + let unit = ModelWorkUnit() + ModelWorkUnit.$current.withValue(unit) { + source.willYield() + + let outer = unit.park() // `node.forEach`'s hook-1 park + #expect(unit.isRunning == false) + + let inner = source.beginWait() + #expect(inner == nil) + #expect(unit.isRunning == true) + + outer.release() // stale + #expect(unit.isRunning == true) + } + } + + /// Termination is a resume too: `finish()` and a cancelled consumer both + /// wake `next()` with `nil`, and the body unwinds from there. + @Test func parkSourceUnparksOnTermination() { + let source = _ParkSource() + let unit = ModelWorkUnit() + ModelWorkUnit.$current.withValue(unit) { + let ticket = source.beginWait() + #expect(unit.isRunning == false) + + source.willTerminate() + #expect(unit.isRunning == true) + + ticket?.release() + #expect(unit.isRunning == true) + } + } +} diff --git a/Tests/SwiftModelTests/Utilities.swift b/Tests/SwiftModelTests/Utilities.swift index bb987a63..3d55eac6 100644 --- a/Tests/SwiftModelTests/Utilities.swift +++ b/Tests/SwiftModelTests/Utilities.swift @@ -257,3 +257,42 @@ final class CapturingIssueReporter: IssueReporter, @unchecked Sendable { lock.withLock { _messages.append(m) } } } + +// MARK: - ParkedClock +// +// Semantic quiescence, design §6b hook 3 (`Docs/test-quiescence-redesign.md`, +// `Docs/Testing.md` → "Telling SwiftModel where a custom clock suspends"). +// +// SwiftModel defines no clock: `node.continuousClock` is the user's own +// dependency, and the test clocks this suite uses (`TestClock`, +// `ImmediateClock`) come from swift-clocks, which knows nothing about +// SwiftModel. An unwrapped `clock.sleep` is therefore a suspension the +// framework cannot see, so the work unit sitting in it reads as RUNNING and the +// combined quiescence rule has to fall back to observing the scheduler. +// +// `ParkedClock` is the three-line adoption, applied to a clock we do not own by +// wrapping it: it forwards everything and wraps only its own `sleep` in +// `withModelParked`. Every `clock.sleep` in every model driven by this clock is +// then declared, with no change to any model or any call site. Injecting it in +// the suite's clock-driven tests is what makes those tests exercise the PARKED +// path rather than the undeclared-work fallback. + +/// A `Clock` that declares its own suspension to SwiftModel — see above. +@available(iOS 16, macOS 13, tvOS 16, watchOS 9, *) +struct ParkedClock: Clock { + typealias Instant = Base.Instant + typealias Duration = Base.Duration + + let base: Base + + init(_ base: Base) { self.base = base } + + var now: Instant { base.now } + var minimumResolution: Duration { base.minimumResolution } + + func sleep(until deadline: Instant, tolerance: Duration?) async throws { + try await withModelParked { + try await base.sleep(until: deadline, tolerance: tolerance) + } + } +}