diff --git a/docs/maintainers/connector/vercel-functions.md b/docs/maintainers/connector/vercel-functions.md index 54b141ad5..a54c24872 100644 --- a/docs/maintainers/connector/vercel-functions.md +++ b/docs/maintainers/connector/vercel-functions.md @@ -44,10 +44,11 @@ the check asserts: - four concurrent cold HTTP invocations start one live owner from the packaged native executable; PostgreSQL has no preparation requests yet; - concurrent HTTP and real pg queries subsequently share that same owner; -- four warm preparation pairs use eight direct IPC requests, zero helper - processes, and zero control requests; -- warm application traffic adds only current-run checks for new visitor - connections: four for pg, plus up to four for HTTP if its pool opens sockets; +- idle background preparation refreshes the current run and its visitor token; +- with control requests held at the fixture proxy, four warm preparation pairs + use eight direct IPC requests and zero helper processes; +- warm application traffic reaches both targets while control remains held, + using the owner's last confirmed preparation within its existing deadline; - assertion rotation adds one identity helper and retains the owner and session; - another cold isolate gets its own owner and key-bound workload session; - missing, expired, and signed wrong-project assertions fail on cold and warm @@ -60,7 +61,8 @@ the check asserts: the request adapter records actual helper launches, IPC operations, and live native PIDs without replacing their implementations. the Go HTTPS front records -actual authority/control requests and forwards responses unchanged. the target +actual authority/control requests. a joined gate holds background preparation +during the warm checks, then resumes normal forwarding. the target counter includes HTTP requests and PostgreSQL startup packets; publisher health probes are excluded. full counters are retained in the invocation evidence. diff --git a/docs/maintainers/transport/peer-packets.md b/docs/maintainers/transport/peer-packets.md index 1347267b0..6e60ae611 100644 --- a/docs/maintainers/transport/peer-packets.md +++ b/docs/maintainers/transport/peer-packets.md @@ -156,7 +156,8 @@ an opt-in native publisher run shares one socket across its two relay connection 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. +exchanges one bounded message per attempt on a dedicated unidirectional stream +each way. a single reader per endpoint bounds queued signaling to one message. 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. @@ -164,9 +165,16 @@ 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. +streams over relay delivery. the connector retries failed setup, lost consent, +or a path that stops passing packet probes after a five-second pause. each attempt +gathers current candidates and fresh credentials; the publisher joins its previous +agent before accepting the next installation. both sides reject reused attempt +numbers and mismatched pair/delivery fields. publisher agent creation is also +limited to once per five seconds, including when offers arrive in a burst. + +attempt numbers belong only to the existing inner connection. they cannot renew +serving authority or survive endpoint-owner replacement. cancellation interrupts +partial signaling, joins its reader and agent, and then releases the shared socket. relay and standalone answer STUN Binding requests on their existing QUIC socket. the server bounds request size and response rate and ignores ICE-authenticated @@ -352,5 +360,5 @@ 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 restart remain required +rebinding and deployed listener source handling remain required before enabling the native path. diff --git a/internal/daemon/peer_packets_integration_test.go b/internal/daemon/peer_packets_integration_test.go index 2ed7c2e33..184106279 100644 --- a/internal/daemon/peer_packets_integration_test.go +++ b/internal/daemon/peer_packets_integration_test.go @@ -157,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, false) + exerciseNativePacketContinuity(t, publisher, carriers.tcp, carriers.loseTCP, false, false, false) }) } }) @@ -166,14 +166,14 @@ func TestIntegrationPeerPacketsKeepNativeTargetAcrossRelayPacketLoss(t *testing. func TestIntegrationPeerPacketsRenewNativeClaimsWithoutReopeningTarget(t *testing.T) { testutil.RequireTestTier(t, testutil.TestTierIntegration) - exerciseNativePacketContinuity(t, "quic", [2]bool{false, true}, 1, true, false) + exerciseNativePacketContinuity(t, "quic", [2]bool{false, true}, 1, true, false, 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, false) + exerciseNativePacketContinuity(t, publisher, [2]bool{true, true}, -2, false, false, false) }) } } @@ -182,23 +182,42 @@ 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) + exerciseNativePacketContinuity(t, publisher, [2]bool{}, -1, false, true, false) }) } } -func exerciseNativePacketContinuity(t *testing.T, publisher string, tcp [2]bool, loseTCP int, renew, direct bool) { +func TestIntegrationPeerPacketsRetryICEWithoutReopeningTarget(t *testing.T) { + testutil.RequireTestTier(t, testutil.TestTierIntegration) + exerciseNativePacketContinuity(t, "default", [2]bool{}, -1, false, true, true) +} + +func exerciseNativePacketContinuity(t *testing.T, publisher string, tcp [2]bool, loseTCP int, renew, direct, restart bool) { t.Helper() f, issuer := newWorkloadTransportFixtureTopology(t, nil, serviceRuntimeTopologyOptions{nativePeers: true}) var directReady [2]atomic.Bool + var checks [2]atomic.Uint64 iceGate := make(chan struct{}) + firstFailure := make(chan struct{}) + var failed sync.Once + releaseFailure := func() { failed.Do(func() { close(firstFailure) }) } + defer releaseFailure() 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 { + attempt := checks[side].Add(1) <-iceGate + if restart && side == 1 && attempt == 1 { + // withhold the first publisher's ICE checks until the connector + // has abandoned that attempt. relay application traffic stays live. + <-firstFailure + } + } + if restart && side == 0 && state == peer.ConnectivityRelay { + releaseFailure() } directReady[side].Store(state == peer.ConnectivityDirect) }} @@ -297,6 +316,7 @@ func exerciseNativePacketContinuity(t *testing.T, publisher string, tcp [2]bool, }() defer func() { releaseICE() + releaseFailure() cancel() select { case err := <-done: @@ -355,9 +375,22 @@ func exerciseNativePacketContinuity(t *testing.T, publisher string, tcp [2]bool, t.Fatal("ICE gate did not keep startup relay-first") } releaseICE() - privateWait(t, 10*time.Second, func(context.Context) (bool, error) { + timeout := 10 * time.Second + if restart { + timeout = 25 * time.Second + } + lastExchange := time.Now() + privateWait(t, timeout, func(context.Context) (bool, error) { + if restart && time.Since(lastExchange) >= time.Second { + counter++ + check(counter) + lastExchange = time.Now() + } return directReady[0].Load() && directReady[1].Load(), nil }) + if restart && (checks[0].Load() < 2 || checks[1].Load() < 2) { + t.Fatal("direct path did not require a second ICE attempt") + } // both relay UDP deliveries are unavailable: only the selected ICE path // can carry the retained inner connection's application bytes. for _, fault := range faults { diff --git a/internal/daemon/workload_vercel_integration_test.go b/internal/daemon/workload_vercel_integration_test.go index d82bbbf1d..9c01dc696 100644 --- a/internal/daemon/workload_vercel_integration_test.go +++ b/internal/daemon/workload_vercel_integration_test.go @@ -103,12 +103,29 @@ func TestIntegrationWorkloadVercelFunction(t *testing.T) { } var mu sync.Mutex calls := map[string]int{} + blocked := map[string]int{} + var paused chan struct{} + var inFlight int + changed := make(chan struct{}) + notify := func() { close(changed); changed = make(chan struct{}) } prepared := map[string]bool{} sessions := map[string]authorityv1.WorkloadSession{} origin, _ := url.Parse(f.controlConfig.ControlOrigin()) proxy := &httputil.ReverseProxy{Rewrite: func(r *httputil.ProxyRequest) { r.SetURL(origin) }} proxy.Transport = splitACMERoundTripFunc(func(r *http.Request) (*http.Response, error) { mu.Lock() + if gate := paused; gate != nil { + blocked[r.Method+" "+r.URL.Path]++ + notify() + mu.Unlock() + select { + case <-r.Context().Done(): + return nil, r.Context().Err() + case <-gate: + } + mu.Lock() + } + inFlight++ calls[r.Method+" "+r.URL.Path]++ for id, name := range services { if strings.Contains(r.URL.Path, "/"+id+"/") { @@ -116,6 +133,12 @@ func TestIntegrationWorkloadVercelFunction(t *testing.T) { } } mu.Unlock() + defer func() { + mu.Lock() + inFlight-- + notify() + mu.Unlock() + }() return f.publicHTTP.Transport.RoundTrip(r) }) proxy.ModifyResponse = func(r *http.Response) error { @@ -189,14 +212,53 @@ func TestIntegrationWorkloadVercelFunction(t *testing.T) { controller := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { mu.Lock() defer mu.Unlock() + wait := func(ready func() bool) bool { + ctx, cancel := context.WithTimeout(r.Context(), 5*time.Second) + defer cancel() + for !ready() { + update := changed + mu.Unlock() + select { + case <-ctx.Done(): + case <-update: + } + mu.Lock() + if ctx.Err() != nil { + http.Error(w, "background preparation did not complete", http.StatusGatewayTimeout) + return false + } + } + return true + } switch r.URL.Path { + case "/background-refresh": + before := calls["POST /v1/visitor-tokens"] + if wait(func() bool { return calls["POST /v1/visitor-tokens"] > before && inFlight == 0 }) { + w.WriteHeader(http.StatusNoContent) + } + case "/pause": + if paused != nil { + w.WriteHeader(http.StatusConflict) + return + } + paused = make(chan struct{}) + blocked = map[string]int{} + if wait(func() bool { return inFlight == 0 && len(blocked) > 0 }) { + w.WriteHeader(http.StatusNoContent) + } + case "/resume": + if paused != nil { + close(paused) + paused = nil + } + w.WriteHeader(http.StatusNoContent) case "/stats": var names []string for name := range prepared { names = append(names, name) } slices.Sort(names) - _ = json.NewEncoder(w).Encode(map[string]any{"sessions": len(sessions), "prepared": names, "targets": targets.Load(), "calls": calls}) + _ = json.NewEncoder(w).Encode(map[string]any{"sessions": len(sessions), "prepared": names, "targets": targets.Load(), "calls": calls, "blocked": blocked}) case "/revoke": client, err := clientauthority.New(f.controlConfig.ControlOrigin(), f.publicHTTP, "") if err == nil { diff --git a/internal/transport/peer/connectivity.go b/internal/transport/peer/connectivity.go index f0b74a34a..0f24101eb 100644 --- a/internal/transport/peer/connectivity.go +++ b/internal/transport/peer/connectivity.go @@ -3,7 +3,7 @@ package peer import ( "context" "errors" - "io" + "math" "net" "net/netip" "strconv" @@ -21,9 +21,11 @@ type ConnectivityState string var ErrConnectivityConsent = errors.New("native direct connectivity consent expired") const ( - ConnectivityChecking ConnectivityState = "checking" - ConnectivityDirect ConnectivityState = "direct" - ConnectivityRelay ConnectivityState = "relay" + ConnectivityChecking ConnectivityState = "checking" + ConnectivityDirect ConnectivityState = "direct" + ConnectivityRelay ConnectivityState = "relay" + connectivitySetupTimeout = 10 * time.Second + connectivityRetryDelay = 5 * time.Second ) // ConnectivityOptions are supplied by native owners, never by application @@ -53,10 +55,10 @@ func (e *ConnectivityError) Unwrap() error { return e.Cause } -// ServeConnectivity is background work on an already-established inner QUIC -// connection. one bounded signaling stream in each direction is independent of -// application admission. the owner joins this function before closing its socket. -// ICE consent installs delivery only; endpoint MAC probes choose the new path. +// ServeConnectivity owns successive ICE attempts on one existing inner QUIC +// connection. only the connector starts attempts. an attempt replaces delivery, +// never application admission, source authority, or the owner's expiry. +// each side has one reader and one agent at a time; cancellation joins both. func ServeConnectivity(ctx context.Context, socket *Socket, connection *quic.Conn, packets *Endpoint, slot *CarrierSlot, binding ConnectivityBinding, controlling bool, options ConnectivityOptions) (retErr error) { if options.DisableDirect { return nil @@ -91,63 +93,135 @@ func ServeConnectivity(ctx context.Context, socket *Socket, connection *quic.Con } urls = append(urls, &stun.URI{Scheme: stun.SchemeTypeSTUN, Proto: stun.ProtoTypeUDP, Host: host, Port: number}) } + life, cancel := context.WithCancel(ctx) + stopConnection := interruptOnCancel(connection.Context(), func() error { cancel(); return nil }) + defer stopConnection() + signals := make(chan connectivitySignal, 1) + readerDone := make(chan struct{}) + go func() { defer close(readerDone); readConnectivitySignals(life, connection, binding, signals) }() + defer func() { cancel(); <-readerDone }() + var attempt uint64 + var offer *tunnelv1.NativePeerConnectivity + var lastStarted time.Time + for { + if life.Err() != nil { + return context.Cause(life) + } + if controlling { + if attempt == math.MaxUint64 { + return &ConnectivityError{Code: tunnelv1.InvalidMessage, Cause: ErrConfiguration} + } + attempt++ + } else { + if offer == nil { + signal, err := nextConnectivitySignal(life, signals) + if err != nil { + return err + } + offer = &signal + } + attempt = offer.Attempt + // bounded input and a local start interval limit an authenticated peer + // to one agent at a time even if it sends offers faster than expected. + if delay := time.Until(lastStarted.Add(connectivityRetryDelay)); delay > 0 { + timer := time.NewTimer(delay) + select { + case <-life.Done(): + timer.Stop() + return context.Cause(life) + case <-timer.C: + } + } + } + lastStarted = time.Now() + var err error + offer, err = runConnectivityAttempt(life, socket, connection, packets, slot, binding, urls, attempt, offer, signals, controlling, options) + if options.Observe != nil { + options.Observe(ConnectivityRelay) + } + if connectivityTerminal(err) { + return err + } + if life.Err() != nil { + return errors.Join(context.Cause(life), err) + } + if controlling { + // a failed setup or lost path never spins agent creation. the previous + // agent and all its socket registrations have been joined before waiting. + if err := waitConnectivityRetry(life, signals, attempt); err != nil { + return err + } + } + } +} + +func waitConnectivityRetry(ctx context.Context, signals <-chan connectivitySignal, attempt uint64) error { + timer := time.NewTimer(connectivityRetryDelay) + defer timer.Stop() + for { + select { + case <-ctx.Done(): + return context.Cause(ctx) + case <-timer.C: + return nil + case signal := <-signals: + if signal.err != nil { + return signal.err + } + // late answers cannot restart the retry timer or install a carrier. + // the single reader rejects reused attempt numbers independently. + if signal.message.Attempt > attempt { + return &ConnectivityError{Code: tunnelv1.InvalidMessage, Cause: tunnelv1.ErrNativePeerConnectivity} + } + } + } +} + +func connectivityTerminal(err error) bool { + var classified *ConnectivityError + return errors.As(err, &classified) && classified != nil && (classified.Code == tunnelv1.InvalidMessage || classified.Code == tunnelv1.Unauthenticated) +} + +func runConnectivityAttempt(ctx context.Context, socket *Socket, connection *quic.Conn, packets *Endpoint, slot *CarrierSlot, binding ConnectivityBinding, urls []*stun.URI, attempt uint64, offer *tunnelv1.NativePeerConnectivity, signals <-chan connectivitySignal, controlling bool, options ConnectivityOptions) (next *tunnelv1.NativePeerConnectivity, retErr error) { agent, err := socket.NewICE(urls, options.IncludeLoopback) if err != nil { - return err + return nil, err } defer func() { retErr = errors.Join(retErr, agent.Close()) }() - setup, cancel := context.WithTimeout(ctx, 10*time.Second) + setup, cancel := context.WithTimeout(ctx, connectivitySetupTimeout) defer cancel() - message := tunnelv1.NativePeerConnectivity{ProtocolVersion: tunnelv1.Version, Kind: "native_peer_connectivity", PairID: binding.PairID, DeliveryID: binding.DeliveryID, DeliveryGeneration: binding.Generation} - message.UsernameFragment, message.Password, err = agent.GetLocalUserCredentials() + local, err := gatherConnectivity(setup, agent, binding, attempt) if err != nil { - return err - } - var mu sync.Mutex - var completed sync.Once - complete := make(chan struct{}) - if err := agent.OnCandidate(func(candidate ice.Candidate) { - if candidate == nil { - completed.Do(func() { close(complete) }) - return - } - mu.Lock() - if len(message.Candidates) < tunnelv1.MaxNativePeerCandidates && validConnectivityCandidate(candidate) { - message.Candidates = append(message.Candidates, candidate.Marshal()) - } - mu.Unlock() - }); err != nil { - return err - } - if err := agent.GatherCandidates(); err != nil { - return err + return nil, err } - select { - case <-setup.Done(): - return context.Cause(setup) - case <-complete: + if err := writeConnectivitySignal(setup, connection, local); err != nil { + return nil, err } - mu.Lock() - local := message - local.Candidates = append([]string(nil), message.Candidates...) - mu.Unlock() - if err := local.Validate(); err != nil { - return &ConnectivityError{Code: tunnelv1.Unavailable, Cause: err} - } - remote, err := exchangeConnectivity(setup, connection, local) - if err != nil { - return err - } - if remote.PairID != binding.PairID || remote.DeliveryID != binding.DeliveryID || remote.DeliveryGeneration != binding.Generation { - return &ConnectivityError{Code: tunnelv1.Unauthenticated, Cause: tunnelv1.ErrNativePeerConnectivity} + var remote tunnelv1.NativePeerConnectivity + if controlling { + for { + remote, err = nextConnectivitySignal(setup, signals) + if err != nil { + return nil, err + } + if remote.Attempt < attempt { + continue + } + if remote.Attempt != attempt { + return nil, &ConnectivityError{Code: tunnelv1.InvalidMessage, Cause: tunnelv1.ErrNativePeerConnectivity} + } + break + } + } else { + remote = *offer } for _, encoded := range remote.Candidates { candidate, err := ice.UnmarshalCandidate(encoded) if err != nil || !validConnectivityCandidate(candidate) { - return &ConnectivityError{Code: tunnelv1.InvalidMessage, Cause: errors.Join(tunnelv1.ErrNativePeerConnectivity, err)} + return nil, &ConnectivityError{Code: tunnelv1.InvalidMessage, Cause: errors.Join(tunnelv1.ErrNativePeerConnectivity, err)} } if err := agent.AddRemoteCandidate(candidate); err != nil { - return err + return nil, err } } if options.Observe != nil { @@ -160,7 +234,7 @@ func ServeConnectivity(ctx context.Context, socket *Socket, connection *quic.Con lost.Do(func() { close(lostConsent) }) } }); err != nil { - return err + return nil, err } var direct *ice.Conn if controlling { @@ -169,76 +243,90 @@ func ServeConnectivity(ctx context.Context, socket *Socket, connection *quic.Con direct, err = agent.Accept(setup, remote.UsernameFragment, remote.Password) } if err != nil { - return err + return nil, err } if ctx.Err() != nil { _ = direct.Close() - return context.Cause(ctx) + return nil, context.Cause(ctx) } if err := slot.Install(ICECarrier{Conn: direct}); err != nil { - return err + return nil, err } defer slot.Install(nil) ticker := time.NewTicker(100 * time.Millisecond) defer ticker.Stop() - last := ConnectivityRelay + last, lastDirect := ConnectivityRelay, time.Now() for { select { case <-ctx.Done(): - return nil - case <-connection.Context().Done(): - return nil + return nil, context.Cause(ctx) case <-lostConsent: - return &ConnectivityError{Code: tunnelv1.Unavailable, Cause: ErrConnectivityConsent} - case <-ticker.C: + return nil, &ConnectivityError{Code: tunnelv1.Unavailable, Cause: ErrConnectivityConsent} + case signal := <-signals: + if signal.err != nil { + return nil, signal.err + } + if controlling || signal.message.Attempt <= attempt { + return nil, &ConnectivityError{Code: tunnelv1.InvalidMessage, Cause: tunnelv1.ErrNativePeerConnectivity} + } + // the reader has already checked binding and ordering. cleanup + // removes this agent before the next offer may install another carrier. + return &signal.message, nil + case now := <-ticker.C: next := ConnectivityRelay if packets.ActivePath() == Direct { - next = ConnectivityDirect + next, lastDirect = ConnectivityDirect, now } if next != last && options.Observe != nil { options.Observe(next) } last = next + if controlling && now.Sub(lastDirect) >= max(5*time.Second, 2*packets.config.PathTimeout) { + return nil, &ConnectivityError{Code: tunnelv1.Unavailable, Cause: ErrConnectivityConsent} + } } } } -func exchangeConnectivity(ctx context.Context, connection *quic.Conn, local tunnelv1.NativePeerConnectivity) (remote tunnelv1.NativePeerConnectivity, retErr error) { - out, err := connection.OpenUniStreamSync(ctx) +func gatherConnectivity(ctx context.Context, agent *ice.Agent, binding ConnectivityBinding, attempt uint64) (tunnelv1.NativePeerConnectivity, error) { + message := tunnelv1.NativePeerConnectivity{ProtocolVersion: tunnelv1.Version, Kind: "native_peer_connectivity", PairID: binding.PairID, DeliveryID: binding.DeliveryID, DeliveryGeneration: binding.Generation, Attempt: attempt} + var err error + message.UsernameFragment, message.Password, err = agent.GetLocalUserCredentials() if err != nil { - return remote, err + return message, err } - stopWrite := interruptOnCancel(ctx, func() error { out.CancelWrite(0); return nil }) - defer stopWrite() - defer func() { - if retErr != nil { - out.CancelWrite(0) + var mu sync.Mutex + var completed sync.Once + complete := make(chan struct{}) + if err := agent.OnCandidate(func(candidate ice.Candidate) { + if candidate == nil { + completed.Do(func() { close(complete) }) + return } - }() - if err := tunnelv1.WriteNativePeerConnectivity(out, local); err != nil { - return remote, err - } - if err := out.Close(); err != nil { - return remote, err + mu.Lock() + if len(message.Candidates) < tunnelv1.MaxNativePeerCandidates && validConnectivityCandidate(candidate) { + message.Candidates = append(message.Candidates, candidate.Marshal()) + } + mu.Unlock() + }); err != nil { + return message, err } - in, err := connection.AcceptUniStream(ctx) - if err != nil { - return remote, err + if err := agent.GatherCandidates(); err != nil { + return message, err } - stopRead := interruptOnCancel(ctx, func() error { in.CancelRead(0); return nil }) - defer stopRead() - defer in.CancelRead(0) - remote, err = tunnelv1.ReadNativePeerConnectivity(in) - if err != nil { - return remote, &ConnectivityError{Code: tunnelv1.InvalidMessage, Cause: err} + select { + case <-ctx.Done(): + return tunnelv1.NativePeerConnectivity{}, context.Cause(ctx) + case <-complete: } - // a signaling stream has exactly one frame. reject trailing data rather than - // leaving an unbounded body or a second credential set attached to the pair. - var extra [1]byte - if n, err := in.Read(extra[:]); n != 0 || !errors.Is(err, io.EOF) { - return remote, &ConnectivityError{Code: tunnelv1.InvalidMessage, Cause: errors.Join(tunnelv1.ErrNativePeerConnectivity, err)} + mu.Lock() + local := message + local.Candidates = append([]string(nil), message.Candidates...) + mu.Unlock() + if err := local.Validate(); err != nil { + return tunnelv1.NativePeerConnectivity{}, &ConnectivityError{Code: tunnelv1.Unavailable, Cause: err} } - return remote, nil + return local, nil } func validConnectivityCandidate(candidate ice.Candidate) bool { diff --git a/internal/transport/peer/connectivity_signal.go b/internal/transport/peer/connectivity_signal.go new file mode 100644 index 000000000..5edf2b153 --- /dev/null +++ b/internal/transport/peer/connectivity_signal.go @@ -0,0 +1,91 @@ +package peer + +import ( + "context" + "errors" + "io" + "time" + + quic "github.com/quic-go/quic-go" + "github.com/tnldotdev/tnl/pkg/protocol/tunnelv1" +) + +type connectivitySignal struct { + message tunnelv1.NativePeerConnectivity + err error +} + +// one dedicated reader owns all signaling streams for the inner connection. a +// one-entry channel bounds queued offers/answers; each stream contains one frame. +func readConnectivitySignals(ctx context.Context, connection *quic.Conn, binding ConnectivityBinding, output chan<- connectivitySignal) { + var previous uint64 + for { + message, err := readConnectivitySignal(ctx, connection) + if err == nil && (message.PairID != binding.PairID || message.DeliveryID != binding.DeliveryID || message.DeliveryGeneration != binding.Generation) { + err = &ConnectivityError{Code: tunnelv1.Unauthenticated, Cause: tunnelv1.ErrNativePeerConnectivity} + } + if err == nil && message.Attempt <= previous { + err = &ConnectivityError{Code: tunnelv1.InvalidMessage, Cause: tunnelv1.ErrNativePeerConnectivity} + } + previous = message.Attempt + select { + case <-ctx.Done(): + return + case output <- connectivitySignal{message, err}: + } + if err != nil { + return + } + } +} + +func nextConnectivitySignal(ctx context.Context, input <-chan connectivitySignal) (tunnelv1.NativePeerConnectivity, error) { + select { + case <-ctx.Done(): + return tunnelv1.NativePeerConnectivity{}, context.Cause(ctx) + case signal := <-input: + return signal.message, signal.err + } +} + +func writeConnectivitySignal(ctx context.Context, connection *quic.Conn, message tunnelv1.NativePeerConnectivity) (retErr error) { + out, err := connection.OpenUniStreamSync(ctx) + if err != nil { + return err + } + stop := interruptOnCancel(ctx, func() error { out.CancelWrite(0); return nil }) + defer stop() + defer func() { + if retErr != nil { + out.CancelWrite(0) + } + }() + if err := tunnelv1.WriteNativePeerConnectivity(out, message); err != nil { + return err + } + return out.Close() +} + +func readConnectivitySignal(ctx context.Context, connection *quic.Conn) (message tunnelv1.NativePeerConnectivity, retErr error) { + in, err := connection.AcceptUniStream(ctx) + if err != nil { + return message, err + } + stop := interruptOnCancel(ctx, func() error { in.CancelRead(0); return nil }) + defer stop() + defer in.CancelRead(0) + // an idle peer need not send another offer. once it opens a stream, bound + // the entire frame and EOF read independently of its claimed attempt number. + if err := in.SetReadDeadline(time.Now().Add(connectivitySetupTimeout)); err != nil { + return message, err + } + message, err = tunnelv1.ReadNativePeerConnectivity(in) + if err != nil { + return message, &ConnectivityError{Code: tunnelv1.InvalidMessage, Cause: err} + } + var extra [1]byte + if n, err := in.Read(extra[:]); n != 0 || !errors.Is(err, io.EOF) { + return message, &ConnectivityError{Code: tunnelv1.InvalidMessage, Cause: errors.Join(tunnelv1.ErrNativePeerConnectivity, err)} + } + return message, nil +} diff --git a/internal/transport/peer/connectivity_test.go b/internal/transport/peer/connectivity_test.go index 7ab490790..13f171b11 100644 --- a/internal/transport/peer/connectivity_test.go +++ b/internal/transport/peer/connectivity_test.go @@ -6,6 +6,7 @@ import ( "errors" "io" "net" + "strings" "sync" "testing" "time" @@ -26,7 +27,7 @@ type connectivityFixture struct { func newConnectivityFixture(t *testing.T) *connectivityFixture { t.Helper() - ctx, cancel := context.WithTimeout(t.Context(), 15*time.Second) + ctx, cancel := context.WithTimeout(t.Context(), 35*time.Second) t.Cleanup(cancel) f := &connectivityFixture{ctx: ctx, binding: ConnectivityBinding{ PairID: "npp_1234567890123456789012", DeliveryID: "AQEBAQEBAQEBAQEBAQEBAQ", Generation: 1, @@ -76,10 +77,8 @@ func newConnectivityFixture(t *testing.T) *connectivityFixture { return f } -func TestNativePeerConnectivityUpgradesAndFallsBackOnSameStream(t *testing.T) { - f := newConnectivityFixture(t) - relayExchange := sharedSocketQUICExchange(t, f.ctx, f.sockets[1]) - relayExchange() +func connectivityCounter(t *testing.T, f *connectivityFixture) func() { + t.Helper() stream, err := f.connections[0].OpenStreamSync(f.ctx) if err != nil { t.Fatal(err) @@ -138,7 +137,7 @@ func TestNativePeerConnectivityUpgradesAndFallsBackOnSameStream(t *testing.T) { workers.Wait() }) var counter uint64 - exchange := func() { + return func() { t.Helper() if err := stream.SetDeadline(time.Now().Add(3 * time.Second)); err != nil { t.Fatal(err) @@ -155,6 +154,13 @@ func TestNativePeerConnectivityUpgradesAndFallsBackOnSameStream(t *testing.T) { t.Fatal("target socket was replaced or bytes replayed") } } +} + +func TestNativePeerConnectivityUpgradesAndFallsBackOnSameStream(t *testing.T) { + f := newConnectivityFixture(t) + relayExchange := sharedSocketQUICExchange(t, f.ctx, f.sockets[1]) + relayExchange() + exchange := connectivityCounter(t, f) exchange() if f.packets[0].ActivePath() == Direct { t.Fatal("application did not start over a relay") @@ -184,6 +190,23 @@ func TestNativePeerConnectivityUpgradesAndFallsBackOnSameStream(t *testing.T) { awaitPath(t, f.ctx, f.packets[0], f.packets[1], Direct) exchange() relayExchange() + old := [2]uint64{f.slots[0].Generation(), f.slots[1].Generation()} + if err := f.slots[0].Install(nil); err != nil { + t.Fatal(err) + } + // losing one ICE agent must trigger new credentials on both sides. application + // bytes continue on relays while the controlling endpoint waits and retries. + exchange() + for f.slots[0].Generation() <= old[0]+1 || f.slots[1].Generation() <= old[1]+1 { + select { + case <-f.ctx.Done(): + t.Fatal("ICE was not restarted") + case <-time.After(10 * time.Millisecond): + } + } + awaitPath(t, f.ctx, f.packets[0], f.packets[1], Direct) + exchange() + relayExchange() stop() deadline, finish := context.WithTimeout(f.ctx, 2*time.Second) defer finish() @@ -253,6 +276,166 @@ func sharedSocketQUICExchange(t *testing.T, ctx context.Context, socket *Socket) } } +func TestNativePeerConnectivityRetriesFailedSetupWithoutReopeningTarget(t *testing.T) { + f := newConnectivityFixture(t) + exchange := connectivityCounter(t, f) + exchange() + ctx, cancel := context.WithCancel(f.ctx) + results := make(chan error, 2) + started := 1 + go func() { + results <- ServeConnectivity(ctx, f.sockets[0], f.connections[0], f.packets[0], f.slots[0], f.binding, true, ConnectivityOptions{IncludeLoopback: true}) + }() + t.Cleanup(func() { + cancel() + for range started { + select { + case err := <-results: + if err != nil && !errors.Is(err, context.Canceled) { + t.Error(err) + } + case <-time.After(time.Second): + t.Error("failed-attempt cleanup did not join") + } + } + }) + first, err := readConnectivitySignal(f.ctx, f.connections[1]) + if err != nil { + t.Fatal(err) + } + if first.Attempt != 1 { + t.Fatal("wrong first attempt") + } + // an unreachable but correctly bound answer consumes only this attempt. + first.UsernameFragment, first.Password = "unreachable", strings.Repeat("x", 22) + first.Candidates = []string{"1 1 udp 2130706431 127.0.0.1 9 typ host"} + if err := writeConnectivitySignal(f.ctx, f.connections[1], first); err != nil { + t.Fatal(err) + } + started++ + go func() { + results <- ServeConnectivity(ctx, f.sockets[1], f.connections[1], f.packets[1], f.slots[1], f.binding, false, ConnectivityOptions{IncludeLoopback: true}) + }() + ticker := time.NewTicker(200 * time.Millisecond) + defer ticker.Stop() + for f.slots[0].Generation() == 0 || f.slots[1].Generation() == 0 { + select { + case <-f.ctx.Done(): + t.Fatal("failed ICE setup never retried") + case <-ticker.C: + exchange() + } + } + awaitPath(t, f.ctx, f.packets[0], f.packets[1], Direct) + exchange() +} + +func TestNativePeerConnectivityRejectsReusedAttempt(t *testing.T) { + f := newConnectivityFixture(t) + ctx, cancel := context.WithCancel(f.ctx) + output := make(chan connectivitySignal, 1) + done := make(chan struct{}) + go func() { defer close(done); readConnectivitySignals(ctx, f.connections[1], f.binding, output) }() + t.Cleanup(func() { cancel(); <-done }) + message := tunnelv1.NativePeerConnectivity{ProtocolVersion: tunnelv1.Version, Kind: "native_peer_connectivity", PairID: f.binding.PairID, DeliveryID: f.binding.DeliveryID, DeliveryGeneration: f.binding.Generation, Attempt: 7, + UsernameFragment: "abcd", Password: strings.Repeat("a", 22), Candidates: []string{"1 1 udp 2130706431 127.0.0.1 1234 typ host"}} + if err := writeConnectivitySignal(ctx, f.connections[0], message); err != nil { + t.Fatal(err) + } + if _, err := nextConnectivitySignal(ctx, output); err != nil { + t.Fatal(err) + } + if err := writeConnectivitySignal(ctx, f.connections[0], message); err != nil { + t.Fatal(err) + } + _, err := nextConnectivitySignal(ctx, output) + var classified *ConnectivityError + if !errors.As(err, &classified) || classified == nil || classified.Code != tunnelv1.InvalidMessage { + t.Fatal("replayed ICE attempt was accepted", err) + } +} + +func TestNativePeerConnectivityAcceptsChangedPeerUDPAddressOnSameInnerConnection(t *testing.T) { + f := newConnectivityFixture(t) + exchange := connectivityCounter(t, f) + ctx, cancel := context.WithCancel(f.ctx) + var results [2]chan error + var stop [2]context.CancelFunc + var replacement *Socket + start := func(side int, socket *Socket) { + life, finish := context.WithCancel(ctx) + stop[side] = finish + results[side] = make(chan error, 1) + result := results[side] + go func() { + result <- ServeConnectivity(life, socket, f.connections[side], f.packets[side], f.slots[side], f.binding, side == 0, ConnectivityOptions{IncludeLoopback: true}) + }() + } + start(0, f.sockets[0]) + start(1, f.sockets[1]) + t.Cleanup(func() { + cancel() + for side, result := range results { + if result == nil { + continue + } + select { + case err := <-result: + if err != nil && !errors.Is(err, context.Canceled) { + var failure *ConnectivityError + if errors.As(err, &failure) && failure != nil { + t.Errorf("remapped peer side=%d code=%s cause=%T: %v", side, failure.Code, failure.Cause, failure.Cause) + } else { + t.Error(err) + } + } + case <-time.After(time.Second): + t.Error("remapped peer did not join") + } + } + if replacement != nil { + _ = replacement.Close() + } + }) + exchange() + awaitPath(t, f.ctx, f.packets[0], f.packets[1], Direct) + exchange() + stop[1]() + select { + case err := <-results[1]: + results[1] = nil + if err != nil && !errors.Is(err, context.Canceled) { + t.Fatal(err) + } + case <-time.After(time.Second): + t.Fatal("old UDP agent did not join") + } + // only the remote UDP mapping changes. retain the exact inner connection, + // packet endpoint, delivery binding and already-open target socket. + var err error + replacement, err = ListenSocket(ctx, "udp4", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + if replacement.LocalAddr().String() == f.sockets[1].LocalAddr().String() { + t.Fatal("UDP address did not change") + } + start(1, replacement) + if err := f.slots[0].Install(nil); err != nil { + t.Fatal(err) + } + exchange() + for f.slots[1].Generation() < 3 { + select { + case <-f.ctx.Done(): + t.Fatal("changed UDP address did not install") + case <-time.After(10 * time.Millisecond): + } + } + awaitPath(t, f.ctx, f.packets[0], f.packets[1], Direct) + exchange() +} + func TestNativePeerConnectivityRejectsCrossBindingAndBadCandidates(t *testing.T) { for name, test := range map[string]struct { change func(*tunnelv1.NativePeerConnectivity) diff --git a/pkg/protocol/tunnelv1/native_peer_connectivity.go b/pkg/protocol/tunnelv1/native_peer_connectivity.go index 62817282c..63a60d4dd 100644 --- a/pkg/protocol/tunnelv1/native_peer_connectivity.go +++ b/pkg/protocol/tunnelv1/native_peer_connectivity.go @@ -15,15 +15,16 @@ const ( var ErrNativePeerConnectivity = errors.New("native peer connectivity message is invalid") -// NativePeerConnectivity travels on the one dedicated unidirectional stream in -// each direction of the existing inner QUIC connection. it grants no application -// permission and carries no source-IP authority or renewable deadline. +// NativePeerConnectivity travels on a dedicated unidirectional stream for each +// attempt in each direction of the existing inner QUIC connection. it grants no +// application permission and carries no source-IP authority or renewable deadline. type NativePeerConnectivity struct { ProtocolVersion uint32 `json:"protocol_version"` Kind string `json:"kind"` PairID string `json:"pair_id"` DeliveryID string `json:"delivery_id"` DeliveryGeneration int64 `json:"delivery_generation"` + Attempt uint64 `json:"attempt"` UsernameFragment string `json:"username_fragment"` Password string `json:"password"` Candidates []string `json:"candidates"` @@ -33,7 +34,7 @@ func (NativePeerConnectivity) String() string { return "native peer connectivi func (NativePeerConnectivity) GoString() string { return "native peer connectivity" } func (m NativePeerConnectivity) Validate() error { - if m.ProtocolVersion != Version || m.Kind != "native_peer_connectivity" || !opaqueid.Valid(m.PairID, "npp_") || m.DeliveryGeneration < 1 { + if m.ProtocolVersion != Version || m.Kind != "native_peer_connectivity" || !opaqueid.Valid(m.PairID, "npp_") || m.DeliveryGeneration < 1 || m.Attempt == 0 { return ErrNativePeerConnectivity } if _, err := ParsePeerDeliveryIdentifier(m.DeliveryID); err != nil { diff --git a/pkg/protocol/tunnelv1/native_peer_connectivity_test.go b/pkg/protocol/tunnelv1/native_peer_connectivity_test.go index 46abf536e..d58f9b680 100644 --- a/pkg/protocol/tunnelv1/native_peer_connectivity_test.go +++ b/pkg/protocol/tunnelv1/native_peer_connectivity_test.go @@ -12,7 +12,7 @@ import ( func connectivityFixture() NativePeerConnectivity { return NativePeerConnectivity{ProtocolVersion: Version, Kind: "native_peer_connectivity", - PairID: "npp_1234567890123456789012", DeliveryID: "AQEBAQEBAQEBAQEBAQEBAQ", DeliveryGeneration: 1, + PairID: "npp_1234567890123456789012", DeliveryID: "AQEBAQEBAQEBAQEBAQEBAQ", DeliveryGeneration: 1, Attempt: 1, UsernameFragment: "abcd", Password: "abcdefghijklmnopqrstuv", Candidates: []string{"1 1 udp 2130706431 192.0.2.1 4443 typ host"}} } @@ -37,6 +37,7 @@ func TestNativePeerConnectivityRejectsInvalidAndBounds(t *testing.T) { "pair": func(m *NativePeerConnectivity) { m.PairID = "npp_invalid" }, "delivery": func(m *NativePeerConnectivity) { m.DeliveryID = strings.Repeat("A", 22) }, "generation": func(m *NativePeerConnectivity) { m.DeliveryGeneration = 0 }, + "attempt": func(m *NativePeerConnectivity) { m.Attempt = 0 }, "ufrag-short": func(m *NativePeerConnectivity) { m.UsernameFragment = "abc" }, "ufrag-oversize": func(m *NativePeerConnectivity) { m.UsernameFragment = strings.Repeat("a", 257) }, "password-short": func(m *NativePeerConnectivity) { m.Password = "short" }, diff --git a/pkg/protocol/tunnelv1/testdata/native-peer-connectivity.frame.hex b/pkg/protocol/tunnelv1/testdata/native-peer-connectivity.frame.hex index 8403a7334..33e9927a1 100644 --- a/pkg/protocol/tunnelv1/testdata/native-peer-connectivity.frame.hex +++ b/pkg/protocol/tunnelv1/testdata/native-peer-connectivity.frame.hex @@ -1 +1 @@ -000001197b2270726f746f636f6c5f76657273696f6e223a312c226b696e64223a226e61746976655f706565725f636f6e6e6563746976697479222c22706169725f6964223a226e70705f31323334353637383930313233343536373839303132222c2264656c69766572795f6964223a2241514542415145424151454241514542415145424151222c2264656c69766572795f67656e65726174696f6e223a312c22757365726e616d655f667261676d656e74223a2261626364222c2270617373776f7264223a226162636465666768696a6b6c6d6e6f70717273747576222c2263616e64696461746573223a5b22312031207564702032313330373036343331203139322e302e322e3120343434332074797020686f7374225d7d +000001257b2270726f746f636f6c5f76657273696f6e223a312c226b696e64223a226e61746976655f706565725f636f6e6e6563746976697479222c22706169725f6964223a226e70705f31323334353637383930313233343536373839303132222c2264656c69766572795f6964223a2241514542415145424151454241514542415145424151222c2264656c69766572795f67656e65726174696f6e223a312c22617474656d7074223a312c22757365726e616d655f667261676d656e74223a2261626364222c2270617373776f7264223a226162636465666768696a6b6c6d6e6f70717273747576222c2263616e64696461746573223a5b22312031207564702032313330373036343331203139322e302e322e3120343434332074797020686f7374225d7d diff --git a/pkg/protocol/tunnelv1/testdata/native-peer-connectivity.json b/pkg/protocol/tunnelv1/testdata/native-peer-connectivity.json index 254cfe273..88a98e806 100644 --- a/pkg/protocol/tunnelv1/testdata/native-peer-connectivity.json +++ b/pkg/protocol/tunnelv1/testdata/native-peer-connectivity.json @@ -4,6 +4,7 @@ "pair_id": "npp_1234567890123456789012", "delivery_id": "AQEBAQEBAQEBAQEBAQEBAQ", "delivery_generation": 1, + "attempt": 1, "username_fragment": "abcd", "password": "abcdefghijklmnopqrstuv", "candidates": ["1 1 udp 2130706431 192.0.2.1 4443 typ host"] diff --git a/tools/vercel-function-check/invoke.ts b/tools/vercel-function-check/invoke.ts index bd1576bc1..b321935a3 100644 --- a/tools/vercel-function-check/invoke.ts +++ b/tools/vercel-function-check/invoke.ts @@ -44,6 +44,7 @@ const statsSchema = z.object({ prepared: z.array(z.string()), targets: z.number().int(), calls: z.record(z.string(), z.number().int()), + blocked: z.record(z.string(), z.number().int()), }); interface Host { child: ChildProcess; @@ -89,30 +90,36 @@ try { const prepared = await stats(); assert.equal(prepared.sessions, 1); assert.deepEqual(prepared.prepared, ["api", "database"]); + // observe periodic run-plus-token preparation without application arrivals, + // then hold all control requests through the warm IPC and target exchanges. + await controlCall("background-refresh"); + const idle = await stats(); + assert( + (idle.calls["POST /v1/visitor-tokens"] ?? 0) > (prepared.calls["POST /v1/visitor-tokens"] ?? 0), + ); + assert( + Object.entries(idle.calls).some( + ([key, count]) => key.endsWith("/run") && count > (prepared.calls[key] ?? 0), + ), + ); + await controlCall("pause"); + const paused = await stats(); const warmAgain = await Promise.all(Array.from({ length: 4 }, () => request(warm, "prepare"))); assert(warmAgain.every((result) => result.ok)); const warmBudget = await evidence(warm); assert.deepEqual(warmBudget.helpers, first.helpers); assert.deepEqual( (await stats()).calls, - prepared.calls, - "verified warm IPC must add no control work", + paused.calls, + "verified warm IPC must complete while control is held", ); assert.equal(warmBudget.ipc.length - both.ipc.length, 8); const warmTraffic = await Promise.all(Array.from({ length: 4 }, () => request(warm, "both"))); assert(warmTraffic.every((result) => result.ok)); - const trafficCalls = (await stats()).calls; - let streamControlDelta = 0; - for (const [key, count] of Object.entries(trafficCalls)) { - const delta = count - (prepared.calls[key] ?? 0); - // warm visitors use prepared local selection. background run refresh may - // overlap these requests, but each visitor opens without a control call. - if (key.endsWith("/run")) { - assert(delta >= 0 && delta <= 4); - streamControlDelta += delta; - } else assert.equal(delta, 0, `unexpected warm control work: ${key}`); - } - assert(streamControlDelta <= 4, "warm visitor traffic triggered excess control work"); + const traffic = await stats(); + assert.deepEqual(traffic.calls, paused.calls, "warm traffic must complete while control is held"); + assert(traffic.targets > paused.targets, "warm traffic must reach the real targets"); + await controlCall("resume"); const rotated = await request(warm, "both", tokens.rotated); assert.equal(rotated.ok, true); @@ -204,7 +211,8 @@ try { coldHelpers: first.helpers, warmHelperDelta: warmBudget.helpers.length - first.helpers.length, warmPrepareControlDelta: 0, - warmStreamControlDelta: streamControlDelta, + warmStreamControlDelta: 0, + warmBlockedControl: traffic.blocked, warmIPCDelta: warmBudget.ipc.length - both.ipc.length, rotationHelperDelta: rotation.helpers.length - warmBudget.helpers.length, ownersPerIsolate: [first.owners.length, second.owners.length], @@ -213,7 +221,9 @@ try { controlBudgets: { coldHTTP: incremental.calls, both: prepared.calls, - warmTraffic: trafficCalls, + idleRefresh: idle.calls, + paused: paused.calls, + warmTraffic: traffic.calls, }, concurrentContextIsolation: true, negatives, @@ -227,6 +237,7 @@ try { } catch (error) { errors.push(error); } finally { + await controlCall("resume").catch((error: unknown) => errors.push(error)); for (const host of hosts) { try { await fetch(`${host.url}/__fixture/shutdown`, { method: "POST" }).catch(() => {});