fix: 修复实时日志与数据库切换可靠性

apps/app-frontend/package.json

@@ -1,7 +1,7 @@
 {
 	"name": "@modrinth/app-frontend",
 	"private": true,
-	"version": "1.0.0-beta12",
+	"version": "1.0.0-beta15",
 	"type": "module",
 	"scripts": {
 		"dev": "vite",

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

@@ -1,120 +0,0 @@
-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

@@ -1,88 +0,0 @@
-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/live-log-reconciler.test.ts

@@ -0,0 +1,70 @@
+import assert from 'node:assert/strict'
+import test from 'node:test'
+
+import {
+	createLiveLogReconciler,
+	type LiveLogSnapshot,
+} from './live-log-reconciler.ts'
+
+test('merges the authoritative snapshot with queued events without gaps or duplicates', async () => {
+	let resolveSnapshot!: (snapshot: LiveLogSnapshot) => void
+	const displayed: number[] = []
+	const reconciler = createLiveLogReconciler(
+		() =>
+			new Promise((resolve) => {
+				resolveSnapshot = resolve
+			}),
+		(snapshot) => {
+			displayed.splice(
+				0,
+				displayed.length,
+				...snapshot.lines.map((line) => line.sequence),
+			)
+		},
+		(event) => displayed.push(event.last_sequence),
+	)
+
+	const syncing = reconciler.sync()
+	reconciler.accept({ first_sequence: 3, last_sequence: 3 })
+	resolveSnapshot({
+		lines: [
+			{ sequence: 1, message: 'one' },
+			{ sequence: 2, message: 'two' },
+		],
+		last_sequence: 2,
+	})
+	await syncing
+	assert.deepEqual(displayed, [1, 2, 3])
+
+	await reconciler.accept({ first_sequence: 3, last_sequence: 3 })
+	assert.deepEqual(displayed, [1, 2, 3])
+})
+
+test('resynchronizes when an event sequence reveals a missed batch', async () => {
+	const displayed: number[] = []
+	let loads = 0
+	const reconciler = createLiveLogReconciler(
+		async () => {
+			loads++
+			return {
+				lines: [1, 2, 3, 4].map((sequence) => ({
+					sequence,
+					message: String(sequence),
+				})),
+				last_sequence: 4,
+			}
+		},
+		(snapshot) => {
+			displayed.splice(
+				0,
+				displayed.length,
+				...snapshot.lines.map((line) => line.sequence),
+			)
+		},
+		(event) => displayed.push(event.last_sequence),
+	)
+
+	await reconciler.accept({ first_sequence: 4, last_sequence: 4 })
+	assert.equal(loads, 1)
+	assert.deepEqual(displayed, [1, 2, 3, 4])
+})

apps/app-frontend/src/composables/live-log-reconciler.ts

@@ -0,0 +1,81 @@
+export interface LiveLogSnapshot {
+	lines: Array<{ sequence: number; message: string }>
+	last_sequence: number
+}
+
+export interface SequencedLogEvent {
+	first_sequence: number
+	last_sequence: number
+}
+
+export function createLiveLogReconciler<T extends SequencedLogEvent>(
+	loadSnapshot: () => Promise<LiveLogSnapshot>,
+	replace: (snapshot: LiveLogSnapshot) => void,
+	append: (event: T) => void,
+) {
+	let lastSequence = 0
+	let queued: T[] = []
+	let syncing: Promise<void> | null = null
+	let disposed = false
+
+	function drainQueued(): boolean {
+		const events = queued
+		queued = []
+		for (let index = 0; index < events.length; index++) {
+			const event = events[index]!
+			if (event.last_sequence <= lastSequence) continue
+			if (event.first_sequence > lastSequence + 1) {
+				queued.push(...events.slice(index))
+				return false
+			}
+			append(event)
+			lastSequence = event.last_sequence
+		}
+		return true
+	}
+
+	function sync(): Promise<void> {
+		if (syncing) return syncing
+		const request = (async () => {
+			do {
+				const snapshot = await loadSnapshot()
+				if (disposed) return
+				replace(snapshot)
+				lastSequence = snapshot.last_sequence
+			} while (!drainQueued())
+		})()
+		const completion = request.finally(() => {
+			if (syncing === completion) syncing = null
+		})
+		syncing = completion
+		return completion
+	}
+
+	function resync(): Promise<void> {
+		return syncing ? syncing.then(sync) : sync()
+	}
+
+	function accept(event: T): Promise<void> | undefined {
+		if (disposed || event.last_sequence <= lastSequence) return
+		if (syncing) {
+			queued.push(event)
+			return syncing
+		}
+		if (event.first_sequence > lastSequence + 1) {
+			queued.push(event)
+			return sync()
+		}
+		append(event)
+		lastSequence = event.last_sequence
+	}
+
+	return {
+		accept,
+		resync,
+		sync,
+		dispose() {
+			disposed = true
+			queued = []
+		},
+	}
+}

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

@@ -2,12 +2,14 @@ import { createConsoleState } from '@modrinth/ui'
 
 import {
 	clear_log_buffer,
-	get_live_log_buffer,
+	get_live_log_snapshot,
 	get_logs,
-	get_minecraft_latest_log_cursor,
 } from '@/helpers/logs'
 
-import { createLatestLogReader } from './latest-log-reader'
+import {
+	createLiveLogReconciler,
+	type SequencedLogEvent,
+} from './live-log-reconciler'
 
 type ConsoleState = ReturnType<typeof createConsoleState>
 
@@ -25,8 +27,17 @@ interface InstanceConsoleEntry {
 	historicalConsole: ConsoleState
 	historicalCache: Map<string, string>
 	logList: LogEntry[] | null
-	liveHistoryHydration: Promise<void> | null
-	latestLog: ReturnType<typeof createLatestLogReader>
+	liveLog: ReturnType<typeof createLiveLogReconciler<ProcessLogPayload>>
+}
+
+interface ProcessLogPayload extends SequencedLogEvent {
+	type: 'log4j' | 'legacy'
+	message?: string
+	logger_name?: string
+	level?: string
+	thread_name?: string
+	timestamp_millis?: number
+	throwable?: string
 }
 
 const instances = new Map<string, InstanceConsoleEntry>()
@@ -36,19 +47,28 @@ function getOrCreate(instanceId: string): InstanceConsoleEntry {
 	if (entry) return entry
 
 	const liveConsole = createConsoleState()
+	const liveLog = createLiveLogReconciler<ProcessLogPayload>(
+		() => get_live_log_snapshot(instanceId),
+		(snapshot) => {
+			liveConsole.clear()
+			void liveConsole.addLegacyLog(
+				snapshot.lines.map((line) => line.message).join('\n'),
+			)
+		},
+		(payload) => {
+			if (payload.type === 'log4j') {
+				liveConsole.addLog4jEvent(payload)
+			} else if (payload.message !== undefined) {
+				void liveConsole.addLegacyLog(payload.message)
+			}
+		},
+	)
 	entry = {
 		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)
-			},
-		),
+		liveLog,
 	}
 	instances.set(instanceId, entry)
 	return entry
@@ -56,28 +76,7 @@ function getOrCreate(instanceId: string): InstanceConsoleEntry {
 
 async function hydrate(instanceId: string): Promise<void> {
 	const entry = getOrCreate(instanceId)
-	if (entry.liveHistoryHydration) {
-		return entry.liveHistoryHydration
-	}
-
-	const hydration = (async () => {
-		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)
-			}
-		}
-	})()
-
-	entry.liveHistoryHydration = hydration
-	try {
-		await hydration
-	} finally {
-		if (entry.liveHistoryHydration === hydration) {
-			entry.liveHistoryHydration = null
-		}
-	}
+	await entry.liveLog.sync()
 }
 
 async function getHistoricalLogs(instanceId: string): Promise<LogEntry[]> {
@@ -109,15 +108,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(() => {})
+	await clear_log_buffer(instanceId)
+	await entry.liveLog.resync()
 }
 
-async function destroy(instanceId: string): Promise<void> {
-	instances.get(instanceId)?.latestLog.reset()
+function destroy(instanceId: string): void {
+	instances.get(instanceId)?.liveLog.dispose()
 	instances.delete(instanceId)
-	await clear_log_buffer(instanceId).catch(() => {})
 }
 
 export function useInstanceConsole(instanceId: string) {
@@ -130,13 +127,8 @@ 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()
-		},
+		appendLive: (payload: ProcessLogPayload) => entry.liveLog.accept(payload),
+		resyncLive: () => entry.liveLog.resync(),
 		destroy: () => destroy(instanceId),
 	}
 }

apps/app-frontend/src/helpers/logs.js

@@ -64,14 +64,9 @@ export async function get_latest_log_cursor(instanceId, cursor) {
 	return await invoke('plugin:logs|logs_get_latest_log_cursor', { instanceId, cursor })
 }
 
-/// Read Minecraft's logs/latest.log from a cursor.
-export async function get_minecraft_latest_log_cursor(instanceId, cursor) {
-	return await invoke('plugin:logs|logs_get_minecraft_latest_log_cursor', { instanceId, cursor })
-}
-
-/// Get all buffered live log lines for an instance from the Rust ring buffer
-export async function get_live_log_buffer(instanceId) {
-	return await invoke('plugin:logs|logs_get_live_log_buffer', { instanceId })
+/// Get the authoritative sequenced live-log snapshot from the Rust ring buffer.
+export async function get_live_log_snapshot(instanceId) {
+	return await invoke('plugin:logs|logs_get_live_log_snapshot', { instanceId })
 }
 
 /// Clear the live log buffer for an instance on the Rust side

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

@@ -86,13 +86,16 @@ const {
 	getHistoricalContent,
 	invalidate,
 	clearLive,
-	refreshLive,
-	flushLive,
-	acceptsProcessLogs,
-	resetLive,
+	appendLive,
+	resyncLive,
 } = useInstanceConsole(instanceId.value)
 
-await hydrate()
+const unlistenLog = await log_listener((payload) => {
+	if (payload.instance_id !== instanceId.value) return
+	void appendLive(payload)?.catch(console.warn)
+})
+
+await hydrate().catch(handleError)
 
 function buildLogList(rawLogs) {
 	return [
@@ -219,28 +222,16 @@ if (!props.playing) {
 	void analyseForCrash()
 }
 
-const unlistenLog = await log_listener((payload) => {
-	if (payload.instance_id !== instanceId.value) return
-	if (!acceptsProcessLogs()) return
-
-	if (payload.type === 'log4j') {
-		liveConsole.addLog4jEvent(payload)
-	} else if (payload.type === 'legacy') {
-		void liveConsole.addLegacyLog(payload.message)
-	}
-})
-
 const unlistenProcesses = await process_listener(async (e) => {
 	if (e.instance_id !== instanceId.value) return
 	if (e.event === 'launched') {
-		resetLive()
+		await resyncLive().catch(console.warn)
 		clearCrashAnalysis()
 		invalidate()
 		selectedLogIndex.value = 0
 	}
 	if (e.event === 'finished') {
-		await refreshLive().catch(console.warn)
-		flushLive()
+		await resyncLive().catch(console.warn)
 		invalidate()
 		const freshLogs = await getHistoricalLogs()
 		logs.value = buildLogList(freshLogs)
@@ -248,20 +239,7 @@ 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()
 })

apps/app/Cargo.toml

@@ -1,7 +1,7 @@
 [package]
 name = "theseus_gui"
 # The actual version is set by the theseus-build workflow on tagging
-version = "1.0.0-beta12"
+version = "1.0.0-beta15"
 edition.workspace = true
 description = "Axolotl Launcher is a cross-platform Minecraft launcher"
 homepage = "https://www.ghs.red"

apps/app/build.rs

@@ -292,8 +292,7 @@ fn main() {
                         "logs_delete_logs",
                         "logs_delete_logs_by_filename",
                         "logs_get_latest_log_cursor",
-                        "logs_get_minecraft_latest_log_cursor",
-                        "logs_get_live_log_buffer",
+                        "logs_get_live_log_snapshot",
                         "logs_clear_live_log_buffer",
                         "logs_analyze_crash",
                         "logs_get_crash_analysis_ai_settings",

apps/app/src/api/logs.rs

@@ -4,7 +4,7 @@ use std::path::PathBuf;
 use theseus::logs::LogType;
 use theseus::logs::{
     self, CensoredString, CrashAnalysis, CrashAnalysisAiExplanation,
-    CrashAnalysisAiSettings, LatestLogCursor, Logs,
+    CrashAnalysisAiSettings, LatestLogCursor, LiveLogSnapshot, Logs,
 };
 
 /*
@@ -26,8 +26,7 @@ pub fn init<R: tauri::Runtime>() -> tauri::plugin::TauriPlugin<R> {
             logs_delete_logs,
             logs_delete_logs_by_filename,
             logs_get_latest_log_cursor,
-            logs_get_minecraft_latest_log_cursor,
-            logs_get_live_log_buffer,
+            logs_get_live_log_snapshot,
             logs_clear_live_log_buffer,
             logs_analyze_crash,
             logs_get_crash_analysis_ai_settings,
@@ -95,24 +94,12 @@ pub async fn logs_get_latest_log_cursor(
     Ok(logs::get_latest_log_cursor(instance_id, cursor).await?)
 }
 
-/// Get Minecraft's logs/latest.log from a cursor.
-#[tauri::command]
-pub async fn logs_get_minecraft_latest_log_cursor(
-    instance_id: &str,
-    cursor: u64, // 0 to start at beginning of file
-) -> Result<LatestLogCursor> {
-    Ok(
-        logs::get_generic_live_log_cursor(instance_id, "latest.log", cursor)
-            .await?,
-    )
-}
-
 /// Get all buffered live log lines for an instance.
 #[tauri::command]
-pub async fn logs_get_live_log_buffer(
+pub async fn logs_get_live_log_snapshot(
     instance_id: &str,
-) -> Result<CensoredString> {
-    Ok(logs::get_live_log_buffer(instance_id).await?)
+) -> Result<LiveLogSnapshot> {
+    Ok(logs::get_live_log_snapshot(instance_id).await?)
 }
 
 /// Clear the live log buffer for an instance.