Citus Reference Table Replica Guard with Kestra

id: citus-reference-table-replica-guard
namespace: company.team

description: |
  Find the active primary workers in a Citus cluster that are missing a reference table, which is
  what a node added with citus_add_node looks like until someone replicates to it. List the exact
  worker and table pairs plus the bytes a copy would move, then call replicate_reference_tables
  behind an approval, read the placement metadata again per worker and fail the run if any worker
  is still short.

inputs:
  - id: jdbc_url
    type: STRING
    displayName: Coordinator JDBC URL
    description: The Citus coordinator. Worker nodes and reference tables are read
      from its metadata.
    defaults: "jdbc:postgresql://localhost:5432/citus"

  - id: shard_transfer_mode
    type: SELECT
    displayName: Shard transfer mode
    description: >-
      How replicate_reference_tables copies the data. block_writes copies with
      writes blocked on the reference tables and works on any wal_level. auto
      and force_logical use logical replication, which needs wal_level=logical
      on the coordinator.
    values:
      - block_writes
      - auto
      - force_logical
    defaults: block_writes

  - id: max_copy_mb
    type: INT
    displayName: Copy budget in MB
    description: Refuse to start the copy when the missing placements add up to more
      than this. Reference tables are usually small, so a large number here
      means something unexpected is distributed as one.
    defaults: 1024

  - id: auto_remediate
    type: BOOL
    displayName: Auto remediate
    description: Allow the flow to call replicate_reference_tables. Off means report only.
    defaults: false

  - id: dry_run
    type: BOOL
    displayName: Dry run
    description: Must be false, together with auto_remediate, before any data is copied.
    defaults: true

variables:
  placement_sql: |
    WITH ref AS (
      SELECT s.shardid, pt.logicalrelid::regclass::text AS table_name
        FROM pg_dist_partition pt
        JOIN pg_dist_shard s ON s.logicalrelid = pt.logicalrelid
       WHERE pt.partmethod = 'n'
    ), node AS (
      SELECT n.groupid, n.nodename, n.nodeport, n.shouldhaveshards
        FROM pg_dist_node n
       WHERE n.noderole = 'primary' AND n.isactive AND n.groupid <> 0
    ), shard_size AS (
      SELECT shard_id, max(size) AS bytes FROM citus_shard_sizes() GROUP BY shard_id
    ), expected AS (
      SELECT n.groupid, n.nodename, n.nodeport, n.shouldhaveshards,
             r.shardid, r.table_name, coalesce(z.bytes, 0) AS bytes,
             EXISTS (SELECT 1 FROM pg_dist_placement p
                      WHERE p.shardid = r.shardid AND p.groupid = n.groupid
                        AND p.shardstate = 1) AS present
        FROM node n
        LEFT JOIN ref r ON true
        LEFT JOIN shard_size z ON z.shard_id = r.shardid
    ), per_node AS (
      SELECT groupid, nodename, nodeport, shouldhaveshards,
             count(shardid) AS ref_tables,
             count(shardid) FILTER (WHERE present) AS present_count,
             count(shardid) FILTER (WHERE NOT present) AS missing_count,
             coalesce(sum(bytes) FILTER (WHERE NOT present), 0) AS missing_bytes,
             coalesce(string_agg(table_name, ', ' ORDER BY table_name)
                        FILTER (WHERE NOT present), '') AS missing_tables
        FROM expected GROUP BY 1, 2, 3, 4
    )

  coverage_sql: |
    {{ render(vars.placement_sql) }}
    SELECT nodename || ':' || nodeport AS worker, shouldhaveshards AS accepts_shards,
           ref_tables, present_count, missing_count,
           missing_bytes, pg_size_pretty(missing_bytes) AS missing_size,
           missing_tables
      FROM per_node ORDER BY missing_count DESC, nodename

  assess_sql: |
    {{ render(vars.placement_sql) }}, agg AS (
      SELECT count(*) AS worker_count,
             coalesce(max(ref_tables), 0) AS ref_table_count,
             coalesce(sum(present_count), 0) AS present_placements,
             coalesce(sum(missing_count), 0) AS missing_placements,
             coalesce(sum(missing_bytes), 0) AS missing_bytes,
             count(*) FILTER (WHERE missing_count > 0) AS affected_workers
        FROM per_node
    )
    SELECT
      CASE WHEN a.missing_placements > 0 THEN 'MISSING_REPLICAS' ELSE 'OK' END AS state,
      a.worker_count, a.ref_table_count, a.affected_workers,
      a.present_placements, a.missing_placements,
      a.worker_count * a.ref_table_count AS expected_placements,
      a.missing_bytes, pg_size_pretty(a.missing_bytes) AS missing_size,
      (a.missing_bytes <= {{ inputs.max_copy_mb }}::bigint * 1048576) AS within_budget,
      to_char(100.0 * a.present_placements
                / greatest(a.worker_count * a.ref_table_count, 1), 'FM990.0') AS coverage_pct,
      coalesce((SELECT string_agg(nodename || ':' || nodeport || ' missing ' || missing_tables,
                                  ', ' ORDER BY nodename)
                  FROM per_node WHERE missing_count > 0), '') AS affected_list,
      coalesce((SELECT string_agg(DISTINCT table_name, ', ' ORDER BY table_name)
                  FROM expected WHERE table_name IS NOT NULL), '') AS ref_table_list,
      CASE WHEN a.worker_count = 0
             THEN 'No active primary worker is registered, so there is nothing to replicate to.'
           WHEN a.ref_table_count = 0
             THEN 'The cluster has no reference table, so every worker is complete already.'
           ELSE '' END AS note
      FROM agg a

tasks:
  - id: coverage
    type: io.kestra.plugin.jdbc.postgresql.Query
    description: >-
      Read-only. One row per active primary worker, built by crossing every
      reference table shard in pg_dist_partition with every worker in
      pg_dist_node, then asking pg_dist_placement whether that pair exists and
      is healthy. A worker that holds nothing still gets a row.
    url: "{{ inputs.jdbc_url }}"
    username: "{{ secret('POSTGRES_USERNAME') }}"
    password: "{{ secret('POSTGRES_PASSWORD') }}"
    fetchType: FETCH
    sql: "{{ render(vars.coverage_sql) }}"

  - id: assess
    type: io.kestra.plugin.jdbc.postgresql.Query
    description: >-
      Read-only. MISSING_REPLICAS when any active primary worker lacks a
      reference table placement, OK otherwise, with the bytes a copy would move
      and whether that fits the budget.
    url: "{{ inputs.jdbc_url }}"
    username: "{{ secret('POSTGRES_USERNAME') }}"
    password: "{{ secret('POSTGRES_PASSWORD') }}"
    fetchType: FETCH_ONE
    sql: "{{ render(vars.assess_sql) }}"

  - id: report
    type: io.kestra.plugin.core.log.Log
    message: >-
      Citus reference tables on {{ inputs.jdbc_url }} are {{
      outputs.assess.row.state }}. {{ outputs.assess.row.ref_table_count }}
      reference tables over {{ outputs.assess.row.worker_count }} active primary
      workers: {{ outputs.assess.row.present_placements }} of {{
      outputs.assess.row.expected_placements }} placements present ({{
      outputs.assess.row.coverage_pct }}%), {{
      outputs.assess.row.missing_placements }} missing on {{
      outputs.assess.row.affected_workers }} workers, {{
      outputs.assess.row.missing_size }} to copy. {{
      outputs.assess.row.ref_table_list == '' ? '' : 'Reference tables: ' ~
      outputs.assess.row.ref_table_list ~ '. ' }}{{
      outputs.assess.row.affected_list == '' ? '' : 'Short workers: ' ~
      outputs.assess.row.affected_list ~ '. ' }}{{ outputs.assess.row.note }} {%
      for r in outputs.coverage.rows %}[{{ r.worker }} {{ r.present_count }}/{{
      r.ref_tables }}{{ r.missing_count > 0 ? ' missing ' ~ r.missing_tables ~ '
      (' ~ r.missing_size ~ ')' : '' }}{{ r.accepts_shards ? '' : ' draining'
      }}] {% endfor %}

  - id: route
    type: io.kestra.plugin.core.flow.If
    condition: "{{ outputs.assess.row.state != 'OK' }}"
    then:
      - id: alert
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        payload: |
          {{ {'text': 'Citus ' ~ inputs.jdbc_url ~ ' is ' ~ outputs.assess.row.state ~ '. ' ~ outputs.assess.row.affected_workers ~ ' of ' ~ outputs.assess.row.worker_count ~ ' workers are missing ' ~ outputs.assess.row.missing_placements ~ ' reference table placements (' ~ outputs.assess.row.missing_size ~ ' to copy). ' ~ outputs.assess.row.affected_list ~ '.' ~ (inputs.auto_remediate and not inputs.dry_run and outputs.assess.row.within_budget ? ' Replication pending approval.' : ' No change applied (auto_remediate=' ~ inputs.auto_remediate ~ ', dry_run=' ~ inputs.dry_run ~ ', within_budget=' ~ outputs.assess.row.within_budget ~ ').')} | toJson }}

      - id: plan
        type: io.kestra.plugin.jdbc.postgresql.Query
        description: >-
          The dry run, read-only. Citus has no plan function for reference
          replication, so the plan is the list of worker and table pairs with no
          healthy placement, each with the bytes that copy would move.
        url: "{{ inputs.jdbc_url }}"
        username: "{{ secret('POSTGRES_USERNAME') }}"
        password: "{{ secret('POSTGRES_PASSWORD') }}"
        fetchType: FETCH
        sql: >-
          {{ render(vars.placement_sql) }} SELECT nodename || ':' || nodeport AS
          worker, table_name, shardid,
                 bytes, pg_size_pretty(bytes) AS size
            FROM expected WHERE NOT present
           ORDER BY bytes DESC, nodename, table_name

      - id: log_plan
        type: io.kestra.plugin.core.log.Log
        message: >-
          {{ outputs.plan.size }} reference table placements would be created,
          {{ outputs.assess.row.missing_size }} of data, with
          shard_transfer_mode {{ inputs.shard_transfer_mode }}. {% for r in
          outputs.plan.rows %}{{ r.table_name }} shard {{ r.shardid }} ({{
          r.size }}) to {{ r.worker }}. {% endfor %}

      - id: remediate_gate
        type: io.kestra.plugin.core.flow.If
        condition: "{{ inputs.auto_remediate and not inputs.dry_run and
          outputs.assess.row.within_budget and outputs.plan.size > 0 }}"
        then:
          - id: approve
            type: io.kestra.plugin.core.flow.Pause
            description: Resume the execution to copy the reference tables listed by
              log_plan. Kill it to cancel. Replication copies the whole table to
              every short worker, so it is never applied without this.

          - id: replicate
            type: io.kestra.plugin.jdbc.postgresql.Query
            description: >-
              replicate_reference_tables copies every under-replicated reference
              table to every worker that is short, in one cluster-wide call. It
              returns void. That return value is deliberately ignored, because
              the gate is the re-read below.
            url: "{{ inputs.jdbc_url }}"
            username: "{{ secret('POSTGRES_USERNAME') }}"
            password: "{{ secret('POSTGRES_PASSWORD') }}"
            fetchType: NONE
            sql: "SELECT replicate_reference_tables('{{ inputs.shard_transfer_mode }}')"

          - id: recoverage
            type: io.kestra.plugin.jdbc.postgresql.Query
            description: Read the placement metadata again, per worker, after the copy.
            url: "{{ inputs.jdbc_url }}"
            username: "{{ secret('POSTGRES_USERNAME') }}"
            password: "{{ secret('POSTGRES_PASSWORD') }}"
            fetchType: FETCH
            sql: "{{ render(vars.coverage_sql) }}"

          - id: reverify
            type: io.kestra.plugin.jdbc.postgresql.Query
            description: >-
              The same assessment, wrapped so the before numbers decide the
              verdict in SQL. fixed is false unless every active primary worker
              now holds every reference table.
            url: "{{ inputs.jdbc_url }}"
            username: "{{ secret('POSTGRES_USERNAME') }}"
            password: "{{ secret('POSTGRES_PASSWORD') }}"
            fetchType: FETCH_ONE
            sql: >-
              WITH after AS (
                {{ render(vars.assess_sql) }}
              ) SELECT a.*, {{ outputs.assess.row.missing_placements }} AS
              before_missing,
                     {{ outputs.assess.row.affected_workers }} AS before_affected,
                     (a.missing_placements = 0) AS fixed
                FROM after a

          - id: verify
            type: io.kestra.plugin.core.flow.If
            condition: "{{ outputs.reverify.row.fixed }}"
            then:
              - id: replicated
                type: io.kestra.plugin.core.log.Log
                message: >-
                  Reference tables replicated with {{ inputs.shard_transfer_mode
                  }}. Missing placements {{ outputs.reverify.row.before_missing
                  }} before, {{ outputs.reverify.row.missing_placements }}
                  after. Short workers {{ outputs.reverify.row.before_affected
                  }} before, {{ outputs.reverify.row.affected_workers }} after.
                  Coverage {{ outputs.assess.row.coverage_pct }}% before, {{
                  outputs.reverify.row.coverage_pct }}% after over {{
                  outputs.reverify.row.worker_count }} workers. Cluster is now
                  {{ outputs.reverify.row.state }}. {% for r in
                  outputs.recoverage.rows %}[{{ r.worker }} {{ r.present_count
                  }}/{{ r.ref_tables }}] {% endfor %}
            else:
              - id: remediation_failed
                type: io.kestra.plugin.core.execution.Fail
                errorMessage: >-
                  replicate_reference_tables returned without error, but {{
                  outputs.reverify.row.missing_placements }} reference table
                  placements are still missing on {{
                  outputs.reverify.row.affected_workers }} workers ({{
                  outputs.reverify.row.before_missing }} missing before the
                  call). Still short: {{ outputs.reverify.row.affected_list }}.
                  A worker that joined between the plan and the call is caught
                  here. So is one the coordinator cannot reach, since the copy
                  is skipped per node. Check pg_dist_node and the coordinator
                  log, then re-run.
        else:
          - id: reported_only
            type: io.kestra.plugin.core.log.Log
            message: >-
              {{ outputs.plan.size }} reference table placements reported, none
              created. {{ outputs.assess.row.within_budget ? 'auto_remediate=' ~
              inputs.auto_remediate ~ ', dry_run=' ~ inputs.dry_run ~ '.' : 'The
              copy is ' ~ outputs.assess.row.missing_size ~ ', over the ' ~
              inputs.max_copy_mb ~ ' MB budget, so it was refused. Raise
              max_copy_mb to allow it. A reference table that large is usually
              one that should have been distributed instead.' }}
    else:
      - id: complete
        type: io.kestra.plugin.core.log.Log
        message: >-
          Every one of {{ outputs.assess.row.worker_count }} active primary
          workers holds all {{ outputs.assess.row.ref_table_count }} reference
          tables, {{ outputs.assess.row.present_placements }} placements in
          total. {{ outputs.assess.row.note }}

errors:
  - id: alert_flow_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    payload: |
      {{ {'text': 'Citus reference table replica guard failed on ' ~ inputs.jdbc_url ~ ' (execution ' ~ execution.id ~ '). Check the logs.'} | toJson }}

triggers:
  - id: daily
    type: io.kestra.plugin.core.trigger.Schedule
    description: Shipped disabled and always in safe mode.
    cron: "0 3 * * *"
    disabled: true
    inputs:
      auto_remediate: false
      dry_run: true

A Citus reference table is meant to exist in full on every worker. That is what lets a join between a distributed table and a reference table run locally on each node instead of shipping rows across the cluster. It is also what lets a foreign key from a distributed table to that reference table be enforced where the shard lives. Add a worker with citus_add_node and Citus does not copy the existing reference tables to it. Nothing errors at add time. The cluster looks bigger, the new node is in pg_dist_node as an active primary. The reference tables are simply not there.

This blueprint crosses every reference table shard with every active primary worker and asks pg_dist_placement whether that pair exists, so a worker that holds nothing still gets a row. It names the worker and the table, adds up the bytes a copy would move, then calls replicate_reference_tables behind an approval and reads the placement metadata again per worker. A call that returns without error while a worker is still short fails the run.

This blueprint was created by zkasuran.

Findings

Five things measured on the rig, each of which changes the design:

A node added with citus_add_node gets no reference tables, even with citus.replicate_reference_tables_on_activate left at its default of on. Verified on Citus 12.1: a cluster with two reference tables and one worker, then a second worker added, came back with 2 placements on the first worker and 0 on the second. So the default configuration is already the broken case, which is why this check is worth running on a schedule rather than only after a known config change.
The missing replica blocks the rebalance, not just the join. citus_move_shard_placement of a distributed table shard onto the new worker failed with ERROR: relation "public.countries_102008" does not exist, CONTEXT: while executing command on <new worker>, because the shard's foreign key to the reference table cannot be created there. The same CREATE TABLE ... REFERENCES <reference table> that failed while the worker was short succeeded once the copy had run. A new node therefore stays unusable for distributed data until the reference tables are on it.
replicate_reference_tables() defaults to shard_transfer_mode => 'auto', which fails outright on a stock cluster. On default wal_level = replica it returned ERROR: logical decoding requires wal_level >= logical after printing the NOTICE that it was replicating. block_writes works on any wal_level, so it is the default here. The copy is real IO over the network, which is what the approval and the byte budget are for.
shouldhaveshards = false does not exclude a node. A worker marked as draining still received the reference tables and still needs them, because reference tables are not what the rebalancer moves. The check covers every active primary worker and reports the draining flag for context instead of skipping the node.
replicate_reference_tables returns void over a placement that exists but is not healthy. A placement row left at shardstate = 3 was not repaired and the call still returned success, so the cluster stayed at 1 of 2 tables on that worker. The check counts a placement as present only at shardstate = 1. The re-read after the call turns that silence into a failed run.

One trap worth knowing while confirming any of this by hand: Citus hides shard tables from pg_class for ordinary clients, so \dt on a worker shows the shell tables and no shards at all. SET citus.show_shards_for_app_name_prefixes = '*' reveals countries_102008 and friends.

Safety

The whole detection path, including the plan, reads pg_dist_partition, pg_dist_shard, pg_dist_placement, pg_dist_node and citus_shard_sizes(). Nothing is written.
The copy runs only when auto_remediate is true, dry_run is false, the plan is not empty and the bytes fit max_copy_mb. Even then it runs only after the Pause is resumed.
shard_transfer_mode is explicit. block_writes blocks writes on the reference tables while they copy and needs nothing from wal_level.
The run fails when the metadata still shows a short worker after the call, because a function that returns void is not evidence that anything landed.

How it works

coverage (io.kestra.plugin.jdbc.postgresql.Query, read-only) crosses every reference table shard with every active primary worker and returns one row per worker: how many reference tables it should hold, how many it holds, which ones are missing and the bytes behind them. partmethod = 'n' in pg_dist_partition is what identifies a reference table.

assess turns the same data into a verdict:

State	Meaning
`MISSING_REPLICAS`	At least one active primary worker has no healthy placement for at least one reference table
`OK`	Every active primary worker holds every reference table, including the cases of no workers and no reference tables

report logs the verdict, the coverage percentage, the named workers and the per-worker table.
route (io.kestra.plugin.core.flow.If):
- OK: complete logs it.
- MISSING_REPLICAS: alert posts to Slack.
  - plan lists every worker and table pair with no healthy placement, with the bytes that copy would move. Citus has no plan function for reference replication, so the plan comes from the metadata. log_plan prints it.
  - remediate_gate continues only with auto_remediate true, dry_run false, a non-empty plan and the bytes inside max_copy_mb. Otherwise reported_only says what would have been copied and why it was not.
  - approve (io.kestra.plugin.core.flow.Pause) waits for a human.
  - replicate calls replicate_reference_tables(shard_transfer_mode). The call is cluster-wide: it copies every under-replicated reference table to every short worker in one go, rather than per table or per node.
  - recoverage and reverify read the metadata again. reverify wraps the assessment with the before numbers so fixed is decided in SQL, not by comparing rendered strings.
  - verify fails the run unless every active primary worker now holds every reference table.
errors alerts Slack when the coordinator cannot be read.

What you get

Per-worker reference table coverage, with the short workers and the missing tables named.
The bytes a copy would move, before anything is copied, plus a budget that refuses an unexpectedly large one.
A cluster-wide replication that runs only after a human resumes it.
A per-worker re-read of the placement metadata that fails the run when the copy did not land.

Who it's for

Teams scaling a Citus cluster out, where citus_add_node is the step people remember and this is the step they do not.
Platform teams whose rebalance onto a new node fails with a missing relation error and who need the cause named.
Anyone running multi-tenant Citus where a lookup table is a reference table and a join silently stops being local.

Why orchestrate this with Kestra

Reference table coverage drifts the moment a node joins. Nothing in the cluster complains until a rebalance or a foreign key fails. Kestra reads the metadata on a schedule, sends the short workers to Slack, holds the copy behind an approval and keeps the plan, the byte count and the before and after coverage on the execution. The read-only path is safe to run every night, which is how the gap gets found before the next rebalance does.

Tested end to end

On Kestra 2.0.5 OSS against Citus 12.1 on PostgreSQL 16, one coordinator and two workers, with a local HTTP sink standing in for Slack. Two reference tables (countries, currencies) and one distributed table (orders, 4 shards, 2000 rows) were created while only the first worker was registered, then a second worker was added with citus_add_node so it joined with no reference table placements. Every number below was checked with psql against pg_dist_placement and pg_dist_node on the coordinator, not only from the Kestra logs.

Run	Inputs	Result
1	defaults	`MISSING_REPLICAS`. 2 of 4 placements present, 50.0% coverage, the new worker missing both tables. Alert sent, 2 copies planned at 64 kB, nothing applied
2	`auto_remediate: true`, `dry_run: true`	Same plan, nothing applied. `psql` confirms 2 placements on the first worker and 0 on the second
3	`auto_remediate: true`, `dry_run: false`	Paused, resumed, copied with `block_writes`. 2 missing placements before, 0 after, coverage 50.0% to 100.0%. `psql` confirms both reference tables on both workers and the first worker's 4 distributed shards untouched
4	defaults, healthy cluster	`OK`, 4 of 4 placements, no alert, nothing planned
5	`auto_remediate: true`, `dry_run: false`, approval killed instead of resumed	Execution killed at the `Pause`. `replicate` never ran and `psql` still showed 0 placements on the new worker
6	`auto_remediate: true`, `dry_run: false`, `max_copy_mb: 0`	`reported_only` refused the copy over the budget and named the byte count. No `Pause`, no copy
7	`shard_transfer_mode: auto`	`replicate` raised `logical decoding requires wal_level >= logical` on a default `wal_level`. The run failed, the `errors` block alerted Slack and the placement counts were unchanged
8	`auto_remediate: true`, `dry_run: false`, one placement row forced to `shardstate = 3`	The re-verify failure. `replicate_reference_tables` returned void and changed nothing. The re-read found 1 placement still missing, `remediation_failed` fired with the before and after numbers and the `errors` block alerted Slack
9	unreachable coordinator	`coverage` failed on connect and the `errors` block alerted Slack

Prerequisites

A Citus cluster, version 11 or later, reached through its coordinator. replicate_reference_tables exists from Citus 9.3. Its shard_transfer_mode argument needs Citus 11.
At least one reference table, created with create_reference_table.
A role that can read pg_dist_* and call citus_shard_sizes() for the detection path. The copy needs a role that can write on the workers, usually a superuser.
shard_transfer_mode: auto or force_logical needs wal_level = logical on the coordinator. Use block_writes when it is not set, which is the default here.
replicate_reference_tables blocks until the copy finishes, so size max_copy_mb and the run window to the reference tables you actually have.

Secrets

POSTGRES_USERNAME / POSTGRES_PASSWORD: credentials for the coordinator.
SLACK_WEBHOOK_URL: used by alert and the errors block.
In Kestra OSS, provide each one as a base64-encoded environment variable prefixed SECRET_.

Inputs

Input	Default	Purpose
`jdbc_url`	`jdbc:postgresql://localhost:5432/citus`	Coordinator
`shard_transfer_mode`	`block_writes`	`block_writes`, `auto` or `force_logical`
`max_copy_mb`	`1024`	Refuse a copy larger than this
`auto_remediate`	`false`	Allow the copy to run
`dry_run`	`true`	Must be `false` before anything is copied

Quick start

Add the POSTGRES_USERNAME, POSTGRES_PASSWORD and SLACK_WEBHOOK_URL secrets.
Import the flow and set jdbc_url to your coordinator.
Run it once with the defaults and read the per-worker coverage and the planned copies.
Run it with auto_remediate: true and dry_run: false, then resume the approve task to copy.
Enable the daily trigger, which always runs in safe mode.

How to extend

Chain it in front of a rebalance flow, so a new node gets its reference tables before any shard is moved onto it.
Add a run_command_on_workers step that checks the shard table really exists on each worker, since replicate_reference_tables trusts the metadata and will not repair a placement row whose table was dropped by hand.
Lower max_copy_mb to the size of your largest reference table, so a table that grows past it shows up as a refusal rather than as a long copy.
Send the per-worker coverage to your metrics store on every run, so the gap between a node joining and the copy landing becomes a number you can alert on.
Run it after every citus_add_node from the same flow that adds the node, with auto_remediate on. Keep the nightly schedule as the backstop.

Citus Reference Table Replica Guard

Findings

Safety

How it works

What you get

Who it's for

Why orchestrate this with Kestra

Tested end to end

Prerequisites

Secrets

Inputs

Quick start

How to extend

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

New to Kestra?

Citus Reference Table Replica Guard

Findings

Safety

How it works

What you get

Who it's for

Why orchestrate this with Kestra

Tested end to end

Prerequisites

Secrets

Inputs

Quick start

How to extend

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

More Infrastructure Blueprints

ChaosBlade Experiment Fault Injection State Sentinel

Kubernetes Chaos Game Day with Parallel Blast-Radius Watch

CI/CD with GitHub Webhooks, Process Pull Request Events and Test Locally with JSON Payloads

Cilium eBPF Map Pressure and Kernel Conntrack Saturation Sentinel

Score change requests on risk and build the CAB agenda, pre-approving the low-risk ones

Chrony NTP Stratum Dispersion and Clock Drift Sentinel

More Data Blueprints

Build a Chat-With-Your-Data System Using Elasticsearch and OpenAI (RAG)

Score change requests on risk and build the CAB agenda, pre-approving the low-risk ones

Chroma Vector Store Dangling Document and Quarantine Eviction Guard

Citus Inactive Worker Reactivate Guard

Chroma Collection Distance Metric Validator

Incremental Docs-to-Chroma RAG Sync with Deletions and a Retrieval Smoke Test

New to Kestra?