Documentation
¶
Index ¶
- Constants
- Variables
- func ParseNodeLabels(raw string) map[string]string
- func SubscribeWakeups(ctx context.Context, bus event.Bus, extra ...<-chan struct{}) <-chan struct{}
- func WakeupURLForNodeAddress(nodeAddress string, apiPort int) (string, error)
- type ClaimInspector
- type Claimer
- type CompletionSink
- type DistributedWakeupConfig
- type DistributedWakeups
- type ExpiredReclaimer
- type LeaseRenewer
- type Pool
- type ReclaimGate
- type ReclaimGateFunc
- type RunLeaseRenewer
- type TaskClaimer
- type TaskExecutor
- type WakeupMessage
- type WakeupPeerResolver
- type WakeupPeerResolverFunc
- type WakeupSignaler
- type Worker
- func (w *Worker) Run(ctx context.Context) error
- func (w *Worker) SubmitDispatched(d dispatch.InboundDispatch) error
- func (w *Worker) WithInboundDispatch(completionToken string) *Worker
- func (w *Worker) WithLeaseRenewal(renewer LeaseRenewer, leaseTTL, renewInterval time.Duration) *Worker
- func (w *Worker) WithReclaimGate(gate ReclaimGate) *Worker
- func (w *Worker) WithReclaimInterval(interval time.Duration) *Worker
- func (w *Worker) WithRunLeaseRenewal(renewer RunLeaseRenewer, leaseTTL time.Duration, nodeID string) *Worker
- func (w *Worker) WithWakeups(ch <-chan struct{}) *Worker
Constants ¶
const ( WakeupFanoutFull = "full" WakeupFanoutGossip = "gossip" )
Variables ¶
var ErrInboundFull = errors.New("worker: no execution capacity for dispatched task")
ErrInboundFull is returned by SubmitDispatched when the worker has no free pool slot to run the task on (it is saturated). The dispatch handler treats this as a rejectable condition and rolls the claim back so the owner re-dispatches — which is exactly right, because the owner has already flipped the row to `running` by the time it asks the worker to take it.
var ErrWorkerNotAccepting = errors.New("worker: not accepting dispatched tasks")
ErrWorkerNotAccepting is returned by SubmitDispatched when the worker is not configured to accept dispatched tasks (WithInboundDispatch was never called).
Functions ¶
func ParseNodeLabels ¶
func SubscribeWakeups ¶
Types ¶
type ClaimInspector ¶
type ClaimInspector interface {
ClaimedTaskRunIDs(ctx context.Context, nodeID string, ids []uuid.UUID) ([]uuid.UUID, error)
}
ClaimInspector answers "of these task runs, which do I still hold the claim on?" without writing anything. Implemented by *run.Store; discovered from the LeaseRenewer by type assertion, the same way ExpiredReclaimer is discovered from the TaskClaimer, so a renewer that cannot answer simply falls back to the RenewLeases row count.
type Claimer ¶
type Claimer struct {
// contains filtered or unexported fields
}
func NewClaimer ¶
func (*Claimer) ClaimNext ¶
ClaimNext claims one ready task, or returns nil when no tasks are available.
func (*Claimer) WithRateLimiter ¶
WithRateLimiter gates claimed tasks before they enter the worker pool.
type CompletionSink ¶
type CompletionSink interface {
// Succeeded finalizes a task that ran to a normal completion (whatever the
// underlying container result was — the result string carries it).
Succeeded(ctx context.Context, taskRun *models.TaskRun, result string, outputs map[string]string, branchSelections []string) error
// Failed finalizes a task whose attempts were exhausted with an error.
Failed(ctx context.Context, taskRun *models.TaskRun, failure error) error
// Cached finalizes a task satisfied from the result cache.
Cached(ctx context.Context, taskRun *models.TaskRun, source run.CacheHitSource, result string, outputs map[string]string, branchSelections []string) error
}
CompletionSink is the abstraction the runtime executor calls to finalize a task's terminal outcome. It exists so a single execution path can route its completion either to the local DB (the ClaimNext pull path, unchanged from Phase 1) or back to the owning node over /internal/complete (the run-owner push-dispatch path).
The three methods mirror the three store finalization calls the executor made directly before this abstraction was introduced:
Succeeded → run.Store.CompleteTaskClaimed Failed → run.Store.FailTaskClaimed Cached → run.Store.CacheHitTaskClaimed
The owner re-derives the real terminal status from the result string in CompleteTaskClaimed, so a "failure" result routed through Succeeded still lands as TaskStatusFailed on the owner — byte-identical to the local path.
func NewLocalSink ¶
func NewLocalSink(store *run.Store) CompletionSink
NewLocalSink returns the default DB-backed completion sink.
type DistributedWakeupConfig ¶
type DistributedWakeupConfig struct {
Token string
FanoutMode string
Signaler *WakeupSignaler
Resolver WakeupPeerResolver
HTTPClient *http.Client
}
type DistributedWakeups ¶
type DistributedWakeups struct {
// contains filtered or unexported fields
}
func NewDistributedWakeups ¶
func NewDistributedWakeups(cfg DistributedWakeupConfig) *DistributedWakeups
func (*DistributedWakeups) HandleRemote ¶
func (d *DistributedWakeups) HandleRemote(ctx context.Context, msg WakeupMessage)
type ExpiredReclaimer ¶
type LeaseRenewer ¶
type LeaseRenewer interface {
RenewLeases(ctx context.Context, nodeID string, ids []uuid.UUID, newExpiresAt time.Time) (int64, error)
}
LeaseRenewer is implemented by any component that can issue a single batched UPDATE extending claim_expires_at for a set of in-flight task runs. Returns the number of rows actually updated; useful for both metric accuracy and detecting concurrent claim reassignment.
type Pool ¶
type Pool struct {
// contains filtered or unexported fields
}
Pool bounds concurrent goroutines using a semaphore.
The reservation half of the API (TryAcquire/Acquire → Go or Release) exists so a caller can prove it has capacity BEFORE it takes an action that is externally visible as "this task is running". Claiming a task flips its row to `running` in the catalog, so a claim issued without a slot advertises work the worker cannot start until some other task finishes — see Worker.Run and Worker.SubmitDispatched.
func (*Pool) Acquire ¶
Acquire reserves one execution slot, blocking until one frees or ctx is done.
func (*Pool) Free ¶
func (p *Pool) Free() <-chan struct{}
Free is the wake-only signal poked whenever a slot is released. Select on it alongside a timer to wait for capacity without polling.
func (*Pool) Go ¶
func (p *Pool) Go(fn func())
Go runs fn on a slot the caller has ALREADY reserved via TryAcquire/Acquire. The slot is released when fn returns.
func (*Pool) Release ¶
func (p *Pool) Release()
Release returns a reserved slot that will not be used.
func (*Pool) TryAcquire ¶
TryAcquire reserves one execution slot without blocking, reporting whether a slot was available. A successful reservation MUST be handed to Go (which consumes it) or returned with Release.
type ReclaimGateFunc ¶
func (ReclaimGateFunc) CanReclaim ¶
func (f ReclaimGateFunc) CanReclaim(ctx context.Context) (bool, error)
type RunLeaseRenewer ¶
type RunLeaseRenewer interface {
// RenewOwnedLeases extends lease_expires_at in a single UPDATE for every
// non-expired lease owned by ownerNode. Returns the number of rows
// renewed, which is also the count of currently owned, non-expired runs.
RenewOwnedLeases(ctx context.Context, ownerNode string, newExpiresAt time.Time) (int64, error)
}
RunLeaseRenewer is implemented by the run.LeaseStore and used by the worker's run-lease renewal goroutine.
type TaskClaimer ¶
type TaskExecutor ¶
func NewRuntimeExecutor ¶
type WakeupMessage ¶
type WakeupPeerResolver ¶
func NewCachedWakeupPeerResolver ¶
func NewCachedWakeupPeerResolver(resolver WakeupPeerResolver, ttl time.Duration) WakeupPeerResolver
type WakeupPeerResolverFunc ¶
func (WakeupPeerResolverFunc) WakeupPeers ¶
func (f WakeupPeerResolverFunc) WakeupPeers(ctx context.Context) ([]string, error)
type WakeupSignaler ¶
type WakeupSignaler struct {
// contains filtered or unexported fields
}
func NewWakeupSignaler ¶
func NewWakeupSignaler() *WakeupSignaler
func (*WakeupSignaler) C ¶
func (s *WakeupSignaler) C() <-chan struct{}
func (*WakeupSignaler) Signal ¶
func (s *WakeupSignaler) Signal()
type Worker ¶
type Worker struct {
// contains filtered or unexported fields
}
func NewWorker ¶
func NewWorker(claimer TaskClaimer, pool *Pool, pollInterval time.Duration, executor TaskExecutor) *Worker
func (*Worker) SubmitDispatched ¶
func (w *Worker) SubmitDispatched(d dispatch.InboundDispatch) error
SubmitDispatched enqueues a dispatched task for execution on the worker's shared pool. It is non-blocking: it returns ErrInboundFull when the worker has no free pool slot and ErrWorkerNotAccepting when the inbound path is disabled. *Worker implements dispatch.WorkerSubmitter via this method, so the dispatch handler can hand it accepted tasks directly (no adapter needed).
The non-blocking contract is deliberate: HandleDispatch runs on an HTTP request goroutine and must not block on a full pool. Saturation surfaces as a 409 the owner retries, rather than holding the dispatch RPC open.
A pool slot is RESERVED here, before the task is accepted, so acceptance and capacity are the same decision. The owner flips the row to `running` before it POSTs the dispatch and rolls that claim back when this method errors — so accepting a task the worker cannot start would leave the catalog claiming work is running while it sits parked behind another task's execution. The reservation travels with the task through the inbound channel and is consumed by drainInbound (Pool.Go), or handed back if the enqueue itself fails.
func (*Worker) WithInboundDispatch ¶
WithInboundDispatch enables the Phase 2 run-owner push path: the worker accepts dispatched tasks via SubmitDispatched and drains them onto the same execution pool as ClaimNext'd tasks. completionToken is the bearer token the worker presents when reporting a dispatched task's completion back to the owner's /internal/complete (the CAESIUM_INTERNAL_WAKEUP_TOKEN).
The buffer is sized to the pool size (floored at 1) because every buffered task already holds a reserved pool slot (SubmitDispatched reserves before it enqueues), so at most one pool's worth of tasks can ever be in the buffer. It is a hand-off queue between the HTTP goroutine and the Run loop, not a backlog: capacity — not buffer space — is what admits a dispatch. Without an explicit call this stays nil and the worker behaves byte-identically to Phase 1.
func (*Worker) WithLeaseRenewal ¶
func (w *Worker) WithLeaseRenewal(renewer LeaseRenewer, leaseTTL, renewInterval time.Duration) *Worker
WithLeaseRenewal configures per-node batched lease renewal.
- renewer issues a single UPDATE for all in-flight claims at once.
- leaseTTL is the configured claim TTL; it drives the renewal cadence and the skip-when-not-needed threshold.
- renewInterval is the override interval; pass 0 to use leaseTTL/4.
func (*Worker) WithReclaimGate ¶
func (w *Worker) WithReclaimGate(gate ReclaimGate) *Worker
func (*Worker) WithReclaimInterval ¶
func (*Worker) WithRunLeaseRenewal ¶
func (w *Worker) WithRunLeaseRenewal(renewer RunLeaseRenewer, leaseTTL time.Duration, nodeID string) *Worker
WithRunLeaseRenewal configures per-node batched run-lease renewal for Phase 2 run-owner mode. Piggybacked on the same ticker cadence as task claim renewals (leaseTTL/4). nodeID is the CAESIUM_NODE_ADDRESS value that identifies this node in run_leases.owner_node.
Source Files