weishaupt/wIRC-desktop
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 6 Wiki Activity

Fix scrollback: burst-proof stick-to-bottom, cap-proof layout wait, serialized buffer updates, atomic history saves

Browse Source
This commit is contained in:
weishaupt
2026-10-04 16:29:21 -04:00
parent 6709db3b13
commit 4f717709d2
3 changed files with 140 additions and 86 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/main/kotlin/irc/ConnectionManager.kt
+69 -47
View File
@@ -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<BufferKey>(extraBufferCapacity = 16)
val newQueryBuffers: SharedFlow<BufferKey> = _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<BufferKey, Mutex>()
private fun mutexFor(key: BufferKey): Mutex = bufferMutexes.computeIfAbsent(key) { Mutex() }
// Pending debounced saves, keyed by buffer - see scheduleSave().
private val pendingSaveJobs = ConcurrentHashMap<BufferKey, Job>()
// 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<ChannelUser>) -> List<ChannelUser>) {
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