Documentation
¶
Overview ¶
Package stream owns the C-S2 visibility watermark and C-S7 retention machinery for the transactional change-event outbox.
Index ¶
Constants ¶
This section is empty.
Variables ¶
var ErrLeaseHeld = errors.New("stream watermark lease is held")
ErrLeaseHeld means another watermarker owns the unexpired singleton lease.
Functions ¶
This section is empty.
Types ¶
type Retention ¶
type Retention struct {
// contains filtered or unexported fields
}
Retention prunes change events without consulting consumer cursors and advances per-stream horizons in the same transaction as each delete batch.
func NewRetention ¶
func NewRetention( pool *pgxpool.Pool, options RetentionOptions, ) (*Retention, error)
NewRetention validates the locked seven-day C-S7 floor.
type RetentionOptions ¶
type RetentionOptions struct {
Age time.Duration
Period time.Duration
BatchSize int
Now func() time.Time
OnPrune func(context.Context, string, int64)
Tracer trace.Tracer
}
RetentionOptions configures independent C-S7 change-event pruning.
type WatermarkObserver ¶
type WatermarkObserver interface {
WatermarkStep(context.Context, WatermarkProgress)
}
WatermarkObserver is M6's C-S2 progress seam.
type WatermarkOptions ¶
type WatermarkOptions struct {
RefreshInterval time.Duration
LeaseTTL time.Duration
FenceLockTimeout time.Duration
Owner string
Observer WatermarkObserver
InstallationID int64
}
WatermarkOptions configures the leased C-S2 maintenance loop.
type WatermarkProgress ¶
type WatermarkProgress struct {
SafeSeq int64
MaxSeq int64
CandidateSeq *int64
Advanced bool
// FenceTimedOut is a bounded, retryable step outcome rather than a
// watermarker failure. Observers meter it separately from advances.
FenceTimedOut bool
}
WatermarkProgress is one observed state of the public stream watermark.
type Watermarker ¶
type Watermarker struct {
// contains filtered or unexported fields
}
Watermarker advances stream_watermark.safe_seq while holding the exclusive side of the outbox writer fence.
func NewWatermarker ¶
func NewWatermarker( pool *pgxpool.Pool, options WatermarkOptions, ) (*Watermarker, error)
NewWatermarker constructs a leader-coordinated C-S2 watermarker.
func (*Watermarker) Close ¶
func (w *Watermarker) Close(ctx context.Context) error
Close releases this runtime's singleton lease. It does not alter the safe watermark.
func (*Watermarker) Run ¶
func (w *Watermarker) Run(ctx context.Context) error
Run maintains the watermark at approximately RefreshInterval. Standby processes keep attempting the expiring lease, so failover needs no external coordinator (C-O2).
func (*Watermarker) Step ¶
func (w *Watermarker) Step( ctx context.Context, ) (WatermarkProgress, error)
Step renews or acquires the singleton lease, waits for all registered outbox writers, and publishes the greatest committed sequence. Read-only snapshots and unrelated transactions do not participate in this fence.
Source Files