oiupoyt/OneLauncher-Cracked
1
1
Fork 0
Code Issues Pull requests Projects Releases 2 Packages Wiki Activity

feat(core): Process recovery (session exit times and live log after restarting launcher)

Browse source
This commit is contained in:
LynithDev 2026-07-16 17:48:16 +02:00
parent ddda1d66ba
commit 8dca2550d1
29 changed files with 1536 additions and 6129 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
Cargo.lock generated
View file

@ -4577,6 +4577,7 @@ dependencies = [
"sha2 0.10.9",
"sqlx",
"strum",
"sysinfo",
"thiserror 2.0.18",
"tokio",
"tokio-stream",

16
packages/oneclient_app/src/layout/cluster_shell.rs
View file

@ -118,14 +118,18 @@ impl Component for ClusterShell {
let show_game_log = game.is_active(cluster_id);
let launch_state = launch_button_state(&game, cluster_id);
// Queried out here rather than where it is used: the shell can mount
// before the cluster list settles, and a hook that only runs once the
// cluster resolves would change this component's hook count mid-life.
let parsed = cluster.as_ref().and_then(|c| parse_mc_version(&c.mc_version));
let metadata = use_version_metadata(
parsed.as_ref().map(|p| p.major),
parsed.and_then(|p| p.key()),
cluster.as_ref().map(|c| c.mc_loader),
);
let header = cluster.as_ref().map(|cluster| {
let title = format!("{} {}", cluster.mc_loader, cluster.mc_version);
let parsed = parse_mc_version(&cluster.mc_version);
let metadata = use_version_metadata(
parsed.as_ref().map(|p| p.major),
parsed.and_then(|p| p.key()),
Some(cluster.mc_loader),
);
let subtitle = metadata
.and_then(|m| m.long_description)
.unwrap_or_else(|| cluster.name.clone());

2
packages/oneclient_app/src/utils.rs
View file

@ -35,7 +35,7 @@ pub fn group_clusters_by_major(clusters: &[Cluster]) -> ClusterGroups {
pub fn major_pretty_name(major: u32) -> String {
if major >= 26 {
format!("{major}.x")
format!("{major}")
} else {
format!("1.{major}")
}

1
packages/oneclient_core/Cargo.toml
View file

@ -124,6 +124,7 @@ url.workspace = true
uuid.workspace = true
chrono.workspace = true
parking_lot.workspace = true
sysinfo.workspace = true
sanitize-filename.workspace = true
dunce.workspace = true
bytes.workspace = true

14
packages/oneclient_core/src/bundles/overrides.rs
View file

@ -331,7 +331,7 @@ mod tests {
}
async fn read(path: &Path) -> Option<Vec<u8>> {
tokio::fs::read(path).await.ok()
polyio::read(path).await.ok()
}
async fn sync(zip: &Path, root: &Path) -> OverrideSyncReport {
@ -403,7 +403,7 @@ mod tests {
write_bundle(&v1, &[("overrides/config/a.toml", b"alpha")]).await;
sync(&v1, &root).await;
tokio::fs::write(root.join("config/a.toml"), b"user-edit")
polyio::write(root.join("config/a.toml"), b"user-edit")
.await
.unwrap();
@ -423,7 +423,7 @@ mod tests {
write_bundle(&v1, &[("overrides/config/a.toml", b"alpha")]).await;
sync(&v1, &root).await;
tokio::fs::write(root.join("config/a.toml"), b"user-edit")
polyio::write(root.join("config/a.toml"), b"user-edit")
.await
.unwrap();
@ -455,7 +455,7 @@ mod tests {
write_bundle(&v1, &[(&format!("overrides/{path}"), b"deps-v1")]).await;
sync(&v1, &root).await;
tokio::fs::write(root.join(path), b"runtime-mutated")
polyio::write(root.join(path), b"runtime-mutated")
.await
.unwrap();
@ -482,7 +482,7 @@ mod tests {
sync(&v1, &root).await;
// User modified kept.toml; gone.toml untouched.
tokio::fs::write(root.join("config/kept.toml"), b"edited")
polyio::write(root.join("config/kept.toml"), b"edited")
.await
.unwrap();
@ -503,10 +503,10 @@ mod tests {
async fn bootstrap_pre_existing_edit_is_never_overwritten() {
let root = tmp_root("bootstrap");
tokio::fs::create_dir_all(root.join("config"))
polyio::create_dir_all(root.join("config"))
.await
.unwrap();
tokio::fs::write(root.join("config/a.toml"), b"pre-existing")
polyio::write(root.join("config/a.toml"), b"pre-existing")
.await
.unwrap();

4
packages/oneclient_core/src/clusters/manager.rs
View file

@ -153,9 +153,9 @@ impl ClusterManager {
let marker = cluster.dedicated_marker()?;
if dedicated {
polyio::create_dir_all(cluster.dir()?).await?;
tokio::fs::write(&marker, b"").await.ok();
polyio::write(&marker, b"").await.ok();
} else if marker.exists() {
tokio::fs::remove_file(&marker).await.ok();
polyio::remove_file(&marker).await.ok();
}
Ok(())
}

2
packages/oneclient_core/src/game/download.rs
View file

@ -43,7 +43,7 @@ pub async fn download_to_path(
child.set_phase(TaskPhase::Verifying);
let actual = sha1_file(dest).await?;
if normalize_hash(&actual) != normalize_hash(expected) {
let _ = tokio::fs::remove_file(dest).await;
let _ = polyio::remove_file(dest).await;
return Err(GameError::HashMismatch {
path: dest.display().to_string(),
expected: expected.to_string(),

320
packages/oneclient_core/src/game/launch.rs
View file

@ -1,23 +1,22 @@
use std::path::Path;
use std::process::Stdio;
use std::sync::Arc;
use std::time::{Duration, Instant};
use std::time::Duration;
use chrono::{DateTime, Utc};
use interfrost::api::minecraft::ArgumentType;
use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
use tokio::process::Command;
use tokio::sync::Mutex as AsyncMutex;
use crate::auth::MinecraftAccount;
use crate::ClusterStage;
use crate::clusters::ClusterManager;
use crate::auth::MinecraftAccount;
use crate::clusters::{Cluster, ClusterManager};
use crate::discord::Presence;
use crate::game::session::SessionRecorder;
use crate::game::tail::spawn_log_tail;
use crate::game::{
GameError, arguments, download_minecraft, download_version_info, get_loader_version,
libraries_missing, resolve_minecraft_version,
};
use crate::game::session::SessionRecorder;
use crate::java::JavaManager;
use crate::notification::{GroupedProgressSession, LaunchStage};
use crate::settings::GameSettingsProfile;
@ -125,9 +124,14 @@ pub async fn launch_cluster(
)
.await?;
let version_info =
download_version_info(&state.services, Some(&progress), &version, loader_version.as_ref(), false)
.await;
let version_info = download_version_info(
&state.services,
Some(&progress),
&version,
loader_version.as_ref(),
false,
)
.await;
(version, updated, loader_version, version_info)
};
@ -141,9 +145,10 @@ pub async fn launch_cluster(
}
};
let version_name = loader_version
.as_ref()
.map_or_else(|| version.id.clone(), |lv| format!("{}-{}", version.id, lv.id));
let version_name = loader_version.as_ref().map_or_else(
|| version.id.clone(),
|lv| format!("{}-{}", version.id, lv.id),
);
tracing::info!(
cluster_id,
@ -172,8 +177,7 @@ pub async fn launch_cluster(
match libraries_missing(&version_info, &java.os_arch, updated) {
Ok(true) => {
tracing::info!(cluster_id, "missing game files; repairing");
let _ =
ClusterManager::set_stage(state, cluster_id, ClusterStage::Repairing).await;
let _ = ClusterManager::set_stage(state, cluster_id, ClusterStage::Repairing).await;
stage(LaunchStage::Downloading);
if let Err(err) = download_minecraft(
&state.services,
@ -200,16 +204,14 @@ pub async fn launch_cluster(
stage(LaunchStage::Launching);
let cwd = game_dir;
tokio::fs::create_dir_all(&cwd).await.ok();
polyio::create_dir_all(&cwd).await.ok();
if let Err(err) = crate::game::write_allowed_symlinks(&cwd).await {
tracing::warn!(cluster_id, error = %err, "failed to write allowed_symlinks.txt");
}
if !dedicated {
if let Err(err) =
crate::game::sync_shared_content(&state.services, &cluster, &cwd).await
{
if let Err(err) = crate::game::sync_shared_content(&state.services, &cluster, &cwd).await {
tracing::warn!(cluster_id, error = %err, "failed to sync shared content");
}
// Redirect the shared dir's `logs`/`crash-reports` into this cluster's
@ -263,11 +265,7 @@ pub async fn launch_cluster(
profile.resolution.unwrap_or_default(),
&java.os_arch,
)?;
arguments::append_profile_game_arguments(
&mut mc_args,
profile.force_fullscreen,
None,
);
arguments::append_profile_game_arguments(&mut mc_args, profile.force_fullscreen, None);
run_hook(profile.hook_pre.as_deref(), &cwd).await;
@ -289,7 +287,39 @@ pub async fn launch_cluster(
.args(mc_args)
.current_dir(&cwd);
command.stdout(Stdio::piped()).stderr(Stdio::piped());
let log_path = crate::logs::cluster_output_log(&cluster)?;
if let Some(parent) = log_path.parent() {
polyio::create_dir_all(parent).await.ok();
}
// The game's output goes straight to the log file rather than through pipes
// held by the launcher. A pipe would tie the game's lifetime to ours: once
// the launcher exits its read end closes, and the game's next write to a
// broken stdout takes it down with us. Writing to a file keeps the two
// independent, and the launcher tails that file for the live console.
// A cloned handle shares the file offset, so stdout and stderr interleave
// into one stream instead of overwriting each other.
// `Stdio` needs owned std handles, so the tokio files are unwrapped only
// once the async work of opening and cloning them is done.
let handles = match tokio::fs::File::create(&log_path).await {
Ok(out) => match out.try_clone().await {
Ok(err) => Ok((out.into_std().await, err.into_std().await)),
Err(err) => Err(err),
},
Err(err) => Err(err),
};
match handles {
Ok((out, err)) => {
command.stdout(Stdio::from(out)).stderr(Stdio::from(err));
}
Err(err) => {
tracing::warn!(cluster_id, error = %err, "failed to open game log; discarding output");
command.stdout(Stdio::null()).stderr(Stdio::null());
}
}
command.stdin(Stdio::null());
detach(&mut command);
let mut child = command
.spawn()
@ -304,39 +334,30 @@ pub async fn launch_cluster(
mc_version: cluster.mc_version.clone(),
});
let log_path = crate::logs::cluster_output_log(&cluster)?;
if let Some(parent) = log_path.parent() {
tokio::fs::create_dir_all(parent).await.ok();
}
let log_file = tokio::fs::File::create(&log_path)
.await
.ok()
.map(|file| Arc::new(tokio::sync::Mutex::new(file)));
let recorder =
SessionRecorder::start(state, cluster_id, profile.mem_max.unwrap_or(2048), &java).await;
let recorder = SessionRecorder::start(
state,
cluster_id,
profile.mem_max.unwrap_or(2048),
&java,
)
.await;
// Pin the process to the session row so that if the launcher exits before
// the game does, the next start can tell whether it is still playing.
if let (Some(recorder), Some(pid)) = (recorder.as_ref(), pid) {
recorder
.record_process(pid, crate::game::process_start_time(pid))
.await;
}
if let Some(stdout) = child.stdout.take() {
spawn_log_reader(cluster_id, stdout, notifier.clone(), log_file.clone(), recorder.clone());
}
if let Some(stderr) = child.stderr.take() {
spawn_log_reader(cluster_id, stderr, notifier.clone(), log_file.clone(), recorder.clone());
}
let started_at = recorder
.as_ref()
.and_then(SessionRecorder::started_at)
.unwrap_or_else(Utc::now);
let tail = spawn_log_tail(cluster_id, log_path, notifier.clone(), recorder.clone());
let (kill_tx, kill_rx) = tokio::sync::oneshot::channel::<()>();
state.games.register_kill(cluster_id, kill_tx);
let state = Arc::clone(state);
let post_hook = profile.hook_post.clone();
let cluster_name = cluster.name.clone();
tokio::spawn(async move {
let cluster = cluster;
let started = Instant::now();
let status = tokio::select! {
status = child.wait() => status,
_ = kill_rx => {
@ -344,79 +365,154 @@ pub async fn launch_cluster(
child.wait().await
}
};
let played = started.elapsed();
tail.stop().await;
let outcome = match status {
Ok(status) => Exit::Observed {
code: status.code().map(i64::from),
success: status.success(),
display: status.to_string(),
},
Err(err) => Exit::Failed(err.to_string()),
};
finalize_session(
&state,
&cluster,
&cwd,
dedicated,
post_hook.as_deref(),
recorder,
SessionEnd {
started_at,
ended_at: Utc::now(),
outcome,
owns_slot: true,
},
)
.await;
});
Ok(LaunchedGame { cluster_id, pid })
}
/// Cut the game loose from the launcher's process group / console, so signals
/// aimed at the launcher (a terminal Ctrl-C, a console window closing) don't
/// reach the game as collateral.
fn detach(command: &mut Command) {
#[cfg(unix)]
{
use std::os::unix::process::CommandExt;
command.as_std_mut().process_group(0);
}
#[cfg(windows)]
{
use std::os::windows::process::CommandExt;
const DETACHED_PROCESS: u32 = 0x0000_0008;
const CREATE_NEW_PROCESS_GROUP: u32 = 0x0000_0200;
command
.as_std_mut()
.creation_flags(DETACHED_PROCESS | CREATE_NEW_PROCESS_GROUP);
}
}
/// How a session ended, as far as anyone could tell.
pub(crate) enum Exit {
/// The launcher was there and saw the process exit.
Observed {
code: Option<i64>,
success: bool,
display: String,
},
/// Waiting on the process itself failed.
Failed(String),
/// The game exited while the launcher was closed; the time was recovered
/// from its log, and the exit code is gone for good.
Inferred,
}
pub(crate) struct SessionEnd {
pub started_at: DateTime<Utc>,
pub ended_at: DateTime<Utc>,
pub outcome: Exit,
/// Whether this session is the one currently holding the cluster's running
/// slot. A stale session recovered from the database may share its cluster
/// with a game that is playing right now — booking the old session's
/// playtime is right, but clearing that slot would report the live game as
/// exited.
pub owns_slot: bool,
}
/// Everything that has to happen once a game is gone: clear its running state,
/// bank the playtime, close the session row, run the post hook and unwind the
/// shared-directory plumbing. Shared by the live exit path and by recovery of
/// sessions that outlived the launcher.
pub(crate) async fn finalize_session(
state: &Arc<LauncherState>,
cluster: &Cluster,
cwd: &Path,
dedicated: bool,
post_hook: Option<&str>,
recorder: Option<SessionRecorder>,
end: SessionEnd,
) {
let cluster_id = cluster.id;
let played = (end.ended_at - end.started_at).to_std().unwrap_or_default();
if end.owns_slot {
state.games.remove(cluster_id);
state.services.notifier.game_stage(cluster_id, LaunchStage::Exited);
state
.services
.notifier
.game_stage(cluster_id, LaunchStage::Exited);
if state.games.running_ids().is_empty() {
state.discord.set_presence(Presence::Idle);
}
}
if played > Duration::from_secs(1) {
let _ = ClusterManager::add_playtime(&state, cluster_id, played).await;
if played > Duration::from_secs(1) {
let _ = ClusterManager::add_playtime(state, cluster_id, played).await;
}
if let Some(recorder) = recorder {
let code = match &end.outcome {
Exit::Observed { code, .. } => *code,
Exit::Failed(_) | Exit::Inferred => None,
};
recorder.finish_at(&end.ended_at.to_rfc3339(), code).await;
}
run_hook(post_hook, cwd).await;
if !dedicated {
crate::game::import_manual_content(&state.services, cluster, cwd).await;
if let Err(err) = crate::game::clear_shared_content(cwd).await {
tracing::warn!(cluster_id, error = %err, "failed to clear shared content on exit");
}
crate::game::unlink_cluster_logs(cwd).await;
}
if let Some(recorder) = recorder {
let exit_code = status.as_ref().ok().and_then(|s| s.code()).map(i64::from);
recorder.finish(exit_code).await;
}
run_hook(post_hook.as_deref(), &cwd).await;
if !dedicated {
crate::game::import_manual_content(&state.services, &cluster, &cwd).await;
if let Err(err) = crate::game::clear_shared_content(&cwd).await {
tracing::warn!(cluster_id, error = %err, "failed to clear shared content on exit");
}
crate::game::unlink_cluster_logs(&cwd).await;
}
match status {
Ok(status) if status.success() => state
.services
.notifier
.send_info("Game closed", &format!("{cluster_name} exited")),
Ok(status) => state.services.notifier.send_error(
"Game crashed",
&format!("{cluster_name} exited with {status}"),
),
Err(err) => state
.services
.notifier
.send_error("Game error", &format!("{cluster_name}: {err}")),
}
});
Ok(LaunchedGame {
cluster_id,
pid,
})
}
fn spawn_log_reader<R>(
cluster_id: i64,
reader: R,
notifier: crate::notification::NotificationService,
file: Option<Arc<AsyncMutex<tokio::fs::File>>>,
recorder: Option<SessionRecorder>,
) where
R: tokio::io::AsyncRead + Unpin + Send + 'static,
{
tokio::spawn(async move {
let mut lines = BufReader::new(reader).lines();
while let Ok(Some(line)) = lines.next_line().await {
if let Some(file) = &file {
let mut handle = file.lock().await;
let _ = handle.write_all(line.as_bytes()).await;
let _ = handle.write_all(b"\n").await;
}
if let Some(recorder) = &recorder {
recorder.observe(&line).await;
}
notifier.game_log(cluster_id, line);
}
});
let name = &cluster.name;
match end.outcome {
Exit::Observed { success: true, .. } => state
.services
.notifier
.send_info("Game closed", &format!("{name} exited")),
Exit::Observed { display, .. } => state
.services
.notifier
.send_error("Game crashed", &format!("{name} exited with {display}")),
Exit::Failed(err) => state
.services
.notifier
.send_error("Game error", &format!("{name}: {err}")),
// Nothing was watching, so there is no crash to report and no news the
// user wants a popup about — the session is simply booked and closed.
Exit::Inferred => {}
}
}
fn base_command(profile: &GameSettingsProfile, java_path: &str) -> Command {

297
packages/oneclient_core/src/game/log_replay.rs Normal file
View file

@ -0,0 +1,297 @@
//! Reconstructing a game session from its log.
//!
//! When the launcher is closed while the game keeps playing, nothing observes
//! the exit. The log is the only witness left, so on the next start we replay
//! it to recover when the session ended and which servers it visited.
//!
//! Minecraft logs a bare wall-clock time (`[12:29:00]`) in the machine's local
//! zone — no date, no offset. Absolute timestamps therefore have to be rebuilt
//! by anchoring at the session start and walking forward.
use chrono::{DateTime, Local, LocalResult, NaiveDate, NaiveDateTime, NaiveTime, TimeZone, Utc};
use super::session::{ServerJoin, parse_server_join};
/// A backwards jump larger than this means the clock wrapped past midnight
/// rather than lines merely arriving out of order between threads.
const ROLLOVER_SLACK_SECS: i64 = 12 * 60 * 60;
/// Lines that mean "no longer connected to the server we were tracking".
/// Absent any of these the span is closed by the next join, or by the session
/// end — a missing marker costs accuracy, never correctness.
const LEAVE_MARKERS: &[&str] = &[
"Stopping!",
"Stopping worker threads",
"Connection lost",
"Lost connection to server",
"Client disconnected with reason",
"Disconnected from server",
];
/// The clean-shutdown marker. Its timestamp is the truest exit time available.
const STOP_MARKER: &str = "Stopping!";
pub(crate) fn parse_log_time(line: &str) -> Option<NaiveTime> {
let stamp = line.strip_prefix('[')?.split(']').next()?.trim();
// `%.f` also matches an empty fraction, so this covers `[12:29:00]` and
// `[12:29:00.123]` alike.
if let Ok(time) = NaiveTime::parse_from_str(stamp, "%H:%M:%S%.f") {
return Some(time);
}
// Some pack-shipped log4j configs prepend the date.
for fmt in ["%Y-%m-%d %H:%M:%S%.f", "%d.%m.%Y %H:%M:%S%.f"] {
if let Ok(dt) = NaiveDateTime::parse_from_str(stamp, fmt) {
return Some(dt.time());
}
}
None
}
pub(crate) fn is_leave_marker(line: &str) -> bool {
LEAVE_MARKERS.iter().any(|marker| line.contains(marker))
}
fn local_to_utc(date: NaiveDate, time: NaiveTime) -> DateTime<Utc> {
let naive = NaiveDateTime::new(date, time);
match Local.from_local_datetime(&naive) {
LocalResult::Single(dt) => dt.with_timezone(&Utc),
// DST end: the same wall-clock time happens twice. The first is the
// better guess for a forward-walking log.
LocalResult::Ambiguous(first, _) => first.with_timezone(&Utc),
// DST start: this wall-clock time never existed, so step over the gap
// rather than reading the local time as if it were UTC — that would be
// wrong by the zone's whole offset.
LocalResult::None => Local
.from_local_datetime(&(naive + chrono::Duration::hours(1)))
.earliest()
.map(|dt| dt.with_timezone(&Utc))
.unwrap_or_else(|| Utc.from_utc_datetime(&naive)),
}
}
/// Turns the log's bare times into absolute instants by walking forward from an
/// anchor and advancing the date whenever the clock wraps.
pub(crate) struct LogClock {
date: NaiveDate,
/// High-water mark, not the previous line: threads interleave, so a line
/// may legitimately sit a second behind the one before it.
peak: NaiveTime,
}
impl LogClock {
pub(crate) fn new(anchor: DateTime<Utc>) -> Self {
let local = anchor.with_timezone(&Local).naive_local();
Self {
date: local.date(),
peak: local.time(),
}
}
pub(crate) fn absolute(&mut self, time: NaiveTime) -> DateTime<Utc> {
let behind = self.peak.signed_duration_since(time).num_seconds();
if behind > ROLLOVER_SLACK_SECS {
self.date = self.date.succ_opt().unwrap_or(self.date);
self.peak = time;
} else if time > self.peak {
self.peak = time;
}
local_to_utc(self.date, time)
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) struct ServerSpan {
pub host: String,
pub port: Option<u16>,
pub joined_at: DateTime<Utc>,
pub disconnected_at: Option<DateTime<Utc>>,
}
#[derive(Debug, Default, Clone)]
pub(crate) struct LogReplay {
/// Timestamp of the last line carrying one — a lower bound on the exit.
pub last_activity: Option<DateTime<Utc>>,
/// Set when the game logged a clean shutdown.
pub stopped_at: Option<DateTime<Utc>>,
pub servers: Vec<ServerSpan>,
}
/// Walk a whole session log, recovering server spans and the last sign of life.
pub(crate) fn replay(content: &str, started_at: DateTime<Utc>) -> LogReplay {
let mut clock = LogClock::new(started_at);
let mut out = LogReplay::default();
// Timestamps only appear on the first line of a multi-line entry (stack
// traces continue underneath), so carry the last one forward.
let mut now = started_at;
for line in content.lines() {
if let Some(time) = parse_log_time(line) {
now = clock.absolute(time);
out.last_activity = Some(now);
}
if let Some(ServerJoin { host, port }) = parse_server_join(line) {
close_open_span(&mut out, now);
out.servers.push(ServerSpan {
host,
port,
joined_at: now,
disconnected_at: None,
});
continue;
}
if is_leave_marker(line) {
close_open_span(&mut out, now);
}
if line.contains(STOP_MARKER) {
out.stopped_at = Some(now);
}
}
out
}
fn close_open_span(replay: &mut LogReplay, at: DateTime<Utc>) {
if let Some(open) = replay
.servers
.iter_mut()
.rev()
.find(|span| span.disconnected_at.is_none())
{
open.disconnected_at = Some(at.max(open.joined_at));
}
}
#[cfg(test)]
mod tests {
use super::*;
use chrono::Duration;
fn anchor() -> DateTime<Utc> {
Local
.with_ymd_and_hms(2026, 7, 16, 12, 0, 0)
.unwrap()
.with_timezone(&Utc)
}
fn local_at(h: u32, m: u32, s: u32) -> DateTime<Utc> {
Local
.with_ymd_and_hms(2026, 7, 16, h, m, s)
.unwrap()
.with_timezone(&Utc)
}
#[test]
fn parses_plain_and_fractional_times() {
assert_eq!(
parse_log_time("[12:29:00] [Client thread/INFO]: Setting user: Lynith"),
NaiveTime::from_hms_opt(12, 29, 0)
);
assert_eq!(
parse_log_time("[12:29:00.123] [Render thread/INFO]: hi"),
NaiveTime::from_hms_milli_opt(12, 29, 0, 123)
);
}
#[test]
fn parses_dated_stamp() {
assert_eq!(
parse_log_time("[2026-07-16 12:29:00] [Render thread/INFO]: hi"),
NaiveTime::from_hms_opt(12, 29, 0)
);
}
#[test]
fn ignores_untimestamped_lines() {
assert!(parse_log_time("\tat net.minecraft.Foo.bar(Foo.java:12)").is_none());
assert!(parse_log_time("[Render thread/INFO]: no time here").is_none());
}
#[test]
fn clock_keeps_same_day_while_moving_forward() {
let mut clock = LogClock::new(anchor());
let at = clock.absolute(NaiveTime::from_hms_opt(12, 30, 0).unwrap());
assert_eq!(at, local_at(12, 30, 0));
}
#[test]
fn clock_rolls_over_midnight() {
let mut clock = LogClock::new(anchor());
clock.absolute(NaiveTime::from_hms_opt(23, 59, 0).unwrap());
let at = clock.absolute(NaiveTime::from_hms_opt(0, 1, 0).unwrap());
assert_eq!(at, local_at(12, 0, 0) + Duration::hours(12) + Duration::minutes(1));
}
#[test]
fn clock_tolerates_out_of_order_threads() {
let mut clock = LogClock::new(anchor());
clock.absolute(NaiveTime::from_hms_opt(12, 29, 8).unwrap());
// A second thread's line lands a second late; this must not add a day.
let at = clock.absolute(NaiveTime::from_hms_opt(12, 29, 7).unwrap());
assert_eq!(at, local_at(12, 29, 7));
}
#[test]
fn replays_server_span_closed_by_quit() {
let log = "\
[12:29:00] [Client thread/INFO]: Setting user: Lynith
[12:30:00] [Render thread/INFO]: Connecting to mc.hypixel.net, 25565
[12:45:00] [Client thread/INFO]: Stopping!
";
let out = replay(log, anchor());
assert_eq!(out.servers.len(), 1);
let span = &out.servers[0];
assert_eq!(span.host, "mc.hypixel.net");
assert_eq!(span.port, Some(25565));
assert_eq!(span.joined_at, local_at(12, 30, 0));
assert_eq!(span.disconnected_at, Some(local_at(12, 45, 0)));
assert_eq!(out.stopped_at, Some(local_at(12, 45, 0)));
assert_eq!(out.last_activity, Some(local_at(12, 45, 0)));
}
#[test]
fn consecutive_joins_close_previous_span() {
let log = "\
[12:30:00] [Render thread/INFO]: Connecting to a.example.com, 25565
[12:40:00] [Render thread/INFO]: Connecting to b.example.com, 25566
";
let out = replay(log, anchor());
assert_eq!(out.servers.len(), 2);
assert_eq!(out.servers[0].disconnected_at, Some(local_at(12, 40, 0)));
assert_eq!(out.servers[1].disconnected_at, None);
}
#[test]
fn span_left_open_when_log_just_stops() {
let log = "[12:30:00] [Render thread/INFO]: Connecting to a.example.com, 25565\n";
let out = replay(log, anchor());
assert_eq!(out.servers[0].disconnected_at, None);
assert_eq!(out.stopped_at, None);
assert_eq!(out.last_activity, Some(local_at(12, 30, 0)));
}
#[test]
fn crash_without_stop_marker_reports_last_activity() {
let log = "\
[12:30:00] [Render thread/INFO]: Connecting to a.example.com, 25565
[12:31:00] [Render thread/ERROR]: Encountered an unexpected exception
\tat net.minecraft.Foo.bar(Foo.java:12)
";
let out = replay(log, anchor());
assert_eq!(out.stopped_at, None);
assert_eq!(out.last_activity, Some(local_at(12, 31, 0)));
}
#[test]
fn empty_log_yields_nothing() {
let out = replay("", anchor());
assert!(out.servers.is_empty());
assert!(out.last_activity.is_none());
assert!(out.stopped_at.is_none());
}
}

8
packages/oneclient_core/src/game/mod.rs
View file

@ -3,11 +3,14 @@ mod arguments;
mod download;
mod error;
mod launch;
mod log_replay;
mod metadata;
mod process;
mod reattach;
mod rules;
mod session;
mod shared_dir;
mod tail;
pub use analytics::{
Analytics, DayPlaytime, Persona, PlaytimeStats, ServerStat, WEEKDAY_LABELS, aggregate_servers,
@ -20,7 +23,10 @@ pub use arguments::{
pub use download::{download_to_path, fetch_bytes_verified};
pub use error::GameError;
pub use launch::{LaunchedGame, is_running, launch_cluster};
pub use process::{GameProcess, GameProcessManager};
pub use process::{
GameProcess, GameProcessManager, is_process_alive, kill_process, process_start_time,
};
pub use reattach::recover_sessions;
pub use metadata::{
download_minecraft, download_version_info, get_game_versions, get_loader_version,
get_loader_versions, get_loaders_for_version, is_version_updated, libraries_missing,

49
packages/oneclient_core/src/game/process.rs
View file

@ -4,10 +4,55 @@ use std::path::{Path, PathBuf};
use std::sync::Mutex;
use std::time::Instant;
use sysinfo::{Pid, ProcessesToUpdate, Signal, System};
use tokio::sync::oneshot;
use crate::notification::LaunchStage;
fn probe(pid: u32) -> Option<(System, Pid)> {
let pid = Pid::from_u32(pid);
let mut sys = System::new();
sys.refresh_processes(ProcessesToUpdate::Some(&[pid]), true);
sys.process(pid)?;
Some((sys, pid))
}
/// Process start time in unix seconds — pids get recycled, this pins identity.
pub fn process_start_time(pid: u32) -> Option<u64> {
let (sys, pid) = probe(pid)?;
Some(sys.process(pid)?.start_time())
}
/// Whether `pid` is still the process we launched. A matching pid whose start
/// time differs is a recycled pid belonging to someone else — never ours.
pub fn is_process_alive(pid: u32, started_at: Option<u64>) -> bool {
let Some(actual) = process_start_time(pid) else {
return false;
};
match started_at {
Some(expected) => actual == expected,
// Pre-migration sessions have no recorded start time. Liveness alone is
// a weaker signal, but treating the game as running is the safer error:
// worst case the exit time is recovered on a later start.
None => true,
}
}
/// Ask a process to exit. Used for games re-adopted after a launcher restart,
/// where no `Child` handle survives to kill through.
pub fn kill_process(pid: u32) -> bool {
let Some((sys, pid)) = probe(pid) else {
return false;
};
let Some(process) = sys.process(pid) else {
return false;
};
// Prefer a graceful terminate so the game can save and shut down cleanly.
process
.kill_with(Signal::Term)
.unwrap_or_else(|| process.kill())
}
#[derive(Debug, Clone)]
pub struct GameProcess {
pub pid: Option<u32>,
@ -82,6 +127,10 @@ impl GameProcessManager {
.map(|(id, _)| *id)
}
pub fn pid(&self, cluster_id: i64) -> Option<u32> {
self.inner.lock().unwrap().get(&cluster_id).and_then(|p| p.pid)
}
pub fn is_running(&self, cluster_id: i64) -> bool {
self.inner
.lock()

414
packages/oneclient_core/src/game/reattach.rs Normal file
View file

@ -0,0 +1,414 @@
//! Picking up game sessions that outlived the launcher.
//!
//! Games are spawned detached, so closing the launcher no longer takes the game
//! down with it. That leaves two loose ends to tidy on the next start, both of
//! them sessions whose row still has no `ended_at`:
//!
//! * the game is **still running** — re-adopt it, so the UI shows it as playing
//! and its exit is recorded properly when it finally happens;
//! * the game **already exited** unobserved — its log is the only witness, so
//! replay it to recover when it ended and which servers it visited.
use std::path::{Path, PathBuf};
use std::sync::Arc;
use std::time::Duration;
use chrono::{DateTime, Utc};
use oneclient_db::dao::game_session as session_dao;
use oneclient_db::models::UnfinishedSession;
use crate::clusters::{Cluster, ClusterManager};
use crate::discord::Presence;
use crate::game::launch::{Exit, SessionEnd, finalize_session};
use crate::game::log_replay::{self, ServerSpan};
use crate::game::process::{is_process_alive, kill_process};
use crate::game::session::SessionRecorder;
use crate::game::tail::spawn_log_tail;
use crate::notification::LaunchStage;
use crate::state::LauncherState;
/// How often to check whether a re-adopted game is still alive. It is only a
/// fallback for a process we don't own, so a slow poll is plenty.
const LIVENESS_POLL: Duration = Duration::from_secs(3);
/// Settle every session left open by a previous launcher run. Safe to call once
/// at startup, before anything can launch a game.
#[tracing::instrument(skip(state), level = "debug")]
pub async fn recover_sessions(state: &Arc<LauncherState>) {
let sessions = match session_dao::unfinished_sessions(&state.services.db).await {
Ok(sessions) => sessions,
Err(err) => {
tracing::warn!(error = %err, "failed to list unfinished game sessions");
return;
}
};
if sessions.is_empty() {
return;
}
tracing::info!(count = sessions.len(), "recovering unfinished game sessions");
let mut live = Vec::new();
let mut dead = Vec::new();
for session in sessions {
match classify(state, session).await {
Some((cluster, session, started_at, Some(pid))) => {
live.push((cluster, session, started_at, pid));
}
Some((cluster, session, started_at, None)) => {
dead.push((cluster, session, started_at));
}
None => {}
}
}
// Re-adopt the live games first. Reconciling an exited session tears down
// its shared game directory, which a still-running cluster may be sharing —
// registering the live ones up front is what makes that check possible.
for (cluster, session, started_at, pid) in live {
readopt(state, cluster, session, started_at, pid).await;
}
for (cluster, session, started_at) in dead {
reconcile(state, cluster, session, started_at).await;
}
}
/// Resolve a session to its cluster and decide whether its process is alive.
/// `None` means the session was closed out here and needs no further work.
async fn classify(
state: &Arc<LauncherState>,
session: UnfinishedSession,
) -> Option<(Cluster, UnfinishedSession, DateTime<Utc>, Option<u32>)> {
let cluster_id = session.cluster_id;
// A session this launcher already owns needs no recovery.
if state.games.is_active(cluster_id) {
return None;
}
let started_at = match DateTime::parse_from_rfc3339(&session.started_at) {
Ok(at) => at.with_timezone(&Utc),
Err(err) => {
tracing::warn!(cluster_id, session = %session.started_at, error = %err, "unparseable session start; closing it");
close_untraceable(state, &session).await;
return None;
}
};
// The cluster is gone, so there is no log to consult and no playtime worth
// attributing. Just close the row so it stops being reconsidered forever.
let Ok(cluster) = ClusterManager::get(state, cluster_id).await else {
tracing::warn!(cluster_id, "cluster missing for unfinished session; closing it");
close_untraceable(state, &session).await;
return None;
};
let alive = session
.pid
.and_then(|pid| u32::try_from(pid).ok())
.filter(|pid| is_process_alive(*pid, session.pid_started_at.map(|t| t as u64)));
Some((cluster, session, started_at, alive))
}
/// Close a session we can say nothing about. Ending it at its start time books
/// no playtime, which beats inventing some.
async fn close_untraceable(state: &Arc<LauncherState>, session: &UnfinishedSession) {
if let Err(err) = session_dao::finish_session_at(
&state.services.db,
&session.started_at,
&session.started_at,
None,
)
.await
{
tracing::warn!(error = %err, "failed to close untraceable session");
}
}
/// The game is still playing. Wire it back up as though we had launched it.
#[tracing::instrument(skip(state, cluster, session), fields(cluster_id = cluster.id, pid), level = "debug")]
async fn readopt(
state: &Arc<LauncherState>,
cluster: Cluster,
session: UnfinishedSession,
started_at: DateTime<Utc>,
pid: u32,
) {
let cluster_id = cluster.id;
let Ok(cwd) = cluster.game_dir() else {
return;
};
tracing::info!(cluster_id, pid, "re-adopting running game");
let open_server = open_server_of(state, &session.started_at).await;
let recorder = SessionRecorder::resume(
state.services.db.clone(),
session.started_at.clone(),
open_server,
);
state.games.set_stage(cluster_id, LaunchStage::Running);
state.games.set_pid(cluster_id, Some(pid));
state.games.set_dir(cluster_id, cwd.clone());
state.services.notifier.game_stage(cluster_id, LaunchStage::Running);
state.discord.set_presence(Presence::Playing {
cluster: cluster.name.clone(),
mc_version: cluster.mc_version.clone(),
});
let Ok(log_path) = crate::logs::cluster_output_log(&cluster) else {
return;
};
let tail = spawn_log_tail(
cluster_id,
log_path,
state.services.notifier.clone(),
Some(recorder.clone()),
);
let (kill_tx, kill_rx) = tokio::sync::oneshot::channel::<()>();
state.games.register_kill(cluster_id, kill_tx);
// The post hook belongs to whichever settings apply now; resolve it up front
// so the watcher task doesn't have to reach back into the state.
let post_hook = ClusterManager::resolve_settings(state, &cluster)
.await
.ok()
.and_then(|profile| profile.hook_post);
let dedicated = cluster.uses_dedicated_dir();
let pid_started_at = session.pid_started_at.map(|t| t as u64);
let state = Arc::clone(state);
tokio::spawn(async move {
tokio::select! {
() = wait_for_exit(pid, pid_started_at) => {}
_ = kill_rx => {
kill_process(pid);
wait_for_exit(pid, pid_started_at).await;
}
}
tail.stop().await;
finalize_session(
&state,
&cluster,
&cwd,
dedicated,
post_hook.as_deref(),
Some(recorder),
SessionEnd {
started_at,
// We watched this one die, so the time is real even though the
// exit code went with the launcher that spawned it.
ended_at: Utc::now(),
outcome: Exit::Inferred,
// Only tear down the cluster's running state if it is still
// ours. With parallel clusters enabled the user can start a
// fresh game on this cluster while the re-adopted one plays on,
// and that newer game now owns the slot.
owns_slot: state.games.pid(cluster_id) == Some(pid),
},
)
.await;
});
}
async fn wait_for_exit(pid: u32, pid_started_at: Option<u64>) {
while is_process_alive(pid, pid_started_at) {
tokio::time::sleep(LIVENESS_POLL).await;
}
}
/// The game exited while nothing was watching. Rebuild what we can from the log.
#[tracing::instrument(skip(state, cluster, session), fields(cluster_id = cluster.id), level = "debug")]
async fn reconcile(
state: &Arc<LauncherState>,
cluster: Cluster,
session: UnfinishedSession,
started_at: DateTime<Utc>,
) {
let cluster_id = cluster.id;
let log = read_session_log(&cluster).await;
let replay = log
.as_ref()
.map(|log| log_replay::replay(&log.content, started_at))
.unwrap_or_default();
// Prefer the log's own testimony over the file's mtime: a clean shutdown
// says exactly when it happened, and the last timestamped line is the last
// proof of life. mtime only helps when nothing in the log carries a time.
let ended_at = replay
.stopped_at
.or(replay.last_activity)
.or_else(|| log.as_ref().and_then(|log| log.modified))
.unwrap_or(started_at)
.clamp(started_at, Utc::now());
tracing::info!(
cluster_id,
session = %session.started_at,
%ended_at,
clean_exit = replay.stopped_at.is_some(),
servers = replay.servers.len(),
"recovered session that outlived the launcher"
);
let open_server = apply_spans(state, &session.started_at, &replay.servers).await;
let recorder = SessionRecorder::resume(
state.services.db.clone(),
session.started_at.clone(),
open_server,
);
let Ok(cwd) = cluster.game_dir() else {
recorder.finish_at(&ended_at.to_rfc3339(), None).await;
return;
};
// Unwinding the shared directory is only safe if nobody is playing in it.
// A cluster re-adopted moments ago may share this exact directory, and
// clearing its content mid-session would pull the game apart underneath it.
let shared_dir_busy = state.games.dir_in_use_by(&cwd, cluster_id).is_some();
if shared_dir_busy {
tracing::debug!(cluster_id, "shared game dir still in use; skipping exit cleanup");
}
finalize_session(
state,
&cluster,
&cwd,
cluster.uses_dedicated_dir() || shared_dir_busy,
// The post hook is deliberately skipped. It is an *on exit* hook, and
// the exit already happened — possibly days ago. Firing it now, during
// startup, would surprise anyone whose hook does something real.
None,
Some(recorder),
SessionEnd {
started_at,
ended_at,
outcome: Exit::Inferred,
// This session is over, but the cluster may have a newer one playing
// right now — only claim the slot if nothing else holds it.
owns_slot: !state.games.is_active(cluster_id),
},
)
.await;
}
/// Rewrite a session's server spans from the log replay, which — unlike the
/// live rows — covers the whole session including the part we missed. Returns
/// the `joined_at` of a span still left open, for the caller to close.
async fn apply_spans(
state: &Arc<LauncherState>,
session_id: &str,
spans: &[ServerSpan],
) -> Option<String> {
let db = &state.services.db;
let existing = session_dao::list_session_servers(db, session_id)
.await
.unwrap_or_default();
// A log that yields fewer spans than we already recorded has been rotated or
// truncated. Live rows are then the better record; keep them and just report
// whichever is still open so it gets closed at the session's end.
if spans.len() < existing.len() {
tracing::debug!(
session = %session_id,
replayed = spans.len(),
existing = existing.len(),
"log replay is thinner than recorded spans; keeping recorded ones"
);
return existing
.into_iter()
.find(|row| row.disconnected_at.is_none())
.map(|row| row.joined_at);
}
if let Err(err) = session_dao::delete_session_servers(db, session_id).await {
tracing::warn!(session = %session_id, error = %err, "failed to clear server spans for replay");
return None;
}
let mut open = None;
for span in spans {
let joined_at = span.joined_at.to_rfc3339();
if let Err(err) = session_dao::insert_server_join_at(
db,
session_id,
&span.host,
span.port.map(i64::from),
&joined_at,
)
.await
{
tracing::warn!(session = %session_id, error = %err, "failed to replay server join");
continue;
}
match span.disconnected_at {
Some(at) => {
if let Err(err) =
session_dao::finish_server_at(db, &joined_at, &at.to_rfc3339()).await
{
tracing::warn!(session = %session_id, error = %err, "failed to replay server leave");
}
}
None => open = Some(joined_at),
}
}
open
}
async fn open_server_of(state: &Arc<LauncherState>, session_id: &str) -> Option<String> {
session_dao::list_session_servers(&state.services.db, session_id)
.await
.unwrap_or_default()
.into_iter()
.find(|row| row.disconnected_at.is_none())
.map(|row| row.joined_at)
}
struct SessionLog {
content: String,
modified: Option<DateTime<Utc>>,
}
/// Read whichever log covers the session. `cluster-output.log` is preferred: the
/// launcher truncates it at every launch, so it belongs to exactly one session,
/// whereas `latest.log` is the game's own and may already have been rotated.
async fn read_session_log(cluster: &Cluster) -> Option<SessionLog> {
let mut candidates: Vec<PathBuf> = Vec::new();
if let Ok(path) = crate::logs::cluster_output_log(cluster) {
candidates.push(path);
}
if let Ok(dir) = cluster.dir() {
candidates.push(dir.join("logs").join("latest.log"));
}
for path in candidates {
match read_log(&path).await {
Some(log) if !log.content.trim().is_empty() => return Some(log),
_ => continue,
}
}
None
}
async fn read_log(path: &Path) -> Option<SessionLog> {
let content = polyio::read_to_string(path).await.ok()?;
let modified = polyio::stat(path)
.await
.ok()
.and_then(|meta| meta.modified().ok())
.map(DateTime::<Utc>::from);
Some(SessionLog { content, modified })
}

66
packages/oneclient_core/src/game/session.rs
View file

@ -15,6 +15,9 @@ pub(crate) struct ServerJoin {
pub port: Option<u16>,
}
/// The session row's identity — its `started_at`, which is the primary key.
pub(crate) type SessionId = String;
pub(crate) fn parse_server_join(line: &str) -> Option<ServerJoin> {
const MARKER: &str = "Connecting to ";
let idx = line.find(MARKER)?;
@ -85,11 +88,49 @@ impl SessionRecorder {
})
}
pub(crate) async fn observe(&self, line: &str) {
let Some(join) = parse_server_join(line) else {
return;
};
/// Re-attach to a session row left open by a previous launcher run, so a
/// game that outlived the launcher keeps recording into the same session.
pub(crate) fn resume(db: DbPool, session_started_at: SessionId, open_server: Option<String>) -> Self {
Self {
session_started_at,
db,
open_server: Arc::new(Mutex::new(open_server)),
}
}
/// When the session row says it began. Playtime is measured against this so
/// it agrees with the session span analytics reads back out of the row.
pub(crate) fn started_at(&self) -> Option<chrono::DateTime<chrono::Utc>> {
chrono::DateTime::parse_from_rfc3339(&self.session_started_at)
.ok()
.map(|at| at.with_timezone(&chrono::Utc))
}
/// Remember which OS process backs this session. Without it, a launcher
/// restart cannot tell a live game from one that exited unobserved.
#[tracing::instrument(skip(self), level = "debug")]
pub(crate) async fn record_process(&self, pid: u32, pid_started_at: Option<u64>) {
if let Err(err) = session_dao::set_session_process(
&self.db,
&self.session_started_at,
Some(i64::from(pid)),
pid_started_at.map(|t| t as i64),
)
.await
{
tracing::warn!(pid, error = %err, "failed to record session process");
}
}
pub(crate) async fn observe(&self, line: &str) {
if let Some(join) = parse_server_join(line) {
self.open(join).await;
} else if super::log_replay::is_leave_marker(line) {
self.close_open().await;
}
}
async fn open(&self, join: ServerJoin) {
let mut open = self.open_server.lock().await;
if let Some(prev) = open.take()
&& let Err(err) = session_dao::finish_server(&self.db, &prev).await
@ -111,16 +152,27 @@ impl SessionRecorder {
}
}
async fn close_open(&self) {
if let Some(prev) = self.open_server.lock().await.take()
&& let Err(err) = session_dao::finish_server(&self.db, &prev).await
{
tracing::warn!(error = %err, "failed to close server row");
}
}
/// Close the session. The time is explicit because an exit is not always
/// observed as it happens — one recovered from a log ended in the past.
#[tracing::instrument(skip(self), fields(exit_code), level = "debug")]
pub(crate) async fn finish(self, exit_code: Option<i64>) {
pub(crate) async fn finish_at(self, ended_at: &str, exit_code: Option<i64>) {
if let Some(open) = self.open_server.lock().await.take()
&& let Err(err) = session_dao::finish_server(&self.db, &open).await
&& let Err(err) = session_dao::finish_server_at(&self.db, &open, ended_at).await
{
tracing::warn!(error = %err, "failed to close open server on exit");
}
if let Err(err) =
session_dao::finish_session(&self.db, &self.session_started_at, exit_code).await
session_dao::finish_session_at(&self.db, &self.session_started_at, ended_at, exit_code)
.await
{
tracing::warn!(session = %self.session_started_at, error = %err, "failed to finish game session");
}

2
packages/oneclient_core/src/game/shared_dir.rs
View file

@ -308,7 +308,7 @@ async fn move_dir_contents(from: &Path, to: &Path) {
let dest = to.join(name);
if tokio::fs::rename(&src, &dest).await.is_err() {
if polyio::rename(&src, &dest).await.is_err() {
tracing::warn!(file = %src.display(), "failed to salvage leaked log file");
}
}

279
packages/oneclient_core/src/game/tail.rs Normal file
View file

@ -0,0 +1,279 @@
use std::path::PathBuf;
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, Ordering};
use std::time::Duration;
use tokio::io::{AsyncBufReadExt, AsyncSeekExt, BufReader};
use crate::game::session::SessionRecorder;
use crate::notification::NotificationService;
const POLL_INTERVAL: Duration = Duration::from_millis(200);
pub(crate) struct LogTail {
stop: Arc<AtomicBool>,
handle: tokio::task::JoinHandle<()>,
}
impl LogTail {
/// Stop tailing, but only after one last pass: the lines explaining *why*
/// the game exited are written moments before it does, and would be lost if
/// the tail stopped the instant the process did.
pub(crate) async fn stop(self) {
self.stop.store(true, Ordering::Relaxed);
let _ = self.handle.await;
}
}
pub(crate) fn spawn_log_tail(
cluster_id: i64,
path: PathBuf,
notifier: NotificationService,
recorder: Option<SessionRecorder>,
) -> LogTail {
let stop = Arc::new(AtomicBool::new(false));
let flag = Arc::clone(&stop);
let handle = tokio::spawn(async move {
if let Err(err) = tail(cluster_id, &path, &notifier, recorder.as_ref(), &flag).await {
tracing::warn!(cluster_id, path = %path.display(), error = %err, "game log tail stopped");
}
});
LogTail { stop, handle }
}
async fn tail(
cluster_id: i64,
path: &PathBuf,
notifier: &NotificationService,
recorder: Option<&SessionRecorder>,
stop: &AtomicBool,
) -> std::io::Result<()> {
// The game may not have created the file yet on a re-adopt, and on launch it
// exists but is empty. Either way, wait for it rather than giving up.
let file = loop {
match tokio::fs::File::open(path).await {
Ok(file) => break file,
Err(_) if !stop.load(Ordering::Relaxed) => {
tokio::time::sleep(POLL_INTERVAL).await;
}
Err(err) => return Err(err),
}
};
let mut reader = BufReader::new(file);
let mut line = String::new();
let mut offset = 0u64;
loop {
line.clear();
let read = reader.read_line(&mut line).await?;
if read == 0 {
// A final pass has already run by the time the flag is seen here,
// since the read above returned EOF on a fully drained file.
if stop.load(Ordering::Relaxed) {
return Ok(());
}
// The log is truncated when a cluster relaunches; if the file
// shrank, our offset is past the end and we would never read again.
let len = polyio::stat(path).await.map(|m| m.len()).unwrap_or(offset);
if len < offset {
reader.seek(std::io::SeekFrom::Start(0)).await?;
offset = 0;
}
tokio::time::sleep(POLL_INTERVAL).await;
continue;
}
offset += read as u64;
// A partial line means the game is mid-write; wait for the newline
// rather than emitting half a line and re-emitting the rest.
if !line.ends_with('\n') {
// Unless the game is already gone — a process killed mid-write
// leaves a newline that will never arrive, and waiting for it would
// hang the caller of `stop()` forever.
if stop.load(Ordering::Relaxed) {
emit(cluster_id, &line, notifier, recorder).await;
return Ok(());
}
reader.seek(std::io::SeekFrom::Start(offset - read as u64)).await?;
offset -= read as u64;
tokio::time::sleep(POLL_INTERVAL).await;
continue;
}
emit(cluster_id, &line, notifier, recorder).await;
}
}
async fn emit(
cluster_id: i64,
line: &str,
notifier: &NotificationService,
recorder: Option<&SessionRecorder>,
) {
// Blank lines are kept: the game prints them, and the console should read
// the way the log does.
let text = line.trim_end_matches(['\n', '\r']).to_string();
if let Some(recorder) = recorder {
recorder.observe(&text).await;
}
notifier.game_log(cluster_id, text);
}
#[cfg(test)]
mod tests {
use super::*;
use crate::notification::Notification;
use tokio::io::AsyncWriteExt;
use tokio::sync::mpsc;
struct Harness {
dir: async_tempfile::TempDir,
rx: mpsc::UnboundedReceiver<Notification>,
notifier: NotificationService,
}
impl Harness {
/// The temp dir removes itself on drop, so nothing here needs cleaning up.
async fn new() -> Self {
let (tx, rx) = mpsc::unbounded_channel();
let dir = polyio::tempdir().await.expect("temp dir");
Self {
dir,
rx,
notifier: NotificationService::new(tx),
}
}
fn path(&self) -> PathBuf {
self.dir.dir_path().join("cluster-output.log")
}
/// Drain whatever the tail has emitted so far.
fn drain(&mut self) -> Vec<String> {
let mut out = Vec::new();
while let Ok(Notification::GameLog { line, .. }) = self.rx.try_recv() {
out.push(line);
}
out
}
}
/// Give the tail a few poll cycles to notice what we wrote.
async fn settle() {
tokio::time::sleep(POLL_INTERVAL * 4).await;
}
#[tokio::test]
async fn follows_lines_appended_after_start() {
let mut h = Harness::new().await;
let path = h.path();
polyio::write(&path, "[12:00:00] first\n").await.unwrap();
let tail = spawn_log_tail(7, path.clone(), h.notifier.clone(), None);
settle().await;
let mut file = tokio::fs::OpenOptions::new()
.append(true)
.open(&path)
.await
.unwrap();
file.write_all(b"[12:00:01] second\n").await.unwrap();
file.flush().await.unwrap();
settle().await;
tail.stop().await;
assert_eq!(h.drain(), vec!["[12:00:00] first", "[12:00:01] second"]);
}
#[tokio::test]
async fn waits_for_a_file_that_does_not_exist_yet() {
let mut h = Harness::new().await;
let path = h.path();
let tail = spawn_log_tail(7, path.clone(), h.notifier.clone(), None);
settle().await;
polyio::write(&path, "[12:00:00] late\n").await.unwrap();
settle().await;
tail.stop().await;
assert_eq!(h.drain(), vec!["[12:00:00] late"]);
}
#[tokio::test]
async fn holds_back_a_partial_line_until_its_newline_lands() {
let mut h = Harness::new().await;
let path = h.path();
polyio::write(&path, "[12:00:00] half").await.unwrap();
let tail = spawn_log_tail(7, path.clone(), h.notifier.clone(), None);
settle().await;
// A line mid-write must not be emitted, or it would arrive twice.
assert!(h.drain().is_empty());
let mut file = tokio::fs::OpenOptions::new()
.append(true)
.open(&path)
.await
.unwrap();
file.write_all(b" and half\n").await.unwrap();
file.flush().await.unwrap();
settle().await;
tail.stop().await;
assert_eq!(h.drain(), vec!["[12:00:00] half and half"]);
}
#[tokio::test]
async fn recovers_when_the_log_is_truncated() {
let mut h = Harness::new().await;
let path = h.path();
polyio::write(&path, "[12:00:00] old session line\n").await.unwrap();
let tail = spawn_log_tail(7, path.clone(), h.notifier.clone(), None);
settle().await;
assert_eq!(h.drain(), vec!["[12:00:00] old session line"]);
// A relaunch truncates the log; the tail would otherwise sit forever at
// an offset past the new end and go silent.
polyio::write(&path, "[13:00:00] new\n").await.unwrap();
settle().await;
tail.stop().await;
assert_eq!(h.drain(), vec!["[13:00:00] new"]);
}
#[tokio::test]
async fn stops_even_when_the_last_line_has_no_newline() {
let mut h = Harness::new().await;
let path = h.path();
// A process killed mid-write leaves a newline that never arrives.
polyio::write(&path, "[12:00:00] cut off mid-writ").await.unwrap();
let tail = spawn_log_tail(7, path.clone(), h.notifier.clone(), None);
settle().await;
// Waiting for that newline would hang the exit path forever.
let stopped = tokio::time::timeout(Duration::from_secs(5), tail.stop()).await;
assert!(stopped.is_ok(), "tail.stop() hung on an unterminated final line");
assert_eq!(h.drain(), vec!["[12:00:00] cut off mid-writ"]);
}
#[tokio::test]
async fn emits_the_final_lines_written_just_before_exit() {
let mut h = Harness::new().await;
let path = h.path();
polyio::write(&path, "[12:00:00] Stopping!\n").await.unwrap();
// Stopping immediately, as the exit path does, must still flush the
// reason the game exited rather than race past it.
let tail = spawn_log_tail(7, path.clone(), h.notifier.clone(), None);
tail.stop().await;
assert_eq!(h.drain(), vec!["[12:00:00] Stopping!"]);
}
}

2
packages/oneclient_core/src/java/install.rs
View file

@ -57,7 +57,7 @@ pub async fn install_package(
.await;
}
let _ = tokio::fs::remove_file(&archive_path).await;
let _ = polyio::remove_file(&archive_path).await;
tracing::info!(vendor = %package.vendor, major, "installed Java runtime");

6
packages/oneclient_core/src/packages/store/download.rs
View file

@ -24,9 +24,9 @@ pub async fn ensure_artifact_file(
if dest.exists() {
let existing = sha1_file(dest).await?;
if normalize_hash(&existing) == normalize_hash(hash) {
return Ok(tokio::fs::metadata(dest).await?.len());
return Ok(polyio::stat(dest).await?.len());
}
tokio::fs::remove_file(dest).await?;
polyio::remove_file(dest).await?;
}
tracing::debug!("downloading artifact file");
@ -45,7 +45,7 @@ pub async fn ensure_artifact_file(
.into());
}
Ok(tokio::fs::metadata(dest).await?.len())
Ok(polyio::stat(dest).await?.len())
}
#[allow(clippy::too_many_arguments)]

2
packages/oneclient_core/src/packages/store/link.rs
View file

@ -74,7 +74,7 @@ pub async fn unlink_cluster_file(
.join(file_name);
if path.exists() {
tokio::fs::remove_file(&path).await?;
polyio::remove_file(&path).await?;
}
Ok(())
}

4
packages/oneclient_core/src/packages/store/mod.rs
View file

@ -270,9 +270,9 @@ impl PackageStore {
if let Some(parent) = dest.parent() {
polyio::create_dir_all(parent).await?;
}
tokio::fs::copy(path, &dest).await?;
polyio::copy(path, &dest).await?;
let size = tokio::fs::metadata(&dest).await?.len();
let size = polyio::stat(&dest).await?.len();
let stored_path = relative_cache_path(&dest)?;
let row = artifact_dao::insert_artifact(

8
packages/oneclient_core/src/recovery.rs
View file

@ -160,7 +160,7 @@ async fn index_cache_file(
.file_name()
.map(|n| n.to_string_lossy().into_owned())
.unwrap_or_default();
let size = tokio::fs::metadata(file).await.ok().map(|m| m.len() as i64);
let size = polyio::stat(file).await.ok().map(|m| m.len() as i64);
artifact_dao::insert_artifact(
pool,
@ -265,7 +265,7 @@ async fn relink_cluster_files(
if artifact_dao::get_artifact_by_hash(pool, &hash).await?.is_none() {
let stored_path = relative_launcher_path(launcher_dir, &file);
let size = tokio::fs::metadata(&file).await.ok().map(|m| m.len() as i64);
let size = polyio::stat(&file).await.ok().map(|m| m.len() as i64);
artifact_dao::insert_artifact(
pool,
&hash,
@ -388,7 +388,7 @@ fn relative_launcher_path(launcher_dir: &Path, abs: &Path) -> String {
async fn list_dir_names(dir: &Path) -> LauncherResult<Vec<String>> {
let mut names = Vec::new();
let Ok(mut entries) = tokio::fs::read_dir(dir).await else {
let Ok(mut entries) = polyio::read_dir(dir).await else {
return Ok(names);
};
while let Some(entry) = entries.next_entry().await? {
@ -401,7 +401,7 @@ async fn list_dir_names(dir: &Path) -> LauncherResult<Vec<String>> {
async fn list_files(dir: &Path) -> LauncherResult<Vec<PathBuf>> {
let mut files = Vec::new();
let Ok(mut entries) = tokio::fs::read_dir(dir).await else {
let Ok(mut entries) = polyio::read_dir(dir).await else {
return Ok(files);
};
while let Some(entry) = entries.next_entry().await? {

2
packages/oneclient_core/src/state.rs
View file

@ -91,6 +91,8 @@ impl LauncherState {
}
};
crate::game::recover_sessions(&background).await;
let (versions_res, bundles_res) = tokio::join!(
background.versions.sync(&background.services),
background.bundles.sync(&background.services),

12
packages/oneclient_db/.sqlx/query-71fbe674b85a1fd6dc07a062e96a35888bb3b6a5a4fda5df3c662ca16b1d801e.json generated Normal file
View file

@ -0,0 +1,12 @@
{
"db_name": "SQLite",
"query": "\n\t\tUPDATE game_sessions\n\t\tSET pid = ?, pid_started_at = ?\n\t\tWHERE started_at = ?\n\t\t",
"describe": {
"columns": [],
"parameters": {
"Right": 3
},
"nullable": []
},
"hash": "71fbe674b85a1fd6dc07a062e96a35888bb3b6a5a4fda5df3c662ca16b1d801e"
}

12
packages/oneclient_db/.sqlx/query-b42a04c7b073b6cceff8fbd48a7b55f9beddb045bc43eaef851da89b3e547933.json generated Normal file
View file

@ -0,0 +1,12 @@
{
"db_name": "SQLite",
"query": "\n\t\tDELETE FROM game_session_servers\n\t\tWHERE session_started_at = ?\n\t\t",
"describe": {
"columns": [],
"parameters": {
"Right": 1
},
"nullable": []
},
"hash": "b42a04c7b073b6cceff8fbd48a7b55f9beddb045bc43eaef851da89b3e547933"
}

62
packages/oneclient_db/.sqlx/query-c49ab599ef0aa31813f2a84d86f79235e2470eeec5b10f1e768fcf54d37d57fe.json generated Normal file
View file

@ -0,0 +1,62 @@
{
"db_name": "SQLite",
"query": "\n\t\tSELECT cluster_id, started_at, pid, pid_started_at\n\t\tFROM game_sessions\n\t\tWHERE ended_at IS NULL\n\t\tORDER BY started_at ASC\n\t\t",
"describe": {
"columns": [
{
"name": "cluster_id",
"ordinal": 0,
"type_info": "Integer",
"origin": {
"Table": {
"table": "game_sessions",
"name": "cluster_id"
}
}
},
{
"name": "started_at",
"ordinal": 1,
"type_info": "Text",
"origin": {
"Table": {
"table": "game_sessions",
"name": "started_at"
}
}
},
{
"name": "pid",
"ordinal": 2,
"type_info": "Integer",
"origin": {
"Table": {
"table": "game_sessions",
"name": "pid"
}
}
},
{
"name": "pid_started_at",
"ordinal": 3,
"type_info": "Integer",
"origin": {
"Table": {
"table": "game_sessions",
"name": "pid_started_at"
}
}
}
],
"parameters": {
"Right": 0
},
"nullable": [
false,
false,
true,
true
]
},
"hash": "c49ab599ef0aa31813f2a84d86f79235e2470eeec5b10f1e768fcf54d37d57fe"
}

6
packages/oneclient_db/migrations/20260716120000_session_process_tracking.sql Normal file
View file

@ -0,0 +1,6 @@
-- Track the OS process behind a session so a restarted launcher can tell a
-- still-running game from one that exited while the launcher was closed.
-- `pid_started_at` is the process start time (unix seconds), used to reject
-- a recycled pid that now belongs to an unrelated process.
ALTER TABLE game_sessions ADD COLUMN pid INTEGER;
ALTER TABLE game_sessions ADD COLUMN pid_started_at INTEGER;

89
packages/oneclient_db/src/dao/game_session.rs
View file

@ -3,6 +3,7 @@ use sqlx::SqlitePool;
use crate::models::{
GameSessionRow, GameSessionServerRow, NewGameSession, ServerJoinCount, SessionSpan,
UnfinishedSession,
};
pub async fn insert_session(
@ -36,7 +37,17 @@ pub async fn finish_session(
started_at: &str,
exit_code: Option<i64>,
) -> Result<(), sqlx::Error> {
let ended_at = Utc::now().to_rfc3339();
finish_session_at(pool, started_at, &Utc::now().to_rfc3339(), exit_code).await
}
/// Close a session at an explicit time. Used when the exit time is inferred
/// from the game's log rather than observed live.
pub async fn finish_session_at(
pool: &SqlitePool,
started_at: &str,
ended_at: &str,
exit_code: Option<i64>,
) -> Result<(), sqlx::Error> {
sqlx::query!(
r#"
UPDATE game_sessions
@ -52,13 +63,58 @@ pub async fn finish_session(
Ok(())
}
pub async fn set_session_process(
pool: &SqlitePool,
started_at: &str,
pid: Option<i64>,
pid_started_at: Option<i64>,
) -> Result<(), sqlx::Error> {
sqlx::query!(
r#"
UPDATE game_sessions
SET pid = ?, pid_started_at = ?
WHERE started_at = ?
"#,
pid,
pid_started_at,
started_at
)
.execute(pool)
.await?;
Ok(())
}
pub async fn unfinished_sessions(pool: &SqlitePool) -> Result<Vec<UnfinishedSession>, sqlx::Error> {
sqlx::query_as!(
UnfinishedSession,
r#"
SELECT cluster_id, started_at, pid, pid_started_at
FROM game_sessions
WHERE ended_at IS NULL
ORDER BY started_at ASC
"#
)
.fetch_all(pool)
.await
}
pub async fn insert_server_join(
pool: &SqlitePool,
session_started_at: &str,
address: &str,
port: Option<i64>,
) -> Result<String, sqlx::Error> {
let joined_at = Utc::now().to_rfc3339();
insert_server_join_at(pool, session_started_at, address, port, &Utc::now().to_rfc3339()).await
}
/// Record a join at an explicit time, for spans replayed out of a game log.
pub async fn insert_server_join_at(
pool: &SqlitePool,
session_started_at: &str,
address: &str,
port: Option<i64>,
joined_at: &str,
) -> Result<String, sqlx::Error> {
sqlx::query!(
r#"
INSERT INTO game_session_servers (session_started_at, address, port, joined_at)
@ -71,11 +127,19 @@ pub async fn insert_server_join(
)
.execute(pool)
.await?;
Ok(joined_at)
Ok(joined_at.to_string())
}
pub async fn finish_server(pool: &SqlitePool, joined_at: &str) -> Result<(), sqlx::Error> {
let disconnected_at = Utc::now().to_rfc3339();
finish_server_at(pool, joined_at, &Utc::now().to_rfc3339()).await
}
/// Close a server span at an explicit time, for spans replayed out of a log.
pub async fn finish_server_at(
pool: &SqlitePool,
joined_at: &str,
disconnected_at: &str,
) -> Result<(), sqlx::Error> {
sqlx::query!(
r#"
UPDATE game_session_servers
@ -128,6 +192,23 @@ pub async fn list_session_servers(
.await
}
/// Drop every server span of a session, so a log replay can rewrite them from scratch.
pub async fn delete_session_servers(
pool: &SqlitePool,
session_started_at: &str,
) -> Result<(), sqlx::Error> {
sqlx::query!(
r#"
DELETE FROM game_session_servers
WHERE session_started_at = ?
"#,
session_started_at
)
.execute(pool)
.await?;
Ok(())
}
pub async fn server_join_counts(
pool: &SqlitePool,
cluster_id: i64,

8
packages/oneclient_db/src/models/game_session.rs
View file

@ -23,6 +23,14 @@ pub struct NewGameSession<'a> {
pub java_version: Option<&'a str>,
}
#[derive(Debug, Clone, FromRow)]
pub struct UnfinishedSession {
pub cluster_id: i64,
pub started_at: String,
pub pid: Option<i64>,
pub pid_started_at: Option<i64>,
}
#[derive(Debug, Clone, FromRow)]
pub struct GameSessionServerRow {
pub session_started_at: String,

2
packages/oneclient_db/src/models/mod.rs
View file

@ -16,7 +16,7 @@ pub use cluster_bundle::{
};
pub use game_session::{
GameSessionId, GameSessionRow, GameSessionServerRow, NewGameSession, ServerJoinCount,
SessionSpan,
SessionSpan, UnfinishedSession,
};
pub use java::JavaVersionRow;
pub use setting_profile::SettingProfileRow;

5975
uncommitted.patch
View file

File diff suppressed because it is too large Load diff