Documentation
¶
Overview ¶
Package cancel implements per-task cancel-signal subscriptions for sqi-worker.
When a job is canceled on sqi-server the scheduler publishes a task.cancel.<taskID> NATS message for each task that was actively executing. Handler subscribes to those subjects and routes the signal to the executor via TaskCanceler.Cancel.
Subject scheme ¶
The server publishes to task.cancel.<taskID>. A per-task NATS core subscription is created in Handler.Register when the executor begins processing each task and torn down in Handler.Deregister when the task completes. Using task-specific subscriptions avoids the routing ambiguity that would arise if multiple workers shared a single consumer on the SQI_CANCEL JetStream stream with WorkQueuePolicy.
Durability ¶
Core subscriptions miss messages published while the worker is disconnected. This is acceptable because: (a) the SQI_CANCEL JetStream stream retains cancel signals for up to 5 minutes, providing a durability backstop; and (b) the server's heartbeat sweep reclaims stale assignments regardless of whether the cancel signal was delivered.
Usage ¶
h := cancel.New(nc, exec, logger) exec.SetCancelRegistrar(h)
Index ¶
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type Handler ¶
type Handler struct {
// contains filtered or unexported fields
}
Handler manages per-task NATS cancel subscriptions.
Create one with New and pass it to [executor.Executor.SetCancelRegistrar] before starting the pull loop. The executor calls [Register] before each task process starts and [Deregister] when the task goroutine exits.
Handler is safe for concurrent use from multiple goroutines.
func New ¶
func New(nc natsSubscriber, canceler TaskCanceler, logger *slog.Logger) *Handler
New creates a ready-to-use Handler.
nc is the worker's NATS connection (implements [natsSubscriber]). canceler is typically the worker's Executor.
func (*Handler) Deregister ¶
Deregister unsubscribes from task.cancel.<taskID>.
Safe to call multiple times; a missing subscription is a no-op.
func (*Handler) Register ¶
Register subscribes to task.cancel.<taskID> so that cancel signals published by the server are delivered to TaskCanceler.Cancel.
Calling Register for the same taskID more than once replaces the existing subscription. Errors are returned so the executor can log them; they do not abort task execution.
type TaskCanceler ¶
type TaskCanceler interface {
// Cancel requests cancellation of the task identified by taskID.
// Returns true if the task was found and cancellation was initiated,
// false if the task is unknown (already finished or never dispatched).
Cancel(taskID string) bool
}
TaskCanceler is implemented by the executor to cancel a specific in-progress task. [*executor.Executor] implements this interface.
Source Files