Documentation
¶
Overview ¶
Package heartbeat implements the worker's periodic liveness signal to sqi-server.
Overview ¶
A Publisher ticks on a configurable interval and publishes a protocol.HeartbeatMsg to the bus.SubjectWorkerHeartbeat NATS subject. Each message carries the worker's current runtime state (active task count, active task IDs, uptime, last assignment time) so the server can detect stale assignments without additional store queries.
Watchdog ¶
Publisher.Run also spawns an internal watchdog goroutine that polls the NATS connection status every [watchdogPollInterval]. When it observes a transition from a non-connected state back to connected it calls Registrar.Register — but only if the reconnect callback installed by [registration.Registrar.SetupReconnectHook] did not already succeed (as indicated by Registrar.LastRegisteredAt). This guards against the case where the callback's publish fails or is lost without producing a duplicate registration on every normal reconnect.
Latency warning ¶
After publishing a heartbeat the Publisher calls [nats.Conn.FlushTimeout] to drain the NATS outbound buffer to the TCP layer. If the flush does not complete within half the configured interval the Publisher logs a warning, indicating that the NATS server or network connection is under strain and backpressure is building in the write buffer.
StateSource ¶
StateSource is the interface through which the Publisher reads live task state. It is satisfied by the executor. Until the executor exists, callers pass a NoopStateSource which reports zero active tasks.
Index ¶
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type NoopStateSource ¶
type NoopStateSource struct{}
NoopStateSource is a StateSource that always reports zero active tasks. Use it before the executor is available.
func (NoopStateSource) ActiveTaskCount ¶
func (NoopStateSource) ActiveTaskCount() int
ActiveTaskCount always returns 0.
func (NoopStateSource) ActiveTaskIDs ¶
func (NoopStateSource) ActiveTaskIDs() []string
ActiveTaskIDs always returns nil.
func (NoopStateSource) LastAssignmentAt ¶
func (NoopStateSource) LastAssignmentAt() *time.Time
LastAssignmentAt always returns nil.
type Publisher ¶
type Publisher struct {
// contains filtered or unexported fields
}
Publisher periodically publishes protocol.HeartbeatMsg to bus.SubjectWorkerHeartbeat and runs an internal watchdog goroutine that triggers re-registration when the NATS connection is restored after a drop and the reconnect callback did not already succeed.
Obtain one with New and call Publisher.Run in a goroutine; it blocks until ctx is canceled.
func New ¶
func New( nc natsConn, workerID string, maxConcurrentTasks int, interval time.Duration, state StateSource, registrar Registrar, logger *slog.Logger, ) *Publisher
New creates a Publisher. nc and registrar must be non-nil.
- workerID: the stable worker identity string published in each message.
- maxConcurrentTasks: the worker's configured concurrency ceiling, included in each heartbeat for server-side capacity tracking.
- interval: how often to publish; must be > 0.
- state: provides live task counts and IDs; pass NoopStateSource until the executor is wired in.
- registrar: called by the watchdog when a reconnect is detected and the callback did not already re-register.
type Registrar ¶
type Registrar interface {
// Register publishes a registration message to the server. Safe to call
// multiple times; the server upserts based on worker ID.
Register(ctx context.Context) error
// LastRegisteredAt returns the time of the most recent successful Register
// call, or the zero time if Register has never been called successfully.
// The watchdog uses this to avoid re-registering when the reconnect callback
// already handled it.
LastRegisteredAt() time.Time
}
Registrar is the subset of [registration.Registrar] used by the watchdog to trigger re-registration when a NATS reconnect is detected.
Defined as an interface so that tests can substitute a lightweight stub without constructing a full Registrar.
type StateSource ¶
type StateSource interface {
// ActiveTaskCount returns the number of task processes currently running.
ActiveTaskCount() int
// ActiveTaskIDs returns a snapshot of the IDs of all currently running
// tasks. Implementations must return a value that is safe for the caller
// to hold across concurrent mutations of the underlying task set — either
// a snapshot copy or an immutable slice.
ActiveTaskIDs() []string
// LastAssignmentAt returns the time of the most recent task assignment,
// or nil if the worker has not yet received any assignment in this process
// lifetime.
LastAssignmentAt() *time.Time
}
StateSource provides the Publisher with live task-execution state. Implemented by the executor package; use NoopStateSource until then.
Source Files