Catch Stale, Broken or Drifting Chainlink Price Feeds Before Your Protocol Uses Them

id: chainlink-price-feed-staleness-guard
namespace: company.team
description: |
  Watch the Chainlink-style price feeds your protocol reads, on any EVM chain.
  Alerts once per change when a feed is older than its heartbeat, returns a
  price of zero or below, carries an answer over from an earlier round, or
  drifts away from a second feed or a public price API. On a critical alert,
  on-call can pause your own consumer contract through a guardian signer.

triggers:
  - id: incident_webhook
    type: io.kestra.plugin.core.trigger.Webhook
    description: |
      Call this from your alerting when a consumer contract reverts on a price check, or when a feed's AnswerUpdated events stop
      (Tenderly, QuickNode Streams, Alertmanager), to diagnose every feed at once:
      curl -X POST http://localhost:8080/api/v1/main/executions/webhook/company.team/chainlink-price-feed-staleness-guard/price-feed-guard-change-me
    key: price-feed-guard-change-me
  - id: every_five_minutes
    type: io.kestra.plugin.core.trigger.Schedule
    description: The backstop. A stale feed emits no event, so only polling catches it.
    cron: "*/5 * * * *"

inputs:
  - id: feeds
    type: ARRAY
    itemType: STRING
    defaults:
      - ETH/USD=0x5f4eC3Df9cbd43714FE2740f5E3616155c5b8419=3600=coingecko:ethereum
      - BTC/USD=0xF4030086522a5bEEa4988F8cA5B36dbC97BeE88c=3600=coingecko:bitcoin
      - USDC/USD=0x8fFfFfd4AfB6115b954Bd326cbe7B4BA576818f6=86400=coingecko:usd-coin
    description: |
      Feeds as `Label=0xfeed=heartbeatSeconds[=reference]`. The reference is either another feed's address or `coingecko:<coin id>`
      (priced in the quote currency of the label). Take the heartbeat from data.chain.link; the defaults are Ethereum mainnet feeds.
  - id: grace_seconds
    type: INT
    defaults: 120
    description: Seconds allowed past the heartbeat before a feed counts as stale,
      for the update transaction to be mined.
  - id: deviation_warn_percent
    type: FLOAT
    defaults: 1
    description: Warn when a feed differs from its reference by this much. Keep it
      above the feed's own deviation threshold (0.5% for ETH/USD).
  - id: deviation_critical_percent
    type: FLOAT
    defaults: 3
    description: Critical deviation; on-call is asked whether to pause.
  - id: price_api_url
    type: STRING
    defaults: https://api.coingecko.com/api/v3/simple/price
    description: CoinGecko-compatible simple price endpoint, used for `coingecko:`
      references. Point it at a Pro or self-hosted mirror if you hit rate
      limits.
  - id: sequencer_uptime_feed
    type: STRING
    required: false
    description: On an L2 (Arbitrum, Optimism, Base, ...), the Chainlink L2
      sequencer uptime feed. While the sequencer is down, prices on that chain
      are frozen.
  - id: sequencer_grace_seconds
    type: INT
    defaults: 3600
    description: After the sequencer comes back, warn until this long has passed, as
      Chainlink recommends for consumers.
  - id: pause_target
    type: STRING
    required: false
    description: Your contract with `pause()` and `paused()`. When set, a critical
      alert asks on-call whether to pause it. Leave empty for alerts only.
  - id: guardian
    type: STRING
    required: false
    description: Address allowed to call `pause()` on `pause_target`. The signer
      behind EVM_SIGNER_URL signs for it.
  - id: remind_minutes
    type: INT
    defaults: 60
    description: Ask on-call again after this long while a feed stays critical.
  - id: explorer_url
    type: STRING
    defaults: https://etherscan.io
    description: Block explorer for links in Slack.
  - id: kestra_url
    type: STRING
    defaults: http://localhost:8080
    description: Base URL of your Kestra UI, used for the approval link.

variables:
  # Every input mistake, as one readable list (empty when the inputs are fine).
  input_problems: >-
    {{ {'feeds': inputs.feeds, 'pause': inputs.pause_target ?? '', 'guardian':
    inputs.guardian ?? '', 'sequencer': inputs.sequencer_uptime_feed ?? ''} |
    jq('[(.feeds[] |
    select(test("^[^=,]+=0x[0-9a-fA-F]{40}=[0-9]+(=(0x[0-9a-fA-F]{40}|coingecko:[a-z0-9-]+))?$")
    | not) | "not Label=0xfeed=heartbeat[=reference]: " + .),
           (.feeds | map(split("=") | .[1] // "" | ascii_downcase) | group_by(.) | map(select(length > 1) | .[0]) | .[] | "listed twice: " + .),
           (if (.pause + .guardian) != "" and ([.pause, .guardian] | map(test("^0x[0-9a-fA-F]{40}$")) | all | not) then "pause_target and guardian must both be addresses" else empty end),
           (if .sequencer != "" and (.sequencer | test("^0x[0-9a-fA-F]{40}$") | not) then "sequencer_uptime_feed is not an address" else empty end)]
           | join("; ")') | first }}
  # One JSON-RPC batch: chain id, latest block, and latestRoundData(), decimals(), description() of every feed and reference feed.
  read_batch: >-
    [{"jsonrpc": "2.0", "id": "chain", "method": "eth_chainId", "params": []},
    {"jsonrpc": "2.0", "id": "block", "method": "eth_getBlockByNumber",
    "params": ["latest", false]}{% for a in inputs.feeds | jq('[.[] | split("=")
    | (.[1], (.[3] // "" | select(startswith("0x"))))] | map(ascii_downcase) |
    unique') | first %}, {"jsonrpc": "2.0", "id": "round/{{ a }}", "method":
    "eth_call", "params": [{"to": "{{ a }}", "data": "0xfeaf968c"}, "latest"]},
    {"jsonrpc": "2.0", "id": "decimals/{{ a }}", "method": "eth_call", "params":
    [{"to": "{{ a }}", "data": "0x313ce567"}, "latest"]}, {"jsonrpc": "2.0",
    "id": "description/{{ a }}", "method": "eth_call", "params": [{"to": "{{ a
    }}", "data": "0x7284e416"}, "latest"]}{% endfor %}{% if
    (inputs.sequencer_uptime_feed ?? '') != '' %}, {"jsonrpc": "2.0", "id":
    "round/{{ inputs.sequencer_uptime_feed | lower }}", "method": "eth_call",
    "params": [{"to": "{{ inputs.sequencer_uptime_feed }}", "data":
    "0xfeaf968c"}, "latest"]}, {"jsonrpc": "2.0", "id": "decimals/{{
    inputs.sequencer_uptime_feed | lower }}", "method": "eth_call", "params":
    [{"to": "{{ inputs.sequencer_uptime_feed }}", "data": "0x313ce567"},
    "latest"]}{% endif %}{% if (inputs.pause_target ?? '') != '' %}, {"jsonrpc":
    "2.0", "id": "paused", "method": "eth_call", "params": [{"to": "{{
    inputs.pause_target }}", "data": "0x5c975abb"}, "latest"]}{% endif %}]
  # CoinGecko coin ids and quote currencies of the coingecko: references.
  coin_ids: "{{ inputs.feeds | jq('[.[] | split(\"=\") | .[3] // \"\" |
    select(startswith(\"coingecko:\")) | ltrimstr(\"coingecko:\")] | unique |
    join(\",\")') | first }}"
  quotes: "{{ inputs.feeds | jq('[.[] | split(\"=\") | select((.[3] // \"\") |
    startswith(\"coingecko:\")) | .[0] | split(\"/\") | .[1] // \"usd\" |
    split(\" \") | .[0] | ascii_downcase] | unique | join(\",\")') | first }}"

tasks:
  - id: validate_inputs
    type: io.kestra.plugin.core.flow.If
    description: Reject malformed entries before any RPC call, so a typo never turns
      into a silently unwatched feed.
    condition: "{{ render(vars.input_problems) | length > 0 }}"
    then:
      - id: invalid_inputs
        type: io.kestra.plugin.core.execution.Fail
        errorMessage: "Invalid inputs: {{ render(vars.input_problems) }}"

  - id: read_feeds
    type: io.kestra.plugin.core.http.Request
    uri: "{{ secret('EVM_RPC_URL') }}"
    method: POST
    contentType: application/json
    body: "{{ render(vars.read_batch) }}"
    retry:
      type: exponential
      interval: PT2S
      maxInterval: PT30S
      maxAttempts: 3

  - id: has_api_references
    type: io.kestra.plugin.core.flow.If
    condition: "{{ render(vars.coin_ids) | length > 0 }}"
    then:
      - id: fetch_reference_prices
        type: io.kestra.plugin.core.http.Request
        description: If the price API is down or rate-limited, deviation is reported as
          not checked; the on-chain checks still run.
        uri: "{{ inputs.price_api_url }}?ids={{ render(vars.coin_ids)
          }}&vs_currencies={{ render(vars.quotes) }}"
        allowFailure: true
        timeout: PT30S
        retry:
          type: exponential
          interval: PT2S
          maxInterval: PT10S
          maxAttempts: 3

  - id: evaluate
    type: io.kestra.plugin.scripts.python.Script
    description: |
      Decode the ABI results (int256 answers, uint80 round ids, the description string) and check each feed against the
      latest block's timestamp, which is the clock your consumer contract sees, rather than Kestra's clock.
    taskRunner:
      type: io.kestra.plugin.core.runner.Process
    inputFiles:
      rpc.json: "{{ outputs.read_feeds.body }}"
      prices.json: "{{ outputs.fetch_reference_prices.body ?? '' }}"
    env:
      FEEDS: "{{ inputs.feeds | toJson }}"
      GRACE: "{{ inputs.grace_seconds }}"
      DEVIATION_WARN: "{{ inputs.deviation_warn_percent }}"
      DEVIATION_CRITICAL: "{{ inputs.deviation_critical_percent }}"
      SEQUENCER: "{{ inputs.sequencer_uptime_feed ?? '' }}"
      SEQUENCER_GRACE: "{{ inputs.sequencer_grace_seconds }}"
    script: |
      import json, os, time

      rpc = json.load(open("rpc.json"))
      if not isinstance(rpc, list):
          raise SystemExit("The RPC endpoint rejected the batch: " + json.dumps(rpc)[:300])
      R = {x.get("id"): x for x in rpc}

      def result(key):
          x = R.get(key) or {}
          return x.get("result"), (x.get("error") or {}).get("message")

      def words(data):
          h = (data or "0x")[2:]
          return [int(h[i:i + 64], 16) for i in range(0, len(h) - 63, 64)]

      def signed(v):
          return v - (1 << 256) if v >= 1 << 255 else v

      def text(data):
          b = bytes.fromhex((data or "0x")[2:])
          return b[64:64 + int.from_bytes(b[32:64], "big")].decode("utf-8", "replace") if len(b) >= 64 else ""

      def ago(s):
          s = int(s)
          if s < 120: return f"{s}s"
          if s < 7200: return f"{s // 60}m"
          if s < 172800: return f"{s // 3600}h {s % 3600 // 60}m"
          return f"{s // 86400}d {s % 86400 // 3600}h"

      def num(x):
          return f"{x:.8f}".rstrip("0").rstrip(".")

      def norm(s):
          return "".join(ch for ch in s.upper() if ch.isalnum())

      chain, err = result("chain")
      block = result("block")[0]
      if err or not block:
          raise SystemExit(f"The RPC endpoint returned no chain id or block: {err or result('block')[1]}")
      chain_id, now, height = int(chain, 16), int(block["timestamp"], 16), int(block["number"], 16)

      def read_round(addr):
          """latestRoundData() and decimals() of a feed, or why they cannot be read."""
          res, err = result(f"round/{addr}")
          if err:
              return None, f"latestRoundData() reverted: {err}"
          w = words(res)
          if len(w) < 5:
              return None, "latestRoundData() returned no data: there is no feed at this address on this chain"
          dres, derr = result(f"decimals/{addr}")
          if derr or not words(dres):
              return None, f"decimals() failed: {derr or 'no data'}"
          dec = words(dres)[0]
          return {"round": w[0], "answer": signed(w[1]), "started": w[2], "updated": w[3], "answered_in": w[4],
                  "decimals": dec, "price": signed(w[1]) / 10 ** dec,
                  "description": text(result(f"description/{addr}")[0])}, None

      try:
          prices = json.load(open("prices.json"))
      except ValueError:
          prices = None
      grace = int(os.environ["GRACE"])
      warn_pct, crit_pct = float(os.environ["DEVIATION_WARN"]), float(os.environ["DEVIATION_CRITICAL"])
      rows = []

      def finish(row, problems):
          row["status"] = "critical" if any(p[0] == "critical" for p in problems) else "warn" if problems else "ok"
          row["problems"] = ",".join(sorted(f"{s}:{c}" for s, c, _ in problems))
          row["details"] = "; ".join(t for _, _, t in problems)
          rows.append(row)
          print(f"{row['status'].upper():8} {row['label']}: {row['details'] or row.get('summary', '')}")

      for entry in json.loads(os.environ["FEEDS"]):
          p = entry.split("=")
          label, addr, heartbeat = p[0].strip(), p[1].lower(), int(p[2])
          ref = p[3] if len(p) > 3 else ""
          row = {"label": label, "address": addr, "heartbeat": heartbeat, "reference": ref}
          problems = []
          r, why = read_round(addr)
          if r is None:
              problems.append(("critical", "unreadable", why))
              row["summary"] = "no price"
              finish(row, problems)
              continue
          age = now - r["updated"]
          row.update(price=num(r["price"]), decimals=r["decimals"], round_id=str(r["round"]), age_seconds=age,
                     updated_at=time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(r["updated"])))
          row["summary"] = f"price {num(r['price'])}, updated {ago(age)} ago (heartbeat {ago(heartbeat)})"
          if r["answer"] <= 0:
              problems.append(("critical", "invalid_answer", f"the answer is {num(r['price'])}; a price must be positive"))
          if r["updated"] == 0:
              problems.append(("critical", "incomplete_round", f"round {r['round']} has no update time"))
          elif age > heartbeat + grace:
              problems.append(("critical", "stale", f"stale: last updated {ago(age)} ago, heartbeat {ago(heartbeat)}"))
          if r["answered_in"] < r["round"]:
              problems.append(("critical", "carried_over",
                               f"answeredInRound {r['answered_in']} < roundId {r['round']}: the answer was carried over from an earlier round"))
          if r["description"] and norm(r["description"]) not in norm(label):
              problems.append(("warn", "wrong_feed", f"the contract describes itself as \"{r['description']}\", not {label}: check the address"))

          ref_price, ref_why = None, None
          if ref.startswith("0x"):
              rr, ref_why = read_round(ref.lower())
              if rr is not None:
                  if rr["answer"] <= 0:
                      ref_why = f"its answer is {num(rr['price'])}"
                  elif now - rr["updated"] > heartbeat + grace:
                      ref_why = f"it was last updated {ago(now - rr['updated'])} ago"
                  else:
                      ref_price = rr["price"]
          elif ref.startswith("coingecko:"):
              coin = ref.split(":", 1)[1]
              quote = label.split("/")[1].split()[0].lower() if "/" in label else "usd"
              value = ((prices or {}).get(coin) or {}).get(quote) if isinstance(prices, dict) else None
              if prices is None:
                  ref_why = "the price API did not answer"
              elif value is None:
                  ref_why = f"the price API has no {coin} price in {quote}"
              else:
                  ref_price = float(value)
          if ref_why:
              problems.append(("warn", "reference_unavailable", f"deviation not checked, reference {ref} unusable: {ref_why}"))
          elif ref_price and r["answer"] > 0:
              dev = abs(r["price"] - ref_price) / ref_price * 100
              row.update(reference_price=num(ref_price), deviation_percent=round(dev, 3))
              row["summary"] += f", {dev:.2f}% from the reference ({num(ref_price)})"
              if dev >= warn_pct:
                  problems.append(("critical" if dev >= crit_pct else "warn", "deviation",
                                   f"{dev:.2f}% away from the reference ({num(ref_price)})"))
          finish(row, problems)

      sequencer = os.environ["SEQUENCER"].lower()
      if sequencer:
          row, problems = {"label": "L2 sequencer uptime", "address": sequencer, "reference": ""}, []
          r, why = read_round(sequencer)
          if r is None:
              problems.append(("critical", "unreadable", why))
          else:
              since = now - r["started"]
              row["summary"] = f"sequencer {'down' if r['answer'] == 1 else 'up'} for {ago(since)}"
              if r["answer"] == 1:
                  problems.append(("critical", "sequencer_down", f"the sequencer has been down for {ago(since)}: prices on this chain are frozen"))
              elif since < int(os.environ["SEQUENCER_GRACE"]):
                  problems.append(("warn", "sequencer_grace", f"the sequencer came back {ago(since)} ago; do not trust prices for {ago(int(os.environ['SEQUENCER_GRACE']))}"))
          finish(row, problems)

      paused_raw = result("paused")[0]
      paused = bool(words(paused_raw)[0]) if words(paused_raw) else None
      print("::" + json.dumps({"outputs": {"feeds": rows, "chain_id": chain_id, "block": height, "consumer_paused": paused}}) + "::")

  - id: ensure_tables
    type: io.kestra.plugin.jdbc.postgresql.Queries
    url: "{{ secret('POSTGRES_URL') }}"
    username: "{{ secret('POSTGRES_USER') }}"
    password: "{{ secret('POSTGRES_PASSWORD') }}"
    sql: |
      CREATE TABLE IF NOT EXISTS price_feed_watch (
        chain_id    BIGINT NOT NULL,
        address     TEXT NOT NULL,
        label       TEXT NOT NULL,
        status      TEXT NOT NULL,
        problems    TEXT NOT NULL,
        since       TIMESTAMPTZ NOT NULL DEFAULT now(),
        asked_at    TIMESTAMPTZ,
        PRIMARY KEY (chain_id, address)
      );
      CREATE TABLE IF NOT EXISTS price_feed_watch_log (
        id            BIGSERIAL PRIMARY KEY,
        at            TIMESTAMPTZ NOT NULL DEFAULT now(),
        chain_id      BIGINT NOT NULL,
        address       TEXT NOT NULL,
        label         TEXT NOT NULL,
        status        TEXT NOT NULL,
        problems      TEXT NOT NULL,
        details       TEXT,
        execution_id  TEXT
      );

  - id: track
    type: io.kestra.plugin.jdbc.postgresql.Query
    description: |
      Turn statuses into changes. A feed is reported when its set of problems changes (and when it recovers), and a critical feed is
      proposed for a pause again every remind_minutes. The conditional UPDATE keeps overlapping runs from reporting the same change twice.
    url: "{{ secret('POSTGRES_URL') }}"
    username: "{{ secret('POSTGRES_USER') }}"
    password: "{{ secret('POSTGRES_PASSWORD') }}"
    fetchType: FETCH
    parameters:
      rows: "{{ outputs.evaluate.vars.feeds | toJson }}"
      chain: "{{ outputs.evaluate.vars.chain_id }}"
      remind: "{{ inputs.remind_minutes }}"
      can_pause: "{{ (inputs.pause_target ?? '') != '' }}"
      execution: "{{ execution.id }}"
    sql: |
      WITH cur AS (
        SELECT x->>'address' AS address, x->>'label' AS label, x->>'status' AS status, x->>'problems' AS problems,
               x->>'details' AS details, x->>'summary' AS summary,
               x->>'status' = 'critical' AND CAST(:can_pause AS boolean) AS actionable
        FROM jsonb_array_elements(CAST(:rows AS jsonb)) x
      ), ins AS (
        INSERT INTO price_feed_watch (chain_id, address, label, status, problems, asked_at)
        SELECT CAST(:chain AS bigint), address, label, status, problems, CASE WHEN actionable THEN now() END FROM cur
        ON CONFLICT (chain_id, address) DO NOTHING
        RETURNING address, status <> 'ok' AS changed, false AS recovered, asked_at IS NOT NULL AS propose, CAST(NULL AS TEXT) AS previous
      ), upd AS (
        UPDATE price_feed_watch w
        SET status = c.status, problems = c.problems, label = c.label,
            since = CASE WHEN w.problems <> c.problems THEN now() ELSE w.since END,
            asked_at = CASE WHEN NOT c.actionable THEN NULL
                            WHEN w.problems <> c.problems OR w.asked_at IS NULL
                                 OR w.asked_at < now() - make_interval(mins => CAST(:remind AS int)) THEN now()
                            ELSE w.asked_at END
        FROM cur c, price_feed_watch old
        WHERE w.chain_id = CAST(:chain AS bigint) AND w.address = c.address
          AND old.chain_id = w.chain_id AND old.address = w.address
          AND (w.problems <> c.problems OR w.label <> c.label
               OR (c.actionable AND (w.asked_at IS NULL OR w.asked_at < now() - make_interval(mins => CAST(:remind AS int))))
               OR (NOT c.actionable AND w.asked_at IS NOT NULL))
        RETURNING w.address, w.since = now() AND w.status <> 'ok' AS changed, w.since = now() AND w.status = 'ok' AS recovered,
                  w.asked_at = now() AS propose, old.problems AS previous
      ), changes AS (
        SELECT * FROM ins UNION ALL SELECT * FROM upd
      ), log AS (
        INSERT INTO price_feed_watch_log (chain_id, address, label, status, problems, details, execution_id)
        SELECT CAST(:chain AS bigint), c.address, c.label, c.status, c.problems, c.details, :execution
        FROM cur c JOIN changes ch USING (address)
        WHERE ch.changed OR ch.recovered
      )
      SELECT c.label, c.address, c.status, c.details, c.summary, coalesce(ch.changed, false) AS changed,
             coalesce(ch.recovered, false) AS recovered, coalesce(ch.propose, false) AS propose, ch.previous
      FROM cur c JOIN changes ch USING (address)
      WHERE ch.changed OR ch.recovered OR ch.propose

  - id: report
    type: io.kestra.plugin.core.flow.If
    condition: "{{ outputs.track.rows | jq('[.[] | select(.changed or .recovered)] |
      length') | first > 0 }}"
    then:
      - id: notify_changes
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        messageText: |
          Price feeds on chain {{ outputs.evaluate.vars.chain_id }}, block {{ outputs.evaluate.vars.block }}:
          {% for x in outputs.track.rows %}{% if x.changed %}- {{ x.status | upper }} {{ x.label }}: {{ x.details }}. Now: {{ x.summary }}. {{ inputs.explorer_url }}/address/{{ x.address }}
          {% elseif x.recovered %}- OK {{ x.label }} is healthy again (was {{ x.previous }}). {{ x.summary }}
          {% endif %}{% endfor %}{% if outputs.track.rows | jq('[.[] | select(.changed and .status == "critical")] | length') | first > 0 %}{% if (inputs.pause_target ?? '') == '' %}Consumers reading these feeds should refuse the price; no pause_target is configured, so no pause will be proposed.{% elseif outputs.evaluate.vars.consumer_paused ?? false %}{{ inputs.pause_target }} is already paused.{% endif %}{% endif %}

  - id: propose
    type: io.kestra.plugin.core.flow.If
    condition: "{{ outputs.track.rows | jq('[.[] | select(.propose)] | length') |
      first > 0 and not (outputs.evaluate.vars.consumer_paused ?? false) }}"
    then:
      - id: ask_on_call
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        messageText: |
          Critical price feeds: {{ outputs.track.rows | jq('[.[] | select(.propose) | "\(.label) (\(.details))"] | join(", ")') | first }}
          Pause {{ inputs.pause_target }} from the guardian {{ inputs.guardian }}? Decide within 1 hour: {{ inputs.kestra_url }}/ui/main/executions/{{ flow.namespace }}/{{ flow.id }}/{{ execution.id }}
          No answer means no pause; you will be asked again in {{ inputs.remind_minutes }} min while a feed stays critical.

      - id: wait_for_decision
        type: io.kestra.plugin.core.flow.Pause
        description: No answer within 1 hour means no pause.
        pauseDuration: PT1H
        behavior: RESUME
        onResume:
          - id: decision
            type: SELECT
            values:
              - Pause the consumer
              - Not now
            defaults: Not now
          - id: note
            type: STRING
            required: false
            description: Optional reason, posted to Slack with the outcome.

      - id: pause_consumer
        type: io.kestra.plugin.core.flow.If
        condition: "{{ outputs.wait_for_decision.onResume.decision == 'Pause the
          consumer' }}"
        then:
          - id: check_not_paused
            type: io.kestra.plugin.core.http.Request
            description: Someone may have paused it while the question waited; a second
              pause() would revert.
            uri: "{{ secret('EVM_RPC_URL') }}"
            method: POST
            contentType: application/json
            body: "{{ {'jsonrpc': '2.0', 'id': 1, 'method': 'eth_call', 'params': [{'to':
              inputs.pause_target, 'data': '0x5c975abb'}, 'latest']} | toJson
              }}"
            retry:
              type: exponential
              interval: PT2S
              maxInterval: PT30S
              maxAttempts: 3

          - id: already_paused
            type: io.kestra.plugin.core.flow.If
            condition: "{{ outputs.check_not_paused.body | jq('.result // \"\" |
              test(\"^0x0*1$\")') | first }}"
            then:
              - id: notify_already_paused
                type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
                url: "{{ secret('SLACK_WEBHOOK_URL') }}"
                messageText: "{{ inputs.pause_target }} was already paused when the answer came;
                  nothing was sent."
            else:
              - id: send_pause
                type: io.kestra.plugin.core.http.Request
                uri: "{{ secret('EVM_SIGNER_URL') }}"
                method: POST
                contentType: application/json
                options:
                  allowFailed: true
                body: "{{ {'jsonrpc': '2.0', 'id': 1, 'method': 'eth_sendTransaction', 'params':
                  [{'from': inputs.guardian, 'to': inputs.pause_target, 'data':
                  '0x8456cb59'}]} | toJson }}"

              - id: pause_sent
                type: io.kestra.plugin.core.flow.If
                condition: "{{ (outputs.send_pause.body | jq('.error.message // \"\"') | first)
                  != '' }}"
                then:
                  - id: pause_refused
                    type: io.kestra.plugin.core.execution.Fail
                    errorMessage: "The signer refused the pause transaction: {{
                      outputs.send_pause.body | jq('.error.message') | first }}.
                      The consumer is NOT paused"

              - id: wait_for_receipt
                type: io.kestra.plugin.core.flow.LoopUntil
                condition: "{{ outputs.pause_receipt.body | jq('.result != null') | first }}"
                checkFrequency:
                  interval: PT3S
                  maxIterations: 100
                failOnMaxReached: true
                tasks:
                  - id: pause_receipt
                    type: io.kestra.plugin.core.http.Request
                    uri: "{{ secret('EVM_RPC_URL') }}"
                    method: POST
                    contentType: application/json
                    body: "{{ {'jsonrpc': '2.0', 'id': 1, 'method': 'eth_getTransactionReceipt',
                      'params': [outputs.send_pause.body | jq('.result') |
                      first]} | toJson }}"

              - id: check_paused
                type: io.kestra.plugin.core.http.Request
                description: Read paused() so the result comes from the contract, not from the
                  transaction alone.
                uri: "{{ secret('EVM_RPC_URL') }}"
                method: POST
                contentType: application/json
                body: "{{ {'jsonrpc': '2.0', 'id': 1, 'method': 'eth_call', 'params': [{'to':
                  inputs.pause_target, 'data': '0x5c975abb'}, 'latest']} |
                  toJson }}"

              - id: pause_worked
                type: io.kestra.plugin.core.flow.If
                condition: "{{ (outputs.pause_receipt.body | jq('.result.status') | first) !=
                  '0x1' or not (outputs.check_paused.body | jq('.result // \"\"
                  | test(\"^0x0*1$\")') | first) }}"
                then:
                  - id: pause_failed
                    type: io.kestra.plugin.core.execution.Fail
                    errorMessage: "Pause transaction {{ outputs.send_pause.body | jq('.result') |
                      first }} {{ (outputs.pause_receipt.body |
                      jq('.result.status') | first) == '0x1' ? 'was mined but
                      paused() is not true' : 'reverted: check that the guardian
                      may call pause()' }}. The consumer is NOT paused"

              - id: notify_paused
                type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
                url: "{{ secret('SLACK_WEBHOOK_URL') }}"
                messageText: |
                  PAUSED: {{ inputs.pause_target }} is paused (paused() = true) by {{ inputs.explorer_url }}/tx/{{ outputs.send_pause.body | jq('.result') | first }}{% if (outputs.wait_for_decision.onResume.note ?? '') != '' %} ("{{ outputs.wait_for_decision.onResume.note }}"){% endif %}
                  Unpause it once the feeds are healthy again; this flow reports when they are.

errors:
  - id: alert_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    messageText: >-
      Price feed guard failed in execution {{ execution.id }} {% if errorLogs()
      | length > 0 %}at `{{ errorLogs()[0]['taskId'] }}`: {{
      errorLogs()[0]['message'] | split(' \[\[') | first }}{% else %}without an
      error message{% endif %}. {% if errorLogs() | length > 0 and
      ['check_not_paused', 'send_pause', 'pause_refused', 'pause_receipt',
      'wait_for_receipt', 'check_paused', 'pause_failed'] contains
      errorLogs()[0]['taskId'] %}Check the consumer contract by hand.{% else
      %}The feeds were NOT checked in this run.{% endif %} {{ inputs.kestra_url
      }}/ui/main/executions/{{ flow.namespace }}/{{ flow.id }}/{{ execution.id
      }}

outputs:
  - id: feeds
    type: JSON
    description: Every feed with its price, age, reference price, deviation, status
      and problems.
    value: "{{ outputs.evaluate.vars.feeds ?? [] }}"