A records export should be safe to run twice.

The pipeline behind this note was built in 2025, before Kernwise existed. It exported patient records and scanned documents from an EHR/EMR system into Azure Blob Storage, indexed them in Azure AI Search, and served keyword and filtered search through FastAPI endpoints. The set ran to tens of thousands of records, processed in incremental batches. This note is about the one property of that kind of pipeline that decides how painful it is to operate: whether running it a second time is safe.

The project is listed on the work page. The code below is a reduced sketch of the pattern, written for this note. It is not the code from that project.

The first run is not the last run

A records export is usually described as a migration, something that happens once. In practice it runs many times over the same documents. A malformed file can stop a batch halfway. The search service can answer 503 under load. A review can find a field the sanitiser should have removed, and then every document that carried it has to go through again. A field can need a different analyzer, which in Azure AI Search means a new index and a full reload.

If each of those events means starting from zero, or worse, means guessing which documents made it and which did not, the pipeline becomes the problem. The design goal is narrower than speed and more useful: any run, interrupted at any point and started again, converges on the same end state.

Three things get you most of the way there. Keys that do not depend on the run. A ledger that records what happened to each document at each stage. And a run definition that is data rather than code.

Keys that do not depend on the run

Every document needs an identity derived from the source, not from the run that happened to process it. If the key is a row number, a timestamp or a freshly generated UUID, a rerun creates duplicates instead of overwriting.

Azure AI Search is strict about keys: letters, digits, underscores, dashes and equal signs only, at most 1,024 characters, and the first character cannot be an underscore. Source identifiers often contain slashes and colons. A hash of the source identity sidesteps all of it and gives a fixed-length key that also works as a blob name.

import hashlib

def doc_key(source_system: str, record_id: str, part: str) -> str:
    """Same source document, same key, on every run."""
    identity = f"{source_system}:{record_id}:{part}"
    return hashlib.sha256(identity.encode("utf-8")).hexdigest()

A hex digest is 64 letters and digits, so it passes the key rules without any encoding. The same key names the blob, the ledger row and the search document, which turns tracing one document through every stage into a single lookup.

With a stable key, writes become overwrites, and the write action starts to matter. Azure AI Search offers upload, which replaces the whole document, and mergeOrUpload, which updates only the fields you send. On a first run they behave the same. On a rerun they do not. If a sanitisation change means a field is no longer sent at all, mergeOrUpload leaves the old value sitting in the index, because fields missing from a merge keep their existing values. When the reason for the rerun is that something should no longer be there, use upload.

A ledger, not a log

Logs tell you what happened. A ledger tells you what state each document is in, and that is the question a rerun has to answer. One row per document per stage is enough.

CREATE TABLE export_ledger (
  doc_key        CHAR(64)     NOT NULL,
  stage          VARCHAR(16)  NOT NULL,  -- export, sanitise, index
  content_sha256 CHAR(64)     NOT NULL,
  rules_version  INT          NOT NULL,  -- sanitiser version used
  status         VARCHAR(12)  NOT NULL,  -- pending, done, failed,
                                         -- quarantined
  attempts       INT          NOT NULL DEFAULT 0,
  last_error     VARCHAR(200) NULL,      -- class and code only
  updated_at     DATETIME2    NOT NULL,
  PRIMARY KEY (doc_key, stage)
);

Two columns do most of the work. content_sha256 lets a run skip a document whose source has not changed since it was last exported. rules_version records which version of the sanitiser produced the stored copy, so raising the version marks every older copy as stale without touching a single file.

The last_error column deserves a rule of its own. In a pipeline that handles patient records, an error message that echoes a field value is a copy of that value in a place nobody reviews. Store the error class and a status code, and nothing from the document itself.

A run then starts by asking the ledger for its work:

SELECT doc_key
FROM export_ledger
WHERE stage = 'index'
  AND attempts < @max_attempts
  AND (status IN ('pending', 'failed')
       OR rules_version < @rules_version);

Stop the run anywhere and the next one picks up exactly the documents that are not done. Nothing is reprocessed because a run was restarted, and nothing is skipped because a run was assumed to have finished.

Retry what is temporary, quarantine what is not

Azure AI Search accepts up to 1,000 documents or 16 MB per indexing request, whichever limit comes first. Scanned documents with extracted text reach the byte limit long before the count limit, so batch by size as well as by count.

import json

MAX_DOCS = 1000
MAX_BYTES = 15 * 1024 * 1024  # headroom under the 16 MB request limit

def batches(docs):
    batch, size = [], 0
    for doc in docs:
        n = len(json.dumps(doc).encode("utf-8"))
        if batch and (len(batch) == MAX_DOCS or size + n > MAX_BYTES):
            yield batch
            batch, size = [], 0
        batch.append(doc)
        size += n
    if batch:
        yield batch

A batch can partly succeed. The service answers 207 and reports a status for each document, and those codes split cleanly into two groups. 409, 422 and 503 are temporary: a concurrent write to the same document, an index briefly offline, a service under load. 400 and 404 are not: the document itself is wrong, and sending it again will fail again.

RETRYABLE = {409, 422, 503}

def record_results(ledger, results):
    for r in results:  # key, succeeded, status_code
        if r.succeeded:
            ledger.mark(r.key, "index", "done")
        elif r.status_code in RETRYABLE:
            ledger.fail(r.key, "index", code=r.status_code)  # attempts += 1
        else:
            ledger.mark(r.key, "index", "quarantined", code=r.status_code)

Quarantine matters as much as retry. A document that will never index should stop consuming attempts and stop appearing in every run's failure count, where it hides the failures that are new. It goes on a list a person reads.

A run full of 503s is a signal to slow down, not to retry harder. Microsoft's own guidance is that if indexing traffic does not ease off, the service can start rejecting every request. Back off between batches with an exponential delay, and let the ledger, not an in-memory loop, carry retry state from one run to the next.

Let configuration choose what a run does

Most reruns are partial. Re-index one document type after a mapping change. Re-sanitise everything produced under an older rules version. Rebuild into a new index after changing an analyzer, which Azure AI Search does not allow on a field that already exists. If each of these needs a code change, each one is a deploy, and a deploy is the wrong unit for an operational task.

The alternative is a run file: a small, reviewed piece of data that says what this run is for.

{
  "stages": ["sanitise", "index"],
  "select": { "document_types": ["scanned_document"] },
  "rules_version": 4,
  "target_index": "records-v3",
  "write_mode": "upload",
  "batch": { "max_docs": 500, "max_bytes": 15728640 },
  "dry_run": true
}

One file per run gives an audit trail at no extra cost: what each run was asked to do sits next to what the ledger says it did. A dry run that prints counts from the ledger before anything is written catches the selection mistake that would otherwise re-index the whole set. And because the search API reads its index name from configuration too, moving readers to records-v3 is a configuration change rather than a release.

Sanitise before storage, and allowlist the index

One ordering decision carries most of the privacy weight. Sanitise before a document lands in storage, not before it is indexed, so no stage downstream ever holds the unsanitised form. Then build each search document from an explicit list of fields, never by passing the stored record through whole.

INDEX_FIELDS = ("doc_key", "document_type", "created_on", "text")

def to_search_document(record: dict) -> dict:
    return {field: record[field] for field in INDEX_FIELDS}

A new field added upstream then stops at the allowlist instead of appearing quietly in search results. Put together with rules_version in the ledger and upload as the write mode, a change to the sanitiser becomes an ordinary rerun: raise the version, select the stale rows, replace them.

What this leaves open

None of this makes search results good. Ranking quality, and automated checks that sanitised output really is clean, are separate pieces of work that this design does not cover. What it does guarantee is that when either of them changes something, getting the change into the index is a run you can start, stop and start again.