Making Media Jobs Idempotent with Content-Hash Keys

Derive each job’s key from what actually determines its output — the SHA-256 of the input object (or its immutable version ID plus ETag) combined with a hash of the recipe (the step name and its parameters) — claim the key with a single conditional insert before doing any work, and write outputs to paths derived from the same key, so a duplicate event, a retried message or a double-click all resolve to one job and one set of files.

Every layer under a media pipeline delivers at least once. S3 notifications are documented as occasionally duplicated; SQS standard queues redeliver when a visibility timeout expires; a Lambda retry resubmits the whole event; users click “reprocess” twice. Without idempotency, each duplicate spends minutes of transcoding CPU and can race to overwrite a finished output with a half-written one. This page is part of media job orchestration in media processing and delivery pipelines. It generalises the per-message deduplication shown in queueing transcode jobs with SQS and Lambda to every step of the pipeline.

When to use this approach

  • Your pipeline has several steps (probe, transcode, thumbnail, captions) triggered by at-least-once events and retries.
  • Jobs are expensive enough that running one twice matters — anything over a few seconds of CPU or a few cents of managed service.
  • The same bytes can arrive more than once: re-uploads of an identical file, copies between buckets, or a user uploading the same video to two posts.

Prerequisites

  1. A transactional store with conditional writes: PostgreSQL (INSERT … ON CONFLICT DO NOTHING), DynamoDB (ConditionExpression: attribute_not_exists(pk)), or Redis SET NX for short-lived locks.
  2. Access to a stable fingerprint of the input: S3’s VersionId plus ETag on a versioned bucket, or a SHA-256 computed at upload time. Computing file checksums in the browser with Web Crypto can supply it before the bytes even arrive.
  3. Node 20+ with pg 8.x for the example below.
  4. Outputs written to deterministic paths — no timestamps or random IDs in output keys.

What goes into the key

A job’s output is a pure function of two things: the input bytes and the recipe. Hash both, and the key names the output exactly.

Composition of a content-hash job key The job key is a hash of the input content hash and the recipe hash. The input hash comes from the object's bytes. The recipe hash covers the step name, the recipe version and the canonical JSON of its parameters. The key names both the job row and the output prefix. key = H(input hash ‖ recipe hash) input hash SHA-256 of object bytes or VersionId + ETag recipe hash step: "hls-package" version: 3 params: canonical JSON job key sha256, 64 hex jobs row (PK) claimed once derived/<key>/ output prefix Change the bytes or the recipe and the key changes; change neither and every attempt finds the same row.
Keying by content and recipe, not by event or message ID, is what makes duplicates from any source collapse into one job.

A message ID is the wrong key: two different S3 events for the same object have different IDs, and a retried Lambda invocation carries a new request ID. An object key alone is also wrong: a user overwriting avatar.jpg with a new photo must produce a new job. Content plus recipe is the only combination that is equal exactly when the output would be equal.

Implementation 

import { createHash } from "node:crypto";
import pg from "pg";

const pool = new pg.Pool({ connectionString: process.env.DATABASE_URL });

/* Schema:
CREATE TABLE media_jobs (
  job_key      text PRIMARY KEY,
  step         text NOT NULL,
  input_hash   text NOT NULL,
  status       text NOT NULL CHECK (status IN ('running','done','failed')),
  lease_until  timestamptz NOT NULL,
  attempts     int  NOT NULL DEFAULT 1,
  output_prefix text,
  error        text,
  updated_at   timestamptz NOT NULL DEFAULT now()
);
*/

export interface Recipe { step: string; version: number; params: Record<string, unknown> }

/** Canonical JSON: sorted keys, so {a:1,b:2} and {b:2,a:1} hash identically. */
function canonical(v: unknown): string {
  if (Array.isArray(v)) return `[${v.map(canonical).join(",")}]`;
  if (v && typeof v === "object") {
    return `{${Object.keys(v as object).sort()
      .map((k) => `${JSON.stringify(k)}:${canonical((v as Record<string, unknown>)[k])}`).join(",")}}`;
  }
  return JSON.stringify(v);
}

export function jobKey(inputHash: string, recipe: Recipe): string {
  const recipeHash = createHash("sha256").update(canonical(recipe)).digest("hex");
  return createHash("sha256").update(`${inputHash}\n${recipeHash}`).digest("hex");
}

type Claim =
  | { kind: "run"; key: string; prefix: string }
  | { kind: "done"; key: string; prefix: string }
  | { kind: "busy"; key: string };

const LEASE_SECONDS = 900;   // longer than the slowest job; a crashed worker's lease expires

export async function claim(inputHash: string, recipe: Recipe): Promise<Claim> {
  const key = jobKey(inputHash, recipe);
  const prefix = `derived/${key}/`;

  // 1. Try to create the row. Exactly one concurrent caller succeeds.
  const inserted = await pool.query(
    `INSERT INTO media_jobs (job_key, step, input_hash, status, lease_until, output_prefix)
     VALUES ($1, $2, $3, 'running', now() + make_interval(secs => $4), $5)
     ON CONFLICT (job_key) DO NOTHING
     RETURNING job_key`,
    [key, recipe.step, inputHash, LEASE_SECONDS, prefix],
  );
  if (inserted.rowCount === 1) return { kind: "run", key, prefix };

  // 2. Row exists. Take it over only if it failed or its lease expired.
  const taken = await pool.query(
    `UPDATE media_jobs
        SET status = 'running', attempts = attempts + 1,
            lease_until = now() + make_interval(secs => $2), updated_at = now()
      WHERE job_key = $1
        AND (status = 'failed' OR (status = 'running' AND lease_until < now()))
      RETURNING job_key`,
    [key, LEASE_SECONDS],
  );
  if (taken.rowCount === 1) return { kind: "run", key, prefix };

  const { rows } = await pool.query(`SELECT status FROM media_jobs WHERE job_key = $1`, [key]);
  return rows[0]?.status === "done" ? { kind: "done", key, prefix } : { kind: "busy", key };
}

export async function finish(key: string, ok: boolean, error?: string): Promise<void> {
  await pool.query(
    `UPDATE media_jobs SET status = $2, error = $3, updated_at = now() WHERE job_key = $1`,
    [key, ok ? "done" : "failed", error ?? null],
  );
}

/** Wrap any step: duplicates return the existing output instead of re-running. */
export async function runOnce(
  inputHash: string,
  recipe: Recipe,
  work: (outputPrefix: string) => Promise<void>,
): Promise<{ prefix: string; reused: boolean } | { busy: true }> {
  const c = await claim(inputHash, recipe);
  if (c.kind === "done") return { prefix: c.prefix, reused: true };
  if (c.kind === "busy") return { busy: true };      // let the queue redeliver later
  try {
    await work(c.prefix);
    await finish(c.key, true);
    return { prefix: c.prefix, reused: false };
  } catch (err) {
    await finish(c.key, false, String(err).slice(0, 2000));
    throw err;
  }
}

// Usage inside a queue consumer
const result = await runOnce(
  "4f9d3c0b7a51e2d86c4f0a9b3e7d1c5a2f8b6e0d4c9a7b3f1e5d2c8a6b4e21a0", // sha256 from the asset row
  { step: "hls-package", version: 3, params: { segment: 4, ladder: ["1080", "720", "360"] } },
  async (prefix) => { console.log("packaging into", prefix); },
);
console.log(result);

Line-by-line on the decisions that matter

  • version in the recipe. When you change the packaging code — a new ladder rule, a bug fix — bump the version. Every asset gets a new key, and re-running the pipeline produces fresh outputs beside the old ones instead of being skipped as “already done”.
  • Canonical JSON. Object key order is not guaranteed stable across code paths; {"ladder":…,"segment":4} and {"segment":4,"ladder":…} must hash the same, or equal recipes create different jobs.
  • INSERT … ON CONFLICT DO NOTHING. The primary key constraint is the lock. Two workers racing on the same key cannot both insert; the database decides, atomically, without an application-level mutex.
  • A lease, not a lock. lease_until lets another worker take over a job whose worker crashed without calling finish. Set it longer than your slowest legitimate job, or two workers will run the same slow transcode.
  • busy returns without error. The duplicate is not a failure; returning lets the consumer leave the message on the queue (or change its visibility) so it is retried after the first worker finishes, at which point claim returns done.
  • Output prefix from the key. If a crashed attempt wrote half its files, the retry writes to the same prefix and overwrites them. Nothing ever reads a prefix until its row is done.

Where duplicates come from, and where they stop

Four duplicate sources converging on one claim A duplicated S3 event, an SQS redelivery after a visibility timeout, a Lambda retry and a user clicking reprocess all compute the same job key. The first claim runs the job; the others see busy or done and do no work. Four ways to ask twice, one place to say no duplicate S3 event SQS redelivery Lambda retry user clicks reprocess ×2 claim(key) INSERT … ON CONFLICT first: run → done during: busy, requeue after: done, reuse Only the first request does work; the rest cost one indexed database round trip each.
Idempotency lives at the claim, not in each upstream system — you cannot stop every source of duplicates, but you can stop them all in one place.

Configuration gotchas

Multipart ETags are not content hashes. An S3 ETag for a multipart upload is an MD5 of the part MD5s plus -N; the same file uploaded with a different part size gets a different ETag. It is still a fine identity for one object version, but not for cross-object deduplication. For “same bytes anywhere” semantics, compute SHA-256 at upload time (or use S3’s x-amz-checksum-sha256 full-object checksum) and store it on the asset — see verifying uploads with S3 additional checksums.

duplicate key value violates unique constraint "media_jobs_pkey". You used a plain INSERT and caught the error. It works, but in PostgreSQL an error aborts the surrounding transaction; use ON CONFLICT DO NOTHING and check rowCount instead.

Jobs stuck in running forever. The lease is too long, or nothing ever takes over expired leases because the queue message was deleted. Delete the queue message only after finish succeeds, so an abandoned job’s message reappears and its retry takes over the expired lease.

Two different recipes produce the same key. You hashed JSON.stringify(params) of objects built in different orders — or you left version out and changed code. Always hash the canonical form and treat the recipe version as part of the contract.

Deduplication beyond retries

Because the key depends on content rather than on the upload, identical files uploaded separately also collapse into one job — a user posting the same video in three places, or a popular meme uploaded by thousands of people. The processing cost is paid once and every asset points at the same derived prefix.

Processing jobs run with and without content-hash keys Across a month of 100 thousand video uploads, keying jobs by message ID ran about 108 thousand transcodes because of retries and duplicates. Keying by content and recipe ran about 91 thousand, because retries collapsed and 9 percent of uploads were byte-identical to an earlier one. Transcodes run per 100k uploads (one month) keyed by message ID ≈108k — retries and duplicates re-run keyed by content ≈91k — each unique input once 8% saved by collapsing redeliveries, 9% by byte-identical re-uploads. Remember deletion: an output shared by many assets is removed only when the last reference goes.
The same mechanism that makes retries safe also deduplicates popular content for free.

Sharing outputs has one consequence: deleting an asset must not delete a derived prefix another asset still references. Keep a reference count (or a join table from assets to job keys) and garbage-collect prefixes with zero references, which is the same bookkeeping as in deduplicating uploads with content hashes.

Verification 

import { strict as assert } from "node:assert";
import { jobKey, runOnce } from "./jobs.ts";

const recipe = { step: "thumb", version: 1, params: { width: 320, format: "webp" } };
const reordered = { version: 1, step: "thumb", params: { format: "webp", width: 320 } };
assert.equal(jobKey("abc", recipe), jobKey("abc", reordered), "key order must not matter");
assert.notEqual(jobKey("abc", recipe), jobKey("abc", { ...recipe, version: 2 }), "version bumps");

// Fire ten concurrent attempts at the same job: exactly one runs.
let runs = 0;
const results = await Promise.all(Array.from({ length: 10 }, () =>
  runOnce("abc", recipe, async () => { runs++; await new Promise((r) => setTimeout(r, 200)); })));
assert.equal(runs, 1, "only one attempt may do the work");
console.log(results.map((r) => ("busy" in r ? "busy" : r.reused ? "reused" : "ran")).join(" "));
// ran busy busy busy busy busy busy busy busy busy

Frequently Asked Questions

Is a Redis lock enough instead of a database row?

A SET key NX PX lock prevents concurrent runs but forgets that a job already finished once the lock expires, so a late duplicate re-runs a completed job. Pair it with a durable “done” marker — or use the database row alone, which provides both.

How is this different from SQS FIFO deduplication?

FIFO queues deduplicate identical message bodies within a five-minute window. That helps with producer retries but not with a Lambda that fails after doing half the work, nor with duplicates spread over hours. Content-hash claims cover all of those and work with standard queues.

Should the hash be computed by the worker or at upload time?

At upload time when possible — the bytes stream through the upload path anyway, and a stored hash lets the orchestrator compute keys without downloading anything. The worker can then verify the hash on read, which doubles as an integrity check.

Part of: Media Job Orchestration

Related

Poison Messages and DLQs

What happens to a job key that fails on every attempt.

Read →
Idempotency Keys for Fetch Uploads

The same idea on the client side of the upload.

Read →
Routing S3 Events with EventBridge

One of the at-least-once sources this pattern absorbs.

Read →

Related