You're processing a folder of evidence files and some of them were already anchored last week. Your script crashed at file 47 and you need to rerun it. Two services are ingesting from the same drop zone at the same time.
All three problems have the same root cause: no deduplication. And they all produce the same outcome: redundant submissions that waste API quota, create ambiguous records, and make downstream bookkeeping messy.
The fix isn't complicated. An idempotent worker checks before it submits. And when the race window catches it anyway, it handles the duplicate_of response correctly.
Check the Hash Before You Touch the API
The first move is a pre-flight check. Before calling the write endpoint, ask the public verification endpoint if the hash already exists:
GET https://proofledger.io/api/v1/verify?hash=<sha256>No auth required. Returns { "found": true, "proof": {...} } or { "found": false, "sha256": "..." }. Rate-limited at 120 requests per hour per IP, so it costs nothing in API quota and nothing in authentication overhead.
Here's the hash computation and pre-flight:
import hashlib
import requests
VERIFY_URL = "https://proofledger.io/api/v1/verify"
def sha256_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()
def check_existing(file_hash: str) -> dict | None:
resp = requests.get(VERIFY_URL, params={"hash": file_hash}, timeout=10)
resp.raise_for_status()
data = resp.json()
return data.get("proof") if data.get("found") else NoneIf check_existing() returns something, you're done for that file. Store the proof ID and move on. If it returns None, proceed to submit.
One thing worth caching locally: once a hash is anchored, it doesn't become unanchored. So if you've seen found: true for a hash, write it to disk and skip the pre-flight forever on that hash. That's how you stay well under the 120/hr rate limit across large batches.
Submitting and the duplicate_of Field
Pre-flight check eliminates most duplicates. But two workers running concurrently can both see found: false for the same hash and both POST it within milliseconds of each other. The API handles this case without error. When the second submission lands, the response includes a duplicate_of field pointing to the ID of the earlier proof.
SUBMIT_URL = "https://proofledger.io/api/v1/proof"
API_KEY = "sk_YOUR_KEY_HERE"
def submit_proof(file_hash: str, filename: str | None = None) -> dict:
headers = {"Authorization": f"Bearer {API_KEY}"}
body = {"sha256": file_hash}
if filename:
body["filename"] = filename
resp = requests.post(SUBMIT_URL, json=body, headers=headers, timeout=15)
resp.raise_for_status()
return resp.json()
def canonical_id(submission: dict) -> str:
return submission.get("duplicate_of") or submission["id"]Always store canonical_id(submission), not submission["id"]. If duplicate_of is present, the earlier proof is the canonical one. A third-party verifier, an auditor, or opposing counsel checking your evidence records should see the same proof ID you have on file. Using a later duplicate's ID creates a mismatch that's technically harmless but annoying to explain.
Checking Anchor Status
After a successful POST, the proof sits in a pending state while the blockchain transaction propagates. You can poll GET /api/v1/proof/:id (auth required, owner only) to check when it confirms:
import time
PROOF_URL = "https://proofledger.io/api/v1/proof"
def wait_for_anchor(proof_id: str, max_polls: int = 12, interval: int = 15) -> dict | None:
headers = {"Authorization": f"Bearer {API_KEY}"}
for attempt in range(max_polls):
resp = requests.get(f"{PROOF_URL}/{proof_id}", headers=headers, timeout=10)
resp.raise_for_status()
data = resp.json()
if data.get("polygon_status") == "anchored":
return data
if attempt < max_polls - 1:
time.sleep(interval)
return NoneFor batch jobs, you usually don't want to block per file waiting for confirmation. Submit everything, store the IDs, check status later. Polygon confirmation is fast. ProofLedger's Bitcoin anchoring runs on a daily batch with Merkle proofs, so that chain takes longer by design.
A Complete Worker
Here's the full thing: hash, pre-flight, submit with retry, handle duplicate_of, save progress.
import hashlib
import json
import os
import time
import requests
VERIFY_URL = "https://proofledger.io/api/v1/verify"
SUBMIT_URL = "https://proofledger.io/api/v1/proof"
API_KEY = "sk_YOUR_KEY_HERE"
PROGRESS_FILE = "anchored.json"
def sha256_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()
def load_progress() -> dict:
if os.path.exists(PROGRESS_FILE):
with open(PROGRESS_FILE) as f:
return json.load(f)
return {}
def save_progress(progress: dict) -> None:
with open(PROGRESS_FILE, "w") as f:
json.dump(progress, f, indent=2)
def anchor_file(path: str, progress: dict) -> None:
file_hash = sha256_file(path)
if file_hash in progress:
print(f" skip (cached): {path}")
return
verify_resp = requests.get(VERIFY_URL, params={"hash": file_hash}, timeout=10)
verify_resp.raise_for_status()
verify_data = verify_resp.json()
if verify_data.get("found"):
proof_id = verify_data["proof"]["id"]
progress[file_hash] = {"id": proof_id, "source": "pre-flight"}
print(f" found (pre-flight): {path} -> {proof_id}")
save_progress(progress)
return
headers = {"Authorization": f"Bearer {API_KEY}"}
body = {"sha256": file_hash, "filename": os.path.basename(path)}
for attempt in range(3):
try:
resp = requests.post(SUBMIT_URL, json=body, headers=headers, timeout=15)
if resp.status_code == 429:
retry_after = int(resp.headers.get("Retry-After", 60))
print(f" rate limited, sleeping {retry_after}s...")
time.sleep(retry_after)
continue
if resp.status_code in (400, 401, 403):
print(f" non-retryable {resp.status_code}: {path}")
return
resp.raise_for_status()
data = resp.json()
proof_id = data.get("duplicate_of") or data["id"]
source = "duplicate_of" if data.get("duplicate_of") else "submitted"
progress[file_hash] = {"id": proof_id, "source": source}
print(f" {source}: {path} -> {proof_id}")
save_progress(progress)
return
except requests.Timeout:
if attempt == 2:
print(f" timeout after 3 attempts: {path}")
return
time.sleep(2 ** attempt)
def run(directory: str) -> None:
progress = load_progress()
files = [
os.path.join(root, fname)
for root, _, filenames in os.walk(directory)
for fname in filenames
]
print(f"{len(files)} files in {directory}")
for path in files:
anchor_file(path, progress)
if __name__ == "__main__":
import sys
run(sys.argv[1] if len(sys.argv) > 1 else ".")A few design notes worth calling out.
Progress is keyed by hash, not by path. If a file gets renamed or moved, it won't be re-submitted. Hash identity is what the anchoring system cares about, so that's what your progress tracker should care about too.
The 429 handler reads Retry-After from the response header instead of sleeping a fixed amount. Hardcoded sleep intervals are almost always wrong. The server's value accounts for its current load and rate window.
Non-retryable errors (400/401/403) exit immediately. A 400 means your hash is malformed. A 401/403 means your key is wrong or expired. Neither gets better with retries.
Running It
pip install requests
python anchor_worker.py ./evidence-files/On the first run, it hits the verify endpoint for every file it hasn't seen. On reruns, anything in anchored.json is skipped entirely, no network calls. Anything found via pre-flight is recorded without touching your submission quota. Only genuinely new hashes reach the POST endpoint.
If you want to check a specific file's proof later, the id in your progress file maps to the certificate: https://proofledger.io/cert/<id>. And the public verify endpoint works independently for anyone who needs to confirm a hash without your records:
GET https://proofledger.io/api/v1/verify?hash=<sha256>That independence is the point. I made that endpoint public so your database isn't the only source of truth for a file's anchor. Anyone can check. That's what a neutral anchor is for.
---
Interested in the Merkle proof structure that backs the Bitcoin anchoring side? Source for the verify-proof PyPI package is at https://github.com/Fulcrum-Enterprises/verify-proof if you want to dig into the offline verification path.
What are you anchoring? Drop a comment with your use case.