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) 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) 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.
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) 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) 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.
Source Files
Directories