Documentation
¶
Overview ¶
Package visibility maintains the eventually-consistent status/flow indexes (spec §9). An Indexer durably consumes the packtrail-events stream and projects each domain event into the packtrail-idx-status and packtrail-idx-flow KV buckets, idempotently and per-revision: an event is applied only if its execution revision is newer than the one already indexed, so duplicate or out-of-order deliveries — and multiple indexer instances on the same durable — are safe.
A periodic Reconcile authoritatively rebuilds the indexes from the source of truth (packtrail-executions), closing any residual drift. The indexes are best-effort: use them for dashboards and operational search, never for correctness decisions (for those, read packtrail-executions directly).
Index ¶
- type Indexer
- func (ix *Indexer) ByFlow(ctx context.Context, flow string) ([]string, error)
- func (ix *Indexer) ByFlowEvents(ctx context.Context, flow string) ([]store.Event, error)
- func (ix *Indexer) ByFlowEventsLimit(ctx context.Context, flow string, limit int) ([]store.Event, error)
- func (ix *Indexer) ByStatus(ctx context.Context, status string) ([]string, error)
- func (ix *Indexer) ByStatusEvents(ctx context.Context, status string) ([]store.Event, error)
- func (ix *Indexer) ByStatusEventsLimit(ctx context.Context, status string, limit int) ([]store.Event, error)
- func (ix *Indexer) GC(ctx context.Context, staleAfter time.Duration) (int, error)
- func (ix *Indexer) Reconcile(ctx context.Context) error
- func (ix *Indexer) ReconcileActive(ctx context.Context) error
- func (ix *Indexer) Run(ctx context.Context) (jetstream.ConsumeContext, error)
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type Indexer ¶
type Indexer struct {
// contains filtered or unexported fields
}
Indexer projects domain events into the visibility indexes and answers lookups by status and by flow.
func New ¶
New returns an Indexer backed by the store's JetStream context, index buckets and namespace.
func (*Indexer) ByFlowEvents ¶ added in v0.0.2
ByFlowEvents returns the index entries for all executions belonging to flow.
func (*Indexer) ByFlowEventsLimit ¶ added in v0.1.0
func (ix *Indexer) ByFlowEventsLimit(ctx context.Context, flow string, limit int) ([]store.Event, error)
ByFlowEventsLimit is ByFlowEvents capped at limit entries (limit <= 0 means no cap). The same arbitrary-subset caveat as ByStatusEventsLimit applies.
func (*Indexer) ByStatusEvents ¶ added in v0.0.2
ByStatusEvents returns the index entries for all executions under status. Each entry is the most recent event stored for that execution, so callers can build summaries (including the error message) without a per-execution round-trip.
func (*Indexer) ByStatusEventsLimit ¶ added in v0.1.0
func (ix *Indexer) ByStatusEventsLimit(ctx context.Context, status string, limit int) ([]store.Event, error)
ByStatusEventsLimit is ByStatusEvents capped at limit entries (limit <= 0 means no cap). Because KV keys have no inherent order, the cap yields an arbitrary subset, not an ordered page; it is a guardrail against transferring an unbounded result set, not a pagination cursor.
func (*Indexer) GC ¶ added in v0.1.0
GC prunes index entries whose execution no longer exists in either the hot bucket or the cold archive — orphans left behind when an archived terminal execution passes its retention and expires. It inspects only terminal (completed/failed) entries older than staleAfter: active executions always exist in the hot bucket, and recently-terminal ones are still within archive retention, so neither needs a source-of-truth read. For each surviving candidate it reads the store; if the execution is gone it deletes the flow and status membership entries. staleAfter should be the archive retention; a non-positive value checks every terminal entry. It returns the count pruned.
func (*Indexer) Reconcile ¶
Reconcile rebuilds the indexes from the source of truth, closing any drift the asynchronous projection may have left behind (spec §9). It streams every hot execution in packtrail-executions, then every retained cold execution in the archive, through last-per-key watches and authoritatively re-asserts each one's membership, so a full pass avoids a key listing plus a Get per execution.
Per §13 this full scan still grows with the hot set plus retained archive set; ReconcileActive is the cheap, frequent counterpart that only touches in-flight executions. Run Reconcile infrequently as the deep backstop.
func (*Indexer) ReconcileActive ¶ added in v0.1.0
ReconcileActive re-asserts the indexes for only the executions the index currently lists as in-flight (running or waiting). It costs O(in-flight) index reads plus a Get per active execution — independent of how many terminal executions have accumulated — so it is cheap enough to run often.
It catches the common drift: an execution that has since completed or failed but whose terminal event was dropped, so the index still shows it as running/waiting. Reading the source of truth and re-asserting moves it to its real status. It cannot resurrect an active execution that is missing from the index entirely (its membership key was never written or was lost); the periodic full Reconcile remains the backstop for that.
func (*Indexer) Run ¶
Run starts the durable consumer on packtrail-events and projects every event until ctx is cancelled. The returned ConsumeContext must be stopped by the caller. DeliverAll lets a fresh or restarted indexer catch up from the start of the stream; projection is idempotent so replays are harmless.
Source Files