fix: follow live logs and reconcile duplicate managed mods

apps/app-frontend/src/composables/console-history.test.ts

@@ -1,23 +0,0 @@
-import assert from 'node:assert/strict'
-import test from 'node:test'
-
-import { mergeLiveLogHistory } from './console-history.ts'
-
-test('retains the persisted latest.log history and appends a live-only tail', () => {
-	assert.equal(
-		mergeLiveLogHistory('first\nsecond\nthird', 'second\nthird\nfourth'),
-		'first\nsecond\nthird\nfourth',
-	)
-})
-
-test('does not duplicate a buffer already persisted in latest.log', () => {
-	assert.equal(mergeLiveLogHistory('first\nsecond\nthird', 'second\nthird'), 'first\nsecond\nthird')
-})
-
-test('keeps live output when it is not present in latest.log', () => {
-	assert.equal(mergeLiveLogHistory('minecraft', 'launcher'), 'minecraft\nlauncher')
-})
-
-test('accepts an unavailable latest.log while the instance is starting', () => {
-	assert.equal(mergeLiveLogHistory('', 'launcher output'), 'launcher output')
-})

apps/app-frontend/src/composables/console-history.ts

@@ -1,31 +0,0 @@
-function nonEmptyLines(content: string): string[] {
-	return content.split(/\r\n|\r|\n/).filter(Boolean)
-}
-
-function overlapLength(history: readonly string[], liveBuffer: readonly string[]): number {
-	const maximum = Math.min(history.length, liveBuffer.length)
-
-	for (let length = maximum; length > 0; length--) {
-		let overlaps = true
-		for (let index = 0; index < length; index++) {
-			if (history[history.length - length + index] !== liveBuffer[index]) {
-				overlaps = false
-				break
-			}
-		}
-		if (overlaps) return length
-	}
-
-	return 0
-}
-
-/**
- * Combines the persisted Minecraft log with output that may not have reached
- * latest.log yet. The ring buffer commonly repeats the tail of latest.log, so
- * retain only its non-overlapping suffix.
- */
-export function mergeLiveLogHistory(latestLog: string, liveBuffer: string): string {
-	const history = nonEmptyLines(latestLog)
-	const buffer = nonEmptyLines(liveBuffer)
-	return [...history, ...buffer.slice(overlapLength(history, buffer))].join('\n')
-}

apps/app-frontend/src/composables/latest-log-reader.test.ts

@@ -0,0 +1,120 @@
+import assert from 'node:assert/strict'
+import test from 'node:test'
+
+import { createLatestLogReader } from './latest-log-reader.ts'
+
+test('follows file-only output after startup, re-entry, clear and file rotation', async () => {
+	let file = ''
+	let displayed = 'stdout startup'
+	const reader = createLatestLogReader(
+		async (cursor) => ({
+			cursor: file.length,
+			output: file.slice(cursor > file.length ? 0 : cursor),
+			new_file: cursor > file.length,
+		}),
+		(output, replace) => {
+			displayed = (replace ? '' : displayed) + output
+		},
+	)
+	await reader.refresh()
+	assert.equal(reader.active, false)
+	file = 'Startup complete\n'
+	await reader.refresh()
+	assert.equal(reader.active, true)
+	assert.equal(displayed, file) // Replaces stdout, not appended a second time.
+	file += 'Joined world\nDisconnected\n'
+	await reader.refresh()
+	await reader.refresh() // Re-entering the view must not replay history.
+	assert.equal(displayed, file)
+	reader.clear()
+	displayed = ''
+	file += 'After clear\n'
+	await reader.refresh()
+	assert.equal(displayed, 'After clear\n')
+	file = 'Rotated\n'
+	await reader.refresh()
+	assert.equal(displayed, file)
+	file = ''
+	await reader.refresh()
+	assert.equal(displayed, '')
+	assert.equal(reader.active, false)
+	file = 'File recreated\n'
+	await reader.refresh()
+	assert.equal(displayed, file)
+	reader.reset()
+	assert.equal(reader.active, false)
+	file = 'New session with a longer file\n'
+	await reader.refresh()
+	assert.equal(displayed, file)
+})
+
+test('serializes reads, ignores pre-restart responses, consumes pre-clear responses', async () => {
+	const resolvers: ((value: { cursor: number; output: string; new_file: boolean }) => void)[] = []
+	const cursors: number[] = []
+	const output: string[] = []
+	const reader = createLatestLogReader(
+		(cursor) => {
+			cursors.push(cursor)
+			return new Promise((resolve) => resolvers.push(resolve))
+		},
+		(text) => output.push(text),
+	)
+	const old = reader.refresh()
+	const concurrent = reader.refresh()
+	assert.deepEqual(cursors, [0])
+	reader.reset()
+	const fresh = reader.refresh()
+	resolvers[0]({ cursor: 100, output: 'old session', new_file: false })
+	await Promise.all([old, concurrent])
+	assert.deepEqual(output, [])
+	reader.clear()
+	resolvers[1]({ cursor: 20, output: 'before clear', new_file: false })
+	await fresh
+	assert.deepEqual(output, [])
+	const next = reader.refresh()
+	assert.deepEqual(cursors, [0, 0, 20])
+	resolvers[2]({ cursor: 40, output: 'after clear\n', new_file: false })
+	await next
+	assert.deepEqual(output, ['after clear\n'])
+})
+
+test('a temporary read failure keeps the cursor and retries without duplicating output', async () => {
+	let failed = true
+	const output: string[] = []
+	const reader = createLatestLogReader(
+		async (cursor) => {
+			assert.equal(cursor, 0)
+			if (failed) throw new Error('file temporarily unavailable')
+			return { cursor: 6, output: 'ready\n', new_file: false }
+		},
+		(text) => output.push(text),
+	)
+	await assert.rejects(reader.refresh())
+	assert.equal(reader.active, false)
+	failed = false
+	await reader.refresh()
+	assert.deepEqual(output, ['ready\n'])
+})
+
+test('joins split lines and flushes the final unterminated line on exit', async () => {
+	let chunk = 'Startup\n[Render thread/INFO] Join'
+	let cursor = 0
+	let displayed = ''
+	const reader = createLatestLogReader(
+		async () => ({
+			cursor: (cursor += chunk.length),
+			output: chunk,
+			new_file: false,
+		}),
+		(text, replace) => {
+			displayed = (replace ? '' : displayed) + text
+		},
+	)
+	await reader.refresh()
+	assert.equal(displayed, 'Startup\n')
+	chunk = 'ed world\nLast line'
+	await reader.refresh()
+	assert.equal(displayed, 'Startup\n[Render thread/INFO] Joined world\n')
+	reader.flush()
+	assert.equal(displayed, 'Startup\n[Render thread/INFO] Joined world\nLast line')
+})

apps/app-frontend/src/composables/latest-log-reader.ts

@@ -0,0 +1,88 @@
+interface LogCursor {
+	cursor: number
+	output: string
+	new_file: boolean
+}
+
+/** One authoritative stream: use process events until latest.log becomes available. */
+export function createLatestLogReader(
+	read: (cursor: number) => Promise<LogCursor>,
+	append: (output: string, replace: boolean) => void,
+) {
+	let cursor = 0
+	let active = false
+	let session = 0
+	let cleared = 0
+	let partialLine = ''
+	let replaceOnAppend = false
+	let pending: Promise<void> | null = null
+
+	function refresh(): Promise<void> {
+		if (pending) return pending
+		const requestSession = session
+		const requestClear = cleared
+		const request = (async () => {
+			const result = await read(cursor)
+			if (requestSession !== session) return
+			const rotated = result.new_file || result.cursor < cursor
+			const replace = !active || rotated
+			cursor = result.cursor
+			if (result.cursor === 0 && !result.output) {
+				if (rotated) {
+					active = false
+					partialLine = ''
+					replaceOnAppend = false
+					if (requestClear === cleared) append('', true)
+				}
+				return
+			}
+			active = true
+			// Clearing during a read consumes its bytes without restoring cleared text.
+			if (requestClear === cleared && (result.output || result.new_file)) {
+				if (replace) {
+					partialLine = ''
+					replaceOnAppend = true
+				}
+				partialLine += result.output
+				// File writes can end mid-line. Do not turn the next chunk into a
+				// separate log entry; bound unterminated output just like the console.
+				const end =
+					partialLine.length > 64 * 1024 ? partialLine.length : partialLine.lastIndexOf('\n') + 1
+				if (end > 0) {
+					append(partialLine.slice(0, end), replaceOnAppend)
+					partialLine = partialLine.slice(end)
+					replaceOnAppend = false
+				}
+			}
+		})()
+		pending = request
+		return request.finally(() => {
+			if (pending === request) pending = null
+		})
+	}
+
+	return {
+		refresh,
+		get active() {
+			return active
+		},
+		clear() {
+			cleared++
+			partialLine = ''
+			replaceOnAppend = false
+		},
+		flush() {
+			if (partialLine) append(partialLine, replaceOnAppend)
+			partialLine = ''
+			replaceOnAppend = false
+		},
+		reset() {
+			session++
+			cursor = 0
+			active = false
+			partialLine = ''
+			replaceOnAppend = false
+			pending = null
+		},
+	}
+}

apps/app-frontend/src/composables/useInstanceConsole.ts

@@ -7,7 +7,7 @@ import {
 	get_minecraft_latest_log_cursor,
 } from '@/helpers/logs'
 
-import { mergeLiveLogHistory } from './console-history'
+import { createLatestLogReader } from './latest-log-reader'
 
 type ConsoleState = ReturnType<typeof createConsoleState>
 
@@ -26,6 +26,7 @@ interface InstanceConsoleEntry {
 	historicalCache: Map<string, string>
 	logList: LogEntry[] | null
 	liveHistoryHydration: Promise<void> | null
+	latestLog: ReturnType<typeof createLatestLogReader>
 }
 
 const instances = new Map<string, InstanceConsoleEntry>()
@@ -34,12 +35,20 @@ function getOrCreate(instanceId: string): InstanceConsoleEntry {
 	let entry = instances.get(instanceId)
 	if (entry) return entry
 
+	const liveConsole = createConsoleState()
 	entry = {
-		liveConsole: createConsoleState(),
+		liveConsole,
 		historicalConsole: createConsoleState(),
 		historicalCache: new Map(),
 		logList: null,
 		liveHistoryHydration: null,
+		latestLog: createLatestLogReader(
+			(cursor) => get_minecraft_latest_log_cursor(instanceId, cursor),
+			(output, replace) => {
+				if (replace) liveConsole.clear()
+				void liveConsole.addLegacyLog(output)
+			},
+		),
 	}
 	instances.set(instanceId, entry)
 	return entry
@@ -47,25 +56,17 @@ function getOrCreate(instanceId: string): InstanceConsoleEntry {
 
 async function hydrate(instanceId: string): Promise<void> {
 	const entry = getOrCreate(instanceId)
-	if (entry.liveConsole.output.value.length > 0) return
-
 	if (entry.liveHistoryHydration) {
 		return entry.liveHistoryHydration
 	}
 
 	const hydration = (async () => {
-		const [latestLog, buffer] = await Promise.all([
-			get_minecraft_latest_log_cursor(instanceId, 0)
-				.then((result) => result.output)
-				.catch(() => ''),
-			get_live_log_buffer(instanceId),
-		])
-
-		if (entry.liveConsole.output.value.length > 0) return
-
-		const history = mergeLiveLogHistory(latestLog, buffer)
-		if (history) {
-			await entry.liveConsole.addLegacyLog(history)
+		await entry.latestLog.refresh().catch(console.warn)
+		if (!entry.latestLog.active && entry.liveConsole.output.value.length === 0) {
+			const buffer = await get_live_log_buffer(instanceId)
+			if (!entry.latestLog.active && entry.liveConsole.output.value.length === 0) {
+				await entry.liveConsole.addLegacyLog(buffer)
+			}
 		}
 	})()
 
@@ -108,11 +109,13 @@ function invalidate(instanceId: string): void {
 
 async function clearLive(instanceId: string): Promise<void> {
 	const entry = getOrCreate(instanceId)
+	entry.latestLog.clear()
 	entry.liveConsole.clear()
 	await clear_log_buffer(instanceId).catch(() => {})
 }
 
 async function destroy(instanceId: string): Promise<void> {
+	instances.get(instanceId)?.latestLog.reset()
 	instances.delete(instanceId)
 	await clear_log_buffer(instanceId).catch(() => {})
 }
@@ -127,6 +130,13 @@ export function useInstanceConsole(instanceId: string) {
 		getHistoricalContent: (filename: string) => getHistoricalContent(instanceId, filename),
 		invalidate: () => invalidate(instanceId),
 		clearLive: () => clearLive(instanceId),
+		refreshLive: () => entry.latestLog.refresh(),
+		flushLive: () => entry.latestLog.flush(),
+		acceptsProcessLogs: () => !entry.latestLog.active,
+		resetLive: () => {
+			entry.latestLog.reset()
+			entry.liveConsole.clear()
+		},
 		destroy: () => destroy(instanceId),
 	}
 }

apps/app-frontend/src/pages/instance/Logs.vue

@@ -86,6 +86,10 @@ const {
 	getHistoricalContent,
 	invalidate,
 	clearLive,
+	refreshLive,
+	flushLive,
+	acceptsProcessLogs,
+	resetLive,
 } = useInstanceConsole(instanceId.value)
 
 await hydrate()
@@ -217,6 +221,7 @@ if (!props.playing) {
 
 const unlistenLog = await log_listener((payload) => {
 	if (payload.instance_id !== instanceId.value) return
+	if (!acceptsProcessLogs()) return
 
 	if (payload.type === 'log4j') {
 		liveConsole.addLog4jEvent(payload)
@@ -228,12 +233,14 @@ const unlistenLog = await log_listener((payload) => {
 const unlistenProcesses = await process_listener(async (e) => {
 	if (e.instance_id !== instanceId.value) return
 	if (e.event === 'launched') {
-		liveConsole.clear()
+		resetLive()
 		clearCrashAnalysis()
 		invalidate()
 		selectedLogIndex.value = 0
 	}
 	if (e.event === 'finished') {
+		await refreshLive().catch(console.warn)
+		flushLive()
 		invalidate()
 		const freshLogs = await getHistoricalLogs()
 		logs.value = buildLogList(freshLogs)
@@ -241,7 +248,20 @@ const unlistenProcesses = await process_listener(async (e) => {
 	}
 })
 
+// Keep following the file even when a mod stops writing to stdout after startup.
+// Schedule after each read so slow disk/IPC cannot pile up concurrent requests.
+let disposed = false
+let refreshTimer
+async function followLatestLog() {
+	await refreshLive().catch(console.warn)
+	if (!props.playing) flushLive()
+	if (!disposed) refreshTimer = setTimeout(followLatestLog, 1000)
+}
+void followLatestLog()
+
 onUnmounted(() => {
+	disposed = true
+	clearTimeout(refreshTimer)
 	unlistenLog()
 	unlistenProcesses()
 })