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

base_http_client/
connection.rs
