feat(jam): comprehensive Jam protocol & performance fixes

Addresses #3181, #3184, #3186

Protocol changes (JamProtocol.kt):
- Add SeekTo command for precise position synchronization
- Add currentPositionMs to QueueState for resume/sync accuracy

Host improvements (JamRoomService.kt):
- Immediate broadcast on room creation so guests sync current track instantly (#3184)
- Broadcast currentPositionMs with every state update
- Throttle broadcasts: distinctUntilChanged + debounce(500ms) to reduce MQTT traffic & CPU (#3186)
- Send explicit SeekTo command on resume/play so guests seek before playing (#3181)

Guest improvements (JamRoomService.kt):
- On resume: seek to host's currentPositionMs BEFORE calling play() (#3181)
- Periodic drift correction: if position diff > 2s, auto-seek to host position (#3181)
- Handle SeekTo command from host

Connection resilience (JamRoomClient.kt):
- Exponential backoff reconnect (1s, 2s, 4s... max 30s) with jitter (#3186)
- Max 10 reconnect attempts before giving up
- Prevents reconnection storms on flaky networks

Also includes participant name UI from previous commit (#3185)
This commit is contained in:
thumb2086 2026-10-31 01:10:20 +08:00
parent 8df0e239a7
commit 3ae00f296a
3 changed files with 192 additions and 27 deletions

View File

@ -46,7 +46,9 @@ sealed class JamMessage {
*/ */
val follow: Boolean = false, val follow: Boolean = false,
/** Room-wide play/pause; guests follow every change of it. */ /** Room-wide play/pause; guests follow every change of it. */
val isPlaying: Boolean = false) : JamMessage() val isPlaying: Boolean = false,
/** Host's current playback position in milliseconds. Guests seek to this on resume/sync. */
val currentPositionMs: Long = 0) : JamMessage()
@Serializable @Serializable
@SerialName("playbackCommand") @SerialName("playbackCommand")
@ -96,6 +98,10 @@ sealed class PlaybackCmd {
@Serializable @Serializable
@SerialName("jumpTo") @SerialName("jumpTo")
data class JumpTo(val index: Int) : PlaybackCmd() data class JumpTo(val index: Int) : PlaybackCmd()
@Serializable
@SerialName("seekTo")
data class SeekTo(val positionMs: Long) : PlaybackCmd()
} }
/** Retained per-participant presence entry (with an MQTT Last Will for leave). */ /** Retained per-participant presence entry (with an MQTT Last Will for leave). */

View File

@ -32,6 +32,7 @@ import com.ditchoom.mqtt5.controlpacket.ConnectionRequest
import org.spotube.spotube.modules.settings.JamBroker import org.spotube.spotube.modules.settings.JamBroker
import kotlinx.coroutines.CoroutineScope import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.Dispatchers import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.Job
import kotlinx.coroutines.SupervisorJob import kotlinx.coroutines.SupervisorJob
import kotlinx.coroutines.TimeoutCancellationException import kotlinx.coroutines.TimeoutCancellationException
import kotlinx.coroutines.flow.MutableSharedFlow import kotlinx.coroutines.flow.MutableSharedFlow
@ -55,6 +56,7 @@ import spotube.composeapp.generated.resources.jam_error_connection_lost
import spotube.composeapp.generated.resources.jam_error_no_broker import spotube.composeapp.generated.resources.jam_error_no_broker
import spotube.composeapp.generated.resources.settings_jam_broker_test_ok import spotube.composeapp.generated.resources.settings_jam_broker_test_ok
import org.koin.core.component.KoinComponent import org.koin.core.component.KoinComponent
import kotlin.random.Random
import kotlin.time.Duration.Companion.seconds import kotlin.time.Duration.Companion.seconds
/** /**
@ -68,6 +70,7 @@ import kotlin.time.Duration.Companion.seconds
* *
* The library keeps the connection alive (auto-reconnect + backoff); this class only * The library keeps the connection alive (auto-reconnect + backoff); this class only
* re-establishes the room subscription and re-publishes presence after a reconnect. * re-establishes the room subscription and re-publishes presence after a reconnect.
* Exponential backoff on reconnect prevents CPU/battery drain on flaky networks (fixes #3186).
*/ */
class JamRoomClient : KoinComponent { class JamRoomClient : KoinComponent {
private val log = Logger.withTag("JamRoomClient") private val log = Logger.withTag("JamRoomClient")
@ -79,9 +82,12 @@ class JamRoomClient : KoinComponent {
} }
private var client: MqttClient? = null private var client: MqttClient? = null
private var reconnectJob: Job? = null
private var roomCode: String? = null private var roomCode: String? = null
private var localPresence: JamPresence? = null private var localPresence: JamPresence? = null
private var currentBroker: JamBroker? = null
private var currentClientId: String = ""
private val _isConnected = MutableStateFlow(false) private val _isConnected = MutableStateFlow(false)
val isConnected: StateFlow<Boolean> = _isConnected.asStateFlow() val isConnected: StateFlow<Boolean> = _isConnected.asStateFlow()
@ -128,6 +134,8 @@ class JamRoomClient : KoinComponent {
if (broker.host.isBlank()) { if (broker.host.isBlank()) {
return Result.failure(IllegalArgumentException(getString(Res.string.jam_error_no_broker))) return Result.failure(IllegalArgumentException(getString(Res.string.jam_error_no_broker)))
} }
currentBroker = broker
currentClientId = clientId
return runCatching { return runCatching {
roomCode = code roomCode = code
localPresence = JamPresence( localPresence = JamPresence(
@ -204,8 +212,12 @@ class JamRoomClient : KoinComponent {
} }
suspend fun disconnect() { suspend fun disconnect() {
reconnectJob?.cancel()
reconnectJob = null
roomCode = null roomCode = null
localPresence = null localPresence = null
currentBroker = null
currentClientId = ""
_isConnected.value = false _isConnected.value = false
_presence.value = emptyMap() _presence.value = emptyMap()
val current = client val current = client
@ -246,6 +258,8 @@ class JamRoomClient : KoinComponent {
is ConnectionState.Connected -> { is ConnectionState.Connected -> {
_isConnected.value = true _isConnected.value = true
_connectionError.value = null _connectionError.value = null
// Reset reconnect attempt counter on successful connection
reconnectAttempt = 0
scope.launch { scope.launch {
// Every connection (initial + reconnects) must re-establish the // Every connection (initial + reconnects) must re-establish the
// broker-side subscription (clean session) and re-publish our // broker-side subscription (clean session) and re-publish our
@ -258,6 +272,10 @@ class JamRoomClient : KoinComponent {
ConnectionState.Disconnected, ConnectionState.Handshaking -> { ConnectionState.Disconnected, ConnectionState.Handshaking -> {
_isConnected.value = false _isConnected.value = false
// Schedule exponential backoff reconnect (fixes #3186 reconnection storm)
if (state == ConnectionState.Disconnected && roomCode != null && currentBroker != null) {
scheduleReconnect()
}
} }
else -> { else -> {
@ -270,6 +288,49 @@ class JamRoomClient : KoinComponent {
} }
} }
private var reconnectAttempt = 0
private const val MAX_RECONNECT_ATTEMPTS = 10
private const val BASE_RECONNECT_DELAY_MS = 1000L
private const val MAX_RECONNECT_DELAY_MS = 30_000L
private fun scheduleReconnect() {
reconnectJob?.cancel()
reconnectJob = scope.launch {
if (reconnectAttempt >= MAX_RECONNECT_ATTEMPTS) {
log.w { "Max reconnect attempts reached, giving up" }
_connectionError.value = getString(Res.string.jam_error_connection_lost)
return@launch
}
val delayMs = (BASE_RECONNECT_DELAY_MS * (2.0.pow(reconnectAttempt))).toLong().coerceAtMost(MAX_RECONNECT_DELAY_MS)
// Add jitter (±20%) to avoid thundering herd
val jitter = (delayMs * 0.2 * (Random.nextDouble() * 2 - 1)).toLong()
val finalDelay = (delayMs + jitter).coerceAtMost(MAX_RECONNECT_DELAY_MS)
log.i { "Scheduling reconnect attempt ${reconnectAttempt + 1} in ${finalDelay}ms" }
delay(finalDelay)
if (roomCode == null || currentBroker == null) return@launch
reconnectAttempt++
val broker = currentBroker!!
val clientId = currentClientId
runCatching {
val mqtt = startClient(broker, clientId)
client = mqtt
mqtt.connectionState
.onEach { onConnectionStateChanged(it) }
.launchIn(scope)
mqtt.connectionState.first { it is ConnectionState.Connected }
}.onFailure { e ->
log.w(e) { "Reconnect attempt $reconnectAttempt failed" }
// Schedule next attempt
scheduleReconnect()
}
}
}
private suspend fun subscribeRoom() { private suspend fun subscribeRoom() {
val mqtt = client ?: return val mqtt = client ?: return
val code = roomCode ?: return val code = roomCode ?: return

View File

@ -31,12 +31,15 @@ import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.Dispatchers import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.Job import kotlinx.coroutines.Job
import kotlinx.coroutines.SupervisorJob import kotlinx.coroutines.SupervisorJob
import kotlinx.coroutines.channels.Channel
import kotlinx.coroutines.delay import kotlinx.coroutines.delay
import kotlinx.coroutines.flow.MutableStateFlow import kotlinx.coroutines.flow.MutableStateFlow
import kotlinx.coroutines.flow.SharingStarted import kotlinx.coroutines.flow.SharingStarted
import kotlinx.coroutines.flow.StateFlow import kotlinx.coroutines.flow.StateFlow
import kotlinx.coroutines.flow.asStateFlow import kotlinx.coroutines.flow.asStateFlow
import kotlinx.coroutines.flow.combine import kotlinx.coroutines.flow.combine
import kotlinx.coroutines.flow.debounce
import kotlinx.coroutines.flow.distinctUntilChanged
import kotlinx.coroutines.flow.launchIn import kotlinx.coroutines.flow.launchIn
import kotlinx.coroutines.flow.mapNotNull import kotlinx.coroutines.flow.mapNotNull
import kotlinx.coroutines.flow.onEach import kotlinx.coroutines.flow.onEach
@ -48,6 +51,7 @@ import spotube.composeapp.generated.resources.Res
import spotube.composeapp.generated.resources.jam_error_invalid_code import spotube.composeapp.generated.resources.jam_error_invalid_code
import spotube.composeapp.generated.resources.jam_error_no_broker import spotube.composeapp.generated.resources.jam_error_no_broker
import kotlin.time.Clock import kotlin.time.Clock
import kotlin.time.Duration
/** /**
* A jam session over MQTT (star topology, host-authoritative queue). * A jam session over MQTT (star topology, host-authoritative queue).
@ -57,7 +61,8 @@ import kotlin.time.Clock
* from anyone are applied by the host, then broadcast via the retained state topic. * from anyone are applied by the host, then broadcast via the retained state topic.
* - Play/pause is global: anyone's play/pause is sent to the host, which applies it * - Play/pause is global: anyone's play/pause is sent to the host, which applies it
* and broadcasts it; guests follow every change of the host's play state. * and broadcasts it; guests follow every change of the host's play state.
* - Seek, volume and loop are local to each device — never broadcast. * - Seek is now global: host broadcasts currentPositionMs; guests seek on resume/sync.
* - Volume and loop are local to each device — never broadcast.
* - Shuffle is host-only; guests mirror the host's shuffle setting. * - Shuffle is host-only; guests mirror the host's shuffle setting.
* - Guests can only add to the queue (suggest); the host applies suggestions. * - Guests can only add to the queue (suggest); the host applies suggestions.
* - If the host leaves, the participant with the lowest client id takes over. * - If the host leaves, the participant with the lowest client id takes over.
@ -66,7 +71,9 @@ private data class PlaybackBroadcast(
val queue: List<QueueEntry>, val queue: List<QueueEntry>,
val current: QueueEntry?, val current: QueueEntry?,
val shuffle: Boolean, val shuffle: Boolean,
val isPlaying: Boolean) val isPlaying: Boolean,
val positionMs: Long
)
class JamRoomService( class JamRoomService(
private val jamClient: JamRoomClient, private val jamClient: JamRoomClient,
@ -76,6 +83,9 @@ class JamRoomService(
private val log = Logger.withTag("JamRoomService") private val log = Logger.withTag("JamRoomService")
private val scope = CoroutineScope(SupervisorJob() + Dispatchers.Default) private val scope = CoroutineScope(SupervisorJob() + Dispatchers.Default)
// Throttle host broadcasts to max 1 per 500ms to reduce MQTT traffic & CPU (fixes #3186)
private val hostBroadcastChannel = Channel<PlaybackBroadcast>(capacity = 10)
/** /**
* Play/pause intent. Transient states (buffering between tracks, loading, completion) * Play/pause intent. Transient states (buffering between tracks, loading, completion)
* keep the last value, so they never pause or resume the room. * keep the last value, so they never pause or resume the room.
@ -120,6 +130,7 @@ class JamRoomService(
get() = localClientId get() = localClientId
private var hostBroadcastJob: Job? = null private var hostBroadcastJob: Job? = null
private var hostBroadcastProcessorJob: Job? = null
/** Guest side: last queue snapshot applied to the local player. */ /** Guest side: last queue snapshot applied to the local player. */
private var lastAppliedItems: List<JamMediaItem> = emptyList() private var lastAppliedItems: List<JamMediaItem> = emptyList()
@ -135,6 +146,8 @@ class JamRoomService(
/** Guest side: the room's play state from the last host broadcast (null until the first one). */ /** Guest side: the room's play state from the last host broadcast (null until the first one). */
private var lastHostIsPlaying: Boolean? = null private var lastHostIsPlaying: Boolean? = null
/** Guest side: last known host position to avoid redundant seeks. */
private var lastHostPositionMs: Long = -1
/** Host side: client ids banned for this session. */ /** Host side: client ids banned for this session. */
private val bannedClientIds = mutableSetOf<String>() private val bannedClientIds = mutableSetOf<String>()
@ -202,7 +215,10 @@ class JamRoomService(
leaving = false leaving = false
autoAdvanceDeadlineMs = 0L autoAdvanceDeadlineMs = 0L
lastHostIsPlaying = null lastHostIsPlaying = null
lastHostPositionMs = -1
startHostBroadcast() startHostBroadcast()
// IMMEDIATELY broadcast current playback state so guests sync instantly (fixes #3184)
broadcastCurrentStateImmediately()
persistLastCode(code) persistLastCode(code)
code code
} }
@ -234,6 +250,7 @@ class JamRoomService(
lastAppliedIndex = -1 lastAppliedIndex = -1
leaving = false leaving = false
lastHostIsPlaying = null lastHostIsPlaying = null
lastHostPositionMs = -1
persistLastCode(normalized) persistLastCode(normalized)
} }
} }
@ -252,6 +269,7 @@ class JamRoomService(
lastAppliedIndex = -1 lastAppliedIndex = -1
bannedClientIds.clear() bannedClientIds.clear()
lastHostIsPlaying = null lastHostIsPlaying = null
lastHostPositionMs = -1
} }
// ---------- Controls (called from the UI) ---------- // ---------- Controls (called from the UI) ----------
@ -308,38 +326,90 @@ class JamRoomService(
private fun startHostBroadcast() { private fun startHostBroadcast() {
if (hostBroadcastJob?.isActive == true) return if (hostBroadcastJob?.isActive == true) return
// Producer: collect playback state changes and send to channel (debounced)
hostBroadcastJob = scope.launch { hostBroadcastJob = scope.launch {
combine( combine(
audioPlayerQueue.queueFlow, audioPlayerQueue.queueFlow,
audioPlayerQueue.currentQueueEntryFlow, audioPlayerQueue.currentQueueEntryFlow,
audioPlayer.shuffleModeFlow, audioPlayer.shuffleModeFlow,
playIntent) { queue, current, shuffle, isPlaying -> playIntent,
PlaybackBroadcast(queue, current, shuffle, isPlaying) audioPlayer.positionFlow) { queue, current, shuffle, isPlaying, position ->
PlaybackBroadcast(queue, current, shuffle, isPlaying, position.inWholeMilliseconds)
} }
.distinctUntilChanged { old, new ->
// Only emit if meaningfully changed: queue content, index, play state, or position > 1s diff
old.queue == new.queue &&
old.current == new.current &&
old.shuffle == new.shuffle &&
old.isPlaying == new.isPlaying &&
kotlin.math.abs(old.positionMs - new.positionMs) < 1000
}
.debounce(500) // max 2 broadcasts/sec (fixes #3186 overheating)
.onEach { broadcast -> .onEach { broadcast ->
if (_role.value != JamRole.Host) return@onEach if (_role.value != JamRole.Host) return@onEach
val index = if (broadcast.current != null) { hostBroadcastChannel.trySend(broadcast)
broadcast.queue.indexOfFirst { it.matchesEntry(broadcast.current) }
} else {
-1
}
_shuffleEnabled.value = broadcast.shuffle
jamClient.publishState(
JamMessage.QueueState(
items = broadcast.queue.map(JamMediaItem::fromQueueEntry),
currentIndex = index.coerceAtLeast(0),
shuffleEnabled = broadcast.shuffle,
follow = now() > autoAdvanceDeadlineMs,
isPlaying = broadcast.isPlaying)
)
} }
.launchIn(this) .launchIn(this)
} }
// Consumer: process broadcasts sequentially, serialize MQTT publishes
hostBroadcastProcessorJob = scope.launch {
for (broadcast in hostBroadcastChannel) {
if (_role.value != JamRole.Host) break
val index = if (broadcast.current != null) {
broadcast.queue.indexOfFirst { it.matchesEntry(broadcast.current) }
} else {
-1
}
_shuffleEnabled.value = broadcast.shuffle
jamClient.publishState(
JamMessage.QueueState(
items = broadcast.queue.map(JamMediaItem::fromQueueEntry),
currentIndex = index.coerceAtLeast(0),
shuffleEnabled = broadcast.shuffle,
follow = now() > autoAdvanceDeadlineMs,
isPlaying = broadcast.isPlaying,
currentPositionMs = broadcast.positionMs)
)
}
}
} }
private fun stopHostBroadcast() { private fun stopHostBroadcast() {
hostBroadcastJob?.cancel() hostBroadcastJob?.cancel()
hostBroadcastJob = null hostBroadcastJob = null
hostBroadcastProcessorJob?.cancel()
hostBroadcastProcessorJob = null
hostBroadcastChannel.close()
}
/** Immediately broadcast current state (used on room creation and host takeover) */
private fun broadcastCurrentStateImmediately() {
scope.launch {
val currentQueue = audioPlayerQueue.queueFlow.value
val currentEntry = audioPlayerQueue.currentQueueEntryFlow.value
val shuffle = audioPlayer.shuffleModeFlow.value
val isPlaying = audioPlayer.playerStateFlow.value == PlayerState.PLAYING
val positionMs = audioPlayer.positionFlow.value.inWholeMilliseconds
if (currentEntry != null) {
val index = currentQueue.indexOfFirst { it.matchesEntry(currentEntry) }
jamClient.publishState(
JamMessage.QueueState(
items = currentQueue.map(JamMediaItem::fromQueueEntry),
currentIndex = index.coerceAtLeast(0),
shuffleEnabled = shuffle,
follow = true,
isPlaying = isPlaying,
currentPositionMs = positionMs)
)
// If playing, also send explicit seek to ensure guests start at exact position
if (isPlaying) {
jamClient.publishCommand(JamMessage.PlaybackCommand(PlaybackCmd.SeekTo(positionMs)))
}
}
}
} }
// ---------- Guest: apply remote state ---------- // ---------- Guest: apply remote state ----------
@ -351,13 +421,28 @@ class JamRoomService(
_shuffleEnabled.value = state.shuffleEnabled _shuffleEnabled.value = state.shuffleEnabled
runCatching { audioPlayer.shuffle(state.shuffleEnabled) } runCatching { audioPlayer.shuffle(state.shuffleEnabled) }
// Play/pause is room-wide: follow every change of the host's play state. The room // Play/pause is room-wide: follow every change of the host's play state.
// state is recorded first, so the resulting local intent isn't sent back as a command. // On resume (isPlaying becomes true), seek to host's currentPositionMs first (fixes #3181).
if (state.isPlaying != lastHostIsPlaying) { if (state.isPlaying != lastHostIsPlaying) {
lastHostIsPlaying = state.isPlaying lastHostIsPlaying = state.isPlaying
if (state.isPlaying != playIntent.value) { if (state.isPlaying) {
runCatching { if (state.isPlaying) audioPlayer.play() else audioPlayer.pause() } // Host resumed: seek to their position BEFORE playing
.onFailure { log.w(it) { "Failed to follow the host's play state" } } if (state.currentPositionMs > 0 && state.currentPositionMs != lastHostPositionMs) {
lastHostPositionMs = state.currentPositionMs
runCatching {
audioPlayer.seekTo(Duration.milliseconds(state.currentPositionMs))
}.onFailure { log.w(it) { "Failed to seek to host position on resume" } }
}
if (state.isPlaying != playIntent.value) {
runCatching { audioPlayer.play() }
.onFailure { log.w(it) { "Failed to follow the host's play state" } }
}
} else {
// Host paused: just pause locally
if (state.isPlaying != playIntent.value) {
runCatching { audioPlayer.pause() }
.onFailure { log.w(it) { "Failed to follow the host's pause" } }
}
} }
} }
@ -404,6 +489,17 @@ class JamRoomService(
audioPlayerQueue.jumpTo(state.currentIndex.coerceAtLeast(0), autoPlay = false) audioPlayerQueue.jumpTo(state.currentIndex.coerceAtLeast(0), autoPlay = false)
}.onFailure { log.w(it) { "Failed to follow jam queue index" } } }.onFailure { log.w(it) { "Failed to follow jam queue index" } }
} }
// Periodic position sync: if host position drifted > 2s, correct it (fixes #3181 drift)
if (state.isPlaying && state.currentPositionMs > 0) {
val guestPosition = audioPlayer.positionFlow.value.inWholeMilliseconds
if (kotlin.math.abs(guestPosition - state.currentPositionMs) > 2000) {
lastHostPositionMs = state.currentPositionMs
runCatching {
audioPlayer.seekTo(Duration.milliseconds(state.currentPositionMs))
}.onFailure { log.w(it) { "Failed to correct position drift" } }
}
}
} }
// ---------- Incoming commands ---------- // ---------- Incoming commands ----------
@ -447,6 +543,9 @@ class JamRoomService(
PlaybackCmd.SkipNext -> runCatching { audioPlayer.skipToNext() } PlaybackCmd.SkipNext -> runCatching { audioPlayer.skipToNext() }
PlaybackCmd.SkipPrevious -> runCatching { audioPlayer.skipToPrevious() } PlaybackCmd.SkipPrevious -> runCatching { audioPlayer.skipToPrevious() }
is PlaybackCmd.JumpTo -> runCatching { audioPlayer.jumpTo(command.index) } is PlaybackCmd.JumpTo -> runCatching { audioPlayer.jumpTo(command.index) }
is PlaybackCmd.SeekTo -> runCatching {
audioPlayer.seekTo(Duration.milliseconds(command.positionMs))
}
} }
} }
@ -490,6 +589,7 @@ class JamRoomService(
_role.value = JamRole.Host _role.value = JamRole.Host
jamClient.claimHost() jamClient.claimHost()
startHostBroadcast() startHostBroadcast()
broadcastCurrentStateImmediately()
} }
} }
@ -537,8 +637,6 @@ class JamRoomService(
MetadataArtist.Basic(id = "", name = artist, thumbnails = emptyList(), externalUri = null) MetadataArtist.Basic(id = "", name = artist, thumbnails = emptyList(), externalUri = null)
), ),
album = null, album = null,
// Keep the cover art flowing to participants — the wire carries it
// as coverUrl, so mirror it back into the reconstructed thumbnails.
thumbnails = coverUrl.takeIf { it.isNotBlank() } thumbnails = coverUrl.takeIf { it.isNotBlank() }
?.let { url -> listOf(Thumbnail(url = url, width = 0, height = 0)) }, ?.let { url -> listOf(Thumbnail(url = url, width = 0, height = 0)) },
explicit = null, explicit = null,