From 4f717709d2e92e5b70197dd2d34d5a9d5c02f98b Mon Sep 17 00:00:00 2001 From: weishaupt Date: Sun, 4 Oct 2026 16:29:21 -0400 Subject: [PATCH] Fix scrollback: burst-proof stick-to-bottom, cap-proof layout wait, serialized buffer updates, atomic history saves --- .../kotlin/data/MessageHistoryRepository.kt | 22 +++- src/main/kotlin/irc/ConnectionManager.kt | 116 +++++++++++------- src/main/kotlin/ui/ChannelScreen.kt | 88 +++++++------ 3 files changed, 140 insertions(+), 86 deletions(-) diff --git a/src/main/kotlin/data/MessageHistoryRepository.kt b/src/main/kotlin/data/MessageHistoryRepository.kt index b5ca645..ef6925c 100644 --- a/src/main/kotlin/data/MessageHistoryRepository.kt +++ b/src/main/kotlin/data/MessageHistoryRepository.kt @@ -3,6 +3,9 @@ package data import org.json.JSONArray import org.json.JSONObject import java.io.File +import java.nio.file.AtomicMoveNotSupportedException +import java.nio.file.Files +import java.nio.file.StandardCopyOption /** * Persists each buffer's scrollback to a JSON file under rootDir/history/, so messages @@ -25,14 +28,31 @@ class MessageHistoryRepository(private val rootDir: File) { val arr = JSONArray(file.readText()) (0 until arr.length()).map { i -> fromJson(arr.getJSONObject(i)) } } catch (e: Exception) { + // Rename the bad file aside rather than silently returning empty - returning + // empty here looks to the caller exactly like "no history yet", and the next + // debounced save would then happily overwrite - and permanently lose - whatever + // was in the corrupt file. + file.renameTo(File(file.parentFile, "${file.name}.corrupt")) emptyList() } } + // @Synchronized because flushAllSaves (on shutdown) calls this directly for every + // buffer while a debounced save for the same key can still be in flight on its own + // coroutine - without this, the two writes race and whichever's atomic move lands + // second wins, which isn't necessarily the one with the more complete message list. + @Synchronized fun saveMessages(key: BufferKey, messages: List) { val arr = JSONArray() messages.takeLast(500).forEach { arr.put(toJson(it)) } - fileFor(key).writeText(arr.toString()) + val file = fileFor(key) + val tmp = File(file.parentFile, "${file.name}.tmp") + tmp.writeText(arr.toString()) + try { + Files.move(tmp.toPath(), file.toPath(), StandardCopyOption.REPLACE_EXISTING, StandardCopyOption.ATOMIC_MOVE) + } catch (e: AtomicMoveNotSupportedException) { + Files.move(tmp.toPath(), file.toPath(), StandardCopyOption.REPLACE_EXISTING) + } } private fun toJson(m: IrcMessage): JSONObject = JSONObject().apply { diff --git a/src/main/kotlin/irc/ConnectionManager.kt b/src/main/kotlin/irc/ConnectionManager.kt index 212c383..1c5b5d6 100644 --- a/src/main/kotlin/irc/ConnectionManager.kt +++ b/src/main/kotlin/irc/ConnectionManager.kt @@ -5,6 +5,7 @@ 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.MutableSharedFlow import kotlinx.coroutines.flow.MutableStateFlow @@ -12,8 +13,6 @@ import kotlinx.coroutines.flow.SharedFlow import kotlinx.coroutines.flow.StateFlow import kotlinx.coroutines.flow.update import kotlinx.coroutines.launch -import kotlinx.coroutines.sync.Mutex -import kotlinx.coroutines.sync.withLock import java.io.File import java.util.concurrent.ConcurrentHashMap @@ -63,6 +62,35 @@ object ConnectionManager { private val _newQueryBuffers = MutableSharedFlow(extraBufferCapacity = 16) val newQueryBuffers: SharedFlow = _newQueryBuffers + // Every buffer state change - incoming lines, local echoes of our own sends - goes + // through this single FIFO queue, processed by one consumer coroutine. Previously each + // addMessage ran as its own independently-launched coroutine on Dispatchers.IO, guarded + // only by a per-key Mutex; a Mutex makes each key's critical section exclusive but gives + // no ordering guarantee between separate launches racing to acquire it, so lines in a + // busy channel could commit to the buffer out of order. Routing everything through one + // queue with one consumer makes "posted first, applied first" an actual guarantee instead + // of a usual case, and lets addMessage/updateUsers/the TOPIC handler go back to being + // plain synchronous code - they're never called from anywhere except this one coroutine. + private val bufferOps = Channel<() -> Unit>(Channel.UNLIMITED) + + init { + scope.launch(Dispatchers.IO) { + for (op in bufferOps) { + try { + op() + } catch (e: Exception) { + // One bad op must not take down the consumer loop - that would silently + // stop ALL future buffer updates for the rest of the process. + e.printStackTrace() + } + } + } + } + + private fun enqueue(op: () -> Unit) { + bufferOps.trySend(op) + } + fun connect(config: ServerConfig) { // TEMP DEBUG (see task: on-connect identify command investigation) - confirms // whether the config this connect() call actually received has the on-connect @@ -84,7 +112,7 @@ object ConnectionManager { val conn = IrcConnection( config = config, scope = scope, - onLine = { line -> handleLine(config.id, line) }, + onLine = { line -> enqueue { handleLine(config.id, line) } }, onStatus = { msg -> _statuses.update { it + (config.id to msg) } } ) connections[config.id] = conn @@ -105,10 +133,9 @@ object ConnectionManager { fun sendChannelMessage(serverId: String, target: String, text: String) { connectionFor(serverId)?.sendMessage(target, text) val myNick = ownNicks[serverId] ?: return - addMessage( - BufferKey(serverId, target), - IrcMessage(System.currentTimeMillis(), MessageKind.CHAT, myNick, text) - ) + val key = BufferKey(serverId, target) + val msg = IrcMessage(System.currentTimeMillis(), MessageKind.CHAT, myNick, text) + enqueue { addMessage(key, msg) } } // Same local-echo problem as sendChannelMessage, but for /me: the wire form is a @@ -117,10 +144,9 @@ object ConnectionManager { fun sendChannelAction(serverId: String, target: String, action: String) { connectionFor(serverId)?.sendMessage(target, "ACTION $action") val myNick = ownNicks[serverId] ?: return - addMessage( - BufferKey(serverId, target), - IrcMessage(System.currentTimeMillis(), MessageKind.ACTION, myNick, action) - ) + val key = BufferKey(serverId, target) + val msg = IrcMessage(System.currentTimeMillis(), MessageKind.ACTION, myNick, action) + enqueue { addMessage(key, msg) } } // Raw passthrough for slash commands that don't have dedicated handling. @@ -131,46 +157,40 @@ object ConnectionManager { // Local-only feedback (e.g. the /reload confirmation) that never touches the // wire - just appended to the buffer the same way any other SYSTEM line is. fun addSystemMessage(serverId: String, target: String, text: String) { - addMessage(BufferKey(serverId, target), IrcMessage(System.currentTimeMillis(), MessageKind.SYSTEM, null, text)) + val key = BufferKey(serverId, target) + val msg = IrcMessage(System.currentTimeMillis(), MessageKind.SYSTEM, null, text) + enqueue { addMessage(key, msg) } } - // One mutex per buffer key, guarding the "is this buffer new -> seed its history from - // disk if so -> append the new message" sequence for that key end to end. Previously - // that sequence was: a synchronous isNewBuffer read, then a separate async coroutine - // that did the disk seed-load and the in-memory update - two steps with a real window - // between them. In a busy channel, several messages for a brand-new buffer (e.g. right - // after JOIN, when NAMES/TOPIC/a flood of chat lines all land within milliseconds) could - // each see isNewBuffer=true before any of them had finished writing, each independently - // re-reading the same stale on-disk snapshot and racing to write - concretely wasteful - // (redundant disk reads) and, since MessageHistoryRepository.saveMessages() does a plain - // unsynchronized file.writeText() with no ordering guarantee between concurrent callers, - // capable of persisting a stale/incomplete snapshot last even when the in-memory state - // was already fully correct. A per-key Mutex makes the whole check-seed-append sequence - // for a given key strictly one-at-a-time, so only the genuinely first caller ever seeds - // from disk and no two writers for the same key can be in that window simultaneously. - private val bufferMutexes = ConcurrentHashMap() - private fun mutexFor(key: BufferKey): Mutex = bufferMutexes.computeIfAbsent(key) { Mutex() } - // Pending debounced saves, keyed by buffer - see scheduleSave(). private val pendingSaveJobs = ConcurrentHashMap() + // If this key's buffer doesn't exist yet, seeds it from disk history before anything + // else touches it. Every code path that can create a buffer (addMessage, updateUsers, + // the TOPIC handler) goes through this first, so a buffer's first appearance in + // _buffers always carries its persisted history - never an empty one that a later + // debounced save would then overwrite the real history file with. This only ever runs + // on the single bufferOps consumer coroutine, so the "does it exist" check and the + // insert-with-seeded-history that follows can't race with another caller doing the + // same thing for the same key. + private fun ensureBuffer(key: BufferKey) { + if (_buffers.value.containsKey(key)) return + val seeded = historyRepo?.loadMessages(key) ?: emptyList() + _buffers.update { map -> map + (key to ChannelBuffer(key, messages = seeded)) } + if (!isChannelName(key.name)) { + // tryEmit, not emit: this runs on the single, non-suspend bufferOps consumer, + // not its own coroutine, so there's nothing to suspend into. The SharedFlow's + // 16-slot extraBufferCapacity means this can't actually fail in practice. + _newQueryBuffers.tryEmit(key) + } + } + private fun addMessage(key: BufferKey, msg: IrcMessage) { - // Do the file I/O (seeding a brand-new buffer from disk, then persisting the - // updated list) off the caller's thread - addMessage gets called both from the - // IO-dispatched read loop and directly from UI-thread local-echo sends. - scope.launch(Dispatchers.IO) { - mutexFor(key).withLock { - val isNewBuffer = !_buffers.value.containsKey(key) - val seeded = if (isNewBuffer) (historyRepo?.loadMessages(key) ?: emptyList()) else emptyList() - _buffers.update { map -> - val existing = map[key] ?: ChannelBuffer(key, messages = seeded) - val trimmed = (existing.messages + msg).takeLast(MESSAGE_HISTORY_CAP) - map + (key to existing.copy(messages = trimmed)) - } - if (isNewBuffer && !isChannelName(key.name)) { - _newQueryBuffers.emit(key) - } - } + ensureBuffer(key) + _buffers.update { map -> + val existing = map[key] ?: ChannelBuffer(key) + val trimmed = (existing.messages + msg).takeLast(MESSAGE_HISTORY_CAP) + map + (key to existing.copy(messages = trimmed)) } scheduleSave(key) } @@ -202,6 +222,7 @@ object ConnectionManager { } private fun updateUsers(key: BufferKey, transform: (List) -> List) { + ensureBuffer(key) _buffers.update { map -> val existing = map[key] ?: ChannelBuffer(key) map + (key to existing.copy(users = transform(existing.users))) @@ -274,12 +295,13 @@ object ConnectionManager { "TOPIC" -> { val channel = line.params.getOrNull(0) ?: return val topic = line.params.getOrNull(1) ?: "" + val key = BufferKey(serverId, channel) + ensureBuffer(key) _buffers.update { map -> - val key = BufferKey(serverId, channel) val existing = map[key] ?: ChannelBuffer(key) map + (key to existing.copy(topic = topic)) } - addMessage(BufferKey(serverId, channel), IrcMessage(now, MessageKind.TOPIC, nick, "$nick changed topic: $topic")) + addMessage(key, IrcMessage(now, MessageKind.TOPIC, nick, "$nick changed topic: $topic")) } "353" -> { // NAMES reply val channel = line.params.getOrNull(2) ?: return diff --git a/src/main/kotlin/ui/ChannelScreen.kt b/src/main/kotlin/ui/ChannelScreen.kt index d7767ad..a9f8ac0 100644 --- a/src/main/kotlin/ui/ChannelScreen.kt +++ b/src/main/kotlin/ui/ChannelScreen.kt @@ -316,58 +316,70 @@ fun WindowScope.ChannelScreen( // pattern real Compose chat apps use for exactly this reason. val reversedMessages = messages.asReversed() + // How close to the very bottom (index 0, offset 0) counts as "still at the bottom" when + // a scroll comes to rest - a little slack so a pixel or two of residual offset doesn't + // read as "scrolled away". + val stickToBottomThresholdPx = with(LocalDensity.current) { 24.dp.toPx() } + // Track whether the user is currently near the bottom (= near index 0, since the list // is reversed) of the scrollback. If they've manually scrolled up to read backscroll, // stop auto-scrolling until they scroll back down themselves. + // + // This only recomputes stickToBottom while isScrollInProgress is true, plus the single + // emission right after a scroll just ended (tracked via wasScrolling), so that scroll's + // final resting position still gets recorded. Previously this derived stickToBottom from + // firstVisibleItemIndex alone, which also changes from pure item-anchoring: reverseLayout + // anchors the previously-first-visible item by key when a new item is inserted at index 0, + // so every new message bumped that index by one with zero actual scrolling - two or more + // messages landing before this effect re-ran pushed the index past the old "<= 1" cutoff + // and flipped stickToBottom off for the rest of the session, silently disabling auto-scroll + // during any message burst. LaunchedEffect(listState) { + var wasScrolling = false snapshotFlow { - listState.layoutInfo.visibleItemsInfo.firstOrNull()?.index - }.collect { firstVisible -> - stickToBottom = firstVisible == null || firstVisible <= 1 + Triple( + listState.isScrollInProgress, + listState.firstVisibleItemIndex, + listState.firstVisibleItemScrollOffset + ) + }.collect { (isScrolling, index, offset) -> + if (isScrolling || wasScrolling) { + stickToBottom = index == 0 && offset <= stickToBottomThresholdPx + } + wasScrolling = isScrolling } } // Whenever a new message actually lands, jump to index 0 (the newest, since the list is // reversed) IF we were already sticking to the bottom. // - // This needs BOTH of two properties that earlier versions of this effect only ever had - // one of at a time: + // The trigger is msg.seq, not totalItemsCount: msg.seq is monotonically increasing and + // never repeats or saturates, while ConnectionManager caps every buffer at + // MESSAGE_HISTORY_CAP (500) messages, evicting the oldest one for every new one once a + // channel hits that cap - so totalItemsCount stops changing forever past that point, and + // an effect keyed on it would silently stop firing for any channel busy enough to fill its + // scrollback. // - // 1) The trigger has to be something that changes on literally every new message, cap or - // no cap. A version of this keyed on listState.layoutInfo.totalItemsCount got that - // wrong: ConnectionManager caps every buffer at MESSAGE_HISTORY_CAP (500) messages, - // evicting the oldest one for every new one once a channel hits that cap, so the count - // - and totalItemsCount with it - stops changing forever past that point, and the - // effect silently stopped firing for any channel busy enough to fill its scrollback. - // msg.seq is monotonically increasing and never repeats or saturates, so it's used as - // the LaunchedEffect's own key here instead - Compose relaunches the effect (with a - // fresh closure over the current `stickToBottom`/`listState`) every time it changes, - // which also sidesteps a staleness trap a `snapshotFlow { messages.lastOrNull() } would - // fall into: `messages` is a plain local val recomputed on every recomposition, not a - // live read through a stable remembered object like `listState`, so a *persisted* - // snapshotFlow closure capturing it would only ever see the value from the first - // composition it ran in. - // - // 2) The actual scrollToItem(0) call has to wait until the LazyColumn's own layout pass - // has genuinely caught up to that new item before it runs. This part was confirmed by - // hand, live, after dropping it: keying the relaunch on msg.seq and calling - // scrollToItem(0) immediately (no wait) reintroduced the exact one-message-behind lag - // this file's history already describes - sending several messages in a row left the - // scroll position permanently one message short, each new send only revealing the - // PREVIOUS one. Composition having the new item (which msg.seq as a key does guarantee) - // is not the same thing as the LazyColumn's layout pass having processed it yet - - // scrollToItem(0) called too early resolves against however many items layout still - // thinks exist, which is one short, and nothing re-corrects it afterwards. Waiting for - // layoutInfo.totalItemsCount to actually reach the expected count - which, unlike using - // it as the trigger, works fine even at the cap, since "count" here just means "layout - // has processed at least this many items so far", not "count changed" - closes that - // window without giving up cap-immunity. + // The actual scrollToItem(0) call still has to wait for the LazyColumn's own layout pass + // to genuinely catch up to that new item - calling it immediately resolves against + // whatever layout still thinks exists, which can be one short, with nothing to + // re-correct it afterwards. Waiting for `totalItemsCount >= expectedItemCount` (an earlier + // version of this wait) is not cap-proof either: once a buffer sits at the 500 cap, + // totalItemsCount was already 500 before the new item got laid out (an old item was + // evicted for it), so that wait is satisfied immediately against stale layout and + // scrollToItem(0) can still land one message short. Instead, this captures the current + // reversedMessages list at launch and waits for the layout to actually reflect it: either + // there are no visible items yet, or the first visible item's key matches what that + // captured list says should be at that index - which can only be true once layout has + // processed up through at least that item. val newestSeq = messages.lastOrNull()?.seq - val expectedItemCount = messages.size LaunchedEffect(newestSeq) { - if (newestSeq != null && stickToBottom) { - snapshotFlow { listState.layoutInfo.totalItemsCount } - .first { it >= expectedItemCount } + if (newestSeq != null && stickToBottom && !listState.isScrollInProgress) { + val capturedList = reversedMessages + snapshotFlow { listState.layoutInfo.visibleItemsInfo.firstOrNull() } + .first { item -> + item == null || (item.index < capturedList.size && item.key == capturedList[item.index].seq) + } listState.scrollToItem(0) } }