feat: retain native streams across connector carrier handoff

docs/maintainers/transport/architecture.md

@@ -133,9 +133,13 @@ control request, authority time, or application admission. the relay rechecks it
 current process lease, publisher assignment, and claim deadline before installing
 packet forwarding. closing a passive loser cannot retire the winner's publisher
 registration. a duplicate activation cannot replace a live registration or reopen
-a retired delivery generation. this is startup selection, not reconnect handoff:
-closing the active outer session still retires that registration. production
-native admission remains disabled until its source authority is qualified.
+a retired delivery generation. startup selection and reconnect handoff have
+separate control messages. `native_leg_handoff` binds the new signed hello to
+the previous selected nonce while retaining the publisher registration and its
+replay state. connector loss detaches the carrier; the bridge retains only its
+last confirmed deadline. rejected or ambiguous handoffs cannot retry selection,
+and expiry still retires the registration. production native admission remains
+disabled until its source authority is qualified.
 
 `httpstream` enables HTTP/1 full duplex before target responses and bounds the
 upload, including trickles, to 30 seconds. Body completion ends the read deadline;

docs/maintainers/transport/peer-packets.md

@@ -186,8 +186,22 @@ deadline from request start plus the observed remaining duration, at most five
 minutes, rejecting a response already expired in transit. it cannot restart this
 deadline when delivery begins. publisher replacement, drain, close, cancellation,
 or claim expiry interrupt and join the packet workers and partial introductions.
-closing one pair removes its live registration while retaining the generation
-tombstone; it leaves sibling pairs and the borrowed publisher session alive.
+connector socket loss detaches its carrier while the relay bridge retains the
+publisher registration, replay history, and last confirmed deadline. bridge
+expiry removes the registration and keeps its generation tombstone; sibling pairs
+and the borrowed publisher session stay alive.
+
+`native_leg_handoff` selects a new signed hello nonce and names the exact previous
+selected nonce. the relay requires the same pair, delivery, source, assignment,
+and process lease, consumes the predecessor once, and clips selection to both
+existing deadlines. it retains at most 64 selected candidate nonces per bridge;
+a consumed nonce cannot return as a later candidate. an ordinary
+`native_leg_activate` cannot perform this handoff or reopen a retired generation.
+the connector joins the failed leg and its renewal work before preparing a
+replacement. setup failures may retry before selection; an ambiguous handoff
+response stops recovery of that slot. replacing either carrier preserves the
+inner QUIC connection and target sockets. assignment, source, and endpoint
+replacement still fence the existing pair rather than inherit its authority.
 
 pair renewal alone never advances an existing relay claim. connector background
 work sends a fresh `native_leg_renew` proof on the existing native control
@@ -308,7 +322,9 @@ the control-backed continuity proof runs with
 it uses disposable PostgreSQL, two real direct-socket relays, and no ingress.
 QUIC and TLS/TCP publishers each serve all connector carrier combinations.
 UDP packet loss and closing a native outer TCP session leave one target socket
-and its monotonic application counter intact. this fixture does not qualify
+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.
 the renewal case also retains that target past the original pair deadline,
 checks that both durable leg claims advanced, then applies the carrier faults.

internal/connector/stream/peer_connection.go

@@ -26,6 +26,7 @@ type peerConnection struct {
 	source    string
 	socket    *peer.Socket
 	legs      [2]*peerRelayLeg
+	paths     [2]*peer.CarrierSlot
 	packets   *peer.Endpoint
 	transport *quic.Transport
 	session   muxsession.Session
@@ -47,6 +48,7 @@ type peerRelayLeg struct {
 	until   time.Time
 	changed chan struct{}
 	retired bool
+	nonce   [32]byte
 }
 
 func (l *peerRelayLeg) Close() {
@@ -81,7 +83,7 @@ func preparePeerConnection(ctx context.Context, config Config, prepared *prepare
 	results := make(chan result, 2)
 	for slot := range 2 {
 		go func() {
-			leg, err := racePeerRelayLeg(startup, life, config, p.socket, initial, signer, slot)
+			leg, err := racePeerRelayLeg(startup, life, config, p.socket, initial, signer, slot, nil)
 			results <- result{slot, leg, err}
 		}()
 	}
@@ -102,7 +104,9 @@ func preparePeerConnection(ctx context.Context, config Config, prepared *prepare
 	}
 	p.source = p.legs[0].receipt.SourceIP
 	for slot, leg := range p.legs {
-		p.renewals.Go(func() { leg.renewPrepared(prepared, initial, signer, slot) })
+		p.paths[slot] = peer.NewCarrierSlot()
+		_ = p.paths[slot].Install(leg.carrier)
+		p.renewals.Go(func() { p.maintainRelayLeg(config, slot) })
 	}
 	// both relay claims can now be discovered by the publisher. do not start
 	// even packet probes until it has installed its independent registrations.
@@ -126,7 +130,7 @@ func preparePeerConnection(ctx context.Context, config Config, prepared *prepare
 	var packetKey [32]byte
 	copy(packetKey[:], key)
 	clear(key)
-	paths := [3]peer.Carrier{p.legs[0].carrier, p.legs[1].carrier}
+	paths := [3]peer.Carrier{p.paths[0], p.paths[1]}
 	p.packets, err = peer.New(life, peer.Config{Kind: peer.NativePeer, Role: 0, Key: packetKey, Paths: paths})
 	clear(packetKey[:])
 	if err != nil {
@@ -169,7 +173,7 @@ func (p *peerConnection) watch() {
 			return
 		case <-ticker.C:
 			_, err := p.prepared.peerRunForRenewal(p.initial)
-			if err != nil || p.legs[0].ctx.Err() != nil && p.legs[1].ctx.Err() != nil {
+			if err != nil {
 				p.cancel(ErrSelectionExpired)
 				return
 			}
@@ -188,12 +192,17 @@ func (p *peerConnection) closeResources() {
 	if p.packets != nil {
 		_ = p.packets.Close()
 	}
+	p.renewals.Wait()
 	for _, leg := range p.legs {
 		if leg != nil {
 			leg.Close()
 		}
 	}
-	p.renewals.Wait()
+	for _, path := range p.paths {
+		if path != nil {
+			_ = path.Close()
+		}
+	}
 	if p.socket != nil {
 		_ = p.socket.Close()
 	}
@@ -205,13 +214,13 @@ func (p *peerConnection) Close() {
 
 // the race joins both attempts, including a late authenticated loser. each
 // attempt gets its own signed nonce so it cannot replay a consumed leg proof.
-func racePeerRelayLeg(ctx, life context.Context, config Config, socket *peer.Socket, binding preparedPeer, signer Signer, slot int) (*peerRelayLeg, error) {
+func racePeerRelayLeg(ctx, life context.Context, config Config, socket *peer.Socket, binding preparedPeer, signer Signer, slot int, previous *peerLegPredecessor) (*peerRelayLeg, error) {
 	quicConnector := muxsession.QUICConnector{TLSConfig: config.TransportTLS, Dialer: socket, Config: muxsession.QUICConfig{Config: &quic.Config{EnableDatagrams: true}}}
 	tcpConnector := muxsession.TLSYamuxConnector{TLSConfig: config.TransportTLS}
-	return racePeerRelayCandidates(ctx, life, [2]muxsession.Connector{quicConnector, tcpConnector}, binding, signer, slot)
+	return racePeerRelayCandidates(ctx, life, [2]muxsession.Connector{quicConnector, tcpConnector}, binding, signer, slot, previous)
 }
 
-func racePeerRelayCandidates(ctx, life context.Context, connectors [2]muxsession.Connector, binding preparedPeer, signer Signer, slot int) (*peerRelayLeg, error) {
+func racePeerRelayCandidates(ctx, life context.Context, connectors [2]muxsession.Connector, binding preparedPeer, signer Signer, slot int, previous *peerLegPredecessor) (*peerRelayLeg, error) {
 	call, cancel := context.WithCancel(ctx)
 	defer cancel()
 	type result struct {
@@ -264,21 +273,41 @@ func racePeerRelayCandidates(ctx, life context.Context, connectors [2]muxsession
 		winner.mu.Lock()
 		until, retired := winner.until, winner.retired
 		winner.mu.Unlock()
+		if previous != nil {
+			until = minSelectionTime(until, previous.until)
+			if winner.receipt.SourceIP != previous.source {
+				winner.Close()
+				return nil, ErrServiceConnection
+			}
+		}
 		if retired || ctx.Err() != nil || winner.ctx.Err() != nil || !until.After(time.Now()) {
 			winner.Close()
 			return nil, errors.Join(context.Cause(ctx), ErrSelectionExpired)
 		}
 		activation, finish := context.WithDeadline(ctx, until)
-		err := winner.session.ActivateNativeLeg(activation, "activate")
+		var err error
+		if previous == nil {
+			err = winner.session.ActivateNativeLeg(activation, "activate")
+		} else {
+			err = winner.session.HandoffNativeLeg(activation, "handoff", previous.nonce[:])
+		}
 		finish()
 		if err != nil {
 			winner.Close()
+			if previous != nil {
+				err = errors.Join(errPeerHandoffConsumed, err)
+			}
 			return nil, err
 		}
 		if ctx.Err() != nil || winner.ctx.Err() != nil || !until.After(time.Now()) {
 			winner.Close()
-			return nil, errors.Join(context.Cause(ctx), ErrSelectionExpired)
+			return nil, errors.Join(errPeerHandoffConsumed, context.Cause(ctx), ErrSelectionExpired)
 		}
+		winner.mu.Lock()
+		winner.until = until
+		close(winner.changed)
+		winner.changed = make(chan struct{})
+		winner.mu.Unlock()
 		return winner, nil
 	}
 	return nil, errors.Join(ErrServiceConnection, context.Cause(ctx), errors.Join(failures...))
@@ -352,6 +381,8 @@ func dialPeerRelayLeg(ctx, lifetime context.Context, connector muxsession.Connec
 		return nil, err
 	}
 	leg := &peerRelayLeg{session: session, carrier: carrier, receipt: receipt, ctx: life, cancel: cancel, done: make(chan struct{}), until: until, changed: make(chan struct{})}
+	leg.carrier = &peerRelayCarrier{Carrier: carrier, stop: cancel}
+	copy(leg.nonce[:], proof.Nonce)
 	go func() {
 		defer close(leg.done)
 		leg.watchAuthority()

internal/connector/stream/peer_connection_test.go

@@ -40,6 +40,7 @@ type peerTestSession struct {
 	packets     atomic.Int32
 	activations atomic.Int32
 	activate    func() error
+	selection   func(tunnelv1.NativeLegBinding)
 }
 
 func newPeerTestSession(t *testing.T, datagrams bool, open func(muxsession.Stream)) *peerTestSession {
@@ -153,16 +154,23 @@ func peerHelloSession(t *testing.T, binding preparedPeer, signer Signer, change
 			if err != nil {
 				return
 			}
-			if request.Type == tunnelv1.NativeLegActivate && request.NativeLegBinding != nil && request.NativeLegBinding.Matches(proof) {
+			if (request.Type == tunnelv1.NativeLegActivate || request.Type == tunnelv1.NativeLegHandoff) && request.NativeLegBinding != nil && request.NativeLegBinding.Matches(proof) {
 				session.activations.Add(1)
+				if session.selection != nil {
+					session.selection(*request.NativeLegBinding)
+				}
 				if session.activate != nil {
 					if err := session.activate(); err != nil {
 						_ = tunnelv1.WriteControl(stream, tunnelv1.Message{Type: tunnelv1.Error, ProtocolVersion: tunnelv1.Version,
 							RequestID: request.RequestID, Code: tunnelv1.Unavailable})
 						return
 					}
 				}
-				if err := tunnelv1.WriteControl(stream, tunnelv1.Message{Type: tunnelv1.NativeLegActivated, ProtocolVersion: tunnelv1.Version,
+				accepted := tunnelv1.NativeLegActivated
+				if request.Type == tunnelv1.NativeLegHandoff {
+					accepted = tunnelv1.NativeLegHandedOff
+				}
+				if err := tunnelv1.WriteControl(stream, tunnelv1.Message{Type: accepted, ProtocolVersion: tunnelv1.Version,
 					RequestID: request.RequestID, NativeLegBinding: request.NativeLegBinding}); err != nil {
 					return
 				}
@@ -260,7 +268,7 @@ func TestNativePeerRelayRaceJoinsCanceledLateConnect(t *testing.T) {
 		}
 		done := make(chan *peerRelayLeg, 1)
 		go func() {
-			leg, err := racePeerRelayCandidates(t.Context(), t.Context(), connectors, binding, owner.signer, 0)
+			leg, err := racePeerRelayCandidates(t.Context(), t.Context(), connectors, binding, owner.signer, 0, nil)
 			if err != nil {
 				t.Error(err)
 			}
@@ -323,7 +331,7 @@ func TestNativePeerRelayActivationFailureClosesWinnerWithoutRetry(t *testing.T)
 				}
 				ctx, cancel := context.WithTimeout(t.Context(), time.Second)
 				defer cancel()
-				leg, err := racePeerRelayCandidates(ctx, t.Context(), connectors, binding, signer, 0)
+				leg, err := racePeerRelayCandidates(ctx, t.Context(), connectors, binding, signer, 0, nil)
 				if leg != nil || err == nil || primary.activations.Load() != 1 || connects.Load() != 1 || signatures.Load() != 1 {
 					t.Fatal("activation failure retried setup or admitted a carrier", err)
 				}
@@ -360,7 +368,7 @@ func TestNativePeerRelayOverlappingHellosActivateOnlySelectedCandidate(t *testin
 			muxsession.ConnectorFunc(func(context.Context, muxsession.Endpoint) (muxsession.Session, error) { return primary, nil }),
 			muxsession.ConnectorFunc(func(context.Context, muxsession.Endpoint) (muxsession.Session, error) { return fallback, nil }),
 		}
-		leg, err := racePeerRelayCandidates(t.Context(), t.Context(), connectors, binding, signer, 0)
+		leg, err := racePeerRelayCandidates(t.Context(), t.Context(), connectors, binding, signer, 0, nil)
 		if err != nil || leg == nil {
 			t.Fatal("overlapping native hellos did not select a winner", err)
 		}
@@ -407,7 +415,7 @@ func TestNativePeerRelayWinnerExpiryWhileJoiningDoesNotActivate(t *testing.T) {
 		}
 		done := make(chan error, 1)
 		go func() {
-			leg, err := racePeerRelayCandidates(t.Context(), t.Context(), connectors, binding, owner.signer, 0)
+			leg, err := racePeerRelayCandidates(t.Context(), t.Context(), connectors, binding, owner.signer, 0, nil)
 			if leg != nil {
 				leg.Close()
 				t.Error("expired winner was returned")
@@ -558,7 +566,7 @@ func TestNativePeerPacketStartupRequiresMatchingSourcesAndPublisherConfirmation(
 					accepted <- struct{}{}
 					controlDone := make(chan error, 1)
 					go func() {
-						controlDone <- session.HandleNativeLegControl(ctx, func(context.Context) error { return nil },
+						controlDone <- session.HandleNativeLegControl(ctx, func(context.Context, tunnelv1.NativeLegBinding) error { return nil },
 							func(context.Context, tunnelv1.NativePairLegProof) (tunnelv1.NativePairLegAccepted, error) {
 								t.Error("startup unexpectedly renewed authority")
 								return tunnelv1.NativePairLegAccepted{}, &tunnel.ProtocolError{Code: tunnelv1.Unauthenticated}

internal/connector/stream/peer_handoff.go

@@ -0,0 +1,107 @@
+package stream
+
+import (
+	"context"
+	"errors"
+	"time"
+
+	"github.com/tnldotdev/tnl/internal/transport/peer"
+)
+
+var errPeerHandoffConsumed = errors.New("native carrier handoff attempt was consumed")
+
+type peerLegPredecessor struct {
+	nonce  [32]byte
+	source string
+	until  time.Time
+}
+
+// a failed packet carrier retires the associated control owner as well. otherwise
+// a live control stream could keep renewing an unusable packet registration.
+type peerRelayCarrier struct {
+	peer.Carrier
+	stop context.CancelFunc
+}
+
+var _ peer.Carrier = (*peerRelayCarrier)(nil)
+
+func (c *peerRelayCarrier) Send(ctx context.Context, frame []byte) error {
+	err := c.Carrier.Send(ctx, frame)
+	if err != nil {
+		c.stop()
+	}
+	return err
+}
+
+func (c *peerRelayCarrier) Receive(ctx context.Context) ([]byte, error) {
+	frame, err := c.Carrier.Receive(ctx)
+	if err != nil {
+		c.stop()
+	}
+	return frame, err
+}
+
+func (c *peerRelayCarrier) Close() error {
+	c.stop()
+	return c.Carrier.Close()
+}
+
+// maintainRelayLeg owns this slot's renewal and reconnect work. replacement
+// touches neither the inner QUIC connection nor any application stream. only
+// failures before handoff may retry, within the predecessor's fixed deadline.
+func (p *peerConnection) maintainRelayLeg(config Config, slot int) {
+	leg := p.legs[slot]
+	defer func() { leg.Close() }()
+	for {
+		renewed := make(chan struct{})
+		go func() {
+			defer close(renewed)
+			leg.renewPrepared(p.prepared, p.initial, p.signer, slot)
+		}()
+		select {
+		case <-p.ctx.Done():
+		case <-leg.ctx.Done():
+		}
+		leg.Close()
+		<-renewed
+		if p.ctx.Err() != nil {
+			return
+		}
+		leg.mu.Lock()
+		previous := peerLegPredecessor{nonce: leg.nonce, source: leg.receipt.SourceIP, until: leg.until}
+		leg.mu.Unlock()
+		_ = p.paths[slot].Install(nil)
+		for {
+			if p.ctx.Err() != nil || !previous.until.After(time.Now()) {
+				return
+			}
+			binding, err := p.prepared.peerCurrent()
+			if err == nil && !samePeerDelivery(p.initial, binding) {
+				return
+			}
+			if err == nil {
+				call, cancel := context.WithTimeout(p.ctx, min(5*time.Second, time.Until(previous.until)))
+				next, err := racePeerRelayLeg(call, p.ctx, config, p.socket, binding, p.signer, slot, &previous)
+				cancel()
+				if err == nil {
+					leg = next
+					p.legs[slot] = next
+					if p.paths[slot].Install(next.carrier) != nil {
+						return
+					}
+					break
+				}
+				if errors.Is(err, errPeerHandoffConsumed) {
+					return
+				}
+			}
+			timer := time.NewTimer(min(250*time.Millisecond, time.Until(previous.until)))
+			select {
+			case <-p.ctx.Done():
+				timer.Stop()
+				return
+			case <-timer.C:
+			}
+		}
+	}
+}