Documentation
¶
Index ¶
- type PrioritySource
- func (s *PrioritySource) Ack(ctx context.Context, msg hermod.Message) error
- func (s *PrioritySource) Close() error
- func (s *PrioritySource) GetLag(ctx context.Context) (uint64, error)
- func (s *PrioritySource) GetState() map[string]string
- func (s *PrioritySource) IsReady(ctx context.Context) error
- func (s *PrioritySource) LastStreamActivity() time.Time
- func (s *PrioritySource) PendingWork() (pending bool, known bool)
- func (s *PrioritySource) Ping(ctx context.Context) error
- func (s *PrioritySource) Read(ctx context.Context) (hermod.Message, error)
- func (s *PrioritySource) SetLogger(l hermod.Logger)
- func (s *PrioritySource) SetState(state map[string]string)
- func (s *PrioritySource) StreamSilenceThreshold() time.Duration
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type PrioritySource ¶
type PrioritySource struct {
// contains filtered or unexported fields
}
PrioritySource wraps two sources and prioritizes reading from 'recovery' before 'primary'.
func NewPrioritySource ¶
func NewPrioritySource(recovery, primary hermod.Source, logger hermod.Logger) *PrioritySource
func (*PrioritySource) Close ¶
func (s *PrioritySource) Close() error
func (*PrioritySource) GetLag ¶
func (s *PrioritySource) GetLag(ctx context.Context) (uint64, error)
GetLag reports the outstanding work held by the wrapped sources.
Embedding or wrapping a source does not carry its optional interfaces across, and every lag-based check downstream — workflow status, WAL-retention alerting, and the stall watchdog's "is any work outstanding?" question (pkg/engine/stall.go:164) — reads zero when the assertion fails. Zero lag on a wedged CDC pipeline is indistinguishable from an idle healthy one, so the wedge goes unreported for as long as it lasts. This is the same defect already found and fixed in MetricsSource (pkg/comm/source/decorators.go:120).
func (*PrioritySource) GetState ¶
func (s *PrioritySource) GetState() map[string]string
func (*PrioritySource) IsReady ¶
func (s *PrioritySource) IsReady(ctx context.Context) error
IsReady runs each wrapped source's deep readiness check, falling back to Ping for sources that do not have one.
Without this, wrapping a source for DLQ priority downgraded the engine's health check from the CDC-aware IsReady (which verifies wal_level, that the replication slot exists and is active, and that the publication is present — pkg/comm/source/postgres/postgres.go:1671) to a plain connection Ping, which keeps returning healthy long after replication has stopped.
func (*PrioritySource) LastStreamActivity ¶
func (s *PrioritySource) LastStreamActivity() time.Time
LastStreamActivity reports the primary source's stream liveness. Only the primary holds a replication stream; the recovery source is a dead-letter queue being drained, which has no server-push cadence to vouch for.
func (*PrioritySource) PendingWork ¶
func (s *PrioritySource) PendingWork() (pending bool, known bool)
PendingWork reports whether either wrapped source is still owed acknowledgements: a DLQ being drained holds outstanding work just as a primary stream does.
func (*PrioritySource) SetLogger ¶
func (s *PrioritySource) SetLogger(l hermod.Logger)
SetLogger forwards the engine's logger to both wrapped sources and keeps a copy for this wrapper's own DLQ-priority reporting.
func (*PrioritySource) SetState ¶
func (s *PrioritySource) SetState(state map[string]string)
func (*PrioritySource) StreamSilenceThreshold ¶
func (s *PrioritySource) StreamSilenceThreshold() time.Duration
StreamSilenceThreshold reports the primary source's silence deadline.
Source Files