Durable storage for graph threads: each thread's checkpoint history, the pending writes recorded against its latest checkpoint, its durable event log, and the claim of the run that holds it. One store owns all of them so that a commit is atomic and fenced. Several GraphRuntimes, in one process or many, may share a store: the claim is what keeps two of them from running one thread at once.
Implementations must be safe to call from several threads and processes, and must:
grant at most one live claim per thread (claim): a claim is live until its expiresAt, by the store's own clock, has passed; a request while one is live is refused with GraphError.ThreadBusy naming its holder, and an expired claim is replaced - taken over;
issue each claim a FencingToken strictly greater than every token issued for that thread before - across releases, takeovers and deleteThread - so a token is never reused;
apply a Commit entirely or not at all, and only if its token is the thread's current claim (GraphError.StaleClaim otherwise, checked first) - so a writer that lost its claim, even one whose token is the highest it has seen, records nothing after another run took over;
accept a new checkpoint only when its parent is the thread's latest checkpoint id - the thread version the writer expects (GraphError.CheckpointConflict otherwise) - and its id is not already in the thread's history, and only pending writes naming the resulting latest checkpoint (GraphError.InvalidCommit);
keep every checkpoint a commit superseded in the thread's history, newest first in history and readable by id with checkpoint, until prune, a commit's retractTurn or deleteThread removes it - but keep pending writes for the latest checkpoint only;
number a thread's events 1, 2, 3, ... in commit order, inside the commit, never reusing a number - including after compaction, pruning or a failed commit;
store checkpoints and pending writes as data (Checkpoint.toJson, PendingWrite's ReadWriter), so nothing executable survives a round trip;
return from awaitEventsAfter with what eventsAfter would, at once if the log already holds an event after afterSeq, and otherwise within about timeout - with the events of a commit made meanwhile through any store over the same storage, or empty;
tolerate the thread calling eventsAfter or awaitEventsAfter being interrupted at any point: cancelling a subscription interrupts its watch mid-read, so an interrupt must end that one call - with Left(CancelledError) or a failed result - and leave the store usable by every other thread and later call (see awaitEventsAfter).
The holder of a claim may commit and renew it after expiresAt for as long as no other claim has replaced it: expiry only lets another run take the thread over. CheckpointerContract in llm4s-agent-testkit is the executable form of this contract; a store should pass it.
Claims threadId for request.holder 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) naming its holder, with nothing changed. A thread need not have a checkpoint to be claimed.
Claims threadId for request.holder 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) naming its holder, with nothing changed. A thread need not have a checkpoint to be claimed.
Removes everything stored for the thread - its checkpoint history, its pending writes, its event log and its claim - so that its id names a new thread again, whose events are numbered from 1; tokens issued after it are still greater than every token issued before. An unknown thread is Right(()). Use GraphRuntime.deleteThread, which claims the thread first, so it refuses a thread with a live run in any runtime, and checks the tenant; the store itself checks neither.
Removes everything stored for the thread - its checkpoint history, its pending writes, its event log and its claim - so that its id names a new thread again, whose events are numbered from 1; tokens issued after it are still greater than every token issued before. An unknown thread is Right(()). Use GraphRuntime.deleteThread, which claims the thread first, so it refuses a thread with a live run in any runtime, and checks the tenant; the store itself checks neither.
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.
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.
Removes the checkpoints and events of the thread that policy does not keep (see RetentionPolicy), by the store's clock. 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(()).
Removes the checkpoints and events of the thread that policy does not keep (see RetentionPolicy), by the store's clock. 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.
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.
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.
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.
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.