# Real-Time Matching Tutorial

_Version 1.0, August 3, 2026_

This tutorial describes how to match records in **real time** with `madmatcher-pro`: instead of running a batch job over two whole tables, you match one incoming record (or a live stream of them) against a fixed set of your own records the moment it arrives. We first discuss how real-time matching works and how it relates to the batch pipeline you may already know from Sparkly, Delex, Semantic blocking, and MatchFlow, and then provide a step-by-step guide that walks you through publishing a serving bundle, loading it, and every way you can drive requests through it. You should read the "How Real-Time Matching Works" section first, because it defines the terms used later.

Real-time matching is a **separately-licensed add-on**. `RealtimeMatcher.load` checks that your license includes the `realtime` entitlement and raises a `LicenseError` without it; a base (batch) license does not grant it. Contact support@madmatcher.ai to enable it.

---

## How Real-Time Matching Works

The batch pipeline answers "which pairs across these two tables are the same entity?" by blocking, featurizing, and predicting over both tables at once on Spark. Real-time matching answers a narrower, faster question: **"which of my known records does this one incoming record match?"**, and answers it immediately. It runs the same three steps (block, featurize, predict) for a single record, from state prepared once and loaded into memory.

Three ideas carry the whole design:

1. **Build once, offline.** Everything that depends only on your corpus, the search index, the preprocessed corpus features, and the trained model, is produced by an ordinary batch job and frozen into a **bundle** on disk. This is the expensive part, and it happens away from the request path.
2. **Load resident.** A serving process loads the bundle into memory once. From then on the index, the corpus feature state, and the model live in the process.
3. **Answer per record.** For each incoming record the matcher **blocks** it against the corpus (retrieving a few candidates), **featurizes** each (record, candidate) pair with the exact batch featurizer, and **predicts**. Because the kernels are the batch kernels, a served prediction equals the batch prediction for the same record and corpus.

We reuse the terminology of the other tutorials, adapted to one-record-at-a-time serving:

- The **corpus** is the fixed set of records you match against. It plays the role of the batch **indexing table** (table A): the bundle is built over it once, and every incoming record is compared to it.
- An **incoming record** is the record you want to match, arriving live or from a stream. It plays the role of one row of the batch **search table** (table B).
- A **candidate** is a corpus record that blocking retrieves as a possible match.
- A **match** is a candidate the model predicts is the same entity.
- A **bundle** is the on-disk package a serving process loads: the index, the corpus feature state, the model, and a manifest that pins the configuration so serving cannot drift from training.

For this tutorial we assume a corpus of catalog products you already matched and cleaned in a batch job, and incoming product records (new listings, supplier feeds, marketplace items) that you match against it as they arrive.

**Your corpus (matched against)**

| \_id | name                       | brand    | price |
| ---- | -------------------------- | -------- | ----- |
| 0    | Sony WH-1000XM5 headphones | Sony     | 399   |
| 1    | Logitech MX Master 3S      | Logitech | 99    |
| …    | …                          | …        | …     |

**An incoming record (matched live)**

| name                                            | brand | price |
| ----------------------------------------------- | ----- | ----- |
| Sony wireless over-ear noise-cancelling headset | Sony  | 349   |

Matching that incoming record returns the corpus records the model predicts it is the same as:

| Column       | Type           | Meaning                                                                                  |
| ------------ | -------------- | ---------------------------------------------------------------------------------------- |
| `id1`        | long           | the matched **corpus** record's id                                                       |
| `id2`        | long or string | the **incoming** record's key: its own id (from `id_col`), or a correlation key you pass |
| `prediction` | float          | `1.0` on every row (the frame holds only predicted matches)                              |
| `confidence` | float          | the model's confidence for that pair, highest first                                      |

The result is a `pandas.DataFrame`, already filtered to predicted matches. An incoming record with no match returns an empty frame with those same columns, so "no match" is distinct from "not processed".

One typing difference from batch is deliberate. In batch, both tables are indexed, so both id columns must be 32- or 64-bit integers; `id1` keeps that rule here, because the corpus is still an indexed table. The incoming record, though, is never indexed: `id2` only identifies the request in its results, so it is not restricted to an integer. An integer id and a string key (a request id, a queue offset, a UUID) both work; pick one kind and use it consistently within a run, because a sink writing to typed storage such as parquet cannot store a column that mixes the two.

`id1` and `id2` are the **result frame's** column names only (the same pair naming batch featurize uses); no input needs a column named either. The corpus carries its ids in a column named **`_id`** (Step 1.1 covers this requirement), and an incoming record carries its own id, when it has one, in the column named by `id_col` (default `_id` as well). Serving maps them into `id1` and `id2` on the way out.

### The choices you make

Standing up serving comes down to a few independent choices, each covered in the steps below:

- **Which blocker the bundle carries** (Step 1): Sparkly (lexical), Delex (lexical), Semantic (dense), a fusion of one lexical blocker plus Semantic, or none at all (you supply candidates per request).
- **Whether the model uses the semantic cosine feature** (Step 1): if you trained with it, serving reproduces the per-candidate cosine from the semantic blocker's scores, from a stored corpus embedding table, or both.
- **How an incoming record is embedded** (Steps 1 and 2), whenever the bundle blocks semantically or reproduces the cosine feature: in-process by a resident embedder (you publish the same provider you used in batch), or attached to the request as a precomputed `probe_embedding` (which is required when the bundle carries no embedder).
- **Whether the matcher is sklearn or SparkML** (Step 1): sklearn/xgboost predicts in-process with no Spark on the request path; SparkML runs one Spark job per request.
- **How you drive requests** (Steps 2 to 8): a single call, a batch, a concurrent pool, or an orchestrated stream, plus how you hand off from a batch build and refresh the corpus over time.

---

## Durability and Progress Tracking

Real-time matching is one of `madmatcher-pro`'s premium features. Two capabilities apply to the operations that run long or must not drop records, both opt-in:

- **Durability.** The orchestrated streaming runner has a durable form (`run_matching_durable`) that records per-record completion, so a crashed run resumes where it stopped, and several workers pointed at the same body of incoming records can divide it between them with no broker or queue. Publishing a bundle can likewise be made crash-resumable, and the batch-to-serving handoff spools arriving records durably so none is lost while the build runs.
- **Progress tracking.** Every serving surface can report live throughput to the same web dashboard the batch engines use (`track=True`), and a publish shows up as a job while it runs.

The [Reference: Durability and Progress Tracking](#reference-durability-and-progress-tracking) section at the end covers the claim protocol, the delivery guarantee, the dashboard, and every argument in full.

---

## Step-by-Step Walkthrough

This section walks through writing serving into your application, in the following steps:

1. Publishing a Serving Bundle
2. Loading the Matcher and Matching One Record
3. Matching a Batch
4. Serving Concurrent Requests
5. Streaming Records Through a Matcher
6. Durable and Multi-Worker Streaming
7. Handing Off From a Batch Build
8. Refreshing the Bundle Without Downtime

Steps 1 and 2 are the minimal path: publish a bundle, then match against it. Steps 3 to 8 add the modes you reach for as your workload grows: batches, concurrency, streaming, durability, and live corpus refresh. Read Steps 1 and 2 first; the rest are independent and you can jump to the one you need.

### Step 1: Publishing a Serving Bundle

Everything the real-time matcher serves from is prepared in this step, in two parts. In **Step 1.1** you produce the batch artifacts serving reuses: a blocker built over your corpus, and a trained matcher. These are made with the engines the other tutorials teach, so this part tells you what to build, which tutorial shows how, and what to keep when you are done. In **Step 1.2** you assemble those artifacts into a **serving bundle** with one call, `publish_serving_artifacts`. The result is a self-contained folder on disk; every later step starts by loading it.

#### Step 1.1: Building the Batch Artifacts

Entity matching runs in two steps: **blocking**, which finds, for a record, a small set of likely matches among your corpus records (its candidates), and **matching**, which predicts which of those candidates truly match. The real-time matcher performs both steps for each incoming record, so the bundle carries an artifact for each, and you have options on both sides.

**The blocking side.** Choose what will find candidates for each incoming record:

- **Sparkly** finds candidates by keyword similarity (BM25) over an inverted index built on your corpus. To review how Sparkly works and build the index, see the [Sparkly tutorial](../sparkly/sparkly-tutorial.md); its Step 3 builds the index. Keep the index directory when you are done (this guide uses `build/lucene`); publishing copies it into the bundle.
- **Delex** finds candidates with blocking rules you write. To review how Delex works and construct a program, see the [Delex tutorial](../delex/delex-tutorial.md); its Step 3 constructs the `BlockingProgram`. Keep the program object and the corpus DataFrame it applies to; there is no index directory to keep, because serving stores both in the bundle and rebuilds Delex's state when the bundle is loaded.
- **Semantic** finds candidates by meaning, comparing embedding vectors. To review how semantic blocking works, embed your corpus, and build the index, see the [semantic tutorial](../semantic/semantic-tutorial.md); its Steps 3 and 4 embed the records and build the index. Keep three things: the index directory, the embedding provider you used (for example the `SentenceTransformerProvider` and its model id), and the list of fields you embedded. Serving reads the default (flat) index; an index built with a compression codec is refused at publish.
- **Bring your own candidates.** Build no blocker at all. Each request then passes the record's candidate ids itself (`match(record, candidates=[...])`, shown in Step 2), for when another system already produces candidates. Nothing to keep.

A bundle can carry Sparkly or Delex (not both), can add semantic beside either one (their candidates are then fused per record, as in batch fusion), can carry semantic alone, or can carry no blocker at all.

**One id column ties everything together.** Your corpus must carry a unique 32- or 64-bit integer id column named **`_id`**: the feature state serving loads is keyed by that exact name, and a result's `id1` is one of its values. Build every blocker over that same column (`IndexConfig(id_col="_id")` for Sparkly, `delex_id_col` for Delex, the semantic index's `id_col`), so the ids a blocker retrieves are the ids the feature state and your corpus know. `id1` and `id2` exist only in the result frame; no input table needs a column with those names.

**If you bring your own embeddings.** Semantic blocking and the cosine feature (Step 1.2) work with embeddings in two shapes:

- The **corpus embedding table**: one row per corpus record, holding the record's id and its vector, as an id column plus an array-of-float column with unique ids. This is exactly the table `create_embeddings` writes in the semantic tutorial's Step 3, so if you followed that tutorial you already have it. A table you embedded outside MadMatcher works too, in that same shape.
- A **per-request probe embedding**: one plain float vector (for example a numpy array) for one incoming record, produced by the same model as the corpus vectors. Step 2 shows where a request passes it.

**The matching side.** The matcher is a model trained with MatchFlow; the [MatchFlow technical guide](../matchflow/matchflow-technical-guide.md) covers that whole flow: creating features, featurizing candidate pairs, getting labels (from labeled data you already have, from seeds, or through active learning, with or without down-sampling), and training. Whichever route you take, serving needs three things out of that run:

- **The trained model**: a `SKLearnModel` (wrapping an sklearn or xgboost estimator) or a `SparkMLModel`.
- **The feature list** that `create_features` produced and `featurize` used. Serving uses the same list to build each incoming pair's feature vector exactly as training did.
- **The `fill_na` value** you passed to `featurize`. Serving fills missing feature values with the same number, and it cannot be recovered from the model afterwards, so note it down.

**Carrying the artifacts to the publish script.** If you publish in the same script that finishes the build, hold the corpus, the features, and the model in memory and continue to Step 1.2. If you build and publish in separate scripts, MatchFlow has a save/load helper for each artifact. In the build script:

```python
from madmatcher_pro.matchflow import save_features, save_dataframe

model.save("build/model")
save_features(features, "build/features.pkl")
save_dataframe(corpus, "build/corpus")
```

And in the publish script:

```python
from madmatcher_pro.matchflow import MLModel, load_features, load_dataframe

model = MLModel.load("build/model")
features = load_features("build/features.pkl")
corpus = load_dataframe("build/corpus", "sparkdf")
```

The reloaded model predicts identically to the one you trained. The index directories are already on disk where you built them (publishing reads them by path), and a Delex `BlockingProgram` is a plain Python object you can rebuild in the publish script or pickle alongside the rest.

#### Step 1.2: Publishing the Bundle

With the artifacts in hand, one call assembles the bundle. `publish_serving_artifacts` takes the output folder plus four required arguments:

- **`out_dir`** (the first, positional argument) is the folder the bundle is written into. Use a fresh folder per publish (more on this below).
- **`indexed_table`** is the corpus as a Spark DataFrame: the records every incoming record is matched against.
- **`features`** is the feature list from Step 1.1.
- **`model`** is the trained matcher from Step 1.1.
- **`fill_na`** is the value you gave `featurize` in training. It is required and must match exactly: a different value silently changes missing-value features at serve time, and it cannot be validated against the model, so publishing refuses to guess it.

The blocker is passed alongside, per option (each shown below). A minimal Sparkly-blocked publish:

```python
from madmatcher_pro.serving import publish_serving_artifacts

publish_serving_artifacts(
    "bundles/v1",
    indexed_table=corpus,                 # the corpus (a Spark DataFrame)
    features=features,                    # the feature list from training
    model=model,                          # the trained model
    sparkly_index_path="build/lucene",    # your built Sparkly index directory
    fill_na=0.0,                          # the same value you gave featurize()
)
```

This writes a self-contained bundle at `bundles/v1` and returns its path. The `_PUBLISHED.json` marker is written last, so a crash mid-publish leaves an incomplete bundle the loader refuses rather than a half-written one it would serve.

**Publish a new directory per version.** `publish_serving_artifacts` rewrites `out_dir` in place. That is safe for a fresh directory, but never publish over a bundle a `RealtimeMatcher` is actively serving (a live reader holds the corpus memory-map open). To refresh a live process, publish the new version to its own directory and swap it in (Step 8).

Two optional arguments matter to every bundle:

- **`id_col`** (default `"_id"`) names the column an incoming record carries its own id in. Serving stamps that id onto each result as `id2` when a request passes no explicit `key`, so a served row always says which record produced it.
- **`block_limit`** (default `50`) caps how many candidates blocking retrieves per incoming record, which is how many pairs the model scores. It is fixed in the bundle at publish.

  **If you are deduplicating, add one.** When the corpus contains the records you serve, every blocker returns each record's own row at rank 1, and that self-match occupies one of the `block_limit` slots. The batch equivalent asked for K+1 candidates and dropped the self pair, so publish `block_limit=K+1` to get the same K real candidates, and drop `id1 == id2` from the results yourself (serving has no self-match exclusion). Getting this wrong is not an error, just one candidate fewer per record, which surfaces later as a small recall gap against the batch run. `block_limit` is read from `MANIFEST.json` at load and used nowhere else, so an already-published bundle can be corrected by editing that one field instead of re-publishing.

The rest of this step shows the publish call for each blocking option from Step 1.1, then the cosine feature, then the two model kinds.

##### Publishing when the driver is not the only machine

The bundle is assembled as ordinary files on the **driver**. Two things follow from that.

**Where the bundle sits decides whether a DataFrame argument works.** `embedding_store` and `delex_corpus` are the two arguments that can be Spark DataFrames, and Spark executes a DataFrame write on the *executors*, against the destination path as each of **them** resolves it:

- **The bundle is on storage every node shares** — a single machine, a network mount (NFS and friends), a lakehouse volume. The executors write exactly where the driver reads, so passing a DataFrame is fine and needs nothing extra.
- **The bundle is on the driver's own disk while the executors are elsewhere** — the usual cluster shape. Each executor writes to *its* local disk under that same path, and the bundle gets nothing. Publishing verifies the driver can read each Spark-written artifact back, so this fails naming the cause rather than surfacing steps later as a parquet schema error.

The question is where the *bundle* is, not whether the cluster has a distributed filesystem: a `file://` write ignores `fs.defaultFS` entirely, so having HDFS or GCS available does not by itself make a DataFrame argument work.

For the second shape, stage the table once to a filesystem **both sides can see** and pass that URI. Both arguments accept a path or URI on any filesystem Spark can address (`hdfs://`, `s3a://`, `gs://`, `abfss://`, or a local/shared-mount path), and publishing pulls it into the bundle itself with no Spark write:

```python
# once: written by the executors, to storage the driver can also read
embeddings.write.parquet("gs://bucket/staging/embeddings")
corpus.write.parquet("gs://bucket/staging/corpus")

publish_serving_artifacts(
    "bundles/v1", …,
    embedding_store="gs://bucket/staging/embeddings",   # copied into the bundle, no Spark write
    delex_corpus="gs://bucket/staging/corpus",
)
```

**Publish in its own Spark application.** A batch featurize run registers its preprocessed corpus with the `SparkContext` (`addFile`), and Spark ships every registered file to all executors of every *later* job in that context. On a large corpus that is tens of gigabytes per executor for jobs that never read it, enough to fill the executors' disks and get the nodes marked unhealthy. Publishing itself registers nothing, but it cannot undo a registration a preceding job made, so run it as its own application rather than as a tail stage of a matching job.

##### Option 1: Sparkly (lexical)

The example above. Pass `sparkly_index_path=` pointing at a built `LuceneIndex` directory. This is the default choice and needs no Spark at serve time.

##### Option 2: Delex (lexical)

Delex has no cross-process resident form, so the bundle stores the raw corpus and the (unbuilt) `BlockingProgram`, and the loader rebuilds the resident probe once at load. Fusing Delex therefore makes the loader require a `SparkSession`, and the rebuild is a one-time Spark job at load, not per request.

```python
publish_serving_artifacts(
    "bundles/v1",
    indexed_table=corpus, features=features, model=model, fill_na=0.0,
    delex_program=program,          # a delex BlockingProgram (see the Delex tutorial)
    delex_corpus=corpus,            # a DataFrame or a parquet path/URI (defaults to indexed_table)
    delex_id_col="_id",
)
```

##### Option 3: Semantic, and fusion

Semantic blocking retrieves candidates by meaning. Publish it by pointing at a built **flat** `SemanticIndex` (a lossy-codec index is refused; serving decodes full vectors only) and giving serving the embedder and the fields it embedded, so it can turn each incoming record into a query vector identically to the corpus:

```python
from madmatcher_pro.semantic import SentenceTransformerProvider

publish_serving_artifacts(
    "bundles/v1",
    indexed_table=corpus, features=features, model=model, fill_na=0.0,
    sparkly_index_path="build/lucene",              # lexical half of the fusion
    semantic_index_path="build/semantic",           # a built flat SemanticIndex
    embed_provider=SentenceTransformerProvider(model_id="BAAI/bge-small-en-v1.5"),
    embed_fields=["name", "brand"],                 # the fields the corpus was embedded from
    nprobe=32,
    fusion_weights={"sparkly": 0.5, "semantic": 0.5},
)
```

With both a lexical index and a semantic index, serving **fuses** them per record with the same weighted reciprocal-rank fusion the batch pipeline uses (`fusion_weights` and `fusion_kappa`). Publish the semantic index alone (omit `sparkly_index_path`) for semantic-only blocking. The embedder is stored as config, not weights; the loader rebuilds one resident embedder that the semantic probe and the cosine store share, and it loads its model lazily on the first request.

The `embed_provider` can be any provider you used for the batch embeddings (a `SentenceTransformerProvider`, `FastEmbedProvider`, `OpenAIEmbeddingProvider`, or a model-id string). Pass `embed_provider=None` to require a precomputed `probe_embedding` on every `match` call instead of embedding in-process.

##### Option 4: Matching-only (bring your own candidates)

Omit every blocker to publish a bundle that only featurizes and predicts. You then supply the candidate ids on each request (`match(record, candidates=[...])`, Step 2). Use this when another system already produces candidates.

```python
publish_serving_artifacts(
    "bundles/v1",
    indexed_table=corpus, features=features, model=model, fill_na=0.0,
    # no sparkly_index_path / semantic_index_path / delex_program
)
```

##### The cosine feature

If you trained the model with the semantic cosine appended as its last feature (see "optional cosine feature for matching" in the Semantic tutorial), serving must reproduce that feature per candidate. Set `cosine_feature=True` and give it a **cosine source**:

- **`semantic_index_path`** lets serving **reuse** the semantic blocker's own cosine for each candidate it ranked.
- **`embedding_store`** (a corpus embedding table with an id column and an embedding column: a parquet path or URI, or a Spark DataFrame — see "publishing when the driver is not the only machine" above) lets serving **recompute** cosine, fetching a candidate's stored vector and dotting it with the incoming record's embedding. This covers candidates a lexical-only blocker surfaced that the semantic blocker never ranked.
- **Both** is the most faithful: serving reuses the semantic score where it has one and recomputes only the gaps.

```python
publish_serving_artifacts(
    "bundles/v1",
    indexed_table=corpus, features=features, model=model, fill_na=0.0,
    sparkly_index_path="build/lucene",
    semantic_index_path="build/semantic",
    embed_provider=SentenceTransformerProvider(model_id="BAAI/bge-small-en-v1.5"),
    embed_fields=["name", "brand"],
    cosine_feature=True,                    # the model was trained with the cosine feature
    embedding_store="build/embeddings",     # recompute cosine for candidates the scores miss
)
```

The embedding store is validated at publish time (each corpus id must map to exactly one vector), so a bad table fails now rather than at serve time. You can validate one up front with `check_embedding_store(table, id_col="_id", embedding_col="embedding")`.

Both cosine sources also need the **incoming record's own embedding**. Serving produces it in-process with the resident embedder you published (`embed_provider`, the same provider class you used in batch), so a plain `match(record)` needs nothing extra, and that one embedding serves both semantic blocking and the cosine feature. To skip in-process embedding, attach a precomputed vector with `match(record, probe_embedding=...)`; publishing with `embed_provider=None` makes that mandatory, so every call must carry its own `probe_embedding`.

##### sklearn vs SparkML

The model kind decides the request path, and publish records it in the manifest:

- An **sklearn/xgboost** model (`SKLearnModel`) predicts in-process with no Spark on the request path. This is the fast path and what real-time serving is built for.
- A **SparkML** model (`SparkMLModel`) has no in-process predict, so serving runs **one Spark job per `match` call**. It is supported for batch interoperability (a bundle that must match your SparkML training exactly), but it is slower and the loader needs a `SparkSession`.

Publish validates the model against the pinned feature set at load: the trained model must expect exactly as many features as the bundle produces (the feature list, plus one when `cosine_feature=True`), so a model trained on a different feature list is caught cleanly before it ever serves.

### Step 2: Loading the Matcher and Matching One Record

`RealtimeMatcher.load` is the only supported way to create a matcher. It checks the `realtime` entitlement, then loads the bundle into resident state:

```python
from madmatcher_pro.serving import RealtimeMatcher

matcher = RealtimeMatcher.load("bundles/v1")

result = matcher.match({"name": "Sony wireless over-ear noise-cancelling headset",
                        "brand": "Sony", "price": 349},
                       key="req-7")
print(result)
#    id1     id2  prediction  confidence
# 0    0   req-7         1.0        0.98
```

`load` takes:

- **`artifacts_dir`** is the bundle from Step 1 (a local path, or a Spark URI such as `s3://...` for the artifacts Spark reads).
- **`spark`** (default `None`) defaults to the active or a new session, and is needed **only** for a Delex or SparkML bundle. A resident sklearn bundle loads without it.
- **`track`** (default `False`) shows this matcher's throughput on the dashboard.
- **`id_col`** overrides which column supplies a record's own id; it wins over what the bundle recorded.

Loading is where the gate and the heavy lifting live: `load` checks the license and the `realtime` entitlement (a `LicenseError` here means the base license is valid but the add-on is not enabled), then opens the index, maps the corpus feature state into memory, and loads the model, so `match` itself does none of that. One exception is deferred: a bundle with a resident embedder loads the embedding model's weights on the first request, so the first `match` is slower than the rest. If first-request latency matters, serve one throwaway record right after `load`.

`match` takes the incoming record (a `dict`, a `pandas.Series`, a one-row `pandas.DataFrame`, or a pyspark `Row`) and returns the `[id1, id2, prediction, confidence]` frame. Its keyword arguments:

- **`key`** is the value stamped as `id2` on every result row, and it can be an integer or a string. Resolution is first-hit-wins: an explicit `key`, else the record's own id under `id_col`, else `None`. Use an explicit key for a record with no id of its own, for example a string request id for a record pulled off a queue, and it is echoed back on every match so the result stays attributable once it leaves the call.
- **`candidates`** is the bring-your-own list for a matching-only bundle (Option 4): when given, blocking is skipped and the record is matched only against those corpus ids. An id absent from the corpus is dropped; an empty or all-unknown list gives the empty result.
- **`probe_embedding`** is a precomputed embedding for the semantic source or cosine store, skipping the resident embedder for this call. It is required when the bundle was published with `embed_provider=None`.

**What fields should the incoming record carry?** The columns your features compare, which are the corpus columns you trained over (the record in the example carries `name`, `brand`, and `price` because the corpus has them). It does not need all of them: a missing field is treated as a null value, exactly as a stored null in the corpus is, so the features that compare it fall back to `fill_na` instead of raising. Fields the features never read are ignored, and each value is cast to the corpus column's type before comparing, so an incoming `1999` where the corpus stores text (or `"1999"` where it stores a number) still compares consistently.

**Reading the result.** The trained model decides what is a match: `prediction` is its call and `confidence` its probability for that call, computed by the same model call the batch `apply_matcher` uses. The frame holds the candidates the model predicts as matches, sorted by confidence descending (ties broken by `id1`, so results are identical run to run). A result carries the matched record's **id**, not its fields: `id1` is a key into your own corpus, so join it back to your corpus table when a consumer needs the record itself. On the streaming paths (Step 5), the **incoming** record's fields can be stored beside each match with `passthrough`.

**Release the matcher when done.** A matcher holds Lucene readers and memory-maps open. Call `matcher.close()` when you are finished, or before loading a new one:

```python
matcher.close()
```

### Step 3: Matching a Batch

When you have several records in hand, `match_batch` is faster than calling `match` on each: a batch shares one embedder forward pass and one model predict (one Spark job, for a SparkML model). The results are exactly what per-record `match` would give, one frame per input record, in order:

```python
results = matcher.match_batch([
    {"name": "Sony wireless noise-cancelling headset", "brand": "Sony"},
    {"name": "wireless ergonomic mouse, USB-C", "brand": "Logitech"},
], keys=["req-1", "req-2"])

for res in results:            # each is a [id1, id2, prediction, confidence] frame
    print(res)
```

`match_batch` takes:

- **`records`** is a list of record mappings (each any shape `match` accepts) or a `pandas.DataFrame` whose rows are the records. An empty input returns an empty list.
- **`keys`** (optional) is a sequence aligned one-to-one with `records`, each entry as `match`'s `key`. Where an entry is `None` (or the whole argument is omitted), that record falls back to its own id under `id_col`.
- **`candidates`** (optional) is an aligned sequence: each entry a bring-your-own candidate list for that record, or `None` to block that record as usual.
- **`probe_embeddings`** (optional) is an aligned sequence of precomputed embeddings, each entry as `match`'s `probe_embedding`.

It returns a list of `[id1, id2, prediction, confidence]` frames, one per input record, in input order; each frame is exactly what `match` would have returned for that record on its own.

### Step 4: Serving Concurrent Requests

Under concurrent load, wrap the matcher in a pool. `match` is thread-safe, and the pools add parallelism and load-shedding on top. Each pool lets you `submit(record)` from any number of request threads and get a `concurrent.futures.Future`, or call the blocking `match(record)`. `MatcherPool` and `ProcessMatcherPool` also stream an iterable through `map` (results in submission order) or `imap_unordered` (results as they finish); the two thread pools (`MatcherPool` and `MicroBatchMatcher`) can `scale(n)` the worker count up or down at any time. They are in-process helpers: no service and no network hop, and the caller owns transport (an HTTP handler, a Kafka consumer).

#### Option 1: MatcherPool (threads)

A pool of worker threads, each serving one record at a time through the loaded matcher. Simplest to reason about, and right for a handful of concurrent requests or when spawning processes is undesirable. Because the featurize string comparisons run in Python and hold the GIL, a thread pool tops out near 2x on CPU-bound matching:

```python
from madmatcher_pro.serving import MatcherPool

with MatcherPool(matcher, n_workers=8) as pool:
    for matches in pool.imap_unordered(incoming_records):   # served in parallel, streamed
        handle(matches)
```

`MatcherPool` takes the loaded `matcher` and **`n_workers`** (default: the machine's CPU count, capped at 16), plus the queue policy covered under Backpressure below (`max_queue`, `on_full`). **`track`** (default `True`) shows a live request counter and worker count on the dashboard whenever a session is active.

#### Option 2: MicroBatchMatcher (coalesce into batches)

Coalesces concurrent single-record requests into `match_batch` windows, for throughput on a request stream without changing the per-request contract. Each worker groups requests arriving close together (up to `max_batch`, or until `max_delay_ms`) and serves the window with one `match_batch`, so the window shares one embedder pass and one predict. Reach for it when embedding or prediction dominates and requests arrive faster than one at a time:

```python
from madmatcher_pro.serving import MicroBatchMatcher

batcher = MicroBatchMatcher(matcher, n_workers=2, max_batch=32, max_delay_ms=10.0)
future = batcher.submit({"name": "apple watch series 9", "brand": "Apple"})
matches = future.result()
batcher.close()
```

**`max_batch`** (default 32) caps the window size and **`max_delay_ms`** (default 10) caps how long a window waits to fill, so a lone request is never stuck: it is served after at most `max_delay_ms`. **`n_workers`** (default 1) sets how many windows are served in parallel. Requests stay isolated even when coalesced: one failing request fails only its own `Future`, never the other requests in its window.

#### Option 3: ProcessMatcherPool (near-linear across cores)

For CPU-bound matching where you want near-linear throughput across cores, worker **processes** bypass the GIL. Because each worker loads its own matcher, this pool takes a **bundle path**, not a loaded matcher, and each process pays its own load cost (its own PyLucene JVM, model, and embedder). Memory-mapped bundle artifacts (the corpus rows, vectorizer arrays, Lucene segments, cosine embeddings) are shared across the processes, but the per-process state is not, so size the pool against measured per-process memory times `n_procs`, not the bundle size:

```python
from madmatcher_pro.serving import ProcessMatcherPool

with ProcessMatcherPool("bundles/v1", n_procs=8) as pool:
    for matches in pool.imap_unordered(incoming_records):   # served across processes
        handle(matches)
```

`ProcessMatcherPool` takes the bundle path, **`n_procs`** (default: the CPU count), and **`load_kwargs`**, a dict forwarded to each worker's `RealtimeMatcher.load` (for example `id_col`). Do not put `spark` in it: a session cannot be sent to another process, and a resident sklearn bundle loads without one (a Delex or SparkML bundle needs Spark in every worker, so prefer `MatcherPool` for those). **`semantic_cache_mb`** is a pool-wide semantic cache budget split evenly across the workers, and **`max_in_flight`** (default 4 per process) bounds outstanding requests the way the thread pools' queue does.

A bundle published with the shared semantic cell store (every bundle published from this version on) does not use that cache at all: all workers memory-map one read-only copy of the semantic index's cells, so the semantic memory cost of the pool is one copy per machine rather than one per worker, and you can size the pool by CPU instead of by RAM. For a bundle published before the cell store existed (or one whose semantic index you rebuilt in place, which invalidates the store and drops that bundle back to the cached path with a warning at load), you have three options: re-publish the bundle, run `madmatcher_pro.semantic.cell_store.stage_cells_durably(<bundle>/blocking/semantic)` once against the bundle's semantic dir to (re)build the shared store in place (much cheaper than a re-publish; the next load picks it up), or keep budgeting `semantic_cache_mb`.

A bundle carrying an embedder or a Delex program is more expensive per process (each worker loads its own copy of the model weights, or rebuilds every Delex plan node), and the pool warns when it detects that so you can choose a smaller pool or fall back to `MatcherPool`.

**How results are keyed in a pool.** The thread pools (`MatcherPool`, `MicroBatchMatcher`) serve each record without a per-request key, so a result's `id2` comes from the record's own id (the `id_col` from `load`), or is `None` for a record carrying none. Feed them records that carry their ids. `ProcessMatcherPool.submit` additionally accepts an explicit `key=`, and the streaming runners in Steps 5 and 6 assign keys for you.

**Backpressure.** Every pool bounds its queue by default, so a producer faster than the pool gets backpressure rather than unbounded memory growth. `on_full="block"` (the default) waits for space; `on_full="reject"` raises `QueueFullError` so a caller can shed load.

**Shutting down.** `close()` stops accepting new requests (further `submit` raises), serves what is already queued, and joins the workers; a request still unserved when the workers exit resolves its `Future` with an error rather than hanging its caller. Using a pool as a `with` block closes it for you.

### Step 5: Streaming Records Through a Matcher

When you have a file or a stream to run start to finish, `run_matching` owns the loop and the buffering: it pulls from a `RecordSource`, matches through the matcher, and writes each result to a `ResultSink`, holding only a bounded window in memory so an unbounded source works.

```python
from madmatcher_pro.serving import run_matching, ParquetSource, ParquetSink

with ParquetSource("incoming/", id_col="_id") as src, \
     ParquetSink("out/matches") as sink:
    stats = run_matching(src, sink, matcher)
print(stats.records, stats.matches)
```

**Sources** yield `(key, record)` pairs and are iterated lazily (never materialized whole): `IterableSource` (any iterable or generator), `ParquetSource`, `CsvSource`, and `QueueSource` (drain a `queue.Queue` a producer thread fills). **Sinks** take each result frame: `MemorySink` and `DataFrameSink` (accumulate in memory), `CallbackSink` (hand each result to your function), `ParquetSink` (append parquet parts through local/HDFS/S3/GCS storage), and `FanOutResultSink` (several at once). See the Appendix for the full tables.

`run_matching` takes `source`, `sink`, and `matcher` positionally; `matcher` can be a `RealtimeMatcher` or any pool from Step 4, so streaming through parallel workers is just passing the pool. Its keyword arguments:

- **`batch_size`** (default `None`) matches in groups of this size through `match_batch`, so each group shares one embedder pass and one predict. `None` matches one record at a time.
- **`passthrough`** (default `False`) also hands the sink each incoming record, so it can store the original fields beside the match (`ParquetSink` selects which with `passthrough_cols`).
- **`progress_every`** (default `0`, silent) logs a progress line every N records.

It returns a `MatchRunStats` whose fields are **`records`** (records matched), **`matches`** (match rows written), and **`elapsed`** (wall-clock seconds). A record whose key resolves to `None` is assigned its position in the stream as its key, so every written row stays attributable to an input. The sink is flushed when the run ends, even when it ends on an error.

**Keeping the incoming record's fields with its matches.** A match row alone carries only ids and scores. To write the incoming record's own fields beside each of its matches, so a consumer reads results without a second lookup, turn on passthrough and name the fields on the sink:

```python
sink = ParquetSink("out/matches", passthrough_cols=["name", "brand"])
stats = run_matching(ParquetSource("incoming/", id_col="_id"), sink, matcher,
                     passthrough=True)
```

Each written row then carries `[id1, id2, prediction, confidence]` plus that record's `name` and `brand`. The same flag serves the other sinks: `MemorySink` keeps the records beside the result frames, and `CallbackSink` receives each `(result, record)` pair.

### Step 6: Durable and Multi-Worker Streaming

`run_matching` is not durable: a crash starts the run over from the beginning of the source. `run_matching_durable` adds a per-record **claim protocol** on shared storage, which buys three things from one mechanism. "The input" here is one body of incoming records, the same `source` (a parquet directory, a queue) each worker reads:

- **Crash resumption.** Re-running the same command after a crash skips what already completed.
- **Multiple workers.** Several processes given the same `source` and the same `checkpoint_dir` split that work: each reads the records and claims what it can, and together they cover the input once, with no broker or coordinator. Workers can join or leave mid-run.
- **Unconnected machines.** The same command on several machines that can all see `checkpoint_dir` cooperates the same way.

```python
from madmatcher_pro.serving import run_matching_durable, ParquetSource, ParquetSink

stats = run_matching_durable(
    ParquetSource("incoming/", id_col="_id"),
    ParquetSink("out/matches"),
    matcher,
    checkpoint_dir="ckpt/run-7",
)
```

Delivery is **at-least-once**: a worker that dies after writing results but before recording completion leaves the record claimable, and a later worker redoes it. Pass `exactly_once=True` to skip a record already marked done; to close the last window (a crash between the sink write and the completion marker) use an idempotent sink that de-duplicates on `(id1, id2)`.

**The key is a record's durable identity.** Claims and completion markers are named by each record's key (integer or string, either works), so "this record is already done" means "a record with this key is done". Keys must therefore be stable across re-runs and unique within the run: a source whose records carry ids (`id_col`) or explicit keys resumes exactly, while unkeyed records fall back to their position in the source, which resumes correctly only if the source replays in the same order.

**Every worker reads the whole source.** Claims decide who processes a record, not who reads it: each worker iterates the source in order and skips past records it does not claim. That is cheap for a parquet or CSV source; for a source that is expensive to read, give each machine its own slice of the input instead of the identical source.

**Why it is opt-in.** Durability is not free, so `run_matching` is the default. At record granularity every record costs a claim marker, a completion marker, and a sink flush (several small writes to `checkpoint_dir`), delivery becomes at-least-once (so you often want an idempotent sink), and you must provision shared checkpoint storage every worker can reach. For a bounded run on one machine that you can simply re-run on failure, plain `run_matching` is simpler and faster. Reach for the durable form when repeating a crashed run is expensive, or when you want several workers on one input.

**Granularity.** The default `granularity="record"` claims one record at a time, so a record is processed the moment it arrives, which is what a live stream needs. `granularity="shard"` batches claims into fixed-size groups for cheaper markers, but a shard is processed only once it fills, so use it only for **bulk backfill** of records that are all already present. One consequence of per-record durability: the sink is flushed after every record (results must be durable before the completion marker), so a `ParquetSink` in record mode writes one part file per matched record. For compact bulk output prefer shard mode or a coalescing sink. The Reference section documents `lease_seconds`, `max_units`, `worker_id`, and the rest.

### Step 7: Handing Off From a Batch Build

A batch build takes a while, and records arriving during it have nowhere to go: the bundle they would match against does not exist yet. `BatchToServing` closes that gap without dropping records, by moving through four phases: it **accepts** arriving records into a durable spool while the build runs, **publishes** the bundle when the build finishes, **drains** the spooled backlog against the new bundle, then **serves** live records from there on.

```python
from madmatcher_pro.serving import BatchToServing, RealtimeMatcher, ParquetSink, publish_serving_artifacts

h = BatchToServing(spool_dir="spool/", bundle_dir="bundles/v1",
                   id_col="_id", checkpoint_dir="ckpt/drain")

# While the batch build runs, records keep arriving. Spool them durably.
h.accept({"_id": 1001, "name": "apple pie", "brand": "acme"})

# The build finishes: publish from its artifacts, load, and go live.
h.publish(lambda out: publish_serving_artifacts(
    out, indexed_table=corpus, features=features, model=model,
    sparkly_index_path="build/lucene", fill_na=0.0))
h.start_serving(RealtimeMatcher.load("bundles/v1"))

# Match the backlog, then serve live.
h.drain(ParquetSink("out/backlog"))
h.match({"_id": 1002, "name": "banana bread", "brand": "acme"})   # live from here on
```

`BatchToServing` takes:

- **`spool_dir`** is where accepted records are buffered durably, as append-only segment files. Construction also recovers any segment a crashed writer left open, so records accepted before a restart are not stranded.
- **`bundle_dir`** is where `publish` writes the bundle.
- **`id_col`** (optional) supplies each accepted record's key, exactly as in Step 2.
- **`checkpoint_dir`** (optional) makes the drain durable with Step 6's per-record claims, so a crash mid-drain resumes instead of restarting the backlog. Without it the drain is a plain `run_matching`.
- **`segment_rows`** (default 1000) is how many records go into one spool segment before it is sealed. Smaller segments become drainable sooner; larger ones make fewer files.

Its methods, in the order the phases use them:

- **`accept(record, key=None)`** buffers one record durably and returns its key. It is safe to call from many threads, and it keeps working after serving starts (it just spools), so a producer never needs to track which phase the handoff is in.
- **`publish(publish_fn)`** seals what has been spooled so far, then calls `publish_fn(bundle_dir)`; you supply the `publish_serving_artifacts` call from Step 1.2, as in the example. Records still arriving keep spooling meanwhile.
- **`start_serving(matcher)`** goes live with a loaded matcher (or pool). `match` raises until this is called; before that, `accept` is the way in.
- **`drain(sink, batch_size=None, passthrough=False, clear=True)`** matches the backlog into a sink and returns the same `MatchRunStats` as Step 5. It snapshots the sealed segments at its start and matches exactly that snapshot, so a record arriving mid-drain is neither double-matched nor lost (it waits for the next drain, or for live serving). `clear=True` (the default) deletes the drained segments, and only after their results are in the sink; pass `clear=False` to keep them, for example to re-drain into a second sink. `batch_size` groups a plain (non-durable) drain through `match_batch`; on a durable drain, pass runner options such as `lease_seconds` as extra keyword arguments instead.
- **`match(record, key=None)`** serves one record live, as in Step 2.
- **`close()`** stops accepting and seals the spool. It does not close the matcher; you own that lifecycle, which usually outlives the handoff.

The underlying `RecordSpool` is itself a `RecordSource`, so you can also point `run_matching` at it directly.

### Step 8: Refreshing the Bundle Without Downtime

To fold new corpus records in, rebuild a bundle offline (this is not a per-record live upsert) and swap it into the running process. `HotSwappableMatcher` serves from a loaded bundle and can atomically switch to a newer one: the new bundle loads and warms while the old one keeps serving, then requests switch over in one step, and the old bundle is retired once its in-flight requests finish (each request finishes on the bundle it started on).

```python
from madmatcher_pro.serving import HotSwappableMatcher

live = HotSwappableMatcher("bundles/v1")
live.match({"name": "apple pie", "brand": "acme"})

# Later, after publishing bundles/v2 offline:
live.swap("bundles/v2")     # v2 now serves; v1 retired once its requests drain
```

`swap` re-checks the license and `realtime` entitlement, so a lapsed license stops future serving, and a failed load (an invalid or half-published bundle) leaves the current bundle serving unchanged.

`HotSwappableMatcher` takes the first bundle to load, `spark` (for a Delex or SparkML bundle), and any `RealtimeMatcher.load` keyword arguments, which it forwards to every load it performs (for example `id_col`). **`drain_timeout`** (default 30 seconds) is how long a swap waits for the old bundle's in-flight requests before giving up on closing it: a request that outlives the timeout leaves its old bundle open rather than being torn down mid-request, and that bundle is retired later, once the request finishes (the `undrained` property counts bundles in that state; normally 0). `version` names the bundle currently serving, and **`track`** (default `True`) shows each swap's load, drain, and go-live on the dashboard.

**Many replicas.** To tell N serving replicas that a new bundle exists, write a pointer file to the storage every replica already reads bundles from, so no message bus is needed. A publisher writes the pointer after publishing, and each replica runs a `BundleWatcher` that polls it and hot-swaps when the version changes:

```python
from madmatcher_pro.serving import publish_bundle_version, BundleWatcher

# On the publisher, after publish_serving_artifacts("s3://.../bundles/v2", ...):
publish_bundle_version("s3://.../bundles", "s3://.../bundles/v2", version="v2")

# On each replica:
live = HotSwappableMatcher("s3://.../bundles/v1")
watcher = BundleWatcher(live, "s3://.../bundles", poll_seconds=30.0).start()
# ... serve; new versions are picked up automatically ...
watcher.stop()
```

Write the pointer only after `publish_serving_artifacts` has returned: the pointer tells replicas the bundle is safe to load, so writing it early would send them at a half-written bundle. The pointer itself is one small JSON file (`CURRENT.json`) written atomically, so a replica polling mid-write sees the old version or the new one, never a half-write; `read_bundle_version(pointer_dir)` reads it back.

`BundleWatcher` takes the live matcher and the pointer directory, plus **`poll_seconds`** (default 30; the check is one small file read, so anywhere from seconds to minutes is normal) and an optional **`on_swap`** callback called after each successful swap. `start()` polls from a background thread and `stop()` ends the polling without closing the matcher; to drive the check from your own loop instead, call `check_once()`, which polls once and returns whether it swapped. A replica that was down picks up the current version when it starts, a failed swap is logged and the replica keeps serving the version it has (so a broken bundle never takes a replica down), and you roll back by writing the pointer again.

## Closing Notes

You built the corpus state once with a batch job, published it as a bundle, and matched incoming records against it in memory. From there you scaled up along independent axes: batches, thread and process pools, orchestrated streaming, durable multi-worker runs, a lossless handoff from the batch build, and live bundle refresh across replicas. Every path reuses the exact batch kernels, so a served match equals the batch match for the same record and corpus. When you need the durability and throughput-tracking details, the **Reference** section documents them in full, and the **Appendix** collects the sources, the sinks, and a table for choosing a serving surface.

---

## Reference: Durability and Progress Tracking

This is the complete reference for the durability and progress features used in the steps above.

### How durable streaming works

`run_matching_durable` coordinates workers through files under `checkpoint_dir`, with no broker:

1. **Claim.** Before processing a unit (a record, or a shard in shard mode), a worker writes a claim file naming that unit. A unit already claimed or completed is skipped, so two workers never process the same unit at once.
2. **Process and write.** The worker matches the unit and writes its results to the sink, then flushes, so results are durable before the next step.
3. **Complete.** The worker writes a completion marker for the unit. The marker, not the claim, is what records the unit as done.
4. **Resume.** Re-running reads the markers, skips completed units, and processes the rest. Several workers sharing the directory each take what they can, so together they finish the input once.

The manifest under `checkpoint_dir` pins the run's configuration (`granularity`, `shard_size`) and refuses a mismatch: claims are named per unit, so mixing settings would double-process some records and skip others. Use a fresh `checkpoint_dir` for a different input.

### The delivery guarantee

Delivery is **at-least-once** by default. A worker that dies after writing results but before its completion marker leaves the record claimable, so a later worker (or a re-run) redoes it. Two options tighten this:

- **`exactly_once=True`** skips any record already marked done, suppressing that redelivery.
- **An idempotent sink** that de-duplicates on `(id1, id2)` closes the remaining window, the gap between the sink write and the completion marker, where a crash could produce a duplicate.

A stalled or dead worker's claim is reclaimed once it is deemed abandoned: a same-host worker whose process is gone, or any claim older than a lease. `lease_seconds` sets that lease (raise it for a `granularity="shard"` run whose shards legitimately take longer than the default, so a slow shard is not reclaimed from its worker). A reclaim is only ever a redo, which at-least-once already permits.

### Resumable publish

Publishing writes the bundle in independent artifact stages (the corpus feature state, the index copies, the embedder, the model), each committed after its data is durable. `enable_crash_recovery=True` with a `checkpoint_dir` makes a crash mid-publish resume from the last committed stage instead of rebuilding the whole bundle:

```python
publish_serving_artifacts(
    "bundles/v1", indexed_table=corpus, features=features, model=model,
    sparkly_index_path="build/lucene", fill_na=0.0,
    enable_crash_recovery=True, checkpoint_dir="ckpt/publish-v1",
)
```

Keep the checkpoint outside the bundle and use a fresh directory per publish: resuming an existing checkpoint continues that exact publish, so a re-publish (a refreshed corpus or a changed config) needs its own directory. With `validate_input=True` (the default) the corpus row count is pinned and re-checked, so a resume after the corpus grew or shrank is refused rather than serving stale artifacts. Reaching a non-local `checkpoint_dir` needs a live `SparkSession`.

### What happens when a job crashes

Nothing is left corrupted. A committed unit (a completed record, or a published bundle) is safe, because its marker is written only after its output is durable. A unit interrupted mid-write has no marker, so it is treated as not done. To recover, **run the same command again** with the same `checkpoint_dir`: there is no separate resume command. A completed record is never reprocessed (except as an at-least-once redo), an interrupted one is redone, and a finished run does no work.

### Viewing throughput on the dashboard

Every serving surface can report to the same web dashboard the batch engines use. Pass `track=True` when loading a matcher or constructing a pool, or create a `MadMatcherSession` near the top of your script so everything that follows is tracked:

```python
from madmatcher_pro import MadMatcherSession

with MadMatcherSession.builder.open_browser(True).getOrCreate() as session:
    print(f"Dashboard: {session.url}")
    matcher = RealtimeMatcher.load("bundles/v1", track=True)
    # each match / match_batch advances a live "Serving requests" counter
```

Each matcher and pool shows a live throughput counter with per-surface KPIs (a pool's worker count, a micro-batcher's window sizes). A `HotSwappableMatcher` shows a swap-lifecycle job counting versions swapped in, and a publish shows up as a job while it runs. Tracking is best-effort: if the dashboard cannot start (a busy port, a headless host) the work still runs. The dashboard has no built-in authentication, so on a shared host reach it by SSH port-forward or secure it at the network layer, exactly as the engine tutorials describe. The read-only viewer (`python -m madmatcher_pro.reliability.dashboard <checkpoint_dir>`) works against a durable run's checkpoint directory from any machine that can read it.

### Durability arguments

**`run_matching_durable(..., checkpoint_dir=...)`** is required and names where claims and completion markers live. It must be reachable by every worker sharing the run, and not reused across different inputs.

**`granularity`** (`"record"` default, or `"shard"`) claims per record (safe for live streams) or per fixed-size group (cheaper markers, bulk backfill only).
_It does not:_ change the result, only how work is claimed and flushed.

**`shard_size`** (default `1000`) is records per shard when `granularity="shard"`.

**`exactly_once`** (default `False`) skips records already marked done instead of re-processing them.
_It does not:_ close the crash-between-write-and-marker window on its own; pair it with an idempotent sink for that.

**`lease_seconds`** (default `None`, the store default) is how long before an incomplete claim is assumed abandoned and reclaimable.
_It does not:_ affect a single-worker crash-resume, where the same host reclaims its own dead process's work regardless.

**`batch_size`** (default `None`, shard mode only) matches within a shard in groups through `match_batch`.

**`max_units`** (default `None`) stops this worker after completing that many units.

**`worker_id`** (default host+pid) identifies this worker in claim files.

**`spark`** (default `None`) is needed only for a non-local `checkpoint_dir`.

**`progress_every`** (default `0`, silent) logs a line every N records.

---

## Appendix

### Record sources

A `RecordSource` yields `(key, record)` pairs lazily, so an unbounded stream works.

| Source                                                               | What it reads                                                                                                                                |
| -------------------------------------------------------------------- | -------------------------------------------------------------------------------------------------------------------------------------------- |
| `IterableSource(records, id_col=None)`                               | any Python iterable or generator of records (or of `(key, record)` pairs). The in-memory case and the adapter for a reader you already have. |
| `ParquetSource(path, id_col=None, columns=None, batch_rows=10_000)`  | a parquet file or directory, streamed one batch at a time through pyarrow.                                                                   |
| `CsvSource(path, id_col=None, chunk_rows=10_000, **read_csv_kwargs)` | a CSV, in chunks. Extra kwargs pass to `pandas.read_csv`.                                                                                    |
| `QueueSource(queue_=None, maxsize=1000, id_col=None)`                | a `queue.Queue` a producer thread fills. Iteration ends when the producer calls `close()`.                                                   |

`id_col` names the column each record carries its own id in, used as its key. `batched(source, size)` groups a source into lists of at most `size` for `match_batch` or shard runs.

### Result sinks

A `ResultSink` takes each result frame. A sink is called from several worker threads at once, so a custom `write` should lock around shared state.

| Sink                                                                         | Where results go                                                                                                  |
| ---------------------------------------------------------------------------- | ----------------------------------------------------------------------------------------------------------------- |
| `MemorySink()`                                                               | accumulated in memory (`.frames`, `.to_frame()`). Grows with the stream; for bounded runs and tests.              |
| `DataFrameSink()`                                                            | accumulated into one `pandas.DataFrame` (`.result`), to hand straight to the next pipeline step.                  |
| `CallbackSink(fn)`                                                           | each result handed to your `fn(result, record)` (or `fn(result)`). Holds nothing, so an unbounded stream is fine. |
| `ParquetSink(out_dir, batch_rows=10_000, passthrough_cols=None, spark=None)` | appended as parquet parts through local/HDFS/S3/GCS storage, each written atomically.                             |
| `FanOutResultSink([...])`                                                    | every result sent to several sinks at once (for example keep it in a DataFrame and also write it durably).        |

Pass `passthrough=True` on the runner (or `passthrough_cols=` on `ParquetSink`) to store the incoming record's fields beside each match.

### Choosing a serving surface

| You have                                     | Use                                       | Why                                          |
| -------------------------------------------- | ----------------------------------------- | -------------------------------------------- |
| one record, want the answer now              | `RealtimeMatcher.match`                   | the base call                                |
| several records in hand                      | `RealtimeMatcher.match_batch`             | one embed pass and one predict for the batch |
| concurrent requests, a few cores             | `MatcherPool`                             | worker threads over one shared matcher       |
| concurrent requests, coalesce for throughput | `MicroBatchMatcher`                       | groups requests into `match_batch` windows   |
| CPU-bound, want near-linear scaling          | `ProcessMatcherPool`                      | worker processes bypass the GIL              |
| a file or stream to run start to finish      | `run_matching`                            | owns the loop and bounded buffering          |
| the above, crash-safe or multi-worker        | `run_matching_durable`                    | per-record claims and resume                 |
| records arriving during the batch build      | `BatchToServing`                          | spool, then drain, then serve                |
| refresh the corpus without downtime          | `HotSwappableMatcher` (+ `BundleWatcher`) | atomic bundle swap, per replica              |

### What a bundle contains

`publish_serving_artifacts` writes a self-contained directory:

- `MANIFEST.json`: the pinned configuration (feature fingerprint, model kind, `fill_na`, `block_limit`, `id_col`, the blocking query spec, and each enabled blocker's settings).
- `_PUBLISHED.json`: the terminal marker, written last. Its absence means an incomplete bundle, which the loader refuses.
- `feature_state/`: the preprocessed corpus feature state.
- `model/`: the trained model in its native (no-pickle) format.
- `blocking/`: the published blockers, `sparkly/` (the Lucene index), `semantic/` (the flat index), `delex/` (the corpus and unbuilt program), `embedding_store/` (the corpus vectors as a shared memmap), and `embed_provider.pkl` (the embedder config), whichever were published.

---

For the full API, see the reference at `https://docs.madmatcher.ai/madmatcher-pro/{version}/madmatcher_pro.serving.html` (`{version}` is your installed release). For help, email support@madmatcher.ai or visit [madmatcher.ai](https://madmatcher.ai).