journal

type AmbiguousAckError ¶

type AmbiguousAckError struct {
	Subject  string
	MsgID    string
	Expected uint64
	Cause    error
}

AmbiguousAckError reports an append whose outcome the backend could not resolve: the persist call was lost or timed out and a bounded retry stayed ambiguous, so the serializer cannot tell whether the record landed. The fence stays unadvanced, so the next Append re-fences on the same tip; the caller decides whether to fail the session or retry later. It carries the record's routing/destination identifier, its idempotency id, the expected sequence the append fenced on, and the underlying cause.

func (e *AmbiguousAckError) Error() string

func (e *AmbiguousAckError) Unwrap() error

type AppendError ¶

type AppendError struct {
	Subject  string
	MsgID    string
	Expected uint64
	Cause    error
}

AppendError wraps a definite failure to persist a record to the session log. It carries the record's routing/destination identifier, its idempotency id, and the expected sequence the append was fenced under, and unwraps to the underlying backend error (a context deadline, a fence rejection, a transport error). The fence stays unadvanced when this is returned, so the next Append re-fences on the same tip.

func (e *AppendError) Error() string

func (e *AppendError) Unwrap() error

type AppendFunc ¶

type AppendFunc func(context.Context, JournalRecord) (uint64, error)

AppendFunc is one journal append operation.

type AppendMiddleware ¶

type AppendMiddleware func(next AppendFunc) AppendFunc

AppendMiddleware decorates one AppendFunc. Implementations must invoke next synchronously exactly once with the supplied record and return its exact sequence and error.

func HookMiddleware(runner *hook.Runner, sessionID uuid.UUID) AppendMiddleware

type AppendResult ¶

type AppendResult struct {
	Sequence uint64
	Appended bool
}

AppendResult reports the outcome of an append issued through an IdempotentJournal: the durable sequence the record occupies, and whether this call durably persisted a NEW frame (Appended=true) or deduplicated an identical retry of an already-durable record (Appended=false; Sequence is then the ORIGINAL append's sequence, not a new one).

type AppenderOption ¶

type AppenderOption func(*JournalEventAppender)

AppenderOption configures a JournalEventAppender at construction. Applied in order over a defaults struct (nop catalog), so a later option overrides an earlier one.

func WithCatalog(c catalogUpdater) AppenderOption

type AroundAppend ¶ added in v0.31.0

type AroundAppend func(ctx context.Context, rec JournalRecord, next func(context.Context) error) error

AroundAppend decorates one durable append of ANY result type. next must be invoked exactly once, with the context the delegated call should run under — a decorator may substitute its own (the operation-hook decorator does) or pass the caller's through (the offload-GC admission gate does). The error next returns is the append's error; returning a different one substitutes it, and returning without calling next skips the append entirely.

The result VALUE never appears here, which is the point: an append seam that returns a sequence, an AppendResult, or a CommittedAppendResult is decorated by the same function, so a decorator is written once and cannot be written wrong for one seam and right for another.

type CommandApplicationDecodeError ¶ added in v0.31.0

type CommandApplicationDecodeError struct {
	Reason string
	Cause  error
}

CommandApplicationDecodeError wraps a failure to decode CommandApplicationRecord bytes at the untrusted restore boundary: malformed JSON, an unknown field, trailing data, or a correlation that no valid admitted record could have produced. It fails secure — a prefix that cannot be trusted must never be read as "this command was already applied", because that answer suppresses the application entirely.

func (e *CommandApplicationDecodeError) Error() string

func (e *CommandApplicationDecodeError) Unwrap() error

type CommandApplicationEncodeError ¶ added in v0.31.0

type CommandApplicationEncodeError struct{ Cause error }

CommandApplicationEncodeError wraps a failure to marshal a CommandApplicationRecord to JSON.

func (e *CommandApplicationEncodeError) Error() string

func (e *CommandApplicationEncodeError) Unwrap() error

type CommandApplicationRecord ¶ added in v0.31.0

type CommandApplicationRecord struct {
	// contains filtered or unexported fields
}

CommandApplicationRecord is the PRIVATE durable record that correlates a public CommandID with the one RuntimeCommandID it maps to and the lease epoch the application runs under. It is appended BEFORE the command's runtime-visible effect, so a redelivery after a crash finds it and replays the original disposition rather than applying the command twice.

It is not an event and not a command: it is never projected to a public wire body, never replayed as an event, and the EventReplayer never decodes it. Existing Harness command headers keep their UUIDs; this record is the durable bridge from Host's opaque identity to those UUIDs, nothing more.

Its idempotency id is derived from the PUBLIC CommandID, which is what makes duplicate delivery detectable at the append: an idempotent journal reports Appended=false for a byte-identical retry and fails closed with an *IdempotencyCollisionError when the same public id names a DIFFERENT mapping.

func NewCommandApplicationRecord(app runtimecommand.Application) CommandApplicationRecord

func UnmarshalCommandApplicationRecord(data []byte) (CommandApplicationRecord, error)

func (r CommandApplicationRecord) Application() runtimecommand.Application

func (r CommandApplicationRecord) IdempotencyID() string

type CommandRecord ¶

type CommandRecord struct {
	// contains filtered or unexported fields
}

CommandRecord wraps a command.Command targeting a specific loop. Unlike an event, a command does not uniformly carry its routing coordinates (Interrupt and Shutdown carry only a Header; the session dispatches them by other means), so the writer supplies the target sessionID/loopID at construction. Its id is the command's physical CommandRecordID. The serializer encodes the wrapped command via command.MarshalCommand.

func NewCommandRecord(sessionID, loopID uuid.UUID, cmd command.Command) CommandRecord

func (r CommandRecord) Command() command.Command

func (r CommandRecord) DeliveryPhase() command.DelegateDeliveryPhase

func (r CommandRecord) IdempotencyID() string

func (r CommandRecord) LogicalCommandID() uuid.UUID

func (r CommandRecord) LoopID() uuid.UUID

func (r CommandRecord) NormalizedDeliveryFingerprint() (Fingerprint, error)

func (r CommandRecord) PhysicalID() CommandRecordID

func (r CommandRecord) SessionID() uuid.UUID

type CommandRecordID ¶

type CommandRecordID struct {
	CommandID uuid.UUID
	Phase     command.DelegateDeliveryPhase
}

CommandRecordID is the physical idempotency identity of one command record. Delivery fallback records retain the logical command UUID while using their phase as a typed suffix, so the intent and fallback frames can both be durable without weakening the journal's ordinary idempotency collision rule.

func (id CommandRecordID) String() string

type CommandRouteMismatchError ¶

type CommandRouteMismatchError struct {
	RecordLoopID uuid.UUID
	TargetLoopID uuid.UUID
}

CommandRouteMismatchError reports disagreement between a durable delegate command's embedded target and the live CommandRecord dispatch route.

func (e *CommandRouteMismatchError) Error() string

type CommittedAppendResult ¶ added in v0.31.0

type CommittedAppendResult struct {
	AppendResult
	Public CommittedPublicBody
}

CommittedAppendResult is AppendResult widened with the committed public bytes. It embeds AppendResult so every existing Sequence/Appended reading applies unchanged.

type CommittedPublicBody ¶ added in v0.31.0

type CommittedPublicBody struct {
	EventID string
	Body    []byte
}

CommittedPublicBody is what a durable journal stored, for one record, on the PUBLIC side of its envelope: the canonical public event id it committed the record under and the exact canonical body bytes it wrote. Both are zero for a private record, for a non-event record, and for a deduplicated retry (which stored nothing). Body is owned by the returned value; the journal must not retain a reference it later mutates.

type CommittedPublicJournal ¶ added in v0.31.0

type CommittedPublicJournal interface {
	IdempotentJournal
	// AppendCommitted behaves exactly like AppendIdempotent — same fencing, same
	// dedup, same errors — and additionally reports the public event id and the
	// exact canonical public body it stored for a newly appended PUBLIC event
	// record. A deduplicated retry reports Appended=false with a zero Public: this
	// call stored nothing, so it has no stored bytes of its own to report.
	AppendCommitted(ctx context.Context, rec JournalRecord) (CommittedAppendResult, error)
}

CommittedPublicJournal is the OPTIONAL extension a SessionJournal implementation may satisfy to report the EXACT canonical public bytes it stored for a record. It embeds IdempotentJournal, so a committed-bytes implementation is usable anywhere a plain or idempotent SessionJournal is expected and the existing seams are never weakened.

The reason the bytes are reported rather than re-derived is that a second projection is a second answer. A consumer joining a durable tail to a live stream dedupes on (sequence, event id) and compares bodies; if the live body were re-projected it could differ from the stored one — in key order, in a field a later projector version adds — and the consumer would render two different things for one event without any error anywhere. Only the journal that wrote the bytes can say what they are.

type DeliveryTransitionError ¶

type DeliveryTransitionError struct {
	CommandID uuid.UUID
	Phase     command.DelegateDeliveryPhase
	Reason    string
}

DeliveryTransitionError reports a phased delegate command that violates the logical intent-to-fallback ordering or otherwise cannot participate in the transition index. It carries only bounded identity/category data; payloads are deliberately excluded.

func (e *DeliveryTransitionError) Error() string

type EventCursor ¶

type EventCursor interface {
	// Next returns the next event and its sequence, or io.EOF when the cold backlog is
	// exhausted. A decode/read error fails secure: the cursor surfaces the typed error
	// rather than skipping or zero-valuing the record.
	Next(ctx context.Context) (event.Event, uint64, error)
	// Close tears down the reader. Idempotent: a second call is a no-op.
	Close() error
}

EventCursor yields a session's Enduring events in sequence order. Next returns the next decoded event with its sequence, io.EOF once the backlog is drained (cold mode), or a typed error on a malformed/missing/corrupt record. Close releases the underlying reader; it is idempotent and safe to call after an error.

type EventRecord ¶

type EventRecord struct {
	// contains filtered or unexported fields
}

EventRecord wraps an Enduring event.Event as a JournalRecord. The event already carries its producer Coordinates and Scope; its id is the event's EventID. The serializer encodes the wrapped event via event.MarshalEvent (which fails closed on an Ephemeral event); the record never re-encodes.

func NewEventRecord(ev event.Event) EventRecord

func (r EventRecord) Event() event.Event

func (r EventRecord) IdempotencyID() string

type EventReplayer ¶

type EventReplayer interface {
	// Open binds a cursor over the session's events selected by req and positioned at
	// req.From.
	Open(ctx context.Context, req ReplayRequest) (EventCursor, error)
}

EventReplayer is the journal's read side: it opens an ordered cursor over a session's Enduring events. It is the narrow counterpart to SessionJournal (the write side) — a caller that only reads history depends on Open alone. The concrete implementation lives in a backend package (e.g. pkg/sessionstore over storage).

type FenceDecodeError ¶

type FenceDecodeError struct {
	Reason string
	Cause  error
}

FenceDecodeError wraps a failure to decode LeaseFence bytes at the untrusted restore boundary: malformed JSON, a wrong field type, a non-object, or trailing data after the object. The codec fails closed with this typed error so callers inspect the cause via errors.As rather than guessing an epoch.

func (e *FenceDecodeError) Error() string

func (e *FenceDecodeError) Unwrap() error

type FenceEncodeError ¶

type FenceEncodeError struct{ Cause error }

FenceEncodeError wraps a failure to marshal a LeaseFence to JSON. A LeaseFence is a single uint64, so this is effectively unreachable, but the codec returns a typed error rather than dropping the json.Marshal error to satisfy the errors-are-typed contract at the package API.

func (e *FenceEncodeError) Error() string

func (e *FenceEncodeError) Unwrap() error

type FenceRecord ¶

type FenceRecord struct {
	// contains filtered or unexported fields
}

FenceRecord wraps a LeaseFence as a JournalRecord. The fence carries no session id of its own, so the writer supplies the target sessionID at construction; its idempotency id is the epoch.

func NewFenceRecord(sessionID uuid.UUID, fence LeaseFence) FenceRecord

func (r FenceRecord) Fence() LeaseFence

func (r FenceRecord) IdempotencyID() string

func (r FenceRecord) SessionID() uuid.UUID

type Fingerprint ¶

type Fingerprint struct {
	// contains filtered or unexported fields
}

Fingerprint identifies a record's persisted kind and payload bytes — exactly what a backend durably writes — independent of any transient in-memory routing a record wrapper additionally carries (CommandRecord's session/loop dispatch target is never persisted, so it never enters a Fingerprint: a backend derives Fingerprint from the SAME (kind, codec-marshaled body) pair it writes to the log, never from the record's Go value). Two records fingerprint equal if and only if a backend would durably persist byte-identical frames for them.

func NewFingerprint(kind string, body []byte) Fingerprint

type FollowUnsupportedError ¶

type FollowUnsupportedError struct {
	Stream string
}

FollowUnsupportedError is returned by Open when ReplayRequest.Follow is true and the backend implements only the cold (Follow:false) backlog read. Failing closed with a typed error is preferable to silently behaving as a cold cursor that EOFs at the current tip. It carries the log identifier the replay targeted.

func (e *FollowUnsupportedError) Error() string

type GatePreparedDecodeError ¶

type GatePreparedDecodeError struct {
	Stage string // "json", "prepared", or "payload"
	Cause error
}

GatePreparedDecodeError wraps a failure to decode GatePreparedRecord bytes at the untrusted restore boundary: malformed JSON, a malformed embedded event, or a malformed embedded payload. It fails secure rather than skipping or zero-valuing the record.

func (e *GatePreparedDecodeError) Error() string

func (e *GatePreparedDecodeError) Unwrap() error

type GatePreparedEncodeError ¶

type GatePreparedEncodeError struct {
	Stage string // "prepared" or "payload"
	Cause error
}

GatePreparedEncodeError wraps a failure to marshal a GatePreparedRecord to JSON: either the embedded GatePrepared event or the gate.OpenPayload failed its codec.

func (e *GatePreparedEncodeError) Error() string

func (e *GatePreparedEncodeError) Unwrap() error

type GatePreparedRecord ¶

type GatePreparedRecord struct {
	// contains filtered or unexported fields
}

GatePreparedRecord is the PRIVATE durable record for a gate's prepare step. It carries the GatePrepared event (the public envelope stored privately until ActivateGate appends the public GateOpened) PLUS the sealed gate.Payload the resolver needs for response validation and restore — a payload that must NEVER be exposed to SSE/history and must NEVER be appended through NewEventRecord or hub.PublishEvent. Its idempotency id is the prepared event's EventID.

func NewGatePreparedRecord(prepared event.GatePrepared, payload gate.OpenPayload) GatePreparedRecord

func UnmarshalGatePreparedRecord(data []byte) (GatePreparedRecord, error)

func (r GatePreparedRecord) IdempotencyID() string

func (r GatePreparedRecord) Payload() gate.OpenPayload

func (r GatePreparedRecord) Prepared() event.GatePrepared

type IdempotencyCollisionError ¶

type IdempotencyCollisionError struct {
	ID  string
	Seq uint64 // the ledger sequence the ORIGINAL (colliding) record already occupies
}

IdempotencyCollisionError reports that a record's idempotency id already names a durable record in the log with a DIFFERENT persisted kind or payload — a genuine id collision (a bug or a forged retry), never a legitimate duplicate retry (which is always byte-identical to what is already durable). The append fails closed rather than silently accepting a differently-shaped record under a reused id.

func (e *IdempotencyCollisionError) Error() string

type IdempotencyIndex ¶

type IdempotencyIndex struct {
	// contains filtered or unexported fields
}

IdempotencyIndex tracks, for every idempotency id already durable in a session's log, the ledger sequence it occupies and the Fingerprint of what was persisted under it. A backend hydrates one from its full durable ledger (see the sessionstore package) before accepting new appends, then consults and updates it — via Check and Observe — under the same lock that already serializes its writes. It is NOT safe for concurrent use on its own; the caller's append-serializing lock is its only synchronization.

func NewIdempotencyIndex() *IdempotencyIndex

func (idx *IdempotencyIndex) Check(id string, fp Fingerprint) (seq uint64, duplicate bool, err error)

func (idx *IdempotencyIndex) Observe(id string, seq uint64, fp Fingerprint)

type IdempotentJournal ¶

type IdempotentJournal interface {
	SessionJournal
	// AppendIdempotent behaves exactly like Append — same fencing, same errors —
	// except a record whose IdempotencyID() already names a durable record with an
	// IDENTICAL persisted kind+payload is detected and reported via
	// AppendResult.Appended=false (carrying the ORIGINAL sequence) rather than
	// durably appended a second time. A record whose id names a durable record with
	// a DIFFERENT persisted kind or payload fails closed with a typed
	// *IdempotencyCollisionError.
	AppendIdempotent(ctx context.Context, rec JournalRecord) (AppendResult, error)
}

IdempotentJournal is the OPTIONAL extension a SessionJournal implementation may satisfy to deduplicate a redelivered append by idempotency id. It embeds SessionJournal so an idempotent implementation is usable anywhere a plain SessionJournal is expected — the existing narrow Append seam is never weakened or replaced. A caller that additionally wants to know whether ITS OWN call produced a new durable frame or deduplicated a retry (e.g. to skip a live broadcast for a duplicate) type-asserts for IdempotentJournal and calls AppendIdempotent instead of Append.

type JournalCommandAppender ¶

type JournalCommandAppender struct {
	// contains filtered or unexported fields
}

JournalCommandAppender adapts a SessionJournal to the narrow "append one command" seam the session depends on for the intent log. The session holds an unexported commandAppender interface (AppendCommand(ctx, CommandRecord) error); this type satisfies it structurally, so the composition root (Phase 10) wires it in without the session importing journal internals beyond the CommandRecord constructor (Dependency Inversion). It carries no state beyond the journal — one method, one responsibility: append a CommandRecord (which the SESSION built with the dispatch target loopID, since a command — Interrupt/Shutdown especially — does not carry its own routing) and return the underlying typed error.

Unlike the event appender, the SESSION treats this seam as AUDIT-ONLY: a non-nil error is logged and the dispatch proceeds (losing a command record must never block the user's action). This façade itself never swallows — it returns the journal's error unchanged so the session owns the log-and-proceed decision.

func NewJournalCommandAppender(journal SessionJournal) *JournalCommandAppender

func NewJournalCommandAppenderChecked(journal SessionJournal) (*JournalCommandAppender, error)

func (a *JournalCommandAppender) AppendCommand(ctx context.Context, rec CommandRecord) error

type JournalEventAppender ¶

type JournalEventAppender struct {
	// contains filtered or unexported fields
}

JournalEventAppender adapts a SessionJournal (the write side) to the narrow "append one Enduring event" seam the session hub depends on. The hub holds an unexported eventAppender interface (AppendEvent(ctx, event.Event) error); this type satisfies it structurally, so the composition root (Phase 10) wires it in via hub.WithAppender without the hub ever importing the journal package (Dependency Inversion). Beyond the journal it holds an optional catalog updater (nop by default): after a successful Append it notifies the catalog best-effort so the replay-free session index stays current. One responsibility: wrap the event in an EventRecord (which self-derives its subject from the event's scope+coordinates and its idempotency id from the EventID), append it, then best-effort index it.

func NewJournalEventAppender(journal SessionJournal, opts ...AppenderOption) *JournalEventAppender

func NewJournalEventAppenderChecked(journal SessionJournal, opts ...AppenderOption) (*JournalEventAppender, error)

func (a *JournalEventAppender) AppendEvent(ctx context.Context, ev event.Event) (uint64, error)

func (a *JournalEventAppender) AppendEventCommitted(ctx context.Context, ev event.Event) (event.AppendCommit, error)

func (a *JournalEventAppender) AppendEventResult(ctx context.Context, ev event.Event) (uint64, bool, error)

func (a *JournalEventAppender) SupportsCommittedPublicBodies() bool

type JournalGateAppender ¶

type JournalGateAppender struct {
	// contains filtered or unexported fields
}

JournalGateAppender adapts a SessionJournal to the session gate directory's strict durable append seam. GatePreparedRecord is appended as a private record; GateOpened and GateResolved are public Enduring events and are wrapped in EventRecord exactly like JournalEventAppender.

func NewJournalGateAppender(journal SessionJournal) *JournalGateAppender

func NewJournalGateAppenderChecked(journal SessionJournal) (*JournalGateAppender, error)

func (a *JournalGateAppender) AppendGateOpened(ctx context.Context, ev event.GateOpened) error

func (a *JournalGateAppender) AppendGatePrepared(ctx context.Context, rec GatePreparedRecord) error

func (a *JournalGateAppender) AppendGateResolved(ctx context.Context, ev event.GateResolved) error

type JournalLeaseLostError ¶

type JournalLeaseLostError struct {
	SessionID uuid.UUID
	Epoch     uint64
}

JournalLeaseLostError reports an Append refused because the journal's ownership lease was lost — released by the holder or overtaken by a higher epoch. Once the lease is gone the journal fails every append fast and never re-fetches or advances its expected sequence: a new owner (higher epoch) has written, or will write, its own LeaseFence, so this stale journal's fence would reject the append anyway. Failing here is the fast-path guard; the backend fence is the hard backstop. It carries the session and the lost lease's epoch and unwraps to a *LeaseLostError for errors.As.

func (e *JournalLeaseLostError) Error() string

func (e *JournalLeaseLostError) Unwrap() error

type JournalNotReadyError ¶

type JournalNotReadyError struct {
	SessionID uuid.UUID
}

JournalNotReadyError reports an Append attempted before the journal's opening LeaseFence was acknowledged. The journal writes the LeaseFence as its first append and only marks itself ready once it lands; an Append before that fails closed with this typed error rather than racing the fence. It carries the session so a caller can correlate the failure.

func (e *JournalNotReadyError) Error() string

type JournalRecord ¶

type JournalRecord interface {

	// IdempotencyID is the stable per-record id a backend uses as its de-dup key so
	// a redelivered append de-duplicates: an event's EventID, a command's physical
	// CommandRecordID, or a fence's epoch.
	IdempotencyID() string
	// contains filtered or unexported methods
}

JournalRecord is the sealed sum a session's serialized writer persists: an Enduring event, a command (the intent log), or an internal LeaseFence. It is a marker plus the one backend-neutral fact the writer needs to persist without re-inspecting the payload — the record's idempotency id (the stable per-record id a backend uses to de-duplicate a redelivered append). The concrete payload codec is the existing event/command marshaler; a record only carries the typed payload and exposes how to identify it. How a record is routed/stored is the backend's concern (a storage ledger name, a subject, …), never the record's.

The set is sealed by the unexported isJournalRecord marker: only the wrapper types in this package implement it, so the serializer's switch over the sum is exhaustive and a foreign type can never masquerade as a record.

type JournalRuntimeCommandAppender ¶ added in v0.31.0

type JournalRuntimeCommandAppender struct {
	// contains filtered or unexported fields
}

JournalRuntimeCommandAppender adapts an IdempotentJournal to the narrow "append one application prefix" seam the session's runtime-command applier depends on. It is a SEPARATE appender from JournalCommandAppender on purpose: the intent log is audit-only and swallows its failures, while this append is the crash-safety barrier in front of a runtime-visible effect and its failure MUST stop the effect.

func NewJournalRuntimeCommandAppenderChecked(journal SessionJournal) (*JournalRuntimeCommandAppender, error)

func (a *JournalRuntimeCommandAppender) AppendCommandApplication(ctx context.Context, app runtimecommand.Application) (AppendResult, error)

type Lease ¶

type Lease interface {

	// SessionID is the session this lease grants single-writer ownership of.
	SessionID() uuid.UUID
	// Release relinquishes the lease: stops whatever the provider does to keep the
	// grant live, marks it no longer held (firing Lost), and best-effort clears the
	// entry so a successor can re-acquire.
	//
	// WHAT ENDS A GRANT, IN THIS BUILD. The upstream storage.Leaser contract calls a
	// grant "renewable" and its storetest.TestLeaserLifecycle suite conforms
	// providers that implement renewal and expiry, so an implementation MAY end a
	// grant by expiry or higher-epoch takeover. Neither pinned provider does:
	// memstore is Release-only ("no TTL, no takeover"), and fsstore's grant is an
	// advisory lock the OS drops when the holding fd closes or the process exits. So
	// on the backends this module is built against, Release is the only prompt end to
	// a grant, and an unreleased one is held for as long as its holder lives. Code
	// here must therefore treat loss as something it CAUSES, never as something that
	// will happen on its own. Idempotent.
	Release(ctx context.Context) error
	// contains filtered or unexported methods
}

Lease is the single-writer ownership token for one session's durable log. A SessionJournal depends on it (DIP): the composition root acquires a Lease from a backend and passes it in; the journal stamps the lease's Epoch into its first LeaseFence and refuses to append once the lease is lost. The holder (the composition root), not the journal, calls Release — the journal only reads Epoch and the validity/loss signals (see ownershipToken, the narrower view it depends on).

type LeaseFence ¶

type LeaseFence struct {
	Epoch uint64 `json:"epoch"`
}

LeaseFence is an internal journal record marking a lease-handover boundary: a monotonically increasing Epoch fenced into the stream when ownership of a session's writer lease changes. It is journal-private (not an event or a command); the EventReplayer never decodes it. Codec: MarshalLeaseFence/UnmarshalLeaseFence in record_json.go.

func UnmarshalLeaseFence(data []byte) (LeaseFence, error)

type LeaseHeldError ¶

type LeaseHeldError struct {
	SessionID uuid.UUID
	Epoch     uint64
}

LeaseHeldError reports that acquiring a lease lost the single-holder race: the session's lease is currently held by a live holder, or a concurrent acquirer won the race. "Live" is not "unexpired" — no pinned provider expires a grant, so a holder stays live until it releases or its process dies, and retrying this later in the same process will keep failing unless the holder acts. It carries the session and the epoch currently fenced so a caller can log who holds it. It is the expected, non-fatal "someone else owns this session" outcome — the loser must not write to the log.

func (e *LeaseHeldError) Error() string

type LeaseLostError ¶

type LeaseLostError struct {
	SessionID uuid.UUID
	Epoch     uint64
}

LeaseLostError reports an operation attempted on a lease that is no longer held: it was released, its holding process exited (fsstore), or — on a provider that implements one, which neither pinned provider does — it expired or was taken over by a higher epoch. It carries the session and the lease's epoch. The journal returns it (wrapped in a JournalLeaseLostError) when an Append is attempted after the lease is lost.

func (e *LeaseLostError) Error() string

type MarshalRecordError ¶

type MarshalRecordError struct {
	Subject string
	Cause   error
}

MarshalRecordError wraps a failure to encode a record's payload before it is persisted. It names the record's routing/destination identifier so a caller can correlate the failure without re-inspecting the payload, and unwraps to the underlying codec error (an *event.EphemeralNotPersistableError, *command.UnknownCommandTypeError, a *FenceEncodeError, …) for errors.As inspection.

func (e *MarshalRecordError) Error() string

func (e *MarshalRecordError) Unwrap() error

type NilJournalError ¶

type NilJournalError struct{}

NilJournalError reports that a JournalEventAppender was constructed over a nil SessionJournal — a composition-wiring bug. The checked constructor fails loud with this typed error rather than letting the nil surface as a panic at the first append.

func (*NilJournalError) Error() string

type NonIdempotentJournalError ¶ added in v0.31.0

type NonIdempotentJournalError struct{}

NonIdempotentJournalError reports that a runtime-command appender was asked to wrap a SessionJournal that cannot deduplicate a redelivered append. It is a CAPABILITY refusal, not a wiring bug: the duplicate-delivery contract is implemented BY the journal's dedup, so a journal without it cannot honor the contract and the seam must decline to advertise it rather than apply an admitted command twice.

func (*NonIdempotentJournalError) Error() string

type RecordCursor ¶

type RecordCursor interface {
	// Next returns the next record and its sequence, or io.EOF when the cold backlog is
	// exhausted. A decode/read error fails secure: the cursor surfaces the typed error
	// rather than skipping or zero-valuing the record.
	Next(ctx context.Context) (JournalRecord, uint64, error)
	// Close tears down the reader. Idempotent: a second call is a no-op.
	Close() error
}

RecordCursor yields a session's journal records in sequence order. Next returns the next decoded JournalRecord (an EventRecord, CommandRecord, or FenceRecord) with its sequence, io.EOF once the cold backlog is drained, or a typed error on a malformed/missing/corrupt record. Close releases the underlying reader; it is idempotent and safe to call after an error. It is the all-records counterpart to EventCursor (which yields events only).

type RecordKindError ¶

type RecordKindError struct {
	Subject string
}

RecordKindError reports a JournalRecord whose concrete type is outside the sealed sum the serializer encodes. It is unreachable for an in-package record (the sum is sealed by the unexported marker); it exists so a serializer's default arm fails closed with a typed error rather than panicking.

func (e *RecordKindError) Error() string

type RecordReplayer ¶

type RecordReplayer interface {
	// Open binds a cursor over the WHOLE session log (every record kind) positioned at
	// req.From. Only the cold path (Follow:false) need be implemented; Follow:true
	// returns a typed *FollowUnsupportedError, matching EventReplayer.
	Open(ctx context.Context, req ReplayRequest) (RecordCursor, error)
}

RecordReplayer is the journal's FULL read side: it opens an ordered cursor over a session's log and surfaces EVERY record — events, commands, AND fences — in sequence order. It is the data seam the transcript export consumes: the narrower EventReplayer yields enduring events only and therefore DROPS every CommandRecord (the user's gate decisions). Reading the whole log in sequence instead yields events and commands interleaved in append/causal order — the merged stream the transcript builder needs. The concrete implementation lives in a backend package (same inputs as EventReplayer).

type RecordTooLargeError ¶

type RecordTooLargeError struct {
	Subject string
	MsgID   string
	Length  int
	Cause   error
}

RecordTooLargeError reports a record whose marshaled payload exceeded the inline threshold but could NOT be offloaded to the backend's content-addressed blob store (the store was unavailable or the upload failed). The journal fails closed with this typed error rather than silently persisting an over-threshold record. It carries the record's routing/destination identifier, its idempotency id, the payload length, and the underlying offload cause.

func (e *RecordTooLargeError) Error() string

func (e *RecordTooLargeError) Unwrap() error

type ReplayRequest ¶

type ReplayRequest struct {
	// SessionID is the session whose log is replayed (required; a zero id yields a
	// setup error rather than a replay over every session).
	SessionID uuid.UUID
	// LoopID, when non-zero, narrows an event replay to that single loop; zero
	// replays the session's events plus every loop's events.
	LoopID uuid.UUID
	// From is where the backlog read begins: Beginning or FromSeq(n).
	From StartPos
	// Follow keeps the cursor live after the backlog drains (tailing new appends). A
	// backend that implements only the cold path returns a typed *FollowUnsupportedError
	// from Open rather than silently behaving as a cold cursor.
	Follow bool
}

ReplayRequest selects which of a session's records to replay and how. Which records (events only, or a single loop's) is derived from SessionID + LoopID; how far back from From; whether to keep tailing from Follow. The concrete filtering is the backend replayer's job — this is the backend-neutral request it honors.

type SessionJournal ¶

type SessionJournal interface {
	// Append serializes rec, persists it under the next expected sequence, and
	// returns the assigned sequence. ctx bounds the caller's willingness to wait; the
	// implementation additionally carries a per-append deadline independent of ctx so
	// one stuck call cannot wedge the serialized writer forever. Appends are totally
	// ordered: the returned sequences are strictly monotonic across calls.
	Append(ctx context.Context, rec JournalRecord) (seq uint64, err error)
}

SessionJournal is the single serialized writer for one session's durable log. Append encodes a JournalRecord's payload, persists it under single-writer fencing, and returns the assigned sequence. It is the only thing that writes a session's log; callers funnel every event, command, and fence through it so the log stays a totally-ordered, gap-free record of the session.

The interface is intentionally narrow (one method): a caller that only needs to persist a record must not depend on any log-management surface. The concrete implementation lives in a backend package (e.g. pkg/sessionstore over storage), wired at the composition root — this package owns only the contract.

func Decorate(inner SessionJournal, around AroundAppend) SessionJournal

func WithHooks(j SessionJournal, runner *hook.Runner, sessionID uuid.UUID) SessionJournal

type StartPos ¶

type StartPos struct {
	// contains filtered or unexported fields
}

StartPos is the closed value type naming where a replay begins: the log beginning (every record) or a specific sequence. It is a value, not an interface, so a caller cannot smuggle a third start mode past a switch; the two constructors Beginning and FromSeq are the only ways to build one, and Seq reads it back.

func Beginning() StartPos

func FromSeq(seq uint64) StartPos

func (p StartPos) Seq() uint64