MeteredCheckpointer

eventsAfter, waiting for a commit when there is nothing to return yet: up to limit events with seq > afterSeq, ascending, at once if the log holds any; otherwise those of a commit that lands within timeout, or an empty Vector once it has passed. It is how a subscription learns of commits made through another GraphRuntime or process sharing the store (see WatchPolicy): each subscription calls it in a loop, from the last event it has queued.

The default polls. It reads once and, finding nothing, waits timeout and returns empty, so the caller's next call reads again: a commit is seen up to timeout after it lands, at the cost of one eventsAfter per timeout per subscription. A store that can be told of commits overrides it to return as soon as one lands, without reading meanwhile - InMemoryCheckpointer wakes on its own commits, and a database store can use its notification channel (PostgreSQL's LISTEN/NOTIFY). Such a store must wake for commits made through every store over the same storage, not only its own, and still return within about timeout when none lands.

Errors are eventsAfter's - GraphError.ReplayUnavailable when afterSeq is behind the replay floor. Interrupting the waiting thread ends the wait with Left(CancelledError), the thread's interrupt flag set again.

'''The reading thread may be interrupted at any point.''' A subscription's watch calls this on a thread of its own, and Subscription.cancel - or the subscription's end - interrupts that thread whether it is waiting or in the middle of a read, so the interrupt can land inside the store's own I/O (eventsAfter included). A store must take that as the end of this one call, never as damage to itself: release what the call held (a pooled connection, a lock), and stay usable by other threads and later calls. Beware drivers whose I/O goes through a java.nio.channels.InterruptibleChannel (a FileChannel or SocketChannel): an interrupt during a read or write closes the channel, with ClosedByInterruptException, so a store that shares one channel - one connection - between threads would lose it for all of them. Such a store reads on a connection of its own for the waiting call, or discards and replaces a connection an interrupt has closed.

Checkpoint checkpointId of the thread's history, latest included; None if it is not there.

Up to limit of parent's direct children (design 4.41), oldest first: every thread whose latest checkpoint's Checkpoint.origin names parent, in the order each first committed a checkpoint with that origin, starting after child after when it is given, so a caller pages through them by passing the last id of each page. A thread deleteThread removed is not listed, nor is a thread whose latest checkpoint names another parent or none. An after the store does not know as a child of parent is Left(ValidationError); a child deleted since it was listed keeps its place as long as its tombstone does (deleteThread), so a page boundary survives retention deleting it. A thread with no children has none; limit below 1 reads none. Needs no claim, and reads no tenant: the runtime checks the parent's tenant (GraphRuntime.children).

Claims threadId for request.holder, recording request.tenantId with the claim, until request.ttl from now, by the store's clock, if no other claim on it is live, and returns the claim with a new FencingToken. A live claim - whoever holds it, the same holder included - is Left(GraphError.ThreadBusy(threadId, latest, holder)) with the thread's latest checkpoint id and nothing changed. holder is GraphError.BusyHolder.Run(holder) only when the live claim's tenant is request.tenantId (None and Some differ), and GraphError.BusyHolder.Unnamed otherwise - also for a claim that recorded no tenant, such as one persisted before claims carried one - so that a refusal never reveals another tenant's RunId; a store never returns GraphError.BusyHolder.Lapsed. A thread need not have a checkpoint to be claimed. A claim that replaces an expired one nobody released names that claim's holder as RunClaim.takenOverFrom, under the same rule: only if it was of request.tenantId, and None otherwise.

Applies commit atomically, fenced by its token, and returns its events with their sequence numbers, each recording that token. A Commit.fence of (other, otherToken) fences it by other's claim too, checked in the same transaction or under the same lock as the write, after the thread's own claim: otherwise Left(GraphError.StaleClaim(other, otherToken, otherCurrent)) with nothing written.

Drops events with seq < beforeSeq; replay can then start no earlier than beforeSeq.

Removes everything stored for the thread - its checkpoint history, its pending writes, its event log, its effect records and its claim - so that its id names a new thread again; tokens issued after it are still greater than every token issued before. The store keeps only where the thread's event numbering ended - a tombstone, holding nothing of the thread's data, and nothing at all for a thread that numbered no event - so that a new thread of the id numbers its events on past the deleted thread's, one number skipped, and a read of the deleted log after the delete is GraphError.ThreadDeleted (see eventsAfter); a store that wakes awaitEventsAfter on commits wakes it on a delete too. Deleting a thread whose tombstone has numbered no event since does not move the numbering on again.

'''The tombstone is kept until sweep reclaims it.''' A deleted id keeps a small record - its next event number and when the store deleted it, nothing of the thread's data - which neither prune nor a later delete removes. It is what lets a reader of the deleted log, in any process, be told the thread was deleted, and a new thread of the id be numbered past it. A store whose threads are deleted often - every failed one-shot Agent.run forgets its thread, and child retention deletes children - grows by one such record per deleted thread that numbered an event until a sweep with a tombstoneAge reclaims the old ones; a reclaimed tombstone's readers are still told the thread was deleted, through the store's reclamation mark (design 4.53). Fenced like a commit: only while token is the thread's current claim, checked atomically with the delete, and otherwise Left(GraphError.StaleClaim) with nothing deleted - so a deleter that lost its claim (it expired and another run took the thread over, or it was released) deletes nothing of the run that holds the thread now. A thread with no claim, an unknown one included, is therefore refused too: claim it first. Use GraphRuntime.deleteThread, which claims the thread, so it refuses a thread with a live run in any runtime, and checks the tenant under its claim.

'''The fence.''' With fence = Some((other, otherToken)) the delete is also fenced by another thread's claim: it happens only if otherToken is other's current claim too, checked in the same atomic step as the thread's own claim and the delete (one transaction, or under one lock), and otherwise is Left(GraphError.StaleClaim(other, otherToken, otherCurrent)) with nothing deleted. The thread's own claim is checked first. GraphRuntime.deleteThread fences a cascade's delete of each descendant by the root's claim, and child retention each delete by the parent's (design 4.41), so a successor run that took the thread above over never finds a child it may reattach gone. None fences by the thread's own claim alone.

The thread's effect record under key (ReplayGuard, design §4.24); None if it has none, or the thread is unknown. Needs no claim.

Up to limit events with seq > afterSeq, ascending. afterSeq 0 reads from the thread's first event, as does the number just before it; one behind the replay floor is GraphError.ReplayUnavailable, and one inside the log of a deleted thread of the id (Checkpointer.deleted) is GraphError.ThreadDeleted.

Up to limit of the thread's checkpoints, newest first - the latest is the first of the first page - starting after checkpoint before when it is given, so a caller pages through the history by passing the last id of each page. A before that is not in the history is Left(GraphError.CheckpointNotFound); a thread with no checkpoint has an empty history.

The incomplete root threads of tenantId (design 4.52): those whose latest checkpoint is incomplete (CheckpointStatus.Running) and that no live claim holds, by the store's clock - ThreadStatus.Idle - oldest first by ThreadSummary.updatedAt (ties by thread id, ascending), starting after after, at most limit (1 to ThreadQuery.MaxLimit, ValidationError otherwise), with the next page's cursor when more match. A thread whose claim expired without a release is listed; one whose cascade delete has begun is not (ThreadStatus.Deleted).

It is for operators, not a worker's recovery query: a cancelled run, a drained one and a crashed one all look alike here, and only a run queue's entry says which should run again (Stage 5, rule 4). Tenant-scoped and bounded as threads is.

The thread's latest checkpoint and its pending writes; None for a new thread.

The run-id lookup (Stage 5 rule 4, design 4.55): the newest checkpoint in threadId's history whose Checkpoint.runId is one of runIds, or None if none of them committed one that is still kept - none of them was admitted, or the thread was deleted since. A run queue's attempt asks it with the RunIds its entry recorded, before it touches the thread, to learn whether the entry's input was admitted, whatever ran on the thread since: the thread's latest checkpoint alone cannot say, because a direct run after the attempt died would have replaced it. Newest is by place in the history, not by createdAt.

A store answers it by an index on the thread and run, within its bound, however long the history; an empty runIds is Right(None). While an open entry of the store's run queue names a run, prune keeps the newest checkpoint that run committed, so the lookup never loses the evidence an open entry depends on. Needs no claim.

Removes the checkpoints and events of the thread that policy does not keep (see RetentionPolicy), by the store's clock: maxAge is judged against the time the store committed each checkpoint and event, by that same clock, never against the producer's Checkpoint.createdAt or EventRecord.timestamp, so a writer whose clock is skewed makes nothing too old or too young. The latest checkpoint and its pending writes are always kept. Pruning a checkpoint also removes the events before the oldest checkpoint kept; maxAge and maxEvents may remove more, including the latest checkpoint's. Removing events raises the replay floor as compactEvents does. Needs no claim: it touches nothing a run reads or writes. An unknown thread is Right(()).

Ends the claim whose token is token, so the thread can be claimed again at once. A token that is not the thread's current claim changes nothing and is Right(()): a run that lost its claim must never release its successor's.

Extends the claim whose token is token to ttl from now, by the store's clock, and returns it. Left(GraphError.StaleClaim) if token is not the thread's current claim (another run took it over, it was released, or the thread was deleted). An expired claim nobody replaced is renewed.

The store-wide retention sweep (design 4.53): one bounded page of the threads the store holds, tombstones included, in an order of the store's own that every call follows - the threads after after in it, whether or not the store still holds after, at most policy.batchSize of them - with the next page's cursor when more follow. A caller sweeps the whole store by calling again with each page's next until it is None; a thread created while a caller pages may be visited in that pass or only in the next.

Before it changes anything of the page it asks guard which of the page's threads are protected (SweepGuard), and a Left from the guard ends the call with nothing changed. Then, for each thread of the page:

• its history is pruned by policy.retention exactly as prune prunes it, judged by the store's clock against the store's commit times - except that, of a protected thread, the newest checkpoint each of its protected RunIds committed (Checkpoint.runId) is kept as well, whatever the policy, while the rest is pruned as usual and the events' floor follows the policy alone; and so, as prune does, is the newest checkpoint each run an open entry of the store's run queue recorded committed (design 4.55), whatever the guard answers;
• its tombstone (deleteThread) is reclaimed - removed, with the place it kept among its parent's children - when policy.tombstoneAge is set, the store last deleted the thread more than that long ago by its clock, nothing has been stored under the id since, and the thread is not protected. Reclaiming raises the store's reclamation mark to the tombstone's first number, so a new thread of the id - or of any id without a tombstone - numbers its events on past the old thread's, and a read from inside the reclaimed log is still GraphError.ThreadDeleted (eventsAfter). Only paging children after a reclaimed child refuses its cursor.

Like prune, it needs no claim and changes nothing a run reads or writes; it reads no tenant, for it is the operator's, store-wide. Each call returns within the store's bound, whatever the store holds. Run one sweep at a time per store (a worker's schedule does: Stage 5, W2); two at once are safe, only wasteful.

The thread registry (design 4.52): a page of the root threads of query.tenantId that pass every filter of query (ThreadQuery.matches), newest first by ThreadSummary.updatedAt - when the store last committed a checkpoint or metadata to the thread, by its clock - ties broken by thread id, descending; starting after query.after, at most query.limit of them, with the cursor of the next page when more match (ThreadPage.of). Each thread's ThreadStatus is judged from its latest checkpoint and whether a live claim holds it, by the store's clock, when it is read.

Listed are only threads with a checkpoint whose latest checkpoint's tenant is query.tenantId (None and Some differ) and whose latest checkpoint names no parent (Checkpoint.origin): a child thread is read through children, and a thread deleteThread removed is never listed. Paging is by position: a thread changed while a caller pages moves to the front, so a later page of that pass does not list it again, and a pass from the start, or with updatedSince, finds it. Needs no claim, and reads no other tenant's thread: the tenant is the query's. A store answers it within its bound however many threads it holds, by an index.