Redis Cache Generation Invalidation Gate with Kestra

id: redis-cache-generation-invalidation-gate
namespace: company.team
description: |
  Compare a Redis "generation" key against the last value this flow saw and,
  when it changed, broadcast an invalidation message on a Redis Pub/Sub
  channel so every subscribed app instance flushes its local cache.
  auto_remediate and dry_run are two independent gates: nothing is
  published, and the last-seen marker is not advanced, unless both allow it.

triggers:
  - id: periodic_generation_check
    type: io.kestra.plugin.core.trigger.Schedule
    description: Check every 5 minutes, frequent enough that a cache invalidation is
      not stale by the time it fans out. Shipped disabled; enable it once
      redis_url/generation_key point at a real instance and a manual run has
      been verified. Always fires in safe mode (auto_remediate false, dry_run
      true) regardless of this flow's own defaults.
    cron: "*/5 * * * *"
    disabled: true
    inputs:
      auto_remediate: "false"
      dry_run: "true"

inputs:
  - id: redis_url
    type: STRING
    displayName: "Redis Connection URL"
    defaults: "redis://localhost:6379/0"
    description: Full Redis connection string, passed directly to the url property
      shared by every io.kestra.plugin.redis.string/pubsub task.

  - id: generation_key
    type: STRING
    displayName: "Generation Key"
    defaults: "cache:generation"
    description: Key the data pipeline is expected to bump (for example to a
      timestamp or a counter) every time the underlying source data changes.

  - id: last_seen_key
    type: STRING
    displayName: "Last Seen Generation Key"
    defaults: "cache:last_seen_generation"
    description: Key this flow uses to remember the generation value it last acted
      on, so repeated runs only broadcast once per actual change.

  - id: invalidation_channel
    type: STRING
    displayName: "Invalidation Channel"
    defaults: "cache:invalidate"
    description: Redis Pub/Sub channel that subscribed application instances listen
      on to know when to flush their local cache.

  - id: auto_remediate
    type: SELECT
    displayName: "Auto Remediate"
    defaults: "false"
    values:
      - "false"
      - "true"
    description: If "true" and the generation changed, consider broadcasting an
      invalidation message (still gated by dry_run below). If "false", the
      change is only reported.

  - id: dry_run
    type: SELECT
    displayName: "Dry Run"
    defaults: "true"
    values:
      - "true"
      - "false"
    description: If "true", the broadcast is only logged, PUBLISH never runs and
      last_seen_key is never advanced. Must be explicitly "false", together with
      auto_remediate "true", for the broadcast to actually happen.

tasks:
  - id: read_cache_generation
    type: io.kestra.plugin.redis.string.Get
    description: Reads the current generation marker. failedOnMissing defaults to
      false, so a key that was never set returns a null data output instead of
      failing the task - this blueprint treats that as a distinct, documented
      pitfall rather than a crash.
    url: "{{ inputs.redis_url }}"
    key: "{{ inputs.generation_key }}"

  - id: read_last_seen_generation
    type: io.kestra.plugin.redis.string.Get
    description: Reads what this flow last acted on. On the very first run this key
      does not exist yet; GET returns null and SerdeType.deserialize(null)
      returns null (verified from source), so this comparison does not need
      special-casing for a missing key.
    url: "{{ inputs.redis_url }}"
    key: "{{ inputs.last_seen_key }}"

  - id: evaluate_generation_drift
    type: io.kestra.plugin.core.flow.If
    description: Drift means the current generation differs from what was last acted
      on, including the first-run case where last_seen_key does not exist yet
      (null != a real value is true).
    condition: "{{ outputs.read_cache_generation.data !=
      outputs.read_last_seen_generation.data }}"
    then:
      - id: route_remediation
        type: io.kestra.plugin.core.flow.Switch
        description: auto_remediate only decides whether a broadcast is considered at
          all; dry_run, checked separately below, decides whether PUBLISH
          actually runs.
        value: "{{ inputs.auto_remediate }}"
        cases:
          "true":
            - id: check_dry_run
              type: io.kestra.plugin.core.flow.If
              condition: "{{ inputs.dry_run == \"false\" }}"
              then:
                - id: broadcast_invalidation
                  type: io.kestra.plugin.redis.pubsub.Publish
                  description: Publishes the new generation value as the invalidation message, so
                    subscribers know not only that they must invalidate but what
                    the new generation is. count in the output is the number of
                    values this task published, not the number of subscribers
                    that received them - Lettuce's own per-message subscriber
                    count is computed internally but discarded, confirmed
                    directly from source.
                  url: "{{ inputs.redis_url }}"
                  channel: "{{ inputs.invalidation_channel }}"
                  from:
                    - "{{ outputs.read_cache_generation.data }}"
                  serdeType: STRING
                - id: update_last_seen_generation
                  type: io.kestra.plugin.redis.string.Set
                  description: Advance the marker only after a successful broadcast, so a failed
                    PUBLISH leaves last_seen_key unchanged and the next run
                    retries the same generation.
                  url: "{{ inputs.redis_url }}"
                  key: "{{ inputs.last_seen_key }}"
                  value: "{{ outputs.read_cache_generation.data }}"
                - id: log_invalidation_broadcast
                  type: io.kestra.plugin.core.log.Log
                  message: "Broadcast invalidation for generation {{
                    outputs.read_cache_generation.data }} on channel {{
                    inputs.invalidation_channel }}. Values published: {{
                    outputs.broadcast_invalidation.count }}."
              else:
                - id: log_dry_run_invalidation
                  type: io.kestra.plugin.core.log.Log
                  message: "DRY RUN: would broadcast generation {{
                    outputs.read_cache_generation.data }} on {{
                    inputs.invalidation_channel }} (dry_run is true). No message
                    published, last_seen_key not advanced."
          "false":
            - id: log_remediation_disabled
              type: io.kestra.plugin.core.log.Log
              message: "auto_remediate is false: generation changed from {{
                outputs.read_last_seen_generation.data }} to {{
                outputs.read_cache_generation.data }}, but no broadcast was
                attempted."
      - id: alert_generation_drift
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        description: Always alert on a detected change, independent of whether a
          broadcast actually ran.
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        payload: |
          {
            "text": "Redis cache generation gate: {{ inputs.generation_key }} changed from {{ outputs.read_last_seen_generation.data }} to {{ outputs.read_cache_generation.data }}. auto_remediate: {{ inputs.auto_remediate }}. dry_run: {{ inputs.dry_run }}. Broadcast sent: {{ outputs.broadcast_invalidation is defined }}. Execution {{ execution.id }}."
          }
    else:
      - id: log_generation_unchanged
        type: io.kestra.plugin.core.log.Log
        message: "{{ inputs.generation_key }} is unchanged at {{
          outputs.read_cache_generation.data }}. No invalidation needed."

  - id: log_audit_complete
    type: io.kestra.plugin.core.log.Log
    description: Final status line for every run, independent of which branch fired,
      so the execution history always shows the headline values.
    message: "Generation check complete. current={{
      outputs.read_cache_generation.data }}. last_seen={{
      outputs.read_last_seen_generation.data }}. auto_remediate={{
      inputs.auto_remediate }}. dry_run={{ inputs.dry_run }}."

errors:
  - id: alert_execution_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    description: Alert when the flow itself fails outside the handled branches, for
      example an unreachable Redis instance.
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    payload: |
      {
        "text": "Redis cache generation gate FAILED in execution {{ execution.id }} for {{ inputs.generation_key }} on {{ inputs.redis_url }}. The generation check did not complete; no invalidation decision can be trusted from this run."
      }

Many systems run several app instances, each holding its own in-memory cache, with no shared signal for when the underlying data changes. The usual fix is a generation or version marker the data pipeline bumps on every change, paired with a fan-out broadcast so every instance knows to invalidate at the same moment. This blueprint is that broadcast, built as a gate rather than a bare PUBLISH: it remembers the last generation it acted on, only considers broadcasting when that value actually changed, and keeps a human-reviewable path between "change detected" and "every subscriber just got told to flush its cache".

How it works

periodic_generation_check (io.kestra.plugin.core.trigger.Schedule) runs every 5 minutes, always forcing auto_remediate: "false" and dry_run: "true" through its own inputs: override regardless of this flow's defaults. Shipped disabled so you can validate a manual run first.
read_cache_generation and read_last_seen_generation (both io.kestra.plugin.redis.string.Get) read the current generation marker and the value this flow last acted on.
evaluate_generation_drift (io.kestra.plugin.core.flow.If) flags a change whenever the two differ - this also covers the first run, where last_seen_key does not exist yet and GET returns null.
- then (changed): route_remediation (io.kestra.plugin.core.flow.Switch on auto_remediate) either considers a broadcast ("true", gated again by dry_run through check_dry_run) or only logs that it is disabled ("false"). When both gates allow it, broadcast_invalidation (io.kestra.plugin.redis.pubsub.Publish) publishes the new generation value to invalidation_channel, then update_last_seen_generation (io.kestra.plugin.redis.string.Set) advances the marker so the same change is not re-broadcast next run. alert_generation_drift always fires to Slack regardless of which path ran.
- else (unchanged): log_generation_unchanged records a clean check.
log_audit_complete always runs last, printing the headline values regardless of which branch fired.
The errors block alerts Slack separately if the flow itself fails - an unreachable Redis instance must not read as "nothing changed".

What you get

A change-triggered broadcast instead of an unconditional one - subscribers are only told to invalidate when the generation actually moved.
A last-seen marker that only advances after a successful broadcast, so a failed PUBLISH leaves the next run free to retry the same generation.
Two independent gates (auto_remediate, dry_run) between "change detected" and "message actually published".
A final status log every run, changed or not, so the execution history doubles as a generation-change timeline.

Who it's for

Platform teams running multiple app instances with local in-memory caches that need a shared invalidation signal.
Teams who want cache invalidation reviewable and auditable, not a bare redis-cli PUBLISH buried in a deploy script.
Anyone evaluating the Redis plugin who wants a worked example of pubsub.Publish paired with a change-detection gate.

Why orchestrate this with Kestra

A raw PUBLISH has no memory of the last value it saw, no threshold or change logic, and no notification path of its own. Kestra supplies the schedule, the change detection as a first-class branch, two independent authorization gates before anything is broadcast, and an execution history that shows exactly when the generation changed and whether subscribers were told.

Prerequisites

A reachable Redis instance that subscribers already listen to on invalidation_channel.
A data pipeline (or anything else) that bumps generation_key whenever the data it governs changes.
A Slack incoming webhook for alerts.

Local testing: docker run -d --name redis-gate -p 6381:6379 redis:7-alpine docker network connect YOUR_KESTRA_NETWORK redis-gate (only needed if the Kestra Worker runs in a separate Docker network than this container; set redis_url to redis://redis-gate:6379/0 from inside that network, or redis://localhost:6381/0 from the host).

Secrets

SLACK_WEBHOOK_URL: Slack incoming webhook used by alert_generation_drift and the errors block.
This blueprint's redis_url assumes an unauthenticated local instance, matching the plugin's own documentation examples. For a password-protected instance, use redis://:{{ secret('REDIS_PASSWORD') }}@<host>:<port>/0 as the redis_url value - never place the password directly in the input.
In Kestra OSS (no Enterprise secrets backend), secrets are supplied as environment variables prefixed SECRET_, base64-encoded, and read back in flows with {{ secret('NAME') }} - for example SECRET_SLACK_WEBHOOK_URL=$(echo -n 'https://hooks.slack.com/...' | base64). This keeps credentials out of the flow YAML but, per Kestra's own documentation, offers no encryption at rest or access control beyond the host environment; use the Enterprise secrets backend for stronger guarantees.

Inputs

redis_url (STRING, default redis://localhost:6379/0): connection string shared by every Redis task.
generation_key (STRING, default cache:generation): marker bumped by the data pipeline on every change.
last_seen_key (STRING, default cache:last_seen_generation): marker this flow uses to avoid re-broadcasting the same change.
invalidation_channel (STRING, default cache:invalidate): Pub/Sub channel subscribers listen on.
auto_remediate (SELECT: "false", "true"; default "false"): must be "true" for a broadcast to even be considered.
dry_run (SELECT: "true", "false"; default "true"): must be explicitly "false", together with auto_remediate: "true", for PUBLISH to actually run.

Outputs

outputs.read_cache_generation.data / outputs.read_last_seen_generation.data: the two compared values for every run.
outputs.broadcast_invalidation.count: number of values published (not subscriber count), present only when a broadcast actually ran.

Quick start

Start Redis locally with the command above and set an initial generation: redis-cli -p 6381 SET cache:generation v1.
Add the SLACK_WEBHOOK_URL secret.
Run the flow manually with the defaults and confirm evaluate_generation_drift detects a change on this first run (no last_seen_key exists yet) and, with auto_remediate: "false", log_remediation_disabled logs without publishing.
Run once with auto_remediate: "true" and dry_run: "true" and confirm log_dry_run_invalidation describes the broadcast without sending it.
Run once with auto_remediate: "true" and dry_run: "false", then bump the generation (redis-cli -p 6381 SET cache:generation v2) and run again to confirm a fresh broadcast fires only on the real change, not on an unchanged generation.
Enable periodic_generation_check once you trust the check; it always runs in the safe auto_remediate: "false" / dry_run: "true" mode regardless of what you leave the flow's own defaults set to.

How to extend

Subscribe a real application instance to invalidation_channel (any Redis client's SUBSCRIBE) to see the full loop end to end.
Use io.kestra.plugin.redis.json.Get instead of string.Get if the generation marker is stored as a JSON document rather than a plain string.
Loop over several generation_key/invalidation_channel pairs with io.kestra.plugin.core.flow.Loop to watch more than one cache domain from a single flow.
Publish a structured JSON payload (serdeType: JSON, from a map with the old and new generation) instead of a bare string, if subscribers need more context than the new value alone.

Pitfalls

A generation_key that was never set is indistinguishable from "unchanged". failedOnMissing defaults to false, so a missing key returns data: null rather than failing; if both generation_key and last_seen_key are missing, null != null is false and no drift is reported. Verify the data pipeline is actually writing generation_key before trusting a quiet run.
count on broadcast_invalidation is not a subscriber count. Verified from source: the task sums each PUBLISH call's own subscriber-count return value internally, then discards it and returns 1 per published value regardless. count only tells you how many values were sent, never how many subscribers received them.
last_seen_key only advances after the broadcast task runs. If broadcast_invalidation fails, update_last_seen_generation never executes, so the next run retries the same change instead of silently skipping it.
auto_remediate and dry_run are independent gates, not a single boolean. Both must be "true"/"false" respectively for PUBLISH to run; setting only one leaves the other still blocking.
Switch case keys "true"/"false" are quoted strings. auto_remediate renders as the literal string "true" or "false"; unquoted true:/false: YAML map keys would parse as booleans instead and would not match.

Redis Cache Generation Change Detector and Invalidation Broadcast Gate

How it works

What you get

Who it's for

Why orchestrate this with Kestra

Prerequisites

Secrets

Inputs

Outputs

Quick start

How to extend

Pitfalls

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

New to Kestra?

Redis Cache Generation Change Detector and Invalidation Broadcast Gate

How it works

What you get

Who it's for

Why orchestrate this with Kestra

Prerequisites

Secrets

Inputs

Outputs

Quick start

How to extend

Pitfalls

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

More Infrastructure Blueprints

Kafka Schema Registry Compatibility and Mode Drift Guard

Kafka Schema Registry Orphaned Subject Cleanup Guard

Kafka Schema Registry Subject Version Pruning Guard

Kafka Consumer Lag Sentinel

Kafka Orphaned Topic Reclaim Guard

Block a breaking Avro schema before a Kafka producer deploys

More Data Blueprints

Full OpenSearch Index Export to Internal Storage with Scroll

Search OpenAlex for Related Scholarly Works

Use OpenSearch as a Config Store with Put and Get Roundtrip

Ingest an HTTP JSON Dataset into OpenSearch with Load

Automated OpenLineage Metadata Extraction and Marquez Governance Pipeline

OpenSearch PPL Missing Field Compliance Gate

New to Kestra?