core.async Retry Strategy: Avoiding Duplicate Writes Without Idempotency Keys

Question

core.async Retry Strategy: Avoiding Duplicate Writes Without Idempotency Keys
In a ClojureScript application that uses core.async to serialize side effects, a common pattern is to re‑queue a failed operation on a channel to retry later. The goal is to guarantee that the underlying write – e.g. a database insert or an HTTP POST – is performed at most once, even if the message is retried multiple times.
Because core.async channels do not enforce semantic idempotency, the developer must decide how to distinguish a genuine retry from a new operation. One approach is to embed a unique identifier in the message and guard the write with an atom or a transactional flag. Another is to rely on an effect‑handler that checks a flag before performing the side effect. The ambiguity lies in whether the same message can be sent again without creating a duplicate write, and how to enforce that boundary when the channel buffer grows or when the retry logic is triggered from multiple goroutines.
Unresolved decisions include: the best way to generate and propagate an idempotency key across retries; the trade‑off between buffering retries in an unbounded channel versus using a bounded channel with back‑pressure; and whether to treat a retry as a separate event or as an idempotent re‑dispatch of the same event.
What mechanisms can be used to prevent duplicate writes when re‑queuing a message in core.async?
How does the size of a core.async channel buffer affect the risk of duplicate writes during rapid retry cycles?
Is it more reliable to guard writes with an atom flag or to rely on a transactional event‑handler that checks a global idempotency flag?

Tasadduq Burney · Answer

Answering the Core Questions
1. Mechanisms to prevent duplicate writes when re‑queuing a message in core.async?

Database‑level uniqueness constraints: Add a unique index on the columns that identify the logical record (e.g., user‑id + timestamp or a UUID). The write will be rejected on a duplicate attempt, and the retry logic can treat the error as a success.
Conditional inserts/updates: Use SQL clauses such as INSERT … ON CONFLICT DO NOTHING or UPDATE … WHERE version = expected so that the operation only takes effect once.
Per‑attempt identifier embedded in the payload: Generate a lightweight ID (e.g., System/nanoTime or a short hash) and include it in the message. On retry, the same payload is sent; the database will reject duplicates via the unique constraint.
Atomic guard in memory: Maintain an atom that records the last processed ID. Before performing the side effect, atomically check and set the flag. This protects against duplicate writes from the same process.
Transactional event‑handler: Wrap the handler in a transaction that checks a global flag (e.g., a Redis key or a database row) before executing the write. If the flag is already set, the transaction aborts.

2. Impact of core.async channel buffer size on duplicate‑write risk during rapid retries

A large or unbounded buffer can hold many in‑flight retry attempts, increasing the chance that two identical messages reach the write target before the first one has been acknowledged. If the write is not protected by a uniqueness check, this leads to duplicates.
A bounded buffer with back‑pressure forces the producer to wait when the channel is full. This throttles retry bursts and reduces the window where duplicate writes could occur.
However, a too‑small buffer may cause message loss if the consumer is blocked by a slow external system; lost messages mean the retry logic may never re‑attempt the operation, violating at‑least‑once guarantees.

3. Atom flag vs. transactional global idempotency check: which is more reliable?

Atom flag:

Fast, in‑process check; no network hop.
Only protects against duplicate writes from the same process.
Susceptible to process restarts – the flag is lost unless persisted.

Transactional global check (e.g., Redis, database row):

Provides cross‑process, cross‑node idempotency.
Relies on the transaction to enforce atomicity; if the write succeeds, the flag is set in the same transaction.
Adds network latency but gives stronger guarantees, especially in distributed deployments.

In most production scenarios where the same operation can be retried from multiple nodes, the transactional guard is the safer choice. For single‑process, high‑performance pipelines, an atom guard combined with a database uniqueness constraint can suffice.

Step‑by‑Step Guide for a Typical Core.async Retry Flow

Define the uniqueness contract

Determine the fields that uniquely identify the logical record (e.g., user-id + event-id).
Create a unique index on those columns in the target database.

Generate a lightweight idempotency token
(defn make-token [payload]
  (str (System/nanoTime) ":" (hash payload)))
Attach the token to the message before putting it on the channel.

Write a retry macro
(defmacro with-retry [max-retries & body]
  `(loop [attempt# 0]
     (try
       ~@body
       (catch Exception e#
         (if (< attempt# max-retries)
           (do
             (js/console.warn "Retry" attempt# ":" e#)
             (recur (inc attempt#)))
           (throw e#))))))

Use conditional SQL inside the macro
(with-retry 5
  (jdbc/execute! db-conn
    ["INSERT INTO events (user_id, event_id, token) VALUES (?,?,?)"
     [user-id event-id token]]))
In databases that support ON CONFLICT, replace the insert with INSERT … ON CONFLICT DO NOTHING to make the operation idempotent.

Guard with an atom if operating in a single process
(def processed (atom #{}))

(defn safe-write [payload]
  (let [id (:token payload)]
    (when (swap! processed (fn [s] (if (contains? s id) s (conj s id))))
      ;; perform the write
      (db/write payload))))

For distributed setups, wrap the write in a transaction that sets a global flag
(jdbc/with-transaction [tx db-conn]
  (let [exists? (jdbc/execute! tx
                 ["SELECT 1 FROM flags WHERE token = ?" [token]])]
    (when-not (seq exists?)
      (jdbc/execute! tx
        ["INSERT INTO flags (token) VALUES (?)" [token]])
      (db/write payload))))

Monitor and log

Log the token, attempt count, and database response.
Set up alerts for repeated failures on the same token.

What Diagnostic Detail Might Change the Recommendation?
Do you currently have a unique constraint on the database columns that identify the record? If not, the atom or transaction guard alone will not prevent duplicate writes; the database itself must enforce uniqueness.

core.async Retry Strategy: Avoiding Duplicate Writes Without Idempotency Keys

core.async Retry Strategy: Avoiding Duplicate Writes Without Idempotency Keys

1 answer

Answering the Core Questions

Step‑by‑Step Guide for a Typical Core.async Retry Flow

What Diagnostic Detail Might Change the Recommendation?

Your perspective belongs here.

1 question comment

Deduplicate retries with a distinct transducer