Store Engine — Single-Writer JSON

Not a database. One process, one worker, one os.replace. If you want Postgres, this is not the service — it is a demo store engineered to be trivially auditable.

Guarantees

• Atomicity — a mutation and its side-effects (version bump, event append, idempotency ledger, attachment finalize) commit together via a single os.replace(tmp, state.json); no half-commit path exists.
• Durability — os.fsync(fd) on the tmp file then fsync on its directory before replace.
• Isolation — asyncio.Lock serializes JsonStore.commit; OperationBarrier gives many shared readers + one exclusive resetter.
• Exclusive runtime — fcntl.flock(LOCK_EX|LOCK_NB) on runtime/runtime.lock; second process raises RuntimeLockedError at open().
• Determinism — no wall-clock read inside commit; caller supplies clock.now().

Snapshot shape

runtime/state.json = StoreSnapshot (see Data Models). Key counters:

• storeRevision: int — starts 0, +=1 per commit, survives restart.
• stateEpoch: uuid — changes only on destructive _dev/reset or reseed — signals clients to discard optimistic state.
• streamEpoch/streamCursor — cursor for GET /sync-snapshot + SSE Last-Event-ID replay.
• events: StoredEvent[] — ring buffer, capacity EVENT_RING_CAPACITY=100, oldest dropped.
• idempotency: Record<scopeKey, {requestDigest, response}> — only 2xx recorded, same atomic commit.

Lifecycle

sequenceDiagram
  participant App
  participant Lock as _RuntimeFileLock
  participant Store as JsonStore
  App->>Lock: acquire(runtime.lock) LOCK_EX|LOCK_NB
  Lock-->>App: pid written + fsync
  App->>Store: open()
  alt state.json exists
    Store->>Store: _load_state() + version/epoch guards
  else no state
    Store->>Store: _build_seed_snapshot() from app/seed/*.json
    Store->>Store: _persist(snapshot) via os.replace
  end
  Store->>Store: _verify_attachments() + _remove_unreferenced_uploads()
  Store->>Store: clear staging dir
  App->>Store: set_publisher(broker.publish)
  App->>Store: _schedule_next_deadline(service)

Marker guard: runtime/.kxinspect-runtime contains RUNTIME_MARKER_UUID=6f2a1e34-5b7c-4d18-9e0a-2c7d5b8f4a13; scripts/reset_runtime.py refuses to delete a directory without it.

Commit protocol

All preparation happens outside the lock: schema validation, attachment staging/hashing, digest, Idempotency-Key UUID check.

Inside commit(mutator) under asyncio.Lock:

acquire async lock
if idempotency replay hit -> return stored response (no events/version)
check expectedStateEpoch vs snapshot.stateEpoch -> 409 store.epoch_mismatch
deadline reconciliation #1  (before transition)
run transition(status, event) # pure
if deferred reconciliation won race -> commit Accepted then surface deferred 409
if contest -> finalize_attachments()
bump version, updatedAt, append DraftEvent -> StoredEvent (id = deterministic event_id)
write idempotency record (2xx only)
events ring trim, storeRevision+=1
serialize -> tmp file -> fsync -> os.replace -> fsync dir
publish(events) via broker
return CommitResult
release lock

MutationOutcome[T] carries value, events, persist, deferred_error — the deferred-error field is the one explicit contract race: reconciliation must persist first, then the caller's command fails.

Operation barrier

barrier.shared()    # reads, uploads, downloads — concurrent
barrier.exclusive() # reset — waits until shared==0, then blocks new shared

Every finite operation holds barrier.shared() unless it self-manages (/events SSE stream, /_dev/reset). Without it, a reset could delete state.json mid-read.

File locking

class _RuntimeFileLock:
  def acquire(self):
    fd = os.open(path, O_CREAT|O_RDWR, 0o600)
    fcntl.flock(fd, LOCK_EX|LOCK_NB)  # EWOULDBLOCK -> RuntimeLockedError
    ftruncate(fd, 0); write(f"{pid}\n"); fsync(fd)

Parent and uploads dirs are chmod 0o700. Lock pid is informational only; the advisory lock is the truth.

Referenced-attachment GC

At open() and after each commit, _referenced_attachment_ids(snapshot) collects every url/thumbnailUrl/downloadUrl last segment across charges/inspections/reports; unreferenced files under runtime/uploads/ are removed.

Scheduler path

_schedule_next_deadline(service, scheduler) finds earliest future deadlineAt among non-terminal charges, calls scheduler.schedule_at(at, fn) where fn = service.reconcile_deadlines. On fire, reconciles then reschedules.

In tests, ManualClock + manual scheduler make this synchronous and observable.

Failure modes

Failure	Surface
Second process	RuntimeLockedError with path
Corrupt state.json	CorruptRuntimeStateError
Version mismatch (incompatible snapshot)	IncompatibleRuntimeStateError
Missing uploads for referenced id	logged + CorruptRuntimeStateError if verification strict
Oversize body	MaxBodySizeMiddleware 413 before parser
Epoch mismatch inside commit	store.epoch_mismatch 409 with currentStateEpoch