Skip to content

Tracking: long-lived read_only connection vs. concurrent writer activity — the lifetime half of reader/writer safety (#615 race #1) #666

Description

@M0nkeyFl0wer

Version: ladybug 0.18.0 (PyPI)
OS: Linux 6.11.0-29-generic (Ubuntu 24.04)

Summary

Tracking issue for "race #1" from the PR #615 discussion, filed at @adsharma's
request (comment
thread). #615 makes the open-time race fail closed (verified in #642). This
issue tracks the lifetime half: a read_only connection that opened
successfully and is then held while a writer process mutates and checkpoints
the same file. The maintainer-identified hazard: the reader's lock files are
released once WALReplayer::replay() returns, after which the reader queries
with no synchronization against the writer — checkpoint could shrink the file
under the reader's cached pages, or reuse pages the reader's open-time
metadata still points at.

Two contributions here: empirical results on stock 0.18.0 (the hazard did not
reproduce, and the file-shrink mechanism appears unreachable in this version),
and a repro harness for regression use when this is worked on.

Empirical status on 0.18.0: does not reproduce

Setup: 3 reader processes each open read_only once, hold the connection
for the whole run, and query in a tight loop; 1 writer process churns the same
file (open / mutate / CHECKPOINT / close per cycle). Seed: 200k rows with a
256-char payload. All children under ulimit -c 0, faulthandler enabled,
reader stats persisted every 200 queries so a crash cannot lose them. The
harness was validated to report -11 for a deliberately SIGSEGV'd child. The
parent samples the data file size at 1 Hz to detect physical shrinkage.

Exp Writer Reader query / buffer pool Writer work Result
A insert-churn (#642 writer) count(n), 128 MiB 349 checkpoints, 6,980 inserts clean
B delete 5k window + reinsert 5k per cycle count(n), 128 MiB 446 checkpoints, 1.1M deletes + 1.1M inserts clean
B-scan same full data scan, 128 MiB ~448 checkpoints clean
C delete 50% + CHECKPOINT + reinsert + CHECKPOINT full scan, 128 MiB 74 checkpoints, ~3.7M rows churned clean
D/D2 delete-window (every original row eventually deleted) full scan, 16 MiB pool (≪ file size, forces disk re-reads every scan) 414–454 checkpoints clean

Totals: ~1,700 writer checkpoints, ~760k statistics-only reader queries plus
~20k full data-page scans. Zero crashes, zero exceptions, zero wrong
results.
In D2, every one of 4,945 scans on every reader returned the exact
identical open-time tuple (200000, 0, 199999, 51200000) — a perfectly stable
snapshot even after the writer had deleted and re-created every row the reader
could see, across hundreds of checkpoints, with a buffer pool too small to
shield the reader from disk.

Why the named mechanism can't fire on 0.18.0

The data file never shrank in any run — monotonic growth only, even in the
50%-delete-then-checkpoint variant. Consistent with the source:
PageManager::reclaimTailPagesIfNeeded (src/storage/page_manager.cpp:71)
adds tail pages to the FreeSpaceManager for reuse, and I found no call
path that truncates the main data file (only the WAL is truncated). So
"reader's cached pages now beyond EOF" has no trigger in this version.

What this issue should track

  1. The overwritten-page variant remains untested-in-anger. This is a null
    result, not a proof: in these runs the writer's page reuse never landed new
    bytes on pages the readers' open-time metadata referenced (the file grew
    instead). A targeted adversarial test — a larger DB where the FSM
    demonstrably hands the writer the reader's hot pages, string overflow-page
    reuse, longer windows — would settle whether lifetime reads are actually
    copy-on-write-safe or just lucky under these workloads.
  2. Frozen-snapshot semantics should be documented if intended. A held
    read-only connection never observes any writer commit (counts stayed
    pinned at the open-time value across every experiment); cross-process
    visibility requires re-opening. Reasonable semantics for an embedded
    snapshot model — but it means "long-lived read-only connection" is also
    "arbitrarily stale connection," and today that contract is discoverable
    only by experiment. If intended, a docs paragraph closes this half.
  3. Regression guard for future file truncation. Checkpoint currently
    never returns disk space (delete-heavy workloads only grow the file). If a
    future version adds data-file truncation on checkpoint — a natural
    space-reclamation feature — the beyond-EOF hazard in this issue becomes
    live for held readers, and the harness below should run against that
    change.

Repro harness

repro.py below; orchestration is: seed, spawn 3 readers + 1 writer as
separate processes under ulimit -c 0 (e.g. bash -c 'ulimit -c 0; exec "$@"' _ python repro.py ...), wait, then check exit codes (-11 = SIGSEGV) and
the per-reader stats JSON (distinct result tuples, error counts). Happy to
re-run it against any future revision, same offer as the #642 harness.

#!/usr/bin/env python3
"""
Long-lived read-only connection vs. writer churn repro for LadybugDB.

Tests maintainer-identified gap #1 of PR #615: a reader that opens
read_only ONCE and then holds the connection while a separate writer
process churns (inserts / deletes + CHECKPOINT) the same DB file.

Subcommands:
  seed          --db PATH --rows N
  reader        --db PATH --duration S --stats FILE
  writer-insert --db PATH --duration S            (issue #642 writer: 20 inserts + CHECKPOINT per cycle)
  writer-delete --db PATH --duration S --window N (sliding-window DETACH DELETE + re-insert + CHECKPOINT)
  writer-half   --db PATH --duration S            (delete 50% then CHECKPOINT, re-seed, repeat)
"""
import argparse
import faulthandler
import json
import os
import sys
import time

faulthandler.enable()

PAYLOAD = "x" * 256
# reader buffer pool overridable so we can force it smaller than the DB,
# making every scan hit disk pages the writer may have overwritten
READER_BP = int(os.environ.get("READER_BP", 128 * 1024 * 1024))
WRITER_BP = 256 * 1024 * 1024


def seed(db_path, rows):
    import ladybug as lb
    db = lb.Database(db_path, buffer_pool_size=WRITER_BP)
    conn = lb.Connection(db)
    conn.execute("CREATE NODE TABLE IF NOT EXISTS N(id INT64, payload STRING, PRIMARY KEY(id))")
    conn.execute(f"UNWIND range(0, {rows - 1}) AS i CREATE (:N {{id: i, payload: '{PAYLOAD}'}})")
    conn.execute("CHECKPOINT")
    conn.close()
    db.close()
    print(f"seeded {rows} rows", flush=True)


def reader(db_path, duration, stats_path, scan=False):
    """Open read_only ONCE (retry until success), hold the connection,
    query in a tight loop for `duration` seconds.

    scan=False: plain `count(n)` (may be answered from table statistics).
    scan=True:  force a real data-page scan by touching every payload
                string (incl. overflow pages) and every id.
    """
    import ladybug as lb

    stats = {
        "pid": os.getpid(),
        "open_retries": 0,
        "open_errors": {},          # error-message-prefix -> count during open retry
        "queries_ok": 0,
        "query_errors": {},         # "ExcType: msg-prefix" -> count
        "min_count": None,
        "max_count": None,
        "absurd_values": [],        # counts that were 0/negative (list, capped)
        "opened": False,
    }

    def dump():
        with open(stats_path, "w") as f:
            json.dump(stats, f, indent=1)

    deadline = time.time() + duration

    # --- open ONCE, retrying on clean errors (the #642/#615-covered race) ---
    db = conn = None
    while time.time() < deadline:
        try:
            db = lb.Database(db_path, read_only=True, buffer_pool_size=READER_BP)
            conn = lb.Connection(db)
            stats["opened"] = True
            break
        except RuntimeError as e:
            stats["open_retries"] += 1
            key = str(e)[:120]
            stats["open_errors"][key] = stats["open_errors"].get(key, 0) + 1
            time.sleep(0.05)
    dump()
    if conn is None:
        print("reader: never managed to open", flush=True)
        sys.exit(3)

    # --- hold the connection open; tight query loop ---
    if scan:
        # forces reading every id and every payload string (overflow pages)
        query = ("MATCH (n:N) WHERE size(n.payload) >= 0 "
                 "RETURN count(n), min(n.id), max(n.id), sum(size(n.payload))")
    else:
        query = "MATCH (n:N) RETURN count(n)"
    n = 0
    while time.time() < deadline:
        try:
            res = conn.execute(query)
            row = res.get_next()
            cnt = row[0]
            if scan:
                # payload bytes must be consistent with the row count:
                # every payload is exactly 256 chars
                if row[3] != cnt * 256 and len(stats.setdefault("scan_anomalies", [])) < 20:
                    stats["scan_anomalies"].append(list(row))
                # a long-lived read-only connection has ONE immutable snapshot:
                # every scan must return the identical (cnt, min, max, sum) tuple.
                # >1 distinct tuple observed = the reader saw writer activity
                # (torn read / overwritten page / snapshot violation).
                tup = repr((row[0], row[1], row[2], str(row[3])))
                tuples = stats.setdefault("distinct_tuples", {})
                if tup in tuples or len(tuples) < 10:
                    tuples[tup] = tuples.get(tup, 0) + 1
            stats["queries_ok"] += 1
            if stats["min_count"] is None or cnt < stats["min_count"]:
                stats["min_count"] = cnt
            if stats["max_count"] is None or cnt > stats["max_count"]:
                stats["max_count"] = cnt
            if cnt <= 0 and len(stats["absurd_values"]) < 20:
                stats["absurd_values"].append(cnt)
        except Exception as e:
            key = f"{type(e).__name__}: {str(e)[:160]}"
            stats["query_errors"][key] = stats["query_errors"].get(key, 0) + 1
        n += 1
        if n % 200 == 0:
            dump()  # persist stats so a SIGSEGV doesn't lose them
    dump()
    conn.close()
    db.close()
    print(f"reader done: {stats['queries_ok']} ok, errors={sum(stats['query_errors'].values())}", flush=True)


def _open_writer(lb, db_path):
    db = lb.Database(db_path, buffer_pool_size=WRITER_BP)
    conn = lb.Connection(db)
    return db, conn


def writer_insert(db_path, duration):
    """Issue #642 writer: loop { open, 20 inserts, CHECKPOINT, close }."""
    import ladybug as lb
    deadline = time.time() + duration
    next_id = 10_000_000
    cycles = 0
    while time.time() < deadline:
        db, conn = _open_writer(lb, db_path)
        for _ in range(20):
            conn.execute(f"CREATE (:N {{id: {next_id}, payload: '{PAYLOAD}'}})")
            next_id += 1
        conn.execute("CHECKPOINT")
        conn.close()
        db.close()
        cycles += 1
        time.sleep(0.05)
    print(f"writer-insert done: {cycles} cycles, {20 * cycles} rows inserted", flush=True)


def writer_delete(db_path, duration, window):
    """Sliding-window delete + re-insert + CHECKPOINT, to churn pages and
    (hopefully) trigger tail-page reclaim on checkpoint."""
    import ladybug as lb
    deadline = time.time() + duration
    lo = 0
    next_id = 20_000_000
    cycles = 0
    while time.time() < deadline:
        db, conn = _open_writer(lb, db_path)
        hi = lo + window
        conn.execute(f"MATCH (n:N) WHERE n.id >= {lo} AND n.id < {hi} DETACH DELETE n")
        # re-insert a similar amount at fresh ids
        conn.execute(
            f"UNWIND range({next_id}, {next_id + window - 1}) AS i "
            f"CREATE (:N {{id: i, payload: '{PAYLOAD}'}})"
        )
        next_id += window
        conn.execute("CHECKPOINT")
        conn.close()
        db.close()
        lo = hi
        cycles += 1
        time.sleep(0.05)
    print(f"writer-delete done: {cycles} cycles of {window} delete+insert", flush=True)


def writer_half(db_path, duration, rows):
    """Delete 50% of rows then CHECKPOINT (forcing maximal page freeing /
    tail reclaim), then re-seed the deleted half and CHECKPOINT again."""
    import ladybug as lb
    deadline = time.time() + duration
    cycles = 0
    while time.time() < deadline:
        db, conn = _open_writer(lb, db_path)
        # delete the top half of the id space currently present
        res = conn.execute("MATCH (n:N) RETURN max(n.id), count(n)")
        max_id, cnt = res.get_next()
        half = cnt // 2
        conn.execute(f"MATCH (n:N) WITH n ORDER BY n.id DESC LIMIT {half} DETACH DELETE n")
        conn.execute("CHECKPOINT")
        # re-insert the same number of rows at fresh ids
        base = max_id + 1
        conn.execute(
            f"UNWIND range({base}, {base + half - 1}) AS i "
            f"CREATE (:N {{id: i, payload: '{PAYLOAD}'}})"
        )
        conn.execute("CHECKPOINT")
        conn.close()
        db.close()
        cycles += 1
        time.sleep(0.05)
    print(f"writer-half done: {cycles} cycles", flush=True)


def main():
    p = argparse.ArgumentParser()
    p.add_argument("mode", choices=["seed", "reader", "writer-insert", "writer-delete", "writer-half"])
    p.add_argument("--db", required=True)
    p.add_argument("--rows", type=int, default=200_000)
    p.add_argument("--duration", type=float, default=75.0)
    p.add_argument("--stats", default="/dev/null")
    p.add_argument("--window", type=int, default=5000)
    p.add_argument("--scan", action="store_true")
    a = p.parse_args()

    if a.mode == "seed":
        seed(a.db, a.rows)
    elif a.mode == "reader":
        reader(a.db, a.duration, a.stats, scan=a.scan)
    elif a.mode == "writer-insert":
        writer_insert(a.db, a.duration)
    elif a.mode == "writer-delete":
        writer_delete(a.db, a.duration, a.window)
    elif a.mode == "writer-half":
        writer_half(a.db, a.duration, a.rows)


if __name__ == "__main__":
    main()

Related

Metadata

Metadata

Assignees

No one assigned

    Labels

    No labels
    No labels

    Type

    No type

    Fields

    No fields configured for issues without a type.

    Projects

    No projects

    Milestone

    No milestone

    Relationships

    None yet

    Development

    No branches or pull requests

    Issue actions