diff --git a/CHANGELOG.md b/CHANGELOG.md
index 32e6741..9bf02aa 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -53,8 +53,17 @@
   leaving peers to notice the outage from their own timeouts. Crashes are
   recorded too: `crash()` writes a trace event and consumes a uid.
 - Every host now has `host.disk`, a mapping that survives its crashes:
-  where state a real process would fsync belongs. Writes are atomic at
-  assignment; there is no partial-write model.
+  where state a real process would fsync belongs. Writes are durable at
+  assignment and `disk.sync()` does nothing — until the host asks for
+  otherwise with `loop.net.set_disk(name, buffered=True)`, which queues
+  writes and deletes until a `sync()` while the host itself reads them
+  back immediately. A crash then takes the queue with it, and `torn=True`
+  keeps a seeded prefix of it instead: the state a machine that lost power
+  mid-batch actually reboots into. A prefix is all it claims to be —
+  nothing is reordered and no value is ever half-written. A disk nobody
+  configured is untouched down to the draw: storage records no trace
+  events and consumes no randomness, so those runs decide everything
+  exactly as they did before.
 - Clocks can lie per host: `loop.net.set_clock(name, offset=...)` skews
   what that host's tasks read from `loop.time()`, and the deadlines they
   hand to `call_at` with it, while durations (`sleep`, `timeout`,
@@ -64,10 +73,12 @@
   exactly the decisions it made without the feature.
 - A second flagship demo: `examples/raft/` is a teaching-sized Raft (leader
   election + log replication, plain asyncio on streams) tested only under
-  simulation — four safety invariants checked over 50,000 chaos seeds, five
+  simulation — four safety invariants checked over 50,000 chaos seeds, six
   safeguard ablations each caught and replayed from a seed, and failing
   schedules minimized toward FIFO — down to a single interesting step in the
-  sharpest case.
+  sharpest case. Its state can live on host disks that buffer and tear, with
+  the syncs Raft owes before it answers an RPC; drop the sync before an
+  append is acknowledged and the first seed loses a committed entry.
 - Campaign evidence at scale, regenerable via `benchmarks/campaign.py`:
   100,000 seeds of jobqueue chaos green in under six minutes on a laptop,
   every ablation caught with its failure density recorded, and 20 sampled
diff --git a/README.md b/README.md
index f3fcb6b..07e223e 100644
--- a/README.md
+++ b/README.md
@@ -233,7 +233,9 @@ net.crash("node2")                              # no reset, just silence
 - **Crashes with a way back** — `net.crash` kills a host's tasks and
   binds, `net.restart` brings it back as a fresh incarnation, and
   `host.disk` is a mapping that outlives both: machines die, reboot, and
-  remember what they wrote down.
+  remember what they wrote down. `net.set_disk(name, buffered=True,
+  torn=True)` makes the disk lie too — writes only land on `sync()`, and a
+  crash keeps a seeded prefix of whatever was still queued.
 - **Clocks that lie** — `net.set_clock(name, offset=...)` skews what one
   host reads from the clock without changing how long anything takes, for
   testing lease and timeout code against machines that disagree about the
diff --git a/docs/design.md b/docs/design.md
index ffd39aa..737e8fe 100644
--- a/docs/design.md
+++ b/docs/design.md
@@ -202,6 +202,17 @@ Decisions inside that model, each doing real work:
   listeners close, and they go silent. Peers cannot distinguish a crash from
   a partition except by timeout — which is the entire epistemology of
   distributed failure detection, enforced by construction.
+- **A disk lies only when asked to.** Storage is durable at assignment
+  unless a host opts into `set_disk(name, buffered=True)`, because the
+  default has to be the one that costs nothing: an unconfigured disk draws
+  no numbers, records no events, and leaves every existing trace hash where
+  it was. Opted in, writes queue until `sync()`, and a crash keeps a seeded
+  *prefix* of the queue. A prefix, deliberately: real hardware also reorders
+  and corrupts individual sectors, and modeling that would need a sector
+  layer nothing else here would use, while a prefix is the failure an
+  application can actually defend against — write, sync, and only then act
+  on it. Skipping that sync is a bug the simulation can now find, which is
+  the whole reason the model exists.
 - **The accept is sequence 0.** The server builds its transport and sends
   `accept` before its protocol's `connection_made` can write; the client
   transport is built when the accept is *dispatched*, not when the connector
diff --git a/docs/supported-api.md b/docs/supported-api.md
index 0a26ac8..a749a46 100644
--- a/docs/supported-api.md
+++ b/docs/supported-api.md
@@ -41,7 +41,8 @@ a host belong to an implicit `driver` host.
 | `loop.net.partition` / `heal` | Silent blackhole: datagrams are lost, stream traffic is held and resumes intact after healing; nothing errors — only your own timeouts fire |
 | `loop.net.crash` | A host's tasks are cancelled and it goes silent; no reset is sent — peers cannot tell a crash from a partition |
 | `loop.net.restart` / `host.restart()` | The counterpart to a crash: the host comes back as a fresh incarnation. Liveness is all that is revived — the old tasks stay cancelled, its listeners and binds are gone, and the caller boots whatever should run on the machine again, the same way it booted it the first time. Cancellation is requested at crash and lands on the next scheduler step, so a restart in the same step can briefly coexist with a dying task that swallows `CancelledError`. A packet is checked against liveness when it arrives, so traffic due during the dead window is lost; a packet that was already in flight and lands after the machine is back is delivered, and finds a host that no longer holds the old incarnation's connections. Peers still learn about the outage only from their own timeouts |
-| `host.disk` | Storage that survives the crash: a `MutableMapping` per host, where state a real process would fsync belongs. Writes are atomic at assignment; there is no partial-write model. Values are stored as given, so mutating a stored object afterwards is the caller's own aliasing, exactly as with a cache in front of a real disk |
+| `host.disk` | Storage that survives the crash: a `MutableMapping` per host, where state a real process would fsync belongs. By default a write is durable the moment it is made. Values are stored as given, so mutating a stored object afterwards is the caller's own aliasing, exactly as with a cache in front of a real disk. `disk.sync()` exists on every disk and does nothing on one that does not buffer, so the code under test is written the same way either way |
+| `loop.net.set_disk` | Makes a host's disk lie about when a write lands. `buffered=True` queues writes and deletes in order and only makes them durable on `sync()`; reads on that host see the queue merged over what is durable, in a fixed order — durable keys where the durable state has them, then the keys the queue invented, in write order, with a queued delete hiding a key. A crash throws the queue away and the reboot finds what was synced; with `torn=True` a seeded prefix of the queue survives instead, which is the state a machine that lost power part-way through a batch comes back with. A prefix is the whole model: writes never land out of order, and no value is ever half-written. Tearing needs a buffer to tear (`torn=True` alone is a `ValueError`), and reconfiguring a disk flushes whatever it was holding. The prefix is drawn from a seed-derived stream of its own, so a torn run makes exactly the network draws it would have made untorn, and a run that never calls `set_disk` draws nothing and records nothing — storage is not a scheduling event and has no trace events at all |
 | `loop.net.set_clock` / `clock_offset` | Per-host clock skew, in seconds. The offset changes what that host's tasks *read*: `loop.time()` (and `sim.time()` with it) returns true time plus the offset, and a deadline handed to `call_at` is interpreted on the calling task's clock. Durations are immune — `asyncio.sleep`, `asyncio.timeout`, `wait_for` and `call_later` cost the same everywhere, which is exactly what a wrong wall clock does to a real machine. By default the driver and unconfigured hosts read true time; the driver can be given an offset too. Trace timestamps stay on the true clock, so skew never perturbs scheduling and traces from skewed runs stay comparable |
 | `transport.abort()` | Peer gets `connection_lost(ConnectionResetError)` |
 | `loop.getaddrinfo` | Resolves against the host table, never DNS: a registered host name, its synthetic address, or a loopback-shaped name (`None`, `""`, `localhost`, `127.0.0.1`, `0.0.0.0`) meaning the calling task's own host. Returns stdlib-shaped rows — `(AF_INET, SOCK_STREAM, IPPROTO_TCP, "", (address, port))` and the `SOCK_DGRAM` / `IPPROTO_UDP` row — filtered by `family`, `type` and `proto`. Ports are numeric (`int`, a digit string, or `None` for 0); resolver `flags` have nothing to vary |
diff --git a/examples/raft/README.md b/examples/raft/README.md
index db6eae1..6a307dd 100644
--- a/examples/raft/README.md
+++ b/examples/raft/README.md
@@ -12,7 +12,9 @@ changes, no client sessions. Submission is at-least-once and says so: a
 command a leader accepts before being deposed can be resubmitted and commit
 twice, under two indices, and the safety claims below deliberately do not
 mind. Storage is an in-memory stand-in that survives a process restart within
-a run, the way a disk survives a reboot.
+a run, the way a disk survives a reboot — or, when a scenario asks for it,
+a simulated host disk that only makes a write durable when the node syncs it
+and keeps a seeded prefix of the rest when the power goes.
 
 ## The claim, stated honestly
 
@@ -24,6 +26,8 @@ Each rule that carries the safety argument sits behind its own flag in
 `Safeguards`, so the tests can switch exactly one off and watch the explorer
 find the schedule it lets through: stale-term rejection, one vote per term
 (§5.2), the log-freshness check on votes (§5.4.1), persistence before reply,
+the two syncs that make that persistence mean something on a disk that
+buffers — before a vote is granted and before an append is acknowledged —
 the own-term commit gate (§5.4.2), and the per-term no-op that lets a quiet
 term commit (§8).
 
@@ -53,13 +57,22 @@ leader that committed it:
 
 - Scenario suite: 11 seeded scenarios (elections, replication, RPC framing)
   × 10 seeds each, alongside unit tests for the log, vote and persistence
-  rules — 63 fast tests plus the slow proofs, green.
+  rules — 71 fast tests plus the slow proofs, green.
 - Campaign: **50,000 seeds** of five-node chaos — three randomized partition
   windows per seed, a process restart after about half of them, 2% message
   drop and 2% duplication throughout, with a client proposing — invariants
   held on every seed. 917.29s (15m17s) with `--simloop-jobs=8` on an M4
   MacBook Air, about 54 seeds a second. The same scenario runs 300 seeds
   sequentially in 27.97s and 2,000 seeds in 36.34s at `--simloop-jobs=8`.
+- The same chaos on disks that lose power: state on `host.disk` configured
+  `buffered=True, torn=True`, every restart a hard crash. **5,000 seeds**
+  green in 51.55s at `--simloop-jobs=8`, and the 300 the suite runs by
+  default in 14.10s sequentially (M4 MacBook Air, approximate — measured
+  with other work on the machine). Sync discipline is exactly what makes
+  that boring: across 25 seeds the power went out 39 times and found an
+  empty write buffer every time, because the node has already flushed
+  whatever it answered with. Switch one sync off and the same power cuts
+  land on disks holding 13 unflushed records apiece.
 - Replay stability: a one-off local measurement re-explored each ablation's
   found-at seed 100 times on a fresh loop — 5 seeds × 100 replays, one trace
   hash apiece, byte-identical throughout. The standing check is
@@ -76,10 +89,11 @@ leader that committed it:
 | 3 | Persistence before reply off (`persist_before_reply=False`) | leader-completeness | 4 | 5 | `... ::test_skipped_persistence_forgets_committed_entries` |
 | 4 | Stale-term rejection off (`reject_stale_term=False`) | state-machine-safety | 0 | 1 | `... ::test_accepting_stale_terms_rewrites_history` |
 | 5 | Commit gate off (`commit_own_term_only=False`, no-op off both sides) | state-machine-safety | 3 | 4 | `... ::test_committing_old_terms_by_count_loses_writes` |
+| 6 | Sync before acking an append off (`sync_before_ack=False`, on buffered, torn disks) | leader-completeness | 0 | 1 | `... ::test_an_unsynced_ack_loses_a_committed_entry` |
 
-Rows 1–4 searched a budget of 300 seeds, row 5 a budget of 500. All five are
-labeled ablations — detection demonstrations, not bugs that were ever
-shipped.
+Rows 1–4 and 6 searched a budget of 300 seeds, row 5 a budget of 500. All
+six are labeled ablations — detection demonstrations, not bugs that were
+ever shipped.
 
 "Invariant violated" records what the found seed actually produced, not the
 only thing that ablation can produce. Three of the five tests deliberately
@@ -88,6 +102,26 @@ instance, takes both stale appends and stale vote grants, so naming one of the
 four claims would say less than letting the checker report which one broke
 first.
 
+Row 6 is the disk's. A node keeps its whole record — term, vote and log —
+under one key, so a power cut can rewind it to an earlier record but can
+never leave it holding this term beside the previous vote; what makes a
+record durable is the `sync()` the node owes before it answers an RPC. Drop
+the one before an append is acknowledged and the leader counts a follower
+that has the entries in a write buffer and nowhere else: cut the power to
+the followers behind a partition and the term the survivors elect next has
+never heard of an entry that was already committed. Seed 0 produces it, and
+the schedule minimizes to FIFO except one step of 1,273. Put the sync back
+and the same scenario holds across 150 seeds
+(`test_the_synced_ack_carries_the_same_scenario`, slow-marked).
+
+The vote's sync is the same argument with the same flag shape
+(`sync_before_vote`), but the explorer is not what proves it here: the
+window it opens is one heartbeat wide — the winner's first AppendEntries
+flushes the vote behind the voter's back — and 4,000 seeds of a scenario
+built to walk through it never did. `tests/test_votes.py` shows the hazard
+directly instead: grant a vote on a buffered disk, cut the power, and the
+machine that boots grants the same term to somebody else.
+
 Two safeguards are also shown to be load-bearing *on their own*, which is the
 other half of the argument: with the commit gate the only thing standing (the
 per-term no-op switched off on both sides), the paper's Figure 8 runs clean
@@ -140,14 +174,15 @@ take `shrink=True` as an argument.
 ## Run it
 
     uv run pytest examples/raft/tests -q            # fast suite: scenarios, units, ablations
-    uv run pytest examples/raft/tests -q -m slow    # campaign, the two safe proofs, the replay guard
+    uv run pytest examples/raft/tests -q -m slow    # campaigns, the safe proofs, the replay guard
 
 Add `-s` to either and the ablations print the explorer's report — the failing
 seed and the trace around it.
 
-Turn the campaign up and spread the seeds over cores:
+Turn a campaign up and spread the seeds over cores:
 
-    uv run pytest examples/raft/tests/test_chaos_campaign.py -q -m slow --simloop-seeds=50000 --simloop-jobs=8
+    uv run pytest 'examples/raft/tests/test_chaos_campaign.py::test_chaos_campaign_holds_the_invariants' -q -m slow --simloop-seeds=50000 --simloop-jobs=8
+    uv run pytest 'examples/raft/tests/test_chaos_campaign.py::test_the_campaign_holds_on_disks_that_lose_power' -q -m slow --simloop-seeds=5000 --simloop-jobs=8
 
 Replay any campaign failure exactly:
 
diff --git a/examples/raft/raft/node.py b/examples/raft/raft/node.py
index d4640a8..3c20a40 100644
--- a/examples/raft/raft/node.py
+++ b/examples/raft/raft/node.py
@@ -41,6 +41,11 @@ class Safeguards:
     one_vote_per_term: bool = True      # never grant two candidates one term
     check_log_up_to_date: bool = True   # §5.4.1: voters gate on log freshness
     persist_before_reply: bool = True   # durability before acknowledgement
+    # A write is only as durable as the disk it landed on. On a buffered disk
+    # these two are what push it past the buffer before the answer that
+    # depends on it goes out; on a disk that never buffers they cost nothing.
+    sync_before_vote: bool = True       # a vote is on the disk before it is granted
+    sync_before_ack: bool = True        # entries are on the disk before they are acked
     # §5.4.2: count replicas only for own-term entries. With leader_noop on,
     # every reachable quorum index already carries the leader's term, so this
     # gate only shows its teeth when the no-op is off too.
@@ -149,8 +154,9 @@ async def _connection(
         except (asyncio.IncompleteReadError, OSError, wire.FrameError):
             # IncompleteReadError is named on its own because it is not an
             # OSError. OSError here means a torn socket -- which also swallows
-            # one raised out of handle(), acceptable only while storage is in
-            # memory; a file-backed Storage needs its own except clause.
+            # one raised out of handle(), acceptable only while no Storage
+            # here can fail a write; a real file-backed one needs its own
+            # except clause.
             pass
         finally:
             writer.close()
@@ -329,7 +335,7 @@ def _handle_request_vote(self, m: dict[str, Any]) -> dict[str, Any]:
         if self._safeguards.check_log_up_to_date and theirs < ours:
             return refused
         self._state.voted_for = m["candidate"]
-        self._persist()
+        self._persist(sync=self._safeguards.sync_before_vote)
         self._reset.set()  # granting a vote defers our own candidacy
         return {"term": self._state.term, "granted": True}
 
@@ -351,7 +357,7 @@ def _handle_append_entries(self, m: dict[str, Any]) -> dict[str, Any]:
                 del log[index - 1 :]
             if index > len(log):
                 log.append(Entry(entry_term, command))
-        self._persist()
+        self._persist(sync=self._safeguards.sync_before_ack)
         if m["leader_commit"] > self.commit_index:
             # max(): a backed-off heartbeat carries a short verified prefix;
             # the commit mark never moves backwards for it.
@@ -380,9 +386,11 @@ def _become_follower(self, term: int) -> None:
         self._persist()
         self.role = FOLLOWER
 
-    def _persist(self) -> None:
+    def _persist(self, *, sync: bool = True) -> None:
         if self._safeguards.persist_before_reply:
             self._storage.save(self._state)
+            if sync:
+                self._storage.sync()
 
     def _last_log_term(self) -> int:
         return self._state.log[-1].term if self._state.log else 0
diff --git a/examples/raft/raft/storage.py b/examples/raft/raft/storage.py
index 82f0848..0b50e66 100644
--- a/examples/raft/raft/storage.py
+++ b/examples/raft/raft/storage.py
@@ -1,9 +1,9 @@
-"""What a Raft peer must remember across restarts, and an in-memory disk."""
+"""What a Raft peer must remember across restarts, and the disks it can keep it on."""
 
 from __future__ import annotations
 
 from dataclasses import dataclass, field
-from typing import Protocol
+from typing import Any, Protocol
 
 
 @dataclass(frozen=True)
@@ -24,6 +24,8 @@ def load(self) -> PersistentState: ...
 
     def save(self, state: PersistentState) -> None: ...
 
+    def sync(self) -> None: ...
+
 
 class MemoryStorage:
     """Survives a node restart within one run, the way a disk survives reboots."""
@@ -42,3 +44,53 @@ def save(self, state: PersistentState) -> None:
         self._saved = PersistentState(
             term=state.term, voted_for=state.voted_for, log=list(state.log)
         )
+
+    def sync(self) -> None:
+        """Nothing to do: a save here is already as durable as this disk gets."""
+
+
+class Disk(Protocol):
+    """The slice of a host's storage this demo uses: a mapping that flushes."""
+
+    def __contains__(self, key: object) -> bool: ...
+
+    def __getitem__(self, key: str) -> Any: ...
+
+    def __setitem__(self, key: str, value: Any) -> None: ...
+
+    def sync(self) -> None: ...
+
+
+class DiskStorage:
+    """Persistent state on a machine's disk, made durable when ``sync`` says so.
+
+    The whole record goes down under one key, as a single immutable value.
+    That is what makes a save atomic against a power cut: a disk that keeps
+    a prefix of what it was holding can rewind the node to an earlier
+    record, but never leave it with this term beside the previous vote.
+    """
+
+    KEY = "raft-state"
+
+    def __init__(self, disk: Disk) -> None:
+        self._disk = disk
+
+    def load(self) -> PersistentState:
+        if self.KEY not in self._disk:
+            return PersistentState()
+        term, voted_for, log = self._disk[self.KEY]
+        return PersistentState(
+            term=term,
+            voted_for=voted_for,
+            log=[Entry(entry_term, command) for entry_term, command in log],
+        )
+
+    def save(self, state: PersistentState) -> None:
+        self._disk[self.KEY] = (
+            state.term,
+            state.voted_for,
+            tuple((entry.term, entry.command) for entry in state.log),
+        )
+
+    def sync(self) -> None:
+        self._disk.sync()
diff --git a/examples/raft/tests/harness.py b/examples/raft/tests/harness.py
index 6382fce..382ed41 100644
--- a/examples/raft/tests/harness.py
+++ b/examples/raft/tests/harness.py
@@ -11,7 +11,7 @@
 
 from raft import wire
 from raft.node import LEADER, PORT, Event, RaftNode, Safeguards
-from raft.storage import Entry, MemoryStorage
+from raft.storage import DiskStorage, Entry, MemoryStorage, Storage
 
 from checks import check_invariants
 
@@ -25,7 +25,7 @@ def sim_loop() -> SimLoop:
 @dataclass
 class Member:
     name: str
-    storage: MemoryStorage
+    storage: Storage
     node: RaftNode
     task: asyncio.Task[Any]
 
@@ -42,8 +42,15 @@ def logs(self) -> dict[str, tuple[Entry, ...]]:
 
 
 async def start_cluster(
-    *, size: int = 3, safeguards: Safeguards | None = None
+    *, size: int = 3, safeguards: Safeguards | None = None, disks: bool = False
 ) -> Cluster:
+    """Boot a cluster; ``disks`` puts its state on host disks that lose power.
+
+    With disks on, each node keeps its record on `host.disk` configured
+    buffered and torn: a write is only durable once the node syncs it, and a
+    crash keeps a seeded prefix of whatever it was still holding. Off, the
+    state sits in a stand-in that is durable the moment it is written.
+    """
     loop = sim_loop()
     loop.net.set_defaults(latency=(0.01, 0.05))
     cluster = Cluster(
@@ -53,12 +60,19 @@ async def start_cluster(
         safeguards=safeguards if safeguards is not None else Safeguards(),
     )
     for name in cluster.names:
-        _boot(cluster, name, MemoryStorage())
+        storage: Storage
+        if disks:
+            host = loop.net.host(name)
+            loop.net.set_disk(name, buffered=True, torn=True)
+            storage = DiskStorage(host.disk)
+        else:
+            storage = MemoryStorage()
+        _boot(cluster, name, storage)
     await asyncio.sleep(0.05)  # let the servers start listening
     return cluster
 
 
-def _boot(cluster: Cluster, name: str, storage: MemoryStorage) -> None:
+def _boot(cluster: Cluster, name: str, storage: Storage) -> None:
     loop = sim_loop()
     node = RaftNode(
         name,
@@ -83,11 +97,33 @@ async def restart(cluster: Cluster, name: str) -> None:
     _boot(cluster, name, member.storage)
 
 
-async def chaos(cluster: Cluster, rng: random.Random) -> None:
+async def power_cut(cluster: Cluster, name: str) -> None:
+    """The machine dies where it stands, then boots from whatever the disk kept.
+
+    Unlike ``restart``, the host itself goes down: a buffered disk loses the
+    writes it had not synced, so the incarnation that boots here can be
+    missing something the old one already told its peers.
+    """
+    loop = sim_loop()
+    member = cluster.members[name]
+    loop.net.crash(name)
+    try:
+        await member.task
+    except asyncio.CancelledError:
+        pass
+    loop.net.restart(name)
+    _boot(cluster, name, member.storage)
+
+
+async def chaos(
+    cluster: Cluster, rng: random.Random, *, power_cuts: bool = False
+) -> None:
     """A seed-derived fault schedule: partition windows and process restarts.
 
     Cut sizes never exceed half the cluster, so a quorum side always exists
     and the driver -- which is never partitioned -- can keep proposing.
+    ``power_cuts`` makes those restarts hard ones, which is only a different
+    fault if the cluster's disks buffer.
     """
     loop = sim_loop()
     for _ in range(3):
@@ -98,7 +134,11 @@ async def chaos(cluster: Cluster, rng: random.Random) -> None:
         await asyncio.sleep(rng.uniform(0.5, 3.0))
         loop.net.heal()
         if rng.random() < 0.5:
-            await restart(cluster, rng.choice(cluster.names))
+            victim = rng.choice(cluster.names)
+            if power_cuts:
+                await power_cut(cluster, victim)
+            else:
+                await restart(cluster, victim)
 
 
 def leader_now(cluster: Cluster) -> str | None:
diff --git a/examples/raft/tests/test_ablations.py b/examples/raft/tests/test_ablations.py
index 44c83c0..f4a9d56 100644
--- a/examples/raft/tests/test_ablations.py
+++ b/examples/raft/tests/test_ablations.py
@@ -100,6 +100,46 @@ def test_skipped_persistence_forgets_committed_entries() -> None:
     )
 
 
+async def unsynced_ack(*, sync_before_ack: bool) -> None:
+    """A committed entry that only ever reached the followers' write buffers.
+
+    The leader counts an ack it should not have been given yet, commits,
+    and applies. Cutting the power to the followers behind a partition
+    takes the entry with it, and the term they elect next has no idea it
+    was ever committed.
+    """
+    cluster = await harness.start_cluster(
+        safeguards=Safeguards(sync_before_ack=sync_before_ack), disks=True
+    )
+    loop = harness.sim_loop()
+    leader = await harness.wait_for_leader(cluster)
+    await harness.propose(cluster, "k0")
+    rest = [name for name in cluster.names if name != leader]
+    loop.net.partition([leader], rest)
+    for name in rest:
+        await harness.power_cut(cluster, name)
+    await asyncio.sleep(4.0)
+    harness.verify(cluster)
+
+
+def test_an_unsynced_ack_loses_a_committed_entry() -> None:
+    report = _find(lambda: unsynced_ack(sync_before_ack=False))
+    assert isinstance(report.exception, InvariantViolation)
+    assert report.exception.invariant in (
+        "leader-completeness", "state-machine-safety",
+    )
+
+
+@pytest.mark.slow
+def test_the_synced_ack_carries_the_same_scenario() -> None:
+    """The other half of the row above: put the sync back and it holds.
+
+    Same power cuts on the same disks -- the only difference is that the
+    entries were on them before the acks went out.
+    """
+    assert explore(lambda: unsynced_ack(sync_before_ack=True), range(150)) is None
+
+
 def test_accepting_stale_terms_rewrites_history() -> None:
     async def scenario() -> None:
         cluster = await harness.start_cluster(
diff --git a/examples/raft/tests/test_chaos_campaign.py b/examples/raft/tests/test_chaos_campaign.py
index 5f20e6d..1651d10 100644
--- a/examples/raft/tests/test_chaos_campaign.py
+++ b/examples/raft/tests/test_chaos_campaign.py
@@ -11,16 +11,14 @@
 import harness
 
 
-@pytest.mark.slow
-@sim_test(seeds=300)
-async def test_chaos_campaign_holds_the_invariants() -> None:
+async def _campaign(*, disks: bool) -> None:
     rng = sim.random
     loop = harness.sim_loop()
-    cluster = await harness.start_cluster(size=5)
+    cluster = await harness.start_cluster(size=5, disks=disks)
     loop.net.set_defaults(latency=(0.01, 0.05), drop=0.02, duplicate=0.02)
     for i in range(2):
         await harness.propose(cluster, f"before.{i}")
-    disorder = loop.create_task(harness.chaos(cluster, rng))
+    disorder = loop.create_task(harness.chaos(cluster, rng, power_cuts=disks))
     sent = 0
     while not disorder.done():
         await harness.propose(cluster, f"during.{sent}", timeout_s=120.0)
@@ -31,3 +29,22 @@ async def test_chaos_campaign_holds_the_invariants() -> None:
     await harness.propose(cluster, "after", timeout_s=120.0)
     await harness.settle(cluster)
     harness.verify(cluster)
+
+
+@pytest.mark.slow
+@sim_test(seeds=300)
+async def test_chaos_campaign_holds_the_invariants() -> None:
+    await _campaign(disks=False)
+
+
+@pytest.mark.slow
+@sim_test(seeds=300)
+async def test_the_campaign_holds_on_disks_that_lose_power() -> None:
+    """The same chaos, with the state on buffered disks and hard crashes.
+
+    Every restart here is a power cut: the machine dies with whatever it had
+    not synced, and a seeded prefix of that is all its reboot can find. The
+    node syncs before every reply it makes, which is exactly the discipline
+    this campaign is holding to account.
+    """
+    await _campaign(disks=True)
diff --git a/examples/raft/tests/test_persistence.py b/examples/raft/tests/test_persistence.py
index 22cb675..7f6d851 100644
--- a/examples/raft/tests/test_persistence.py
+++ b/examples/raft/tests/test_persistence.py
@@ -2,7 +2,9 @@
 
 from __future__ import annotations
 
-from raft.storage import Entry, MemoryStorage, PersistentState
+from simloop import SimLoop
+
+from raft.storage import DiskStorage, Entry, MemoryStorage, PersistentState
 
 
 def test_a_fresh_disk_loads_empty_state() -> None:
@@ -35,3 +37,60 @@ def test_mutating_saved_state_does_not_write_through() -> None:
     state.term = 9
     assert disk.load().log == [Entry(1, "a")]
     assert disk.load().term == 1
+
+
+def _powered(seed: int = 0, *, torn: bool = True) -> SimLoop:
+    loop = SimLoop(seed=seed)
+    loop.net.host("n1")
+    loop.net.set_disk("n1", buffered=True, torn=torn)
+    return loop
+
+
+def test_disk_storage_round_trips_through_a_host_disk() -> None:
+    loop = _powered()
+    storage = DiskStorage(loop.net.host("n1").disk)
+    storage.save(PersistentState(term=3, voted_for="n2", log=[Entry(1, "a")]))
+    storage.sync()
+    state = storage.load()
+    assert (state.term, state.voted_for, state.log) == (3, "n2", [Entry(1, "a")])
+    loop.close()
+
+
+def test_a_fresh_host_disk_loads_empty_state() -> None:
+    loop = _powered()
+    state = DiskStorage(loop.net.host("n1").disk).load()
+    assert (state.term, state.voted_for, state.log) == (0, None, [])
+    loop.close()
+
+
+def test_the_power_cut_takes_what_was_never_synced() -> None:
+    loop = _powered(torn=False)
+    storage = DiskStorage(loop.net.host("n1").disk)
+    storage.save(PersistentState(term=1, voted_for="n1", log=[]))
+    storage.sync()
+    storage.save(PersistentState(term=2, voted_for="n3", log=[Entry(2, "a")]))
+    loop.net.crash("n1")
+    loop.net.restart("n1")
+    state = DiskStorage(loop.net.host("n1").disk).load()
+    assert (state.term, state.voted_for, state.log) == (1, "n1", [])
+    loop.close()
+
+
+def test_a_torn_crash_lands_on_a_whole_record() -> None:
+    # Each save is one write, so a torn crash can rewind the state to any
+    # record the node wrote -- never to half of one, which is why the state
+    # goes down as a single value.
+    written = [
+        PersistentState(term=term, voted_for=f"n{term}", log=[Entry(term, "a")])
+        for term in range(1, 9)
+    ]
+    for seed in range(6):
+        loop = _powered(seed)
+        storage = DiskStorage(loop.net.host("n1").disk)
+        for state in written:
+            storage.save(state)
+        loop.net.crash("n1")
+        loop.net.restart("n1")
+        landed = DiskStorage(loop.net.host("n1").disk).load()
+        loop.close()
+        assert landed in [PersistentState()] + written
diff --git a/examples/raft/tests/test_votes.py b/examples/raft/tests/test_votes.py
index cb966d3..185d565 100644
--- a/examples/raft/tests/test_votes.py
+++ b/examples/raft/tests/test_votes.py
@@ -4,8 +4,10 @@
 
 import random
 
+from simloop import SimLoop
+
 from raft.node import CANDIDATE, FOLLOWER, RaftNode, Safeguards
-from raft.storage import Entry, MemoryStorage, PersistentState
+from raft.storage import DiskStorage, Entry, MemoryStorage, PersistentState
 
 
 def make_node(
@@ -96,6 +98,47 @@ def test_without_the_freshness_check_a_stale_log_wins_votes() -> None:
     assert ask(node, term=2, last_index=0, last_term=0)["granted"] is True
 
 
+def _voted_twice_across_a_power_cut(*, sync_before_vote: bool) -> bool:
+    """Grant a vote, cut the power, and ask the machine again in the same term.
+
+    The disk buffers, so the grant is only as durable as the sync that
+    follows it -- which is the whole difference between the two callers.
+    """
+    loop = SimLoop(seed=0)
+    loop.net.host("n1")
+    loop.net.set_disk("n1", buffered=True)
+    safeguards = Safeguards(sync_before_vote=sync_before_vote)
+    node = RaftNode(
+        "n1",
+        ["n2", "n3"],
+        DiskStorage(loop.net.host("n1").disk),
+        rng=random.Random(7),
+        safeguards=safeguards,
+    )
+    ask(node, term=1, candidate="n2")
+    loop.net.crash("n1")
+    loop.net.restart("n1")
+    reborn = RaftNode(
+        "n1",
+        ["n2", "n3"],
+        DiskStorage(loop.net.host("n1").disk),
+        rng=random.Random(7),
+        safeguards=safeguards,
+    )
+    granted = ask(reborn, term=1, candidate="n3")["granted"]
+    loop.close()
+    assert isinstance(granted, bool)
+    return granted
+
+
+def test_a_synced_vote_outlives_the_power_cut() -> None:
+    assert _voted_twice_across_a_power_cut(sync_before_vote=True) is False
+
+
+def test_a_vote_left_in_the_buffer_is_granted_again() -> None:
+    assert _voted_twice_across_a_power_cut(sync_before_vote=False) is True
+
+
 def test_without_persistence_a_reload_forgets_the_vote() -> None:
     disk = MemoryStorage()
     relaxed = Safeguards(persist_before_reply=False)
diff --git a/src/simloop/_net.py b/src/simloop/_net.py
index f5ba09a..38641d0 100644
--- a/src/simloop/_net.py
+++ b/src/simloop/_net.py
@@ -232,34 +232,129 @@ def _check_latency(value: tuple[float, float]) -> tuple[float, float]:
     return (lo, hi)
 
 
+class _Deleted:
+    """The journal's way of writing down that a key went away."""
+
+    __slots__ = ()
+
+
+_DELETED = _Deleted()
+
+
 class SimDisk(MutableMapping[str, object]):
     """A host's storage that survives crashes and restarts.
 
     A crash loses everything volatile — tasks, connections, binds — but not
     what was written here, which is the whole point: this is where state
-    that a real process would fsync belongs. Writes are atomic at
-    assignment; there is no partial-write model. Values are stored as
-    given, so mutating a stored object later is the caller's own aliasing,
-    exactly as it would be with a cache in front of a real disk.
+    that a real process would fsync belongs. By default a write is durable
+    the moment it is made and there is no partial-write model. Values are
+    stored as given, so mutating a stored object later is the caller's own
+    aliasing, exactly as it would be with a cache in front of a real disk.
+
+    ``loop.net.set_disk(name, buffered=True)`` makes the disk lie the way a
+    real one does: writes and deletes queue in order and only reach durable
+    state on ``sync()``, while reads on the machine itself see them
+    immediately. A crash throws the queue away — durable state is what the
+    reboot finds. With ``torn=True`` a seeded prefix of the queue survives
+    instead, so the reboot can find a machine that got halfway through its
+    last batch. ``sync()`` exists on every disk and does nothing on an
+    unbuffered one, so the code under test is written once either way.
     """
 
     def __init__(self) -> None:
         self._data: dict[str, object] = {}
+        self._buffered = False
+        self._torn = False
+        # The journal is the ordered record a crash tears, the overlay is the
+        # same writes keyed for reading. Both are needed: the journal keeps
+        # repeated writes to one key as separate operations, which is what a
+        # prefix has to be a prefix of.
+        self._journal: list[tuple[str, object]] = []
+        self._overlay: dict[str, object] = {}
+
+    @property
+    def buffered(self) -> bool:
+        return self._buffered
+
+    @property
+    def torn(self) -> bool:
+        return self._torn
+
+    def sync(self) -> None:
+        """Make everything written so far durable, in the order it was written."""
+        for key, value in self._journal:
+            if value is _DELETED:
+                self._data.pop(key, None)
+            else:
+                self._data[key] = value
+        self._journal.clear()
+        self._overlay.clear()
+
+    def _configure(self, *, buffered: bool, torn: bool) -> None:
+        # Whatever is queued belongs to the disk as it was configured when the
+        # writes were made, so it lands before the new configuration applies.
+        self.sync()
+        self._buffered = buffered
+        self._torn = torn
+
+    def _crash(self, rng: random.Random) -> None:
+        journal, self._journal = self._journal, []
+        self._overlay.clear()
+        if not journal:
+            # Nothing was in flight, so there is nothing to decide and the
+            # seeded stream stays where it was. An unbuffered disk is always
+            # here: it has never queued anything.
+            return
+        kept = rng.randint(0, len(journal)) if self._torn else 0
+        for key, value in journal[:kept]:
+            if value is _DELETED:
+                self._data.pop(key, None)
+            else:
+                self._data[key] = value
 
     def __getitem__(self, key: str) -> object:
+        if key in self._overlay:
+            value = self._overlay[key]
+            if value is _DELETED:
+                raise KeyError(key)
+            return value
         return self._data[key]
 
     def __setitem__(self, key: str, value: object) -> None:
-        self._data[key] = value
+        if not self._buffered:
+            self._data[key] = value
+            return
+        self._journal.append((key, value))
+        self._overlay[key] = value
 
     def __delitem__(self, key: str) -> None:
-        del self._data[key]
+        if not self._buffered:
+            del self._data[key]
+            return
+        self[key]  # a delete of what is not there raises, buffered or not
+        self._journal.append((key, _DELETED))
+        self._overlay[key] = _DELETED
 
     def __iter__(self) -> Iterator[str]:
-        return iter(self._data)
+        if not self._overlay:
+            return iter(self._data)
+        return self._merged()
+
+    def _merged(self) -> Iterator[str]:
+        # Durable keys hold the place they were first written, which is where
+        # a flush would leave them too; keys the queue invented follow in the
+        # order they were written.
+        for key in self._data:
+            if self._overlay.get(key, None) is not _DELETED:
+                yield key
+        for key, value in self._overlay.items():
+            if key not in self._data and value is not _DELETED:
+                yield key
 
     def __len__(self) -> int:
-        return len(self._data)
+        if not self._overlay:
+            return len(self._data)
+        return sum(1 for _ in self._merged())
 
 
 class Host:
@@ -299,6 +394,10 @@ def __init__(self, loop: SimLoop) -> None:
         # Fault decisions draw from their own seed-derived stream so they can
         # never perturb the scheduler's draws or the sim.* user streams.
         self._rng = random.Random(f"{loop.seed}:net")
+        # Torn writes draw from a stream of their own for the same reason: a
+        # run that tears a disk must make exactly the network's own draws it
+        # would have made without one.
+        self._disk_rng = random.Random(f"{loop.seed}:disk")
         self._hosts: dict[str, Host] = {}
         self._addresses: dict[str, str] = {}
         self._names: dict[str, str] = {}
@@ -474,6 +573,28 @@ def set_clock(self, name: str, *, offset: float) -> None:
         self._require_host(name)
         self._clock_offsets[name] = float(offset)
 
+    def set_disk(
+        self, name: str, *, buffered: bool = False, torn: bool = False
+    ) -> None:
+        """Choose how honest a host's storage is about when a write lands.
+
+        A buffered disk holds writes and deletes until ``sync()``: the host
+        reads them back straight away, a crash before the flush loses them,
+        and the reboot sees only what was synced. ``torn`` decides what a
+        crash does with the queue — dropped whole by default, or cut at a
+        seeded point, keeping a prefix of it, which is the failure a machine
+        that lost power mid-batch actually leaves behind. Tearing needs a
+        buffer to tear, and configuring a disk flushes whatever it was
+        already holding. Contents are untouched: this says how the disk
+        behaves from here, not what is on it.
+        """
+        self._require_host(name)
+        if torn and not buffered:
+            raise ValueError(
+                "torn=True needs buffered=True: an unbuffered write is already durable"
+            )
+        self.host(name).disk._configure(buffered=buffered, torn=torn)
+
     def clock_offset(self, name: str) -> float:
         self._require_host(name)
         return self._clock_offsets.get(name, 0.0)
@@ -827,7 +948,8 @@ def crash(self, name: str) -> None:
 
         A crashed machine sends no reset — peers see nothing at all, which is
         what makes crashes indistinguishable from partitions to the code
-        under test until a timeout says otherwise.
+        under test until a timeout says otherwise. A buffered disk loses
+        whatever it had not synced; see ``set_disk``.
         """
         self._require_host(name)
         if name == DRIVER:
@@ -835,6 +957,9 @@ def crash(self, name: str) -> None:
         if not self._alive[name]:
             raise ValueError(f"host {name!r} already crashed")
         self._alive[name] = False
+        disk = self._disks.get(name)
+        if disk is not None:
+            disk._crash(self._disk_rng)
         for task in list(self._tasks[name]):
             task.cancel()
         for key in [key for key in self._listeners if key[0] == name]:
diff --git a/tests/test_disk.py b/tests/test_disk.py
index 6ce1f03..2906458 100644
--- a/tests/test_disk.py
+++ b/tests/test_disk.py
@@ -2,6 +2,8 @@
 
 from __future__ import annotations
 
+import asyncio
+
 import pytest
 
 from simloop import SimLoop
@@ -52,3 +54,236 @@ def test_the_driver_has_a_disk_too() -> None:
     loop.net.host("driver").disk["x"] = 1
     assert loop.net.host("driver").disk["x"] == 1
     loop.close()
+
+
+def test_sync_is_a_no_op_on_an_unbuffered_disk() -> None:
+    loop = _network()
+    disk = loop.net.host("server").disk
+    disk["term"] = 7
+    disk.sync()  # application code calls it whether or not the disk buffers
+    loop.net.crash("server")
+    assert loop.net.host("server").disk["term"] == 7
+    loop.close()
+
+
+def test_a_buffered_write_is_not_durable_until_sync() -> None:
+    loop = _network()
+    loop.net.set_disk("server", buffered=True)
+    disk = loop.net.host("server").disk
+    disk["term"] = 7
+    loop.net.crash("server")
+    loop.net.restart("server")
+    assert "term" not in loop.net.host("server").disk
+    loop.close()
+
+
+def test_sync_makes_a_buffered_write_survive_the_crash() -> None:
+    loop = _network()
+    loop.net.set_disk("server", buffered=True)
+    disk = loop.net.host("server").disk
+    disk["term"] = 7
+    disk.sync()
+    disk["vote"] = "n2"  # written after the flush, so this one is lost
+    loop.net.crash("server")
+    loop.net.restart("server")
+    assert dict(loop.net.host("server").disk) == {"term": 7}
+    loop.close()
+
+
+def test_a_buffered_disk_reads_its_own_writes() -> None:
+    loop = _network()
+    loop.net.set_disk("server", buffered=True)
+    disk = loop.net.host("server").disk
+    disk["a"] = 1
+    disk.sync()
+    disk["a"] = 2
+    disk["b"] = 3
+    assert disk["a"] == 2
+    assert disk["b"] == 3
+    assert len(disk) == 2
+    loop.close()
+
+
+def test_a_pending_delete_hides_a_durable_key() -> None:
+    loop = _network()
+    loop.net.set_disk("server", buffered=True)
+    disk = loop.net.host("server").disk
+    disk["a"] = 1
+    disk.sync()
+    del disk["a"]
+    assert "a" not in disk
+    assert len(disk) == 0
+    with pytest.raises(KeyError):
+        disk["a"]
+    with pytest.raises(KeyError):
+        del disk["a"]
+    loop.net.crash("server")
+    loop.net.restart("server")
+    assert dict(loop.net.host("server").disk) == {"a": 1}  # the delete never landed
+    loop.close()
+
+
+def test_the_merged_view_iterates_durable_order_then_write_order() -> None:
+    loop = _network()
+    loop.net.set_disk("server", buffered=True)
+    disk = loop.net.host("server").disk
+    disk["a"] = 1
+    disk["b"] = 2
+    disk.sync()
+    disk["z"] = 26
+    disk["a"] = 10  # an overwrite keeps the key where the durable state has it
+    disk["c"] = 3
+    del disk["b"]
+    assert list(disk) == ["a", "z", "c"]
+    disk.sync()
+    assert list(disk) == ["a", "z", "c"]  # the flush leaves the order it showed
+    loop.close()
+
+
+def test_a_crash_with_nothing_pending_keeps_everything() -> None:
+    loop = _network()
+    loop.net.set_disk("server", buffered=True, torn=True)
+    disk = loop.net.host("server").disk
+    disk["a"] = 1
+    disk.sync()
+    loop.net.crash("server")
+    loop.net.restart("server")
+    assert dict(loop.net.host("server").disk) == {"a": 1}
+    loop.close()
+
+
+def _torn_prefix(seed: int, *, torn: bool = True, writes: int = 20) -> list[str]:
+    loop = SimLoop(seed=seed)
+    loop.net.host("server")
+    loop.net.set_disk("server", buffered=True, torn=torn)
+    disk = loop.net.host("server").disk
+    for i in range(writes):
+        disk[f"k{i}"] = i
+    loop.net.crash("server")
+    loop.net.restart("server")
+    survivors = list(loop.net.host("server").disk)
+    loop.close()
+    return survivors
+
+
+def test_a_torn_crash_keeps_a_prefix_of_the_journal() -> None:
+    survivors = _torn_prefix(0)
+    assert survivors == [f"k{i}" for i in range(len(survivors))]
+    assert 0 <= len(survivors) <= 20
+
+
+def test_the_same_seed_tears_at_the_same_place() -> None:
+    assert _torn_prefix(3) == _torn_prefix(3) == _torn_prefix(3)
+
+
+def test_different_seeds_tear_at_different_places() -> None:
+    lengths = {len(_torn_prefix(seed)) for seed in range(12)}
+    assert len(lengths) > 1
+
+
+def test_without_torn_a_crash_keeps_nothing_pending() -> None:
+    assert _torn_prefix(3, torn=False) == []
+
+
+def test_torn_writes_need_a_buffer_to_tear() -> None:
+    loop = _network()
+    with pytest.raises(ValueError):
+        loop.net.set_disk("server", torn=True)
+    with pytest.raises(OSError):
+        loop.net.set_disk("ghost", buffered=True)
+    loop.close()
+
+
+def test_reconfiguring_flushes_what_was_pending() -> None:
+    loop = _network()
+    loop.net.set_disk("server", buffered=True)
+    disk = loop.net.host("server").disk
+    disk["a"] = 1
+    loop.net.set_disk("server", buffered=False)
+    assert not disk.buffered
+    loop.net.crash("server")
+    loop.net.restart("server")
+    assert dict(loop.net.host("server").disk) == {"a": 1}
+    loop.close()
+
+
+def test_a_configured_disk_keeps_what_it_already_held() -> None:
+    loop = _network()
+    loop.net.host("server").disk["a"] = 1
+    loop.net.set_disk("server", buffered=True, torn=True)
+    disk = loop.net.host("server").disk
+    assert disk["a"] == 1
+    assert (disk.buffered, disk.torn) == (True, True)
+    loop.close()
+
+
+def _traced(configure: bool) -> str:
+    loop = SimLoop(seed=1)
+    loop.net.host("server")
+    if configure:
+        loop.net.set_disk("server", buffered=True)
+
+    async def main() -> None:
+        disk = loop.net.host("server").disk
+        for i in range(5):
+            disk[f"k{i}"] = i
+            await asyncio.sleep(0.1)
+        disk.sync()
+
+    try:
+        loop.run_until_complete(loop.net.host("server").create_task(main()))
+        return loop.trace_hash()
+    finally:
+        loop.close()
+
+
+def test_buffering_a_disk_changes_no_scheduling_decision() -> None:
+    # Storage is not a scheduling event: the buffer changes when a value
+    # becomes durable and nothing about what ran when.
+    assert _traced(False) == _traced(True)
+
+
+def test_a_buffered_crash_leaves_the_network_draws_alone() -> None:
+    def run(buffered: bool) -> tuple[str, list[str]]:
+        loop = SimLoop(seed=5)
+        loop.net.host("server")
+        loop.net.host("client")
+        loop.net.set_defaults(latency=(0.01, 0.05), drop=0.2, duplicate=0.2)
+        if buffered:
+            loop.net.set_disk("server", buffered=True, torn=True)
+
+        async def main() -> None:
+            class Echo(asyncio.DatagramProtocol):
+                def datagram_received(self, data: bytes, addr: object) -> None:
+                    seen.append(data)
+
+            seen: list[bytes] = []
+            await loop.net.host("server").create_task(
+                loop.create_datagram_endpoint(Echo, local_addr=("server", 9000))
+            )
+            transport, _ = await loop.net.host("client").create_task(
+                loop.create_datagram_endpoint(
+                    asyncio.DatagramProtocol,
+                    local_addr=("client", 9001),
+                    remote_addr=("server", 9000),
+                )
+            )
+            disk = loop.net.host("server").disk
+            for i in range(10):
+                disk[f"k{i}"] = i
+                transport.sendto(f"{i}".encode())
+                await asyncio.sleep(0.1)
+            loop.net.crash("server")
+            await asyncio.sleep(0.5)
+            arrived.extend(d.decode() for d in seen)
+
+        arrived: list[str] = []
+        try:
+            loop.run_until_complete(main())
+            return loop.trace_hash(), arrived
+        finally:
+            loop.close()
+
+    # The torn prefix draws from the disk's own seed-derived stream, so which
+    # datagrams the network dropped is the same either way.
+    assert run(False) == run(True)