Function stream_subscribe_flushing 

async fn stream_subscribe_flushing<C>(
    client: &mut C,
    portal: SubscriptionPortal,
    send_row_desc: bool,
    result_format: Option<&Format>,
) -> PgWireResult<()>
where
    C: Sink<PgWireBackendMessage> + Unpin + Send,
    C::Error: Debug,
    PgWireError: From<<C as Sink<PgWireBackendMessage>>::Error>,

Stream a SUBSCRIBE, flushing the Sink after every batch.

Workaround: pgwire feed()s DataRows and only flushes at end-of-response (never, for an unbounded SUBSCRIBE) or at its ~8 KB buffer, so a sparse stream stalls. Per-batch flush is unconditional (fine — batches amortise; not per-row). Retire when pgwire flushes streaming responses upstream. Both paths need it (psql=simple, psycopg/JDBC=extended).

send_row_desc: simple query carries RowDescription; extended already sent it via Describe (caller returns Response::Execution for CommandComplete). Returns Ok(()) only on pump exit.