Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -28,6 +28,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
- `ConfigurationError.emptyVisualizerTypes` is absent because empty requests are valid.

### Fixed
- Sync correction pauses across output rebuilds, and telemetry reports pending sync until output is ready.
- Initial player format negotiation reflects the known output route before composing `client/hello`.
- Player streams emit `streamStarted` once on their first successful engine outcome and `streamFormatChanged` on later outcomes.
- Server reactivation updates session state without repeating `serverConnected` with an empty name.
Expand Down
12 changes: 7 additions & 5 deletions Sources/SendspinKit/Audio/AudioEngine.swift
Original file line number Diff line number Diff line change
Expand Up @@ -1472,10 +1472,12 @@ actor AudioEngine {
let estErrUs = Int64(syncSnap.estimatedError.rounded())
let driftPpm = syncSnap.drift * 1_000_000.0

let syncErrorUs = tSnap.syncErrorUs
let dropN = tSnap.correctionSchedule.dropEveryNFrames
let insertN = tSnap.correctionSchedule.insertEveryNFrames
let correcting = tSnap.correctionSchedule.isCorrecting
let syncErrorUs = outputHasStarted ? tSnap.syncErrorUs : nil
let syncText = syncErrorUs.map { "\($0)us" } ?? "pending"
let schedule = syncErrorUs == nil ? CorrectionSchedule() : tSnap.correctionSchedule
let dropN = schedule.dropEveryNFrames
let insertN = schedule.insertEveryNFrames
let correcting = schedule.isCorrecting

let telemetry = "sched=\(framesScheduled) played=\(framesPlayed)"
+ " late=\(framesDroppedLate)"
Expand All @@ -1486,7 +1488,7 @@ actor AudioEngine {
+ " drift=\(String(format: "%.2f", driftPpm))ppm"
+ " samples=\(syncSnap.sampleCount)"
+ " queue=\(currentStats.queueSize)"
+ " sync=\(syncErrorUs)us"
+ " sync=\(syncText)"
+ " correcting=\(correcting)"
+ " drop=\(dropN) insert=\(insertN)"
+ " timingCodec=\(chunkTimingFormat?.codec.rawValue ?? "none")"
Expand Down
40 changes: 33 additions & 7 deletions Sources/SendspinKit/Audio/AudioPlayer.swift
Original file line number Diff line number Diff line change
Expand Up @@ -72,7 +72,7 @@ private let volumeRampStepDuration: Duration = .milliseconds(10)
/// (no copy), but `@unchecked Sendable` means the compiler won't catch
/// accidental copies elsewhere. Only one instance should ever exist,
/// owned by the `OSAllocatedUnfairLock` in `AudioPlayer`.
private struct LockedState: @unchecked Sendable {
struct LockedState: @unchecked Sendable {
// Ring buffer
var pcmRingBuffer: PCMRingBuffer
var frameSize: Int = 0
Expand All @@ -94,6 +94,21 @@ private struct LockedState: @unchecked Sendable {

/// Latest sync error in µs, written by audio callback, read by telemetry
var lastSyncErrorUs: Int64 = 0

var syncOutputIsReady: Bool {
!prewarming && queueStartAbsoluteUs > 0 && spinUpUs >= 0
}

var telemetrySyncErrorUs: Int64? {
syncOutputIsReady ? lastSyncErrorUs : nil
}

mutating func suspendCorrection() {
correctionSchedule = CorrectionSchedule()
dropCounter = 0
insertCounter = 0
}

var pendingReanchorServerTime: Int64 = 0
var reanchorRequested: Bool = false
/// Grace period: suppress sync correction after AudioQueue rebuild.
Expand Down Expand Up @@ -235,7 +250,7 @@ actor AudioPlayer {
/// All state shared between the actor and the audio thread, protected by
/// `OSAllocatedUnfairLock` with priority donation. Access is structurally
/// enforced: every read/write goes through `withLock`.
private nonisolated let lockedState: OSAllocatedUnfairLock<LockedState>
nonisolated let lockedState: OSAllocatedUnfairLock<LockedState>

private var currentVolume: Float = 1.0
private var appliedVolume: Float = 1.0
Expand Down Expand Up @@ -610,6 +625,12 @@ actor AudioPlayer {
let currentPlayerID = playerID
let currentlyPlaying = _isPlaying
Log.audio.debug("stop player=\(currentPlayerID, privacy: .public) isPlaying=\(currentlyPlaying, privacy: .public)")
lockedState.withLock { state in
state.queueStartAbsoluteUs = 0
state.spinUpUs = -1
state.prewarming = true
state.suspendCorrection()
}
cancelVolumeRamp()
outputDeviceLiveContinuation?.resume()
outputDeviceLiveContinuation = nil
Expand Down Expand Up @@ -936,7 +957,7 @@ actor AudioPlayer {
struct TelemetrySnapshot {
let cursorMicroseconds: Int64
let sampleRate: Int
let syncErrorUs: Int64
let syncErrorUs: Int64?
let correctionSchedule: CorrectionSchedule
let underrunCount: Int64
let pcmBytesDropped: Int64
Expand Down Expand Up @@ -975,7 +996,7 @@ actor AudioPlayer {
let startupLateUs: Int64

init(
cursorMicroseconds: Int64, sampleRate: Int, syncErrorUs: Int64, correctionSchedule: CorrectionSchedule,
cursorMicroseconds: Int64, sampleRate: Int, syncErrorUs: Int64?, correctionSchedule: CorrectionSchedule,
underrunCount: Int64, pcmBytesDropped: Int64, startupOffsetUs: Int64?, spinUpUs: Int64,
startupPadFrames: Int64, framesConsumed: Int64, silentBuffers: Int64, enqueueFailures: Int64,
peakOutputLevel: Float, appliedVolume: Float, queueGain: Float, deviceVolume: Float,
Expand Down Expand Up @@ -1039,8 +1060,8 @@ actor AudioPlayer {
return TelemetrySnapshot(
cursorMicroseconds: state.cursorMicroseconds,
sampleRate: state.sampleRate,
syncErrorUs: state.lastSyncErrorUs,
correctionSchedule: state.correctionSchedule,
syncErrorUs: state.telemetrySyncErrorUs,
correctionSchedule: state.syncOutputIsReady ? state.correctionSchedule : CorrectionSchedule(),
underrunCount: state.underrunCount,
pcmBytesDropped: state.pcmBytesDropped,
startupOffsetUs: state.startupOffsetUs,
Expand Down Expand Up @@ -1094,13 +1115,18 @@ actor AudioPlayer {
return snapshot.localTimeToServer(audibleLocalTime)
}

private static func updateCorrectionSchedule(
static func updateCorrectionSchedule(
state: inout LockedState,
capacity: Int,
frameSize: Int,
sampleRate: Int,
inFlightAtCallback: Int64?
) {
// A stopped or prewarming queue has no output cursor to compare with the clock.
guard state.syncOutputIsReady else {
state.suspendCorrection()
return
}
guard state.cursorMicroseconds > 0, let snapshot = state.timeSnapshot else { return }
let nowAbsolute = MonotonicClock.absoluteMicroseconds()

Expand Down
2 changes: 1 addition & 1 deletion Sources/SendspinKit/Audio/NoOpAudioOutput.swift
Original file line number Diff line number Diff line change
Expand Up @@ -12,7 +12,7 @@ actor NoOpAudioOutput: AudioOutput {
AudioPlayer.TelemetrySnapshot(
cursorMicroseconds: 0,
sampleRate: 0,
syncErrorUs: 0,
syncErrorUs: nil,
correctionSchedule: CorrectionSchedule(),
underrunCount: 0,
pcmBytesDropped: 0,
Expand Down
99 changes: 99 additions & 0 deletions Tests/SendspinKitTests/Audio/OutputRebuildCorrectionTests.swift
Original file line number Diff line number Diff line change
@@ -0,0 +1,99 @@
@testable import SendspinKit
import Testing

struct OutputRebuildCorrectionTests {
@Test func stopResetsReadinessBeforeNextCorrectionPass() async throws {
let format = try AudioFormatSpec(codec: .pcm, channels: 2, sampleRate: 48_000, bitDepth: 16)
let player = AudioPlayer()
let capacity = Int(audioQueueBufferByteSize)
let frameSize = format.channels * (format.bitDepth / 8)
let cadence = UInt32(capacity / frameSize)
player.lockedState.withLock { state in
state.prewarming = false
state.queueStartAbsoluteUs = MonotonicClock.absoluteMicroseconds()
state.spinUpUs = 0
state.lastSyncErrorUs = audioQueueTeardownSlowThresholdUs
}
let ready = await player.telemetrySnapshot
#expect(ready.syncErrorUs == audioQueueTeardownSlowThresholdUs)

await player.stop()
player.lockedState.withLock { state in
state.correctionSchedule = CorrectionSchedule(insertEveryNFrames: cadence, dropEveryNFrames: cadence)
state.dropCounter = cadence
state.insertCounter = cadence
AudioPlayer.updateCorrectionSchedule(
state: &state, capacity: capacity, frameSize: frameSize,
sampleRate: format.sampleRate, inFlightAtCallback: nil
)
#expect(state.correctionSchedule == CorrectionSchedule())
#expect(state.dropCounter == 0)
#expect(state.insertCounter == 0)
}
let stopped = await player.telemetrySnapshot
#expect(stopped.syncErrorUs == nil)
#expect(stopped.correctionSchedule == CorrectionSchedule())
}

@Test func staleCursorDoesNotDriveCorrectionUntilOutputIsReady() throws {
let format = try AudioFormatSpec(codec: .pcm, channels: 2, sampleRate: 48_000, bitDepth: 16)
let capacity = Int(audioQueueBufferByteSize)
let frameSize = format.channels * (format.bitDepth / 8)
var state = LockedState(pcmRingBuffer: PCMRingBuffer(capacity: capacity))
defer { state.deallocateResources() }
state.sampleRate = format.sampleRate
state.cursorMicroseconds = Int64(format.sampleRate)
state.timeSnapshot = TimeFilterSnapshot(
offset: 0, drift: 0, lastUpdate: 0, useDrift: false, clientProcessStartAbsolute: 0
)
let sentinelError = Int64(audioQueueTeardownSlowThresholdUs)
state.lastSyncErrorUs = sentinelError
state.correctionGraceFrames = Int64(format.sampleRate)
let cadence = UInt32(capacity / frameSize)

for phase in 0 ... 2 {
state.queueStartAbsoluteUs = phase == 0 ? 0 : MonotonicClock.absoluteMicroseconds()
state.prewarming = phase != 2
state.spinUpUs = -1
state.correctionSchedule = CorrectionSchedule(insertEveryNFrames: cadence, dropEveryNFrames: cadence)
state.dropCounter = cadence
state.insertCounter = cadence
AudioPlayer.updateCorrectionSchedule(
state: &state, capacity: capacity, frameSize: frameSize,
sampleRate: format.sampleRate, inFlightAtCallback: nil
)
#expect(state.correctionSchedule == CorrectionSchedule())
#expect(state.dropCounter == 0)
#expect(state.insertCounter == 0)
#expect(state.lastSyncErrorUs == sentinelError)
#expect(state.correctionGraceFrames == Int64(format.sampleRate))
#expect(state.telemetrySyncErrorUs == nil)
}

state.spinUpUs = 0
AudioPlayer.updateCorrectionSchedule(
state: &state, capacity: capacity, frameSize: frameSize,
sampleRate: format.sampleRate, inFlightAtCallback: nil
)
#expect(state.lastSyncErrorUs != sentinelError)
#expect(state.telemetrySyncErrorUs == state.lastSyncErrorUs)
#expect(state.correctionGraceFrames == Int64(format.sampleRate - capacity / frameSize))
}

@Test func telemetryReportsPendingForStoppedQueueAndNumberAfterReadiness() async {
let player = AudioPlayer()
let stopped = await player.telemetrySnapshot
#expect(stopped.syncErrorUs == nil)
#expect(stopped.correctionSchedule == CorrectionSchedule())

var state = LockedState(pcmRingBuffer: PCMRingBuffer(capacity: Int(audioQueueBufferByteSize)))
defer { state.deallocateResources() }
state.lastSyncErrorUs = audioQueueTeardownSlowThresholdUs
#expect(state.telemetrySyncErrorUs == nil)
state.queueStartAbsoluteUs = MonotonicClock.absoluteMicroseconds()
state.prewarming = false
#expect(state.telemetrySyncErrorUs == nil)
state.spinUpUs = 0
#expect(state.telemetrySyncErrorUs == audioQueueTeardownSlowThresholdUs)
}
}
1 change: 1 addition & 0 deletions docs/audio-timing-model.md
Original file line number Diff line number Diff line change
Expand Up @@ -47,6 +47,7 @@ is deliberately not combined with drift correction.

## What the implementation now does

- The correction loop stays idle and telemetry reports `sync=pending` across an output rebuild until the new queue places its first frame and receives its first device callback.
- The correction target is one shared mapping: `snapshot.localTimeToServer(localNow + physicalPipeline + localOutputDelay)`, with saturating local arithmetic. The callback error, grace-expiry rebaseline, and reanchor target all use that exact helper, including nonzero clock drift.
- `L` includes the device path, read from the HAL at `prepare()` (`OutputDeviceLatency`), while the commanded output delay remains a separate local-domain term.
- A runtime output-delay change shifts every pending scheduler/startup/deferred local play instant by `oldDelay - newDelay` exactly once. Wire timestamps and decoded cadence remain unchanged; chunks already yielded to the output are immutable and are corrected by render pacing rather than rewriting PCM.
Expand Down