From c0c282ca3ab4edbd7ef07d8dd583690156843074 Mon Sep 17 00:00:00 2001 From: evgeny Date: Wed, 28 Oct 2026 10:32:38 +0100 Subject: [PATCH] Make voice recording ownership and playback failures explicit --- .../utun/chat/data/AudioRecorderManager.kt | 335 ++++++++++-------- .../utun/chat/data/RecorderHostEnvironment.kt | 5 +- .../chat/data/AudioRecorderManagerTest.kt | 29 ++ 3 files changed, 213 insertions(+), 156 deletions(-) diff --git a/tools/chatgui-android/app/src/main/java/com/utun/chat/data/AudioRecorderManager.kt b/tools/chatgui-android/app/src/main/java/com/utun/chat/data/AudioRecorderManager.kt index 29e46d8a..41035d7a 100644 --- a/tools/chatgui-android/app/src/main/java/com/utun/chat/data/AudioRecorderManager.kt +++ b/tools/chatgui-android/app/src/main/java/com/utun/chat/data/AudioRecorderManager.kt @@ -10,6 +10,16 @@ import android.media.MediaRecorder import androidx.core.content.ContextCompat import com.utun.chat.ChatApplication +/** + * Запись голосового сообщения/тест микрофона и отдельный плеер native Opus-файла. + * Запись и тест делят один native recorder: успешный voiceStart даёт владение до stop/cancel. + * Lifecycle сериализован монитором менеджера; stopRecording/cancelRecording вызываются также из IO worker. + * Захват — PCM16 48 кГц mono, чтение без блокировки; stop сначала останавливает reader, ждёт join и освобождает Android. + * Ошибка захвата/очистки запрещает публикацию частичной записи: stopRecording вернёт -1 и отменит native recorder. + * После отказа worker владелец вызывает stop/cancel; неосвобождённый Android device сохраняется для повтора cleanup. + * Плеер владеет своим decoder/AudioTrack в worker; прогресс/завершение — быстрые callbacks этого потока. + * Pause/stop меняют поколение и прерывают worker, старое поколение больше не уведомляет UI. + */ class AudioRecorderManager { companion object { const val SAMPLE_RATE = 48000 @@ -19,14 +29,13 @@ class AudioRecorderManager { const val MIN_DURATION_MS = 500 } + private enum class CaptureMode { RECORD, TEST } + @Volatile private var captureMode: CaptureMode? = null private var audioRecord: AudioRecord? = null private var recordingThread: Thread? = null - @Volatile private var isRecording = false - - private var testRecord: AudioRecord? = null - private var testThread: Thread? = null - @Volatile var isTesting = false - private set + @Volatile private var captureRunning = false + @Volatile private var captureFailed = false + val isTesting: Boolean get() = captureMode == CaptureMode.TEST && captureRunning fun hasPermission(): Boolean { val ctx = ChatApplication.instance @@ -34,167 +43,160 @@ class AudioRecorderManager { PackageManager.PERMISSION_GRANTED } - fun startRecording(channelId: String, replyTo: String = "", requestId: Long = 0, generation: Long = 0): Boolean { - if (isRecording) return false - if (!hasPermission()) return false + /** target — канал либо dm:conv_id; reply/request относятся к каналу, generation — к PM. + * true — Android/native захват запущен; false — отказ с диагностикой и откатом принадлежащих ресурсов. */ + @Synchronized fun startRecording(channelId: String, replyTo: String = "", requestId: Long = 0, generation: Long = 0): Boolean = + startCapture(CaptureMode.RECORD, channelId, replyTo, requestId, generation) - val ok = NativeLib.voiceStart(channelId, replyTo, requestId, generation) - if (!ok) { - LogManager.addLog("ERROR", "AudioRecorder", "native voiceStart failed") + private fun startCapture(mode: CaptureMode, target: String, reply: String = "", request: Long = 0, generation: Long = 0): Boolean { + if (captureMode != null) { + LogManager.addLog("WARN", "AudioRecorder", "capture start rejected mode=$mode owned=$captureMode running=$captureRunning") return false } - - val bufferSize = AudioRecord.getMinBufferSize( - SAMPLE_RATE, - AudioFormat.CHANNEL_IN_MONO, - AudioFormat.ENCODING_PCM_16BIT - ).coerceAtLeast(FRAME_SAMPLES * 2) - - audioRecord = try { - AudioRecord( - MediaRecorder.AudioSource.MIC, - SAMPLE_RATE, - AudioFormat.CHANNEL_IN_MONO, - AudioFormat.ENCODING_PCM_16BIT, - bufferSize * 2 - ) - } catch (e: Exception) { - LogManager.addLog("ERROR", "AudioRecorder", "AudioRecord create failed: ${e.message}") - NativeLib.voiceCancel() - return false - } - - if (audioRecord?.state != AudioRecord.STATE_INITIALIZED) { - LogManager.addLog("ERROR", "AudioRecorder", "AudioRecord not initialized") - audioRecord?.release() - audioRecord = null - NativeLib.voiceCancel() - return false - } - - isRecording = true - LogManager.addLog("INFO", "AudioRecorder", "started ch=$channelId bufferSize=$bufferSize") - audioRecord?.startRecording() - val device = audioRecord ?: return false - - recordingThread = Thread { - val buffer = ShortArray(FRAME_SAMPLES) - var totalRead = 0L - - while (isRecording && device.recordingState == AudioRecord.RECORDSTATE_RECORDING) { - val read = try { device.read(buffer, 0, FRAME_SAMPLES) } catch (e: Exception) { - if (isRecording) LogManager.addLog("ERROR", "AudioRecorder", "capture read failed: ${e.message}") - break - } - if (read <= 0) { - if (isRecording) LogManager.addLog("ERROR", "AudioRecorder", "capture read failed code=$read") - break - } - NativeLib.voiceFeed(buffer.copyOf(read)) - totalRead += read + if (!releaseCaptureDevice()) return false + try { + if (!hasPermission()) { + LogManager.addLog("WARN", "AudioRecorder", "microphone permission missing mode=$mode") + return false } - LogManager.addLog("DEBUG", "AudioRecorder", "thread done totalSamples=$totalRead") - }.apply { - priority = Thread.MAX_PRIORITY - start() + if (!NativeLib.voiceStart(target, reply, request, generation)) { + LogManager.addLog("ERROR", "AudioRecorder", "native voiceStart rejected mode=$mode target=$target") + return false + } + captureMode = mode; captureFailed = false + val minimum = AudioRecord.getMinBufferSize(SAMPLE_RATE, AudioFormat.CHANNEL_IN_MONO, AudioFormat.ENCODING_PCM_16BIT) + check(minimum > 0) { "AudioRecord minimum=$minimum" } + val bytes = minimum.coerceAtLeast(FRAME_SAMPLES * 2) * 2 + val device = AudioRecord(MediaRecorder.AudioSource.MIC, SAMPLE_RATE, AudioFormat.CHANNEL_IN_MONO, + AudioFormat.ENCODING_PCM_16BIT, bytes) + audioRecord = device + check(device.state == AudioRecord.STATE_INITIALIZED) { "AudioRecord not initialized" } + device.startRecording() + check(device.recordingState == AudioRecord.RECORDSTATE_RECORDING) { "AudioRecord did not start" } + captureRunning = true + recordingThread = Thread({ captureLoop(device, mode) }, "utun-voice-capture").apply { priority = Thread.MAX_PRIORITY } + recordingThread!!.start() + LogManager.addLog("INFO", "AudioRecorder", "capture started mode=$mode target=$target bufferBytes=$bytes") + return true + } catch (e: Exception) { + LogManager.addLog("ERROR", "AudioRecorder", "capture start failed mode=$mode target=$target: $e") + if (captureMode != null) { captureFailed = true; finishCapture(false) } + return false } - - return true } - fun stopRecording(): Int { - if (!isRecording) return 0 - isRecording = false - - audioRecord?.apply { - try { stop() } catch (e: Exception) { LogManager.addLog("WARN", "AudioRecorder", "capture stop failed: ${e.message}") } + private fun captureLoop(device: AudioRecord, mode: CaptureMode) { + val buffer = ShortArray(FRAME_SAMPLES) + var total = 0L + try { + while (captureRunning) { + check(device.recordingState == AudioRecord.RECORDSTATE_RECORDING) { "AudioRecord stopped unexpectedly" } + val read = device.read(buffer, 0, buffer.size, AudioRecord.READ_NON_BLOCKING) + if (!captureRunning) break + check(read in 0..buffer.size) { "AudioRecord.read=$read" } + if (read == 0) { Thread.sleep(2); continue } + if (!NativeLib.voiceFeed(buffer.copyOf(read))) { + captureFailed = true + error("native voiceFeed rejected samples=$read") + } + total += read + } + } catch (e: Exception) { + if (captureRunning || captureFailed) { + captureFailed = true + LogManager.addLog("ERROR", "AudioRecorder", "capture failed mode=$mode samples=$total: $e") + } + } finally { + captureRunning = false + LogManager.addLog("DEBUG", "AudioRecorder", "capture worker exited mode=$mode samples=$total failed=$captureFailed") } - recordingThread?.join() - recordingThread = null - audioRecord?.release() - audioRecord = null - - val duration = NativeLib.voiceStop() - LogManager.addLog("INFO", "AudioRecorder", "stopped duration=${duration}ms") - return duration } - fun cancelRecording() { - if (!isRecording) return - isRecording = false - - audioRecord?.apply { - try { stop() } catch (e: Exception) { LogManager.addLog("WARN", "AudioRecorder", "capture cancel failed: ${e.message}") } + /** Под монитором lifecycle: native освобождается только после join и попытки Android release. */ + private fun finishCapture(publish: Boolean): Int { + val mode = captureMode ?: return 0 + captureRunning = false + recordingThread?.interrupt() + var failed = captureFailed + var interrupted = Thread.interrupted() + try { + try { audioRecord?.stop() } + catch (e: Exception) { failed = true; LogManager.addLog("WARN", "AudioRecorder", "capture stop failed mode=$mode: $e") } + while (recordingThread?.isAlive == true) { + try { recordingThread?.join() } + catch (e: InterruptedException) { + interrupted = true + LogManager.addLog("WARN", "AudioRecorder", "capture join interrupted mode=$mode; waiting for worker") + } + } + recordingThread = null + failed = failed || captureFailed + if (!releaseCaptureDevice()) failed = true + val duration = if (publish && !failed) NativeLib.voiceStop() else { NativeLib.voiceCancel(); if (publish) -1 else 0 } + LogManager.addLog(if (duration < 0) "ERROR" else "INFO", "AudioRecorder", + "capture finished mode=$mode publish=$publish failed=$failed durationMs=$duration") + return duration + } catch (e: Exception) { + LogManager.addLog("ERROR", "AudioRecorder", "native capture finish failed mode=$mode: $e") + try { NativeLib.voiceCancel() } + catch (cleanup: Exception) { LogManager.addLog("ERROR", "AudioRecorder", "native capture cancel failed: $cleanup") } + return -1 + } finally { + captureMode = null + if (interrupted) Thread.currentThread().interrupt() } - recordingThread?.join() - recordingThread = null - audioRecord?.release() - audioRecord = null - - NativeLib.voiceCancel() - LogManager.addLog("INFO", "AudioRecorder", "cancelled") } - fun isActive(): Boolean = isRecording + /** Под монитором lifecycle; отказ release сохраняет владение для следующего cleanup/start. */ + private fun releaseCaptureDevice(): Boolean { + val device = audioRecord ?: return true + return try { + device.release(); audioRecord = null + LogManager.addLog("DEBUG", "AudioRecorder", "capture device released mode=$captureMode") + true + } catch (e: Exception) { + LogManager.addLog("WARN", "AudioRecorder", "capture release failed mode=$captureMode; retained for retry: $e") + false + } + } - fun startMicTest(): Boolean { - if (StreamController.blocksCapture()) return false + /** Длительность native в мс, -1 — отказ/запись отброшена после ошибки, 0 — нет записи. + * Успех означает передачу подготовки в native, публикация сообщения подтверждается отдельно событиями. */ + @Synchronized fun stopRecording(): Int { + if (captureMode == CaptureMode.RECORD) return finishCapture(true) + if (captureMode == null) releaseCaptureDevice() + return 0 + } + @Synchronized fun cancelRecording() { + if (captureMode == CaptureMode.RECORD) finishCapture(false) else if (captureMode == null) releaseCaptureDevice() + } + fun isActive(): Boolean = captureMode == CaptureMode.RECORD && captureRunning + + /** Индикатор уровня через native recorder; успешный тест нужно завершить stopMicTest без публикации. */ + @Synchronized fun startMicTest(): Boolean { + if (StreamController.blocksCapture()) { + LogManager.addLog("WARN", "AudioRecorder", "microphone test rejected: stream owns capture") + return false + } if (isTesting) return true - if (!hasPermission()) return false - - NativeLib.voiceStart("__test__") - val bufferSize = AudioRecord.getMinBufferSize( - SAMPLE_RATE, - AudioFormat.CHANNEL_IN_MONO, - AudioFormat.ENCODING_PCM_16BIT - ).coerceAtLeast(FRAME_SAMPLES * 2) - - testRecord = try { - AudioRecord( - MediaRecorder.AudioSource.MIC, - SAMPLE_RATE, - AudioFormat.CHANNEL_IN_MONO, - AudioFormat.ENCODING_PCM_16BIT, - bufferSize * 2 - ) - } catch (e: Exception) { - NativeLib.voiceCancel(); return false - } - if (testRecord?.state != AudioRecord.STATE_INITIALIZED) { - testRecord?.release(); testRecord = null; NativeLib.voiceCancel(); return false - } - - isTesting = true - testRecord?.startRecording() - - testThread = Thread { - val buffer = ShortArray(FRAME_SAMPLES) - while (isTesting && testRecord?.recordingState == AudioRecord.RECORDSTATE_RECORDING) { - val read = testRecord?.read(buffer, 0, FRAME_SAMPLES) ?: break - if (read <= 0) break - NativeLib.voiceFeed(buffer.copyOf(read)) - } - }.apply { priority = Thread.MAX_PRIORITY; start() } - return true + return startCapture(CaptureMode.TEST, "__test__") } - - fun stopMicTest() { - isTesting = false - testThread?.join(1000); testThread = null - testRecord?.apply { try { stop() } catch (_: Exception) {}; release() } - testRecord = null - NativeLib.voiceCancel() + @Synchronized fun stopMicTest() { + if (captureMode == CaptureMode.TEST) finishCapture(false) else if (captureMode == null) releaseCaptureDevice() } @Volatile private var playing = false private var playThread: Thread? = null @Volatile private var playGeneration = 0 + /** Запустить с начала; callbacks приходят из worker только для текущего поколения, completion также при ошибке. */ fun playVoiceFile(filePath: String, onProgress: ((Float, Int) -> Unit)? = null, onComplete: () -> Unit = {}) { playFromPosition(filePath, 0, onProgress, onComplete) } - fun playFromPosition(filePath: String, startMs: Int, onProgress: ((Float, Int) -> Unit)? = null, onComplete: () -> Unit = {}) { + /** Запустить со startMs, округлённого вниз до 20 мс; seek декодирует и отбрасывает предшествующий PCM. + * Отрицательный startMs ограничивается нулём; прогресс учитывает только принятые AudioTrack сэмплы. */ + @Synchronized fun playFromPosition(filePath: String, startMs: Int, onProgress: ((Float, Int) -> Unit)? = null, onComplete: () -> Unit = {}) { stopPlayback() playing = true val myGen = playGeneration @@ -210,41 +212,56 @@ class AudioRecorderManager { return@Thread } val sampleRate = info[0]; val channels = info[1]; val totalSamples = info[2] + check(sampleRate in listOf(8000, 12000, 16000, 24000, 48000) && channels in 1..2 && totalSamples > 0) { + "invalid decode info rate=$sampleRate channels=$channels samples=$totalSamples" + } - /* skip to startMs by decoding and discarding frames */ + // Seek выполняется по полным 20мс кадрам; позиция берётся из фактически пропущенных сэмплов. val frameSamples = sampleRate / 50 * channels - val skipFrames = startMs / FRAME_MS + val skipFrames = startMs.coerceAtLeast(0) / FRAME_MS val skipBuf = ShortArray(frameSamples) var skippedSamples = 0L for (f in 0 until skipFrames) { if (myGen != playGeneration) return@Thread val dec = NativeLib.voiceDecodeRead(handle, skipBuf) - if (dec <= 0) break + check(dec in 0..skipBuf.size) { "decode seek failed result=$dec" } + if (dec == 0) break skippedSamples += dec } val bufSize = AudioTrack.getMinBufferSize(sampleRate, if (channels == 2) AudioFormat.CHANNEL_OUT_STEREO else AudioFormat.CHANNEL_OUT_MONO, AudioFormat.ENCODING_PCM_16BIT) + check(bufSize > 0) { "AudioTrack minimum=$bufSize" } val output = AudioTrack(AudioManager.STREAM_MUSIC, sampleRate, if (channels == 2) AudioFormat.CHANNEL_OUT_STEREO else AudioFormat.CHANNEL_OUT_MONO, AudioFormat.ENCODING_PCM_16BIT, bufSize, AudioTrack.MODE_STREAM) track = output + check(output.state == AudioTrack.STATE_INITIALIZED) { "AudioTrack not initialized" } output.play() + LogManager.addLog("INFO", "AudioPlayer", "playback started generation=$myGen rate=$sampleRate channels=$channels " + + "skippedSamples=$skippedSamples totalSamples=$totalSamples") val pcm = ShortArray(frameSamples) var readTotal = 0L var read: Int while (playing && myGen == playGeneration) { read = NativeLib.voiceDecodeRead(handle, pcm) - if (read <= 0) break + check(read in 0..pcm.size) { "decode read failed result=$read" } + if (read == 0) break if (myGen != playGeneration) break - output.write(pcm, 0, read) - readTotal += read + var offset = 0 + while (playing && myGen == playGeneration && offset < read) { + val written = output.write(pcm, offset, read - offset) + check(written in 1..(read - offset)) { "AudioTrack.write=$written remaining=${read - offset}" } + offset += written; readTotal += written + } val total = (skippedSamples + readTotal).toFloat() / totalSamples.coerceAtLeast(1).toFloat() - val eMs = startMs + (readTotal * FRAME_MS / frameSamples).toInt() - onProgress?.invoke(total, eMs) + val eMs = ((skippedSamples + readTotal) * 1000 / (sampleRate * channels)).toInt() + synchronized(this@AudioRecorderManager) { + if (myGen == playGeneration) onProgress?.invoke(total, eMs) + } } val drainMs = (bufSize * 1000L) / (sampleRate.toLong() * channels * 2) + 100 if (myGen == playGeneration) Thread.sleep(drainMs) @@ -255,20 +272,30 @@ class AudioRecorderManager { } finally { track?.apply { try { stop() } catch (e: Exception) { LogManager.addLog("WARN", "AudioPlayer", "stop failed: ${e.message}") } - release() + try { release() } catch (e: Exception) { LogManager.addLog("WARN", "AudioPlayer", "release failed: $e") } } - if (handle != 0L) NativeLib.voiceDecodeClose(handle) - if (playGeneration == myGen) { playing = false; onComplete() } + try { if (handle != 0L) NativeLib.voiceDecodeClose(handle) } + catch (e: Exception) { LogManager.addLog("ERROR", "AudioPlayer", "decode close failed generation=$myGen: $e") } + synchronized(this@AudioRecorderManager) { + if (playGeneration == myGen) { + playing = false; playThread = null + try { onComplete() } + catch (e: Exception) { LogManager.addLog("ERROR", "AudioPlayer", "completion callback failed generation=$myGen: $e") } + } + } + LogManager.addLog("DEBUG", "AudioPlayer", "playback worker exited generation=$myGen current=$playGeneration") } }.apply { start() } } - fun pausePlayback() { + /** Асинхронно остановить текущий worker; позицию для следующего playFromPosition хранит вызывающий код. */ + @Synchronized fun pausePlayback() { playGeneration++; playing = false playThread?.interrupt(); playThread = null } - fun stopPlayback() { + /** Инвалидировать callbacks и прервать worker; собственные AudioTrack/decoder он освобождает в finally. */ + @Synchronized fun stopPlayback() { playGeneration++; playing = false playThread?.interrupt(); playThread = null } diff --git a/tools/chatgui-android/capture-tests/src/main/kotlin/com/utun/chat/data/RecorderHostEnvironment.kt b/tools/chatgui-android/capture-tests/src/main/kotlin/com/utun/chat/data/RecorderHostEnvironment.kt index 05723a6f..c6cb0d00 100644 --- a/tools/chatgui-android/capture-tests/src/main/kotlin/com/utun/chat/data/RecorderHostEnvironment.kt +++ b/tools/chatgui-android/capture-tests/src/main/kotlin/com/utun/chat/data/RecorderHostEnvironment.kt @@ -9,6 +9,7 @@ object NativeLib { @Volatile var active = false var acceptStart = true var acceptFeed = true + var onFeed: () -> Unit = {} var stopResult = 800 val starts = CopyOnWriteArrayList>() val feeds = CopyOnWriteArrayList() @@ -18,7 +19,7 @@ object NativeLib { val decoderCloses = AtomicInteger() var decoderInfo = intArrayOf(48000, 1, 960) fun reset() { - active = false; acceptStart = true; acceptFeed = true; stopResult = 800 + active = false; acceptStart = true; acceptFeed = true; stopResult = 800; onFeed = {} starts.clear(); feeds.clear(); stops.set(0); cancels.set(0) decoderReads.clear(); decoderCloses.set(0); decoderInfo = intArrayOf(48000, 1, 960) } @@ -27,7 +28,7 @@ object NativeLib { if (!acceptStart || active) return false active = true; return true } - fun voiceFeed(pcm: ShortArray): Boolean { feeds.add(pcm.copyOf()); return active && acceptFeed } + fun voiceFeed(pcm: ShortArray): Boolean { feeds.add(pcm.copyOf()); onFeed(); return active && acceptFeed } @Synchronized fun voiceStop(): Int { stops.incrementAndGet(); active = false; return stopResult } @Synchronized fun voiceCancel() { cancels.incrementAndGet(); active = false } fun voiceDecodeOpen(path: String, info: IntArray): Long { decoderInfo.copyInto(info); return 1 } diff --git a/tools/chatgui-android/capture-tests/src/test/kotlin/com/utun/chat/data/AudioRecorderManagerTest.kt b/tools/chatgui-android/capture-tests/src/test/kotlin/com/utun/chat/data/AudioRecorderManagerTest.kt index 10f28e21..79456c45 100644 --- a/tools/chatgui-android/capture-tests/src/test/kotlin/com/utun/chat/data/AudioRecorderManagerTest.kt +++ b/tools/chatgui-android/capture-tests/src/test/kotlin/com/utun/chat/data/AudioRecorderManagerTest.kt @@ -98,6 +98,35 @@ class AudioRecorderManagerTest { val result = runCatching { manager.stopRecording() } assertEquals(-1, result.getOrNull()); assertFalse(NativeLib.active) assertEquals(1, NativeLib.cancels.get()); diagnostic("release") + device.failures = emptySet() + manager.cancelRecording() + assertTrue("cleanup lost the unreleased device", device.released) + assertEquals(1, NativeLib.cancels.get()) + } + + @Test fun feedFailureWhileStopIsJoiningCannotPublishPartialPcm() { + val entered = CountDownLatch(1); val leave = CountDownLatch(1) + NativeLib.acceptFeed = false + NativeLib.onFeed = { + entered.countDown() + var interrupted = false + while (true) { + try { check(leave.await(1, TimeUnit.SECONDS)); break } + catch (_: InterruptedException) { interrupted = true } + } + if (interrupted) Thread.currentThread().interrupt() + } + val device = plan(); assertTrue(manager.startRecording("7")) + device.chunks.add(AudioRecord.Chunk()) + assertTrue(entered.await(1, TimeUnit.SECONDS)) + val stopped = java.util.concurrent.FutureTask { manager.stopRecording() } + val worker = Thread(stopped, "voice-stop-race").apply { start() } + try { + assertTrue(device.stopEntered.await(1, TimeUnit.SECONDS)) + leave.countDown() + assertEquals(-1, stopped.get(1, TimeUnit.SECONDS)) + assertEquals(0, NativeLib.stops.get()); assertEquals(1, NativeLib.cancels.get()) + } finally { leave.countDown(); worker.join(2_000) } } @Test fun testReaderIsStoppedBeforeJoinAndStopFailureCannotHangIt() {