# Durable Source Projection WorkItem

## Scope contract

This change fixes one amplification boundary in L0 connectors: a state-projection
route must not materialize one executable WorkItem per immutable inbound Event.

The existing invocation mode
`latest_by_source_full_record_preserve_payload_settled` opts a route into this
contract. Other route modes remain event-per-work-item commands.

Dead-letter reads and replay reservations follow
[`workitem-dlq-replay.md`](workitem-dlq-replay.md).

## Sources of truth

- `_System_ConnectorEvent` remains the delivery/audit fact stream. Provider
  payload, routing snapshot, delivery identity, source order, and legacy
  lifecycle fields are immutable for WorkItem-owned generations. Event-only
  compatibility generations retain their bounded legacy lifecycle path.
- `_System_ConnectorWorkItem` is the executable projection intent and the sole
  mutable executor-state owner for WorkItem generations.
- `ProjectionKey` is the stable hash of the installed pack, event kind, route
  version, companion pack/flow identity, work type, and source lane.
- `WorkId` identifies one executable generation. The initial compatibility
  generation may use the `ProjectionKey` as its `WorkId`; every successor or
  replay after a terminal generation uses a distinct `WorkId`.
- `ProjectionActiveKey` is the nullable unique pointer to the current
  nonterminal generation. Terminal generations retain `ProjectionKey`, clear
  `ProjectionActiveKey`, and record `TerminalAt`.
- `WorkItem.EventId` points to the event snapshot currently selected for the
  next execution. Its source-projection generation is the immutable Event range
  after the prior executed Event through that selected Event's source-order key.
  Membership follows the claim path's pairwise comparison: `(EnqueueOrdinal,
  Id)` only when both rows are non-zero, otherwise `(CreatedDate,
  DeliverySequence, Id)`. Since that mixed-version comparison is not a total
  order, Flow input topologically replays its predecessor edges; only a true
  cycle falls back to `(CreatedDate, DeliverySequence, Id)`. Superseded source
  Events remain unchanged inputs to the generation selected by WorkItem
  history. A terminal WorkItem generation bounds the next range. A replayed
  terminal also bounds the range while its original Event is excluded from the
  replacement execution. No second projection registry or schema column is
  introduced.

## Transition protocol

1. Enqueue always inserts every accepted Event fact.
2. Before the active projection generation is claimed, a newer source Event can
   advance that unclaimed WorkItem's `EventId`, readiness, and immutable order
   snapshot. The replaced Event remains unchanged and stays in the selected
   generation's input range; DeliveryId never truncates that range.
3. While the active WorkItem is processing or waiting to retry, newer events
   collapse into at most one unclaimed successor WorkItem with a distinct
   `WorkId`, the same `ProjectionKey`, and a null `ProjectionActiveKey`. A newer
   arrival may replace that queued successor's snapshot before claim; it cannot
   mutate the claimed generation. WorkItem readiness and lane blocking exclude
   that null-active-key successor until the terminal handoff activates it, even
   while the current generation has a future retry backoff.
4. During rolling upgrade, the first WorkItem claim atomically changes the
   selected Event from `event` ownership to the old-binary-visible
   `work_item`/`skipped` fence while acquiring the WorkItem lease. A fresh
   legacy Event lease finishes under Event ownership; its terminal result is
   then copied to the attempt-zero WorkItem and the normal successor handoff
   runs without executing the projection a second time. A stale Event lease
   can be transferred by the WorkItem claim under the shared timeout.
5. Successful, skipped, or dead completion terminalizes only the claimed
   WorkItem generation, records `TerminalAt`, clears its
   `ProjectionActiveKey`, and activates the queued successor when one exists.
   Enqueue and completion serialize on WorkItem rows, so a process crash cannot
   reopen terminal history or expose two active generations.
6. Retrying a claimed projection updates only the same nonterminal WorkItem
   generation's retry state and `RunAfter`. The queued successor remains
   inactive while the current generation retries. A dead transition
   terminalizes that generation; it is never reopened. Lease loss and stale
   reclaim converge through WorkItem lease compare-and-swap.
7. A missing source lane fails closed to the existing per-event behavior.
8. A settled event with a source lane, which therefore enters source projection,
   rejects `TargetMutationLane*`. Collapsing source state and target-mutation
   ordering into one projection identity would make lane ownership ambiguous
   across generations. A missing source lane keeps the existing per-event
   behavior and preserves its target-mutation lane.
9. Existing Event rematerialization uses the Event's route-version snapshot even
   after that route is no longer current or has been soft-deleted. Duplicate
   delivery retains the stable projection identity. Dead-event replay creates a
   distinct WorkItem generation, records `ReplayOfWorkId`, and never reopens the
   terminal original. A replay targeted at the same pack, route, companion,
   work type, and source lane retains `ProjectionKey`. Re-resolving to a
   different pack incarnation or route version creates a different
   `ProjectionKey`. Before materialization, replay locks and rejects every other
   nonterminal generation in the same namespace, canonical event kind, and
   source lane, so retargeting cannot overlap an older pack, route, companion,
   work type, or projection identity. The original WorkItem generation identity
   also pins source-projection execution class. Event-only compatibility rows
   recover that class from their immutable route snapshot. Replay rejects a
   per-event target route or a missing Event-only class proof.
10. A durable `csp:` WorkItem bypasses the legacy pre-execution source-coalescing
   bridge. It activates its queued successor only through the atomic WorkItem
   completion handoff, so that bridge cannot replace the snapshot already
   claimed for work.
11. Ordinary materialization first checks whether any retained generation of
    the exact `ProjectionKey` already represents the incoming Event. Covered
    input is acknowledged without selecting it again, creating a generation,
    or requesting lane readiness reconciliation. This is coverage, not execution
    success: dead history does not authorize an implicit replay. Missing coverage
    still follows the existing task creation/advancement path. Explicit replay
    retains its separate conflict and lineage protocol.
12. Replacing either unclaimed selection must preserve the union of represented
    inputs. Positive ordinal and ingress-time advancement prove inclusion on the
    normal path; historical mixed ordering uses the existing pairwise membership
    predicate and subtracts other retained generations before rejecting an actual
    loss. Unsafe replacement fails the admitted transaction atomically; it neither
    acknowledges the receipt nor removes the currently executable task.
13. Ingress retains the existing namespace/pack admission locks and uses
    pessimistic READ COMMITTED isolation. A waiter sees immutable Event facts
    selected by a preceding committed materializer without acquiring Event locks
    after WorkItem locks. The normal coverage lookup uses the existing
    `(ProjectionKey, LaneOrdinal, EventId, WorkId)` index; provider timestamps and
    mutable Event lifecycle fields are not the materialization authority.

## Invariants

1. Event provider payload, routing snapshot, delivery identity, source order,
   and legacy executor fields are never overwritten by WorkItem-owned
   execution.
2. At most one nonterminal WorkItem owns `ProjectionActiveKey` for a projection.
   Any number of terminal historical generations may retain the same
   `ProjectionKey`.
3. A late event cannot be lost: it either becomes the WorkItem's selected Event
   before claim or is selected by a distinct successor generation after the
   current claim terminalizes.
4. A queued successor is a durable WorkItem generation with its own `EventId`
   and remains non-active until the prior generation terminalizes.
5. A route sweep or dead-event replay cannot change an existing Event from
   source projection back to per-event materialization. Same-identity replay
   retains `ProjectionKey`; an identity-changing replay uses a new key,
   preserves cross-identity lineage through `ReplayOfWorkId`, and requires every
   other generation in the immutable source scope to be terminal. Missing
   Event-only route proof and a per-event target route both fail closed.
6. A claimed `csp:` WorkItem generation executes its selected Event; the legacy
   bridge cannot complete it early or claim a competing successor.
7. Terminal WorkItem history and the distinct `csp:` successor handoff commit
   atomically; no restart can reopen a terminal generation or expose two active
   owners.
8. Retry, failure budget, lease, and terminal state have one mutable owner and
   one read path: the claimed WorkItem generation.
9. A preserve-payload source projection receives every raw Event after the
   prior executed generation boundary through its selected Event. Membership
   uses the claim path's pairwise source-lane key, while a mixed-version Flow
   input topologically replays its predecessor edges and uses ingress order only
   to break a true cycle. It never sees a later successor or prior history.
   Superseded immutable Event siblings remain inputs, while a terminal WorkItem
   generation is the boundary. Replaying a terminal preserves its boundary
   while excluding the replayed original from the replacement Flow payload.
10. A mixed-version Event/WorkItem race has one execution winner. WorkItem
    ownership freezes Event executor state; Event ownership converges only the
    attempt-zero WorkItem terminal and successor handoff.
11. Duplicate delivery is not projection advancement. Already represented input
    cannot regress unclaimed selection, create a redundant generation after
    claim/completion, or erase another input's task coverage.

## Acceptance evidence

The focused TiDB test must prove that 400 independent sources with 10 events
each produce 4,000 Event rows, 400 active WorkItems, and at most one active
generation per source. A race test must enqueue a successor while a projection
is processing and prove that the claimed WorkItem becomes immutable terminal
history while one distinct queued successor becomes active.
A retry test must prove that divergent Event lifecycle fields cannot change the
claimed WorkItem's retry eligibility and that a queued late successor remains
inactive until the current WorkItem terminalizes.
A mixed-version TiDB race must prove one claim and finalizer winner, stale Event
lease transfer, terminal Event-to-WorkItem convergence without a business
claim, and atomic activation of the next source-projection generation.

The duplicate regression suite additionally proves older/equal/missing provider
timestamps, processing with and without a successor, terminal coverage, missing
task repair, covered and mixed receipt batches, and a transaction waiting across
another generation's enqueue and completion. Historical admission tests prove
mixed-order rejection/rollback, retained-history subtraction, legacy-zero repair,
and route isolation. See the current measurements and owner map in
[`projection-materialization-performance.md`](projection-materialization-performance.md).

## SC400 materialization performance record

This record predates the R1.3 Event-immutable cut. Event lifecycle labels below
describe the measured historical implementation; the surviving protocol uses
WorkItem generations for selection, retry, and terminal state.

Baseline `sc400-pr124-v012-r3-20260711T071523Z` ran NexusCRM
`971ff4e8e4f470073b744e1f71ed3649983f9cd3` with connectors `v0.1.12`, 16
inbound workers, and 32 maximum processing rows. The 400 action callbacks were
durably accepted, but their Event rows materialized over 826 seconds. Exact
source-scoped readback showed a receipt-to-Event delay p50 of 787 seconds, p95
of 839 seconds, and maximum of 841 seconds. The first large receipt-router
group committed 2,070 projection Events and only 3 action Events; the next
visible batch did not commit for another 436 seconds. Once projection traffic
was gone, action-only groups of 50 materialized every 3-6 seconds.

The root is the source-projection carve-out in `EnqueueBatch`: ordinary fresh
routes use multi-row Event and WorkItem inserts, while projection routes fell
back to the per-Event transaction loop. That loop performed duplicate lookup,
Event insert, projection WorkItem select/update, and superseded-Event update for
every immutable Event. The single ordered receipt router therefore held one
large transaction while later interactive receipts waited behind bulk facts.
The receipt FIFO and the worker count are not the primary defect.

The selected repair is a `PATCH` on the existing materialization SSoT. It keeps
one transaction and the current receipt/event idempotency, pack/route/companion
locks, delivery order, and projection completion protocol. Fresh Event facts
are inserted with the existing multi-row path; ordinary command WorkItems stay
one per Event; projection WorkItems are reduced in memory to the newest Event
per deterministic `csp:` identity and advanced once per source. No schema,
migration, new queue, L1/L2 behavior, timeout increase, worker increase, or
test-specific branch is selected.

The next proof loop must compare the same 400x10 projection plus 400 interactive
action workload against this baseline. Required evidence is receipt-to-Event
latency, Event and WorkItem creation spans, 4,000 retained projection facts,
approximately 400 projection WorkItems, action progress while bulk is active,
terminal Event/WorkItem counts, and post-last-receipt drain time.

Local partition benchmark after the first implementation pass processed the
4,000 projection plus 400 action shape in 280-310 microseconds, but allocated
about 2.2 MB in 418 allocations per operation. The next narrow hypothesis is
that storing full Event structs as map values causes avoidable copying; retain a
small WorkId-to-head-index map and update a compact head slice instead.

That change reduced the same benchmark to 179-192 microseconds, about 939 KB,
and 36 allocations per operation. Partition CPU is therefore not the remaining
risk; the decisive next proof is database round trips and transaction duration
on the focused TiDB 4,000/400 integration workload.

The first isolated xiaopeng TiDB test-binary run did not execute the workload:
the generic integration helper built an insecure DSN, while the target TiDB
requires verified TLS. This is a harness transport gap, not a performance
result. The next change is limited to making the existing TiDB integration
helper honor `TIDB_TLS` with the driver's verified `tls=true` mode; no workload
or assertion is relaxed.

After that transport repair, the isolated xiaopeng TiDB test passed the exact
4,000 Event / 400 source workload in 70.13 seconds, including temporary database
setup and cleanup. Readback was 4,000 retained Events, 400 WorkItems, 400 pending
heads, 3,600 skipped superseded Events, and maximum one live Event per source.
This is the first database-backed proof that the materialization boundary is
under the existing two-minute test budget; it is not yet the final deployed
receipt/action E2E proof.

The same isolated TiDB run also passed late-event reopen, retry with a
mirrorless successor, target-lane rejection, no-source fallback, replay identity,
and 12/12 concurrent completion/enqueue races. The batch fast path therefore
changes write amplification without weakening the projection's recovery or
ordering protocol.

The final isolated xiaopeng TiDB workload combined 4,000 projection Events
(400 sources x 10 changes) with 400 interactive action Events in one fresh
batch. Materialization itself completed in 117.02 seconds and the complete
test, including temporary database setup and cleanup, passed in 118.93 seconds.
Readback proved 4,000 retained projection Events, 400 deterministic projection
WorkItems, 400 retained action Events, 400 per-event action WorkItems, 400
pending projection heads, 3,600 skipped superseded projection Events, and at
most one live projection Event per source. This bounds interactive Event
materialization with the bulk projection traffic present instead of validating
the two lanes separately.

The first deployed PR-head SC400 run, `sc400-pr126-4858770-r1-20260711T084207Z`,
passed the business oracle (400 Leads and 400 CampaignMembers) in 779.34 seconds,
versus the baseline failure with 89 Leads at 943.7 seconds. Action Event
materialization narrowed from 826 seconds to 381 seconds and all 400 actions
processed. The terminal audit then found a separate correctness residue: 12 of
400 deterministic projection WorkItems remained `blocked` although their
selected Events were already terminal. Nine selected skipped Events and three
selected processed Events; no competing WorkItem existed on those 12 lanes.

The timestamps and transaction shape identify a completion/enqueue snapshot
race rather than queue starvation. Projection completion locks the WorkItem,
then reads a pending successor. Under TiDB repeatable-read, a completion
transaction that waited on the WorkItem can keep an older snapshot and miss an
Event committed by the materializer while it was waiting. The materializer or
backlog bridge can subsequently leave the durable mirror blocked on an Event
that has become terminal.

The first repair tried a successor `FOR UPDATE` current read. It fixed the
snapshot visibility but was superseded before merge: completion already holds
the WorkItem lock, while enqueue writes Event then locks WorkItem. Locking Event
from completion would invert that order. The replacement makes only the
source-projection completion transaction pessimistic `READ COMMITTED`, keeping
the successor read nonlocking. Each statement can then see an Event committed
while completion waited, without creating an Event/WorkItem deadlock cycle.

Local TiDB v8.5.3 integration evidence covers both invariants: a successor
committed after completion began was selected after the fresh multi-row race,
and completion did not lock an uncommitted successor Event while enqueue waited
on the durable WorkItem. A deployed SC400 rerun remains the later acceptance
step: it must finish with zero non-terminal source-projection WorkItems.

The baseline's post-materialization drain is a separate, bounded phase. Action
Events terminalized at a steady approximately 60 per minute once the action-only
materialization groups became visible; their created-to-terminal p50/p95/max
were 154/329/445 seconds. The last action Event to last action terminal tail was
336 seconds. Property projection terminalization finished 148 seconds after the
last action. This is continued worker/Flow progress, not the earlier receipt
starvation shape. Moving Event materialization from an 826-second span to the
bounded batch path lets that existing drain overlap ingress instead of starting
near the 15-minute gate.

## Explicit non-goals

- reducing or deleting raw Event audit facts
- changing command/work-action delivery semantics
- changing worker count, queue-class weights, or scheduler capacity
- adding Salesforce, HubSpot, or tenant-specific branches to L0