Documentation
¶
Index ¶
Constants ¶
const DefaultReconcileInterval = 60 * time.Second
DefaultReconcileInterval is the default interval between reconcile-loop passes (spec 005), overridable per deployment via RECONCILE_INTERVAL. 60s is a reasonable recovery latency for a task whose deferral was lost across a restart.
const DefaultReconcilePassTimeout = 45 * time.Second
DefaultReconcilePassTimeout bounds one reconcile pass so a slow or large vault cannot hold a pass open indefinitely (spec 005 Failure Modes: "Pass bounded by a per-pass timeout; logs evaluate count so cost is visible"). Three quarters of the interval: a pass completes within a tick and leaves headroom for the next one. Raised from 30s on 2026-10-19 — a pass over the openclaw vault aborted on `context_cancelled` every minute after evaluating only 8-73 of ~200 non-terminal tasks, so the tasks ordered after the cut were never reached. The per-assignee active-Job cache added in the same change removes most of the cost; this is the headroom that keeps a slow pass from re-opening the same window. Overridable per deployment via RECONCILE_PASS_TIMEOUT: on 2026-10-21 a sun executor reading the 15k-file openclaw vault aborted every pass at 45s with evaluated=0.
Variables ¶
This section is empty.
Functions ¶
func IsTerminal ¶
IsTerminal reports whether the given phase is in the terminal set. Tasks at a terminal phase must not be re-spawned; operator intervention is required.
func NewAgentsHandler ¶
NewAgentsHandler returns an http.Handler that lists all known agent configs from the in-memory store as JSON. If authSecret is non-empty, requests must include X-Agent-Auth header with the secret.
func NewHealthcheckTriggerHandler ¶
func NewHealthcheckTriggerHandler( runner probe.HealthcheckRunner, ) http.Handler
NewHealthcheckTriggerHandler returns an HTTP handler that fires the healthcheck runner once per invocation with fire-and-forget + single-flight semantics. Concurrent invocations collapse into one in-flight run (second request is silently dropped).
The handler uses context.Background() for its internal operations so it is not affected by the caller's context cancellation. The handler only stops when the HTTP server shuts down.
Types ¶
type HighwaterMarksFunc ¶ added in v0.25.0
type HighwaterMarksFunc func(ctx context.Context, topic libkafka.Topic) (libkafka.PartitionOffsets, error)
HighwaterMarksFunc returns the high-water mark of every partition of topic.
type TaskEventHandler ¶
type TaskEventHandler interface {
ConsumeMessage(ctx context.Context, msg *sarama.ConsumerMessage) error
// EvalDeferredRespawns evaluates all pending deferred-respawn entries immediately.
// Called by RunDeferredRespawnLoop on each tick; also callable directly in tests.
EvalDeferredRespawns(ctx context.Context) error
// RunDeferredRespawnLoop polls evalDeferredRespawns every deferredRespawnInterval
// until ctx is cancelled. Must be run alongside the Kafka consumer.
RunDeferredRespawnLoop(ctx context.Context) error
// ReconcileOnce runs one reconcile pass: it snapshots the in-memory task
// view, applies the eligibility filter, and re-drives every eligible owned
// task through the existing spawn path. Called by RunReconcileLoop on each
// tick; also callable directly in tests.
ReconcileOnce(ctx context.Context) error
// RunReconcileLoop runs one pass immediately at startup, then every
// reconcile interval until ctx is cancelled. Must be run
// alongside the Kafka consumer.
RunReconcileLoop(ctx context.Context) error
}
TaskEventHandler processes task event messages from Kafka and manages deferred respawns.
func NewTaskEventHandler ¶
func NewTaskEventHandler( jobSpawner spawner.JobSpawner, branch base.Branch, cluster string, stageFilter bool, runUntagged bool, resolver pkg.ConfigResolver, resultPublisher pkg.ResultPublisher, taskStore *pkg.TaskStore, currentDateTime libtime.CurrentDateTimeGetter, view TaskView, taskVault string, reconcilePassTimeout time.Duration, reconcileInterval time.Duration, ) TaskEventHandler
NewTaskEventHandler creates a new TaskEventHandler.
type TaskView ¶ added in v0.25.0
type TaskView interface {
// CaptureHighWaterMarks records the topic's per-partition high-water marks at
// start. It must be called once, before the replay consumer starts.
CaptureHighWaterMarks(ctx context.Context) error
// ConsumeMessage applies one task event. It is the MessageHandler the replay
// consumer is built with.
ConsumeMessage(ctx context.Context, msg *sarama.ConsumerMessage) error
// Ready reports whether every partition reached the mark captured at start.
Ready() bool
// Snapshot returns a point-in-time copy of every task in the view.
Snapshot() []lib.Task
// Get returns the task for id, or false when the view has none.
Get(id lib.TaskIdentifier) (lib.Task, bool)
}
TaskView is an in-memory, latest-wins view of the compacted task event topic.
func NewTaskView ¶ added in v0.25.0
func NewTaskView( highwaterMarks HighwaterMarksFunc, topic libkafka.Topic, currentDateTimeGetter libtime.CurrentDateTimeGetter, ) TaskView
NewTaskView creates an empty view for topic.
Source Files