diff --git a/Packages/ValarAudio/Sources/ValarAudio/AVFoundationAudioDecoder.swift b/Packages/ValarAudio/Sources/ValarAudio/AVFoundationAudioDecoder.swift index 8db0224..ce1f2b1 100644 --- a/Packages/ValarAudio/Sources/ValarAudio/AVFoundationAudioDecoder.swift +++ b/Packages/ValarAudio/Sources/ValarAudio/AVFoundationAudioDecoder.swift @@ -1,7 +1,9 @@ import AVFoundation public struct AVFoundationAudioDecoder: AudioDecoder { - private static let maximumTemporaryFileSize = 500_000_000 + private static let maximumTemporaryFileSize = 128 * 1_024 * 1_024 + static let maximumDecodedPCMBytes = 256 * 1_024 * 1_024 + private static let framesPerDecodeRead: AVAudioFrameCount = 16_384 public init() {} @@ -24,19 +26,115 @@ public struct AVFoundationAudioDecoder: AudioDecoder { try data.write(to: tempURL) defer { try? FileManager.default.removeItem(at: tempURL) } - let file = try AVAudioFile(forReading: tempURL) + return try decodeAudioFile(at: tempURL, hint: hint) + } + + public func decode(fileAt url: URL, hint: String?) async throws -> AudioPCMBuffer { + let resourceValues = try url.resourceValues( + forKeys: [.fileSizeKey, .isDirectoryKey, .isRegularFileKey] + ) + guard resourceValues.isDirectory != true, + resourceValues.isRegularFile == true else { + throw AudioPipelineError.decodingFailed("Audio source must be a regular file") + } + if let fileSize = resourceValues.fileSize, + fileSize > Self.maximumTemporaryFileSize { + throw AudioDecoderError.fileTooLarge( + size: fileSize, + maximumSize: Self.maximumTemporaryFileSize + ) + } + if hint?.lowercased().hasPrefix("pcm_f32le") == true { + if let fileSize = resourceValues.fileSize, + fileSize > Self.maximumDecodedPCMBytes { + throw AudioDecoderError.decodedAudioTooLarge( + size: fileSize, + maximumSize: Self.maximumDecodedPCMBytes + ) + } + return try await decode( + Data(contentsOf: url, options: .mappedIfSafe), + hint: hint + ) + } + + return try decodeAudioFile(at: url, hint: hint) + } + + private func decodeAudioFile(at url: URL, hint: String?) throws -> AudioPCMBuffer { + let file = try AVAudioFile(forReading: url) let format = file.processingFormat + let channelCount = Int(format.channelCount) + guard format.sampleRate.isFinite, + format.sampleRate >= 1_000, + format.sampleRate <= 768_000, + (1...64).contains(channelCount) else { + throw AudioPipelineError.decodingFailed("Decoded audio format is outside supported bounds") + } guard file.length >= 0, file.length <= Int64(AVAudioFrameCount.max) else { throw AudioPipelineError.decodingFailed("Frame count \(file.length) exceeds AVAudioPCMBuffer capacity") } - let frameCount = AVAudioFrameCount(file.length) - guard let pcmBuffer = AVAudioPCMBuffer(pcmFormat: format, frameCapacity: frameCount) else { + _ = try Self.validatedDecodedPCMByteCount( + frameCount: file.length, + channelCount: channelCount + ) + let frameCount = Int(file.length) + var channels = (0.. 0 else { + return AudioPCMBuffer( + channels: channels, + format: AudioFormatDescriptor( + sampleRate: format.sampleRate, + channelCount: channelCount, + sampleFormat: .float32, + interleaved: false, + container: hint ?? "wav" + ) + ) + } + let decodeCapacity = min( + Self.framesPerDecodeRead, + AVAudioFrameCount(frameCount) + ) + guard let pcmBuffer = AVAudioPCMBuffer( + pcmFormat: format, + frameCapacity: decodeCapacity + ) else { throw AudioPipelineError.decodingFailed("Could not allocate PCM buffer") } - try file.read(into: pcmBuffer) - - let channelCount = Int(format.channelCount) - let channels = PCMCoding.channels(from: pcmBuffer) + var destinationOffset = 0 + while destinationOffset < frameCount { + let requestedFrames = AVAudioFrameCount( + min( + Int(Self.framesPerDecodeRead), + frameCount - destinationOffset + ) + ) + try file.read(into: pcmBuffer, frameCount: requestedFrames) + let framesRead = Int(pcmBuffer.frameLength) + guard framesRead > 0 else { break } + for channelIndex in 0...size - - // Bulk reinterpret: on LE platforms this is a single memcpy with no conversion. - let flat: [Float] = data.withUnsafeBytes { rawBuffer in - Array(rawBuffer.bindMemory(to: Float.self)) + guard totalSamples.isMultiple(of: channelCount) else { + throw AudioPipelineError.decodingFailed( + "pcm_f32le sample count \(totalSamples) is not divisible by channel count \(channelCount)" + ) } + _ = try Self.validatedDecodedPCMByteCount( + frameCount: Int64(totalSamples / channelCount), + channelCount: channelCount + ) - let channels: [[Float]] - if channelCount == 1 { - channels = [flat] - } else { - channels = (0.. (sampleRate: Double, channelCount: Int) { + private func parsePCMHint( + _ hint: String + ) throws -> (sampleRate: Double, channelCount: Int) { let parts = hint.lowercased().split(separator: ":", maxSplits: 2) - let sampleRate = parts.count > 1 ? Double(parts[1]) ?? 24_000 : 24_000 - let channelCount = parts.count > 2 ? max(1, Int(parts[2]) ?? 1) : 1 + let sampleRate: Double + if parts.count > 1 { + guard let parsed = Double(parts[1]), + parsed.isFinite, + parsed >= 1_000, + parsed <= 768_000 else { + throw AudioPipelineError.decodingFailed( + "pcm_f32le sample rate must be finite and between 1000 and 768000 Hz" + ) + } + sampleRate = parsed + } else { + sampleRate = 24_000 + } + let channelCount: Int + if parts.count > 2 { + guard let parsed = Int(parts[2]), parsed >= 1, parsed <= 64 else { + throw AudioPipelineError.decodingFailed( + "pcm_f32le channel count must be between 1 and 64" + ) + } + channelCount = parsed + } else { + channelCount = 1 + } return (sampleRate, channelCount) } + + static func validatedDecodedPCMByteCount( + frameCount: Int64, + channelCount: Int + ) throws -> Int { + guard frameCount >= 0, + UInt64(frameCount) <= UInt64(Int.max), + (1...64).contains(channelCount) else { + throw AudioDecoderError.decodedAudioTooLarge( + size: nil, + maximumSize: maximumDecodedPCMBytes + ) + } + let (sampleCount, sampleOverflowed) = Int(frameCount) + .multipliedReportingOverflow(by: channelCount) + let (byteCount, byteOverflowed) = sampleCount + .multipliedReportingOverflow(by: MemoryLayout.size) + guard !sampleOverflowed, + !byteOverflowed, + byteCount <= maximumDecodedPCMBytes else { + throw AudioDecoderError.decodedAudioTooLarge( + size: byteOverflowed || sampleOverflowed ? nil : byteCount, + maximumSize: maximumDecodedPCMBytes + ) + } + return byteCount + } } private enum AudioDecoderError: Error, LocalizedError, Sendable { case fileTooLarge(size: Int, maximumSize: Int) + case decodedAudioTooLarge(size: Int?, maximumSize: Int) var errorDescription: String? { switch self { case let .fileTooLarge(size, maximumSize): return "Audio payload size \(size) bytes exceeds temporary decode limit of \(maximumSize) bytes" + case let .decodedAudioTooLarge(size, maximumSize): + let observed = size.map { "\($0) bytes" } ?? "an unrepresentable size" + return "Decoded PCM would require \(observed), exceeding the \(maximumSize)-byte allocation limit" } } } diff --git a/Packages/ValarAudio/Sources/ValarAudio/AVFoundationAudioExporter.swift b/Packages/ValarAudio/Sources/ValarAudio/AVFoundationAudioExporter.swift index 1ab1f0d..9fc225a 100644 --- a/Packages/ValarAudio/Sources/ValarAudio/AVFoundationAudioExporter.swift +++ b/Packages/ValarAudio/Sources/ValarAudio/AVFoundationAudioExporter.swift @@ -1,7 +1,17 @@ import AVFoundation import CoreMedia +/// Incremental M4A sink used by project exporters that must not retain an +/// entire multi-chapter render in memory. +public protocol M4AChapterStreamWriting: Sendable { + func append(_ buffer: AudioPCMBuffer, chapterTitle: String) async throws + func finalize() async throws -> AudioExportedFile + func cancel() async +} + public struct AVFoundationAudioExporter: AudioExporter { + private static let maximumInMemoryExportBytes = 256 * 1_024 * 1_024 + public init() {} public func export(_ buffer: AudioPCMBuffer, as format: AudioFormatDescriptor) async throws -> AudioExportedAsset { @@ -10,6 +20,13 @@ public struct AVFoundationAudioExporter: AudioExporter { defer { try? FileManager.default.removeItem(at: tempURL) } _ = try await export(buffer, as: format, to: tempURL, chapterMarkers: []) + let values = try tempURL.resourceValues(forKeys: [.fileSizeKey]) + if let fileSize = values.fileSize, + fileSize > Self.maximumInMemoryExportBytes { + throw AudioPipelineError.exportFailed( + "Encoded audio exceeds the \(Self.maximumInMemoryExportBytes)-byte in-memory export limit; use file export instead." + ) + } let data = try Data(contentsOf: tempURL) return AudioExportedAsset( @@ -26,34 +43,63 @@ public struct AVFoundationAudioExporter: AudioExporter { ) async throws -> AudioExportedFile { let container = Self.normalizedContainer(format.container) let normalizedFormat = Self.normalizedFormat(format, container: container) + try Self.validate(buffer: buffer, format: normalizedFormat) let normalizedMarkers = Self.normalizedChapterMarkers(chapterMarkers) + let fileManager = FileManager.default + let destinationURL = destinationURL.standardizedFileURL + let destinationDirectory = destinationURL.deletingLastPathComponent() - try FileManager.default.createDirectory( - at: destinationURL.deletingLastPathComponent(), + try fileManager.createDirectory( + at: destinationDirectory, withIntermediateDirectories: true ) - if FileManager.default.fileExists(atPath: destinationURL.path) { - do { - try FileManager.default.removeItem(at: destinationURL) - } catch let error as CocoaError where error.code == .fileNoSuchFile { - // Another process or AVFoundation may have already removed the stale target. + if fileManager.fileExists(atPath: destinationURL.path) { + let values = try destinationURL.resourceValues( + forKeys: [.isDirectoryKey, .isRegularFileKey, .isSymbolicLinkKey] + ) + guard values.isDirectory != true, + values.isRegularFile == true, + values.isSymbolicLink != true else { + throw AudioPipelineError.exportFailed( + "The audio destination must be a regular, non-symbolic-link file." + ) } } + let stagingURL = destinationDirectory + .appendingPathComponent( + ".\(destinationURL.deletingPathExtension().lastPathComponent)-valar-\(UUID().uuidString)", + isDirectory: false + ) + .appendingPathExtension(container) + defer { try? fileManager.removeItem(at: stagingURL) } + switch container { case "wav": - try writeWaveFile(buffer, as: normalizedFormat, to: destinationURL) + try await writeWaveFile(buffer, as: normalizedFormat, to: stagingURL) case "m4a": try await writeM4AFile( buffer, as: normalizedFormat, - to: destinationURL, + to: stagingURL, chapterMarkers: normalizedMarkers ) default: throw AudioPipelineError.exportFailed("Unsupported container '\(container)'") } + try Task.checkCancellation() + if fileManager.fileExists(atPath: destinationURL.path) { + _ = try fileManager.replaceItemAt( + destinationURL, + withItemAt: stagingURL, + backupItemName: nil, + options: [] + ) + } else { + try fileManager.moveItem(at: stagingURL, to: destinationURL) + } + return AudioExportedFile( url: destinationURL, format: normalizedFormat, @@ -65,7 +111,7 @@ public struct AVFoundationAudioExporter: AudioExporter { _ buffer: AudioPCMBuffer, as format: AudioFormatDescriptor, to destinationURL: URL - ) throws { + ) async throws { let settings: [String: Any] = [ AVFormatIDKey: kAudioFormatLinearPCM, AVSampleRateKey: format.sampleRate, @@ -82,7 +128,21 @@ public struct AVFoundationAudioExporter: AudioExporter { commonFormat: .pcmFormatFloat32, interleaved: false ) - try file.write(from: try makePCMBuffer(from: buffer, format: format)) + let framesPerChunk = 16_384 + var startFrame = 0 + while startFrame < buffer.frameCount { + try Task.checkCancellation() + let frameCount = min(framesPerChunk, buffer.frameCount - startFrame) + try file.write( + from: try makePCMBuffer( + from: buffer, + startFrame: startFrame, + frameCount: frameCount, + format: format + ) + ) + startFrame += frameCount + } } private func writeM4AFile( @@ -122,13 +182,31 @@ public struct AVFoundationAudioExporter: AudioExporter { let metadataAdaptor = try makeMetadataAdaptor(for: writer, chapterMarkers: chapterMarkers) - writer.startWriting() + guard writer.startWriting() else { + throw writer.error + ?? AudioPipelineError.exportFailed("Asset writer could not start.") + } writer.startSession(atSourceTime: .zero) - try await appendPCMChunks(from: buffer, using: pcmFormat, to: audioInput, writer: writer) - try await appendChapterMarkers(chapterMarkers, with: metadataAdaptor) - - try await finishWriting(writer) + do { + try await appendPCMChunks( + from: buffer, + using: pcmFormat, + to: audioInput, + writer: writer + ) + try await appendChapterMarkers( + chapterMarkers, + with: metadataAdaptor, + writer: writer + ) + try await finishWriting(writer) + } catch { + if writer.status == .writing || writer.status == .unknown { + writer.cancelWriting() + } + throw error + } } private func appendPCMChunks( @@ -147,7 +225,10 @@ public struct AVFoundationAudioExporter: AudioExporter { var nextFrame = 0 while nextFrame < totalFrames { - try await waitUntilReadyForMoreMediaData(on: audioInput) + try await waitUntilReadyForMoreMediaData( + on: audioInput, + writer: writer + ) let count = min(framesPerChunk, totalFrames - nextFrame) let sampleBuffer = try makePCMSampleBuffer( @@ -170,12 +251,16 @@ public struct AVFoundationAudioExporter: AudioExporter { private func appendChapterMarkers( _ chapterMarkers: [AudioChapterMarker], - with metadataAdaptor: MetadataAdaptor? + with metadataAdaptor: MetadataAdaptor?, + writer: AVAssetWriter ) async throws { guard let metadataAdaptor else { return } for marker in chapterMarkers { - try await waitUntilReadyForMoreMediaData(on: metadataAdaptor.input) + try await waitUntilReadyForMoreMediaData( + on: metadataAdaptor.input, + writer: writer + ) let item = AVMutableMetadataItem() item.identifier = .quickTimeUserDataChapter @@ -214,14 +299,50 @@ public struct AVFoundationAudioExporter: AudioExporter { } } - private func waitUntilReadyForMoreMediaData(on input: AVAssetWriterInput) async throws { + private func waitUntilReadyForMoreMediaData( + on input: AVAssetWriterInput, + writer: AVAssetWriter + ) async throws { + try Task.checkCancellation() + var backoffMilliseconds = 1 while !input.isReadyForMoreMediaData { try Task.checkCancellation() - await Task.yield() + try Self.requireActiveWriter(writer) + try await Task.sleep(for: .milliseconds(Int64(backoffMilliseconds))) + backoffMilliseconds = min(backoffMilliseconds * 2, 20) } + try Task.checkCancellation() + try Self.requireActiveWriter(writer) } - private func makePCMBuffer(from buffer: AudioPCMBuffer, format: AudioFormatDescriptor) throws -> AVAudioPCMBuffer { + private static func requireActiveWriter(_ writer: AVAssetWriter) throws { + switch writer.status { + case .writing: + return + case .failed: + throw writer.error + ?? AudioPipelineError.exportFailed("Asset writer failed without an error.") + case .cancelled: + throw CancellationError() + case .completed: + throw AudioPipelineError.exportFailed( + "Asset writer completed before all media was appended." + ) + case .unknown: + throw AudioPipelineError.exportFailed( + "Asset writer returned to an unknown state while exporting." + ) + @unknown default: + throw AudioPipelineError.exportFailed("Asset writer entered an unsupported state.") + } + } + + private func makePCMBuffer( + from buffer: AudioPCMBuffer, + startFrame: Int, + frameCount: Int, + format: AudioFormatDescriptor + ) throws -> AVAudioPCMBuffer { guard format.channelCount > 0, format.sampleRate > 0 else { throw AudioPipelineError.exportFailed("Invalid format: channelCount=\(format.channelCount) sampleRate=\(format.sampleRate)") } @@ -234,10 +355,14 @@ public struct AVFoundationAudioExporter: AudioExporter { throw AudioPipelineError.exportFailed("Unable to create PCM buffer format") } - guard buffer.frameCount >= 0, buffer.frameCount <= Int(AVAudioFrameCount.max) else { - throw AudioPipelineError.exportFailed("Frame count \(buffer.frameCount) exceeds AVAudioPCMBuffer capacity") + guard startFrame >= 0, + frameCount >= 0, + startFrame <= buffer.frameCount, + frameCount <= buffer.frameCount - startFrame, + frameCount <= Int(AVAudioFrameCount.max) else { + throw AudioPipelineError.exportFailed("Invalid PCM export frame range") } - let frameCapacity = AVAudioFrameCount(buffer.frameCount) + let frameCapacity = AVAudioFrameCount(frameCount) guard let pcmBuffer = AVAudioPCMBuffer( pcmFormat: avFormat, frameCapacity: frameCapacity @@ -246,7 +371,12 @@ public struct AVFoundationAudioExporter: AudioExporter { } pcmBuffer.frameLength = frameCapacity - PCMCoding.fill(pcmBuffer, from: buffer.channels, frameCount: buffer.frameCount) + PCMCoding.fill( + pcmBuffer, + from: buffer.channels, + startFrame: startFrame, + frameCount: frameCount + ) return pcmBuffer } @@ -255,8 +385,15 @@ public struct AVFoundationAudioExporter: AudioExporter { from buffer: AudioPCMBuffer, startFrame: Int, frameCount: Int, - format: AVAudioFormat + format: AVAudioFormat, + presentationFrameOffset: Int = 0 ) throws -> CMSampleBuffer { + let (presentationFrame, overflowed) = presentationFrameOffset.addingReportingOverflow(startFrame) + guard !overflowed, + presentationFrame >= 0, + presentationFrame <= Int(CMTimeValue.max) else { + throw AudioPipelineError.exportFailed("AAC presentation timestamp overflow") + } let interleavedData = makeInterleavedPCMData( from: buffer, startFrame: startFrame, @@ -295,7 +432,7 @@ public struct AVFoundationAudioExporter: AudioExporter { } let presentationTime = CMTime( - value: CMTimeValue(startFrame), + value: CMTimeValue(presentationFrame), timescale: CMTimeScale(format.sampleRate) ) var sampleBuffer: CMSampleBuffer? @@ -339,11 +476,12 @@ public struct AVFoundationAudioExporter: AudioExporter { return data } - private func makeMetadataAdaptor( + fileprivate func makeMetadataAdaptor( for writer: AVAssetWriter, - chapterMarkers: [AudioChapterMarker] + chapterMarkers: [AudioChapterMarker], + includeWhenEmpty: Bool = false ) throws -> MetadataAdaptor? { - guard !chapterMarkers.isEmpty else { return nil } + guard includeWhenEmpty || !chapterMarkers.isEmpty else { return nil } let specification: NSDictionary = [ kCMMetadataFormatDescriptionMetadataSpecificationKey_Identifier as NSString: @@ -376,7 +514,7 @@ public struct AVFoundationAudioExporter: AudioExporter { ) } - private func recommendedBitRate(for channelCount: Int) -> Int { + fileprivate func recommendedBitRate(for channelCount: Int) -> Int { max(64_000, channelCount * 128_000) } @@ -404,12 +542,464 @@ public struct AVFoundationAudioExporter: AudioExporter { private static func normalizedChapterMarkers(_ chapterMarkers: [AudioChapterMarker]) -> [AudioChapterMarker] { chapterMarkers - .filter { !$0.title.trimmingCharacters(in: .whitespacesAndNewlines).isEmpty && $0.duration > 0 } + .filter { + !$0.title.trimmingCharacters(in: .whitespacesAndNewlines).isEmpty + && $0.startTime.isFinite + && $0.startTime >= 0 + && $0.duration.isFinite + && $0.duration > 0 + } .sorted { $0.startTime < $1.startTime } } + + fileprivate static func validate( + buffer: AudioPCMBuffer, + format: AudioFormatDescriptor + ) throws { + guard format.sampleRate.isFinite, + format.sampleRate >= 1_000, + format.sampleRate <= 768_000, + (1...64).contains(format.channelCount), + buffer.channels.count == format.channelCount, + buffer.format.sampleRate.isFinite, + abs(buffer.format.sampleRate - format.sampleRate) + <= max(0.001, format.sampleRate * 1e-9), + buffer.channels.dropFirst().allSatisfy({ + $0.count == buffer.channels[0].count + }) else { + throw AudioPipelineError.exportFailed( + "Audio export requires 1–64 equal-length channels and a finite 1000–768000 Hz sample rate." + ) + } + } } -private struct MetadataAdaptor { +fileprivate struct MetadataAdaptor { let input: AVAssetWriterInput let adaptor: AVAssetWriterInputMetadataAdaptor } + +/// A bounded-memory AAC writer. Each `append` call encodes one chapter in +/// 4,096-frame pieces before returning, so callers can release that chapter's +/// PCM before producing the next one. The completed file is staged beside the +/// destination and atomically installed only after AVAssetWriter succeeds. +public actor StreamingM4AWriter: M4AChapterStreamWriting { + private enum State: Equatable { + case writing + case appending + case finishing + case finished + case cancelled + } + + private static let framesPerChunk = 4_096 + + private let destinationURL: URL + private let stagingURL: URL + private let format: AudioFormatDescriptor + private let pcmFormat: AVAudioFormat + private let writer: AVAssetWriter + private let audioInput: AVAssetWriterInput + private let metadataAdaptor: MetadataAdaptor + private var chapterMarkers: [AudioChapterMarker] = [] + private var totalFrameCount = 0 + private var state = State.writing + private var finalizedAsset: AudioExportedFile? + + public init( + destinationURL: URL, + sampleRate: Double, + channelCount: Int + ) throws { + guard destinationURL.isFileURL, + sampleRate.isFinite, + sampleRate >= 1_000, + sampleRate <= 768_000, + (1...64).contains(channelCount) else { + throw AudioPipelineError.exportFailed( + "M4A streaming export requires a file URL, 1–64 channels, and a finite 1000–768000 Hz sample rate." + ) + } + + let destinationURL = destinationURL.standardizedFileURL + let fileManager = FileManager.default + let destinationDirectory = destinationURL.deletingLastPathComponent() + try fileManager.createDirectory( + at: destinationDirectory, + withIntermediateDirectories: true + ) + try Self.validateDestination(destinationURL, fileManager: fileManager) + + let stagingURL = destinationDirectory + .appendingPathComponent( + ".\(destinationURL.deletingPathExtension().lastPathComponent)-valar-\(UUID().uuidString)", + isDirectory: false + ) + .appendingPathExtension("m4a") + + guard let pcmFormat = AVAudioFormat( + commonFormat: .pcmFormatFloat32, + sampleRate: sampleRate, + channels: AVAudioChannelCount(channelCount), + interleaved: true + ) else { + throw AudioPipelineError.exportFailed("Unable to create PCM export format") + } + + do { + let writer = try AVAssetWriter(outputURL: stagingURL, fileType: .m4a) + let audioInput = AVAssetWriterInput( + mediaType: .audio, + outputSettings: [ + AVFormatIDKey: kAudioFormatMPEG4AAC, + AVSampleRateKey: sampleRate, + AVNumberOfChannelsKey: channelCount, + AVEncoderBitRateKey: AVFoundationAudioExporter() + .recommendedBitRate(for: channelCount), + ], + sourceFormatHint: pcmFormat.formatDescription + ) + audioInput.expectsMediaDataInRealTime = false + guard writer.canAdd(audioInput) else { + throw AudioPipelineError.exportFailed("Unable to add AAC writer input") + } + writer.add(audioInput) + + guard let metadataAdaptor = try AVFoundationAudioExporter() + .makeMetadataAdaptor( + for: writer, + chapterMarkers: [], + includeWhenEmpty: true + ) else { + throw AudioPipelineError.exportFailed("Unable to add chapter metadata input") + } + + guard writer.startWriting() else { + throw writer.error + ?? AudioPipelineError.exportFailed("Asset writer could not start.") + } + writer.startSession(atSourceTime: .zero) + + self.destinationURL = destinationURL + self.stagingURL = stagingURL + self.format = AudioFormatDescriptor( + sampleRate: sampleRate, + channelCount: channelCount, + sampleFormat: .float32, + interleaved: false, + container: "m4a" + ) + self.pcmFormat = pcmFormat + self.writer = writer + self.audioInput = audioInput + self.metadataAdaptor = metadataAdaptor + } catch { + try? fileManager.removeItem(at: stagingURL) + throw error + } + } + + public func append( + _ buffer: AudioPCMBuffer, + chapterTitle: String + ) async throws { + guard state == .writing else { + throw AudioPipelineError.exportFailed( + "Another append or finalization operation is already active on this M4A writer" + ) + } + state = .appending + do { + try AVFoundationAudioExporter.validate(buffer: buffer, format: format) + + let (newTotalFrameCount, overflowed) = totalFrameCount + .addingReportingOverflow(buffer.frameCount) + guard !overflowed, + newTotalFrameCount <= Int(CMTimeValue.max) else { + throw AudioPipelineError.exportFailed("M4A frame count overflow") + } + + var nextFrame = 0 + while nextFrame < buffer.frameCount { + try await waitUntilReadyForMoreMediaData(on: audioInput) + let frameCount = min( + Self.framesPerChunk, + buffer.frameCount - nextFrame + ) + let sampleBuffer = try makePCMSampleBuffer( + from: buffer, + startFrame: nextFrame, + frameCount: frameCount, + presentationFrameOffset: totalFrameCount + ) + guard audioInput.append(sampleBuffer) else { + throw writer.error + ?? AudioPipelineError.exportFailed("Failed to append AAC sample buffer") + } + nextFrame += frameCount + } + + if let marker = Self.chapterMarker( + title: chapterTitle, + startFrame: totalFrameCount, + frameCount: buffer.frameCount, + sampleRate: format.sampleRate + ) { + try await appendChapterMarker(marker) + chapterMarkers.append(marker) + } + totalFrameCount = newTotalFrameCount + state = .writing + } catch { + cancelWriterAndRemoveStagingFile() + throw error + } + } + + nonisolated static func chapterMarker( + title: String, + startFrame: Int, + frameCount: Int, + sampleRate: Double + ) -> AudioChapterMarker? { + let title = title.trimmingCharacters(in: .whitespacesAndNewlines) + guard !title.isEmpty, + startFrame >= 0, + frameCount > 0, + sampleRate.isFinite, + sampleRate > 0 else { + return nil + } + return AudioChapterMarker( + title: title, + startTime: Double(startFrame) / sampleRate, + duration: Double(frameCount) / sampleRate + ) + } + + public func finalize() async throws -> AudioExportedFile { + if let finalizedAsset { + return finalizedAsset + } + guard state == .writing else { + throw AudioPipelineError.exportFailed("Cannot finalize a cancelled M4A writer") + } + + state = .finishing + audioInput.markAsFinished() + metadataAdaptor.input.markAsFinished() + do { + await withCheckedContinuation { continuation in + writer.finishWriting { + continuation.resume() + } + } + if let error = writer.error { + throw error + } + guard writer.status == .completed else { + throw AudioPipelineError.exportFailed( + "Asset writer finished with status \(writer.status.rawValue)" + ) + } + try Task.checkCancellation() + try Self.commitStagedFile( + stagingURL, + to: destinationURL, + fileManager: .default + ) + let asset = AudioExportedFile( + url: destinationURL, + format: format, + chapterMarkers: chapterMarkers + ) + finalizedAsset = asset + state = .finished + return asset + } catch { + cancelWriterAndRemoveStagingFile() + throw error + } + } + + public func cancel() async { + switch state { + case .writing, .appending, .finishing: + cancelWriterAndRemoveStagingFile() + case .finished, .cancelled: + return + } + } + + private func appendChapterMarker(_ marker: AudioChapterMarker) async throws { + try await waitUntilReadyForMoreMediaData(on: metadataAdaptor.input) + + let item = AVMutableMetadataItem() + item.identifier = .quickTimeUserDataChapter + item.dataType = kCMMetadataBaseDataType_UTF8 as String + item.extendedLanguageTag = "und" + item.value = marker.title as NSString + + let group = AVTimedMetadataGroup( + items: [item], + timeRange: CMTimeRange( + start: CMTime(seconds: marker.startTime, preferredTimescale: 600), + duration: CMTime(seconds: marker.duration, preferredTimescale: 600) + ) + ) + guard metadataAdaptor.adaptor.append(group) else { + throw writer.error + ?? AudioPipelineError.exportFailed("Failed to append chapter metadata") + } + } + + private func makePCMSampleBuffer( + from buffer: AudioPCMBuffer, + startFrame: Int, + frameCount: Int, + presentationFrameOffset: Int + ) throws -> CMSampleBuffer { + let (presentationFrame, overflowed) = presentationFrameOffset + .addingReportingOverflow(startFrame) + guard !overflowed, + presentationFrame >= 0, + presentationFrame <= Int(CMTimeValue.max) else { + throw AudioPipelineError.exportFailed("AAC presentation timestamp overflow") + } + + let interleavedData = AVFoundationAudioExporter().makeInterleavedPCMData( + from: buffer, + startFrame: startFrame, + frameCount: frameCount, + channelCount: format.channelCount + ) + var blockBuffer: CMBlockBuffer? + let status = CMBlockBufferCreateWithMemoryBlock( + allocator: kCFAllocatorDefault, + memoryBlock: nil, + blockLength: interleavedData.count, + blockAllocator: nil, + customBlockSource: nil, + offsetToData: 0, + dataLength: interleavedData.count, + flags: 0, + blockBufferOut: &blockBuffer + ) + guard status == kCMBlockBufferNoErr, let blockBuffer else { + throw AudioPipelineError.exportFailed("Unable to allocate PCM block buffer") + } + try interleavedData.withUnsafeBytes { bytes in + guard let baseAddress = bytes.baseAddress else { return } + let replaceStatus = CMBlockBufferReplaceDataBytes( + with: baseAddress, + blockBuffer: blockBuffer, + offsetIntoDestination: 0, + dataLength: interleavedData.count + ) + guard replaceStatus == kCMBlockBufferNoErr else { + throw AudioPipelineError.exportFailed("Unable to populate PCM block buffer") + } + } + + var sampleBuffer: CMSampleBuffer? + let sampleStatus = CMAudioSampleBufferCreateReadyWithPacketDescriptions( + allocator: kCFAllocatorDefault, + dataBuffer: blockBuffer, + formatDescription: pcmFormat.formatDescription, + sampleCount: frameCount, + presentationTimeStamp: CMTime( + value: CMTimeValue(presentationFrame), + timescale: CMTimeScale(format.sampleRate) + ), + packetDescriptions: nil, + sampleBufferOut: &sampleBuffer + ) + guard sampleStatus == noErr, let sampleBuffer else { + throw AudioPipelineError.exportFailed("Unable to create PCM sample buffer") + } + return sampleBuffer + } + + private func waitUntilReadyForMoreMediaData( + on input: AVAssetWriterInput + ) async throws { + try Task.checkCancellation() + var backoffMilliseconds = 1 + while !input.isReadyForMoreMediaData { + try Task.checkCancellation() + try requireActiveWriter() + try await Task.sleep(for: .milliseconds(Int64(backoffMilliseconds))) + backoffMilliseconds = min(backoffMilliseconds * 2, 20) + } + try Task.checkCancellation() + try requireActiveWriter() + } + + private func requireActiveWriter() throws { + switch writer.status { + case .writing: + return + case .failed: + throw writer.error + ?? AudioPipelineError.exportFailed("Asset writer failed without an error.") + case .cancelled: + throw CancellationError() + case .completed: + throw AudioPipelineError.exportFailed( + "Asset writer completed before all media was appended." + ) + case .unknown: + throw AudioPipelineError.exportFailed( + "Asset writer returned to an unknown state while exporting." + ) + @unknown default: + throw AudioPipelineError.exportFailed("Asset writer entered an unsupported state.") + } + } + + private func cancelWriterAndRemoveStagingFile() { + if writer.status == .writing || writer.status == .unknown { + writer.cancelWriting() + } + try? FileManager.default.removeItem(at: stagingURL) + state = .cancelled + } + + private static func validateDestination( + _ destinationURL: URL, + fileManager: FileManager + ) throws { + guard !destinationURL.lastPathComponent.isEmpty else { + throw AudioPipelineError.exportFailed("The M4A destination has no file name.") + } + guard fileManager.fileExists(atPath: destinationURL.path) else { return } + let values = try destinationURL.resourceValues( + forKeys: [.isDirectoryKey, .isRegularFileKey, .isSymbolicLinkKey] + ) + guard values.isDirectory != true, + values.isRegularFile == true, + values.isSymbolicLink != true else { + throw AudioPipelineError.exportFailed( + "The audio destination must be a regular, non-symbolic-link file." + ) + } + } + + private static func commitStagedFile( + _ stagingURL: URL, + to destinationURL: URL, + fileManager: FileManager + ) throws { + try validateDestination(destinationURL, fileManager: fileManager) + if fileManager.fileExists(atPath: destinationURL.path) { + _ = try fileManager.replaceItemAt( + destinationURL, + withItemAt: stagingURL, + backupItemName: nil, + options: [] + ) + } else { + try fileManager.moveItem(at: stagingURL, to: destinationURL) + } + } +} diff --git a/Packages/ValarAudio/Sources/ValarAudio/AccelerateAudioResampler.swift b/Packages/ValarAudio/Sources/ValarAudio/AccelerateAudioResampler.swift index 19419ae..f1952f5 100644 --- a/Packages/ValarAudio/Sources/ValarAudio/AccelerateAudioResampler.swift +++ b/Packages/ValarAudio/Sources/ValarAudio/AccelerateAudioResampler.swift @@ -97,7 +97,7 @@ private extension AccelerateAudioResampler { return desample( channel, decimationFactor: exactFactor, - filter: [1], + filter: downsamplePlan.filter, outputCount: targetFrameCount ) } @@ -143,7 +143,10 @@ private extension AccelerateAudioResampler { return DownsamplePlan( exactFactor: roundedFactor, decimationFactor: roundedFactor, - filter: [1], + filter: cachedLowPassFilter( + overallRatio: targetRate / sourceRate, + decimationFactor: roundedFactor + ), residualRatio: 1 ) } diff --git a/Packages/ValarAudio/Sources/ValarAudio/AudioEnginePlayer.swift b/Packages/ValarAudio/Sources/ValarAudio/AudioEnginePlayer.swift index 2fca77c..d99ddd4 100644 --- a/Packages/ValarAudio/Sources/ValarAudio/AudioEnginePlayer.swift +++ b/Packages/ValarAudio/Sources/ValarAudio/AudioEnginePlayer.swift @@ -2,7 +2,14 @@ import AVFoundation import AudioToolbox public actor AudioEnginePlayer { + enum QueueDrainDisposition: Equatable { + case awaitMoreAudio + case finishPlayback(didPlayAudio: Bool) + } + private let engine: AVAudioEngine + private let playbackQueueLimits: AudioPlaybackQueueLimits + private let playbackAdmission: AudioPlaybackQueueAdmissionController private var playerNode: AVAudioPlayerNode? private var pendingBuffers = AudioChunkRingBuffer() @@ -10,6 +17,7 @@ public actor AudioEnginePlayer { private var totalEnqueuedFrames: Int64 = 0 private var playedFrames: Int64 = 0 private var playbackBaseFrames: Int64 = 0 + private var playbackNodeBaseSampleTime: Int64 = 0 private var scheduledBufferCount = 0 private var finishedStreaming = false private var isPlaying = false @@ -18,32 +26,87 @@ public actor AudioEnginePlayer { private var playbackSessionID: UInt64 = 0 public init() { + let limits = AudioPlaybackQueueLimits.interactive + self.engine = AVAudioEngine() + self.playbackQueueLimits = limits + self.playbackAdmission = Self.makeDefaultPlaybackAdmission(limits: limits) + } + + public init(playbackQueueLimits: AudioPlaybackQueueLimits) throws { self.engine = AVAudioEngine() + self.playbackQueueLimits = playbackQueueLimits + self.playbackAdmission = try AudioPlaybackQueueAdmissionController( + limits: playbackQueueLimits + ) } public func play(_ buffer: AudioPCMBuffer) async throws { - stop() - try feedChunk(buffer) - finishStream() + await stop() + do { + try Self.validatePlaybackBufferShape(buffer) + let chunkFrameCount = try maximumPlaybackChunkFrameCount( + for: buffer.format + ) + var startFrame = 0 + while startFrame < buffer.frameCount { + try Task.checkCancellation() + let endFrame = startFrame + min( + chunkFrameCount, + buffer.frameCount - startFrame + ) + let channels = buffer.channels.map { channel -> [Float] in + guard startFrame < channel.count else { return [] } + return Array(channel[startFrame ..< min(endFrame, channel.count)]) + } + try await feedChunk( + AudioPCMBuffer(channels: channels, format: buffer.format) + ) + startFrame = endFrame + } + finishStream() + } catch { + await stop() + throw error + } } - public func feedChunk(_ buffer: AudioPCMBuffer) throws { - try ensureEngineReady(for: buffer.format) + public func feedChunk(_ buffer: AudioPCMBuffer) async throws { + try Self.validatePlaybackBufferShape(buffer) + let sessionID = playbackSessionID + let cost = AudioPlaybackQueueCost( + frameCount: Int64(buffer.frameCount), + sampleRate: buffer.format.sampleRate, + channelCount: buffer.format.channelCount, + bytesPerSample: MemoryLayout.stride + ) + let reservation = try await playbackAdmission.acquire(cost: cost) + guard sessionID == playbackSessionID, !Task.isCancelled else { + await playbackAdmission.release(reservation) + throw CancellationError() + } - finishedStreaming = false - didFinishPlayback = false + do { + try ensureEngineReady(for: buffer.format) - let avBuffer = try Self.makeAVAudioBuffer(from: buffer) - let queuedBuffer = QueuedAudioBuffer( - buffer: avBuffer, - frameCount: Int64(buffer.frameCount) - ) + finishedStreaming = false + didFinishPlayback = false - pendingBuffers.append(queuedBuffer) - totalEnqueuedFrames += queuedBuffer.frameCount + let avBuffer = try Self.makeAVAudioBuffer(from: buffer) + let queuedBuffer = QueuedAudioBuffer( + buffer: avBuffer, + frameCount: Int64(buffer.frameCount), + reservation: reservation + ) - schedulePendingBuffers() - beginPlaybackIfNeeded() + pendingBuffers.append(queuedBuffer) + totalEnqueuedFrames += queuedBuffer.frameCount + + schedulePendingBuffers() + beginPlaybackIfNeeded() + } catch { + await playbackAdmission.release(reservation) + throw error + } } /// Fast path that feeds a raw Float array directly into `AVAudioPCMBuffer.floatChannelData` @@ -52,24 +115,46 @@ public actor AudioEnginePlayer { /// - Parameters: /// - samples: Mono PCM samples in the range [-1, 1]. /// - sampleRate: Sample rate of the incoming audio (e.g. 24_000). - public func feedSamples(_ samples: [Float], sampleRate: Double) throws { + public func feedSamples(_ samples: [Float], sampleRate: Double) async throws { + let sessionID = playbackSessionID let format = AudioFormatDescriptor(sampleRate: sampleRate, channelCount: 1) - try ensureEngineReady(for: format) + let cost = AudioPlaybackQueueCost( + frameCount: Int64(samples.count), + sampleRate: sampleRate, + channelCount: 1, + bytesPerSample: MemoryLayout.stride + ) + let reservation = try await playbackAdmission.acquire(cost: cost) + guard sessionID == playbackSessionID, !Task.isCancelled else { + await playbackAdmission.release(reservation) + throw CancellationError() + } - finishedStreaming = false - didFinishPlayback = false + do { + try ensureEngineReady(for: format) - let avBuffer = try Self.makeAVAudioBufferFromSamples(samples, sampleRate: sampleRate) - let queuedBuffer = QueuedAudioBuffer( - buffer: avBuffer, - frameCount: Int64(samples.count) - ) + finishedStreaming = false + didFinishPlayback = false - pendingBuffers.append(queuedBuffer) - totalEnqueuedFrames += queuedBuffer.frameCount + let avBuffer = try Self.makeAVAudioBufferFromSamples( + samples, + sampleRate: sampleRate + ) + let queuedBuffer = QueuedAudioBuffer( + buffer: avBuffer, + frameCount: Int64(samples.count), + reservation: reservation + ) - schedulePendingBuffers() - beginPlaybackIfNeeded() + pendingBuffers.append(queuedBuffer) + totalEnqueuedFrames += queuedBuffer.frameCount + + schedulePendingBuffers() + beginPlaybackIfNeeded() + } catch { + await playbackAdmission.release(reservation) + throw error + } } public func finishStream() { @@ -77,7 +162,7 @@ public actor AudioEnginePlayer { updateTerminalStateIfNeeded() } - public func stop() { + public func stop() async { playbackSessionID &+= 1 playerNode?.stop() if engine.isRunning { @@ -85,6 +170,7 @@ public actor AudioEnginePlayer { } resetPlaybackState(didFinishPlayback: false) + await playbackAdmission.reset() } public func playbackSnapshot() -> AudioPlaybackSnapshot { @@ -100,9 +186,81 @@ public actor AudioEnginePlayer { didFinish: didFinishPlayback ) } + + nonisolated static func queueDrainDisposition( + finishedStreaming: Bool, + totalEnqueuedFrames: Int64 + ) -> QueueDrainDisposition { + if finishedStreaming { + return .finishPlayback(didPlayAudio: totalEnqueuedFrames > 0) + } + return .awaitMoreAudio + } + + private nonisolated static func makeDefaultPlaybackAdmission( + limits: AudioPlaybackQueueLimits + ) + -> AudioPlaybackQueueAdmissionController + { + do { + return try AudioPlaybackQueueAdmissionController(limits: limits) + } catch { + preconditionFailure("Invalid library-owned interactive playback limits: \(error)") + } + } } extension AudioEnginePlayer { + private func maximumPlaybackChunkFrameCount( + for format: AudioFormatDescriptor + ) throws -> Int { + guard format.channelCount > 0, + AVAudioChannelCount(exactly: format.channelCount) != nil, + format.sampleRate.isFinite, + format.sampleRate > 0 else { + throw AudioEnginePlayerError.invalidFormat( + sampleRate: format.sampleRate, + channelCount: format.channelCount + ) + } + + let (bytesPerFrame, byteWidthOverflow) = format.channelCount + .multipliedReportingOverflow(by: MemoryLayout.stride) + guard !byteWidthOverflow, bytesPerFrame > 0 else { + throw AudioEnginePlayerError.invalidFormat( + sampleRate: format.sampleRate, + channelCount: format.channelCount + ) + } + let framesByBytes = playbackQueueLimits.maximumScheduledBytes + / Int64(bytesPerFrame) + + let durationFrameCount = ( + playbackQueueLimits.highWaterDuration * format.sampleRate + ).rounded(.down) + guard durationFrameCount.isFinite, durationFrameCount >= 1 else { + throw AudioEnginePlayerError.invalidFormat( + sampleRate: format.sampleRate, + channelCount: format.channelCount + ) + } + let framesByDuration = durationFrameCount >= Double(Int64.max) + ? Int64.max + : Int64(durationFrameCount) + let frameCount = min( + framesByBytes, + framesByDuration, + Int64(Int.max) + ) + guard frameCount > 0 else { + throw AudioEnginePlayerError.invalidFormat( + sampleRate: format.sampleRate, + channelCount: format.channelCount + ) + } + return Int(frameCount) + } + private func resolvedPlayerNode() throws -> AVAudioPlayerNode { if let playerNode { return playerNode @@ -157,6 +315,7 @@ extension AudioEnginePlayer { while let queuedBuffer = pendingBuffers.popFirst() { scheduledBufferCount += 1 let frameCount = queuedBuffer.frameCount + let reservation = queuedBuffer.reservation let sessionID = playbackSessionID playerNode.scheduleBuffer( @@ -164,7 +323,11 @@ extension AudioEnginePlayer { completionCallbackType: .dataPlayedBack ) { [self] _ in Task { - await markBufferPlayedBack(frameCount: frameCount, sessionID: sessionID) + await markBufferPlayedBack( + frameCount: frameCount, + reservation: reservation, + sessionID: sessionID + ) } } } @@ -174,6 +337,12 @@ extension AudioEnginePlayer { guard let playerNode else { return } guard scheduledBufferCount > 0 else { return } guard !playerNode.isPlaying else { + if isBuffering { + // The player clock continues through an underrun. Rebase it so + // silent wait time is not counted as newly played audio. + playbackBaseFrames = playedFrames + playbackNodeBaseSampleTime = currentNodeSampleTime() ?? 0 + } isPlaying = true isBuffering = false return @@ -181,11 +350,17 @@ extension AudioEnginePlayer { playbackBaseFrames = playedFrames playerNode.play() + playbackNodeBaseSampleTime = currentNodeSampleTime() ?? 0 isPlaying = true isBuffering = false } - private func markBufferPlayedBack(frameCount: Int64, sessionID: UInt64) { + private func markBufferPlayedBack( + frameCount: Int64, + reservation: AudioPlaybackQueueReservation, + sessionID: UInt64 + ) async { + await playbackAdmission.release(reservation) guard sessionID == playbackSessionID else { return } playedFrames += frameCount scheduledBufferCount = max(scheduledBufferCount - 1, 0) @@ -196,15 +371,21 @@ extension AudioEnginePlayer { guard let playerNode else { return } guard scheduledBufferCount == 0, pendingBuffers.isEmpty else { return } - playerNode.stop() - - if finishedStreaming { + switch Self.queueDrainDisposition( + finishedStreaming: finishedStreaming, + totalEnqueuedFrames: totalEnqueuedFrames + ) { + case .finishPlayback(let didPlayAudio): + playerNode.stop() if engine.isRunning { engine.stop() } - resetPlaybackState(didFinishPlayback: totalEnqueuedFrames > 0) - } else { + resetPlaybackState(didFinishPlayback: didPlayAudio) + case .awaitMoreAudio: + // Keep the engine and node running across a temporary underrun. + // Scheduling the next buffer then resumes without a stop/start cycle. playbackBaseFrames = playedFrames + playbackNodeBaseSampleTime = currentNodeSampleTime() ?? 0 isPlaying = false isBuffering = totalEnqueuedFrames > 0 didFinishPlayback = false @@ -217,6 +398,7 @@ extension AudioEnginePlayer { totalEnqueuedFrames = 0 playedFrames = 0 playbackBaseFrames = 0 + playbackNodeBaseSampleTime = 0 scheduledBufferCount = 0 finishedStreaming = false isPlaying = false @@ -228,16 +410,23 @@ extension AudioEnginePlayer { guard let playerNode else { return playedFrames } var positionFrames = playedFrames - if playerNode.isPlaying, - let renderTime = playerNode.lastRenderTime, - let playerTime = playerNode.playerTime(forNodeTime: renderTime) { - let liveFrames = max(Int64(playerTime.sampleTime), 0) + if playerNode.isPlaying, let nodeSampleTime = currentNodeSampleTime() { + let liveFrames = max(nodeSampleTime - playbackNodeBaseSampleTime, 0) positionFrames = max(positionFrames, playbackBaseFrames + liveFrames) } return min(positionFrames, totalEnqueuedFrames) } + private func currentNodeSampleTime() -> Int64? { + guard let playerNode, + let renderTime = playerNode.lastRenderTime, + let playerTime = playerNode.playerTime(forNodeTime: renderTime) else { + return nil + } + return Int64(playerTime.sampleTime) + } + private static func playbackComponentIsAvailable() -> Bool { var description = AudioComponentDescription( componentType: kAudioUnitType_Generator, @@ -251,11 +440,18 @@ extension AudioEnginePlayer { } private static func makeAVAudioFormat(from format: AudioFormatDescriptor) throws -> AVAudioFormat { - guard let avFormat = AVAudioFormat( + guard format.sampleRate.isFinite, + format.sampleRate > 0, + let channelCount = AVAudioChannelCount(exactly: format.channelCount), + channelCount > 0, + let avFormat = AVAudioFormat( commonFormat: .pcmFormatFloat32, sampleRate: format.sampleRate, - channels: AVAudioChannelCount(format.channelCount), - interleaved: format.interleaved + channels: channelCount, + // AudioPCMBuffer stores one Float array per channel regardless of + // the source container layout. Keep the AVFoundation buffer planar + // so PCMCoding can copy every channel through floatChannelData. + interleaved: false ) else { throw AudioEnginePlayerError.invalidFormat( sampleRate: format.sampleRate, @@ -266,6 +462,20 @@ extension AudioEnginePlayer { return avFormat } + private static func validatePlaybackBufferShape( + _ buffer: AudioPCMBuffer + ) throws { + guard buffer.channels.count == buffer.format.channelCount, + AVAudioChannelCount(exactly: buffer.format.channelCount) != nil, + buffer.format.sampleRate.isFinite, + buffer.format.sampleRate > 0 else { + throw AudioEnginePlayerError.invalidFormat( + sampleRate: buffer.format.sampleRate, + channelCount: buffer.format.channelCount + ) + } + } + private static func makeAVAudioBuffer(from buffer: AudioPCMBuffer) throws -> AVAudioPCMBuffer { guard buffer.format.channelCount > 0, buffer.format.sampleRate > 0 else { throw AudioEnginePlayerError.invalidFormat( @@ -340,6 +550,7 @@ extension AudioEnginePlayer { private struct QueuedAudioBuffer { let buffer: AVAudioPCMBuffer let frameCount: Int64 + let reservation: AudioPlaybackQueueReservation } private struct AudioChunkRingBuffer { diff --git a/Packages/ValarAudio/Sources/ValarAudio/AudioPlaybackQueueAdmissionController.swift b/Packages/ValarAudio/Sources/ValarAudio/AudioPlaybackQueueAdmissionController.swift new file mode 100644 index 0000000..432befc --- /dev/null +++ b/Packages/ValarAudio/Sources/ValarAudio/AudioPlaybackQueueAdmissionController.swift @@ -0,0 +1,433 @@ +import Foundation + +/// Low/high-water limits for buffers submitted to an audio playback queue. +/// +/// Decoded bytes provide the hard PCM memory bound. Duration limits producer +/// lead, `maximumScheduledBufferCount` bounds overhead from many tiny chunks, +/// and `maximumWaitingProducerCount` prevents suspended producers (and their +/// source PCM) from accumulating without limit. +public struct AudioPlaybackQueueLimits: Codable, Equatable, Sendable { + public let lowWaterDuration: TimeInterval + public let highWaterDuration: TimeInterval + public let maximumScheduledBufferCount: Int + public let maximumWaitingProducerCount: Int + public let maximumScheduledBytes: Int64 + + public init( + lowWaterDuration: TimeInterval, + highWaterDuration: TimeInterval, + maximumScheduledBufferCount: Int, + maximumWaitingProducerCount: Int = 1, + maximumScheduledBytes: Int64 = 8 * 1_024 * 1_024 + ) { + self.lowWaterDuration = lowWaterDuration + self.highWaterDuration = highWaterDuration + self.maximumScheduledBufferCount = maximumScheduledBufferCount + self.maximumWaitingProducerCount = maximumWaitingProducerCount + self.maximumScheduledBytes = maximumScheduledBytes + } + + /// A deliberately small interactive window. Callers should split larger + /// source buffers before requesting admission. + public static let interactive = Self( + lowWaterDuration: 1.5, + highWaterDuration: 4, + maximumScheduledBufferCount: 8, + maximumWaitingProducerCount: 1, + maximumScheduledBytes: 8 * 1_024 * 1_024 + ) +} +/// The amount of PCM represented by one scheduled playback buffer. +public struct AudioPlaybackQueueCost: Codable, Equatable, Sendable { + public let frameCount: Int64 + public let sampleRate: Double + public let channelCount: Int + public let bytesPerSample: Int + + public init( + frameCount: Int64, + sampleRate: Double, + channelCount: Int = 1, + bytesPerSample: Int = MemoryLayout.stride + ) { + self.frameCount = frameCount + self.sampleRate = sampleRate + self.channelCount = channelCount + self.bytesPerSample = bytesPerSample + } + + public var duration: TimeInterval { + guard frameCount >= 0, sampleRate.isFinite, sampleRate > 0 else { return 0 } + return TimeInterval(frameCount) / sampleRate + } + + public var decodedByteCount: Int64? { + guard frameCount >= 0, channelCount > 0, bytesPerSample > 0 else { + return nil + } + let (sampleCount, sampleOverflow) = frameCount.multipliedReportingOverflow( + by: Int64(channelCount) + ) + guard !sampleOverflow else { return nil } + let (byteCount, byteOverflow) = sampleCount.multipliedReportingOverflow( + by: Int64(bytesPerSample) + ) + return byteOverflow ? nil : byteCount + } +} + +/// An idempotently releasable claim on playback queue capacity. +public struct AudioPlaybackQueueReservation: Equatable, Sendable, Identifiable { + public let id: UUID + public let cost: AudioPlaybackQueueCost + + fileprivate init(id: UUID, cost: AudioPlaybackQueueCost) { + self.id = id + self.cost = cost + } +} + +public struct AudioPlaybackQueueAdmissionSnapshot: Equatable, Sendable { + public let limits: AudioPlaybackQueueLimits + public let scheduledDuration: TimeInterval + public let scheduledFrameCount: Int64 + public let scheduledBufferCount: Int + public let scheduledByteCount: Int64 + public let waitingProducerCount: Int + public let isBackpressuring: Bool + + public init( + limits: AudioPlaybackQueueLimits, + scheduledDuration: TimeInterval, + scheduledFrameCount: Int64, + scheduledBufferCount: Int, + scheduledByteCount: Int64, + waitingProducerCount: Int, + isBackpressuring: Bool + ) { + self.limits = limits + self.scheduledDuration = scheduledDuration + self.scheduledFrameCount = scheduledFrameCount + self.scheduledBufferCount = scheduledBufferCount + self.scheduledByteCount = scheduledByteCount + self.waitingProducerCount = waitingProducerCount + self.isBackpressuring = isBackpressuring + } +} + +public enum AudioPlaybackQueueAdmissionError: LocalizedError, Equatable, Sendable { + case invalidLimits(AudioPlaybackQueueLimits) + case invalidCost(AudioPlaybackQueueCost) + case bufferExceedsHighWater(AudioPlaybackQueueCost, AudioPlaybackQueueLimits) + case bufferExceedsByteLimit(AudioPlaybackQueueCost, AudioPlaybackQueueLimits) + case tooManyWaitingProducers(limit: Int) + + public var errorDescription: String? { + switch self { + case .invalidLimits: + return "Playback queue limits require a finite nonnegative low water, a larger finite high water, and positive counts." + case .invalidCost: + return "Playback queue costs require a nonnegative frame count, a finite positive sample rate, and positive channel/sample widths." + case .bufferExceedsHighWater: + return "The playback buffer is larger than the configured high-water duration; split it before scheduling." + case .bufferExceedsByteLimit: + return "The decoded playback buffer is larger than the configured byte limit; split it before scheduling." + case .tooManyWaitingProducers(let limit): + return "The playback queue already has its limit of \(limit) waiting producer(s)." + } + } +} + +/// FIFO, cancellation-aware admission for a bounded playback queue. +/// +/// Once a producer reaches the high-water mark, new work stays suspended until +/// released audio reaches the low-water mark. This hysteresis avoids waking a +/// producer for every render callback. `reset()` cancels all suspended producers +/// and invalidates outstanding reservations, making it suitable for `stop()`. +/// +/// The controller owns no PCM or `AVAudioPCMBuffer` objects. A playback engine +/// acquires immediately before materializing/scheduling a buffer and releases +/// from its render-completion callback. +public actor AudioPlaybackQueueAdmissionController { + private struct Waiter { + let id: UUID + let cost: AudioPlaybackQueueCost + let durationNanoseconds: UInt64 + let decodedByteCount: Int64 + let continuation: CheckedContinuation + } + + private let limits: AudioPlaybackQueueLimits + private let lowWaterNanoseconds: UInt64 + private let highWaterNanoseconds: UInt64 + + private var scheduledNanoseconds: UInt64 = 0 + private var scheduledFrameCount: Int64 = 0 + private var scheduledByteCount: Int64 = 0 + private var activeReservations: [ + UUID: ( + cost: AudioPlaybackQueueCost, + durationNanoseconds: UInt64, + decodedByteCount: Int64 + ) + ] = [:] + private var waiters: [Waiter] = [] + private var isBackpressuring = false + + public init(limits: AudioPlaybackQueueLimits = .interactive) throws { + guard let lowWaterNanoseconds = Self.nanoseconds(for: limits.lowWaterDuration), + let highWaterNanoseconds = Self.nanoseconds(for: limits.highWaterDuration), + lowWaterNanoseconds < highWaterNanoseconds, + limits.maximumScheduledBufferCount > 0, + limits.maximumWaitingProducerCount > 0, + limits.maximumScheduledBytes > 0 else { + throw AudioPlaybackQueueAdmissionError.invalidLimits(limits) + } + + self.limits = limits + self.lowWaterNanoseconds = lowWaterNanoseconds + self.highWaterNanoseconds = highWaterNanoseconds + } + + /// Suspends until the buffer can be scheduled without crossing the high + /// water or buffer-count limit. Requests remain FIFO once backpressure + /// begins; a smaller request cannot bypass a blocked earlier request. + public func acquire( + cost: AudioPlaybackQueueCost + ) async throws -> AudioPlaybackQueueReservation { + guard let durationNanoseconds = Self.durationNanoseconds(for: cost) else { + throw AudioPlaybackQueueAdmissionError.invalidCost(cost) + } + guard let decodedByteCount = cost.decodedByteCount else { + throw AudioPlaybackQueueAdmissionError.invalidCost(cost) + } + guard durationNanoseconds <= highWaterNanoseconds else { + throw AudioPlaybackQueueAdmissionError.bufferExceedsHighWater(cost, limits) + } + guard decodedByteCount <= limits.maximumScheduledBytes else { + throw AudioPlaybackQueueAdmissionError.bufferExceedsByteLimit(cost, limits) + } + try Task.checkCancellation() + + if waiters.isEmpty, + !isBackpressuring, + canAdmit( + cost: cost, + durationNanoseconds: durationNanoseconds, + decodedByteCount: decodedByteCount + ) { + let reservation = makeReservation( + cost: cost, + durationNanoseconds: durationNanoseconds, + decodedByteCount: decodedByteCount + ) + do { + try Task.checkCancellation() + return reservation + } catch { + release(reservation) + throw error + } + } + + guard waiters.count < limits.maximumWaitingProducerCount else { + throw AudioPlaybackQueueAdmissionError.tooManyWaitingProducers( + limit: limits.maximumWaitingProducerCount + ) + } + + isBackpressuring = true + let requestID = UUID() + let reservation = try await withTaskCancellationHandler { + try await withCheckedThrowingContinuation { continuation in + waiters.append( + Waiter( + id: requestID, + cost: cost, + durationNanoseconds: durationNanoseconds, + decodedByteCount: decodedByteCount, + continuation: continuation + ) + ) + // Cancellation may reach the actor before this continuation is + // installed. Close that race after insertion so the producer + // cannot remain suspended forever. + if Task.isCancelled { + let cancelled = waiters.removeLast() + cancelled.continuation.resume(throwing: CancellationError()) + if waiters.isEmpty { + isBackpressuring = false + } + } else { + drainWaitersIfReady() + } + } + } onCancel: { + Task { await self.cancelPendingRequest(id: requestID) } + } + + do { + // `reset()` can interleave after a waiter is resumed but before + // this actor method runs again. Never return that stale claim. + guard activeReservations[reservation.id] != nil else { + throw CancellationError() + } + try Task.checkCancellation() + return reservation + } catch { + release(reservation) + throw error + } + } + + /// Releases a completed buffer. Releasing an old or already released + /// reservation is intentionally a no-op. + public func release(_ reservation: AudioPlaybackQueueReservation) { + guard let active = activeReservations.removeValue(forKey: reservation.id) else { + return + } + + scheduledNanoseconds = scheduledNanoseconds >= active.durationNanoseconds + ? scheduledNanoseconds - active.durationNanoseconds + : 0 + scheduledFrameCount = active.cost.frameCount <= scheduledFrameCount + ? scheduledFrameCount - active.cost.frameCount + : 0 + scheduledByteCount = active.decodedByteCount <= scheduledByteCount + ? scheduledByteCount - active.decodedByteCount + : 0 + drainWaitersIfReady() + } + + /// Invalidates active reservations and wakes every suspended producer with + /// `CancellationError`. Intended to be called by playback stop/reset. + public func reset() { + activeReservations.removeAll(keepingCapacity: true) + scheduledNanoseconds = 0 + scheduledFrameCount = 0 + scheduledByteCount = 0 + isBackpressuring = false + + let cancelledWaiters = waiters + waiters.removeAll(keepingCapacity: true) + for waiter in cancelledWaiters { + waiter.continuation.resume(throwing: CancellationError()) + } + } + + public func snapshot() -> AudioPlaybackQueueAdmissionSnapshot { + AudioPlaybackQueueAdmissionSnapshot( + limits: limits, + scheduledDuration: TimeInterval(scheduledNanoseconds) / 1_000_000_000, + scheduledFrameCount: scheduledFrameCount, + scheduledBufferCount: activeReservations.count, + scheduledByteCount: scheduledByteCount, + waitingProducerCount: waiters.count, + isBackpressuring: isBackpressuring + ) + } + + private func cancelPendingRequest(id: UUID) { + guard let index = waiters.firstIndex(where: { $0.id == id }) else { + return + } + + let waiter = waiters.remove(at: index) + waiter.continuation.resume(throwing: CancellationError()) + if waiters.isEmpty { + isBackpressuring = false + } else { + drainWaitersIfReady() + } + } + + private func drainWaitersIfReady() { + guard !waiters.isEmpty else { + isBackpressuring = false + return + } + + if isBackpressuring, scheduledNanoseconds > lowWaterNanoseconds { + return + } + + isBackpressuring = false + while let waiter = waiters.first, + canAdmit( + cost: waiter.cost, + durationNanoseconds: waiter.durationNanoseconds, + decodedByteCount: waiter.decodedByteCount + ) { + waiters.removeFirst() + let reservation = makeReservation( + cost: waiter.cost, + durationNanoseconds: waiter.durationNanoseconds, + decodedByteCount: waiter.decodedByteCount + ) + waiter.continuation.resume(returning: reservation) + } + + if !waiters.isEmpty { + isBackpressuring = true + } + } + + private func canAdmit( + cost: AudioPlaybackQueueCost, + durationNanoseconds: UInt64, + decodedByteCount: Int64 + ) -> Bool { + guard activeReservations.count < limits.maximumScheduledBufferCount, + durationNanoseconds <= highWaterNanoseconds - scheduledNanoseconds, + decodedByteCount <= limits.maximumScheduledBytes - scheduledByteCount, + cost.frameCount <= Int64.max - scheduledFrameCount else { + return false + } + return true + } + + private func makeReservation( + cost: AudioPlaybackQueueCost, + durationNanoseconds: UInt64, + decodedByteCount: Int64 + ) -> AudioPlaybackQueueReservation { + let reservation = AudioPlaybackQueueReservation(id: UUID(), cost: cost) + activeReservations[reservation.id] = ( + cost, + durationNanoseconds, + decodedByteCount + ) + scheduledNanoseconds += durationNanoseconds + scheduledFrameCount += cost.frameCount + scheduledByteCount += decodedByteCount + return reservation + } + + private static func durationNanoseconds( + for cost: AudioPlaybackQueueCost + ) -> UInt64? { + guard cost.frameCount >= 0, + cost.sampleRate.isFinite, + cost.sampleRate > 0 else { + return nil + } + + let duration = TimeInterval(cost.frameCount) / cost.sampleRate + guard duration.isFinite, duration >= 0 else { + return nil + } + return nanoseconds(for: duration, roundingUp: true) + } + + private static func nanoseconds( + for duration: TimeInterval, + roundingUp: Bool = false + ) -> UInt64? { + guard duration.isFinite, duration >= 0 else { return nil } + let scaled = duration * 1_000_000_000 + // `Double(UInt64.max)` rounds up to 2^64. Require a strict bound so + // conversion can never trap at that rounded, unrepresentable endpoint. + guard scaled.isFinite, scaled < TimeInterval(UInt64.max) else { return nil } + return UInt64(roundingUp ? scaled.rounded(.up) : scaled.rounded(.down)) + } +} diff --git a/Packages/ValarAudio/Sources/ValarAudio/PCMCoding.swift b/Packages/ValarAudio/Sources/ValarAudio/PCMCoding.swift index 43f0fde..3ab7f41 100644 --- a/Packages/ValarAudio/Sources/ValarAudio/PCMCoding.swift +++ b/Packages/ValarAudio/Sources/ValarAudio/PCMCoding.swift @@ -33,24 +33,34 @@ enum PCMCoding { /// /// The number of channels written is clamped to `avBuffer.format.channelCount`. /// Channels present in `avBuffer` but absent from `channels` are zero-filled entirely. - static func fill(_ avBuffer: AVAudioPCMBuffer, from channels: [[Float]], frameCount: Int) { + static func fill( + _ avBuffer: AVAudioPCMBuffer, + from channels: [[Float]], + startFrame: Int = 0, + frameCount: Int + ) { precondition( frameCount == Int(avBuffer.frameLength), "PCMCoding.fill frameCount must match avBuffer.frameLength" ) + precondition(startFrame >= 0, "PCMCoding.fill startFrame must be nonnegative") let channelCount = Int(avBuffer.format.channelCount) for channelIndex in 0.. 0 { + destination.update( + from: baseAddress.advanced(by: startFrame), + count: availableCount + ) } } - if source.count < frameCount { - destination.advanced(by: source.count) - .initialize(repeating: 0, count: frameCount - source.count) + if availableCount < frameCount { + destination.advanced(by: availableCount) + .initialize(repeating: 0, count: frameCount - availableCount) } } } diff --git a/Packages/ValarAudio/Sources/ValarAudio/PlaybackPositionPublisher.swift b/Packages/ValarAudio/Sources/ValarAudio/PlaybackPositionPublisher.swift index ad31cad..86908e0 100644 --- a/Packages/ValarAudio/Sources/ValarAudio/PlaybackPositionPublisher.swift +++ b/Packages/ValarAudio/Sources/ValarAudio/PlaybackPositionPublisher.swift @@ -59,7 +59,9 @@ public final class PlaybackPositionPublisher { self.bridge = bridge var capturedContinuation: AsyncStream.Continuation? - self.stream = AsyncStream { continuation in + // Playback position is coalescible UI state, not lossless event data. + // Keep only the latest display-frame value if the consumer falls behind. + self.stream = AsyncStream(bufferingPolicy: .bufferingNewest(1)) { continuation in capturedContinuation = continuation } self.continuation = capturedContinuation diff --git a/Packages/ValarAudio/Sources/ValarAudio/StreamingWAVWriter.swift b/Packages/ValarAudio/Sources/ValarAudio/StreamingWAVWriter.swift index 1f97a04..02c868a 100644 --- a/Packages/ValarAudio/Sources/ValarAudio/StreamingWAVWriter.swift +++ b/Packages/ValarAudio/Sources/ValarAudio/StreamingWAVWriter.swift @@ -1,4 +1,5 @@ import AVFoundation +import Darwin import Foundation /// A streaming WAV writer that accepts audio sample chunks incrementally. @@ -26,6 +27,7 @@ import Foundation /// `StreamingWAVWriter` is an actor. All append and finalize calls are actor-isolated /// and safe to call from concurrent async contexts. public actor StreamingWAVWriter { + private static let framesPerWrite = 16_384 // MARK: - State @@ -47,8 +49,8 @@ public actor StreamingWAVWriter { /// Opens a new WAV file at `url` for streaming writes. /// - /// Intermediate directories are created if needed. Any existing file at `url` - /// is removed before opening. + /// Intermediate directories are created if needed. An existing regular file + /// at `url` is replaced; symbolic links and other entry types are rejected. /// /// - Parameters: /// - url: Destination path for the WAV file. @@ -57,23 +59,38 @@ public actor StreamingWAVWriter { /// - Throws: `AudioPipelineError.exportFailed` if the format is invalid or the /// file cannot be opened. public init(url: URL, sampleRate: Double, channelCount: Int) throws { - guard channelCount > 0, sampleRate > 0 else { + guard (1...64).contains(channelCount), + sampleRate.isFinite, + (1_000...768_000).contains(sampleRate) else { throw AudioPipelineError.exportFailed( "Invalid format: channelCount=\(channelCount) sampleRate=\(sampleRate)" ) } + guard url.isFileURL, !url.lastPathComponent.isEmpty else { + throw AudioPipelineError.exportFailed( + "Streaming WAV output must be a local file URL." + ) + } // Create intermediate directories if needed. + let url = url.standardizedFileURL let directory = url.deletingLastPathComponent() try FileManager.default.createDirectory(at: directory, withIntermediateDirectories: true) + try Self.requireRealDirectory(at: directory) - // Remove any stale file at the target location. - if FileManager.default.fileExists(atPath: url.path) { - do { - try FileManager.default.removeItem(at: url) - } catch let error as CocoaError where error.code == .fileNoSuchFile { - // Another process may have removed it concurrently — ignore. - _ = error + // AVAudioFile writes directly to its destination. Inspect the directory + // entry without following symbolic links before replacing anything. + if try Self.destinationEntryExists(at: url) { + // `unlink` cannot recursively remove a directory if the entry is + // swapped after lstat. It also removes a swapped link itself rather + // than following it. + if Darwin.unlink(url.path) != 0 { + let errorCode = errno + guard errorCode == ENOENT else { + throw AudioPipelineError.exportFailed( + "Unable to replace the streaming WAV destination (errno \(errorCode))." + ) + } } } @@ -126,10 +143,21 @@ public actor StreamingWAVWriter { let frameCount = samples.count / channelCount guard frameCount > 0 else { return } - let pcmBuffer = try makePCMBuffer(frameCount: frameCount) - deinterleave(samples: samples, frameCount: frameCount, channelCount: channelCount, into: pcmBuffer) - try file!.write(from: pcmBuffer) - totalFramesWritten += frameCount + var startFrame = 0 + while startFrame < frameCount { + let writeFrameCount = min(Self.framesPerWrite, frameCount - startFrame) + let pcmBuffer = try makePCMBuffer(frameCount: writeFrameCount) + deinterleave( + samples: samples, + startFrame: startFrame, + frameCount: writeFrameCount, + channelCount: channelCount, + into: pcmBuffer + ) + try file!.write(from: pcmBuffer) + try addWrittenFrames(writeFrameCount) + startFrame += writeFrameCount + } } /// Appends an `AudioPCMBuffer` to the WAV file. @@ -144,10 +172,29 @@ public actor StreamingWAVWriter { try checkOpen() let frameCount = buffer.frameCount - let pcmBuffer = try makePCMBuffer(frameCount: frameCount) - PCMCoding.fill(pcmBuffer, from: buffer.channels, frameCount: frameCount) - try file!.write(from: pcmBuffer) - totalFramesWritten += frameCount + guard buffer.channels.count == descriptor.channelCount, + buffer.channels.allSatisfy({ $0.count == frameCount }), + buffer.format.sampleRate.isFinite, + abs(buffer.format.sampleRate - descriptor.sampleRate) + <= max(0.001, descriptor.sampleRate * 1e-9) else { + throw AudioPipelineError.exportFailed( + "Streaming WAV input must match the writer's sample rate and equal-length channel layout." + ) + } + var startFrame = 0 + while startFrame < frameCount { + let writeFrameCount = min(Self.framesPerWrite, frameCount - startFrame) + let pcmBuffer = try makePCMBuffer(frameCount: writeFrameCount) + PCMCoding.fill( + pcmBuffer, + from: buffer.channels, + startFrame: startFrame, + frameCount: writeFrameCount + ) + try file!.write(from: pcmBuffer) + try addWrittenFrames(writeFrameCount) + startFrame += writeFrameCount + } } // MARK: - Finalize @@ -194,7 +241,9 @@ public actor StreamingWAVWriter { "StreamingWAVWriter: file handle is not open" ) } - guard let pcmBuffer = AVAudioPCMBuffer( + guard frameCount > 0, + frameCount <= Int(AVAudioFrameCount.max), + let pcmBuffer = AVAudioPCMBuffer( pcmFormat: processingFormat, frameCapacity: AVAudioFrameCount(frameCount) ) else { @@ -206,12 +255,25 @@ public actor StreamingWAVWriter { return pcmBuffer } + private func addWrittenFrames(_ frameCount: Int) throws { + let (nextTotal, overflowed) = totalFramesWritten.addingReportingOverflow( + frameCount + ) + guard !overflowed else { + throw AudioPipelineError.exportFailed( + "Streaming WAV frame count exceeds the supported integer range." + ) + } + totalFramesWritten = nextTotal + } + /// De-interleaves interleaved `samples` into the per-channel pointers of `pcmBuffer`. /// /// For mono, this is a single contiguous copy. For multi-channel, each channel /// is extracted from the interleaved layout. private func deinterleave( samples: [Float], + startFrame: Int, frameCount: Int, channelCount: Int, into pcmBuffer: AVAudioPCMBuffer @@ -220,16 +282,53 @@ public actor StreamingWAVWriter { guard let dst = pcmBuffer.floatChannelData?[0] else { return } samples.withUnsafeBufferPointer { src in if let base = src.baseAddress { - dst.update(from: base, count: frameCount) + dst.update( + from: base.advanced(by: startFrame), + count: frameCount + ) } } } else { for ch in 0.. Bool { + var metadata = stat() + if Darwin.lstat(url.path, &metadata) == 0 { + guard metadata.st_mode & S_IFMT == S_IFREG else { + throw AudioPipelineError.exportFailed( + "The streaming WAV destination must be a regular, non-symbolic-link file." + ) + } + return true + } + + let errorCode = errno + guard errorCode == ENOENT else { + throw AudioPipelineError.exportFailed( + "Unable to inspect the streaming WAV destination (errno \(errorCode))." + ) + } + return false + } } diff --git a/Packages/ValarAudio/Sources/ValarAudio/ValarAudio.swift b/Packages/ValarAudio/Sources/ValarAudio/ValarAudio.swift index 4692550..56e1fef 100644 --- a/Packages/ValarAudio/Sources/ValarAudio/ValarAudio.swift +++ b/Packages/ValarAudio/Sources/ValarAudio/ValarAudio.swift @@ -136,6 +136,32 @@ public struct AudioWaveformSummary: Codable, Sendable, Equatable { public protocol AudioDecoder: Sendable { func decode(_ data: Data, hint: String?) async throws -> AudioPCMBuffer + func decode(fileAt url: URL, hint: String?) async throws -> AudioPCMBuffer +} + +extension AudioDecoder { + /// Compatibility fallback for decoders that only accept in-memory bytes. + /// File-aware decoders should override this to avoid an encoded-file copy. + public func decode(fileAt url: URL, hint: String?) async throws -> AudioPCMBuffer { + let maximumEncodedFileBytes = 128 * 1_024 * 1_024 + let values = try url.resourceValues( + forKeys: [.fileSizeKey, .isDirectoryKey, .isRegularFileKey] + ) + guard values.isDirectory != true, + values.isRegularFile == true, + let fileSize = values.fileSize, + fileSize >= 0, + fileSize <= maximumEncodedFileBytes else { + throw AudioPipelineError.decodingFailed( + "Compatibility audio input must be a regular file no larger than \(maximumEncodedFileBytes) bytes" + ) + } + try Task.checkCancellation() + return try await decode( + Data(contentsOf: url, options: .mappedIfSafe), + hint: hint + ) + } } public protocol AudioResampler: Sendable { @@ -257,6 +283,10 @@ enum TemporaryAudioFileSecurity { } public actor AudioPipeline { + /// Compatibility-only full-buffer concatenation ceiling. Long-form callers + /// should use the streaming WAV/M4A sinks instead of duplicating PCM here. + static let maximumConcatenatedPCMBytes = 64 * 1_024 * 1_024 + private let decoder: any AudioDecoder private let resampler: any AudioResampler private let exporter: any AudioExporter @@ -275,6 +305,10 @@ public actor AudioPipeline { try await decoder.decode(data, hint: hint) } + public func decode(fileAt url: URL, hint: String? = nil) async throws -> AudioPCMBuffer { + try await decoder.decode(fileAt: url, hint: hint) + } + public func resample(_ buffer: AudioPCMBuffer, to sampleRate: Double) async throws -> AudioPCMBuffer { try await resampler.resample(buffer, to: sampleRate) } @@ -353,10 +387,49 @@ public actor AudioPipeline { public func concatenate(_ buffers: [AudioPCMBuffer]) -> AudioPCMBuffer? { guard let first = buffers.first else { return nil } - var combined = Array(repeating: [Float](), count: first.channels.count) + guard first.format.sampleRate.isFinite, + first.format.sampleRate > 0, + !first.channels.isEmpty, + first.format.channelCount == first.channels.count, + first.channels.dropFirst().allSatisfy({ + $0.count == first.channels[0].count + }) else { + return nil + } + var channelCapacities = Array(repeating: 0, count: first.channels.count) + + for buffer in buffers { + guard buffer.channels.count == first.channels.count, + buffer.format.channelCount == first.format.channelCount, + buffer.format.sampleFormat == first.format.sampleFormat, + buffer.format.interleaved == first.format.interleaved, + buffer.format.sampleRate.isFinite, + abs(buffer.format.sampleRate - first.format.sampleRate) + <= max(0.001, first.format.sampleRate * 1e-9), + buffer.channels.dropFirst().allSatisfy({ + $0.count == buffer.channels[0].count + }) else { + return nil + } + for index in buffer.channels.indices { + let (capacity, overflowed) = channelCapacities[index] + .addingReportingOverflow(buffer.channels[index].count) + guard !overflowed else { return nil } + channelCapacities[index] = capacity + } + } + guard Self.validatedConcatenatedPCMByteCount( + channelSampleCounts: channelCapacities + ) != nil else { + return nil + } + var combined = channelCapacities.map { capacity -> [Float] in + var channel: [Float] = [] + channel.reserveCapacity(capacity) + return channel + } for buffer in buffers { - guard buffer.channels.count == combined.count else { return nil } for index in buffer.channels.indices { combined[index].append(contentsOf: buffer.channels[index]) } @@ -365,6 +438,26 @@ public actor AudioPipeline { return AudioPCMBuffer(channels: combined, format: first.format) } + static func validatedConcatenatedPCMByteCount( + channelSampleCounts: [Int] + ) -> Int? { + var totalSampleCount = 0 + for sampleCount in channelSampleCounts { + guard sampleCount >= 0 else { return nil } + let (nextTotal, overflowed) = totalSampleCount.addingReportingOverflow(sampleCount) + guard !overflowed else { return nil } + totalSampleCount = nextTotal + } + let (byteCount, overflowed) = totalSampleCount.multipliedReportingOverflow( + by: MemoryLayout.stride + ) + guard !overflowed, + byteCount <= maximumConcatenatedPCMBytes else { + return nil + } + return byteCount + } + public func waveform(for buffer: AudioPCMBuffer, bucketCount: Int = 32) -> AudioWaveformSummary { AudioWaveformAnalyzer.summarize(buffer, bucketCount: bucketCount) } diff --git a/Packages/ValarAudio/Tests/ValarAudioTests/AudioPlaybackQueueAdmissionControllerTests.swift b/Packages/ValarAudio/Tests/ValarAudioTests/AudioPlaybackQueueAdmissionControllerTests.swift new file mode 100644 index 0000000..b4e9d4c --- /dev/null +++ b/Packages/ValarAudio/Tests/ValarAudioTests/AudioPlaybackQueueAdmissionControllerTests.swift @@ -0,0 +1,272 @@ +import XCTest +@testable import ValarAudio + +final class AudioPlaybackQueueAdmissionControllerTests: XCTestCase { + func testBackpressureWaitsUntilLowWaterBeforeAdmittingFIFOHead() async throws { + let controller = try AudioPlaybackQueueAdmissionController( + limits: AudioPlaybackQueueLimits( + lowWaterDuration: 1, + highWaterDuration: 4, + maximumScheduledBufferCount: 4, + maximumWaitingProducerCount: 1 + ) + ) + let twoSeconds = AudioPlaybackQueueCost(frameCount: 48_000, sampleRate: 24_000) + let first = try await controller.acquire(cost: twoSeconds) + let second = try await controller.acquire(cost: twoSeconds) + let blocked = Task { + try await controller.acquire(cost: twoSeconds) + } + + await waitForWaitingProducerCount(1, controller: controller) + await controller.release(first) + + var snapshot = await controller.snapshot() + XCTAssertEqual(snapshot.scheduledDuration, 2, accuracy: 0.000_001) + XCTAssertEqual(snapshot.waitingProducerCount, 1) + XCTAssertTrue(snapshot.isBackpressuring) + + await controller.release(second) + let third = try await blocked.value + + snapshot = await controller.snapshot() + XCTAssertEqual(snapshot.scheduledDuration, 2, accuracy: 0.000_001) + XCTAssertEqual(snapshot.scheduledFrameCount, 48_000) + XCTAssertEqual(snapshot.scheduledBufferCount, 1) + XCTAssertEqual(snapshot.scheduledByteCount, 48_000 * 4) + XCTAssertEqual(snapshot.waitingProducerCount, 0) + XCTAssertFalse(snapshot.isBackpressuring) + await controller.release(third) + } + + func testBufferCountBoundsTinyChunksIndependentlyOfDuration() async throws { + let controller = try AudioPlaybackQueueAdmissionController( + limits: AudioPlaybackQueueLimits( + lowWaterDuration: 50, + highWaterDuration: 100, + maximumScheduledBufferCount: 2, + maximumWaitingProducerCount: 1 + ) + ) + let tiny = AudioPlaybackQueueCost(frameCount: 1, sampleRate: 24_000) + let first = try await controller.acquire(cost: tiny) + let second = try await controller.acquire(cost: tiny) + let blocked = Task { + try await controller.acquire(cost: tiny) + } + + await waitForWaitingProducerCount(1, controller: controller) + var snapshot = await controller.snapshot() + XCTAssertEqual(snapshot.scheduledBufferCount, 2) + + await controller.release(first) + let third = try await blocked.value + snapshot = await controller.snapshot() + XCTAssertEqual(snapshot.scheduledBufferCount, 2) + + await controller.release(second) + await controller.release(third) + } + + func testCancellationRemovesSuspendedProducer() async throws { + let controller = try makeSingleBufferController() + let oneSecond = AudioPlaybackQueueCost(frameCount: 24_000, sampleRate: 24_000) + let active = try await controller.acquire(cost: oneSecond) + let blocked = Task { + try await controller.acquire(cost: oneSecond) + } + + await waitForWaitingProducerCount(1, controller: controller) + blocked.cancel() + + do { + _ = try await blocked.value + XCTFail("A cancelled producer must not receive a reservation") + } catch is CancellationError { + // Expected. + } + + let snapshot = await controller.snapshot() + XCTAssertEqual(snapshot.waitingProducerCount, 0) + XCTAssertFalse(snapshot.isBackpressuring) + await controller.release(active) + } + + func testResetWakesWaiterAndInvalidatesExistingReservations() async throws { + let controller = try makeSingleBufferController() + let oneSecond = AudioPlaybackQueueCost(frameCount: 24_000, sampleRate: 24_000) + let active = try await controller.acquire(cost: oneSecond) + let blocked = Task { + try await controller.acquire(cost: oneSecond) + } + + await waitForWaitingProducerCount(1, controller: controller) + await controller.reset() + + do { + _ = try await blocked.value + XCTFail("Reset must wake a suspended producer with cancellation") + } catch is CancellationError { + // Expected. + } + + await controller.release(active) + let snapshot = await controller.snapshot() + XCTAssertEqual(snapshot.scheduledDuration, 0) + XCTAssertEqual(snapshot.scheduledFrameCount, 0) + XCTAssertEqual(snapshot.scheduledBufferCount, 0) + XCTAssertEqual(snapshot.scheduledByteCount, 0) + XCTAssertEqual(snapshot.waitingProducerCount, 0) + } + + func testWaitingProducerLimitPreventsUnboundedSuspendedPCM() async throws { + let controller = try makeSingleBufferController() + let oneSecond = AudioPlaybackQueueCost(frameCount: 24_000, sampleRate: 24_000) + let active = try await controller.acquire(cost: oneSecond) + let firstBlocked = Task { + try await controller.acquire(cost: oneSecond) + } + + await waitForWaitingProducerCount(1, controller: controller) + + do { + _ = try await controller.acquire(cost: oneSecond) + XCTFail("A second suspended producer should be rejected") + } catch let error as AudioPlaybackQueueAdmissionError { + XCTAssertEqual(error, .tooManyWaitingProducers(limit: 1)) + } + + firstBlocked.cancel() + _ = try? await firstBlocked.value + await controller.release(active) + } + + func testOversizedAndInvalidBuffersFailBeforeWaiting() async throws { + let limits = AudioPlaybackQueueLimits( + lowWaterDuration: 1, + highWaterDuration: 2, + maximumScheduledBufferCount: 2 + ) + let controller = try AudioPlaybackQueueAdmissionController(limits: limits) + let oversized = AudioPlaybackQueueCost(frameCount: 48_001, sampleRate: 24_000) + + do { + _ = try await controller.acquire(cost: oversized) + XCTFail("An oversized buffer should require caller-side splitting") + } catch let error as AudioPlaybackQueueAdmissionError { + XCTAssertEqual(error, .bufferExceedsHighWater(oversized, limits)) + } + + let invalid = AudioPlaybackQueueCost(frameCount: 1, sampleRate: .nan) + do { + _ = try await controller.acquire(cost: invalid) + XCTFail("A non-finite sample rate must be rejected") + } catch let error as AudioPlaybackQueueAdmissionError { + guard case .invalidCost(let rejectedCost) = error else { + XCTFail("Expected invalidCost, received \(error)") + return + } + XCTAssertEqual(rejectedCost.frameCount, invalid.frameCount) + XCTAssertTrue(rejectedCost.sampleRate.isNaN) + } + } + + func testDecodedByteBudgetBackpressuresAndRejectsOversizedBuffers() async throws { + let limits = AudioPlaybackQueueLimits( + lowWaterDuration: 0.1, + highWaterDuration: 10, + maximumScheduledBufferCount: 2, + maximumWaitingProducerCount: 1, + maximumScheduledBytes: 100 + ) + let controller = try AudioPlaybackQueueAdmissionController(limits: limits) + let eightyBytes = AudioPlaybackQueueCost( + frameCount: 10, + sampleRate: 10, + channelCount: 2, + bytesPerSample: 4 + ) + let twentyFourBytes = AudioPlaybackQueueCost( + frameCount: 6, + sampleRate: 10, + channelCount: 1, + bytesPerSample: 4 + ) + let tooLarge = AudioPlaybackQueueCost( + frameCount: 13, + sampleRate: 10, + channelCount: 2, + bytesPerSample: 4 + ) + + do { + _ = try await controller.acquire(cost: tooLarge) + XCTFail("A single decoded buffer larger than the byte budget must fail") + } catch let error as AudioPlaybackQueueAdmissionError { + XCTAssertEqual(error, .bufferExceedsByteLimit(tooLarge, limits)) + } + + let first = try await controller.acquire(cost: eightyBytes) + let blocked = Task { + try await controller.acquire(cost: twentyFourBytes) + } + await waitForWaitingProducerCount(1, controller: controller) + var snapshot = await controller.snapshot() + XCTAssertEqual(snapshot.scheduledByteCount, 80) + + await controller.release(first) + let second = try await blocked.value + snapshot = await controller.snapshot() + XCTAssertEqual(snapshot.scheduledByteCount, 24) + await controller.release(second) + } + + func testInvalidLimitsFailClosed() { + let invalid = AudioPlaybackQueueLimits( + lowWaterDuration: 4, + highWaterDuration: 4, + maximumScheduledBufferCount: 0, + maximumWaitingProducerCount: 0 + ) + + XCTAssertThrowsError( + try AudioPlaybackQueueAdmissionController(limits: invalid) + ) { error in + XCTAssertEqual( + error as? AudioPlaybackQueueAdmissionError, + .invalidLimits(invalid) + ) + } + } + + private func makeSingleBufferController() throws -> AudioPlaybackQueueAdmissionController { + try AudioPlaybackQueueAdmissionController( + limits: AudioPlaybackQueueLimits( + lowWaterDuration: 0.25, + highWaterDuration: 1, + maximumScheduledBufferCount: 1, + maximumWaitingProducerCount: 1 + ) + ) + } + + private func waitForWaitingProducerCount( + _ expectedCount: Int, + controller: AudioPlaybackQueueAdmissionController, + file: StaticString = #filePath, + line: UInt = #line + ) async { + for _ in 0 ..< 100 { + if await controller.snapshot().waitingProducerCount == expectedCount { + return + } + await Task.yield() + } + + XCTFail( + "Timed out waiting for \(expectedCount) queued playback producer(s)", + file: file, + line: line + ) + } +} diff --git a/Packages/ValarAudio/Tests/ValarAudioTests/ValarAudioTests.swift b/Packages/ValarAudio/Tests/ValarAudioTests/ValarAudioTests.swift index 0816e97..54b2fed 100644 --- a/Packages/ValarAudio/Tests/ValarAudioTests/ValarAudioTests.swift +++ b/Packages/ValarAudio/Tests/ValarAudioTests/ValarAudioTests.swift @@ -1,10 +1,30 @@ import AVFoundation +import Darwin import Foundation import XCTest @testable import ValarAudio import Dispatch final class ValarAudioTests: XCTestCase { + func testStreamingM4AMarkerUsesCumulativeFrameOffset() throws { + let marker = try XCTUnwrap(StreamingM4AWriter.chapterMarker( + title: " Chapter Two ", + startFrame: 24_000, + frameCount: 12_000, + sampleRate: 24_000 + )) + + XCTAssertEqual(marker.title, "Chapter Two") + XCTAssertEqual(marker.startTime, 1, accuracy: 0.000_001) + XCTAssertEqual(marker.duration, 0.5, accuracy: 0.000_001) + XCTAssertNil(StreamingM4AWriter.chapterMarker( + title: " ", + startFrame: 0, + frameCount: 1, + sampleRate: 24_000 + )) + } + func testExportProducesValidData() async throws { let buffer = AudioPCMBuffer(mono: [0, 0.25, 0.5, 1.0], sampleRate: 24_000, container: "wav") let exporter = AVFoundationAudioExporter() @@ -54,6 +74,33 @@ final class ValarAudioTests: XCTestCase { XCTAssertEqual(decoded.format.channelCount, 1) } + func testFileURLDecodeAvoidsEncodedDataAPI() async throws { + let buffer = AudioPCMBuffer( + mono: [0, 0.25, 0.5, 0.75], + sampleRate: 24_000 + ) + let url = makeTemporaryURL(pathExtension: "wav") + defer { try? FileManager.default.removeItem(at: url) } + _ = try await AVFoundationAudioExporter().export( + buffer, + as: AudioFormatDescriptor( + sampleRate: 24_000, + channelCount: 1, + container: "wav" + ), + to: url, + chapterMarkers: [] + ) + + let decoded = try await AVFoundationAudioDecoder().decode( + fileAt: url, + hint: "wav" + ) + + XCTAssertEqual(decoded.frameCount, buffer.frameCount) + XCTAssertEqual(decoded.format.sampleRate, buffer.format.sampleRate) + } + func testM4AExportProducesAACFile() async throws { let buffer = AudioPCMBuffer( mono: [Float](repeating: 0.1, count: 24_000), @@ -142,17 +189,26 @@ final class ValarAudioTests: XCTestCase { XCTAssertEqual(result.channels[0], expected) } - func testResamplerDownsampleIntegerRatioMatchesNaiveLoopExactly() async throws { - let samples = (0..<24).map { index in - Float(index % 7) / 6 - 0.5 + func testResamplerDownsampleIntegerRatioRejectsAboveNyquistTone() async throws { + let sourceRate = 48_000.0 + let targetRate = 24_000.0 + let samples = (0..<48_000).map { index in + Float(sin((2 * Double.pi * 18_000 * Double(index)) / sourceRate)) } - let buffer = AudioPCMBuffer(mono: samples, sampleRate: 48_000) + let buffer = AudioPCMBuffer(mono: samples, sampleRate: sourceRate) let resampler = AccelerateAudioResampler() - let result = try await resampler.resample(buffer, to: 24_000) - let expected = naiveResample(samples, sourceRate: 48_000, targetRate: 24_000) + let result = try await resampler.resample(buffer, to: targetRate) + let output = result.channels[0] + let edgeFrames = 128 + let settledOutput = output.dropFirst(edgeFrames).dropLast(edgeFrames) + let rms = sqrt( + settledOutput.reduce(Float.zero) { $0 + ($1 * $1) } + / Float(settledOutput.count) + ) - XCTAssertEqual(result.channels[0], expected) + XCTAssertEqual(result.frameCount, 24_000) + XCTAssertLessThan(rms, 0.05, "Above-Nyquist energy should be filtered before exact-ratio decimation") } func testResamplerHandlesNonIntegerRatioShortBuffer() async throws { @@ -363,6 +419,220 @@ final class ValarAudioTests: XCTestCase { XCTAssertEqual(combined?.frameCount, 5) } + func testConcatenateRejectsMismatchedSampleRates() async { + let first = AudioPCMBuffer(mono: [0, 1], sampleRate: 24_000) + let second = AudioPCMBuffer(mono: [2, 3], sampleRate: 48_000) + let pipeline = AudioPipeline() + + let combined = await pipeline.concatenate([first, second]) + + XCTAssertNil(combined) + } + + func testConcatenateRejectsUnequalPlanarChannelLengths() async { + let malformed = AudioPCMBuffer( + channels: [[0, 1], [0]], + format: AudioFormatDescriptor(sampleRate: 24_000, channelCount: 2) + ) + let pipeline = AudioPipeline() + + let combined = await pipeline.concatenate([malformed]) + + XCTAssertNil(combined) + } + + func testConcatenateByteCeilingIsCheckedWithoutAllocatingPCM() { + let maximumSampleCount = + AudioPipeline.maximumConcatenatedPCMBytes / MemoryLayout.stride + + XCTAssertEqual( + AudioPipeline.validatedConcatenatedPCMByteCount( + channelSampleCounts: [maximumSampleCount] + ), + AudioPipeline.maximumConcatenatedPCMBytes + ) + XCTAssertNil( + AudioPipeline.validatedConcatenatedPCMByteCount( + channelSampleCounts: [maximumSampleCount, 1] + ) + ) + XCTAssertNil( + AudioPipeline.validatedConcatenatedPCMByteCount( + channelSampleCounts: [Int.max, 1] + ) + ) + } + + func testDecoderRejectsOversizedDecodedPCMBeforeAllocation() { + let maximumFrames = + AVFoundationAudioDecoder.maximumDecodedPCMBytes / MemoryLayout.size + + XCTAssertNoThrow(try AVFoundationAudioDecoder.validatedDecodedPCMByteCount( + frameCount: Int64(maximumFrames), + channelCount: 1 + )) + XCTAssertThrowsError(try AVFoundationAudioDecoder.validatedDecodedPCMByteCount( + frameCount: Int64(maximumFrames) + 1, + channelCount: 1 + )) + XCTAssertThrowsError(try AVFoundationAudioDecoder.validatedDecodedPCMByteCount( + frameCount: .max, + channelCount: 64 + )) + } + + func testExporterRejectsFormatThatWouldReadMissingChannels() async { + let exporter = AVFoundationAudioExporter() + let mono = AudioPCMBuffer(mono: [0, 0.25], sampleRate: 24_000) + + do { + _ = try await exporter.export( + mono, + as: AudioFormatDescriptor( + sampleRate: 24_000, + channelCount: 2, + container: "wav" + ) + ) + XCTFail("Expected channel mismatch rejection") + } catch { + XCTAssertTrue(error is AudioPipelineError) + } + } + + func testFailedExportPreservesExistingDestination() async throws { + let exporter = AVFoundationAudioExporter() + let destinationURL = makeTemporaryURL(pathExtension: "wav") + let original = Data("existing-audio".utf8) + try original.write(to: destinationURL) + defer { try? FileManager.default.removeItem(at: destinationURL) } + + do { + _ = try await exporter.export( + AudioPCMBuffer(mono: [0, 0.25], sampleRate: 24_000), + as: AudioFormatDescriptor( + sampleRate: 24_000, + channelCount: 1, + container: "unsupported" + ), + to: destinationURL, + chapterMarkers: [] + ) + XCTFail("Expected unsupported export format") + } catch { + XCTAssertEqual(try Data(contentsOf: destinationURL), original) + } + } + + func testWaveExportReplacesExistingRegularFileAfterSuccess() async throws { + let exporter = AVFoundationAudioExporter() + let destinationURL = makeTemporaryURL(pathExtension: "wav") + try Data("existing-audio".utf8).write(to: destinationURL) + defer { try? FileManager.default.removeItem(at: destinationURL) } + + _ = try await exporter.export( + AudioPCMBuffer(mono: [0, 0.25], sampleRate: 24_000), + as: AudioFormatDescriptor( + sampleRate: 24_000, + channelCount: 1, + container: "wav" + ), + to: destinationURL, + chapterMarkers: [] + ) + + let data = try Data(contentsOf: destinationURL) + XCTAssertEqual(String(decoding: data.prefix(4), as: UTF8.self), "RIFF") + } + + func testExportRejectsSymbolicLinkDestination() async throws { + let root = FileManager.default.temporaryDirectory + .appendingPathComponent(UUID().uuidString, isDirectory: true) + let targetURL = root.appendingPathComponent("target.wav") + let linkURL = root.appendingPathComponent("link.wav") + try FileManager.default.createDirectory(at: root, withIntermediateDirectories: true) + try Data("do-not-replace".utf8).write(to: targetURL) + try FileManager.default.createSymbolicLink(at: linkURL, withDestinationURL: targetURL) + defer { try? FileManager.default.removeItem(at: root) } + + do { + _ = try await AVFoundationAudioExporter().export( + AudioPCMBuffer(mono: [0, 0.25], sampleRate: 24_000), + as: AudioFormatDescriptor( + sampleRate: 24_000, + channelCount: 1, + container: "wav" + ), + to: linkURL, + chapterMarkers: [] + ) + XCTFail("Expected symbolic-link destination rejection") + } catch { + XCTAssertEqual( + try Data(contentsOf: targetURL), + Data("do-not-replace".utf8) + ) + } + } + + func testStreamingWriterRejectsSymbolicLinkDestination() throws { + let root = FileManager.default.temporaryDirectory + .appendingPathComponent(UUID().uuidString, isDirectory: true) + let targetURL = root.appendingPathComponent("target.wav") + let linkURL = root.appendingPathComponent("link.wav") + try FileManager.default.createDirectory(at: root, withIntermediateDirectories: true) + try Data("keep-target".utf8).write(to: targetURL) + try FileManager.default.createSymbolicLink( + at: linkURL, + withDestinationURL: targetURL + ) + defer { try? FileManager.default.removeItem(at: root) } + + XCTAssertThrowsError( + try StreamingWAVWriter( + url: linkURL, + sampleRate: 24_000, + channelCount: 1 + ) + ) + XCTAssertEqual(try Data(contentsOf: targetURL), Data("keep-target".utf8)) + } + + func testStreamingWriterRejectsDirectoryDestination() throws { + let destinationURL = FileManager.default.temporaryDirectory + .appendingPathComponent(UUID().uuidString, isDirectory: true) + try FileManager.default.createDirectory( + at: destinationURL, + withIntermediateDirectories: true + ) + defer { try? FileManager.default.removeItem(at: destinationURL) } + + XCTAssertThrowsError( + try StreamingWAVWriter( + url: destinationURL, + sampleRate: 24_000, + channelCount: 1 + ) + ) + } + + func testStreamingWriterRejectsNonRegularDestination() throws { + let root = FileManager.default.temporaryDirectory + .appendingPathComponent(UUID().uuidString, isDirectory: true) + let fifoURL = root.appendingPathComponent("audio.fifo") + try FileManager.default.createDirectory(at: root, withIntermediateDirectories: true) + XCTAssertEqual(Darwin.mkfifo(fifoURL.path, mode_t(0o600)), 0) + defer { try? FileManager.default.removeItem(at: root) } + + XCTAssertThrowsError( + try StreamingWAVWriter( + url: fifoURL, + sampleRate: 24_000, + channelCount: 1 + ) + ) + } + func testWaveformSummary() async { let buffer = AudioPCMBuffer(mono: [0, 0.5, -0.5, 1.0, -1.0], sampleRate: 24_000) let pipeline = AudioPipeline() @@ -565,6 +835,37 @@ final class ValarAudioTests: XCTestCase { } } + func testDecodePCMFloat32LERejectsInvalidFormatMetadata() async throws { + let decoder = AVFoundationAudioDecoder() + let data = makePCMFloat32LEData([0, 0.1]) + + for hint in [ + "pcm_f32le:nan:1", + "pcm_f32le:999:1", + "pcm_f32le:24000:0", + "pcm_f32le:24000:65", + ] { + do { + _ = try await decoder.decode(data, hint: hint) + XCTFail("Expected invalid PCM hint rejection for \(hint)") + } catch AudioPipelineError.decodingFailed { + // Expected. + } + } + } + + func testDecodePCMFloat32LERejectsIncompleteInterleavedFrame() async throws { + let decoder = AVFoundationAudioDecoder() + let data = makePCMFloat32LEData([0, 0.1, 0.2]) + + do { + _ = try await decoder.decode(data, hint: "pcm_f32le:24000:2") + XCTFail("Expected incomplete stereo frame rejection") + } catch AudioPipelineError.decodingFailed { + // Expected. + } + } + func testDecodePCMFloat32LERoundTripsAgainstMLXEncoding() async throws { // Mirror the encoding MLXModelHandle.pcmFloat32LEData uses: withUnsafeBufferPointer + Data(bytes:count:) let original: [Float] = [0, 0.1, -0.3, 0.7, -1.0, 1.0] @@ -734,6 +1035,26 @@ final class ValarAudioTests: XCTestCase { final class AudioEnginePlayerFeedSamplesTests: XCTestCase { + func testTemporaryQueueDrainWaitsForMoreAudio() { + XCTAssertEqual( + AudioEnginePlayer.queueDrainDisposition( + finishedStreaming: false, + totalEnqueuedFrames: 2_400 + ), + .awaitMoreAudio + ) + } + + func testFinishedQueueDrainStopsAfterPlayback() { + XCTAssertEqual( + AudioEnginePlayer.queueDrainDisposition( + finishedStreaming: true, + totalEnqueuedFrames: 2_400 + ), + .finishPlayback(didPlayAudio: true) + ) + } + /// `makeAVAudioBufferFromSamples` must copy Float values into `floatChannelData` /// byte-for-byte, with no intermediate Data conversion. func testMakeAVAudioBufferFromSamplesMatchesInput() throws { diff --git a/Packages/mlx-audio-swift-valar/Sources/MLXAudioTTS/Models/VibeVoice/VibeVoiceTTSModel.swift b/Packages/mlx-audio-swift-valar/Sources/MLXAudioTTS/Models/VibeVoice/VibeVoiceTTSModel.swift index dca4fe4..8878d9c 100644 --- a/Packages/mlx-audio-swift-valar/Sources/MLXAudioTTS/Models/VibeVoice/VibeVoiceTTSModel.swift +++ b/Packages/mlx-audio-swift-valar/Sources/MLXAudioTTS/Models/VibeVoice/VibeVoiceTTSModel.swift @@ -103,6 +103,13 @@ public final class VibeVoiceTTSModel: Module, SpeechGenerationModel, @unchecked return hash } + /// Retain only the condition consumed by the next diffusion step. Historical + /// transformer state already lives in the corresponding KV cache, so keeping + /// every hidden output here would duplicate an ever-growing sequence. + static func latestConditionState(_ hidden: MLXArray) -> MLXArray { + hidden[0..., (-1)..., 0...] + } + // MARK: - Init init(config: VibeVoiceModelConfig) { @@ -381,25 +388,16 @@ public final class VibeVoiceTTSModel: Module, SpeechGenerationModel, @unchecked let uncondEps = eps[batchSize...] let guidedEps = uncondEps + MLXArray(cfgScale) * (condEps - uncondEps) - // Duplicate for scheduler - let fullEps = MLX.concatenated([guidedEps, guidedEps], axis: 0) - let fullSpeech = MLX.concatenated([speech, speech], axis: 0) - // Scheduler step let output = noiseScheduler.step( - modelOutput: fullEps, + modelOutput: guidedEps, timestep: tVal, - sample: fullSpeech, + sample: speech, prevX0: prevX0 ) - // Extract positive conditioning result - speech = output.prevSample[..