Documentation
¶
Overview ¶
Package bus provides the embedded NATS JetStream message broker and the subject/stream definitions used for sqi-server's internal messaging.
Subject naming ¶
Each subject class follows a hierarchical naming scheme so that JetStream stream subject-filters and consumer subject-bindings are intuitive:
work.lease.<queue> — worker → server: request/reply work-lease batch task.status.<job> — worker → server: task state transitions task.logs.<task> — worker → server: log chunk ingestion worker.heartbeat — worker → server: periodic liveness pings worker.register — worker → server: capability advertisement worker.deregister — worker → server: graceful departure notification
The leaf token (<queue>, <job>, <task>) is the opaque string identifier of the corresponding entity in the SQLite store. Callers build full subject strings using the helper functions below rather than constructing them by hand.
Index ¶
- Constants
- func TaskCancelSubject(taskID string) string
- func TaskLogsSubject(taskID string) string
- func TaskStatusSubject(jobID string) string
- func WorkLeaseSubject(queueID string) string
- func WorkerDiagSubject(workerID string) string
- type Broker
- type BrokerConfig
- type Client
- func (c *Client) Close()
- func (c *Client) ConsumeTaskLogs(ctx context.Context, handler jetstream.MessageHandler) (jetstream.ConsumeContext, error)
- func (c *Client) ConsumeTaskStatus(ctx context.Context, handler jetstream.MessageHandler) (jetstream.ConsumeContext, error)
- func (c *Client) ConsumeWorker(ctx context.Context, handler jetstream.MessageHandler) (jetstream.ConsumeContext, error)
- func (c *Client) Drain(ctx context.Context) error
- func (c *Client) PublishTaskCancel(ctx context.Context, taskID string, data []byte) error
- func (c *Client) PublishTaskLog(ctx context.Context, taskID string, data []byte) error
- func (c *Client) PublishTaskStatus(ctx context.Context, jobID string, data []byte) error
- func (c *Client) PublishWorkerDiag(_ context.Context, workerID string, data []byte) error
- func (c *Client) PublishWorkerHeartbeat(ctx context.Context, data []byte) error
- func (c *Client) PublishWorkerRegister(ctx context.Context, data []byte) error
- func (c *Client) RequestLease(ctx context.Context, queueID string, data []byte, timeout time.Duration) ([]byte, error)
- func (c *Client) SubscribeLease(handler func(queueID string, data []byte) []byte) (*nats.Subscription, error)
- func (c *Client) SubscribeWorkerDiag(handler func(subject string, data []byte)) (*nats.Subscription, error)
Constants ¶
const ( // StreamTask captures task-status transitions reported by workers back to // the server. WorkQueuePolicy ensures each update is processed exactly once // by the server-side status consumer. StreamTask = "SQI_TASK" // StreamLogs captures log chunks emitted by running tasks. LimitsPolicy // retains chunks so they can be served via the REST API even after the // producing worker has moved on. StreamLogs = "SQI_LOGS" // StreamWorker captures worker-registration and worker-heartbeat messages. // WorkQueuePolicy ensures the server processes each registration and // heartbeat ping exactly once. StreamWorker = "SQI_WORKER" // StreamCancel captures task-cancellation signals published by the server // to assigned workers. WorkQueuePolicy ensures each signal is // delivered exactly once; MaxAge of 5 min keeps the stream small so stale // signals for already-completed tasks are discarded automatically. StreamCancel = "SQI_CANCEL" )
Stream name constants. Every JetStream stream owned by sqi-server uses the "SQI_" prefix so they are clearly identifiable if the embedded NATS instance is ever inspected externally.
const ( // SubjectTaskStatusPrefix is the prefix for task-status subjects. // Full subject: SubjectTaskStatusPrefix + "." + jobID. SubjectTaskStatusPrefix = "task.status" // SubjectTaskLogsPrefix is the prefix for task-log subjects. // Full subject: SubjectTaskLogsPrefix + "." + taskID. SubjectTaskLogsPrefix = "task.logs" // SubjectTaskCancelPrefix is the prefix for task-cancellation subjects. // Full subject: SubjectTaskCancelPrefix + "." + taskID. // The server publishes to this subject; the worker assigned to the task // consumes it and interrupts the running process. SubjectTaskCancelPrefix = "task.cancel" // SubjectWorkerHeartbeat is the subject workers publish liveness pings to. SubjectWorkerHeartbeat = "worker.heartbeat" // SubjectWorkerRegister is the subject workers publish registration // messages to when they first connect or reconnect. SubjectWorkerRegister = "worker.register" // SubjectWorkerDeregister is the subject workers publish to on graceful // shutdown so the server can mark the worker offline immediately rather // than waiting for heartbeat timeout. The server handler for this subject // calls [store.WorkerStore.UpdateWorkerStatus] with WorkerStatusOffline. SubjectWorkerDeregister = "worker.deregister" // SubjectWorkLeasePrefix is the prefix for worker work-lease requests. // Full subject: SubjectWorkLeasePrefix + "." + queueID. Core NATS // request/reply — workers ask for work; the server replies with a batch. SubjectWorkLeasePrefix = "work.lease" // WildcardQueueToken is the lease-subject leaf a queue-unaffiliated worker // (empty QueueIDs — "serve any queue") uses in place of a real queue ID, so // it requests on a valid subject (work.lease._any) that the server's // work.lease.> subscription actually receives. An empty leaf would produce // the invalid subject "work.lease." (no responders). The leaf is reserved // (underscore prefix) so it cannot collide with a real UUID queue ID; the // server selects tasks farm-wide and gates by worker eligibility, so the // token's only role is subject routing and wake-up bucketing. WildcardQueueToken = "_any" )
Subject prefix and fixed-subject constants.
const SubjectWorkerDiagPrefix = "worker.diag"
SubjectWorkerDiagPrefix is the prefix for worker diagnostic-log subjects. Full subject: SubjectWorkerDiagPrefix + "." + workerID.
Unlike task/work/status subjects, diagnostic logs use CORE NATS (not JetStream): they are best-effort and never retained on the broker. The server keeps a bounded in-memory ring buffer instead.
Variables ¶
This section is empty.
Functions ¶
func TaskCancelSubject ¶
TaskCancelSubject returns the full NATS subject for a task-cancellation signal targeting the worker that holds the given task.
func TaskLogsSubject ¶
TaskLogsSubject returns the full NATS subject for log-chunk messages belonging to the given task.
func TaskStatusSubject ¶
TaskStatusSubject returns the full NATS subject for task-status messages belonging to the given job.
func WorkLeaseSubject ¶
WorkLeaseSubject returns the full NATS subject a worker requests work on for the given queue.
func WorkerDiagSubject ¶
WorkerDiagSubject returns the full core-NATS subject a worker publishes its diagnostic log records to.
Types ¶
type Broker ¶
type Broker struct {
// contains filtered or unexported fields
}
Broker wraps an in-process NATS server with JetStream enabled and manages its lifecycle (start, health check, graceful shutdown).
After Broker.Start returns successfully, callers can obtain the server's client URL via Broker.ClientURL to establish their own NATS connections. The typed per-subsystem client wrapper is Client in client.go.
func New ¶
func New(cfg BrokerConfig, logger *slog.Logger) *Broker
New creates a Broker with the given configuration and logger. Call Broker.Start to boot the underlying NATS server.
func (*Broker) Check ¶
Check implements [health.Checker]. It returns nil when the embedded NATS server is running and the admin connection is open, or a descriptive error otherwise. Registered with the health registry during server startup so that GET /readyz reflects broker health.
func (*Broker) ClientURL ¶
ClientURL returns the nats:// URL that in-process clients should connect to. Returns an empty string if the broker has not been started yet.
func (*Broker) NewClient ¶
NewClient dials the embedded NATS server and returns a connected Client configured for automatic reconnection. It must be called after Broker.Start returns successfully.
The returned Client should be shared across goroutines; create one per server component (or one shared instance for the whole server) rather than dialing repeatedly.
func (*Broker) Shutdown ¶
func (b *Broker) Shutdown()
Shutdown drains the admin connection and stops the embedded NATS server, waiting until it has fully exited before returning. It is safe to call Shutdown more than once; subsequent calls are no-ops.
func (*Broker) Start ¶
Start boots the embedded NATS server, enables JetStream, waits for the server to accept connections, and then provisions the sqi-server stream topology via [ensureStreams]. It returns an error if any step fails.
Start must be called before any other Broker method and must not be called more than once.
type BrokerConfig ¶
type BrokerConfig struct {
// Addr is the TCP address the embedded NATS server binds to, in
// "host:port" form. Defaults to "0.0.0.0:4222" (all interfaces) so that
// workers which discover the server via mDNS can reach the broker at the
// advertised LAN host. Broker authentication is not yet in place; it
// arrives in phase 3.
Addr string
// DataDir is the directory JetStream uses for file-backed stream storage.
// Created at startup if it does not exist.
DataDir string
// MaxStoreMB is the maximum disk space JetStream may use, in megabytes.
// A value of 0 means unlimited (not recommended for production).
MaxStoreMB int
}
BrokerConfig holds the parameters needed to start the embedded NATS server.
type Client ¶
type Client struct {
// contains filtered or unexported fields
}
Client is the typed in-process NATS client used by all sqi-server code.
Rather than importing the raw nats package directly, every component that needs to publish or subscribe goes through Client. This indirection keeps the raw nats API out of callers, making it easy to add instrumentation, swap transport, or substitute a [FakeClient] in unit tests.
Obtain a Client via Broker.NewClient after the broker has been started. A Client should be shared across goroutines — its methods are safe for concurrent use.
func NewClient ¶
NewClient dials the embedded NATS server at url and returns a connected Client. The connection is configured for unlimited reconnects with exponential backoff, so transient disruptions (e.g. a brief broker restart during development) recover automatically.
Prefer Broker.NewClient over calling NewClient directly; the Broker supplies the correct loopback URL without the caller needing to know it.
func (*Client) Close ¶
func (c *Client) Close()
Close closes the underlying NATS connection immediately without draining. Prefer Client.Drain for graceful shutdown; use Close only when draining has already completed or is not possible.
func (*Client) ConsumeTaskLogs ¶
func (c *Client) ConsumeTaskLogs(ctx context.Context, handler jetstream.MessageHandler) (jetstream.ConsumeContext, error)
ConsumeTaskLogs creates or updates the server-side durable push consumer for log-chunk messages (SQI_LOGS stream) and begins delivering them to handler. The returned ConsumeContext is tracked internally and stopped automatically during Client.Drain.
Ack / redelivery policy:
- AckWait 30 s — the server has 30 s to persist a log chunk to SQLite. The SQLite write path is a single INSERT so this window is very generous.
- MaxDeliver 5 — allows for transient DB errors without re-filling the stream indefinitely. After exhaustion the chunk is discarded; log gaps are recoverable from the NATS JetStream replay if MaxAge permits.
- DeliverAllPolicy — replays unprocessed log chunks from before the last server restart so no output is permanently lost across restarts.
func (*Client) ConsumeTaskStatus ¶
func (c *Client) ConsumeTaskStatus(ctx context.Context, handler jetstream.MessageHandler) (jetstream.ConsumeContext, error)
ConsumeTaskStatus creates or updates the server-side durable push consumer for task-status messages (SQI_TASK stream) and begins delivering them to handler. The returned ConsumeContext is tracked internally and stopped automatically during Client.Drain.
Ack / redelivery policy:
- AckWait 30 s — the server has 30 s to persist a status update to SQLite before NATS redelivers.
- MaxDeliver 10 — allows for transient DB errors without dropping updates.
- DeliverAllPolicy — on first consumer creation, replay any unprocessed messages from before the last server restart so no state transitions are silently lost. NATS tracks the consumer sequence for existing durables, so subsequent starts pick up where they left off.
func (*Client) ConsumeWorker ¶
func (c *Client) ConsumeWorker(ctx context.Context, handler jetstream.MessageHandler) (jetstream.ConsumeContext, error)
ConsumeWorker creates or updates the server-side durable push consumer for worker-registration and heartbeat messages (SQI_WORKER stream) and begins delivering them to handler. The returned ConsumeContext is tracked internally and stopped automatically during Client.Drain.
Ack / redelivery policy:
- AckWait 10 s — heartbeats are time-sensitive; a 10 s window is well within the scheduler's heartbeat-timeout sweep interval (≈30 s) so delays are caught before workers are incorrectly marked offline.
- MaxDeliver 3 — heartbeats are cheap and frequent; a low retry limit avoids piling up stale pings.
- DeliverAllPolicy — replays unprocessed registration/heartbeat messages on startup, ensuring no worker comes online without the server noticing.
func (*Client) Drain ¶
Drain stops all push consumers started by this Client, waits for in-flight message handlers to finish, and then drains the underlying NATS connection (flushing pending publish-acks and waiting for all outstanding messages to be delivered). It respects the given context deadline.
Call Drain before Client.Close or Broker.Shutdown to guarantee a clean shutdown — no in-flight messages are dropped and no pending publishes are lost. After Drain returns, no further messages will arrive at handlers.
func (*Client) PublishTaskCancel ¶
PublishTaskCancel publishes a task-cancellation signal to the task.cancel.<taskID> subject. The scheduler calls this for each task that was actively assigned or running at the time a job is canceled. The worker holding the task consumes this message and interrupts the running process without waiting for the next heartbeat timeout.
func (*Client) PublishTaskLog ¶
PublishTaskLog publishes a log chunk to the task.logs.<taskID> subject. Workers call this continuously as a task produces output; the server retains chunks in JetStream for later retrieval via the REST and WebSocket APIs.
func (*Client) PublishTaskStatus ¶
PublishTaskStatus publishes a task-status transition to the task.status.<jobID> subject. Workers call this when a task transitions to running, succeeded, failed, or canceled.
func (*Client) PublishWorkerDiag ¶
PublishWorkerDiag publishes a diagnostic log record to worker.diag.<workerID> over core NATS (fire-and-forget, no JetStream ack). Errors are returned but callers should treat them as non-fatal: diagnostics are best-effort.
func (*Client) PublishWorkerHeartbeat ¶
PublishWorkerHeartbeat publishes a heartbeat ping on the worker.heartbeat subject. Workers call this on a regular interval; the server-side heartbeat sweep marks workers offline when pings stop.
func (*Client) PublishWorkerRegister ¶
PublishWorkerRegister publishes a capability advertisement on the worker.register subject. Workers call this on first connect and after any reconnect so the server always has a current view of their capabilities.
func (*Client) RequestLease ¶
func (c *Client) RequestLease(ctx context.Context, queueID string, data []byte, timeout time.Duration) ([]byte, error)
RequestLease sends a core-NATS work-lease request for queueID and waits up to timeout for the server's reply. Returns nats.ErrNoResponders immediately if no server is subscribed, or nats.ErrTimeout if no reply arrives in time.
func (*Client) SubscribeLease ¶
func (c *Client) SubscribeLease(handler func(queueID string, data []byte) []byte) (*nats.Subscription, error)
SubscribeLease subscribes to work-lease requests for all queues (work.lease.>) and replies with the bytes handler returns. handler does its own long-poll blocking before returning; it must respect no work by returning an empty/again-marker payload of the caller's choosing.
func (*Client) SubscribeWorkerDiag ¶
func (c *Client) SubscribeWorkerDiag(handler func(subject string, data []byte)) (*nats.Subscription, error)
SubscribeWorkerDiag subscribes to all worker diagnostic subjects (worker.diag.>) over core NATS and invokes handler for each message with the concrete subject and raw payload. The returned subscription must be unsubscribed/drained by the caller (the server does this at shutdown).
Source Files