feat: restart native ICE without reopening application streams

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.
 

docs/maintainers/transport/peer-packets.md

@@ -156,17 +156,25 @@ 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.
 
 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.

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 {

internal/daemon/workload_vercel_integration_test.go

@@ -103,19 +103,42 @@ 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+"/") {
 				prepared[name] = true
 			}
 		}
 		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 {