The highest tagged major version is
Documentation
¶
Overview ¶
Package trigger is the single-owner trigger broker shared by the fleet's socket-shaped scheduler daemons (docker-renovate-scheduler, docker-rsync-scheduler, docker-fclones-scheduler).
The shape: one daemon process (PID 1) owns every job execution; triggers — a built-in ticker, each `run`/`sync`/`scan` client exec — only submit requests. A bounded FIFO Queue carries the requests to the daemon's single executor goroutine (mutual exclusion is that loop; nothing else may start a job), a Server accepts requests on an owner-only in-container unix socket and streams lifecycle events back, and Submit is the thin synchronous client that forwards one request and blocks until the run's own result. There is no coalescing: every accepted request gets its own run and its own true result, in arrival order.
The request payload is a type parameter. A daemon whose runs take arguments declares a struct (docker-renovate-scheduler forwards repo slugs plus its complete environment); an argless daemon uses an empty struct, which frames as `{}` on the wire. Client and daemon ship in one binary inside one image, so the wire format (newline-delimited JSON, see Event) carries no version field and payload evolution needs only ordinary optional-field care.
The package brokers requests and results; it deliberately owns no policy. What a job does, how its outcome maps to health, when shutdown cancels versus drains, and the exact wording of the app's lifecycle log lines all stay in the consuming app (the mechanism-vs-policy split the parent scheduler package applies to SlotFile). The library's own log lines are payload-free transport diagnostics only.
Unix-only, like the parent package: the socket hygiene in Listen relies on umask(2) and unix domain sockets.
Example ¶
Example wires the whole broker: a daemon-side queue served by one executor goroutine and exposed on a unix socket, and a client that submits one run and waits for its result — the fleet's single-owner scheduler shape.
package main
import (
"fmt"
"os"
"path/filepath"
"time"
"github.com/cplieger/scheduler/v2/trigger"
)
// payload is an app's request type: the arguments one triggered run carries.
// An argless daemon uses struct{} instead.
type payload struct {
Repos []string `json:"repos,omitempty"`
}
// Example wires the whole broker: a daemon-side queue served by one executor
// goroutine and exposed on a unix socket, and a client that submits one run
// and waits for its result — the fleet's single-owner scheduler shape.
func main() {
dir, err := os.MkdirTemp("/tmp", "trigger-example-")
if err != nil {
fmt.Println("tempdir:", err)
return
}
defer func() { _ = os.RemoveAll(dir) }()
socketPath := filepath.Join(dir, "trigger.sock")
// Daemon side: one bounded queue, one executor goroutine (the single
// owner of execution), one socket server bridging requests in.
queue := trigger.NewQueue[payload](16)
executorDone := make(chan struct{})
go func() {
defer close(executorDone)
for job := range queue.Jobs() {
job.Start()
// The app's real work runs here, with the job's exact payload.
job.Finish(trigger.Outcome{OK: true, Duration: 42 * time.Millisecond})
}
}()
ln, err := trigger.Listen(socketPath)
if err != nil {
fmt.Println("listen:", err)
return
}
srv := &trigger.Server[payload]{Queue: queue}
srv.Serve(ln)
// Client side (the `run` subcommand): submit one request, block until
// its own result, map it to an exit code.
final, err := trigger.Submit(socketPath, payload{Repos: []string{"owner/repo"}}, nil)
if err != nil {
fmt.Println("submit:", err)
return
}
fmt.Println("ok:", final.OK)
// Daemon shutdown: stop admission, drain, wait for the handlers.
_ = ln.Close()
queue.Close()
<-executorDone
srv.Wait()
}
Output: ok: true
Index ¶
Examples ¶
Constants ¶
const ( EventQueued = "queued" EventStarted = "started" EventDone = "done" )
Event kinds, in wire order.
const DialTimeout = 5 * time.Second
DialTimeout bounds the connection attempt: the daemon is PID 1 in the same container, so anything slower than instant means it is not accepting.
const TriggerExternal = "external"
TriggerExternal is the Trigger label the Server stamps on socket-submitted jobs.
Variables ¶
var ( // ErrUnreachable wraps a failed dial: no daemon is accepting on the // socket (container down, or a mismatched exec user against the // owner-only socket file). ErrUnreachable = errors.New("cannot reach the scheduler daemon") // ErrSend wraps a failed request write. ErrSend = errors.New("cannot send trigger request") // ErrConnectionLost wraps an event stream that ended before the final // done event: the daemon died or was stopped mid-run. ErrConnectionLost = errors.New("connection lost before the run completed") )
Submit failure classes, distinguishable with errors.Is so the app can log each in its own vocabulary.
var ( // ErrClosed rejects submissions once Close has stopped admission. ErrClosed = errors.New("scheduler is shutting down") // ErrFull rejects submissions while the queue is at capacity — honest // backpressure, never unbounded queueing. ErrFull = errors.New("run queue is full") )
Queue rejection errors. Their messages travel the wire verbatim as the rejection Reason a waiting client logs, so they are part of the trigger contract.
Functions ¶
Types ¶
type Event ¶
type Event struct {
// Kind is the event discriminator: EventQueued, EventStarted, EventDone.
Kind string `json:"event"`
// Reason explains a not-OK outcome that isn't a plain job failure (queue
// full, cancelled by shutdown), or annotates an OK outcome that carries a
// caveat (an app-defined skip tolerance).
Reason string `json:"reason,omitempty"`
// DurationMs is the elapsed execution time on EventDone. Zero when the
// request was rejected or cancelled before running.
DurationMs int64 `json:"duration_ms,omitempty"`
// OK is meaningful only on EventDone: the run's outcome (never omitted,
// so a failed run is explicit on the wire).
OK bool `json:"ok"`
}
Event is one status line the daemon streams back. The client receives EventQueued on acceptance, EventStarted when the executor picks the request up (the gap between the two is queue wait behind an in-flight run), and exactly one EventDone as the final line.
func Submit ¶
Submit performs one triggered run via the daemon at socketPath: it sends payload as the request line, relays each intermediate lifecycle event to onEvent (EventQueued, EventStarted; nil onEvent skips relaying; unknown kinds are ignored for forward compatibility), and returns the final done event. A non-nil error wraps ErrUnreachable, ErrSend, or ErrConnectionLost; the Event is only meaningful when the error is nil. Submit blocks for the run's full queue-wait plus execution — triggered runs are synchronous by contract (the trigger's exit code is the run's result), so there is no read deadline on the event stream.
type Job ¶
type Job[P any] struct { // Payload carries the request's arguments (the app's own type; an // argless daemon uses an empty struct). Payload P // Trigger labels the run's origin in logs: TriggerExternal for socket // requests; apps use their own labels (startup, interval) for ticker // jobs. Trigger string // contains filtered or unexported fields }
Job is one queued run request. The executor signals lifecycle through it: Start the moment the run begins, then Finish with the single result — exactly once per accepted job, from the run itself or from shutdown cancellation.
func (*Job[P]) Finish ¶
Finish delivers the job's single result. Exactly one Finish per accepted job; the buffered result channel means the caller never blocks on a departed waiter.
func (*Job[P]) Result ¶
Result receives the job's exactly-one outcome — from the run itself, or from shutdown cancellation.
type Outcome ¶
type Outcome struct {
// Reason explains a not-OK outcome that isn't a plain job failure
// (cancelled by shutdown, a failed preflight), or annotates an OK outcome
// that carries a caveat.
Reason string
// Duration is the elapsed execution time; zero when the job never ran.
Duration time.Duration
// OK is the run's outcome.
OK bool
}
Outcome is a Job's final result.
type Queue ¶
type Queue[P any] struct { // contains filtered or unexported fields }
Queue is the bounded FIFO between triggers and the executor. Submission is non-blocking: a full or closed queue rejects immediately. The channel is the queue; the executor is its only receiver.
func NewQueue ¶
NewQueue builds a queue holding at most capacity pending jobs. Size it for the realistic trigger set (a periodic job plus a trigger burst), not for storage: a client hitting a full queue is rejected immediately with a clear reason rather than queued unboundedly.
func (*Queue[P]) Close ¶
func (q *Queue[P]) Close()
Close stops admission and closes the channel, letting the executor's range loop drain the already-queued jobs (the executor cancels each once shutdown is signalled) and terminate. Idempotent; called at shutdown.
type Server ¶
type Server[P any] struct { // Queue receives every decoded request as a TriggerExternal job. Queue *Queue[P] // OnAccepted, when non-nil, runs after a request is queued (the app's // "triggered run queued" line). OnAccepted func(payload P) // OnRejected, when non-nil, runs after a submission is rejected with // ErrFull or ErrClosed (the app's rejection warning). Undecodable // requests never reach it; the library logs those without the payload. OnRejected func(payload P, err error) // contains filtered or unexported fields }
Server accepts run requests and bridges them onto the queue.
The zero value is not usable; set Queue. The hooks are optional: they exist so the app can log acceptance and rejection in its own vocabulary and with its own payload attributes (the library never logs payload contents — a forwarded environment can carry secrets).
Source Files