[BUG] UDS Connection::recv is not cancel-safe — run_session drops it mid-frame

[lxsaah]

Summary

run_session's select loop drops and rebuilds the Connection::recv() future on
every iteration. Two of the three transports are cancel-safe under that; the UDS
one is not, so an outbound event arriving while a UDS frame is still being read
silently truncates that inbound frame.

The mechanism

aimdb-core/src/session/server.rs:130-153 constructs the per-iteration futures
inside a scoped block:

let step = {
    let mut recv = conn.recv().fuse();
    let mut event = core::pin::pin!(event_rx.recv().fuse());
    select_biased! {
        r  = recv  => ...,
        ev = event => ...,
        sub_id = subs.select_next_some() => ...,
    }
};

select_biased! only prioritizes among futures that are ready. If recv is
Pending and event is Ready, the select resolves on event and recv — a local
— is dropped at the end of the block. Next iteration builds a fresh one. So
Connection::recv must be cancel-safe, but nothing in the trait says so.

Transport-by-transport

Transport	recv impl	Cancel-safe?
TCP (aimdb-tcp-connector/src/tokio_transport.rs:59)	reads into the persistent self.acc accumulator, AsyncRead::read on an Unpin stream	yes
WebSocket (aimdb-websocket-connector/src/transport.rs:140)	self.ws.next()	yes
UDS (aimdb-uds-connector/src/transport.rs:41)	read_line into a local String	no

The UDS impl:

fn recv(&mut self) -> BoxFut<'_, TransportResult<Option<Vec<u8>>>> {
    Box::pin(async move {
        let mut line = String::new();
        match self.reader.read_line(&mut line).await { ... }
    })
}

Tokio documents AsyncBufReadExt::read_line as not cancellation-safe: bytes
already consumed from the BufReader are appended to line before the await, so
dropping the future discards them. The BufReader has advanced; the data is
gone.

Impact

A truncated frame lands in the codec, fails to decode, and is skipped by the
malformed-frame arm at server.rs:182-186 — the session survives, so this
degrades rather than crashes. But the request is lost with no reply, and the
remainder of the truncated line is then read as the next frame, so it can
mis-decode too.

Requires a frame that does not arrive in a single read (a large record.set
params, record.drain with a big payload) concurrent with subscription traffic
on the same connection — so it is rare, and correspondingly hard to diagnose when
it does happen. AimX-over-UDS with an active wildcard subscription is the
realistic trigger.

Fix

Adopt the TCP shape: hoist the partial-line state into UdsConnection so it
survives across cancellations.

struct UdsConnection {
    reader: BufReader<OwnedReadHalf>,
    pending: String,   // survives a dropped recv future
    ...
}

Read into self.pending, and only take/clear it once a complete line is
assembled. (read_until/read_line still append rather than overwrite, so
resuming into the same buffer is the natural form.)

Also worth doing

Document the requirement on the trait — Connection::recv in
aimdb-core/src/session/mod.rs:345 should state that implementations must be
cancel-safe, because run_session drops the future whenever an outbound event or
a finished subscription pump wins the select. There is currently no mention of
cancel safety anywhere in session/ or in any connector.

Test

A regression test can drive it deterministically: write half a frame to the UDS
socket, fire a subscription event so the event arm wins the select, then write
the rest — and assert the request still gets its reply.

Environment

• Branch: feat/retire-aimdb-ws-protocol
• Found by inspection while reviewing per-event allocations on the AimX fan-out
  path (docs/benchmarks/wi4-fanout-encode.md); not observed in the wild.

[thaodt]

I'd love to investigate this bug if you don't mind @lxsaah