mod async_body

The synchronous ChunkSource (flare.http.body) is a pull: each next returns the next chunk or None for EOF. That is exactly right for an in-process FIFO (e.g. SseChannel) but it can never wait -- it has no way to say "nothing yet, the bytes are coming on another fd." A streaming proxy whose chunks arrive on a UDS / pipe / eventfd needs that third state, or every front re-implements an epoll loop and a hand-rolled fd->client copy.

This module adds the missing state as a small, total API:

• ChunkPoll -- a tri-state poll result. Exactly one of: a ready chunk, "pending, wake me on this fd", or EOF. The three states are constructed by named factories (ready / pending / eof) and read by predicates, so an illegal combination (a chunk and a wait fd) is unrepresentable.
• AsyncChunkSource -- the async sibling of ChunkSource: one method, poll(cancel) -> ChunkPoll. Synchronous sources keep using ChunkSource; this is purely additive.
• UpstreamChunkSource -- the concrete impl a streaming proxy needs: a single framed logical stream over a non-blocking connection. It composes the frame codec (FrameDemux) for parsing and reports pending(fd) on EAGAIN so the reactor parks instead of busy-polling.

Composition with the typed streaming surface

A handler drives an AsyncChunkSource with no bespoke reactor code, no file descriptors, and no per-connection bookkeeping -- attach the source and let the framework pump it::

def on_open(mut self, mut conn: StreamConn) raises:
    conn.attach_upstream(UpstreamChunkSource.connect(self.worker))

def on_upstream(mut self, mut conn: StreamConn) raises:
    conn.relay_upstream()              # drains ready chunks, EOF -> close

attach_upstream takes the source (it reads the fd to watch internally and owns the source for the connection's lifetime, closing it on teardown), and relay_upstream is the standard drain loop: pull ready chunks into the client with backpressure, request close on EOF, park on pending. A front that needs custom per-chunk handling can still drive the source directly via conn.upstream().poll(...).

The reactor calls on_upstream only when the attached fd is readable, so the pending branch is a genuine park, not a poll loop (no busy-poll between gaps). Watermark backpressure additionally gates the upstream read interest when the client is write-blocked.