writer: per-writer queues + work-stealing encode pool

[rgankema]

Summary

Replaces the encode pool's single MPMC channel + per-writer mutex with per-writer FIFO queues + work-stealing workers.

The problem this fixes

Workers committed to a specific writer at task-pull time. If two tasks for the same writer landed adjacently in the global FIFO, the second worker blocked on the writer's mutex even when tasks for other idle writers were queued behind it. Profiling showed ~25% pool waste in a real workload (8 workers, 14 concurrent writers).

The new design

• Each WriterState owns a Mutex<VecDeque<RecordBatch>> + atomic queue_len + atomic busy flag.
• A worker scans the active-writer snapshot, skips writers with empty queues, and CAS busy: false → true to claim one. It then drains that writer's queue to empty before releasing.
• Wakeups use a single shared Notify with two safeguards:
  • Cascade on claim: a worker that wins a claim notifies a peer before draining, so concurrent producer notifications that collapse into one stored permit still rouse enough workers.
  • Re-check after release: after busy.store(false), the worker reloads queue_len and notifies again if a producer pushed during the release window.

Invariants

• Per-writer FIFO: busy serializes pops, and pushes go to the back.
• No stranded tasks: the cascade + post-release re-check covers the missed-wakeup races.
• Close semantics unchanged: pending / done_notify are decremented and signaled exactly once per batch.
• Panic handling unchanged: catch_unwind + first-error-wins on state.error.
• FFI surface unchanged: iceberg_set_encode_workers, submit_batch, iceberg_writer_* all keep their signatures and semantics.

Test plan

• `cargo build --release` clean
• `cargo clippy --tests` clean (no new warnings)
• `cargo test --lib` — 27 passed (25 pre-existing + 2 new)
• Fairness test (new): 4 writers × 8 batches, 20ms delay; asserts per-writer FIFO and that every 4 consecutive completions span all 4 writers.
• Stranded-task stress (new): 4 producer threads × 8 writers × 50 batches; asserts `pending` and `queue_len` reach zero on every writer and exactly 1,600 completions land.
• Performance validation on the originating workload (waiting on author).
• `make test` end-to-end against MinIO.

🤖 Generated with Claude Code