Handle duplicate events in a media pipeline with a local outbox

Handle duplicate media events at the business-action boundary: several observations of one completed job should create one pending action for that workflow step. Record acceptance and the pending action in the same database transaction. Then treat delivery to an external consumer as a separate operation with its own recovery contract.

This guide implements that local acceptance step. It complements the webhooks versus polling guide, which explains how both notification paths can discover the same result. It does not promise exactly-once execution across a database and a remote service.

Separate notification identity from action identity

A repeated notification and a repeated business action are different problems. You might receive a webhook twice, or receive one webhook after polling has already discovered completion. Filtering only repeated delivery IDs addresses the first case, not the second.

Tornado's webhook reference distinguishes organization endpoints from explicit per-job callbacks. Organization events carry delivery_id and place job fields inside data; explicit per-job callbacks have a different payload contract. Configuring both can produce two notifications. Parse the actual route you configured rather than assuming every callback has the same envelope.

Keep a notification receipt if it helps diagnostics, but derive the business-action identity from your own workflow record. One useful key consists of tenant, ingestion attempt and processing step version. For example, the same authorized attempt entering analyze-v1 through a webhook or a status lookup must resolve to the same key.

Do not create a fresh action ID on every delivery. Conversely, an intentional new attempt needs its own identity. If a changed input or processing version should produce new work, represent that explicitly rather than silently reusing an old key.

Authenticate and map before accepting work

Verify callbacks using the mode configured for that endpoint. The Tornado reference describes Simple key verification and HMAC verification; do not substitute a job ID for authentication. For HMAC, follow the documented treatment of the original request bytes and timestamp.

Resolve the received job ID against your trusted application record and tenant before choosing an action. Keep the expected source, output settings and destination there. A syntactically valid callback is not permission to attach another tenant's job to the current workflow.

Only enter the acceptance function after the observation qualifies for your chosen step. Completion may qualify a file for validation; it does not establish that transcription or analysis has finished. Older progress observations should not undo accepted work. Conflicting terminal observations need reconciliation, not a last-arrival-wins update.

Commit acceptance and pending work together

A common failure window is: mark the event handled, then send work to a queue. If the process stops between those operations, a later duplicate can be discarded even though nothing was queued.

The following Python 3.12+ example uses a local outbox table: a durable list of actions a separate dispatcher must process. The accepted row and the outbox row are written together. A duplicate returns False only when the same action already refers to the same job.

The IDs are synthetic application identifiers. This is not a Tornado webhook parser, HTTP receiver or queue worker.

import sqlite3
from contextlib import closing


def create_schema(db):
    with db:
        db.execute("CREATE TABLE accepted ("
                   "action_id TEXT PRIMARY KEY NOT NULL, "
                   "job_id TEXT NOT NULL)")
        db.execute("CREATE TABLE outbox ("
                   "action_id TEXT PRIMARY KEY NOT NULL, "
                   "job_id TEXT NOT NULL)")


def accept_action(db, action_id, job_id):
    if any(not isinstance(v, str) or not v.strip()
           for v in (action_id, job_id)):
        raise ValueError("Nonempty application IDs required")
    with db:
        inserted = db.execute(
            "INSERT INTO accepted VALUES (?, ?) "
            "ON CONFLICT(action_id) DO NOTHING",
            (action_id, job_id),
        ).rowcount
        if inserted == 0:
            existing = db.execute(
                "SELECT job_id FROM accepted WHERE action_id = ?",
                (action_id,),
            ).fetchone()
            if existing != (job_id,):
                raise ValueError("Action ID already maps to another job")
            return False
        db.execute("INSERT INTO outbox VALUES (?, ?)",
                   (action_id, job_id))
    return True


with closing(sqlite3.connect(":memory:", autocommit=False)) as db:
    create_schema(db)
    key = "tenant-demo:attempt-1:analyze-v1"
    assert accept_action(db, key, "job-example")
    assert not accept_action(db, key, "job-example")
    assert db.execute("SELECT count(*) FROM outbox").fetchone() == (1,)

SQLite's UPSERT reference documents the targeted uniqueness conflict used here. The conflict is resolved inside the insert, instead of using a separate unprotected “does this key exist?” check. Other database failures remain errors.

The Python SQLite reference explains how the connection context manager commits on success and rolls back on failure. Use a dedicated connection with this transaction configuration and no unrelated pending work: the helper owns its transaction. closing closes the connection separately.

The in-memory database is disposable. Real acceptance requires persistent storage and an appropriate database deployment. A database timeout, failed commit or unexpected constraint failure must not be reported as a successfully accepted event. Return an appropriate failure response and keep a reconciliation path; do not assume the provider will retry forever.

Preserve the key across external delivery

The local transaction removes the gap between acceptance and creation of pending work. It does not make sending to a remote service atomic with marking the outbox row finished.

Consider a dispatcher that sends an analysis request successfully but stops before recording the response. On restart, the row still appears pending. Sending it again may repeat the remote action.

If the downstream service documents an idempotency mechanism, reuse one stable key for the same action and obey its scope, retention and payload rules. A locally generated key alone does nothing unless the receiver implements that contract. If no such mechanism exists, preserve uncertain outcomes and reconcile them using the receiver's available records before resubmitting.

A production dispatcher also needs concurrent-worker claims, retry scheduling, attempt records and completion state. Those are deliberately outside this small example. Do not delete acceptance records as soon as a row is dispatched: a late callback could otherwise recreate the action. Choose a retention policy that accounts for delayed observations and your recovery process.

Test the failure windows

The example was executed locally, along with checks for a conflicting job mapping, empty IDs, a failed outbox insert followed by rollback and retry, persistence through reopening, and two concurrent connections accepting the same key. These checks used synthetic IDs and temporary SQLite storage. No live callbacks or media jobs were sent.

For your actual integration, exercise these cases before enabling a downstream action:

The local tests establish only the demonstrated database behavior, not a power-loss durability guarantee or exactly-once remote execution. Track accepted actions that remain pending and downstream actions whose outcome is unknown; counting received callbacks alone misses both.

Start with the first media workflow, then route verified webhook observations and status lookups through the same acceptance boundary. Keep worker restart recovery underneath that flow so a restart resumes recorded work instead of inventing a new attempt.