fix: the secure-update crash and the camera ladder (0.6.11)

app/build.gradle.kts

@@ -37,8 +37,8 @@ android {
         minSdk = 33
         targetSdk = 35
         // Remain newer than the 0.6.3 rendezvous candidate (code 26).
-        versionCode = 33
-        versionName = "0.6.10"
+        versionCode = 34
+        versionName = "0.6.11"
 
         testInstrumentationRunner = "androidx.test.runner.AndroidJUnitRunner"
 

app/src/main/kotlin/dev/forgesworn/kithmoot/media/LocalMedia.kt

@@ -79,6 +79,9 @@ class LocalMedia(
 
     private var cameraCapturer: CameraVideoCapturer? = null
     private var cameraSource: VideoSource? = null
+
+    /** What the camera source hands the encoders; see [VideoLadder]. */
+    @Volatile private var cameraRung: VideoRung = VideoLadder.FULL
     private var cameraTrack: VideoTrack? = null
     private var cameraHelper: SurfaceTextureHelper? = null
     private var frontFacing = true
@@ -191,6 +194,9 @@ class LocalMedia(
         // frame reaches the encoder without having been through the rule in
         // `routeFor`. With no scene chosen it forwards frames untouched.
         source.setVideoProcessor(newBackgroundProcessor())
+        // The sensor runs at the full format; the source scales and drops to
+        // the rung before the processor and every encoder see a frame.
+        source.adaptOutputFormat(cameraRung.width, cameraRung.height, cameraRung.fps)
         capturer.initialize(helper, context, source.capturerObserver)
         capturer.startCapture(CAMERA_WIDTH, CAMERA_HEIGHT, CAMERA_FPS)
         val track = factory.createVideoTrack(trackId(Roles.CAMERA), source)
@@ -203,6 +209,19 @@ class LocalMedia(
         return track
     }
 
+    /**
+     * Fit the camera to a call of this size.
+     *
+     * One downscale at the source, so a phone on a four-way call encodes
+     * 640 by 360 three times rather than 720p three times. Takes effect on
+     * the next frame, or when the camera next starts.
+     */
+    @Synchronized
+    fun adaptCamera(rung: VideoRung) {
+        cameraRung = rung
+        runCatching { cameraSource?.adaptOutputFormat(rung.width, rung.height, rung.fps) }
+    }
+
     @Synchronized
     fun stopCamera() {
         runCatching { cameraCapturer?.stopCapture() }

app/src/main/kotlin/dev/forgesworn/kithmoot/media/VideoLadder.kt

@@ -0,0 +1,62 @@
+package dev.forgesworn.kithmoot.media
+
+import dev.forgesworn.kithmoot.session.Roles
+import org.webrtc.RtpSender
+
+/**
+ * What the camera sends for a call of this size.
+ *
+ * Every remote device on a call has its own peer connection and so its own
+ * encoder for this device's camera. A phone on a four-way call runs three
+ * encoders and three decoders at once, and with nothing to bound them each
+ * encodes the full capture (1280 by 720 at 30) at libwebrtc's own ceiling of
+ * about 2.5 Mbps. That is the heat, and the uplink.
+ *
+ * A rung is one downscale at the source, done once and fed to every encoder,
+ * and one bitrate ceiling per sender, which is where libwebrtc takes it.
+ * Nothing on the wire changes: resolution and bitrate are a sender's own
+ * choice, and the far end plays what arrives.
+ */
+data class VideoRung(val width: Int, val height: Int, val fps: Int, val maxBitrateBps: Int)
+
+object VideoLadder {
+    /** One far end: the camera as captured. */
+    val FULL: VideoRung = VideoRung(1280, 720, 30, 1_200_000)
+
+    /** Two or three far ends. */
+    val MEDIUM: VideoRung = VideoRung(960, 540, 24, 800_000)
+
+    /** Four or more. */
+    val SMALL: VideoRung = VideoRung(640, 360, 15, 500_000)
+
+    /** The rung for a call where this device's camera goes to [peers] devices. */
+    fun rungFor(peers: Int): VideoRung = when {
+        peers <= 1 -> FULL
+        peers <= 3 -> MEDIUM
+        else -> SMALL
+    }
+}
+
+/**
+ * Whether a track id names this device's camera.
+ *
+ * Track ids are `role-uuid` (see `LocalMedia.trackId`), so the role is
+ * readable off the id alone, which is all a sender has. Screen shares are
+ * left at libwebrtc's defaults: text needs the bits, and they already run at
+ * 15 frames a second.
+ */
+internal fun isCameraTrackId(trackId: String): Boolean = trackId.startsWith("${Roles.CAMERA}-")
+
+/**
+ * Put a bitrate ceiling on one sender. False when nothing was set: no
+ * ceiling asked for, no encodings to set it on yet, or libwebrtc refused.
+ */
+internal fun capSender(sender: RtpSender, maxBitrateBps: Int): Boolean {
+    if (maxBitrateBps <= 0) return false
+    return runCatching {
+        val parameters = sender.parameters
+        if (parameters.encodings.isEmpty()) return@runCatching false
+        for (encoding in parameters.encodings) encoding.maxBitrateBps = maxBitrateBps
+        sender.setParameters(parameters)
+    }.getOrDefault(false)
+}

app/src/main/kotlin/dev/forgesworn/kithmoot/media/WebRtcEngine.kt

@@ -6,10 +6,14 @@ import dev.forgesworn.kithmoot.protocol.SignalBody
 import dev.forgesworn.kithmoot.protocol.TrackRef
 import dev.forgesworn.kithmoot.session.CALL_PROFILE_2
 import dev.forgesworn.kithmoot.session.CALL_PROFILE_2_ENABLED
+import dev.forgesworn.kithmoot.session.Roles
 import dev.forgesworn.kithmoot.session.RoomSession
 import kotlinx.coroutines.CancellationException
+import kotlinx.coroutines.CoroutineExceptionHandler
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.Job
+import kotlinx.coroutines.SupervisorJob
 import kotlinx.coroutines.flow.MutableStateFlow
 import kotlinx.coroutines.flow.StateFlow
 import kotlinx.coroutines.flow.asStateFlow
@@ -81,6 +85,21 @@ class WebRtcEngine(
     private val iceServers: List<PeerConnection.IceServer>,
 ) {
 
+    /**
+     * Where the engine's jobs run.
+     *
+     * A supervisor under the caller's scope, with a handler: a job that throws
+     * is a logged failure, not the end of the process, and does not take the
+     * other jobs with it. Cancelling the caller's scope still cancels all of
+     * it. The caller's dispatcher is kept; the collectors pick
+     * [MediaDispatcher] themselves.
+     */
+    private val engineScope = CoroutineScope(
+        scope.coroutineContext +
+            SupervisorJob(scope.coroutineContext[Job]) +
+            CoroutineExceptionHandler { _, failure -> Log.e("KithMootMedia", "engine job failed", failure) },
+    )
+
     val eglBase: EglBase = EglBase.create()
 
     private val factory: PeerConnectionFactory
@@ -161,7 +180,7 @@ class WebRtcEngine(
 
     fun setCallActive(active: Boolean) {
         synchronized(lock) { callActive = active }
-        if (active) scope.launch(MediaDispatcher) { reconcile(session.remoteDevices.value) } else stop()
+        if (active) engineScope.launch(MediaDispatcher) { reconcile(session.remoteDevices.value) } else stop()
     }
 
     /**
@@ -174,8 +193,8 @@ class WebRtcEngine(
         // The set of devices to connect to is derived from the roster, so a
         // device that joins, leaves or lapses is reconciled here rather than
         // being handled as an event.
-        scope.launch(MediaDispatcher) { session.remoteDevices.collect { reconcile(it) } }
-        scope.launch(MediaDispatcher) {
+        engineScope.launch(MediaDispatcher) { session.remoteDevices.collect { reconcile(it) } }
+        engineScope.launch(MediaDispatcher) {
             session.signals.collect { signal ->
                 val target = managedLinkFor(signal.from) ?: return@collect
                 try {
@@ -193,8 +212,8 @@ class WebRtcEngine(
                 }
             }
         }
-        scope.launch(MediaDispatcher) { localMedia.tracks.collect { onLocalTracksChanged(it) } }
-        scope.launch(MediaDispatcher) {
+        engineScope.launch(MediaDispatcher) { localMedia.tracks.collect { onLocalTracksChanged(it) } }
+        engineScope.launch(MediaDispatcher) {
             while (isActive) {
                 delay(10_000)
                 synchronized(lock) { links.values().forEach { it.reportMediaProgress() } }
@@ -205,7 +224,7 @@ class WebRtcEngine(
         // the only honest evidence that a direction is alive: a connection can
         // be `connected`, the signalling quiet, every object healthy, and one
         // direction carrying nothing at all.
-        scope.launch(MediaDispatcher) {
+        engineScope.launch(MediaDispatcher) {
             while (isActive) {
                 delay(HEALTH_SAMPLE_MS)
                 val sampling = synchronized(lock) { links.values().filter { it.profileTwo } }
@@ -260,6 +279,7 @@ class WebRtcEngine(
             _connections.update { it - device }
             _remoteTracks.update { current -> current.filterNot { it.device == device } }
         }
+        applyRung()
     }
 
     private fun managedLinkFor(device: String): ManagedLink? = links[device]
@@ -381,6 +401,28 @@ class WebRtcEngine(
         synchronized(lock) {
             for ((device, link) in links.snapshot()) link.syncLocalTracks(tracksFor(device, tracks))
         }
+        applyRung()
+    }
+
+    /** What the camera sends right now; see [VideoLadder]. */
+    @Volatile private var rung: VideoRung = VideoLadder.FULL
+
+    /**
+     * Fit the camera to the number of devices it goes to.
+     *
+     * Called after the links or the audience move, and outside the engine
+     * lock: adapting the source and setting sender parameters both wait on
+     * libwebrtc's threads, and nothing that waits on them may hold the lock
+     * (see [reconcile]). A sender added later is capped as it is added.
+     */
+    private fun applyRung() {
+        val peers = links.devices.count { runCatching { audience(it) }.getOrDefault(false) }
+        val next = VideoLadder.rungFor(peers)
+        if (next == rung) return
+        rung = next
+        Log.i("KithMootMedia", "camera rung peers=$peers ${next.width}x${next.height}@${next.fps} maxBitrateBps=${next.maxBitrateBps}")
+        localMedia.adaptCamera(next)
+        for (link in links.values()) link.capCamera(next.maxBitrateBps)
     }
 
     private fun tracksFor(device: String, tracks: List<LocalTrack> = localMedia.tracks.value): List<LocalTrack> =
@@ -460,8 +502,10 @@ class WebRtcEngine(
 
         @Volatile private var closed = false
         private var connection: PeerConnection? = null
-        private val senders = mutableMapOf<String, RtpSender>()
+        // Read outside the engine lock by `capCamera`; written under it.
+        private val senders = java.util.concurrent.ConcurrentHashMap<String, RtpSender>()
         private val received = java.util.concurrent.ConcurrentHashMap<String, MediaStreamTrack>()
+        private var handle: WebRtcPeerConnection? = null
         lateinit var link: PeerLink
             private set
 
@@ -499,7 +543,7 @@ class WebRtcEngine(
 
             override fun onIceCandidate(candidate: IceCandidate?) {
                 val value = candidate ?: return
-                scope.launch { link.onLocalCandidate(IceCandidateData(value.sdp, value.sdpMid, value.sdpMLineIndex)) }
+                engineScope.launch { link.onLocalCandidate(IceCandidateData(value.sdp, value.sdpMid, value.sdpMLineIndex)) }
             }
 
             override fun onRenegotiationNeeded() {
@@ -538,21 +582,24 @@ class WebRtcEngine(
 
         fun attach(connection: PeerConnection) {
             this.connection = connection
+            val handle = WebRtcPeerConnection(
+                connection,
+                ::refreshRemoteTracks,
+                // Profile 1 adds and removes senders on the connection
+                // directly, so this map is the only thing that can say
+                // whether a repeated offer would be answered the same way.
+                localMedia = { runCatching { senders.keys.toSet() }.getOrNull() },
+                cameraBitrate = { rung.maxBitrateBps },
+            )
+            this.handle = handle
             link = PeerLink(
                 localDevice = session.identity.devicePubkey,
                 remoteDevice = device,
-                connection = WebRtcPeerConnection(
-                    connection,
-                    ::refreshRemoteTracks,
-                    // Profile 1 adds and removes senders on the connection
-                    // directly, so this map is the only thing that can say
-                    // whether a repeated offer would be answered the same way.
-                    localMedia = { runCatching { senders.keys.toSet() }.getOrNull() },
-                ),
+                connection = handle,
                 roomId = session.room.roomId,
                 send = ::sendEnvelope,
                 callProfile = if (profileTwo) CALL_PROFILE_2 else 1,
-                scope = scope,
+                scope = engineScope,
                 onRebuild = { generation, open -> rebuildLink(device, generation, open) },
                 onDowngrade = {
                     // The far end reloaded into a build that does not speak
@@ -689,7 +736,25 @@ class WebRtcEngine(
             // catches up.
             runCatching { connection.addTrack(track.track, listOf(STREAM_ID)) }
                 .getOrNull()
-                ?.let { senders[track.trackId] = it }
+                ?.let { sender ->
+                    senders[track.trackId] = sender
+                    if (track.role == Roles.CAMERA) capSender(sender, rung.maxBitrateBps)
+                }
+        }
+
+        /**
+         * Bound what this pair's camera sender may spend; see [VideoLadder].
+         *
+         * A profile-1 pair's senders are the ones `addLocalTrack` kept; a
+         * profile-2 pair's live in its slots, which the handle reads.
+         */
+        fun capCamera(maxBitrateBps: Int) {
+            if (closed) return
+            if (profileTwo) {
+                handle?.capCameraSenders(maxBitrateBps)
+                return
+            }
+            for ((id, sender) in senders) if (isCameraTrackId(id)) capSender(sender, maxBitrateBps)
         }
 
         /**
@@ -707,7 +772,7 @@ class WebRtcEngine(
         fun syncLocalTracks(tracks: List<LocalTrack>) {
             if (profileTwo) {
                 if (!::link.isInitialized) return
-                scope.launch { runCatching { link.applyTracks(tracks.map { it.slot() }) } }
+                engineScope.launch { runCatching { link.applyTracks(tracks.map { it.slot() }) } }
                 return
             }
             val wanted = tracks.associateBy { it.trackId }

app/src/main/kotlin/dev/forgesworn/kithmoot/media/WebRtcPeerConnection.kt

@@ -7,6 +7,7 @@ import org.webrtc.PeerConnection
 import org.webrtc.RtpTransceiver
 import org.webrtc.SdpObserver
 import org.webrtc.SessionDescription
+import org.webrtc.VideoTrack
 import kotlin.coroutines.resume
 import kotlin.coroutines.resumeWithException
 
@@ -32,6 +33,8 @@ class WebRtcPeerConnection(
      * being told again, expensively.
      */
     private val localMedia: () -> Set<String>? = { null },
+    /** The camera sender's bitrate ceiling at the moment a slot binds it; 0 for none. See [VideoLadder]. */
+    private val cameraBitrate: () -> Int = { 0 },
 ) : PeerConnectionHandle {
 
     override fun localMedia(): Set<String>? = runCatching { localMedia.invoke() }.getOrNull()
@@ -98,7 +101,26 @@ class WebRtcPeerConnection(
         if (media != null && media !is MediaStreamTrack) return false
         // `takeOwnership = false`: the track belongs to LocalMedia and outlives
         // any one connection, so the sender must not dispose it.
-        return withTransceiver(mid) { it.sender.setTrack(media as MediaStreamTrack?, false) }
+        return withTransceiver(mid) {
+            val bound = it.sender.setTrack(media as MediaStreamTrack?, false)
+            if (bound && media is VideoTrack && isCameraTrackId(media.id())) capSender(it.sender, cameraBitrate())
+            bound
+        }
+    }
+
+    /**
+     * Bound every camera sender on this connection; see [VideoLadder].
+     *
+     * The transceivers are read once and never held, for the reason
+     * [withTransceiver] gives.
+     */
+    fun capCameraSenders(maxBitrateBps: Int) {
+        runCatching {
+            for (transceiver in connection.transceivers) {
+                val track = transceiver.sender.track() ?: continue
+                if (track is VideoTrack && isCameraTrackId(track.id())) capSender(transceiver.sender, maxBitrateBps)
+            }
+        }
     }
 
     /**

app/src/main/kotlin/dev/forgesworn/kithmoot/session/RoomSession.kt

@@ -483,9 +483,18 @@ class RoomSession(
      */
     fun calls(): List<CallView> = callsOf(_participants.value)
 
+    /**
+     * What this device is publishing, for the roster.
+     *
+     * Best-effort like a heartbeat, not fail-closed like a chat message: the
+     * engine calls this on every local track change, and a camera toggle
+     * does not stop for a secure update. While the gate is shut the set is
+     * kept and nothing is published; the successor epoch's first
+     * announcement (`applyEpoch`) and every heartbeat after it carry it.
+     */
     fun setTracks(tracks: List<TrackRef>) {
         synchronized(lock) { this.tracks = tracks }
-        announce()
+        announceIfPublishing()
     }
 
     /**