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: every firing is independent.
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
Logger *slog.Logger
}
Deps wires the Sensor reconciler (ADR-0002 §1 — every dep a port).
type HTTPInvoker ¶
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-Put with Attempts reset (never lost) and the delivery error returned, marked as re-parked once the re-Put succeeded. A missing DeadLetter / Sensor / action ⇒ NotFound (the operator discards). 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)
RunRetryWorkers starts the delivery retry workers and blocks until ctx is cancelled, then shuts down the queue and waits for in-flight attempts to drain (a graceful drain narrows — does not close — the in-memory-retry crash window; ADR-0118 Temporary workarounds). Wire it as a pkg/funcd background goroutine. A nil DeadLetters store means dead-lettering is off, so there is no retry loop 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