From ce0f41c973ff779ce90e999e759034028019a6d8 Mon Sep 17 00:00:00 2001 From: Disy920 Date: Sun, 27 Sep 2026 14:01:02 +0800 Subject: [PATCH] =?UTF-8?q?fix:=20=E4=BF=AE=E5=A4=8D=E5=AE=9E=E6=97=B6?= =?UTF-8?q?=E6=97=A5=E5=BF=97=E4=B8=8E=E6=95=B0=E6=8D=AE=E5=BA=93=E5=88=87?= =?UTF-8?q?=E6=8D=A2=E5=8F=AF=E9=9D=A0=E6=80=A7?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- Cargo.lock | 4 +- apps/app-frontend/package.json | 2 +- .../src/composables/latest-log-reader.test.ts | 120 ------------- .../src/composables/latest-log-reader.ts | 88 ---------- .../composables/live-log-reconciler.test.ts | 70 ++++++++ .../src/composables/live-log-reconciler.ts | 81 +++++++++ .../src/composables/useInstanceConsole.ts | 88 +++++----- apps/app-frontend/src/helpers/logs.js | 11 +- apps/app-frontend/src/pages/instance/Logs.vue | 42 ++--- apps/app/Cargo.toml | 2 +- apps/app/build.rs | 3 +- apps/app/src/api/logs.rs | 23 +-- packages/app-lib/Cargo.toml | 2 +- packages/app-lib/src/api/drop_classifier.rs | 15 +- packages/app-lib/src/api/logs.rs | 42 +++-- packages/app-lib/src/event/mod.rs | 2 + .../app-lib/src/install/missing_content.rs | 32 ++-- packages/app-lib/src/install/model.rs | 22 ++- packages/app-lib/src/install/recovery.rs | 7 +- .../app-lib/src/launcher/direct_ensure.rs | 17 +- packages/app-lib/src/launcher/direct_link.rs | 30 +++- packages/app-lib/src/state/db.rs | 157 +++++++++++++++--- packages/app-lib/src/state/db_backup.rs | 94 ++++++++++- .../instances/adapters/sqlite/content_rows.rs | 15 +- .../commands/apply_content_install.rs | 2 +- .../commands/create_direct_link_instance.rs | 4 +- packages/app-lib/src/state/process.rs | 133 +++++++++++++-- packages/app-lib/src/util/fetch.rs | 95 ++++++++--- .../console/components/LogViewport.test.ts | 21 +++ .../shared/console/components/LogViewport.vue | 51 ++++-- 30 files changed, 814 insertions(+), 461 deletions(-) delete mode 100644 apps/app-frontend/src/composables/latest-log-reader.test.ts delete mode 100644 apps/app-frontend/src/composables/latest-log-reader.ts create mode 100644 apps/app-frontend/src/composables/live-log-reconciler.test.ts create mode 100644 apps/app-frontend/src/composables/live-log-reconciler.ts create mode 100644 packages/ui/src/layouts/shared/console/components/LogViewport.test.ts diff --git a/Cargo.lock b/Cargo.lock index edfb4d4..5354349 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -6795,7 +6795,7 @@ dependencies = [ [[package]] name = "theseus" -version = "1.0.0-beta12" +version = "1.0.0-beta15" dependencies = [ "ariadne", "async-minecraft-ping", @@ -6899,7 +6899,7 @@ dependencies = [ [[package]] name = "theseus_gui" -version = "1.0.0-beta12" +version = "1.0.0-beta15" dependencies = [ "ashpd", "async_zip", diff --git a/apps/app-frontend/package.json b/apps/app-frontend/package.json index d0a30b2..86ec06a 100644 --- a/apps/app-frontend/package.json +++ b/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", diff --git a/apps/app-frontend/src/composables/latest-log-reader.test.ts b/apps/app-frontend/src/composables/latest-log-reader.test.ts deleted file mode 100644 index 146381b..0000000 --- a/apps/app-frontend/src/composables/latest-log-reader.test.ts +++ /dev/null @@ -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') -}) diff --git a/apps/app-frontend/src/composables/latest-log-reader.ts b/apps/app-frontend/src/composables/latest-log-reader.ts deleted file mode 100644 index 4449391..0000000 --- a/apps/app-frontend/src/composables/latest-log-reader.ts +++ /dev/null @@ -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, - 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 | null = null - - function refresh(): Promise { - 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 - }, - } -} diff --git a/apps/app-frontend/src/composables/live-log-reconciler.test.ts b/apps/app-frontend/src/composables/live-log-reconciler.test.ts new file mode 100644 index 0000000..22e43de --- /dev/null +++ b/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]) +}) diff --git a/apps/app-frontend/src/composables/live-log-reconciler.ts b/apps/app-frontend/src/composables/live-log-reconciler.ts new file mode 100644 index 0000000..b28fb98 --- /dev/null +++ b/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( + loadSnapshot: () => Promise, + replace: (snapshot: LiveLogSnapshot) => void, + append: (event: T) => void, +) { + let lastSequence = 0 + let queued: T[] = [] + let syncing: Promise | 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 { + 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 { + return syncing ? syncing.then(sync) : sync() + } + + function accept(event: T): Promise | 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 = [] + }, + } +} diff --git a/apps/app-frontend/src/composables/useInstanceConsole.ts b/apps/app-frontend/src/composables/useInstanceConsole.ts index 64ff099..6d65c47 100644 --- a/apps/app-frontend/src/composables/useInstanceConsole.ts +++ b/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 @@ -25,8 +27,17 @@ interface InstanceConsoleEntry { historicalConsole: ConsoleState historicalCache: Map logList: LogEntry[] | null - liveHistoryHydration: Promise | null - latestLog: ReturnType + liveLog: ReturnType> +} + +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() @@ -36,19 +47,28 @@ function getOrCreate(instanceId: string): InstanceConsoleEntry { if (entry) return entry const liveConsole = createConsoleState() + const liveLog = createLiveLogReconciler( + () => 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 { 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 { @@ -109,15 +108,13 @@ function invalidate(instanceId: string): void { async function clearLive(instanceId: string): Promise { 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 { - 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), } } diff --git a/apps/app-frontend/src/helpers/logs.js b/apps/app-frontend/src/helpers/logs.js index 641e2bf..7ed4794 100644 --- a/apps/app-frontend/src/helpers/logs.js +++ b/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 diff --git a/apps/app-frontend/src/pages/instance/Logs.vue b/apps/app-frontend/src/pages/instance/Logs.vue index 8f2f5fb..29d50de 100644 --- a/apps/app-frontend/src/pages/instance/Logs.vue +++ b/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() }) diff --git a/apps/app/Cargo.toml b/apps/app/Cargo.toml index a56bb36..f2ae9ae 100644 --- a/apps/app/Cargo.toml +++ b/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" diff --git a/apps/app/build.rs b/apps/app/build.rs index 9b16e66..f09716c 100644 --- a/apps/app/build.rs +++ b/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", diff --git a/apps/app/src/api/logs.rs b/apps/app/src/api/logs.rs index f314ac0..6cfb9d2 100644 --- a/apps/app/src/api/logs.rs +++ b/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() -> tauri::plugin::TauriPlugin { 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 { - 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 { - Ok(logs::get_live_log_buffer(instance_id).await?) +) -> Result { + Ok(logs::get_live_log_snapshot(instance_id).await?) } /// Clear the live log buffer for an instance. diff --git a/packages/app-lib/Cargo.toml b/packages/app-lib/Cargo.toml index 95fb731..ef4a990 100644 --- a/packages/app-lib/Cargo.toml +++ b/packages/app-lib/Cargo.toml @@ -1,7 +1,7 @@ [package] name = "theseus" # 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 [dependencies] diff --git a/packages/app-lib/src/api/drop_classifier.rs b/packages/app-lib/src/api/drop_classifier.rs index c27be1b..e3a9e35 100644 --- a/packages/app-lib/src/api/drop_classifier.rs +++ b/packages/app-lib/src/api/drop_classifier.rs @@ -441,13 +441,14 @@ impl ZipEntrySet { let mut seen: std::collections::HashSet<&str> = std::collections::HashSet::new(); for dir in &self.dirs { - if let Some(rest) = dir.strip_prefix(base) - && let Some((first, _)) = rest.split_once('/') - && !first.is_empty() - && !is_noise_entry(first) - && seen.insert(first) - { - children.push(first.to_string()); + if let Some(rest) = dir.strip_prefix(base) { + let first = rest.split('/').next().unwrap_or_default(); + if !first.is_empty() + && !is_noise_entry(first) + && seen.insert(first) + { + children.push(first.to_string()); + } } } for path in &self.files { diff --git a/packages/app-lib/src/api/logs.rs b/packages/app-lib/src/api/logs.rs index dc63d76..4ea104b 100644 --- a/packages/app-lib/src/api/logs.rs +++ b/packages/app-lib/src/api/logs.rs @@ -54,6 +54,18 @@ pub struct LatestLogCursor { pub new_file: bool, } +#[derive(Serialize, Debug)] +pub struct LiveLogLine { + pub sequence: u64, + pub message: CensoredString, +} + +#[derive(Serialize, Debug)] +pub struct LiveLogSnapshot { + pub lines: Vec, + pub last_sequence: u64, +} + #[derive(Serialize, Debug)] // Not deserialize #[serde(transparent)] pub struct CensoredString(String); @@ -651,31 +663,31 @@ pub async fn delete_logs_by_filename( } #[tracing::instrument] -pub async fn get_live_log_buffer( +pub async fn get_live_log_snapshot( instance_id: &str, -) -> crate::Result { +) -> crate::Result { let state = State::get().await?; - let lines = crate::state::get_log_buffer(instance_id); - let joined = lines.join("\n"); - let compacted = - cap_log_for_display(compact_duplicate_lines(&joined), false); - let credentials = Credentials::get_all(&state.pool) .await? .into_iter() .map(|x| x.1) .collect::>(); - maybe_emit_log_compaction_warning("live log", compacted.stats).await; - maybe_emit_log_display_truncation_warning( - "live log", - compacted.display_truncated, - ) - .await; - Ok(CensoredString::censor(compacted.output, &credentials)) + let snapshot = crate::state::get_live_log_snapshot(instance_id); + Ok(LiveLogSnapshot { + lines: snapshot + .lines + .into_iter() + .map(|line| LiveLogLine { + sequence: line.sequence, + message: CensoredString::censor(line.message, &credentials), + }) + .collect(), + last_sequence: snapshot.last_sequence, + }) } pub fn clear_live_log_buffer(instance_id: &str) { - crate::state::remove_log_buffer(instance_id); + crate::state::clear_log_buffer(instance_id); } #[tracing::instrument] diff --git a/packages/app-lib/src/event/mod.rs b/packages/app-lib/src/event/mod.rs index debc375..bcc6886 100644 --- a/packages/app-lib/src/event/mod.rs +++ b/packages/app-lib/src/event/mod.rs @@ -409,6 +409,8 @@ mod log_types { #[derive(Serialize, Clone)] pub struct LogPayload { pub instance_id: String, + pub first_sequence: u64, + pub last_sequence: u64, #[serde(flatten)] pub event: LogEvent, } diff --git a/packages/app-lib/src/install/missing_content.rs b/packages/app-lib/src/install/missing_content.rs index 1fd1c3f..fa8c10f 100644 --- a/packages/app-lib/src/install/missing_content.rs +++ b/packages/app-lib/src/install/missing_content.rs @@ -1590,6 +1590,15 @@ mod tests { #[cfg(not(feature = "tauri"))] #[tokio::test] async fn import_api_trusts_persisted_job_context_and_keeps_other_missing_files_waiting() + { + Box::pin( + run_import_api_trusts_persisted_job_context_and_keeps_other_missing_files_waiting(), + ) + .await; + } + + #[cfg(not(feature = "tauri"))] + async fn run_import_api_trusts_persisted_job_context_and_keeps_other_missing_files_waiting() { crate::event::EventState::init().await.unwrap(); let root = tempfile::tempdir().unwrap().keep(); @@ -1788,18 +1797,17 @@ mod tests { .await .unwrap(); assert_eq!(observed.pending, 1); - let (scan_result, manual_result) = tokio::join!( - scan_missing_modpack_files_in_at( - job_id, - &race_downloads, - first_seen + DOWNLOAD_STABILITY_WINDOW, - ), - import_missing_modpack_file( - job_id, - first.item_id.clone(), - selected, - ) - ); + let scan = Box::pin(scan_missing_modpack_files_in_at( + job_id, + &race_downloads, + first_seen + DOWNLOAD_STABILITY_WINDOW, + )); + let manual = Box::pin(import_missing_modpack_file( + job_id, + first.item_id.clone(), + selected, + )); + let (scan_result, manual_result) = tokio::join!(scan, manual); let scan_result = scan_result.unwrap(); assert!(manual_result.is_ok()); assert!( diff --git a/packages/app-lib/src/install/model.rs b/packages/app-lib/src/install/model.rs index f609c2e..847c8d6 100644 --- a/packages/app-lib/src/install/model.rs +++ b/packages/app-lib/src/install/model.rs @@ -13,7 +13,10 @@ use crate::state::{ use chrono::{DateTime, Utc}; use modrinth_content_management::{ContentType, ResolutionPreferences}; use serde::{Deserialize, Serialize}; -use std::{collections::HashMap, path::PathBuf}; +use std::{ + collections::{HashMap, HashSet}, + path::PathBuf, +}; use uuid::Uuid; pub type InstallModpackPreview = CreatePackInstance; @@ -483,7 +486,7 @@ mod tests { assert_eq!(summary.bytes_total, Some(300)); let items = job.download_items(); assert_eq!(items.len(), 3); - assert_eq!(items[0].status, DownloadItemStatus::Verifying); + assert_eq!(items[0].status, DownloadItemStatus::Completed); assert_eq!(items[0].attempt, Some(1)); assert_eq!(items[0].max_attempts, Some(4)); assert_eq!( @@ -2395,6 +2398,7 @@ impl InstallJobState { fn download_items_from_events(&self) -> Vec { let mut items = Vec::::new(); let mut indices = HashMap::::new(); + let mut content_items = HashSet::::new(); for event in &self.events { match &event.kind { InstallJobEventKind::ContentFileQueued { @@ -2402,6 +2406,7 @@ impl InstallJobState { bytes_total, max_attempts, } => { + content_items.insert(path.clone()); if let Some(item) = indices .get(path) .and_then(|&index| items.get_mut(index)) @@ -2482,6 +2487,7 @@ impl InstallJobState { attempt, max_attempts, } => { + content_items.insert(path.clone()); if let Some(item) = indices .get(path) .and_then(|&index| items.get_mut(index)) @@ -2562,10 +2568,14 @@ impl InstallJobState { .get(path) .and_then(|&index| items.get_mut(index)) { - // Network transfer completion is not content - // finalization. `ContentFileCompleted` is the event - // that confirms verification and DB registration. - item.status = DownloadItemStatus::Verifying; + // Content downloads still need verification and DB + // registration. Standalone Java/Minecraft requests + // are final once the request itself finishes. + item.status = if content_items.contains(path) { + DownloadItemStatus::Verifying + } else { + DownloadItemStatus::Completed + }; item.bytes_downloaded = *bytes; item.bytes_total = item.bytes_total.or(Some(*bytes)); } diff --git a/packages/app-lib/src/install/recovery.rs b/packages/app-lib/src/install/recovery.rs index 87b9d48..b921da9 100644 --- a/packages/app-lib/src/install/recovery.rs +++ b/packages/app-lib/src/install/recovery.rs @@ -1361,12 +1361,11 @@ mod tests { assert_eq!(persisted.status, InstallJobStatus::Canceled); assert!(persisted.state.rollback.as_ref().unwrap().content.is_none()); assert!(persisted.state.rollback_error.is_none()); - assert_eq!( - crate::state::list_instances(&state.pool) + assert!( + crate::state::get_instance(&instance_id, &state.pool) .await .unwrap() - .len(), - 1 + .is_some() ); let new_instance = crate::api::instance::create( diff --git a/packages/app-lib/src/launcher/direct_ensure.rs b/packages/app-lib/src/launcher/direct_ensure.rs index f730f7b..458dcc4 100644 --- a/packages/app-lib/src/launcher/direct_ensure.rs +++ b/packages/app-lib/src/launcher/direct_ensure.rs @@ -1107,12 +1107,17 @@ mod tests { "org.lwjgl.lwjgl:lwjgl-platform:2.9.4-nightly-20150209:{classifier}" ) ); - assert!( - plan.destination - .to_string_lossy() - .ends_with(format!("{classifier}.jar").as_str()) - ); - assert_eq!(plan.sha1.as_deref(), Some("aaa")); + let (declared_path, declared_sha1) = match classifier.as_str() { + "natives-linux" => ( + "org/lwjgl/lwjgl/lwjgl-platform/2.9.4-nightly-20150209/lwjgl-platform-2.9.4-nightly-20150209-natives-linux.jar", + "aaa", + ), + "natives-osx" => ("x-osx.jar", "bbb"), + "natives-windows" => ("x-win.jar", "ccc"), + other => panic!("unexpected native classifier {other}"), + }; + assert!(plan.destination.to_string_lossy().ends_with(declared_path)); + assert_eq!(plan.sha1.as_deref(), Some(declared_sha1)); } #[test] diff --git a/packages/app-lib/src/launcher/direct_link.rs b/packages/app-lib/src/launcher/direct_link.rs index d8c2ce5..0d53c3c 100644 --- a/packages/app-lib/src/launcher/direct_link.rs +++ b/packages/app-lib/src/launcher/direct_link.rs @@ -1940,13 +1940,16 @@ mod tests { assert_eq!( direct.version_json, Some( - version_dir - .join("1.20.1-fabric.json") - .canonicalize() - .unwrap() + crate::util::io::canonicalize( + version_dir.join("1.20.1-fabric.json"), + ) + .unwrap(), ) ); - assert_eq!(direct.dot_minecraft, root.path().canonicalize().unwrap()); + assert_eq!( + direct.dot_minecraft, + crate::util::io::canonicalize(root.path()).unwrap() + ); } #[test] @@ -2002,6 +2005,15 @@ mod tests { let root = tempfile::tempdir().unwrap(); let version = root.path().join("versions/demo"); std::fs::create_dir_all(&version).unwrap(); + std::fs::write( + version.join("demo.json"), + serde_json::to_vec(&json!({ + "id": "demo", + "mainClass": "net.minecraft.client.main.Main" + })) + .unwrap(), + ) + .unwrap(); assert_eq!( resolve_content_game_dir(root.path(), &version).unwrap(), @@ -2469,7 +2481,13 @@ mod tests { ); assert_eq!( candidates, - vec![launcher.path().join("runtime/jdk17/bin/java")] + vec![ + launcher + .path() + .join("runtime/jdk17") + .join("bin") + .join(PCL_JAVA_BIN) + ] ); // A relative path escaping the launcher directory degrades to the diff --git a/packages/app-lib/src/state/db.rs b/packages/app-lib/src/state/db.rs index fffdfce..99a68c4 100644 --- a/packages/app-lib/src/state/db.rs +++ b/packages/app-lib/src/state/db.rs @@ -4,7 +4,7 @@ use sqlx::migrate::{Migration, Migrator}; use sqlx::sqlite::{ SqliteConnectOptions, SqliteJournalMode, SqlitePoolOptions, }; -use sqlx::{Pool, Sqlite}; +use sqlx::{ConnectOptions, Connection, Pool, Sqlite}; use std::path::{Path, PathBuf}; use std::time::Duration; @@ -126,17 +126,7 @@ pub async fn copy_release_database_to_beta( .into()); } - let release_pool = open_app_db_pool(&release_path).await?; - let escaped_path = beta_path.to_string_lossy().replace('\'', "''"); - sqlx::query(&format!("VACUUM INTO '{escaped_path}'")) - .execute(&release_pool) - .await?; - release_pool.close().await; - - if let Err(error) = open_app_db_pool(&beta_path).await { - let _ = tokio::fs::remove_file(&beta_path).await; - return Err(error); - } + copy_app_database(&release_path, &beta_path).await?; tracing::info!( source = %release_path.display(), @@ -207,16 +197,7 @@ pub async fn copy_database_between_channels( } crate::util::io::create_dir_all(&target_dir).await?; - let temporary_path = target_dir.join(format!("{LEGACY_APP_DB_FILE}.tmp")); - let source_pool = open_app_db_pool(&source_path).await?; - let escaped_path = temporary_path.to_string_lossy().replace('\'', "''"); - let result = sqlx::query(&format!("VACUUM INTO '{escaped_path}'")) - .execute(&source_pool) - .await; - source_pool.close().await; - result?; - - tokio::fs::rename(&temporary_path, &target_path).await?; + copy_app_database(&source_path, &target_path).await?; tracing::info!( source = %source_path.display(), destination = %target_path.display(), @@ -225,6 +206,87 @@ pub async fn copy_database_between_channels( Ok(()) } +async fn copy_app_database(source: &Path, target: &Path) -> crate::Result<()> { + let temporary = sidecar_path(target, ".copying"); + remove_database_files(&temporary).await?; + + let source_pool = open_app_db_pool(source).await?; + let escaped_path = temporary.to_string_lossy().replace('\'', "''"); + let snapshot_result = sqlx::query(&format!("VACUUM INTO '{escaped_path}'")) + .execute(&source_pool) + .await; + source_pool.close().await; + if let Err(error) = snapshot_result { + let _ = remove_database_files(&temporary).await; + return Err(error.into()); + } + + if let Err(error) = verify_database_file(&temporary).await { + let _ = remove_database_files(&temporary).await; + return Err(error); + } + + let previous = sidecar_path(target, ".replacing"); + remove_database_files(&previous).await?; + let had_target = target.try_exists()?; + if had_target { + super::db_backup::archive_database_files(target, &previous).await?; + } else { + super::db_backup::remove_database_sidecars(target).await?; + } + + if let Err(error) = tokio::fs::rename(&temporary, target).await { + if had_target { + super::db_backup::restore_archived_database_files( + &previous, target, + ) + .await?; + } + let _ = remove_database_files(&temporary).await; + return Err(error.into()); + } + + if let Err(error) = remove_database_files(&previous).await { + tracing::warn!( + database = %previous.display(), + error = %error, + "Copied channel database but could not remove the previous snapshot" + ); + } + Ok(()) +} + +async fn verify_database_file(path: &Path) -> crate::Result<()> { + let options = SqliteConnectOptions::new() + .filename(path) + .busy_timeout(Duration::from_secs(30)) + .read_only(true) + .create_if_missing(false); + let mut connection = options.connect().await?; + let rows = sqlx::query_scalar::<_, String>("PRAGMA quick_check(1)") + .fetch_all(&mut connection) + .await?; + connection.close().await?; + if rows.as_slice() != ["ok"] { + return Err(crate::ErrorKind::FSError(format!( + "Database snapshot {} failed its integrity check: {}", + path.display(), + rows.join("; ") + )) + .into()); + } + Ok(()) +} + +async fn remove_database_files(database: &Path) -> crate::Result<()> { + match tokio::fs::remove_file(database).await { + Ok(()) => {} + Err(error) if error.kind() == std::io::ErrorKind::NotFound => {} + Err(error) => return Err(error.into()), + } + super::db_backup::remove_database_sidecars(database).await +} + pub async fn backup_current_app_db_for_update( app_identifier: &str, target_version: &str, @@ -1461,6 +1523,53 @@ mod tests { const INSTANCE_POST_UPGRADE_NOTICES_MIGRATION_VERSION: i64 = 20260824000000; const MCARCHIVE_PROVIDER_FILE_ID_MIGRATION_VERSION: i64 = 20260823020000; + #[tokio::test] + async fn channel_copy_replaces_database_and_removes_stale_sidecars() { + let temp = tempfile::tempdir().unwrap(); + let source = temp.path().join("beta").join(LEGACY_APP_DB_FILE); + let target = temp.path().join("release").join(LEGACY_APP_DB_FILE); + tokio::fs::create_dir_all(source.parent().unwrap()) + .await + .unwrap(); + tokio::fs::create_dir_all(target.parent().unwrap()) + .await + .unwrap(); + + for (path, marker) in [(&source, "beta"), (&target, "release")] { + let pool = open_app_db_pool(path).await.unwrap(); + sqlx::query("CREATE TABLE marker (value TEXT NOT NULL)") + .execute(&pool) + .await + .unwrap(); + sqlx::query("INSERT INTO marker (value) VALUES (?)") + .bind(marker) + .execute(&pool) + .await + .unwrap(); + pool.close().await; + } + tokio::fs::write(sidecar_path(&target, "-wal"), b"stale wal") + .await + .unwrap(); + tokio::fs::write(sidecar_path(&target, "-shm"), b"stale shm") + .await + .unwrap(); + + copy_app_database(&source, &target).await.unwrap(); + + assert!(!sidecar_path(&target, "-wal").exists()); + assert!(!sidecar_path(&target, "-shm").exists()); + let pool = open_app_db_pool(&target).await.unwrap(); + let marker: String = sqlx::query_scalar("SELECT value FROM marker") + .fetch_one(&pool) + .await + .unwrap(); + pool.close().await; + assert_eq!(marker, "beta"); + assert!(!sidecar_path(&target, ".copying").exists()); + assert!(!sidecar_path(&target, ".replacing").exists()); + } + #[tokio::test] async fn post_upgrade_notices_migrate_fresh_and_existing_databases() { for previous_schema in [false, true] { @@ -3956,7 +4065,7 @@ mod tests { assert_eq!( translation, ( - "microsoft".to_string(), + "google".to_string(), String::new(), String::new(), String::new(), @@ -4065,7 +4174,7 @@ mod tests { assert_eq!( translation, ( - "microsoft".to_string(), + "google".to_string(), String::new(), String::new(), "https://api.openai.com/v1".to_string(), diff --git a/packages/app-lib/src/state/db_backup.rs b/packages/app-lib/src/state/db_backup.rs index abfe0e4..900cd42 100644 --- a/packages/app-lib/src/state/db_backup.rs +++ b/packages/app-lib/src/state/db_backup.rs @@ -39,6 +39,10 @@ async fn restore_corrupt_app_db_from( "App database integrity check failed" ); + if restore_app_db_without_corrupt_sidecars(db_path, backup_dir).await? { + return Ok(()); + } + let Some(backup_path) = latest_healthy_app_db_backup(backup_dir).await? else { return Err(crate::ErrorKind::FSError(format!( @@ -98,6 +102,49 @@ async fn restore_corrupt_app_db_from( Ok(()) } +async fn restore_app_db_without_corrupt_sidecars( + db_path: &Path, + backup_dir: &Path, +) -> crate::Result { + let mut has_sidecars = false; + for suffix in ["-wal", "-shm"] { + has_sidecars |= sqlite_sidecar_path(db_path, suffix).try_exists()?; + } + if !has_sidecars { + return Ok(false); + } + + let staging_path = next_restore_staging_path(db_path).await?; + tokio::fs::copy(db_path, &staging_path).await?; + let healthy_main = matches!( + check_database_integrity(&staging_path).await?, + IntegrityStatus::Healthy + ) && is_app_database(&staging_path).await?; + if !healthy_main { + cleanup_staged_database(&staging_path).await; + return Ok(false); + } + + crate::util::io::create_dir_all(backup_dir).await?; + let corrupt_path = next_corrupt_database_path(backup_dir).await?; + if let Err(error) = archive_database_files(db_path, &corrupt_path).await { + cleanup_staged_database(&staging_path).await; + return Err(error); + } + if let Err(error) = tokio::fs::rename(&staging_path, db_path).await { + restore_archived_database_files(&corrupt_path, db_path).await?; + cleanup_staged_database(&staging_path).await; + return Err(error.into()); + } + + tracing::warn!( + database = %db_path.display(), + corrupt_archive = %corrupt_path.display(), + "Recovered app database by removing corrupt SQLite sidecars" + ); + Ok(true) +} + async fn check_database_integrity( db_path: &Path, ) -> crate::Result { @@ -200,7 +247,7 @@ async fn is_app_database(db_path: &Path) -> crate::Result { Ok(table_count == REQUIRED_APP_DB_TABLES.len() as i64) } -async fn archive_database_files( +pub(crate) async fn archive_database_files( db_path: &Path, archive_path: &Path, ) -> crate::Result<()> { @@ -226,7 +273,7 @@ async fn archive_database_files( Ok(()) } -async fn restore_archived_database_files( +pub(crate) async fn restore_archived_database_files( archive_path: &Path, db_path: &Path, ) -> crate::Result<()> { @@ -285,7 +332,9 @@ fn sqlite_sidecar_path(db_path: &Path, suffix: &str) -> PathBuf { PathBuf::from(path) } -async fn remove_database_sidecars(db_path: &Path) -> crate::Result<()> { +pub(crate) async fn remove_database_sidecars( + db_path: &Path, +) -> crate::Result<()> { for suffix in ["-wal", "-shm"] { let path = sqlite_sidecar_path(db_path, suffix); match tokio::fs::remove_file(path).await { @@ -766,6 +815,45 @@ mod tests { assert_eq!(tokio::fs::read(&db_path).await.unwrap(), corrupt_bytes); } + #[tokio::test] + async fn corrupt_sidecars_recover_from_the_healthy_main_database() { + let temp = tempfile::tempdir().unwrap(); + let db_path = temp.path().join("release").join("app.db"); + let backup_dir = + temp.path().join("Backups").join("app-db").join("release"); + tokio::fs::create_dir_all(db_path.parent().unwrap()) + .await + .unwrap(); + create_test_app_db(&db_path, "release").await; + tokio::fs::write(sqlite_sidecar_path(&db_path, "-wal"), b"bad wal") + .await + .unwrap(); + tokio::fs::write(sqlite_sidecar_path(&db_path, "-shm"), b"bad shm") + .await + .unwrap(); + + assert!( + restore_app_db_without_corrupt_sidecars(&db_path, &backup_dir) + .await + .unwrap() + ); + + assert_eq!(read_marker(&db_path).await, "release"); + assert!(!sqlite_sidecar_path(&db_path, "-wal").exists()); + assert!(!sqlite_sidecar_path(&db_path, "-shm").exists()); + let mut entries = tokio::fs::read_dir(&backup_dir).await.unwrap(); + let mut archived_database = false; + let mut archived_wal = false; + let mut archived_shm = false; + while let Some(entry) = entries.next_entry().await.unwrap() { + let name = entry.file_name().to_string_lossy().into_owned(); + archived_database |= name.ends_with(".db"); + archived_wal |= name.ends_with(".db-wal"); + archived_shm |= name.ends_with(".db-shm"); + } + assert!(archived_database && archived_wal && archived_shm); + } + #[tokio::test] async fn update_backup_creates_a_healthy_channel_snapshot() { let temp = tempfile::tempdir().unwrap(); diff --git a/packages/app-lib/src/state/instances/adapters/sqlite/content_rows.rs b/packages/app-lib/src/state/instances/adapters/sqlite/content_rows.rs index b1cb298..edc3047 100644 --- a/packages/app-lib/src/state/instances/adapters/sqlite/content_rows.rs +++ b/packages/app-lib/src/state/instances/adapters/sqlite/content_rows.rs @@ -16,6 +16,12 @@ use chrono::{DateTime, TimeZone, Utc}; use sqlx::{Executor, Row, Sqlite, SqlitePool, Transaction}; use uuid::Uuid; +async fn begin_content_write( + pool: &SqlitePool, +) -> crate::Result> { + Ok(pool.begin_with("BEGIN IMMEDIATE").await?) +} + /// Ensures the instance a content write is about to reference still exists. /// /// Runs inside the write transaction so a concurrent instance deletion that @@ -662,7 +668,7 @@ pub(crate) async fn upsert_instance_file_from_parts( input: UpsertInstanceFile<'_>, pool: &SqlitePool, ) -> crate::Result { - let mut tx = pool.begin().await?; + let mut tx = begin_content_write(pool).await?; let file = upsert_instance_file_from_parts_in_transaction(input, &mut tx).await?; tx.commit().await?; @@ -726,7 +732,7 @@ pub(crate) async fn rename_instance_file( enabled: bool, pool: &SqlitePool, ) -> crate::Result> { - let mut tx = pool.begin().await?; + let mut tx = begin_content_write(pool).await?; let file = rename_instance_file_in_transaction( instance_id, @@ -1106,7 +1112,7 @@ pub(crate) async fn upsert_content_entry_from_parts( input: UpsertContentEntry<'_>, pool: &SqlitePool, ) -> crate::Result { - let mut tx = pool.begin().await?; + let mut tx = begin_content_write(pool).await?; let entry = upsert_content_entry_from_parts_in_transaction(input, &mut tx).await?; tx.commit().await?; @@ -1230,7 +1236,7 @@ pub(crate) async fn upsert_content_provider_ref( origin: bool, pool: &SqlitePool, ) -> crate::Result<()> { - let mut tx = pool.begin().await?; + let mut tx = begin_content_write(pool).await?; upsert_content_provider_ref_in_transaction( content_entry_id, provider_ref, @@ -2583,6 +2589,7 @@ mod tests { provider TEXT NOT NULL, provider_project_id TEXT NOT NULL, provider_release_id TEXT NULL, + provider_file_id TEXT NULL, is_origin INTEGER NOT NULL DEFAULT 0 ); CREATE UNIQUE INDEX instance_content_provider_refs_identity diff --git a/packages/app-lib/src/state/instances/commands/apply_content_install.rs b/packages/app-lib/src/state/instances/commands/apply_content_install.rs index c245521..7c7e8c2 100644 --- a/packages/app-lib/src/state/instances/commands/apply_content_install.rs +++ b/packages/app-lib/src/state/instances/commands/apply_content_install.rs @@ -539,7 +539,7 @@ pub(crate) async fn persist_resolved_plan_dependency_edges( .chain(plan.dependencies.iter()) .collect::>(); - let mut tx = state.pool.begin().await?; + let mut tx = begin_content_write(&state.pool).await?; for (index, dependency) in plan.dependencies.iter().enumerate() { let Some(parent) = dependency .dependent_on_version_id diff --git a/packages/app-lib/src/state/instances/commands/create_direct_link_instance.rs b/packages/app-lib/src/state/instances/commands/create_direct_link_instance.rs index eba44bc..23df8d4 100644 --- a/packages/app-lib/src/state/instances/commands/create_direct_link_instance.rs +++ b/packages/app-lib/src/state/instances/commands/create_direct_link_instance.rs @@ -254,8 +254,8 @@ mod tests { instance_rows::get_instance_by_id(&instance.id, &state.pool) .await? .expect("instance row"); - let canonical_root = minecraft.path().canonicalize()?; - let canonical_json = json_path.canonicalize()?; + let canonical_root = crate::util::io::canonicalize(minecraft.path())?; + let canonical_json = crate::util::io::canonicalize(&json_path)?; assert_eq!(stored.linked_launcher.as_deref(), Some("generic")); assert_eq!( stored.linked_launcher_root.as_deref(), diff --git a/packages/app-lib/src/state/process.rs b/packages/app-lib/src/state/process.rs index eb38453..7d9a18e 100644 --- a/packages/app-lib/src/state/process.rs +++ b/packages/app-lib/src/state/process.rs @@ -117,8 +117,21 @@ async fn record_post_upgrade_launch_best_effort( } struct LogRingBuffer { - lines: VecDeque, + lines: VecDeque, byte_len: usize, + last_sequence: u64, +} + +#[derive(Clone, Debug, Eq, PartialEq)] +pub struct SequencedLogLine { + pub sequence: u64, + pub message: String, +} + +#[derive(Clone, Debug, Eq, PartialEq)] +pub struct LiveLogSnapshot { + pub lines: Vec, + pub last_sequence: u64, } impl LogRingBuffer { @@ -126,25 +139,37 @@ impl LogRingBuffer { Self { lines: VecDeque::new(), byte_len: 0, + last_sequence: 0, } } - fn push(&mut self, line: String) { - let line_len = line.len(); + fn push(&mut self, message: String) -> u64 { + self.last_sequence = self.last_sequence.saturating_add(1); + let sequence = self.last_sequence; + let line = SequencedLogLine { sequence, message }; + let line_len = line.message.len(); while self.lines.len() >= LOG_BUFFER_CAPACITY || self.byte_len.saturating_add(line_len) > LOG_BUFFER_BYTE_CAPACITY { let Some(removed) = self.lines.pop_front() else { break; }; - self.byte_len = self.byte_len.saturating_sub(removed.len()); + self.byte_len = self.byte_len.saturating_sub(removed.message.len()); } self.byte_len += line_len; self.lines.push_back(line); + sequence } fn get_all(&self) -> Vec { - self.lines.iter().cloned().collect() + self.lines.iter().map(|line| line.message.clone()).collect() + } + + fn snapshot(&self) -> LiveLogSnapshot { + LiveLogSnapshot { + lines: self.lines.iter().cloned().collect(), + last_sequence: self.last_sequence, + } } fn clear(&mut self) { @@ -157,11 +182,24 @@ static LOG_BUFFERS: LazyLock> = LazyLock::new(DashMap::new); pub fn push_log_line(instance_id: &str, line: String) { - let line = truncate_live_log_text(&line); - LOG_BUFFERS + push_log_lines(instance_id, std::iter::once(line)); +} + +fn push_log_lines( + instance_id: &str, + lines: impl IntoIterator, +) -> Option<(u64, u64)> { + let mut buffer = LOG_BUFFERS .entry(instance_id.to_string()) - .or_insert_with(LogRingBuffer::new) - .push(line); + .or_insert_with(LogRingBuffer::new); + let mut first_sequence = None; + let mut last_sequence = 0; + for line in lines { + let sequence = buffer.push(truncate_live_log_text(&line)); + first_sequence.get_or_insert(sequence); + last_sequence = sequence; + } + first_sequence.map(|first| (first, last_sequence)) } fn truncate_live_log_text(value: &str) -> String { @@ -228,6 +266,16 @@ pub fn get_log_buffer(instance_id: &str) -> Vec { .unwrap_or_default() } +pub fn get_live_log_snapshot(instance_id: &str) -> LiveLogSnapshot { + LOG_BUFFERS + .get(instance_id) + .map(|buffer| buffer.snapshot()) + .unwrap_or(LiveLogSnapshot { + lines: Vec::new(), + last_sequence: 0, + }) +} + pub fn clear_log_buffer(instance_id: &str) { if let Some(mut buf) = LOG_BUFFERS.get_mut(instance_id) { buf.clear(); @@ -920,14 +968,20 @@ impl Process { event.throwable = event.throwable.as_deref().map(truncate_live_log_text); + let mut lines = Vec::new(); if let Some(formatted) = Self::format_log4j_entry(&event) { - push_log_line(instance_id, formatted.trim_end().to_string()); + lines.push(formatted.trim_end().to_string()); } if let Some(ref throwable) = event.throwable { for line in throwable.lines().filter(|l| !l.is_empty()) { - push_log_line(instance_id, line.to_string()); + lines.push(line.to_string()); } } + let Some((first_sequence, last_sequence)) = + push_log_lines(instance_id, lines) + else { + return; + }; #[cfg(feature = "tauri")] { @@ -936,6 +990,8 @@ impl Process { "log", LogPayload { instance_id: instance_id.to_string(), + first_sequence, + last_sequence, event: LogEvent::Log4j(event), }, ); @@ -943,13 +999,17 @@ impl Process { } #[cfg(not(feature = "tauri"))] { - let _ = (instance_id, event); + let _ = (instance_id, event, first_sequence, last_sequence); } } fn emit_legacy_log(instance_id: &str, message: &str) { let message = truncate_live_log_text(message); - push_log_line(instance_id, message.clone()); + let Some((first_sequence, last_sequence)) = + push_log_lines(instance_id, std::iter::once(message.clone())) + else { + return; + }; #[cfg(feature = "tauri")] { @@ -958,6 +1018,8 @@ impl Process { "log", LogPayload { instance_id: instance_id.to_string(), + first_sequence, + last_sequence, event: LogEvent::Legacy { message }, }, ); @@ -965,7 +1027,7 @@ impl Process { } #[cfg(not(feature = "tauri"))] { - let _ = (instance_id, message); + let _ = (instance_id, message, first_sequence, last_sequence); } } @@ -1312,6 +1374,49 @@ fn take_next_log4j_frame(buffer: &mut String) -> Option { mod post_upgrade_tests { use super::*; + #[test] + fn live_log_snapshot_sequences_batches_and_preserves_position_after_clear() + { + let instance_id = Uuid::new_v4().to_string(); + assert_eq!( + push_log_lines( + &instance_id, + ["first".to_string(), "second".to_string()] + ), + Some((1, 2)) + ); + assert_eq!( + get_live_log_snapshot(&instance_id), + LiveLogSnapshot { + lines: vec![ + SequencedLogLine { + sequence: 1, + message: "first".to_string(), + }, + SequencedLogLine { + sequence: 2, + message: "second".to_string(), + }, + ], + last_sequence: 2, + } + ); + + clear_log_buffer(&instance_id); + push_log_line(&instance_id, "third".to_string()); + assert_eq!( + get_live_log_snapshot(&instance_id), + LiveLogSnapshot { + lines: vec![SequencedLogLine { + sequence: 3, + message: "third".to_string(), + }], + last_sequence: 3, + } + ); + remove_log_buffer(&instance_id); + } + #[cfg(windows)] #[tokio::test] async fn waiting_for_one_process_leaves_registry_available() { diff --git a/packages/app-lib/src/util/fetch.rs b/packages/app-lib/src/util/fetch.rs index 714299f..882dd81 100644 --- a/packages/app-lib/src/util/fetch.rs +++ b/packages/app-lib/src/util/fetch.rs @@ -958,6 +958,34 @@ fn official_fallback_route(routes: &[DownloadRoute]) -> Option { .cloned() } +fn integrity_fallback_route( + routes: &[DownloadRoute], + current_index: usize, + terminal_routes: &HashSet, +) -> Option { + let current = routes.get(current_index)?; + if !is_official_route(current) + && let Some(official) = routes.iter().enumerate().find_map( + |(candidate_index, candidate)| { + (candidate_index != current_index + && is_official_route(candidate) + && !terminal_routes.contains(&candidate.url)) + .then(|| candidate.clone()) + }, + ) + { + return Some(official); + } + + ((current_index + 1)..routes.len()) + .chain(0..current_index) + .find_map(|candidate_index| { + let candidate = &routes[candidate_index]; + (!terminal_routes.contains(&candidate.url)) + .then(|| candidate.clone()) + }) +} + fn url_with_base(original: &Url, base: &str, path: &str) -> Option { let mut target = Url::parse(base).ok()?; target.set_path(path); @@ -6048,13 +6076,15 @@ async fn download_to_path_inner( ); last_error = Some(error); terminal_routes.insert(route.url.clone()); - if !is_official_route(route) - && let Some(official) = - official_fallback_route(&routes) - { + if let Some(fallback) = integrity_fallback_route( + &routes, + route_index, + &terminal_routes, + ) { remove_if_exists(&part_path).await?; - official_integrity_retry = true; - preferred_route = Some(official); + official_integrity_retry = + is_official_route(&fallback); + preferred_route = Some(fallback); break; } if official_integrity_retry { @@ -6210,6 +6240,9 @@ async fn download_to_path_inner( "File download request failed; trying the next source or retry" ); last_error = Some(error); + if can_switch_route { + continue; + } break; } Err(_) => { @@ -6244,6 +6277,9 @@ async fn download_to_path_inner( "File download stalled before receiving a response" ); last_error = Some(error); + if can_switch_route { + continue; + } break; } }; @@ -6346,6 +6382,11 @@ async fn download_to_path_inner( Some(http_version), ); last_error = Some(error); + if (terminal_status || cooldown_and_switch) + && can_switch_route + { + continue; + } break; } @@ -6773,19 +6814,24 @@ async fn download_to_path_inner( } else { remove_if_exists(&part_path).await?; } - let official = (!is_official_route(route)) - .then(|| official_fallback_route(&routes)) - .flatten(); - let decision = if official.is_some() { - "fallback_official" - } else if attempts >= 2 - || (is_official_route(route) - && official_integrity_retry) - { - "drop_route_after_clean_retry" - } else { - "clear_partial_and_retry" - }; + let fallback = integrity_fallback_route( + &routes, + route_index, + &terminal_routes, + ); + let decision = + if fallback.as_ref().is_some_and(is_official_route) { + "fallback_official" + } else if fallback.is_some() { + "switch_route" + } else if attempts >= 2 + || (is_official_route(route) + && official_integrity_retry) + { + "drop_route_after_clean_retry" + } else { + "clear_partial_and_retry" + }; record_download_attempt_failure( &mut attempt_history, route, @@ -6803,10 +6849,10 @@ async fn download_to_path_inner( )); } last_error = Some(error); - if let Some(official) = official { + if let Some(fallback) = fallback { terminal_routes.insert(route.url.clone()); - official_integrity_retry = true; - preferred_route = Some(official); + official_integrity_retry = is_official_route(&fallback); + preferred_route = Some(fallback); } else if (is_official_route(route) && official_integrity_retry) || attempts >= 2 @@ -8121,7 +8167,7 @@ mod tests { } #[test] - fn h2_fallback_is_limited_to_one_second() { + fn h2_fallback_uses_the_configured_ttl() { let _guard = H2_FALLBACK_TEST_LOCK.lock(); let authority = "h2-cooldown.example:443"; record_authority_h2_failure(authority); @@ -8131,7 +8177,8 @@ mod tests { .get(authority) .map(|until| until.saturating_duration_since(Instant::now())) .unwrap(); - assert!(remaining <= MAX_FAILURE_COOLDOWN); + assert!(remaining <= H2_FALLBACK_TTL); + assert!(remaining > H2_FALLBACK_TTL - Duration::from_secs(1)); H2_FALLBACK_AUTHORITIES.lock().remove(authority); } diff --git a/packages/ui/src/layouts/shared/console/components/LogViewport.test.ts b/packages/ui/src/layouts/shared/console/components/LogViewport.test.ts new file mode 100644 index 0000000..75a105f --- /dev/null +++ b/packages/ui/src/layouts/shared/console/components/LogViewport.test.ts @@ -0,0 +1,21 @@ +import assert from 'node:assert/strict' +import { readFileSync } from 'node:fs' +import test from 'node:test' + +const source = readFileSync( + new URL('./LogViewport.vue', import.meta.url), + 'utf8', +) + +test('log rows use content-driven grid height for wrapped lines', () => { + assert.match(source, /grid-template-columns:\s*52px minmax\(0, 1fr\)/) + assert.match(source, /\.log-line-content\s*\{[^}]*display:\s*block/s) + assert.match( + source, + /\.log-viewport-wrap \.log-line\s*\{[^}]*white-space:\s*pre-wrap/s, + ) + assert.doesNotMatch( + source, + /content-visibility|contain-intrinsic-size|estimateHeight|heightPrefix/, + ) +}) diff --git a/packages/ui/src/layouts/shared/console/components/LogViewport.vue b/packages/ui/src/layouts/shared/console/components/LogViewport.vue index e6a31af..5b04c8d 100644 --- a/packages/ui/src/layouts/shared/console/components/LogViewport.vue +++ b/packages/ui/src/layouts/shared/console/components/LogViewport.vue @@ -6,7 +6,10 @@ :style="{ fontSize: fontSize + 'px' }" @scroll="handleScroll" > -
+
- {{ item.originalIndex + 1 }} - + {{ item.originalIndex + 1 }} +
-
+