src/corosio/src/tls/detail/engine_driver.hpp

97.3% Lines (107 / 110) 100.0% Functions (34 / 34)

[ ] prev/next file n p prev/next uncovered line

engine_driver.hpp

f(x) Functions (34)

Function Calls Lines Blocks

Line	Hits	Source Code
1		//
2		// Copyright (c) 2026 Steve Gerbino
3		//
4		// Distributed under the Boost Software License, Version 1.0. (See accompanying
5		// file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt)
6		//
7		// Official repository: https://github.com/cppalliance/corosio
8		//
9
10		#ifndef SRC_TLS_DETAIL_ENGINE_DRIVER_HPP
11		#define SRC_TLS_DETAIL_ENGINE_DRIVER_HPP
12
13		#include <boost/corosio/tls_context.hpp>
14		#include <boost/corosio/tls_stream.hpp>
15
16		#include "src/tls/detail/engine_types.hpp"
17
18		#include <boost/capy/buffers.hpp>
19		#include <boost/capy/detail/buffer_array.hpp>
20		#include <boost/capy/ex/async_mutex.hpp>
21		#include <boost/capy/io/any_stream.hpp>
22		#include <boost/capy/results.hpp>
23		#include <boost/capy/task.hpp>
24		#include <boost/capy/write.hpp>
25
26		#include <atomic>
27		#include <concepts>
28		#include <cstddef>
29		#include <cstdint>
30		#include <cstring>
31		#include <limits>
32		#include <string>
33		#include <string_view>
34		#include <system_error>
35		#include <tuple>
36		#include <utility>
37		#include <vector>
38
39		// This header is instantiated in both backends' TUs, whose vendor
40		// headers cannot coexist (WolfSSL's OpenSSL-compat layer clashes
41		// with genuine OpenSSL declarations on some builds); it must stay
42		// vendor-free.
43		#ifdef OPENSSL_VERSION_NUMBER
44		#error engine_driver.hpp must stay vendor-header-free
45		#endif
46		#ifdef WOLFSSL_VERSION
47		#error engine_driver.hpp must stay vendor-header-free
48		#endif
49
50		/*
51		TLS Driver Architecture
52
53		TLS layer wrapping an underlying stream (via any_stream). One
54		read_some and one write_some may be in flight concurrently.
55
56		Engine / Driver Split. The engine (each backend's detail/engine.hpp)
57		owns the TLS session, its byte interface, and every library-result-
58		to-error-code decision; it is synchronous and transport-free. This
59		driver owns the transport, the per-direction claims, and all
60		buffering, and loops on engine verdicts:
61
62		1. Call eng_.perform(op, data, len)
63		2. output_then_retry: flush pending ciphertext, then retry
64		3. input: feed transport bytes into the engine, then retry
65		4. done / output_then_done: flush if pending, report the
66		(already mapped) ec
67
68		Renegotiation causes cross-direction I/O: a read may need to
69		write handshake data, a write may need to read. Each operation
70		services whatever direction the engine requests.
71
72		Full-Duplex Concurrency. rd_cm_ and wr_cm_ are separate transport
73		claims: a read parked on the network cannot stall a concurrent
74		write's flush, and vice versa. A cross-direction request (e.g. a
75		read needing to flush queued handshake output) takes the other
76		direction's claim itself.
77
78		read_gen_ counts successful deposits into the engine. Each engine
79		call snapshots it before running; if fill_input later finds the
80		generation has moved (a concurrent operation's flush suspension let
81		a deposit land in the meantime) it retries the engine instead of
82		issuing a redundant transport read.
83
84		The transport reads ciphertext straight into the engine's input
85		staging via input_area(), so no driver-side buffer or deposit copy
86		sits on the read path. The write path stays copy-based on purpose:
87		ciphertext drained from the engine into out_buf_[0, out_len_) is
88		decoupled from the engine's shared output staging, so a concurrent
89		reader can still emit its own records (an alert, a key update) while
90		a write is in flight. out_len_ holds only the abandoned tail a failed
91		or canceled write kept for retry; the length being written is a local,
92		never published, so a concurrent flush probe cannot mistake in-flight
93		bytes for a tail and park on the write claim behind a healthy write.
94
95		A read or write that fully satisfies the caller's buffer still owes
96		a trailing flush (queued handshake or session-ticket output must
97		reach the peer before the call can report success). If that flush
98		fails, the transfer already succeeded and the stream contract
99		forbids reporting both a full count and an error in the same
100		result, so the error is stashed in pending_flush_ec_ and surfaced
101		by the next read, write, or shutdown call instead.
102		*/
103
104		namespace boost::corosio {
105
106		namespace detail {
107
108		/** The engine surface `engine_driver` drives.
109
110		A synchronous, transport-free TLS record engine: `perform`
111		advances one operation and returns a fully mapped verdict, while
112		`put_input` / `get_output` shuttle wire bytes. The handshake
113		hooks (`check_context`, `check_session`, `prepare`) let each
114		backend gate and configure a handshake at the protocol-mandated
115		points without the driver knowing the backend's session model.
116		*/
117		template<class Engine>
118		concept tls_engine = requires(
119		Engine& e,
120		Engine const& ce,
121		engine_op op,
122		void* buf,
123		unsigned char const* in,
124		unsigned char* out,
125		std::size_t n,
126		tls_context const& ctx,
127		tls_role role,
128		std::string const& hostname,
129		std::string& alpn) {
130		{ e.perform(op, buf, n) } -> std::same_as<engine_result>;
131		{ e.put_input(in, n) } -> std::same_as<std::size_t>;
132		{ e.input_area() } -> std::same_as<std::pair<unsigned char*, std::size_t>>;
133		{ e.input_committed(n) };
134		{ ce.pending_output() } -> std::same_as<std::size_t>;
135		{ e.get_output(out, n) } -> std::same_as<std::size_t>;
136		{ ce.received_shutdown() } -> std::same_as<bool>;
137		{ ce.capture_alpn(alpn) };
138		{ e.reset() };
139		{ ce.check_context() } -> std::same_as<std::error_code>;
140		{ ce.check_session() } -> std::same_as<std::error_code>;
141		{ e.prepare(ctx, role, hostname) } -> std::same_as<std::error_code>;
142		};
143
144		/** Coroutine driver shared by every TLS backend.
145
146		Hosts the transport, the per-direction claims, the buffering, and
147		the four operation loops; the engine type supplies all TLS
148		mechanics. Each backend's stream impl instantiates this template
149		with its own engine so the driver logic exists exactly once.
150
151		@par Thread Safety
152		Distinct objects: Safe.@n
153		Shared objects: Unsafe.
154		*/
155		template<tls_engine Engine>
156		class engine_driver
157		{
158		// Large enough to hold the largest possible TLS record.
159		static constexpr std::size_t buffer_size_ = std::size_t{17} * 1024;
160
161		// Operations hold a reference across their suspensions, so the
162		// driver outlives the stream that created it until they finish.
163		std::atomic<std::size_t> refs_{1};
164
165		// Set when the stream is destroyed: no transport call may follow.
166		std::atomic<bool> orphaned_{false};
167
168		// Transport calls in flight. Once orphaned, the owned transport is
169		// destroyed when this reaches zero: right away for a stream with no
170		// call parked, so it never outlives its context waiting on an
171		// operation nobody awaits.
172		std::atomic<std::size_t> calls_{0};
173		std::atomic<bool> transport_gone_{false};
174
175		// The transport in owning mode; s_ refers to it without owning.
176		tls_owned_transport owned_;
177		capy::any_stream s_;
178		tls_context ctx_;
179		Engine eng_;
180
181		// A handshake was attempted (successfully or not); the stream
182		// must be reset before the next handshake.
183		bool used_ = false;
184
185		// Per-stream SNI/verification hostname, set via set_hostname().
186		std::string hostname_;
187
188		// ALPN protocol negotiated during the handshake (empty if none).
189		std::string alpn_selected_;
190
191		std::vector<char> out_buf_;
192
193		// One transport claim per direction so a parked reader cannot block
194		// a writer's flush. read_gen_ counts deposits into the engine: each
195		// operation captures it at its engine call and passes it to
196		// fill_input, which retries the engine instead of issuing a stale
197		// transport read when a deposit landed anywhere after the engine's
198		// verdict (including during an intervening flush suspension).
199		capy::async_mutex rd_cm_;
200		capy::async_mutex wr_cm_;
201		std::uint64_t read_gen_ = 0;
202
203		// Engine ciphertext drained but not yet written: the abandoned
204		// tail a failed or canceled transport write kept for retry.
205		// Zero while a write is in flight (that length is a local).
206		std::size_t out_len_ = 0;
207
208		// A trailing flush that fails after the engine already accepted a
209		// full payload cannot be reported alongside n == size (the stream
210		// contract forbids error + full transfer, and capy::write's
211		// composed loop discards ec exactly in that case); the error is
212		// held here and surfaced by the next operation instead.
213		std::error_code pending_flush_ec_;
214
215	43210x	capy::task<std::error_code> flush_output()
216		{
217		if (eng_.pending_output() == 0 && out_len_ == 0)
218		co_return std::error_code{};
219
220		auto [lec] = co_await wr_cm_.lock();
221		if (lec)
222		co_return lec;
223		capy::async_mutex::lock_guard wr_guard(&wr_cm_);
224
225		// Drain under the claim so records reach the wire in engine
226		// order even when both directions produce output.
227		while (eng_.pending_output() > 0 \|\| out_len_ > 0)
228		{
229		if (orphaned())
230		co_return make_error_code(capy::error::canceled);
231
232		// Take the abandoned tail (from an earlier failed write in
233		// this loop or a prior flush) into a local and append fresh
234		// engine ciphertext behind it, preserving record order.
235		// out_len_ drops to zero for the duration of the write: the
236		// in-flight length lives only in `n`, so a concurrent flush
237		// probe never sees these bytes as a tail and never parks on
238		// the write claim behind a healthy write.
239		std::size_t n = out_len_;
240		out_len_ = 0;
241		if (n < out_buf_.size())
242		n += eng_.get_output(
243		reinterpret_cast<unsigned char*>(out_buf_.data()) + n,
244		out_buf_.size() - n);
245		// The loop guard just confirmed pending bytes exist, so a
246		// drain failure here is unreachable in practice; fail loudly
247		// rather than silently drop already-accepted ciphertext.
248		if (n == 0) // LCOV_EXCL_LINE unreachable: pending bytes confirmed
249		co_return make_error_code(
250		std::errc::
251		no_buffer_space); // LCOV_EXCL_LINE unreachable: transport returned 0 with no error unreachable: pending bytes confirmed
252		// capy::write's loop, with the stream's destruction checked
253		// between partial writes: once orphaned, the transport is
254		// no longer this driver's to write to.
255		std::error_code ec;
256		std::size_t wn = 0;
257		while (wn < n)
258		{
259		transport_call call(*this);
260		if (!call)
261		{
262		ec = make_error_code(capy::error::canceled);
263		break;
264		}
265		auto [wec, k] = co_await s_.write_some(
266		capy::const_buffer(out_buf_.data() + wn, n - wn));
267		call.end();
268		wn += k;
269		if (wn < n && wec)
270		{
271		ec = wec;
272		break;
273		}
274		if (wn < n && orphaned())
275		{
276		ec = make_error_code(capy::error::canceled);
277		break;
278		}
279		}
280		if (ec)
281		{
282		// wn bytes already reached the peer; keep only the unsent
283		// remainder so a post-cancellation flush retry resends
284		// neither the delivered prefix nor loses the rest.
285		std::memmove(out_buf_.data(), out_buf_.data() + wn, n - wn);
286		out_len_ = n - wn;
287		co_return ec;
288		}
289		}
290		co_return std::error_code{};
291	86420x	}
292
293		// Flushes on a fatal/alert exit path: the engine may have queued a
294		// close_notify or alert that should still reach the peer, but a
295		// caller already reporting the triggering error cannot also report
296		// this flush's own failure. flush_output() itself no-ops when
297		// nothing is pending.
298	20x	capy::task<void> best_effort_flush()
299		{
300		std::ignore = co_await flush_output();
301	40x	}
302
303		// gen is the caller's read_gen_ snapshot from its engine call; a
304		// later snapshot would miss deposits landing while the caller's
305		// flush_output was suspended.
306	3651x	capy::task<std::error_code> fill_input(std::uint64_t gen)
307		{
308		// Input already arrived since the engine's verdict; retry the
309		// engine rather than queue behind a re-parked reader.
310		if (read_gen_ != gen)
311		co_return std::error_code{};
312
313		auto [lec] = co_await rd_cm_.lock();
314		if (lec)
315		co_return lec;
316		capy::async_mutex::lock_guard rd_guard(&rd_cm_);
317
318		// Input arrived while we queued for the claim; the caller's
319		// engine retry consumes it.
320		if (read_gen_ != gen)
321		co_return std::error_code{};
322
323		if (orphaned())
324		co_return make_error_code(capy::error::canceled);
325
326		// Read the transport straight into the engine's input staging:
327		// input_area() hands back the contiguous writable run, so no
328		// staging buffer or deposit copy sits between the socket and the
329		// engine.
330		auto [dst, cap] = eng_.input_area();
331		// The staging is full while the caller still wants input: its own
332		// engine retry must decrypt the staged record to free space, so
333		// report success to re-run it rather than issue a zero-length read.
334		if (cap == 0)
335		co_return std::error_code{};
336
337		transport_call call(*this);
338		if (!call)
339		co_return make_error_code(capy::error::canceled);
340		auto [ec, n] = co_await s_.read_some(capy::mutable_buffer(dst, cap));
341		call.end();
342
343		// The stream went away while the read was parked; its result
344		// belongs to no one.
345		if (orphaned())
346		co_return make_error_code(capy::error::canceled);
347
348		// ReadStream permits n>0 alongside ec (IOCP forwards
349		// bytes_transferred on failed completions; a canceled read can
350		// likewise deliver bytes); commit whatever arrived before
351		// surfacing ec, or the record stream loses wire bytes the
352		// transport already handed over.
353		if (n > 0)
354		{
355		eng_.input_committed(n);
356		++read_gen_;
357		co_return ec;
358		}
359
360		if (ec)
361		co_return ec;
362
363		// The transport delivered nothing without an error, so it cannot
364		// make progress: fail loudly rather than spin the engine's input
365		// retry against a staging that will never fill.
366		co_return make_error_code(
367		std::errc::
368		no_buffer_space); // LCOV_EXCL_LINE unreachable: staging cannot stay empty
369	7302x	}
370
371	4588x	void destroy_transport() noexcept
372		{
373	4588x	if (transport_gone_.exchange(true, std::memory_order_acq_rel))
374	2294x	return;
375	2294x	auto const owned = std::exchange(owned_, tls_owned_transport{});
376	2294x	if (owned.destroy)
377	6x	owned.destroy(owned.p);
378		}
379
380		// Return false once orphaned: the transport may already be gone.
381		// The orphan check follows the count's increment, pairing with
382		// orphan()'s store-then-load, so one side always sees the other.
383	43160x	bool begin_transport_call() noexcept
384		{
385	43160x	calls_.fetch_add(1, std::memory_order_seq_cst);
386	43160x	if (!orphaned_.load(std::memory_order_seq_cst))
387	43160x	return true;
388	✗	end_transport_call();
389	✗	return false;
390		}
391
392	45454x	void end_transport_call() noexcept
393		{
394	136232x	if (calls_.fetch_sub(1, std::memory_order_seq_cst) == 1 &&
395	45324x	orphaned_.load(std::memory_order_seq_cst))
396	2294x	destroy_transport();
397	45454x	}
398
399		// A transport call that also ends when its frame is unwound while
400		// parked, so the count never pins an orphaned transport.
401		class transport_call
402		{
403		engine_driver* d_;
404
405		public:
406	43158x	~transport_call()
407		{
408	43158x	end();
409	43158x	}
410
411	43158x	explicit transport_call(engine_driver& d) noexcept
412	43158x	: d_(d.begin_transport_call() ? &d : nullptr)
413		{
414	43158x	}
415
416		transport_call(transport_call const&) = delete;
417		transport_call& operator=(transport_call const&) = delete;
418
419	43158x	explicit operator bool() const noexcept
420		{
421	43158x	return d_ != nullptr;
422		}
423
424	86315x	void end() noexcept
425		{
426	86315x	if (d_)
427	43158x	std::exchange(d_, nullptr)->end_transport_call();
428	86315x	}
429		};
430
431		// A prior read/write already reported its full transfer as success;
432		// the trailing flush error it deferred surfaces on the next call
433		// instead. Each entry point takes it exactly once.
434	73960x	std::error_code take_pending_flush_ec() noexcept
435		{
436	73960x	std::error_code ec = pending_flush_ec_;
437	73960x	pending_flush_ec_ = {};
438	73960x	return ec;
439		}
440
441		public:
442		/** Construct a driver over a transport stream.
443
444		The driver starts with one reference, the creating stream's.
445
446		@param s The transport.
447		@param owned The transport `s` refers to, when the stream owns
448		it; the driver destroys it on @ref orphan.
449		@param ctx The TLS context handed to the engine's handshake
450		preparation.
451		*/
452	2294x	engine_driver(
453		capy::any_stream s, tls_owned_transport owned, tls_context ctx)
454	2294x	: owned_(owned)
455	2294x	, s_(std::move(s))
456	4588x	, ctx_(std::move(ctx))
457		{
458	2294x	out_buf_.resize(buffer_size_);
459	2294x	}
460
461	2294x	~engine_driver()
462		{
463	2294x	destroy_transport();
464	2294x	}
465
466		engine_driver(engine_driver const&) = delete;
467		engine_driver& operator=(engine_driver const&) = delete;
468
469		/// Add a reference for an operation in flight.
470	75710x	void add_ref() noexcept
471		{
472	75710x	refs_.fetch_add(1, std::memory_order_relaxed);
473	75710x	}
474
475		/// Drop a reference; return true if it was the last.
476	78004x	bool release_ref() noexcept
477		{
478	156008x	return refs_.fetch_sub(1, std::memory_order_acq_rel) == 1;
479		}
480
481		/** Detach the driver from its destroyed stream.
482
483		Operations still in flight finish as canceled without touching
484		the transport again. An owned transport is destroyed now when
485		no transport call is parked on it; otherwise it is told to
486		cancel, if it can, and destroyed when the last parked call
487		returns.
488		*/
489	2294x	void orphan() noexcept
490		{
491		// Counted as a call itself, so a parked call returning meanwhile
492		// cannot destroy the transport under the cancel.
493	2294x	calls_.fetch_add(1, std::memory_order_seq_cst);
494	2294x	orphaned_.store(true, std::memory_order_seq_cst);
495	2294x	if (owned_.cancel)
496	4x	owned_.cancel(owned_.p);
497	2294x	end_transport_call();
498	2294x	}
499
500		/// Cancel the owned transport's pending operations, if it can.
501	2x	void cancel_transport() noexcept
502		{
503	2x	if (!begin_transport_call())
504	✗	return;
505	2x	if (owned_.cancel)
506	1x	owned_.cancel(owned_.p);
507	2x	end_transport_call();
508		}
509
510		/// Return true once the stream has been destroyed.
511	46808x	bool orphaned() const noexcept
512		{
513	46808x	return orphaned_.load(std::memory_order_acquire);
514		}
515
516		/// Return the transport.
517	2x	capy::any_stream& stream() noexcept
518		{
519	2x	return s_;
520		}
521
522		/// Return the transport.
523		capy::any_stream const& stream() const noexcept
524		{
525		return s_;
526		}
527
528		/// Return the engine for backend-specific setup.
529		Engine& engine() noexcept
530		{
531		return eng_;
532		}
533
534		/// Return the TLS context this driver was constructed with.
535		tls_context const& context() const noexcept
536		{
537		return ctx_;
538		}
539
540		/// Set the hostname applied to the next client handshake.
541	13x	void set_hostname(std::string_view hostname)
542		{
543	13x	hostname_ = hostname;
544	13x	}
545
546		/// Return the ALPN protocol negotiated by the last handshake.
547	3x	std::string_view alpn_protocol() const noexcept
548		{
549	3x	return alpn_selected_;
550		}
551
552		/// Reset the engine and every per-connection driver state.
553	65x	void reset()
554		{
555	65x	eng_.reset();
556
557	65x	out_len_ = 0;
558
559	65x	alpn_selected_.clear();
560	65x	pending_flush_ec_ = {};
561	65x	used_ = false;
562	65x	}
563
564	36880x	capy::task<capy::results<std::size_t>> do_read_some(
565		capy::detail::mutable_buffer_array<capy::detail::max_iovec_> buffers)
566		{
567		if (auto ec = take_pending_flush_ec())
568		co_return {ec, 0};
569
570		std::error_code ec;
571		std::size_t total_read = 0;
572		std::size_t const bufs_size = capy::buffer_size(buffers);
573
574		for (auto& buf : buffers)
575		{
576		char* const dest = static_cast<char*>(buf.data());
577		// Clamp for the engines' int-sized transfer API: a >=2 GiB
578		// buffer becomes a partial read, never a truncated length.
579		int const remaining = static_cast<int>(
580		(std::min)(buf.size(),
581		std::size_t((std::numeric_limits<int>::max)())));
582		if (remaining == 0)
583		continue;
584
585		// Exits by co_return: success after the first transferred
586		// chunk, or any terminal error.
587		while (true)
588		{
589		auto const gen = read_gen_;
590		auto r = eng_.perform(
591		engine_op::read, dest, static_cast<std::size_t>(remaining));
592
593		if (r.ec)
594		{
595		// Terminal, already mapped by the engine. eof (a
596		// received close_notify) arrives as a plain done:
597		// it queues no output, so no flush is needed.
598		if (r.want == engine_want::output_then_done)
599		co_await best_effort_flush();
600		co_return {r.ec, total_read};
601		}
602
603		if (r.want == engine_want::done \|\|
604		r.want == engine_want::output_then_done)
605		{
606		total_read += r.bytes;
607
608		// r.bytes > 0 already satisfies ReadStream's "at
609		// least one byte transferred" success condition;
610		// report now rather than loop for more (another
611		// engine call could park on input). out_len_ > 0
612		// covers a retained tail the engine cannot see.
613		if (r.want == engine_want::output_then_done \|\| out_len_ > 0)
614		ec = co_await flush_output();
615		if (ec && total_read == bufs_size)
616		{
617		// First failure wins: concurrent directions share
618		// one transport failure domain; the earliest
619		// error is the meaningful one.
620		if (!pending_flush_ec_)
621		pending_flush_ec_ = ec;
622		ec = {};
623		}
624		co_return {ec, total_read};
625		}
626
627		if (r.want == engine_want::output_then_retry)
628		{
629		ec = co_await flush_output();
630		if (ec)
631		co_return {ec, total_read};
632		continue;
633		}
634
635		// want == input. Flush first: a retained tail (an
636		// earlier failed write) may hold bytes the peer needs
637		// before it will send more; no-op when nothing pends.
638		ec = co_await flush_output();
639		if (ec)
640		co_return {ec, total_read};
641
642		ec = co_await fill_input(gen);
643		if (ec)
644		{
645		ec = map_fill_error(
646		engine_op::read, ec, eng_.received_shutdown());
647		co_return {ec, total_read};
648		}
649		}
650		}
651
652		co_return {std::error_code{}, total_read};
653	73760x	}
654
655	36931x	capy::task<capy::results<std::size_t>> do_write_some(
656		capy::detail::const_buffer_array<capy::detail::max_iovec_> buffers)
657		{
658		if (auto ec = take_pending_flush_ec())
659		co_return {ec, 0};
660
661		std::error_code ec;
662		std::size_t total_written = 0;
663		std::size_t const bufs_size = capy::buffer_size(buffers);
664
665		for (auto const& buf : buffers)
666		{
667		// The engine only reads through this pointer for a write
668		// op; the cast satisfies perform's single transfer
669		// signature.
670		void* const src = const_cast<void*>(buf.data());
671		// Clamp for the engines' int-sized transfer API: a >=2 GiB
672		// buffer becomes a partial write, never a truncated length.
673		int const remaining = static_cast<int>(
674		(std::min)(buf.size(),
675		std::size_t((std::numeric_limits<int>::max)())));
676		if (remaining == 0)
677		continue;
678
679		// Exits by co_return: success after the first transferred
680		// chunk, or any terminal error.
681		while (true)
682		{
683		auto const gen = read_gen_;
684		auto r = eng_.perform(
685		engine_op::write, src, static_cast<std::size_t>(remaining));
686
687		if (r.ec)
688		{
689		// Terminal, already mapped. eof means the peer's
690		// close_notify WAS received (an announced close);
691		// it arrives as a plain done and needs no flush.
692		if (r.want == engine_want::output_then_done)
693		co_await best_effort_flush();
694		co_return {r.ec, total_written};
695		}
696
697		if (r.want == engine_want::done \|\|
698		r.want == engine_want::output_then_done)
699		{
700		total_written += r.bytes;
701
702		// r.bytes > 0 already satisfies WriteStream's "at
703		// least one byte transferred" success condition;
704		// report now rather than loop for more. out_len_ > 0
705		// covers a retained tail the engine cannot see.
706		if (r.want == engine_want::output_then_done \|\| out_len_ > 0)
707		ec = co_await flush_output();
708		if (ec && total_written == bufs_size)
709		{
710		// First failure wins: concurrent directions share
711		// one transport failure domain; the earliest
712		// error is the meaningful one.
713		if (!pending_flush_ec_)
714		pending_flush_ec_ = ec;
715		ec = {};
716		}
717		co_return {ec, total_written};
718		}
719
720		if (r.want == engine_want::output_then_retry)
721		{
722		ec = co_await flush_output();
723		if (ec)
724		co_return {ec, total_written};
725		continue;
726		}
727
728		// want == input: a write can need a read mid-rekey; the
729		// transport eof this surfaces means the same thing it
730		// means on the read path, so map it the same way. Flush
731		// first for the same retained-tail reason as the read
732		// path.
733		ec = co_await flush_output();
734		if (ec)
735		co_return {ec, total_written};
736
737		ec = co_await fill_input(gen);
738		if (ec)
739		{
740		ec = map_fill_error(
741		engine_op::write, ec, eng_.received_shutdown());
742		co_return {ec, total_written};
743		}
744		}
745		}
746
747		co_return {std::error_code{}, total_written};
748	73862x	}
749
750	1748x	capy::task<capy::results<>> do_handshake(tls_role role)
751		{
752		// Refuse the handshake while the engine's configuration is
753		// unusable, before consuming any per-connection state.
754		if (auto cec = eng_.check_context())
755		co_return {cec};
756
757		if (used_)
758		reset();
759
760		// reset() may not have restored a usable session; refuse to
761		// hand out a handshake on it.
762		if (auto sec = eng_.check_session())
763		co_return {sec};
764
765		// A failed attempt leaves the session in a dead state; any
766		// attempt, not just a completed handshake, consumes the stream
767		// so the next handshake() starts fresh.
768		used_ = true;
769
770		if (auto pec = eng_.prepare(ctx_, role, hostname_))
771		co_return {pec};
772
773		auto const op = role == tls_role::client ? engine_op::handshake_client
774		: engine_op::handshake_server;
775
776		std::error_code ec;
777
778		while (true)
779		{
780		auto const gen = read_gen_;
781		auto r = eng_.perform(op, nullptr, 0);
782
783		if (r.ec)
784		{
785		if (r.want == engine_want::output_then_done)
786		co_await best_effort_flush();
787		co_return {r.ec};
788		}
789
790		if (r.want == engine_want::done \|\|
791		r.want == engine_want::output_then_done)
792		{
793		eng_.capture_alpn(alpn_selected_);
794		ec = co_await flush_output();
795		co_return {ec};
796		}
797
798		if (r.want == engine_want::output_then_retry)
799		{
800		// Must flush (e.g. ClientHello) before reading the
801		// peer's reply.
802		ec = co_await flush_output();
803		if (ec)
804		co_return {ec};
805		continue;
806		}
807
808		// want == input. The flush is a no-op unless a retained
809		// tail pends; the fill error passes through map_fill_error
810		// unchanged (no close is clean before the session is
811		// established).
812		ec = co_await flush_output();
813		if (ec)
814		co_return {ec};
815
816		ec = co_await fill_input(gen);
817		if (ec)
818		{
819		ec = map_fill_error(op, ec, eng_.received_shutdown());
820		co_return {ec};
821		}
822		}
823	3496x	}
824
825	149x	capy::task<capy::results<>> do_shutdown()
826		{
827		if (auto ec = take_pending_flush_ec())
828		co_return {ec};
829
830		std::error_code ec;
831
832		while (true)
833		{
834		auto const gen = read_gen_;
835		auto r = eng_.perform(engine_op::shutdown, nullptr, 0);
836
837		if (r.ec)
838		{
839		if (r.want == engine_want::output_then_done)
840		co_await best_effort_flush();
841		co_return {r.ec};
842		}
843
844		if (r.want == engine_want::done \|\|
845		r.want == engine_want::output_then_done)
846		{
847		// Covers both the bidirectional-complete result and the
848		// engine's shutdown-after-close success mapping.
849		ec = co_await flush_output();
850		co_return {ec};
851		}
852
853		if (r.want == engine_want::output_then_retry)
854		{
855		// Sends our close_notify before parking for the peer's.
856		ec = co_await flush_output();
857		if (ec)
858		co_return {ec};
859		continue;
860		}
861
862		// want == input: awaiting the peer's close_notify. It may
863		// already have been deposited (and consumed by a concurrent
864		// reader) during a flush; fill_input then returns without
865		// reading and the loop retries the engine instead of parking.
866		ec = co_await flush_output();
867		if (ec)
868		co_return {ec};
869
870		ec = co_await fill_input(gen);
871		if (ec)
872		{
873		ec = map_fill_error(
874		engine_op::shutdown, ec, eng_.received_shutdown());
875		co_return {ec};
876		}
877		}
878	298x	}
879		};
880
881		/** A reference to a TLS stream's driver held by an operation.
882
883		Taken when the operation is created, so a stream destroyed or
884		moved before the operation is awaited or finishes leaves its
885		driver alive until the operation lets go.
886
887		@tparam Impl The stream's implementation, derived from
888		@ref engine_driver.
889		*/
890		template<class Impl>
891		class driver_ref
892		{
893		Impl* p_;
894
895		public:
896		/// Construct, adding a reference to @p p.
897	75710x	explicit driver_ref(Impl* p) noexcept : p_(p)
898		{
899	75710x	p_->add_ref();
900	75710x	}
901
902		/// Construct by taking @p other's reference.
903	75710x	driver_ref(driver_ref&& other) noexcept
904	75710x	: p_(std::exchange(other.p_, nullptr))
905		{
906	75710x	}
907
908		driver_ref(driver_ref const&) = delete;
909		driver_ref& operator=(driver_ref const&) = delete;
910		driver_ref& operator=(driver_ref&&) = delete;
911
912		/// Destroy, deleting the driver if this was its last reference.
913	151420x	~driver_ref()
914		{
915	151420x	if (p_ && p_->release_ref())
916	6x	delete p_;
917	151420x	}
918
919		/// Return the driver.
920	75708x	Impl* operator->() const noexcept
921		{
922	75708x	return p_;
923		}
924		};
925
926		/// Read through a driver an operation keeps alive.
927		template<class Impl>
928		capy::task<capy::results<std::size_t>>
929	36882x	driver_read_some(
930		driver_ref<Impl> d,
931		capy::detail::mutable_buffer_array<capy::detail::max_iovec_> buffers)
932		{
933		co_return co_await d->do_read_some(buffers);
934	73764x	}
935
936		/// Write through a driver an operation keeps alive.
937		template<class Impl>
938		capy::task<capy::results<std::size_t>>
939	36931x	driver_write_some(
940		driver_ref<Impl> d,
941		capy::detail::const_buffer_array<capy::detail::max_iovec_> buffers)
942		{
943		co_return co_await d->do_write_some(buffers);
944	73862x	}
945
946		/// Handshake through a driver an operation keeps alive.
947		template<class Impl>
948		capy::task<capy::results<>>
949	1748x	driver_handshake(driver_ref<Impl> d, tls_role role)
950		{
951		co_return co_await d->do_handshake(role);
952	3496x	}
953
954		/// Shut down through a driver an operation keeps alive.
955		template<class Impl>
956		capy::task<capy::results<>>
957	149x	driver_shutdown(driver_ref<Impl> d)
958		{
959		co_return co_await d->do_shutdown();
960	298x	}
961
962		} // namespace detail
963
964		} // namespace boost::corosio
965
966		#endif
967