QUIC prototype

cmd/neofs-node/config.go

@@ -50,6 +50,7 @@ import (
 	"github.com/nspcc-dev/neofs-sdk-go/user"
 	"github.com/nspcc-dev/neofs-sdk-go/version"
 	"github.com/panjf2000/ants/v2"
+	"github.com/quic-go/quic-go"
 	"go.uber.org/zap"
 	"go.uber.org/zap/zapcore"
 	"golang.org/x/term"
@@ -236,6 +237,8 @@ type cfgGRPC struct {
 	listeners []net.Listener
 	servers   []*grpc.Server
 
+	quicListeners []*quic.Listener
+
 	// serviceRegistrators stores functions that register gRPC service
 	// implementations into a gRPC server.
 	serviceRegistrators []func(*grpc.Server)

cmd/neofs-node/object.go

@@ -368,6 +368,8 @@ func initObjectService(c *cfg) {
 	c.cfgGRPC.registerService(func(srv *grpc.Server) {
 		srv.RegisterService(&svcDesc, server)
 	})
+
+	initQUIC(c, server)
 }
 
 type reputationClientConstructor struct {

cmd/neofs-node/quic.go

@@ -0,0 +1,65 @@
+package main
+
+import (
+	"context"
+	"errors"
+
+	"github.com/nspcc-dev/neofs-node/internal/qstream"
+	objectService "github.com/nspcc-dev/neofs-node/pkg/services/object"
+	"github.com/quic-go/quic-go"
+	"go.uber.org/zap"
+)
+
+func initQUIC(c *cfg, server *objectService.Server) {
+	for _, sc := range c.appCfg.GRPC {
+		endpoint := sc.Endpoint
+
+		tlsCfg, err := qstream.ServerTLSConfig()
+		if err != nil {
+			c.log.Error("QUIC: failed to build TLS config", zap.String("endpoint", endpoint), zap.Error(err))
+			continue
+		}
+
+		ln, err := quic.ListenAddr(endpoint, tlsCfg, qstream.Config())
+		if err != nil {
+			c.log.Error("QUIC: failed to listen", zap.String("endpoint", endpoint), zap.Error(err))
+			continue
+		}
+
+		c.cfgGRPC.quicListeners = append(c.cfgGRPC.quicListeners, ln)
+		c.log.Info("QUIC: start listening GET-stream endpoint", zap.String("endpoint", endpoint))
+
+		c.wg.Go(func() {
+			serveQUICListener(c, ln, server)
+		})
+	}
+
+	c.onShutdown(func() {
+		for _, ln := range c.cfgGRPC.quicListeners {
+			_ = ln.Close()
+		}
+	})
+}
+
+func serveQUICListener(c *cfg, ln *quic.Listener, server *objectService.Server) {
+	for {
+		conn, err := ln.Accept(context.Background())
+		if err != nil {
+			if !errors.Is(err, quic.ErrServerClosed) {
+				c.log.Error("QUIC: accept connection failed", zap.Stringer("endpoint", ln.Addr()), zap.Error(err))
+			}
+			return
+		}
+		go serveQUICConn(conn, server)
+	}
+}
+
+func serveQUICConn(conn *quic.Conn, server *objectService.Server) {
+	for {
+		st, err := conn.AcceptStream(context.Background())
+		if err != nil {
+			return
+		}
+		go server.ServeGetStream(st.Context(), st)
+	}
+}

go.mod

@@ -26,6 +26,7 @@ require (
 	github.com/nspcc-dev/tzhash v1.8.4
 	github.com/panjf2000/ants/v2 v2.11.5
 	github.com/prometheus/client_golang v1.23.2
+	github.com/quic-go/quic-go v0.59.1
 	github.com/spf13/cast v1.10.0
 	github.com/spf13/cobra v1.10.2
 	github.com/spf13/pflag v1.0.10

go.sum

@@ -230,6 +230,8 @@ github.com/prometheus/common v0.66.1 h1:h5E0h5/Y8niHc5DlaLlWLArTQI7tMrsfQjHV+d9Z
 github.com/prometheus/common v0.66.1/go.mod h1:gcaUsgf3KfRSwHY4dIMXLPV0K/Wg1oZ8+SbZk/HH/dA=
 github.com/prometheus/procfs v0.16.1 h1:hZ15bTNuirocR6u0JZ6BAHHmwS1p8B4P6MRqxtzMyRg=
 github.com/prometheus/procfs v0.16.1/go.mod h1:teAbpZRB1iIAJYREa1LsoWUXykVXA1KlTmWl8x/U+Is=
+github.com/quic-go/quic-go v0.59.1 h1:0Gmua0HW1Tv7ANR7hUYwRyD0MG5OJfgvYSZasGZzBic=
+github.com/quic-go/quic-go v0.59.1/go.mod h1:upnsH4Ju1YkqpLXC305eW3yDZ4NfnNbmQRCMWS58IKU=
 github.com/rivo/uniseg v0.2.0/go.mod h1:J6wj4VEh+S6ZtnVlnTBMWIodfgj8LQOQFoIToxlJtxc=
 github.com/rivo/uniseg v0.4.7 h1:WUdvkW8uEhrYfLC4ZzdpI2ztxP1I582+49Oc5Mq64VQ=
 github.com/rivo/uniseg v0.4.7/go.mod h1:FN3SvrM+Zdj16jyLfmOkMNblXMcoc8DfTHruCPUcx88=
@@ -294,6 +296,8 @@ go.opentelemetry.io/otel/trace v1.41.0 h1:Vbk2co6bhj8L59ZJ6/xFTskY+tGAbOnCtQGVVa
 go.opentelemetry.io/otel/trace v1.41.0/go.mod h1:U1NU4ULCoxeDKc09yCWdWe+3QoyweJcISEVa1RBzOis=
 go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto=
 go.uber.org/goleak v1.3.0/go.mod h1:CoHD4mav9JJNrW/WLlf7HGZPjdw8EucARQHekz1X6bE=
+go.uber.org/mock v0.5.2 h1:LbtPTcP8A5k9WPXj54PPPbjcI4Y6lhyOZXn+VS7wNko=
+go.uber.org/mock v0.5.2/go.mod h1:wLlUxC2vVTPTaE3UD51E0BGOAElKrILxhVSDYQLld5o=
 go.uber.org/multierr v1.11.0 h1:blXXJkSxSSfBVBlC76pxqeO+LN3aDfLQo+309xJstO0=
 go.uber.org/multierr v1.11.0/go.mod h1:20+QtiLqy0Nd6FdQB9TLXag12DsQkrbs3htMFfDN80Y=
 go.uber.org/zap v1.27.1 h1:08RqriUEv8+ArZRYSTXy1LeBScaMpVSTBhCeaZYfMYc=

internal/qstream/qstream.go

@@ -0,0 +1,77 @@
+package qstream
+
+import (
+	"crypto/ecdsa"
+	"crypto/elliptic"
+	"crypto/rand"
+	"crypto/tls"
+	"crypto/x509"
+	"encoding/pem"
+	"fmt"
+	"math/big"
+	"time"
+
+	"github.com/quic-go/quic-go"
+)
+
+// ALPN is the TLS application protocol negotiated by both ends of the raw GET
+// stream. It must match on the node (listener) and on every client (S3 gateway
+// and node-to-node forwarder).
+const ALPN = "neofs-get-quic"
+
+const (
+	StatusOK    byte = 0
+	StatusError byte = 1
+)
+
+const MaxRequestSize = 64 * 1024
+
+func Config() *quic.Config {
+	return &quic.Config{
+		MaxIncomingStreams: 1 << 16,
+		MaxIdleTimeout:     5 * time.Minute,
+		KeepAlivePeriod:    30 * time.Second,
+	}
+}
+
+// ServerTLSConfig builds a server TLS config with an ephemeral self-signed
+// certificate.
+func ServerTLSConfig() (*tls.Config, error) {
+	key, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader)
+	if err != nil {
+		return nil, fmt.Errorf("generate key: %w", err)
+	}
+	tmpl := &x509.Certificate{
+		SerialNumber: big.NewInt(1),
+		NotBefore:    time.Unix(0, 0),
+		NotAfter:     time.Now().AddDate(10, 0, 0),
+	}
+	der, err := x509.CreateCertificate(rand.Reader, tmpl, tmpl, &key.PublicKey, key)
+	if err != nil {
+		return nil, fmt.Errorf("create certificate: %w", err)
+	}
+	keyDER, err := x509.MarshalPKCS8PrivateKey(key)
+	if err != nil {
+		return nil, fmt.Errorf("marshal key: %w", err)
+	}
+	cert, err := tls.X509KeyPair(
+		pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE", Bytes: der}),
+		pem.EncodeToMemory(&pem.Block{Type: "PRIVATE KEY", Bytes: keyDER}),
+	)
+	if err != nil {
+		return nil, fmt.Errorf("build key pair: %w", err)
+	}
+	return &tls.Config{
+		Certificates: []tls.Certificate{cert},
+		NextProtos:   []string{ALPN},
+		MinVersion:   tls.VersionTLS13,
+	}, nil
+}
+
+func ClientTLSConfig() *tls.Config {
+	return &tls.Config{
+		InsecureSkipVerify: true, //nolint:gosec // prototype: ephemeral self-signed server cert
+		NextProtos:         []string{ALPN},
+		MinVersion:         tls.VersionTLS13,
+	}
+}

pkg/core/client/client.go

@@ -12,6 +12,7 @@ import (
 	oid "github.com/nspcc-dev/neofs-sdk-go/object/id"
 	reputationSDK "github.com/nspcc-dev/neofs-sdk-go/reputation"
 	"github.com/nspcc-dev/neofs-sdk-go/user"
+	"github.com/quic-go/quic-go"
 	"google.golang.org/grpc"
 )
 
@@ -50,4 +51,8 @@ type MultiAddressClient interface {
 	// ForAnyGRPCConn continues. If this happens on all endpoints, ForAnyGRPCConn
 	// returns [ErrAllConnectionsSkipped].
 	ForAnyGRPCConn(ctx context.Context, f func(context.Context, *grpc.ClientConn) error) error
+
+	// ForAnyQUICStream executes op over a QUIC connection to the multi-address
+	// endpoint-by-endpoint until success.
+	ForAnyQUICStream(context.Context, func(context.Context, *quic.Conn, string) error) error
 }

pkg/network/cache/clients.go

@@ -10,20 +10,24 @@ import (
 	"iter"
 	"maps"
 	"slices"
+	"strings"
 	"sync"
 	"time"
 
+	"github.com/nspcc-dev/neofs-node/internal/qstream"
 	"github.com/nspcc-dev/neofs-node/internal/uriutil"
 	clientcore "github.com/nspcc-dev/neofs-node/pkg/core/client"
 	"github.com/nspcc-dev/neofs-node/pkg/network"
 	"github.com/nspcc-dev/neofs-sdk-go/client"
+	apistatus "github.com/nspcc-dev/neofs-sdk-go/client/status"
 	cid "github.com/nspcc-dev/neofs-sdk-go/container/id"
 	neofscrypto "github.com/nspcc-dev/neofs-sdk-go/crypto"
 	"github.com/nspcc-dev/neofs-sdk-go/netmap"
 	"github.com/nspcc-dev/neofs-sdk-go/object"
 	oid "github.com/nspcc-dev/neofs-sdk-go/object/id"
 	apireputation "github.com/nspcc-dev/neofs-sdk-go/reputation"
 	"github.com/nspcc-dev/neofs-sdk-go/user"
+	"github.com/quic-go/quic-go"
 	"go.uber.org/zap"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/backoff"
@@ -197,6 +201,7 @@ func (x *Clients) initConnections(ctx context.Context, pub []byte, addrs iter.Se
 		log:    x.log.With(zap.String("SN public key", hexKey)),
 		nodeID: hexKey,
 		m:      m,
+		mq:     make(map[string]*quic.Conn),
 	}, nil
 }
 
@@ -261,6 +266,7 @@ type connections struct {
 
 	mtx sync.RWMutex
 	m   map[string]*client.Client // keys are multiaddrs
+	mq  map[string]*quic.Conn     // keys are multiaddrs; lazily dialed for the raw GET-over-QUIC prototype
 }
 
 func (x *connections) closeAll() {
@@ -271,6 +277,95 @@ func (x *connections) closeAll() {
 		}
 		x.log.Info("connection successfully closed", zap.String("address", ma))
 	}
+	x.mtx.Lock()
+	for _, qc := range x.mq {
+		_ = qc.CloseWithError(0, "shutdown")
+	}
+	x.mq = make(map[string]*quic.Conn)
+	x.mtx.Unlock()
+}
+
+func (x *connections) quicConn(ctx context.Context, ma string) (*quic.Conn, error) {
+	x.mtx.RLock()
+	qc := x.mq[ma]
+	x.mtx.RUnlock()
+	if qc != nil && qc.Context().Err() == nil {
+		return qc, nil
+	}
+
+	var a network.Address
+	if err := a.FromString(ma); err != nil {
+		return nil, fmt.Errorf("parse network address %q: %w", ma, err)
+	}
+	addr := a.URIAddr()
+	if i := strings.Index(addr, "://"); i >= 0 {
+		addr = addr[i+3:]
+	}
+
+	conn, err := quic.DialAddr(ctx, addr, qstream.ClientTLSConfig(), qstream.Config())
+	if err != nil {
+		return nil, fmt.Errorf("dial QUIC %q: %w", addr, err)
+	}
+
+	x.mtx.Lock()
+	if existing := x.mq[ma]; existing != nil && existing.Context().Err() == nil {
+		x.mtx.Unlock()
+		_ = conn.CloseWithError(0, "duplicate")
+		return existing, nil
+	}
+	x.mq[ma] = conn
+	x.mtx.Unlock()
+	return conn, nil
+}
+
+// dropQUICConn removes and closes the cached QUIC connection for the given
+// address so the next use re-dials it.
+func (x *connections) dropQUICConn(ma string) {
+	x.mtx.Lock()
+	if c := x.mq[ma]; c != nil {
+		delete(x.mq, ma)
+		_ = c.CloseWithError(0, "drop on error")
+	}
+	x.mtx.Unlock()
+}
+
+// quicForwardAttempts bounds how many times a QUIC forward to one endpoint is
+// retried (with a fresh dial) on transport failures before moving on.
+const quicForwardAttempts = 3
+
+func (x *connections) ForAnyQUICStream(ctx context.Context, f func(context.Context, *quic.Conn, string) error) error {
+	var firstErr error
+	x.mtx.RLock()
+	addrs := slices.Collect(maps.Keys(x.m))
+	x.mtx.RUnlock()
+
+	for _, ma := range addrs {
+		var err error
+		for range quicForwardAttempts {
+			var conn *quic.Conn
+			conn, err = x.quicConn(ctx, ma)
+			if err == nil {
+				err = f(ctx, conn, ma)
+				if err == nil {
+					return nil
+				}
+			}
+			if errors.Is(err, apistatus.Error) {
+				break
+			}
+			x.dropQUICConn(ma)
+			if ctx.Err() != nil {
+				break
+			}
+		}
+		if errors.Is(err, apistatus.Error) && !errors.Is(err, apistatus.ErrObjectNotFound) {
+			return newEndpointError(ma, err)
+		}
+		if firstErr == nil {
+			firstErr = newEndpointError(ma, err)
+		}
+	}
+	return newMultiEndpointError(x.nodeID, firstErr)
 }
 
 func (x *connections) all(f func(ma string, c *client.Client) bool) {

pkg/services/object/get/remote.go

@@ -29,9 +29,13 @@ func (exec *execCtx) processNode(info netmap.NodeInfo) bool {
 	switch {
 	default:
 		exec.status = statusUndefined
-		exec.err = apistatus.ErrObjectNotFound
+		exec.err = err
 
-		exec.log.Debug("remote call failed", nodeLog, zap.Error(err))
+		if errors.Is(err, apistatus.ErrObjectNotFound) {
+			exec.log.Debug("remote node has no object", nodeLog)
+		} else {
+			exec.log.Warn("remote call failed", nodeLog, zap.Error(err))
+		}
 	case err == nil:
 		exec.status = statusOK
 		exec.err = nil

pkg/services/object/put/service_test.go

@@ -51,6 +51,7 @@ import (
 	"github.com/nspcc-dev/neofs-sdk-go/version"
 	"github.com/nspcc-dev/tzhash/tz"
 	"github.com/panjf2000/ants/v2"
+	"github.com/quic-go/quic-go"
 	"github.com/stretchr/testify/require"
 	"go.uber.org/zap"
 	"go.uber.org/zap/zaptest"
@@ -1125,6 +1126,10 @@ func (m *serviceClient) ForAnyGRPCConn(context.Context, func(context.Context, *g
 	panic("unimplemented")
 }
 
+func (m *serviceClient) ForAnyQUICStream(context.Context, func(context.Context, *quic.Conn, string) error) error {
+	panic("unimplemented")
+}
+
 type testPayloadStream Streamer
 
 func (x *testPayloadStream) Write(p []byte) (int, error) {

pkg/services/object/server.go

@@ -858,6 +858,7 @@ func (s *Server) sendStatusGetResponse(stream protoobject.ObjectService_GetServe
 
 type getStream struct {
 	base    protoobject.ObjectService_GetServer
+	w       io.Writer // set instead of base for the raw QUIC GET stream
 	srv     *Server
 	reqInfo aclsvc.RequestInfo