Use fabric WebSocket dataplane for Android VPN

agents/rap-node-agent/internal/mesh/server.go

@@ -8,6 +8,7 @@ import (
 	"encoding/binary"
 	"encoding/hex"
 	"encoding/json"
+	"errors"
 	"fmt"
 	"io"
 	"net/http"
@@ -950,12 +951,7 @@ func (s Server) readVPNPacketWebSocket(ctx context.Context, conn *websocket.Conn
 			}
 			continue
 		}
-		var sendErr error
-		if classIngress, ok := s.VPNPacketIngress.(VPNPacketIngressTrafficClass); ok {
-			sendErr = classIngress.SendClientPacketBatchWithTrafficClass(ctx, clusterID, vpnConnectionID, trafficClass, packets)
-		} else {
-			sendErr = s.VPNPacketIngress.SendClientPacketBatch(ctx, clusterID, vpnConnectionID, packets)
-		}
+		sendErr := s.sendVPNPacketWebSocketBatch(ctx, clusterID, vpnConnectionID, trafficClass, packets, !backendFallbackAllowed)
 		if sendErr != nil {
 			if !backendFallbackAllowed {
 				s.logFabricServiceChannelViolation(nil, clusterID, channelID, vpnConnectionID, backendRelayPolicy, "fabric_route_send_failed_backend_fallback_blocked", sendErr.Error())
@@ -968,6 +964,44 @@ func (s Server) readVPNPacketWebSocket(ctx context.Context, conn *websocket.Conn
 	}
 }
 
+func (s Server) sendVPNPacketWebSocketBatch(ctx context.Context, clusterID string, vpnConnectionID string, trafficClass string, packets [][]byte, retryRouteErrors bool) error {
+	const maxAttempts = 6
+	var lastErr error
+	for attempt := 0; attempt < maxAttempts; attempt++ {
+		if err := ctx.Err(); err != nil {
+			return err
+		}
+		var sendErr error
+		if classIngress, ok := s.VPNPacketIngress.(VPNPacketIngressTrafficClass); ok {
+			sendErr = classIngress.SendClientPacketBatchWithTrafficClass(ctx, clusterID, vpnConnectionID, trafficClass, packets)
+		} else {
+			sendErr = s.VPNPacketIngress.SendClientPacketBatch(ctx, clusterID, vpnConnectionID, packets)
+		}
+		if sendErr == nil {
+			return nil
+		}
+		lastErr = sendErr
+		if !retryRouteErrors || !isRetryableVPNPacketIngressError(sendErr) {
+			return sendErr
+		}
+		timer := time.NewTimer(time.Duration(75+attempt*50) * time.Millisecond)
+		select {
+		case <-ctx.Done():
+			timer.Stop()
+			return ctx.Err()
+		case <-timer.C:
+		}
+	}
+	return lastErr
+}
+
+func isRetryableVPNPacketIngressError(err error) bool {
+	return errors.Is(err, ErrRouteNotFound) ||
+		errors.Is(err, ErrForwardRuntimeUnavailable) ||
+		errors.Is(err, ErrForwardPeerUnavailable) ||
+		errors.Is(err, ErrSyntheticPeerUnavailable)
+}
+
 func (s Server) writeVPNPacketWebSocket(ctx context.Context, conn *websocket.Conn, clusterID string, channelID string, vpnConnectionID string, forceBackendFallback bool, backendFallbackAllowed bool, backendRelayPolicy string) error {
 	lastPing := time.Now()
 	for {