# Changelog
## Unreleased
## 0.3.0.0 — 2026-07-11
This release requires migrations `0003`–`0007` from `kiroku-store-migrations`.
### Breaking Changes
* Backward reads now paginate. `readStreamBackward` and `readAllBackward` treat a
nonzero cursor as an *exclusive upper bound*, so a page returns events older
than the cursor instead of newer ones. The interpreter maps the caller cursor
`0` to `maxBound`, so `StreamVersion 0` / `GlobalPosition 0` still means "start
from the latest event" (versions and positions start at 1, so the sentinel can
never collide with a real value). Previously a nonzero cursor read in the wrong
direction, which made backward pagination impossible: feeding the last returned
version back in returned the same newest page again. Any caller passing a
nonzero backward cursor gets different results, and any mock `Store`
interpreter must mirror the `0 -> maxBound` mapping.
* Oversized stream names are rejected before any database work. `StoreError`
gains `StreamNameTooLong !StreamName !Int` (the `Int` is the offending UTF-8
byte length), and `Kiroku.Store.Error` exports `maxStreamNameBytes` (512) and
`validateStreamName :: StreamName -> Either StoreError ()`. Every
application-stream entry point — `appendToStream`, `appendMultiStream`,
`linkToStream`, `softDeleteStream`, `hardDeleteStream`, `undeleteStream`, and
the `runTransactionAppending*` wrappers — runs the same validation (reserved
`$all` plus the length bound) up front instead of only rejecting `$all`. The
bound exists because the append notification payload embeds the stream name and
PostgreSQL rejects `pg_notify` payloads of 8,000 bytes or more; a name that
used to fail late as an opaque trigger error now fails immediately with a typed
error. Exhaustive matches on `StoreError` must handle the new constructor.
* Publisher queues are created only for non-group `AllStreams` subscriptions.
`Category` subscriptions and all consumer-group members no longer register a
bounded queue with the `EventPublisher`; they are DB-driven in live mode. As a
consequence `SubscriptionConfig.queueCapacity` and
`SubscriptionConfig.overflowPolicy` have no effect for those targets, and they
can no longer be paused, dropped, or terminated with `SubscriptionOverflowed`.
In the exposed `Kiroku.Store.Subscription.Worker` module, `runWorker`'s
signature changed from taking a `TBQueue (Vector RecordedEvent)` plus a `TVar
SubscriberStatus` to taking a single new exported sum `LiveSource
(LiveFromPublisherQueue | LiveFromCategoryNotify | LiveFromGroupPolling)`,
fixed at `subscribe` time from the config's `(consumerGroup, target)` shape.
* Empty append batches are rejected. `StoreError` gains `EmptyAppendBatch
!StreamName` and `AppendConflict` gains `EmptyAppendBatchConflict !StreamName`
(mapped by `appendConflictToStoreError`). `appendToStream` with `[]` now fails
with `EmptyAppendBatch` before any pool work; previously it took the global
`$all` row lock, fired NOTIFY triggers, created an empty stream under
`NoStream`, and surfaced a misleading `WrongExpectedVersion` /
`StreamNotFound`. `appendMultiStream []` is an explicit no-op returning `[]`
without a round trip, while a per-stream empty event list inside a non-empty op
list is rejected with `EmptyAppendBatch`.
* Link failures are typed. `StoreError` gains `EventAlreadyLinked !StreamName
!(Maybe EventId)` (the `23505` unique violation on `stream_events_pkey`,
carrying the offending event id when PostgreSQL's detail string is parseable)
and `LinkSourceEventMissing !StreamName` (the `23502` not-null violation raised
when `linkToStream` references an event id that does not exist; the whole batch
rolls back). Both previously surfaced as an opaque `ConnectionError` carrying a
stringified `UsageError`.
* `StreamInfo` gains a `truncateBefore :: !StreamVersion` field. Code
constructing or exhaustively matching `StreamInfo` must be updated.
* The `Store` effect gains `SetStreamTruncateBefore :: StreamName ->
StreamVersion -> Store m (Maybe StreamId)` and `EventExistsInStream ::
StreamName -> EventId -> Store m Bool`. Mock and test interpreters of the
`Store` effect must add arms for both.
* `Kiroku.Store.SQL`'s hard-delete statements were resplit.
`deleteStreamJunctionsStmt` is removed, replaced by `deleteAllRowsForOriginStmt`,
`deleteJunctionsByEventIdsStmt`, `deleteStreamOwnJunctionsStmt`, and
`deleteDeadLettersForOrphanedEventsStmt`, which the interpreter runs in sequence
before `deleteOrphanedEventsStmt`.
* Checkpoint-load failures now fail the subscription loudly. A `UsageError` while
reading a subscription's saved checkpoint at startup surfaces through the
handle's `wait` as `Left e` and the worker stops; it no longer falls back to
`GlobalPosition 0` and silently re-processes the whole history.
* `subscriptionAckStream` and `subscriptionStream` reject a zero buffer, throwing
the new `InvalidStreamBufferSize` exception (exported from
`Kiroku.Store.Subscription.Stream`) when `bufferSize` is `0`; a zero-capacity
`TBQueue` would deadlock the bridge handler on its first delivery.
* Duplicate events inside transaction bodies are typed. `runTransaction`,
`runTransactionNoRetry`, and the `runTransactionAppending*` wrappers map a
`23505` violation on `events_pkey` / `stream_events_pkey` to `DuplicateEvent
(Just eid)` (new `mapTransactionUsageError` in `Kiroku.Store.Error`) instead of
the previous opaque `ConnectionError`.
### New Features
* Logical truncate-before marker (close-the-book compaction).
`Kiroku.Store.Lifecycle.setStreamTruncateBefore :: StreamName -> StreamVersion
-> Eff es (Maybe StreamId)` and `clearStreamTruncateBefore :: StreamName -> Eff
es (Maybe StreamId)` set a per-stream cursor below which ordered per-stream
reads hide events. `readStreamForward`, `readStreamBackward`, and the paged
`readStreamForwardStream` return only events whose `streamVersion` is at or
above the marker; the `$all` global log, `readCategory`, subscriptions, and
existence probes are deliberately unaffected, so global history stays complete
and future projections can still be built from it. No events are deleted and
the operation is fully reversible. The typical use is snapshot-and-compact:
append a snapshot event at version `V`, then call `setStreamTruncateBefore name
V` so rehydration starts there. The marker is surfaced on
`StreamInfo.truncateBefore` via `getStream`, is idempotent, returns `Nothing`
for a missing or soft-deleted stream, and rejects `$all` with
`ReservedStreamName`. Backed by the new `streams.truncate_before` column
shipped by `kiroku-store-migrations`.
* `Kiroku.Store.Read.eventExistsInStream` adds an indexed point lookup for
idempotency checks that need to know whether a specific event id is present in
a live stream without scanning that stream. Soft-deleted streams report
`False`, matching `readStreamForward` visibility.
* `Kiroku.Store.Types` exports `categoryName :: StreamName -> CategoryName` (the
substring before the first `-`) and its dual `streamNameInCategory ::
CategoryName -> Text -> StreamName`. This is the canonical Haskell mirror of
the `streams.category` generated column and of the rule used by `readCategory`
and category-targeted subscriptions; previously every consumer hand-rolled it.
`Notification.categoryFromPayload` now routes through `categoryName`, so the
rule has one definition.
* New observability event `KirokuEventPublisherLoopError !SomeException`, emitted
when the publisher's broadcast iteration throws a synchronous non-`UsageError`
exception (typically a user-supplied `decodeHook` or observability handler).
Exhaustive matches on `KirokuEvent` must handle it.
### Bug Fixes
* The publisher survives failing callbacks. A synchronous exception from the
broadcast iteration is caught, reported as `KirokuEventPublisherLoopError`, and
retried on the next notification or the 30-second safety poll without advancing
`lastPublished`; previously it killed the publisher loop and stalled every live
subscription. All store-internal threads (publisher, worker, notifier) now emit
observability events through a hardened `emitOrDrop` that drops synchronous
exceptions thrown by the handler and rethrows async ones, so a throwing
metrics or logging callback can no longer kill a store thread.
* Streamly bridges terminate with the worker's outcome. `subscriptionStream` and
`subscriptionAckStream` end the stream normally when the worker stops cleanly
or is cancelled, and rethrow the worker's exception from the next stream pull
when it dies for any other reason (overflow, handler exception, dead-letter
database error, decode-hook exception); previously the consumer blocked
forever. The cancel action no longer writes a sentinel into the bounded queue,
so it cannot block behind a full bridge buffer.
* Subscription startup no longer leaks registrations. The pre-fork window in
`subscribe` is masked and each acquired registration is paired with
`bracketOnError` cleanup, so an exception between publisher registration and
the worker fork can no longer leave a subscriber registered with no reader.
Notifier startup releases its connection if `LISTEN` fails.
* Hard delete purges dead letters. `hardDeleteStream` deletes
`kiroku.dead_letters` rows for exactly the events whose payloads it orphans,
before deleting those payloads. Because `dead_letters.event_id` has a foreign
key to `events(event_id)` with no cascade, hard-deleting a stream that had
dead-lettered events previously failed on that constraint.
* Single-stream appends retry transient conflicts. `appendToStream` retries once
when the first attempt fails with PostgreSQL `40001` (serialization failure) or
`40P01` (deadlock detected) — the same SQLSTATE set `hasql-transaction`
retries. Event ids are prepared before the first attempt, so the retry is
idempotent.
* `appendToStream` and `appendMultiStream` surface duplicate events raised from
inside transaction bodies as `DuplicateEvent` rather than a stringified
connection error.
### Performance
* `appendMultiStream` now pipelines its deterministic pre-lock and per-stream
append statements through Hasql pipeline mode, while preserving all-or-nothing
rollback and transient deadlock retry behavior. The local benchmark regression
gate measured `All.reliability-audit.appendMultiStream 3 existing streams` at
268 us, 29% faster than the refreshed baseline.
* The publisher does no row fetches while idle. With an empty subscriber registry
the publisher loop advances `lastPublished` from the `$all` tail with a
single-row query instead of fetching and decoding full event rows, so a store
whose only subscriptions are `Category` or consumer-group members pays no
fan-out cost. It re-checks registry emptiness in the same STM transaction as
the position write, and offers an in-flight batch to subscribers that
registered after the snapshot, so a subscriber attaching mid-tick never skips
events.
* No empty read round trips. `readStreamForwardStream` stops after a page shorter
than `pageSize` rather than always probing for one more empty page (an exact
multiple of `pageSize` still costs one final probe), flattens pages without an
intermediate list, and `lookupStreamNames []` short-circuits to an empty map in
the interpreter with no database round trip.
* Index hygiene and `streams` fillfactor tuning (migration `0005`).
### Other Changes
* `GlobalPosition`'s documented contract is weakened to "strictly increasing and
totally ordered" — nothing more. Treat values as opaque cursors: construct one
only from `0` or from a value this store returned, never by arithmetic, and
never assume density (`pos + 1` may not exist). The current implementation
still assigns contiguous positions, but contiguity is now explicitly an
implementation detail, not an API guarantee.
* `linkToStream` is marked provisional. It is the only public feature that
requires the `stream_events` junction-table layout, which a future
global-position migration may replace; it may be removed or redesigned.
* `RetryPolicy.retryMaxAttempts` is documented as a bound on total deliveries,
not redeliveries: the first delivery counts as attempt 1, so
`defaultRetryPolicy` (5) means the first delivery plus four redeliveries before
`DeadLetterMaxAttempts`. Behavior is unchanged; the previous haddock was wrong.
* `appendToStreamTx` and the `runTransactionAppending*` wrappers document the
`$all` lock-hold hazard: the append updates Kiroku's global `$all` row and
PostgreSQL holds that row lock until the transaction ends, so every statement
the continuation runs afterwards blocks all other appends store-wide. Keep
continuations minimal and run append-independent work first.
* `WrongExpectedVersion` and `WrongExpectedVersionConflict`'s third field is
documented as a placeholder: the append statement returns zero rows on a
mismatch and the store does not issue an extra read, so it is always
`StreamVersion 0`. Callers needing the live version must use `getStream`.
* The append NOTIFY trigger is now split into INSERT/UPDATE triggers that exclude
the internal `$all` row and only fire when `stream_version` actually advances,
so soft-delete and undelete no longer emit spurious append notifications
(migration `0003`).
## 0.2.0.0 — 2026-05-31
### Breaking Changes
* `currentState` changed type from `m SubscriptionState` to
`m (Maybe SubscriptionState)` and is now resolved through the central registry
by `(name, member)` and this handle's token: `Just s` while the worker is live
and still owns the entry, `Nothing` once it has stopped/cancelled/crashed (its
registry key is removed), before it starts, or after a newer worker supersedes
the same key. This makes the registry genuinely the single source of truth for
live state and unifies the "stopped = absent" rule across `currentState` and
the snapshot. (The worker still solely writes its `stateVar`; only the handle's
read path moved to a registry lookup of that same cell.) A deliberate pre-1.0
breaking change.
### New Features
#### Per-batch delivery event (plan 46)
* New additive `KirokuEvent` constructor
`KirokuEventSubscriptionDelivered !SubscriptionName !Int !SubscriptionDeliveryPhase !SubscriptionGroupContext`,
emitted **once per non-empty batch** from the single delivery primitive
`processEvents` — on **every** target (`$all`, category, consumer group) and in
**both** the catch-up and live phases. The `Int` is the batch row count
(always `>= 1`); the new `SubscriptionDeliveryPhase` (`DeliveredCatchUp` /
`DeliveredLive`, also exported) is derived from the worker's driving FSM state.
This makes an `$all` subscription's live deliveries observable, closing the gap
where the `(Nothing, AllStreams)` live path emitted no per-batch event.
* `KirokuEventSubscriptionFetched` is **unchanged** and still emitted by the
DB-driven live loops per fetch (the live-fetch-rate signal); a DB-driven live
batch therefore emits both `Fetched` (the fetch) and `Delivered` (the delivery).
#### Central subscription-state registry (plan 45)
* `KirokuStore` gains a `subscriptionRegistry` field: a central, in-memory map
keyed by `(SubscriptionName, member)` holding every live subscription worker's
FSM-state cell, each guarded by a per-worker `Data.Unique.Unique` token. Each
`subscribe` registers its worker's existing state `TVar` on start and removes
it on any exit (stop, cancel, crash) via the worker's `finally` cleanup; the
delete is token-conditional, so an older duplicate-key worker cannot delete a
newer worker's replacement entry. The worker's writes are unchanged.
* New public view type `SubscriptionStateView { subscriptionName, member, state,
statePhase, cursor }` (`deriving stock (Show, Generic)`) — the committed
observability surface external consumers read via `^. #field`. New accessor
`subscriptionStates :: KirokuStore -> IO (Map (SubscriptionName, Int32)
SubscriptionStateView)` returns a near-instant snapshot of every live
subscription as view records: it snapshots the outer map and reads each cell
with `readTVarIO` (no large STM read set, so the reader never retries). Both
re-exported from `Kiroku.Store`. This map of view records is the foundation
for a future Prometheus exporter and admin tool (named future consumers, not
built here), and the performant live-state layer that closes the OpenTelemetry
export-on-end blind spot at zero per-event cost. The `cursor` field is the
worker FSM cursor, not a guaranteed durable checkpoint row. A
stopped/cancelled/crashed subscription is represented by absence, never a
`"stopped"` phase.
* New `stateName :: SubscriptionState -> Text` (a stable low-cardinality state
label) in `Kiroku.Store.Subscription.Fsm`; `stateName` and `stateCursor` are
re-exported from `Kiroku.Store`.
#### Per-event retry / dead-letter dispositions (plan 40)
* `SubscriptionResult` gains two Kiroku-native dispositions:
`Retry RetryDelay` (redeliver the same event after a delay, bounded by the new
`SubscriptionConfig.retryPolicy`; default five attempts) and
`DeadLetter DeadLetterReason` (record the event in `kiroku.dead_letters` and
atomically advance the checkpoint past it). `RetryDelay`, `DeadLetterReason`,
`RetryPolicy`, and `defaultRetryPolicy` are exported from
`Kiroku.Store.Subscription.Types`; `RetryDelay` / `DeadLetterReason` live in
`Kiroku.Store.Subscription.Fsm` and are also re-exported from
`Kiroku.Store.Observability`.
* `SubscriptionState` gains a surfaced `Retrying` observability state, visible
through the `currentState` handle accessor while a redelivery is pending. New
lifecycle events `KirokuEventSubscriptionRetrying` and
`KirokuEventSubscriptionDeadLettered`.
* New ack-coupled Streamly bridge `subscriptionAckStream` (emitting `AckItem`
`{ event, attempt, reply }`); the consumer fills the reply to drive the
worker's per-event disposition. `subscriptionStream` is unchanged in behavior
(reimplemented on top of it, always replying `Continue`).
* New `kiroku.dead_letters` schema (shipped by `kiroku-store-migrations`) and the
`SQL.insertDeadLetterAndCheckpointStmt` / `SQL.readDeadLettersStmt` statements;
the dead-letter insert and checkpoint advance run in one atomic statement.
#### Per-subscription event-type filtering and selector (plan 43)
* `SubscriptionConfig.eventTypeFilter :: EventTypeFilter` — a declarative,
closed filter (`AllEventTypes | OnlyEventTypes (Set EventType)`, default
`AllEventTypes`) applied worker-side in `processEvents` before the handler:
non-matching events skip delivery (never reaching the ack-coupled bridge, so
they are never retried or dead-lettered) while the batch-tail checkpoint
advances past them, so a highly selective subscription never stalls. Exported
with `eventTypeMatches` from `Kiroku.Store.Subscription.Types` and forwarded
through the Streamly bridge and the Shibuya adapter.
* `SubscriptionConfig.selector :: Maybe (RecordedEvent -> Bool)` — an optional
opaque per-event predicate (default `Nothing`), the escape hatch for filtering
that `eventTypeFilter` cannot express (e.g. on payload, metadata, or
correlation/causation ids — typically a one-off catch-up reprocess). It is
deliberately kept *separate* from `eventTypeFilter` rather than replacing it:
the closed sum stays introspectable, `Eq`/`Show`-able, and SQL-pushdown-able,
while the selector covers arbitrary predicates. The two compose as a logical
AND via the new `shouldDeliver` helper (an event is delivered only when it
passes both), with the same no-stall / checkpoint-advances guarantee. Both
`eventTypeFilter` and `selector` are forwarded through the Shibuya adapter
(`KirokuAdapterConfig` / `KirokuConsumerGroupConfig`).
#### Subscription worker driven by an explicit FSM (plan 41)
* The subscription worker is now driven by an explicit finite state machine
(`Kiroku.Store.Subscription.Fsm`): a `SubscriptionState`
(`CatchingUp`/`Live`/`Paused`/`Reconnecting`/`Stopped`) plus a single,
exhaustively pattern-matched `step` transition function. Behavior of existing
catch-up/live/stopped paths is unchanged; two recoverable states are added (see
below). `SubscriptionStopReason` moved to the `Fsm` module and is re-exported
from `Kiroku.Store.Observability`, so the public `KirokuEvent` API is unchanged.
* **Recoverable backpressure.** The new `PauseAndResume` `OverflowPolicy` is now
the `defaultSubscriptionConfig` default (previously `DropSubscription`). When a
bounded subscriber queue fills, the publisher pauses delivery and the worker
drains the stale queue and re-catches-up from its checkpoint, so no event is
lost and the checkpoint stays monotonic — instead of terminating the
subscriber. `DropSubscription` (fail-fast `SubscriptionOverflowed`) and
`DropOldest` (lossy) remain available.
* **Worker-level live reconnect.** A `Category` or consumer-group worker that
hits a database error on a live-mode fetch now enters `Reconnecting` — it backs
off and re-catches-up from its checkpoint — instead of retrying silently
in place. (`AllStreams` live delivery is publisher-fed and has no worker fetch.)
#### Subscription observability events
* `KirokuEventSubscriptionPaused`, `KirokuEventSubscriptionResumed`, and
`KirokuEventSubscriptionReconnecting` observability events for the new states.
* `currentState` on `SubscriptionHandleM`: a point-in-time read of the worker's
`SubscriptionState` for observability, backed by a `TVar` the worker writes on
every transition.
### Bug Fixes
#### Subscription catch-up DB errors
* Subscription catch-up now retries transient `FetchBatch` database errors at the
same cursor with capped backoff instead of treating the error as an empty
batch. This prevents `$all` subscribers from switching to live mode at a stale
cursor and missing pre-live events. See
`docs/plans/39-catchup-must-not-treat-db-errors-as-caught-up.md`.
#### Publisher restart history rebroadcast
* `EventPublisher` now initializes its in-memory cursor from the current `$all`
stream tail at startup instead of `GlobalPosition 0`. Restarting a store no
longer re-reads and re-broadcasts the full event history into live subscriber
queues, avoiding spurious `DropSubscription` overflows for healthy
`AllStreams` subscribers while they are still catching up. See
`docs/plans/38-publisher-tail-init-to-avoid-restart-rebroadcast-and-allstreams-overflow.md`.
#### Category/consumer-group live subscriptions busy-spinning (plan 37)
* Live `Category` subscriptions and consumer-group members no longer busy-spin
when idle while other categories/partitions advance the global `$all`
position. `Category` subscriptions now wake from a per-category NOTIFY signal
(the Notifier previously discarded the notification payload), so an idle
category does zero DB work; consumer-group members now gate on the last
observed global position instead of the per-category cursor. Delivery and
checkpoint semantics are unchanged. See
`docs/plans/37-fix-category-subscription-live-loop-busy-spin.md`.
#### Live fetch observability
* `KirokuEventSubscriptionFetched` observability event (per live DB-driven
fetch), exposing per-subscription live-fetch activity.
## 0.1.0.0 — 2026-05-23
### Changed — migrations own schema lifecycle
* `kiroku-store` no longer embeds schema SQL or runs DDL during `withStore`.
Apply `kiroku-store-migrations` before opening the store. This removes the
duplicated bootstrap path between the runtime package and the migration
package.
* Removed `Kiroku.Store.Schema`, `SchemaInitialization`,
`InitializeSchemaOnAcquire`, `SkipSchemaInitialization`, and
`ConnectionSettingsM.schemaInitialization`.
### Added — `lookupStreamNames` / `lookupStreamName` (plan 36)
* `Kiroku.Store.Read.lookupStreamNames :: [StreamId] -> Eff es (Map StreamId
StreamName)` resolves a batch of surrogate stream ids to their names in a
single round trip (and `lookupStreamName :: StreamId -> Eff es (Maybe
StreamName)` for one). This is the inverse of `lookupStreamId` and the
supported way to recover the source stream name from a `RecordedEvent`'s
`originalStreamId` on fan-in reads — `$all`, categories,
causation/correlation queries, and subscriptions — which carry only the
surrogate id.
* Chosen over carrying a `originalStreamName` field on every `RecordedEvent`:
benchmarking showed returning the name on every read row costs ~13% on
`$all` 100-event pages (decoding ~100 extra `text` values per page), whether
the name came from a join or a denormalized column. The lookup API keeps the
read hot path at its prior latency and makes consumers pay only when, and
only as much as, they actually need names (one round trip per batch for the
distinct ids). See `docs/plans/36-add-originalstreamname-to-recordedevent.md`.
* A fresh installation now creates and uses a dedicated `kiroku` PostgreSQL
schema instead of installing into `public`. The
`kiroku-store-migrations` bootstrap creates the schema and sets
`search_path` first, and every pooled connection runs
`SET search_path TO "<schema>", pg_catalog` before any statement.
* `defaultConnectionSettings` now defaults `schema = "kiroku"` (was `"public"`).
The `schema` field is now authoritative for object location, table
resolution, and the `LISTEN <schema>.events` channel — previously it only
named the notification channel while table resolution depended on the
connection-string `search_path`.
* The `kiroku-store-migrations` codd bootstrap installs into `kiroku`; run it
with `CODD_SCHEMAS=kiroku`. The runtime role needs privileges on the
`kiroku` schema rather than on `public`.
### Added — consumer groups for partitioned subscriptions (MasterPlan 4, plans 28–31)
* Static, hash-partitioned consumer groups: a named subscription can be split
across `N` members, each processing a disjoint, per-stream-ordered slice in
parallel, with per-member checkpoints. Works for `Category` and `$all`
targets.
* `Kiroku.Store.Subscription.Types` (re-exported from `Kiroku.Store`):
* `ConsumerGroup { member :: Int32, size :: Int32 }` — static membership
descriptor. A stream is assigned to member
`(((hashtextextended(stream_id::text, 0) % size) + size) % size)`.
* Two new fields on `SubscriptionConfigM m`: `consumerGroup :: Maybe
ConsumerGroup` (default `Nothing`) and `consumerGroupGuard :: Bool`
(default `False`, an opt-in startup advisory-lock conflict probe).
`defaultSubscriptionConfig` sets both defaults, so existing callers compile
unchanged.
* Exceptions `InvalidConsumerGroup` (thrown by `subscribe` when `size < 1` or
`member` is out of range) and `ConsumerGroupGuardConflict` (thrown at
startup when the guard detects a duplicate member).
* `subscriptions` table gains `consumer_group_member` and `consumer_group_size`
columns; the checkpoint key becomes the composite unique
`(subscription_name, consumer_group_member)` (index
`ix_subscriptions_name_member`). The change is applied by the migration
package.
* The four `KirokuEventSubscription*` observability events carry a trailing
`SubscriptionGroupContext` (`NonGroup` | `GroupMember member size`),
re-exported from `Kiroku.Store`.
* Surfaced through every subscription entry point: the `MonadIO`
`subscribe`/`withSubscription`, the effectful `Subscription` effect, the
Streamly `subscriptionStream` bridge, and the Shibuya adapter
(`KirokuAdapterConfig` gains `consumerGroup :: Maybe ConsumerGroup`).
* New user guide `docs/user/consumer-groups.md` and a runnable example,
`cabal run kiroku-store:kiroku-consumer-group-example`.
### Added — causation chain and correlation walkers (plan 14)
* `Kiroku.Store.Causation` (new module, re-exported from `Kiroku.Store`):
* `findCausationDescendants :: EventId -> Eff es (Vector RecordedEvent)`
— walk the causation graph forward from a seed event. Returns the
seed plus every event whose `causation_id` chain leads back to it,
in ascending `global_position` order. Backed by the existing partial
index `ix_events_causation_id`, so cost scales as O(depth · log n)
against the total event count.
* `findCausationAncestors :: EventId -> Eff es (Vector RecordedEvent)`
— symmetric walk in the other direction: returns the seed plus
every ancestor reachable by following `causation_id` upward, in
leaf-first depth order. Uses the same index.
* `findByCorrelation :: UUID -> Eff es (Vector RecordedEvent)` — return
every event whose `correlation_id` equals the input, in ascending
`global_position` order. Backed by `ix_events_correlation_id`.
* `Kiroku.Store.Effect.Store` gains a single new constructor
`FindEvents :: EventFilter -> Store m (Vector RecordedEvent)` whose
argument is a closed sum (`Kiroku.Store.Types.EventFilter` —
`FilterCorrelation`, `FilterCausationDescendants`, `FilterCausationAncestors`).
The closed-sum shape preserves exhaustiveness checking for downstream
mock interpreters.
* The new reads honor `StoreSettings.decodeHook`, so any interpreter-level
decode customization wired through `ConnectionSettings` applies on
parity with `readStreamForward` / `readAllForward` / `readCategory`.
### Added — interpreter-level event-data hooks (plan 13)
* `Kiroku.Store.Settings` (new module, re-exported from `Kiroku.Store`):
* `StoreSettings { enrichEvent :: Maybe (EventData -> IO EventData),
decodeHook :: Maybe (RecordedEvent -> IO RecordedEvent) }` — optional
interpreter-level hooks. `enrichEvent` runs on the append path before
encoding (on the typed `EventData` the caller supplied); `decodeHook`
runs on the read and subscription paths after decoding (on the typed
`RecordedEvent` about to be surfaced). Both default to `Nothing`,
taking a `pure` fast path that adds no traversal or allocation when
the hook is absent.
* `defaultStoreSettings` — both fields `Nothing` (no-op).
* `enrichEvents :: StoreSettings -> [EventData] -> IO [EventData]` and
`decodeEvents :: StoreSettings -> Vector RecordedEvent -> IO (Vector
RecordedEvent)` — internal helpers reused by the interpreter, the
subscription publisher, the subscription worker, and the new
`enrichEventsIO` convenience.
* `Kiroku.Store.Connection.ConnectionSettings` and
`Kiroku.Store.Connection.KirokuStore` gain a `storeSettings ::
StoreSettings` field. `defaultConnectionSettings` seeds it with
`defaultStoreSettings` so existing callers see no behaviour change.
`withStore` copies the value onto the runtime handle for the
interpreter, the publisher, and the worker to read.
* `Kiroku.Store.Transaction`:
* `runTransactionAppendingResource` /
`runTransactionAppendingResourceNoRetry` — hook-aware variants of
`runTransactionAppending` / `runTransactionAppendingNoRetry` for
callers running under a `KirokuStoreResource` effect. They apply
`enrichEvent` to every `EventData` before opening the transaction.
* `enrichEventsIO :: KirokuStore -> [EventData] -> IO [EventData]` —
public convenience for direct callers of `appendToStreamTx`, who
bypass the interpreter.
A typical use case is enriching every appended event with an
OpenTelemetry trace context:
```haskell
storeSettings = defaultStoreSettings
{ enrichEvent = Just $ \ed -> do
ctx <- captureCurrentSpan
pure (ed & #metadata %~ injectTraceContext ctx)
}
```
Wired through `ConnectionSettings`'s `storeSettings` field, the hook
fires uniformly across `appendToStream`, `appendMultiStream`,
`runTransactionAppendingResource`, every read path, and the
subscription pipeline (live + catch-up).
### Added — streaming single-stream forward read
* `Kiroku.Store.Read.readStreamForwardStream` (re-exported from
`Kiroku.Store`): a Streamly `Stream (Eff es) RecordedEvent` companion to
`readStreamForward`. Internally paginates `readStreamForward` at a
caller-supplied page size and yields events one at a time, enabling
constant-memory folds over long streams. Shares SQL path and error
semantics with `readStreamForward`.
### Added — single-stream `RunTransaction` combinator (plan 11)
* `Kiroku.Store.Transaction` (new module, re-exported from
`Kiroku.Store`):
* `runTransaction` / `runTransactionNoRetry` — bare escape hatch for
running an arbitrary `Hasql.Transaction.Transaction a` against the
store's connection pool. The default variant retries the body on
PostgreSQL serialization conflicts; the `-NoRetry` variant runs
exactly once. Both execute at `ReadCommitted` isolation in `Write`
mode.
* `appendToStreamTx :: StreamName -> ExpectedVersion ->
[PreparedEvent] -> UTCTime -> Tx.Transaction (Either AppendConflict
AppendResult)` — the `Tx`-flavored single-stream append building
block. Useful inside a `runTransaction` block when the caller needs
full control over conflict handling. Does not enforce reserved-stream
rejection; pair with `prepareEventsIO` for UUIDv7 / `createdAt`
prep.
* `runTransactionAppending` /
`runTransactionAppendingNoRetry` — recommended high-level wrapper
for the `keiro` use case. Atomically composes a single-stream append
with a caller-supplied `Tx.Transaction` continuation. Rejects `$all`
up front, `Tx.condemn`s on append conflicts (the continuation never
runs), and surfaces both conflict and connection failures through
the `Either StoreError` return type. Carries `IOE :> es` in addition
to `Store :> es`.
* `Kiroku.Store.Effect`:
* New `Store` constructors `RunTransaction` and
`RunTransactionNoRetry`. Mock interpreters are expected to reject
these.
* Internal building blocks `PreparedEvent`, `prepareEvents`,
`buildAppendParams`, and `appendDispatchTx` are now exported (under
the `$internal` haddock group) to support
`Kiroku.Store.Transaction`. Not part of the supported public
surface.
* `Kiroku.Store.Error`:
* `AppendConflict` — sum of append-precondition failures observable
inside a `Tx.Transaction` body
(`WrongExpectedVersionConflict` / `StreamNotFoundConflict` /
`StreamAlreadyExistsConflict`). 1:1 with the corresponding
`StoreError` constructors.
* `appendConflictToStoreError`, `emptyResultConflict` — supporting
pure helpers.
### Changed
* `appendMultiStream`'s interpreter now dispatches per-stream appends
through the shared `appendDispatchTx` helper. Behavior is unchanged.
* `appendToStream`'s haddock now describes when to reach for
`runTransactionAppending` instead.