Documentation
¶
Overview ¶
Package sagaprojection bridges kernel/saga/journal.GlobalReader to the projection harness (projection.ReplaySource + projection.Cursor + cellvocab.ProjectionEvent).
PR-03 of EPIC #1609 (#1627). Design decisions at: docs/architecture/202606051200-1609-adr-saga-journal-projection-source.md §D2.
Carrier identity ¶
Every saga journal event is delivered under the "system" principal sentinel (projection.SystemPrincipalActor) via sagaProjectionEvent.RestoreContext. Saga journal events carry no per-event author identity; the projection Apply function must never inherit an ambient admin identity from the Rebuild trigger context (see ADR #1609 §5, F1 flip). InstallSystemPrincipal performs an unconditional overwrite — contrast with outbox's no-overwrite RestoreToContext.
Position scheme ¶
Positions are the GlobalSeq values assigned by the journal backend (MemJournal: in-process monotonic counter; PG: BIGINT IDENTITY column, PR-PG). GlobalSeq is 1-based and stable across process restarts. The "unseeded" sentinel is globalSeq == 0 (the zero value; a conforming GlobalReader never emits seq 0). SagaJournalSource.Position returns a permanent error for the sentinel so that RunCursorConformance's "PermanentErrorOnUnknownEntry" sub-test passes.
Principal allowlist (PROJECTION-SYSTEM-PRINCIPAL-INSTALL-CALLER-01) ¶
This file is the single sanctioned caller of projection.InstallSystemPrincipal. The archtest in tools/archtest/projection_system_principal_install_caller_test.go locks this down with an anti-vacuity guard. Any other file calling InstallSystemPrincipal will fail CI immediately.
Index ¶
Constants ¶
const SagaJournalStream = "saga.journal.v1"
SagaJournalStream is the stream name reported by sagaProjectionEvent.Stream(). The value is stable across restarts so it can be used as a projection filter or checkpoint namespace key.
pre-v1.0 GA: this stream name may evolve directly; ".v1" is a naming convention, not a wire-contract lock. Post-GA evolution requires a version bump.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type SagaJournalSource ¶
type SagaJournalSource struct {
// contains filtered or unexported fields
}
SagaJournalSource implements both projection.ReplaySource and projection.Cursor for the saga journal, backed by a journal.GlobalReader.
Position scheme ¶
The cursor position of each event IS its GlobalSeq. This is intrinsic: a GlobalReader guarantees globally monotonic, 1-based GlobalSeq values, satisfying the Cursor invariants (1-based, monotonic, gap-allowed) directly.
Thread safety ¶
SagaJournalSource holds no mutable state; all methods are safe for concurrent use from multiple goroutines. The underlying GlobalReader is assumed to be safe for concurrent use (MemJournal satisfies this with a sync.RWMutex).
func NewSagaJournalSource ¶
func NewSagaJournalSource(gr journal.GlobalReader) (*SagaJournalSource, error)
NewSagaJournalSource returns a SagaJournalSource backed by the given GlobalReader, or a non-nil error if gr is nil.
The returned source implements both projection.ReplaySource and projection.Cursor; the SAME instance must be wired into both CoordinatorConfig.Replay and CoordinatorConfig.Cursor to ensure the Position encoding (GlobalSeq) is consistent across the two interfaces.
ref: kernel/projection.NewCoordinator (error-return pattern for required deps)
func (*SagaJournalSource) Head ¶
func (s *SagaJournalSource) Head(ctx context.Context) (int64, error)
Head returns the highest GlobalSeq in the journal, or 0 if the journal is empty. This satisfies projection.ReplaySource.Head.
func (*SagaJournalSource) Position ¶
func (s *SagaJournalSource) Position(entry projection.ProjectionEvent) (int64, error)
Position returns the GlobalSeq of entry as its stable cursor position. If entry is not a *sagaProjectionEvent (wrong carrier type) or is the unseeded sentinel (GlobalSeq == 0), Position returns a permanent error per Cursor invariant #4 (cursor.go): an unresolvable entry is unrecoverable.
This satisfies projection.Cursor.Position.
func (*SagaJournalSource) Replay ¶
func (s *SagaJournalSource) Replay(ctx context.Context, fromOffset int64, fn func(projection.ProjectionEvent) error) error
Replay iterates journal events with GlobalSeq strictly greater than fromOffset, calling fn for each in ascending GlobalSeq order. fromOffset == 0 replays all events. If fn returns a non-nil error, Replay stops immediately and returns it.
Internally, Replay paginates via LoadSince with batchSize == 256 to bound per-call allocations. The loop stops when a batch returns fewer than batchSize events (caught up or empty journal).
This satisfies projection.ReplaySource.Replay.
Source Files