Keep concurrent usage collectors off one shared temp file (#6654)
Two Claude collectors running at once both wrote the cache through a temp path derived from the target, so the second replace found the file already moved away and crashed the update with a FileNotFoundError. Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
co-authored by
Claude Opus 5
parent
f8c235a5e4
commit
e4d85bd037
@@ -23,6 +23,7 @@ import os
|
||||
import re
|
||||
import sqlite3
|
||||
import sys
|
||||
import tempfile
|
||||
import time
|
||||
import urllib.error
|
||||
import urllib.request
|
||||
@@ -230,9 +231,21 @@ def read_fresh_json(path: Path, max_age_seconds: float) -> dict[str, Any] | None
|
||||
|
||||
|
||||
def write_json(path: Path, payload: dict[str, Any]) -> None:
|
||||
tmp = path.with_suffix(path.suffix + ".tmp")
|
||||
tmp.write_text(json.dumps(payload, separators=(",", ":"), sort_keys=True) + "\n", encoding="utf-8")
|
||||
tmp.replace(path)
|
||||
# A temp name unique to this writer, not derived from the target: several
|
||||
# collectors can run at once (the update command backgrounds one per agent,
|
||||
# the panel refreshes on its own), and a shared temp path means the second
|
||||
# replace finds the first one's file already moved away.
|
||||
handle_fd, tmp_name = tempfile.mkstemp(dir=path.parent, prefix=path.name + ".", suffix=".tmp")
|
||||
tmp = Path(tmp_name)
|
||||
try:
|
||||
with os.fdopen(handle_fd, "w", encoding="utf-8") as handle:
|
||||
handle.write(json.dumps(payload, separators=(",", ":"), sort_keys=True) + "\n")
|
||||
# mkstemp opens at 0600; these caches were world-readable before.
|
||||
tmp.chmod(0o644)
|
||||
tmp.replace(path)
|
||||
except BaseException:
|
||||
tmp.unlink(missing_ok=True)
|
||||
raise
|
||||
|
||||
|
||||
def cached_scan(projects_path: Path, max_age_seconds: float) -> dict[str, Any]:
|
||||
|
||||
@@ -125,3 +125,55 @@ result=$(HOME="$PI_HOME" XDG_CACHE_HOME="$PI_HOME/.cache" XDG_DATA_HOME="$PI_HOM
|
||||
[[ $(jq -c '.modelUsage' <<<"$result") == '{"claude-omp":{"cacheCreationInputTokens":1,"cacheReadInputTokens":4,"inputTokens":20,"outputTokens":5},"claude-pi":{"cacheCreationInputTokens":2,"cacheReadInputTokens":3,"inputTokens":10,"outputTokens":4}}' ]] ||
|
||||
fail "Claude collector filters pi and omp sessions to Anthropic providers" "$result"
|
||||
pass "Claude collector counts pi and omp subscription usage"
|
||||
|
||||
# Collectors overlap in practice: the update command backgrounds one per agent
|
||||
# while the panel refreshes on its own. Two writers aiming at one cache file
|
||||
# must both land, not trip over a shared temp path.
|
||||
race_output=$(python3 - "$ROOT/bin/omarchy-agent-usage-claude" "$TEST_HOME/race.json" <<'PY'
|
||||
import importlib.util
|
||||
import json
|
||||
import sys
|
||||
import threading
|
||||
from importlib.machinery import SourceFileLoader
|
||||
from pathlib import Path
|
||||
|
||||
# The collector has no .py suffix, so name its loader explicitly.
|
||||
spec = importlib.util.spec_from_loader("collector", SourceFileLoader("collector", sys.argv[1]))
|
||||
collector = importlib.util.module_from_spec(spec)
|
||||
spec.loader.exec_module(collector)
|
||||
|
||||
target = Path(sys.argv[2])
|
||||
failures = []
|
||||
start = threading.Barrier(8)
|
||||
|
||||
def hammer(writer):
|
||||
start.wait()
|
||||
for round in range(25):
|
||||
try:
|
||||
collector.write_json(target, {"writer": writer, "round": round})
|
||||
except Exception as error:
|
||||
failures.append(repr(error))
|
||||
|
||||
threads = [threading.Thread(target=hammer, args=(writer,)) for writer in range(8)]
|
||||
for thread in threads:
|
||||
thread.start()
|
||||
for thread in threads:
|
||||
thread.join()
|
||||
|
||||
leftovers = sorted(path.name for path in target.parent.glob(target.name + ".*"))
|
||||
print(json.dumps({
|
||||
"failures": failures[:3],
|
||||
"mode": oct(target.stat().st_mode & 0o777),
|
||||
"payload": json.loads(target.read_text(encoding="utf-8")),
|
||||
"leftovers": leftovers,
|
||||
}))
|
||||
PY
|
||||
)
|
||||
|
||||
[[ $(jq -c '.failures' <<<"$race_output") == "[]" ]] ||
|
||||
fail "Claude collector survives concurrent writes to one cache file" "$race_output"
|
||||
[[ $(jq -r '.payload.writer != null and (.leftovers | length) == 0' <<<"$race_output") == "true" ]] ||
|
||||
fail "Claude collector leaves one intact cache file and no temp files" "$race_output"
|
||||
[[ $(jq -r '.mode' <<<"$race_output") == "0o644" ]] ||
|
||||
fail "Claude collector keeps cache files readable" "$race_output"
|
||||
pass "Claude collector survives concurrent writes to one cache file"
|
||||
|
||||
Reference in New Issue
Block a user