mirror of
https://github.com/Niko1221/Strata.git
synced 2026-11-07 06:15:58 -10:00
Windows file in use: every state-file swap goes through atomic.replace (retry); a sharing violation no longer says run as yourself
This commit is contained in:
1 parent
4509f2f997
commit
d0afabb6ab
17 files changed
+114
-24
No files matched your search
+2
-1
@@ -15,6 +15,7 @@ import shutil
|
||||
from pathlib import Path
|
||||
|
||||
from strata_control import user_args
|
||||
from strata_control import atomic as _atomic_mod
|
||||
|
||||
# (key, kind, help). kind: "bool", "int>=0", "num>=0", ("enum", values), "names", or ("sampling", check) for a key
|
||||
# of the "sampling" block, ("arg", flag) for an engine option kept in "args".
|
||||
@@ -179,7 +180,7 @@ def save(path: str | Path, cfg: dict) -> Path:
|
||||
shutil.copyfile(path, bak)
|
||||
tmp = path.with_name(path.name + ".tmp")
|
||||
tmp.write_text(json.dumps(cfg, indent=1), encoding="utf-8")
|
||||
os.replace(tmp, path)
|
||||
_atomic_mod.replace(tmp, path)
|
||||
return bak
|
||||
|
||||
|
||||
|
||||
+2
-1
@@ -33,6 +33,7 @@ from pathlib import Path
|
||||
|
||||
from strata_cli.client import ClientError, Endpoint, _NoRedirect
|
||||
from strata_control.profiles import ProfileError
|
||||
from strata_control import atomic as _atomic_mod
|
||||
|
||||
|
||||
SCHEMA = 1
|
||||
@@ -685,7 +686,7 @@ def _progress_write(path: Path, record: dict, *, first: bool = False) -> None:
|
||||
temporary = path.with_name(".progress-" + uuid.uuid4().hex + ".tmp")
|
||||
try:
|
||||
_write_new(temporary, encoded)
|
||||
os.replace(temporary, path)
|
||||
_atomic_mod.replace(temporary, path)
|
||||
finally:
|
||||
temporary.unlink(missing_ok=True)
|
||||
|
||||
|
||||
+7
-2
@@ -1982,10 +1982,15 @@ def _access_retry(a: argparse.Namespace, error: BaseException) -> int | None:
|
||||
if not (isinstance(error, PermissionError) or "PermissionError" in str(error)):
|
||||
return None
|
||||
as_json = bool(getattr(a, "json", False) or getattr(a, "output", None) in ("json", "jsonl"))
|
||||
in_use = private_files.is_sharing_violation(error)
|
||||
if getattr(a, "_access_retried", False):
|
||||
return _error(f"{private_files.ACCESS_MESSAGE} (details: {error})", 5, as_json)
|
||||
message = private_files.IN_USE_MESSAGE if in_use else private_files.ACCESS_MESSAGE
|
||||
return _error(f"{message} (details: {error})", 5, as_json)
|
||||
a._access_retried = True
|
||||
private_files.repair_known_roots()
|
||||
if in_use:
|
||||
time.sleep(1.5) # a scanner or a sync tool lets go of the file in a moment; repairing would not help
|
||||
else:
|
||||
private_files.repair_known_roots()
|
||||
return _run(a)
|
||||
|
||||
|
||||
|
||||
@@ -6,12 +6,18 @@ CLI resolve the same saved connection without importing server or installer code
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
try:
|
||||
from strata_control.private_files import private_mkdir as _private_mkdir
|
||||
except ImportError: # a thin bundle (monitor only) ships without strata_control
|
||||
def _private_mkdir(path, *, parents=False, exist_ok=True):
|
||||
path.mkdir(mode=0o700, parents=parents, exist_ok=exist_ok)
|
||||
return path
|
||||
try:
|
||||
from strata_control import atomic as _atomic_mod
|
||||
except ImportError: # the same thin bundle: a plain replace
|
||||
class _atomic_mod: # noqa: N801
|
||||
replace = staticmethod(os.replace)
|
||||
import hashlib
|
||||
import json
|
||||
import os
|
||||
@@ -158,7 +164,7 @@ def _replace(path: Path, data: bytes) -> None:
|
||||
stream.write(data)
|
||||
stream.flush()
|
||||
os.fsync(stream.fileno())
|
||||
os.replace(name, path)
|
||||
_atomic_mod.replace(name, path)
|
||||
if os.name == "posix":
|
||||
try:
|
||||
parent = os.open(path.parent, os.O_RDONLY)
|
||||
|
||||
@@ -21,6 +21,7 @@ import sys
|
||||
import tempfile
|
||||
import threading
|
||||
import time
|
||||
from strata_control import atomic as _atomic_mod
|
||||
from dataclasses import dataclass, field
|
||||
from pathlib import Path
|
||||
from typing import Any, Callable
|
||||
@@ -277,7 +278,7 @@ def write_private(path: Path, data: dict) -> None:
|
||||
os.chmod(tmp, 0o600)
|
||||
except OSError:
|
||||
pass
|
||||
os.replace(tmp, path)
|
||||
_atomic_mod.replace(tmp, path)
|
||||
finally:
|
||||
if os.path.exists(tmp):
|
||||
os.unlink(tmp)
|
||||
|
||||
@@ -13,6 +13,7 @@ import json
|
||||
import os
|
||||
import tempfile
|
||||
import time
|
||||
from strata_control import atomic as _atomic_mod
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
@@ -47,7 +48,7 @@ def _private_write(path: Path, payload: bytes) -> None:
|
||||
os.chmod(tmp, 0o600)
|
||||
except OSError:
|
||||
pass
|
||||
os.replace(tmp, path)
|
||||
_atomic_mod.replace(tmp, path)
|
||||
finally:
|
||||
if os.path.exists(tmp):
|
||||
os.unlink(tmp)
|
||||
|
||||
@@ -10,6 +10,7 @@ import sys
|
||||
import tempfile
|
||||
import threading
|
||||
import time
|
||||
from strata_control import atomic as _atomic_mod
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
@@ -151,7 +152,7 @@ def _write_status(path: str | None, payload: dict[str, Any]) -> None:
|
||||
try:
|
||||
tmp = path + ".tmp"
|
||||
Path(tmp).write_text(json.dumps(payload, ensure_ascii=False, default=str), encoding="utf-8")
|
||||
os.replace(tmp, path)
|
||||
_atomic_mod.replace(tmp, path)
|
||||
except OSError:
|
||||
pass
|
||||
|
||||
|
||||
@@ -26,6 +26,7 @@ from pathlib import Path
|
||||
from strata_control import profiles
|
||||
from strata_control.doctor import inspect as inspect_components
|
||||
from strata_control.profiles import ProfileError
|
||||
from strata_control import atomic as _atomic_mod
|
||||
|
||||
|
||||
SUITE = "legacy-decode-v1"
|
||||
@@ -412,7 +413,7 @@ def _replace_bytes(path: Path, payload: bytes) -> None:
|
||||
stream.write(payload)
|
||||
stream.flush()
|
||||
os.fsync(stream.fileno())
|
||||
os.replace(temp, path)
|
||||
_atomic_mod.replace(temp, path)
|
||||
finally:
|
||||
if os.path.exists(temp):
|
||||
os.unlink(temp)
|
||||
|
||||
@@ -52,6 +52,7 @@ from typing import Any, Iterable, Sequence
|
||||
SCHEMA_VERSION = 1
|
||||
|
||||
from strata_control.targets import BACKEND_MATRIX, ComponentError, backend_supported, host_platform # noqa: E402,F401
|
||||
from strata_control import atomic as _atomic_mod
|
||||
#: Component kinds, so a manifest cannot claim a role the architecture does not have.
|
||||
KINDS = ("application", "runtime", "agent-runtime", "tuning", "engine", "prepared-data")
|
||||
|
||||
@@ -426,7 +427,7 @@ class ComponentStore:
|
||||
def _save_state(self, data: dict[str, Any]) -> None:
|
||||
temp = self.state_path.parent / (self.state_path.name + ".tmp")
|
||||
temp.write_text(json.dumps(data, indent=1, sort_keys=True) + "\n", encoding="utf-8")
|
||||
os.replace(temp, self.state_path)
|
||||
_atomic_mod.replace(temp, self.state_path)
|
||||
|
||||
# ------------------------------------------------------------------ cache and installation
|
||||
|
||||
@@ -487,7 +488,7 @@ class ComponentStore:
|
||||
indent=1) + "\n", encoding="utf-8")
|
||||
if destination.exists():
|
||||
shutil.rmtree(destination, ignore_errors=True)
|
||||
os.replace(staging, destination)
|
||||
_atomic_mod.replace(staging, destination)
|
||||
except BaseException:
|
||||
# Leave nothing that looks complete.
|
||||
shutil.rmtree(staging, ignore_errors=True)
|
||||
@@ -590,7 +591,7 @@ class ComponentStore:
|
||||
if active.exists():
|
||||
if previous.exists():
|
||||
shutil.rmtree(previous, ignore_errors=True)
|
||||
os.replace(active, previous)
|
||||
_atomic_mod.replace(active, previous)
|
||||
moved_aside = str(previous)
|
||||
|
||||
# Record what is being displaced BEFORE the activation, so a crash between the two leaves a
|
||||
@@ -611,7 +612,7 @@ class ComponentStore:
|
||||
if incoming.exists():
|
||||
shutil.rmtree(incoming, ignore_errors=True)
|
||||
shutil.copytree(source, incoming, symlinks=False)
|
||||
os.replace(incoming, active)
|
||||
_atomic_mod.replace(incoming, active)
|
||||
|
||||
state = self.state()
|
||||
entry = state["components"].setdefault(component.id, {})
|
||||
@@ -636,8 +637,8 @@ class ComponentStore:
|
||||
abandoned = self.root / "previous" / f"{component_id}.replaced"
|
||||
if abandoned.exists():
|
||||
shutil.rmtree(abandoned, ignore_errors=True)
|
||||
os.replace(active, abandoned)
|
||||
os.replace(previous, active)
|
||||
_atomic_mod.replace(active, abandoned)
|
||||
_atomic_mod.replace(previous, active)
|
||||
state = self.state()
|
||||
entry = state["components"].get(component_id, {})
|
||||
restored = entry.get("rollback", {})
|
||||
|
||||
@@ -26,6 +26,7 @@ import uuid
|
||||
from pathlib import Path
|
||||
|
||||
from strata_control import paths
|
||||
from strata_control import atomic as _atomic_mod
|
||||
|
||||
SCHEMA_VERSION = 1
|
||||
IDENTITY_KEYS = ("version", "build_id", "manifest_sha256", "backend", "platform", "variant")
|
||||
@@ -58,7 +59,7 @@ def _write(path: Path, data: dict) -> None:
|
||||
with os.fdopen(descriptor, "w", encoding="utf-8") as stream:
|
||||
json.dump(data, stream, indent=1, sort_keys=True)
|
||||
stream.write("\n")
|
||||
os.replace(temp, path)
|
||||
_atomic_mod.replace(temp, path)
|
||||
finally:
|
||||
if os.path.exists(temp):
|
||||
os.unlink(temp)
|
||||
|
||||
@@ -29,6 +29,7 @@ from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
from strata_control import tune_knobs, tune_machine
|
||||
from strata_control import atomic as _atomic_mod
|
||||
|
||||
SCHEMA = "strata.preset/1"
|
||||
RECHECK_SET = "recheck/1"
|
||||
@@ -57,7 +58,7 @@ def _atomic(path: Path, payload: bytes) -> None:
|
||||
fh.write(payload)
|
||||
fh.flush()
|
||||
os.fsync(fh.fileno())
|
||||
os.replace(tmp, path)
|
||||
_atomic_mod.replace(tmp, path)
|
||||
finally:
|
||||
if os.path.exists(tmp):
|
||||
os.unlink(tmp)
|
||||
|
||||
@@ -383,7 +383,19 @@ def read_bytes_repairing(path: str | os.PathLike) -> bytes:
|
||||
return target.read_bytes()
|
||||
|
||||
|
||||
ACCESS_MESSAGE = "Strata can't open its settings file. Run Strata once as yourself (not as administrator) or press Repair."
|
||||
ACCESS_MESSAGE = ("Strata is not allowed to open one of its own files. Run `strata repair-access` (or press Repair in the app); "
|
||||
"these files were probably made while Strata ran as administrator, so run it as your normal user from now on.")
|
||||
IN_USE_MESSAGE = ("A file Strata needs is in use by another program (usually antivirus, a backup or a sync tool). "
|
||||
"Wait a moment and run the same command again.")
|
||||
|
||||
|
||||
def is_sharing_violation(error: BaseException) -> bool:
|
||||
"""Windows "file in use" (WinError 32 or 33), as opposed to a real refusal (WinError 5): waiting helps, repairing does not."""
|
||||
code = getattr(error, "winerror", None)
|
||||
if code in (32, 33):
|
||||
return True
|
||||
text = str(error)
|
||||
return "WinError 32" in text or "WinError 33" in text or "being used by another process" in text
|
||||
|
||||
|
||||
_SKIP_DIRS = {"models", "node_modules", "harness", "python", "host", "node", "control", "__pycache__", ".git"}
|
||||
|
||||
@@ -30,6 +30,7 @@ from dataclasses import dataclass
|
||||
from pathlib import Path
|
||||
|
||||
from strata_control import paths
|
||||
from strata_control import atomic as _atomic_mod
|
||||
|
||||
WINDOWS = os.name == "nt"
|
||||
SCHEMA = 1
|
||||
@@ -192,7 +193,7 @@ def write_record(pid: int, role: str, *, profile: str | None = None, cmd: list[s
|
||||
path = procs_dir() / f"{pid}.json"
|
||||
tmp = path.with_suffix(f".{os.getpid()}.tmp")
|
||||
tmp.write_text(json.dumps(record), encoding="utf-8")
|
||||
os.replace(tmp, path)
|
||||
_atomic_mod.replace(tmp, path)
|
||||
return token
|
||||
except (OSError, ValueError):
|
||||
return None
|
||||
|
||||
@@ -17,6 +17,7 @@ import shlex
|
||||
import stat
|
||||
import subprocess
|
||||
import tempfile
|
||||
from strata_control import atomic as _atomic_mod
|
||||
from contextlib import contextmanager
|
||||
from pathlib import Path
|
||||
|
||||
@@ -216,7 +217,7 @@ def _atomic(path: Path, data: bytes) -> None:
|
||||
stream.write(data)
|
||||
stream.flush()
|
||||
os.fsync(stream.fileno())
|
||||
os.replace(temp_name, path)
|
||||
_atomic_mod.replace(temp_name, path)
|
||||
if os.name != "nt":
|
||||
try:
|
||||
parent_fd = os.open(path.parent, os.O_RDONLY)
|
||||
|
||||
@@ -42,6 +42,7 @@ from pathlib import Path
|
||||
from typing import Any, Callable, Sequence
|
||||
|
||||
from strata_control import user_args
|
||||
from strata_control import atomic as _atomic_mod
|
||||
|
||||
SCHEMA_VERSION = 1
|
||||
|
||||
@@ -66,7 +67,7 @@ def _atomic_write(path: Path, payload: bytes) -> None:
|
||||
stream.write(payload)
|
||||
stream.flush()
|
||||
os.fsync(stream.fileno())
|
||||
os.replace(temp, path)
|
||||
_atomic_mod.replace(temp, path)
|
||||
finally:
|
||||
if os.path.exists(temp):
|
||||
os.unlink(temp)
|
||||
|
||||
@@ -44,6 +44,7 @@ from typing import Callable
|
||||
|
||||
from strata_control import acquisition, components, engine_component as ec, feed as feedmod, longpath, preset_marks, signing, transfer
|
||||
from strata_control.targets import host_platform
|
||||
from strata_control import atomic as _atomic_mod
|
||||
|
||||
STATE_DIR = ".updates"
|
||||
STAGE_DIR = ".s" # staging is short on purpose: <components>\.s\<4 hex>\engine\<folder> (see longpath.py)
|
||||
@@ -152,7 +153,7 @@ class State:
|
||||
with os.fdopen(descriptor, "w", encoding="utf-8") as stream:
|
||||
json.dump(self.data, stream, indent=1, sort_keys=True)
|
||||
stream.write("\n")
|
||||
os.replace(temp, self.path)
|
||||
_atomic_mod.replace(temp, self.path)
|
||||
finally:
|
||||
if os.path.exists(temp):
|
||||
os.unlink(temp)
|
||||
@@ -948,7 +949,7 @@ def update_control(ctx: Context, feed: feedmod.Feed, state: State) -> dict:
|
||||
pointer = control_dir(root) / "active.json"
|
||||
temp = pointer.with_suffix(".tmp")
|
||||
temp.write_text(json.dumps({"version": pkg.version, "folder": pkg.version, "sha256": pkg.sha256}), encoding="utf-8")
|
||||
os.replace(temp, pointer)
|
||||
_atomic_mod.replace(temp, pointer)
|
||||
keep = {pkg.version} | ({previous["folder"]} if previous else set())
|
||||
for entry_dir in control_dir(root).iterdir():
|
||||
if entry_dir.is_dir() and entry_dir.name not in keep:
|
||||
@@ -1022,7 +1023,7 @@ def update_app(ctx: Context, feed: feedmod.Feed) -> dict:
|
||||
target_dir.mkdir(parents=True, exist_ok=True)
|
||||
name = Path(transfer.resolve_ref(transfer.base_of(feed.source), entry["url"]).replace("\\", "/")).name or f"strata-app-{feed.app.version}"
|
||||
target = target_dir / name
|
||||
os.replace(archive, target)
|
||||
_atomic_mod.replace(archive, target)
|
||||
result.update({"changed": True, "status": "downloaded", "installed_after": None, "version": feed.app.version, "path": str(target),
|
||||
"note": "Verified. Run the installer to finish; models and settings are kept."})
|
||||
return result
|
||||
|
||||
@@ -0,0 +1,54 @@
|
||||
"""Windows 'file in use': every swap of a state file waits and retries (atomic.replace), and the message does not blame permissions."""
|
||||
|
||||
import re
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
from unittest import mock
|
||||
|
||||
from strata_control import atomic, private_files
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[2]
|
||||
SITES = ["strata_control/calibration.py", "strata_control/components.py", "strata_control/presets.py",
|
||||
"strata_control/preset_marks.py", "strata_control/procs.py", "strata_control/profiles.py",
|
||||
"strata_control/tune_apply.py", "strata_control/updater.py", "strata_cli/bench.py", "strata_cli/connections.py",
|
||||
"strata_cli/launch/runner.py", "strata_cli/launch/store.py", "strata_cli/tune_commands.py", "serve/runconfig.py"]
|
||||
|
||||
|
||||
class Sites(unittest.TestCase):
|
||||
def test_no_bare_os_replace_is_left_in_the_state_file_writers(self):
|
||||
for rel in SITES:
|
||||
text = (ROOT / rel).read_text(encoding="utf-8")
|
||||
bare = [l for l in text.splitlines() if re.search(r"(?<![_\w.])os\.replace\(", l) or re.search(r"[^_]os\.replace\(", l)]
|
||||
# updater._move_dir keeps its own longer retry loop
|
||||
bare = [l for l in bare if not (rel.endswith("updater.py") and "source, destination" in l)]
|
||||
self.assertEqual(bare, [], rel)
|
||||
self.assertIn("_atomic_mod.replace(", text, rel)
|
||||
|
||||
|
||||
class Retry(unittest.TestCase):
|
||||
def test_a_held_file_is_waited_for_on_windows(self):
|
||||
calls = []
|
||||
|
||||
def flaky(src, dst):
|
||||
calls.append(1)
|
||||
if len(calls) < 3:
|
||||
raise PermissionError(13, "in use")
|
||||
with mock.patch.object(atomic.os, "replace", flaky), mock.patch.object(atomic.os, "name", "nt"):
|
||||
atomic.replace("a", "b", sleep=lambda s: None)
|
||||
self.assertEqual(len(calls), 3)
|
||||
|
||||
|
||||
class Message(unittest.TestCase):
|
||||
def test_sharing_violation_is_told_from_a_refusal(self):
|
||||
in_use = PermissionError(13, "The process cannot access the file because it is being used by another process")
|
||||
in_use.winerror = 32
|
||||
denied = PermissionError(13, "Access is denied")
|
||||
denied.winerror = 5
|
||||
self.assertTrue(private_files.is_sharing_violation(in_use))
|
||||
self.assertFalse(private_files.is_sharing_violation(denied))
|
||||
self.assertNotIn("as yourself", private_files.ACCESS_MESSAGE)
|
||||
self.assertIn("another program", private_files.IN_USE_MESSAGE)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in new issue
Block a user