Skip to content

Idempotent (Redis-backed)

pkg/processor/idempotent_redis.go defines the context-aware Idempotent Consumer EIP — for messaging environments that may deliver the same message more than once (Kafka at-least-once, AMQP redelivery, retried HTTP calls). It deduplicates by computing a key from each exchange and rejecting keys already seen.

This file is the V2 evolution of the simpler in-memory repository in routing.go. V2 adds:

  • context.Context propagation (required by Redis client calls)
  • Error returns from every repository method (Redis can fail)
  • A Clear operation
  • A Redis-backed implementation with TTL-based expiry

Processors in this file

DistributedIdempotentRepository interface

The repository contract for V2:

type DistributedIdempotentRepository interface {
    Contains(ctx context.Context, key string) (bool, error)
    Add(ctx context.Context, key string) error
    Remove(ctx context.Context, key string) error
    Clear(ctx context.Context) error
}

Two implementations ship in-tree:

NewMemoryIdempotentRepositoryV2

In-memory DistributedIdempotentRepository backed by sync.Map. Same shape as the legacy MemoryIdempotentRepository (routing.go) but with the new context-aware signature.

No TTL, no size cap — for that, see InMemoryIdempotentRepository which uses a different (non-context) interface.

NewRedisIdempotentRepository

Stores keys in Redis under a configurable prefix with TTL-based expiry.

client := redis.NewClient(&redis.Options{Addr: "localhost:6379"})
repo := processor.NewRedisIdempotentRepository(client, "helix:idem:", 24 * time.Hour)
Constructor arg Description
client A *redis.Client from github.com/redis/go-redis/v9. The caller owns its lifecycle.
keyPrefix Prefix prepended to every key. Use a unique prefix per route to avoid cross-route collisions.
ttl Per-key expiry. Redis evicts keys automatically.

Operations:

Method Redis command Notes
Contains EXISTS Returns true if the prefixed key exists.
Add SET NX EX <ttl> Atomic: only sets if absent. Existing-key case is not an error.
Remove DEL
Clear SCAN + DEL Iterates with cursor batch size 100 — safe for large keysets.

IdempotentConsumerV2

The processor that wraps a downstream pipeline and gates it on uniqueness.

steps:
  - idempotent-consumer:
      starlang: header("X-Message-Id")
      repository: redis
      redis-url: redis://localhost:6379/0
      redis-prefix: "helix:idem:orders:"
      ttl: 24h
      steps:
        - target: postgres://orders
Field Type Default Description
starlang string required Expression yielding the dedup key.
repository string memory memory or redis.
redis-url string required if redis e.g. redis://localhost:6379/0.
redis-prefix string helix:idem: Prefix for all keys in this consumer.
ttl duration 24h TTL for each key.
steps list required Inner pipeline executed for new (non-duplicate) keys.

Constructor:

NewIdempotentConsumerV2(
    expr engine.Expression,
    repo DistributedIdempotentRepository,
    processor engine.Processor,
) *IdempotentConsumerV2

How it works:

  1. Evaluates the key expression. Empty (whitespace-only) keys return an explicit error.
  2. Calls repo.Contains(ctx, key). Any Redis error short-circuits with idempotent consumer v2: contains check: ....
  3. If the key already exists, sets HelixDuplicateMessage = true on the exchange and returns nil — duplicates are silently dropped, not errored.
  4. Otherwise calls repo.Add(ctx, key). If the add fails (Redis unreachable, etc.), propagates the error.
  5. Runs the inner Processor for new keys.

V1 vs V2 — quick comparison

V1 (IdempotentConsumer, routing.go) V2 (this file)
Repository interface IdempotentRepository DistributedIdempotentRepository
AddIfAbsent atomic yes (single method) no (separate Contains + Add)
context.Context no yes
Errors from repo none every method
Redis backend no yes

The DSL reifier picks V2 (idempotent-consumer with repository: redis) when Redis is requested and falls back to V1 (MemoryIdempotentRepository) for the default repository: memory.

Pitfalls

  • Contains + Add is a two-call sequence, not atomic in Redis. Concurrent processors can both observe Contains == false. The Redis Add uses SETNX so the second caller's Add is a no-op, but both will run the inner processor. If you need strict at-most-once execution, key off the SETNX result directly (build a custom processor).
  • Empty keys are rejected to prevent every blank header from collapsing into one bucket.
  • Choose a TTL long enough for your largest expected redelivery window. Too short and duplicates slip through; too long and Redis memory grows.

Examples

Redis-backed dedup with a 1-hour window:

workflows:
  - id: order-ingest
    trigger: { uri: kafka://orders?group=ingest }
    steps:
      - idempotent-consumer:
          starlang: header("X-Order-Id")
          repository: redis
          redis-url: redis://localhost:6379/0
          redis-prefix: "helix:idem:orders:"
          ttl: 1h
          steps:
            - target: postgres://orders

Source

pkg/processor/idempotent_redis.go

See also