| Safe Haskell | None |
|---|---|
| Language | GHC2024 |
Free.Agent.Bus
Description
STM callback bus over a global JSONL log.
The live log image is held in a TVar; subscribers block via STM retry
until posts matching their names appear. A background thread persists new
posts to a single log.jsonl file under a file lock, so the lock never
appears in the agent path.
This is the in-process / single-runtime form of the bus. An out-of-process
form can replace the TVar with file-change events without changing the
PostStampedLog image.
Polymorphic in the body type a. File persistence uses PostBody for
JSON encoding/decoding at the storage boundary.
Synopsis
- data Bus a
- busLogPath :: Bus a -> FilePath
- openBus :: PostBody a => FilePath -> IO (Bus a)
- closeBus :: Bus a -> IO ()
- withBus :: PostBody a => FilePath -> (Bus a -> IO b) -> IO b
- scribe :: Bus a -> UTCTime -> Post a -> STM (Stamped a, TMVar ())
- scribeIO :: Bus a -> Post a -> IO (Stamped a)
- postLocal :: PostBody a => FilePath -> Post a -> IO (Stamped a)
- appendStoredPosts :: PostBody a => FilePath -> [Stamped a] -> IO ()
- appendStoredPostsUnlocked :: PostBody a => FilePath -> [Stamped a] -> IO ()
- readSince :: Bus a -> [Name] -> PostId -> STM [Stamped a]
- awaitSince :: Bus a -> [Name] -> PostId -> STM [Stamped a]
- runSeatBus :: Bus Text -> Name -> [Name] -> FreeSeat -> IO ()
Bus handle
busLogPath :: Bus a -> FilePath Source #
Path to the underlying log.jsonl.
openBus :: PostBody a => FilePath -> IO (Bus a) Source #
Open or create a bus at the given root directory.
Loads any existing log.jsonl into memory and starts the persistence
thread. The lock file lives at root/log.jsonl.lock.
closeBus :: Bus a -> IO () Source #
Stop the persistence thread.
Does not flush pending posts; call this only when durability is not required or after ensuring the log is quiescent.
Scribe
scribe :: Bus a -> UTCTime -> Post a -> STM (Stamped a, TMVar ()) Source #
Append a bare post to the live log inside one STM transaction.
The returned 'Stamped a' carries the absolute line id assigned by the
scribe. The caller must supply the timestamp. The returned TMVar is
filled once the post has been persisted to disk.
scribeIO :: Bus a -> Post a -> IO (Stamped a) Source #
Synchronous scribe: assign the current timestamp, append, and wait for the post to be persisted.
postLocal :: PostBody a => FilePath -> Post a -> IO (Stamped a) Source #
File-truth scribe: assign the id from the file itself, under the lock.
The id is the current line count, read and appended under the exclusive
file lock, so concurrent processes can never assign the same id twice.
This is the posting path for anything that shares the log with other
processes (CLI posts, seat replies). The TVar bus (scribeIO) is for
a single runtime that owns all writes; a long-lived seat's in-memory
image goes stale the moment another process posts, and stale images
assign colliding ids.
Durable append
appendStoredPosts :: PostBody a => FilePath -> [Stamped a] -> IO () Source #
Append stamped posts under the exclusive file lock.
Shared durable image primitive for the live bus persistence loop. Does not assign ids or timestamps.
appendStoredPostsUnlocked :: PostBody a => FilePath -> [Stamped a] -> IO () Source #
Append stamped posts without taking the lock. Caller must already hold
path.lock (or otherwise guarantee exclusive writers).
Subscription
readSince :: Bus a -> [Name] -> PostId -> STM [Stamped a] Source #
Read all posts matching any of the names with id at or after the cursor.
The cursor is the next unprocessed id, matching the file-cursor convention. Does not retry; returns an empty list if nothing matches.
awaitSince :: Bus a -> [Name] -> PostId -> STM [Stamped a] Source #
Wait until at least one matching post exists after the cursor.
Agent runtime
runSeatBus :: Bus Text -> Name -> [Name] -> FreeSeat -> IO () Source #
Run a FreeSeat as a bus agent.
Blocks via awaitSince (STM retry) for posts addressed to any of the
names, feeds the batch into the seat, and scribes any emitted replies.
This is the callback loop: no polling, no file locks in the agent path.
Replies carry thread edges citing the parent stamp. To preserve the
input-to-output mapping we process one 'Stamped Text' at a time; the seat
still sees a singleton batch, and the parent id is prepended to each
emitted post's thread.
Decided quiet: a delivered post carrying a halt (🟢 / 🔵) or escalation (🔴) mark stops the loop. Marks are control, not content: they are not handed to the seat.
Self-halt: a 🔵 reply is the seat deciding its own quiet — scribe it and stop, mid-batch if need be; later posts go unanswered (the seat is gone). 🟢 stays exchange-level: a seat may land one exchange and host more.