Configuring SFTP Drop Ingestion for Carrier Invoices

This page fixes the three ways a naive SFTP poller corrupts freight-bill intake: it picks up a file the carrier is still writing, it reprocesses the same file after a re-drop, and it misses files that land during the run itself.

The failure you are hitting

You pointed a scheduled job at a carrier’s SFTP directory, it globbed *.edi, read each file, and published it. On the sample folder it was flawless. In production it fails in three ways that never raise an exception:

• The poller reads ABCD_20260714.edi while the carrier’s SFTP session is still streaming bytes into it. You ingest a truncated interchange — the IEA control trailer is missing, the B305 total is whatever had been written so far — and it parses into a confidently wrong invoice.
• A carrier re-drops the same file the next night because their side never got an acknowledgement (SFTP has none). Your poller sees a filename it has already seen, or a filename it has not seen because they renamed it, reads it again, and the same freight bill enters the ledger twice.
• Between the moment your poller lists the directory and the moment it finishes reading, three new files land. The listing was a snapshot; those three files are invisible to this run and — if you delete-after-read on the ones you did see — may sit untouched until someone notices the bill was never audited.

This is the SFTP terminal inside Ingestion Transport Protocols. Its only job is to hand whole, deduplicated bytes to the intake gateway, and all three failures are it failing that job.

Root cause analysis

None of these are SFTP bugs. They are the direct consequence of treating a filesystem as if it were a queue, when it offers none of a queue’s guarantees:

1. No atomic-write signal. Plain SFTP put writes bytes progressively into the final path. There is no server-side flag that says “this file is complete”, so a poller that keys off existence reads whatever is present at that instant, partial or not.
2. No delivery receipt, so carriers re-drop. Because the carrier never learns you consumed the file, their retry policy is to send it again — sometimes same name, sometimes a new timestamped name for byte-identical content. Filename-based dedup catches the first case and misses the second.
3. Directory listing is a race, not a transaction. listdir returns a point-in-time snapshot. Files arriving mid-run are simply absent from it, and a poll interval that assumes “what I listed is all there is” silently skips them until the next cycle — or forever, if the run also prunes.
4. No processed manifest. With no durable record of what was already ingested, every restart, overlap, or re-drop is indistinguishable from a genuinely new file.

Reproducible diagnostic

Confirm which failure you have before touching the poller. This snippet lists a directory twice with a short gap and reports, per file, whether its size is still changing and whether its content hash is already known:

import hashlib
import time
import paramiko


def sha256_of(sftp, path: str) -> str:
    h = hashlib.sha256()
    with sftp.open(path, "rb") as fh:
        for chunk in iter(lambda: fh.read(65536), b""):
            h.update(chunk)
    return h.hexdigest()


def probe(host, user, key_path, remote_dir, known_hashes: set[str]):
    client = paramiko.SSHClient()
    client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
    client.connect(host, username=user, key_filename=key_path)
    sftp = client.open_sftp()

    first = {a.filename: a.st_size for a in sftp.listdir_attr(remote_dir)}
    time.sleep(5)
    second = {a.filename: a.st_size for a in sftp.listdir_attr(remote_dir)}

    for name in sorted(set(first) | set(second)):
        s1, s2 = first.get(name), second.get(name)
        growing = s1 is not None and s2 is not None and s1 != s2
        digest = sha256_of(sftp, f"{remote_dir}/{name}") if not growing and s2 else None
        seen = digest in known_hashes if digest else False
        print(f"{name:32} size1={s1} size2={s2} growing={growing} "
              f"hash={'-' if digest is None else digest[:12]} dup={seen}")
    sftp.close(); client.close()

Read the output as a decision table:

A file whose size changes between two listings must never be read this cycle — that single check eliminates the truncation class outright.

Resolution path

The fix is a poller that proves completeness, claims exclusively, and dedups on content. Pin the deps so CI and prod agree:

# requirements.txt
paramiko==3.5.0
structlog==24.4.0

Step 1 — Detect a stable (fully written) file

Do not trust existence. Require a file’s size to be identical across consecutive polls — and, defensively, older than a small quiet-period — before it is a candidate. A file still growing is skipped and retried next cycle.

import time
from dataclasses import dataclass


@dataclass
class Candidate:
    name: str
    size: int


def stable_files(sftp, remote_dir: str, checks: int = 2,
                 interval: float = 5.0) -> list[Candidate]:
    """Return only files whose size held constant across `checks` polls."""
    history: dict[str, list[int]] = {}
    for _ in range(checks):
        for attr in sftp.listdir_attr(remote_dir):
            history.setdefault(attr.filename, []).append(attr.st_size)
        time.sleep(interval)

    stable = []
    for name, sizes in history.items():
        # Present in every poll AND size never changed AND non-empty.
        if len(sizes) == checks and len(set(sizes)) == 1 and sizes[0] > 0:
            stable.append(Candidate(name=name, size=sizes[0]))
    return stable

Common mistake: using mtime alone. A carrier that flushes in bursts can leave mtime untouched for a second while more bytes are still coming. Comparing the actual size across polls is what catches the in-progress write; mtime is at best a secondary guard.

Step 2 — Claim the file atomically by rename

rename on the same SFTP filesystem is atomic — exactly one worker’s rename succeeds, the rest get an error. Moving the file into a /processing/ subdirectory both claims it and removes it from the next directory listing, so overlapping runs cannot both grab it.

import errno


def claim(sftp, remote_dir: str, name: str, processing_dir: str) -> str | None:
    """Atomically move the file into /processing/. Returns the new path, or None if lost the race."""
    src = f"{remote_dir}/{name}"
    dst = f"{processing_dir}/{name}"
    try:
        sftp.rename(src, dst)               # atomic on a POSIX SFTP backend
        return dst
    except IOError as exc:
        if exc.errno in (errno.ENOENT, errno.EEXIST):
            return None                     # another worker already claimed it
        raise

Common mistake: “claiming” by writing a .lock sidecar file. Two workers can both stat the lock as absent and both create it; the check-then-create is not atomic. A single rename is, so let the filesystem arbitrate.

Step 3 — Dedup on content hash, not filename

Read the claimed bytes once, hash them, and consult a durable manifest. A re-drop under any filename hashes identically and is rejected; a genuinely new bill has an unseen hash. The manifest insert must be atomic so a restart mid-batch cannot double-record.

import hashlib
import structlog

logger = structlog.get_logger()


def read_and_hash(sftp, path: str) -> tuple[bytes, str]:
    with sftp.open(path, "rb") as fh:
        payload = fh.read()
    return payload, hashlib.sha256(payload).hexdigest()


def claim_hash(manifest_conn, content_hash: str, source_name: str) -> bool:
    """INSERT the hash; return True only if this call is the first to record it."""
    with manifest_conn.cursor() as cur:
        cur.execute(
            """INSERT INTO ingest_manifest (content_hash, source_name, ingested_at)
               VALUES (%s, %s, now())
               ON CONFLICT (content_hash) DO NOTHING""",
            (content_hash, source_name),
        )
        inserted = cur.rowcount == 1        # 0 rows => hash already present
    manifest_conn.commit()
    if not inserted:
        logger.info("sftp_redrop_skipped", hash=content_hash[:12], name=source_name)
    return inserted

Common mistake: deduping on (filename, size). A carrier that re-exports the same invoice gets a new timestamped name and possibly a byte or two of different whitespace only when the content genuinely differs — but same content under a new name is the common re-drop, and only a content hash catches it reliably.

Step 4 — Emit with structured logging, then move to done

Only after the manifest claim succeeds do you build the IntakeEvent and publish. Persist raw bytes first (the gateway does this), then move the file out of /processing/ into /done/ so a crash mid-publish leaves it claimed, not lost.

def ingest_stable_files(sftp, gateway, remote_dir, processing_dir, done_dir,
                        carrier_scac, manifest_conn) -> int:
    published = 0
    for cand in stable_files(sftp, remote_dir):
        claimed = claim(sftp, remote_dir, cand.name, processing_dir)
        if claimed is None:
            continue                        # lost the race; another worker owns it
        payload, digest = read_and_hash(sftp, claimed)
        if not claim_hash(manifest_conn, digest, cand.name):
            sftp.rename(claimed, f"{done_dir}/{cand.name}")   # re-drop: archive, don't emit
            continue
        # gateway persists bytes, dedups on the same key, and publishes to the broker
        gateway.emit(payload, digest, carrier_scac, "sftp", content_type="application/edi-x12")
        sftp.rename(claimed, f"{done_dir}/{cand.name}")
        logger.info("sftp_ingested", name=cand.name, hash=digest[:12], bytes=len(payload))
        published += 1
    return published

Common mistake: deleting the file after reading instead of moving it. A delete-after-read that races a crash between publish and delete re-emits on restart; leaving files in a dated /done/ archive gives you a replay buffer and an audit trail for free.

Verification

Prove each failure is closed, not hidden. These run against a local SFTP fixture in CI:

def test_growing_file_is_not_stable(fake_sftp):
    fake_sftp.write("ABCD_partial.edi", size=400)
    fake_sftp.schedule_growth("ABCD_partial.edi", to=900)   # grows between polls
    assert "ABCD_partial.edi" not in {c.name for c in stable_files(fake_sftp, "/in", interval=0.01)}


def test_redrop_same_content_is_skipped(fake_sftp, manifest_conn):
    payload = b"ISA*..~IEA*1*000000001~"
    digest = hashlib.sha256(payload).hexdigest()
    assert claim_hash(manifest_conn, digest, "night1.edi") is True
    assert claim_hash(manifest_conn, digest, "night2_renamed.edi") is False


def test_only_one_worker_claims(fake_sftp):
    fake_sftp.write("ABCD_1.edi", size=500, stable=True)
    first = claim(fake_sftp, "/in", "ABCD_1.edi", "/processing")
    second = claim(fake_sftp, "/in", "ABCD_1.edi", "/processing")
    assert first is not None and second is None

In production the healthy signal is a steady sftp_ingested count matched by a near-zero sftp_redrop_skipped, and no growing=True file older than one extra poll interval. A spike in re-drops means a carrier stopped seeing your downstream effects — usually a broken acknowledgement further along, not an SFTP problem.

Preventive configuration

Encode the guarantees so the regression cannot return:

sftp_ingest:
  poll_interval_seconds: 30
  stability_checks: 2            # consecutive equal-size polls before claiming
  min_file_age_seconds: 10       # quiet period; belt-and-suspenders with size check
  claim_strategy: rename_into_processing
  dedup: content_sha256          # never filename or size
  on_redrop: archive_to_done     # keep for replay/audit, never re-emit
  manifest_retention_days: 90

• Atomic manifest. Back the manifest with a UNIQUE(content_hash) constraint and ON CONFLICT DO NOTHING, so concurrency and restarts are handled by the database, not application logic.
• Alert on stuck files. Anything sitting in /processing/ longer than a few intervals is an orphaned claim from a crashed worker — sweep it back to the inbox on a schedule.
• Shared canonical key. The content hash used here is the same dedup_key the Ingestion Transport Protocols gateway keys on, so the file-level and event-level dedup never disagree.

FAQ

Why not just wait a fixed number of seconds after a file appears and then read it?

Because upload duration is not fixed. A 200 KB interchange finishes in under a second; a batched 80 MB drop over a slow VAN link can take minutes. A fixed sleep is either too short (you still read a partial file) or wastefully long. Comparing the file’s size across consecutive polls adapts to the actual transfer and only releases the file once it has genuinely stopped growing.

The carrier re-drops the same invoice under a new timestamped filename. How do I catch that?

Dedup on the SHA-256 of the file contents, not the name. Byte-identical content produces an identical hash regardless of filename, so a re-drop is rejected by the manifest before it becomes a duplicate ledger row. Filename or size-based dedup misses exactly this case, which is the most common carrier re-drop pattern.

Two poller instances run for redundancy. How do I stop both from ingesting the same file?

Claim by rename into a /processing/ directory. On a POSIX SFTP backend a rename is atomic, so exactly one instance’s rename succeeds and the other gets an error and moves on. Do not use a lock sidecar file — creating it is a non-atomic check-then-write that both workers can win.

Should I delete files after ingesting them?

Move them to a dated /done/ archive instead. A delete-after-read that crashes between the broker publish and the delete will re-emit the file on restart. An archive gives you an idempotent move, a replay buffer if a downstream parser needs the original bytes, and an audit trail of exactly what arrived and when.

Related

• Ingestion Transport Protocols — the parent transport edge this SFTP adapter plugs into.
• Setting Up AS2 Transmission for EDI Freight Bills — the acknowledged-transport sibling where dedup keys off an AS2 message-id instead of a content hash.
• Async Batch Processing Workflows — how the emitted intake events fan out across workers once the files are safely on the broker.
• EDI 210/810 Processing — the parser that consumes the whole interchanges this poller guarantees.

Up one level: Ingestion Transport Protocols · Section: Automated Invoice Parsing & EDI/XML Ingestion