drpcquic: add QUIC transport for drpc using quic-go

cmd/quicdemo/main.go

@@ -0,0 +1,89 @@
+package main
+
+import (
+	"context"
+	"crypto/ecdsa"
+	"crypto/elliptic"
+	"crypto/rand"
+	"crypto/tls"
+	"crypto/x509"
+	"crypto/x509/pkix"
+	"fmt"
+	"log"
+	"math/big"
+	"sync"
+	"time"
+
+	"storj.io/drpc"
+	"storj.io/drpc/drpcconn"
+	"storj.io/drpc/drpcquic"
+	"storj.io/drpc/drpcserver"
+)
+
+// trivial string message + encoding (stands in for protobuf)
+type msg struct{ S string }
+type enc struct{}
+
+func (enc) Marshal(m drpc.Message) ([]byte, error)   { return []byte(m.(*msg).S), nil }
+func (enc) Unmarshal(b []byte, m drpc.Message) error { m.(*msg).S = string(b); return nil }
+
+// a handler that echoes the request
+type echo struct{}
+
+func (echo) HandleRPC(stream drpc.Stream, rpc string) error {
+	in := new(msg)
+	if err := stream.MsgRecv(in, enc{}); err != nil {
+		return err
+	}
+	return stream.MsgSend(&msg{S: "echo:" + in.S}, enc{})
+}
+
+func devTLS() (server, client *tls.Config) {
+	key, _ := ecdsa.GenerateKey(elliptic.P256(), rand.Reader)
+	tmpl := &x509.Certificate{
+		SerialNumber: big.NewInt(1),
+		Subject:      pkix.Name{CommonName: "demo"},
+		NotBefore:    time.Now().Add(-time.Hour),
+		NotAfter:     time.Now().Add(time.Hour),
+		DNSNames:     []string{"localhost"},
+	}
+	der, _ := x509.CreateCertificate(rand.Reader, tmpl, tmpl, &key.PublicKey, key)
+	cert := tls.Certificate{Certificate: [][]byte{der}, PrivateKey: key}
+	server = &tls.Config{Certificates: []tls.Certificate{cert}, NextProtos: []string{drpcquic.ALPN}}
+	client = &tls.Config{InsecureSkipVerify: true, NextProtos: []string{drpcquic.ALPN}} //nolint:gosec
+	return
+}
+
+func main() {
+	ctx := context.Background()
+	serverTLS, clientTLS := devTLS()
+
+	ln, err := drpcquic.Listen("127.0.0.1:0", serverTLS, drpcquic.Options{})
+	if err != nil {
+		log.Fatal(err)
+	}
+	go func() { _ = drpcquic.Serve(ctx, ln, drpcserver.New(echo{}), drpcquic.Options{}) }()
+
+	mt, err := drpcquic.Dial(ctx, ln.Addr().String(), clientTLS, drpcquic.Options{})
+	if err != nil {
+		log.Fatal(err)
+	}
+	conn := drpcconn.NewFromMultiplexed(mt, drpcconn.Options{})
+	defer conn.Close()
+
+	var wg sync.WaitGroup
+	for i := range 5 { // 5 RPCs concurrently, each on its own QUIC stream
+		wg.Add(1)
+		go func(i int) {
+			defer wg.Done()
+			out := new(msg)
+			if err := conn.Invoke(ctx, "/echo", enc{}, &msg{S: fmt.Sprintf("req-%d", i)}, out); err != nil {
+				log.Printf("RPC %d error: %v", i, err)
+				return
+			}
+			fmt.Printf("RPC %d -> %s\n", i, out.S)
+		}(i)
+	}
+	wg.Wait()
+	fmt.Println("done — drpc is running over QUIC")
+}

drpc.go

@@ -37,6 +37,16 @@ type Transport interface {
 	io.Closer
 }
 
+// MultiplexedTransport represents a connection that can open and accept
+// independent bidirectional streams (e.g. a QUIC connection). Each returned
+// Transport is a single ordered byte stream within the connection.
+type MultiplexedTransport interface {
+	OpenStream(ctx context.Context) (Transport, error)
+	AcceptStream(ctx context.Context) (Transport, error)
+	Close() error
+	Closed() <-chan struct{}
+}
+
 // Message is a protobuf message. It is expected to be used with an Encoding.
 // This exists so that one can use whatever protobuf library/runtime they want.
 type Message interface{}

drpcconn/conn.go

@@ -39,13 +39,28 @@ type Options struct {
 	Metrics drpcmetrics.ClientMetrics
 }
 
+// manager is the subset of drpcmanager behavior that Conn needs. It is
+// satisfied by both the single-stream TCP *drpcmanager.Manager and the
+// multiplexed *drpcmanager.QUICManager.
+type manager interface {
+	NewClientStream(ctx context.Context, rpc string) (*drpcstream.Stream, error)
+	Close() error
+	Closed() <-chan struct{}
+	Unblocked() <-chan struct{}
+}
+
 // Conn is a drpc client connection.
 type Conn struct {
 	tr   drpc.Transport
-	man  *drpcmanager.Manager
+	man  manager
 	mu   sync.Mutex
 	wbuf []byte
 
+	// concurrent is true for connections over a MultiplexedTransport (e.g.
+	// QUIC), where each RPC gets its own stream and so concurrent Invoke and
+	// NewStream calls are allowed.
+	concurrent bool
+
 	stats map[string]*drpcstats.Stats // TODO (server): deprecate
 }
 
@@ -82,6 +97,24 @@ func NewWithOptions(tr drpc.Transport, opts Options) *Conn {
 	return c
 }
 
+// NewFromMultiplexed returns a conn that runs over a MultiplexedTransport (e.g.
+// a QUIC connection), opening a new stream per RPC. Unlike New/NewWithOptions,
+// concurrent Invoke and NewStream calls are allowed: each maps to its own
+// independent stream.
+func NewFromMultiplexed(mt drpc.MultiplexedTransport, opts Options) *Conn {
+	c := &Conn{concurrent: true}
+
+	// TODO: (server): deprecate
+	if opts.CollectStats {
+		drpcopts.SetManagerStatsCB(&opts.Manager.Internal, c.getStats)
+		c.stats = make(map[string]*drpcstats.Stats)
+	}
+
+	c.man = drpcmanager.NewQUIC(mt, opts.Manager)
+
+	return c
+}
+
 // Stats returns the collected stats grouped by rpc.
 func (c *Conn) Stats() map[string]drpcstats.Stats {
 	c.mu.Lock()
@@ -136,11 +169,32 @@ func (c *Conn) Invoke(ctx context.Context, rpc string, enc drpc.Encoding, in, ou
 	if err != nil {
 		return err
 	}
-	defer func() { err = errs.Combine(err, stream.Close()) }()
+	defer func() {
+		cerr := stream.Close()
+		// Over a multiplexed transport the per-RPC stream's Close is cleanup:
+		// the peer may have already torn its side down (sending STOP_SENDING),
+		// making a final KindClose write fail benignly. Don't let that override
+		// a completed RPC. The single-stream (TCP) path keeps the old behavior.
+		if !c.concurrent {
+			err = errs.Combine(err, cerr)
+		}
+	}()
+
+	if c.concurrent {
+		// Multiplexed transport: each RPC has its own stream, so marshal into a
+		// per-call buffer and do not serialize on c.mu. This is what allows many
+		// unary Invokes to run concurrently, each on its own stream.
+		data, merr := drpcenc.MarshalAppend(in, enc, nil)
+		if merr != nil {
+			return merr
+		}
+		return c.doInvoke(stream, enc, rpc, data, metadata, out)
+	}
 
-	// we have to protect c.wbuf here even though the manager only allows one
-	// stream at a time because the stream may async close allowing another
-	// concurrent call to Invoke to proceed.
+	// Single-stream transport (TCP): reuse c.wbuf under c.mu. We have to protect
+	// c.wbuf here even though the manager only allows one stream at a time
+	// because the stream may async close allowing another concurrent call to
+	// Invoke to proceed.
 	c.mu.Lock()
 	defer c.mu.Unlock()
 

drpcmanager/quic_manager.go

@@ -0,0 +1,158 @@
+// Copyright (C) 2026 Storj Labs, Inc.
+// See LICENSE for copying information.
+
+package drpcmanager
+
+import (
+	"context"
+	"time"
+
+	grpcmetadata "google.golang.org/grpc/metadata"
+
+	"storj.io/drpc"
+	"storj.io/drpc/drpcmetadata"
+	"storj.io/drpc/drpcstream"
+	"storj.io/drpc/drpcwire"
+	"storj.io/drpc/internal/drpcopts"
+)
+
+// QUICManager manages a drpc.MultiplexedTransport (e.g. a QUIC connection),
+// mapping each drpc logical stream onto its own native stream. Unlike Manager,
+// there is no shared reader, no semaphore, no streamBuffer, and no pkts/pdone
+// signaling: each stream owns its underlying stream and reads it directly via
+// drpcstream.NewForReader.
+type QUICManager struct {
+	mt   drpc.MultiplexedTransport
+	opts Options
+}
+
+// NewQUIC returns a new QUICManager over the multiplexed transport.
+func NewQUIC(mt drpc.MultiplexedTransport, opts Options) *QUICManager {
+	return &QUICManager{mt: mt, opts: opts}
+}
+
+// Close closes the multiplexed transport, which closes all of its streams.
+func (m *QUICManager) Close() error { return m.mt.Close() }
+
+// Closed returns a channel closed when the multiplexed transport is closed.
+func (m *QUICManager) Closed() <-chan struct{} { return m.mt.Closed() }
+
+// Unblocked never blocks in multiplexed mode: there is no single-stream
+// semaphore, so a new stream can always be created. It returns an already
+// closed channel to satisfy the manager interface used by drpcconn.Conn.
+func (m *QUICManager) Unblocked() <-chan struct{} { return closedCh }
+
+// streamOpts builds per-stream options for a stream on transport tr. It copies
+// m.opts.Stream so concurrent NewClientStream calls do not race on shared
+// option state. The fin channel is deliberately left unset: in multiplexed mode
+// the stream is not coordinated through a manager semaphore, and checkFinished
+// nil-guards the fin channel.
+func (m *QUICManager) streamOpts(kind drpc.StreamKind, rpc string, tr drpc.Transport) drpcstream.Options {
+	opts := m.opts.Stream
+	drpcopts.SetStreamKind(&opts.Internal, kind)
+	drpcopts.SetStreamRPC(&opts.Internal, rpc)
+	drpcopts.SetStreamTransport(&opts.Internal, tr)
+	if cb := drpcopts.GetManagerStatsCB(&m.opts.Internal); cb != nil {
+		drpcopts.SetStreamStats(&opts.Internal, cb(rpc))
+	}
+	return opts
+}
+
+// NewClientStream opens a new outbound stream and wraps it in a drpc stream with
+// its own Reader/Writer and read loop. The caller (drpcconn) writes the invoke
+// sequence on the returned stream.
+func (m *QUICManager) NewClientStream(ctx context.Context, rpc string) (*drpcstream.Stream, error) {
+	tr, err := m.mt.OpenStream(ctx)
+	if err != nil {
+		return nil, err
+	}
+	rd := drpcwire.NewReaderWithOptions(tr, m.opts.Reader)
+	wr := drpcwire.NewWriter(tr, m.opts.WriterBufferSize)
+	// sid MUST be 1: the reader starts expecting ID{Stream:1, Message:1}.
+	return drpcstream.NewForReader(ctx, 1, tr, rd, wr, m.streamOpts(drpc.StreamKindClient, rpc, tr)), nil
+}
+
+// AcceptTransport accepts the next inbound stream and returns its raw transport
+// WITHOUT reading the invoke. Hand the transport to ServerStream from a
+// per-stream goroutine so that reading one stream's invoke never blocks
+// accepting (or serving) the next stream on the same connection. Doing the
+// invoke read here (as a combined accept+read) would serialize the whole
+// connection behind whichever stream is slowest to send its invoke — exactly
+// the head-of-line blocking running over QUIC is meant to avoid.
+func (m *QUICManager) AcceptTransport(ctx context.Context) (drpc.Transport, error) {
+	return m.mt.AcceptStream(ctx)
+}
+
+// ServerStream reads the invoke (and any preceding metadata) packets off an
+// already-accepted transport tr, then hands the SAME reader to the stream's read
+// loop so no buffered bytes are lost and no second reader races the first. On any
+// error it closes tr, so a failed or slow stream leaks nothing and never affects
+// other streams on the connection.
+func (m *QUICManager) ServerStream(ctx context.Context, tr drpc.Transport) (stream *drpcstream.Stream, rpc string, err error) {
+	defer func() {
+		if err != nil {
+			_ = tr.Close() // don't leak the stream on a failed parse
+		}
+	}()
+
+	// Optionally bound how long we wait for the client's invoke.
+	if to := m.opts.InactivityTimeout; to > 0 {
+		if d, ok := tr.(interface{ SetReadDeadline(time.Time) error }); ok {
+			_ = d.SetReadDeadline(time.Now().Add(to))
+			defer func() { _ = d.SetReadDeadline(time.Time{}) }()
+		}
+	}
+
+	rd := drpcwire.NewReaderWithOptions(tr, m.opts.Reader)
+
+	var meta map[string]string
+	var metaID uint64
+	for {
+		pkt, perr := rd.ReadPacketUsing(nil)
+		if perr != nil {
+			return nil, "", perr
+		}
+		switch pkt.Kind {
+		case drpcwire.KindInvokeMetadata:
+			meta, err = drpcmetadata.Decode(pkt.Data)
+			if err != nil {
+				return nil, "", err
+			}
+			metaID = pkt.ID.Stream
+
+		case drpcwire.KindInvoke:
+			rpc = string(pkt.Data)
+			if metaID == pkt.ID.Stream {
+				if m.opts.GRPCMetadataCompatMode {
+					grpcMeta := make(map[string][]string, len(meta))
+					for k, v := range meta {
+						grpcMeta[k] = []string{v}
+					}
+					ctx = grpcmetadata.NewIncomingContext(ctx, grpcMeta)
+				} else {
+					ctx = drpcmetadata.NewIncomingContext(ctx, meta)
+				}
+			}
+			wr := drpcwire.NewWriter(tr, m.opts.WriterBufferSize)
+			stream = drpcstream.NewForReader(ctx, pkt.ID.Stream, tr, rd, wr,
+				m.streamOpts(drpc.StreamKindServer, rpc, tr))
+			return stream, rpc, nil
+
+		default:
+			return nil, "", drpc.ProtocolError.New("expected invoke, got %s", pkt.Kind)
+		}
+	}
+}
+
+// NewServerStream accepts a new inbound stream and reads its invoke. It is
+// AcceptTransport followed by ServerStream. Prefer the split form when serving
+// many concurrent streams (AcceptTransport in the accept loop, ServerStream in a
+// per-stream goroutine) so the invoke read does not serialize accepts; this
+// combined form remains for single-stream callers and tests.
+func (m *QUICManager) NewServerStream(ctx context.Context) (stream *drpcstream.Stream, rpc string, err error) {
+	tr, err := m.mt.AcceptStream(ctx)
+	if err != nil {
+		return nil, "", err
+	}
+	return m.ServerStream(ctx, tr)
+}