fix: publish media without replacing concurrent readers

This commit is contained in:
zenfun
2026-09-15 20:54:03 +08:00
parent c09dc2324e
commit b60de9e097
4 changed files with 37 additions and 27 deletions
@@ -44,6 +44,7 @@ jobs:
- name: Run focused tests
run: |
mkdir -p "$RUNNER_TEMP/image-memory-validation/results"
uv run pytest -q \
tests/unit/test_image_preparation_budget.py \
tests/unit/test_image_media_store.py \
@@ -56,7 +57,8 @@ jobs:
tests/unit/test_image_memory_benchmark.py \
tests/unit/test_provider_request_retry.py \
tests/unit/test_manage_image_history.py \
tests/unit/test_conversation_media_api.py
tests/unit/test_conversation_media_api.py \
--junitxml="$RUNNER_TEMP/image-memory-validation/results/tests.xml"
- name: Generate reproducible fixtures outside the measured process
run: |
+9 -3
View File
@@ -95,7 +95,7 @@ class ImageMediaStore:
raise OSError("invalid durable media root")
object_path = self.root / f"{media_id}.bin"
metadata_path = self.root / f"{media_id}.json"
detected_mime = mime_type or "application/octet-stream"
detected_mime = "application/octet-stream"
width: int | None = None
height: int | None = None
try:
@@ -151,7 +151,10 @@ class ImageMediaStore:
output.write(data)
output.flush()
os.fsync(output.fileno())
os.replace(temporary_paths[0], object_path)
try:
os.link(temporary_paths[0], object_path)
except FileExistsError:
pass
if hashlib.sha256(object_path.read_bytes()).hexdigest() != media_id:
raise OSError("media hash verification failed")
with tempfile.NamedTemporaryFile(
@@ -166,7 +169,10 @@ class ImageMediaStore:
metadata_file.write(json.dumps(canonical, sort_keys=True) + "\n")
metadata_file.flush()
os.fsync(metadata_file.fileno())
os.replace(metadata_path_tmp, metadata_path)
try:
os.link(metadata_path_tmp, metadata_path)
except FileExistsError:
pass
if json.loads(metadata_path.read_text()) != canonical:
raise OSError("durable media metadata verification failed")
finally:
+14 -14
View File
@@ -139,20 +139,20 @@ def test_deduplicated_corrupt_object_is_rejected(tmp_path):
store.put(_png())
@pytest.mark.parametrize("failure_point", ["replace", "fsync"])
@pytest.mark.parametrize("failure_point", ["link", "fsync"])
def test_write_failure_leaves_no_usable_reference(tmp_path, monkeypatch, failure_point):
import astrbot.core.utils.image_media_store as media_store
store = ImageMediaStore(tmp_path / "media")
original_replace = os.replace
original_link = os.link
original_fsync = os.fsync
if failure_point == "replace":
if failure_point == "link":
def fail_replace(source, target):
raise OSError("injected replace failure")
def fail_link(source, target):
raise OSError("injected link failure")
monkeypatch.setattr(media_store.os, "replace", fail_replace)
monkeypatch.setattr(media_store.os, "link", fail_link)
else:
def fail_fsync(fd):
@@ -162,27 +162,27 @@ def test_write_failure_leaves_no_usable_reference(tmp_path, monkeypatch, failure
with pytest.raises(OSError):
store.put(_png())
assert not list(store.root.glob("*.bin"))
assert not list(store.root.glob("*.json"))
assert not list(store.root.iterdir())
assert original_replace and original_fsync
if failure_point == "fsync":
assert not list(store.root.iterdir())
assert original_link and original_fsync
def test_metadata_replace_failure_is_repaired_by_next_put(tmp_path, monkeypatch):
import astrbot.core.utils.image_media_store as media_store
store = ImageMediaStore(tmp_path / "media")
original_replace = os.replace
original_link = os.link
failed = False
def fail_metadata_replace(source, target):
def fail_metadata_link(source, target):
nonlocal failed
if target.suffix == ".json" and not failed:
failed = True
raise OSError("injected metadata replace failure")
return original_replace(source, target)
raise OSError("injected metadata link failure")
return original_link(source, target)
monkeypatch.setattr(media_store.os, "replace", fail_metadata_replace)
monkeypatch.setattr(media_store.os, "link", fail_metadata_link)
with pytest.raises(OSError):
store.put(_png())
assert not list(store.root.glob("*.json"))
+11 -9
View File
@@ -116,6 +116,15 @@ async def test_cancelled_worker_cleans_after_exit(tmp_path, monkeypatch, timeout
output = tmp_path / "worker-output"
entered = threading.Event()
release = threading.Event()
cleaned = asyncio.Event()
original_unlink = Path.unlink
def unlink(path, *args, **kwargs):
original_unlink(path, *args, **kwargs)
if path == output:
cleaned.set()
monkeypatch.setattr(Path, "unlink", unlink)
def blocked_worker(*args, **kwargs):
entered.set()
@@ -136,15 +145,8 @@ async def test_cancelled_worker_cleans_after_exit(tmp_path, monkeypatch, timeout
await task
finally:
release.set()
# Let the shielded worker and its cleanup callback finish.
for _ in range(100):
await asyncio.sleep(0.01)
if not any(
not pending.done()
for pending in asyncio.all_tasks()
if pending is not asyncio.current_task()
):
break
# A finished task can still have its cleanup callback queued on the loop.
await asyncio.wait_for(cleaned.wait(), timeout=5)
assert not output.exists()
assert source.read_bytes() == b"source"