perf: 3x throughput and half the CPU by removing the ingest queue (py-only)

[drake-nominal]

Touches only py-nominal-streaming. The core crate is not modified.

Bottom line

Writing through this package used to be bounded by a queue that served no purpose. Removing it
means the same hardware moves several times the data for a fifth of the CPU, and Ctrl+C now
flushes instead of hanging.

Measured against the real staging backend, not a local file -- so the uploader is a genuine
gRPC round trip and backpressure is real. Interleaved A/B, three rounds each, batch_size=250_000,
max_wait=0.1s.

	before	after
Sustained throughput to staging, 6,480 channels at one timestamp	0.48 M points/s	2.22 M points/s	4.6x
CPU per million points	2.45 s	0.50 s	5x less
Time blocked inside enqueue, same shape, local target	10,766 us	762 us	14x
Time blocked inside enqueue, single point, local target	1.86 us	0.23 us	8x

Every point was verified delivered, not merely accepted into a buffer: reading the channels back
out of staging reconciles exactly with what each run enqueued (e.g. 1,250 records expected and
1,250 stored). That check matters here because the request dispatcher currently counts a failed
upload as a success, so throughput alone could have measured discarding data faster.

It depends on the shape you write

shape	before	after	throughput	CPU
wide -- many channels, one timestamp (enqueue_from_dict)	0.48 Mp/s	2.22 Mp/s	4.6x	5x less
deep -- one channel, a long series (enqueue_batch)	2.7 Mp/s	2.7 Mp/s	unchanged	unchanged
grouped -- many channels, a short series each	3.6 Mp/s	3.8 Mp/s	unchanged	23% less
float array, width 8 (enqueue_float_array)	960 kcall/s	1,534 kcall/s	1.6x	2.3x less
float array, width 64	639 kcall/s	996 kcall/s	1.56x	2.2x less
float array, width 512	184 kcall/s	266 kcall/s	1.45x	1.9x less
string array, width 32 (enqueue_string_array)	507 kcall/s	746 kcall/s	1.47x	1.8x less

Deep and grouped writes already produced one item per call, so the queue never bound them -- and
note they were already 5-8x faster than a wide write on main. A caller that groups points per
channel before writing was therefore never hurt by this bug, and gains CPU here rather than
throughput. A caller using enqueue_from_dict was hit hardest, and gains the most.

Array-valued channels write one point per call, so they sit on the same per-call path as scalar
enqueue and gain accordingly. The gain shrinks as the array widens -- 1.6x at width 8 down to
1.45x at width 512 -- which is what a removed fixed per-call cost should do as the payload per
call grows. In element terms the wide case moves 136 M elements/s after, against 94 M before.

The practical guidance is unchanged by this PR and worth stating: batch per channel where you
can. grouped remains the fastest scalar shape at 3.8 Mp/s, and a wide array is the most
efficient way to move bulk volume.

What is fundamentally different

Every point used to cross a hardcoded bounded(4) tokio mpsc channel into a single forwarding
task. That cost a thread park/unpark round trip per item, not per call — so
enqueue_from_dict, the API that looks like a bulk write, was the worst case in the package: one
boundary crossing but one item per channel, and measurably slower than a python loop over
enqueue.

The channel bought nothing. NominalDatasetStream::enqueue is synchronous and Sync (confirmed
with a compile probe), so python threads now call it directly under py.detach(). The tokio
runtime stays, but only for the uploader.

Two consequences beyond raw speed:

• Ingest is now genuinely concurrent. Python threads contend on the stream's buffer lock
  instead of serializing through one task.
• Backpressure is honest. Blocking now means the uploader is actually behind, not that a
  4-deep queue is full.

An earlier analysis attributed the fixed per-call cost to the Python->Rust boundary and proposed
"fewer, larger calls". The numbers reproduce; the attribution does not. enqueue_from_dict with
6,480 channels is one pyo3 crossing and 6,480 items: if the boundary were the cost it
would take ~119us, and it took 10,923us. The lever was item count.

Is it safe

Yes, and it fixes two ways the old code could lose data.

• Avro output is byte-identical to main across scalars, series, wide dicts, tags, structs,
  arrays and datetime timestamps, compared by digest.
• Ctrl+C now flushes. Previously the handler abandoned buffered points. Now it refuses further
  writes and drains what is already enqueued. Verified against a live process with a real SIGINT:
  10,442,916 points enqueued single-threaded and 10,442,916 in the file (exit in 3.4s); 1,498,653
  across four writer threads and 1,498,653 in the file, with all four threads correctly refused.
  Refusing first is what makes the drain converge — KeyboardInterrupt reaches only the main
  thread, so workers would otherwise keep feeding the buffer indefinitely.
• A stream left to the garbage collector now drains instead of stalling the interpreter. The
  pyclass had no Drop; dropping the stream drains it, and pyclass deallocation holds the GIL.
  The new Drop releases it. Verified: 400,000 points all written, 0.54s, off the GIL.
• The stream is owned rather than held in an Arc, so the drain cannot silently become a
  no-op if something later clones it.
• 8-thread concurrent ingest lands exactly 40,400 points, no loss or duplication.
• cargo test, mypy, ruff and clippy are clean (one pre-existing clippy warning at lib.rs:267).

The two commits are not independently safe and should land together: with only the perf
commit, Ctrl+C on a multi-threaded writer deadlocks, because close() needs &mut self and
cannot take the pyclass borrow while workers sit inside enqueue holding it. Confirmed by test —
the process hung rather than draining. #321 was folded in here for that reason.

Smaller things included

• Absent and empty tags both normalize to None, so a channel written with tags=None and one
  written with tags={} share a series rather than splitting. Identical on the wire.
• Skip timestamp normalization when the caller already passed integral nanoseconds, and stop
  copying the caller's mapping in enqueue_from_dict, which can hold thousands of keys.
• open() raises when the stream fails to build, rather than logging and returning a stream that
  rejects every write.

Deliberately not here

• The remaining win on tagged wide records (3,852us -> 1,029us) needs a breaking change to
  ChannelDescriptor in the core crate. Separate PR.
• enqueue_many, worth a further ~7% on wide records, is a core-crate addition. Separate PR.
• Waking the batch processor on buffer fill, worth 1.8-2.5x when buffers fill before the flush
  timer. Core crate, and it currently costs 4-5% throughput at saturation. Separate PR.

Known, not introduced here

• open() installs a SIGINT handler whose closure captures self, so the signal module holds a
  strong reference and a main-thread stream is never garbage collected. Wants a weakref.
• max_points_per_record is a soft limit under concurrency: when_capacity reads the count
  without the lock. Unchanged code, but it had one caller before this PR and now has N. Overshoot
  is bounded per racing writer and the buffer already tolerates exceeding its cap.
• Python integers silently become doubles (pyo3's f64::extract accepts anything with
  __float__), losing precision above 2^53.

🤖 Generated with Claude Code