Make voice recording ownership and playback failures explicit

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
     }

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<List<Any>>()
     val feeds = CopyOnWriteArrayList<ShortArray>()
@@ -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 }

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() {