fix(android): cast relay serves a clean header and restarts on its ports

The init is what precedes the first moof; ports are reused like Node's; the
SDK is read on the main thread; a cast that fails says why on screen, and
success waits until the receiver actually plays. Never a VPN's address.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

5 files changed, 159 insertions, 12 deletions

diff --git a/packages/meshbay-android/app/src/main/kotlin/org/meshbay/client/MainActivity.kt b/packages/meshbay-android/app/src/main/kotlin/org/meshbay/client/MainActivity.kt
index 41f5457..23df907 100644
--- a/packages/meshbay-android/app/src/main/kotlin/org/meshbay/client/MainActivity.kt
+++ b/packages/meshbay-android/app/src/main/kotlin/org/meshbay/client/MainActivity.kt
@@ -80,7 +80,8 @@ class MainActivity : Activity() {
                               startActivity = { intent -> runOnUiThread { startActivity(intent) } })
         // A process killed mid-download left unfinished files; nothing else will.
         Thread { saves.cleanUpAfterAKilledProcess() }.start()
-        cast = CastChannels(this, onCasting = { on -> runOnUiThread { casting(on) } })
+        cast = CastChannels(this, onCasting = { on -> runOnUiThread { casting(on) } },
+                            tell = { m -> runOnUiThread { android.widget.Toast.makeText(this, m, android.widget.Toast.LENGTH_LONG).show() } })
         channels = Channels(hub, onHubChanged = { runOnUiThread { reloadForHub() } },
                             hasCatalogue = { code -> hasAsset("ui/locales/$code.js") }, keys = keys, saves = saves,
                             cast = cast)
diff --git a/packages/meshbay-android/app/src/main/kotlin/org/meshbay/client/cast/CastChannels.kt b/packages/meshbay-android/app/src/main/kotlin/org/meshbay/client/cast/CastChannels.kt
index 0efae9d..1044209 100644
--- a/packages/meshbay-android/app/src/main/kotlin/org/meshbay/client/cast/CastChannels.kt
+++ b/packages/meshbay-android/app/src/main/kotlin/org/meshbay/client/cast/CastChannels.kt
@@ -19,6 +19,7 @@ import java.util.Base64
 class CastChannels(
     private val context: Context,
     private val onCasting: (Boolean) -> Unit,
+    private val tell: (String) -> Unit = {},
 ) {
     val control = CastControl(context)
     val relay = CastRelay(::lanAddress)
@@ -49,16 +50,29 @@ class CastChannels(
         "cast:devices" -> control.devices()
         "cast:chromecast:connect" -> {
             val o = a.optJSONObject(0) ?: JSONObject()
-            control.connect(o.optString("deviceId", ""), relayUrl(o), subtitleOf(o))
+            reported { control.connect(o.optString("deviceId", ""), relayUrl(o), subtitleOf(o)) }
         }
         "cast:chromecast:reload" -> {
             val o = a.optJSONObject(0) ?: JSONObject()
-            control.reload(relayUrl(o), subtitleOf(o))
+            reported { control.reload(relayUrl(o), subtitleOf(o)) }
         }
         "cast:chromecast:disconnect" -> control.disconnect()
         else -> throw Refused("Refused: no such channel")
     }
 
+    /**
+     * A failed cast says why, on the screen. The player catches the error and
+     * stops the relay without a word, which leaves the television on the
+     * receiver's idle screen and nobody knowing whether it ever reached the
+     * phone — so the receiver's reason and that fact are shown here.
+     */
+    private fun <T> reported(block: () -> T): T = try { block() } catch (e: Exception) {
+        val reached = if (relay.streamRequests > 0) "the television did reach this phone"
+                      else "the television never reached this phone at ${relay.url?.substringBefore("/stream") ?: "?"}"
+        tell("Cast failed: ${e.message} — $reached")
+        throw e
+    }
+
     fun push(bytes: ByteArray, offset: Int, length: Int): Boolean { relay.push(bytes, offset, length); return true }
 
     /**
@@ -77,11 +91,17 @@ class CastChannels(
     private fun subtitleOf(o: JSONObject): JSONObject? =
         if (o.has("subtitle") && o.get("subtitle") != JSONObject.NULL) o.optJSONObject("subtitle") else relay.subtitleInfo()
 
-    /** The Wi-Fi (or Ethernet) address, never the mobile network's: a TV cannot be reached there. */
+    /**
+     * The Wi-Fi (or Ethernet) address, never the mobile network's: a TV cannot
+     * be reached there. Nor a VPN's: a VPN network carries the transports of
+     * the network under it, Wi-Fi included, and its address is a tunnel the
+     * television cannot route to.
+     */
     private fun lanAddress(): InetAddress? {
         val cm = context.getSystemService(ConnectivityManager::class.java)
         for (network in cm.allNetworks) {
             val caps = cm.getNetworkCapabilities(network) ?: continue
+            if (caps.hasTransport(NetworkCapabilities.TRANSPORT_VPN)) continue
             if (!caps.hasTransport(NetworkCapabilities.TRANSPORT_WIFI) &&
                 !caps.hasTransport(NetworkCapabilities.TRANSPORT_ETHERNET)) continue
             val addr = cm.getLinkProperties(network)?.linkAddresses?.map { it.address }
diff --git a/packages/meshbay-android/app/src/main/kotlin/org/meshbay/client/cast/CastControl.kt b/packages/meshbay-android/app/src/main/kotlin/org/meshbay/client/cast/CastControl.kt
index 88e3a03..e2cd58a 100644
--- a/packages/meshbay-android/app/src/main/kotlin/org/meshbay/client/cast/CastControl.kt
+++ b/packages/meshbay-android/app/src/main/kotlin/org/meshbay/client/cast/CastControl.kt
@@ -111,7 +111,10 @@ class CastControl(private val context: Context) {
             var result: CastSession? = null
             val listener = object : SessionManagerListener<CastSession> {
                 override fun onSessionStarted(s: CastSession, id: String) { result = s; started.countDown() }
-                override fun onSessionStartFailed(s: CastSession, error: Int) { started.countDown() }
+                override fun onSessionStartFailed(s: CastSession, error: Int) {
+                    Log.w(TAG, "session start failed: error=$error")
+                    started.countDown()
+                }
                 override fun onSessionStarting(s: CastSession) {}
                 override fun onSessionEnding(s: CastSession) {}
                 override fun onSessionEnded(s: CastSession, error: Int) {}
@@ -121,6 +124,7 @@ class CastControl(private val context: Context) {
                 override fun onSessionSuspended(s: CastSession, reason: Int) {}
             }
             cast.sessionManager.addSessionManagerListener(listener, CastSession::class.java)
+            Log.i(TAG, "connect: selecting route '${route.name}'")
             r.selectRoute(route)
             deviceName = route.name
             Pending(started, { result }, { cast.sessionManager.removeSessionManagerListener(listener, CastSession::class.java) })
@@ -134,8 +138,12 @@ class CastControl(private val context: Context) {
                 else -> throw IllegalStateException()
             }
         }
+        // The cast framework's objects answer on the main thread only, the
+        // device's name included: read it there, never from this worker.
+        val name = deviceName ?: onMain { session.castDevice?.friendlyName }
+        Log.i(TAG, "session up on '$name', loading the relay stream")
         val state = load(session, mediaUrl, subtitle)
-        return JSONObject().put("deviceName", deviceName ?: session.castDevice?.friendlyName).put("playerState", state)
+        return JSONObject().put("deviceName", name ?: JSONObject.NULL).put("playerState", state)
     }
 
     private class Pending(val done: CountDownLatch, val session: () -> CastSession?, val cleanup: () -> Unit)
@@ -177,12 +185,47 @@ class CastControl(private val context: Context) {
             .apply { if (subUrl != null) setActiveTrackIds(longArrayOf(TEXT_TRACK_ID)) }.build()
         val loaded = CountDownLatch(1)
         var ok = false
+        var reason = ""
         onMain {
             val client = session.remoteMediaClient ?: throw IllegalStateException("The receiver has no media channel")
-            client.load(request).setResultCallback { r -> ok = r.status.isSuccess; loaded.countDown() }
+            client.load(request).setResultCallback { r ->
+                ok = r.status.isSuccess
+                reason = "code ${r.status.statusCode}" + (r.status.statusMessage?.let { ", $it" } ?: "")
+                // The receiver's own reason, which is the only one there is:
+                // the page is told "refused" and nothing else.
+                Log.i(TAG, "load result: ok=$ok code=${r.status.statusCode} " +
+                    "message=${r.status.statusMessage} state=${stateName(client.playerState)} " +
+                    "idleReason=${client.idleReason} subtitles=${subUrl != null}")
+                loaded.countDown()
+            }
+        }
+        if (!loaded.await(20, TimeUnit.SECONDS)) {
+            Log.w(TAG, "load: no answer from the receiver in 20 s")
+            throw IllegalStateException("The receiver did not load the stream")
+        }
+        if (!ok) throw IllegalStateException("The receiver refused the stream ($reason)")
+        // A load the receiver accepted is not a film on the screen: it answers
+        // IDLE, then fetches the stream, and only then plays — or gives up.
+        // So the answer waits for what the receiver actually did (measured
+        // against this receiver: OK/IDLE, then BUFFERING, then PLAYING).
+        val deadline = System.currentTimeMillis() + PLAY_WAIT_MS
+        while (System.currentTimeMillis() < deadline) {
+            val (state, idle) = onMain {
+                val c = session.remoteMediaClient
+                (c?.playerState ?: MediaStatus.PLAYER_STATE_UNKNOWN) to (c?.idleReason ?: MediaStatus.IDLE_REASON_NONE)
+            }
+            if (state == MediaStatus.PLAYER_STATE_PLAYING || state == MediaStatus.PLAYER_STATE_BUFFERING ||
+                state == MediaStatus.PLAYER_STATE_PAUSED) {
+                Log.i(TAG, "receiver is ${stateName(state)}")
+                return stateName(state)
+            }
+            if (state == MediaStatus.PLAYER_STATE_IDLE && idle != MediaStatus.IDLE_REASON_NONE) {
+                Log.w(TAG, "receiver gave up: idle reason ${idleName(idle)}")
+                throw IllegalStateException("The receiver gave up on the stream (${idleName(idle)})")
+            }
+            Thread.sleep(300)
         }
-        if (!loaded.await(20, TimeUnit.SECONDS)) throw IllegalStateException("The receiver did not load the stream")
-        if (!ok) throw IllegalStateException("The receiver refused the stream")
+        Log.w(TAG, "receiver still not playing after ${PLAY_WAIT_MS / 1000} s")
         return onMain { stateName(session.remoteMediaClient?.playerState ?: MediaStatus.PLAYER_STATE_UNKNOWN) }
     }
 
@@ -207,11 +250,22 @@ class CastControl(private val context: Context) {
 
     companion object {
         const val SCAN_DURATION_MS = 6000L
+        private const val TAG = "MeshBayCast"
         // The one track the receiver is ever told about; changing subtitles
         // reloads with a new URL under this same id.
         const val TEXT_TRACK_ID = 1L
         private val SCAN_TOKEN = Any()
 
+        private const val PLAY_WAIT_MS = 15000L
+
+        fun idleName(r: Int) = when (r) {
+            MediaStatus.IDLE_REASON_FINISHED -> "finished"
+            MediaStatus.IDLE_REASON_CANCELED -> "cancelled"
+            MediaStatus.IDLE_REASON_INTERRUPTED -> "interrupted"
+            MediaStatus.IDLE_REASON_ERROR -> "error"
+            else -> "reason $r"
+        }
+
         fun stateName(s: Int) = when (s) {
             MediaStatus.PLAYER_STATE_PLAYING -> "PLAYING"
             MediaStatus.PLAYER_STATE_PAUSED -> "PAUSED"
diff --git a/packages/meshbay-android/app/src/main/kotlin/org/meshbay/client/cast/CastRelay.kt b/packages/meshbay-android/app/src/main/kotlin/org/meshbay/client/cast/CastRelay.kt
index 5cd5f3e..c56d57d 100644
--- a/packages/meshbay-android/app/src/main/kotlin/org/meshbay/client/cast/CastRelay.kt
+++ b/packages/meshbay-android/app/src/main/kotlin/org/meshbay/client/cast/CastRelay.kt
@@ -1,5 +1,6 @@
 package org.meshbay.client.cast
 
+import android.util.Log
 import org.json.JSONObject
 import java.io.BufferedReader
 import java.io.InputStreamReader
@@ -47,6 +48,9 @@ class CastRelay(private val lanAddress: () -> InetAddress?) {
     @Volatile private var initSegment: ByteArray? = null
     @Volatile private var subtitle: Subtitle? = null
     @Volatile private var subtitleVersion = 0
+    /** How many times a receiver asked for the stream since the relay started. */
+    @Volatile var streamRequests = 0
+        private set
     private val ring = ArrayDeque<ByteArray>()
     private var ringBytes = 0L
     private val clients = ConcurrentHashMap.newKeySet<Client>()
@@ -83,8 +87,11 @@ class CastRelay(private val lanAddress: () -> InetAddress?) {
         if (server != null) stop()
         val address = lanAddress() ?: throw IllegalStateException("Casting needs this device on Wi-Fi")
         token = randomToken()
+        streamRequests = 0
+        pushes = 0
+        fragments = 0
         host = address.hostAddress
-        initSegment = init
+        initSegment = init?.let(::headerOnly)
         synchronized(ring) { ring.clear(); ringBytes = 0 }
         clients.clear()
         accum = BoxAccumulator()
@@ -95,7 +102,11 @@ class CastRelay(private val lanAddress: () -> InetAddress?) {
         for (i in 0 until PORT_COUNT) {
             val s = ServerSocket()
             try {
-                s.reuseAddress = false
+                // As Node's server does (and so the desktop relay): a port just
+                // closed sits in TIME_WAIT, and every seek restarts the relay —
+                // without this, four quick seeks used all four ports. It does
+                // not let two live listeners share a port.
+                s.reuseAddress = true
                 s.bind(InetSocketAddress(address, PORT_BASE + i))
                 bound = s; port = PORT_BASE + i
                 break
@@ -105,13 +116,20 @@ class CastRelay(private val lanAddress: () -> InetAddress?) {
         }
         server = bound ?: throw IllegalStateException("All cast relay ports are in use")
         Thread({ acceptLoop(bound) }, "cast-relay-accept").apply { isDaemon = true }.start()
+        log("started on ${hostPart()}:$port, init ${init?.size ?: 0} bytes, header ${initSegment?.size ?: 0} bytes")
         return JSONObject().put("url", url).put("port", port).put("token", token)
             .put("subtitle", subtitleInfo() ?: JSONObject.NULL)
     }
 
+    @Volatile private var pushes = 0
+    @Volatile private var fragments = 0
+
     fun push(data: ByteArray, offset: Int = 0, length: Int = data.size - offset) {
         if (server == null) return
+        pushes++
+        if (pushes <= 3 || pushes % 100 == 0) log("push #$pushes: $length bytes, $fragments fragments so far")
         for (frag in accum.push(data, offset, length)) {
+            fragments++
             synchronized(ring) {
                 ring.addLast(frag); ringBytes += frag.size
                 // Bounded in fragments, as the desktop is, and in bytes too: a
@@ -173,13 +191,14 @@ class CastRelay(private val lanAddress: () -> InetAddress?) {
             if (method == "OPTIONS") return respond(out, socket, 204, CORS_HEADERS)
 
             val path = target.substringBefore('?')
+            log("$method $path from ${socket.inetAddress?.hostAddress}")
             val query = target.substringAfter('?', "").split('&').associate {
                 it.substringBefore('=') to it.substringAfter('=', "")
             }
             if (token == null || query["t"] != token) return respond(out, socket, 403, emptyMap())
             when (path) {
                 "/subs.vtt" -> serveSubtitle(out, socket)
-                "/stream.mp4" -> serveStream(out, socket)
+                "/stream.mp4" -> { streamRequests++; serveStream(out, socket) }
                 else -> respond(out, socket, 404, emptyMap())
             }
         } catch (e: Exception) {
@@ -212,6 +231,7 @@ class CastRelay(private val lanAddress: () -> InetAddress?) {
         for (frag in backlog) chunk(out, frag)
         out.flush()
         clients.add(client)
+        log("client ${socket.inetAddress?.hostAddress} served init + ${backlog.size} fragments")
         try {
             while (!client.closed) {
                 val frag = client.queue.take()
@@ -223,12 +243,37 @@ class CastRelay(private val lanAddress: () -> InetAddress?) {
         } catch (e: Exception) {
             // The receiver went away; nothing to tell anyone.
         } finally {
+            log("client ${socket.inetAddress?.hostAddress} gone")
             clients.remove(client)
             try { socket.close() } catch (e: Exception) {}
         }
     }
 
     companion object {
+        /**
+         * The init segment is what comes before the first moof, and only that.
+         *
+         * The page hands over the first chunk it decrypted, which is a slice of
+         * the stream and not a box: on a real film it was 65536 bytes — ftyp,
+         * moov, then the first moof and the start of its mdat. Served whole,
+         * the receiver read that partial fragment, then the same fragment again
+         * from the accumulator (the page pushes that chunk too), and the box
+         * structure was broken from the first fragment: the receiver gave up
+         * within three seconds, on the television's idle screen. Whether the
+         * first chunk carries film depends on when the node read ffmpeg's
+         * output, so the same film can work once and not the next time.
+         */
+        fun headerOnly(init: ByteArray): ByteArray {
+            var at = 0
+            while (at + 8 <= init.size) {
+                if (BoxAccumulator.u32(init, at + 4) == BoxAccumulator.MOOF) return init.copyOfRange(0, at)
+                val size = BoxAccumulator.u32(init, at)
+                if (size < 8) break
+                at += size.toInt()
+            }
+            return init
+        }
+
         const val RING_CAP = 64
         const val RING_MAX_BYTES = 32L * 1024 * 1024
         const val BACKPRESSURE_HIGH = 8L * 1024 * 1024
@@ -236,6 +281,9 @@ class CastRelay(private val lanAddress: () -> InetAddress?) {
         const val PORT_COUNT = 4
         private val END = ByteArray(0)
 
+        // Never the URL: it carries the token.
+        private fun log(m: String) = try { Log.i("MeshBayCast", "[relay] $m") } catch (e: RuntimeException) { /* JVM tests */ }
+
         // A receiver reads a side-loaded subtitle with XHR from its own
         // origin, so the headers it sends are allowed by name — Range included.
         val CORS_HEADERS = mapOf(
diff --git a/packages/meshbay-android/app/src/test/kotlin/org/meshbay/client/CastRelayTest.kt b/packages/meshbay-android/app/src/test/kotlin/org/meshbay/client/CastRelayTest.kt
index 9fe1cfd..e11c8c0 100644
--- a/packages/meshbay-android/app/src/test/kotlin/org/meshbay/client/CastRelayTest.kt
+++ b/packages/meshbay-android/app/src/test/kotlin/org/meshbay/client/CastRelayTest.kt
@@ -92,6 +92,19 @@ class CastRelayTest {
         assertArrayEquals(init + fragment(1) + fragment(2), reply.body)
     }
 
+    @Test fun `a first chunk that carries film is served as its header only`() {
+        // As the page delivers it from a real film: a 64 KB slice holding the
+        // header, the first moof and the start of its mdat — pushed as well.
+        val header = box("ftyp", 20) + box("moov", 1200)
+        val stream = header + fragment(1) + fragment(2)
+        val firstChunk = stream.copyOfRange(0, header.size + 300)
+        val started = relay.start(firstChunk, null)
+        var at = 0
+        while (at < stream.size) { val n = minOf(300, stream.size - at); relay.push(stream, at, n); at += n }
+        val reply = get(started.getString("url"), readBytes = stream.size)
+        assertArrayEquals("header, then each fragment once", stream, reply.body)
+    }
+
     @Test fun `the token is required and unguessable`() {
         val url = relay.start(null, null).getString("url")
         val token = url.substringAfter("t=")
@@ -142,6 +155,17 @@ class CastRelayTest {
         assertTrue(relay.backlogBytes() >= CastRelay.RING_MAX_BYTES - big.size)
     }
 
+    @Test fun `seek after seek, the relay restarts on its ports`() {
+        // A seek restarts the relay, and a receiver was connected each time:
+        // the ports it closed sit in TIME_WAIT.
+        repeat(8) {
+            val url = relay.start(null, null).getString("url")
+            relay.push(fragment(it))
+            get(url, readBytes = fragment(it).size)
+            relay.stop()
+        }
+    }
+
     @Test fun `stopping closes the port`() {
         val url = relay.start(null, null).getString("url")
         relay.stop()