Idempotent File Anchoring: SHA-256 Dedup Before You Call the API

Building any intake pipeline, you'll hit the same problem eventually. Files arrive from multiple sources. Some you've already processed: re-uploads of the same document, copies from two different intake paths, items your worker errored on last run and re-queued. Call the anchoring API blindly and you end up with multiple proof records for identical bytes.

The ProofLedger v1 API returns a duplicate_of field in its 201 response when it detects a hash it's already seen. But that's only half the solution. A network round-trip costs time and quota even when it comes back as a duplicate. Hash-based local deduplication is the other half.

Here's how to build a worker that handles both layers.

Hash Locally First

The core pattern: compute the SHA-256 digest before making any API call. If you've seen this digest before, skip it. If you haven't, submit it.

Two things you need: a persistent record of digests you've already anchored, and chunked hashing so large files don't blow memory.

import hashlib
import json
from pathlib import Path

SEEN_DB = Path("anchored_hashes.json")

def load_seen():
    if SEEN_DB.exists():
        with open(SEEN_DB) as f:
            return json.load(f)
    return {}

def save_seen(db):
    with open(SEEN_DB, "w") as f:
        json.dump(db, f, indent=2)

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

65536-byte chunks keep memory flat regardless of file size. The load_seen / save_seen pair gives you a persistent record that survives worker restarts.

Submitting and Reading duplicate_of

When duplicate_of appears in the API response, its value is the proof ID of the earliest anchor for that hash. That's the canonical ID. The new proof ID from this call is irrelevant.

import requests

API_URL = "https://proofledger.io/api/v1/proof"
API_KEY = "sk_YOUR_KEY_HERE"

def anchor_file(file_path: str, seen: dict) -> dict:
    digest = hash_file(file_path)

    if digest in seen:
        return {
            "status": "skipped",
            "reason": "already_anchored",
            "digest": digest,
            "original_proof_id": seen[digest]["proof_id"],
        }

    response = requests.post(
        API_URL,
        headers={"Authorization": f"Bearer {API_KEY}"},
        json={
            "sha256": digest,
            "filename": Path(file_path).name,
            "bitcoin_requested": True,
        },
        timeout=30,
    )

    if response.status_code == 201:
        data = response.json()

        if data.get("duplicate_of"):
            canonical_id = data["duplicate_of"]
            seen[digest] = {"proof_id": canonical_id}
            save_seen(seen)
            return {
                "status": "duplicate",
                "canonical_proof_id": canonical_id,
                "digest": digest,
            }

        proof_id = data["id"]
        seen[digest] = {"proof_id": proof_id}
        save_seen(seen)
        return {"status": "anchored", "digest": digest, "proof_id": proof_id}

    if response.status_code == 429:
        retry_after = response.headers.get("Retry-After", "unknown")
        raise RuntimeError(f"Rate limited. Retry-After: {retry_after}")

    if response.status_code in (400, 401, 403):
        raise RuntimeError(f"Non-retryable {response.status_code}: {response.text}")

    raise RuntimeError(f"Unexpected status {response.status_code}")

The 429 path raises instead of silently skipping. That's by design. Backoff logic belongs at the call site, where you can log which file triggered the limit and decide whether to wait or drain a queue before retrying.

One Hash, One Record

When duplicate_of comes back, the earliest anchor is already established. That's your temporal claim. Any later anchor for the same bytes doesn't strengthen anything. In a legal or claims context, it creates confusion about which proof is authoritative.

The rule is simple: one hash, one canonical proof ID. Store the duplicate_of value and discard the rest.

import os

def process_directory(directory: str):
    seen = load_seen()
    counts = {"anchored": 0, "skipped": 0, "duplicate": 0, "errors": 0}

    for root, _, files in os.walk(directory):
        for filename in files:
            file_path = os.path.join(root, filename)
            try:
                result = anchor_file(file_path, seen)
                status = result["status"]
                counts[status] += 1
                short_hash = result["digest"][:12]
                print(f"{status.upper()}: {filename} ({short_hash}...)")
            except RuntimeError as e:
                counts["errors"] += 1
                print(f"ERROR: {filename}: {e}")

    print(f"\nDone. {counts}")

Point this at a directory with a mix of new and previously processed files. The output breaks down new anchors, local hits, API-level duplicates, and errors. Everything accounted for before you report results upstream.

Closing the Crash Window with SQLite

One failure mode to think through: your worker anchors a file, gets a 201 back, then crashes before writing to anchored_hashes.json. On the next run, it re-hashes the same file, hits the API again, and receives duplicate_of. No data loss; one wasted API call.

For small batches, that's acceptable. For anything serious, swap the JSON file for SQLite with a unique constraint on the digest column.

import sqlite3

def init_db(db_path="anchored.db"):
    conn = sqlite3.connect(db_path)
    conn.execute("""
        CREATE TABLE IF NOT EXISTS anchors (
            digest TEXT PRIMARY KEY,
            proof_id TEXT NOT NULL,
            anchored_at TEXT
        )
    """)
    conn.commit()
    return conn

def is_seen(conn, digest):
    row = conn.execute(
        "SELECT proof_id FROM anchors WHERE digest = ?", (digest,)
    ).fetchone()
    return row[0] if row else None

def record_anchor(conn, digest, proof_id, anchored_at=None):
    conn.execute(
        "INSERT OR IGNORE INTO anchors (digest, proof_id, anchored_at) VALUES (?, ?, ?)",
        (digest, proof_id, anchored_at),
    )
    conn.commit()

INSERT OR IGNORE handles concurrent workers cleanly. The first process to insert wins. The second sees the row already there and skips. If it got far enough to call the API, it receives duplicate_of and lands on the same canonical ID anyway.

Write the row immediately after getting the 201. Don't batch the writes. A crash between anchoring and writing costs one API call per file on retry. A crash between a full batched anchor run and a batched write costs the whole batch.

What to Build Next

Wire a rate-limit backoff loop around the anchor_file call. On 429, read Retry-After from the response header and sleep for exactly that many seconds before retrying. Don't spin; the header tells you what the server needs.

If intake volume outpaces synchronous directory walking, feed file paths into a queue and run workers in parallel. The dedup layer here is already concurrency-safe via INSERT OR IGNORE.

Third-party verification doesn't need your API key. The public GET /api/v1/verify?hash=<sha256> endpoint is unauthenticated, rate-limited to 120 requests per hour per IP, and returns the full proof record including blockchain explorer URLs. An auditor or opposing counsel can verify the anchor themselves without touching your credentials.

For offline verification once you've got a proof record, the verify-proof package on PyPI checks the hash and walks the Merkle path locally, no network call required.

What about intake systems that already anchored duplicates before you added this dedup layer? The duplicate_of field handles that retroactively too. Feed your archive through the worker above, collect the canonical IDs, and let the others go. The earliest anchor is still on-chain, unchanged.

Have you built dedup into a file ingestion pipeline before? What did you end up using for the seen-hash store?