Fix webtransport conn not being served due to missing EnableStreamResetPartialDelivery

.changeset/enable_streamresetpartialdelivery_when_listening_for_quic_connections.md

@@ -0,0 +1,5 @@
+---
+default: patch
+---
+
+# Enable StreamResetPartialDelivery when listening for QUIC connections.

rhp/v4/quic/quic.go

@@ -130,10 +130,11 @@ func Dial(ctx context.Context, addr string, peerKey types.PublicKey, opts ...Cli
 		NextProtos: []string{TLSNextProtoRHP4},
 	}
 	qc := &quic.Config{
-		EnableDatagrams:      true,
-		HandshakeIdleTimeout: 15 * time.Second,
-		KeepAlivePeriod:      30 * time.Second,
-		MaxIdleTimeout:       30 * time.Minute,
+		EnableDatagrams:                  true,
+		HandshakeIdleTimeout:             15 * time.Second,
+		KeepAlivePeriod:                  30 * time.Second,
+		MaxIdleTimeout:                   30 * time.Minute,
+		EnableStreamResetPartialDelivery: true,
 	}
 	cc := &clientConfig{
 		streamMiddleware: func(nc net.Conn) net.Conn { return nc },
@@ -224,10 +225,11 @@ func Listen(conn net.PacketConn, certs CertManager) (*quic.Listener, error) {
 		GetCertificate: certs.GetCertificate,
 		NextProtos:     []string{TLSNextProtoRHP4, http3.NextProtoH3},
 	}, &quic.Config{
-		EnableDatagrams:    true,
-		KeepAlivePeriod:    30 * time.Second,
-		MaxIdleTimeout:     30 * time.Minute,
-		MaxIncomingStreams: maxIncomingStreams,
+		EnableDatagrams:                  true,
+		KeepAlivePeriod:                  30 * time.Second,
+		MaxIdleTimeout:                   30 * time.Minute,
+		MaxIncomingStreams:               maxIncomingStreams,
+		EnableStreamResetPartialDelivery: true,
 	})
 }
 
@@ -279,6 +281,9 @@ func Serve(l *quic.Listener, s *rhp4.Server, opts ...ServeOption) {
 	}
 	defer wts.Close()
 
+	// configure the HTTP/3 server for WebTransport (enables extended CONNECT, etc.)
+	webtransport.ConfigureHTTP3Server(wts.H3)
+
 	mux.HandleFunc("/sia/rhp/v4", func(w http.ResponseWriter, r *http.Request) {
 		sess, err := wts.Upgrade(w, r)
 		if err != nil {
@@ -311,13 +316,15 @@ func Serve(l *quic.Listener, s *rhp4.Server, opts ...ServeOption) {
 		case http3.NextProtoH3: // webtransport
 			go func() {
 				defer conn.CloseWithError(0, "")
-				wts.ServeQUICConn(conn)
+				if err := wts.ServeQUICConn(conn); err != nil {
+					log.Debug("failed to serve webtransport connection", zap.Error(err))
+				}
 			}()
 		case TLSNextProtoRHP4: // quic
 			go func() {
 				defer conn.CloseWithError(0, "")
 				if err := s.Serve(&transport{qc: conn, streamMiddleware: o.streamMiddleware}, log); err != nil {
-					log.Debug("failed to serve connection", zap.Error(err))
+					log.Debug("failed to serve quic connection", zap.Error(err))
 				}
 			}()
 		default:

rhp/v4/rpc_test.go

@@ -4,6 +4,7 @@ import (
 	"bytes"
 	"context"
 	"crypto/tls"
+	"fmt"
 	"math"
 	"net"
 	"reflect"
@@ -13,6 +14,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/quic-go/webtransport-go"
 	"go.sia.tech/core/consensus"
 	proto4 "go.sia.tech/core/rhp/v4"
 	"go.sia.tech/core/types"
@@ -68,6 +70,50 @@ func (fs *fundAndSign) Address() types.Address {
 	return fs.w.Address()
 }
 
+// webTransportClient wraps a WebTransport session to implement rhp4.TransportClient
+type webTransportClient struct {
+	sess    *webtransport.Session
+	peerKey types.PublicKey
+}
+
+func (c *webTransportClient) Close() error {
+	return c.sess.CloseWithError(0, "")
+}
+
+func (c *webTransportClient) PeerKey() types.PublicKey {
+	return c.peerKey
+}
+
+func (c *webTransportClient) FrameSize() int {
+	return 1440 * 3
+}
+
+func (c *webTransportClient) DialStream(ctx context.Context) (net.Conn, error) {
+	stream, err := c.sess.OpenStreamSync(ctx)
+	if err != nil {
+		return nil, err
+	}
+	return &webTransportStream{
+		Stream:     stream,
+		localAddr:  c.sess.LocalAddr(),
+		remoteAddr: c.sess.RemoteAddr(),
+	}, nil
+}
+
+// webTransportStream wraps a WebTransport stream to implement net.Conn
+type webTransportStream struct {
+	*webtransport.Stream
+	localAddr, remoteAddr net.Addr
+}
+
+func (s *webTransportStream) LocalAddr() net.Addr {
+	return s.localAddr
+}
+
+func (s *webTransportStream) RemoteAddr() net.Addr {
+	return s.remoteAddr
+}
+
 func testRenterHostPairSiaMux(tb testing.TB, hostKey types.PrivateKey, cm rhp4.ChainManager, w rhp4.Wallet, c rhp4.Contractor, sr rhp4.Settings, ss rhp4.Sectors, log *zap.Logger) rhp4.TransportClient {
 	rs := rhp4.NewServer(hostKey, cm, c, w, sr, ss, rhp4.WithPriceTableValidity(2*time.Minute))
 	hostAddr := testutil.ServeSiaMux(tb, rs, log.Named("siamux"))
@@ -95,6 +141,27 @@ func testRenterHostPairQUIC(tb testing.TB, hostKey types.PrivateKey, cm rhp4.Cha
 	return transport
 }
 
+func testRenterHostPairWebTransport(tb testing.TB, hostKey types.PrivateKey, cm rhp4.ChainManager, w rhp4.Wallet, c rhp4.Contractor, sr rhp4.Settings, ss rhp4.Sectors, log *zap.Logger) rhp4.TransportClient {
+	rs := rhp4.NewServer(hostKey, cm, c, w, sr, ss, rhp4.WithPriceTableValidity(2*time.Minute))
+	hostAddr := testutil.ServeQUIC(tb, rs, quic.WithServeLogger(log.Named("webtransport")))
+
+	dialer := webtransport.Dialer{
+		TLSClientConfig: &tls.Config{
+			InsecureSkipVerify: true,
+		},
+	}
+	tb.Cleanup(func() { dialer.Close() })
+
+	url := fmt.Sprintf("https://%s/sia/rhp/v4", hostAddr)
+	_, sess, err := dialer.Dial(context.Background(), url, nil)
+	if err != nil {
+		tb.Fatal(err)
+	}
+	tb.Cleanup(func() { sess.CloseWithError(0, "") })
+
+	return &webTransportClient{sess: sess, peerKey: hostKey.PublicKey()}
+}
+
 func startTestNode(tb testing.TB, n *consensus.Network, genesis types.Block) (*chain.Manager, *wallet.SingleAddressWallet) {
 	db, tipstate, err := chain.NewDBStore(chain.NewMemDB(), n, genesis, nil)
 	if err != nil {
@@ -268,6 +335,11 @@ func TestFormContract(t *testing.T) {
 		transport := testRenterHostPairQUIC(t, hostKey, cm, w, c, sr, ss, zap.NewNop())
 		testFormContract(t, transport)
 	})
+
+	t.Run("webtransport", func(t *testing.T) {
+		transport := testRenterHostPairWebTransport(t, hostKey, cm, w, c, sr, ss, zap.NewNop())
+		testFormContract(t, transport)
+	})
 }
 
 func TestFormContractBasis(t *testing.T) {