diff --git a/.github/workflows/build.yml b/.github/workflows/build.yml index 2592fefc..657a2884 100644 --- a/.github/workflows/build.yml +++ b/.github/workflows/build.yml @@ -8,6 +8,11 @@ # # This is a open-source software, liscensed under the AGPL-3.0 License. # See /License for more information. +# +# NOTE: This build.yml is SELF-CONTAINED and intentionally does NOT depend on +# daeuniverse/ci-seed-jobs (pre-actions / post-actions). It calls the local +# seed-build.yml directly. Keep it that way so the workflow runs without the +# external ci-seed-jobs repo. name: Build (Main) @@ -27,28 +32,9 @@ on: - "Makefile" jobs: - pre-actions: - uses: daeuniverse/ci-seed-jobs/.github/workflows/pre-actions.yml@master - with: - repository: ${{ github.repository }} - ref: ${{ github.sha }} - fetch-depth: 0 - check-runs: '["build", "main-build-passed"]' - secrets: inherit - build: - # uses: daeuniverse/dae/.github/workflows/seed-build.yml@main uses: ./.github/workflows/seed-build.yml with: ref: ${{ github.sha }} build-type: main-build secrets: inherit - - post-actions: - if: always() - needs: [build] - uses: daeuniverse/ci-seed-jobs/.github/workflows/dae-post-actions.yml@master - with: - check-run-id: "dae-bot[bot]/main-build-passed" - check-run-conclusion: ${{ needs.build.result }} - secrets: inherit diff --git a/.github/workflows/seed-build.yml b/.github/workflows/seed-build.yml index e3a846cd..caccf15f 100644 --- a/.github/workflows/seed-build.yml +++ b/.github/workflows/seed-build.yml @@ -27,22 +27,13 @@ on: jobs: build: + # Restricted matrix: only amd64 (v1/v2/v3) + arm64 for kdae-custom fork + # to keep CI fast. Expand if other targets are needed. strategy: matrix: goos: [ linux ] - goarch: [ arm64, 386, riscv64, loong64, mips64, mips64le, mipsle, mips, ppc64, ppc64le, s390x ] + goarch: [ arm64 ] include: - # BEGIN Linux ARM 5 6 7 - - goos: linux - goarch: arm - goarm: 7 - - goos: linux - goarch: arm - goarm: 6 - - goos: linux - goarch: arm - goarm: 5 - # END Linux ARM 5 6 7 # BEGIN Linux AMD64 v1 v2 v3 - goos: linux goarch: amd64 @@ -54,17 +45,6 @@ jobs: goarch: amd64 goamd64: v3 # END Linux AMD64 v1 v2 v3 - # BEGIN Linux RISCV64 rva20u64 rva22u64 rva23u64 - - goos: linux - goarch: riscv64 - goriscv64: rva20u64 - - goos: linux - goarch: riscv64 - goriscv64: rva22u64 - - goos: linux - goarch: riscv64 - goriscv64: rva23u64 - # END Linux RISCV64 rva20u64 rva22u64 rva23u64 fail-fast: false runs-on: ubuntu-22.04 diff --git a/common/consts/dialer.go b/common/consts/dialer.go index cef35b4a..2ad6991f 100644 --- a/common/consts/dialer.go +++ b/common/consts/dialer.go @@ -26,6 +26,10 @@ const ( DialerSelectionPolicy_Random DialerSelectionPolicy = "random" // DialerSelectionPolicy_Fixed always selects the first dialer. DialerSelectionPolicy_Fixed DialerSelectionPolicy = "fixed" + // DialerSelectionPolicy_FixedWithFallback always selects the n-th dialer when alive; + // falls back to min_moving_avg among other alive dialers when dead. + // When the fixed dialer revives, traffic will automatically return to it. + DialerSelectionPolicy_FixedWithFallback DialerSelectionPolicy = "fixed_fallback" // DialerSelectionPolicy_MinAverage10Latencies selects the dialer with minimum average latency of last 10 checks. DialerSelectionPolicy_MinAverage10Latencies DialerSelectionPolicy = "min_avg10" // DialerSelectionPolicy_MinMovingAverageLatencies selects the dialer with minimum moving average latency. diff --git a/component/outbound/dialer/connectivity_check.go b/component/outbound/dialer/connectivity_check.go index ca959378..a6fd9743 100644 --- a/component/outbound/dialer/connectivity_check.go +++ b/component/outbound/dialer/connectivity_check.go @@ -151,6 +151,29 @@ func (d *Dialer) MustGetAlive(typ *NetworkType) bool { return d.mustGetCollection(typ).Alive.Load() } +// AliveForRetry reports whether the dialer should be considered alive for +// fixed_fallback retry/fallback decisions on network type typ. +// +// When udp_check_dns is not configured, the DNS-UDP collection is never probed +// by the health-check loop and its Alive flag stays at its initial value +// (true) forever. That false-positive "alive" would make a Select() over +// DNS-UDP reset the fixed_fallback retry counter on every DNS resolution +// cycle, so fallback never triggers even though the node's TCP — and thus the +// node itself — is dead. +// +// To fix that without coupling the DNS-UDP and TCP collections (which would +// break UDP health-domain independence and snapshot/restore semantics, since +// the two network types must remain independently markable), we mirror the +// liveness decision to the same IP family's TCP collection only at this +// retry-decision site. +func (d *Dialer) AliveForRetry(typ *NetworkType) bool { + if typ != nil && typ.L4Proto == consts.L4ProtoStr_UDP && typ.IsDns && len(d.CheckDnsOptionRaw.Raw) == 0 { + mirrored := &NetworkType{L4Proto: consts.L4ProtoStr_TCP, IpVersion: typ.IpVersion} + return d.MustGetAlive(mirrored) + } + return d.MustGetAlive(typ) +} + func (d *Dialer) SnapshotLastProbe(typ *NetworkType) DialerProbeObservationSnapshot { if d == nil || typ == nil { return DialerProbeObservationSnapshot{} @@ -466,6 +489,31 @@ func releaseConnectivityCheckDialer() { } } +// shouldSkipIpFamily6 returns true when raw explicitly lists only IPv4 addresses +// (no explicit IPv6 entries). This avoids unnecessary IPv6 probes when the user's +// network doesn't support IPv6. +// Returns false (keep IPv6 probes) when: +// - Explicit IPv6 addresses are found in config +// - No explicit IPs are given (DNS resolution might return IPv6) +func shouldSkipIpFamily6(raw []string) bool { + hasIpv6 := false + hasExplicitIpv4 := false + + for i := 1; i < len(raw); i++ { + addr, err := netip.ParseAddr(raw[i]) + if err != nil { + continue + } + if addr.Is6() { + hasIpv6 = true + } else { + hasExplicitIpv4 = true + } + } + + return hasExplicitIpv4 && !hasIpv6 +} + func getActiveDialerCount() int { poolMu.Lock() defer poolMu.Unlock() @@ -473,6 +521,13 @@ func getActiveDialerCount() int { } func (d *Dialer) aliveBackground() { + // If check_interval is 0 or not configured, skip connectivity check entirely + if d.CheckInterval == 0 { + d.Log.WithField("dialer", d.Property().Name). + Warnln("Connectivity check disabled (check_interval=0)") + return + } + cycle := d.CheckInterval var tcpSomark uint32 var mptcp bool @@ -571,7 +626,45 @@ func (d *Dialer) aliveBackground() { }, CheckFunc: makeDnsCheckFunc(func(o *CheckDnsOption) netip.Addr { return o.Ip6 }, &udpNetwork), } - var CheckOpts = []*CheckOption{tcp4CheckOpt, tcp6CheckOpt, udp4CheckDnsOpt, udp6CheckDnsOpt} + // Build CheckOpts dynamically based on configuration: + // - Only add TCP checks if tcp_check_url is configured + // - Only add UDP DNS checks if udp_check_dns is configured + // - Skip IPv6 probes when only IPv4 addresses are explicitly configured + useTcpCheck := len(d.TcpCheckOptionRaw.Raw) > 0 + useUdpDns := len(d.CheckDnsOptionRaw.Raw) > 0 + skipTcp6 := useTcpCheck && shouldSkipIpFamily6(d.TcpCheckOptionRaw.Raw) + skipUdp6 := useUdpDns && shouldSkipIpFamily6(d.CheckDnsOptionRaw.Raw) + + var CheckOpts []*CheckOption + if useTcpCheck { + CheckOpts = append(CheckOpts, tcp4CheckOpt) + if !skipTcp6 { + CheckOpts = append(CheckOpts, tcp6CheckOpt) + } + } + if useUdpDns { + CheckOpts = append(CheckOpts, udp4CheckDnsOpt) + if !skipUdp6 { + CheckOpts = append(CheckOpts, udp6CheckDnsOpt) + } + } + + // If neither TCP nor UDP checks are configured, return early + if len(CheckOpts) == 0 { + d.Log.WithField("dialer", d.Property().Name). + Warnln("No connectivity checks configured, skipping") + return + } + + if d.Log.IsLevelEnabled(logrus.DebugLevel) { + d.Log.WithFields(logrus.Fields{ + "dialer": d.property.Name, + "tcp4": useTcpCheck, + "tcp6": useTcpCheck && !skipTcp6, + "udp4_dns": useUdpDns, + "udp6_dns": useUdpDns && !skipUdp6, + }).Debugln("Connectivity check probes configured") + } var unusedOnce bool checkUnused := func() bool { @@ -685,13 +778,23 @@ func (d *Dialer) aliveBackground() { case <-waitDone: case <-d.ctx.Done(): return + case <-time.After(cycle + 5*time.Second): + // Probe(s) appear stuck — log diagnostic and continue. + // The stuck probe will eventually resolve, but we don't block + // the entire check cycle waiting for it. + if d.Log != nil { + d.Log.WithField("dialer", d.Property().Name). + Warnln("Health check probe appears stuck; continuing cycle") + } } if checkFamily == "" { // Stability-based wash white: only reset stability if a protocol family had failures // WITHOUT any successes in this cycle. This allows partially-working dual-stack // nodes (e.g. V4 OK, V6 broken) to eventually wash white their penalty. d.NotifyPeriodicCheckResult(consts.L4ProtoStr_TCP, cycleRes.tcpSuccess, cycleRes.tcpFailure && !cycleRes.tcpSuccess) - d.NotifyPeriodicCheckResultForType(udp4CheckDnsOpt.networkType, cycleRes.udpSuccess, cycleRes.udpFailure && !cycleRes.udpSuccess) + if useUdpDns { + d.NotifyPeriodicCheckResultForType(udp4CheckDnsOpt.networkType, cycleRes.udpSuccess, cycleRes.udpFailure && !cycleRes.udpSuccess) + } } // Targeted checks don't disturb the periodic timer — only full checks do. @@ -967,6 +1070,25 @@ func (d *Dialer) markUnavailableInternal(typ *NetworkType, force bool, isTraffic wasAlive := collection.Alive.Load() collection.Alive.Store(alive) + // Log alive/dead transitions for operational visibility. + if d.Log != nil { + nodeName := "" + if d.property != nil { + nodeName = d.property.Name + } + if wasAlive && !alive { + d.Log.WithFields(logrus.Fields{ + "dialer": nodeName, + "network": typ.String(), + }).Warnln("Node became DEAD") + } else if !wasAlive && alive { + d.Log.WithFields(logrus.Fields{ + "dialer": nodeName, + "network": typ.String(), + }).Infoln("Node became ALIVE") + } + } + update := collectionUpdate{ alive: alive, movingAverage: collection.MovingAverage, @@ -1015,6 +1137,18 @@ func (d *Dialer) markAvailable(typ *NetworkType, latency time.Duration) (collect isRevival := !wasAlive d.NotifyHealthCheckResult(typ, true, isRevival) if isRevival { + // Log node revival for operational visibility. + if d.Log != nil { + nodeName := "" + if d.property != nil { + nodeName = d.property.Name + } + d.Log.WithFields(logrus.Fields{ + "dialer": nodeName, + "network": typ.String(), + "latency": latency.String(), + }).Infoln("Node became ALIVE") + } d.notifyAliveTransition(typ, true) } @@ -1040,6 +1174,17 @@ func (d *Dialer) markAvailableTraffic(typ *NetworkType) collectionUpdate { d.NotifyHealthCheckResult(typ, true, isRevival) if isRevival { d.notifyAliveTransition(typ, true) + // Log dead→alive transitions for operational visibility. + if d.Log != nil { + nodeName := "" + if d.property != nil { + nodeName = d.property.Name + } + d.Log.WithFields(logrus.Fields{ + "dialer": nodeName, + "network": typ.String(), + }).Infoln("Node became ALIVE (traffic)") + } } return update } diff --git a/component/outbound/dialer/dialer.go b/component/outbound/dialer/dialer.go index bc0c0f72..eccdce13 100644 --- a/component/outbound/dialer/dialer.go +++ b/component/outbound/dialer/dialer.go @@ -327,6 +327,14 @@ func (d *Dialer) CloneWithGlobalOptionContext(ctx context.Context, option *Globa return clone } +// Done returns a channel that is closed when the dialer is shut down +// (via Close or a reload that replaces it). It lets callers outside the +// dialer package observe cancellation without reaching into the unexported +// ctx field. +func (d *Dialer) Done() <-chan struct{} { + return d.ctx.Done() +} + // RetireForEstablishedFlows releases control-plane health state while keeping // the underlying transport available to already-established connections. func (d *Dialer) RetireForEstablishedFlows() { diff --git a/component/outbound/dialer_group.go b/component/outbound/dialer_group.go index 6d8cea52..a9e46c98 100644 --- a/component/outbound/dialer_group.go +++ b/component/outbound/dialer_group.go @@ -40,6 +40,31 @@ type DialerGroup struct { resuscitateLastTime atomic.Int64 noAliveLogLastTimes [8]atomic.Int64 + // fixed_fallback retry state (protected by fixedFallbackMu) + fixedFallbackMu sync.Mutex + fixedFallbackDeadSince int64 + fixedFallbackRetryCount int64 + fixedFallbackLastRetryNano int64 + fixedFallbackDone bool + // Background retry goroutine for fixed_fallback. + // Started when the fixed node is first detected dead. + // Stopped when the node recovers (MustGetAlive=true). + fixedFallbackRunning atomic.Bool + + // fixedFallbackNt is the networkType the background retry goroutine is + // currently probing. Set when the goroutine starts, cleared when it exits. + // Keyed so a select on a different networkType (partial node death, e.g. + // TCP alive while UDP dead) does not reset this networkType's retry state, + // which would otherwise cause log jitter and prevent the goroutine from + // settling (see M1 review). + fixedFallbackNt *dialer.NetworkType + + // fixed_fallback log rate limit + fixedFallbackLastLogMark atomic.Int64 + + // fixed_fallback detail log rate limit (timestamp-based, 10s cooldown) + fixedFallbackDetailLog atomic.Int64 + cachedMinCheckInterval time.Duration } @@ -75,6 +100,49 @@ func NewDialerGroup( group.selectionState.Store(state) group.cachedMinCheckInterval = group.MinCheckInterval() + // Register a callback on the fixed dialer so the background retry + // goroutine starts when the health check marks the node as dead, + // not only when traffic flows through Select(). + // + // NOTE: RegisterAliveTransitionCallback only appends and is never + // unregistered (see dialer.go). This is currently safe because a + // config reload builds a fresh DialerGroup via Clone(), discarding + // the old dialer instances and their callbacks, so no callback + // leak accumulates across reloads. If DialerGroup ever gains an + // in-place update path that reuses dialers, an Unregister (or a + // one-shot guard) will be required to avoid duplicate goroutines. + if p.Policy == consts.DialerSelectionPolicy_FixedWithFallback && + p.FixedIndex >= 0 && p.FixedIndex < len(dialers) { + // H1: if health check is disabled (check_interval==0) but the policy + // asks for retries, the background retry probes are silently dropped + // (aliveBackground returns early when CheckInterval==0), so timeout/ + // retries never take effect — the node falls back permanently on the + // first failure. Warn loudly at startup so the misconfiguration is + // visible (the generic "Health check is DISABLED" warning does not + // mention this policy-specific consequence). + if p.FixedFallbackRetries > 0 && group.cachedMinCheckInterval == 0 { + log.Warnf("fixed_fallback: retries=%d but check_interval=0 (health check disabled). "+ + "Retry probes will be silently dropped; the fixed node will fallback permanently on first failure. "+ + "Set check_interval>0 to enable retry probes, or set retries=0 to skip retry explicitly.", + p.FixedFallbackRetries) + } + fixed := dialers[p.FixedIndex] + if fixed != nil { + fixed.RegisterAliveTransitionCallback(func(nt *dialer.NetworkType, alive bool) { + if alive { + return + } + if group.fixedFallbackRunning.CompareAndSwap(false, true) { + group.fixedFallbackMu.Lock() + group.fixedFallbackDeadSince = time.Now().UnixNano() + group.fixedFallbackRetryCount = 0 + group.fixedFallbackMu.Unlock() + go group.runFixedFallbackRetry(fixed, group.currentSelectionState().policy, nt) + } + }) + } + } + for _, nt := range standardSelectionNetworkTypes() { aliveChangeCallback(true, nt, true) } @@ -312,6 +380,113 @@ func (g *DialerGroup) logNoAlive( }).Warn("no alive dialer for selection (rate-limited)") } +// LogNoAliveDialer logs a warning when no alive dialer is found for selection. +// It is rate-limited per network type to prevent log spam. +func (g *DialerGroup) LogNoAliveDialer( + origNetworkType string, + selectionNetworkType *dialer.NetworkType, + src netip.AddrPort, + dst netip.AddrPort, + domain string, + strictIpVersion bool, +) { + idx := selectionNetworkType.Index() + interval := max(g.cachedMinCheckInterval*5, 10*time.Second) + + if g.tryDoRateLimitedAction(&g.noAliveLogLastTimes[idx], interval) { + g.logNoAlive(origNetworkType, selectionNetworkType, src, dst, domain, strictIpVersion, interval) + } +} + +// logFixedFallback records state transitions for the fixed_fallback policy. +// Mark values: 0=alive/recovery, 1=dead_detected, >=10=retry step, +// -1=fallen back to alternative. +func (g *DialerGroup) logFixedFallback(state int64, fixed *dialer.Dialer, nt *dialer.NetworkType) { + if g.log == nil { + return + } + nodeName := "" + if fixed != nil && fixed.Property() != nil { + nodeName = fixed.Property().Name + } + + switch { + case state == 0: + // Recovery: fixed dialer is alive again + old := g.fixedFallbackLastLogMark.Swap(0) + if old != 0 { + g.log.WithFields(logrus.Fields{ + "group": g.Name, + "dialer": nodeName, + "network": nt.String(), + }).Infoln("Fixed dialer is ALIVE, traffic restored") + } + case state == 1: + // First time detecting dead: log and update state + old := g.fixedFallbackLastLogMark.Swap(1) + if old != 1 { + g.log.WithFields(logrus.Fields{ + "group": g.Name, + "dialer": nodeName, + "network": nt.String(), + }).Warnln("Fixed dialer DEAD, starting retry") + } + case state >= 10: + // Retry: log the actual retry count (state - 10) + retryCount := state - 10 + old := g.fixedFallbackLastLogMark.Swap(state) + if old != state { + g.log.WithFields(logrus.Fields{ + "group": g.Name, + "dialer": nodeName, + "network": nt.String(), + }).Infoln("Fixed dialer retry", retryCount) + } + case state < 0: + // Fallen back to alternative + old := g.fixedFallbackLastLogMark.Swap(-1) + if old >= 0 { + g.log.WithFields(logrus.Fields{ + "group": g.Name, + "dialer": nodeName, + "network": nt.String(), + }).Warnln("Fixed dialer DEAD, fallen back to alternative") + } + } +} + +// logFixedFallbackDetail logs the actual fallback target after selection. +// Rate-limited to once per 10 seconds to prevent log spam under high traffic. +func (g *DialerGroup) logFixedFallbackDetail(fixed, fallbackDialer *dialer.Dialer, nt *dialer.NetworkType, latency time.Duration) { + if g.log == nil { + return + } + // Rate limit: allow one log per 10 seconds. + now := time.Now().UnixNano() + last := g.fixedFallbackDetailLog.Load() + if now-last < 10*int64(time.Second) { + return + } + if !g.fixedFallbackDetailLog.CompareAndSwap(last, now) { + return + } + fixedName := "" + if fixed != nil && fixed.Property() != nil { + fixedName = fixed.Property().Name + } + fallbackName := "" + if fallbackDialer != nil && fallbackDialer.Property() != nil { + fallbackName = fallbackDialer.Property().Name + } + g.log.WithFields(logrus.Fields{ + "group": g.Name, + "fixed": fixedName, + "fallback": fallbackName, + "network": nt.String(), + "latency": latency.String(), + }).Warnln("Fixed dialer DEAD, fallback to", fallbackName) +} + // Select is a backward-compatible wrapper for SelectWithExclusion. func (g *DialerGroup) Select(networkType *dialer.NetworkType, strictIpVersion bool) (d *dialer.Dialer, latency time.Duration, err error) { d, latency, _, err = g.SelectWithExclusionResult(networkType, strictIpVersion, nil) @@ -385,6 +560,120 @@ func (g *DialerGroup) _select(networkType *dialer.NetworkType, state *dialerGrou selected := preferAlternateSelectionNetworkType(g.Dialers[policy.FixedIndex], networkType) return g.Dialers[policy.FixedIndex], 0, selected, nil + case consts.DialerSelectionPolicy_FixedWithFallback: + if policy.FixedIndex < 0 || policy.FixedIndex >= len(g.Dialers) { + return nil, 0, nil, fmt.Errorf("selected dialer index is out of range") + } + fixed := g.Dialers[policy.FixedIndex] + + fallbackPolicy := DialerSelectionPolicy{Policy: policy.FallbackPolicy} + networkTypes, count := g.selectionNetworkTypes(networkType, fallbackPolicy) + + for i := range count { + a := state.aliveDialerSets[networkTypes[i].Index()] + if a == nil { + continue + } + nt := &networkTypes[i] + + // Try fixed dialer first + if fixed != nil && fixed.AliveForRetry(nt) { + g.fixedFallbackMu.Lock() + // Only a select on the networkType the retry goroutine is + // tracking may clear the dead/retry state. A different + // networkType (e.g. TCP alive while UDP is dead) must not + // reset UDP's retry, otherwise partial death causes log + // jitter and the goroutine never settles (see M1 review). + if g.fixedFallbackNt == nil || nt.Index() == g.fixedFallbackNt.Index() { + // Node is alive → reset retry state and use it + wasDead := g.fixedFallbackDeadSince != 0 + g.fixedFallbackDeadSince = 0 + g.fixedFallbackRetryCount = 0 + g.fixedFallbackLastRetryNano = 0 + g.fixedFallbackDone = false + if wasDead { + g.logFixedFallback(0, fixed, nt) + } + } + g.fixedFallbackMu.Unlock() + selected := preferAlternateSelectionNetworkType(fixed, nt) + return fixed, 0, selected, nil + } + + // Fixed dialer is dead → fallback. + // Retries are handled by the background goroutine. + var ( + nowNano int64 + deadSinceNano int64 + ) + + g.fixedFallbackMu.Lock() + nowNano = time.Now().UnixNano() + deadSinceNano = g.fixedFallbackDeadSince + done := g.fixedFallbackDone + + if done { + // Retries exhausted by background goroutine. + // Fallback until health check recovers the node. + g.fixedFallbackMu.Unlock() + g.logFixedFallback(-1, fixed, nt) + goto doFallback + } + + if deadSinceNano == 0 { + // First Select() finding this node dead. + g.fixedFallbackDeadSince = nowNano + g.fixedFallbackRetryCount = 0 + g.fixedFallbackMu.Unlock() + g.logFixedFallback(1, fixed, nt) + + // Start background retry goroutine if not already running + // (may have been started by aliveTransitionCallback already). + if g.fixedFallbackRunning.CompareAndSwap(false, true) { + go g.runFixedFallbackRetry(fixed, policy, nt) + } + + // Background goroutine handles retries separately. + // Natural traffic falls back immediately. + goto doFallback + } + + // Node already known dead. Background goroutine owns retries. + // Fallback immediately — no retryCount/elapsed check. + g.fixedFallbackMu.Unlock() + g.logFixedFallback(-1, fixed, nt) + goto doFallback + + doFallback: + switch policy.FallbackPolicy { + case consts.DialerSelectionPolicy_Random: + d := a.GetRandExcluded(excluded) + if d != nil { + g.logFixedFallbackDetail(fixed, d, nt, 0) + selected := preferAlternateSelectionNetworkType(d, nt) + return d, 0, selected, nil + } + case consts.DialerSelectionPolicy_MinLastLatency, + consts.DialerSelectionPolicy_MinAverage10Latencies, + consts.DialerSelectionPolicy_MinMovingAverageLatencies: + d, lat := a.GetMinLatency(excluded) + if d != nil { + g.logFixedFallbackDetail(fixed, d, nt, lat) + selected := preferAlternateSelectionNetworkType(d, nt) + return d, lat, selected, nil + } + default: + // M3: defensive fallback. FallbackPolicy should only ever be + // one of the four handled cases above (the parser rejects + // fixed/fixed_fallback and unknown names), but guard against a + // zero value or a future policy constant so a misconfigured + // FallbackPolicy is surfaced instead of silently returning + // ErrNoAliveDialer. + g.log.Warnf("fixed_fallback: FallbackPolicy %q is not handled by doFallback; no fallback dialer selected", policy.FallbackPolicy) + } + } + return nil, time.Hour, nil, ErrNoAliveDialer + case consts.DialerSelectionPolicy_MinLastLatency, consts.DialerSelectionPolicy_MinAverage10Latencies, consts.DialerSelectionPolicy_MinMovingAverageLatencies: @@ -409,6 +698,7 @@ func (g *DialerGroup) selectionNetworkTypes(networkType *dialer.NetworkType, pol count = 1 if policy.Policy == consts.DialerSelectionPolicy_Fixed || + policy.Policy == consts.DialerSelectionPolicy_FixedWithFallback || networkType.L4Proto != consts.L4ProtoStr_UDP || networkType.EffectiveUdpHealthDomain() != dialer.UdpHealthDomainData { return networkTypes, count @@ -448,13 +738,21 @@ func (g *DialerGroup) buildSelectionState(policy DialerSelectionPolicy, setAlive return state } + // Determine the policy to use for AliveDialerSet creation. + // FixedWithFallback uses its FallbackPolicy so latency is tracked + // for min_moving_avg / min / random fallback selection. + aliveSetPolicy := policy.Policy + if policy.Policy == consts.DialerSelectionPolicy_FixedWithFallback { + aliveSetPolicy = policy.FallbackPolicy + } + specs := standardSelectionNetworkTypes() keys := dialer.StandardHealthKeys() for i, nt := range specs { networkType := *nt set := dialer.NewAliveDialerSet( - g.log, g.Name, &networkType, g.checkTolerance, policy.Policy, + g.log, g.Name, &networkType, g.checkTolerance, aliveSetPolicy, g.Dialers, g.dialersAnnotations, func(networkType *dialer.NetworkType) func(alive bool) { return func(alive bool) { g.aliveChangeCallback(alive, networkType, false) } @@ -499,7 +797,8 @@ func policyNeedsAliveState(policy consts.DialerSelectionPolicy) bool { case consts.DialerSelectionPolicy_Random, consts.DialerSelectionPolicy_MinLastLatency, consts.DialerSelectionPolicy_MinAverage10Latencies, - consts.DialerSelectionPolicy_MinMovingAverageLatencies: + consts.DialerSelectionPolicy_MinMovingAverageLatencies, + consts.DialerSelectionPolicy_FixedWithFallback: return true case consts.DialerSelectionPolicy_Fixed: return false @@ -567,3 +866,102 @@ func alternateNetworkType(networkType *dialer.NetworkType) *dialer.NetworkType { return nil } } + +// resetFixedFallback clears all fixed_fallback retry state. +func (g *DialerGroup) resetFixedFallback() { + g.fixedFallbackMu.Lock() + g.fixedFallbackDeadSince = 0 + g.fixedFallbackRetryCount = 0 + g.fixedFallbackLastRetryNano = 0 + g.fixedFallbackDone = false + g.fixedFallbackMu.Unlock() +} + +// runFixedFallbackRetry is a background goroutine that drives the +// timeout × retries cycle for the fixed_fallback policy independently +// of traffic. It fires probes at each FixedFallbackTimeout interval, +// and after maxRetries, marks the node for fallback. +func (g *DialerGroup) runFixedFallbackRetry(fixed *dialer.Dialer, policy DialerSelectionPolicy, nt *dialer.NetworkType) { + defer g.fixedFallbackRunning.Store(false) + // Track which networkType this goroutine is probing so a select on a + // different networkType (partial node death, e.g. TCP alive / UDP dead) + // does not reset this networkType's retry state (see M1 review). + defer func() { + g.fixedFallbackMu.Lock() + g.fixedFallbackNt = nil + g.fixedFallbackMu.Unlock() + }() + g.fixedFallbackMu.Lock() + g.fixedFallbackNt = nt + g.fixedFallbackMu.Unlock() + + // retries <= 0: give up immediately, no ticker needed. + // Node stays marked as done until health check recovers it. + if policy.FixedFallbackRetries <= 0 { + g.fixedFallbackMu.Lock() + g.fixedFallbackDone = true + g.fixedFallbackDeadSince = time.Now().UnixNano() - 1 + g.fixedFallbackMu.Unlock() + g.logFixedFallback(-1, fixed, nt) + return + } + + actualTimeout := policy.FixedFallbackTimeout + if actualTimeout < 2*time.Second { + actualTimeout = 2 * time.Second + nodeName := "" + if fixed != nil && fixed.Property() != nil { + nodeName = fixed.Property().Name + } + g.log.WithFields(logrus.Fields{ + "group": g.Name, + "configured": policy.FixedFallbackTimeout.String(), + "actual": actualTimeout.String(), + "node": nodeName, + "network_type": nt.String(), + }).Warnln("fixed_fallback timeout too low, clamped to minimum 2s to prevent probe storm") + } + ticker := time.NewTicker(actualTimeout) + defer ticker.Stop() + + for { + select { + case <-ticker.C: + case <-fixed.Done(): + // dae is shutting down or the dialer was replaced by a reload; + // stop probing instead of blocking until retries are exhausted. + return + } + + // Check if node has recovered + if fixed.AliveForRetry(nt) { + g.resetFixedFallback() + return + } + + // Advance retry count + g.fixedFallbackMu.Lock() + g.fixedFallbackRetryCount++ + g.fixedFallbackLastRetryNano = time.Now().UnixNano() + + shouldFallback := g.fixedFallbackRetryCount >= int64(policy.FixedFallbackRetries) + if shouldFallback { + g.fixedFallbackDone = true + g.fixedFallbackDeadSince = time.Now().UnixNano() - 1 + } else { + g.fixedFallbackDeadSince = time.Now().UnixNano() + } + g.fixedFallbackMu.Unlock() + + // Fire probes — also gives the node a chance to be marked alive + // before the next tick. + fixed.NotifyCheckTcp() + fixed.NotifyCheckDnsUdp() + + if shouldFallback { + g.logFixedFallback(-1, fixed, nt) + return + } + g.logFixedFallback(10+g.fixedFallbackRetryCount, fixed, nt) + } +} diff --git a/component/outbound/dialer_selection_policy.go b/component/outbound/dialer_selection_policy.go index 667312d8..7f6fe9e3 100644 --- a/component/outbound/dialer_selection_policy.go +++ b/component/outbound/dialer_selection_policy.go @@ -8,14 +8,19 @@ package outbound import ( "fmt" "strconv" + "strings" + "time" "github.com/daeuniverse/dae/common/consts" "github.com/daeuniverse/dae/config" ) type DialerSelectionPolicy struct { - Policy consts.DialerSelectionPolicy - FixedIndex int + Policy consts.DialerSelectionPolicy + FixedIndex int + FixedFallbackTimeout time.Duration // 节点超时时间 + FixedFallbackRetries int // 超时重试次数 + FallbackPolicy consts.DialerSelectionPolicy // 重试耗尽后的回退策略，默认 min_moving_avg } func NewDialerSelectionPolicyFromGroupParam(param *config.Group) (policy *DialerSelectionPolicy, err error) { @@ -53,7 +58,133 @@ func NewDialerSelectionPolicyFromGroupParam(param *config.Group) (policy *Dialer FixedIndex: index, }, nil + case consts.DialerSelectionPolicy_FixedWithFallback: + + if f.Not { + return nil, fmt.Errorf("policy param does not support not operator: !%v()", f.Name) + } + if len(f.Params) < 1 || len(f.Params) > 4 { + return nil, fmt.Errorf(`invalid "%v" param format: expected 1-4 params, got %v`, f.Name, len(f.Params)) + } + // Parse index (required, first param) + if f.Params[0].Key != "" { + return nil, fmt.Errorf(`invalid "%v" param format: first param must be index (no key)`, f.Name) + } + index, err := strconv.Atoi(f.Params[0].Val) + if err != nil { + return nil, fmt.Errorf(`invalid "%v" param format: %w`, f.Name, err) + } + // Parse timeout (optional, second param, with unit suffix: ms/s/m) + timeout := 3 * time.Second // default + if len(f.Params) >= 2 { + if f.Params[1].Key != "" { + return nil, fmt.Errorf(`invalid "%v" param format: second param must be timeout (no key)`, f.Name) + } + timeout, err = parseDurationWithUnit(f.Params[1].Val) + if err != nil { + return nil, fmt.Errorf(`invalid "%v" param format: %w`, f.Name, err) + } + } + // Parse retries (optional, third param). Default 3. 0 means the node + // falls back immediately on first failure with no background retry + // (matches the canonical "retries<=0 = do not retry" semantics). + retries := 3 // default + if len(f.Params) >= 3 { + if f.Params[2].Key != "" { + return nil, fmt.Errorf(`invalid "%v" param format: third param must be retry count (no key)`, f.Name) + } + retries, err = strconv.Atoi(f.Params[2].Val) + if err != nil { + return nil, fmt.Errorf(`invalid "%v" param format: retries must be an integer: %w`, f.Name, err) + } + if retries < 0 { + return nil, fmt.Errorf(`invalid "%v" param format: retries must be >= 0`, f.Name) + } + } + // Parse fallback policy (optional, fourth param) + fallbackPolicy := consts.DialerSelectionPolicy_MinMovingAverageLatencies // default + if len(f.Params) >= 4 { + if f.Params[3].Key != "" { + return nil, fmt.Errorf(`invalid "%v" param format: fourth param must be fallback policy name (no key)`, f.Name) + } + fp, err := parsePolicyName(f.Params[3].Val) + if err != nil { + return nil, fmt.Errorf(`invalid "%v" param format: fallback policy: %w`, f.Name, err) + } + fallbackPolicy = fp + } + return &DialerSelectionPolicy{ + Policy: consts.DialerSelectionPolicy_FixedWithFallback, + FixedIndex: index, + FixedFallbackTimeout: timeout, + FixedFallbackRetries: retries, + FallbackPolicy: fallbackPolicy, + }, nil + default: return nil, fmt.Errorf("unexpected policy: %v", f.Name) } } + +// parsePolicyName maps a policy name string to the corresponding DialerSelectionPolicy constant. +// Supported fallback policies: random, min_moving_avg, min_last_latency, min_avg10. +// Fixed and fixed_fallback are not supported as fallback policies (would cause infinite recursion). +func parsePolicyName(s string) (consts.DialerSelectionPolicy, error) { + s = strings.TrimSpace(s) + switch s { + case "random": + return consts.DialerSelectionPolicy_Random, nil + case "min_moving_avg": + return consts.DialerSelectionPolicy_MinMovingAverageLatencies, nil + case "min_last_latency": + return consts.DialerSelectionPolicy_MinLastLatency, nil + case "min_avg10": + return consts.DialerSelectionPolicy_MinAverage10Latencies, nil + default: + return consts.DialerSelectionPolicy(""), fmt.Errorf("unsupported fallback policy %q (supported: random, min_moving_avg, min_last_latency, min_avg10)", s) + } +} + +// Supported: "ms" (milliseconds), "s" (seconds), "m" (minutes). +// No suffix defaults to seconds for backward compatibility. +// Examples: "500ms", "5s", "2m", "10". +func parseDurationWithUnit(s string) (time.Duration, error) { + s = strings.TrimSpace(s) + if s == "" { + return 0, fmt.Errorf("empty duration string") + } + + // Check "ms" first (must precede "s" check) + if strings.HasSuffix(s, "ms") { + val, err := strconv.ParseFloat(strings.TrimSuffix(s, "ms"), 64) + if err != nil { + return 0, fmt.Errorf("invalid duration %q: %w", s, err) + } + return time.Duration(val * float64(time.Millisecond)), nil + } + + // "s" suffix + if strings.HasSuffix(s, "s") { + val, err := strconv.ParseFloat(strings.TrimSuffix(s, "s"), 64) + if err != nil { + return 0, fmt.Errorf("invalid duration %q: %w", s, err) + } + return time.Duration(val * float64(time.Second)), nil + } + + // "m" suffix + if strings.HasSuffix(s, "m") { + val, err := strconv.ParseFloat(strings.TrimSuffix(s, "m"), 64) + if err != nil { + return 0, fmt.Errorf("invalid duration %q: %w", s, err) + } + return time.Duration(val * float64(time.Minute)), nil + } + + // No suffix: treat as seconds (backward compatible) + val, err := strconv.ParseFloat(s, 64) + if err != nil { + return 0, fmt.Errorf("invalid duration %q: %w", s, err) + } + return time.Duration(val * float64(time.Second)), nil +} diff --git a/control/control_plane.go b/control/control_plane.go index 2dee5641..7cd72473 100644 --- a/control/control_plane.go +++ b/control/control_plane.go @@ -749,6 +749,16 @@ func newControlPlaneWithContextOptions( if global.AllowInsecure { log.Warnln("AllowInsecure is enabled, but it is not recommended. Please make sure you have to turn it on.") } + // Warn about health-check configuration that would silently disable probing. + // check_interval=0 disables ALL probing (including any configured URLs), + // so it must be reported even when tcp_check_url/udp_check_dns are set. + if global.CheckInterval == 0 { + log.Warnln("Health check is DISABLED: check_interval is 0. Nodes will NOT be probed even if " + + "tcp_check_url/udp_check_dns are configured. Set check_interval (>0) to enable health checks.") + } else if len(global.TcpCheckUrl) == 0 && len(global.UdpCheckDns) == 0 { + log.Warnln("Health check has no probe target: tcp_check_url and udp_check_dns are both empty. " + + "Nodes will not be probed. Configure at least one to enable health checks.") + } locationFinder := assets.NewLocationFinder(externGeoDataDirs) option := dialer.NewGlobalOption(global, log) option.DaeDNS, err = daedns.NewWithOption(log, global, dnsConfig, &daedns.NewOption{LocationFinder: locationFinder})