Ingestion Transport Protocols

Before a single segment of an EDI interchange is parsed or a single XML node is traversed, the invoice has to physically arrive. Carriers do not agree on how to deliver it: one drops a file on an SFTP server overnight, another pushes a signed AS2 payload through an EDI VAN, a third POSTs a JSON envelope to a webhook the moment the bill is cut. This topic covers the transport edge — the thin, unglamorous layer where those three delivery mechanisms terminate — and its single job is to turn every one of them into the same thing: one canonical intake event published to the message broker, deduplicated, with the raw bytes already persisted. Everything downstream in Automated Invoice Parsing & EDI/XML Ingestion assumes that event already exists.

Get the transport edge wrong and the failures are quiet and expensive. A poller that reads a half-written SFTP file emits a truncated interchange that parses into a wrong total. An AS2 endpoint that fails to return its receipt makes the carrier retransmit, and the same invoice is ingested twice. A webhook that does its parsing inside the request handler times out under a burst, the carrier retries, and now you have both a duplicate and a dropped bill in the same minute. The discipline here is to keep the transport layer dumb and idempotent: accept the bytes, prove they are complete, stamp a stable event key, hand off, and get out of the way. Parsing is somebody else’s problem.

Prerequisites

This layer sits upstream of every parser and owns no business logic. It assumes the following are already in place:

Two configuration contracts matter before any adapter runs. First, each carrier is registered with a carrier_scac and the transport it uses, so the intake event can be attributed without inspecting the payload. Second, the dedup key strategy is fixed per transport — content hash for SFTP, message-id for AS2, event-id for webhooks — because each transport carries a different natural idempotency token, and mixing them up is how replays slip through.

Architecture: three transports, one canonical event

The topology is a funnel. Three transport-specific adapters terminate their protocol, and each produces the identical IntakeEvent — the same shape whether the bytes came from a file, an AS2 message, or an HTTP body. That event is deduplicated once, then published to the broker, from which the format-specific parsers pull. No parser ever knows or cares which transport a bill arrived on.

The three transports are not interchangeable. They differ on the guarantees they offer, what an acknowledgement even means, whether ordering is preserved, and how the sender is authenticated. Those differences drive the whole design of each adapter, so it is worth pinning them down in one place:

The recurring theme across all three columns is at-least-once. No mainstream freight transport gives you exactly-once delivery, so the intake layer must be the place where at-least-once becomes effectively-once. That is what the dedup gate is for, and it is why every adapter must produce a stable key.

Step-by-step implementation

The implementation is a small transport-agnostic core — one IntakeEvent and one adapter interface — plus three concrete adapters that live on their own detail pages. The core is what guarantees every transport ends up identical on the broker.

Step 1 — Define the canonical intake event

The IntakeEvent is the contract. Every adapter, regardless of protocol, fills in exactly these fields. The dedup_key is mandatory and transport-specific; the raw_ref points at bytes already in the object store, never at bytes held in memory.

from __future__ import annotations
import hashlib
from dataclasses import dataclass, field
from datetime import datetime, timezone
from enum import Enum
from typing import Optional


class Transport(str, Enum):
    SFTP = "sftp"
    AS2 = "as2"
    WEBHOOK = "webhook"


@dataclass(frozen=True)
class IntakeEvent:
    """The single canonical shape every transport adapter emits."""
    carrier_scac: str
    transport: Transport
    dedup_key: str            # content hash, AS2 message-id, or webhook event-id
    raw_ref: str              # object-store URI of the already-persisted bytes
    byte_len: int
    received_at: str = field(
        default_factory=lambda: datetime.now(timezone.utc).isoformat()
    )
    content_type: Optional[str] = None

    def __post_init__(self) -> None:
        if not self.dedup_key:
            raise ValueError("IntakeEvent requires a non-empty dedup_key")
        if len(self.carrier_scac) != 4 or not self.carrier_scac.isalpha():
            raise ValueError(f"invalid SCAC: {self.carrier_scac!r}")


def content_hash(payload: bytes) -> str:
    """Stable content-addressed key; identical bytes always hash identically."""
    return hashlib.sha256(payload).hexdigest()

Common mistake: letting one transport smuggle an extra field into the event “just for now”. The moment the webhook adapter adds a retry_count that the SFTP adapter never sets, downstream consumers start branching on transport, and the whole point of the canonical event is lost. Keep transport-specific detail inside the adapter; publish only the shared shape.

Step 2 — Define the adapter interface

Every transport implements the same two-method protocol: receive() yields raw payloads with their transport-natural dedup key, and the shared emit() handles persistence, dedup, and publish once. Only receive() differs per transport.

from typing import Iterator, Protocol, Tuple

# (payload_bytes, dedup_key, carrier_scac, content_type)
RawDelivery = Tuple[bytes, str, str, Optional[str]]


class TransportAdapter(Protocol):
    """Each transport implements receive(); emit() is shared and identical."""
    transport: Transport

    def receive(self) -> Iterator[RawDelivery]:
        """Terminate the protocol and yield completed, whole payloads only."""
        ...

Common mistake: doing the persist-and-publish inside each adapter. That duplicates the idempotency logic three times, and the three copies drift. Centralize emit() so there is exactly one place that decides whether an event is new.

Step 3 — Persist raw bytes, then dedup, then publish

Order is load-bearing. Persist the bytes first so a parser can always re-read the original; only then check the dedup key; only then publish. If you publish before persisting, a consumer can race ahead to a raw_ref that does not exist yet.

import structlog

logger = structlog.get_logger()


class IntakeGateway:
    def __init__(self, raw_store, dedup_store, broker, topic: str = "invoice.intake"):
        self.raw_store = raw_store        # put(key, bytes) -> uri
        self.dedup_store = dedup_store     # add(key) -> True if newly added
        self.broker = broker               # publish(topic, event_dict)
        self.topic = topic

    def emit(self, payload: bytes, dedup_key: str,
             carrier_scac: str, transport: Transport,
             content_type: Optional[str] = None) -> Optional[IntakeEvent]:
        # 1. Dedup FIRST is tempting but wrong: persist so bytes always exist.
        raw_ref = self.raw_store.put(dedup_key, payload)

        # 2. Atomic claim: add() is a set-if-absent; only the first caller wins.
        if not self.dedup_store.add(dedup_key):
            logger.info("intake_duplicate", key=dedup_key, scac=carrier_scac,
                        transport=transport.value)
            return None                    # already ingested; drop silently

        event = IntakeEvent(
            carrier_scac=carrier_scac, transport=transport,
            dedup_key=dedup_key, raw_ref=raw_ref, byte_len=len(payload),
            content_type=content_type,
        )
        self.broker.publish(self.topic, event.__dict__)
        logger.info("intake_published", key=dedup_key, scac=carrier_scac,
                    transport=transport.value, bytes=len(payload))
        return event

    def run(self, adapter: TransportAdapter) -> int:
        published = 0
        for payload, key, scac, ctype in adapter.receive():
            if self.emit(payload, key, scac, adapter.transport, ctype):
                published += 1
        return published

Common mistake: using a non-atomic “check then set” on the dedup store (if key not in store: store.add(key)). Two workers polling the same SFTP folder both see the key absent and both publish. The add() here must be atomic — a Redis SET NX, or a Postgres INSERT ... ON CONFLICT DO NOTHING whose affected-row count tells you whether you won the claim.

Validation & testing

Because the gateway is a pure orchestration of three injectable dependencies, it tests without any real network. Fake the store, keystore, and broker, then assert on the exact effectively-once behaviour.

import pytest


class FakeStore:
    def __init__(self): self.data = {}
    def put(self, key, payload): self.data[key] = payload; return f"mem://{key}"


class FakeDedup:
    def __init__(self): self.seen = set()
    def add(self, key):
        if key in self.seen:
            return False
        self.seen.add(key); return True


class FakeBroker:
    def __init__(self): self.published = []
    def publish(self, topic, event): self.published.append((topic, event))


def make_gateway():
    return IntakeGateway(FakeStore(), FakeDedup(), FakeBroker())


def test_first_delivery_publishes_once():
    gw = make_gateway()
    ev = gw.emit(b"raw-bytes", "abc123", "ABCD", Transport.SFTP)
    assert ev is not None
    assert len(gw.broker.published) == 1


def test_replayed_delivery_is_dropped():
    gw = make_gateway()
    gw.emit(b"raw-bytes", "abc123", "ABCD", Transport.SFTP)
    second = gw.emit(b"raw-bytes", "abc123", "ABCD", Transport.SFTP)
    assert second is None                       # replay dropped
    assert len(gw.broker.published) == 1        # still exactly one publish


def test_bad_scac_rejected_before_publish():
    gw = make_gateway()
    with pytest.raises(ValueError):
        gw.emit(b"x", "k1", "12", Transport.WEBHOOK)

The fixtures worth keeping cover the effectively-once boundary from every angle: the same key arriving twice on the same transport, the same key arriving from two transports (a carrier that both drops a file and pushes a webhook for the same bill), and a burst of distinct keys to confirm the counter tracks published-versus-dropped honestly.

Performance & tuning

The gateway itself is I/O-bound — the object-store put and the broker publish dominate; the hashing and dedup lookup are sub-millisecond. Tune the transports independently because their bottlenecks differ.

The high-volume fan-out that happens after the broker — parsing thousands of interchanges in parallel — is deliberately not this layer’s concern. That elasticity belongs to Async Batch Processing Workflows; the transport edge stays thin and simply keeps the broker fed.

Failure modes

Three failure signatures dominate the transport edge, one per transport, and each is what its detail page exists to close.

1. Partial upload (SFTP). The poller reads a file the carrier is still writing and emits a truncated payload. Diagnostic: the content hash of a “complete” file changes on the next poll. Resolution: require consecutive equal-size polls (or a rename-into-place convention) before claiming — the full fix is in Configuring SFTP Drop Ingestion for Carrier Invoices.

2. Missing MDN (AS2). Verification or decryption succeeds but the signed receipt never goes back, so the carrier retransmits and the same message-id arrives again. Diagnostic: duplicate message-id rows minutes apart. Resolution: dedup on message-id and return the MDN before considering the message done — see Setting Up AS2 Transmission for EDI Freight Bills.

3. Replayed webhook. The handler parses synchronously, exceeds the sender’s timeout, and the carrier retries the same event-id. Diagnostic: identical event-id with a 5xx or slow-2xx in the access log. Resolution: verify, persist, ack 202 fast, and process async — covered in Wiring Webhook Endpoints for Real-Time Invoice Intake.

Integration points

The output of this layer is a stream of IntakeEvent records on the invoice.intake topic, each pointing at persisted raw bytes. That is a stable contract the parsers consume without ever touching a socket. The content_type and the leading bytes at raw_ref let a router dispatch each event to the right parser: X12 interchanges to EDI 210/810 Processing, and markup payloads to XML Freight Bill Ingestion. Because the event already guarantees whole, deduplicated bytes, those parsers can be pure transforms that never worry about half-files or replays.

In this section

• Configuring SFTP Drop Ingestion for Carrier Invoices — stops the poller from reading half-written files, reprocessing a carrier re-drop, or missing files that land mid-run, using size-stability detection, an atomic claim, and a content-hash manifest.
• Setting Up AS2 Transmission for EDI Freight Bills — closes the silent double-ingest that happens when an MDN receipt is not returned, plus signature/decryption drops and certificate rotation, with message-id dedup.
• Wiring Webhook Endpoints for Real-Time Invoice Intake — keeps a real-time intake endpoint from dropping invoices under burst load or blocking the sender past its timeout, with a thin handler, HMAC verification, and event-id dedup.

Related

• Automated Invoice Parsing & EDI/XML Ingestion — the parent ingestion architecture this transport edge feeds.
• EDI 210/810 Processing — the X12 parser that consumes intake events carrying interchanges.
• XML Freight Bill Ingestion — the DOM-based parser for markup payloads on the same topic.
• Async Batch Processing Workflows — how the published events fan out across workers without blocking intake.

---

Up: Automated Invoice Parsing & EDI/XML Ingestion