Pin Replace Failures and Temp Leftovers Before One Atomic Write Extract

Do not extract an atomic writer from mixed batch code. Freeze leftover temps, replace errors, and exact bytes first. Characterization tests make the smallest safe change measurable.

Messy repos hide file writes inside business loops. A later crash then truncates the only good summary. This workflow pins observed write behavior before any helper extract.

The failure this workflow targets

A god function often lists inputs, scores rows, and writes JSON. The write still uses a plain Python open call. A crash during json.dump leaves a truncated file.

Readers then treat the truncated file as truth. Downstream jobs then ingest those partial counts as complete. Recovery becomes a guess about missing keys.

The extract you want is an atomic replace. Tests must exist before that extract lands.

Inventory the mixed write path

Walk the module with a single grep pass. Record every write site before touching structure.

1. List each write call with path, mode, and encoding.
2. Note whether the destination file already exists.
3. Note whether a temporary name is already used.
4. Note exception types the caller already catches.

rg -n "open\(|write_text\(|json.dump|os.replace|os.rename" batch_job.py

Keep this inventory in the test file docstring. Do not trust memory during the later extract.

Snapshot the messy write before moving it

Leave listing and scoring inside the original function. Treat the summary write as the only candidate. Label this module snippet as unexecuted sample code.

# batch_job.py — write still inlined with business flow
import json
from pathlib import Path


def run(inbox, outbox, summary_path):
    rows = list(Path(inbox).glob("*.csv"))
    summary = {"count": len(rows), "ok": True}
    with open(summary_path, "w", encoding="utf-8") as handle:
        json.dump(summary, handle)
        handle.write("\n")

The open(..., "w") call truncates on entry. A dump failure then destroys the previous good file. That is the behavior the harness must freeze.

Build a characterization harness

Create a temp inbox that never touches production disks. Drive the god function through its public entry. Assert output bytes, leftover files, and exception types.

Label the following tests as unexecuted sample code. Adapt names to the real module under change.

# test_atomic_write_chars.py
import json
from pathlib import Path

import pytest

from batch_job import run


def _write_old_summary(path: Path) -> bytes:
    old = {"ok": True, "count": 7}
    payload = (json.dumps(old) + "\n").encode("utf-8")
    path.write_bytes(payload)
    return payload


def test_run_replaces_summary_bytes(tmp_path: Path):
    inbox = tmp_path / "inbox"
    outbox = tmp_path / "outbox"
    inbox.mkdir()
    outbox.mkdir()
    (inbox / "a.csv").write_text("id\n1\n", encoding="utf-8")
    summary = tmp_path / "summary.json"
    old = _write_old_summary(summary)

    run(str(inbox), str(outbox), str(summary))

    new = summary.read_bytes()
    assert new != old
    data = json.loads(new.decode("utf-8"))
    assert data["count"] == 1
    assert new.endswith(b"\n")

Pin leftover scans on the same public entry. Do not treat a green byte assert as complete.

def test_run_leaves_no_tmp_siblings(tmp_path: Path):
    inbox = tmp_path / "inbox"
    outbox = tmp_path / "outbox"
    inbox.mkdir()
    outbox.mkdir()
    (inbox / "a.csv").write_text("id\n1\n", encoding="utf-8")
    summary = tmp_path / "summary.json"

    run(str(inbox), str(outbox), str(summary))

    siblings = sorted(p.name for p in tmp_path.iterdir() if p.name.startswith("summary.json"))
    assert siblings == ["summary.json"]


def test_run_keeps_old_bytes_on_dump_error(tmp_path: Path, monkeypatch):
    inbox = tmp_path / "inbox"
    outbox = tmp_path / "outbox"
    inbox.mkdir()
    outbox.mkdir()
    (inbox / "a.csv").write_text("id\n1\n", encoding="utf-8")
    summary = tmp_path / "summary.json"
    old = _write_old_summary(summary)

    def boom(*args, **kwargs):
        raise TypeError("unserializable")

    monkeypatch.setattr("batch_job.json.dump", boom)
    with pytest.raises(TypeError):
        run(str(inbox), str(outbox), str(summary))
    assert summary.read_bytes() == old

The dump-error test fails on a naive truncate-open. That red result is the needed characterization pin. Record the traceback before any helper extract.

Pin encoding as bytes

Text equality still hides BOM and newline translation. Read the destination with Path.read_bytes after every run. Compare against a literal byte string built in the test.

Default json.dumps will escape non-ASCII with ensure_ascii. If the messy module already dumps raw Unicode, pin that output. Changing ensure_ascii is a second behavior change.

python -c "from pathlib import Path; print(repr(Path('summary.json').read_bytes()))"

Store that repr in the test as the oracle. Human reviewers still miss a missing trailing newline.

Why leftover globs belong in the pin

Atomic helpers often pick unique names with mkstemp. Unique names break a naive glob on summary.json*. Pin the prefix, not a random suffix, when mkstemp is required.

Prefer dest.name plus a .tmp suffix when only one writer exists. Unique names need a directory scan for prefix matches. Record that policy in the test, not in a helper comment.

A passing content assert can still leak temps. Later jobs then glob the wrong sibling file. Leftover pins catch that class of extract mistake.

Decision table: extract or leave

Use this table after the harness is green or honestly red.

Observed signal	Extract now	Leave the write mixed
Leftover tmp files after a successful run	Yes, after names are pinned	No
Destination truncated on a dump error	Yes, if truncate is not a contract	No, if callers parse partial JSON
Write already uses a unique temp directory	Maybe, pin the directory first	Yes, if cleanup policy is undocumented
Caller catches selected OSError subclasses	Pin errno and filename first	Do not widen the except
Windows and POSIX paths are both required	Pin os.replace overwrite rules	Do not invent rename fallbacks

Extract only one helper in the first patch. Leave listing, scoring, and copy logic untouched. One behavior change per pull request keeps blame readable.

Smallest safe change

Add a private helper beside the god function. Keep the public entry signature completely frozen during the extract. Route only the summary bytes through the helper.

import os
from pathlib import Path


def _atomic_write_bytes(path: str, data: bytes) -> None:
    dest = Path(path)
    tmp = dest.with_name(dest.name + ".tmp")
    try:
        tmp.write_bytes(data)
        os.replace(tmp, dest)
    except Exception:
        if tmp.exists():
            tmp.unlink()
        raise

Call it with encoded JSON, not a live file object. Keep json.dumps inside the original caller function. That split preserves dump errors as dump errors.

payload = json.dumps(summary, separators=(",", ":")) + "\n"
_atomic_write_bytes(summary_path, payload.encode("utf-8"))

Re-run the three characterization tests after the helper lands. Leftover and truncation pins should now pass cleanly. If a new temp suffix appears, update the glob pin first.

Do not move json.dumps into the helper yet. Keep serializer options as a separate characterization pin. Mixing them hides which change fixed the truncation.

Commands that keep the extract honest

Run the narrow file, then the package, then a leftover hunt.

1. Execute the characterization file with quiet pytest output.
2. Execute the whole test package after the helper lands.
3. Grep the tree for extra tmp writers after review.

pytest -q test_atomic_write_chars.py
pytest -q
rg -n "\\.tmp|os\\.replace|write_bytes" -g "*.py"

Fail the change if grep finds a second writer. Two writers reintroduce truncation races under concurrent load. Keep a single suffix owned by the helper.

Where a free model session fits

Those tests must exist before any generated extract. A model without pins will invent rename fallbacks. Those fallbacks still differ across POSIX and Windows.

Disclosure: This article was prepared as part of MonkeyCode's product outreach. MonkeyCode offers free model access and a free server option. Use one session to draft the private helper against the frozen tests.

Reject any patch that widens the public run signature. Do not ask the session for hardware claims or speed numbers. Re-run pytest locally after the draft returns.

The harness is the contract, not the prompt text. If the helper changes leftover names, update pins first.

Limitations

Characterization tests freeze current bugs as well as features. If callers already parse truncated JSON, atomic replace is a behavior change. Split that behavior change from the structural extract.

The os.replace call overwrites on both POSIX and Windows. The os.rename call does not share that overwrite rule. Do not simplify the helper to rename calls.

Cross-device destination paths still fail hard at replace. This helper still does not copy across filesystems. Pin that OSError if production paths can span mounts.

Byte writes skip text newline translation on Windows. That newline skip remains intentional for this extract. If the messy code used text mode, pins must record CRLF.

Do not use this approach in the following cases.

1. Public APIs that promise streaming writes to callers.
2. Append-only logs that must survive a crash mid-line.
3. Untrusted path names taken from network input.
4. Binary formats with checksums you have not pinned.

Checklist before merge

1. Write-site inventory sits in the test docstring.
2. Byte identity, leftovers, and dump errors are pinned.
3. Only one private helper was introduced this patch.
4. Public run arguments remain unchanged after the extract.
5. Grep shows a single temporary suffix in the module.
6. Windows replace behavior is not inferred from Linux runs.

Stop there after the checklist is green. Do not extract listing or scoring in the same patch. The next pin starts from this green harness.