Multi-writer coordination: optimistic locking, conflicts, and retries

An h5i-db database is a directory, and nothing stops two processes from opening it at the same time: a feed handler and a corrections job, or two teammates' notebooks.

The answer to concurrent writers is optimistic concurrency. Every commit can carry an expected_version, and if the table head has moved since you read it, the commit is rejected with an explicit ConflictError instead of silently interleaving.

That is the right trade for teams. A lost update in a positions or marks table is a silent P&L error. A ConflictError is a retry.

In this recipe we:

simulate two writers with two Database handles on the same path,
scale up to three threads racing to ingest a chunked feed,
hit the same conflict machinery through the plan/apply mutation flow.

import threading

import pyarrow as pa

import h5i_db
import cookbook_utils as cu

path = cu.fresh_db("prod_writers")
writer_a = h5i_db.Database(path, create=True)

1. The data#

cu.make_trades gives one session of ticks for three names, our "feed". One row per print.

column	type	meaning
`ts`	`timestamp[us, tz=UTC]`	trade timestamp, ascending
`symbol`	`string`	ticker
`price`	`float64`	trade price
`size`	`int64`	shares traded
`exchange`	`string`	reporting venue
`side`	`string`	`B` buyer-initiated, `S` seller-initiated

Sections below carve it into time-contiguous chunks. The table is sorted by ts, so slices are windows.

feed = cu.make_trades(symbols=["AAPL", "MSFT", "NVDA"], days=1, trades_per_day=8_000, seed=7)
print(f"feed: {feed.num_rows:,} rows x {feed.num_columns} columns")
feed.to_pandas().head()

output

feed: 21,180 rows x 6 columns

	ts	symbol	price	size	exchange	side
0	2026-06-01 13:30:00.138654+00:00	NVDA	319.22	100	BATS	S
1	2026-06-01 13:30:00.858361+00:00	AAPL	265.00	1	BATS	S
2	2026-06-01 13:30:01.734996+00:00	MSFT	362.96	1	NYSE	S
3	2026-06-01 13:30:01.982844+00:00	MSFT	363.05	1	NYSE	B
4	2026-06-01 13:30:01.996525+00:00	AAPL	265.12	600	BATS	B

schema = pa.schema(
    [
        pa.field("ts", pa.timestamp("us", tz="UTC"), nullable=False),
        pa.field("symbol", pa.string()),
        pa.field("price", pa.float64()),
        pa.field("size", pa.int64()),
        pa.field("exchange", pa.string()),
        pa.field("side", pa.string()),
    ]
)
writer_a.create_table("trades", schema, time_column="ts", sort_key=["ts", "symbol"])

output

{'table': 'trades',
 'sequence': 0,
 'op': 'create',
 'rows_total': 0,
 'segments_total': 0,
 'segments_added': 0,
 'segments_deduped': 0,
 'committed_at_ns': 1785195773830604369}

2. Two handles, one database#

writer_b opens the same directory. Think of it as a second process.

Commits made through one handle are immediately visible through the other. Both handles read the same manifest on disk, and there is no per-handle cache of the table head to go stale.

writer_b = h5i_db.Database(path)  # no create: the db already exists

writer_a.append("trades", feed.slice(0, 3_000), note="A: chunk 0")
head_a = writer_a.versions("trades")[-1]["sequence"]
head_b = writer_b.versions("trades")[-1]["sequence"]
print(f"head seen by A: v{head_a}, by B: v{head_b}")

output

head seen by A: v1, by B: v1

3. A stale `expected_version` fails loudly#

B reads the head at v1 and plans to append. Before it does, A commits another chunk.

B's append(..., expected_version=1) is now a compare-and-swap against a head that no longer exists, and h5i-db rejects it.

The error is machine-readable: .code for dispatch, .retryable telling you this is safe to retry, .hint for the operator.

head_b_saw = writer_b.versions("trades")[-1]["sequence"]  # B reads: v1

writer_a.append("trades", feed.slice(3_000, 1_500), note="A: chunk 1")  # head -> v2

try:
    writer_b.append("trades", feed.slice(4_500, 1_500), expected_version=head_b_saw)
except h5i_db.ConflictError as e:
    print(f"code      = {e.code}")
    print(f"retryable = {e.retryable}")
    print(f"hint      = {e.hint}")

output

code      = version_conflict
retryable = True
hint      = re-read the head of "trades" and retry against it; pure appends rebase safely (the CLI and Python bindings already auto-retry those)

Nothing was written. B's rows are not in the table, and the version chain is untouched.

Contrast this with "last writer wins" storage such as plain Parquet directories or CSV drops. There, B's write would have clobbered or interleaved with A's, and nobody would know until reconciliation.

print(f"rows in table: {len(writer_b.read('trades')):,}  (B's 1,500 rows were NOT committed)")

output

rows in table: 4,500  (B's 1,500 rows were NOT committed)

4. The retry pattern#

Since retryable=True, the fix is mechanical. Re-read the head, re-issue the append against it, and give up after a few attempts.

This is the same CAS-retry loop you would write against any optimistic-concurrency store.

def append_with_retry(handle, table, data, note=None, max_attempts=5):
    """Append with optimistic locking; retry on conflict."""
    for attempt in range(1, max_attempts + 1):
        head = handle.versions(table)[-1]["sequence"]
        try:
            commit = handle.append(table, data, expected_version=head, note=note)
            return commit, attempt
        except h5i_db.ConflictError:
            continue  # head moved between read and commit - re-read and retry
    raise RuntimeError(f"gave up after {max_attempts} attempts")


commit, attempts = append_with_retry(writer_b, "trades", feed.slice(4_500, 1_500), note="B: chunk 2 (retried)")
print(f"B committed v{commit['sequence']} on attempt {attempts}; rows_total={commit['rows_total']:,}")

output

B committed v3 on attempt 1; rows_total=6,000

5. Three threads racing on one feed#

Now the stress test. The rest of the feed is split into 9 time-contiguous chunks, and three writer threads, each with its own Database handle, race to ingest them.

Coordination happens entirely through the CAS. Each thread derives which chunk is next from the committed head sequence, and appends it with expected_version=head.

When two threads pick the same chunk, exactly one commit lands. The loser gets ConflictError, re-reads the head, and moves on to the next chunk. No locks and no queue: the version chain is the queue.

N_CHUNKS = 9
base_rows = 6_000  # rows already committed in sections 1-3
late_rows = 1_500  # held back for section 5
chunk_rows = (len(feed) - base_rows - late_rows) // N_CHUNKS
chunks = [feed.slice(base_rows + i * chunk_rows, chunk_rows) for i in range(N_CHUNKS)]
thread_rows = sum(len(c) for c in chunks)
print(f"{N_CHUNKS} chunks x {chunk_rows:,} rows for the race")

base_seq = writer_a.versions("trades")[-1]["sequence"]
rows_before = len(writer_a.read("trades"))
stats = {}


def feed_worker(name: str) -> None:
    handle = h5i_db.Database(path)  # per-thread handle, like a separate process
    wins = conflicts = 0
    try:
        while True:
            head = handle.versions("trades")[-1]["sequence"]
            next_chunk = head - base_seq  # chunk index is derived from the committed head
            if next_chunk >= N_CHUNKS:
                break
            try:
                handle.append("trades", chunks[next_chunk], expected_version=head,
                              note=f"{name}: chunk {next_chunk}")
                wins += 1
            except h5i_db.ConflictError:
                conflicts += 1  # another writer landed this chunk first - retry
    finally:
        handle.close()
    stats[name] = {"commits": wins, "conflicts": conflicts}


threads = [threading.Thread(target=feed_worker, args=(f"writer-{i}",)) for i in range(3)]
for t in threads:
    t.start()
for t in threads:
    t.join()

stats

output

9 chunks x 1,520 rows for the race

{'writer-0': {'commits': 9, 'conflicts': 0},
 'writer-2': {'commits': 0, 'conflicts': 1},
 'writer-1': {'commits': 0, 'conflicts': 1}}

Which thread lands which chunk varies run to run. The outcome is deterministic: every chunk committed exactly once, in order, with a linear version chain. Below we verify no rows were lost or duplicated.

rows_after = len(writer_a.read("trades"))
expected = rows_before + thread_rows
assert rows_after == expected, f"lost updates! {rows_after} != {expected}"

seqs = [v["sequence"] for v in writer_a.versions("trades")]
assert seqs == list(range(len(seqs))), "version chain is not linear"

total_commits = sum(s["commits"] for s in stats.values())
total_conflicts = sum(s["conflicts"] for s in stats.values())
print(f"rows: {rows_before:,} -> {rows_after:,} (all {N_CHUNKS} chunks landed, none lost)")
print(f"commits: {total_commits}, conflicts absorbed by retries: {total_conflicts}")
print(f"version chain: v0..v{seqs[-1]}, strictly linear")

output

rows: 6,000 -> 19,680 (all 9 chunks landed, none lost)
commits: 9, conflicts absorbed by retries: 2
version chain: v0..v12, strictly linear

[
    {k: v[k] for k in ("sequence", "op", "rows", "note") if k in v}
    for v in writer_a.versions("trades")[-4:]
]

output

[{'sequence': 9, 'op': 'append', 'rows': 15120, 'note': 'writer-0: chunk 5'},
 {'sequence': 10, 'op': 'append', 'rows': 16640, 'note': 'writer-0: chunk 6'},
 {'sequence': 11, 'op': 'append', 'rows': 18160, 'note': 'writer-0: chunk 7'},
 {'sequence': 12, 'op': 'append', 'rows': 19680, 'note': 'writer-0: chunk 8'}]

6. Plan/apply hits the same wall#

The previewable-mutation flow, plan_delete_range then inspect then apply, is CAS-guarded too. A plan is built against a specific base version, and apply() refuses if the head has moved since.

Here A plans to delete a window of suspect prints, but B commits fresh data before A applies. That is exactly the race you want caught when a feed and an ops job share a table.

Range arguments are raw microseconds, the ts column's unit, and the end bound is exclusive.

ts0 = feed["ts"][0].value  # raw us since epoch
bad_lo, bad_hi = ts0, ts0 + 60_000_000  # first minute of the day

plan = writer_a.plan_delete_range("trades", bad_lo, bad_hi, note="drop suspect open prints")
print("planned:", plan.summary["rows_affected"], "rows to delete",
      f"({plan.summary['rows_before']:,} -> {plan.summary['rows_after']:,})")

# B lands one more chunk while A's plan sits unapplied:
writer_b.append("trades", feed.slice(base_rows + thread_rows, late_rows), note="B: late chunk")

try:
    plan.apply()
except h5i_db.ConflictError as e:
    print(f"\napply failed - code={e.code}, retryable={e.retryable}")
    print(f"hint = {e.hint}")

output

planned: 158 rows to delete (19,680 -> 19,522)

apply failed - code=version_conflict, retryable=True
hint = re-read the head of "trades" and retry against it; pure appends rebase safely (the CLI and Python bindings already auto-retry those)

The stale plan is dead, but re-planning is cheap. The second plan is built against the new head, so its preview reflects B's late chunk too.

That is the point of the CAS. You re-decide with current facts, instead of blindly mutating a table that changed under you.

plan.discard()  # drop the stale plan
plan2 = writer_a.plan_delete_range("trades", bad_lo, bad_hi, note="drop suspect open prints (re-planned)")
commit = plan2.apply()
print(f"re-planned and applied as v{commit['sequence']} ({commit['op']}), "
      f"rows_total={commit['rows_total']:,}")

[
    {k: v[k] for k in ("sequence", "op", "rows", "note") if k in v}
    for v in writer_a.versions("trades")[-3:]
]

output

re-planned and applied as v14 (delete_range), rows_total=21,022

[{'sequence': 12, 'op': 'append', 'rows': 19680, 'note': 'writer-0: chunk 8'},
 {'sequence': 13, 'op': 'append', 'rows': 21180, 'note': 'B: late chunk'},
 {'sequence': 14,
  'op': 'delete_range',
  'rows': 21022,
  'note': 'drop suspect open prints (re-planned)'}]

Takeaways#

Multiple Database handles on one path are first-class. Commits through one handle are immediately visible to the others, and the version chain stays linear no matter who writes.
expected_version turns append into a compare-and-swap. A stale write raises ConflictError carrying .code, .retryable and .hint: an explicit, retryable failure instead of a silent lost update.
The retry loop is five lines: re-read head, re-append, bounded attempts. Three racing threads, coordinated through nothing but the CAS, ingested a chunked feed with zero lost rows.
plan.apply() is guarded by the same mechanism. Plans bind to a base version, and a moved head forces a re-plan, so your preview can never be stale at apply time.
Explicit conflict beats last-writer-wins for shared research and production tables. The failure mode is a retry, not a reconciliation break.

writer_a.close()
writer_b.close()

Multi-writer coordination: optimistic locking, conflicts, and retries

1. The data#

2. Two handles, one database#

3. A stale expected_version fails loudly#

4. The retry pattern#

5. Three threads racing on one feed#

6. Plan/apply hits the same wall#

Takeaways#

3. A stale `expected_version` fails loudly#