hub

package
v0.33.0 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Sep 10, 2026 License: Apache-2.0 Imports: 11 Imported by: 0

README

pkg/hub

pkg/hub is the session-level event fan-in: a publish/subscribe hub with a federated-quiescence model. Loops publish events through the narrow eventPublisher contract; consumers (TUI, CLI, durable journal, HTTP SSE) subscribe with an event.EventFilter. The hub aggregates loop activity into one sessionState so a headless run can WaitIdle without any session goroutine.

What is hub?

A *hub.Hub is owned by a Session. It exposes:

  • To loops — only PublishEvent / PublishEventChecked (the eventPublisher interface). Loops never see subscribers, state, or waiters.
  • To consumers — SubscribeEvents, which returns an *EventSubscription carrying one bounded egress channel of event.Delivery values.
  • To the session — ExpectTurn / CancelExpectTurn / StopSession / WaitIdle for the quiescence model.

The hub is constructed with hub.New(sessionID, opts...). Default options install a nop journal appender, a real-clock/real-uuid event.Factory, and a nop fault reporter — i.e. a hub that publishes and fans out but persists nothing. The composition root injects the real trio via WithAppender / WithFactory / WithFaultReporter.

How to use

Consumers reach the hub through Session.SubscribeEvents:

sub, err := session.SubscribeEvents(nil)  // nil = all events
if err != nil { return err }
defer sub.Close()
for delivery := range sub.Events() {
    fmt.Println(delivery.Event.Scope(), delivery.Event.Class())
    if delivery.Event.EndsTurn() { /* this turn is over */ }
    if delivery.JournalSeq > 0    { /* an enduring event was persisted */ }
}
_ = sub.Err()  // nil on Close; *SubscriptionLossError on hub-forced loss

A subscriber filters by passing an event.EventFilter:

onlyTurns := event.EventFilter{
    MatchTurns: true,
}
sub, _ := session.SubscribeEvents(onlyTurns)

A headless runner blocks on quiescence without a session goroutine:

// internally, the session:
hub.ExpectTurn(turnID)
hub.WaitIdle(ctx)  // returns when the active set drains and the durable edge completes

Sibling packages

  • pkg/event — event.Event, event.EventFilter, event.Delivery, the lifecycle and scope mixins the hub routes by.
  • pkg/identity — the event.Header producer identity the hub stamps on synthesized session events.

How it is designed

The hub is a single fan-in point with three concurrency domains, each under its own lock:

   Loops ──PublishEvent──►  ┌────────────────────────────────────────┐
                            │ Hub                                     │
                            │                                         │
                            │  publishMu   ── admission seal (abort)   │
                            │  activityMu  ── active-set ↔ durable edge │
                            │  mu          ── subs, state, waiters     │
                            │                                         │
                            │  appender (durable write, OUTSIDE mu)    │
                            │  factory  (event.Header stamper)         │
                            │  reporter (fault escalation seam)        │
                            │  idleBoundary (native durable completion) │
                            └──────┬─────────────────────────────────┬──┘
                                   │                                 │
                                   ▼                                 ▼
                          ┌────────────────┐               ┌────────────────┐
                          │ Subscribers     │               │ WaitIdle        │
                          │ (TUI / SSE /    │               │ waiters         │
                          │  journal / test)│               │ (headless run)  │
                          └────────────────┘               └────────────────┘
  • publishMu is the construction-abort admission seal. AbortSession closes admission atomically and returns publishDrained so the session retains journal ownership until every already-admitted publisher has left the appender path.
  • activityMu orders every active-set mutation with its derived durable edge. It may be held across I/O, unlike mu; no state is read or written under this lock alone.
  • mu guards subs, state, and waiters together. One lock keeps the subscriber-set snapshot consistent with the active/phase transition. The critical section only copies subscribers; mu is always released before durable I/O, workspace boundaries, reporting, or delivery.
Bounded egress and the overflow policy

Each subscription owns one bounded egress channel (default 256). A slow subscriber never blocks a publisher or another subscriber, so delivery is a non-blocking send into this buffer; on overflow the class-aware policy applies:

  • Ephemeral events are dropped (a TokenDelta is expendable).
  • Enduring events fail the subscription with a typed *SubscriptionLossError. The subscriber learns it lost the stream — so it can re-subscribe and re-sync — rather than silently missing an authoritative event.

SessionStopped is an event, not a stream terminator: a subscription ends only on Close, loss, or hub teardown.

Federated quiescence

WaitIdle blocks until the active set is empty and the durable edge that corresponds to that transition has completed. Idle-boundary generations close the fast-path window between an in-memory Active→Idle transition and its native durable completion: a generation prevents an older overlapping boundary from clearing a newer pending edge. A sticky waiterFailure survives until the owning recoverable operation clears its exact generation token, so stale recovery cannot erase a later fault.

Documentation

Overview

Package hub implements the session-level event fan-in: a publish/subscribe hub with a federated-quiescence model. Loops publish events through the narrow eventPublisher contract; consumers (TUI/CLI now, a durable journal later) subscribe with an EventFilter. The hub aggregates loop activity into a single sessionState so a headless run can WaitIdle without any session goroutine.

Concurrency contract: mu guards the subscriber set, sessionState (active/phase), and WaitIdle waiter registry. activityMu serializes each active-set mutation with its derived durable edge but never guards state itself. The critical section under mu only applies state and copies subscribers; mu is always released before durable I/O, workspace boundaries, reporting, or delivery.

Index

Constants

This section is empty.

Variables

View Source 
var ErrCommitEventMismatch = errors.New("committed append result does not belong to the delivered event")

ErrCommitEventMismatch is the cause recorded when a delivery's committed append result belongs to a DIFFERENT event than the one being delivered. It is a hub programming error, not a runtime condition: sessionwire.Project stamps a public EventID from the event's own header, so a committed result and its event always agree unless a delivery path paired the wrong two values.

It is guarded rather than trusted because the failure is silent where it lands. A consumer joining a durable tail to this stream dedupes on (sequence, EventID); two deliveries carrying one identity make it either drop an event as a duplicate or render another twice, with nothing anywhere reporting an error. The compiler cannot help — every append result is the same type — so the pairing is checked at the one place every delivery passes through.

View Source 
var ErrCommittedBodyMissing = errors.New("enduring public event carried no committed public body")

ErrCommittedBodyMissing is the cause recorded on a committed-public-event subscription the hub failed because an ENDURING public event reached its fan-out without the canonical bytes its durable append was supposed to report. It is a broken invariant of the committed stream, not congestion, and naming it separately is what keeps the loss legible: without a cause, this failure is indistinguishable from an egress overflow, and a Host consumer that treated it as backpressure would resubscribe forever against a hub that can never satisfy the contract.

Its text carries no "hub:" prefix because it is always surfaced through *SubscriptionLossError, which supplies one; prefixing here would double it.

View Source 
var ErrResidencyReleased = errors.New("hub: session residency released")

ErrResidencyReleased is the AbortSession cause a session passes when it gives up its residency NONTERMINALLY. It reaches subscribers as the subscription loss cause, so a consumer can tell "this process released the session" apart from "construction failed" and from the terminal ErrSessionStopped.

View Source 
var ErrSessionStopped = errors.New("hub: session stopped")

ErrSessionStopped is returned by WaitIdle when the session has stopped (either already at entry or stopped while a caller was waiting). It is a leaf sentinel with no additional context fields.

Functions

This section is empty.

Types

type CommittedPublicEventsUnavailableError added in v0.31.0 

type CommittedPublicEventsUnavailableError struct{}

CommittedPublicEventsUnavailableError reports that this hub cannot serve the segregated committed-public-event capability, because its injected appender does not report the exact canonical public bytes a durable append stored. It is a capability REFUSAL, not a runtime failure: a no-persistence hub and a hub over a legacy journal are both permanently incapable, and saying so at subscribe time is what stops a consumer from persisting coverage it was never actually given.

func (*CommittedPublicEventsUnavailableError) Error added in v0.31.0 

func (*CommittedPublicEventsUnavailableError) Error() string

type EventSubscription 

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

EventSubscription is a consumer's handle to the session fan-in. It owns exactly one bounded egress channel. Events closes when the subscriber Closes it or when the hub fails it for loss. Err returns nil for an intentional Close and the typed SubscriptionLossError for a hub-forced termination. SessionStopped is an event, not a stream terminator: a subscription ends only on Close, loss, or hub teardown.

func (*EventSubscription) Close 

func (s *EventSubscription) Close() error

Close is the subscriber's intentional teardown. It is idempotent and records no error (Err stays nil). The first terminal — Close or a hub fail — wins, so a Close after a loss does not clobber the recorded loss error.

func (*EventSubscription) Err 

func (s *EventSubscription) Err() error

Err returns nil for an intentional Close and the typed SubscriptionLossError for a hub-forced loss. It returns nil while the subscription is still live.

func (*EventSubscription) Events 

func (s *EventSubscription) Events() <-chan event.Delivery

Events is the receive end of the subscription's egress channel. It is closed on Close, loss, or hub teardown.

type FaultReporter 

type FaultReporter interface {
	ReportFault(ctx context.Context, fault *SessionPersistenceFault)
}

FaultReporter is the hub's escalation seam for a required-durable-append failure. The hub depends only on this narrow interface (Dependency Inversion): it never sees the Session's closing latch or its WaitIdle registry, and the Session (which implements it) never sees the hub's append path. The implementation must be fail-secure — on a fault the owning session stops accepting new Submit/NewLoop and wakes any WaitIdle waiter with the fault — and must not block the hub's publish path (the hub calls it inline, outside the hub lock).

ctx is the publish context: the reporter may use it to bound any work it does, but must not depend on it staying live (a fault may arrive on a cancelled publish).

type Hub 

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

Hub is the session's event fan-in. It is owned by Session; loops see only its PublishEvent method (the narrow eventPublisher), and consumers see only SubscribeEvents. ExpectTurn/CancelExpectTurn/StopSession are session-owned (the session calls them; loops depend on the eventPublisher interface, which excludes them).

func New 

func New(sessionID uuid.UUID, opts ...Option) *Hub

New builds an idle hub for sessionID. The returned hub has no subscribers and a zero-value (idle, empty) sessionState. Without options it installs the durable-tap defaults: a nop appender (persists nothing, never fails — headless/no-persistence mode), a real-clock/real-uuid event Factory, and a nop fault reporter. The composition root (Phase 10) injects the real trio via WithAppender/WithFactory/ WithFaultReporter.

func (*Hub) AbortSession 

func (h *Hub) AbortSession(cause error) <-chan struct{}

AbortSession closes the hub LOCALLY — clearing activity, forcing the in-memory phase to SessionStopped, waking WaitIdle waiters and failing every subscription with cause — WITHOUT appending or delivering the durable SessionStopped lifecycle event.

Two callers need exactly that, and the shared "no append" is why they share this method rather than each growing their own copy:

  • an unpublished/failed construction, which must not journal a stop for a session that never started;
  • a NONTERMINAL residency release (cause ErrResidencyReleased), which must not journal a stop for a session that is still restorable elsewhere.

The cause is what tells them apart at a subscriber.

func (*Hub) AcquireHustleActivity 

func (h *Hub) AcquireHustleActivity(ctx context.Context, runID hustle.RunID) (*HustleActivityLease, error)

AcquireHustleActivity inserts runID as blocking session work. The returned lease removes exactly that entry. If the Idle->Active edge cannot be durably committed, a non-nil partial lease is returned with the fault so the caller can silently roll back the in-memory insertion.

func (*Hub) CancelExpectTurn 

func (h *Hub) CancelExpectTurn(ctx context.Context, subagentLoopID uuid.UUID)

CancelExpectTurn releases a {wake, subagentLoopID} token when its hand-back is rejected or explicitly discarded. It derives SessionIdle if this emptied active. Exported for the session only (see ExpectTurn).

func (*Hub) ClearWaiterFailure 

func (h *Hub) ClearWaiterFailure(token uint64)

ClearWaiterFailure clears the sticky waiter failure only when token still owns the current generation. A stale recovery token is a no-op, preserving any newer persistence or root-lease fault.

func (*Hub) CommittedPublicEventsSupported added in v0.31.0 

func (h *Hub) CommittedPublicEventsSupported() bool

CommittedPublicEventsSupported reports whether SubscribeCommittedPublicEvents will succeed on this hub.

func (*Hub) ExpectTurn 

func (h *Hub) ExpectTurn(ctx context.Context, subagentLoopID uuid.UUID)

ExpectTurn takes a {wake, subagentLoopID} token at subagent spawn so a finished subagent's in-flight hand-back cannot empty active and fire a false SessionIdle. It derives SessionActive if the session was idle. It is exported for the session (its sole caller); loops depend only on the narrow eventPublisher interface, which excludes it, so a loop can never reach it.

There is no triggering EVENT here (the wake token is hub-internal), so only the derived session event is durable: it is minted + appended OUTSIDE the lock before delivery, fail-secure (a failed append delivers nothing and raises a fault).

func (*Hub) FailWaiters 

func (h *Hub) FailWaiters(err error) uint64

FailWaiters latches err as the current sticky waiter failure, returns its monotonically increasing generation token, wakes every WaitIdle waiter, and clears the registry. It is the session's escalation lever on a SessionPersistenceFault: a faulted session is neither idle nor cleanly stopped, so its blocked WaitIdle callers must be released with the fault rather than left hanging or falsely told "idle". Exported for the session only (its FaultReporter implementation); loops never see it. It takes the lock itself (called outside it). A waiter that arrives after this point observes the same sticky error before consulting idle state or registering.

func (*Hub) IsIdle 

func (h *Hub) IsIdle() bool

IsIdle is the non-blocking quiescence probe used by the session's manual checkpoint control plane. It is intentionally narrower than exposing hub state.

func (*Hub) PublishEvent 

func (h *Hub) PublishEvent(ctx context.Context, ev event.Event) error

PublishEvent is the durable tap (design "Hub tap algorithm"): for an Enduring event it appends BEFORE applying it to hub state (durable-first, fail-secure), then applies the quiescence transition (which may DERIVE a session event D), then mints+appends D, then delivers the triggering event followed by D — in causal order. The precise ordering, honoring the lock rule (no I/O under the hub lock):

  1. Ephemeral event: never persisted — fan out only (the unchanged path).
  2. Enduring event: appendCommitted(ev) OUTSIDE the lock, which asks the injected appender both whether the append durably persisted a NEW frame and — when the appender can report them — for the exact canonical public bytes it stored. On error → ReportFault, deliver NOTHING, return (do not apply a transition for an event that did not persist). On a deduplicated retry (Appended=false) → apply NOTHING and deliver NOTHING; the event was already applied and delivered by its original, genuinely new append, so this call reports the same success without repeating either.
  3. Under the lock: apply ev's active/phase mutation, which may derive D (SessionActive/SessionIdle); snapshot the subscriber set. Unlock.
  4. If D was derived: mint it (Factory: EventID+CreatedAt) and append it OUTSIDE the lock. On error → ReportFault, deliver NEITHER ev nor D, return.
  5. Deliver live, outside the lock (existing fan-out policy): ev, then D. If D was a SessionIdle (the Active->Idle edge), wake WaitIdle waiters AFTER its durable append — never before (a failed append must not falsely report idle).

Subscriber delivery never blocks a publisher: each send into the bounded egress channel is non-blocking. Activity-affecting publishers do serialize with other activity transitions until their derived durable edge completes. It returns nil even with no subscribers (the headless case) — the sessionState transition still runs.

After SessionStopped, the event is still appended+delivered (filtered) but no longer mutates active/phase and never derives SessionIdle/SessionActive.

func (*Hub) PublishEventChecked 

func (h *Hub) PublishEventChecked(ctx context.Context, ev event.Event) error

PublishEventChecked is the construction-transaction variant of PublishEvent. It keeps the same durable-first reporting semantics but also returns the persistence fault to a caller that must not make an object reachable unless its creation event committed.

func (*Hub) PublishEventCommitted added in v0.31.0 

func (h *Hub) PublishEventCommitted(ctx context.Context, ev event.Event) (event.AppendCommit, error)

PublishEventCommitted is PublishEventChecked that ALSO reports the primary event's own durable append result. It exists for one caller shape: a publisher that must durably record WHICH record its event committed as — the session's nonterminal residency release, whose SessionResidencyReleased carries the sequence of the workspace checkpoint it is anchored to.

The commit describes the PRIMARY event only. A derived SessionActive/SessionIdle carries its own separate append and is never reported here. On a deduplicated retry the commit reports the sequence of the ORIGINAL append with Appended false, so a caller records the same anchor it would have recorded the first time.

func (*Hub) PublishInternalEventChecked 

func (h *Hub) PublishInternalEventChecked(ctx context.Context, ev event.Event) error

PublishInternalEventChecked durably appends one recognized private hustle-lifecycle or permission-review event. It deliberately bypasses quiescence mutation, workspace idle boundaries, and subscriber delivery: the separate hustle activity lease owns the blocking state, while this method owns only the private audit record.

func (*Hub) ReserveTurnStart 

func (h *Hub) ReserveTurnStart(loopID uuid.UUID) (*TurnStartReservation, error)

ReserveTurnStart establishes the global activity-before-checkpoint lock order for one loop's opening TurnStarted. The returned capability must be released on every path that does not publish that exact event.

func (*Hub) StopSession 

func (h *Hub) StopSession(ctx context.Context)

StopSession is the session-owned teardown transition. It is idempotent: if already SessionStopped it returns without effect. Otherwise — honoring the durable tap's append-before-apply, fail-secure rule — it:

  1. Mints the synthesized SessionStopped (Factory: EventID+CreatedAt) and durably appends it OUTSIDE the lock, BEFORE any state change. On error → ReportFault, do NOT flip phase, wake nobody, deliver nothing, return (the session faults instead of stopping; its FaultReporter wakes WaitIdle waiters with the fault).
  2. Under the lock: clear active, force phase=SessionStopped (bypassing applyActivity so no SessionIdle is derived), wake every WaitIdle waiter with ErrSessionStopped, snapshot the subscriber set. Unlock.
  3. Deliver the (already durable) SessionStopped live.

Exported for the session only (see ExpectTurn).

func (*Hub) SubscribeCommittedPublicEvents added in v0.31.0 

func (h *Hub) SubscribeCommittedPublicEvents(filter event.EventFilter) (*EventSubscription, error)

SubscribeCommittedPublicEvents registers a subscription on the SEGREGATED committed-public-event stream: every delivery it yields carries the exact canonical public body the durable append stored, its committed public EventID, and a CoveredThrough watermark equal to that append's own sequence.

It is a separate call from SubscribeEvents rather than a flag on it because the two contracts differ. SubscribeEvents is the compatibility stream every TUI/CLI consumer already uses: it delivers ephemeral events too, works with no persistence at all, and promises nothing about bytes. This one promises committed bytes on every delivery and therefore cannot be offered by a hub that has none — it refuses with *CommittedPublicEventsUnavailableError instead of degrading silently.

func (*Hub) SubscribeEvents 

func (h *Hub) SubscribeEvents(filter event.EventFilter) (*EventSubscription, error)

SubscribeEvents registers a new subscription with the given filter and returns its handle. The subscriber reads ev from sub.Events(); it must Close the subscription when done.

func (*Hub) WaitIdle 

func (h *Hub) WaitIdle(ctx context.Context) error

WaitIdle blocks until the session is quiescent (active empty), ctx is done, or the session stops. It returns nil on idle, ctx.Err() on cancellation, and ErrSessionStopped if the session is or becomes stopped. With no session goroutine, waiters are woken by applyActivity (Active->Idle) and StopSession.

Sticky waiter failures are checked under the same lock before the idle fast path or waiter registration, closing the fault-before-registration race. Stopped takes precedence so every post-stop call returns ErrSessionStopped.

type HustleActivityError 

type HustleActivityError struct {
	Reason HustleActivityReason
	RunID  hustle.RunID
}

HustleActivityError reports that a blocking hustle activity could not be inserted into the hub's quiescence set.

func (*HustleActivityError) Error 

func (e *HustleActivityError) Error() string

type HustleActivityLease 

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

HustleActivityLease owns one blocking hustle entry in the hub's quiescence set. Its concrete return type is intentional: sessionruntime adapts it to the narrow controller-owned lease interface without weakening Go's invariant return types.

func (*HustleActivityLease) Release 

func (l *HustleActivityLease) Release(ctx context.Context) error

Release removes this lease's exact run activity at most once. A partial lease performs an in-memory rollback only and returns the cached acquisition fault; a committed lease persists any resulting Active->Idle edge through the native idle boundary and caches that result for later calls.

type HustleActivityReason 

type HustleActivityReason string

HustleActivityReason identifies an activity-acquisition contract violation. 

const (
	HustleActivityInvalidRunID HustleActivityReason = "invalid_run_id"
	HustleActivityDuplicate    HustleActivityReason = "duplicate_run_id"
	HustleActivityStopped      HustleActivityReason = "session_stopped"
)

type Option 

type Option func(*Hub)

Option configures an optional hub dependency at construction. The bare New(sessionID) installs the nop appender, a real-clock/real-uuid Factory, and the nop fault reporter; an Option overrides one of them. This keeps existing callers (hub.New(id)) working while the composition root injects the durable trio.

func WithAppender 

func WithAppender(a eventAppender) Option

WithAppender injects the durable event appender (the composition root's adapter over SessionJournal). A nil appender is ignored (the nop default stays installed) so a caller can never accidentally null out the field and skip the nil-safe publish path.

func WithCommitObserver 

func WithCommitObserver(observer func(event.Event)) Option

WithCommitObserver installs a callback invoked after the primary event and any derived session event have durably committed, but before either event is fanned out to subscribers. It is an observation hook: the callback must not mutate durable state or return an error. The session runtime uses it to keep its bounded live status view ordered before consumers can observe the same committed terminal.

func WithFactory 

func WithFactory(f *event.Factory) Option

WithFactory injects the event Factory the hub mints EventID+CreatedAt from for the session events it SYNTHESIZES (SessionActive/SessionIdle/SessionStopped). A nil factory is ignored (the default real-clock Factory stays). The session-scoped events the hub derives currently carry no header identity; this factory stamps them so the journal sees a stable idempotency key and creation time.

func WithFaultReporter 

func WithFaultReporter(r FaultReporter) Option

WithFaultReporter injects the fail-secure escalation seam invoked when a required durable append fails. A nil reporter is ignored (the nop default stays). The Session implements it to reject new Submit/NewLoop and wake WaitIdle waiters.

type PublishBoundaryError 

type PublishBoundaryError struct {
	Reason    PublishBoundaryReason
	EventType string
	Cause     error
}

PublishBoundaryError reports a fail-closed event publication denial.

func (*PublishBoundaryError) Error 

func (e *PublishBoundaryError) Error() string

func (*PublishBoundaryError) Unwrap 

func (e *PublishBoundaryError) Unwrap() error

type PublishBoundaryReason 

type PublishBoundaryReason string

PublishBoundaryReason identifies why an event was rejected at a hub publication boundary. The closed reason set lets callers inspect the denial without parsing an error string or retaining the event payload in the error. 

const (
	PublishBoundaryNilEvent   PublishBoundaryReason = "nil_event"
	PublishBoundaryVisibility PublishBoundaryReason = "visibility"
	PublishBoundaryClass      PublishBoundaryReason = "class"
	PublishBoundarySession    PublishBoundaryReason = "session"
	PublishBoundaryType       PublishBoundaryReason = "type"
	PublishBoundaryInvalid    PublishBoundaryReason = "invalid"
)

type SessionAbortedError 

type SessionAbortedError struct{ Cause error }

SessionAbortedError reports a publish rejected after construction teardown sealed the unpublished session's durable event tap.

func (*SessionAbortedError) Error 

func (e *SessionAbortedError) Error() string

func (*SessionAbortedError) Unwrap 

func (e *SessionAbortedError) Unwrap() error

type SessionPersistenceFault 

type SessionPersistenceFault struct {
	// Event is the Enduring event whose required durable append failed — either the
	// triggering event the loop published or a hub-synthesized session event
	// (SessionActive/SessionIdle/SessionStopped). Never nil on a real fault.
	Event event.Event
	// Cause is the underlying journal failure (typed). It may be nil only in the
	// degenerate construction a test exercises; a real fault always chains the
	// append error.
	Cause error
}

SessionPersistenceFault is the typed error the hub raises when a REQUIRED durable append of an Enduring event fails (the durable tap's append-before-apply step). It carries the offending Event (so an operator log names what could not be persisted) and the underlying Cause from the journal (a *journal.AppendError, *FenceViolationError, …) so a reporter can errors.As the root failure. It is the fail-secure signal: the live event whose append failed is NOT delivered, and the session that owns the hub stops accepting new work.

func (*SessionPersistenceFault) Error 

func (e *SessionPersistenceFault) Error() string

func (*SessionPersistenceFault) FatalPublication 

func (*SessionPersistenceFault) FatalPublication() bool

FatalPublication marks a required-journal failure for dependency-inverted producers. Callers can fail closed without importing hub's concrete type.

func (*SessionPersistenceFault) Unwrap 

func (e *SessionPersistenceFault) Unwrap() error

type SessionPhase 

type SessionPhase uint8

SessionPhase is the coarse quiescence phase of the whole session. SessionIdle is the zero value, so a freshly built session is idle until its first turn. 

const (
	SessionIdle    SessionPhase = iota // quiescent — user may type again; zero value
	SessionActive                      // >=1 loop busy, or a hand-back in flight
	SessionStopped                     // after Shutdown (distinct from Idle)
)

type SubscriptionLossError 

type SubscriptionLossError struct {
	DroppedClass event.Class
	Cause        error
}

SubscriptionLossError is the typed terminal recorded on a subscription the hub fails rather than let it silently miss an authoritative event. DroppedClass is the class of the event that triggered the loss; Cause names the reason when it is not the default one.

The reasons call for OPPOSITE responses, which is why Cause exists rather than one undifferentiated loss. A nil Cause is egress overflow: the subscriber fell behind, and re-subscribing to re-sync is the right answer. ErrCommittedBodyMissing and ErrCommitEventMismatch are both broken invariants — an enduring public event that arrived without its committed bytes, and a delivery whose committed append belongs to a different event — where re-subscribing loops forever against a hub that cannot satisfy the contract.

func (*SubscriptionLossError) Error 

func (e *SubscriptionLossError) Error() string

func (*SubscriptionLossError) Unwrap 

func (e *SubscriptionLossError) Unwrap() error

type TurnStartReservation 

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

TurnStartReservation is an opaque one-shot publisher that owns the Hub activity transition from immediately before a loop acquires its first checkpoint reader through publication of that loop's exact opening TurnStarted.

func (*TurnStartReservation) PublishTurnStarted 

func (r *TurnStartReservation) PublishTurnStarted(ctx context.Context, started event.TurnStarted) error

PublishTurnStarted consumes this capability for exactly one matching value event. Generic Hub publication cannot discover or claim it from event coordinates. This legacy form preserves unchecked Hub reporting semantics; construction paths that must not install live state without the event use PublishTurnStartedChecked.

func (*TurnStartReservation) PublishTurnStartedChecked 

func (r *TurnStartReservation) PublishTurnStartedChecked(ctx context.Context, started event.TurnStarted) (committed bool, err error)

PublishTurnStartedChecked consumes the reservation with checked durable publication. Committed distinguishes failure of the primary TurnStarted append from a later failure while publishing its derived session activity transition.

func (*TurnStartReservation) Release 

func (r *TurnStartReservation) Release()

Release cancels an unused capability. A publication already in progress owns the activity release; a published or previously released capability is a no-op.

type TurnStartReservationError 

type TurnStartReservationError struct {
	Reason TurnStartReservationReason
	LoopID uuid.UUID
}

TurnStartReservationError reports a denied or mismatched one-shot turn-start activity reservation.

func (*TurnStartReservationError) Error 

func (e *TurnStartReservationError) Error() string

type TurnStartReservationReason 

type TurnStartReservationReason string

TurnStartReservationReason identifies why a loop could not reserve the Hub's activity transition for its opening TurnStarted publication. 

const (
	TurnStartReservationInvalidLoop TurnStartReservationReason = "invalid_loop_id"
	TurnStartReservationStopped     TurnStartReservationReason = "session_stopped"
	TurnStartReservationMismatch    TurnStartReservationReason = "publication_mismatch"
	TurnStartReservationReleased    TurnStartReservationReason = "released"
	TurnStartReservationReused      TurnStartReservationReason = "reused"
)

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.