Merge hotfix for large local database startup

Author: Yohjisakamoto

.github/workflows/ci.yml

@@ -10,6 +10,7 @@ on:
     branches:
       - main
       - "fix/**"
+      - "hotfix/**"
       - "release/**"
       - "staging/runtime_session_refactor"
     tags:

apps/tui/src/gateway/autostart.ts

@@ -1,5 +1,13 @@
-import { execFile, spawn, type ChildProcess } from "node:child_process";
-import { closeSync, existsSync, mkdirSync, openSync, realpathSync } from "node:fs";
+import { execFile, spawn, spawnSync, type ChildProcess } from "node:child_process";
+import {
+  closeSync,
+  existsSync,
+  mkdirSync,
+  openSync,
+  readSync,
+  realpathSync,
+  statSync,
+} from "node:fs";
 import { dirname, join, resolve } from "node:path";
 import { promisify } from "node:util";
 import { fileURLToPath } from "node:url";
@@ -19,7 +27,7 @@ import {
 type StartupStep = "checking";
 
 const HEALTH_POLL_INTERVAL_MS = 500;
-const GATEWAY_START_TIMEOUT_MS = 20_000;
+const GATEWAY_START_TIMEOUT_MS = 120_000;
 
 interface GatewayIdentity {
   root: string;
@@ -271,6 +279,8 @@ async function launchGatewayProcess(request: GatewayLaunchRequest): Promise<stri
   if (!executable) throw new Error(t("gatewayMissingBinary"));
   const targetUrl = stripTrailingSlash(request.targetUrl);
   let exited: { code: number | null; signal: NodeJS.Signals | null } | undefined;
+  const stderrPath = gatewayAutostartLogPath(request.instanceHome, "stderr");
+  const stderrOffset = fileSize(stderrPath);
   const stdio = gatewayLogStdio(request.instanceHome);
   const child = spawn(executable, [], {
     detached: true,
@@ -283,10 +293,16 @@ async function launchGatewayProcess(request: GatewayLaunchRequest): Promise<stri
     exited = { code, signal };
   });
   child.unref();
-  const deadline = Date.now() + GATEWAY_START_TIMEOUT_MS;
+  const deadline = Date.now() + gatewayStartTimeoutMs;
   while (Date.now() < deadline) {
     if (exited) {
-      throw new Error(`Gateway exited before becoming healthy (${exitDescription(exited)}).`);
+      throw new Error(
+        gatewayLaunchError(
+          `Gateway exited before becoming healthy (${exitDescription(exited)}).`,
+          stderrPath,
+          stderrOffset,
+        ),
+      );
     }
     const activeUrl = readActiveGatewayUrl(request.instanceHome);
     const candidateUrl = activeUrl ? stripTrailingSlash(activeUrl) : targetUrl;
@@ -300,7 +316,7 @@ async function launchGatewayProcess(request: GatewayLaunchRequest): Promise<stri
     await delay(HEALTH_POLL_INTERVAL_MS);
   }
   stopUnreadyChild(child);
-  throw new Error(t("gatewayStartTimeout"));
+  throw new Error(gatewayLaunchError(t("gatewayStartTimeout"), stderrPath, stderrOffset));
 }
 
 function gatewayProcessEnv(request: GatewayLaunchRequest): NodeJS.ProcessEnv {
@@ -344,19 +360,64 @@ function exitDescription(exit: { code: number | null; signal: NodeJS.Signals | n
 
 function stopUnreadyChild(child: ChildProcess): void {
   try {
-    if (child.exitCode === null && child.signalCode === null) child.kill();
+    if (child.exitCode !== null || child.signalCode !== null) return;
+    if (process.platform === "win32" && child.pid) {
+      const result = spawnSync("taskkill.exe", ["/PID", String(child.pid), "/T", "/F"], {
+        stdio: "ignore",
+        windowsHide: true,
+      });
+      if (!result.error && result.status === 0) return;
+    }
+    child.kill();
   } catch {
     // Best effort: the gateway may have already detached or exited.
   }
 }
 
+function gatewayAutostartLogPath(instanceHome: string, stream: "stdout" | "stderr"): string {
+  return join(instanceHome, ".tura", "logs", `gateway-autostart.${stream}.log`);
+}
+
+function fileSize(path: string): number {
+  try {
+    return statSync(path).size;
+  } catch {
+    return 0;
+  }
+}
+
+function gatewayLaunchError(message: string, stderrPath: string, offset: number): string {
+  try {
+    const size = fileSize(stderrPath);
+    if (size <= offset) return message;
+    const start = Math.max(offset, size - 8_192);
+    const buffer = Buffer.alloc(size - start);
+    const file = openSync(stderrPath, "r");
+    try {
+      readSync(file, buffer, 0, buffer.length, start);
+    } finally {
+      closeSync(file);
+    }
+    const detail = buffer
+      .toString("utf8")
+      .split(/\r?\n/u)
+      .map((line) => line.trim())
+      .filter(Boolean)
+      .slice(-8)
+      .join(" | ");
+    return detail ? `${message} Gateway stderr: ${detail}` : message;
+  } catch {
+    return message;
+  }
+}
+
 function gatewayLogStdio(instanceHome: string): ["ignore", number, number] {
   const logDir = join(instanceHome, ".tura", "logs");
   mkdirSync(logDir, { recursive: true });
   return [
     "ignore",
-    openSync(join(logDir, "gateway-autostart.stdout.log"), "a"),
-    openSync(join(logDir, "gateway-autostart.stderr.log"), "a"),
+    openSync(gatewayAutostartLogPath(instanceHome, "stdout"), "a"),
+    openSync(gatewayAutostartLogPath(instanceHome, "stderr"), "a"),
   ];
 }
 

crates/gateway/src/router_process.rs

@@ -21,7 +21,10 @@ use std::{
     time::{Duration, Instant},
 };
 
-const ROUTER_HEALTH_TIMEOUT: Duration = Duration::from_secs(20);
+// This must exceed the session_db startup budget (30 seconds). Gateway may
+// perform two router attempts, while the TUI allows 120 seconds for the whole
+// process tree to become healthy.
+const ROUTER_HEALTH_TIMEOUT: Duration = Duration::from_secs(45);
 const DEFAULT_ROUTER_EXECUTION_TIMEOUT: Duration = Duration::from_secs(35 * 60);
 const ROUTER_PROBE_CONNECT_TIMEOUT: Duration = Duration::from_millis(100);
 const ROUTER_STARTUP_POLL_INTERVAL: Duration = Duration::from_millis(200);

crates/gateway/src/session_feed.rs

@@ -80,7 +80,21 @@ pub fn start_session_feed_tailer() -> Result<(SessionFeedTailer, tokio::sync::on
                 match subscription.poll_next_entry(SUBSCRIPTION_POLL_INTERVAL) {
                     Ok(Poll::Ready(Some(entry))) => {
                         if let Err(error) = reducer.apply(entry) {
-                            break Err(error.context("failed to reduce durable Session feed"));
+                            if !is_session_feed_cursor_gap(&error) {
+                                break Err(error.context("failed to reduce durable Session feed"));
+                            }
+                            match reconnect_session_feed(
+                                &mut reducer,
+                                &thread_stopping,
+                                &thread_cancellation,
+                            ) {
+                                Ok(Some(reconnected)) => subscription = reconnected,
+                                Ok(None) => break Ok(()),
+                                Err(error) => {
+                                    break Err(error
+                                        .context("failed to resynchronize durable Session feed"));
+                                }
+                            }
                         }
                     }
                     Ok(Poll::Ready(None)) if thread_stopping.load(Ordering::SeqCst) => break Ok(()),
@@ -187,6 +201,14 @@ fn reconnect_session_feed(
     }
 }
 
+fn is_session_feed_cursor_gap(error: &anyhow::Error) -> bool {
+    error.chain().any(|cause| {
+        cause
+            .to_string()
+            .starts_with("session feed cursor gap for ")
+    })
+}
+
 fn replay_all_sessions(client: &SessionDbClient, reducer: &mut SessionFeedReducer) -> Result<()> {
     let mut canonical_session_ids = HashSet::new();
     for workspace in client.list_workspaces()? {

crates/router/src/services/session_db.rs

@@ -6,7 +6,8 @@
 use anyhow::{anyhow, Result};
 use serde_json::json;
 use session_log_contract::client::{
-    call_service, service_addr_path, service_is_running, unreachable_owner_lock_message,
+    call_service, service_addr_path, service_is_running, session_db_owner_record_path,
+    unreachable_owner_lock_message,
 };
 use std::{
     path::{Path, PathBuf},
@@ -56,7 +57,9 @@ impl SessionDbService {
         if service_is_running() {
             return Ok(self.status_payload("running"));
         }
-        if let Some(message) = unreachable_owner_lock_message() {
+        if let Some(message) = unreachable_owner_lock_message()
+            && !terminate_orphaned_session_db_owner()?
+        {
             return Err(anyhow!(message));
         }
         let service_bin = session_db_binary()
@@ -66,6 +69,13 @@ impl SessionDbService {
         let mut command = Command::new(&service_bin);
         command
             .env("TURA_HOME", tura_path::instance_home())
+            .env("TURA_ROUTER_PARENT_PID", std::process::id().to_string())
+            .env(
+                "TURA_ROUTER_PARENT_START_TIME",
+                current_process_start_time(std::process::id())
+                    .unwrap_or_default()
+                    .to_string(),
+            )
             .stdin(Stdio::null())
             .stdout(Stdio::null())
             .stderr(Stdio::null());
@@ -200,6 +210,108 @@ impl SessionDbService {
     }
 }
 
+#[derive(Default)]
+struct SessionDbOwnerRecord {
+    pid: Option<u32>,
+    process_start_time: Option<u64>,
+    kind: Option<String>,
+    build_kind: Option<String>,
+    home: Option<String>,
+    parent_pid: Option<u32>,
+    parent_process_start_time: Option<u64>,
+}
+
+fn terminate_orphaned_session_db_owner() -> Result<bool> {
+    let raw = match std::fs::read_to_string(session_db_owner_record_path()) {
+        Ok(raw) => raw,
+        Err(_) => return Ok(false),
+    };
+    let mut record = SessionDbOwnerRecord::default();
+    for line in raw.lines() {
+        let Some((key, value)) = line.split_once('=') else {
+            continue;
+        };
+        let value = value.trim();
+        match key.trim() {
+            "pid" => record.pid = value.parse().ok(),
+            "process_start_time" => record.process_start_time = value.parse().ok(),
+            "kind" => record.kind = Some(value.to_string()),
+            "build_kind" => record.build_kind = Some(value.to_string()),
+            "home" => record.home = Some(value.to_string()),
+            "parent_pid" => record.parent_pid = value.parse().ok(),
+            "parent_process_start_time" => record.parent_process_start_time = value.parse().ok(),
+            _ => {}
+        }
+    }
+    if record.kind.as_deref() != Some("session_db")
+        || record.build_kind.as_deref() != Some(tura_path::build_kind())
+        || !record
+            .home
+            .as_deref()
+            .is_some_and(|home| same_path(home, &tura_path::instance_home()))
+    {
+        return Ok(false);
+    }
+    let (Some(pid), Some(start_time), Some(parent_pid), Some(parent_start_time)) = (
+        record.pid,
+        record.process_start_time,
+        record.parent_pid,
+        record.parent_process_start_time,
+    ) else {
+        return Ok(false);
+    };
+    let mut system = sysinfo::System::new_all();
+    system.refresh_processes();
+    if system
+        .process(sysinfo::Pid::from_u32(parent_pid))
+        .is_some_and(|parent| parent.start_time() == parent_start_time)
+    {
+        return Ok(false);
+    }
+    let Some(process) = system.process(sysinfo::Pid::from_u32(pid)) else {
+        return Ok(false);
+    };
+    if process.start_time() != start_time
+        || !process
+            .name()
+            .trim_end_matches(".exe")
+            .eq_ignore_ascii_case("tura_session_db")
+        || service_is_running()
+    {
+        return Ok(false);
+    }
+    if !process.kill() {
+        return Ok(false);
+    }
+    let started = Instant::now();
+    while started.elapsed() < Duration::from_secs(10) {
+        system.refresh_processes();
+        if system.process(sysinfo::Pid::from_u32(pid)).is_none() {
+            return Ok(true);
+        }
+        std::thread::sleep(Duration::from_millis(100));
+    }
+    Ok(false)
+}
+
+fn current_process_start_time(pid: u32) -> Option<u64> {
+    let mut system = sysinfo::System::new_all();
+    system.refresh_processes();
+    system
+        .process(sysinfo::Pid::from_u32(pid))
+        .map(sysinfo::Process::start_time)
+}
+
+fn same_path(left: &str, right: &Path) -> bool {
+    let left = tura_path::normalize_path(Path::new(left));
+    let right = tura_path::normalize_path(right);
+    if cfg!(windows) {
+        left.to_string_lossy().to_lowercase() == right.to_string_lossy().to_lowercase()
+    } else {
+        left == right
+    }
+}
+
 fn wait_for_child_exit(child: &mut Child, timeout: Duration) -> bool {
     let started = Instant::now();
     while started.elapsed() < timeout {

crates/session_log/Cargo.toml

@@ -51,11 +51,11 @@ lifecycle.workspace = true
 rusqlite.workspace = true
 serde.workspace = true
 serde_json.workspace = true
+sysinfo.workspace = true
 tracing.workspace = true
 tura_path.workspace = true
 session_log_contract.workspace = true
 uuid.workspace = true
 
 [dev-dependencies]
-sysinfo.workspace = true
 tempfile.workspace = true