PostgreSQL Logical Subscription Recovery with Kestra

id: postgres-logical-subscription-recovery
namespace: company.team
description: |
  Detect an apply worker that repeatedly fails without advancing a PostgreSQL
  logical subscription. Report a new stall once, show the evidence to on-call,
  and permit a logged transaction skip only after approval and a fresh safety check.
  Skipping loses every change in that publisher transaction; repairing the conflicting
  row on the subscriber is the preferred manual alternative.

concurrency: { limit: 1, behavior: QUEUE }

triggers:
  - id: replication_alert
    type: io.kestra.plugin.core.trigger.Webhook
    description: Call POST
      /api/v1/main/executions/webhook/company.team/postgres-logical-subscription-recovery/pg-subscription-change-me
      from an Alertmanager subscription-error alert.
    key: pg-subscription-change-me
  - id: minute_backstop
    type: io.kestra.plugin.core.trigger.Schedule
    description: Enable only after setting the subscriber secret; the sweep detects
      stalls even when Alertmanager misses an event.
    cron: "*/1 * * * *"
    disabled: true

inputs:
  - id: subscription_name
    type: STRING
    defaults: demo_sub
    description: Name of one subscription on the subscriber database; use a separate
      flow copy for each subscription.
  - id: stale_seconds
    type: INT
    defaults: 60
    description: Seconds of unchanged apply position with repeated errors before
      declaring a stall; lower it only for a short local test.
  - id: force_review
    type: BOOL
    defaults: false
    description: Reopen an existing stall for a new human decision after a previous
      operator chose no action.
  - id: kestra_url
    type: STRING
    defaults: http://localhost:8080
    description: Browser-reachable Kestra base URL for the actionable on-call alert.

tasks:
  - id: validate_inputs
    type: io.kestra.plugin.core.flow.If
    condition: "{{ inputs.subscription_name == '' or inputs.stale_seconds < 1 }}"
    then:
      - id: reject_inputs
        type: io.kestra.plugin.core.execution.Fail
        errorMessage: Supply a nonempty subscription_name and stale_seconds of at least 1.

  - id: check_subscription
    type: io.kestra.plugin.jdbc.postgresql.Query
    description: Read the subscription, apply worker and cumulative error counters
      from the subscriber only.
    url: "{{ secret('POSTGRES_URL') }}"
    username: "{{ secret('POSTGRES_USERNAME') }}"
    password: "{{ secret('POSTGRES_PASSWORD') }}"
    fetchType: FETCH_ONE
    parameters:
      name: "{{ inputs.subscription_name }}"
    sql: |
      SELECT CAST(s.oid AS TEXT) AS subid, s.subname, s.subenabled AS enabled,
             s.subtwophasestate <> 'd' AS two_phase,
             CAST(s.subskiplsn AS TEXT) AS skip_lsn,
             COALESCE(st.apply_error_count, 0) AS apply_errors,
             COALESCE(st.sync_error_count, 0) AS sync_errors,
             COALESCE(CAST(ro.remote_lsn AS TEXT), '0/0') AS end_lsn,
             CAST(a.latest_end_time AS TEXT) AS latest_end_time,
             (a.pid IS NOT NULL) AS worker_running,
             CAST(current_setting('server_version_num') AS INT) AS version_num
      FROM pg_subscription s
      LEFT JOIN pg_stat_subscription_stats st ON st.subid = s.oid
      LEFT JOIN pg_replication_origin_status ro ON ro.external_id = 'pg_' || CAST(s.oid AS TEXT)
      LEFT JOIN LATERAL (
        SELECT pid, latest_end_time
        FROM pg_stat_subscription
        WHERE subid = s.oid AND relid IS NULL AND leader_pid IS NULL
        LIMIT 1
      ) a ON true
      WHERE s.subname = :name
    retry: { type: exponential, interval: PT2S, maxInterval: PT30S, maxAttempts: 3 }

  - id: check_supported
    type: io.kestra.plugin.core.flow.If
    condition: "{{ (outputs.check_subscription.row ?? {}) == {} or
      (outputs.check_subscription.row.version_num ?? 0) < 160000 }}"
    then:
      - id: reject_subscription
        type: io.kestra.plugin.core.execution.Fail
        errorMessage: Subscription missing or PostgreSQL older than 16. Check the name
          and connect to the subscriber, not the publisher.

  - id: ensure_watch
    type: io.kestra.plugin.jdbc.postgresql.Query
    description: An idempotent local ledger makes changes and approvals visible across runs.
    url: "{{ secret('POSTGRES_URL') }}"
    username: "{{ secret('POSTGRES_USERNAME') }}"
    password: "{{ secret('POSTGRES_PASSWORD') }}"
    sql: |
      CREATE TABLE IF NOT EXISTS subscription_recovery_watch (
        subid OID PRIMARY KEY,
        subname TEXT NOT NULL,
        last_lsn TEXT NOT NULL,
        last_apply_errors BIGINT NOT NULL,
        first_bad_at TIMESTAMPTZ,
        level TEXT NOT NULL DEFAULT 'ok',
        updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
      )

  - id: sample
    type: io.kestra.plugin.jdbc.postgresql.Query
    description: Carry the start of one stalled incident forward only while the
      confirmed apply LSN stays unchanged.
    url: "{{ secret('POSTGRES_URL') }}"
    username: "{{ secret('POSTGRES_USERNAME') }}"
    password: "{{ secret('POSTGRES_PASSWORD') }}"
    fetchType: FETCH_ONE
    parameters:
      subid: "{{ outputs.check_subscription.row.subid }}"
      name: "{{ outputs.check_subscription.row.subname }}"
      lsn: "{{ outputs.check_subscription.row.end_lsn }}"
      errors: "{{ outputs.check_subscription.row.apply_errors }}"
    sql: |
      INSERT INTO subscription_recovery_watch AS w
        (subid, subname, last_lsn, last_apply_errors, first_bad_at, updated_at)
      VALUES (CAST(:subid AS OID), :name, :lsn, CAST(:errors AS BIGINT), NULL, now())
      ON CONFLICT (subid) DO UPDATE SET
        subname = EXCLUDED.subname,
        first_bad_at = CASE
          WHEN w.last_lsn IS DISTINCT FROM EXCLUDED.last_lsn
            OR EXCLUDED.last_apply_errors < w.last_apply_errors THEN NULL
          WHEN EXCLUDED.last_apply_errors > w.last_apply_errors THEN COALESCE(w.first_bad_at, now())
          ELSE w.first_bad_at END,
        last_lsn = EXCLUDED.last_lsn,
        last_apply_errors = EXCLUDED.last_apply_errors,
        updated_at = now()
      RETURNING CAST(subid AS TEXT) AS subid, subname, last_lsn, last_apply_errors,
                CAST(first_bad_at AS TEXT) AS first_bad_at

  - id: classify
    type: io.kestra.plugin.jdbc.postgresql.Query
    description: The level transition is one database update, so overlapping
      executions cannot announce the same stall twice.
    url: "{{ secret('POSTGRES_URL') }}"
    username: "{{ secret('POSTGRES_USERNAME') }}"
    password: "{{ secret('POSTGRES_PASSWORD') }}"
    fetchType: FETCH_ONE
    parameters:
      subid: "{{ outputs.check_subscription.row.subid }}"
      stale: "{{ inputs.stale_seconds }}"
    sql: |
      WITH state AS (
        SELECT subid, level AS old_level,
          CASE WHEN {{ outputs.check_subscription.row.enabled }} AND first_bad_at IS NOT NULL
                     AND now() - first_bad_at >= CAST(:stale AS INT) * interval '1 second'
                THEN 'stalled' ELSE 'ok' END AS new_level
        FROM subscription_recovery_watch WHERE subid = CAST(:subid AS OID)
      ), changed AS (
        UPDATE subscription_recovery_watch w SET level = state.new_level
        FROM state WHERE w.subid = state.subid AND w.level <> state.new_level
        RETURNING w.subid
      )
      SELECT state.old_level, state.new_level, EXISTS(SELECT 1 FROM changed) AS changed
      FROM state

  - id: report_change
    type: io.kestra.plugin.core.flow.If
    condition: "{{ outputs.classify.row.changed or (inputs.force_review and
      outputs.classify.row.new_level == 'stalled') }}"
    then:
      - id: alert_transition
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        description: One alert per transition, with both the evidence and a safe
          alternative to skipping.
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        messageText: >-
          PostgreSQL subscription {{ inputs.subscription_name }}: {{
          outputs.classify.row.old_level }} -> {{ outputs.classify.row.new_level
          }}. Apply errors {{ outputs.check_subscription.row.apply_errors }},
          applied LSN {{ outputs.check_subscription.row.end_lsn }}, worker
          running {{ outputs.check_subscription.row.worker_running }}. {% if
          outputs.classify.row.new_level == 'stalled' %} Inspect the subscriber
          server log for the conflicting row and transaction finish LSN. Prefer
          fixing the row or permissions, then retry. A transaction skip discards
          all changes in that transaction. Review this execution: {% else %}
          Apply progress resumed. Verify publisher/subscriber row counts and
          reconcile any transaction that was skipped. Execution: {% endif %} {{
          inputs.kestra_url }}/ui/main/executions/company.team/{{ flow.id }}/{{
          execution.id }}
        retry: { type: exponential, interval: PT2S, maxInterval: PT30S, maxAttempts: 3 }

      - id: recovery_gate
        type: io.kestra.plugin.core.flow.If
        condition: "{{ outputs.classify.row.new_level == 'stalled' }}"
        then:
          - id: operator_decision
            type: io.kestra.plugin.core.flow.Pause
            description: Fix the conflicting row manually, or explicitly approve skipping
              the whole transaction using its finish LSN from the subscriber
              log.
            pauseDuration: PT2H
            behavior: FAIL
            onResume:
              - id: action
                type: SELECT
                values: [ NO_CHANGE, SKIP_TRANSACTION ]
                defaults: NO_CHANGE
                description: Choose NO_CHANGE after repairing the row, or SKIP_TRANSACTION only
                  if discarding the entire publisher transaction is acceptable.
              - id: finish_lsn
                type: STRING
                required: false
                description: Exact transaction finish LSN from the subscriber server log, for
                  example 0/14C0378; required for SKIP_TRANSACTION.
              - id: rationale
                type: STRING
                required: false
                description: Incident reference and reason for accepting data loss; required for
                  SKIP_TRANSACTION.

          - id: approved_skip
            type: io.kestra.plugin.core.flow.If
            condition: "{{ outputs.operator_decision.onResume.action == 'SKIP_TRANSACTION'
              }}"
            then:
              - id: prepare_skip
                type: io.kestra.plugin.jdbc.postgresql.Query
                description: Check the approver fields and live subscription again; build a
                  quoted command only if the observed position has not moved.
                url: "{{ secret('POSTGRES_URL') }}"
                username: "{{ secret('POSTGRES_USERNAME') }}"
                password: "{{ secret('POSTGRES_PASSWORD') }}"
                fetchType: FETCH_ONE
                parameters:
                  subid: "{{ outputs.check_subscription.row.subid }}"
                  observed_lsn: "{{ outputs.check_subscription.row.end_lsn }}"
                  finish_lsn: "{{ outputs.operator_decision.onResume.finish_lsn ?? '' }}"
                  rationale: "{{ outputs.operator_decision.onResume.rationale ?? '' }}"
                sql: |
                  SELECT CASE
                    WHEN s.subenabled AND s.subtwophasestate = 'd'
                      AND s.subskiplsn = '0/0'
                      AND COALESCE(CAST(ro.remote_lsn AS TEXT), '0/0') = :observed_lsn
                      AND :finish_lsn ~ '^[0-9A-Fa-f]+/[0-9A-Fa-f]+$'
                      AND length(trim(:rationale)) >= 8
                      AND st.apply_error_count > 0
                    THEN format('ALTER SUBSCRIPTION %I SKIP (lsn = %L)', s.subname, upper(:finish_lsn))
                    ELSE '' END AS command,
                    COALESCE(CAST(ro.remote_lsn AS TEXT), '0/0') AS current_lsn,
                    CAST(s.subskiplsn AS TEXT) AS current_skip_lsn
                  FROM pg_subscription s
                  JOIN pg_stat_subscription_stats st ON st.subid = s.oid
                  LEFT JOIN pg_replication_origin_status ro ON ro.external_id = 'pg_' || CAST(s.oid AS TEXT)
                  WHERE s.oid = CAST(:subid AS OID)
                retry: { type: exponential, interval: PT2S, maxInterval: PT30S, maxAttempts: 3 }

              - id: apply_if_safe
                type: io.kestra.plugin.core.flow.If
                condition: "{{ (outputs.prepare_skip.row.command ?? '') != '' }}"
                then:
                  - id: set_skip_lsn
                    type: io.kestra.plugin.jdbc.postgresql.Query
                    description: ALTER SUBSCRIPTION only; the approved finish LSN is quoted by
                      PostgreSQL format() and the command is recorded in this
                      execution.
                    url: "{{ secret('POSTGRES_URL') }}"
                    username: "{{ secret('POSTGRES_USERNAME') }}"
                    password: "{{ secret('POSTGRES_PASSWORD') }}"
                    sql: "{{ outputs.prepare_skip.row.command }}"
                  - id: verify_skip_setting
                    type: io.kestra.plugin.jdbc.postgresql.Query
                    description: Read back the requested skip setting; the apply worker may clear it
                      once it passes that transaction.
                    url: "{{ secret('POSTGRES_URL') }}"
                    username: "{{ secret('POSTGRES_USERNAME') }}"
                    password: "{{ secret('POSTGRES_PASSWORD') }}"
                    fetchType: FETCH_ONE
                    parameters:
                      subid: "{{ outputs.check_subscription.row.subid }}"
                    sql: |
                      SELECT CAST(subskiplsn AS TEXT) AS current_skip_lsn
                      FROM pg_subscription WHERE oid = CAST(:subid AS OID)
                  - id: announce_skip
                    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
                    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
                    messageText: >-
                      Approved by {{ outputs.operator_decision.resumed.by ??
                      'operator' }}: {{ outputs.prepare_skip.row.command }} for
                      subscription {{ inputs.subscription_name }}. Reason: {{
                      outputs.operator_decision.onResume.rationale }}. Setting
                      now {{ outputs.verify_skip_setting.row.current_skip_lsn
                      }}; check subscriber row counts and apply LSN on the next
                      run. Command accepted is not proof of data consistency.
                    retry: { type: exponential, interval: PT2S, maxInterval: PT30S, maxAttempts: 3 }
                else:
                  - id: stale_or_invalid
                    type: io.kestra.plugin.core.log.Log
                    message: >-
                      Skip rejected: invalid finish LSN or short rationale,
                      disabled/two-phase subscription, existing skip, or apply
                      position changed since the proposal. No SQL mutation ran.
            else:
              - id: manual_repair
                type: io.kestra.plugin.core.log.Log
                message: No transaction skipped. Repair the conflicting subscriber row or
                  permissions, then run the flow again.

errors:
  - id: alert_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    messageText: >-
      Subscription recovery execution {{ execution.id }} failed {% if
      errorLogs() | length > 0 %}at {{ errorLogs()[0]['taskId'] }}: {{
      errorLogs()[0]['message'] }}{% else %}without an error message{% endif %}
      Check the subscriber connection, Kestra task logs and whether a Pause
      expired. No automatic transaction skip is attempted.
    retry: { type: exponential, interval: PT2S, maxInterval: PT30S, maxAttempts: 3 }

outputs:
  - id: observation
    type: JSON
    value: "{{ (outputs.check_subscription.row ?? {}) | toJson }}"
  - id: classification
    type: JSON
    value: "{{ (outputs.classify.row ?? {}) | toJson }}"

A conflicting row can stop a PostgreSQL logical subscriber while the publisher keeps writing. The subscriber's apply-error counter rises, but the confirmed position does not. This blueprint samples both signals, reports a new stall once, and presents the operator with a choice. It runs against a local PostgreSQL 16 subscriber by default after secrets are set.

How it works

replication_alert (Webhook) starts diagnosis immediately; the disabled minute_backstop (Schedule) can be enabled as a sweep.
check_subscription reads pg_subscription, pg_stat_subscription_stats, pg_stat_subscription, and the durable pg_replication_origin_status.remote_lsn. sample remembers the first repeated error while the applied LSN stays fixed, even if the worker crashes and disappears from the live statistics view.
classify changes one ledger row from ok to stalled after stale_seconds, so repeated runs do not repeat the Slack alert.
operator_decision (Pause) asks whether to repair the conflict manually or skip the whole transaction. The operator must copy the transaction finish LSN from the subscriber log and explain the data-loss decision.
prepare_skip checks current subscription identity, enablement, two-phase mode, existing skip setting and unchanged apply position. Only then does set_skip_lsn run a PostgreSQL-quoted ALTER SUBSCRIPTION ... SKIP.

Error handling

Invalid subscription names and unsupported PostgreSQL versions fail before the ledger changes. Database calls retry with backoff; flow failures alert with the task and cause.
A denied, malformed, stale or expired approval never runs the skip. No automated conflicting-row deletion is offered because the correct repair depends on application semantics.
The accepted command is reported as a setting, not proof of recovered replication. PostgreSQL may clear subskiplsn after it has skipped the transaction. Operators must verify the apply LSN and reconcile the skipped transaction's rows.

What this blueprint teaches about Kestra

A webhook and optional schedule share one diagnosis. A database ledger deduplicates alerts across executions, while a Pause and a second read guard a destructive SQL statement after a potentially long human delay.

Prerequisites

PostgreSQL 16+ subscriber with a logical subscription. The Kestra database role must read subscription catalogs and own the subscription for an approved skip. PostgreSQL 16 is used in the quick start.
An incoming Slack webhook and a Kestra worker that can connect to the subscriber.

Secrets

POSTGRES_URL: JDBC URL of the subscriber database, not the publisher.
POSTGRES_USERNAME, POSTGRES_PASSWORD: subscriber role credentials.
SLACK_WEBHOOK_URL: incoming webhook for alerts and approved actions.

Inputs

subscription_name (STRING, demo_sub): subscription to monitor.
stale_seconds (INT, 60): minimum time of unchanged position with apply errors.
force_review (BOOL, false): request another decision for an existing stall.
kestra_url (STRING, http://localhost:8080): link in the alert.

Outputs

observation (JSON): raw subscription state, including error counts and confirmed LSN.
classification (JSON): old/new level and whether it changed. Outputs on failed runs may be empty.

Quick start

Create publication and subscription as in the PostgreSQL logical replication documentation. Connect POSTGRES_URL to the subscriber.
Set all four secrets, change the webhook key, then run twice with at least stale_seconds between samples.
On a stalled alert, inspect the subscriber log's transaction ... finished at LSN context. Repair the conflicting row if possible. If the whole transaction may be discarded, resume with SKIP_TRANSACTION, the finish LSN and a reason.
After a skip, compare publisher and subscriber rows for that transaction, then run again to confirm apply progress.