Idempotent deduplication for maritime document queues #

An AS2 gateway that never receives its MDN acknowledgment will resend the same interchange; an SFTP partner whose cron overlaps will drop the same manifest twice; and every at-least-once broker between the wire and the worker is licensed to redeliver on any timeout. Idempotent deduplication for maritime document queues is the layer that lets all of that redundancy happen safely — it computes a stable identity for each inbound document, records that identity in a fast dedup store before any side effect fires, and guarantees that the first copy of a Bill of Lading books a berth or files a customs entry exactly once while every retransmission collapses to an audited no-op. The hard part is doing this without swallowing a legitimate resend: a corrected VGM or an amended B/L must supersede its predecessor, not be silently discarded as a duplicate.

Architecture Alignment #

Deduplication is a control that runs throughout the Async Batch Processing Pipelines topic area, the throughput and back-pressure tier of the wider Document Ingestion & EDI Parsing Workflows domain. The queue and worker wiring it sits on top of is specified in Building Celery queues for maritime doc ingestion; the structural, semantic, and regulatory checks that decide whether a de-duplicated payload is even valid live in Schema Validation Frameworks. The dedup layer has one responsibility and one only: decide whether the effects of this document have already been applied. It does not parse, it does not validate, and it never mutates the payload — it answers a single boolean question against a shared store, then lets the pipeline proceed or short-circuit. Getting that boundary right is what turns at-least-once delivery from a correctness hazard into a durability feature.

Prerequisites & Environment Setup #

The dedup store targets Python 3.11+ and a Redis instance shared across the worker pool. Install the Redis client, structured logging, and typed-validation stack:

pip install "redis>=5.0" structlog pydantic

Structured JSON logging is not optional here: every dedup verdict — accepted, dropped, or superseded — is an audit event that a customs or port-state-control review may replay, so bare print() and unstructured strings are unacceptable. Configure structlog once, at worker boot, to emit JSON:

import structlog

structlog.configure(
    processors=[
        structlog.contextvars.merge_contextvars,
        structlog.processors.add_log_level,
        structlog.processors.TimeStamper(fmt="iso", utc=True),
        structlog.processors.JSONRenderer(),
    ],
)
log = structlog.get_logger("maritime.dedup")

Environment variables the dedup layer expects:

Size the TTL to the longest window over which a partner might legitimately retransmit the same interchange — a week comfortably covers AS2 MDN timeout storms and weekend SFTP re-drops, while keeping the Redis keyspace bounded. A TTL that is too short reopens the duplicate window; one that never expires turns the dedup store into an unbounded liability.

Step-by-step Implementation #

Step 1 — Derive a stable idempotency key #

A dedup key must be identical for two byte-for-byte-equivalent documents and different for two genuinely different ones. Two derivations serve maritime EDI well. The first is content-addressed: canonicalize the payload — strip the volatile UNB timestamp (data element S004), normalize segment terminators and whitespace, uppercase the syntax envelope — then take the SHA-256 of the result, so cosmetic reformatting by an intermediary VAN does not defeat the match. The second is identity-addressed and cheaper: the UN/EDIFACT interchange control reference (UNB data element 0020) is guaranteed unique per sender for the life of their numbering scheme, so pairing it with the sender identification (0004, inside composite S002) yields a natural key without hashing a megabyte of segments.

from __future__ import annotations

import hashlib
import re

_TERMINATOR_RE = re.compile(rb"[\r\n]+")
_WS_RUN_RE = re.compile(rb"[ \t]{2,}")


def canonicalize(raw: bytes) -> bytes:
    """Normalize transport-level noise so equivalent interchanges hash alike."""
    normalized = _TERMINATOR_RE.sub(b"", raw.strip())
    normalized = _WS_RUN_RE.sub(b" ", normalized)
    # Drop the UNB date/time (S004) — it varies per transmission, not per document.
    normalized = re.sub(rb"(UNB\+[^+]+\+[^+]+\+[^+]+)\+\d{6,8}:\d{4}", rb"\1", normalized)
    return normalized.upper()


def content_key(raw: bytes) -> str:
    return hashlib.sha256(canonicalize(raw)).hexdigest()


def interchange_key(sender_id: str, control_ref_0020: str) -> str:
    # UNB sender identification (0004) + interchange control reference (0020).
    return f"{sender_id.strip().upper()}:{control_ref_0020.strip()}"

Step 2 — Back the guard with a Redis SETNX store #

The dedup store is a thin, typed wrapper over Redis SET key value NX EX ttl — an atomic “set if not present with expiry” that is the canonical primitive for distributed idempotency. first_seen returns True exactly once per key across the entire worker fleet, because Redis serializes the conditional set; every later caller gets False. Storing the version alongside the key is what later lets a real amendment be told apart from a plain duplicate.

from __future__ import annotations

import redis
import structlog

log = structlog.get_logger("maritime.dedup")


class DedupStore:
    """Redis SETNX-backed idempotency store for maritime document queues."""

    def __init__(self, client: redis.Redis, *, prefix: str, ttl_seconds: int) -> None:
        self._redis = client
        self._prefix = prefix
        self._ttl = ttl_seconds

    def _namespaced(self, key: str) -> str:
        return f"{self._prefix}:{key}"

    def first_seen(self, key: str, *, version: int = 0) -> bool:
        """True only for the first delivery of `key`; False for any duplicate."""
        stored: bool = bool(
            self._redis.set(self._namespaced(key), version, nx=True, ex=self._ttl)
        )
        if stored:
            log.info("dedup_first_seen", key=key, version=version, ttl=self._ttl)
        else:
            log.info("dedup_duplicate_suppressed", key=key, version=version)
        return stored

    def stored_version(self, key: str) -> int | None:
        raw = self._redis.get(self._namespaced(key))
        return int(raw) if raw is not None else None

Step 3 — Choose publish-time versus consume-time deduplication #

There are two places to run the guard, and they defend different failure modes. Consume-time dedup — checking the key as the worker pulls a task — collapses broker redeliveries: a task that ran, applied its effect, then died before acking is redelivered, and the guard stops the second run. Publish-time dedup — checking before the downstream emit — collapses upstream duplicates that arrived as two distinct broker messages, such as an AS2 retransmission the ingestion edge enqueued twice. Production pipelines want both, but the load-bearing one is publish-time, keyed on the same identity the whole pipeline agrees on, because it is the last gate before an irreversible side effect (a filed customs entry, a booked berth). The guard must wrap the effect, not merely precede it.

from typing import Callable, TypeVar

import structlog

log = structlog.get_logger("maritime.dedup")
T = TypeVar("T")


def once(store: DedupStore, key: str, effect: Callable[[], T]) -> T | None:
    """Run `effect` exactly once per key; a duplicate returns None as a no-op."""
    if not store.first_seen(key):
        log.info("effect_skipped_duplicate", key=key)
        return None
    try:
        return effect()
    except Exception:
        # The effect failed after we claimed the key — release it so a retry can
        # re-attempt, otherwise a transient fault would be masked as a duplicate.
        store._redis.delete(store._namespaced(key))
        log.exception("effect_failed_key_released", key=key)
        raise

Step 4 — Let a legitimate amendment supersede its predecessor #

At-least-once delivery is not the only reason two documents share a key. A carrier that discovers a wrong consignee or an under-declared weight issues a corrected B/L or a revised VGM — same interchange lineage, new content. UN/EDIFACT carries this intent in the BGM beginning-of-message segment: data element 1225 is the message function code, where 9 is original and 5 is replace. Keying dedup on content alone would let the correction through as a new document (good), but keying on the interchange reference alone would swallow it as a duplicate (a serious defect — the wrong weight would stand). The fix is a compound decision: treat an incoming version strictly greater than the stored version as a supersede, apply it, and advance the stored version atomically.

import redis
import structlog

log = structlog.get_logger("maritime.dedup")

# Atomically store the new version only if it is strictly greater than the old.
_SUPERSEDE_LUA = """
local cur = redis.call('GET', KEYS[1])
if cur == false or tonumber(ARGV[1]) > tonumber(cur) then
    redis.call('SET', KEYS[1], ARGV[1], 'EX', ARGV[2])
    return 1
end
return 0
"""


def accept_or_supersede(store: DedupStore, key: str, version: int) -> bool:
    """True if this is a first sight or a higher-version amendment; else False."""
    script = store._redis.register_script(_SUPERSEDE_LUA)
    accepted = bool(script(keys=[store._namespaced(key)], args=[version, store._ttl]))
    log.info(
        "dedup_versioned_decision",
        key=key,
        version=version,
        outcome="supersede" if accepted else "duplicate",
    )
    return accepted

Edge Cases & Carrier Deviations #

A specific trap is the reused interchange control reference. Data element 0020 is only required to be unique per sender within their own numbering cycle, and a partner whose counter is a five-digit field will eventually wrap and reissue 00042 for a brand-new interchange. Keyed on 0020 alone, that new document would be discarded as a duplicate of a year-old one. Pairing 0020 with the sender identification and, where the retransmission window is long, a coarse date bucket keeps the key genuinely unique; when in doubt, the content hash from Step 1 is collision-proof by construction.

Verification & Testing #

Assert the three outcomes — first-seen accepts, duplicate drops, amendment supersedes — against fakeredis so no live broker is needed, and confirm the audit event fires.

import fakeredis
import pytest

from maritime_dedup.store import DedupStore, accept_or_supersede
from maritime_dedup.keys import content_key


@pytest.fixture
def store() -> DedupStore:
    client = fakeredis.FakeRedis()
    return DedupStore(client, prefix="dedup:test", ttl_seconds=3600)


def test_first_delivery_is_seen_once(store: DedupStore) -> None:
    key = content_key(b"UNB+UNOA:2++++42'UNH+1+IFCSUM:D:96A:UN'")
    assert store.first_seen(key) is True
    assert store.first_seen(key) is False   # AS2 retransmission collapses


def test_reformatted_duplicate_matches(store: DedupStore) -> None:
    a = content_key(b"UNB+UNOA:2+SENDER+RECV+250717:0900+42'UNH+1'")
    b = content_key(b"UNB+UNOA:2+SENDER+RECV+250718:1130+42'UNH+1'\r\n")
    assert a == b   # only the volatile UNB timestamp differed


def test_higher_version_supersedes(store: DedupStore) -> None:
    key = "SENDER:42"
    assert accept_or_supersede(store, key, version=1) is True   # original BGM+9
    assert accept_or_supersede(store, key, version=1) is False  # exact duplicate
    assert accept_or_supersede(store, key, version=2) is True   # corrected BGM+5

A passing run emits one JSON line per verdict; a suppressed duplicate looks like {"event": "dedup_duplicate_suppressed", "key": "SENDER:42", "version": 0, "log_level": "info", "timestamp": "..."}. Assert on the structured keys, never on a formatted string.

Frequently Asked Questions #

Related #

• Building Celery queues for maritime doc ingestion — the queue, prefetch, and DLQ wiring this dedup guard runs on top of
• Async Batch Processing Pipelines — the parent topic area defining at-least-once delivery and idempotent downstream publishing
• Schema Validation Frameworks — the structural, semantic, and regulatory checks a de-duplicated payload still has to pass
• Document Ingestion & EDI Parsing Workflows — the domain framework governing ingestion, schema, and resilience contracts

Up: Async Batch Processing Pipelines — the topic area governing asynchronous document throughput and duplicate-safe delivery.