Reconcile a Redis cache against PostgreSQL with Kestra

id: redis-cache-vs-postgres-staleness-reconcile-guard
namespace: company.team

description: |
  Reconcile a Redis cache against its PostgreSQL source of truth. PostgreSQL is read first for the
  authoritative keyset and a per-row version, then exactly those keys are probed in Redis. A cached
  entry whose version is behind the database is STALE. A key whose row is gone from the database
  is ORPHANED. On auto_remediate the orphans are evicted and the stale keys refreshed from the
  database, then the same keys are read again and the run fails if any divergence remains.

inputs:
  - id: pg_host
    type: STRING
    displayName: PostgreSQL host
    defaults: "localhost"

  - id: pg_port
    type: INT
    displayName: PostgreSQL port
    defaults: 5432

  - id: pg_db
    type: STRING
    displayName: PostgreSQL database
    defaults: "postgres"

  - id: pg_user
    type: STRING
    displayName: PostgreSQL user
    defaults: "postgres"

  - id: pg_password
    type: STRING
    displayName: PostgreSQL password
    defaults: "{{ secret('POSTGRES_PASSWORD') }}"

  - id: pg_table
    type: STRING
    displayName: Source of truth table
    description: Table with the authoritative rows. Needs an id column, a version
      column and a soft-delete timestamp.
    defaults: "catalog_item"

  - id: redis_host
    type: STRING
    displayName: Redis host
    defaults: "localhost"

  - id: redis_port
    type: INT
    displayName: Redis port
    defaults: 6379

  - id: redis_password
    type: STRING
    displayName: Redis password
    description: Leave empty for an instance with no password.
    defaults: "{{ secret('REDIS_PASSWORD') }}"

  - id: key_prefix
    type: STRING
    displayName: Cache key prefix
    description: The cache key for a row is this prefix plus the row id, for example
      item:42.
    defaults: "item:"

  - id: batch_limit
    type: INT
    displayName: Rows per run
    description: Upper bound on the keyset read in one run, so a huge table cannot
      build an unbounded command.
    defaults: 5000

  - id: network
    type: STRING
    displayName: Docker network for redis-cli
    defaults: ""

  - id: auto_remediate
    type: BOOL
    displayName: Auto remediate
    defaults: false

  - id: dry_run
    type: BOOL
    displayName: Dry run
    description: Must be false, together with auto_remediate, for any eviction or
      refresh to run.
    defaults: true

variables:
  jdbc_url: "jdbc:postgresql://{{ inputs.pg_host }}:{{ inputs.pg_port }}/{{
    inputs.pg_db }}"
  redis_url: "redis://{{ inputs.redis_password == '' ? '' : ':' ~
    inputs.redis_password ~ '@' }}{{ inputs.redis_host }}:{{ inputs.redis_port
    }}"
  mget_prog: >-
    .prefix as $p | "MGET " + ([.rows[] | $p + (.item_id|tostring)] | join(" "))
  affected_mget_prog: >-
    "MGET " + ([.affected[] | .key] | join(" "))
  diff_prog: >-
    .prefix as $p | (.mget | split("\\n")) as $L | [ .rows | to_entries[]
        | .key as $i | .value as $r
        | ($p + ($r.item_id|tostring)) as $key
        | ($L[$i] // "") as $line
        | ($line | length > 0) as $present
        | (if $present then (try ($line | fromjson | .v | tonumber) catch null) else null end) as $cv
        | ($r.version | tonumber) as $dv
        | if ($r.deleted and $present) then {code: "ORPHAN", key: $key, item_id: $r.item_id, db_version: $dv, cached_version: $cv}
          elif (($r.deleted | not) and $present and ($cv != $dv)) then {code: "STALE", key: $key, item_id: $r.item_id, db_version: $dv, cached_version: $cv, fresh: ({v: $dv, name: $r.name} | tojson)}
          else empty end ]
  reverify_prog: >-
    (.mget | split("\\n")) as $L | [ .affected | to_entries[]
        | .key as $i | .value as $a
        | ($L[$i] // "") as $line
        | ($line | length > 0) as $present
        | if ($a.kind == "ORPHAN" and $present) then {key: $a.key, problem: "still_cached"}
          elif ($a.kind == "STALE" and ((if $present then (try ($line | fromjson | .v | tonumber) catch -1) else -1 end) != ($a.expect | tonumber))) then {key: $a.key, problem: "version_mismatch"}
          else empty end ]

tasks:
  - id: pg_read
    type: io.kestra.plugin.jdbc.postgresql.Query
    description: Read-only. The authoritative keyset, the version of each row and
      whether it is deleted.
    url: "{{ render(vars.jdbc_url) }}"
    username: "{{ inputs.pg_user }}"
    password: "{{ inputs.pg_password }}"
    fetchType: FETCH
    sql: >-
      SELECT item_id, version, (deleted_at IS NOT NULL) AS deleted, name FROM {{
      inputs.pg_table }} ORDER BY item_id LIMIT {{ inputs.batch_limit }}

  - id: keyset
    type: io.kestra.plugin.core.output.OutputValues
    description: Build one MGET over exactly the keys PostgreSQL knows about, in id order.
    values:
      mget: >-
        {{ {'rows': outputs.pg_read.rows, 'prefix': inputs.key_prefix} |
        jq(vars.mget_prog) | first }}

  - id: redis_read
    type: io.kestra.plugin.redis.cli.RedisCLI
    description: Read-only. One MGET returns the cached value of every source key,
      nil for a miss.
    host: "{{ inputs.redis_host }}"
    port: "{{ inputs.redis_port }}"
    password: "{{ inputs.redis_password }}"
    taskRunner:
      type: io.kestra.plugin.scripts.runner.docker.Docker
      networkMode: "{{ inputs.network }}"
    commands:
      - "{{ outputs.keyset.values.mget }}"

  - id: diff
    type: io.kestra.plugin.core.output.OutputValues
    description: Diff each source row against its cached value. Orphan when the row
      is deleted but cached, stale when the cached version is not the database
      version.
    values:
      findings: >-
        {{ {'rows': outputs.pg_read.rows, 'mget':
        outputs.redis_read.vars.command_1, 'prefix': inputs.key_prefix} |
        jq(vars.diff_prog) | first | toJson }}

  - id: plan
    type: io.kestra.plugin.core.output.OutputValues
    description: The orphan keys to evict, the stale keys to refresh and the full
      affected set to re-read.
    values:
      orphans: >-
        {{ fromJson(outputs.diff.values.findings) | jq('[.[] |
        select(.code=="ORPHAN") | .key]') | first | toJson }}
      stales: >-
        {{ fromJson(outputs.diff.values.findings) | jq('[.[] |
        select(.code=="STALE") | {key: .key, value: .fresh}]') | first | toJson
        }}
      affected: >-
        {{ fromJson(outputs.diff.values.findings) | jq('[.[] | {key: .key, kind:
        .code, expect: (if .code=="STALE" then .db_version else null end)}]') |
        first | toJson }}
      affected_mget: >-
        {{ {'affected': fromJson(outputs.diff.values.findings)} |
        jq(vars.affected_mget_prog) | first }}

  - id: summary
    type: io.kestra.plugin.core.log.Log
    message: >-
      Reconcile of {{ inputs.redis_host }} against {{ render(vars.jdbc_url) }}:
      {{ fromJson(outputs.diff.values.findings) | length }} divergences ({{
      fromJson(outputs.plan.values.orphans) | length }} orphan, {{
      fromJson(outputs.plan.values.stales) | length }} stale). {{
      fromJson(outputs.diff.values.findings) | jq('map("\(.code) \(.key)
      db=\(.db_version) cache=\(.cached_version // "miss")") | join("; ")') |
      first }}

  - id: route
    type: io.kestra.plugin.core.flow.If
    condition: "{{ fromJson(outputs.diff.values.findings) | length > 0 }}"
    then:
      - id: alert
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        payload: |
          {{ {'text': 'Redis vs PostgreSQL reconcile on ' ~ inputs.redis_host ~ ': ' ~ (fromJson(outputs.diff.values.findings) | jq('map("\(.code) \(.key)") | join(", ")') | first) ~ '.' ~ (inputs.auto_remediate and not inputs.dry_run ? ' Remediating now.' : ' No change applied (auto_remediate=' ~ inputs.auto_remediate ~ ', dry_run=' ~ inputs.dry_run ~ ').')} | toJson }}

      - id: remediate_gate
        type: io.kestra.plugin.core.flow.If
        condition: "{{ inputs.auto_remediate and not inputs.dry_run }}"
        then:
          - id: evict_gate
            type: io.kestra.plugin.core.flow.If
            condition: "{{ fromJson(outputs.plan.values.orphans) | length > 0 }}"
            then:
              - id: evict_orphans
                type: io.kestra.plugin.redis.string.Delete
                description: Evict the keys whose row no longer exists in the source of truth.
                url: "{{ render(vars.redis_url) }}"
                keys: "{{ fromJson(outputs.plan.values.orphans) }}"
                failedOnMissing: false

          - id: refresh_stale
            type: io.kestra.plugin.core.flow.Loop
            description: Remediate stale keys one at a time, writing the current value from
              the source of truth. Serial so writes do not race.
            concurrencyLimit: 1
            values: "{{ outputs.plan.values.stales }}"
            tasks:
              - id: set_fresh
                type: io.kestra.plugin.redis.string.Set
                url: "{{ render(vars.redis_url) }}"
                key: "{{ fromJson(item.value).key }}"
                value: "{{ fromJson(item.value).value }}"
                serdeType: STRING

          - id: reverify_read
            type: io.kestra.plugin.redis.cli.RedisCLI
            description: Read the affected keys again, so the proof comes from Redis itself
              and not from the write tasks.
            host: "{{ inputs.redis_host }}"
            port: "{{ inputs.redis_port }}"
            password: "{{ inputs.redis_password }}"
            taskRunner:
              type: io.kestra.plugin.scripts.runner.docker.Docker
              networkMode: "{{ inputs.network }}"
            commands:
              - "{{ outputs.plan.values.affected_mget }}"

          - id: reverify_diff
            type: io.kestra.plugin.core.output.OutputValues
            description: An orphan still cached or a stale key still off the database
              version is unresolved.
            values:
              remaining: >-
                {{ {'affected': fromJson(outputs.plan.values.affected), 'mget':
                outputs.reverify_read.vars.command_1} | jq(vars.reverify_prog) |
                first | toJson }}

          - id: verify_result
            type: io.kestra.plugin.core.flow.If
            condition: "{{ fromJson(outputs.reverify_diff.values.remaining) | length > 0 }}"
            then:
              - id: remediation_failed
                type: io.kestra.plugin.core.execution.Fail
                errorMessage: >-
                  Reconcile did not converge: {{
                  fromJson(outputs.reverify_diff.values.remaining) |
                  jq('map("\(.key) \(.problem)") | join(", ")') | first }}.
                  Check that the connecting Redis user may run DEL and SET on
                  these keys.
            else:
              - id: remediated
                type: io.kestra.plugin.core.log.Log
                message: >-
                  Reconciled and verified: evicted {{
                  fromJson(outputs.plan.values.orphans) | length }} orphan keys,
                  refreshed {{ fromJson(outputs.plan.values.stales) | length }}
                  stale keys, 0 divergence left.
    else:
      - id: clean
        type: io.kestra.plugin.core.log.Log
        message: "Cache matches the source of truth: every live row is cached at its
          current version and no deleted row is still cached."

errors:
  - id: alert_flow_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    payload: |
      {{ {'text': 'Redis vs PostgreSQL reconcile guard failed on ' ~ inputs.redis_host ~ ' (execution ' ~ execution.id ~ '). Check the logs.'} | toJson }}

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

A read-through or write-behind cache drifts from its database in two ways that no single system can see on its own. A row deleted in PostgreSQL can leave its key behind in Redis, so the application keeps serving a record that no longer exists. A row updated in PostgreSQL can leave an older copy in the cache, so the application serves a value that is behind the truth.

This blueprint reconciles the two. It reads the authoritative keyset and a per-row version from PostgreSQL first, then probes exactly those keys in Redis with one MGET. A SCAN of Redis is never used, because its order is not stable and its counts drift between runs, so a cache diff built on it is not reproducible. Driving the keyset from the source of truth makes every run deterministic.

It reports two kinds of divergence, alerts Slack and on request evicts the orphans with DEL and refreshes the stale keys with SET from the database value, then reads the same keys again and fails the run if anything is still wrong.

This blueprint was created by zkasuran.

Findings

Finding	Meaning	Fix
`ORPHAN`	The row is deleted in PostgreSQL (its `deleted_at` is set) but the key is still cached	`DEL` the key
`STALE`	The row is live but the cached version is not the database version	`SET` the key to the current database value

A live row with no cached key is a cache miss, not a divergence, so it is counted and left alone. A deleted row with no cached key is already correct. The source of truth records deletions with a deleted_at timestamp, which is why the orphan set can be derived from PostgreSQL rather than from a scan of Redis.

How it works

pg_read (io.kestra.plugin.jdbc.postgresql.Query, read-only) returns every row with its id, its version and whether it is deleted, in id order, up to batch_limit rows.
keyset builds one MGET over the key of each row (key_prefix plus the id). redis_read (io.kestra.plugin.redis.cli.RedisCLI) runs it, so the cached value of every source key comes back in one round trip, with a nil for a miss.
diff compares the two by position: an orphan is a deleted row still cached, a stale key is a live row whose cached version is not the database version. plan splits the result into keys to evict, keys to refresh and the full affected set. summary logs it.
route (io.kestra.plugin.core.flow.If):
- clean: logs it, nothing to do.
- divergence: alert posts to Slack.
  - remediate_gate runs only when auto_remediate is true and dry_run is false.
  - evict_orphans (io.kestra.plugin.redis.string.Delete) removes the orphan keys.
  - refresh_stale (io.kestra.plugin.core.flow.Loop, concurrencyLimit: 1) writes each stale key back from the database value with io.kestra.plugin.redis.string.Set.
  - reverify_read reads the affected keys again and verify_result fails the run if any orphan is still cached or any refreshed key is still off the database version.
errors alerts Slack when the flow fails.

What you get

A per-key reconcile across two live systems, with PostgreSQL as the source of truth, not a diff of two Redis scans.
The orphans and the stale keys named, with the database version next to the cached version.
Eviction and refresh applied only behind two flags, one key at a time for the writes, then a second read of the same keys as proof.

Who it's for

Teams that cache PostgreSQL rows in Redis and have served a deleted record or a stale value.
SREs who want a scheduled check that the cache has not drifted from the database.
Teams evaluating the PostgreSQL and Redis plugins who want one flow that reads from both and writes to one.

Why orchestrate this with Kestra

A cache drifts quietly and the drift spans two systems that each look healthy alone. Kestra reads the authoritative keyset from PostgreSQL, probes exactly those keys in Redis, sends the divergence to Slack, evicts and refreshes only behind auto_remediate and dry_run, verifies the result and keeps a history of every reconcile.

Tested end to end

On Kestra 2.0.5 OSS against PostgreSQL 16 (w26-rec-pg) and Redis 7 (w26-rec-redis, password set), with a local HTTP sink standing in for Slack. The source table held 8 rows: 5 live (item:1 v5, item:2 v5, item:3 v7, item:4 v3, item:5 v2) and 3 deleted (item:6, item:7, item:8). The cache held item:1 v5, item:2 v5, item:3 v4, item:4 v2, item:6 v9, item:7 v4. So item:3 plus item:4 were stale, item:6 plus item:7 were orphaned, item:5 was a plain miss and item:8 was already gone.

Run	Inputs	Result
1	defaults	4 divergences (orphan `item:6`, `item:7`; stale `item:3` db=7 cache=4, `item:4` db=3 cache=2). Alert sent. `redis-cli` confirmed the 6 keys unchanged
2	`auto_remediate: true`, `dry_run: true`	Same 4 divergences. The gate stayed closed. `redis-cli` confirmed no key changed
3	`auto_remediate: true`, `dry_run: false`	Evicted 2 orphans, refreshed 2 stale keys, re-read reported 0 remaining. `redis-cli` after: `item:6` plus `item:7` gone, `item:3` now v7, `item:4` now v3, `item:1` plus `item:2` untouched. `psql` confirmed the table unchanged
4	clean cache, defaults	0 divergences, clean branch, no alert. The miss (`item:5`) plus the already-evicted delete (`item:8`) were not flagged
5	unreachable PostgreSQL host	`pg_read` failed, the run failed, the `errors` block posted to Slack
6	`auto_remediate: true`, `dry_run: false`, with a writer re-adding orphan `item:6` during the run	Re-read still found `item:6` cached. `verify_result` failed the run rather than reporting success, which is the point of the re-verify step

Detector proof: the degraded cache reported 4 and the clean cache reported 0, both matching the counts measured straight from psql plus redis-cli.

Prerequisites

A PostgreSQL table that is the source of truth, with an item_id, a version that increases on every update and a deleted_at timestamp set when a row is logically removed.
A Redis instance that caches one JSON value per row under key_prefix plus the id, carrying a v field with the version it was written at.
For the audit, a PostgreSQL role that can read the table. For remediation, a Redis user allowed to run DEL and SET.
Local testing:
- docker run -d --name pg --network kw7 -e POSTGRES_PASSWORD=pw postgres:16;
- docker run -d --name cache --network kw7 redis:7-alpine redis-server --requirepass pw;
- seed a few rows, cache some at the right version, cache others a version behind, then leave a cached key for a deleted row.

Secrets

POSTGRES_PASSWORD: the password for pg_user, read by every query.
REDIS_PASSWORD: the Redis password, left empty for an instance with no password.
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
`pg_host` / `pg_port` / `pg_db` / `pg_user`	localhost, 5432, postgres, postgres	PostgreSQL connection
`pg_password`	`secret('POSTGRES_PASSWORD')`	PostgreSQL password
`pg_table`	`catalog_item`	Source of truth table
`redis_host` / `redis_port`	localhost, 6379	Redis connection
`redis_password`	`secret('REDIS_PASSWORD')`	Redis password, empty for none
`key_prefix`	`item:`	Cache key is this prefix plus the row id
`batch_limit`	`5000`	Upper bound on the keyset read in one run
`network`	empty	Docker network for the `redis-cli` container
`auto_remediate`	`false`	Allow `DEL` and `SET`
`dry_run`	`true`	Must be `false` for a change to apply

Quick start

Add the POSTGRES_PASSWORD, REDIS_PASSWORD and SLACK_WEBHOOK_URL secrets.
Import the flow and set the connection inputs and key_prefix to match your cache.
Run it once and read the orphan and stale keys.
Run it with auto_remediate: true and dry_run: false to evict and refresh, then read the remediated log for the proof.
Enable the hourly trigger, which always runs in safe mode.

How to extend

Compare on updated_at instead of a numeric version by returning it from the query and storing it in the cached value.
Page through a large table by ordering on the id and carrying a cursor in a KV store.
Reconcile several caches by looping over a list of prefixes and hosts.
Add a staleness budget so a key behind by one version is tolerated for a short window.

Redis Cache vs PostgreSQL Staleness Reconcile Guard

Findings

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?

Redis Cache vs PostgreSQL Staleness Reconcile Guard

Findings

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 Data Blueprints

AI Data Quality Anomaly Advisor and SQL Remediation Generator

Ingest Zendesk data into DuckDB using dlt

Ingest a public JSON API into DocumentDB

AI Document Data Extraction: Parse PDFs and Web Pages into Structured JSON with OpenAI

AI-Powered Inventory Demand Forecasting, Auto-Generate Purchase Orders with OpenAI

Aerospike Stop-Writes Headroom Sentinel

More Infrastructure Blueprints

ADLS World-Writable ACL Sentinel

AI Incident Post-Mortem and Root Cause Generator

Automatically Review GitHub Pull Requests with OpenAI GPT-4o and Post Comments

AI-Powered Incident Response Automation

Aerospike Stop-Writes Headroom Sentinel

Autonomous Database Incident Investigation Agent with DuckDB and AI

New to Kestra?