diff --git a/composeApp/src/commonMain/composeResources/values/strings.xml b/composeApp/src/commonMain/composeResources/values/strings.xml
index 2fbae5e0..364d2231 100644
--- a/composeApp/src/commonMain/composeResources/values/strings.xml
+++ b/composeApp/src/commonMain/composeResources/values/strings.xml
@@ -319,6 +319,8 @@
Password (optional)
Client ID prefix
Used to identify this Spotube instance
+ Participant name
+ Your display name in jam rooms
Encrypt the broker connection
Check the current broker settings before starting a room.
Test connection
diff --git a/composeApp/src/commonMain/kotlin/org/spotube/spotube/core/jam/JamProtocol.kt b/composeApp/src/commonMain/kotlin/org/spotube/spotube/core/jam/JamProtocol.kt
index 03231e66..afa9309c 100644
--- a/composeApp/src/commonMain/kotlin/org/spotube/spotube/core/jam/JamProtocol.kt
+++ b/composeApp/src/commonMain/kotlin/org/spotube/spotube/core/jam/JamProtocol.kt
@@ -46,7 +46,9 @@ sealed class JamMessage {
*/
val follow: Boolean = false,
/** 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
@SerialName("playbackCommand")
@@ -96,6 +98,10 @@ sealed class PlaybackCmd {
@Serializable
@SerialName("jumpTo")
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). */
diff --git a/composeApp/src/commonMain/kotlin/org/spotube/spotube/core/jam/JamRoomClient.kt b/composeApp/src/commonMain/kotlin/org/spotube/spotube/core/jam/JamRoomClient.kt
index 07810202..f3312f58 100644
--- a/composeApp/src/commonMain/kotlin/org/spotube/spotube/core/jam/JamRoomClient.kt
+++ b/composeApp/src/commonMain/kotlin/org/spotube/spotube/core/jam/JamRoomClient.kt
@@ -32,6 +32,7 @@ import com.ditchoom.mqtt5.controlpacket.ConnectionRequest
import org.spotube.spotube.modules.settings.JamBroker
import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.Job
import kotlinx.coroutines.SupervisorJob
import kotlinx.coroutines.TimeoutCancellationException
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.settings_jam_broker_test_ok
import org.koin.core.component.KoinComponent
+import kotlin.random.Random
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
* 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 {
private val log = Logger.withTag("JamRoomClient")
@@ -79,9 +82,12 @@ class JamRoomClient : KoinComponent {
}
private var client: MqttClient? = null
+ private var reconnectJob: Job? = null
private var roomCode: String? = null
private var localPresence: JamPresence? = null
+ private var currentBroker: JamBroker? = null
+ private var currentClientId: String = ""
private val _isConnected = MutableStateFlow(false)
val isConnected: StateFlow = _isConnected.asStateFlow()
@@ -128,6 +134,8 @@ class JamRoomClient : KoinComponent {
if (broker.host.isBlank()) {
return Result.failure(IllegalArgumentException(getString(Res.string.jam_error_no_broker)))
}
+ currentBroker = broker
+ currentClientId = clientId
return runCatching {
roomCode = code
localPresence = JamPresence(
@@ -204,8 +212,12 @@ class JamRoomClient : KoinComponent {
}
suspend fun disconnect() {
+ reconnectJob?.cancel()
+ reconnectJob = null
roomCode = null
localPresence = null
+ currentBroker = null
+ currentClientId = ""
_isConnected.value = false
_presence.value = emptyMap()
val current = client
@@ -246,6 +258,8 @@ class JamRoomClient : KoinComponent {
is ConnectionState.Connected -> {
_isConnected.value = true
_connectionError.value = null
+ // Reset reconnect attempt counter on successful connection
+ reconnectAttempt = 0
scope.launch {
// Every connection (initial + reconnects) must re-establish the
// broker-side subscription (clean session) and re-publish our
@@ -258,6 +272,10 @@ class JamRoomClient : KoinComponent {
ConnectionState.Disconnected, ConnectionState.Handshaking -> {
_isConnected.value = false
+ // Schedule exponential backoff reconnect (fixes #3186 reconnection storm)
+ if (state == ConnectionState.Disconnected && roomCode != null && currentBroker != null) {
+ scheduleReconnect()
+ }
}
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() {
val mqtt = client ?: return
val code = roomCode ?: return
diff --git a/composeApp/src/commonMain/kotlin/org/spotube/spotube/core/jam/JamRoomService.kt b/composeApp/src/commonMain/kotlin/org/spotube/spotube/core/jam/JamRoomService.kt
index f317e6ec..0f340b50 100644
--- a/composeApp/src/commonMain/kotlin/org/spotube/spotube/core/jam/JamRoomService.kt
+++ b/composeApp/src/commonMain/kotlin/org/spotube/spotube/core/jam/JamRoomService.kt
@@ -31,12 +31,15 @@ import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.Job
import kotlinx.coroutines.SupervisorJob
+import kotlinx.coroutines.channels.Channel
import kotlinx.coroutines.delay
import kotlinx.coroutines.flow.MutableStateFlow
import kotlinx.coroutines.flow.SharingStarted
import kotlinx.coroutines.flow.StateFlow
import kotlinx.coroutines.flow.asStateFlow
import kotlinx.coroutines.flow.combine
+import kotlinx.coroutines.flow.debounce
+import kotlinx.coroutines.flow.distinctUntilChanged
import kotlinx.coroutines.flow.launchIn
import kotlinx.coroutines.flow.mapNotNull
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_no_broker
import kotlin.time.Clock
+import kotlin.time.Duration
/**
* 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.
* - 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.
- * - 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.
* - 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.
@@ -66,7 +71,9 @@ private data class PlaybackBroadcast(
val queue: List,
val current: QueueEntry?,
val shuffle: Boolean,
- val isPlaying: Boolean)
+ val isPlaying: Boolean,
+ val positionMs: Long
+)
class JamRoomService(
private val jamClient: JamRoomClient,
@@ -76,6 +83,9 @@ class JamRoomService(
private val log = Logger.withTag("JamRoomService")
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(capacity = 10)
+
/**
* Play/pause intent. Transient states (buffering between tracks, loading, completion)
* keep the last value, so they never pause or resume the room.
@@ -120,6 +130,7 @@ class JamRoomService(
get() = localClientId
private var hostBroadcastJob: Job? = null
+ private var hostBroadcastProcessorJob: Job? = null
/** Guest side: last queue snapshot applied to the local player. */
private var lastAppliedItems: List = emptyList()
@@ -135,6 +146,8 @@ class JamRoomService(
/** Guest side: the room's play state from the last host broadcast (null until the first one). */
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. */
private val bannedClientIds = mutableSetOf()
@@ -202,7 +215,10 @@ class JamRoomService(
leaving = false
autoAdvanceDeadlineMs = 0L
lastHostIsPlaying = null
+ lastHostPositionMs = -1
startHostBroadcast()
+ // IMMEDIATELY broadcast current playback state so guests sync instantly (fixes #3184)
+ broadcastCurrentStateImmediately()
persistLastCode(code)
code
}
@@ -234,6 +250,7 @@ class JamRoomService(
lastAppliedIndex = -1
leaving = false
lastHostIsPlaying = null
+ lastHostPositionMs = -1
persistLastCode(normalized)
}
}
@@ -252,6 +269,7 @@ class JamRoomService(
lastAppliedIndex = -1
bannedClientIds.clear()
lastHostIsPlaying = null
+ lastHostPositionMs = -1
}
// ---------- Controls (called from the UI) ----------
@@ -308,38 +326,90 @@ class JamRoomService(
private fun startHostBroadcast() {
if (hostBroadcastJob?.isActive == true) return
+
+ // Producer: collect playback state changes and send to channel (debounced)
hostBroadcastJob = scope.launch {
combine(
audioPlayerQueue.queueFlow,
audioPlayerQueue.currentQueueEntryFlow,
audioPlayer.shuffleModeFlow,
- playIntent) { queue, current, shuffle, isPlaying ->
- PlaybackBroadcast(queue, current, shuffle, isPlaying)
+ playIntent,
+ 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 ->
if (_role.value != JamRole.Host) return@onEach
- 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)
- )
+ hostBroadcastChannel.trySend(broadcast)
}
.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() {
hostBroadcastJob?.cancel()
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 ----------
@@ -351,13 +421,28 @@ class JamRoomService(
_shuffleEnabled.value = state.shuffleEnabled
runCatching { audioPlayer.shuffle(state.shuffleEnabled) }
- // Play/pause is room-wide: follow every change of the host's play state. The room
- // state is recorded first, so the resulting local intent isn't sent back as a command.
+ // Play/pause is room-wide: follow every change of the host's play state.
+ // On resume (isPlaying becomes true), seek to host's currentPositionMs first (fixes #3181).
if (state.isPlaying != lastHostIsPlaying) {
lastHostIsPlaying = state.isPlaying
- if (state.isPlaying != playIntent.value) {
- runCatching { if (state.isPlaying) audioPlayer.play() else audioPlayer.pause() }
- .onFailure { log.w(it) { "Failed to follow the host's play state" } }
+ if (state.isPlaying) {
+ // Host resumed: seek to their position BEFORE playing
+ 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)
}.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 ----------
@@ -447,6 +543,9 @@ class JamRoomService(
PlaybackCmd.SkipNext -> runCatching { audioPlayer.skipToNext() }
PlaybackCmd.SkipPrevious -> runCatching { audioPlayer.skipToPrevious() }
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
jamClient.claimHost()
startHostBroadcast()
+ broadcastCurrentStateImmediately()
}
}
@@ -537,8 +637,6 @@ class JamRoomService(
MetadataArtist.Basic(id = "", name = artist, thumbnails = emptyList(), externalUri = 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() }
?.let { url -> listOf(Thumbnail(url = url, width = 0, height = 0)) },
explicit = null,
@@ -587,4 +685,4 @@ class JamRoomService(
private const val HOST_TAKEOVER_DELAY_MS = 1_500L
private const val AUTO_ADVANCE_WINDOW_MS = 2_000L
}
-}
+}
\ No newline at end of file
diff --git a/composeApp/src/commonMain/kotlin/org/spotube/spotube/modules/jam/JamViewModel.kt b/composeApp/src/commonMain/kotlin/org/spotube/spotube/modules/jam/JamViewModel.kt
index 5041e371..939660f2 100644
--- a/composeApp/src/commonMain/kotlin/org/spotube/spotube/modules/jam/JamViewModel.kt
+++ b/composeApp/src/commonMain/kotlin/org/spotube/spotube/modules/jam/JamViewModel.kt
@@ -102,7 +102,9 @@ class JamViewModel(
codeCopied,
snapshotFlow { codeInput.text.toString() },
participantMenu,
- isLeaving
+ isLeaving,
+ // Local client ID to exclude self from moderation menu
+ jamRoomService.participantClientId
) { values ->
@Suppress("UNCHECKED_CAST")
val role = values[0] as JamRole?
@@ -116,6 +118,7 @@ class JamViewModel(
val localError = values[6] as String?
val busy = values[8] as Boolean
val brokerConfigured = !settings?.jamBroker?.host.isNullOrBlank()
+ val localClientId = values[13] as String
JamUiState(
isActive = role != null,
@@ -127,7 +130,8 @@ class JamViewModel(
displayName = participant.displayName,
initial = participant.displayName.firstOrNull()?.uppercase() ?: "?",
isHost = participant.isHost,
- canModerate = role == JamRole.Host && !participant.isHost,
+ // Host can moderate OTHER participants only (not self)
+ canModerate = role == JamRole.Host && !participant.isHost && participant.id != localClientId,
menuExpanded = values[11] == participant.id
)
},
diff --git a/composeApp/src/commonMain/kotlin/org/spotube/spotube/modules/settings/JamSettingsScreen.kt b/composeApp/src/commonMain/kotlin/org/spotube/spotube/modules/settings/JamSettingsScreen.kt
index 0643cb4b..ac107939 100644
--- a/composeApp/src/commonMain/kotlin/org/spotube/spotube/modules/settings/JamSettingsScreen.kt
+++ b/composeApp/src/commonMain/kotlin/org/spotube/spotube/modules/settings/JamSettingsScreen.kt
@@ -58,6 +58,8 @@ import spotube.composeapp.generated.resources.settings_jam_broker_content_descri
import spotube.composeapp.generated.resources.settings_jam_broker_username_placeholder
import spotube.composeapp.generated.resources.settings_jam_broker_test_fail
import spotube.composeapp.generated.resources.jam_error_connection_failed
+import spotube.composeapp.generated.resources.settings_jam_participant_name
+import spotube.composeapp.generated.resources.settings_jam_participant_name_placeholder
@Composable
fun JamSettingsScreen(
@@ -197,6 +199,23 @@ internal fun JamSettingsContent(
settingsSectionHeader(Res.string.settings_jam_section_identity)
settingsSectionCard(
items = listOf(
+ {
+ TextInputSettingCard(
+ title = stringResource(Res.string.settings_jam_participant_name),
+ value = uiState.settings.jamParticipantName,
+ placeholder = stringResource(Res.string.settings_jam_participant_name_placeholder),
+ onValueSaved = { name ->
+ viewModel.update {
+ copy(jamParticipantName = name)
+ }
+ },
+ icon = {
+ SettingsItemIcon(
+ Iconsax.User,
+ stringResource(Res.string.settings_jam_participant_name)
+ )
+ })
+ },
{
TextInputSettingCard(
title = stringResource(Res.string.settings_jam_broker_client_id),
@@ -260,4 +279,4 @@ internal fun JamSettingsContent(
}
}
}
-}
+}
\ No newline at end of file