PostgreSQL Logical Failover Rehearsal

id: postgres-subscriber-continuity-rehearsal
namespace: company.rehearsal
description: Rehearse a disposable PostgreSQL 17 publisher switchover and verify
  the existing logical subscriber.
concurrency:
  limit: 1

inputs:
  - id: allow_disposable_switchover
    type: BOOL
    defaults: false
  - id: expected_publisher_system_id
    type: STRING
    required: true
  - id: expected_subscriber_system_id
    type: STRING
    required: true

tasks:
  - id: require_consent
    type: io.kestra.plugin.core.flow.If
    condition: "{{ not inputs.allow_disposable_switchover }}"
    then:
      - id: consent_missing
        type: io.kestra.plugin.core.execution.Fail
        errorMessage: This flow stops its disposable publisher. Explicit lab consent is
          required.

  - id: inspect_publisher
    type: io.kestra.plugin.jdbc.postgresql.Query
    url: jdbc:postgresql://publisher:5432/rehearsal
    username: lab
    password: "{{ secret('LAB_DB_PASSWORD') }}"
    fetchType: FETCH_ONE
    sql: |
      SELECT system_identifier::text AS system_id,
             pg_is_in_recovery() AS recovery,
             current_setting('cluster_name') AS cluster_name,
             current_setting('server_version_num')::int AS version_num,
             pg_current_wal_lsn()::text AS barrier_lsn,
             (SELECT string_agg(id::text || ':' || value, ',' ORDER BY id)
              FROM rehearsal_items) AS baseline
      FROM pg_control_system();

  - id: publisher_observation
    type: io.kestra.plugin.core.debug.Return
    format: "id={{ outputs.inspect_publisher.row.system_id }}, expected={{
      inputs.expected_publisher_system_id }}, recovery={{
      outputs.inspect_publisher.row.recovery }}, cluster={{
      outputs.inspect_publisher.row.cluster_name }}, version={{
      outputs.inspect_publisher.row.version_num }}, baseline={{
      outputs.inspect_publisher.row.baseline }}"

  - id: require_expected_publisher
    type: io.kestra.plugin.core.flow.If
    condition: "{{ outputs.inspect_publisher.row.system_id !=
      inputs.expected_publisher_system_id or
      outputs.inspect_publisher.row.recovery or
      outputs.inspect_publisher.row.cluster_name != 'rehearsal_publisher' or
      outputs.inspect_publisher.row.version_num < 170000 or
      outputs.inspect_publisher.row.baseline != '1:keep,2:before,3:remove' }}"
    then:
      - id: unexpected_publisher
        type: io.kestra.plugin.core.execution.Fail
        errorMessage: Publisher identity, role, version, or disposable fixture does not
          match.

  - id: inspect_subscriber
    type: io.kestra.plugin.jdbc.postgresql.Query
    url: jdbc:postgresql://subscriber:5432/rehearsal
    username: lab
    password: "{{ secret('LAB_DB_PASSWORD') }}"
    fetchType: FETCH_ONE
    sql: |
      SELECT (SELECT system_identifier::text FROM pg_control_system()) AS system_id,
             (SELECT oid::text FROM pg_subscription WHERE subname='rehearsal_sub') AS subscription_oid,
             (SELECT roident::text FROM pg_replication_origin
              WHERE roname='pg_' || (SELECT oid::text FROM pg_subscription
                                    WHERE subname='rehearsal_sub')) AS origin_id,
             (SELECT count(*)=1 AND bool_and(subname='rehearsal_sub' AND subenabled
                     AND subfailover AND subslotname='rehearsal_sub')
              FROM pg_subscription) AS subscription_ready,
             (SELECT count(*)=1 AND bool_and(srsubstate='r')
              FROM pg_subscription_rel) AS tables_ready,
             (SELECT string_agg(id::text || ':' || value, ',' ORDER BY id)
              FROM rehearsal_items) AS baseline;

  - id: require_expected_subscriber
    type: io.kestra.plugin.core.flow.If
    condition: "{{ outputs.inspect_subscriber.row.system_id !=
      inputs.expected_subscriber_system_id or
      outputs.inspect_subscriber.row.system_id ==
      inputs.expected_publisher_system_id or
      outputs.inspect_subscriber.row.origin_id == null or not
      outputs.inspect_subscriber.row.subscription_ready or not
      outputs.inspect_subscriber.row.tables_ready or
      outputs.inspect_subscriber.row.baseline != '1:keep,2:before,3:remove' }}"
    then:
      - id: unexpected_subscriber
        type: io.kestra.plugin.core.execution.Fail
        errorMessage: Expected one enabled failover subscription, one ready table, and
          the known baseline.

  - id: inspect_standby
    type: io.kestra.plugin.jdbc.postgresql.Query
    url: jdbc:postgresql://standby:5432/rehearsal
    username: lab
    password: "{{ secret('LAB_DB_PASSWORD') }}"
    fetchType: FETCH_ONE
    sql: |
      SELECT system_identifier::text AS system_id,
             pg_is_in_recovery() AS recovery,
             current_setting('cluster_name') AS cluster_name,
             pg_last_wal_replay_lsn() >= '{{ outputs.inspect_publisher.row.barrier_lsn }}'::pg_lsn AS caught_up,
             (SELECT count(*)=1 AND coalesce(bool_and(synced AND NOT temporary
                     AND invalidation_reason IS NULL),false)
              FROM pg_replication_slots WHERE slot_name='rehearsal_sub') AS slot_ready
      FROM pg_control_system();

  - id: require_failover_slot
    type: io.kestra.plugin.core.flow.If
    condition: "{{ outputs.inspect_standby.row.system_id !=
      inputs.expected_publisher_system_id or not
      outputs.inspect_standby.row.recovery or
      outputs.inspect_standby.row.cluster_name != 'rehearsal_standby' or not
      outputs.inspect_standby.row.caught_up or not
      outputs.inspect_standby.row.slot_ready }}"
    then:
      - id: standby_not_ready
        type: io.kestra.plugin.core.execution.Fail
        errorMessage: Standby identity, recovery, WAL replay, or synchronized failover
          slot is not ready.

  - id: pause_subscriber
    type: io.kestra.plugin.jdbc.postgresql.Query
    url: jdbc:postgresql://subscriber:5432/rehearsal
    username: lab
    password: "{{ secret('LAB_DB_PASSWORD') }}"
    sql: ALTER SUBSCRIPTION rehearsal_sub DISABLE;

  - id: stop_old_publisher
    type: io.kestra.plugin.docker.cli.Stop
    containerId: kestra-rehearsal-publisher
    kill: false
    delete: false

  - id: promote_standby
    type: io.kestra.plugin.jdbc.postgresql.Query
    url: jdbc:postgresql://standby:5432/rehearsal
    username: lab
    password: "{{ secret('LAB_DB_PASSWORD') }}"
    fetchType: FETCH_ONE
    sql: SELECT pg_promote(true,30) AS promoted;

  - id: verify_promotion
    type: io.kestra.plugin.jdbc.postgresql.Query
    url: jdbc:postgresql://standby:5432/rehearsal
    username: lab
    password: "{{ secret('LAB_DB_PASSWORD') }}"
    fetchType: FETCH_ONE
    sql: |
      SELECT NOT pg_is_in_recovery() AS primary_now,
             current_setting('synchronized_standby_slots') = '' AS no_stale_wait,
             system_identifier::text AS system_id
      FROM pg_control_system();

  - id: require_new_primary
    type: io.kestra.plugin.core.flow.If
    condition: "{{ not outputs.promote_standby.row.promoted or not
      outputs.verify_promotion.row.primary_now or not
      outputs.verify_promotion.row.no_stale_wait or
      outputs.verify_promotion.row.system_id !=
      inputs.expected_publisher_system_id }}"
    then:
      - id: promotion_unverified
        type: io.kestra.plugin.core.execution.Fail
        errorMessage: Promotion was not verified or inherited standby waits would block
          this disposable topology.

  - id: reconnect_subscriber
    type: io.kestra.plugin.jdbc.postgresql.Query
    url: jdbc:postgresql://subscriber:5432/rehearsal
    username: lab
    password: "{{ secret('LAB_DB_PASSWORD') }}"
    sql: |
      ALTER SUBSCRIPTION rehearsal_sub CONNECTION
      'host=standby port=5432 dbname=rehearsal user=lab passfile=/var/lib/postgresql/.pgpass';

  - id: resume_subscriber
    type: io.kestra.plugin.jdbc.postgresql.Query
    url: jdbc:postgresql://subscriber:5432/rehearsal
    username: lab
    password: "{{ secret('LAB_DB_PASSWORD') }}"
    sql: ALTER SUBSCRIPTION rehearsal_sub ENABLE;

  - id: mutate_new_primary
    type: io.kestra.plugin.jdbc.postgresql.Queries
    url: jdbc:postgresql://standby:5432/rehearsal
    username: lab
    password: "{{ secret('LAB_DB_PASSWORD') }}"
    sql: |
      UPDATE rehearsal_items SET value='after' WHERE id=2;
      DELETE FROM rehearsal_items WHERE id=3;
      INSERT INTO rehearsal_items VALUES (4,'inserted');

  - id: await_subscriber_data
    type: io.kestra.plugin.core.flow.LoopUntil
    condition: "{{ outputs.verify_subscriber.row.matches }}"
    failOnMaxReached: true
    checkFrequency:
      interval: PT1S
      maxIterations: 30
      maxDuration: PT1M
    tasks:
      - id: verify_subscriber
        type: io.kestra.plugin.jdbc.postgresql.Query
        url: jdbc:postgresql://subscriber:5432/rehearsal
        username: lab
        password: "{{ secret('LAB_DB_PASSWORD') }}"
        fetchType: FETCH_ONE
        sql: |
          SELECT string_agg(id::text || ':' || value, ',' ORDER BY id) AS observed,
                 string_agg(id::text || ':' || value, ',' ORDER BY id) =
                   '1:keep,2:after,4:inserted' AS matches
          FROM rehearsal_items;

  - id: verify_consumer_identity
    type: io.kestra.plugin.jdbc.postgresql.Query
    url: jdbc:postgresql://subscriber:5432/rehearsal
    username: lab
    password: "{{ secret('LAB_DB_PASSWORD') }}"
    fetchType: FETCH_ONE
    sql: |
      SELECT (SELECT oid::text FROM pg_subscription WHERE subname='rehearsal_sub') AS subscription_oid,
             (SELECT roident::text FROM pg_replication_origin
              WHERE roname='pg_' || (SELECT oid::text FROM pg_subscription
                                    WHERE subname='rehearsal_sub')) AS origin_id;

  - id: require_same_consumer
    type: io.kestra.plugin.core.flow.If
    condition: "{{ outputs.verify_consumer_identity.row.subscription_oid !=
      outputs.inspect_subscriber.row.subscription_oid or
      outputs.verify_consumer_identity.row.origin_id !=
      outputs.inspect_subscriber.row.origin_id }}"
    then:
      - id: consumer_recreated
        type: io.kestra.plugin.core.execution.Fail
        errorMessage: Subscriber identity changed; recreation is not continuity.

  - id: continuity_evidence
    type: io.kestra.plugin.core.output.OutputValues
    values:
      verdict: verified_in_disposable_lab
      publisher_system_id: "{{ outputs.inspect_publisher.row.system_id }}"
      subscriber_system_id: "{{ outputs.inspect_subscriber.row.system_id }}"
      subscription_oid: "{{ outputs.verify_consumer_identity.row.subscription_oid }}"
      origin_id: "{{ outputs.verify_consumer_identity.row.origin_id }}"
      observed_rows: "{{ outputs.verify_subscriber.row.observed }}"

triggers:
  - id: scheduled_rehearsal
    type: io.kestra.plugin.core.trigger.Schedule
    disabled: true
    cron: "0 6 * * 0"
    inputs:
      allow_disposable_switchover: false
      expected_publisher_system_id: not-configured
      expected_subscriber_system_id: not-configured

Exercise a deliberately disposable PostgreSQL 17 topology before trusting a switchover procedure. The existing subscriber must retain its identity and receive exact fixture changes after the candidate is promoted.

How it works

Require explicit consent and the two expected database system identifiers.
Check roles, baseline data, one ready subscription and a synchronized slot.
Disable the subscriber, stop the old Docker publisher, then promote the standby.
Reconnect the same subscription, mutate three records and poll for exact results.

What you get

A Kestra execution showing the real shutdown, promotion and consumer checks.
Refusal before shutdown for unexpected identities, data or slot readiness.
A verified fixture result with unchanged subscription and replication-origin IDs.
Explicit failure on timeout; no zero-data-loss or automatic rollback claim.

Who it's for

Data and platform engineers testing a PostgreSQL logical-replication switchover in an isolated, quiescent lab. This is not a production HA controller.

Why orchestrate this with Kestra

Native JDBC, Docker, conditions and bounded polling expose each phase and its evidence. PostgreSQL owns replication and promotion; Kestra sequences the drill.

Prerequisites

Kestra OSS 2.0.6, JDBC PostgreSQL and Docker plugins, and the companion disposable Docker fixture. It provides publisher, standby and subscriber DNS names, the rehearsal database/table, one failover subscription and a subscriber passfile. The Kestra worker needs access to the lab Docker socket. The fixture must have no concurrent application writers. Keep it separate from production resources.

Secrets

LAB_DB_PASSWORD: generated fixture password, supplied through OSS SECRET_ encoding.

Quick start

Start and bootstrap the companion fixture; record both database system identifiers.
Import this flow and supply those identifiers with disposable-switchover consent.
Inspect the execution and independently query the subscriber's three final records.
Reset the fixture before another drill. The schedule remains disabled and nonconsenting.

How to extend

Adapt a separately validated fixture to test another PostgreSQL version or HA interface. Do not reuse this shutdown mechanism for live traffic. Multiple subscribers, incomplete copies and recovery after partial promotion need new tests.

Rehearse PostgreSQL failover and verify logical subscriber continuity

How it works

What you get

Who it's for

Why orchestrate this with Kestra

Prerequisites

Secrets

Quick start

How to extend

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

New to Kestra?

Rehearse PostgreSQL failover and verify logical subscriber continuity

How it works

What you get

Who it's for

Why orchestrate this with Kestra

Prerequisites

Secrets

Quick start

How to extend

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

More Data Blueprints

Enforce Retention on a Pinecone Namespace with a Weekly Purge

Load Precomputed Embeddings into Pinecone with a Daily Upsert Pipeline

Build a point-in-time correct training set and block feature leakage

Spot-Check Pinecone Retrieval Quality with a Query by Vector Id

Apache Pinot Consuming Segment Commit Stall and Consumer Lag Guard

Execute GDPR Right-to-Erasure on Pinecone Vectors with Proof of Deletion

New to Kestra?