feat: upgrade retained native connections to direct UDP

docs/maintainers/transport/peer-packets.md

@@ -152,6 +152,22 @@ and eligible local host candidates come from that mux. closing an ICE agent
 leaves sibling agents and relay QUIC alive. library logging is disabled; role
 owners report bounded outcomes through the existing observability boundary.
 
+an opt-in native publisher run shares one socket across its two relay connections
+and peer agents. the connector shares its socket across both relay attempts and
+its peer agent. after the inner QUIC connection starts over relay delivery,
+[connectivity preparation](../../../internal/transport/peer/connectivity.go)
+exchanges one bounded message on a dedicated unidirectional stream each way.
+the exact pair and delivery bind at most 32 IP-only UDP candidates and short-lived
+ICE credentials. neither endpoint accepts source authority, application permission,
+or a serving deadline from that message.
+
+gathering and consent run outside application admission. successful ICE consent
+installs a direct carrier; fresh packet MAC probes must succeed before selection.
+consent loss removes that carrier while the inner QUIC connection retains its
+streams over relay delivery. cancellation interrupts partial signaling and joins
+the agent before releasing the shared socket. this initial negotiation does not
+yet restart ICE after network changes or a failed setup.
+
 relay and standalone answer STUN Binding requests on their existing QUIC socket.
 the server bounds request size and response rate and ignores ICE-authenticated
 requests intended for a peer. its operational counter has fixed result labels.
@@ -326,12 +342,15 @@ and its monotonic application counter intact. the reconnect case replaces each
 TCP carrier, then closes both replacements together and requires recovery on
 the same target. this fixture does not qualify
 load-balancer source handling or assignment replacement.
+relay-fault cases explicitly disable direct delivery. a separate direct case
+holds ICE until the first target reply, then upgrades and serves through a loss
+of both relay UDP paths. it includes the default publisher socket-sharing dialer.
 the renewal case also retains that target past the original pair deadline,
 checks that both durable leg claims advanced, then applies the carrier faults.
 
 production publishers and connectors do not enable `NativePeers` yet, and
 production relays leave native admission disabled. a relay hello receipt proves
 its own checked claim, not publisher installation or application permission.
 pair renewal does not bypass the separate checked relay renewal. assignment/source
-rebinding, deployed listener source handling, and ICE integration remain required
+rebinding, deployed listener source handling, and ICE restart remain required
 before enabling the native path.

internal/connector/stream/connect.go

@@ -17,6 +17,7 @@ import (
 	"github.com/tnldotdev/tnl/internal/naming/opaqueid"
 	"github.com/tnldotdev/tnl/internal/observability"
 	"github.com/tnldotdev/tnl/internal/transport/muxsession"
+	"github.com/tnldotdev/tnl/internal/transport/peer"
 	"github.com/tnldotdev/tnl/internal/transport/streamcopy"
 	"github.com/tnldotdev/tnl/internal/transport/targettransport"
 	"github.com/tnldotdev/tnl/internal/transport/tunnel"
@@ -90,7 +91,8 @@ type Config struct {
 	CleanupContext func() context.Context `json:"-"`
 	// NativePeers requires prepared relay packet delivery. it never falls back
 	// to ingress; production owners leave it off until rollout qualification.
-	NativePeers bool
+	NativePeers      bool
+	PeerConnectivity peer.ConnectivityOptions
 }
 
 func (Config) String() string   { return "private connector configuration" }

internal/connector/stream/peer_connection.go

@@ -27,6 +27,7 @@ type peerConnection struct {
 	socket    *peer.Socket
 	legs      [2]*peerRelayLeg
 	paths     [2]*peer.CarrierSlot
+	direct    *peer.CarrierSlot
 	packets   *peer.Endpoint
 	transport *quic.Transport
 	session   muxsession.Session
@@ -130,7 +131,8 @@ func preparePeerConnection(ctx context.Context, config Config, prepared *prepare
 	var packetKey [32]byte
 	copy(packetKey[:], key)
 	clear(key)
-	paths := [3]peer.Carrier{p.paths[0], p.paths[1]}
+	p.direct = peer.NewCarrierSlot()
+	paths := [3]peer.Carrier{p.paths[0], p.paths[1], p.direct}
 	p.packets, err = peer.New(life, peer.Config{Kind: peer.NativePeer, Role: 0, Key: packetKey, Paths: paths})
 	clear(packetKey[:])
 	if err != nil {
@@ -142,7 +144,7 @@ func preparePeerConnection(ctx context.Context, config Config, prepared *prepare
 	connection, err := p.transport.Dial(startup, p.packets.RemoteAddr(), verification, &quic.Config{
 		InitialPacketSize: tunnelv1.MaxPeerPacketBytes, DisablePathMTUDiscovery: true,
 		MaxIdleTimeout: 30 * time.Second, KeepAlivePeriod: 10 * time.Second,
-		MaxIncomingStreams: -1, MaxIncomingUniStreams: -1,
+		MaxIncomingStreams: -1, MaxIncomingUniStreams: 1,
 	})
 	if err != nil {
 		return nil, err
@@ -151,6 +153,12 @@ func preparePeerConnection(ctx context.Context, config Config, prepared *prepare
 	if err != nil {
 		return nil, err
 	}
+	p.renewals.Go(func() {
+		_ = peer.ServeConnectivity(life, p.socket, connection, p.packets, p.direct, peer.ConnectivityBinding{
+			PairID: initial.pair.PairId, DeliveryID: initial.pair.DeliveryId, Generation: initial.pair.DeliveryGeneration,
+			RelayAddresses: [2]string{initial.pair.Relays[0].RelayAddress, initial.pair.Relays[1].RelayAddress},
+		}, true, config.PeerConnectivity)
+	})
 	go p.watch()
 	return p, nil
 }
@@ -204,6 +212,9 @@ func (p *peerConnection) closeResources() {
 		}
 	}
 	if p.socket != nil {
+		if p.direct != nil {
+			_ = p.direct.Close()
+		}
 		_ = p.socket.Close()
 	}
 }

internal/daemon/peer_packets_integration_test.go

@@ -15,6 +15,7 @@ import (
 	"github.com/tnldotdev/tnl/internal/connector/workload"
 	"github.com/tnldotdev/tnl/internal/testutil"
 	"github.com/tnldotdev/tnl/internal/testutil/githubtest"
+	"github.com/tnldotdev/tnl/internal/transport/peer"
 	controlv1 "github.com/tnldotdev/tnl/pkg/api/control/public/v1"
 )
 
@@ -38,7 +39,16 @@ func startNativePacketFaultSocket(t *testing.T, address, target string) *atomic.
 	done := make(chan struct{})
 	go func() {
 		defer close(done)
-		var client *net.UDPAddr
+		// preserve each sender's UDP mapping. the default publisher and connector
+		// both use this advertised address; a last-sender proxy mixes their QUIC.
+		upstreams := make(map[string]*net.UDPConn)
+		var readers sync.WaitGroup
+		defer func() {
+			for _, upstream := range upstreams {
+				_ = upstream.Close()
+			}
+			readers.Wait()
+		}()
 		packet := make([]byte, 64*1024)
 		for {
 			n, source, err := socket.ReadFromUDP(packet)
@@ -48,14 +58,30 @@ func startNativePacketFaultSocket(t *testing.T, address, target string) *atomic.
 			if drop.Load() {
 				continue
 			}
-			if source.String() == remote.String() {
-				if client != nil {
-					_, _ = socket.WriteToUDP(packet[:n], client)
+			upstream := upstreams[source.String()]
+			if upstream == nil {
+				if len(upstreams) == 64 {
+					continue
+				}
+				upstream, err = net.DialUDP("udp4", &net.UDPAddr{IP: net.IPv4(127, 0, 0, 1)}, remote)
+				if err != nil {
+					return
 				}
-			} else {
-				client = source
-				_, _ = socket.WriteToUDP(packet[:n], remote)
+				upstreams[source.String()] = upstream
+				readers.Go(func() {
+					response := make([]byte, 64*1024)
+					for {
+						n, err := upstream.Read(response)
+						if err != nil {
+							return
+						}
+						if !drop.Load() {
+							_, _ = socket.WriteToUDP(response[:n], source)
+						}
+					}
+				})
 			}
+			_, _ = upstream.Write(packet[:n])
 		}
 	}()
 	t.Cleanup(func() { _ = socket.Close(); <-done })
@@ -131,7 +157,7 @@ func TestIntegrationPeerPacketsKeepNativeTargetAcrossRelayPacketLoss(t *testing.
 				{"tcp-tcp-lose-b", [2]bool{true, true}, 1},
 			} {
 				t.Run(carriers.name, func(t *testing.T) {
-					exerciseNativePacketContinuity(t, publisher, carriers.tcp, carriers.loseTCP, false)
+					exerciseNativePacketContinuity(t, publisher, carriers.tcp, carriers.loseTCP, false, false)
 				})
 			}
 		})
@@ -140,21 +166,44 @@ func TestIntegrationPeerPacketsKeepNativeTargetAcrossRelayPacketLoss(t *testing.
 
 func TestIntegrationPeerPacketsRenewNativeClaimsWithoutReopeningTarget(t *testing.T) {
 	testutil.RequireTestTier(t, testutil.TestTierIntegration)
-	exerciseNativePacketContinuity(t, "quic", [2]bool{false, true}, 1, true)
+	exerciseNativePacketContinuity(t, "quic", [2]bool{false, true}, 1, true, false)
 }
 
 func TestIntegrationPeerPacketsReconnectBothNativeCarriersWithoutReopeningTarget(t *testing.T) {
 	testutil.RequireTestTier(t, testutil.TestTierIntegration)
 	for _, publisher := range []string{"quic", "tls-tcp"} {
 		t.Run(publisher, func(t *testing.T) {
-			exerciseNativePacketContinuity(t, publisher, [2]bool{true, true}, -2, false)
+			exerciseNativePacketContinuity(t, publisher, [2]bool{true, true}, -2, false, false)
+		})
+	}
+}
+
+func TestIntegrationPeerPacketsUpgradeNativeTargetToDirectUDP(t *testing.T) {
+	testutil.RequireTestTier(t, testutil.TestTierIntegration)
+	for _, publisher := range []string{"default", "tls-tcp"} {
+		t.Run(publisher, func(t *testing.T) {
+			exerciseNativePacketContinuity(t, publisher, [2]bool{}, -1, false, true)
 		})
 	}
 }
 
-func exerciseNativePacketContinuity(t *testing.T, publisher string, tcp [2]bool, loseTCP int, renew bool) {
+func exerciseNativePacketContinuity(t *testing.T, publisher string, tcp [2]bool, loseTCP int, renew, direct bool) {
 	t.Helper()
 	f, issuer := newWorkloadTransportFixtureTopology(t, nil, serviceRuntimeTopologyOptions{nativePeers: true})
+	var directReady [2]atomic.Bool
+	iceGate := make(chan struct{})
+	var release sync.Once
+	releaseICE := func() { release.Do(func() { close(iceGate) }) }
+	defer releaseICE()
+	connectivity := func(side int) peer.ConnectivityOptions {
+		return peer.ConnectivityOptions{DisableDirect: !direct, IncludeLoopback: true, Observe: func(state peer.ConnectivityState) {
+			if state == peer.ConnectivityChecking {
+				<-iceGate
+			}
+			directReady[side].Store(state == peer.ConnectivityDirect)
+		}}
+	}
+	f.peerConnectivity = connectivity(1)
 	// health probes may connect and close, but only a socket that receives
 	// application bytes gets an ID. a reopened target cannot restart its counter.
 	target := serviceRuntimeListener(t)
@@ -243,10 +292,11 @@ func exerciseNativePacketContinuity(t *testing.T, publisher string, tcp [2]bool,
 	defer cancel()
 	ready, done := make(chan struct{}), make(chan error, 1)
 	go func() {
-		done <- connectorstream.Serve(ctx, listener, connectorstream.Config{Control: control, NativePeers: true, ServiceID: run.ServiceID, Protocol: run.Protocol,
+		done <- connectorstream.Serve(ctx, listener, connectorstream.Config{Control: control, NativePeers: true, PeerConnectivity: connectivity(0), ServiceID: run.ServiceID, Protocol: run.Protocol,
 			TrustRevision: run.TrustRevision, Endpoint: run.Endpoint, TransportTLS: f.roots, ServiceTLS: f.visitorRoots, Ready: func() error { close(ready); return nil }})
 	}()
 	defer func() {
+		releaseICE()
 		cancel()
 		select {
 		case err := <-done:
@@ -300,6 +350,25 @@ func exerciseNativePacketContinuity(t *testing.T, publisher string, tcp [2]bool,
 	}
 	check(1)
 	counter := uint64(1)
+	if direct {
+		if directReady[0].Load() || directReady[1].Load() {
+			t.Fatal("ICE gate did not keep startup relay-first")
+		}
+		releaseICE()
+		privateWait(t, 10*time.Second, func(context.Context) (bool, error) {
+			return directReady[0].Load() && directReady[1].Load(), nil
+		})
+		// both relay UDP deliveries are unavailable: only the selected ICE path
+		// can carry the retained inner connection's application bytes.
+		for _, fault := range faults {
+			fault.Store(true)
+		}
+		counter++
+		check(counter)
+		for _, fault := range faults {
+			fault.Store(false)
+		}
+	}
 	if renew {
 		var original, firstClaim time.Time
 		if err := f.inspect.QueryRowContext(t.Context(), `SELECT expires_at FROM control.native_peer_pairs WHERE service_id=$1`, run.ServiceID).Scan(&original); err != nil {
@@ -342,7 +411,7 @@ func exerciseNativePacketContinuity(t *testing.T, publisher string, tcp [2]bool,
 		t.Logf("one target survived the original pair deadline; both relay claims advanced by at least %s", lastClaim.Sub(firstClaim).Round(time.Millisecond))
 	}
 	for slot := range 2 {
-		if tcp[slot] {
+		if tcp[slot] || direct {
 			continue
 		}
 		faults[slot].Store(true)

internal/daemon/service_separated_fixture_test.go

@@ -30,6 +30,7 @@ import (
 	"github.com/tnldotdev/tnl/internal/serviceapi"
 	"github.com/tnldotdev/tnl/internal/testutil"
 	"github.com/tnldotdev/tnl/internal/transport/muxsession"
+	"github.com/tnldotdev/tnl/internal/transport/peer"
 )
 
 const serviceRuntimeClusterSecret = "0123456789abcdef0123456789abcdef"
@@ -62,6 +63,7 @@ type privateSeparatedFixture struct {
 	visitorRootPEM              []byte
 	processContext              func(string) context.Context
 	nativePeers                 bool
+	peerConnectivity            peer.ConnectivityOptions
 	relays                      []*daemon
 	nativePacketDrops           map[string]*atomic.Bool
 	nativeTCPListeners          map[string]*nativeTrackedListener

internal/daemon/workload_transport_integration_test.go

@@ -336,9 +336,14 @@ func startWorkloadPublishers(t *testing.T, f *privateSeparatedFixture, group, tr
 		}
 		ready, done := make(chan struct{}), make(chan error, 1)
 		go func() {
-			done <- publisherservice.Publish(ctx, publisherservice.RunConfig{Control: control, State: f.identity.state, Access: native.Access, Identity: identity, ServiceID: service.Id, Target: assignment.Configuration.Target, Metadata: publisheridentity.ServiceMetadata(service), HeartbeatInterval: time.Second, Ready: func() error { close(ready); return nil }, Connect: func(ctx context.Context, bound *publisherservice.Bound, slot uint8) (publisherservice.SlotSession, error) {
-				return bound.ConnectSlotWithTransports(ctx, slot, false, quic, tcp)
-			}, NativePeers: f.nativePeers})
+			config := publisherservice.RunConfig{Control: control, State: f.identity.state, Access: native.Access, Identity: identity, ServiceID: service.Id, Target: assignment.Configuration.Target, Metadata: publisheridentity.ServiceMetadata(service), HeartbeatInterval: time.Second, Ready: func() error { close(ready); return nil },
+				NativePeers: f.nativePeers, PeerConnectivity: f.peerConnectivity, TransportTLS: f.roots}
+			if transport != "default" {
+				config.Connect = func(ctx context.Context, bound *publisherservice.Bound, slot uint8) (publisherservice.SlotSession, error) {
+					return bound.ConnectSlotWithTransports(ctx, slot, false, quic, tcp)
+				}
+			}
+			done <- publisherservice.Publish(ctx, config)
 		}()
 		t.Cleanup(func() {
 			cancel()

internal/publisher/service/lifecycle.go

@@ -21,6 +21,7 @@ import (
 	"github.com/tnldotdev/tnl/internal/region"
 	"github.com/tnldotdev/tnl/internal/security/credentials"
 	"github.com/tnldotdev/tnl/internal/security/targettrust"
+	"github.com/tnldotdev/tnl/internal/transport/peer"
 	"github.com/tnldotdev/tnl/internal/transport/targettransport"
 	"github.com/tnldotdev/tnl/internal/transport/tunnel"
 	authorityv1 "github.com/tnldotdev/tnl/pkg/api/control/authority/v1"
@@ -65,7 +66,8 @@ type RunConfig struct {
 	ObserveReadiness  func(RunStatus)
 	// NativePeers opts this run into background pair discovery and guarded
 	// packet serving. server-side admission must be prepared independently.
-	NativePeers bool
+	NativePeers      bool
+	PeerConnectivity peer.ConnectivityOptions
 }
 
 // RunStatus separates current readiness from the one-time Ready notification.
@@ -259,9 +261,19 @@ func Publish(ctx context.Context, config RunConfig) (result error) {
 	if err != nil {
 		return err
 	}
+	// all native pairs and both relay connections borrow this run-owned socket.
+	// invalidate joins their packet work before the physical socket is closed.
+	if config.NativePeers {
+		bound.peerSocket, err = peer.ListenSocket(ctx, "udp", ":0")
+		if err != nil {
+			return err
+		}
+		defer bound.peerSocket.Close()
+	}
 	defer bound.Invalidate()
 	bound.streamCapacity = int(config.StreamCapacity)
 	bound.nativePeers = config.NativePeers
+	bound.peerConnectivity = config.PeerConnectivity
 	policy, err := visitorpolicy.Prepare(ctx, managedControl, visitorscope.Scope{Kind: visitorscope.Service, ID: setup.ServiceId}, setup.ServiceRunId, setup.RunNumber, setup.RunToken, nil)
 	if err != nil {
 		return err

internal/publisher/service/peer_packets.go

@@ -3,6 +3,7 @@ package service
 import (
 	"context"
 	"crypto/tls"
+	"encoding/base64"
 	"errors"
 	"sync"
 	"time"
@@ -48,6 +49,12 @@ func (p *PeerSession) servePackets(ctx context.Context, prepare func(PeerBinding
 		return err
 	}
 	defer closePeerCarriers(paths)
+	var direct *peer.CarrierSlot
+	if p.bound.peerSocket != nil {
+		direct = peer.NewCarrierSlot()
+		paths[peer.Direct] = direct
+		defer direct.Close()
+	}
 	packets, err := peer.New(life, peer.Config{Kind: peer.NativePeer, Key: binding.key, Role: 1, Paths: paths})
 	if err != nil {
 		return err
@@ -62,7 +69,7 @@ func (p *PeerSession) servePackets(ctx context.Context, prepare func(PeerBinding
 			}
 			return p.bound.prepared.currentCertificate()
 		}}, &quic.Config{InitialPacketSize: tunnelv1.MaxPeerPacketBytes, DisablePathMTUDiscovery: true,
-		MaxIncomingStreams: 64, MaxIncomingUniStreams: -1, MaxIdleTimeout: 30 * time.Second, KeepAlivePeriod: 10 * time.Second})
+		MaxIncomingStreams: 64, MaxIncomingUniStreams: 1, MaxIdleTimeout: 30 * time.Second, KeepAlivePeriod: 10 * time.Second})
 	if err != nil {
 		return err
 	}
@@ -85,6 +92,14 @@ func (p *PeerSession) servePackets(ctx context.Context, prepare func(PeerBinding
 	closed := make(chan struct{})
 	stopSession := context.AfterFunc(life, func() { _ = session.Close(); close(closed) })
 	var workers sync.WaitGroup
+	if direct != nil {
+		workers.Go(func() {
+			_ = peer.ServeConnectivity(life, p.bound.peerSocket, connection, packets, direct, peer.ConnectivityBinding{
+				PairID: binding.pairID, DeliveryID: base64.RawURLEncoding.EncodeToString(binding.deliveryID[:]), Generation: binding.generation,
+				RelayAddresses: [2]string{binding.relays[0].RelayAddress, binding.relays[1].RelayAddress},
+			}, false, p.bound.peerConnectivity)
+		})
+	}
 	slots := make(chan struct{}, 64)
 	defer func() {
 		cancel()