perf: enqueue directly into the stream, and share channel tag maps

[drake-nominal]

Two commits, both aimed at making bulk writes cost what they should. Measured against a local
file target (no uploader) at --release on both sides.

What was wrong

Every enqueued point crossed a hardcoded bounded(4) tokio mpsc channel drained by a single
forwarding task, costing a park/unpark round-trip per EnqueueItem rather than per call. That
made enqueue_from_dict — one boundary crossing, but one item per channel — no faster than looping
over enqueue, and measurably slower on a 6,480-channel record (10,923us vs 8,964us).

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

With that gone, the next cost became visible: a wide record is thousands of channels sharing one
timestamp and one set of tags, but ChannelDescriptor owned its tags, so each channel copied the
map — once building the descriptor, once more when the buffer first saw the channel.

Results

Measured on a local file target, --release on both sides. Numbers were taken on a workstation
with other load present; repeating a measurement on the same build moves the median by 11-17%, so
treat one significant figure as meaningful and ignore differences smaller than that.

Three write shapes, min of 7:

shape	before	after
6,480 channels @ one timestamp	10,766us	714us	15.1x
10,000 points @ one channel	576us	182us	3.2x
single point	1.86us	0.22us	8.4x

Realistic workload -- 6,480 tagged channels at 10Hz, median of 200 samples per call:

max_request_delay	main	+ direct enqueue	+ shared tags
0.1s	10,462us	4,260us	1,532us	6.8x
1.0s	12,224us	6,011us	2,756us	4.4x

Both commits carry their weight at both settings: direct enqueue is 2.0-2.5x, shared tags a
further 2.2-2.8x.

Commits

commit	what
d38cb9d	Delete the ingest channel; python calls stream.enqueue directly. Add enqueue_many so a wide record takes one capacity reservation and one buffer lock.
8c8fbf6	ChannelDescriptor::tags becomes Option<Arc<BTreeMap<..>>>, so a wide record shares one tag map instead of copying it per channel. Breaking for direct field access.
bce5cc2	Chunk enqueue_many to max_points_per_record, bounding both request size and lock hold time; own the stream rather than holding it in an Arc, so the drain cannot become a silent no-op.

Also in here

• Absent and empty tags both normalize to None, so a channel written with tags=None and one
  written with tags={} share a series instead of splitting into two. Identical on the wire.
• Python-side: skip timestamp normalization when the caller already passed integral nanoseconds,
  and stop copying the caller's mapping in enqueue_from_dict (it can hold thousands of channels).
• open() now raises when the stream fails to build, rather than logging and returning a stream
  that rejects every write.
• Concurrent ingest from several python threads now contends only on the buffer lock, rather than
  serializing through one forwarding task.

Breaking

ChannelDescriptor::tags is now Option<Arc<BTreeMap<String, String>>>. new() and with_tags()
are unaffected; only code reading or assigning the field directly needs a change.

Verification

• Avro round-trip output is byte-identical to main across every write shape: scalars, series,
  wide dicts, tags, structs, arrays, and non-integral timestamps.
• 21 cargo tests pass, including two new ones covering enqueue_many.
• 8-thread concurrent ingest lands exactly 40,400 points, no loss or duplication.
• ruff, mypy, and clippy clean (one pre-existing clippy warning at lib.rs:267, untouched).

Known follow-up

Per-call cost is about 1.8x higher at max_request_delay=1.0s than at 0.1s (2,756us vs
1,532us), which is well outside the 11-17% repeat spread. At 0.1s there is essentially nothing
left on the table: a tagged wide write costs 1,532us against a 1,489us floor measured with
flushing switched off entirely.

The mechanism is identified but not yet fixed: batch_processor parks for the full
max_request_delay after each flush and is only woken early by a write that no longer fits, so a
writer that fills the buffer can block for the remainder of the delay. Two candidate fixes
(building the protobuf series off the buffer lock, and waking the processor at a fill threshold)
were tried and neither showed a reliable win; both are worth retrying, since the measurements that
rejected them were taken in a noisy window. That work is not part of this PR.

🤖 Generated with Claude Code

[drake-nominal]

Superseded by a split into independently reviewable PRs, so the core-crate changes can be judged on their own terms:

• perf: 3x throughput and half the CPU by removing the ingest queue (py-only) #320 — the direct-enqueue change, py-only, core crate untouched. Carries nearly all the win: 4.6x throughput and 5x less CPU against the staging backend.
• perf: add enqueue_many so a wide record takes one buffer lock, not one per channel #322 — enqueue_many (core, additive): ~10% CPU on top.
• feat: share a channel descriptor's tag map instead of copying it per channel #323 — shared tag maps (core, breaking): ~35% CPU on tagged wide records.
• perf: wake the batch processor when a buffer nears capacity #326 — flush-on-fill (core): 1.7x lower blocking under fill pressure, 3.6% less peak throughput.

All numbers here were also re-measured against the real staging backend rather than a local file target, with delivery verified by reading the data back out.