Skip to content
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@ import CmuxNextCompat
import CmuxNextDaemon
import CmuxNextDaemonService
import CmuxNextOnboarding
import CmuxNextSidebar
import CmuxNextWakeups
import Foundation

Expand Down Expand Up @@ -119,15 +120,24 @@ extension AppOnboardingServices {
}

/// Waits for a new workspace's slot placement: listed (queued) within 5 s, then
/// through the queue. A workspace whose window went away is never listed.
/// through the queue. A workspace whose window went away is never listed. On the
/// sidebar document the slot is a document move instead (`sidebar.apply`, not the
/// queue): its reply must come before the group write too (#28425).
private static func placementLanded(_ listed: FirstOutcome<Void>, queue: PersonalPlacementQueue) async {
let bound = DemandTimer(owner: "ClassicImport.placementLanded")
bound.schedule(after: .seconds(5)) { listed.resolve(()) }
await listed.value()
bound.cancel()
await queue.drained()
let started = ContinuousClock.now
await SidebarDocStore.current?.settled()
// The wait this adds per imported workspace, recorded only while a DEBUG bench runs (ClassicUpgradeSimulationPerfTests).
BenchSpans.mark(Self.documentSettledSpan, milliseconds: BenchSpans.ms(.now - started))
}

/// The bench span of ``placementLanded(_:queue:)``'s wait for the sidebar document.
static let documentSettledSpan = "classicImport.documentSettled"

/// The daemon a classic workspace opens on: this Mac's, or the one for
/// its SSH host once connected. Classic's identity file and SSH options
/// do not come over; `~/.ssh/config` and the SSH agent apply. A machine
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -30,10 +30,12 @@ enum SidebarDocWiring {
return shown.items[ref] == nil
}

static func start(_ services: AppServices, layout: SidebarLayoutService) {
/// `wrapping`: a test seam over the home daemon's document remote (a live-daemon test holds a reply).
static func start(_ services: AppServices, layout: SidebarLayoutService,
wrapping: @MainActor (any SidebarDocRemote) -> any SidebarDocRemote = { $0 }) {
let local = services.machines.local, machines = services.machines
let serves: @MainActor () -> Bool = { local.supports(DaemonCapabilities.sidebarDoc) }
let store = SidebarDocStore(remote: DaemonSidebarDocRemote(connection: { local.connection }, serves: serves))
let store = SidebarDocStore(remote: wrapping(DaemonSidebarDocRemote(connection: { local.connection }, serves: serves)))
SidebarDocStore.current = store
// Under `top` a new workspace lists at the daemon's slot for it until its item comes (sidebar-ordering-DESIGN.md);
// `afterCurrent` keeps its row at the window's slot (PendingWorkspaceRows), `bottom` lists it last.
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,39 @@
public import Foundation

/// A change sent with no answer (a timeout, a closed connection): the daemon may have applied it. The store sends it
/// again under the same idempotency key (the daemon replays a committed change) instead of dropping it.
public nonisolated struct SidebarDocUnsettled: Error, Hashable, Sendable {
public init() {}
}

/// The daemon side of the document (the App's home daemon connection).
@MainActor
public protocol SidebarDocRemote: AnyObject {
/// The home daemon is connected and serves `sidebar-doc-v1`.
var isAvailable: Bool { get }
/// `sidebar.get`, as JSON.
func get() async throws -> Data
func move(_ items: [String], to target: SidebarDocTarget, key: String) async throws -> SidebarDocReply
func pin(_ items: [String], key: String) async throws -> SidebarDocReply
func unpin(_ items: [String], key: String) async throws -> SidebarDocReply
func adopt(_ refs: [SidebarDocWorkspaceRef], key: String) async throws -> SidebarDocReply
/// `sidebar.apply {ops}`; `ops` is a JSON array.
func apply(ops: Data, key: String) async throws -> SidebarDocReply
}

/// A mutation's reply: the revision after it and its inverse ops (a JSON array).
public nonisolated struct SidebarDocReply: Hashable, Sendable {
public var revision: UInt64?
public var inverse: Data

public init(revision: UInt64?, inverse: Data) {
self.revision = revision
self.inverse = inverse
}

/// The inverse holds at least one op.
public var hasInverse: Bool {
guard let ops = try? JSONSerialization.jsonObject(with: inverse) as? [Any] else { return false }
return !ops.isEmpty
}
}
Original file line number Diff line number Diff line change
@@ -0,0 +1,46 @@
extension SidebarDocStore {
/// Nothing this store asked the daemon is still on its way: every change (move, pin, apply)
/// and every adopt it sent has its reply or was given up, and no document load runs (a change
/// left unanswered is settled by the next load). The daemon committed each answered change
/// before its reply, so a write sent after this lands after them; with the store's events
/// applied through a barrier taken after this (`DaemonStore.appliedThrough`), the document
/// holds their echoes too.
public var isSettled: Bool { inFlight == 0 && !loading }

/// Returns once ``isSettled`` holds: now, or when the last reply or load in flight ends.
/// Event-driven (each reply and load end resumes the waiters), no polling and no quiet window.
/// A classic import waits here so a workspace's group write comes after its slot move: a slot
/// move landing after it would take the workspace out of its group again (#28425).
public func settled() async {
while !isSettled {
await withCheckedContinuation { waiter in
settledWaiters.append(waiter)
onSettledWait?()
}
}
}

/// What keeps ``isSettled`` false, for a test's hang report: the changes and adopts in flight, a
/// running load, the changes a load must settle, and who waits.
var pendingDescription: String {
"\(inFlight) sidebar document changes or adopts unanswered, load running: \(loading), "
+ "settled by the next load: \(unansweredCount), waiting in settled(): \(settledWaiters.count)"
}

/// Someone waits in ``settled()`` (a test seam: a live-daemon test releases the replies it holds).
var waitsForSettled: Bool { !settledWaiters.isEmpty }

/// One change or adopt got its reply (or was given up).
func answered() {
inFlight -= 1
resumeIfSettled()
}

/// Resumes everyone in ``settled()`` once nothing is in flight.
func resumeIfSettled() {
guard isSettled, !settledWaiters.isEmpty else { return }
let waiting = settledWaiters
settledWaiters = []
for waiter in waiting { waiter.resume() }
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -65,7 +65,12 @@ public final class SidebarDocStore {
@ObservationIgnored public var onUndoRefused: @MainActor (AnyObject) -> Void = { _ in }
/// Set by the App around running an undo step's inverse; each change sent meanwhile carries it to `onUndoable`.
@MainActor public static var undoOrigin: AnyObject?
@ObservationIgnored private var loading = false
@ObservationIgnored private(set) var loading = false
/// Changes and adopts sent and not answered yet, and who waits for none (``settled()``).
@ObservationIgnored var inFlight = 0
@ObservationIgnored var settledWaiters: [CheckedContinuation<Void, Never>] = []
/// Runs when someone starts to wait in ``settled()`` (a test seam; the app sets none).
@ObservationIgnored var onSettledWait: (@MainActor () -> Void)?
/// A refetch asked for while one was in flight runs after it.
@ObservationIgnored private var refetchAgain = false
/// Batches that arrived while the first load was in flight; applied after it (the mirror drops the stale ones).
Expand All @@ -84,6 +89,7 @@ public final class SidebarDocStore {
/// echo may have gone to a closed connection). They stay pending until the next document load, which says where
/// the rows really are: the load drops them (cx-ml98q).
@ObservationIgnored private var unanswered: Set<String> = []
var unansweredCount: Int { unanswered.count }
/// Loads failed in a row while unanswered changes wait (a `sidebar.get` that timed out under the same load).
@ObservationIgnored private var failedLoads = 0
/// Refs already sent to `sidebar.adopt` this session.
Expand Down Expand Up @@ -160,6 +166,7 @@ public final class SidebarDocStore {
defer {
self?.loading = false
if self?.refetchAgain == true { self?.refetchAgain = false; self?.refresh() }
self?.resumeIfSettled()
}
guard let data = try? await remote.get(), let records = try? SidebarDocRecords(document: data), let self,
let dropped = mirror.load(data) else {
Expand Down Expand Up @@ -234,9 +241,11 @@ public final class SidebarDocStore {
let fresh = refs.filter { !adopting.contains($0) }
guard isActive, !isProvisional, !fresh.isEmpty, let remote else { return } // adopts wait for the daemon's document
adopting.formUnion(fresh)
inFlight += 1
// task-owner: one sidebar.adopt; its batch places the rows. A failed adopt is
// tried again on the next structural change.
Task { [weak self] in
defer { self?.answered() }
if (try? await remote.adopt(fresh, key: Self.newKey())) == nil { self?.adopting.subtract(fresh) }
}
}
Expand Down Expand Up @@ -381,8 +390,10 @@ public final class SidebarDocStore {
let key = intent.idempotencyKey
// An undo's or redo's inverse fills the step the undo put; a user's change puts its own step now (cx-kpicb).
let origin = Self.undoOrigin ?? (undoable ? onUndoPending(title) : nil)
inFlight += 1
// task-owner: one sidebar mutation; settles its pending intent
Task { [weak self] in
defer { self?.answered() }
// wakeup-allow: bounded retry only after an awaited remote call; this is not polling.
// Each ask either returns or, unanswered before the last one, asks again: at most `unsettledAsks` sends.
for asked in 1...Self.unsettledAsks {
Expand Down Expand Up @@ -456,45 +467,7 @@ public final class SidebarDocStore {
static func newKey() -> String { "sidebar-" + UUID().uuidString.lowercased() }
}

/// A change sent with no answer (a timeout, a closed connection): the daemon may have applied it. The store sends it
/// again under the same idempotency key (the daemon replays a committed change) instead of dropping it.
public nonisolated struct SidebarDocUnsettled: Error, Hashable, Sendable {
public init() {}
}

/// A change refused before it was sent, with the message the refusal HUD shows.
public nonisolated struct SidebarDocSendRefusal: Error, Hashable, Sendable {
public var message: String
}

/// The daemon side of the document (the App's home daemon connection).
@MainActor
public protocol SidebarDocRemote: AnyObject {
/// The home daemon is connected and serves `sidebar-doc-v1`.
var isAvailable: Bool { get }
/// `sidebar.get`, as JSON.
func get() async throws -> Data
func move(_ items: [String], to target: SidebarDocTarget, key: String) async throws -> SidebarDocReply
func pin(_ items: [String], key: String) async throws -> SidebarDocReply
func unpin(_ items: [String], key: String) async throws -> SidebarDocReply
func adopt(_ refs: [SidebarDocWorkspaceRef], key: String) async throws -> SidebarDocReply
/// `sidebar.apply {ops}`; `ops` is a JSON array.
func apply(ops: Data, key: String) async throws -> SidebarDocReply
}

/// A mutation's reply: the revision after it and its inverse ops (a JSON array).
public nonisolated struct SidebarDocReply: Hashable, Sendable {
public var revision: UInt64?
public var inverse: Data

public init(revision: UInt64?, inverse: Data) {
self.revision = revision
self.inverse = inverse
}

/// The inverse holds at least one op.
public var hasInverse: Bool {
guard let ops = try? JSONSerialization.jsonObject(with: inverse) as? [Any] else { return false }
return !ops.isEmpty
}
}
Original file line number Diff line number Diff line change
@@ -1,24 +1,97 @@
@testable import CmuxNextApp
import CmuxNextDaemon
import CmuxNextDaemonService
@testable import CmuxNextSidebar
import Foundation
import Testing

extension AppServices {
/// Waits until the personal placement queue is through and the store has
/// applied every daemon event sent before it was: the sidebar order the
/// queued placements wrote is then the order the store reads. A live-daemon
/// test reads order only after this; the queue can trail a run on a loaded
/// runner (#26153), well past any quiet window.
@MainActor func placementsSettled(timeout: Duration = .seconds(60)) async {
/// Waits until every sidebar placement asked for so far has landed, and records an issue naming
/// what is still in flight if that takes longer than `timeout` (a hang guard, not a quiet window):
/// - the personal placement queue is through (#26153);
/// - the sidebar document has the reply to every move and adopt it sent, and no load runs
/// (`SidebarDocStore.isSettled`; a classic import's slot moves go there, not the queue);
/// - the store applied every daemon event sent before both were (their echoes, the personal
/// rows the document's write-back changed);
/// - no live workspace still waits for its new-workspace placement;
/// - the document projection the sidebar draws is the current one (`shownVersion == version`).
/// A live-daemon test reads order and groups only after this. Each part is a completion signal
/// the app itself sets; the loop repeats because applying an event can start another placement.
@MainActor func placementsSettled(timeout: Duration = .seconds(60), sourceLocation: SourceLocation = #_sourceLocation) async {
let clock = ContinuousClock()
let deadline = clock.now + timeout
while clock.now < deadline {
while !personalPlacements.isIdle, clock.now < deadline { try? await Task.sleep(for: .milliseconds(20)) }
if let barrier = await daemon.connection?.eventSequence() {
while daemon.store.appliedSequence < barrier, clock.now < deadline { try? await Task.sleep(for: .milliseconds(20)) }
while let unsettled = await placementsInFlight(until: deadline) {
guard clock.now < deadline else {
Issue.record("sidebar placements still in flight after \(timeout): \(unsettled)", sourceLocation: sourceLocation)
return
}
// A store event can enqueue another placement; settled only once both are through.
if personalPlacements.isIdle { return }
}
}

/// One pass: waits for each signal in turn (polling only its condition, up to `deadline`), then
/// returns nil when all of them hold at once, else the first one that no longer holds.
@MainActor private func placementsInFlight(until deadline: ContinuousClock.Instant) async -> String? {
let clock = ContinuousClock()
let document = SidebarDocStore.current
func wait(_ done: () -> Bool) async {
while !done(), clock.now < deadline { try? await Task.sleep(for: .milliseconds(20)) }
}
await wait { personalPlacements.isIdle }
// The document's own completion signal (each reply resumes it), not a poll; the deadline is a hang guard only.
if let document {
let settled = await Self.settled(document, before: deadline)
if !settled { return document.pendingDescription }
}
// Taken after the replies: every echo and write-back event of the changes above is before it.
if let barrier = await daemon.connection?.eventSequence() {
await wait { daemon.store.appliedSequence >= barrier }
if daemon.store.appliedSequence < barrier { return "daemon events up to \(barrier) (applied \(daemon.store.appliedSequence))" }
}
await wait { awaitingPlacement().isEmpty }
await wait { document.map { !$0.isActive || $0.shownVersion == $0.version } ?? true }
if !personalPlacements.isIdle { return "a personal placement queued or running" }
if let document, !document.isSettled { return document.pendingDescription }
// Not the barrier again: a later event (a shell's output) is no placement's, and new placement
// work since would show in the queue or the document above.
let awaiting = awaitingPlacement()
if !awaiting.isEmpty { return "the new-workspace placement of \(awaiting.sorted())" }
if let document, document.isActive, document.shownVersion != document.version {
return "the sidebar projection (shown \(document.shownVersion), document \(document.version))"
}
return nil
}

/// Whether `document` settled (``SidebarDocStore/settled()``, event-driven) before `deadline`; false
/// leaves the wait behind so the caller can report what is pending instead of hanging past the test's limit.
@MainActor private static func settled(_ document: SidebarDocStore, before deadline: ContinuousClock.Instant) async -> Bool {
if document.isSettled { return true }
@MainActor final class First {
var continuation: CheckedContinuation<Bool, Never>?
func resolve(_ value: Bool) {
continuation?.resume(returning: value)
continuation = nil
}
}
let first = First()
return await withCheckedContinuation { continuation in
first.continuation = continuation
// task-owner: the hang guard of one test wait; cancelled once the store settles
let guardTask = Task { @MainActor in
try? await Task.sleep(until: deadline)
if !Task.isCancelled { first.resolve(false) }
}
// task-owner: one test wait; ends with the store's settle, or stays behind it after the hang guard
Task { @MainActor in
await document.settled()
guardTask.cancel()
first.resolve(true)
}
}
}

/// Live workspaces whose new-workspace placement has not run yet (one for a workspace that never
/// became live, a create that failed, is not waited for).
@MainActor private func awaitingPlacement() -> [String] {
windows.pendingPlacementOrder.filter { machines.workspace(id: $0) != nil && windows.pendingPlacements[$0] != nil }
}
}
Loading
Loading