Documentation
¶
Overview ¶
Package sensor implements the F69 Sensor (ADR-0109): the reusable event→action binder. It subscribes a Sensor's named event dependencies to the ADR-0108 in-process Fanout and, on a firing, runs the matching kind-keyed actions — start a WorkflowRun (`workflow:`) or invoke a Function (`function:`) — projecting the firing CloudEvent into the target's input via the F73 engine, and recording an Invocation per action. It owns only KindSensor (one-reconciler-per-gvk, ADR-0015). Stateless in V1, except that a Sensor's firings share its bounded delivery queue (ADR-0156).
Index ¶
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type Deps ¶
type Deps struct {
Store store.Store
Subscriber Subscriber
Invoker Invoker
// DeadLetters is the DLQ (ADR-0118). Optional: a nil store disables dead-lettering — the pre-ADR
// behavior stands (one delivery attempt, one Invocation, no retry loop).
DeadLetters deadletter.Store
// DeliveryAttempts is the bounded-retry cap before an action-delivery is dead-lettered (ADR-0118);
// 0 ⇒ the default (3). Ignored when DeadLetters is nil.
DeliveryAttempts int
// MaxDeliveriesInFlight (workers), MaxInFlightPerTarget and MaxQueuedPerSensor size the delivery queue
// (ADR-0156); a value < 1 ⇒ its default (32, 4, 4096). Ignored when DeadLetters is nil.
MaxDeliveriesInFlight int
MaxInFlightPerTarget int
MaxQueuedPerSensor int
// DeliveryBackoffInitial and DeliveryBackoffMax are eventing.deliveryBackoffInitial/Max (ADR-0163): the first
// delivery retry wait, doubled up to the max; 0 ⇒ 100ms and 10s.
DeliveryBackoffInitial time.Duration
DeliveryBackoffMax time.Duration
Logger *slog.Logger
}
Deps wires the Sensor reconciler (ADR-0002 §1 — every dep a port).
type HTTPInvoker ¶
type HTTPInvoker struct {
Endpoints activator.Endpoints
Waker Waker
Client *http.Client // nil: a client of the invoker's own, built on the first call
// contains filtered or unexported fields
}
HTTPInvoker invokes a Function with a CloudEvent, waking a cold target first (ADR-0033). This is the invoke/wake logic ADR-0108 REMOVED from the eventing Source, re-created here where the action lives — the Sensor's `function:` action. Built in pkg/funcd with the activator's Endpoints + Waker.
func (*HTTPInvoker) Invoke ¶
func (i *HTTPInvoker) Invoke(ctx context.Context, ns v1.NamespaceName, fn v1.ObjectName, ev eventing.CloudEvent) error
Invoke resolves fn's ready upstream (waking it if cold) and POSTs the CloudEvent. A transport/5xx/4xx failure is Unavailable, carrying the start of the function's answer; the caller records it on the Invocation (never silently dropped).
type Invoker ¶
type Invoker interface {
Invoke(ctx context.Context, ns v1.NamespaceName, fn v1.ObjectName, ev eventing.CloudEvent) error
}
Invoker delivers a CloudEvent to a Function's upstream, waking a cold one (ADR-0033). The concrete impl (HTTPInvoker, built in pkg/funcd) holds the activator's Endpoints + Waker; the reconciler only calls it.
type Reconciler ¶
type Reconciler struct {
// contains filtered or unexported fields
}
Reconciler owns KindSensor (ADR-0109).
func NewReconciler ¶
func NewReconciler(d Deps) (*Reconciler, error)
NewReconciler builds the Sensor reconciler. Store/Subscriber/Invoker are required; DeadLetters + DeliveryAttempts are optional (ADR-0118 — nil DeadLetters keeps the pre-ADR one-attempt behavior).
func (*Reconciler) Reconcile ¶
func (r *Reconciler) Reconcile(ctx context.Context, req controller.Request) (controller.Result, error)
Reconcile owns KindSensor: static-check the Sensor, (idempotently) subscribe its dependencies to the Fanout, and set Ready — or NotReady with a reason on a static defect; unsubscribe on delete.
func (*Reconciler) Replay ¶
func (r *Reconciler) Replay(ctx context.Context, ns v1.NamespaceName, id string) error
Replay performs ONE synchronous delivery attempt of a stored DeadLetter's CloudEvent through the LIVE Sensor's action path (ADR-0118 §3) — it does NOT re-enter the async bounded-retry loop. The replay is an action-delivery, so it records its Invocation (Ready or Failed). On success the entry is Deleted and nil returned; on failure the entry is re-parked with Attempts reset (never lost) and the delivery error returned, marked as re-parked once the re-park succeeded. The re-park only updates an entry that still exists, so a discard or a retention sweep made during the delivery stays in effect. A missing DeadLetter / Sensor / action ⇒ NotFound (the operator discards); a replay of an id already being replayed ⇒ Conflict. Idempotent: a repeat replay of a still-broken target re-parks with a fresh attempt count.
func (*Reconciler) RunRetryWorkers ¶
func (r *Reconciler) RunRetryWorkers(ctx context.Context, drain time.Duration)
RunRetryWorkers starts the delivery workers and blocks until ctx is cancelled. It then shuts the queue down, parks the deliveries still queued until the drain bound passes and drops the rest with one count log, and waits for the attempts in flight (ADR-0156 §7). An attempt still in flight at the bound is cancelled, so shutdown keeps its bound (ADR-0028). Wire it as a pkg/funcd background goroutine. A nil DeadLetters store means dead-lettering is off: delivery stays inline and there is no queue to run.
type Subscriber ¶
type Subscriber interface {
Subscribe(ns v1.NamespaceName, source, event v1.ObjectName, fn func(context.Context, eventing.CloudEvent)) (cancel func())
}
Subscriber is the subscribe seam the Sensor needs from the ADR-0108 Fanout — a single-method port (the concrete *eventing.Fanout satisfies it; a value type would copy its lock). The returned cancel deregisters.
Source Files