Fix/mqtt connection reliability | Android 11 / Nvidia Shield

[phoenixxx-1]

Publishing as draft; will test for a further 24 hours before marking as ready to merge to ensure stability across other apps

Whilst running v1.1.5 on an NVIDIA SHIELD TV (Android TV 11) publishing to a mosquitto broker, sensor updates would silently stop after some time (typically frozen in a stale playing state) until the application process was restarted. Debugging on-device traced this to three related problems, fixed in separate commits:

1. Dead connections were never detected (5f5a2f4)
   The client connects with keepAlive = 0, so when the connection dies silently (device standby, network drop, broker restart), neither side notices. Failed publishes were swallowed by tryConnectAndPublish whilst previous-value tracking in MainWorker advanced anyway, so the lost update (typically the final idle state at the end of a session) was never sent again, leaving retained topics permanently stale.
   Fixed by advertising a 60-second keep-alive (so the broker drops dead sessions), retrying a failed publish once on a fresh connection inside KMQTTClient, and replacing fire-and-forget publishing with publishWithRetry + collectLatest: a failed publish retries with backoff (1s → 30s) until it succeeds or a newer value supersedes it. The Home Assistant discovery configuration receives the same treatment, ensuring it is not lost when the application starts without connectivity.

2. KMQTT's client silently drops messages once stopped (091afdb)
   This was the main cause of the issue where publishing ceased entirely. Following any connection error, KMQTT's MQTTClient sets its internal running flag to false, after which step() is a silent no-op (if (running.value) check()) and a QoS 0 publish() is quietly discarded. Since getConnectedClient() never checked isRunning(), a single failed reconnection (e.g. RST during a broker restart) left a permanently dead client that accepted every subsequent publish without error, meaning no recovery path was ever triggered. Fixed by checking isRunning() after stepping (to replace the dead client) and after publishing (to raise an exception so retry logic takes over).
   Reproduced reliably with Emby, which updates the playback position 1 to 2 times per second: each update was a retained publish, and each publish takes ~500 ms (KMQTT's 250 ms socket read timeout), saturating the single-threaded dispatcher and flooding the broker's retained-message store. Position publications are now rate-limited to once every 5 seconds via conflate(), consistently converging on the latest value; state transitions still publish immediately.

3. The pipeline had no recovery from unexpected failures (57b2d94)

Defensive hardening rather than a fix for an observed crash: MainWorker.start() launches the publishing pipeline exactly once, so any exception escaping a media session flow would terminate all publisher coroutines permanently while the process keeps running. This can happen in practice — for example MediaSessionManager.getActiveSessions() is documented to throw SecurityException if notification access is revoked while the app runs. The two top-level loops are now supervised and restart after 5 seconds on unexpected failure, with the error logged.

Testing: verified on a SHIELD TV against mosquitto with Home Assistant MQTT discovery: YouTube and Emby playback, pause/stop/app-switch transitions, device standby during playback (the final idle state now survives connection loss and arrives on retry), broker-dropped connections (CLOSE_WAIT sockets detected and replaced), and a multi-hour soak test with no stuck sensors.

Relationship to #19: complementary rather than overlapping

#19 addresses the notification listener failing to bind after reboot, while this PR addresses the MQTT publishing side dying at runtime with the listener healthy. Both present identically to users ("sensors stop updating until a restart"), so existing reports of frozen sensors may be attributable to either cause.

[cbeyls]

Thank you for the report.

I have questions regarding the proposed fixes though:

What is the purpose of launchSupervised() ? It restarts a coroutine in case of Exceptions but you use it in 2 places that are supposed to not throw Exceptions:

• In the coroutine that monitors the settings and publishes the results. All network errors are supposed to be caught by tryConnectAndPublish(). The rest of the code is just callbacks that are not supposed to throw any exceptions. And when the service stops listening, it calls stopListening() which clears the state and does not throw exceptions either. So if you saw exceptions being thrown in this coroutine, can you tell me which methods threw those exceptions exactly so we can catch them properly at the source?
• In the "watchdog" coroutine calling MediaSessionListenerService.requestRebind() when the isListening boolean changes from true to false. requestRebind() already catches all exceptions internally so I'm pretty sure it's impossible that an Exception is thrown in that coroutine. Did I miss anything?

If an exception is due to a missing permission like you suggested then the entire app should probably be restarted after granting the permission anyway.

Regarding KMQTTClient, you found an issue indeed: step() will not throw exception if the client is already stopped so we should always call isRunning() before sending a message. The simpler fix is to add the call at the beginning of getConnectedClient() and always create a new instance in that case.

The current code in KMQTTClient auto-retries when the initial step() call fails, which I believe tries to clear the queue but probably doesn't do anything if the queue is empty so it's not a good enough check to determine if the connection is alive. So I'll change the code to make sure it also recreates the client and retries sending once if the second step() fails (the one happening after enqueuing the message). I will also add a small delay before the retry, directly in KMQTTClient.

I believe this should be enough to ensure that no message will be lost and we don't need to use the keepalive mechanism. If the client doesn't detect a disconnection, the next message sending failure will signal that the connection has been closed and it will re-establish the connection. If the broker decides to keep the connection open: good. If the broker decides to close the connection at some point: same thing, the client will notice the next time it tries to send a message. So I don't see any benefit from using keepalives with the current model, unless we also change the entire code to call step() regularly on the client in a loop to keep it talking to the server even when we have nothing to send and keep the connection "warm", which we don't do in the current code. The only benefit is allowing the broker to close the connection earlier to save resources. Do you agree or is there something I missed?

Position publications are now rate-limited to once every 5 seconds via conflate(),

All data should be conflated indeed. In MediaSessionFlows I already conflate the callbacks but indeed I overlooked that the flatMapLatest() calls on top of them are also buffered. Good catch! I'll add the 2 required extra conflate() calls at the source.
However the rate limiting fix is something else. It's not a good idea to add a delay between one playback state update and the next because it will introduce an artificial 5 seconds delay. Imagine you stop the video right after the playback state has been sent: your home assistant will only react to the playback stop action after 5 seconds. The point of this app is to have instant updates. So that's not the correct fix.

I'll treat this as a separate issue and make sure that the playback position is only updated once during a state switch (for example from pause to play) and not repeatedly while in the play state.

Finally, as far as I know, #19 has been fixed in the latest version. I was just waiting for confirmation from the PR author before closing their pull request.

[phoenixxx-1]

Good Morning!

Thanks for the detailed review, please see responses per point:

1. launchSupervised(): fair! I never caught an exception in either coroutine; this was something I added when chasing the KMQTTClient issue (a dead client silently "succeeding" produces exactly the same symptom as a dead coroutine, which is what I was trying to rule out at the time). Upon reflection, this should be dropped from the PR - my bad!

2. KMQTTClient / isRunning: agreed on checking at the start of getConnectedClient() and recreating - its cleaner than my implementation. One thing I think needs preserving in whatever shape the fix takes: a bounded retry inside KMQTTClient alone won't cover the failure that motivated this PR. The lost message is typically the final idle published at the exact moment the device enters standby and the network drops for minutes. This surfaced itself for me as I'm using the play state to drive light automations; I was turning on the shield and it was stuck in a "playing" state from the previous session (dimming my lights in turn. With the publisher-level publishWithRetry + collectLatest; even if that state arrived when I turned my shield on; "idle" was detected and the automation didn't trigger. Retry once is a good complement for network timeouts, but wouldn't correct this in my scenario alone. That being said, with send-failure detection plus the publisher-level retry in place, I agree the keepalive adds little and can go.

3. Rate Limiting Position: I believe that the 5s interval applies only to the position topic; playback state, title, etc. have separate collectors and always publish immediately - therefore a pause should be reflected instantly (and indeed is on my test APK). I'm not wedded to the update frequency being 5s; my thought process was that this would still be sufficient for any progress/time based automations, but not overload the broker with significant updates (Emby publishes these around every 500ms - conflate protects against slow consumers, but not high-frequency producers). Moving to publishing position only on a state switch would certainly stop the flood, though two things worth considering: (a) it may break the aforementioned progress-based automations, and (b) seeks: when the user jumps position during playback, the player reports it only as a new position in the PlaybackState callback with the state unchanged, under switch-only publishing the seek is never published and the last-known position silently becomes wrong until the next state switch. Comparing the reported position against the expected one (last position + elapsed × speed) would catch these discontinuities and publish just those.

Ref 19; I think it's probably a separate issue too; It felt worth highlighting just incase though!

Thanks again for the detailed review; I'm happy to adjust the PR in-line with the above (and any further feedback);

I'm equally as happy if you would prefer to cherry pick this and close this :)

[cbeyls]

Hello again, I was just writing a follow-up when you posted your answer.

First I want to correct what I said about the delay and rate-limiting (3): indeed I realized later that you changed the code to observe the playback state and position in 2 different coroutines and the delay is only introduced for the position, not the state. However I don't like this either because now the state can get desynchronized from its associated position. The most important issue was the missing conflate: we don't want to queue events and send everything, at the risk of ending up publishing events much later and not reflecting the current state.

MQTT is supposed to be a lightweight protocol so my initial thoughts is that it's not an issue to send messages continuously, as long as we always send the most recent one next and drop intermediate ones in case of a slow consumer. And my proposal of sending the position only in case of state change is not enough because it needs to take care of intermediate states like seeking and buffering.

So I suggest that either:

• we just keep the conflate which could send a lot of messages but they will always reflect the latest state
• or we introduce a separate flow that only reports the current position when the underlying state (buffering seeking etc.) changes (and the actual value changes as well).

I prefer the second solution as I'm not a big fan of reporting the position (the app initially did not and I added that feature later on request) but I first need to make tests to make sure it works correctly.

Now regarding the dropped messages (2): If I understand correctly, the point of the delay you want to add is for the case where the device is turned off, so it will come back online in the middle of a delay and try sending the state it could not send when the device was turned off ? I get the idea but I think that a better solution would be to monitor the device network connectivity and restart everything when it comes back online. I'll work on that solution.

I'll update the code immediately to improve the MQTTClient error detection and add the missing conflates. Then later I'll create test branches for the new playback position logic and the network state logic and I'll ask you to test them to make sure they fix your issues. Than you!

[phoenixxx-1]

That all sounds good, and thanks for re-checking point 3.

Desync: fair point, agreed. The split collectors do allow the state and position to momentarily disagree. Happy to drop the interval approach entirely; your option 2 is also my preference (a dedicated flow reporting position on underlying state changes).

One caveat to test on option 2: transient states (buffering/seeking) will catch most seeks, but I believe some players jump position with the state remaining PLAYING throughout; no transient state at all (I suspect Emby direct-play does this). If so, a position-delta check (reported vs. last position + elapsed × speed) may still be needed as a fallback. I can verify exactly what Emby emits during seeks on your test branch rather than guessing.

Network monitoring: yes, you've understood the intent of the retry correctly, and restarting the pipeline on reconnect achieves the same outcome more cleanly for that case. One gap from my debugging though: I captured a failure where the device's network was fully healthy and the broker reset the connection (fresh connects received RST - e.g. during a broker restart; my original stuck-playing incident also happened mid-playback with the network up). A connectivity callback never fires in that scenario, so the lost message would stay lost until the next state change. Suggestion: treat a publish that still fails after the internal retry as a second trigger for the same restart/recovery path - then both device-side and broker-side interruptions converge on one mechanism.

Testing: very happy to test both branches. Just ping here when the branches are up.

Thank you (and thankyou for the project - this saved me building my own - MQTT is a much more elegant solution than the HA ADB connections!)

[cbeyls]

@phoenixxx-1 Hello again and thank you for the feedback.

I believe I now have a better algorithm to publish playback positions. It will publish on state changes, as well as when the position is "drifting" during playback: after an instant seek or a playback speed change for example. I don't have an app that updates position every second during playback so I would be grateful if you can test it.

I added the network connectivity check as well in a temporary branch that I plan to merge soon. The entire publishing process is now stopped on connectivity loss and restarted from scratch with a fresh client when recovered. This should ensure that your session state is always properly reset after turning the device back on.

I didn't have this issue myself because both my TVs properly dispatch the playback stop event when I turn them off because they keep the network connection alive for a few minutes after turning off the screen (which also stops the player). But even on my TVs this change is an improvement because it's better to properly resend all MQTT events to the broker after the TV is turned back on.

That being said I'm still not conviced that the process should also be restarted in case of transient network errors or messages to be retried multiple times with a backoff strategy: the messages must be sent as fast as possible and everything is queued so if they can't go out because of a temporary failure they should be dropped and retried immediately once rather than delaying the queue (I know each message type is conflated but messages of different types can still queue up and the reported state becomes stale). If the broker is down or restarted, it's not the client's responsibility to care about it with a protocol like MQTT. With the new logic, the client properly takes care of its own failures and always starts with a fresh connection after recovering connectivity which is the most important. The retry logic should be enough to take care of stale connections while the device and the broker are both online.

But please try for yourself with the code in that branch and give me your feedback when you have time:
https://github.com/cbeyls/MediaSession2MQTT/tree/feat/connectivity_checker

I believe it should work much more reliably than the production release and I'll publish a new release soon so everyone can benefit from it.

Thank you for your help!

[phoenixxx-1]

Hi! Sorry this took a week or so to get around to; life gets in the way sometimes!

Tested feat/connectivity_checker on my setup (SHIELD TV / Android 11, mosquitto, Home Assistant, Emby + YouTube). Results:

• Position flood: fixed. During 60+ seconds of steady Emby playback (which emits position updates ~every 500ms), zero position messages were published. Previously this produced 1–2 retained messages per second.
• Seeks: detected. A fast-forward jump published the new position immediately via the drift check.
• Pause/stop/standby: all clean. State changes publish instantly with their positions; entering standby mid-session delivered paused then idle with all topics cleared before the network dropped.
• Connectivity recovery: works. I forced a network loss (WiFi off/on) — on recovery the publishing pipeline restarted with a fresh client and reconnected correctly.

One bug found: the initial position is never published at playback start when the player's first PLAYING state carries position -1 (PLAYBACK_POSITION_UNKNOWN), which Emby does. Since Idle.positionInMillis is also -1, the fold considers the position unchanged and skips it — and drift suppression then correctly filters all subsequent steady-play updates, so the position topic stays empty for the entire session until the first pause or seek. A simple fix: also publish the position whenever the state name changes (if (previousPlaybackState?.name != name || previousPlaybackState.positionInMillis != positionInMillis)), which additionally guarantees state and position always update together; in line with your desync concern earlier.

Overall this branch feels much more robust than the former release and the positioning bug doesnt impact my automations so I'll carry on running this build for some more "real world" testing and report back if I notice anything else noteworthy!

Thanks for your efforts!

[cbeyls]

Thank you for the detailed feedback. Indeed I forgot to take PLAYBACK_POSITION_UNKNOWN into account, so I just pushed a fix in the main branch to make sure transitioning to and from an unknown position is always reported.

You mentioned forcing a network loss, but did you also test simply turning the device off for a while then back on? This should also be fixed because gaining initial connectivity after turning the device back on should create the creation of a new fresh client.

I'll now merge the feat/connectivity_checker branch into main. You can do more tests from the latest main branch if you want to confirm that the Emby bug (initial position not reported until the first pause) has been fixed.

Then I'll publish a new release soon because I want everyone to benefit from these fixes. I'll credit your efforts in the release notes. Thank you again.

[phoenixxx-1]

Hi again :)

"You mentioned forcing a network loss, but did you also test simply turning the device off for a while then back on?"

I've been running the build for the past 4 days and haven't had any publishing issues after a restart or when recovering from sleep - I'd say this is solid too.

Thanks for the final changes - really appreciated. I'll go ahead and close this PR now to keep things clean - have a fantastic weekend.

[kmcc049]

Thanks for this, I've been testing a similar local change for the last month and just went to raise an issue and PR. However I don't need to now! Will a new release be cut soon?