Stage 07 · Manifest 发布崩溃矩阵

目标

在 manifest rename 前后注入崩溃，钉死其线性化点。

动手任务

从stage-06开始，通过 CrashOnce 覆盖 DiskStorage.persist_bucket 的每个发布边界。 行为必须留在下列源码同构边界中；不要先复制补丁。

交付文件

• tests/test_storage.py

自查

1. 本阶段的可见性或状态迁移由谁负责？

   答案

   rename 前恢复到旧状态；rename 后看到完整新状态。

2. 如果绕过新边界，哪个测试会最先失败？

   答案

   阅读 tests.txt，找出最窄的新节点，并说出它覆盖的公开调用。

通关命令

uv run pytest -q $(cat journey/stages/07-publication-crash-matrix/tests.txt)

对应真实 S3 的一课

rename 前恢复到旧状态；rename 后看到完整新状态。

教材

第 5 章

在 GitHub 查看阶段差异

完成后可运行 git checkout stage-07 对照你的结果。

先做后看：stage.patch

diff --git a/tests/test_storage.py b/tests/test_storage.py
index c96a7a6..5faad97 100644
--- a/tests/test_storage.py
+++ b/tests/test_storage.py
@@ -30,3 +30,69 @@ def test_restart_restores_versions_bodies_and_counter(tmp_path: Path) -> None:

     assert reopened.get_object("b", "k", version_id=first.version_id).body == b"one"
     assert second.version_id != first.version_id
+
+
+def test_crash_before_manifest_publish_leaves_old_state(tmp_path: Path) -> None:
+    MiniS3(tmp_path).create_bucket("b")
+    crashing = MiniS3(
+        tmp_path,
+        counter=SequenceCounter(10),
+        crash_injector=CrashOnce("before_manifest_publish"),
+    )
+
+    with pytest.raises(InjectedCrash):
+        crashing.put_object("b", "new", b"value")
+
+    reopened = MiniS3(tmp_path)
+    with pytest.raises(NoSuchKey):
+        reopened.get_object("b", "new")
+    assert not list(tmp_path.rglob("*.tmp-*"))
+    assert not list(tmp_path.rglob("e00000010.*"))
+
+
+@pytest.mark.parametrize(
+    "crash_point",
+    [
+        "after_object_directory_create",
+        "after_object_directory_parent_fsync",
+        "before_manifest_publish",
+    ],
+)
+def test_crash_before_manifest_publication_matrix_leaves_old_state(
+    tmp_path: Path,
+    crash_point: str,
+) -> None:
+    MiniS3(tmp_path).create_bucket("b")
+    crashing = MiniS3(
+        tmp_path,
+        counter=SequenceCounter(10),
+        crash_injector=CrashOnce(crash_point),
+    )
+
+    with pytest.raises(InjectedCrash):
+        crashing.put_object("b", "new", b"value")
+
+    reopened = MiniS3(tmp_path)
+    with pytest.raises(NoSuchKey):
+        reopened.get_object("b", "new")
+    assert not list(tmp_path.rglob("*.tmp-*"))
+    assert not list(tmp_path.rglob("e00000010.*"))
+
+
+def test_crash_after_manifest_publish_exposes_complete_new_state(
+    tmp_path: Path,
+) -> None:
+    MiniS3(tmp_path).create_bucket("b")
+    crashing = MiniS3(
+        tmp_path,
+        counter=SequenceCounter(10),
+        crash_injector=CrashOnce("after_manifest_publish"),
+    )
+
+    with pytest.raises(InjectedCrash):
+        crashing.put_object("b", "new", b"value")
+
+    reopened = MiniS3(tmp_path)
+    visible = reopened.get_object("b", "new")
+    assert visible.body == b"value"
+    assert visible.etag == '"2063c1608d6e0baf80249c42e2be5804"'