Integration sync: duplex fetch/write — async bulk writes behind a committed-watermark gate

## Problem

The per-lane sync cycle is serial: `fetch page → write page → fetch next`. The savings available from overlapping two steps equal the **smaller** step, which makes a serial design age badly as fetch improves:

| era | fetch | write | serial cycle | duplex cycle | overlap wins |
|---|---|---|---|---|---|
| today (tuned cursor) | ~6s | ~1s | 7s | 6s | ~15% |
| partitioned fetch (4x) | ~1.5s | ~1s | 2.5s | 1.5s | ~40% |
| vendor bulk fetch | ~0.5s | ~1s | 1.5s | 1.0s | serial is now **write-bound** |

Every future fetch improvement (sub-query partitioning — #4007 — or vendor bulk APIs) makes the fixed write leg a larger fraction of a serial cycle, until it becomes the new wall. Measured live (2026-08): a fully tuned single lane sustains ~20k rows/min and spends ~85% of its cycle waiting on fetch — the write leg is small *today*, which is exactly why the serial shape's cost is invisible until fetch speeds up.

## Proposal: duplex (write-behind) lanes

Each lane runs two concurrent streams:

1. **Fetch stream**: fetches continuously; never awaits a write.
2. **Write stream**: each batch's bulk write is fired async as its batch arrives.

Rules that keep it exactly as safe as the serial design:

- **The watermark checkpoint is the only synchronization point.** It advances only to the *committed low-water mark* — the highest point below which every write has committed. A crash refetches the small in-flight window; it can never strand rows behind a watermark that overstates coverage.
- **Bounded in-flight count** (e.g. 3–4 batches) caps memory; a fast fetch back-pressures on the cap, not on individual writes.
- **Failures journal and route records to `BaseEntity.Save()`** — the canonical write path (not a "fallback": `Save()` is the definition of a correct write; the bulk path is an optimization that must prove equivalence, fenced to provably-new records only).

## Write topology: per-map direct bulk through the provider pool

A controlled A/B (same code, same database, same rows; 16 concurrent lanes) compared per-deposit **direct bulk transactions** against funneling all lanes through a serializing accumulator:

| regime | direct (pool-parallel) | accumulator funnel | funnel ratio |
|---|---|---|---|
| trickle (16 lanes × 20-row deposits) | 917k rows/min | 126k rows/min | 0.14x |
| backlog (16 lanes × 500-row deposits) | 2.44M rows/min | 1.55M rows/min | 0.63x |

Bulk transport amortizes per-call cost by itself and the connection pool supplies the parallelism — consolidation-era reasoning (correct when writes were statement-compile-bound; measured 324x there) no longer applies, and a serializing funnel only adds gather latency and removes parallelism. The design therefore uses **per-map direct bulk transactions** (entity rows + record maps committed atomically per call, column set and integration stamps identical to entity saves, types/nullability from entity metadata). A coordinator remains as **state manager only** — in-flight registry, committed low-water mark, memory budget, failure routing — never in the data path.

## Prerequisite / sibling: native `BulkCreate` provider capability

The clean home for the transport is the metadata layer, not the integration engine: a provider capability `BulkCreate(entities[])` whose default implementation loops `Save()` (every provider works unchanged) and whose SQL Server implementation streams typed TDS bulk with record maps in-transaction. The engine then calls the metadata layer like everything else in MJ. Working prototype evidence: a local end-to-end against real SQL Server (206-column tables) sustained **~727k rows/min** with exact typed roundtrip, transactional record maps, and verified rollback → canonical-path routing.

## Evidence (live, 2026-08)

- Production ladder on a multi-million-row object: ~250 → ~1,000 → ~14,000 → ~20,000 rows/min sustained as write-path and watermark defects were fixed; final state is fetch-bound with writes ~2% of capacity.
- Fetch/process overlap already validated in isolation (prefetch pipelining: next page in flight while the current one processes).
- The A/B above; and durable per-batch watermark checkpoints proven in production (runs killed and resumed repeatedly with zero lost progress).

## Per-provider transport (this must NOT live in the dialect layer)

The dialect (`@memberjunction/sql-dialect`) renders SQL *text* — quoting identifiers and literals. Bulk loading bypasses SQL text entirely: it is a **wire protocol**, and each engine has a different one. So `BulkCreate` belongs to the **provider capability**, with the dialect used only to quote the target table/column names:

| provider | transport | driver surface |
|---|---|---|
| SQL Server | TDS bulk copy (`INSERT BULK`) — typed `Table` streamed on a `Request` | `mssql` pool via `DatabaseConnection` |
| PostgreSQL | `COPY () FROM STDIN` — streamed rows | `pg` pool via `DatabaseConnection` + `pg-copy-streams` |
| any other | default implementation loops `Save()` — correct, unoptimized | — |

### PostgreSQL specifics worth designing for up front

- **COPY needs a dedicated client, not `pool.query()`** — check one out (`pool.connect()`), run `BEGIN`, stream the entity-row COPY and the record-map COPY on that same client, then `COMMIT`. This preserves the atomic rows+maps guarantee exactly as the SQL Server path does.
- **Format choice**: text/CSV format is simple (escape `\`, tab, newline; `\N` for NULL) and typically an order of magnitude faster than per-row INSERTs; binary format is faster still but requires per-type encoders and type OIDs. Text first, binary as an optimization.
- **Failure semantics are already compatible**: COPY is all-or-nothing with no per-row error reporting — identical to a bulk transaction, so the contract ("failed batch rolls back and its records route to the canonical `Save()` path") needs no change.
- **No `ON CONFLICT` under COPY** — not a limitation here, because the fast path is fenced to provably-new records in both providers.
- **Identifier quoting must go through the dialect**: PostgreSQL folds unquoted identifiers to lowercase, and MJ deployments use mixed-case entity/column names.

### Why this shape

A provider-level capability with a `Save()`-looping default means: every existing provider keeps working untouched, two providers get native speed, and the integration engine (and any other caller) writes to the metadata layer without knowing which transport is underneath.

Start by tracing the integration engine, metadata-layer provider boundary, BaseEntity.Save(), DatabaseConnection, and dialect responsibilities described here. Define the BulkCreate contract and coordinator behavior, then validate that default Save-looping providers, native SQL Server and PostgreSQL transactions, committed low-water marks, bounded in-flight batches, and failure routing preserve the stated guarantees.

Written by the indexing model from the issue text.