codesnips
sql python 101 lines · 3 tabs

Atomic “Read + Mark Processed” with UPDATE … RETURNING

Shared by codesnips Jan 2026
3 tabs 
CREATE TYPE outbox_status AS ENUM ('pending', 'retry', 'processing', 'done', 'dead');

CREATE TABLE outbox_events (
    id           BIGINT GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
    dedupe_key   TEXT        NOT NULL,
    topic        TEXT        NOT NULL,
    payload      JSONB       NOT NULL,
    status       outbox_status NOT NULL DEFAULT 'pending',
    attempts     INT         NOT NULL DEFAULT 0,
    max_attempts INT         NOT NULL DEFAULT 5,
    available_at TIMESTAMPTZ NOT NULL DEFAULT now(),
    locked_by    TEXT,
    locked_at    TIMESTAMPTZ,
    created_at   TIMESTAMPTZ NOT NULL DEFAULT now()
);

-- producers stay idempotent: a repeated enqueue is a no-op
CREATE UNIQUE INDEX idx_outbox_dedupe ON outbox_events (dedupe_key);

-- partial index so the claim query only touches unfinished work
CREATE INDEX idx_outbox_claimable
    ON outbox_events (available_at)
    WHERE status IN ('pending', 'retry');

INSERT INTO outbox_events (dedupe_key, topic, payload)
VALUES ('order-42-created', 'orders.created', '{"order_id": 42}')
ON CONFLICT (dedupe_key) DO NOTHING;
3 files · sql, python Explain with highlit

A common failure mode in worker queues is the read-then-update race: one query selects pending rows, another updates them, and between the two calls a second worker grabs the same rows. This snippet shows the PostgreSQL idiom that closes that gap by folding the claim and the state transition into a single statement using UPDATE ... RETURNING combined with FOR UPDATE SKIP LOCKED.

The schema.sql tab defines the outbox_events table as a durable queue. Each row carries a status, an attempts counter, a locked_by worker id, and a locked_at timestamp. The partial index idx_outbox_claimable is the performance heart of the design: it indexes only rows in the pending or retry state and orders them by available_at, so the claim query never scans already-processed history.

The claim_events.sql tab is where atomicity happens. The CTE claimed selects up to :batch_size claimable rows, and crucially applies FOR UPDATE SKIP LOCKED inside the subquery. FOR UPDATE row-locks the candidates; SKIP LOCKED tells Postgres to silently ignore rows another transaction already locked instead of blocking on them. The outer UPDATE then flips those exact rows to processing, stamps locked_by/locked_at, bumps attempts, and RETURNING hands the full claimed payloads back to the caller. Because select-and-mark is one statement in one implicit transaction, no two workers can ever claim the same event.

The worker.py tab shows the consumer loop. claim_batch runs the claim query and returns rows already marked processing, so the worker owns them the moment it sees them. On success mark_processed transitions to done; on failure mark_failed either reschedules with a backoff via available_at or moves the row to dead once max_attempts is exceeded. The dedupe_key column and ON CONFLICT in the schema make producers idempotent, guarding against duplicate enqueues.

The trade-off is that claimed-but-crashed rows stay in processing until a reaper resets stale locks by locked_at, so at-least-once, not exactly-once, delivery is the realistic guarantee. This pattern scales horizontally without an external broker and is the standard way to build a reliable outbox on plain Postgres.

Related snips

typescript 
export interface RetryOptions {
  retries: number;
  baseMs: number;
  maxMs: number;
  signal?: AbortSignal;
  onRetry?: (attempt: number, delay: number, err: unknown) => void;

Exponential backoff with jitter for retries

typescript reliability retry
by codesnips 2 tabs
go 
package dbutil

import (
  "context"

  "github.com/jackc/pgconn"

Retry Postgres serialization failures with bounded attempts

go postgres transactions
by Leah Thompson 1 tab
ruby 
require "csv"

class PeopleCsvStream
  include Enumerable

  HEADERS = %w[id full_name email signed_up_at plan].freeze

Resilient CSV Export as a Streamed Response

rails performance streaming
by codesnips 3 tabs
ruby 
class SignupForm
  include ActiveModel::Model
  include ActiveModel::Attributes

  attribute :account_name, :string
  attribute :email, :string

Shallow Controller, Deep Params: Form Object Pattern

rails activemodel form-object
by codesnips 3 tabs
rust 
use crossbeam::channel::unbounded;
use std::thread;

fn main() {
    let (tx, rx) = unbounded();

Crossbeam for advanced concurrent data structures

rust concurrency lock-free
by Marcus Chen 1 tab
javascript 
// Creating a Promise
const myPromise = new Promise((resolve, reject) => {
  const success = true;

  setTimeout(() => {
    if (success) {

Promises and async/await patterns for asynchronous JavaScript

javascript promises async-await
by Alex Chang 1 tab

Share this code

Here's the card — post it anywhere.

Atomic “Read + Mark Processed” with UPDATE … RETURNING — share card
Link copied