Event-driven triggers (faucet serve)

faucet serve --triggers <file> loads a static triggers file at startup and spawns long-lived watcher tasks. When a watcher fires, it enqueues a run through the same runner::submit pipeline as a normal POST /v1/runs, inheriting the full queue/semaphore/idempotency/history machinery.

Requires the triggers Cargo feature (included in full):

cargo install faucet-cli --features triggers                   # framework + webhook only
cargo install faucet-cli --features "triggers,triggers-object-store"   # + S3/GCS
cargo install faucet-cli --features "triggers,triggers-redis"          # + Redis queue-depth
cargo install faucet-cli --features full                               # everything

Triggers file grammar

version: 1          # required; must be 1
triggers:
  - name: <string>          # unique; used in metrics, idempotency keys, webhook path
    enabled: true           # optional; default true — set false to disable without deleting
    config: <path|inline>   # pipeline config: a file path string OR an inline pipeline doc
    template:               # …or a registered template instead of `config` (exactly one)
      id: <template id>
      version: stable       # optional: a number or a channel
      params: {}            # optional; string values may use ${trigger.*}
      sink: <sink template> # optional: for a source template
      overlay: <overlay id> # optional deployment overlay
      allow_untested: false # optional: run a pinned version whose tests are not passing (#856)
    tenants: all            # optional: `all` or [tenant ids] — one run per tenant
    run:                    # optional run-shaping
      name: <template>      # run name; supports {name}, {object_key}, {bucket}, etc.
      labels: {}            # static labels merged with the auto-derived trigger labels
      timeout_secs: null    # per-run timeout in seconds
      selection: null       # optional: run only some rows / streams (see below)
      pool: null            # optional: execution pool for the fired runs (#957); wins over the config's `pool:`
    type: <trigger type>    # required; one of object_arrival, webhook, queue_depth
    # … type-specific fields below

The config: field accepts either a path string (resolved relative to the triggers file, not the process CWD) or an inline pipeline document ({ pipeline: … }). template: runs a registered template instead (needs the templates feature). A trigger pinned to a version that holds no label and whose tests are not passing needs allow_untested: true; each such run is labelled and audited, the server logs a warning naming the trigger when the file loads, and allow_untested together with tenants: is refused (see Runs follow labels). tenants: fans each fire out to one run per tenant, each run as that tenant — its connections, ${tenant.*} values, state namespace and limits (embedded integrations; needs the tenants feature). Each tenant’s idempotency key is the trigger’s key suffixed :<tenant>; a tenant that is suspended, at its limit, or missing a connection is skipped and logged. A tenant whose run is refused for a reason a retry cannot fix (a missing ${tenant.labels.*} label, a config that fails validation) is reported for that tenant only — a WARN log, faucet_serve_trigger_tenant_failures_total and a run_failure event through the tenant’s own notifications: — and the fire still counts for every other tenant. Only a transient failure (the run history unavailable, a full queue) retries the whole fire.

The triggers file is validated strictly at load time: an unknown or misspelled field on a trigger entry (e.g. debounce_sec for debounce_secs) or inside its nested store: / queue: block fails fast with an error naming the field and the trigger, rather than being silently dropped. Keys inside an inline config: pipeline document are validated by the pipeline loader, not here.

Trigger types

object_arrival

Polls an object store (S3 or GCS) for new objects under a prefix.

Requires the triggers-object-store Cargo feature.

type: object_arrival
store:
  type: s3                  # s3 | gcs
  bucket: my-bucket         # required
  prefix: incoming/         # key prefix to watch (optional; defaults to root)
  region: us-east-1         # S3 only (optional)
  endpoint: null            # S3 only — override endpoint URL for S3-compatible stores
poll_interval_secs: 30      # how often to list the prefix (default 30)
mode: per_object            # per_object (one run per new object) | batch (one run for all new objects)
start_at: now               # now (only objects after the first startup) | beginning (all objects, incl. existing)
lookback_secs: 86400        # how far behind the newest fired object a late-listed one still fires (default 24h)
poll_timeout_secs: null     # optional limit on one whole poll (default: none — see below)

An object is fired once, identified by its key, ETag and last-modified time (a rewritten object fires again). Object stores list an object only once it is complete, stamped with when it started — a multipart upload — or an earlier write time, so it can appear after newer objects already fired; any object last modified within lookback_secs of the newest fired one is still fired. A late batch gets its own idempotency key, so it is not taken for the earlier batch that shares its newest time.

Every poll lists the whole prefix — object stores have no “changed since” listing — so its request cost grows with the number of objects under it; only objects not yet fired are kept in memory. Point the trigger at a landing prefix that is cleaned up or archived, and size poll_interval_secs to its listing cost.

A large prefix may take minutes to list; that is not cut off. Instead each wait for the next listing page is bounded by max(poll_interval_secs, 60) seconds: a store that stops answering fails the poll (the watcher backs off and, after three failures in a row, reports unhealthy) rather than freezing it. Set poll_timeout_secs to also bound the whole poll — listing plus firing.

A key the object-store client cannot read as a path (an empty, . or .. segment, or a control character, such as in//x.csv) does not stop the prefix: the listing narrows to that key, skips it with a warning naming it (counted in faucet_serve_trigger_errors_total), and resumes right after it, so every other object still fires.

${trigger.*} tokens injected into the run config:

Tokens are substituted inside each string value of the parsed config, with the raw value: bucket: "${trigger.bucket}", prefix: in/${trigger.object_key} and bucket: ${trigger.bucket} all yield a plain string, and a value holding :, quotes or newlines stays inside its scalar. A whole-value token is always a string (size: ${trigger.size} gives "42"). Mapping keys are not substituted.

A token spliced into a longer string is plain text of that string — never a bound SQL parameter or an escaped path. A value the event’s sender chooses (a webhook body, query value or header, an object key) is checked against where it lands:

• spliced into a longer string: no .. path segment and no control character; as a whole value: no .. path segment when the field names a path or URL (a body used whole as data may hold one);
• spliced into a SQL statement (a field the connector’s schema marks as SQL, such as a SQL source’s query) or into free text (a string with spaces or quotes around the token): no quote (', ", `), ; or backslash. A path, URL or name (in/${trigger.object_key}, s3://bucket/${trigger.object_key}) may hold them;
• spliced unquoted into SQL text (WHERE id = ${trigger.query.id}): a number only (42, -4.5), the rule ${param.*} follows. Quote the token (WHERE id = '${trigger.query.id}') to splice text;
• spliced into an http:// / https:// URL: no ?, #, &, whitespace or .. segment, so a value cannot add a query parameter or a fragment;
• inside a URL’s host (https://${trigger.header.x-region}.api.example): host name characters only (letters, digits, ., -);
• values next to each other, or next to text the config writes, may not combine into an interpolation directive (${env:…}, ${file:…}): a value $ beside a value {env:HOME} is refused;
• as the whole of a SQL field: refused — a sender never supplies a whole statement. A GraphQL query or a search or document-store filter is not SQL and takes a whole value. Outside any connector or transform the trigger can name, a whole query, sql, statement, where or filter value is refused;
• a URL, or an absolute path that starts the value, must be one the server’s caller reference policy allows (--allow-url-host, --allow-url-scheme, --allow-path-root, and never the system trees or the server’s credential files).

The trigger’s config is the operator’s and loads as the server’s own, but every value that holds sender text is then held to the caller reference policy where it lands, read with that field’s grammar: a brokers list, an ADO.NET or Azure Storage connection string, an annotated option key, an absolute URL, an auth: provider endpoint. A scalar that holds any sender text is judged whole, operator text included, because a fragment can change what the rest means (a later key= in a connection string wins). A sender value placed in a params: default is followed to every ${param.NAME} that uses it. A refusal names the field and the token (${trigger.header.x-brokers}), shows only the sender’s text, and is audited as policy.denied on trigger values. Allow fixed hosts with --allow-url-host. A bare host:port in a field no schema marks as a host is not read as one, as for a caller’s own config.

A template: trigger’s params are checked the same way. Where the template places a param is not known when the trigger fires, so a sender value in a param is judged by the param’s name and the value itself: no .. segment in a value that reads as a path, no whole value for a param named query, sql, statement, where or filter, no quote, ; or backslash when spliced into free text, and a URL or a leading absolute path must pass the caller reference policy. Once the template is bound, a param that holds sender text is checked where the template places it, as a config trigger’s value is.

A refused fire never runs: a webhook answers 422 (the trigger stays healthy), and an object_arrival watcher skips the object and reports it, then fires the objects after it. A webhook also answers 422 when the config uses a header or query value the request did not send.

A webhook trigger that cannot run whatever is sent — its config does not validate, its template or tenant is gone, a policy refuses it — answers 500 and, after three such fires in a row, reports unhealthy on /readyz: the operator has something to fix, not the sender.

mode: per_object — one token set per object:

Token	Value
${trigger.name}	The trigger’s name field
${trigger.type}	object_arrival
${trigger.fired_at}	ISO 8601 timestamp when the trigger fired
${trigger.object_key}	The S3/GCS object key
${trigger.bucket}	The S3/GCS bucket name
${trigger.size}	Object size in bytes
${trigger.last_modified}	RFC 3339 last-modified timestamp of the object

mode: batch — one token set for the entire batch of new objects:

Token	Value
${trigger.name}	The trigger’s name field
${trigger.type}	object_arrival
${trigger.fired_at}	ISO 8601 timestamp when the trigger fired
${trigger.bucket}	The S3/GCS bucket name
${trigger.object_count}	Number of new objects in the batch

Note

${trigger.object_key}, ${trigger.size}, and ${trigger.last_modified} are not available in mode: batch (they are per-object fields).

Idempotency key:

• mode: per_object: trig:<name>:<bucket>:<object_key>:<last_modified> — deterministic per object version; re-listing a processed object does not enqueue a duplicate run.
• mode: batch: trig:<name>:<watermark> where <watermark> is the maximum last_modified timestamp across the batch.

start_at: now behaviour: on its first startup the watcher ignores every object last modified before that moment; only objects after it are treated as new. Set start_at: beginning to fire for all objects currently in the prefix (use mode: batch to coalesce them into one run).

The cursor survives a restart. The watcher keeps its cursor (the start point and the objects it fired within the look-back window) in the run-history settings store, keyed by trigger name, and resumes from it after a restart: an object that landed while the server was down is fired, and an object already fired is not fired again. start_at applies only when no cursor is stored (or the trigger’s bucket or prefix changed). With the in-memory history the cursor lasts as long as the process.

A fire that cannot start: when an object’s run is refused for a reason a retry cannot fix (its substituted config does not load, a policy or validation refusal), the object is skipped — the cursor moves past it, an ERROR log names it — and the objects after it still fire. A transient failure (the queue is full, the history store is unreachable) stops the poll and retries the same object next time. Either way the poll counts as failed: the watcher backs off, faucet_serve_trigger_errors_total counts it, and after repeated failures the watcher reports healthy: false (with consecutive_failures); the reason is in the server log (trigger poll failed) — /readyz is unauthenticated, so it carries no error text.

webhook

Exposes POST /v1/triggers/{name} on the faucet serve listener. The endpoint is bearer-authenticated (same token as /v1/runs). Returns 202 on success, 404 for an unknown trigger name, and 400 when the HTTP method is not in the configured methods list.

No additional Cargo features are required (the route is part of the base triggers feature, which implies serve).

type: webhook
methods: [POST]             # allowed HTTP methods (default [POST]); PUT also supported
dedupe_header: null         # header used as idempotency key (optional; else a per-request UUID)
debounce_secs: 0            # leading-edge debounce window in seconds (default 0 = off)
content_types: [application/json, application/*+json]  # accepted request Content-Types (default: JSON)

content_types

the media types (parameters such as charset are ignored) a request body may be sent as; anything else gets 415. The default accepts application/json and the structured JSON types (application/*+json, e.g. application/cloudevents+json). type/* accepts any subtype, type/*+suffix any subtype ending in +suffix, and * anything. A request without a Content-Type is accepted only when its body is empty. A form-encoded sender, or one that posts a body without a Content-Type, is refused until its type is listed — e.g. content_types: [application/json, application/x-www-form-urlencoded] (or * for a sender that sends no Content-Type at all).

Leading-edge debounce

when debounce_secs > 0, the first request is accepted and any further requests that arrive within debounce_secs of that accepted fire are coalesced — they return 200 { "status": "coalesced" } and enqueue no run. A request that arrives while the accepted fire is still being submitted gets 503 and should retry: that fire may not go through, and if it fails (or its request is abandoned) the window is undone so the retry fires. The window re-arms once debounce_secs have fully elapsed since the last accepted fire. Debounce is webhook-only; polling triggers (object_arrival, queue_depth) pace themselves via poll_interval_secs.

dedupe_header trust boundary

the caller-supplied header value becomes the run’s idempotency key (namespaced by the trigger name). A caller who controls this value can suppress a legitimate run by reusing a key from a prior run. Only set dedupe_header when callers are trusted or the header value is verified upstream (e.g. by a gateway signing scheme or HMAC validation).

Disallowed methods

a request whose HTTP method is not in methods returns 400 in v1 (not 405). This is intentional: the route itself exists for all methods; the 400 carries a descriptive message.

${trigger.*} tokens:

Token	Value
${trigger.name}	The trigger’s name field
${trigger.type}	webhook
${trigger.fired_at}	ISO 8601 timestamp when the trigger fired
${trigger.method}	HTTP method of the request (POST, PUT, …)
${trigger.body}	Raw request body (string)
${trigger.header.<name>}	Value of HTTP request header <name>
${trigger.query.<name>}	Value of query parameter <name>

Idempotency key: trig:<name>:<value>, where <value> is the dedupe_header value when configured and present in the request, otherwise a fresh per-request UUID. The trigger name keeps keys apart: one event id delivered to two webhook triggers runs both pipelines, and never collides with a key a caller used on POST /v1/runs.

Fire the webhook with curl:

curl -XPOST http://127.0.0.1:8080/v1/triggers/sync-hook \
     -H "Authorization: Bearer s3cret" \
     -H "Idempotency-Key: run-20260612-001" \
     -H "Content-Type: application/json" \
     -d '{}'

queue_depth

Polls a Redis list/stream or a Kafka consumer group lag metric. When the observed depth crosses threshold, the watcher fires once (edge-triggered). It will not fire again until the depth drops below the threshold and rises back.

type: queue_depth
queue:
  type: redis               # redis | kafka
  # Redis fields:
  url: redis://localhost:6379
  key: jobs                 # list key or stream name
  kind: list                # list | stream (default list)
  # Kafka fields:
  # brokers: localhost:9092
  # topic: events
  # group: my-consumer-group
threshold: 1                # fire when depth >= threshold (default 1)
poll_interval_secs: 30      # polling interval (default 30)
poll_timeout_secs: 60       # a poll taking longer counts as failed (default max(poll_interval_secs, 60))

Redis requires the triggers-redis feature; Kafka requires triggers-kafka.

${trigger.*} tokens:

Token	Value
${trigger.name}	The trigger’s name field
${trigger.type}	queue_depth
${trigger.fired_at}	ISO 8601 timestamp when the trigger fired
${trigger.queue}	The queue key / topic name
${trigger.depth}	Observed depth (as a string) that crossed the threshold

Idempotency key: trig:<name>:edge:<edge_ordinal>. The edge — whether it is armed, and its ordinal — is stored in the run history, not in the watcher: a rising crossing disarms it and takes the next ordinal, a drain below the threshold re-arms it. So a restart never reuses an ordinal (the next crossing always gets a new key) and, on a shared SQL history, exactly one instance fires each crossing. A fire that is dropped or fails gives its ordinal back so the next poll retries the same crossing. A crossing whose fire was never confirmed — the process stopped between disarming the edge and firing, or giving the ordinal back failed — is fired again under the same ordinal while the depth stays above the threshold; the stable key makes a fire that did start a run a replay. After the idempotency window has passed, that re-fire starts a second run for the crossing. With the in-memory history the edge is per-process, like the idempotency claims it pairs with.

Schedules moved

Cron schedules are their own concept: see Schedules (faucet serve --schedules, /v1/schedules, the console’s Schedules page). A type: schedule entry in a triggers file fails to load with a message naming the entry and pointing there.

Labels on enqueued runs

Every trigger-fired run receives these automatic labels (visible in GET /v1/runs responses and Prometheus metrics):

Label	Value
faucet.trigger.name	Trigger name
faucet.trigger.type	Trigger type (object_arrival, webhook, queue_depth)

Additional labels can be added per trigger via run.labels:.

Running a subset of rows (run.selection)

run.selection runs only some of the pipeline’s matrix rows (or a template’s streams) on every fire — the same object POST /v1/runs takes (select, only, skip, tags, status, include_parents). A schedule takes the same run.selection, so the cheap streams can run hourly and the heavy one nightly:

triggers:
  - name: crm-landed
    type: object_arrival
    store: { type: s3, bucket: exports, prefix: crm/ }
    template: { id: acme/crm, sink: faucet-hq/bigquery }
    run:
      selection: { skip: [activities] }

The selection is validated when the trigger fires (an unknown row fails that fire and counts in faucet_serve_trigger_errors_total), labels the run selection=<canonical>, and applies to every tenant of a tenants: fan-out. See Running a subset of streams.

/readyz — trigger health

GET /readyz includes a triggers array when --triggers is active:

{
  "status": "ready",
  "history_ok": true,
  "queue_ok": true,
  "cluster": { "enabled": false, "instances": 0 },
  "triggers": [
    { "name": "load-dropped-files", "healthy": true },
    { "name": "sync-hook",          "healthy": true },
    { "name": "drain-jobs",         "healthy": true }
  ]
}

A watcher whose client cannot be built at startup (an object-store or queue client error) is reported healthy: false straight away, with the reason in the server log — it never polls. A degraded watcher (crashed and backing off) sets its healthy flag to false but does not flip the top-level status to not_ready — the server keeps accepting runs from the other trigger paths.

Schema introspection

faucet schema triggers     # print the JSON Schema for the triggers file

Metrics

Metric	Type	Labels	Description
faucet_serve_triggers_active	Gauge	—	Number of enabled, running trigger watchers
faucet_serve_trigger_healthy	Gauge	trigger	1 = healthy, 0 = in error backoff
faucet_serve_trigger_last_fire_unix_seconds	Gauge	trigger	Unix timestamp of last fire
faucet_serve_triggers_fired_total	Counter	trigger, type	Total trigger fires
faucet_serve_trigger_runs_enqueued_total	Counter	trigger	Runs successfully enqueued
faucet_serve_trigger_runs_coalesced_total	Counter	trigger	Fires coalesced — webhook debounce, or an idempotency-conflict no-op
faucet_serve_trigger_runs_dropped_total	Counter	trigger, reason	Fires dropped because the run queue was full (reason="queue_full")
faucet_serve_trigger_errors_total	Counter	trigger, type	Watcher errors (poll failures, etc.)
faucet_serve_trigger_tenant_failures_total	Counter	trigger, tenant	Fan-out fires refused for one tenant, or skipped at its max_concurrent_runs limit, while the other tenants ran

Cluster note

When running a cluster (--cluster + shared --history DB), every instance loads the same --triggers file and spawns independent watchers. Idempotency keys are deterministic (derived from object key + last_modified, the dedupe header value, or a rising-edge ordinal stored in the shared history), so concurrent fires from multiple instances resolve to a single run via the shared idempotency claim. No additional coordination is required.

Feature flags

Feature	Contents
triggers	Framework, supervisor, webhook trigger (implies serve)
triggers-object-store	object_arrival watcher (S3/GCS listing)
triggers-redis	queue_depth watcher backed by Redis
triggers-kafka	queue_depth watcher backed by Kafka consumer-group lag
templates (with triggers)	template: on a trigger
tenants (with triggers)	tenants: on a trigger

All four are included in full and none are in default.

See also

• Event-driven triggers — the task-oriented recipe for wiring up file-arrival, webhook, and queue-depth triggers end to end.