Experimental tunnel API for out-of-package network transports

docs/exp/SER009.md

@@ -0,0 +1,22 @@
+`RESPite.Transports.DuplexTransport` (and `TransportReceiver`) are an experimental transport
+abstraction: staged-and-flushed outbound (the transport itself is the `IBufferWriter<byte>`), push
+inbound with transport-owned memory, and an explicit batch-end notification. They exist to let
+alternative IO engines plug in beneath the library (see `Tunnel.ConnectTransportAsync`) without being
+constrained to `Stream` or pipe semantics.
+
+**This API is not intended for external use at this time.** It exists to support internal
+experimentation with alternative transports, and is public only because a seam has to be public to be
+implemented. Specifically:
+
+1. **No stability whatsoever is implied.** The shape may gain members, change semantics, be renamed,
+   move assembly or namespace, or be removed outright — between *any* two versions, including
+   patches, with no breaking-change ceremony, no deprecation period, and no migration notes.
+2. **Do not implement or consume it in code you ship.** If you build on it anyway, you are accepting
+   that any update may break you without warning, and issues asking for compatibility or support for
+   it will be closed.
+3. The `[Experimental]` diagnostic (`SER009`) is the enforcement mechanism: suppressing it is you
+   signing up to the above.
+
+If a transport seam useful beyond this experiment emerges, it will be stabilised deliberately — with
+its own announcement and a removed `[Experimental]` marker — rather than by this surface quietly
+hardening into a contract.

src/RESPite/PublicAPI/PublicAPI.Unshipped.txt

@@ -1 +1,14 @@
 #nullable enable
+[SER009]RESPite.Transports.DuplexTransport
+[SER009]RESPite.Transports.DuplexTransport.DuplexTransport() -> void
+[SER009]abstract RESPite.Transports.DuplexTransport.Advance(int count) -> void
+[SER009]abstract RESPite.Transports.DuplexTransport.DisposeAsync() -> System.Threading.Tasks.ValueTask
+[SER009]abstract RESPite.Transports.DuplexTransport.Flush() -> bool
+[SER009]abstract RESPite.Transports.DuplexTransport.GetMemory(int sizeHint = 0) -> System.Memory<byte>
+[SER009]virtual RESPite.Transports.DuplexTransport.GetSpan(int sizeHint = 0) -> System.Span<byte>
+[SER009]abstract RESPite.Transports.DuplexTransport.Start(RESPite.Transports.TransportReceiver! receiver) -> void
+[SER009]RESPite.Transports.TransportReceiver
+[SER009]RESPite.Transports.TransportReceiver.TransportReceiver() -> void
+[SER009]abstract RESPite.Transports.TransportReceiver.OnReceived(System.ReadOnlySpan<byte> payload) -> bool
+[SER009]virtual RESPite.Transports.TransportReceiver.OnBatchEnd() -> void
+[SER009]virtual RESPite.Transports.TransportReceiver.OnClosed(System.Exception? fault) -> void

src/RESPite/Shared/Experiments.cs

@@ -20,6 +20,7 @@ internal static class Experiments
         public const string UnitTesting = "SER005";
         public const string GeoRedundantFailover = "SER007";
         public const string Server_8_10 = "SER008";
+        public const string Transport = "SER009";
 
         // ReSharper restore InconsistentNaming
 

src/RESPite/Transports/DuplexTransport.cs

@@ -0,0 +1,72 @@
+using System;
+using System.Buffers;
+using System.Diagnostics.CodeAnalysis;
+using System.Threading.Tasks;
+
+namespace RESPite.Transports;
+
+/// <summary>
+/// A duplex byte transport, deliberately NOT a <see cref="System.IO.Stream"/> and NOT a pipe: outbound
+/// is staged-and-flushed (batching is an explicit contract point, not an implementation accident), and
+/// inbound is PUSH — the transport delivers bytes to a <see cref="TransportReceiver"/> on the
+/// transport's own schedule, in transport-owned memory.
+///
+/// The shape is derived from measured transport work rather than taste: any-thread copying writes with
+/// an explicit flush (batching at the caller's natural boundaries was the largest single lever
+/// measured), push delivery (pull adapters over a push transport measured 24-40% overhead), and a
+/// batch-end notification (coalescing responses produced during a delivery burst into one flush
+/// eliminated a measured 3x send amplification).
+///
+/// The transport IS the outbound <see cref="IBufferWriter{T}"/> — there is no separate output object.
+/// Staging is callable from any thread (single logical writer at a time); bytes are owned by the
+/// transport once <see cref="Advance"/> returns; <see cref="Flush"/> hands the staged bytes to the
+/// wire. Passing the transport AS <see cref="IBufferWriter{T}"/> deliberately grants stage-only
+/// access: the holder composes, the owner flushes at its batch boundary.
+/// </summary>
+[Experimental(Experiments.Transport, UrlFormat = Experiments.UrlFormat)]
+public abstract class DuplexTransport : IBufferWriter<byte>, IAsyncDisposable
+{
+    /// <summary>Request writable space to stage outbound bytes (see <see cref="IBufferWriter{T}"/>).</summary>
+    public abstract Memory<byte> GetMemory(int sizeHint = 0);
+
+    /// <inheritdoc cref="GetMemory"/>
+    /// <remarks>Defaults to <c>GetMemory(sizeHint).Span</c>; override when the transport has a cheaper
+    /// span path than a <see cref="Memory{T}"/> round-trip.</remarks>
+    public virtual Span<byte> GetSpan(int sizeHint = 0) => GetMemory(sizeHint).Span;
+
+    /// <summary>Commit <paramref name="count"/> bytes obtained via <see cref="GetMemory"/> or
+    /// <see cref="GetSpan"/>; the transport owns them when this returns, so caller state need not
+    /// survive it.</summary>
+    public abstract void Advance(int count);
+
+    /// <summary>Hand everything staged since the last flush to the wire, as one send where the
+    /// transport allows. Returns false if the transport is closed (staged bytes are dropped).</summary>
+    public abstract bool Flush();
+
+    /// <summary>Begin inbound delivery. Exactly one receiver, set once, before any data is expected;
+    /// delivery runs on the transport's schedule and threads.</summary>
+    public abstract void Start(TransportReceiver receiver);
+
+    public abstract ValueTask DisposeAsync();
+}
+
+/// <summary>
+/// The consumer half of <see cref="DuplexTransport"/>. Callbacks run on the transport's threads and
+/// must be bounded and non-blocking; anything long-running belongs on the consumer's own scheduler.
+/// </summary>
+[Experimental(Experiments.Transport, UrlFormat = Experiments.UrlFormat)]
+public abstract class TransportReceiver
+{
+    /// <summary>Bytes arrived. <paramref name="payload"/> is TRANSPORT-OWNED and valid only for the
+    /// duration of the call — copy anything retained. Return false to request the transport close.</summary>
+    public abstract bool OnReceived(ReadOnlySpan<byte> payload);
+
+    /// <summary>A delivery burst has ended (for loop transports: the event batch is drained). Flush
+    /// anything staged in response to the burst HERE, once, rather than per <see cref="OnReceived"/> —
+    /// per-callback flushing measurably amplifies peer segmentation.</summary>
+    public virtual void OnBatchEnd() { }
+
+    /// <summary>The transport closed; fires exactly once. <paramref name="fault"/> is the failure when
+    /// the transport can attribute one, else null (a clean or unattributed close).</summary>
+    public virtual void OnClosed(Exception? fault) { }
+}

src/StackExchange.Redis/Configuration/Tunnel.cs

@@ -1,5 +1,6 @@
 using System;
 using System.Buffers;
+using System.Diagnostics.CodeAnalysis;
 using System.IO;
 using System.Net;
 using System.Net.Sockets;
@@ -35,6 +36,16 @@ public abstract class Tunnel
         /// </summary>
         public virtual ValueTask<Stream?> BeforeAuthenticateAsync(EndPoint endpoint, ConnectionType connectionType, Socket? socket, CancellationToken cancellationToken) => default;
 
+        /// <summary>
+        /// Optionally supply the ENTIRE transport for this connection — the same hijack as
+        /// <see cref="BeforeAuthenticateAsync"/> one level deeper: instead of yielding a
+        /// <see cref="Stream"/> over a socket the library owns, yield a
+        /// <see cref="RESPite.Transports.DuplexTransport"/> the tunnel owns, and no socket is created at
+        /// all. Return null (the default for every existing tunnel) for the standard socket path.
+        /// </summary>
+        [Experimental(RESPite.Experiments.Transport, UrlFormat = RESPite.Experiments.UrlFormat)]
+        public virtual ValueTask<RESPite.Transports.DuplexTransport?> ConnectTransportAsync(EndPoint endpoint, ConnectionType connectionType, CancellationToken cancellationToken) => default;
+
         private sealed class HttpProxyTunnel : Tunnel
         {
             public EndPoint Proxy { get; }

src/StackExchange.Redis/PublicAPI/PublicAPI.Unshipped.txt

@@ -1 +1,2 @@
 #nullable enable
+[SER009]virtual StackExchange.Redis.Configuration.Tunnel.ConnectTransportAsync(System.Net.EndPoint! endpoint, StackExchange.Redis.ConnectionType connectionType, System.Threading.CancellationToken cancellationToken) -> System.Threading.Tasks.ValueTask<RESPite.Transports.DuplexTransport?>