Fix SD-WAN connection dropout: isolate mangle to PREROUTING, add conntrack dead-flow reclassification, and prevent probe flapping (v1.8.20)

VERSION

@@ -1 +1 @@
-1.8.19
+1.8.20

internal/routing/manager.go

@@ -896,6 +896,43 @@ func normalizeDst(dst string) string {
 	return dst
 }
 
+func detectGatewayForInterface(ifaceName string) string {
+	link, err := netlink.LinkByName(ifaceName)
+	if err == nil {
+		routes, err := netlink.RouteList(link, netlink.FAMILY_V4)
+		if err == nil {
+			for _, r := range routes {
+				if (r.Dst == nil || r.Dst.IP.IsUnspecified()) && r.Gw != nil && !r.Gw.IsUnspecified() {
+					return r.Gw.String()
+				}
+			}
+			for _, r := range routes {
+				if r.Gw != nil && !r.Gw.IsUnspecified() {
+					return r.Gw.String()
+				}
+			}
+		}
+	}
+	out, err := exec.Command("ip", "route", "show", "dev", ifaceName).CombinedOutput()
+	if err == nil {
+		lines := strings.Split(string(out), "\n")
+		for _, l := range lines {
+			if strings.Contains(l, "via ") {
+				fields := strings.Fields(l)
+				for i, f := range fields {
+					if f == "via" && i+1 < len(fields) {
+						gw := fields[i+1]
+						if net.ParseIP(gw) != nil {
+							return gw
+						}
+					}
+				}
+			}
+		}
+	}
+	return ""
+}
+
 func (m *RoutingManager) syncRouteGroupToKernel(dstAddress, routingTable string, distance int) error {
 	normDst := normalizeDst(dstAddress)
 	tbl := strings.TrimSpace(routingTable)
@@ -985,18 +1022,27 @@ func (m *RoutingManager) syncRouteGroupToKernel(dstAddress, routingTable string,
 			args = []string{"route", "replace", "blackhole", normDst}
 		} else {
 			args = []string{"route", "replace", normDst}
+			iface := strings.TrimSpace(r.Interface)
 			if gw != "" {
 				if net.ParseIP(gw) != nil {
 					args = append(args, "via", gw)
 				} else {
-					args = append(args, "dev", gw)
+					// Gateway field contains an interface name
+					targetDev := gw
+					if iface == "" {
+						iface = targetDev
+					}
+					if detectedGw := detectGatewayForInterface(targetDev); detectedGw != "" {
+						args = append(args, "via", detectedGw)
+					} else {
+						args = append(args, "dev", targetDev)
+					}
 				}
 			}
-			iface := strings.TrimSpace(r.Interface)
-			if iface != "" && gw != iface && (gw == "" || net.ParseIP(gw) != nil) {
+			if iface != "" && (gw == "" || net.ParseIP(gw) != nil || (gw != "" && detectGatewayForInterface(gw) != "")) {
 				args = append(args, "dev", iface)
 			}
-			if gw != "" && iface != "" && net.ParseIP(gw) != nil {
+			if iface != "" && (gw != "" || detectGatewayForInterface(iface) != "") {
 				args = append(args, "onlink")
 			}
 		}
@@ -1028,18 +1074,19 @@ func (m *RoutingManager) syncRouteGroupToKernel(dstAddress, routingTable string,
 	for _, r := range activeRoutes {
 		args = append(args, "nexthop")
 		gw := strings.TrimSpace(r.Gateway)
-		if gw != "" && net.ParseIP(gw) != nil {
-			args = append(args, "via", gw)
-		}
 		iface := strings.TrimSpace(r.Interface)
 		if iface == "" && gw != "" && net.ParseIP(gw) == nil {
 			iface = gw
 		}
-		if iface != "" {
-			args = append(args, "dev", iface)
+		if gw != "" && net.ParseIP(gw) != nil {
+			args = append(args, "via", gw)
+		} else if iface != "" {
+			if detectedGw := detectGatewayForInterface(iface); detectedGw != "" {
+				args = append(args, "via", detectedGw)
+			}
 		}
-		if gw != "" && iface != "" && net.ParseIP(gw) != nil {
-			args = append(args, "onlink")
+		if iface != "" {
+			args = append(args, "dev", iface, "onlink")
 		}
 		args = append(args, "weight", "1")
 	}

internal/sdwan/engine.go

@@ -67,9 +67,9 @@ func (e *KernelEngine) Sync(groups []*SDWANGroup) error {
 	_ = exec.Command("iptables", "-t", "mangle", "-N", "DEBTIK_SDWAN").Run()
 	_ = exec.Command("iptables", "-t", "mangle", "-F", "DEBTIK_SDWAN").Run()
 
-	// Ensure jumps from PREROUTING and OUTPUT
+	// Ensure jumps from PREROUTING only. Local router management, WireGuard tunnel
+	// underlays, routing daemons (FRR/BGP/EIGRP), and health check probes must not be hijacked.
 	e.ensureMangleJump("PREROUTING")
-	e.ensureMangleJump("OUTPUT")
 
 	// 1. Bypass inbound traffic from all member WAN interfaces so return traffic is never classified/marked
 	seenInIfaces := make(map[string]bool)
@@ -91,14 +91,17 @@ func (e *KernelEngine) Sync(groups []*SDWANGroup) error {
 	// 3. Ensure high-priority policy routing rules for private subnets so return traffic and LAN traffic always lookup main table
 	e.ensurePrivateRoutingRules()
 
-	// Restore connection mark: if a flow has already been classified, keep its mark (Session Stickiness!)
-	_ = exec.Command("iptables", "-t", "mangle", "-A", "DEBTIK_SDWAN", "-j", "CONNMARK", "--restore-mark").Run()
-	_ = exec.Command("iptables", "-t", "mangle", "-A", "DEBTIK_SDWAN", "-m", "mark", "!", "--mark", "0", "-j", "RETURN").Run()
-
 	markBase := 0x5d00
 	tableBase := 1100
 	memberCounter := 1
 
+	type groupPlan struct {
+		group      *SDWANGroup
+		activePool []*SDWANMember
+	}
+	var plans []groupPlan
+	var allActiveMembers []*SDWANMember
+
 	for _, g := range groups {
 		if g.Disabled || len(g.Members) == 0 {
 			g.Running = false
@@ -151,7 +154,26 @@ func (e *KernelEngine) Sync(groups []*SDWANGroup) error {
 			}
 		}
 
-		// Apply Mangle classification rules based on group mode
+		plans = append(plans, groupPlan{group: g, activePool: activePool})
+		allActiveMembers = append(allActiveMembers, activePool...)
+	}
+
+	// 4. Restore connection mark:
+	_ = exec.Command("iptables", "-t", "mangle", "-A", "DEBTIK_SDWAN", "-j", "CONNMARK", "--restore-mark").Run()
+
+	// Only return for active, healthy link marks (prevents sticky blackholes on failed links)
+	for _, m := range allActiveMembers {
+		_ = exec.Command("iptables", "-t", "mangle", "-A", "DEBTIK_SDWAN", "-m", "mark", "--mark", m.MarkHex, "-j", "RETURN").Run()
+	}
+
+	// If mark does not belong to any currently active link (new connection or failed link), reset to 0 to reclassify:
+	_ = exec.Command("iptables", "-t", "mangle", "-A", "DEBTIK_SDWAN", "-j", "MARK", "--set-mark", "0").Run()
+
+	// 5. Apply Mangle classification rules based on group mode
+	for _, p := range plans {
+		g := p.group
+		activePool := p.activePool
+
 		switch g.Mode {
 		case ModeWeighted:
 			e.applyWeightedRules(activePool)
@@ -182,16 +204,20 @@ func (e *KernelEngine) applyMemberRoute(m *SDWANMember) {
 	// Flush table first
 	_ = exec.Command("ip", "route", "flush", "table", tableStr).Run()
 
-	gw := m.Gateway
+	gw := strings.TrimSpace(m.Gateway)
 	if gw == "" {
 		gw = e.detectInterfaceGateway(m.Interface)
 	}
 
 	args := []string{"route", "replace", "default"}
-	if gw != "" {
+	if gw != "" && net.ParseIP(gw) != nil {
 		args = append(args, "via", gw)
 	}
-	args = append(args, "dev", m.Interface, "table", tableStr)
+	args = append(args, "dev", m.Interface)
+	if gw != "" && net.ParseIP(gw) != nil {
+		args = append(args, "onlink")
+	}
+	args = append(args, "table", tableStr)
 
 	if out, err := exec.Command("ip", args...).CombinedOutput(); err != nil {
 		log.Printf("[SD-WAN] Warning: failed adding route for %s (table %s): %s (%v)", m.Interface, tableStr, strings.TrimSpace(string(out)), err)
@@ -200,16 +226,40 @@ func (e *KernelEngine) applyMemberRoute(m *SDWANMember) {
 
 func (e *KernelEngine) detectInterfaceGateway(ifaceName string) string {
 	link, err := netlink.LinkByName(ifaceName)
-	if err != nil {
-		return ""
-	}
-	routes, err := netlink.RouteList(link, netlink.FAMILY_V4)
-	if err != nil {
-		return ""
+	if err == nil {
+		routes, err := netlink.RouteList(link, netlink.FAMILY_V4)
+		if err == nil {
+			// Prefer default route gateway first
+			for _, r := range routes {
+				if (r.Dst == nil || r.Dst.IP.IsUnspecified()) && r.Gw != nil && !r.Gw.IsUnspecified() {
+					return r.Gw.String()
+				}
+			}
+			// Fallback to any non-empty gateway on this link
+			for _, r := range routes {
+				if r.Gw != nil && !r.Gw.IsUnspecified() {
+					return r.Gw.String()
+				}
+			}
+		}
 	}
-	for _, r := range routes {
-		if r.Gw != nil && !r.Gw.IsUnspecified() {
-			return r.Gw.String()
+
+	// Also check ip route show dev <ifaceName> via CLI as fallback
+	out, err := exec.Command("ip", "route", "show", "dev", ifaceName).CombinedOutput()
+	if err == nil {
+		lines := strings.Split(string(out), "\n")
+		for _, l := range lines {
+			if strings.Contains(l, "via ") {
+				fields := strings.Fields(l)
+				for i, f := range fields {
+					if f == "via" && i+1 < len(fields) {
+						gw := fields[i+1]
+						if net.ParseIP(gw) != nil {
+							return gw
+						}
+					}
+				}
+			}
 		}
 	}
 	return ""
@@ -275,11 +325,11 @@ func (e *KernelEngine) applyPCCRules(pool []*SDWANMember) {
 		return
 	}
 
-	// Using nth packet mode for new connection distribution
+	// Using nth packet mode for unclassified / failed-flow distribution (mark 0)
 	for i, m := range pool {
 		args := []string{
 			"-t", "mangle", "-A", "DEBTIK_SDWAN",
-			"-m", "conntrack", "--ctstate", "NEW",
+			"-m", "mark", "--mark", "0",
 			"-m", "statistic", "--mode", "nth",
 			"--every", strconv.Itoa(n - i),
 			"--packet", "0",
@@ -289,12 +339,10 @@ func (e *KernelEngine) applyPCCRules(pool []*SDWANMember) {
 
 		// Save mark into conntrack
 		_ = exec.Command("iptables", "-t", "mangle", "-A", "DEBTIK_SDWAN",
-			"-m", "conntrack", "--ctstate", "NEW",
 			"-m", "mark", "--mark", m.MarkHex,
 			"-j", "CONNMARK", "--save-mark").Run()
 
 		_ = exec.Command("iptables", "-t", "mangle", "-A", "DEBTIK_SDWAN",
-			"-m", "conntrack", "--ctstate", "NEW",
 			"-m", "mark", "--mark", m.MarkHex,
 			"-j", "RETURN").Run()
 	}
@@ -344,23 +392,21 @@ func (e *KernelEngine) applyFailoverRules(pool []*SDWANMember) {
 func (e *KernelEngine) addMarkRule(markHex string, prob float64) {
 	if prob >= 1.0 {
 		_ = exec.Command("iptables", "-t", "mangle", "-A", "DEBTIK_SDWAN",
-			"-m", "conntrack", "--ctstate", "NEW",
+			"-m", "mark", "--mark", "0",
 			"-j", "MARK", "--set-mark", markHex).Run()
 	} else {
 		probStr := fmt.Sprintf("%.4f", prob)
 		_ = exec.Command("iptables", "-t", "mangle", "-A", "DEBTIK_SDWAN",
-			"-m", "conntrack", "--ctstate", "NEW",
+			"-m", "mark", "--mark", "0",
 			"-m", "statistic", "--mode", "random", "--probability", probStr,
 			"-j", "MARK", "--set-mark", markHex).Run()
 	}
 
 	_ = exec.Command("iptables", "-t", "mangle", "-A", "DEBTIK_SDWAN",
-		"-m", "conntrack", "--ctstate", "NEW",
 		"-m", "mark", "--mark", markHex,
 		"-j", "CONNMARK", "--save-mark").Run()
 
 	_ = exec.Command("iptables", "-t", "mangle", "-A", "DEBTIK_SDWAN",
-		"-m", "conntrack", "--ctstate", "NEW",
 		"-m", "mark", "--mark", markHex,
 		"-j", "RETURN").Run()
 }

internal/sdwan/health.go

@@ -106,15 +106,20 @@ func (h *HealthMonitor) runProbeIteration(groups []*SDWANGroup) {
 			// Calculate telemetry
 			m.LastChecked = time.Now()
 			if success {
+				m.ConsecutiveSuccesses++
 				m.ConsecutiveFails = 0
 			} else {
 				m.ConsecutiveFails++
+				m.ConsecutiveSuccesses = 0
 			}
 
 			h.calculateMetrics(m, history)
 
-			// Determine status based on SLA thresholds
-			if m.ConsecutiveFails >= 3 {
+			// Determine status based on SLA thresholds with anti-flapping hysteresis
+			if m.ConsecutiveFails >= 5 {
+				m.Status = StatusDown
+			} else if prevStatus == StatusDown && m.ConsecutiveSuccesses < 2 {
+				// Require 2 consecutive successful probe cycles before recovering from DOWN
 				m.Status = StatusDown
 			} else {
 				maxLat := float64(m.MaxLatency)
@@ -149,15 +154,21 @@ func (h *HealthMonitor) runProbeIteration(groups []*SDWANGroup) {
 }
 
 func (h *HealthMonitor) executeProbe(iface, target string) (float64, bool) {
-	// ping -c 1 -W 2 -I <iface> <target>
-	cmd := exec.Command("ping", "-c", "1", "-W", "2", "-I", iface, target)
+	// Sanitize probe target: fallback if empty, ending in .0, or known rate-limited anycast target
+	cleanTarget := strings.TrimSpace(target)
+	if cleanTarget == "" || cleanTarget == "188.114.97.0" || strings.HasSuffix(cleanTarget, ".0") {
+		cleanTarget = "1.1.1.1"
+	}
+
+	// ping -c 2 -W 2 -i 0.2 -I <iface> <cleanTarget>
+	cmd := exec.Command("ping", "-c", "2", "-W", "2", "-i", "0.2", "-I", iface, cleanTarget)
 	out, err := cmd.CombinedOutput()
 	if err != nil {
 		return 0, false
 	}
 
 	str := string(out)
-	if !strings.Contains(str, "1 received") && !strings.Contains(str, "1 packets received") {
+	if strings.Contains(str, "100% packet loss") || strings.Contains(str, " 0 received") || strings.Contains(str, "0 packets received") {
 		return 0, false
 	}
 

internal/sdwan/sdwan_test.go

@@ -146,6 +146,58 @@ func TestHealthMetricsCalculation(t *testing.T) {
 	}
 }
 
+func TestHealthHysteresisAndThresholds(t *testing.T) {
+	mem := &SDWANMember{
+		Interface:  "eth0",
+		MaxLatency: 250,
+		MaxLoss:    20,
+		Status:     StatusHealthy,
+	}
+
+	// 1-4 fails: should NOT transition to StatusDown
+	for i := 1; i <= 4; i++ {
+		mem.ConsecutiveFails = i
+		mem.ConsecutiveSuccesses = 0
+		if mem.ConsecutiveFails >= 5 {
+			mem.Status = StatusDown
+		}
+		if mem.Status == StatusDown {
+			t.Fatalf("member became DOWN prematurely at %d fails", i)
+		}
+	}
+
+	// 5th fail: transitions to StatusDown
+	mem.ConsecutiveFails = 5
+	if mem.ConsecutiveFails >= 5 {
+		mem.Status = StatusDown
+	}
+	if mem.Status != StatusDown {
+		t.Fatalf("expected StatusDown after 5 fails, got %s", mem.Status)
+	}
+
+	// 1st recovery success: hysteresis must keep it StatusDown
+	prevStatus := mem.Status
+	mem.ConsecutiveSuccesses = 1
+	mem.ConsecutiveFails = 0
+	if prevStatus == StatusDown && mem.ConsecutiveSuccesses < 2 {
+		mem.Status = StatusDown
+	}
+	if mem.Status != StatusDown {
+		t.Fatalf("expected StatusDown during 1st success cycle (hysteresis), got %s", mem.Status)
+	}
+
+	// 2nd recovery success: now allowed to recover
+	mem.ConsecutiveSuccesses = 2
+	if prevStatus == StatusDown && mem.ConsecutiveSuccesses < 2 {
+		mem.Status = StatusDown
+	} else {
+		mem.Status = StatusHealthy
+	}
+	if mem.Status != StatusHealthy {
+		t.Fatalf("expected StatusHealthy after 2 consecutive successes, got %s", mem.Status)
+	}
+}
+
 func TestSDWANVirtualDeviceLifecycle(t *testing.T) {
 	tmpDir := t.TempDir()
 	storePath := filepath.Join(tmpDir, "sdwan.json")

internal/sdwan/types.go

@@ -42,8 +42,9 @@ type SDWANMember struct {
 	PacketLoss       float64      `json:"packet_loss"`       // Packet loss percentage (0.0 - 100.0)
 	Jitter           float64      `json:"jitter"`            // Latency jitter in ms
 	LastChecked      time.Time    `json:"last_checked"`
-	ConsecutiveFails int          `json:"consecutive_fails"` // Counter for sequential probe failures
-	TableID          int          `json:"table_id"`          // Kernel routing table ID (e.g. 1101)
+	ConsecutiveFails     int          `json:"consecutive_fails"`     // Counter for sequential probe failures
+	ConsecutiveSuccesses int          `json:"consecutive_successes"` // Counter for sequential probe successes
+	TableID              int          `json:"table_id"`              // Kernel routing table ID (e.g. 1101)
 	MarkHex          string       `json:"mark_hex"`          // Fwmark hex string (e.g. "0x5d01")
 	MarkInt          uint32       `json:"mark_int"`          // Fwmark integer
 }