diff --git a/Cargo.lock b/Cargo.lock index 9308dfc5..36a730e6 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3780,6 +3780,7 @@ dependencies = [ "governor", "indicatif", "interpulse", + "libc", "md-5", "merge", "murmur2", diff --git a/apps/oneclient/desktop/src/ext/updater.rs b/apps/oneclient/desktop/src/ext/updater.rs index 6c137fb8..13474c3e 100644 --- a/apps/oneclient/desktop/src/ext/updater.rs +++ b/apps/oneclient/desktop/src/ext/updater.rs @@ -1,5 +1,5 @@ use onelauncher_core::store::Core; -use tauri::{Emitter, Manager, Runtime}; +use tauri::{Manager, Runtime}; use tauri_plugin_updater::{Update as TauriPluginUpdate, UpdaterExt}; use tokio::sync::Mutex; @@ -44,28 +44,9 @@ pub enum UpdateEvent { pub async fn check_for_update( app: tauri::AppHandle, ) -> Result, String> { - app.emit("updater", UpdateEvent::Loading).ok(); - - let update = match get_update(app.clone()).await { - Ok(update) => update, - Err(e) => { - app.emit("updater", UpdateEvent::Error { error: e.clone() }) - .ok(); - return Err(e); - } - }; - - let update = update.map(|u| Update::new(&u)); - - app.emit( - "updater", - update.clone().map_or(UpdateEvent::NoUpdateAvailable, |u| { - UpdateEvent::UpdateAvailable { update: u } - }), - ) - .ok(); - - Ok(update) + get_update(app) + .await + .map(|update| update.map(|u| Update::new(&u))) } pub async fn install_update(app: tauri::AppHandle) -> Result<(), String> { @@ -74,8 +55,6 @@ pub async fn install_update(app: tauri::AppHandle) -> Result<(), return Err("Update already installing".into()); }; - app.emit("updater", UpdateEvent::Installing).ok(); - get_update(app.clone()) .await? .ok_or_else(|| "No update required".to_string())? diff --git a/apps/oneclient/desktop/src/oneclient/bundle_updates.rs b/apps/oneclient/desktop/src/oneclient/bundle_updates.rs index 636cbd9a..787d1c40 100644 --- a/apps/oneclient/desktop/src/oneclient/bundle_updates.rs +++ b/apps/oneclient/desktop/src/oneclient/bundle_updates.rs @@ -197,6 +197,25 @@ async fn check_bundle_updates_inner( "Retrieved available bundles" ); + let tracked_bundle_names: std::collections::HashSet = bundle_packages + .iter() + .filter_map(|bp| bp.bundle_name.clone()) + .collect(); + let loaded_bundle_names: std::collections::HashSet = + bundles.iter().map(|b| b.manifest.name.clone()).collect(); + let unavailable_tracked_bundles: std::collections::HashSet = tracked_bundle_names + .into_iter() + .filter(|name| !loaded_bundle_names.contains(name)) + .collect(); + + if !unavailable_tracked_bundles.is_empty() { + tracing::warn!( + cluster_id = %cluster_id, + unavailable_tracked_bundles = ?unavailable_tracked_bundles, + "Some tracked bundles could not be loaded; skipping updates/removals for those bundles" + ); + } + let mut bundle_versions: std::collections::HashMap< String, std::collections::HashMap, // key: managed(provider+id) or external(sha1) @@ -358,6 +377,7 @@ async fn check_bundle_updates_inner( let mut skipped_no_package_id = 0; let mut skipped_no_version_id = 0; let mut skipped_no_provider = 0; + let mut skipped_unavailable_bundle = 0; let mut not_in_bundle = 0; for bundle_pkg in &bundle_packages { @@ -395,6 +415,17 @@ async fn check_bundle_updates_inner( }; if let Some(ref bundle_name) = bundle_pkg.bundle_name { + if unavailable_tracked_bundles.contains(bundle_name) { + skipped_unavailable_bundle += 1; + tracing::warn!( + package_id = %pkg_id, + package_hash = %bundle_pkg.package_hash, + bundle_name = %bundle_name, + "Skipping package update check because tracked bundle metadata is unavailable" + ); + continue; + } + let mut matched_target: Option<(String, String, ModpackFile)> = bundle_versions.get(bundle_name).and_then(|files_map| { files_map @@ -571,6 +602,7 @@ async fn check_bundle_updates_inner( skipped_no_package_id = %skipped_no_package_id, skipped_no_version_id = %skipped_no_version_id, skipped_no_provider = %skipped_no_provider, + skipped_unavailable_bundle = %skipped_unavailable_bundle, not_in_bundle = %not_in_bundle, "Bundle update check completed" ); diff --git a/apps/oneclient/desktop/src/oneclient/bundles.rs b/apps/oneclient/desktop/src/oneclient/bundles.rs index 79c070ee..2e76ec1f 100644 --- a/apps/oneclient/desktop/src/oneclient/bundles.rs +++ b/apps/oneclient/desktop/src/oneclient/bundles.rs @@ -191,40 +191,53 @@ async fn download_and_load_bundle( if disk_path.exists() { // Check if the remote file differs from the local cached copy. - let res = http::request(Method::HEAD, &url).await?; + // If the validation request itself fails, trust the local cache instead of + // treating the bundle as unavailable. + match http::request(Method::HEAD, &url).await { + Ok(res) if res.status().is_success() => { + let server_etag = res + .headers() + .get(reqwest::header::ETAG) + .and_then(|v| v.to_str().ok()) + .map(|s| s.to_string()); + let stored_etag: Option = tokio::fs::read_to_string(&etag_path).await.ok(); - if !res.status().is_success() { - return Err(anyhow::anyhow!("failed to download bundle from remote: {}", url).into()); - } + if let (Some(server), Some(stored)) = (&server_etag, &stored_etag) { + // ETag comparison: most reliable cache validation method. + if server == stored { + tracing::debug!("bundle cache hit via ETag for: {url}"); + return Ok(ModpackFormat::from_file(disk_path).await?); + } + } else { + // Fall back to Content-Length comparison when no ETag is available. + let content_length = res + .headers() + .get(header::CONTENT_LENGTH) + .and_then(|v| v.to_str().ok()) + .and_then(|v| v.parse::().ok()); - let server_etag = res - .headers() - .get(reqwest::header::ETAG) - .and_then(|v| v.to_str().ok()) - .map(|s| s.to_string()); - let stored_etag: Option = tokio::fs::read_to_string(&etag_path).await.ok(); - - if let (Some(server), Some(stored)) = (&server_etag, &stored_etag) { - // ETag comparison: most reliable cache validation method. - if server == stored { - tracing::debug!("bundle cache hit via ETag for: {url}"); - return Ok(ModpackFormat::from_file(disk_path).await?); - } - } else { - // Fall back to Content-Length comparison when no ETag is available. - let content_length = res - .headers() - .get(header::CONTENT_LENGTH) - .and_then(|v| v.to_str().ok()) - .and_then(|v| v.parse::().ok()); - - if let Some(length) = content_length { - let file_size = io::stat(disk_path).await.map(|m| m.len()).unwrap_or(0); - if length == file_size { - return Ok(ModpackFormat::from_file(disk_path).await?); + if let Some(length) = content_length { + let file_size = io::stat(disk_path).await.map(|m| m.len()).unwrap_or(0); + if length == file_size { + return Ok(ModpackFormat::from_file(disk_path).await?); + } + } + // If neither ETag nor Content-Length are available, fall through to re-download. } } - // If neither ETag nor Content-Length are available, fall through to re-download. + Ok(res) => { + tracing::warn!( + status = %res.status(), + "failed to validate remote bundle cache for {url}; using cached local bundle" + ); + return Ok(ModpackFormat::from_file(disk_path).await?); + } + Err(e) => { + tracing::warn!( + "failed to validate remote bundle cache for {url}: {e}; using cached local bundle" + ); + return Ok(ModpackFormat::from_file(disk_path).await?); + } } } diff --git a/apps/oneclient/frontend/src/bindings.gen.ts b/apps/oneclient/frontend/src/bindings.gen.ts index 59f0b667..94a3f209 100644 --- a/apps/oneclient/frontend/src/bindings.gen.ts +++ b/apps/oneclient/frontend/src/bindings.gen.ts @@ -222,6 +222,8 @@ export type Process = { pid: number; started_at: string; cluster_id: number; pos export type ProcessError = { type: "HookUnsuccessful"; data: string } | { type: "NoPid"; data: string } +export type ProcessLogTail = { content: string; truncated: boolean } + export type ProcessPayload = { cluster_id: number; kind: ProcessPayloadKind } export type ProcessPayloadKind = { type: "Starting"; command: string } | { type: "Started"; pid: number; started_at: string; post_hook: string | null; account_id: string } | { type: "Stopped"; pid: number; exit_code: number } | { type: "Output"; pid: number; output: string } @@ -311,8 +313,19 @@ export type VersionType = */ "old_beta" -const ARGS_MAP = { 'folders':'{"openCluster":["folder_name"],"fromCluster":["folder_name"]}', 'oneclient':'{"getClustersGroupedByMajor":[],"checkForUpdate":[],"downloadPackageFromBundle":["package","cluster_id","bundle_name","skip_compatibility"],"extractBundleOverrides":["bundle_path","cluster_id"],"updateBundlePackages":["cluster_id"],"getVersions":[],"isBundleSyncing":[],"installUpdate":[],"getBundlesFor":["cluster_id"]}', 'events':'{"ingress":["event"],"message":["event"],"process":["event"]}', 'debug':'{"getGitCommitHash":[],"getFamily":[],"getArch":[],"getType":[],"isInDev":[],"getPlatform":[],"openDevTools":[],"getOsVersion":[],"getBuildTimestamp":[],"getPackageVersion":[],"getLocale":[]}', 'core':'{"downloadPackage":["provider","package_id","version_id","cluster_id","skip_compatibility"],"uploadSkinBytes":["access_token","skin_data","image_format","skin_variant"],"getPackageBody":["provider","body"],"createCluster":["options"],"getMultiplePackages":["provider","slugs"],"refreshAccounts":[],"changeSkin":["access_token","skin_url","skin_variant"],"fetchLoggedInProfile":["access_token"],"getLinkedPackages":["cluster_id"],"searchPackages":["provider","query"],"removeCape":["access_token"],"installModpack":["modpack","cluster_id"],"updateClusterProfile":["name","profile"],"getLoadersForVersion":["mc_version"],"getUsers":[],"isClusterRunning":["cluster_id"],"setDefaultUser":["uuid"],"getUsersFromAuthor":["provider","author"],"syncCluster":["cluster_id"],"fetchMinecraftProfile":["uuid"],"killProcess":["pid"],"setDiscordRPCMessage":["message"],"getPackage":["provider","slug"],"changeCape":["access_token","cape_uuid"],"getGameVersions":[],"getLogs":["id"],"getRunningProcessesByClusterId":["cluster_id"],"getProfileOrDefault":["name"],"getPackageVersions":["provider","slug","mc_version","loader","offset","limit"],"setClusterStage":["id","stage"],"createSettingsProfile":["name"],"getDefaultUser":["fallback"],"removeUser":["uuid"],"writeSettings":["setting"],"downloadExternalPackage":["package","cluster_id","force","skip_compatibility"],"refreshAccount":["uuid"],"getClusterById":["id"],"removePackage":["cluster_id","package_hash"],"getUser":["uuid"],"getWorlds":["id"],"togglePackage":["cluster_id","package_hash"],"convertUsernameUUID":["username_uuid"],"getLogByName":["id","name"],"launchCluster":["id","uuid","search_for_java"],"updateClusterById":["id","request"],"removeCluster":["id"],"getScreenshots":["id"],"getClusters":[],"getRunningProcesses":[],"getGlobalProfile":[],"openMsaLogin":[],"readSettings":[],"open":["input"]}' } -export type Router = { 'oneclient': { getClustersGroupedByMajor: () => Promise>, +const ARGS_MAP = { 'folders':'{"openCluster":["folder_name"],"fromCluster":["folder_name"]}', 'oneclient':'{"downloadPackageFromBundle":["package","cluster_id","bundle_name","skip_compatibility"],"isBundleSyncing":[],"getBundlesFor":["cluster_id"],"updateBundlePackages":["cluster_id"],"extractBundleOverrides":["bundle_path","cluster_id"],"getVersions":[],"getClustersGroupedByMajor":[],"installUpdate":[],"checkForUpdate":[]}', 'core':'{"open":["input"],"searchPackages":["provider","query"],"togglePackage":["cluster_id","package_hash"],"getLogByName":["id","name"],"fetchLoggedInProfile":["access_token"],"removeCluster":["id"],"removeCape":["access_token"],"getRunningProcessesByClusterId":["cluster_id"],"createSettingsProfile":["name"],"killProcess":["pid"],"setClusterStage":["id","stage"],"writeSettings":["setting"],"getLoadersForVersion":["mc_version"],"readSettings":[],"createCluster":["options"],"launchCluster":["id","uuid","search_for_java"],"getUser":["uuid"],"getPackageBody":["provider","body"],"getMultiplePackages":["provider","slugs"],"getPackageVersions":["provider","slug","mc_version","loader","offset","limit"],"setDefaultUser":["uuid"],"getLinkedPackages":["cluster_id"],"syncCluster":["cluster_id"],"downloadExternalPackage":["package","cluster_id","force","skip_compatibility"],"getDefaultUser":["fallback"],"removeUser":["uuid"],"fetchMinecraftProfile":["uuid"],"removePackage":["cluster_id","package_hash"],"uploadSkinBytes":["access_token","skin_data","image_format","skin_variant"],"changeCape":["access_token","cape_uuid"],"getRunningProcesses":[],"getProfileOrDefault":["name"],"convertUsernameUUID":["username_uuid"],"getScreenshots":["id"],"getLogs":["id"],"getProcessLogTail":["id","max_lines"],"getClusterById":["id"],"getClusters":[],"updateClusterById":["id","request"],"updateClusterProfile":["name","profile"],"refreshAccounts":[],"refreshAccount":["uuid"],"getGameVersions":[],"getPackage":["provider","slug"],"getUsersFromAuthor":["provider","author"],"getWorlds":["id"],"getUsers":[],"downloadPackage":["provider","package_id","version_id","cluster_id","skip_compatibility"],"installModpack":["modpack","cluster_id"],"setDiscordRPCMessage":["message"],"getGlobalProfile":[],"isClusterRunning":["cluster_id"],"changeSkin":["access_token","skin_url","skin_variant"],"openMsaLogin":[]}', 'events':'{"ingress":["event"],"process":["event"],"message":["event"]}', 'debug':'{"getPackageVersion":[],"getType":[],"openDevTools":[],"isInDev":[],"getArch":[],"getPlatform":[],"getOsVersion":[],"getFamily":[],"getGitCommitHash":[],"getLocale":[],"getBuildTimestamp":[]}' } +export type Router = { 'debug': { openDevTools: () => Promise, +isInDev: () => Promise, +getArch: () => Promise, +getFamily: () => Promise, +getLocale: () => Promise, +getType: () => Promise, +getPlatform: () => Promise, +getOsVersion: () => Promise, +getGitCommitHash: () => Promise, +getBuildTimestamp: () => Promise, +getPackageVersion: () => Promise }, +'oneclient': { getClustersGroupedByMajor: () => Promise>, getBundlesFor: (clusterId: number) => Promise, getVersions: () => Promise, extractBundleOverrides: (bundlePath: string, clusterId: number) => Promise, @@ -321,11 +334,6 @@ installUpdate: () => Promise, downloadPackageFromBundle: (package: ModpackFileKind, clusterId: number, bundleName: string, skipCompatibility: boolean | null) => Promise, updateBundlePackages: (clusterId: number) => Promise, isBundleSyncing: () => Promise }, -'events': { ingress: (event: IngressPayload) => Promise, -message: (event: MessagePayload) => Promise, -process: (event: ProcessPayload) => Promise }, -'folders': { fromCluster: (folderName: string) => Promise, -openCluster: (folderName: string) => Promise }, 'core': { getClusters: () => Promise, getClusterById: (id: number) => Promise, removeCluster: (id: number) => Promise, @@ -337,6 +345,7 @@ getScreenshots: (id: number) => Promise, getWorlds: (id: number) => Promise, getLogs: (id: number) => Promise, getLogByName: (id: number, name: string) => Promise, +getProcessLogTail: (id: number, maxLines: number) => Promise, getRunningProcesses: () => Promise, getRunningProcessesByClusterId: (clusterId: number) => Promise, isClusterRunning: (clusterId: number) => Promise, @@ -379,17 +388,11 @@ removeCape: (accessToken: string) => Promise, convertUsernameUUID: (usernameUuid: string) => Promise, setDiscordRPCMessage: (message: string) => Promise, open: (input: string) => Promise }, -'debug': { openDevTools: () => Promise, -isInDev: () => Promise, -getArch: () => Promise, -getFamily: () => Promise, -getLocale: () => Promise, -getType: () => Promise, -getPlatform: () => Promise, -getOsVersion: () => Promise, -getGitCommitHash: () => Promise, -getBuildTimestamp: () => Promise, -getPackageVersion: () => Promise } }; +'events': { ingress: (event: IngressPayload) => Promise, +message: (event: MessagePayload) => Promise, +process: (event: ProcessPayload) => Promise }, +'folders': { fromCluster: (folderName: string) => Promise, +openCluster: (folderName: string) => Promise } }; export type { InferCommandOutput } diff --git a/apps/oneclient/frontend/src/components/LogViewer.tsx b/apps/oneclient/frontend/src/components/LogViewer.tsx index d3299761..61e3094e 100644 --- a/apps/oneclient/frontend/src/components/LogViewer.tsx +++ b/apps/oneclient/frontend/src/components/LogViewer.tsx @@ -1,34 +1,68 @@ import { useVirtualizer } from '@tanstack/react-virtual'; import { OverlayScrollbarsComponent } from 'overlayscrollbars-react'; -import { useCallback, useImperativeHandle, useMemo, useRef, useState } from 'react'; +import { useCallback, useEffect, useImperativeHandle, useRef, useState } from 'react'; import styles from './LogViewer.module.css'; export interface LogViewerProps { content: string; scrollRef: React.RefObject; autoScroll?: boolean; + maxLines?: number; + onTrimmedLines?: (trimmedLines: number) => void; ref?: React.Ref; } export interface LogViewerRef { push: (line: string) => void; - // search: (query: string) => void; + scrollToEnd: () => void; } const LINE_HEIGHT = 16; const REGEX_PATTERN = /\[((?:\S+ )?\d+:\d+:\d+(?:\.\d+)?)\] \[(.[^(\n\r/\u2028\u2029]*)\/(\w+)\]:? (?:\[(CHAT)\])?/; +function normalizeLineLimit(maxLines?: number): number { + if (maxLines === undefined || !Number.isFinite(maxLines)) + return Number.POSITIVE_INFINITY; + + return Math.max(1, Math.floor(maxLines)); +} + +interface TrimResult { + lines: Array; + trimmedCount: number; +} + +function trimLines( + lines: Array, + lineLimit: number, +): TrimResult { + if (!Number.isFinite(lineLimit) || lines.length <= lineLimit) + return { + lines, + trimmedCount: 0, + }; + + const trimmedCount = lines.length - lineLimit; + return { + lines: lines.slice(trimmedCount), + trimmedCount, + }; +} + export function LogViewer({ content, scrollRef, autoScroll = false, + maxLines, + onTrimmedLines, ref, }: LogViewerProps) { - const lines = useMemo>(() => content.split('\n'), [content]); + const lineLimit = normalizeLineLimit(maxLines); const containerRef = useRef(null); - - // used to force re-render (and re-init of virtualizer hook) - const [_lastUpdated, setLastUpdated] = useState(Date.now()); + const pendingLinesRef = useRef>([]); + const frameRef = useRef(null); + const [lines, setLines] = useState>(() => trimLines(content.split('\n'), lineLimit).lines); + const linesRef = useRef>(lines); const virtualizer = useVirtualizer({ count: lines.length, @@ -45,13 +79,52 @@ export function LogViewer({ requestAnimationFrame(() => scrollRef.current?.scrollTo({ top: scrollRef.current.scrollHeight })); }, [scrollRef]); - const appendLine = useCallback((line: string) => { - setLastUpdated(Date.now()); - lines.push(line); + useEffect(() => { + pendingLinesRef.current = []; + if (frameRef.current !== null) { + cancelAnimationFrame(frameRef.current); + frameRef.current = null; + } - if (autoScroll) - scrollToEnd(); - }, [autoScroll, lines, scrollToEnd]); + const nextState = trimLines(content.split('\n'), lineLimit); + linesRef.current = nextState.lines; + setLines(nextState.lines); + + if (nextState.trimmedCount > 0) + onTrimmedLines?.(nextState.trimmedCount); + }, [content, lineLimit, onTrimmedLines]); + + useEffect(() => { + return () => { + if (frameRef.current !== null) + cancelAnimationFrame(frameRef.current); + }; + }, []); + + const appendLine = useCallback((line: string) => { + pendingLinesRef.current.push(line); + if (frameRef.current !== null) + return; + + frameRef.current = requestAnimationFrame(() => { + frameRef.current = null; + if (pendingLinesRef.current.length === 0) + return; + + const pending = pendingLinesRef.current; + pendingLinesRef.current = []; + + const nextState = trimLines(linesRef.current.concat(pending), lineLimit); + linesRef.current = nextState.lines; + setLines(nextState.lines); + + if (nextState.trimmedCount > 0) + onTrimmedLines?.(nextState.trimmedCount); + + if (autoScroll) + scrollToEnd(); + }); + }, [autoScroll, lineLimit, onTrimmedLines, scrollToEnd]); useImperativeHandle(ref, () => ({ push: appendLine, diff --git a/apps/oneclient/frontend/src/hooks/useClusterDirWatch.ts b/apps/oneclient/frontend/src/hooks/useClusterDirWatch.ts index 6f39aef3..30ae072c 100644 --- a/apps/oneclient/frontend/src/hooks/useClusterDirWatch.ts +++ b/apps/oneclient/frontend/src/hooks/useClusterDirWatch.ts @@ -1,8 +1,125 @@ +import type { DebouncedWatchOptions, WatchEvent } from '@tauri-apps/plugin-fs'; import { useLastPlayedClusters } from '@/hooks/useClusters'; import { bindings } from '@/main'; import { useQueryClient } from '@tanstack/react-query'; -import { watch } from '@tauri-apps/plugin-fs'; -import { useEffect } from 'react'; +import { invoke, SERIALIZE_TO_IPC_FN, transformCallback } from '@tauri-apps/api/core'; +import { useEffect, useMemo } from 'react'; + +const WATCH_DELAY_MS = 1000; +const SYNC_DEBOUNCE_MS = 1000; +const STALE_WATCH_CALLBACK_TTL_MS = 30_000; + +interface WatchTarget { + id: number; + folderName: string; +} + +interface ChannelMessage { + message: T; + index: number; +} + +interface ChannelEndMessage { + end: true; + index: number; +} + +type RawChannelMessage = ChannelMessage | ChannelEndMessage; + +class WatchChannel { + private readonly callbackId: number; + private disposed = false; + + constructor(onMessage: (message: T) => void) { + this.callbackId = transformCallback((rawMessage: RawChannelMessage) => { + if (this.disposed) + return; + + if (!('message' in rawMessage)) + return; + + onMessage(rawMessage.message); + }, false); + } + + [SERIALIZE_TO_IPC_FN]() { + return `__CHANNEL__:${this.callbackId}`; + } + + toJSON() { + return this[SERIALIZE_TO_IPC_FN](); + } + + scheduleCleanup() { + const callbackKey = `_${this.callbackId}`; + const callbacks = window as unknown as Record; + this.disposed = true; + + // Keep a noop callback briefly after unwatch to absorb late native events. + callbacks[callbackKey] = () => {}; + window.setTimeout(() => { + Reflect.deleteProperty(window, callbackKey); + }, STALE_WATCH_CALLBACK_TTL_MS); + } +} + +async function watchSafely( + path: string, + callback: (event: WatchEvent) => void, + options: DebouncedWatchOptions, +) { + const onEvent: WatchChannel = new WatchChannel(callback); + const rid = await invoke('plugin:fs|watch', { + paths: [path], + options: { + recursive: false, + delayMs: 2000, + ...options, + }, + onEvent, + }); + + let closed = false; + + return () => { + if (closed) + return; + + closed = true; + void invoke('plugin:fs|unwatch', { rid }).finally(() => { + onEvent.scheduleCleanup(); + }); + }; +} + +function normalizePath(path: string): string { + return path.replaceAll('\\', '/').toLowerCase(); +} + +function isStructuralChange(event: WatchEvent): boolean { + if (typeof event.type === 'string') + return false; + + return 'create' in event.type || 'remove' in event.type; +} + +function shouldIgnoreEvent(event: WatchEvent): boolean { + // Ignore plain file edits (modify/metadata/data writes). Sync only on + // structural changes (add/remove file or folder). + if (!isStructuralChange(event)) + return true; + + if (event.paths.length === 0) + return false; + + return event.paths.every((rawPath) => { + const path = normalizePath(rawPath); + return path.includes('/logs/') + || path.includes('/crash-reports/') + || path.includes('/saves/') + || path.endsWith('/latest.log'); + }); +} /** * Watches ALL cluster directories for file-system changes (files added, @@ -12,41 +129,129 @@ import { useEffect } from 'react'; export function useAllClusterDirWatch() { const { data: clusters } = useLastPlayedClusters(); const queryClient = useQueryClient(); + const watchTargets: Array = useMemo( + () => clusters.map(cluster => ({ + id: cluster.id, + folderName: cluster.folder_name, + })), + [clusters], + ); useEffect(() => { - if (clusters.length === 0) + if (watchTargets.length === 0) return; + const lifecycle = { + disposed: false, + }; const unwatchers: Array<() => void> = []; + const pendingTimeouts: Map = new Map(); + const syncInFlight: Set = new Set(); + const syncQueued: Set = new Set(); - for (const cluster of clusters) - bindings.folders.fromCluster(cluster.folder_name).then((clusterDir) => { - watch( + const clearPendingSync = (clusterId: number) => { + const timeoutId = pendingTimeouts.get(clusterId); + if (timeoutId === undefined) + return; + + window.clearTimeout(timeoutId); + pendingTimeouts.delete(clusterId); + }; + + const runSync = async (clusterId: number) => { + if (lifecycle.disposed) + return; + + if (syncInFlight.has(clusterId)) { + syncQueued.add(clusterId); + return; + } + + syncInFlight.add(clusterId); + try { + // Skip sync while the backend is applying bundle updates to avoid + // race conditions (watcher would re-sync mid-download). + const bundleSyncing = await bindings.oneclient.isBundleSyncing().catch(() => false); + if (bundleSyncing) + return; + + // Skip sync while cluster is running to avoid constant scans while + // the game is actively writing files. + const running = await bindings.core.isClusterRunning(clusterId).catch(() => false); + if (running) + return; + + await bindings.core.syncCluster(clusterId); + await queryClient.invalidateQueries({ queryKey: ['getLinkedPackages', clusterId] }); + } + catch (e) { + console.error('[watch] sync failed for cluster', clusterId, ':', e); + } + finally { + syncInFlight.delete(clusterId); + if (syncQueued.has(clusterId)) { + syncQueued.delete(clusterId); + scheduleSync(clusterId); + } + } + }; + + const scheduleSync = (clusterId: number) => { + if (lifecycle.disposed) + return; + + clearPendingSync(clusterId); + const timeoutId = window.setTimeout(() => { + pendingTimeouts.delete(clusterId); + void runSync(clusterId); + }, SYNC_DEBOUNCE_MS); + pendingTimeouts.set(clusterId, timeoutId); + }; + + const isDisposed = () => lifecycle.disposed; + + const registerWatch = async (target: WatchTarget) => { + try { + const clusterDir = await bindings.folders.fromCluster(target.folderName); + if (isDisposed()) + return; + + const unwatch = await watchSafely( clusterDir, - async () => { - // Skip sync while the backend is applying bundle updates to - // avoid race conditions (watcher would re-sync mid-download). - const syncing = await bindings.oneclient.isBundleSyncing().catch(() => false); - if (syncing) + (event) => { + if (isDisposed() || shouldIgnoreEvent(event)) return; - try { - await bindings.core.syncCluster(cluster.id); - await queryClient.invalidateQueries({ queryKey: ['getLinkedPackages', cluster.id] }); - } - catch (e) { - console.error('[watch] sync failed for cluster', cluster.id, ':', e); - } + scheduleSync(target.id); }, - { delayMs: 500, recursive: true }, - ) - .then(fn => unwatchers.push(fn)) - .catch(e => console.error('[watch] failed to register watch:', e)); - }).catch(e => console.error('[watch] failed to resolve cluster dir:', e)); + { delayMs: WATCH_DELAY_MS, recursive: true }, + ); + + if (isDisposed()) { + unwatch(); + return; + } + + unwatchers.push(unwatch); + } + catch (e) { + console.error('[watch] failed to register watch:', e); + } + }; + + for (const target of watchTargets) + void registerWatch(target); return () => { + lifecycle.disposed = true; + for (const timeoutId of pendingTimeouts.values()) + window.clearTimeout(timeoutId); + pendingTimeouts.clear(); + syncQueued.clear(); + syncInFlight.clear(); + for (const unwatch of unwatchers) unwatch(); }; - }, [clusters, queryClient]); + }, [queryClient, watchTargets]); } diff --git a/apps/oneclient/frontend/src/hooks/useClusters.ts b/apps/oneclient/frontend/src/hooks/useClusters.ts index ce09f056..ad87320e 100644 --- a/apps/oneclient/frontend/src/hooks/useClusters.ts +++ b/apps/oneclient/frontend/src/hooks/useClusters.ts @@ -1,8 +1,8 @@ import { bindings } from '@/main'; import useAppShellStore from '@/stores/appShellStore'; -import { useAsyncEffect, useCommandSuspense } from '@onelauncher/common'; +import { useAsyncEffect, useCommand, useCommandSuspense } from '@onelauncher/common'; import { useQueryClient } from '@tanstack/react-query'; -import { useEffect, useState } from 'react'; +import { useEffect } from 'react'; export function useLastPlayedClusters() { return useCommandSuspense( @@ -35,39 +35,55 @@ export function useActiveCluster() { } export function useIsRunning(clusterId: number | undefined | null) { - const [running, setRunning] = useState(false); const queryClient = useQueryClient(); - - useEffect(() => { - if (!clusterId) - return; - - const query = queryClient.ensureQueryData({ - queryKey: ['isClusterRunning', clusterId], - queryFn: () => bindings.core.isClusterRunning(clusterId), + const enabled = clusterId !== null && clusterId !== undefined; + const queryClusterId = clusterId ?? -1; + const { data: running } = useCommand( + ['isClusterRunning', queryClusterId], + () => bindings.core.isClusterRunning(queryClusterId), + { + enabled, staleTime: 0, gcTime: 0, - }); + refetchOnWindowFocus: true, + refetchInterval: 2000, + refetchIntervalInBackground: true, + }, + ); - query.then(setRunning); + useEffect(() => { + if (!enabled) + return; + + // Keep state accurate when the game is force-closed and a process event + // is delayed/missed. + const interval = window.setInterval(() => { + queryClient.invalidateQueries({ + queryKey: ['isClusterRunning', queryClusterId], + }); + }, 2000); return () => { - setRunning(false); + window.clearInterval(interval); }; - }, [clusterId, queryClient]); + }, [enabled, queryClient, queryClusterId]); useAsyncEffect(async () => { + if (!enabled) + return; + const unlisten = await bindings.events.process.on((e) => { - if (e.cluster_id !== clusterId) + if (e.cluster_id !== queryClusterId) return; - setRunning(e.kind.type !== 'Stopped'); + const isRunning = e.kind.type !== 'Stopped'; + queryClient.setQueryData(['isClusterRunning', queryClusterId], isRunning); }); return () => { unlisten(); }; - }, []); + }, [enabled, queryClient, queryClusterId]); - return running; + return enabled ? Boolean(running) : false; } diff --git a/apps/oneclient/frontend/src/routes/__root.tsx b/apps/oneclient/frontend/src/routes/__root.tsx index 6e8beb48..769e0282 100644 --- a/apps/oneclient/frontend/src/routes/__root.tsx +++ b/apps/oneclient/frontend/src/routes/__root.tsx @@ -3,7 +3,7 @@ import type { NavigateOptions, ToOptions } from '@tanstack/react-router'; import { Toasts } from '@/components'; import { useSettings } from '@/hooks/useSettings'; import { bindings } from '@/main'; -import { checkForUpdate, installUpdate, listenForUpdateEvents } from '@/utils/updater'; +import { checkForUpdate, installUpdate } from '@/utils/updater'; import { useCommand } from '@onelauncher/common'; import { TanStackDevtools } from '@tanstack/react-devtools'; import { ReactQueryDevtoolsPanel } from '@tanstack/react-query-devtools'; @@ -82,12 +82,15 @@ function useDiscordRPC() { function useAutoUpdate() { useEffect(() => { - const unlistenPromise = listenForUpdateEvents(async (event) => { - // eslint-disable-next-line no-console -- Used for debugging - aka important - console.log('Updater event:', event); - if (event.status === 'updateAvailable') { + void (async () => { + try { + const update = await checkForUpdate(); + if (!update) + return; + // eslint-disable-next-line no-console -- Used for debugging - aka important - console.log('Update available, installing...'); + console.log('Update found on initial check:', update.version); + try { await installUpdate(); } @@ -95,17 +98,10 @@ function useAutoUpdate() { console.error('Failed to install update:', e); } } - }); - - checkForUpdate().then((update) => { - if (update) - // eslint-disable-next-line no-console -- Used for debugging - aka important - console.log('Update found on initial check:', update.version); - }).catch(e => console.error('Failed to check for update:', e)); - - return () => { - unlistenPromise.then(unlisten => unlisten()); - }; + catch (e) { + console.error('Failed to check for update:', e); + } + })(); }, []); } diff --git a/apps/oneclient/frontend/src/routes/app/cluster/process.tsx b/apps/oneclient/frontend/src/routes/app/cluster/process.tsx index 16e45a6d..65607858 100644 --- a/apps/oneclient/frontend/src/routes/app/cluster/process.tsx +++ b/apps/oneclient/frontend/src/routes/app/cluster/process.tsx @@ -33,11 +33,14 @@ function LogContent({ const { cluster } = Route.useRouteContext(); const logViewerRef = useRef(null); - const { data: content, error, isLoading, refetch } = useCommand( - ['getLogByName', cluster.id, 'latest.log'], - () => bindings.core.getLogByName(cluster.id, 'latest.log'), + const { data: tail, error, isLoading, refetch } = useCommand( + ['getProcessLogTail', cluster.id, 0], + () => bindings.core.getProcessLogTail(cluster.id, 0), { staleTime: 0, + gcTime: 0, + refetchOnWindowFocus: false, + refetchOnReconnect: false, }, ); @@ -54,7 +57,9 @@ function LogContent({ if (e.kind.type !== 'Output') return; - logViewerRef.current?.push(e.kind.output); + const outputLines = e.kind.output.split('\n'); + for (const line of outputLines) + logViewerRef.current?.push(line); }); return () => { @@ -66,7 +71,7 @@ function LogContent({ diff --git a/packages/core/Cargo.toml b/packages/core/Cargo.toml index 94ed1232..0a721242 100644 --- a/packages/core/Cargo.toml +++ b/packages/core/Cargo.toml @@ -79,6 +79,7 @@ byteorder = { workspace = true } merge = { workspace = true } opener = { workspace = true } shadow-rs = { workspace = true } +libc = { workspace = true } # integration discord-rich-presence = { workspace = true } diff --git a/packages/core/src/api/cluster/content.rs b/packages/core/src/api/cluster/content.rs index 5476527e..69e02a21 100644 --- a/packages/core/src/api/cluster/content.rs +++ b/packages/core/src/api/cluster/content.rs @@ -1,9 +1,22 @@ +use std::path::Path; + use onelauncher_entity::clusters; +use tokio::io::{AsyncReadExt, AsyncSeekExt}; use crate::error::LauncherResult; use crate::store::Dirs; use crate::utils::io; +const PROCESS_LOG_FILE_NAME: &str = "latest.log"; +const TAIL_READ_CHUNK_SIZE: u64 = 64 * 1024; + +#[onelauncher_macro::specta] +#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)] +pub struct ProcessLogTail { + pub content: String, + pub truncated: bool, +} + /// Returns a list of screenshot file names pub async fn get_screenshots(cluster: &clusters::Model) -> LauncherResult> { let dir = cluster.folder_name.clone(); @@ -100,3 +113,175 @@ pub async fn get_log_by_name( Ok(Some(content)) } + +/// Returns a tail window from a running cluster's `latest.log`. +/// `max_lines = 0` means no line limit (return full content). +pub async fn get_process_log_tail( + cluster: &clusters::Model, + max_lines: usize, +) -> LauncherResult> { + let dir = cluster.folder_name.clone(); + let path = Dirs::get_clusters_dir() + .await? + .join(dir) + .join("logs") + .join(PROCESS_LOG_FILE_NAME); + + if !path.exists() { + return Ok(None); + } + + let tail = read_process_log_tail(&path, max_lines).await?; + + Ok(Some(tail)) +} + +async fn read_process_log_tail( + path: &Path, + max_lines: usize, +) -> Result { + if max_lines == 0 { + let bytes = tokio::fs::read(path).await?; + let content = String::from_utf8_lossy(&bytes).to_string(); + return Ok(ProcessLogTail { + content, + truncated: false, + }); + } + + let max_lines = max_lines.max(1); + let mut file = tokio::fs::File::open(path).await?; + let file_size = file.metadata().await?.len(); + + if file_size == 0 { + return Ok(ProcessLogTail { + content: String::new(), + truncated: false, + }); + } + + let mut cursor = file_size; + let mut buffer = Vec::new(); + let mut newline_count = 0usize; + + while cursor > 0 && newline_count <= max_lines { + let bytes_to_read = TAIL_READ_CHUNK_SIZE.min(cursor) as usize; + cursor -= bytes_to_read as u64; + + file.seek(std::io::SeekFrom::Start(cursor)).await?; + + let mut chunk = vec![0u8; bytes_to_read]; + file.read_exact(&mut chunk).await?; + newline_count += chunk.iter().filter(|byte| **byte == b'\n').count(); + + if buffer.is_empty() { + buffer = chunk; + } else { + chunk.extend_from_slice(&buffer); + buffer = chunk; + } + } + + let raw = String::from_utf8_lossy(&buffer); + let (content, trimmed_by_line_limit) = trim_to_last_lines(&raw, max_lines); + + Ok(ProcessLogTail { + content, + truncated: cursor > 0 || trimmed_by_line_limit, + }) +} + +fn trim_to_last_lines(content: &str, max_lines: usize) -> (String, bool) { + let lines: Vec<&str> = content.lines().collect(); + if lines.len() <= max_lines { + return (content.to_string(), false); + } + + let start_index = lines.len() - max_lines; + (lines[start_index..].join("\n"), true) +} + +#[cfg(test)] +mod tests { + use std::path::PathBuf; + use std::time::{SystemTime, UNIX_EPOCH}; + + use super::read_process_log_tail; + + fn temp_file_path(name: &str) -> PathBuf { + let nonce = SystemTime::now() + .duration_since(UNIX_EPOCH) + .expect("clock went backwards") + .as_nanos(); + std::env::temp_dir().join(format!("onelauncher_{name}_{nonce}.log")) + } + + #[tokio::test] + async fn process_log_tail_returns_full_content_when_small() { + let path = temp_file_path("small_tail"); + tokio::fs::write(&path, "a\nb\nc\n").await.unwrap(); + + let tail = read_process_log_tail(&path, 10_000).await.unwrap(); + + assert_eq!(tail.content, "a\nb\nc\n"); + assert!(!tail.truncated); + + let _ = tokio::fs::remove_file(path).await; + } + + #[tokio::test] + async fn process_log_tail_returns_only_last_requested_lines() { + let path = temp_file_path("large_tail"); + let content = (0..20_000) + .map(|n| format!("line-{n}")) + .collect::>() + .join("\n"); + tokio::fs::write(&path, format!("{content}\n")) + .await + .unwrap(); + + let tail = read_process_log_tail(&path, 10_000).await.unwrap(); + + let lines: Vec<&str> = tail.content.lines().collect(); + assert_eq!(lines.len(), 10_000); + assert_eq!(lines.first(), Some(&"line-10000")); + assert_eq!(lines.last(), Some(&"line-19999")); + assert!(tail.truncated); + + let _ = tokio::fs::remove_file(path).await; + } + + #[tokio::test] + async fn process_log_tail_handles_no_trailing_newline() { + let path = temp_file_path("no_trailing_newline"); + let content = (0..5) + .map(|n| format!("line-{n}")) + .collect::>() + .join("\n"); + tokio::fs::write(&path, content).await.unwrap(); + + let tail = read_process_log_tail(&path, 3).await.unwrap(); + + let lines: Vec<&str> = tail.content.lines().collect(); + assert_eq!(lines, vec!["line-2", "line-3", "line-4"]); + assert!(tail.truncated); + + let _ = tokio::fs::remove_file(path).await; + } + + #[tokio::test] + async fn process_log_tail_handles_crlf_input() { + let path = temp_file_path("crlf_tail"); + tokio::fs::write(&path, "a\r\nb\r\nc\r\nd\r\n") + .await + .unwrap(); + + let tail = read_process_log_tail(&path, 2).await.unwrap(); + + let lines: Vec<&str> = tail.content.lines().collect(); + assert_eq!(lines, vec!["c", "d"]); + assert!(tail.truncated); + + let _ = tokio::fs::remove_file(path).await; + } +} diff --git a/packages/core/src/api/proxy/impl/proxy_tauri.rs b/packages/core/src/api/proxy/impl/proxy_tauri.rs index f2b090e4..15b88d3b 100644 --- a/packages/core/src/api/proxy/impl/proxy_tauri.rs +++ b/packages/core/src/api/proxy/impl/proxy_tauri.rs @@ -60,8 +60,22 @@ impl LauncherProxy for ProxyTauri { () } - LauncherEvent::Message(message) => self.emitter.message(message)?, - LauncherEvent::Process(process) => self.emitter.process(process)?, + LauncherEvent::Message(message) => { + let emitter = Arc::clone(&self.emitter); + self.handle.run_on_main_thread(move || { + if let Err(err) = emitter.message(message) { + tracing::error!(?err, "failed to emit tauri message event on main thread"); + } + })?; + } + LauncherEvent::Process(process) => { + let emitter = Arc::clone(&self.emitter); + self.handle.run_on_main_thread(move || { + if let Err(err) = emitter.process(process) { + tracing::error!(?err, "failed to emit tauri process event on main thread"); + } + })?; + } }) } diff --git a/packages/core/src/api/tauri/commands.rs b/packages/core/src/api/tauri/commands.rs index 0a7721ce..9fbfe55a 100644 --- a/packages/core/src/api/tauri/commands.rs +++ b/packages/core/src/api/tauri/commands.rs @@ -66,6 +66,12 @@ pub trait TauriLauncherApi { #[taurpc(alias = "getLogByName")] async fn get_log_by_name(id: ClusterId, name: String) -> LauncherResult>; + + #[taurpc(alias = "getProcessLogTail")] + async fn get_process_log_tail( + id: ClusterId, + max_lines: usize, + ) -> LauncherResult>; // endregion: clusters // MARK: API: processes @@ -462,6 +468,18 @@ impl TauriLauncherApi for TauriLauncherApiImpl { api::cluster::content::get_log_by_name(&cluster, &name).await } + + async fn get_process_log_tail( + self, + id: ClusterId, + max_lines: usize, + ) -> LauncherResult> { + let cluster = api::cluster::dao::get_cluster_by_id(id) + .await? + .ok_or_else(|| anyhow::anyhow!("cluster with id {} not found", id))?; + + api::cluster::content::get_process_log_tail(&cluster, max_lines).await + } // endregion: clusters // MARK: Impl: processes diff --git a/packages/core/src/logger.rs b/packages/core/src/logger.rs index 54333806..db5cf05e 100644 --- a/packages/core/src/logger.rs +++ b/packages/core/src/logger.rs @@ -3,9 +3,15 @@ use tracing_subscriber::prelude::*; use tracing_subscriber::util::SubscriberInitExt; // Handling for the live development logging -// This will log to the console, and will not log to a file +// This will log to both the console and a file in the logs directory #[cfg(debug_assertions)] pub async fn start_logger() { + use chrono::Local; + use std::fs::OpenOptions; + use tracing_subscriber::fmt::time::ChronoLocal; + + use crate::store::Dirs; + let filter = tracing_subscriber::EnvFilter::try_from_default_env().unwrap_or_else(|_| { tracing_subscriber::EnvFilter::new( Core::get() @@ -15,7 +21,7 @@ pub async fn start_logger() { ) }); - let mut fmt_layer = tracing_subscriber::fmt::layer() + let mut console_fmt_layer = tracing_subscriber::fmt::layer() .with_ansi(true) .with_file(true) .with_line_number(true) @@ -24,15 +30,52 @@ pub async fn start_logger() { .pretty(); if let Some(span) = Core::get().logger_span_formatting.clone() { - fmt_layer = fmt_layer.with_span_events(span); + console_fmt_layer = console_fmt_layer.with_span_events(span); } - if let Err(err) = tracing_subscriber::registry() - .with(fmt_layer) - .with(filter) - .with(tracing_error::ErrorLayer::default()) - .try_init() + let logs_dir = Dirs::get_launcher_logs_dir().await.unwrap_or_default(); + let log_file_name = format!("launcher_{}.log", Local::now().format("%Y%m%d_%H%M%S")); + let log_file_path = logs_dir.join(log_file_name); + + if let Err(err) = std::fs::create_dir_all(&logs_dir) { + eprintln!("Could not create logs directory: {err}"); + } + + let init_result = match OpenOptions::new() + .create(true) + .append(true) + .open(&log_file_path) { + Ok(file) => { + let mut file_fmt_layer = tracing_subscriber::fmt::layer() + .compact() + .with_writer(file) + .with_ansi(false) + .with_timer(ChronoLocal::rfc_3339()); + + if let Some(span) = Core::get().logger_span_formatting.clone() { + file_fmt_layer = file_fmt_layer.with_span_events(span); + } + + tracing_subscriber::registry() + .with(console_fmt_layer) + .with(file_fmt_layer) + .with(filter) + .with(tracing_error::ErrorLayer::default()) + .try_init() + } + Err(err) => { + eprintln!("Could not open log file in dev mode: {err}"); + + tracing_subscriber::registry() + .with(console_fmt_layer) + .with(filter) + .with(tracing_error::ErrorLayer::default()) + .try_init() + } + }; + + if let Err(err) = init_result { eprintln!("Could not set default logger: {err}"); } } diff --git a/packages/core/src/store/processes.rs b/packages/core/src/store/processes.rs index 75003c53..7d4df49e 100644 --- a/packages/core/src/store/processes.rs +++ b/packages/core/src/store/processes.rs @@ -24,6 +24,10 @@ use crate::utils::io::IOError; use super::State; use super::credentials::MinecraftCredentials; +const PROCESS_OUTPUT_FLUSH_MS: u64 = 50; +const PROCESS_OUTPUT_BATCH_MAX_LINES: usize = 128; +const PROCESS_OUTPUT_BATCH_MAX_CHARS: usize = 16 * 1024; + #[onelauncher_macro::error] #[derive(Debug, thiserror::Error)] pub enum ProcessError { @@ -155,15 +159,76 @@ impl ProcessStore { /// Returns whether a cluster has a running process. pub async fn has_running(&self, cluster_id: ClusterId) -> bool { - let read_guard = self.processes.read().await; + let processes = { + let read_guard = self.processes.read().await; + read_guard + .iter() + .map(|(pid, process)| (*pid, process.clone())) + .collect::>() + }; - for process in read_guard.values() { - if process.read().await.cluster_id == cluster_id { - return true; + let mut stale_pids = Vec::new(); + let mut running = false; + + for (pid, process) in processes { + let (process_cluster_id, process_loop_finished, child_type) = { + let process_guard = process.read().await; + ( + process_guard.cluster_id, + process_guard + .process_loop + .as_ref() + .map_or(true, |handle| handle.is_finished()), + process_guard.child_type.clone(), + ) + }; + let is_target_cluster = process_cluster_id == cluster_id; + + if !is_target_cluster { + continue; + } + + if process_loop_finished { + stale_pids.push(pid); + continue; + } + + let stopped = { + let mut child = child_type.write().await; + match child.try_wait() { + Ok(Some(_)) => true, + Ok(None) => { + let exists = process_exists_os(pid); + if exists { false } else { true } + } + Err(err) => { + tracing::warn!( + pid, + cluster_id, + ?err, + "failed to poll process in has_running; removing stale process", + ); + true + } + } + }; + + if stopped { + stale_pids.push(pid); + continue; + } + + running = true; + } + + if !stale_pids.is_empty() { + let mut write_guard = self.processes.write().await; + for pid in &stale_pids { + write_guard.remove(pid); } } - false + running } /// Spawns a new process (Minecraft Instance) @@ -220,19 +285,75 @@ impl ProcessStore { tokio::spawn(async move { let mut stdout = BufReader::new(stdout).lines(); let mut stderr = BufReader::new(stderr).lines(); + let mut stdout_closed = false; + let mut stderr_closed = false; + let mut pending_lines = Vec::with_capacity(PROCESS_OUTPUT_BATCH_MAX_LINES); + let mut pending_chars = 0usize; - while let Ok(Some(mut line)) = tokio::select! { - line = stdout.next_line() => line, - line = stderr.next_line() => line, - } { - censor_line(&censors, &mut line); + let mut flush_tick = + tokio::time::interval(std::time::Duration::from_millis(PROCESS_OUTPUT_FLUSH_MS)); + flush_tick.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); - send_event(LauncherEvent::Process(ProcessPayload { - cluster_id, - kind: ProcessPayloadKind::Output { pid, output: line }, - })) - .await; + loop { + if stdout_closed && stderr_closed { + break; + } + + tokio::select! { + line = stdout.next_line(), if !stdout_closed => { + match line { + Ok(Some(mut line)) => { + censor_line(&censors, &mut line); + pending_chars += line.len(); + pending_lines.push(line); + } + Ok(None) => { + stdout_closed = true; + } + Err(err) => { + stdout_closed = true; + tracing::warn!("stdout read failed for pid {}: {}", pid, err); + } + } + } + + line = stderr.next_line(), if !stderr_closed => { + match line { + Ok(Some(mut line)) => { + censor_line(&censors, &mut line); + pending_chars += line.len(); + pending_lines.push(line); + } + Ok(None) => { + stderr_closed = true; + } + Err(err) => { + stderr_closed = true; + tracing::warn!("stderr read failed for pid {}: {}", pid, err); + } + } + } + + _ = flush_tick.tick(), if !pending_lines.is_empty() => { + flush_process_output_event(cluster_id, pid, &mut pending_lines, &mut pending_chars).await; + } + } + + if pending_lines.len() >= PROCESS_OUTPUT_BATCH_MAX_LINES + || pending_chars >= PROCESS_OUTPUT_BATCH_MAX_CHARS + { + flush_process_output_event( + cluster_id, + pid, + &mut pending_lines, + &mut pending_chars, + ) + .await; + } } + + flush_process_output_event(cluster_id, pid, &mut pending_lines, &mut pending_chars) + .await; }); let child_type = Arc::new(RwLock::new(ChildType::new(child))); @@ -284,14 +405,26 @@ impl ProcessStore { ) -> LauncherResult { let pid = child.read().await.id(); - let exit_code = wait_for_exit(cluster_id, &child).await?; + let exit_code = match wait_for_exit(cluster_id, &child).await { + Ok(exit_code) => exit_code, + Err(err) => { + tracing::warn!( + pid, + cluster_id, + ?err, + "failed while waiting for process exit; marking process as stopped", + ); + -1 + } + }; let state = State::get().await?; let store = &state.processes; - let mut write_guard = store.processes.write().await; - - write_guard.remove(&pid); + { + let mut write_guard = store.processes.write().await; + write_guard.remove(&pid); + } tracing::debug!("removed process with pid {}", pid); if let Some(post_hook) = post_hook @@ -310,6 +443,30 @@ impl ProcessStore { } } +async fn flush_process_output_event( + cluster_id: ClusterId, + pid: u32, + pending_lines: &mut Vec, + pending_chars: &mut usize, +) -> usize { + if pending_lines.is_empty() { + return 0; + } + + let emitted_lines = pending_lines.len(); + let output = pending_lines.join("\n"); + pending_lines.clear(); + *pending_chars = 0; + + send_event(LauncherEvent::Process(ProcessPayload { + cluster_id, + kind: ProcessPayloadKind::Output { pid, output }, + })) + .await; + + emitted_lines +} + async fn run_hook(hook: &str, cwd: PathBuf) -> LauncherResult> { let mut split = hook.split(' '); @@ -344,13 +501,34 @@ async fn wait_for_exit( cluster_id: ClusterId, child: &Arc>, ) -> LauncherResult { + let pid = child.read().await.id(); + let mut last_updated = Utc::now(); loop { - let mut child = child.write().await; - if let Some(status) = child.try_wait()? { - tracing::info!("process exited with status: {:?}", status); - return Ok(status); + let exit_status = { + let mut child = child.write().await; + child.try_wait() + }; + + match exit_status { + Ok(Some(status)) => { + tracing::info!("process exited with status: {:?}", status); + return Ok(status); + } + Ok(None) => { + if !process_exists_os(pid) { + return Ok(-1); + } + } + Err(err) => { + tracing::warn!( + cluster_id, + ?err, + "failed to poll process status; treating as already stopped", + ); + return Ok(-1); + } } // Every 60 seconds, update the playtime of the process @@ -366,3 +544,17 @@ async fn wait_for_exit( tokio::time::sleep(std::time::Duration::from_millis(100)).await; } } + +fn process_exists_os(pid: u32) -> bool { + // SAFETY: `kill` with signal 0 performs existence/permission checks only and + // does not deliver a signal. + let rc = unsafe { libc::kill(pid as i32, 0) }; + if rc == 0 { + return true; + } + + let err = std::io::Error::last_os_error() + .raw_os_error() + .unwrap_or_default(); + err == libc::EPERM +}