Reduce bounded observation backlog drain work

Sources/Switch2Kit/Public/Observation.swift

@@ -40,15 +40,74 @@ package protocol EventSink: Sendable {
     func cancel()
 }
 
+// FIFO specialized for the hub's monotonically numbered envelopes. Storage grows
+// lazily up to the observation's existing bound. Popping releases the envelope
+// immediately without shifting the rest of the backlog under a producer lock.
+struct PendingControllerEvents: Sendable {
+    private let capacity: Int
+    private var storage: [EventEnvelope?] = []
+    private var head = 0
+    private(set) var count = 0
+    var isEmpty: Bool { count == 0 }
+    var allocatedCount: Int { storage.count }
+
+    init(capacity: Int) {
+        precondition((1...4096).contains(capacity))
+        self.capacity = capacity
+    }
+    mutating func append(_ event: EventEnvelope) {
+        precondition(count < capacity)
+        if count == storage.count {
+            var grown = [EventEnvelope?](repeating: nil,
+                count: min(capacity, max(1, storage.count * 2)))
+            for index in 0..<count { grown[index] = storage[(head + index) % storage.count] }
+            storage = grown; head = 0
+        }
+        storage[(head + count) % storage.count] = event
+        count += 1
+    }
+    mutating func popFirst() -> EventEnvelope? {
+        guard count != 0 else { return nil }
+        let event = storage[head]
+        storage[head] = nil
+        head = (head + 1) % storage.count
+        count -= 1
+        return event
+    }
+    mutating func takeFirst(_ maximum: Int) -> [EventEnvelope] {
+        var result: [EventEnvelope] = []
+        result.reserveCapacity(min(maximum, count))
+        for _ in 0..<min(maximum, count) {
+            if let event = popFirst() { result.append(event) }
+        }
+        return result
+    }
+    mutating func discard(through sequence: UInt64) {
+        // Only resynchronization may advance past queued input. Since the hub
+        // publishes in order, obsolete envelopes form a prefix, not a full scan.
+        while count != 0, let first = storage[head], first.sequence <= sequence {
+            _ = popFirst()
+        }
+    }
+    mutating func removeAll(keepingCapacity: Bool = false) {
+        if keepingCapacity {
+            while popFirst() != nil {}
+        } else {
+            storage.removeAll(); count = 0
+        }
+        head = 0
+    }
+}
+
 package final class EventMailbox: EventSink {
     private struct State: Sendable {
-        var pending: [EventEnvelope] = []
+        var pending: PendingControllerEvents
         var overflow = false
         var scheduled = false
         var cancelled = false
         var delivered: UInt64 = 0
     }
-    private let state = Mutex(State())
+    private let state: Mutex<State>
     private let capacity: Int
     private let queue: DispatchQueue
     private let interval: TimeInterval
@@ -58,6 +117,7 @@ package final class EventMailbox: EventSink {
                  current: @escaping @Sendable () -> EventEnvelope,
                  handler: @escaping @Sendable (Switch2ControllerEvent) -> Void) {
         self.capacity = min(4096, max(1, capacity)); self.queue = queue
+        self.state = Mutex(State(pending: PendingControllerEvents(capacity: self.capacity)))
         self.interval = interval; self.current = current; self.handler = handler
     }
     package func enqueue(_ event: EventEnvelope) {
@@ -79,7 +139,7 @@ package final class EventMailbox: EventSink {
                 guard !value.cancelled else { return (false, nil) }
                 if value.overflow { value.overflow = false; return (true, nil) }
                 guard !value.pending.isEmpty else { return (false, nil) }
-                return (false, value.pending.removeFirst())
+                return (false, value.pending.popFirst())
             }
             var envelope: EventEnvelope
             if next.0 { envelope = current() }
@@ -93,7 +153,7 @@ package final class EventMailbox: EventSink {
             let deliver = state.withLock { value in
                 guard !value.cancelled, envelope.sequence > value.delivered else { return false }
                 value.delivered = envelope.sequence
-                value.pending.removeAll { $0.sequence <= value.delivered }
+                value.pending.discard(through: value.delivered)
                 return true
             }
             if deliver { handler(envelope.event) }
@@ -209,18 +269,19 @@ package final class ControllerEventHub: Sendable {
 // The only producer is ControllerEventHub; one host thread drains each reader.
 package final class ControllerEventReader: EventSink {
     private struct State: Sendable {
-        var pending: [EventEnvelope] = []
+        var pending: PendingControllerEvents
         var overflow = true
         var cancelled = false
         var delivered: UInt64 = 0
     }
-    private let state = Mutex(State())
+    private let state: Mutex<State>
     private let capacity: Int
     private let current: @Sendable () -> EventEnvelope
     private let remove: @Sendable () -> Void
     package init(capacity: Int, current: @escaping @Sendable () -> EventEnvelope,
                  remove: @escaping @Sendable () -> Void) {
         self.capacity = capacity; self.current = current; self.remove = remove
+        self.state = Mutex(State(pending: PendingControllerEvents(capacity: capacity)))
     }
     package func enqueue(_ event: EventEnvelope) {
         state.withLock { value in
@@ -237,8 +298,7 @@ package final class ControllerEventReader: EventSink {
         let picked: (events: [EventEnvelope], resync: Bool) = state.withLock { value in
             guard !value.cancelled, maximum > 0 else { return ([], false) }
             if value.overflow { value.overflow = false; return ([], true) }
-            let events = Array(value.pending.prefix(maximum))
-            value.pending.removeFirst(events.count)
+            let events = value.pending.takeFirst(maximum)
             return (events, events.contains { $0.lifetime?.isActive == false || $0.snapshotLifetimes.contains { !$0.isActive } })
         }
         let now = current()
@@ -251,7 +311,7 @@ package final class ControllerEventReader: EventSink {
             guard !value.cancelled else { return ([], snapshot, false, false) }
             if resync {
                 value.delivered = max(value.delivered, now.sequence)
-                value.pending.removeAll { $0.sequence <= value.delivered }
+                value.pending.discard(through: value.delivered)
                 // A concurrent producer may have overflowed after current() was read.
                 // Keep that overflow bit, forcing another authoritative snapshot next read.
                 return ([], snapshot, true, value.overflow || !value.pending.isEmpty)

Tests/Switch2KitTests/PendingControllerEventsTests.swift

@@ -0,0 +1,149 @@
+import Foundation
+import Synchronization
+import XCTest
+@testable import Switch2Kit
+
+final class PendingControllerEventsTests: XCTestCase {
+    private func envelope(_ sequence: UInt64, lifetime: SessionLifetime? = nil) -> EventEnvelope {
+        .init(sequence: sequence, event: .status(.init()), lifetime: lifetime)
+    }
+
+    func testLazyGrowthAndRepeatedWrapStayWithinConfiguredCapacity() {
+        for capacity in [1, 2, 3, 7, 256, 4096] {
+            var inbox = PendingControllerEvents(capacity: capacity)
+            var expected: [UInt64] = []
+            XCTAssertEqual(inbox.allocatedCount, 0, "Idle observers must not preallocate their maximum backlog")
+            for sequence in UInt64(1)...10_000 {
+                if expected.count == capacity {
+                    XCTAssertEqual(inbox.popFirst()?.sequence, expected.removeFirst())
+                }
+                inbox.append(envelope(sequence)); expected.append(sequence)
+                if sequence % 3 == 0 {
+                    XCTAssertEqual(inbox.popFirst()?.sequence, expected.removeFirst())
+                }
+                XCTAssertEqual(inbox.count, expected.count)
+                XCTAssertLessThanOrEqual(inbox.allocatedCount, capacity)
+            }
+            XCTAssertEqual(inbox.takeFirst(capacity).map(\.sequence), expected)
+            XCTAssertTrue(inbox.isEmpty)
+            XCTAssertNil(inbox.popFirst())
+        }
+    }
+
+    func testGrowthPreservesWrappedEntriesAndPartialReads() {
+        var inbox = PendingControllerEvents(capacity: 7)
+        for n in UInt64(1)...4 { inbox.append(envelope(n)) }
+        XCTAssertEqual(inbox.takeFirst(3).map(\.sequence), [1, 2, 3])
+        for n in UInt64(5)...10 { inbox.append(envelope(n)) }
+        XCTAssertEqual(inbox.allocatedCount, 7)
+        XCTAssertTrue(inbox.takeFirst(0).isEmpty)
+        XCTAssertEqual(inbox.takeFirst(2).map(\.sequence), [4, 5])
+        XCTAssertEqual(inbox.takeFirst(100).map(\.sequence), [6, 7, 8, 9, 10])
+        XCTAssertTrue(inbox.isEmpty)
+    }
+
+    func testResynchronizationDiscardsOnlyObsoletePrefixAfterWrap() {
+        var inbox = PendingControllerEvents(capacity: 4)
+        for n in UInt64(1)...4 { inbox.append(envelope(n)) }
+        _ = inbox.takeFirst(3)
+        for n in UInt64(5)...7 { inbox.append(envelope(n)) }
+        inbox.discard(through: 3)
+        XCTAssertEqual(inbox.count, 4)
+        inbox.discard(through: 5)
+        XCTAssertEqual(inbox.takeFirst(4).map(\.sequence), [6, 7])
+        inbox.append(envelope(8)); inbox.discard(through: .max)
+        XCTAssertTrue(inbox.isEmpty)
+    }
+
+    func testPoppedAndDiscardedEnvelopesReleaseTheirLifetimeImmediately() {
+        var inbox = PendingControllerEvents(capacity: 8)
+        weak var popped: SessionLifetime?
+        weak var discarded: SessionLifetime?
+        do {
+            let first = SessionLifetime(), second = SessionLifetime()
+            popped = first; discarded = second
+            inbox.append(envelope(1, lifetime: first))
+            inbox.append(envelope(2, lifetime: second))
+        }
+        XCTAssertNotNil(popped); XCTAssertNotNil(discarded)
+        _ = inbox.popFirst()
+        XCTAssertNil(popped, "Consumed storage must not pin a retired attempt")
+        XCTAssertNotNil(discarded)
+        inbox.discard(through: 2)
+        XCTAssertNil(discarded)
+    }
+
+    func testClearReleasesSnapshotTokensAndOptionallyRetainsOnlyEmptySlots() {
+        for keep in [true, false] {
+            var inbox = PendingControllerEvents(capacity: 4)
+            weak var token: SessionLifetime?
+            do {
+                let lifetime = SessionLifetime(); token = lifetime
+                var item = envelope(1)
+                item.snapshotLifetimes = [lifetime]
+                inbox.append(item)
+            }
+            XCTAssertNotNil(token)
+            inbox.removeAll(keepingCapacity: keep)
+            XCTAssertNil(token)
+            XCTAssertTrue(inbox.isEmpty)
+            XCTAssertEqual(inbox.allocatedCount, keep ? 1 : 0)
+            inbox.append(envelope(2))
+            XCTAssertEqual(inbox.popFirst()?.sequence, 2)
+        }
+    }
+
+    func testFullRateObservationDrainsRepeatedMaximumBacklogsInOrder() throws {
+        let hub = ControllerEventHub(), lifetime = SessionLifetime()
+        let id = Switch2ControllerID(rawValue: UUID())
+        let queue = DispatchQueue(label: "test.observation.circular")
+        let reports = Mutex<[UInt64]>([])
+        let done = DispatchSemaphore(value: 0)
+        let observation = try hub.observe(queue: queue, capacity: 4096) { event in
+            if case .input(let controller) = event {
+                reports.withLock { $0.append(controller.state.sequence) }
+                if controller.state.sequence % 4095 == 0 { done.signal() }
+            }
+        }
+        defer { observation.cancel() }
+        for round in 0..<3 {
+            queue.suspend()
+            for n in 1...4095 {
+                let sequence = UInt64(round * 4095 + n)
+                let value = Switch2Controller(id: id, model: .joyCon2Left,
+                    state: .init(buttons: sequence % 2 == 0 ? [] : .slL, sequence: sequence),
+                    connectedAt: Date(timeIntervalSince1970: 0), bodyColor: nil, buttonColor: nil,
+                    serialNumber: nil, sessionGeneration: lifetime.id, lastActivityAt: 0)
+                hub.publish(.init(controllers: [value]), event: .input(value), lifetime: lifetime)
+            }
+            queue.resume()
+            XCTAssertEqual(done.wait(timeout: .now() + 10), .success)
+        }
+        XCTAssertEqual(reports.withLock { $0 }, Array(UInt64(1)...12_285))
+    }
+
+    func testPullReaderPreservesPartialReadFlagsAndOverflowRecovery() throws {
+        let hub = ControllerEventHub(), reader = try hub.makeReader(capacity: 7)
+        defer { reader.cancel() }
+        XCTAssertTrue(reader.read(maximum: 1).resync)
+        for _ in 0..<1000 {
+            for _ in 0..<7 { hub.publish(.init(), event: .status(.init())) }
+            for count in [3, 2, 2] {
+                let batch = reader.read(maximum: count)
+                XCTAssertFalse(batch.resync)
+                XCTAssertEqual(batch.events.count, count)
+                XCTAssertEqual(batch.more, reader.pendingCount > 0)
+            }
+            XCTAssertEqual(reader.pendingCount, 0)
+        }
+        for _ in 0..<100 { hub.publish(.init(), event: .status(.init())) }
+        let recovered = reader.read(maximum: 7)
+        XCTAssertTrue(recovered.resync); XCTAssertTrue(recovered.events.isEmpty)
+        XCTAssertFalse(recovered.more)
+        hub.publish(.init(), event: .status(.init()))
+        XCTAssertEqual(reader.read(maximum: 1).events.count, 1)
+        reader.cancel()
+        hub.publish(.init(), event: .status(.init()))
+        XCTAssertTrue(reader.read(maximum: 7).events.isEmpty)
+    }
+}

tests/observation-buffer/Benchmark.swift

@@ -0,0 +1,30 @@
+import Foundation
+import Synchronization
+
+@main enum Benchmark {
+    static func main() {
+        for capacity in [256, 4096] {
+            var elapsed: [Double] = []
+            for _ in 0..<9 {
+                let queue = DispatchQueue(label: "buffer.benchmark")
+                queue.suspend()
+                let done = DispatchSemaphore(value: 0)
+                let calls = Mutex(0)
+                let box = EventMailbox(capacity: capacity, queue: queue, current: {
+                    EventEnvelope(sequence: UInt64(capacity + 1), event: .snapshot(.init()), lifetime: nil)
+                }, handler: { _ in
+                    let count = calls.withLock { $0 += 1; return $0 }
+                    if count == capacity { done.signal() }
+                })
+                for n in 1...capacity { box.enqueue(.init(sequence: UInt64(n), event: .status(.init()), lifetime: nil)) }
+                let start = DispatchTime.now().uptimeNanoseconds
+                queue.resume()
+                precondition(done.wait(timeout: .now() + 20) == .success)
+                elapsed.append(Double(DispatchTime.now().uptimeNanoseconds - start) / 1e6)
+                box.cancel(); queue.sync {}
+                precondition(calls.withLock { $0 } == capacity)
+            }
+            print("capacity=\(capacity) median_drain_ms=\(elapsed.sorted()[4]) samples_ms=\(elapsed)")
+        }
+    }
+}

tests/observation-buffer/README.md

@@ -0,0 +1,23 @@
+# Observation inbox microbenchmark
+
+From any working directory:
+
+```sh
+bash /path/to/Switch2Kit/tests/observation-buffer/benchmark.sh
+```
+
+This optional benchmark compiles the real `EventMailbox` in release mode. It queues
+exactly 256 or 4096 status envelopes while its delivery queue is suspended, then
+measures resume-to-last-handler time for nine repetitions. The output contains all
+samples and their median. It checks completion and delivered count, not a timing
+threshold. It is not a Bluetooth, SDL, emulator-frame, or physical input latency test.
+
+For a before/after comparison, run the identical `Benchmark.swift` and compile
+command against each revision's `Observation.swift` and its own source dependencies
+in separate checkouts. Do not compare a debug build against a release build. CPU
+load, scheduler behavior and platform affect the measured times.
+
+Correctness and storage-bound tests are in
+`Tests/Switch2KitTests/PendingControllerEventsTests.swift`; the existing observation,
+reader, generation and native consumer regressions remain mandatory. This optional
+microbenchmark is deliberately not discovered as a timing-sensitive CI test.

tests/observation-buffer/benchmark.sh

@@ -0,0 +1,11 @@
+#!/bin/bash
+# Optional queue-only microbenchmark; not a hardware latency or CI timing assertion.
+set -euo pipefail
+cd "$(dirname "$0")/../.."
+source tests/support/kit-sources.sh
+work=$(mktemp -d)
+trap 'rm -rf "$work"' EXIT
+swiftc "${kit_flags[@]}" -swift-version 6 -O -warnings-as-errors "${kit_sources[@]}" \
+    Sources/Switch2Kit/Public/Observation.swift tests/observation-buffer/Benchmark.swift \
+    -o "$work/benchmark"
+"$work/benchmark"