GeezerVison

Pipelines::new recomputes output_assignment from scratch on every start, and nothing records the previous assignment. So any change to the computed placement between two runs from the same checkpoint does the same damage.

What can move streams: anything that feeds assign_output_streams, which sorts streams by cost and breaks ties by name.
- Adding or removing a plain output connector on any stream, or turning distributed on or off for one. Distributed connectors don't count toward cost.
- Adding, removing or renaming a table or view, which changes the set of streams and their name order.
- A query change that makes the coordinator estimate a view as small (aggregated) instead of large, or the reverse.
- A coordinator release that changes the cost model or the tie-breaking. This is the upgrade case.

A distributed Delta part can silently drop its snapshot or a transaction's changes (6; verified)
- Why: parts commit only in steps_processed, which runs only for queue entries that carry a step count. But DeltaTableWriter::batch_start clears pending_actions on every batch.
- Effect: a snapshot queued during an open transaction carries no count, so the commit's own change batch wipes the snapshot's files, or the reverse. No error is raised, and snapshot_sent is still checkpointed, so the snapshot is never resent.
- Fix: keep the current batch's files and the current step's files in separate lists.

Setup

- Pipeline: multihost, 2 hosts, dev_tweaks.view_integrals: all_hosts, which is the default for a new pipeline. So each host holds part of every view.
- View: materialized view v = SELECT * FROM t. It already holds 300 committed rows, about 150 on each host.
- Output sink: delta_table_output with mode: append, update_mode: cdc, distributed: true, send_snapshot: true, paused: true.
- Leader: host 1, the host assigned to v. Host 0's part is a follower.

Sequence

Each step shows host 0; host 1 does the same with its own share.

1. While sink is paused: each step, CircuitThread::push_output gives it an empty entry. The snapshot stays pending until the connector starts.
2. POST /start_transaction:
   - From now on, CircuitThread::processed_records returns None after every step, because a transaction is open.
   - update_snapshot doesn't run during a transaction, so the newest stored snapshot is still rows 0..300.
3. The client inserts rows 300..310 into t inside the transaction.
4. POST /output_endpoints/sink/start: the coordinator relays it to both hosts.
5. The next step, still inside the transaction:
   - paused_outputs no longer contains sink, and its snapshot is pending.
   - Because the endpoint is distributed, push_output passes the step's own count along, which is None.
   - enqueue_latest_snapshot takes host 0's part of the stored snapshot (about 150 rows) and calls mark_snapshot_delivered.
   - It queues {snapshot, 150 rows, no step count}.
6. Host 0's output thread:
   - push_batch_to_encoder calls DeltaTableWriter::batch_start, which finds nothing pending.
   - encode writes Parquet files A for the 150 rows, so pending_actions = [A].
   - batch_end returns early, because a part keeps its files for the leader.
   - The entry has no step count, so steps_processed isn't called. Files A sit in the table directory, uncommitted.
7. POST /commit_transaction: at the step where the commit completes, processed_records becomes Some(C). The transaction's change stream is published; host 0's share is, say, 5 of the 10 new rows.
8. push_output: the snapshot is no longer pending, so the normal path queues {changes, 5 rows, step C}.
9. Host 0's output thread:
   - batch_start runs pending_actions.clear(), and files A are dropped.
   - encode writes files D0, so pending_actions = [D0], and batch_end returns early.
   - steps_processed(C) takes [D0], and OutputPart::commit reports only D0 to the leader.
10. Host 1, the leader:
    - Its own snapshot files B were dropped the same way when its change batch arrived.
    - CommitLeader::gather(C) receives D0, adds the leader's own D1, checks the paths, and commits D0 and D1 in one Delta commit.

Result

- The table holds only rows 300..310. The 300 snapshot rows never reach it.
- Files A and B remain in the table directory as orphans; VACUUM eventually removes them.
- No error is raised, and the connector stays healthy.
- The loss is permanent. Step 5 set the delivered flag. The next checkpoint stores snapshot_sent, so after a restart ControllerInner::add_output_endpoint treats the snapshot as already sent. Only changing the connector's definition clears the flag.

When it doesn't happen

- Host with no new rows in the commit: if a host has no new rows of its own, its change entry carries no data. batch_start never runs, and steps_processed(C) commits its snapshot files. With few new rows, only some hosts lose their share, so the table ends up with part of the snapshot.
- No transaction open at the start: the snapshot carries the step's count and is committed at once. I haven't traced whether a commit that spans several steps opens the same window without an API transaction.
- Plain Delta connectors: not affected, because their batch_end commits each batch immediately.

Test to reproduce it: two in-process hosts with this connector. Ingest 300 rows, start a transaction, insert 10 more, start the connector, step, then commit. Expect 310 rows in the table; I expect it to hold about 10.

Possible second bug (unverified): a start that lands on the step where a host publishes its share of a multi-step commit may make the snapshot skip that share. The pending-snapshot branch drops the step's changes, but the stored snapshot predates the transaction. If real, this would also affect plain connectors on main.

Next: waiting for your command.

H3. A distributed input plus a distributed Delta output to a new table can never start (5–6; verified)
- Why: when a distributed Delta Lake or Iceberg input needs host 0 to choose a value first, the coordinator activates host 0 alone. Host 0 publishes its choices only after all its connectors, outputs included, have initialized.
- Effect: if the output's leader is on another host, host 0's part waits for a table that only the leader creates. It gives up after about 10 attempts, and every restart repeats this.
- Fix: publish choices once the inputs have initialized, or let the other parts open the table at their first commit.

H4. Multihost snapshots are encoded one piece at a time (3; verified)
- Why: encode_snapshot_parts calls encode once per local worker part and once per ~1 MiB chunk from another host.
- Effect on Delta Lake:
  - Each call writes its own files, so thousands of small files (inferred).
  - With update_mode: merge, each call is its own commit, so readers see the snapshot half-applied (traced).
  - An encode error clears all earlier parts' files while the snapshot carries on, so the commit holds only the later parts.
- Fix: encode the local parts as one reader (whole), merge remote chunks into larger readers, and make Delta's error path drop only the failing call's files.

Medium

- M5. Leases expire mid-snapshot (verified)
  - Why: all leases are taken up front, a lease's idle time resets only after a scan, hosts are read in order after the local parts, and the idle limit is 60 minutes.
  - Effect: if the earlier parts take over an hour to drain into the sink, the later hosts fail. This also happens under one_host, where those hosts' parts are empty, so it is a regression for large snapshots that used to be purely local.
- M6. A part's report larger than 2 MiB stalls the connector forever (verified)
  - Why: /coordination/output/commit uses actix's default 2 MiB JSON body limit, and PeerClient::report_output retries the resulting 413 error forever.
  - When: about 1–2k file entries per host in one step reaches the limit, for example the snapshot of a large view (size inferred).
- M7. send_snapshot on a distributed connector under one_host duplicates or loses rows (verified; both reviewers found it)
  - Why: only host H's part has a snapshot, taken at its own start step. The other parts send changes from their own start steps, and the coordinator starts each host independently.
  - Tables are not affected: a table's integral uses the same partition function (shard_batch) as the distributed changes. I discarded one reviewer's claim that tables are affected.
  - Fix: refuse this combination unless the layout is all_hosts, or have the leader pick one snapshot step for all parts.
- M8. A failed snapshot looks complete, or disappears
  - HTTP egress ends normally with partial data; the PR's own test asserts exactly this (verified).
  - A failed configured connector vanishes from /stats (traced).
  - A checkpoint taken mid-snapshot can store snapshot_sent for a snapshot that then fails, so it is never resent (traced).
- M9. A distributed output's pause state is never made consistent across hosts (traced)
  - The coordinator does this only for inputs.
  - After a restart some parts can stay paused and drop their share, while the merged status still says running.
- M10. Switching layouts under concurrent bootstrap serves empty views (runtime traced; I didn't check whether multihost pipelines use concurrent bootstrap)
  - Why: the switch marks no relation as affected.
  - Effect: ad hoc queries and new send_snapshot connectors see empty views until cutover.
- M11. HTTP/2 flow control shared between scans and leases (inferred)
  - When: over HTTPS without client certificates.
  - Effect: unread scan data can block a lease reply. The lease runs on the circuit thread and is retried every step, up to 30 s each time.

Low

- Version skew:
  - Old coordinator with a new runtime: host H's leader waits forever for reports that never come, because only the follower checks for host URLs (verified).
  - New coordinator with an old runtime: every host runs a full writer, and they race to create the table (traced).
- Background fetch task:
  - tokio::join!(fetch, forward) hangs when the reader goes away, which pins the other host's lease and data until restart.
  - An egress client that disconnects doesn't stop the fetch.
  - A closed channel counts as success, so a panic commits a partial snapshot.
- Leasing: leases are taken one host at a time on the circuit thread, up to 30 s × (hosts − 1) per step. A failed lease waits for the next step, which an idle pipeline may never take.
- Commit leader:
  - A part that skips a step deadlocks instead of getting an error.
  - CommitLeader::gather ignores close().
  - The 60 s "unhealthy" warning fires on large but healthy steps.
- Endpoint access:
  - The new host-to-host endpoints accept any caller the pipeline accepts, even on Solo.
  - The leader trusts report.host, so a forged report can complete a step.
  - Leases are unbounded.
  - Main's /coordination/* endpoints have the same exposure.
- Unused index names: the index names in the distributed set are never used. A distributed connector on view V gives V and all its indexes a distributed change stream.
- Clock skew: __feldera_ts now comes from several hosts' clocks, which can break the documented MERGE watermark when one host's clock lags.
- Chunk size: chunks have no upper bound for indexes with few distinct keys.
- Statistics: queued_batches drifts, and some merged counters mislead.