Your batch script is 3,200 files into 5,000 when the SSH session drops, the laptop goes to sleep, or the OOM killer steps in. What state did it leave behind? And what happens when you run it again?
For most scripts the honest answer is "I don't know", and that's the problem. A job that can't be safely re-run turns every crash into manual cleanup.
This post makes a script safe to kill at any moment, then proves it with a test that kills it at random moments and checks what's left. The complete code is at the end, and every number below comes from running it.
Three ways a batch job dies badly
A crash leaves behind one or more of these:
Problem | What you see | Fix |
|---|---|---|
Torn output | A half-written file that looks real | Atomic writes |
Lost progress | The restart begins again at item 1 | A checkpoint |
Repeated side effects | The customer gets two emails | Idempotent steps |
The three fixes work together, and the order matters. We'll take them one at a time.
How the safe version works
flowchart TD
S["Start the run"] --> C["Delete stale temp files"]
C --> L{"Next item?"}
L -- "None left" --> D["Done"]
L -- "Yes" --> E{"Output file exists?"}
E -- "Yes" --> L
E -- "No" --> W["Do the work"]
W --> X["Side effect, with an idempotency key"]
X --> T["Write the output to a temp file"]
T --> R["Rename the temp file into place"]
R --> LA kill can land anywhere in that loop. Here is what each moment leaves behind:
Before the side effect: nothing happened yet, so the restart simply does the item.
After the side effect, before the rename: the effect is recorded under its key, so repeating it does nothing. A temp file may be left over, and the next run deletes it.
After the rename: the item is complete, and the restart skips it.
No moment leaves a half-finished file where a real output should be.
Fix 1: atomic writes
The rule is simple: never write to the final path. Write a temp file next to it, then rename it into place. On POSIX systems a rename within one filesystem is atomic, so anything reading the path sees the old file or the complete new one, never half of one.
def atomic_write(path, write):
"""Call write(f) on a temp file next to `path`, then swap it into place in one step.
Readers (and a restarted job) see either the old file or the complete new one,
never half of it. The temp file must be in the same directory so the rename stays
on one filesystem."""
path = Path(path)
tmp = path.with_name(f".{path.name}.{os.getpid()}.tmp")
try:
with open(tmp, "xb") as f: # "x" fails loudly if a stale temp file is in the way
write(f)
f.flush()
os.fsync(f.fileno())
os.replace(tmp, path)
fsync_dir(path.parent)
except BaseException:
tmp.unlink(missing_ok=True)
raiseA few details in there matter:
Same directory. The temp file must sit next to the target. A rename across filesystems fails, and a copy-then-delete fallback is not atomic.
flushandfsyncbefore the rename. Otherwise the rename can survive a power cut while the data doesn't, which leaves an empty file.fsyncon the directory. This makes the rename itself durable. It's skipped on Windows, where it isn't possible."xb"mode. It fails loudly if a stale temp file is in the way instead of silently reusing it.except BaseException. Ctrl+C removes the temp file too.
One honest caveat. The rename protects you from a killed process. Surviving a power cut also depends on the fsync calls, and on macOS fsync has historically not forced data out of the drive's own cache (fcntl.F_FULLFSYNC does). Check your platform's guarantees if power loss matters to you.
Fix 2: the outputs are the checkpoint
Once writes are atomic, "the output file exists" means "this item is complete". That's a checkpoint with no extra file to maintain. On restart the loop skips finished items and picks up where it stopped.
There is a trap here. "Skip it if the file exists" is the obvious cure for restarting from scratch, and without atomic writes it's worse than having no cure. A half-written file exists, so the restart skips it and keeps the corrupt data without telling you. The kill test below shows this happening.
When results aren't files, such as rows in a database, keep the checkpoint in SQLite: a table of finished item ids. SQLite is crash-safe itself, because its journal rolls back a transaction that was cut off.
Fix 3: make side effects safe to repeat
Ordering alone can't give you "exactly once". If the side effect happens before the checkpoint, a crash in between repeats it on restart. If the checkpoint comes first, a crash in between means it never happens.
The practical answer is to accept "at least once" and make repeating harmless. Give each effect a key, here the item id, and let the receiver ignore a key it has already seen. In the demo that's a SQLite row with the id as primary key and INSERT OR IGNORE.
For an external service such as email or payments, look for an idempotency-key feature and send the item id as the key. If the service has none, keep your own ledger of keys already sent.
Here is the worker that puts the three fixes together:
def run_worker(root, mode):
root = Path(root)
out = root / "outputs"
out.mkdir(exist_ok=True)
db = sqlite3.connect(root / "effects.db") # stands in for an email service, a payment API...
key = " PRIMARY KEY" if mode == "safe" else ""
db.execute(f"CREATE TABLE IF NOT EXISTS effects (item INTEGER{key})")
db.commit()
if mode == "safe":
for stale in out.glob(".*.tmp"): # left behind by a run that was killed mid-write
stale.unlink()
for src in sorted((root / "inputs").glob("item-*.txt")):
i = int(src.stem.split("-")[1])
dest = out / f"item-{i:04d}.json"
if mode != "naive" and dest.exists():
continue # safe to trust only if every write is atomic
with open(root / "attempts.log", "a") as log: # instrumentation for the test, not part of the pattern
log.write(f"{i}\n")
data = process(root, i)
time.sleep(0.003)
verb = "INSERT OR IGNORE" if mode == "safe" else "INSERT"
db.execute(f"{verb} INTO effects (item) VALUES (?)", (i,)) # the side effect comes first
db.commit()
if mode == "safe":
atomic_write(dest, lambda f: slow_write(f, data))
else:
with open(dest, "wb") as f:
slow_write(f, data)The kill test
A test for crash safety has to crash. The harness starts the job as a subprocess, waits a random 0.15 to 0.9 seconds, kills it with SIGKILL so it can't clean up, and repeats. Then it does one clean, uninterrupted run and checks what's left:
missing: outputs that don't exist
corrupt: outputs whose bytes differ from a recomputed expectation
duplicate effects: side effects that fired more than once for an item
redone items: items attempted more times than there are items
leftover temp files: stale
.tmpfiles
It runs the job three ways. naive re-does everything on every run and writes in place. skip skips items whose output exists but still writes in place. safe uses everything above.
To make the write window easy to hit, each 256 KB output is written in 32 KB chunks with a 4 ms pause, like a slow disk or network share. That's a simulation, and the post says so on purpose.
python crashsafe_demo.py killtest100 items, 10 kills at random moments, then one clean run
mode missing corrupt duplicate effects redone items leftover temp files
naive 0 0 101 102 0
skip 0 5 0 4 0
safe 0 0 0 10 0
safe mode: PASSYour numbers will differ on every run, because the kills land at random moments. The pattern won't:
naive: no corrupt files, because the final clean run rewrote everything. But 101 extra side effects fired and 102 items were redone. Every kill threw away all progress.
skip: 5 outputs exist, look finished, and hold wrong data. Nothing complained. Only the verifier noticed.
safe: zeros everywhere. The 10 redone items are the one item that was in flight at each of the 10 kills. They were redone, but not duplicated, because the key made the repeated side effect a no-op.
Running safe mode four more times with 25 kills each and different random seeds also passed every time.
What the test doesn't prove
Power loss. SIGKILL simulates a dead process, not a dead machine. The
fsynccalls are there for the second case, but this test can't confirm them.Two copies running at once. The start-up cleanup deletes temp files, which would break a second instance that is mid-write. Use a lock the OS drops when the process dies, such as
fcntl.flock. A PID file goes stale after a crash.Changed code. Skipping by existence assumes the output depends only on the input. If you change the logic, delete the old outputs or put a version in the filename.
Jobs that write several files per item. Each rename is atomic on its own, but the set isn't. Write a small manifest or
donemarker last, and treat that as the checkpoint.Speed. An
fsyncper file adds up over thousands of tiny files. Measure before you decide to batch them.
A checklist for your own script
Find every place the job writes a file, and switch it to
atomic_write.Decide what "done" means for one item, and make it observable: an output file or a database row.
Give every side effect a key so that repeating it is harmless.
Delete stale temp files when the job starts.
Write a kill test: random kills, one clean run, then a verifier that recomputes what the outputs should be.
The complete code
Save this as crashsafe_demo.py. It's about 190 lines of standard-library Python and needs Python 3.8 or newer.
#!/usr/bin/env python3
"""
crashsafe_demo - three ways to run a batch job, and a kill test that tells them apart.
python crashsafe_demo.py killtest # run all three modes, print a comparison
python crashsafe_demo.py killtest --mode safe # just one mode
python crashsafe_demo.py killtest --kills 30 # be meaner
Modes:
naive re-does everything on every run, writes outputs in place
skip skips items whose output file exists, but still writes in place
safe atomic writes + idempotent side effects (the pattern this post is about)
Standard library only. The kill test sends SIGKILL (TerminateProcess on Windows) at random
moments. That simulates a crashed or killed process, not a power cut.
"""
import argparse
import hashlib
import json
import os
import random
import shutil
import sqlite3
import subprocess
import sys
import tempfile
import time
from pathlib import Path
CHUNK = 32 * 1024
CHUNK_DELAY = 0.004 # pretend the disk or network share is slow, so the write window is easy to hit
MODES = ("naive", "skip", "safe")
# ---------------------------------------------------------------- building blocks
def fsync_dir(directory):
"""Make the rename itself durable. Not possible on Windows, so skip it there."""
if os.name == "nt":
return
fd = os.open(directory, os.O_RDONLY)
try:
os.fsync(fd)
finally:
os.close(fd)
def atomic_write(path, write):
"""Call write(f) on a temp file next to `path`, then swap it into place in one step.
Readers (and a restarted job) see either the old file or the complete new one,
never half of it. The temp file must be in the same directory so the rename stays
on one filesystem."""
path = Path(path)
tmp = path.with_name(f".{path.name}.{os.getpid()}.tmp")
try:
with open(tmp, "xb") as f: # "x" fails loudly if a stale temp file is in the way
write(f)
f.flush()
os.fsync(f.fileno())
os.replace(tmp, path)
fsync_dir(path.parent)
except BaseException:
tmp.unlink(missing_ok=True)
raise
def slow_write(f, data):
for i in range(0, len(data), CHUNK):
f.write(data[i:i + CHUNK])
f.flush()
time.sleep(CHUNK_DELAY)
# ---------------------------------------------------------------- the job
def make_inputs(root, n):
inputs = Path(root) / "inputs"
inputs.mkdir(parents=True, exist_ok=True)
for i in range(1, n + 1):
(inputs / f"item-{i:04d}.txt").write_text(f"item {i} " * 50)
def process(root, i):
"""The 'work': deterministic, so a verifier can recompute what every output must contain."""
text = (Path(root) / "inputs" / f"item-{i:04d}.txt").read_text()
digest = hashlib.sha256(text.encode()).hexdigest()
result = {"item": i, "sha256": digest, "words": len(text.split()), "payload": (digest * 4096)[:256 * 1024]}
return json.dumps(result, sort_keys=True).encode()
def run_worker(root, mode):
root = Path(root)
out = root / "outputs"
out.mkdir(exist_ok=True)
db = sqlite3.connect(root / "effects.db") # stands in for an email service, a payment API...
key = " PRIMARY KEY" if mode == "safe" else ""
db.execute(f"CREATE TABLE IF NOT EXISTS effects (item INTEGER{key})")
db.commit()
if mode == "safe":
for stale in out.glob(".*.tmp"): # left behind by a run that was killed mid-write
stale.unlink()
for src in sorted((root / "inputs").glob("item-*.txt")):
i = int(src.stem.split("-")[1])
dest = out / f"item-{i:04d}.json"
if mode != "naive" and dest.exists():
continue # safe to trust only if every write is atomic
with open(root / "attempts.log", "a") as log: # instrumentation for the test, not part of the pattern
log.write(f"{i}\n")
data = process(root, i)
time.sleep(0.003)
verb = "INSERT OR IGNORE" if mode == "safe" else "INSERT"
db.execute(f"{verb} INTO effects (item) VALUES (?)", (i,)) # the side effect comes first
db.commit()
if mode == "safe":
atomic_write(dest, lambda f: slow_write(f, data))
else:
with open(dest, "wb") as f:
slow_write(f, data)
# ---------------------------------------------------------------- the test
def verify(root, n):
root = Path(root)
out = root / "outputs"
missing = corrupt = 0
for i in range(1, n + 1):
f = out / f"item-{i:04d}.json"
if not f.exists():
missing += 1
elif f.read_bytes() != process(root, i):
corrupt += 1
db = sqlite3.connect(root / "effects.db")
duplicates = sum(c - 1 for (c,) in db.execute("SELECT COUNT(*) FROM effects GROUP BY item"))
lines = (root / "attempts.log").read_text().split() if (root / "attempts.log").exists() else []
return {"missing": missing, "corrupt": corrupt, "duplicate effects": duplicates,
"redone items": max(0, len([x for x in lines if x.isdigit()]) - n),
"leftover temp files": len(list(out.glob(".*.tmp")))}
def killtest(mode, n, kills, seed):
root = Path(tempfile.mkdtemp(prefix=f"killtest-{mode}-"))
try:
make_inputs(root, n)
rng = random.Random(seed)
for _ in range(kills):
proc = subprocess.Popen([sys.executable, __file__, "run", str(root), mode])
time.sleep(rng.uniform(0.15, 0.9))
proc.kill() # SIGKILL: the process gets no chance to clean up
proc.wait()
subprocess.run([sys.executable, __file__, "run", str(root), mode], check=True) # one clean run
return verify(root, n)
finally:
shutil.rmtree(root, ignore_errors=True)
def cmd_killtest(args):
modes = MODES if args.mode == "all" else (args.mode,)
rows = {m: killtest(m, args.items, args.kills, args.seed) for m in modes}
cols = list(next(iter(rows.values())))
print(f"{args.items} items, {args.kills} kills at random moments, then one clean run\n")
print(f"{'mode':<7}" + "".join(f"{c:>{len(c) + 2}}" for c in cols))
for m, r in rows.items():
print(f"{m:<7}" + "".join(f"{r[c]:>{len(c) + 2}}" for c in cols))
safe = rows.get("safe")
if safe:
ok = (not safe["missing"] and not safe["corrupt"] and not safe["duplicate effects"]
and not safe["leftover temp files"] and safe["redone items"] <= args.kills)
print("\nsafe mode:", "PASS" if ok else "FAIL")
return 0 if ok else 1
return 0
def main():
p = argparse.ArgumentParser(prog="crashsafe_demo", description=__doc__,
formatter_class=argparse.RawDescriptionHelpFormatter)
sub = p.add_subparsers(dest="cmd", required=True)
k = sub.add_parser("killtest", help="kill the job at random moments and check what is left")
k.add_argument("--mode", choices=("all",) + MODES, default="all")
k.add_argument("--items", type=int, default=100)
k.add_argument("--kills", type=int, default=10)
k.add_argument("--seed", type=int, default=1)
k.set_defaults(fn=cmd_killtest)
r = sub.add_parser("run") # used by the test harness
r.add_argument("root")
r.add_argument("mode", choices=MODES)
r.set_defaults(fn=lambda a: run_worker(a.root, a.mode))
args = p.parse_args()
return args.fn(args) or 0
if __name__ == "__main__":
sys.exit(main())Run the test with python crashsafe_demo.py killtest, or one mode at a time with --mode safe. The whole run takes about 20 seconds.
Comments(0)
Join the discussion by logging into your account.
No comments yet. Be the first to comment!