Documentation
¶
Overview ¶
Package saga implements kernel/saga/journal.Journal over PostgreSQL.
Two tables under one transaction per mutation (see migration 040):
- saga_instances (mutable projection): status + current_version + lease
- saga_events (append-only log): authoritative event history, PK (instance_id, version) enforces version monotonicity.
Fencing model mirrors adapters/postgres outbox (ADR 202605051600-adr-pg-outbox-fencing.md): ClaimPending mints a fresh UUID lease per batch; subsequent Append / Heartbeat / MarkTerminal CAS-fence on (id, lease_id, lease_expires_at >= injected_now) so a zombie leader's write must miss the row and surface as KindConflict (Append) or (false, nil) (Heartbeat / MarkTerminal). The `>=` boundary (vs strict `>`) matches memjournal's `!Before(now)` semantic — at exact equality the lease is STILL VALID — and is verified by conformance subtests {Append,ClaimPending,Heartbeat}_ExactlyAtLeaseExpiry_*. Behavior matches the in-memory reference implementation (kernel/saga/journal.MemJournal) and is verified by sagajournaltest.RunConformanceSuite over a real PostgreSQL backend (testcontainers).
Status / EventKind columns are SMALLINT mappings of saga.Status and journal.EventKind enums (iota+1; the zero value is the invalid sentinel and is never persisted). The mapping is centralized in store.go statusFromInt16 / kindFromInt16 — no magic numbers appear in SQL bodies.
ref: adapters/postgres/outbox_store.go (lease_id CAS pattern) ref: tools/archtest/user_repo_conformance_enrollment_test.go (SAGA-JOURNAL-CONFORMANCE-ENROLLMENT-01)
Index ¶
- type PGJournal
- func (s *PGJournal) Append(ctx context.Context, instanceID, leaseID idutil.SafeID, event journal.Event) (int64, error)
- func (s *PGJournal) ClaimPending(ctx context.Context, batchSize int, leaseDuration time.Duration) ([]journal.ClaimedInstance, idutil.SafeID, error)
- func (s *PGJournal) Enqueue(ctx context.Context, instance saga.Instance) error
- func (s *PGJournal) HeadSeq(ctx context.Context) (int64, error)
- func (s *PGJournal) Heartbeat(ctx context.Context, instanceID, leaseID idutil.SafeID, ...) (bool, error)
- func (s *PGJournal) Load(ctx context.Context, instanceID idutil.SafeID) ([]journal.Event, error)
- func (s *PGJournal) LoadSince(ctx context.Context, afterGlobalSeq int64, limit int) ([]journal.GlobalEvent, error)
- func (s *PGJournal) MarkTerminal(ctx context.Context, instanceID, leaseID idutil.SafeID, ...) (bool, error)
- func (s *PGJournal) RepoReady(ctx context.Context) error
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type PGJournal ¶
type PGJournal struct {
// contains filtered or unexported fields
}
PGJournal implements kernel/saga/journal.Journal over PostgreSQL.
All SQL ops route through pgExecutor (pg_executor.go), which is ambient-tx aware via kernel/persistence.TxFromContext. Multi-statement mutations (Append / MarkTerminal / ClaimPending) acquire a transaction via pgExecutor.acquireTx — joining the ambient tx when one is present (Coordinator.commitStep wraps Append + outbox.Emit + RegisterAfterCommit in one txRunner.RunInTx; saga MUST join so an Emit failure rolls back the journal write together with the outbox row, preserving the L2 OutboxFact invariant). When no ambient tx is present, acquireTx opens a fresh one and the caller takes ownership of Commit / Rollback.
func NewJournal ¶
NewJournal constructs a PGJournal backed by pool. clk is required; a nil or typed-nil Clock panics via clock.MustHaveClock (programmer error, matching the kernel/ wiring convention shared with MemJournal / outbox).
func (*PGJournal) Append ¶
func (s *PGJournal) Append(ctx context.Context, instanceID, leaseID idutil.SafeID, event journal.Event) (int64, error)
Append acquires a transaction (joining the ambient tx if present), reads the projection (FOR UPDATE), checks the lease fence, validates the event, advances the status via saga.AdvanceSaga, bumps current_version, and inserts the event row. Returns the assigned version on success.
Error precedence matches memjournal: instance-existence and lease fence are checked BEFORE Event.ValidateForAppend so a bad payload on an unknown instance returns ErrSagaNotFound (not ErrValidationFailed). Conformance suite locks this with Append_UnknownInstanceWithBadPayload_PrefersNotFound.
Commit-order correctness (F1, #1630): immediately before the insertEvent INSERT this method acquires sagaEventsGlobalAppendLockKey via pg_advisory_xact_lock. The xact-scoped lock serializes all saga_events INSERTs so that the IDENTITY-assigned global_seq order equals commit order (a tailer advancing its cursor by delivered global_seq will never permanently skip a lower seq that committed later). Lock-ordering: per- instance FOR UPDATE is acquired first, then the single global advisory key in that consistent order across every transaction; no deadlock is possible because each transaction holds at most one instance row lock before requesting the shared advisory key. The throughput tradeoff (serialized appends) is replaced by a safe-lag committed-prefix reader in #2070.
func (*PGJournal) ClaimPending ¶
func (s *PGJournal) ClaimPending( ctx context.Context, batchSize int, leaseDuration time.Duration, ) ([]journal.ClaimedInstance, idutil.SafeID, error)
ClaimPending leases up to batchSize claimable instances under a fresh batch UUID, returning them stamped with the lease the caller MUST echo on every lease-fenced mutation for each claimed instance.
func (*PGJournal) Enqueue ¶
Enqueue inserts a Pending instance. Duplicate IDs surface as errcode.ErrSagaDuplicateInstance (KindConflict).
func (*PGJournal) HeadSeq ¶
HeadSeq implements journal.GlobalReader.HeadSeq.
Returns 0 when saga_events is empty (COALESCE(MAX(global_seq), 0)) matching the MemJournal contract: GlobalSeq 0 is the "empty journal" sentinel and a conforming backend never returns an event with GlobalSeq 0.
func (*PGJournal) Heartbeat ¶
func (s *PGJournal) Heartbeat(ctx context.Context, instanceID, leaseID idutil.SafeID, leaseDuration time.Duration) (bool, error)
Heartbeat extends an active lease. Returns (false, nil) on stale lease / missing instance — distinguishing the two would leak existence to a zombie leader, so the silent semantic is deliberate.
func (*PGJournal) Load ¶
Load returns the full ordered event history. A never-enqueued instance returns errcode.ErrSagaNotFound (KindNotFound); an enqueued instance with no events yet returns an empty non-nil slice.
func (*PGJournal) LoadSince ¶
func (s *PGJournal) LoadSince(ctx context.Context, afterGlobalSeq int64, limit int) ([]journal.GlobalEvent, error)
LoadSince implements journal.GlobalReader.LoadSince.
Argument validation mirrors MemJournal exactly so that the shared sagajournaltest.RunGlobalReaderConformance suite passes against both backends:
- limit <= 0 → journal.NewNonPositiveLimitError
- afterGlobalSeq < 0 → journal.NewNegativeGlobalSeqError
Returns a non-nil empty slice (not nil) when no rows are found, matching the interface contract ("caught-up" ≡ empty slice + nil error).
func (*PGJournal) MarkTerminal ¶
func (s *PGJournal) MarkTerminal(ctx context.Context, instanceID, leaseID idutil.SafeID, finalStatus saga.Status) (bool, error)
MarkTerminal transitions the instance to finalStatus and appends the matching terminal event atomically. Joins the ambient tx when one is present (Coordinator wraps commitStep in RunInTx). (false, nil) on stale lease / missing instance; KindInvalid on a non-terminal finalStatus or illegal transition.
Commit-order correctness (F1, #1630): same advisory-lock protocol as Append — acquires sagaEventsGlobalAppendLockKey immediately before the insertEvent INSERT so global_seq order equals commit order. See Append godoc and sagaEventsGlobalAppendLockKey for the full rationale.
Source Files
Directories