From 86e587ecf803b80b592a71fa77e47874e2e10b2c Mon Sep 17 00:00:00 2001 From: zy C Date: Tue, 10 Nov 2026 04:11:46 -0800 Subject: [PATCH 1/3] Isolate the OCaml application core on a serial thread --- .../JournalRuntimeStartupBackend.swift | 4 + .../JournalRuntimeStartupBridge.c | 53 ++- .../JournalRuntimeStartupTests.swift | 49 ++- swift/JournalLUIEvents.swift | 4 +- swift/JournalRuntime.swift | 369 +++++++++++++++--- tool/test_journal_runtime_startup.py | 2 +- 6 files changed, 405 insertions(+), 76 deletions(-) diff --git a/apple-tests/platform-services/JournalRuntimeStartupBackend.swift b/apple-tests/platform-services/JournalRuntimeStartupBackend.swift index 99896010..a79ccd60 100644 --- a/apple-tests/platform-services/JournalRuntimeStartupBackend.swift +++ b/apple-tests/platform-services/JournalRuntimeStartupBackend.swift @@ -6,9 +6,13 @@ public struct LUIAppleExtensionRegistry { public init() {} } @MainActor public final class LUIAppleBackend { public var rootIDs: [Int] = [] public var onEvent: ((LUIEvent) -> Void)? + public var applied: [String] = [] + public var onApply: ((String) -> Void)? public init(appIcons: [String: String], extensionRegistry: LUIAppleExtensionRegistry) throws {} public func apply(json: String) throws { if json == "invalid" { throw NSError(domain: "fixture", code: 1) } rootIDs = [42] + applied.append(json) + onApply?(json) } } diff --git a/apple-tests/platform-services/JournalRuntimeStartupBridge.c b/apple-tests/platform-services/JournalRuntimeStartupBridge.c index 995155a5..98e5e0aa 100644 --- a/apple-tests/platform-services/JournalRuntimeStartupBridge.c +++ b/apple-tests/platform-services/JournalRuntimeStartupBridge.c @@ -1,8 +1,27 @@ #include #include #include -int32_t journal_ocaml_failure_reason(int32_t operation){return 2;} +#include static void (*request)(const char*,int32_t); static int mode,inside,calls,reentrant; +static void (*patches)(const char*), (*wake)(void); +static pthread_t owner; +static int owner_set, pump_count, event_count, pumped, stop_count; +static pthread_mutex_t pump_lock = PTHREAD_MUTEX_INITIALIZER; +static pthread_cond_t pump_ready = PTHREAD_COND_INITIALIZER; +static void check_owner(void) { + if(mode<6)return; + assert(!pthread_main_np()); + if(!owner_set){owner=pthread_self();owner_set=1;} + assert(pthread_equal(owner,pthread_self())); +} +int32_t journal_ocaml_failure_reason(int32_t operation){check_owner();return 2;} +void fixture_pump_while_ui_busy(void){ + pthread_mutex_lock(&pump_lock); + int previous=pumped; + wake(); + while(pumped==previous)pthread_cond_wait(&pump_ready,&pump_lock); + pthread_mutex_unlock(&pump_lock); +} int32_t fixture_inside(void){return inside;} int32_t fixture_calls(void){return calls;} int32_t fixture_reentrant(void){return reentrant;} @@ -10,14 +29,32 @@ void fixture_mode(int32_t m){mode=m;} void fixture_emit(void){char bytes[32]={76,74,80,50,2,0,6,0};if(mode==5){request(bytes,32);memset(bytes,9,32);}else{char small[]={1,2};request(small,2);memset(small,9,2);}} void* worker(void*x){fixture_emit();return 0;} int32_t lui_ocaml_start(void (*patch)(const char*),int32_t p,int32_t h,const char*d,int32_t n){ + check_owner();patches=patch; inside=1; if(mode==3){pthread_t t;pthread_create(&t,0,worker,0);pthread_join(t,0);}else{fixture_emit();if(mode==4)fixture_emit();} - patch(mode==2?"invalid":"valid");inside=0;return mode==1?0:1; + patch(mode==2?"invalid":mode==6?"1":"valid");inside=0;return mode==1?0:1; +} +int32_t lui_ocaml_stop(void){ + check_owner(); + if(mode==7 && ++stop_count<3){patches("retired");return 0;} + return 1; } -int32_t lui_ocaml_stop(void){return 1;} -void journal_ocaml_set_wakeup_callback(void(*f)(void)){} +void journal_ocaml_set_wakeup_callback(void(*f)(void)){wake=f;} void journal_ocaml_set_platform_request_callback(void(*f)(const char*,int32_t)){request=f;} -int32_t journal_ocaml_pump(void){return 1;} -int32_t journal_ocaml_platform_event(const char*d,int32_t n){if(inside)reentrant++;return 1;} -int32_t journal_ocaml_platform_response(const char*d,int32_t n){calls++;if(inside)reentrant++;return 1;} -int32_t journal_ocaml_platform_failure(const char*d,int32_t n){calls++;if(inside)reentrant++;return 1;} +int32_t journal_ocaml_pump(void){ + check_owner(); + if(mode==6){ + if(++pump_count==1){ + patches("2");patches("3"); + }else{patches(pump_count==2?"6":"7");} + pthread_mutex_lock(&pump_lock);pumped=pump_count;pthread_cond_signal(&pump_ready);pthread_mutex_unlock(&pump_lock); + } + return 1; +} +int32_t journal_ocaml_platform_event(const char*d,int32_t n){ + check_owner();if(inside)reentrant++; + if(mode==6){patches(event_count++==0?"4":"5");if(event_count==1)wake();} + return 1; +} +int32_t journal_ocaml_platform_response(const char*d,int32_t n){check_owner();calls++;if(inside)reentrant++;return 1;} +int32_t journal_ocaml_platform_failure(const char*d,int32_t n){check_owner();calls++;if(inside)reentrant++;return 1;} diff --git a/apple-tests/platform-services/JournalRuntimeStartupTests.swift b/apple-tests/platform-services/JournalRuntimeStartupTests.swift index 3f216e5c..0827c5dd 100644 --- a/apple-tests/platform-services/JournalRuntimeStartupTests.swift +++ b/apple-tests/platform-services/JournalRuntimeStartupTests.swift @@ -6,8 +6,9 @@ import LUIAppleBackend @_silgen_name("fixture_reentrant") func reentrant() -> Int32 @_silgen_name("fixture_mode") func mode(_ m: Int32) @_silgen_name("fixture_emit") func emit() +@_silgen_name("fixture_pump_while_ui_busy") func pumpWhileUIBusy() let journalAppIcons: [String: String] = [:] -@MainActor enum JournalLUIEvents { static func dispatch(_ e: LUIEvent) -> Int32 { 1 } } +enum JournalLUIEvents { static func dispatch(_ e: LUIEvent) -> Int32 { 1 } } @MainActor final class JournalApplicationPlatform { var connected = false, hold = true var begins = 0 @@ -25,7 +26,7 @@ let journalAppIcons: [String: String] = [:] @MainActor @main struct Test { static func main() async throws { let scenario = CommandLine.arguments[1] - mode(["rejected": 1, "patch-failure": 2, "worker": 3, "multiple": 4,"same-tag":5][scenario] ?? 0) + mode(["rejected": 1, "patch-failure": 2, "worker": 3, "multiple": 4,"same-tag":5,"core-order":6,"core-lease":7][scenario] ?? 0) let p = JournalApplicationPlatform() let r = try JournalRuntime( platform: p, startupPayload: Data(), extensionRegistry: LUIAppleExtensionRegistry()) @@ -35,7 +36,49 @@ let journalAppIcons: [String: String] = [:] try await r.waitForInitialPlatformRequests() published = true } - if scenario == "rejected" || scenario == "patch-failure" { + if scenario == "core-lease" { + while p.waiters.isEmpty { await Task.yield() } + p.waiters.removeFirst().resume(returning: Data([1, 2])) + try await boot.value + let replacement = try JournalRuntime( + platform: p, startupPayload: Data(), extensionRegistry: LUIAppleExtensionRegistry()) + do { try replacement.start(); preconditionFailure("replaced a live lease") } + catch {} + precondition(p.connected) + precondition(!r.stop(), "fixture stop must fail") + do { try replacement.start(); preconditionFailure("started before failed cleanup settled") } + catch {} + precondition(!p.connected) + p.hold = false + try replacement.start() + try await replacement.waitForInitialPlatformRequests() + precondition(r.stop(), "late stop did not retire the old facade") + for _ in 0..<20 { await Task.yield() } + precondition(p.connected && replacement.failure == nil && r.backend.applied == ["valid"], + "retired facade disconnected its replacement or consumed an old patch") + precondition(replacement.stop()) + } else if scenario == "core-order" { + while p.waiters.isEmpty { await Task.yield() } + p.waiters.removeFirst().resume(returning: Data([1, 2])) + try await boot.value + pumpWhileUIBusy() + var deferred: CheckedContinuation? + r.backend.onApply = { patch in + if patch == "2" { r.sendPlatformEvent(Data([1])) } + if patch == "6" { deferred?.resume(); deferred = nil } + } + r.sendPlatformEvent(Data([1])) + precondition(r.backend.applied == ["1", "2", "3", "4", "5"], + "asynchronous pump, synchronous transaction and nested patch order changed") + await withCheckedContinuation { deferred = $0 } + precondition(r.backend.applied == ["1", "2", "3", "4", "5", "6"], + "the deferred wake was not delivered after the synchronous transaction") + pumpWhileUIBusy() + r.stop() + for _ in 0..<20 { await Task.yield() } + precondition(r.backend.applied == ["1", "2", "3", "4", "5", "6"], + "a retired generation applied its queued patch") + } else if scenario == "rejected" || scenario == "patch-failure" { do { try await boot.value preconditionFailure("rejected startup completed") diff --git a/swift/JournalLUIEvents.swift b/swift/JournalLUIEvents.swift index 9bbaa00b..8c68c21f 100644 --- a/swift/JournalLUIEvents.swift +++ b/swift/JournalLUIEvents.swift @@ -3,7 +3,7 @@ import LUIAppleBackend // Match LUI's typed native ABI. Journal's C adapter also acquires/releases the // OCaml runtime lock and delivers an owned patch after releasing it, before -// returning. A backend patch can therefore synchronously dispatch another event. +// returning. The host serializes these calls on its fixed application-core thread. @_silgen_name("lui_ocaml_appear") private func luiOCamlAppear(_ node: Int64) -> Int32 @_silgen_name("lui_ocaml_press") @@ -66,7 +66,7 @@ private func journalOCamlExtensionEvent( _ node: Int64, _ name: UnsafePointer?, _ payload: UnsafePointer? ) -> Int32 -@MainActor enum JournalLUIEvents { +enum JournalLUIEvents { static func dispatch(_ event: LUIEvent) -> Int32 { switch event { case .appear(let node): return luiOCamlAppear(Int64(node)) diff --git a/swift/JournalRuntime.swift b/swift/JournalRuntime.swift index 8fefad55..2f746a08 100644 --- a/swift/JournalRuntime.swift +++ b/swift/JournalRuntime.swift @@ -46,8 +46,8 @@ import Observation /// C bridge entries exported by app/journal_lui_bridge.c (see also /// platform/native/lui_ocaml_bridge.c in the lui repository). Every entry that /// produces a patch copies it into owned bytes and releases the OCaml runtime -/// before invoking the synchronous patch callback. All entries run on the main -/// actor; applying a patch may synchronously reenter an event entry. +/// before invoking the synchronous patch callback. Entries share one fixed +/// core thread; the callback only copies into the UI mailbox. private typealias PatchCallback = @convention(c) (UnsafePointer?) -> Void private typealias WakeupCallback = @convention(c) () -> Void private typealias PlatformRequestCallback = @@ -90,24 +90,232 @@ private func journalOCamlSetPlatformRequestCallback(_ callback: PlatformRequestC } } -nonisolated(unsafe) private var activeRuntime: JournalRuntime? +private struct JournalCoreResult: Sendable { + let accepted: Int32 + let reason: Int32 +} + +/// The OCaml primordial thread lives for the process, not a SwiftUI session. +/// Only this thread enters OCaml, and it never waits for the main actor. +private final class JournalCoreExecutor: @unchecked Sendable { + static let shared = JournalCoreExecutor() + private let condition = NSCondition() + private var jobs: [@Sendable () -> Void] = [] + private var nextJob = 0 + private var thread: Thread? + private var publishedIdentity: UUID? + private var uiTransactions = 0 + private var deferredWake: UUID? + private var owner: UUID? + private var ingress: JournalPatchIngress? + private var callbacksInstalled = false + private var started = false + private var retiring = false + + private final class Receipt: @unchecked Sendable { + let completed = DispatchSemaphore(value: 0) + var result: JournalCoreResult? + } + + private init() { + let worker = Thread { [self] in + while true { + condition.lock() + while nextJob == jobs.count { condition.wait() } + let job = jobs[nextJob] + nextJob += 1 + if nextJob == jobs.count { + jobs = [] + nextJob = 0 + } + condition.unlock() + autoreleasepool { job() } + } + } + worker.name = "Journal application core" + worker.qualityOfService = .userInitiated + thread = worker + worker.start() + } + + private func enqueue(_ job: @escaping @Sendable () -> Void) { + condition.lock() + jobs.append(job) + condition.signal() + condition.unlock() + } + + private func sync(_ action: @escaping @Sendable () -> JournalCoreResult?) -> JournalCoreResult? { + precondition(Thread.current !== thread) + let receipt = Receipt() + enqueue { + receipt.result = action() + receipt.completed.signal() + } + receipt.completed.wait() + return receipt.result + } + + func claim(_ identity: UUID, ingress: JournalPatchIngress) -> Bool { + sync { [self] in + if owner != nil, retiring { + guard luiOCamlStop() == 1 else { return nil } + owner = nil + self.ingress = nil + } + guard owner == nil else { return nil } + owner = identity + started = false + retiring = false + self.ingress = ingress + condition.lock() + publishedIdentity = identity + condition.unlock() + if !callbacksInstalled { + journalOCamlSetWakeupCallback(wakeup) + journalOCamlSetPlatformRequestCallback(platformRequest) + callbacksInstalled = true + } + return JournalCoreResult(accepted: 1, reason: 0) + } != nil + } -/// The C bridge invokes this on the main actor after releasing the runtime. -/// The copied bytes remain valid until the callback returns, including reentry. + func perform( + _ identity: UUID, operation: Int32, _ action: @escaping @Sendable () -> Int32 + ) -> JournalCoreResult? { + sync { [self] in + guard owner == identity else { return nil } + let accepted = action() + if operation == 1 { started = accepted == 1 } + return JournalCoreResult( + accepted: accepted, + reason: accepted == 1 ? 0 : journalOCamlFailureReason(operation)) + } + } + + func stop(_ identity: UUID) -> JournalCoreResult? { + sync { [self] in + guard owner == identity else { return JournalCoreResult(accepted: 1, reason: 0) } + retiring = true + started = false + let accepted = luiOCamlStop() + let result = JournalCoreResult( + accepted: accepted, + reason: accepted == 1 ? 0 : journalOCamlFailureReason(2)) + if accepted == 1 { + started = false + owner = nil + ingress = nil + condition.lock() + publishedIdentity = nil + condition.unlock() + } + return result + } + } + + func receive(_ json: String) { + precondition(Thread.current === thread) + ingress?.submit(.patch(json)) + } + + func wake() { + condition.lock() + let identity = publishedIdentity + condition.unlock() + guard let identity else { return } + enqueuePump(identity) + } + + @MainActor func beginUITransaction() { + condition.lock() + uiTransactions += 1 + condition.unlock() + } + + @MainActor func endUITransaction() { + condition.lock() + uiTransactions -= 1 + let identity = uiTransactions == 0 ? deferredWake : nil + if uiTransactions == 0 { deferredWake = nil } + condition.unlock() + if let identity { enqueuePump(identity) } + } + + private func enqueuePump(_ identity: UUID) { + enqueue { [self] in + guard owner == identity, started else { return } + condition.lock() + if uiTransactions != 0 { + deferredWake = identity + condition.unlock() + return + } + condition.unlock() + if journalOCamlPump() != 1 { + ingress?.submit(.failure(.pump, journalOCamlFailureReason(3))) + } + } + } +} + +/// Sync UI transactions and asynchronous pumps drain the same ordered mailbox. +/// Pop one batch at a time so a batch's deferred UI events can safely reenter. +private final class JournalPatchIngress: @unchecked Sendable { + enum Delivery: Sendable { + case patch(String) + case failure(JournalRuntime.Failure, Int32) + } + private let lock = NSLock() + private var pending: [Delivery] = [] + private var next = 0 + private var scheduled = false + private let receive: @MainActor @Sendable (Delivery) -> Void + + init(receive: @escaping @MainActor @Sendable (Delivery) -> Void) { + self.receive = receive + } + + func submit(_ delivery: Delivery) { + lock.lock() + pending.append(delivery) + let needsSchedule = !scheduled + scheduled = true + lock.unlock() + if needsSchedule { DispatchQueue.main.async { self.drain() } } + } + + @MainActor func drain() { + JournalCoreExecutor.shared.beginUITransaction() + defer { JournalCoreExecutor.shared.endUITransaction() } + while true { + lock.lock() + guard next < pending.count else { + scheduled = false + lock.unlock() + return + } + let delivery = pending[next] + next += 1 + if next == pending.count { + pending = [] + next = 0 + } + lock.unlock() + receive(delivery) + } + } +} + +/// The bridge releases the runtime before calling; copy before it frees bytes. private let receivePatch: PatchCallback = { source in guard let source else { return } - let json = String(cString: source) - MainActor.assumeIsolated { - activeRuntime?.apply(json: json) - } + JournalCoreExecutor.shared.receive(String(cString: source)) } -/// Fired on whichever OCaml worker thread enqueued cross-thread work; hop to -/// the main actor before draining the pump queue. +/// A worker wakeup never needs the UI actor to advance application state. private let wakeup: WakeupCallback = { - Task { @MainActor in - activeRuntime?.pump() - } + JournalCoreExecutor.shared.wake() } /// One bridge session owns both buffered startup requests and later deliveries. @@ -166,7 +374,7 @@ private let platformRequest: PlatformRequestCallback = { data, length in /// Owns the lui backend, the OCaml runtime, and the platform bridge for one /// journal session. Replaces `BonsaiApplicationView` + `BonsaiApplicationBridge`. @Observable @MainActor final class JournalRuntime { - enum Failure: String, Error, LocalizedError { + enum Failure: String, Error, LocalizedError, Sendable { case start = "runtime.start.failed" case patch = "runtime.patch.failed" case pump = "runtime.pump.failed" @@ -216,8 +424,7 @@ private let platformRequest: PlatformRequestCallback = { data, length in private func record(_ operation:Failure, _ reason:JournalSafeDiagnostic) { failure = operation; failureReason = reason; failureCause = nil } - private func recordBridge(_ failure: Failure, _ operation: Int32) { - let code = journalOCamlFailureReason(operation) + private func recordBridge(_ failure: Failure, _ code: Int32) { let reason: JournalSafeDiagnostic switch code { case 1: reason = .missingHandler @@ -248,6 +455,8 @@ private let platformRequest: PlatformRequestCallback = { data, length in private(set) var failure: Failure? private var started = false @ObservationIgnored private var generation = UUID() + @ObservationIgnored private var coreIdentity: UUID? + @ObservationIgnored private var patchIngress: JournalPatchIngress? @ObservationIgnored private var initialPlatformTasks: [Task] = [] private var stopPending = false @@ -272,44 +481,53 @@ private let platformRequest: PlatformRequestCallback = { data, length in func start() throws { if let reportedFailure { throw reportedFailure } guard !started else { return } + JournalCoreExecutor.shared.beginUITransaction() + defer { JournalCoreExecutor.shared.endUITransaction() } generation = UUID() let identity = generation + let ingress = JournalPatchIngress { [weak self] delivery in + guard let self, self.generation == identity else { return } + switch delivery { + case .patch(let json): self.apply(json: json) + case .failure(let operation, let code): self.recordBridge(operation, code) + } + } + guard JournalCoreExecutor.shared.claim(identity, ingress: ingress) else { + throw FailureDetail(operation: .start, reason: .unknown) + } + coreIdentity = identity + patchIngress = ingress platformRequestIngress.begin(identity) defer { if !started { platformRequestIngress.close(identity) } } - journalOCamlSetWakeupCallback(wakeup) - journalOCamlSetPlatformRequestCallback(platformRequest) - activeRuntime = self #if os(macOS) let operatingSystem: Int32 = 1 #else let operatingSystem: Int32 = 2 #endif - let accepted = JournalStartupDiagnostics.measure("ocaml_start") { - startupPayload.withUnsafeBytes { bytes in - luiOCamlStart( - receivePatch, - operatingSystem, - 2, - bytes.baseAddress?.assumingMemoryBound(to: CChar.self), - Int32(bytes.count)) + let payload = startupPayload + let result = JournalStartupDiagnostics.measure("ocaml_start") { + JournalCoreExecutor.shared.perform(identity, operation: 1) { + payload.withUnsafeBytes { bytes in + luiOCamlStart( + receivePatch, operatingSystem, 2, + bytes.baseAddress?.assumingMemoryBound(to: CChar.self), Int32(bytes.count)) + } } } + ingress.drain() guard generation == identity else { throw CancellationError() } - guard accepted == 1 else { - recordBridge(.start,1) + guard result?.accepted == 1 else { + recordBridge(.start, result?.reason ?? 0) let reported = reportedFailure! - _ = luiOCamlStop() - activeRuntime = nil + _ = stopCore() throw reported } if let reportedFailure { - _ = luiOCamlStop() - activeRuntime = nil + _ = stopCore() throw reportedFailure } guard rootID != nil else { - _ = luiOCamlStop() - activeRuntime = nil + _ = stopCore() record(.start,.missingRoot) throw reportedFailure! } @@ -347,19 +565,47 @@ private let platformRequest: PlatformRequestCallback = { data, length in } @discardableResult func stop() -> Bool { - guard started || stopPending else { return true } - platformRequestIngress.close(generation) + guard coreIdentity != nil else { return true } + JournalCoreExecutor.shared.beginUITransaction() + defer { JournalCoreExecutor.shared.endUITransaction() } + if started { + platformRequestIngress.close(generation) + for task in initialPlatformTasks { task.cancel() } + initialPlatformTasks = [] + platform.disconnect() + } generation = UUID() - for task in initialPlatformTasks { task.cancel() } - initialPlatformTasks = [] - platform.disconnect() started = false - activeRuntime = nil - stopPending = luiOCamlStop() != 1 - if stopPending { recordBridge(.stop,2) } + return stopCore() + } + + private func stopCore() -> Bool { + guard let identity = coreIdentity else { return true } + generation = UUID() + let result = JournalCoreExecutor.shared.stop(identity) + stopPending = result?.accepted != 1 + if stopPending { + recordBridge(.stop, result?.reason ?? 0) + } else { + coreIdentity = nil + patchIngress = nil + } return !stopPending } + private func perform( + _ failure: Failure, operation: Int32, _ action: @escaping @Sendable () -> Int32 + ) { + JournalCoreExecutor.shared.beginUITransaction() + defer { JournalCoreExecutor.shared.endUITransaction() } + let identity = generation + let result = JournalCoreExecutor.shared.perform(identity, operation: operation, action) + patchIngress?.drain() + if generation == identity, started, let result, result.accepted != 1 { + recordBridge(failure, result.reason) + } + } + func apply(json: String) { let identity = generation do { @@ -375,8 +621,7 @@ private let platformRequest: PlatformRequestCallback = { data, length in func pump() { guard started else { return } - let identity = generation, accepted = journalOCamlPump() - if generation == identity, started, accepted != 1 { recordBridge(.pump,3) } + perform(.pump, operation: 3) { journalOCamlPump() } } /// Marshals one LJP2 request onto the platform actor and ships its response @@ -398,35 +643,35 @@ private let platformRequest: PlatformRequestCallback = { data, length in // result and still completes the original continuation without a ledger row. guard let response else { return } if response.prefix(4) == Data("LJF1".utf8) { - let accepted = response.withUnsafeBytes { buffer in - journalOCamlPlatformFailure(buffer.baseAddress?.assumingMemoryBound(to:CChar.self),Int32(buffer.count)) + perform(.platformFailure, operation: 6) { + response.withUnsafeBytes { buffer in + journalOCamlPlatformFailure( + buffer.baseAddress?.assumingMemoryBound(to: CChar.self), Int32(buffer.count)) + } } - if generation == identity, started, accepted != 1 { recordBridge(.platformFailure,6) } return } - let accepted = response.withUnsafeBytes { buffer in - journalOCamlPlatformResponse( - buffer.baseAddress?.assumingMemoryBound(to: CChar.self), - Int32(buffer.count)) + perform(.platformResponse, operation: 5) { + response.withUnsafeBytes { buffer in + journalOCamlPlatformResponse( + buffer.baseAddress?.assumingMemoryBound(to: CChar.self), Int32(buffer.count)) + } } - if generation == identity, started, accepted != 1 { recordBridge(.platformResponse,5) } } /// Pushes one host-originated LJP2 event envelope to OCaml. func sendPlatformEvent(_ bytes: Data) { guard started else { return } - let identity = generation - let accepted = bytes.withUnsafeBytes { buffer in - journalOCamlPlatformEvent( - buffer.baseAddress?.assumingMemoryBound(to: CChar.self), - Int32(buffer.count)) + perform(.platformEvent, operation: 4) { + bytes.withUnsafeBytes { buffer in + journalOCamlPlatformEvent( + buffer.baseAddress?.assumingMemoryBound(to: CChar.self), Int32(buffer.count)) + } } - if generation == identity, started, accepted != 1 { recordBridge(.platformEvent,4) } } private func handle(_ event: LUIEvent) { guard started else { return } - let identity = generation, accepted = JournalLUIEvents.dispatch(event) - if generation == identity, started, accepted != 1 { recordBridge(.event,7) } + perform(.event, operation: 7) { JournalLUIEvents.dispatch(event) } } } diff --git a/tool/test_journal_runtime_startup.py b/tool/test_journal_runtime_startup.py index 9b7a9124..90edab14 100644 --- a/tool/test_journal_runtime_startup.py +++ b/tool/test_journal_runtime_startup.py @@ -24,7 +24,7 @@ def run(c):subprocess.run(c,cwd=W,check=True) run(['xcrun','clang','-c',str(W/'bridge.c'),'-o',str(W/'bridge.o')]) run(['xcrun','swiftc',*flags,'-Xfrontend','-disable-sandbox','-I',str(W),'-L',str(W),'-lLUIAppleBackend','-Xlinker','-rpath','-Xlinker',str(W),str(ROOT/'swift/JournalSafeDiagnostic.swift'),str(ROOT/'swift/JournalPlatformWire.swift'),str(source.resolve()),str(W/'Tests.swift'),str(W/'bridge.o'),'-o',str(W/'tests')]) rows=[] -for scenario in ['normal','worker','multiple','repeated-start','rejected','patch-failure','stopped','cancelled','stale-cancel','nil-response','same-tag']: +for scenario in ['normal','worker','multiple','repeated-start','rejected','patch-failure','stopped','cancelled','stale-cancel','nil-response','same-tag','core-order','core-lease']: r=subprocess.run([str(W/'tests'),scenario],capture_output=True,text=True,timeout=20) rows.append(r.returncode);print(scenario,r.returncode,r.stdout.strip(),r.stderr.splitlines()[:1],flush=True) From 8bdfadb2148fe9dba3d75ace69d854ccbc394963 Mon Sep 17 00:00:00 2001 From: zy C Date: Tue, 10 Nov 2026 04:19:56 -0800 Subject: [PATCH 2/3] Align the failure fixture with serial-core event dispatch --- apple-tests/platform-services/JournalRuntimeFailureTests.swift | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/apple-tests/platform-services/JournalRuntimeFailureTests.swift b/apple-tests/platform-services/JournalRuntimeFailureTests.swift index f8db7d25..02b6fde1 100644 --- a/apple-tests/platform-services/JournalRuntimeFailureTests.swift +++ b/apple-tests/platform-services/JournalRuntimeFailureTests.swift @@ -20,7 +20,7 @@ import LUIAppleBackend } } let journalAppIcons: [String: String] = [:] -@MainActor enum JournalLUIEvents { +enum JournalLUIEvents { static func dispatch(_ event: LUIEvent) -> Int32 { 1 } } From ef7f7916b18725b822bdba7eee9e82941e8c9770 Mon Sep 17 00:00:00 2001 From: zy C Date: Tue, 10 Nov 2026 05:04:43 -0800 Subject: [PATCH 3/3] Scope graph-picker sign-out fixtures to the active account menu --- test/application_view_test.ml | 15 +++++++++++---- 1 file changed, 11 insertions(+), 4 deletions(-) diff --git a/test/application_view_test.ml b/test/application_view_test.ml index 64c0d545..616d4640 100644 --- a/test/application_view_test.ml +++ b/test/application_view_test.ml @@ -7670,6 +7670,13 @@ let test_graph_picker_controls scenario () = if List.assoc_opt key values = Some (`String value) then Some id else found) props None in let node name = match find "accessibility-identifier" name with | Some id -> id | None -> failwith ("Missing " ^ name) in + let picker_account_action title = + let menu = node "journal-startup-menu" in + let rec belongs id = id = menu || match Hashtbl.find_opt page_parents id with + | None -> false | Some parent -> belongs parent in + Hashtbl.to_seq props |> Seq.find_map (fun (id, fields) -> + if belongs id && List.assoc_opt "text" fields = Some (`String title) + then Some id else None) |> Option.get in let picker_list () = let marker=node "graph-picker-list" in let rec belongs id = id=marker || match Hashtbl.find_opt page_parents id with @@ -7826,7 +7833,7 @@ let test_graph_picker_controls scenario () = Alcotest.(check bool) "superseded native identity is never installed" false (List.mem(Some "next-user")!auth_classified)) | `Sign_out | `Sign_out_unknown -> - press(Option.get(find "text" "Sign out")); + press(picker_account_action "Sign out"); wait "sign out consumes actual old-account permission before reconcile" (fun()-> Atomic.get host_applied=1 && not(Atomic.get current).snapshot.startup.authenticated); Alcotest.(check int) "one sign-out prepare" 1(Atomic.get host_prepared); @@ -7853,7 +7860,7 @@ let test_graph_picker_controls scenario () = ((Atomic.get current).snapshot.resident_graphs=[first;uuid 4;uuid 3]) | `Remove_queued_sign_out -> press(node("graph-resident-remove:" ^ Logseq_db_types.Graph_types.Uuid.to_string second)); - press(Option.get(find "text" "Sign out")); + press(picker_account_action "Sign out"); wait "new account intent cancels an unstarted graph close" (fun()-> !native_sign_outs=1); Alcotest.(check int) "only the account/stop prepare reaches the Service" 1(List.length(Atomic.get prepare_attempts)); Alcotest.(check int) "only the newer host permission is applied" 1(Atomic.get host_applied) @@ -7875,7 +7882,7 @@ let test_graph_picker_controls scenario () = wait "provider admits a fresh generation of the same graph" (fun()-> List.exists(fun(scope:Service.asset_scope)->scope.graph_id=second && scope.graph_generation=4) (Atomic.get current).snapshot.resident_scopes)); - if scenario=`Remove_retry_sign_out then press(Option.get(find "text" "Sign out")); + if scenario=`Remove_retry_sign_out then press(picker_account_action "Sign out"); if scenario=`Remove_retry_terminate then ( let packet=Bytes.make 32 '\000' in Bytes.blit_string "LJP2" 0 packet 0 4; Bytes.set_uint16_le packet 4 2;Bytes.set_uint16_le packet 6 12; @@ -7916,7 +7923,7 @@ let test_graph_picker_controls scenario () = if scenario=`Remove_select then press(node("graph-picker:" ^ Logseq_db_types.Graph_types.Uuid.to_string(uuid 3))); if scenario=`Remove_cancel then press(node "graph-picker-close"); - if scenario=`Remove_sign_out then press(Option.get(find "text" "Sign out")); + if scenario=`Remove_sign_out then press(picker_account_action "Sign out"); let target=if scenario=`Remove_select then uuid 3 else if scenario=`Remove_cancel || scenario=`Remove_sign_out then first else second in for _=0 to 32 do consume(hooks.pump());Unix.sleepf 0.001 done;