diff --git a/desktop/internal/engine/archive.mbt b/desktop/internal/engine/archive.mbt index d23349a36..dc78ade21 100644 --- a/desktop/internal/engine/archive.mbt +++ b/desktop/internal/engine/archive.mbt @@ -336,12 +336,35 @@ async fn migrate_legacy_archive( ) -> Unit { guard manager.runtime_dir is Some(dir) else { return } let list = dir.join(LegacyArchivedListFile) + // Probe before locking: this migration has already run for almost every + // launch, and a lock sidecar should not outlive the file it guards. + guard (@fsx.try_read_text(list) catch { + error if @async.is_being_cancelled() => raise error + _ => None + }) + is Some(_) else { + return + } + guard resolved_session_root() is Some(root) else { return } + // Another instance may be migrating the same list. The record moves are + // renames, so without this the loser of a race reads its own "already + // gone" as a failed move and writes the id back for the next launch to + // retry — forever. + let shared = @fsx.FileLock::acquire(list, Exclusive) catch { + error if @async.is_being_cancelled() => raise error + // Every other filesystem failure here is best-effort, and the lock is no + // different: a runtime directory that turned read-only, or one on a + // filesystem without advisory locks, must leave the caller listing the + // archives it already has rather than failing the whole request over a + // one-time migration. + _ => return + } + defer shared.release() let text = (@fsx.try_read_text(list) catch { error if @async.is_being_cancelled() => raise error _ => None }).unwrap_or("") guard !text.is_empty() else { return } - guard resolved_session_root() is Some(root) else { return } let remaining : Array[String] = [] for line in text.split("\n") { let session = line.trim().to_owned() @@ -358,7 +381,7 @@ async fn migrate_legacy_archive( _ => () } } else { - @fsx.write_text(list, "\{remaining.join("\n")}\n") catch { + @fsx.write_text_atomic(list, "\{remaining.join("\n")}\n") catch { error if @async.is_being_cancelled() => raise error _ => () } diff --git a/desktop/internal/engine/settings.mbt b/desktop/internal/engine/settings.mbt index 5b4f458d3..e6b357eee 100644 --- a/desktop/internal/engine/settings.mbt +++ b/desktop/internal/engine/settings.mbt @@ -207,6 +207,12 @@ pub async fn update_settings( guard manager.runtime_dir is Some(dir) else { raise EngineError("engine settings cannot persist: no runtime directory") } + // `settings_lock` orders this process's clients; the runtime directory is + // shared with every other instance running against it. The load below has + // to follow this lock — settings read before it would carry back the very + // fields another instance just changed. + let shared = @fsx.FileLock::acquire(settings_file(dir), Exclusive) + defer shared.release() let current = load_engine_settings(manager.runtime_dir) guard current.version <= EngineSettingsVersion else { raise EngineError( @@ -274,36 +280,15 @@ fn apply_text_patch( } ///| -/// Atomic save: write a 0600 temp file, then rename over the target. The -/// file holds a credential, so a crash mid-write must never leave a -/// truncated store — the temp file keeps the full content until the rename -/// lands. Windows may refuse to rename over an existing file; removing the -/// target first narrows atomicity to that one platform's small window. +/// Atomic save at 0600. The file holds a credential, so a crash mid-write +/// must never leave a truncated store, and every reader of it — this process, +/// another instance, the next launch — sees one complete version or the +/// other. Reads therefore take no lock; only the read-modify-write in +/// `update_settings` does. async fn write_settings_file(runtime_dir : @pathx.Path, text : String) -> Unit { - let target = settings_file(runtime_dir) - let temp = runtime_dir.join(EngineSettingsFileName + ".tmp") - @fs.write_file( - temp.to_string(), - text, - create_mode=CreateOrTruncate, - permission=0o600, - ) catch { + @fsx.write_text_atomic(settings_file(runtime_dir), text, permission=0o600) catch { error if @async.is_being_cancelled() => raise error - error => raise EngineError("could not write engine settings: \{error}") - } - @fs.rename(temp.to_string(), target.to_string()) catch { - error if @async.is_being_cancelled() => raise error - _ => { - @fsx.remove_if_exists(target) catch { - error if @async.is_being_cancelled() => raise error - error => - raise EngineError("could not replace engine settings: \{error}") - } - @fs.rename(temp.to_string(), target.to_string()) catch { - error if @async.is_being_cancelled() => raise error - error => raise EngineError("could not save engine settings: \{error}") - } - } + error => raise EngineError("could not save engine settings: \{error}") } } diff --git a/desktop/internal/engine/workspaces.mbt b/desktop/internal/engine/workspaces.mbt index 577e3c82e..f82e76fbb 100644 --- a/desktop/internal/engine/workspaces.mbt +++ b/desktop/internal/engine/workspaces.mbt @@ -14,6 +14,12 @@ const WorkspacesFile = "workspaces.json" /// lock. Remote requests run in independent tasks; without this serialization /// two concurrent add/remove operations can both read the same old list and /// let the later write silently erase the earlier one. +/// +/// It orders this process only. The registry is shared with every other +/// writer of the same session root — a second desktop instance, a TUI in the +/// same workspace — so each mutation also takes the cross-process +/// `@fsx.FileLock` on the registry file. Reads need neither lock: the write +/// is atomic, so a reader sees one complete version or the other. let workspace_registry_lock : @async.Mutex = Mutex() ///| @@ -23,6 +29,19 @@ fn workspaces_file() -> @pathx.Absolute? { resolved_session_root().map(root => root.join(WorkspacesFile)) } +///| +/// The registry file a mutation must lock and write. Unlike reads, which +/// degrade to "no workspaces", a mutation with nowhere to persist has to say +/// so rather than silently succeed against nothing. +fn required_workspaces_file() -> @pathx.Absolute raise EngineError { + guard workspaces_file() is Some(file) else { + raise EngineError( + "cannot determine a session root; set OPENSEEK_SESSION_ROOT", + ) + } + file +} + ///| /// The canonical (symlink-resolved) spelling of a workspace directory — the /// registry's one identity per project, so `/tmp/x` and `/private/tmp/x` @@ -77,21 +96,20 @@ pub async fn registered_workspaces() -> Array[@pathx.Absolute] { } ///| +/// Save the registry. Atomic, so the concurrent readers this file has — other +/// desktop instances, a TUI, the next launch after a crash — never observe a +/// truncated list and read it as "no workspaces attached". async fn write_registry(paths : Array[@pathx.Absolute]) -> Unit { - guard resolved_session_root() is Some(root) else { - raise EngineError( - "cannot determine a session root; set OPENSEEK_SESSION_ROOT", - ) - } - @fsx.ensure_dir(root.to_path()) catch { + let file = required_workspaces_file() + @fsx.ensure_dir(file.to_path().dirname()) catch { error if @async.is_being_cancelled() => raise error - error => raise EngineError("could not create \{root}: \{error}") + error => + raise EngineError( + "could not create \{file.to_path().dirname()}: \{error}", + ) } let registry : Json = { "workspaces": paths.map(path => path.0).to_json() } - @fsx.write_text( - root.join(WorkspacesFile).to_path(), - registry.stringify(indent=2), - ) catch { + @fsx.write_text_atomic(file.to_path(), registry.stringify(indent=2)) catch { error if @async.is_being_cancelled() => raise error error => raise EngineError("could not save the workspace list: \{error}") } @@ -133,6 +151,14 @@ pub async fn add_workspace( exclude_from_git(dir, ".openseek") workspace_registry_lock.acquire() defer workspace_registry_lock.release() + // The registry read has to follow both locks: a list read before another + // process released this one is exactly the stale snapshot whose write-back + // would erase that process's entry. + let shared = @fsx.FileLock::acquire( + required_workspaces_file().to_path(), + Exclusive, + ) + defer shared.release() let paths = read_registry_unlocked() // The write and broadcast callback are one finite, cancellation-shielded // linearization point while the lock is held. Once disk changes, every @@ -159,6 +185,11 @@ pub async fn remove_workspace( let target = canonical_workspace_path(expanded_workspace_path(path)) workspace_registry_lock.acquire() defer workspace_registry_lock.release() + let shared = @fsx.FileLock::acquire( + required_workspaces_file().to_path(), + Exclusive, + ) + defer shared.release() let paths = read_registry_unlocked().filter(p => p != target) @async.protect_from_cancel(() => { write_registry(paths) diff --git a/desktop/internal/engine/worktrees.mbt b/desktop/internal/engine/worktrees.mbt index 0d567d3e1..50a65c994 100644 --- a/desktop/internal/engine/worktrees.mbt +++ b/desktop/internal/engine/worktrees.mbt @@ -23,7 +23,10 @@ const WorktreeBranchPrefix = "openseek/" /// Registry reads and read-modify-write mutations share one process-wide /// lock, exactly like `workspaces.json` — without it two concurrent /// create/remove operations could both read the same old list and let the -/// later write silently erase the earlier one. +/// later write silently erase the earlier one. And exactly like +/// `workspaces.json`, it orders this process only: each mutation also takes +/// the cross-process `@fsx.FileLock` on that workspace's registry file, while +/// reads rely on the write being atomic. let worktree_registry_lock : @async.Mutex = Mutex() ///| @@ -130,7 +133,9 @@ async fn write_worktrees( error => raise EngineError("could not create \{store}: \{error}") } let registry : Json = { "worktrees": entries.to_json() } - @fsx.write_text( + // Atomic, so a concurrent reader never sees a truncated list and mistakes + // a bound conversation for one that runs in the workspace root. + @fsx.write_text_atomic( worktrees_file(workspace).to_path(), registry.stringify(indent=2), ) catch { @@ -612,6 +617,11 @@ pub async fn create_worktree( ensure_session_not_archived(session) worktree_registry_lock.acquire() defer worktree_registry_lock.release() + // Held across the git work below as well as the registry write: a create + // that released between them would let another instance's create pick the + // same `wt-N` after this one probed it as free. + let shared = @fsx.FileLock::acquire(worktrees_file(dir).to_path(), Exclusive) + defer shared.release() let entries = read_worktrees_unlocked(dir) // This conversation already holds its environment. Two shapes land here: // the idempotent replay of a create whose reply was lost, and the repair @@ -707,11 +717,11 @@ pub async fn create_worktree( } exclude_from_git(dir, WorktreesDirName) let base = git_in(dir, ["rev-parse", "HEAD"]) - // The branch comes first, on its own: our locks are in-process only, so - // another process can race the probes above — `git branch` then fails - // atomically having created nothing, and from here on the branch is - // provably THIS invocation's, which is what makes the rollback's - // `branch -D` safe. + // The branch comes first, on its own: our locks bind desktop instances, + // not the user's own git, so anything outside the app can still race the + // probes above — `git branch` then fails atomically having created + // nothing, and from here on the branch is provably THIS invocation's, + // which is what makes the rollback's `branch -D` safe. ignore(git_in(dir, ["branch", branch, base])) ignore( git_in(dir, ["worktree", "add", "\{WorktreesDirName}/\{name}", branch]) catch { @@ -894,8 +904,11 @@ pub async fn remove_worktree( } worktree_registry_lock.acquire() defer worktree_registry_lock.release() + let shared = @fsx.FileLock::acquire(worktrees_file(dir).to_path(), Exclusive) + defer shared.release() // Re-read: another worktree may have been created while the checkout was - // being removed (creation takes only the registry lock). + // being removed — by this process, which takes only the registry lock for + // that, or by another one, which this file lock has just been waiting on. let remaining = read_worktrees_unlocked(dir).filter(entry => { entry.name != name }) diff --git a/desktop/internal/fsx/lock.mbt b/desktop/internal/fsx/lock.mbt new file mode 100644 index 000000000..78583a3f7 --- /dev/null +++ b/desktop/internal/fsx/lock.mbt @@ -0,0 +1,370 @@ +// Cross-process serialization for shared durable state. This process is not +// the only writer of the files under the session root and the runtime +// directory: a second desktop instance, a TUI launched in the same workspace, +// and a leftover process from a previous run all reach the same registries. An +// `@async.Mutex` orders this process's own tasks and nothing else, so a +// read-modify-write that must not lose a concurrent update pairs one with the +// advisory file lock here. + +///| +/// The advisory lock beside `path`. It is a sidecar rather than the data file +/// itself because an atomic save replaces the target by rename: a lock taken +/// on the data file would be attached to the inode that rename orphans, and +/// two writers would each hold "the lock" on a different file. +fn lock_file(path : @pathx.Path) -> @pathx.Path { + path.dirname().join(path.basename().to_owned() + ".lock") +} + +///| +/// An advisory lock on one shared file, held across processes until `release`. +/// It is acquired beside the `@async.Mutex` that orders this process's own +/// tasks, and in the same shape: +/// +/// ```moonbit skip +/// registry_lock.acquire() +/// defer registry_lock.release() +/// let shared = @fsx.FileLock::acquire(registry_file, Exclusive) +/// defer shared.release() +/// ``` +/// +/// `Exclusive` must span a complete read-modify-write: the read has to happen +/// after the lock is granted, since a value read before it may already be +/// stale. `Shared` covers a read that must not overlap a mutation — a reader +/// of a file saved by `write_text_atomic` needs no lock at all. +/// +/// The lock only orders writers that take it. Keeping unlocked readers safe +/// is `write_text_atomic`'s job: another tool, an older build, or a plain +/// `cat` never takes this lock and must still see a complete file. +struct FileLock { + file : @fs.File +} + +///| +/// Take the advisory lock guarding `path`, waiting for whichever process +/// holds it. Blocking is the point: a mutation that gave up here would be the +/// lost update the lock exists to prevent. +pub async fn FileLock::acquire(path : @pathx.Path, kind : @fs.Lock) -> FileLock { + // The lock precedes the data file: a first-run registry write creates the + // directory itself, and there is nothing to open a lock in until it exists. + ensure_dir(path.dirname()) + let file = open_lock_file(lock_file(path).to_string()) + // A wait interrupted by cancellation leaves the descriptor ours to close; + // leaking it would also leak the lock for as long as the process lives. + file.lock(kind) catch { + error => { + file.close() + raise error + } + } + { file, } +} + +///| +/// Take the lock only if nothing holds it right now. +/// +/// For the caller that is asking *whether* someone holds it rather than +/// waiting for a turn — blocking there would mean waiting on the very +/// process the question is about. `None` means someone else holds it. +pub async fn FileLock::try_acquire( + path : @pathx.Path, + kind : @fs.Lock, +) -> FileLock? { + ensure_dir(path.dirname()) + let file = open_lock_file(lock_file(path).to_string()) + let taken = file.try_lock(kind) catch { + error => { + file.close() + raise error + } + } + guard taken else { + file.close() + return None + } + Some({ file, }) +} + +///| +/// Open the sidecar without being redirected by a link at its name. The +/// sidecar has to live beside the data file — every writer, this app or a +/// TUI, must agree on where it is — and beside the data file can mean inside +/// an attached repository, which is free to ship its own +/// `.openseek/worktrees.json.lock`. +/// +/// A plain `OpenOrCreate` follows such a link twice over: it creates a file +/// wherever the link points, and it locks that inode instead, so two +/// instances following a link that moved would each hold a lock on a +/// different file and both proceed into the registry write this exists to +/// serialize. +/// +/// `CreateNew` (O_CREAT|O_EXCL) never follows a link, so the create path is +/// closed outright. The already-exists path — the ordinary one — rejects +/// anything that is not a regular file by its own `lstat`. That leaves a +/// window for a link swapped in between the check and the open: closing it +/// needs `O_NOFOLLOW`, which this filesystem API does not expose. What it +/// does close is the case that gets here without an attacker present at the +/// keyboard, which is a link committed to a repository. +async fn open_lock_file(sidecar : String) -> @fs.File { + let created = Some( + @fs.open(sidecar, mode=ReadWrite, create_mode=CreateNew, permission=0o600), + ) catch { + error if @async.is_being_cancelled() => raise error + _ => None + } + if created is Some(file) { + return file + } + let regular = (@fs.kind(sidecar, follow_symlink=false) is Regular) catch { + error if @async.is_being_cancelled() => raise error + _ => false + } + guard regular else { + fail("\{sidecar} is not a regular file; refusing to lock through it") + } + @fs.open(sidecar, mode=ReadWrite, create_mode=OpenExisting) +} + +///| +pub fn FileLock::release(self : FileLock) -> Unit { + self.file.unlock() + self.file.close() +} + +///| +/// Replace `path`'s contents in one step: write a private file beside it, then +/// rename it over the target. A reader therefore sees either the old contents +/// or the new ones and never a truncated prefix, whether or not it takes the +/// lock, and a crash mid-write leaves the previous file intact. +pub async fn write_text_atomic( + path : @pathx.Path, + content : String, + permission? : Int = 0o644, +) -> Unit { + let target = path.to_string() + // The directory is not always ours. A workspace is whatever repository the + // user attached, and a repository can carry a symlink at a guessable + // `.tmp` — committed before the attach that adds the git exclusion. + // Writing through it would truncate whatever it points at and then rename + // the link itself into place as the registry. So: an unguessable name, and + // `CreateNew` (O_CREAT|O_EXCL), which refuses any existing name including a + // symlink. The random suffix also keeps a leftover from a crashed run from + // blocking every later save, which a fixed name plus O_EXCL would. + guard @entropy.random_hex(8) is Some(token) else { + fail("could not name a private temporary file beside \{path}") + } + let temp = path.dirname().join(path.basename().to_owned() + ".tmp-" + token) + // Nothing reuses this name, so every failure from here on has to take the + // file away itself — including a write that raised partway (a cancelled + // task, a full disk), which would otherwise leave one abandoned file per + // attempt, holding whatever the caller was saving. Cancellation is among + // those failures, hence the shield around the removal. + try { + @fs.write_file( + temp.to_string(), + content, + create_mode=CreateNew, + permission~, + ) + commit_temp_file(temp, target) + } catch { + error => { + @async.protect_from_cancel(() => discard_temp_file(temp)) + raise error + } + } +} + +///| +/// Move a finished temp file onto `target`. Windows may refuse to rename over +/// an existing file; removing the target first narrows the window where +/// neither name holds the new content to that one platform. +async fn commit_temp_file(temp : @pathx.Path, target : String) -> Unit { + @fs.rename(temp.to_string(), target) catch { + error if @async.is_being_cancelled() => raise error + rename_failure => { + // The retry exists only for that Windows refusal, which is a regular + // file standing in the way. Anything else at the target is a different + // failure and must surface as one: a directory there means the path is + // wrong, and removing it would recursively destroy whatever it holds + // rather than report that. + let replaceable = (@fs.kind(target) is Regular) catch { + error if @async.is_being_cancelled() => raise error + _ => false + } + guard replaceable else { raise rename_failure } + // Between the remove and the rename neither name holds the content. + // Cancellation there would leave the target absent and let the caller's + // cleanup take the temp file too, destroying the file this call was + // supposed to update — so the pair commits as one. + @async.protect_from_cancel(() => { + @fs.remove(target) + @fs.rename(temp.to_string(), target) + }) + } + } +} + +///| +/// Best-effort removal of an abandoned temp file: the caller is already +/// raising the failure that matters, and a cleanup error must not replace it. +async fn discard_temp_file(temp : @pathx.Path) -> Unit { + @fs.remove(temp.to_string()) catch { + _ => () + } +} + +// Tests. The lock is flock-based — it belongs to the open file description, +// not the process — so two acquires of one path contend inside a single +// process through exactly the mechanism a second instance would hit. That is +// what makes these tests meaningful without spawning a helper process. + +///| +/// Every leftover matters now that the temp name is random: a save that +/// forgot to clean up would litter the directory instead of overwriting one +/// reused name. +async fn temp_leftovers(root : @pathx.Path, base : String) -> Array[String] { + @fs.readdir(root.to_string()).filter(name => { + name != base && name.has_prefix(base) + }) +} + +///| +async test "an atomic save replaces the target and leaves no temp behind" { + let root : @pathx.Path = @fs.tmpdir(prefix="openseek-fsx-atomic-") + let file = root.join("registry.json") + write_text_atomic(file, "first") + assert_eq(read_text(file), "first") + write_text_atomic(file, "second") + assert_eq(read_text(file), "second") + assert_eq(temp_leftovers(root, "registry.json"), []) + @fs.rmdir(root.to_string(), recursive=true) +} + +///| +/// The lock must not be redirectable either: an attached repository can ship +/// `.openseek/worktrees.json.lock` as a link, and following it would both +/// create a file wherever it points and lock the wrong inode. +async test "a lock refuses a sidecar that is not a regular file" { + let root : @pathx.Path = @fs.tmpdir(prefix="openseek-fsx-locklink-") + let victim = root.join("victim.txt") + let file = root.join("registry.json") + @fs.symlink( + root.join("registry.json.lock").to_string(), + target=victim.to_string(), + ) + let mut refused = false + let taken = Some(FileLock::acquire(file, Exclusive)) catch { + _ => { + refused = true + None + } + } + if taken is Some(lock) { + lock.release() + } + assert_true(refused) + // Following the link would have created the file it points at. + assert_false(exists(victim)) + @fs.rmdir(root.to_string(), recursive=true) +} + +///| +/// The save must not be redirectable out of its directory. A repository can +/// carry a symlink at the guessable `.tmp` before it is ever attached +/// as a workspace; writing through it would truncate the link's target and +/// then rename the link itself into place as the registry. +async test "an atomic save never writes through a guessable temp name" { + let root : @pathx.Path = @fs.tmpdir(prefix="openseek-fsx-symlink-") + let victim = root.join("victim.txt") + write_text_atomic(victim, "precious") + let planted = root.join("registry.json.tmp") + @fs.symlink(planted.to_string(), target=victim.to_string()) + let file = root.join("registry.json") + write_text_atomic(file, "saved") + assert_eq(read_text(file), "saved") + // The planted link is untouched, and so is what it points at. + assert_eq(read_text(victim), "precious") + assert_true(@fs.kind(planted.to_string(), follow_symlink=false) is SymLink) + @fs.rmdir(root.to_string(), recursive=true) +} + +///| +async test "an exclusive lock makes the next holder wait for the release" { + let root : @pathx.Path = @fs.tmpdir(prefix="openseek-fsx-exclusive-") + let file = root.join("registry.json") + let log : Array[String] = [] + @async.with_task_group(group => { + let first = FileLock::acquire(file, Exclusive) + group.spawn(() => { + let second = FileLock::acquire(file, Exclusive) + defer second.release() + log.push("second acquired") + }) + |> ignore + // Long enough for the waiter to reach the lock and block on it. + @async.sleep(100) + log.push("first releasing") + first.release() + }) + assert_eq(log, ["first releasing", "second acquired"]) + @fs.rmdir(root.to_string(), recursive=true) +} + +///| +/// Readers must not queue behind each other, or every unlocked read this +/// design relies on would have been the wrong call. +async test "shared locks admit each other" { + let root : @pathx.Path = @fs.tmpdir(prefix="openseek-fsx-shared-") + let file = root.join("registry.json") + let log : Array[String] = [] + @async.with_task_group(group => { + let first = FileLock::acquire(file, Shared) + group.spawn(() => { + let second = FileLock::acquire(file, Shared) + defer second.release() + log.push("second acquired") + }) + |> ignore + @async.sleep(100) + log.push("first releasing") + first.release() + }) + assert_eq(log, ["second acquired", "first releasing"]) + @fs.rmdir(root.to_string(), recursive=true) +} + +///| +/// The Windows retry must not turn "something else already occupies this +/// path" into a recursive delete of whatever that something holds. +async test "an atomic save refuses a target that is not a regular file" { + let root : @pathx.Path = @fs.tmpdir(prefix="openseek-fsx-kind-") + let occupied = root.join("registry.json") + ensure_dir(occupied) + let kept = occupied.join("keep.txt") + write_text_atomic(kept, "precious") + let mut refused = false + write_text_atomic(occupied, "clobber") catch { + _ => refused = true + } + assert_true(refused) + assert_eq(read_text(kept), "precious") + // The refusal must also not leave its half-finished temp file behind. + assert_eq(temp_leftovers(root, "registry.json"), []) + @fs.rmdir(root.to_string(), recursive=true) +} + +///| +/// A released lock must leave nothing behind that keeps the next acquire out +/// — the failure this catches is `release` closing the descriptor without +/// unlocking, which only shows up on the second acquire. +async test "a lock can be retaken after release" { + let root : @pathx.Path = @fs.tmpdir(prefix="openseek-fsx-retake-") + let file = root.join("registry.json") + for _ in 0..<3 { + let lock = FileLock::acquire(file, Exclusive) + write_text_atomic(file, "held") + lock.release() + } + assert_eq(read_text(file), "held") + @fs.rmdir(root.to_string(), recursive=true) +} diff --git a/desktop/internal/fsx/moon.pkg b/desktop/internal/fsx/moon.pkg index c48dbc240..30206a2ef 100644 --- a/desktop/internal/fsx/moon.pkg +++ b/desktop/internal/fsx/moon.pkg @@ -2,6 +2,7 @@ import { "moonbitlang/async", "moonbitlang/async/os_error", "moonbitlang/async/fs", + "openseek_desktop/internal/entropy", "openseek_desktop/internal/pathx", "tonyfettes/platform", } diff --git a/desktop/internal/fsx/pkg.generated.mbti b/desktop/internal/fsx/pkg.generated.mbti index a8e8b1c5a..4c57e5106 100644 --- a/desktop/internal/fsx/pkg.generated.mbti +++ b/desktop/internal/fsx/pkg.generated.mbti @@ -2,6 +2,7 @@ package "openseek_desktop/internal/fsx" import { + "moonbitlang/async/fs", "openseek_desktop/internal/pathx", } @@ -30,6 +31,8 @@ pub async fn write_bytes(@pathx.Path, Bytes) -> Unit pub async fn write_text(@pathx.Path, String) -> Unit +pub async fn write_text_atomic(@pathx.Path, String, permission? : Int) -> Unit + // Errors // Types and methods @@ -42,6 +45,11 @@ type DirectoryEntry pub fn DirectoryEntry::is_dir(Self) -> Bool pub fn DirectoryEntry::name(Self) -> String +type FileLock +pub async fn FileLock::acquire(@pathx.Path, @fs.Lock) -> Self +pub fn FileLock::release(Self) -> Unit +pub async fn FileLock::try_acquire(@pathx.Path, @fs.Lock) -> Self? + // Type aliases // Traits diff --git a/desktop/internal/host/host.mbt b/desktop/internal/host/host.mbt index 7a0c85a14..7e92ef128 100644 --- a/desktop/internal/host/host.mbt +++ b/desktop/internal/host/host.mbt @@ -86,11 +86,17 @@ pub fn new_host_state( ///| /// The background tasks behind the host ops, spawned into the caller's /// app-lifetime task group: the moon-check diagnostics forwarder and the -/// self-update pump with its startup leftover sweep. These are shared by +/// self-update download pump. These are shared by /// every connection; /// connection-owned filesystem watcher tasks start in `HostConnection::run`. /// Every background task is best-effort: a language-service hiccup degrades /// that feature, never the connection. +/// +/// The update pump is safe to start before this process knows whether it owns +/// the runtime directory: it only ever acts on requests from a page, and a +/// launch that gets handed to another host never has one. The one piece of +/// update work that touches shared state unprompted is +/// `sweep_update_leftovers`, which is separate for exactly that reason. pub fn[G] spawn_host_pumps( group : @async.TaskGroup[G], state : HostState, @@ -104,6 +110,20 @@ pub fn[G] spawn_host_pumps( group.spawn_bg(allow_failure=true, () => state.updates.run(emit)) } +///| +/// Clear what a previous update left in the runtime directory. Call this only +/// from the host that owns that directory, and only once it does: the sweep +/// discards a relaunch marker, which belongs to the owner and may be the one +/// it is about to act on. +pub async fn HostState::sweep_update_leftovers(self : HostState) -> Unit { + self.updates.cleanup() catch { + error if @async.is_being_cancelled() => raise error + error => + @xlog.warn(category="update") Self pub async fn HostConnection::run(Self, async (@protocol.Notification) -> Unit) -> Unit type HostState +pub async fn HostState::sweep_update_leftovers(Self) -> Unit // Type aliases diff --git a/desktop/internal/host/update_check.mbt b/desktop/internal/host/update_check.mbt index 8b6f612f4..e0d5dd496 100644 --- a/desktop/internal/host/update_check.mbt +++ b/desktop/internal/host/update_check.mbt @@ -53,7 +53,17 @@ fn SelfUpdate::new( ///| /// Sweep debris a previous update may have left behind (a staging dir, the -/// old bundle parked aside) — the host's only unprompted update work. +/// old bundle parked aside, a relaunch marker nobody acted on) — the host's +/// only unprompted update work. +/// +/// The relaunch marker is why this is not pump work. The marker is shared +/// state belonging to whichever host owns the runtime directory, and it lives +/// exactly across the gap between that host applying an update and its run +/// loop returning. A launch that is about to be handed to that host runs its +/// own pumps first and would sweep the marker out from under it — the bundle +/// swapped on disk, and nothing left to say it should reopen. So the sweep +/// belongs to the host that owns the directory, and runs only once that is +/// settled. async fn SelfUpdate::cleanup(self : SelfUpdate) -> Unit { if self.work_dir is Some(dir) && self.bundle_path is Some(bundle) { @update.cleanup_leftovers(work_dir=dir, bundle_path=bundle) @@ -211,10 +221,11 @@ async fn SelfUpdate::download_package( } ///| -/// The app-lifetime pump: sweep leftovers once, then perform accepted -/// downloads one at a time and report each outcome as a notification. Only -/// this task downloads and installs staged state, so `busy` stays true from -/// acceptance until the matching completion notification. +/// The app-lifetime pump: perform accepted downloads one at a time and report +/// each outcome as a notification. Only this task downloads and installs +/// staged state, so `busy` stays true from acceptance until the matching +/// completion notification. The leftover sweep is `SelfUpdate::cleanup`, +/// which the owning host runs separately. async fn SelfUpdate::run( self : SelfUpdate, emit : async (@protocol.Notification) -> Unit, @@ -222,12 +233,6 @@ async fn SelfUpdate::run( // A closed queue turns later `update.download` requests into rejections // instead of accepted work nobody will perform. defer self.downloads.close() - self.cleanup() catch { - error if @async.is_being_cancelled() => raise error - error => - @xlog.warn(category="update") { + group.spawn_bg(allow_failure=true, () => updates.run(fn(_) { })) + // Well past the point where a sweep at the pump's start would have run. + @async.sleep(50) + group.return_immediately(()) + }) + assert_true(@update.take_relaunch_target(work_dir~) is Some(_)) + @update.write_relaunch_marker(work_dir~, bundle_path~) + updates.cleanup() + assert_true(@update.take_relaunch_target(work_dir~) is None) + @fs.rmdir(work_dir, recursive=true) +} + ///| async test "self-update apply refuses while a download is active" { let updates : SelfUpdate = { diff --git a/desktop/internal/instance/instance.mbt b/desktop/internal/instance/instance.mbt new file mode 100644 index 000000000..850023171 --- /dev/null +++ b/desktop/internal/instance/instance.mbt @@ -0,0 +1,313 @@ +// One host per runtime directory, and the liveness signal that decides when +// that rule may be enforced. +// +// `App::single_instance` hands a later launch to whichever process holds the +// identity's lock. That process is the right destination only while it is +// still running its event loop. A host stuck in CEF's shutdown is not: it +// keeps its lock, and its instance listener runs on its own thread, so the +// launch is accepted, enqueued, acknowledged — and then never acted on, +// because the thread that would act on it is the stuck one. The user sees +// the app simply stop opening, with no error anywhere for us to react to. +// +// So the rule is enforced conditionally. A host stamps a heartbeat from its +// event loop for as long as it owns the runtime directory; a launch that +// finds a *live* process whose stamp has stopped advancing declines to hand +// over and starts on its own instead. That is the pre-`single_instance` +// behaviour, which is degraded but usable — and the shared state it lands in +// is the state `@fsx.FileLock` already serializes. + +///| +/// The file a host stamps while it owns the runtime directory. Its sidecar +/// lock is what "owning" means: the holder is the host, and only the holder +/// stamps, so a stamp never describes a process other than the lock's owner. +fn heartbeat_file(runtime_dir : @pathx.Path) -> @pathx.Path { + runtime_dir.join("host-heartbeat") +} + +///| +/// How often the owning host restamps. Short enough that a launch never +/// waits on it, long enough to stay invisible next to everything else the +/// event loop does. +const HeartbeatIntervalMs = 5_000 + +///| +/// How far behind a stamp may fall before its host looks like it has stopped. +/// Well past the interval, so ordinary lateness never reaches it. +const HeartbeatStaleMs : UInt64 = 60_000 + +///| +/// How long a stamp that already looks abandoned is watched before its host +/// is declared stopped. Longer than one interval, so a host that is merely +/// late — the machine was suspended, and its heartbeat task has not run since +/// the wake — is guaranteed the chance to produce one stamp. +const HeartbeatConfirmMs : Int = HeartbeatIntervalMs * 2 + +///| +/// The single-instance identity for `runtime_dir`. +/// +/// Keyed by the runtime directory rather than the bundle, because the +/// directory is what one host owns: two builds sharing it must not both run, +/// and two checkouts with separate `dev-state` directories share nothing and +/// must not exclude each other. It also keeps a development host from +/// silently swallowing the launch of the installed app. +/// +/// The path is canonicalized first, because two spellings of one directory +/// are still one host. `/tmp` and `/private/tmp`, or a checkout reached once +/// through a symlinked parent, would otherwise win separate elections and +/// both run — and the one that then lost the claim would carry on with no +/// heartbeat at all, invisible to `can_hand_off_to`. The workspace registry +/// collapses aliases for the same reason. +pub async fn identity(runtime_dir : @pathx.Path) -> String { + let canonical = @fs.realpath(runtime_dir.to_string()) catch { + error if @async.is_being_cancelled() => raise error + // A directory that cannot be resolved keeps its lexical spelling: every + // launch that cannot resolve it agrees on that same fallback. + _ => runtime_dir.to_string() + } + "community.moonbit.proton.openseek-desktop:" + canonical +} + +///| +/// This host's claim on `runtime_dir`, held for as long as the process runs. +/// Dropping it is the kernel's job: a host that dies — cleanly, by signal, or +/// stuck and then killed — releases it without leaving anything to clean up, +/// which is why the claim is a lock and not a recorded pid. +priv struct Claim { + lock : @fsx.FileLock + runtime_dir : @pathx.Path +} + +///| +/// Whether this launch may be given to whoever owns `runtime_dir` — the +/// question `single_instance` asks, answered before enabling it. +/// +/// True in two different situations, because both want the rule enforced: +/// nobody owns the directory, so this process becomes the owner; or the owner +/// is running its event loop and will act on what it is handed. False only +/// for the state in between — an owner that is alive but has stopped +/// stamping, where a hand-off is accepted by its listener thread and then +/// never acted on, and the launch simply disappears. +/// +/// A probe, deliberately: it takes the claim only to see whether it can, and +/// lets go again. Keeping it would pair the claim with a process that has not +/// yet won the single-instance election and may be the one that forwards and +/// exits, leaving the surviving primary with no heartbeat and every later +/// launch mistaking it for healthy. +/// +/// Only a live owner's stamp is ever read. A stamp left behind by a host that +/// exited holds no lock, so it is never consulted at all. +pub async fn can_hand_off_to(runtime_dir : @pathx.Path) -> Bool { + match @fsx.FileLock::try_acquire(heartbeat_file(runtime_dir), Exclusive) { + Some(lock) => { + lock.release() + true + } + None => { + let stamped = read_stamp(runtime_dir) + guard stamp_is_behind(stamped) else { return true } + // A machine resuming from suspend leaves exactly this evidence: the + // owner's heartbeat task has not been scheduled since the wake, so its + // stamp is as old as the sleep was long, and its loop is perfectly + // healthy. One stamp settles it. The wait falls only on this already + // suspicious path, and a launch that takes a few extra seconds beats + // one that starts a rival host against a working one. + @async.sleep(HeartbeatConfirmMs) + read_stamp(runtime_dir) != stamped + } + } +} + +///| +/// Own `runtime_dir` and stamp for it for as long as this host runs, taking +/// the claim as soon as it is free. +/// +/// Call this from the process that won the single-instance election and +/// nowhere else. That is what keeps the claim and the primary the same +/// process, so the heartbeat always describes the host a later launch would +/// actually be handed to. +/// +/// It waits rather than gives up, because the primary can legitimately find +/// the claim taken: a host that started on the stopped-owner path holds no +/// single-instance identity, so its claim outlives the election it never +/// entered. Giving up there would leave this — the process every later launch +/// is handed to — with no heartbeat at all, and a launch reading the other +/// host's fresh stamp would go on handing itself to this one however stuck it +/// became. Waiting costs nothing and ends the moment that host exits. +pub async fn hold( + runtime_dir : @pathx.Path, + retry_interval_ms? : Int = HeartbeatIntervalMs, +) -> Unit { + for ;; { + if claim(runtime_dir) is Some(held) { + return held.run() + } + @async.sleep(retry_interval_ms) + } +} + +///| +/// Take ownership of `runtime_dir` for this host, or `None` if another +/// process holds it. +async fn claim(runtime_dir : @pathx.Path) -> Claim? { + guard @fsx.FileLock::try_acquire(heartbeat_file(runtime_dir), Exclusive) + is Some(lock) else { + return None + } + // Stamp before anyone can read it, so a launch racing this one never finds + // a fresh lock beside the previous host's abandoned stamp. + stamp(runtime_dir) + Some({ lock, runtime_dir }) +} + +///| +/// Restamp for as long as the host runs. Spawned on the app's task group, so +/// the stamp advances exactly while the loop that would act on a hand-off is +/// running — which is the whole signal. +async fn Claim::run(self : Claim) -> Unit { + for ;; { + @async.sleep(HeartbeatIntervalMs) + stamp(self.runtime_dir) + } +} + +///| +/// Record the current wall clock. Written atomically: a launch reading this +/// file concurrently must see one complete stamp or the previous one, never a +/// half-written number that would parse as ancient and read as a dead host. +async fn stamp(runtime_dir : @pathx.Path) -> Unit { + @fsx.write_text_atomic( + heartbeat_file(runtime_dir), + @env.now().to_string(), + permission=0o600, + ) catch { + error if @async.is_being_cancelled() => raise error + // A stamp that cannot be written leaves the previous one to go stale, so + // the failure mode is "a later launch starts its own instance" — the + // degraded-but-working side, which is where an unwritable runtime + // directory belongs anyway. + _ => () + } +} + +///| +/// The owning host's last stamp, or `None` when there is nothing readable +/// where one belongs. +async fn read_stamp(runtime_dir : @pathx.Path) -> UInt64? { + let text = (@fsx.try_read_text(heartbeat_file(runtime_dir)) catch { + error if @async.is_being_cancelled() => raise error + _ => None + }).unwrap_or("") + Some(@string.parse_uint64(text.trim())) catch { + _ => None + } +} + +///| +/// Whether a stamp is far enough behind the clock to look abandoned. Absent +/// or unparsable counts as behind: something holds the lock while the file +/// that should describe it says nothing. +fn stamp_is_behind(stamped : UInt64?) -> Bool { + guard stamped is Some(stamped) else { return true } + let now = @env.now() + // A stamp from the future is a clock that moved backwards, not a dead host. + guard now > stamped else { return false } + now - stamped > HeartbeatStaleMs +} + +///| +async test "the identity separates runtime directories and repeats for one" { + let installed : @pathx.Path = "/Users/u/Library/Application Support/SeekMoon" + let checkout : @pathx.Path = "/src/openseek/desktop/target/dev-state" + assert_eq(identity(installed), identity(installed)) + assert_not_eq(identity(installed), identity(checkout)) +} + +///| +/// The claim is the liveness fact: a launch that can take the lock knows no +/// host is running, whatever any leftover stamp says. +async test "an unclaimed runtime directory is claimable" { + let root : @pathx.Path = @fs.tmpdir(prefix="openseek-instance-free-") + guard claim(root) is Some(held) else { fail("expected a free claim") } + held.lock.release() + @fs.rmdir(root.to_string(), recursive=true) +} + +///| +/// A host that exited leaves its stamp behind. Nothing holds the lock, so +/// that stamp must never be read — the next launch is simply the new owner. +async test "a stamp left by an exited host is not read at all" { + let root : @pathx.Path = @fs.tmpdir(prefix="openseek-instance-stale-") + @fsx.write_text_atomic(heartbeat_file(root), "1") + assert_true(can_hand_off_to(root)) + guard claim(root) is Some(held) else { + fail("an abandoned stamp must not block a claim") + } + held.lock.release() + @fs.rmdir(root.to_string(), recursive=true) +} + +///| +/// The owner is alive and stamping: hand the launch over. +async test "a live owner with a fresh stamp takes the launch" { + let root : @pathx.Path = @fs.tmpdir(prefix="openseek-instance-live-") + guard claim(root) is Some(owner) else { fail("expected the first claim") } + assert_true(can_hand_off_to(root)) + assert_true(claim(root) is None) + owner.lock.release() + @fs.rmdir(root.to_string(), recursive=true) +} + +///| +/// The owner is alive and has stopped stamping: keep the launch for ourselves +/// rather than feed it to a process that will not act on it. +async test "a live owner with a stopped stamp cannot be handed to" { + let root : @pathx.Path = @fs.tmpdir(prefix="openseek-instance-stopped-") + guard claim(root) is Some(owner) else { fail("expected the first claim") } + // The owner's loop stopped long ago; only the process is still there. + @fsx.write_text_atomic(heartbeat_file(root), "1") + assert_false(can_hand_off_to(root)) + owner.lock.release() + @fs.rmdir(root.to_string(), recursive=true) +} + +///| +/// The primary can find the claim already taken — a host that started on the +/// stopped-owner path holds one without ever entering an election. Giving up +/// there would leave the process every later launch is handed to with no +/// heartbeat at all, so it waits the other host out instead. +async test "a primary that loses the claim takes it when the holder leaves" { + let root : @pathx.Path = @fs.tmpdir(prefix="openseek-instance-wait-") + guard claim(root) is Some(other) else { fail("expected the first claim") } + @async.with_task_group(group => { + group.spawn_bg(allow_failure=true, () => hold(root, retry_interval_ms=10)) + // Many retries' worth of waiting changes nothing while the other host is + // there — the claim stays that host's. + @async.sleep(50) + assert_true(claim(root) is None) + other.lock.release() + // Now nobody else holds it, so a claim that still cannot be taken can + // only be the waiting host's — and it stamps, which is what makes the + // heartbeat describe the process a launch would be handed to. + @async.sleep(50) + assert_true(claim(root) is None) + assert_true(can_hand_off_to(root)) + group.return_immediately(()) + }) + @fs.rmdir(root.to_string(), recursive=true) +} + +///| +/// The probe must not keep what it takes. Two launches racing here both find +/// the directory free and both go on to the single-instance election; if the +/// probe held on, the loser would carry the claim out of the process and the +/// winning primary would run with no heartbeat at all. +async test "the probe leaves the claim free for the election winner" { + let root : @pathx.Path = @fs.tmpdir(prefix="openseek-instance-probe-") + assert_true(can_hand_off_to(root)) + assert_true(can_hand_off_to(root)) + guard claim(root) is Some(held) else { + fail("probing must not consume the claim") + } + held.lock.release() + @fs.rmdir(root.to_string(), recursive=true) +} diff --git a/desktop/internal/instance/moon.pkg b/desktop/internal/instance/moon.pkg new file mode 100644 index 000000000..8340d82d3 --- /dev/null +++ b/desktop/internal/instance/moon.pkg @@ -0,0 +1,12 @@ +import { + "moonbitlang/async", + "moonbitlang/async/fs", + "moonbitlang/core/env", + "moonbitlang/core/string", + "openseek_desktop/internal/fsx", + "openseek_desktop/internal/pathx", +} + +warnings = "+missing_doc+test_unqualified_package+unnecessary_annotation+unnecessary_view_op+ambiguous_range_direction+unqualified_local_using" + +supported_targets = "+native" diff --git a/desktop/internal/instance/pkg.generated.mbti b/desktop/internal/instance/pkg.generated.mbti new file mode 100644 index 000000000..6c55e5058 --- /dev/null +++ b/desktop/internal/instance/pkg.generated.mbti @@ -0,0 +1,21 @@ +// Generated using `moon info`, DON'T EDIT IT +package "openseek_desktop/internal/instance" + +import { + "openseek_desktop/internal/pathx", +} + +// Values +pub async fn can_hand_off_to(@pathx.Path) -> Bool + +pub async fn hold(@pathx.Path, retry_interval_ms? : Int) -> Unit + +pub async fn identity(@pathx.Path) -> String + +// Errors + +// Types and methods + +// Type aliases + +// Traits diff --git a/desktop/main.mbt b/desktop/main.mbt index c479a9839..c0252ce32 100644 --- a/desktop/main.mbt +++ b/desktop/main.mbt @@ -37,6 +37,25 @@ fn main { log.info() + if @instance.can_hand_off_to(dir) { + Some(@instance.identity(dir)) + } else { + log.warn() None + } // The relay sign-in store: a persisted session from auth.json, unless the // environment override pins the connector config directly (development). let auth = @auth.AuthStore::load( @@ -58,6 +77,11 @@ fn main { let relay = @remote.RelayActor::new(state, announce=device => { log.info() () } }) + // Everything that acts on state belonging to whoever owns the runtime + // directory starts here and nowhere earlier: this hook runs only in the + // process that won the single-instance election. A launch that is + // handed to another host returns from `app.run()` without ever reaching + // it, so it can neither stamp a heartbeat the other host's would be + // mistaken for, nor sweep away the relaunch marker that host is about + // to act on. Both are spawned on the application task group, so they + // stop exactly when the loop they belong to does. + .app_lifecycle( + on_start=async fn(context) { + is_primary.val = true + let tasks = context.task_group() + match runtime_dir { + Some(dir) => + tasks.spawn_bg(allow_failure=true, () => @instance.hold(dir)) + None => () + } + tasks.spawn_bg(allow_failure=true, () => { + state.host_state().sweep_update_leftovers() + }) + }, + // The claim needs no shutdown of its own: the kernel releases it when + // the process dies, however it dies. + on_shutdown=fn(_) { }, + ) + let app = match instance_identity { + Some(identity) => app.single_instance(identity) + None => app + } @async.with_task_group(group => { // The bridge actor, engine actor, and host actors run for the app's // lifetime — they feed every page/client rather than borrowing a @@ -163,18 +216,31 @@ fn main { // relaunch marker: reopen the new version now that the run loop is done. // `open -n` hands the launch to LaunchServices, so the new instance is // not our child and this process can exit right after. - if runtime_dir is Some(dir) && - @update.take_relaunch_target(work_dir=dir.to_string()) is Some(target) { - log.info() { - log.error() { + log.error()