base_http_client/
connection.rs

1// Released under the MIT License.
2// Copyright, 2026, by Samuel Williams.
3
4use crate::{
5    Endpoint, Error, Options, ResponseBody,
6    body::Consumption,
7    driver::{Assigned, Operation, Queue, checked, linked, lock},
8    outgoing::Interim,
9    writable::{
10        COMPLETE, ExchangeCompletion, FINALIZE, Open, OpenFactory, Opening, Return, ReturnTo,
11    },
12};
13use base_endpoint::{
14    AddressEndpoint, ConnectEndpoint,
15    stream::{ReadableStream, WritableStream},
16};
17use base_http::{
18    InterimResponses, Response,
19    body::{BodyError, BodyReader, ReadableBody},
20};
21use base_http1::{RequestWriter, ResponseReader};
22use base_pool::{Admission, MultiplexBuildPermit, MultiplexResource};
23use base_scheduler::{
24    Cancellation, SocketClose, SocketConnect, SocketOpen, SocketStreamReading, SocketStreamWriting,
25};
26use event_listener::Event;
27use futures_channel::oneshot;
28use futures_util::future::join;
29use std::sync::{
30    Arc, Mutex, Weak,
31    atomic::{AtomicBool, Ordering},
32};
33
34struct Inner<'scope> {
35    assigned: Mutex<Option<Assigned<'scope>>>,
36    direct: Mutex<Option<Arc<dyn DirectConnection<'scope> + 'scope>>>,
37    changed: Event,
38    stop: Cancellation,
39    result: Mutex<Option<Result<(), Error>>>,
40}
41
42/// An established HTTP/1 session retaining its transport and codec buffers.
43/// Direct requests take and return its directions in the calling task;
44/// supervised work handles construction, queued calls and retirement.
45#[derive(Clone)]
46pub struct Connection<'scope> {
47    inner: Arc<Inner<'scope>>,
48}
49impl<'scope> Connection<'scope> {
50    fn new() -> Self {
51        Self {
52            inner: Arc::new(Inner {
53                assigned: Mutex::new(None),
54                direct: Mutex::new(None),
55                changed: Event::new(),
56                stop: Cancellation::new(),
57                result: Mutex::new(None),
58            }),
59        }
60    }
61    pub(crate) fn clone_handle(&self) -> Self {
62        self.clone()
63    }
64    pub(crate) fn submit(&self, assigned: Assigned<'scope>) {
65        let mut slot = lock(&self.inner.assigned);
66        assert!(slot.is_none(), "HTTP/1 admits one exchange at a time");
67        *slot = Some(assigned);
68        drop(slot);
69        self.inner.changed.notify(usize::MAX);
70    }
71    pub(crate) fn open(&self, opening: Open<'scope>) -> Opening<'scope> {
72        // Admission is published only after the session installs its direct state, which it never removes.
73        let direct = lock(&self.inner.direct)
74            .clone()
75            .expect("an admitted connection has direct state");
76        direct.open(opening)
77    }
78    async fn next(&self, lifetime: &Cancellation) -> Option<Assigned<'scope>> {
79        loop {
80            event_listener::listener!(self.inner.changed => listener);
81            if let Some(assigned) = lock(&self.inner.assigned).take() {
82                return Some(assigned);
83            }
84            if checked(
85                checked(
86                    async {
87                        listener.await;
88                        Ok(())
89                    },
90                    &self.inner.stop,
91                ),
92                lifetime,
93            )
94            .await
95            .is_err()
96            {
97                return None;
98            }
99        }
100    }
101    pub(crate) fn stop(&self) {
102        self.inner.stop.cancel();
103    }
104    pub(crate) async fn wait_closed(&self) -> Result<(), Error> {
105        loop {
106            event_listener::listener!(self.inner.changed => listener);
107            if let Some(result) = lock(&self.inner.result).clone() {
108                return result;
109            }
110            listener.await;
111        }
112    }
113    fn finish(&self, result: Result<(), Error>) {
114        let mut current = lock(&self.inner.result);
115        if current.is_none() {
116            *current = Some(result);
117        }
118        drop(current);
119        self.inner.changed.notify(usize::MAX);
120    }
121}
122
123struct Owner<'scope> {
124    connection: Connection<'scope>,
125    registration: Option<MultiplexResource<Connection<'scope>>>,
126}
127impl Drop for Owner<'_> {
128    fn drop(&mut self) {
129        if let Some(registration) = &self.registration {
130            registration.retire();
131        }
132        let abandoned = lock(&self.connection.inner.assigned).take();
133        drop(abandoned);
134        self.connection.finish(Err(Error::Closed));
135    }
136}
137
138#[allow(clippy::too_many_arguments)]
139pub(crate) async fn run<'scope, S>(
140    scheduler: &'scope S,
141    endpoint: Endpoint,
142    options: Options,
143    build: MultiplexBuildPermit<Connection<'scope>>,
144    cancellation: Cancellation,
145    lifetime: Cancellation,
146    ready: oneshot::Sender<Result<(), Error>>,
147    queue: Weak<Queue<'scope>>,
148) where
149    S: SocketOpen + SocketConnect + SocketClose + SocketStreamReading + SocketStreamWriting,
150    S::Socket: 'static,
151{
152    let connection = Connection::new();
153    let mut owner = Owner {
154        connection: connection.clone(),
155        registration: None,
156    };
157    let mut build = Some(build);
158    let mut ready = Some(ready);
159    let result = async {
160        let address = endpoint.address;
161        let endpoint_transport = AddressEndpoint::new(address);
162        #[cfg(feature = "rustls")]
163        if let Some(tls) = endpoint.tls {
164            use base_endpoint::tls::{TlsClientEndpoint, TlsSession, rustls::RustlsConnector};
165            let mut config = (*tls.config).clone();
166            config.alpn_protocols = options.protocols.alpn_protocols();
167            let endpoint = TlsClientEndpoint::new(
168                endpoint_transport,
169                RustlsConnector::new(Arc::new(config)),
170                tls.server_name,
171            );
172            let transport = endpoint
173                .connect(scheduler, &cancellation)
174                .await
175                .map_err(|error| Error::connect(address, error))?;
176            let selected = transport.negotiated_protocol().map(<[u8]>::to_vec);
177            return transport
178                .with_duplex(scheduler, &lifetime, |reader, writer| async {
179                    let protocol = options.protocols.negotiated(selected.as_deref())?;
180                    let (input, output) = protocol
181                        .client(reader, writer, options.http1.clone())
182                        .expect("Client::new validated the HTTP/1 options");
183                    session(
184                        input,
185                        output,
186                        &mut owner,
187                        &mut build,
188                        &mut ready,
189                        &options,
190                        &cancellation,
191                        &lifetime,
192                        queue.clone(),
193                    )
194                    .await
195                })
196                .await;
197        }
198        let socket = endpoint_transport
199            .connect(scheduler, &cancellation)
200            .await
201            .map_err(|error| Error::connect(address, error))?;
202        let initialized = async {
203            let mut reader = scheduler.socket_stream_reader(&socket)?;
204            let writer = match scheduler.socket_stream_writer(&socket) {
205                Ok(writer) => writer,
206                Err(error) => {
207                    let _ = ReadableStream::close(&mut reader).await;
208                    return Err(Error::from(error));
209                }
210            };
211            Ok::<_, Error>(
212                options
213                    .protocols
214                    .plaintext()
215                    .client(reader, writer, options.http1.clone())
216                    .expect("Client::new validated the HTTP/1 options"),
217            )
218        }
219        .await;
220        let result = match initialized {
221            Ok((input, output)) => {
222                session(
223                    input,
224                    output,
225                    &mut owner,
226                    &mut build,
227                    &mut ready,
228                    &options,
229                    &cancellation,
230                    &lifetime,
231                    queue.clone(),
232                )
233                .await
234            }
235            Err(error) => Err(error),
236        };
237        let closed = scheduler.socket_close(socket).await;
238        result?;
239        closed?;
240        Ok(())
241    }
242    .await;
243    // Failed construction remains charged through socket/lane cleanup.
244    drop(build);
245    if let Some(ready) = ready {
246        let _ = ready.send(Err(result.as_ref().err().cloned().unwrap_or(Error::Closed)));
247    }
248    connection.finish(result);
249}
250
251// A connection retains its typed HTTP buffers and transport directions. Only
252// an admitted exchange may take them; short locks protect transfer, never I/O.
253trait DirectConnection<'scope>: Send + Sync {
254    fn open(self: Arc<Self>, opening: Open<'scope>) -> Opening<'scope>;
255}
256struct Codecs<R, W> {
257    input: Option<ResponseReader<R>>,
258    output: Option<RequestWriter<W>>,
259}
260struct Http1Connection<'scope, R, W> {
261    codecs: Mutex<Codecs<R, W>>,
262    changed: Event,
263    closing: AtomicBool,
264    registration: MultiplexResource<Connection<'scope>>,
265    queue: Weak<Queue<'scope>>,
266    open: OpenFactory<'scope, R, W>,
267}
268impl<R: ReadableStream, W: WritableStream> Http1Connection<'_, R, W> {
269    fn take(&self) -> Result<(ResponseReader<R>, RequestWriter<W>), Error> {
270        let mut codecs = lock(&self.codecs);
271        if self.closing.load(Ordering::Acquire) || codecs.input.is_none() || codecs.output.is_none()
272        {
273            return Err(Error::Closed);
274        }
275        Ok((codecs.input.take().unwrap(), codecs.output.take().unwrap()))
276    }
277    fn returned(&self) {
278        if self.closing.load(Ordering::Acquire) {
279            self.changed.notify(usize::MAX);
280        }
281    }
282    async fn close(&self) -> Result<(), Error> {
283        self.closing.store(true, Ordering::Release);
284        let (mut input, mut output) = loop {
285            event_listener::listener!(self.changed => listener);
286            let returned = {
287                let mut codecs = lock(&self.codecs);
288                if codecs.input.is_some() && codecs.output.is_some() {
289                    Some((codecs.input.take().unwrap(), codecs.output.take().unwrap()))
290                } else {
291                    None
292                }
293            };
294            if let Some(codecs) = returned {
295                break codecs;
296            }
297            listener.await;
298        };
299        let a = input.close().await;
300        let b = output.close().await;
301        drop(input);
302        drop(output);
303        a?;
304        b?;
305        Ok(())
306    }
307}
308impl<'scope, R: ReadableStream + 'scope, W: WritableStream + 'scope> DirectConnection<'scope>
309    for Http1Connection<'scope, R, W>
310{
311    fn open(self: Arc<Self>, opening: Open<'scope>) -> Opening<'scope> {
312        let (input, output) = match self.take() {
313            Ok(codecs) => codecs,
314            Err(error) => {
315                self.registration.retire();
316                opening.active.complete();
317                return crate::writable::failed_open(error);
318            }
319        };
320        let completion = ExchangeCompletion::new(
321            opening.active,
322            opening.pending,
323            opening.cancellation.clone(),
324            self.registration.clone(),
325        );
326        // Construct guards before the future: even dropping an unpolled open
327        // returns both directions and acknowledges their retirement obligation.
328        let input = Return::new(input, completion.clone(), self.clone(), false);
329        let output = Return::new(output, completion, self.clone(), true);
330        (self.open)(
331            input,
332            output,
333            opening.metadata,
334            opening.interim,
335            opening.lifetime,
336        )
337    }
338}
339impl<'scope, R: ReadableStream + 'scope, W: WritableStream + 'scope>
340    ReturnTo<'scope, ResponseReader<R>> for Http1Connection<'scope, R, W>
341{
342    fn give_back(
343        self: Arc<Self>,
344        input: ResponseReader<R>,
345        mode: u8,
346        completion: Arc<ExchangeCompletion<'scope>>,
347    ) {
348        let reusable = mode == COMPLETE && input.is_reusable();
349        lock(&self.codecs).input = Some(input);
350        self.returned();
351        completion.returned(1, reusable, None);
352    }
353}
354impl<'scope, R: ReadableStream + 'scope, W: WritableStream + 'scope>
355    ReturnTo<'scope, RequestWriter<W>> for Http1Connection<'scope, R, W>
356{
357    fn give_back(
358        self: Arc<Self>,
359        output: RequestWriter<W>,
360        mode: u8,
361        completion: Arc<ExchangeCompletion<'scope>>,
362    ) {
363        if mode == FINALIZE {
364            // The exchange's pending admission retains the queue until both directions return.
365            let queue = self
366                .queue
367                .upgrade()
368                .expect("an unfinished exchange retains its queue");
369            let cancellation = completion.cancellation().clone();
370            let mut owner = Return::new(output, completion, self, true);
371            let work = Box::pin(async move {
372                let result = owner
373                    .value
374                    .as_mut()
375                    .unwrap()
376                    .write_frame(base_http::body::BodyFrame::<Vec<u8>>::End, &cancellation)
377                    .await;
378                match result {
379                    Ok(()) => owner.complete(),
380                    Err(error) => owner.report_error(error.into()),
381                }
382                // Failure/drop returns the poisoned writer through the guard.
383            });
384            // An abandoned driver drops the work, whose guard abandons the output.
385            let _ = queue.push_work(work);
386            return;
387        }
388        let reusable = mode == COMPLETE && output.is_reusable();
389        lock(&self.codecs).output = Some(output);
390        self.returned();
391        completion.returned(
392            2,
393            reusable,
394            Some(if mode == COMPLETE {
395                Ok(())
396            } else {
397                Err(Error::Abandoned)
398            }),
399        );
400    }
401}
402
403#[allow(clippy::too_many_arguments)]
404async fn session<'scope, R, W>(
405    mut input: ResponseReader<R>,
406    mut output: RequestWriter<W>,
407    owner: &mut Owner<'scope>,
408    build: &mut Option<MultiplexBuildPermit<Connection<'scope>>>,
409    ready: &mut Option<oneshot::Sender<Result<(), Error>>>,
410    options: &Options,
411    cancellation: &Cancellation,
412    lifetime: &Cancellation,
413    queue: Weak<Queue<'scope>>,
414) -> Result<(), Error>
415where
416    R: ReadableStream + 'scope,
417    W: WritableStream + 'scope,
418{
419    if cancellation.is_cancelled() {
420        let _ = input.close().await;
421        let _ = output.close().await;
422        return Err(Error::Cancelled);
423    }
424    // Set the direct interface before publication; no admitted caller can see
425    // a connection without its codec storage.
426    let Ok(registration) = build
427        .take()
428        .unwrap()
429        .commit(owner.connection.clone(), Admission::concurrent(0))
430    else {
431        // Commit fails only once shutdown has closed the pool, which queues this connection for retirement. Its waiter sees the closed client, and only a failed close is a driver failure.
432        let ready = ready.take().expect("a session reports readiness once");
433        let _ = ready.send(Err(Error::Closed));
434        let input = input.close().await;
435        let output = output.close().await;
436        input?;
437        output?;
438        return Ok(());
439    };
440    owner.registration = Some(registration.clone());
441    let state = Arc::new(Http1Connection {
442        codecs: Mutex::new(Codecs {
443            input: Some(input),
444            output: Some(output),
445        }),
446        changed: Event::new(),
447        closing: AtomicBool::new(false),
448        registration,
449        queue,
450        open: crate::writable::factory(),
451    });
452    *lock(&owner.connection.inner.direct) = Some(state.clone());
453    // A registration that is already draining publishes no admission; its retirement stops the session below.
454    state.registration.update(Admission::concurrent(1));
455    // Construction runs one session, and reports readiness itself only if the session never does.
456    let ready = ready.take().expect("a session reports readiness once");
457    let _ = ready.send(Ok(()));
458    // Only queued pull-body calls run here. Imperative open/write/read acquires the retained state directly on the calling task.
459    while let Some(Assigned {
460        mut operation,
461        active,
462    }) = owner.connection.next(lifetime).await
463    {
464        // Direct exchanges restore both directions before releasing their use, and closing begins only after this loop.
465        let (mut input, mut output) = state
466            .take()
467            .expect("an admitted exchange finds both codec directions");
468        let local = operation.cancellation.clone();
469        let caller = operation.lifetime.clone();
470        let outcome = linked(
471            exchange(&mut input, &mut output, &mut operation, options),
472            &local,
473            &caller,
474            lifetime,
475        )
476        .await;
477        operation.request.body.close(None);
478        if let Some(reply) = operation.reply.take() {
479            let _ = reply.send(Err(outcome
480                .as_ref()
481                .err()
482                .cloned()
483                .unwrap_or(Error::Closed)));
484        }
485        let reusable = matches!(outcome, Ok(true));
486        if !reusable {
487            state.registration.retire();
488        }
489        {
490            let mut codecs = lock(&state.codecs);
491            codecs.input = Some(input);
492            codecs.output = Some(output);
493        }
494        drop(operation);
495        active.complete();
496        if !reusable {
497            break;
498        }
499    }
500    state.registration.retire();
501    state.close().await
502}
503
504async fn exchange<R: ReadableStream, W: WritableStream>(
505    input: &mut ResponseReader<R>,
506    output: &mut RequestWriter<W>,
507    operation: &mut Operation<'_>,
508    options: &Options,
509) -> Result<bool, Error> {
510    let Operation {
511        request,
512        interim,
513        reply,
514        cancellation,
515        ..
516    } = operation;
517    let metadata = request.metadata.clone();
518    let headers = AtomicBool::new(false);
519    let send = async {
520        let result = checked(
521            async {
522                output
523                    .write_request(request, cancellation)
524                    .await
525                    .map_err(Error::from)
526            },
527            cancellation,
528        )
529        .await;
530        // Preserve a readable final response when the peer refuses further
531        // upload. The connection is retired even if that response succeeds.
532        if result.is_err() && !headers.load(Ordering::Acquire) {
533            cancellation.cancel();
534        }
535        result
536    };
537    let receive = async {
538        let mut writer = None;
539        let result = checked(
540            async {
541                let mut observer = InterimResponses::new(Interim(interim.as_mut()));
542                let mut response = input
543                    .read_response(&metadata, &mut observer, cancellation)
544                    .await?;
545                let consumption = Consumption::new(cancellation.clone());
546                let empty = response.body.is_empty();
547                let (reader, mut producer) = BodyReader::bounded(
548                    options.response_buffer_bytes,
549                    options.response_buffer_frames,
550                );
551                // Publish known-empty input as completed before handing it to
552                // another task. Its consumer may drop it immediately, without
553                // racing a redundant End write or retiring a healthy session.
554                if empty {
555                    consumption.complete();
556                    producer.close_write(None);
557                }
558                let body = ResponseBody {
559                    reader,
560                    consumption: consumption.clone(),
561                    length: response.body.length(),
562                    trailers: response.body.may_have_trailers(),
563                };
564                writer = Some(producer);
565                let returned = Response::from_parts(response.metadata, body);
566                if reply.take().unwrap().send(Ok(returned)).is_err() {
567                    return Err(Error::Abandoned);
568                }
569                headers.store(true, Ordering::Release);
570                if !empty {
571                    loop {
572                        let frame = response.body.read(cancellation).await?;
573                        let end = frame.is_end();
574                        writer
575                            .as_mut()
576                            .unwrap()
577                            .write_frame(frame, cancellation)
578                            .await?;
579                        if end {
580                            break;
581                        }
582                    }
583                }
584                consumption.wait().await
585            },
586            cancellation,
587        )
588        .await;
589        if let Err(error) = &result {
590            if let Some(writer) = &mut writer {
591                writer.close_write(Some(match error {
592                    Error::Body(error) => error.clone(),
593                    error => BodyError::new(error.clone()),
594                }));
595            }
596            cancellation.cancel();
597        }
598        result
599    };
600    let (sent, received) = join(send, receive).await;
601    match (sent, received) {
602        (Ok(persistent), Ok(())) => Ok(persistent && input.is_reusable()),
603        (Err(_), Ok(())) => Ok(false),
604        (Err(error), Err(Error::Cancelled)) if !matches!(error, Error::Cancelled) => Err(error),
605        (_, Err(error)) => Err(error),
606    }
607}
608
609#[cfg(test)]
610mod tests {
611    use super::*;
612    use crate::{Client, driver::Pending, outgoing::tests::ignore};
613    use base_http::{Method, RequestMetadata};
614    use base_scheduler_direct::DirectScheduler;
615    use futures_lite::future::{block_on, yield_now, zip};
616    use std::{io::Read, net::TcpListener, thread};
617
618    #[test]
619    fn opening_an_admitted_use_after_the_session_closes_fails_and_releases_it() {
620        let listener = TcpListener::bind("127.0.0.1:0").unwrap();
621        let address = listener.local_addr().unwrap();
622        let peer = thread::spawn(move || {
623            let (mut stream, _) = listener.accept().unwrap();
624            let mut bytes = Vec::new();
625            stream.read_to_end(&mut bytes).unwrap();
626            assert!(bytes.is_empty());
627        });
628        let scheduler = DirectScheduler::new(Default::default()).unwrap();
629        let (client, driver) =
630            Client::new(&scheduler, Endpoint::new(address), Options::default()).unwrap();
631        let lifetime = Cancellation::new();
632        let application = async {
633            let active = client
634                .context
635                .acquire(&Cancellation::never())
636                .await
637                .unwrap();
638            let connection = active.clone_handle();
639            let pending = Pending::new(client.queue.clone()).unwrap();
640            // Another thread can close the session between admission and opening its codecs.
641            lifetime.cancel();
642            while lock(&connection.inner.result).is_none() {
643                yield_now().await;
644            }
645            let opening = Open {
646                metadata: RequestMetadata::new(Method::GET),
647                interim: Box::new(ignore),
648                active,
649                pending,
650                cancellation: Cancellation::new(),
651                lifetime: Cancellation::never(),
652            };
653            let error = connection.open(opening).await.err().unwrap();
654            assert_eq!(error.to_string(), Error::Closed.to_string());
655        };
656        let (result, ()) = block_on(zip(driver.run(&lifetime), application));
657        result.unwrap();
658        assert_eq!(client.pool_stats().total(), 0);
659        peer.join().unwrap();
660        block_on(scheduler.shutdown());
661    }
662}
base_http_client/connection.rs

base_http_client/
connection.rs
