Vercel Drive fsync durability

#!/usr/bin/env python3
"""Does a Vercel Drive hold a file once fsync has returned?

Sandbox A mounts a Drive read-write, writes two fresh files (4 KiB and 64 MiB), fsyncs each file and the directory,
and reports each file's sha256. After a chosen delay, the script reads the Drive through a path that doesn't go
through sandbox A, and checks both files:

  snapshot       mount drive.snapshot() in a new sandbox B (A stays mounted and idle)
  snapshot-busy  the same, while A keeps writing other data to the Drive
  fork           drive.fork(), then mount the fork read-write in a new sandbox B
  stop           sandbox.stop() on A right after fsync, then mount the Drive read-write in a new sandbox B
                 (no delay sweep; records how long until the Drive can be mounted again)

Each check is "ok", "missing" or "corrupt". A file that is missing or corrupt after its fsync returned means that
path did not see data the writer had already fsynced.

Each trial runs in its own child process with a hard timeout, so a stuck SDK call costs one trial, not the run.

Requires: pip install vercel (the Vercel Python SDK); VERCEL_TOKEN, VERCEL_TEAM_ID and VERCEL_PROJECT_ID set.
usage:  python vercel-fsync-durability.py [--region pdx1] [--trials 10] [--delays 0,0.1,1,5,30]
                                          [--variants snapshot,snapshot-busy,fork,stop] [--out results.jsonl]
Everything the script creates (sandboxes, the drive, forks) is named fsync-<run>-* and deleted at the end.
"""
import argparse
import json
import secrets
import subprocess
import sys
import time
from collections import Counter

from vercel.sandbox import DriveQueryByName, SandboxQueryByName, SandboxResources, sync

MNT = "/data"
BIG_BYTES = 64 << 20
LIMIT_S = 4 * 3600
REREAD_AFTER_S = 60

# Payloads are a deterministic AES-CTR keystream per file id, so a reader can regenerate the expected bytes and say
# exactly which 1 MiB chunks differ when a file is corrupt.
GEN = "openssl enc -aes-128-ctr -nosalt -pbkdf2 -pass pass:$1-{k} < /dev/zero 2>/dev/null | head -c {n}"

WRITE = (
    GEN.format(k="small", n=4096) + " > /tmp/$1.small\n" + GEN.format(k="big", n=BIG_BYTES) + " > /tmp/$1.big\n"
    + r"""set -e
s=$(sha256sum < /tmp/$1.small | cut -c1-64)
b=$(sha256sum < /tmp/$1.big | cut -c1-64)
dd if=/tmp/$1.small of=/data/$1.small bs=4M conv=fsync,excl status=none
dd if=/tmp/$1.big of=/data/$1.big bs=4M conv=fsync,excl status=none
sync /data
rm -f /tmp/$1.small /tmp/$1.big
echo "FSYNCED $s $b"
"""
)

CHECK = r"""
for k in small big; do
  f=$2/$1.$k
  if [ -e "$f" ]; then echo "$k $(stat -c %s "$f") $(sha256sum < "$f" | cut -c1-64)"; else echo "$k MISSING"; fi
done
"""

# For a corrupt file: the exact byte ranges that differ from the expected bytes, and whether each range is all zeros.
DIFF = GEN.format(k="big", n=BIG_BYTES) + r""" > /tmp/want
python3 - "$2/$1.big" <<'PY'
import sys, json
want = open('/tmp/want', 'rb').read(); got = open(sys.argv[1], 'rb').read()
ranges, i, n = [], 0, min(len(want), len(got))
B = 4096
while i < n:
    if want[i:i+B] == got[i:i+B]:
        i += B
        continue
    j = i
    while j < n and want[j:j+B] != got[j:j+B]:
        j += B
    # Narrow both ends to the byte.
    a = i
    while want[a] == got[a]:
        a += 1
    b = min(j, n)
    while want[b-1] == got[b-1]:
        b -= 1
    ranges.append({"start": a, "end": b, "all_zero": got[a:b].count(0) == b - a})
    i = j
print("DIFF " + json.dumps({"size": len(got), "ranges": ranges[:20], "range_count": len(ranges),
                            "bytes_wrong": sum(r["end"] - r["start"] for r in ranges)}))
PY
"""

BUSY = "nohup setsid bash -c 'while :; do dd if=/dev/urandom of=/data/noise bs=1M count=16 conv=fsync status=none; done' >/dev/null 2>&1 &"


def sh(sb, script, *args, timeout=600):
    r = sb.run_process("bash", ["-c", script, "x", *args], sudo=True, capture_output=True, kill_after=timeout)
    return r.returncode, (r.stdout or "") + (r.stderr or "")


def new_sandbox(region, name, mounts, retry_s=60):
    t, last = time.monotonic(), None
    while True:
        try:
            sb = sync.create_sandbox(name=name, region=region, mounts=mounts, resources=SandboxResources(vcpus=2),
                                     execution_time_limit=LIMIT_S, persistent=False)
            return sb, time.monotonic() - t, last
        except Exception as e:  # noqa: BLE001
            last = str(e)[:200]
            if time.monotonic() - t > retry_s:
                raise
            time.sleep(1)


def drop(sb):
    for step in (sb.stop, sb.destroy):
        try:
            step()
        except Exception:  # noqa: BLE001
            pass


def check(sb, file_id, want, directory):
    _, out = sh(sb, CHECK, file_id, directory)
    result = {}
    for k in ("small", "big"):
        line = next((l for l in out.splitlines() if l.startswith(k + " ")), f"{k} MISSING")
        parts = line.split()
        result[k] = "missing" if parts[1] == "MISSING" else ("ok" if parts[2] == want[k] else "corrupt")
        if result[k] == "corrupt":
            result[f"{k}_size"] = int(parts[1])
    if result["big"] == "corrupt":
        diff = sh(sb, DIFF, file_id, directory)[1]
        line = next((l for l in diff.splitlines() if l.startswith("DIFF ")), None)
        result["big_diff"] = json.loads(line[5:]) if line else diff[-300:]
        # Does the same reader see the right bytes later? Drop its page cache so the re-read goes to the device.
        time.sleep(REREAD_AFTER_S)
        _, again = sh(sb, "sync; echo 3 > /proc/sys/vm/drop_caches; sha256sum < \"$2/$1.big\" | cut -c1-64", file_id, directory)
        result["big_reread_after_s"] = REREAD_AFTER_S
        result["big_reread"] = "ok" if want["big"] in again else "still corrupt"
    return result


def child(job):
    """One trial. Prints one RESULT line; for `stop` it also names the sandbox that now holds the Drive."""
    region, run, variant, delay, trial, writer, seq = (job[k] for k in ("region", "run", "variant", "delay", "trial", "writer", "seq"))
    drive = sync.get_drive(name_or_id=f"fsync-{run}")
    a_sb = sync.get_sandbox(name=writer)
    file_id = f"{run}-{variant}-{delay}-{trial}"
    code, out = sh(a_sb, WRITE, file_id)
    line = next((l for l in out.splitlines() if l.startswith("FSYNCED ")), None)
    if code != 0 or not line:
        raise RuntimeError(f"write failed ({code}): {out[-300:]}")
    _, s, b = line.split()
    want = {"small": s, "big": b}
    t_fsync, w_fsync = time.monotonic(), time.time()
    rec = {"variant": variant, "delay_s": delay, "trial": trial, "fsync_returned_utc": iso(w_fsync)}
    if variant == "stop":
        a_sb.stop()
        rec["stop_call_s"] = round(time.monotonic() - t_fsync, 3)
        b_sb, wait_s, last = new_sandbox(region, f"fsync-{run}-{seq}", {MNT: drive}, retry_s=1500)
        rec.update({"remount_after_stop_s": round(time.monotonic() - t_fsync, 3), "remount_wait_s": round(wait_s, 3),
                    "last_remount_error": last, "next_writer": b_sb.name})
        rec.update(check(b_sb, file_id, want, MNT))
    else:
        time.sleep(delay)
        rec["gap_fsync_to_request_s"] = round(time.monotonic() - t_fsync, 4)
        rec["request_utc"] = iso(time.time())
        rec["wall_gap_s"] = round(time.time() - w_fsync, 4)
        # Monotonic time stops while the orchestrating machine sleeps; a wall/monotonic mismatch flags the trial.
        rec["orchestrator_slept"] = rec["wall_gap_s"] - rec["gap_fsync_to_request_s"] > 1
        t_call = time.monotonic()
        if variant == "fork":
            # A fork call can hit the SDK's HTTP read timeout; retry under a new name (cleanup removes any orphan).
            for attempt in range(3):
                try:
                    mount = drive.fork(name=f"fsync-{run}-f{seq}-{attempt}")
                    break
                except Exception as e:  # noqa: BLE001
                    rec[f"fork_error_{attempt}"] = str(e)[:120]
                    if attempt == 2:
                        raise
        else:
            mount = drive.snapshot()
        rec["fork_or_snapshot_call_s"] = round(time.monotonic() - t_call, 3)
        b_sb, create_s, _ = new_sandbox(region, f"fsync-{run}-{seq}", {"/check": mount})
        rec["read_sandbox_create_s"] = round(create_s, 3)
        rec.update(check(b_sb, file_id, want, "/check"))
        drop(b_sb)
    print("RESULT " + json.dumps(rec), flush=True)


def iso(t):
    return time.strftime("%Y-%m-%dT%H:%M:%S", time.gmtime(t)) + f".{int(t % 1 * 1000):03d}Z"


def cleanup(run):
    for sb in sync.query_sandboxes(query=SandboxQueryByName(name_prefix=f"fsync-{run}")):
        drop(sb)
    for _ in range(60):
        drives = list(sync.query_drives(query=DriveQueryByName(name_prefix=f"fsync-{run}")))
        if not drives:
            return
        for d in drives:
            try:
                sync.get_drive(name_or_id=d.name).delete()
            except Exception:  # noqa: BLE001
                pass
        time.sleep(5)


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--region", default="pdx1")
    ap.add_argument("--trials", type=int, default=10)
    ap.add_argument("--delays", default="0,0.1,1,5,30")
    ap.add_argument("--variants", default="snapshot,snapshot-busy,fork,stop")
    ap.add_argument("--out", default="vercel-fsync-durability.jsonl")
    ap.add_argument("--child", help=argparse.SUPPRESS)
    a = ap.parse_args()
    if a.child:
        return child(json.loads(a.child))

    run = secrets.token_hex(3)
    delays = [float(d) for d in a.delays.split(",")]
    out = open(a.out, "a")
    tally = Counter()
    seq = 0
    try:
        drive, _ = sync.get_or_create_drive(name=f"fsync-{run}", region=a.region)
        writer, _, _ = new_sandbox(a.region, f"fsync-{run}-a", {MNT: drive})
        writer_name = writer.name
        for variant in a.variants.split(","):
            if variant == "snapshot-busy":
                sh(writer, BUSY)
            for delay in ([None] if variant == "stop" else delays):
                for trial in range(a.trials):
                    seq += 1
                    job = {"region": a.region, "run": run, "variant": variant, "delay": delay, "trial": trial,
                           "writer": writer_name, "seq": seq}
                    try:
                        p = subprocess.run([sys.executable, __file__, "--child", json.dumps(job)], capture_output=True,
                                           text=True, timeout=(delay or 0) + REREAD_AFTER_S + (1800 if variant == "stop" else 600))
                        line = next((l for l in p.stdout.splitlines() if l.startswith("RESULT ")), None)
                        rec = json.loads(line[7:]) if line else {"variant": variant, "delay_s": delay, "trial": trial,
                                                                 "error": (p.stderr or p.stdout)[-300:]}
                    except subprocess.TimeoutExpired:
                        rec = {"variant": variant, "delay_s": delay, "trial": trial, "error": "trial timed out"}
                    rec = {"run": run, "t": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), **rec}
                    out.write(json.dumps(rec) + "\n")
                    out.flush()
                    print(json.dumps(rec), flush=True)
                    for k in ("small", "big"):
                        tally[(variant, delay, k, rec.get(k, "error"))] += 1
                    if rec.get("next_writer"):
                        writer_name = rec["next_writer"]
                        writer = sync.get_sandbox(name=writer_name)
                    elif variant == "stop" and "error" in rec:
                        raise RuntimeError("stop trial failed; the Drive's writer is unknown")
            if variant == "snapshot-busy":
                sh(writer, "pkill -f 'of=/data/noise' || true")
    finally:
        cleanup(run)
        print("\nvariant        delay_s file  result     count")
        for (variant, delay, k, res), n in sorted(tally.items(), key=lambda x: str(x)):
            print(f"{variant:14s} {str(delay):7s} {k:5s} {res:10s} {n}")


if __name__ == "__main__":
    sys.exit(main())