fix(alerts): evaluate fresh reports and order notification delivery (#1242)

* fix(alerts): evaluate fresh reports and make notification transitions reliable

* fix(alerts): identify equal-timestamp reports by sequence

* fix(alerts): preserve notification transition semantics

---------

Co-authored-by: naiba <hi@nai.ba>

cmd/dashboard/controller/alertrule.go

@@ -178,6 +178,9 @@ func validateRule(c *gin.Context, r *model.AlertRule) error {
 			if !rule.IsSupportedType() {
 				return singleton.Localizer.ErrorT("unsupported rule type")
 			}
+			if !rule.HasSafeThresholds() {
+				return singleton.Localizer.ErrorT("invalid rule thresholds")
+			}
 			switch rule.Cover {
 			case model.RuleCoverAll, model.RuleCoverIgnoreAll:
 			default:
@@ -192,6 +195,9 @@ func validateRule(c *gin.Context, r *model.AlertRule) error {
 					return singleton.Localizer.ErrorT("duration is too large")
 				}
 			} else {
+				if !rule.HasSafeCycleUnit() {
+					return singleton.Localizer.ErrorT("invalid cycle unit")
+				}
 				if rule.CycleInterval < 1 {
 					return singleton.Localizer.ErrorT("cycle_interval need to be at least 1")
 				}

model/alert_evaluation_test.go

@@ -0,0 +1,205 @@
+package model
+
+import (
+	"math"
+	"testing"
+	"time"
+
+	"gorm.io/driver/sqlite"
+	"gorm.io/gorm"
+)
+
+func TestAlertEvaluationDistinguishesUnknownFromBreach(t *testing.T) {
+	server := &Server{Common: Common{ID: 1}}
+	rule := &Rule{Type: "cpu", Max: 80, Cover: RuleCoverAll}
+	if passed, known := rule.Evaluate(nil, server, nil); !passed || known {
+		t.Fatalf("missing state = (%v, %v), want unknown", passed, known)
+	}
+	lease := server.AttachStateStream(runtimeOwnershipStream{})
+	if !lease.UpdateState(&HostState{CPU: 99}, time.Now()) {
+		t.Fatal("state update rejected")
+	}
+	if passed, known := rule.Evaluate(nil, server, nil); passed || !known {
+		t.Fatalf("fresh high CPU = (%v, %v), want known breach", passed, known)
+	}
+	if !lease.Clear() {
+		t.Fatal("stream clear rejected")
+	}
+	// The raw metric remains available for historical display. The sentinel
+	// enforces freshness and must not count this value after disconnect.
+	if !server.RuntimeSnapshot().LastActive.IsZero() {
+		t.Fatal("disconnect did not clear online visibility")
+	}
+}
+
+func newCycleTestDB(t *testing.T, migrate bool) *gorm.DB {
+	t.Helper()
+	db, err := gorm.Open(sqlite.Open("file:"+t.Name()+"?mode=memory&cache=shared"), &gorm.Config{})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if migrate {
+		if err := db.AutoMigrate(&Transfer{}); err != nil {
+			t.Fatal(err)
+		}
+	}
+	return db
+}
+
+func newCycleTestStats() *CycleTransferStats {
+	return &CycleTransferStats{
+		ServerName: make(map[uint64]string),
+		Transfer:   make(map[uint64]uint64),
+		NextUpdate: make(map[uint64]time.Time),
+	}
+}
+
+func TestCycleRuleRejectsInvalidConfiguration(t *testing.T) {
+	start := time.Now().Add(-time.Hour)
+	rule := &Rule{Type: "transfer_out_cycle", CycleStart: &start, CycleInterval: 1, CycleUnit: "hour", Max: 100}
+	if !rule.HasSafeCycleConfiguration() {
+		t.Fatal("normal hourly cycle rejected")
+	}
+	for _, tc := range []struct {
+		name   string
+		mutate func(*Rule)
+	}{
+		{"unknown unit", func(r *Rule) { r.CycleUnit = "fortnight" }},
+		{"negative limit", func(r *Rule) { r.Max = -1 }},
+		{"NaN limit", func(r *Rule) { r.Max = math.NaN() }},
+		{"reversed limits", func(r *Rule) { r.Min, r.Max = 200, 100 }},
+		{"future start", func(r *Rule) { future := time.Now().Add(time.Hour); r.CycleStart = &future }},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			copy := *rule
+			tc.mutate(&copy)
+			if copy.HasSafeCycleConfiguration() {
+				t.Fatal("unsafe cycle accepted")
+			}
+		})
+	}
+}
+
+func TestCycleBoundsUseOneClockAtBoundary(t *testing.T) {
+	start := time.Date(2026, 9, 1, 0, 0, 0, 0, time.UTC)
+	rule := &Rule{Type: "transfer_out_cycle", CycleStart: &start, CycleInterval: 1, CycleUnit: "hour", Max: 100}
+	before := start.Add(59*time.Minute + 59*time.Second)
+	from, to := rule.transferDurationBounds(before)
+	if !from.Equal(start) || !to.Equal(start.Add(time.Hour)) {
+		t.Fatalf("before boundary = %v..%v", from, to)
+	}
+	from, to = rule.transferDurationBounds(start.Add(time.Hour))
+	if !from.Equal(start.Add(time.Hour)) || !to.Equal(start.Add(2*time.Hour)) {
+		t.Fatalf("at boundary = %v..%v", from, to)
+	}
+}
+
+func TestCycleQueryFailureIsUnknownAndDoesNotCache(t *testing.T) {
+	start := time.Now().Add(-time.Hour)
+	rule := &Rule{Type: "transfer_out_cycle", CycleStart: &start, CycleInterval: 1, CycleUnit: "hour", Max: 50}
+	server := &Server{Common: Common{ID: 1}, State: &HostState{}}
+	passed, known := rule.Evaluate(newCycleTestStats(), server, newCycleTestDB(t, false))
+	if !passed || known || len(rule.NextTransferAt) != 0 {
+		t.Fatalf("SQL error yielded passed=%v known=%v cache=%v", passed, known, rule.NextTransferAt)
+	}
+}
+
+func TestCycleQueryExcludesFutureRowsAndHandlesMinOnly(t *testing.T) {
+	start := time.Now().Add(-90 * time.Minute)
+	db := newCycleTestDB(t, true)
+	server := &Server{Common: Common{ID: 1}, State: &HostState{}}
+	for _, row := range []Transfer{
+		{Common: Common{CreatedAt: time.Now().Add(-time.Minute)}, ServerID: 1, Out: 10},
+		{Common: Common{CreatedAt: time.Now().Add(10 * time.Minute)}, ServerID: 1, Out: 100},
+	} {
+		if err := db.Create(&row).Error; err != nil {
+			t.Fatal(err)
+		}
+	}
+	rule := &Rule{Type: "transfer_out_cycle", CycleStart: &start, CycleInterval: 1, CycleUnit: "hour", Max: 50}
+	stats := newCycleTestStats()
+	passed, known := rule.Evaluate(stats, server, db)
+	if !passed || !known || stats.Transfer[1] != 10 {
+		t.Fatalf("future row included: passed=%v known=%v amount=%d", passed, known, stats.Transfer[1])
+	}
+	minOnly := &Rule{Type: "transfer_out_cycle", CycleStart: &start, CycleInterval: 1, CycleUnit: "hour", Min: 20}
+	passed, known = minOnly.Evaluate(newCycleTestStats(), server, db)
+	if passed || !known {
+		t.Fatalf("min-only cycle = (%v,%v), want known breach", passed, known)
+	}
+}
+
+func TestAlertEvaluationAllNetworkSpeedUsesBothDirections(t *testing.T) {
+	for _, tc := range []struct {
+		name    string
+		in, out uint64
+		breach  bool
+	}{
+		{"inbound only", 100, 0, true},
+		{"outbound below combined limit", 0, 30, false},
+		{"both directions", 25, 30, true},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			server := &Server{Common: Common{ID: 1}, State: &HostState{NetInSpeed: tc.in, NetOutSpeed: tc.out}}
+			rule := &Rule{Type: "net_all_speed", Max: 50, Cover: RuleCoverAll}
+			passed, known := rule.Evaluate(nil, server, nil)
+			if !known || passed == tc.breach {
+				t.Fatalf("got passed=%v known=%v, want breach=%v", passed, known, tc.breach)
+			}
+		})
+	}
+}
+
+func TestAlertEvaluationUnavailableSensorsAreUnknown(t *testing.T) {
+	tests := []struct {
+		name  string
+		rule  string
+		state *HostState
+		host  *Host
+	}{
+		{"missing memory host", "memory", &HostState{MemUsed: 1}, nil},
+		{"zero memory total", "memory", &HostState{MemUsed: 1}, &Host{}},
+		{"missing GPU", "gpu_max", &HostState{}, &Host{}},
+		{"missing temperature", "temperature_max", &HostState{}, &Host{}},
+	}
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			server := &Server{Common: Common{ID: 1}, State: tc.state, Host: tc.host}
+			passed, known := (&Rule{Type: tc.rule, Max: 50, Cover: RuleCoverAll}).Evaluate(nil, server, nil)
+			if !passed || known {
+				t.Fatalf("got passed=%v known=%v, want unknown", passed, known)
+			}
+		})
+	}
+}
+
+func TestOfflineRuleDoesNotRequireMetricState(t *testing.T) {
+	server := &Server{Common: Common{ID: 1}}
+	InitServer(server)
+	passed, known := (&Rule{Type: "offline", Cover: RuleCoverAll}).Evaluate(nil, server, nil)
+	if passed || !known {
+		t.Fatalf("offline without state = (%v, %v), want known breach", passed, known)
+	}
+}
+
+func TestCompoundAlertUsesOneImmutableAgentReport(t *testing.T) {
+	server := &Server{Common: Common{ID: 1}}
+	alert := &AlertRule{Rules: []*Rule{
+		{Type: "cpu", Max: 80, Cover: RuleCoverAll},
+		{Type: "memory", Max: 80, Cover: RuleCoverAll},
+	}}
+	oldReport := RuntimeSnapshot{
+		State: &HostState{CPU: 99, MemUsed: 0},
+		Host:  &Host{MemTotal: 100},
+	}
+	newReport := RuntimeSnapshot{
+		State: &HostState{CPU: 0, MemUsed: 99},
+		Host:  &Host{MemTotal: 100},
+	}
+	for _, report := range []RuntimeSnapshot{oldReport, newReport} {
+		point, known := alert.SnapshotStatusWithRuntime(nil, server, report, nil)
+		if !known || len(point) != 2 || point[0] == point[1] {
+			t.Fatalf("compound point combined incompatible reports: known=%v point=%v", known, point)
+		}
+	}
+}

model/alert_timed_test.go

@@ -0,0 +1,104 @@
+package model
+
+import (
+	"testing"
+	"time"
+)
+
+func timedRule(kind string, duration uint64) *AlertRule {
+	return &AlertRule{Rules: []*Rule{{Type: kind, Duration: duration, Max: 80, Cover: RuleCoverAll}}}
+}
+
+func timedPoints(start time.Time, values []bool, spacing time.Duration) []TimedAlertPoint {
+	points := make([]TimedAlertPoint, len(values))
+	for i, value := range values {
+		points[i] = TimedAlertPoint{At: start.Add(time.Duration(i) * spacing), Values: []bool{value}}
+	}
+	return points
+}
+
+func TestTimedAlertNeedsElapsedWindowAndDistinctReports(t *testing.T) {
+	start := time.Unix(1_700_000_000, 0)
+	rule := timedRule("cpu", 5)
+	points := timedPoints(start, []bool{false, false, false, false}, AlertSampleInterval)
+	if known, _ := rule.CheckTimed(points, points[len(points)-1].At); known {
+		t.Fatal("four reports decided a five-tick window")
+	}
+	points = append(points, TimedAlertPoint{At: start.Add(12 * time.Second), Values: []bool{false}})
+	if known, passed := rule.CheckTimed(points, points[len(points)-1].At); !known || passed {
+		t.Fatalf("complete high-CPU window: known=%v passed=%v", known, passed)
+	}
+	// Repeating the same observation does not advance wall-clock coverage.
+	if known, _ := rule.CheckTimed(points, start.Add(30*time.Second)); known {
+		t.Fatal("stale sample created a new complete window")
+	}
+}
+
+func TestTimedAlertGapIsUnknownAndCannotResolve(t *testing.T) {
+	start := time.Unix(1_700_000_000, 0)
+	rule := timedRule("cpu", 5)
+	points := []TimedAlertPoint{
+		{At: start, Values: []bool{false}},
+		{At: start.Add(30 * time.Second), Values: []bool{true}},
+	}
+	if known, _ := rule.CheckTimed(points, start.Add(30*time.Second)); known {
+		t.Fatal("telemetry gap must not be interpreted as healthy")
+	}
+}
+
+func TestTimedAlertIrregularReportsUseElapsedTime(t *testing.T) {
+	start := time.Unix(1_700_000_000, 0)
+	rule := timedRule("cpu", 6)
+	points := timedPoints(start, []bool{false, false, false, false}, 5*time.Second)
+	if known, passed := rule.CheckTimed(points, start.Add(15*time.Second)); !known || passed {
+		t.Fatalf("irregular reports cover the wall-clock window: known=%v passed=%v", known, passed)
+	}
+}
+
+func TestTimedAlertExactSeventyPercentIsNotIncident(t *testing.T) {
+	start := time.Unix(1_700_000_000, 0)
+	rule := timedRule("cpu", 10)
+	values := []bool{false, false, false, false, false, false, false, true, true, true}
+	points := timedPoints(start, values, AlertSampleInterval)
+	if known, passed := rule.CheckTimed(points, points[len(points)-1].At); !known || !passed {
+		t.Fatalf("70%% failed is below the >70%% trigger: known=%v passed=%v", known, passed)
+	}
+}
+
+func TestTimedAlertTenTickWindowDoesNotDecideAtNine(t *testing.T) {
+	start := time.Unix(1_700_000_000, 0)
+	rule := timedRule("cpu", 10)
+	points := timedPoints(start, []bool{false, false, false, false, false, false, false, false, false}, AlertSampleInterval)
+	if known, _ := rule.CheckTimed(points, points[len(points)-1].At); known {
+		t.Fatal("nine reports decided a ten-tick window one tick early")
+	}
+	points = append(points, TimedAlertPoint{At: start.Add(27 * time.Second), Values: []bool{false}})
+	if known, passed := rule.CheckTimed(points, points[len(points)-1].At); !known || passed {
+		t.Fatalf("full ten-tick breach: known=%v passed=%v", known, passed)
+	}
+}
+
+func TestTimedOfflineRequiresSustainedAbsence(t *testing.T) {
+	start := time.Unix(1_700_000_000, 0)
+	rule := timedRule("offline", 5)
+	points := timedPoints(start, []bool{false, false, false, false, false}, AlertSampleInterval)
+	if known, passed := rule.CheckTimed(points, points[len(points)-1].At); !known || passed {
+		t.Fatalf("continuous offline interval: known=%v passed=%v", known, passed)
+	}
+	points = append(points, TimedAlertPoint{At: start.Add(15 * time.Second), Values: []bool{true}})
+	if known, passed := rule.CheckTimed(points, points[len(points)-1].At); !known || !passed {
+		t.Fatalf("online observation should resolve offline: known=%v passed=%v", known, passed)
+	}
+}
+
+func TestTimedCompoundRuleWaitsForAllKnown(t *testing.T) {
+	start := time.Unix(1_700_000_000, 0)
+	rule := &AlertRule{Rules: []*Rule{
+		{Type: "cpu", Duration: 5, Max: 80, Cover: RuleCoverAll},
+		{Type: "memory", Duration: 5, Max: 80, Cover: RuleCoverAll},
+	}}
+	points := timedPoints(start, []bool{false, false, false, false, false}, AlertSampleInterval)
+	if known, _ := rule.CheckTimed(points, points[len(points)-1].At); known {
+		t.Fatal("missing second condition treated as decisive")
+	}
+}

model/alertrule.go

@@ -2,12 +2,134 @@ package model
 
 import (
 	"slices"
+	"time"
 
 	"github.com/gin-gonic/gin"
 	"github.com/goccy/go-json"
 	"gorm.io/gorm"
 )
 
+const (
+	AlertSampleInterval = 3 * time.Second
+	AlertSampleMaxAge   = 6 * time.Second
+)
+
+// TimedAlertPoint records one distinct observation. Its timestamp is the
+// dashboard receipt time of the Agent report, or the sentinel tick for an
+// offline-only rule.
+type TimedAlertPoint struct {
+	At     time.Time
+	Values []bool
+}
+
+// CheckTimed evaluates the legacy Duration as Duration * 3 seconds, avoiding
+// both repeated stale samples and cadence-dependent early incidents. Unknown
+// coverage cannot create an incident or a recovery.
+func (r *AlertRule) CheckTimed(points []TimedAlertPoint, now time.Time) (known, passed bool) {
+	if r == nil || len(r.Rules) == 0 {
+		return false, false
+	}
+	allFailed := true
+	for index, rule := range r.Rules {
+		if rule == nil || !rule.IsSupportedType() {
+			return false, false
+		}
+		ruleKnown, rulePassed := rule.checkTimed(points, index, now)
+		if !ruleKnown {
+			return false, false
+		}
+		if rulePassed {
+			allFailed = false
+		}
+	}
+	return true, !allFailed
+}
+
+func (rule *Rule) checkTimed(points []TimedAlertPoint, index int, now time.Time) (known, passed bool) {
+	if len(points) == 0 {
+		return false, false
+	}
+	if rule.IsTransferDurationRule() {
+		last := points[len(points)-1]
+		if index >= len(last.Values) || now.Sub(last.At) > AlertSampleMaxAge {
+			return false, false
+		}
+		return true, last.Values[index]
+	}
+	window := time.Duration(rule.Duration) * AlertSampleInterval
+	if window <= 0 {
+		return false, false
+	}
+	end := now.Add(AlertSampleInterval) // the current observation represents one tick
+	start := end.Add(-window)
+	if points[0].At.After(start) {
+		return false, false // the configured wall-clock window has not elapsed
+	}
+	var covered, failed time.Duration
+	for i, point := range points {
+		if index >= len(point.Values) {
+			return false, false
+		}
+		segmentStart := point.At
+		if start.After(segmentStart) {
+			segmentStart = start
+		}
+		segmentEnd := point.At.Add(AlertSampleMaxAge)
+		if end.Before(segmentEnd) {
+			segmentEnd = end
+		}
+		if i+1 < len(points) {
+			if points[i+1].At.Before(segmentEnd) {
+				segmentEnd = points[i+1].At
+			}
+		} else {
+			if tickEnd := point.At.Add(AlertSampleInterval); tickEnd.Before(segmentEnd) {
+				segmentEnd = tickEnd
+			}
+		}
+		if !segmentEnd.After(segmentStart) {
+			continue
+		}
+		span := segmentEnd.Sub(segmentStart)
+		covered += span
+		if !point.Values[index] {
+			failed += span
+		}
+	}
+	if covered < window-window/10 {
+		return false, false
+	}
+	if rule.IsOfflineRule() {
+		if covered-failed >= AlertSampleInterval {
+			return true, true
+		}
+		if failed >= window-AlertSampleInterval && covered >= window-AlertSampleInterval {
+			return true, false
+		}
+		return false, false
+	}
+	if float64(failed) > float64(window)*0.70 {
+		return true, false
+	}
+	if float64(failed+window-covered) <= float64(window)*0.70 {
+		return true, true
+	}
+	return false, false
+}
+
+func (r *AlertRule) TimedRetention() time.Duration {
+	longest := AlertSampleMaxAge
+	for _, rule := range r.Rules {
+		if rule == nil || rule.IsTransferDurationRule() {
+			continue
+		}
+		if duration, ok := rule.DurationInt(); ok {
+			longest = max(longest, time.Duration(duration)*AlertSampleInterval+AlertSampleMaxAge)
+		}
+	}
+	return longest
+}
+
 const (
 	ModeAlwaysTrigger  = 0
 	ModeOnetimeTrigger = 1
@@ -75,6 +197,9 @@ func (r *AlertRule) IsSafeToEvaluate() bool {
 		if rule == nil || !rule.IsSupportedType() {
 			return false
 		}
+		if !rule.HasSafeThresholds() {
+			return false
+		}
 		switch rule.Cover {
 		case RuleCoverAll, RuleCoverIgnoreAll:
 		default:
@@ -152,18 +277,38 @@ func (r *AlertRule) HasPermission(ctx *gin.Context) bool {
 
 // Snapshot 对传入的Server进行该报警规则下所有type的检查 返回每项检查结果
 func (r *AlertRule) Snapshot(cycleTransferStats *CycleTransferStats, server *Server, db *gorm.DB) []bool {
+	point, _ := r.SnapshotStatus(cycleTransferStats, server, db)
+	return point
+}
+
+// SnapshotStatus preserves the distinction between a healthy measurement and
+// telemetry that is unavailable. A compound alert cannot be decided while any
+// of its applicable conditions is unknown.
+func (r *AlertRule) SnapshotStatus(cycleTransferStats *CycleTransferStats, server *Server, db *gorm.DB) ([]bool, bool) {
+	if server == nil {
+		return nil, false
+	}
+	return r.SnapshotStatusWithRuntime(cycleTransferStats, server, server.RuntimeSnapshot(), db)
+}
+
+// SnapshotStatusWithRuntime uses one immutable report for every condition.
+func (r *AlertRule) SnapshotStatusWithRuntime(cycleTransferStats *CycleTransferStats, server *Server, runtime RuntimeSnapshot, db *gorm.DB) ([]bool, bool) {
 	point := make([]bool, len(r.Rules))
+	known := true
 
 	for i, rule := range r.Rules {
 		if rule == nil || !rule.IsSupportedType() {
 			// Invalid persisted rules are ignored instead of being interpreted as
 			// a failed condition or allowed to panic the sentinel.
 			point[i] = true
+			known = false
 			continue
 		}
-		point[i] = rule.Snapshot(cycleTransferStats, server, db)
+		var valid bool
+		point[i], valid = rule.EvaluateRuntime(cycleTransferStats, server, runtime, db)
+		known = known && valid
 	}
-	return point
+	return point, known
 }
 
 // Check 传入包含当前报警规则下所有type检查结果 返回报警持续时间与是否通过报警检查(通过则返回true)

model/notification.go

@@ -180,63 +180,65 @@ func (ns *NotificationServerBundle) replaceParamsInString(str string, message st
 
 	if ns.Server != nil {
 		runtime := ns.Server.RuntimeSnapshot()
-		if runtime.State == nil || runtime.Host == nil {
-			return str
-		}
-		state := runtime.State
-		host := runtime.Host
 		replacements = append(replacements,
 			"#SERVER.NAME#", mod(ns.Server.Name),
 			"#SERVER.ID#", mod(fmt.Sprintf("%d", ns.Server.ID)),
-
-			// Converted metrics
-			"#SERVER.CPU#", mod(ns.formatUsage(false, state.CPU)),
-			"#SERVER.MEM#", mod(ns.formatUsage(true, float64(state.MemUsed)/float64(host.MemTotal))),
-			"#SERVER.SWAP#", mod(ns.formatUsage(true, float64(state.SwapUsed)/float64(host.SwapTotal))),
-			"#SERVER.DISK#", mod(ns.formatUsage(true, float64(state.DiskUsed)/float64(host.DiskTotal))),
-			"#SERVER.SPEEDIN#", mod(fmt.Sprintf("%s/s", ns.formatSize(state.NetInSpeed))),
-			"#SERVER.SPEEDOUT#", mod(fmt.Sprintf("%s/s", ns.formatSize(state.NetOutSpeed))),
-			"#SERVER.TRANSFERIN#", mod(ns.formatSize(state.NetInTransfer)),
-			"#SERVER.TRANSFEROUT#", mod(ns.formatSize(state.NetOutTransfer)),
-
-			// Raw metrics
-			"#SERVER.CPUUSED#", mod(fmt.Sprintf("%f", state.CPU)),
-			"#SERVER.MEMUSED#", mod(fmt.Sprintf("%d", state.MemUsed)),
-			"#SERVER.SWAPUSED#", mod(fmt.Sprintf("%d", state.SwapUsed)),
-			"#SERVER.DISKUSED#", mod(fmt.Sprintf("%d", state.DiskUsed)),
-			"#SERVER.MEMTOTAL#", mod(fmt.Sprintf("%d", host.MemTotal)),
-			"#SERVER.SWAPTOTAL#", mod(fmt.Sprintf("%d", host.SwapTotal)),
-			"#SERVER.DISKTOTAL#", mod(fmt.Sprintf("%d", host.DiskTotal)),
-			"#SERVER.NETINSPEED#", mod(fmt.Sprintf("%d", state.NetInSpeed)),
-			"#SERVER.NETOUTSPEED#", mod(fmt.Sprintf("%d", state.NetOutSpeed)),
-			"#SERVER.NETINTRANSFER#", mod(fmt.Sprintf("%d", state.NetInTransfer)),
-			"#SERVER.NETOUTTRANSFER#", mod(fmt.Sprintf("%d", state.NetOutTransfer)),
-			"#SERVER.LOAD1#", mod(fmt.Sprintf("%f", state.Load1)),
-			"#SERVER.LOAD5#", mod(fmt.Sprintf("%f", state.Load5)),
-			"#SERVER.LOAD15#", mod(fmt.Sprintf("%f", state.Load15)),
-			"#SERVER.TCPCONNCOUNT#", mod(fmt.Sprintf("%d", state.TcpConnCount)),
-			"#SERVER.UDPCONNCOUNT#", mod(fmt.Sprintf("%d", state.UdpConnCount)),
 		)
+		if runtime.State != nil && runtime.Host != nil {
+			state := runtime.State
+			host := runtime.Host
+			replacements = append(replacements,
+				// Converted metrics
+				"#SERVER.CPU#", mod(ns.formatUsage(false, state.CPU)),
+				"#SERVER.MEM#", mod(ns.formatUsage(true, float64(state.MemUsed)/float64(host.MemTotal))),
+				"#SERVER.SWAP#", mod(ns.formatUsage(true, float64(state.SwapUsed)/float64(host.SwapTotal))),
+				"#SERVER.DISK#", mod(ns.formatUsage(true, float64(state.DiskUsed)/float64(host.DiskTotal))),
+				"#SERVER.SPEEDIN#", mod(fmt.Sprintf("%s/s", ns.formatSize(state.NetInSpeed))),
+				"#SERVER.SPEEDOUT#", mod(fmt.Sprintf("%s/s", ns.formatSize(state.NetOutSpeed))),
+				"#SERVER.TRANSFERIN#", mod(ns.formatSize(state.NetInTransfer)),
+				"#SERVER.TRANSFEROUT#", mod(ns.formatSize(state.NetOutTransfer)),
 
-		var ipv4, ipv6, validIP string
-		ip := ns.Server.GeoIP.IP
-		if ip.IPv4Addr != "" && ip.IPv6Addr != "" {
-			ipv4 = ip.IPv4Addr
-			ipv6 = ip.IPv6Addr
-			validIP = ipv4
-		} else if ip.IPv4Addr != "" {
-			ipv4 = ip.IPv4Addr
-			validIP = ipv4
-		} else {
-			ipv6 = ip.IPv6Addr
-			validIP = ipv6
+				// Raw metrics
+				"#SERVER.CPUUSED#", mod(fmt.Sprintf("%f", state.CPU)),
+				"#SERVER.MEMUSED#", mod(fmt.Sprintf("%d", state.MemUsed)),
+				"#SERVER.SWAPUSED#", mod(fmt.Sprintf("%d", state.SwapUsed)),
+				"#SERVER.DISKUSED#", mod(fmt.Sprintf("%d", state.DiskUsed)),
+				"#SERVER.MEMTOTAL#", mod(fmt.Sprintf("%d", host.MemTotal)),
+				"#SERVER.SWAPTOTAL#", mod(fmt.Sprintf("%d", host.SwapTotal)),
+				"#SERVER.DISKTOTAL#", mod(fmt.Sprintf("%d", host.DiskTotal)),
+				"#SERVER.NETINSPEED#", mod(fmt.Sprintf("%d", state.NetInSpeed)),
+				"#SERVER.NETOUTSPEED#", mod(fmt.Sprintf("%d", state.NetOutSpeed)),
+				"#SERVER.NETINTRANSFER#", mod(fmt.Sprintf("%d", state.NetInTransfer)),
+				"#SERVER.NETOUTTRANSFER#", mod(fmt.Sprintf("%d", state.NetOutTransfer)),
+				"#SERVER.LOAD1#", mod(fmt.Sprintf("%f", state.Load1)),
+				"#SERVER.LOAD5#", mod(fmt.Sprintf("%f", state.Load5)),
+				"#SERVER.LOAD15#", mod(fmt.Sprintf("%f", state.Load15)),
+				"#SERVER.TCPCONNCOUNT#", mod(fmt.Sprintf("%d", state.TcpConnCount)),
+				"#SERVER.UDPCONNCOUNT#", mod(fmt.Sprintf("%d", state.UdpConnCount)),
+			)
 		}
 
-		replacements = append(replacements,
-			"#SERVER.IP#", mod(validIP),
-			"#SERVER.IPV4#", mod(ipv4),
-			"#SERVER.IPV6#", mod(ipv6),
-		)
+		var ipv4, ipv6, validIP string
+		if ns.Server.GeoIP != nil {
+			ip := ns.Server.GeoIP.IP
+			if ip.IPv4Addr != "" && ip.IPv6Addr != "" {
+				ipv4 = ip.IPv4Addr
+				ipv6 = ip.IPv6Addr
+				validIP = ipv4
+			} else if ip.IPv4Addr != "" {
+				ipv4 = ip.IPv4Addr
+				validIP = ipv4
+			} else {
+				ipv6 = ip.IPv6Addr
+				validIP = ipv6
+			}
+
+			replacements = append(replacements,
+				"#SERVER.IP#", mod(validIP),
+				"#SERVER.IPV4#", mod(ipv4),
+				"#SERVER.IPV6#", mod(ipv6),
+			)
+		}
 	}
 
 	replacer := strings.NewReplacer(replacements...)

model/rule.go

@@ -1,6 +1,8 @@
 package model
 
 import (
+	"log"
+	"math"
 	"slices"
 	"strings"
 	"time"
@@ -90,36 +92,80 @@ func (u *Rule) DurationInt() (duration int, ok bool) {
 // which is the documented legacy spelling for hours.
 func (u *Rule) HasSafeCycleConfiguration() bool {
 	return u != nil && u.IsTransferDurationRule() && u.CycleStart != nil &&
-		u.CycleInterval > 0 && u.CycleInterval <= MaxAlertRuleCycleInterval
+		!u.CycleStart.After(time.Now()) && u.CycleInterval > 0 && u.CycleInterval <= MaxAlertRuleCycleInterval &&
+		u.HasSafeThresholds() && u.HasSafeCycleUnit()
 }
 
-// Snapshot 未通过规则返回 false, 通过返回 true
-func (u *Rule) Snapshot(cycleTransferStats *CycleTransferStats, server *Server, db *gorm.DB) bool {
-	if u == nil || server == nil || !u.IsSupportedType() {
+func (u *Rule) HasSafeCycleUnit() bool {
+	if u == nil {
+		return false
+	}
+	switch strings.ToLower(u.CycleUnit) {
+	case "", "hour", "day", "week", "month", "year":
 		return true
+	default:
+		return false
+	}
+}
+
+func (u *Rule) HasSafeThresholds() bool {
+	if u == nil || math.IsNaN(u.Min) || math.IsNaN(u.Max) || math.IsInf(u.Min, 0) || math.IsInf(u.Max, 0) || u.Min < 0 || u.Max < 0 {
+		return false
+	}
+	return u.Min == 0 || u.Max == 0 || u.Min <= u.Max
+}
+
+// Snapshot is retained for callers which only need a boolean result. The alert
+// sentinel uses Evaluate so missing telemetry is not mistaken for a breach.
+func (u *Rule) Snapshot(cycleTransferStats *CycleTransferStats, server *Server, db *gorm.DB) bool {
+	passed, _ := u.Evaluate(cycleTransferStats, server, db)
+	return passed
+}
+
+// Evaluate returns known=false when a metric cannot be evaluated from the
+// current report. Unknown must neither trigger nor resolve an incident.
+func (u *Rule) Evaluate(cycleTransferStats *CycleTransferStats, server *Server, db *gorm.DB) (passed, known bool) {
+	if server == nil {
+		return true, false
+	}
+	return u.EvaluateRuntime(cycleTransferStats, server, server.RuntimeSnapshot(), db)
+}
+
+// EvaluateRuntime evaluates all conditions against the same immutable Agent
+// report. Callers evaluating a compound rule must share this snapshot.
+func (u *Rule) EvaluateRuntime(cycleTransferStats *CycleTransferStats, server *Server, runtime RuntimeSnapshot, db *gorm.DB) (passed, known bool) {
+	if u == nil || server == nil || !u.IsSupportedType() {
+		return true, false
 	}
 	if u.IsTransferDurationRule() && (!u.HasSafeCycleConfiguration() || cycleTransferStats == nil) {
-		return true
+		return true, false
 	}
 
 	// 监控全部但是排除了此服务器
 	if u.Cover == RuleCoverAll && u.Ignore[server.ID] {
-		return true
+		return true, true
 	}
 	// 忽略全部但是指定监控了此服务器
 	if u.Cover == RuleCoverIgnoreAll && !u.Ignore[server.ID] {
-		return true
+		return true, true
 	}
 
+	evalNow := time.Now()
+	var cycleStart, cycleEnd time.Time
+	if u.IsTransferDurationRule() {
+		cycleStart, cycleEnd = u.transferDurationBounds(evalNow)
+	}
 	// 循环区间流量检测 · 短期无需重复检测
-	if u.IsTransferDurationRule() && u.NextTransferAt[server.ID].After(time.Now()) {
-		return u.LastCycleStatus[server.ID]
+	if u.IsTransferDurationRule() && u.NextTransferAt[server.ID].After(evalNow) {
+		return u.LastCycleStatus[server.ID], true
 	}
 
 	var src float64
-	runtime := server.RuntimeSnapshot()
+	if u.IsOfflineRule() {
+		return !runtime.LastActive.IsZero() && time.Since(runtime.LastActive) <= 6*time.Second, true
+	}
 	if runtime.State == nil {
-		return false
+		return true, false
 	}
 	state := runtime.State
 
@@ -128,22 +174,31 @@ func (u *Rule) Snapshot(cycleTransferStats *CycleTransferStats, server *Server,
 		src = float64(state.CPU)
 	case "gpu", "gpu_max":
 		if len(state.GPU) == 0 {
-			return true
+			return true, false
 		}
 		src = slices.Max(state.GPU)
 	case "memory":
 		if runtime.Host == nil {
-			return false
+			return true, false
+		}
+		if runtime.Host.MemTotal == 0 {
+			return true, false
 		}
 		src = percentage(state.MemUsed, runtime.Host.MemTotal)
 	case "swap":
 		if runtime.Host == nil {
-			return false
+			return true, false
+		}
+		if runtime.Host.SwapTotal == 0 {
+			return true, false
 		}
 		src = percentage(state.SwapUsed, runtime.Host.SwapTotal)
 	case "disk":
 		if runtime.Host == nil {
-			return false
+			return true, false
+		}
+		if runtime.Host.DiskTotal == 0 {
+			return true, false
 		}
 		src = percentage(state.DiskUsed, runtime.Host.DiskTotal)
 	case "net_in_speed":
@@ -151,38 +206,50 @@ func (u *Rule) Snapshot(cycleTransferStats *CycleTransferStats, server *Server,
 	case "net_out_speed":
 		src = float64(state.NetOutSpeed)
 	case "net_all_speed":
-		src = float64(state.NetOutSpeed + state.NetOutSpeed)
+		src = float64(state.NetInSpeed) + float64(state.NetOutSpeed)
 	case "transfer_in":
 		src = float64(state.NetInTransfer)
 	case "transfer_out":
 		src = float64(state.NetOutTransfer)
 	case "transfer_all":
 		src = float64(state.NetOutTransfer + state.NetInTransfer)
-	case "offline":
-		if runtime.LastActive.IsZero() {
-			src = 0
-		} else {
-			src = float64(runtime.LastActive.Unix())
-		}
 	case "transfer_in_cycle":
+		if db == nil {
+			return true, false
+		}
 		src = float64(utils.SubUintChecked(state.NetInTransfer, runtime.PrevTransferInSnapshot))
 		if u.CycleInterval != 0 {
 			var res NResult
-			db.Model(&Transfer{}).Select("SUM(`in`) AS n").Where("datetime(`created_at`) >= datetime(?) AND server_id = ?", u.GetTransferDurationStart().UTC(), server.ID).Scan(&res)
+			if err := db.Model(&Transfer{}).Select("SUM(`in`) AS n").Where("datetime(`created_at`) >= datetime(?) AND datetime(`created_at`) < datetime(?) AND datetime(`created_at`) <= datetime(?) AND server_id = ?", cycleStart.UTC(), cycleEnd.UTC(), evalNow.UTC(), server.ID).Scan(&res).Error; err != nil {
+				log.Printf("NEZHA>> Alert cycle transfer query failed for rule %s server %d: %v", u.Type, server.ID, err)
+				return true, false
+			}
 			src += float64(res.N)
 		}
 	case "transfer_out_cycle":
+		if db == nil {
+			return true, false
+		}
 		src = float64(utils.SubUintChecked(state.NetOutTransfer, runtime.PrevTransferOutSnapshot))
 		if u.CycleInterval != 0 {
 			var res NResult
-			db.Model(&Transfer{}).Select("SUM(`out`) AS n").Where("datetime(`created_at`) >= datetime(?) AND server_id = ?", u.GetTransferDurationStart().UTC(), server.ID).Scan(&res)
+			if err := db.Model(&Transfer{}).Select("SUM(`out`) AS n").Where("datetime(`created_at`) >= datetime(?) AND datetime(`created_at`) < datetime(?) AND datetime(`created_at`) <= datetime(?) AND server_id = ?", cycleStart.UTC(), cycleEnd.UTC(), evalNow.UTC(), server.ID).Scan(&res).Error; err != nil {
+				log.Printf("NEZHA>> Alert cycle transfer query failed for rule %s server %d: %v", u.Type, server.ID, err)
+				return true, false
+			}
 			src += float64(res.N)
 		}
 	case "transfer_all_cycle":
+		if db == nil {
+			return true, false
+		}
 		src = float64(utils.SubUintChecked(state.NetOutTransfer, runtime.PrevTransferOutSnapshot) + utils.SubUintChecked(state.NetInTransfer, runtime.PrevTransferInSnapshot))
 		if u.CycleInterval != 0 {
 			var res NResult
-			db.Model(&Transfer{}).Select("SUM(`in`+`out`) AS n").Where("datetime(`created_at`) >= datetime(?) AND server_id = ?", u.GetTransferDurationStart().UTC(), server.ID).Scan(&res)
+			if err := db.Model(&Transfer{}).Select("SUM(`in`+`out`) AS n").Where("datetime(`created_at`) >= datetime(?) AND datetime(`created_at`) < datetime(?) AND datetime(`created_at`) <= datetime(?) AND server_id = ?", cycleStart.UTC(), cycleEnd.UTC(), evalNow.UTC(), server.ID).Scan(&res).Error; err != nil {
+				log.Printf("NEZHA>> Alert cycle transfer query failed for rule %s server %d: %v", u.Type, server.ID, err)
+				return true, false
+			}
 			src += float64(res.N)
 		}
 	case "load1":
@@ -205,23 +272,32 @@ func (u *Rule) Snapshot(cycleTransferStats *CycleTransferStats, server *Server,
 			}
 		}
 		if len(temp) == 0 {
-			return true
+			return true, false
 		}
 		src = slices.Max(temp)
 	default:
-		return true
+		return true, false
 	}
 
 	// 循环区间流量检测 · 更新下次需要检测时间
 	if u.IsTransferDurationRule() {
-		seconds := max(1800*((u.Max-src)/u.Max), 180)
+		seconds := float64(180)
+		if u.Max > 0 {
+			seconds = max(1800*((u.Max-src)/u.Max), 180)
+		} else if u.Min > 0 {
+			seconds = 1800
+		}
 		if u.NextTransferAt == nil {
 			u.NextTransferAt = make(map[uint64]time.Time)
 		}
 		if u.LastCycleStatus == nil {
 			u.LastCycleStatus = make(map[uint64]bool)
 		}
-		u.NextTransferAt[server.ID] = time.Now().Add(time.Second * time.Duration(seconds))
+		nextAt := evalNow.Add(time.Second * time.Duration(seconds))
+		if nextAt.After(cycleEnd) {
+			nextAt = cycleEnd
+		}
+		u.NextTransferAt[server.ID] = nextAt
 		if (u.Max > 0 && src > u.Max) || (u.Min > 0 && src < u.Min) {
 			u.LastCycleStatus[server.ID] = false
 		} else {
@@ -233,17 +309,15 @@ func (u *Rule) Snapshot(cycleTransferStats *CycleTransferStats, server *Server,
 		cycleTransferStats.Transfer[server.ID] = uint64(src)
 		cycleTransferStats.NextUpdate[server.ID] = u.NextTransferAt[server.ID]
 		// 自动更新周期流量展示起止时间
-		cycleTransferStats.From = u.GetTransferDurationStart()
-		cycleTransferStats.To = u.GetTransferDurationEnd()
+		cycleTransferStats.From = cycleStart
+		cycleTransferStats.To = cycleEnd
 	}
 
-	if u.Type == "offline" && float64(time.Now().Unix())-src > 6 {
-		return false
-	} else if (u.Max > 0 && src > u.Max) || (u.Min > 0 && src < u.Min) {
-		return false
+	if (u.Max > 0 && src > u.Max) || (u.Min > 0 && src < u.Min) {
+		return false, true
 	}
 
-	return true
+	return true, true
 }
 
 // IsTransferDurationRule 判断该规则是否属于周期流量规则 属于则返回true
@@ -263,83 +337,49 @@ func (u *Rule) IsOfflineRule() bool {
 	return u != nil && u.Type == "offline"
 }
 
-// GetTransferDurationStart 获取周期流量的起始时间
+// transferDurationBounds computes both edges from the same clock reading.
+// Separate time.Now calls can otherwise straddle a cycle boundary.
+func (u *Rule) transferDurationBounds(now time.Time) (time.Time, time.Time) {
+	if !u.HasSafeCycleConfiguration() {
+		return time.Time{}, time.Time{}
+	}
+	start := *u.CycleStart
+	var end time.Time
+	switch strings.ToLower(u.CycleUnit) {
+	case "year":
+		end = start.AddDate(int(u.CycleInterval), 0, 0)
+		for !now.Before(end) {
+			start, end = end, end.AddDate(int(u.CycleInterval), 0, 0)
+		}
+	case "month":
+		end = start.AddDate(0, int(u.CycleInterval), 0)
+		for !now.Before(end) {
+			start, end = end, end.AddDate(0, int(u.CycleInterval), 0)
+		}
+	case "week":
+		end = start.AddDate(0, 0, 7*int(u.CycleInterval))
+		for !now.Before(end) {
+			start, end = end, end.AddDate(0, 0, 7*int(u.CycleInterval))
+		}
+	case "day":
+		end = start.AddDate(0, 0, int(u.CycleInterval))
+		for !now.Before(end) {
+			start, end = end, end.AddDate(0, 0, int(u.CycleInterval))
+		}
+	default: // empty is the legacy spelling for hour
+		interval := 3600 * int64(u.CycleInterval)
+		start = time.Unix(u.CycleStart.Unix()+(now.Unix()-u.CycleStart.Unix())/interval*interval, 0)
+		end = time.Unix(start.Unix()+interval, 0)
+	}
+	return start, end
+}
+
 func (u *Rule) GetTransferDurationStart() time.Time {
-	// Accept uppercase and lowercase
-	unit := strings.ToLower(u.CycleUnit)
-	startTime := *u.CycleStart
-	var nextTime time.Time
-	switch unit {
-	case "year":
-		nextTime = startTime.AddDate(int(u.CycleInterval), 0, 0)
-		for time.Now().After(nextTime) {
-			startTime = nextTime
-			nextTime = nextTime.AddDate(int(u.CycleInterval), 0, 0)
-		}
-	case "month":
-		nextTime = startTime.AddDate(0, int(u.CycleInterval), 0)
-		for time.Now().After(nextTime) {
-			startTime = nextTime
-			nextTime = nextTime.AddDate(0, int(u.CycleInterval), 0)
-		}
-	case "week":
-		nextTime = startTime.AddDate(0, 0, 7*int(u.CycleInterval))
-		for time.Now().After(nextTime) {
-			startTime = nextTime
-			nextTime = nextTime.AddDate(0, 0, 7*int(u.CycleInterval))
-		}
-	case "day":
-		nextTime = startTime.AddDate(0, 0, int(u.CycleInterval))
-		for time.Now().After(nextTime) {
-			startTime = nextTime
-			nextTime = nextTime.AddDate(0, 0, int(u.CycleInterval))
-		}
-	default:
-		// For hour unit or not set.
-		interval := 3600 * int64(u.CycleInterval)
-		startTime = time.Unix(u.CycleStart.Unix()+(time.Now().Unix()-u.CycleStart.Unix())/interval*interval, 0)
-	}
-
-	return startTime
+	start, _ := u.transferDurationBounds(time.Now())
+	return start
 }
 
-// GetTransferDurationEnd 获取周期流量结束时间
 func (u *Rule) GetTransferDurationEnd() time.Time {
-	// Accept uppercase and lowercase
-	unit := strings.ToLower(u.CycleUnit)
-	startTime := *u.CycleStart
-	var nextTime time.Time
-	switch unit {
-	case "year":
-		nextTime = startTime.AddDate(int(u.CycleInterval), 0, 0)
-		for time.Now().After(nextTime) {
-			startTime = nextTime
-			nextTime = nextTime.AddDate(int(u.CycleInterval), 0, 0)
-		}
-	case "month":
-		nextTime = startTime.AddDate(0, int(u.CycleInterval), 0)
-		for time.Now().After(nextTime) {
-			startTime = nextTime
-			nextTime = nextTime.AddDate(0, int(u.CycleInterval), 0)
-		}
-	case "week":
-		nextTime = startTime.AddDate(0, 0, 7*int(u.CycleInterval))
-		for time.Now().After(nextTime) {
-			startTime = nextTime
-			nextTime = nextTime.AddDate(0, 0, 7*int(u.CycleInterval))
-		}
-	case "day":
-		nextTime = startTime.AddDate(0, 0, int(u.CycleInterval))
-		for time.Now().After(nextTime) {
-			startTime = nextTime
-			nextTime = nextTime.AddDate(0, 0, int(u.CycleInterval))
-		}
-	default:
-		// For hour unit or not set.
-		interval := 3600 * int64(u.CycleInterval)
-		startTime = time.Unix(u.CycleStart.Unix()+(time.Now().Unix()-u.CycleStart.Unix())/interval*interval, 0)
-		nextTime = time.Unix(startTime.Unix()+interval, 0)
-	}
-
-	return nextTime
+	_, end := u.transferDurationBounds(time.Now())
+	return end
 }

model/server.go

@@ -16,6 +16,7 @@ import (
 )
 
 var runtimeHolderInitMu sync.Mutex
+var nextServerReportSequence atomic.Uint64
 
 type Server struct {
 	Common
@@ -73,15 +74,16 @@ type taskStreamHolder struct {
 }
 
 type serverRuntimeHolder struct {
-	mu         sync.Mutex
-	canonical  *Server
-	stream     pb.NezhaService_ReportSystemStateServer
-	generation uint64
-	state      *HostState
-	host       *Host
-	lastActive time.Time
-	prevIn     uint64
-	prevOut    uint64
+	mu             sync.Mutex
+	canonical      *Server
+	stream         pb.NezhaService_ReportSystemStateServer
+	generation     uint64
+	state          *HostState
+	host           *Host
+	lastActive     time.Time
+	reportSequence uint64
+	prevIn         uint64
+	prevOut        uint64
 }
 
 type StateStreamLease struct {
@@ -204,6 +206,7 @@ func (lease StateStreamLease) updateState(receiver *Server, state *HostState, la
 			return false
 		}
 	}
+	lease.holder.reportSequence = nextServerReportSequence.Add(1)
 	return true
 }
 
@@ -334,6 +337,7 @@ type RuntimeSnapshot struct {
 	State                   *HostState
 	Host                    *Host
 	LastActive              time.Time
+	ReportSequence          uint64
 	PrevTransferInSnapshot  uint64
 	PrevTransferOutSnapshot uint64
 }
@@ -360,7 +364,7 @@ func (s *Server) RuntimeSnapshot() RuntimeSnapshot {
 			holder.host = cloneHost(s.Host)
 		}
 	}
-	return RuntimeSnapshot{State: cloneHostState(holder.state), Host: cloneHost(holder.host), LastActive: holder.lastActive, PrevTransferInSnapshot: holder.prevIn, PrevTransferOutSnapshot: holder.prevOut}
+	return RuntimeSnapshot{State: cloneHostState(holder.state), Host: cloneHost(holder.host), LastActive: holder.lastActive, ReportSequence: holder.reportSequence, PrevTransferInSnapshot: holder.prevIn, PrevTransferOutSnapshot: holder.prevOut}
 }
 
 func (s *Server) SetTransferSnapshots(inbound, outbound uint64) bool {

service/singleton/alertsentinel.go

@@ -4,6 +4,7 @@ import (
 	"fmt"
 	"log"
 	"sync"
+	"sync/atomic"
 	"time"
 
 	"github.com/jinzhu/copier"
@@ -26,11 +27,122 @@ type NotificationHistory struct {
 var (
 	AlertsLock                    sync.RWMutex
 	Alerts                        []*model.AlertRule
-	alertsStore                   map[uint64]map[uint64][][]bool       // [alert_id][server_id] -> [timeTick][ruleId] 时间点对应的rule的检查结果
-	alertsPrevState               map[uint64]map[uint64]uint8          // [alert_id][server_id] -> 对应报警规则的上一次报警状态
-	AlertsCycleTransferStatsStore map[uint64]*model.CycleTransferStats // [alert_id] -> 对应报警规则的周期流量统计
+	alertsStore                   map[uint64]map[uint64][]model.TimedAlertPoint // distinct reports per alert/server
+	alertsPrevState               map[uint64]map[uint64]uint8                   // [alert_id][server_id] -> 对应报警规则的上一次报警状态
+	alertsLastMetricSeq           map[uint64]map[uint64]uint64                  // last distinct Agent metric report per alert/server
+	AlertsCycleTransferStatsStore map[uint64]*model.CycleTransferStats          // [alert_id] -> 对应报警规则的周期流量统计
+	alertDeliveryMu               sync.Mutex
+	alertDeliveries               map[uint64]map[uint64]*alertDeliveryEntry
+	alertEventSequence            atomic.Uint64
 )
 
+type alertDeliveryEntry struct {
+	cancel chan struct{}
+	done   chan struct{}
+	sendMu *sync.Mutex
+}
+
+type alertDeliveryPhase uint8
+
+const (
+	alertDeliveryIncident alertDeliveryPhase = iota
+	alertDeliveryRecovery
+)
+
+func cancelAlertDeliveries(alertID uint64) {
+	alertDeliveryMu.Lock()
+	defer alertDeliveryMu.Unlock()
+	for _, entry := range alertDeliveries[alertID] {
+		close(entry.cancel)
+	}
+	delete(alertDeliveries, alertID)
+}
+
+func cancelAlertDeliveriesForServers(serverIDs []uint64) {
+	alertDeliveryMu.Lock()
+	defer alertDeliveryMu.Unlock()
+	for alertID, byServer := range alertDeliveries {
+		for _, serverID := range serverIDs {
+			if entry := byServer[serverID]; entry != nil {
+				close(entry.cancel)
+				delete(byServer, serverID)
+			}
+		}
+		if len(byServer) == 0 {
+			delete(alertDeliveries, alertID)
+		}
+	}
+}
+
+func runAlertDelivery(cancel <-chan struct{}, send func() bool, always bool, retryDelay time.Duration) {
+	for {
+		select {
+		case <-cancel:
+			return
+		default:
+		}
+		delivered := send()
+		if delivered && !always {
+			return
+		}
+		delay := retryDelay
+		if delivered {
+			delay = firstNotificationDelay
+		}
+		timer := time.NewTimer(delay)
+		select {
+		case <-cancel:
+			timer.Stop()
+			return
+		case <-timer.C:
+		}
+	}
+}
+
+func startAlertDelivery(alert *model.AlertRule, server *model.Server, message, muteLabel string, phase alertDeliveryPhase) {
+	// A mute entry belongs to one transition, not all later incidents of the
+	// same phase. Concurrent in-flight sends may finish after a recovery.
+	muteLabel = fmt.Sprintf("%s:event-%d", muteLabel, alertEventSequence.Add(1))
+	alertDeliveryMu.Lock()
+	if alertDeliveries == nil {
+		alertDeliveries = make(map[uint64]map[uint64]*alertDeliveryEntry)
+	}
+	if alertDeliveries[alert.ID] == nil {
+		alertDeliveries[alert.ID] = make(map[uint64]*alertDeliveryEntry)
+	}
+	sendMu := new(sync.Mutex)
+	if old := alertDeliveries[alert.ID][server.ID]; old != nil {
+		close(old.cancel)
+		sendMu = old.sendMu
+	}
+	entry := &alertDeliveryEntry{cancel: make(chan struct{}), done: make(chan struct{}), sendMu: sendMu}
+	alertDeliveries[alert.ID][server.ID] = entry
+	alertDeliveryMu.Unlock()
+
+	groupID := alert.NotificationGroupID
+	always := phase == alertDeliveryIncident && alert.TriggerMode == model.ModeAlwaysTrigger
+	go func() {
+		defer close(entry.done)
+		runAlertDelivery(entry.cancel, func() bool {
+			// Serialize incident/recovery delivery for this alert/server pair.
+			// A recovery can cancel a pending incident, but never overtake an
+			// already-running network request.
+			entry.sendMu.Lock()
+			defer entry.sendMu.Unlock()
+			select {
+			case <-entry.cancel:
+				return true
+			default:
+			}
+			return NotificationShared.SendNotification(groupID, message, muteLabel, server)
+		}, always, 30*time.Second)
+	}()
+}
+
+func shouldRunAlertFailTasks(triggerMode uint8, newIncident bool) bool {
+	return newIncident || triggerMode == model.ModeAlwaysTrigger
+}
+
 // addCycleTransferStatsInfo 向AlertsCycleTransferStatsStore中添加周期流量报警统计信息
 func addCycleTransferStatsInfo(alert *model.AlertRule) {
 	if alert == nil || !alert.Enabled() || !alert.IsSafeToEvaluate() {
@@ -59,8 +171,17 @@ func addCycleTransferStatsInfo(alert *model.AlertRule) {
 
 // AlertSentinelStart 报警器启动
 func AlertSentinelStart() {
-	alertsStore = make(map[uint64]map[uint64][][]bool)
+	alertDeliveryMu.Lock()
+	for _, byServer := range alertDeliveries {
+		for _, entry := range byServer {
+			close(entry.cancel)
+		}
+	}
+	alertDeliveries = make(map[uint64]map[uint64]*alertDeliveryEntry)
+	alertDeliveryMu.Unlock()
+	alertsStore = make(map[uint64]map[uint64][]model.TimedAlertPoint)
 	alertsPrevState = make(map[uint64]map[uint64]uint8)
+	alertsLastMetricSeq = make(map[uint64]map[uint64]uint64)
 	AlertsCycleTransferStatsStore = make(map[uint64]*model.CycleTransferStats)
 	AlertsLock.Lock()
 	if err := DB.Find(&Alerts).Error; err != nil {
@@ -71,8 +192,9 @@ func AlertSentinelStart() {
 			log.Printf("NEZHA>> Skipping invalid nil alert rule loaded from database")
 			continue
 		}
-		alertsStore[alert.ID] = make(map[uint64][][]bool)
+		alertsStore[alert.ID] = make(map[uint64][]model.TimedAlertPoint)
 		alertsPrevState[alert.ID] = make(map[uint64]uint8)
+		alertsLastMetricSeq[alert.ID] = make(map[uint64]uint64)
 		if !alert.IsSafeToEvaluate() {
 			log.Printf("NEZHA>> Skipping invalid alert rule %d loaded from database", alert.ID)
 			continue
@@ -101,8 +223,10 @@ func AlertSentinelStart() {
 func OnRefreshOrAddAlert(alert *model.AlertRule) {
 	AlertsLock.Lock()
 	defer AlertsLock.Unlock()
+	cancelAlertDeliveries(alert.ID)
 	delete(alertsStore, alert.ID)
 	delete(alertsPrevState, alert.ID)
+	delete(alertsLastMetricSeq, alert.ID)
 	var isEdit bool
 	for i := range Alerts {
 		if Alerts[i].ID == alert.ID {
@@ -113,8 +237,9 @@ func OnRefreshOrAddAlert(alert *model.AlertRule) {
 	if !isEdit {
 		Alerts = append(Alerts, alert)
 	}
-	alertsStore[alert.ID] = make(map[uint64][][]bool)
+	alertsStore[alert.ID] = make(map[uint64][]model.TimedAlertPoint)
 	alertsPrevState[alert.ID] = make(map[uint64]uint8)
+	alertsLastMetricSeq[alert.ID] = make(map[uint64]uint64)
 	delete(AlertsCycleTransferStatsStore, alert.ID)
 	addCycleTransferStatsInfo(alert)
 }
@@ -123,8 +248,10 @@ func OnDeleteAlert(id []uint64) {
 	AlertsLock.Lock()
 	defer AlertsLock.Unlock()
 	for _, i := range id {
+		cancelAlertDeliveries(i)
 		delete(alertsStore, i)
 		delete(alertsPrevState, i)
+		delete(alertsLastMetricSeq, i)
 		currentAlerts := Alerts[:0]
 		for _, alert := range Alerts {
 			if alert.ID != i {
@@ -176,25 +303,67 @@ func checkStatusForServer(alert *model.AlertRule, server *model.Server) {
 		}
 	}()
 
-	alertsStore[alert.ID][server.ID] = append(alertsStore[alert.
-		ID][server.ID], alert.Snapshot(AlertsCycleTransferStatsStore[alert.ID], server, DB))
+	// Offline-only rules deliberately observe elapsed time on every tick. Metric
+	// rules, including compound rules, must never count the same Agent report
+	// repeatedly or treat a disconnected Agent's retained State as fresh data.
+	metricRule := false
+	runtime := server.RuntimeSnapshot()
+	for _, rule := range alert.Rules {
+		if !rule.IsOfflineRule() {
+			metricRule = true
+			break
+		}
+	}
+	if metricRule {
+		if runtime.LastActive.IsZero() || runtime.State == nil || time.Since(runtime.LastActive) > model.AlertSampleMaxAge || time.Since(runtime.LastActive) < 0 {
+			alertsStore[alert.ID][server.ID] = nil
+			return
+		}
+		if runtime.ReportSequence == 0 || runtime.ReportSequence == alertsLastMetricSeq[alert.ID][server.ID] {
+			return
+		}
+		alertsLastMetricSeq[alert.ID][server.ID] = runtime.ReportSequence
+	}
+	point, known := alert.SnapshotStatusWithRuntime(AlertsCycleTransferStatsStore[alert.ID], server, runtime, DB)
+	if !known {
+		alertsStore[alert.ID][server.ID] = nil
+		return
+	}
+	sampledAt := time.Now()
+	if metricRule {
+		sampledAt = runtime.LastActive
+	}
+	alertsStore[alert.ID][server.ID] = append(alertsStore[alert.ID][server.ID], model.TimedAlertPoint{At: sampledAt, Values: point})
+	// Bound memory even while coverage is insufficient to decide a state.
+	cutoff := time.Now().Add(-alert.TimedRetention())
+	samples := alertsStore[alert.ID][server.ID]
+	first := 0
+	for first < len(samples) && samples[first].At.Before(cutoff) {
+		first++
+	}
+	if first > 0 {
+		alertsStore[alert.ID][server.ID] = append([]model.TimedAlertPoint(nil), samples[first:]...)
+	}
 	// 发送通知，分为触发报警和恢复通知
-	_, passed := alert.Check(alertsStore[alert.ID][server.ID])
+	known, passed := alert.CheckTimed(alertsStore[alert.ID][server.ID], time.Now())
+	if !known {
+		return
+	}
 	// 保存当前服务器状态信息
 	curServer := model.Server{}
 	copier.Copy(&curServer, server)
 
 	// 本次未通过检查
 	if !passed {
-		// 始终触发模式或上次检查不为失败时触发报警（跳过单次触发+上次失败的情况）
-		if alert.TriggerMode == model.ModeAlwaysTrigger || alertsPrevState[alert.ID][server.ID] != _RuleCheckFail {
+		newIncident := alertsPrevState[alert.ID][server.ID] != _RuleCheckFail
+		if newIncident {
 			alertsPrevState[alert.ID][server.ID] = _RuleCheckFail
 			message := fmt.Sprintf("[%s] %s(%s) %s", Localizer.T("Incident"),
 				server.Name, IPDesensitize(server.GeoIP.IP.Join()), alert.Name)
+			startAlertDelivery(alert, &curServer, message, NotificationMuteLabel.ServerIncident(server.ID, alert.ID), alertDeliveryIncident)
+		}
+		if shouldRunAlertFailTasks(alert.TriggerMode, newIncident) {
 			go CronShared.SendTriggerTasks(alert.FailTriggerTasks, curServer.ID, alert.UserID)
-			go NotificationShared.SendNotification(alert.NotificationGroupID, message, NotificationMuteLabel.ServerIncident(server.ID, alert.ID), &curServer)
-			// 清除恢复通知的静音缓存
-			NotificationShared.UnMuteNotification(alert.NotificationGroupID, NotificationMuteLabel.ServerIncidentResolved(server.ID, alert.ID))
 		}
 	} else {
 		// 本次通过检查但上一次的状态为失败，则发送恢复通知
@@ -202,20 +371,8 @@ func checkStatusForServer(alert *model.AlertRule, server *model.Server) {
 			message := fmt.Sprintf("[%s] %s(%s) %s", Localizer.T("Resolved"),
 				server.Name, IPDesensitize(server.GeoIP.IP.Join()), alert.Name)
 			go CronShared.SendTriggerTasks(alert.RecoverTriggerTasks, curServer.ID, alert.UserID)
-			go NotificationShared.SendNotification(alert.NotificationGroupID, message, NotificationMuteLabel.ServerIncidentResolved(server.ID, alert.ID), &curServer)
-			// 清除失败通知的静音缓存
-			NotificationShared.UnMuteNotification(alert.NotificationGroupID, NotificationMuteLabel.ServerIncident(server.ID, alert.ID))
+			startAlertDelivery(alert, &curServer, message, NotificationMuteLabel.ServerIncidentResolved(server.ID, alert.ID), alertDeliveryRecovery)
 		}
 		alertsPrevState[alert.ID][server.ID] = _RuleCheckPass
 	}
-	// 清理旧数据：保留窗口由规则定义决定（各规则 Duration 的最大值），
-	// 而非 Check 的判定结果。window==0 表示没有任何有效规则需要回看历史
-	// （例如全部 Duration<=0），此时清空采样避免切片无限增长。
-	window := alert.RetentionWindow()
-	samples := alertsStore[alert.ID][server.ID]
-	if window <= 0 {
-		alertsStore[alert.ID][server.ID] = samples[:0]
-	} else if window < len(samples) {
-		alertsStore[alert.ID][server.ID] = samples[len(samples)-window:]
-	}
 }

service/singleton/alertsentinel_freshness_test.go

@@ -0,0 +1,281 @@
+package singleton
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/patrickmn/go-cache"
+	"google.golang.org/grpc/metadata"
+
+	"github.com/nezhahq/nezha/model"
+	pb "github.com/nezhahq/nezha/proto"
+)
+
+type alertTestStateStream struct{}
+
+func (alertTestStateStream) Send(*pb.Receipt) error       { return nil }
+func (alertTestStateStream) Recv() (*pb.State, error)     { return nil, context.Canceled }
+func (alertTestStateStream) SetHeader(metadata.MD) error  { return nil }
+func (alertTestStateStream) SendHeader(metadata.MD) error { return nil }
+func (alertTestStateStream) SetTrailer(metadata.MD)       {}
+func (alertTestStateStream) Context() context.Context     { return context.Background() }
+func (alertTestStateStream) SendMsg(any) error            { return nil }
+func (alertTestStateStream) RecvMsg(any) error            { return nil }
+
+func TestAlertSentinelCountsOnlyFreshDistinctMetricReports(t *testing.T) {
+	oldStore, oldPrev, oldLast, oldCycle := alertsStore, alertsPrevState, alertsLastMetricSeq, AlertsCycleTransferStatsStore
+	alertsStore = map[uint64]map[uint64][]model.TimedAlertPoint{7: {1: nil}}
+	alertsPrevState = map[uint64]map[uint64]uint8{7: {}}
+	alertsLastMetricSeq = map[uint64]map[uint64]uint64{7: {}}
+	AlertsCycleTransferStatsStore = map[uint64]*model.CycleTransferStats{}
+	t.Cleanup(func() {
+		alertsStore, alertsPrevState, alertsLastMetricSeq, AlertsCycleTransferStatsStore = oldStore, oldPrev, oldLast, oldCycle
+	})
+
+	alert := &model.AlertRule{
+		Common: model.Common{ID: 7},
+		Rules:  []*model.Rule{{Type: "cpu", Max: 80, Duration: 100, Cover: model.RuleCoverAll}},
+	}
+	server := &model.Server{Common: model.Common{ID: 1}}
+	model.InitServer(server)
+	lease := server.AttachStateStream(alertTestStateStream{})
+	fixedReportTime := time.Now()
+	if !lease.UpdateState(&model.HostState{CPU: 99}, fixedReportTime) {
+		t.Fatal("state update rejected")
+	}
+	checkStatusForServer(alert, server)
+	if got := len(alertsStore[7][1]); got != 1 {
+		t.Fatalf("first report produced %d samples, want 1", got)
+	}
+	for i := 0; i < 100; i++ {
+		checkStatusForServer(alert, server)
+	}
+	if got := len(alertsStore[7][1]); got != 1 {
+		t.Fatalf("same report was counted %d times, want 1", got)
+	}
+	if !lease.Clear() {
+		t.Fatal("stream clear rejected")
+	}
+	checkStatusForServer(alert, server)
+	if got := len(alertsStore[7][1]); got != 0 {
+		t.Fatalf("disconnect retained %d high samples, want 0", got)
+	}
+	alertsPrevState[7][1] = _RuleCheckFail // incident was active before disconnect
+	newLease := server.AttachStateStream(alertTestStateStream{})
+	// Distinct reports may share a clock timestamp on coarse-resolution hosts.
+	if !newLease.UpdateState(&model.HostState{CPU: 99}, fixedReportTime) {
+		t.Fatal("reconnect state update rejected")
+	}
+	checkStatusForServer(alert, server)
+	if got := len(alertsStore[7][1]); got != 1 || alertsStore[7][1][0].Values[0] {
+		t.Fatalf("reconnect window = %v, want one high sample", alertsStore[7][1])
+	}
+	if alertsPrevState[7][1] != _RuleCheckFail {
+		t.Fatal("incomplete warm-up falsely resolved active incident")
+	}
+	replacement := &model.Server{Common: model.Common{ID: 1}}
+	model.InitServer(replacement)
+	replacementLease := replacement.AttachStateStream(alertTestStateStream{})
+	if !replacementLease.UpdateState(&model.HostState{CPU: 99}, fixedReportTime) {
+		t.Fatal("replacement server state update rejected")
+	}
+	checkStatusForServer(alert, replacement)
+	if got := len(alertsStore[7][1]); got != 2 {
+		t.Fatalf("replacement holder reused a report identity; samples=%d", got)
+	}
+}
+
+func TestAlertNewIncidentNotMutedByEarlierInflightIncident(t *testing.T) {
+	oldCache, oldNotifications := Cache, NotificationShared
+	Cache = cache.New(time.Minute, time.Minute)
+	n := &model.Notification{Common: model.Common{ID: 601}, Name: "hook"}
+	nc := newNotificationClassWithItems(n)
+	nc.UpdateGroup(&model.NotificationGroup{Common: model.Common{ID: 603}, Name: "alert-test"}, []uint64{n.ID})
+	NotificationShared = nc
+	t.Cleanup(func() {
+		cancelAlertDeliveries(602)
+		NotificationShared, Cache = oldNotifications, oldCache
+	})
+	started := make(chan struct{})
+	release := make(chan struct{})
+	delivered := make(chan string, 3)
+	nc.sendForTest = func(_ *model.Notification, desc string, _ *model.Server) error {
+		if desc == "first incident" {
+			close(started)
+			<-release
+		}
+		delivered <- desc
+		return nil
+	}
+	alert := &model.AlertRule{
+		Common:              model.Common{ID: 602},
+		NotificationGroupID: 603,
+		TriggerMode:         model.ModeOnetimeTrigger,
+	}
+	server := &model.Server{Common: model.Common{ID: 604}}
+	label := NotificationMuteLabel.ServerIncident(server.ID, alert.ID)
+	startAlertDelivery(alert, server, "first incident", label, alertDeliveryIncident)
+	<-started
+	startAlertDelivery(alert, server, "recovered", NotificationMuteLabel.ServerIncidentResolved(server.ID, alert.ID), alertDeliveryRecovery)
+	startAlertDelivery(alert, server, "second incident", label, alertDeliveryIncident)
+	close(release)
+	timer := time.NewTimer(time.Second)
+	defer timer.Stop()
+	seen := map[string]bool{}
+	for len(seen) < 2 {
+		select {
+		case message := <-delivered:
+			seen[message] = true
+		case <-timer.C:
+			t.Fatalf("new incident was suppressed after old send completed: %v", seen)
+		}
+	}
+	if !seen["first incident"] || !seen["second incident"] || seen["recovered"] {
+		t.Fatalf("unexpected transition deliveries: %v", seen)
+	}
+	alertDeliveryMu.Lock()
+	done := alertDeliveries[alert.ID][server.ID].done
+	alertDeliveryMu.Unlock()
+	select {
+	case <-done:
+	case <-time.After(time.Second):
+		t.Fatal("new incident delivery did not finish")
+	}
+}
+
+func TestAlertRecoveryDeliveryStopsAfterSuccessInAlwaysMode(t *testing.T) {
+	oldCache, oldNotifications := Cache, NotificationShared
+	Cache = cache.New(time.Minute, time.Minute)
+	n := &model.Notification{Common: model.Common{ID: 611}, Name: "hook"}
+	nc := newNotificationClassWithItems(n)
+	nc.UpdateGroup(&model.NotificationGroup{Common: model.Common{ID: 613}, Name: "recovery-test"}, []uint64{n.ID})
+	NotificationShared = nc
+	t.Cleanup(func() {
+		cancelAlertDeliveries(612)
+		NotificationShared, Cache = oldNotifications, oldCache
+	})
+
+	delivered := make(chan struct{}, 2)
+	nc.sendForTest = func(*model.Notification, string, *model.Server) error {
+		delivered <- struct{}{}
+		return nil
+	}
+	alert := &model.AlertRule{
+		Common:              model.Common{ID: 612},
+		NotificationGroupID: 613,
+		TriggerMode:         model.ModeAlwaysTrigger,
+	}
+	server := &model.Server{Common: model.Common{ID: 614}}
+	startAlertDelivery(alert, server, "recovered", "recovery-event", alertDeliveryRecovery)
+
+	select {
+	case <-delivered:
+	case <-time.After(time.Second):
+		t.Fatal("recovery notification was not delivered")
+	}
+	alertDeliveryMu.Lock()
+	done := alertDeliveries[alert.ID][server.ID].done
+	alertDeliveryMu.Unlock()
+	select {
+	case <-done:
+	case <-time.After(time.Second):
+		t.Fatal("successful recovery delivery kept repeating in always mode")
+	}
+	select {
+	case <-delivered:
+		t.Fatal("recovery notification was delivered more than once")
+	default:
+	}
+}
+
+func TestServerDeleteCancelsAlertDelivery(t *testing.T) {
+	alertDeliveryMu.Lock()
+	oldDeliveries := alertDeliveries
+	cancel := make(chan struct{})
+	alertDeliveries = map[uint64]map[uint64]*alertDeliveryEntry{
+		621: {622: {cancel: cancel, done: make(chan struct{})}},
+	}
+	alertDeliveryMu.Unlock()
+	t.Cleanup(func() {
+		alertDeliveryMu.Lock()
+		alertDeliveries = oldDeliveries
+		alertDeliveryMu.Unlock()
+	})
+
+	servers := &ServerClass{
+		class: class[uint64, *model.Server]{
+			list: map[uint64]*model.Server{
+				622: {Common: model.Common{ID: 622}, UUID: "delete-alert-worker"},
+			},
+		},
+		uuidToID: map[string]uint64{"delete-alert-worker": 622},
+	}
+	servers.Delete([]uint64{622})
+
+	select {
+	case <-cancel:
+	default:
+		t.Fatal("server deletion did not cancel its alert delivery")
+	}
+	alertDeliveryMu.Lock()
+	_, retained := alertDeliveries[621]
+	alertDeliveryMu.Unlock()
+	if retained {
+		t.Fatal("server deletion retained an empty alert delivery map")
+	}
+}
+
+func TestAlwaysAlertContinuesFailureTriggerTasks(t *testing.T) {
+	if !shouldRunAlertFailTasks(model.ModeAlwaysTrigger, false) {
+		t.Fatal("always mode stopped failure trigger tasks after the initial incident")
+	}
+	if shouldRunAlertFailTasks(model.ModeOnetimeTrigger, false) {
+		t.Fatal("one-time mode repeated failure trigger tasks")
+	}
+	if !shouldRunAlertFailTasks(model.ModeOnetimeTrigger, true) {
+		t.Fatal("one-time mode skipped failure trigger tasks for a new incident")
+	}
+}
+
+func TestAlertDeliveryRetriesFailureAndStopsAfterSuccess(t *testing.T) {
+	cancel := make(chan struct{})
+	done := make(chan struct{})
+	attempts := 0
+	go func() {
+		runAlertDelivery(cancel, func() bool {
+			attempts++
+			return attempts == 2
+		}, false, time.Millisecond)
+		close(done)
+	}()
+	select {
+	case <-done:
+	case <-time.After(time.Second):
+		close(cancel)
+		t.Fatal("delivery did not retry and finish")
+	}
+	if attempts != 2 {
+		t.Fatalf("got %d attempts, want 2", attempts)
+	}
+}
+
+func TestAlertDeliveryCancelsAfterRecovery(t *testing.T) {
+	cancel := make(chan struct{})
+	done := make(chan struct{})
+	started := make(chan struct{})
+	go func() {
+		runAlertDelivery(cancel, func() bool {
+			close(started)
+			return true
+		}, true, time.Millisecond)
+		close(done)
+	}()
+	<-started
+	close(cancel)
+	select {
+	case <-done:
+	case <-time.After(time.Second):
+		t.Fatal("recovered incident delivery was not cancelled")
+	}
+}

service/singleton/alertsentinel_test.go

@@ -122,7 +122,7 @@ func TestCheckStatusForServerRecoversUnexpectedEvaluatorPanic(t *testing.T) {
 	originalStore := alertsStore
 	originalPrevState := alertsPrevState
 	originalCycleStore := AlertsCycleTransferStatsStore
-	alertsStore = map[uint64]map[uint64][][]bool{}
+	alertsStore = map[uint64]map[uint64][]model.TimedAlertPoint{}
 	alertsPrevState = map[uint64]map[uint64]uint8{}
 	AlertsCycleTransferStatsStore = map[uint64]*model.CycleTransferStats{}
 	t.Cleanup(func() {

service/singleton/notification.go

@@ -22,8 +22,228 @@ type NotificationClass struct {
 	groupToIDList map[uint64]map[uint64]*model.Notification
 	idToGroupList map[uint64]map[uint64]struct{}
 
-	groupList map[uint64]string
-	groupMu   sync.RWMutex
+	groupList       map[uint64]string
+	groupMu         sync.RWMutex
+	deliveryMu      sync.Mutex
+	deliveryKeys    map[string]*notificationKeyLock
+	asyncInFlight   map[string]struct{}
+	asyncRetryAfter map[string]time.Time
+	asyncRetryDelay map[string]time.Duration
+	muteGeneration  map[string]uint64
+	serviceDelivery map[uint64]*orderedServiceDelivery
+	sendForTest     func(*model.Notification, string, *model.Server) error
+}
+
+type serviceDeliveryRequest struct {
+	groupID uint64
+	desc    string
+	label   string
+}
+
+type orderedServiceDelivery struct {
+	current serviceDeliveryRequest
+	pending *serviceDeliveryRequest
+	update  chan struct{}
+	cancel  chan struct{}
+	done    chan struct{}
+}
+
+// SendServiceState preserves incident→recovery order without allowing a
+// slow webhook to accumulate one goroutine per probe result. Only the latest
+// pending phase is retained; a failed current phase retries with backoff.
+func (c *NotificationClass) SendServiceState(serviceID, groupID uint64, desc, label string) {
+	req := serviceDeliveryRequest{groupID: groupID, desc: desc, label: label}
+	c.deliveryMu.Lock()
+	if c.serviceDelivery == nil {
+		c.serviceDelivery = make(map[uint64]*orderedServiceDelivery)
+	}
+	state := c.serviceDelivery[serviceID]
+	if state == nil {
+		state = &orderedServiceDelivery{current: req, update: make(chan struct{}, 1), cancel: make(chan struct{}), done: make(chan struct{})}
+		c.serviceDelivery[serviceID] = state
+		c.deliveryMu.Unlock()
+		go c.runServiceDelivery(serviceID, state)
+		return
+	}
+	if state.pending == nil && state.current.label == label {
+		c.deliveryMu.Unlock()
+		return
+	}
+	if state.pending == nil || state.pending.label != label {
+		state.pending = &req
+		select {
+		case state.update <- struct{}{}:
+		default:
+		}
+	}
+	c.deliveryMu.Unlock()
+}
+
+func (c *NotificationClass) runServiceDelivery(serviceID uint64, state *orderedServiceDelivery) {
+	defer close(state.done)
+	delay := 30 * time.Second
+	for {
+		select {
+		case <-state.cancel:
+			return
+		default:
+		}
+		delivered := c.SendNotification(state.current.groupID, state.current.desc, state.current.label)
+		c.deliveryMu.Lock()
+		// Do not advance to a later phase until every recipient has received the
+		// current one. Otherwise a failed incident can be discarded as soon as a
+		// recovery is queued, leaving that recipient with a recovery but no incident.
+		if delivered && state.pending != nil {
+			next := state.pending
+			state.current = *next
+			state.pending = nil
+			delay = 30 * time.Second
+			select {
+			case <-state.update:
+			default:
+			}
+			c.deliveryMu.Unlock()
+			continue
+		}
+		if delivered {
+			if c.serviceDelivery[serviceID] == state {
+				delete(c.serviceDelivery, serviceID)
+			}
+			c.deliveryMu.Unlock()
+			return
+		}
+		c.deliveryMu.Unlock()
+		timer := time.NewTimer(delay)
+		select {
+		case <-state.cancel:
+			timer.Stop()
+			return
+		case <-state.update:
+			timer.Stop()
+		case <-timer.C:
+			delay = min(delay*2, 15*time.Minute)
+		}
+	}
+}
+
+func (c *NotificationClass) CancelServiceState(serviceID uint64) {
+	if c == nil {
+		return
+	}
+	c.deliveryMu.Lock()
+	if state := c.serviceDelivery[serviceID]; state != nil {
+		close(state.cancel)
+		delete(c.serviceDelivery, serviceID)
+	}
+	c.deliveryMu.Unlock()
+}
+
+// SendNotificationAsync coalesces concurrent service-probe reports for the
+// same event before allocating a webhook goroutine. Alert transitions use the
+// synchronous SendNotification path so they retain their retry semantics.
+func (c *NotificationClass) SendNotificationAsync(groupID uint64, desc, muteLabel string, ext ...*model.Server) {
+	c.sendNotificationAsync(groupID, desc, muteLabel, nil, ext...)
+}
+
+func (c *NotificationClass) sendNotificationAsync(groupID uint64, desc, muteLabel string, onSuccess func(), ext ...*model.Server) {
+	key := fmt.Sprintf("%d:%s", groupID, muteLabel)
+	c.deliveryMu.Lock()
+	if c.asyncInFlight == nil {
+		c.asyncInFlight = make(map[string]struct{})
+	}
+	if _, busy := c.asyncInFlight[key]; busy {
+		c.deliveryMu.Unlock()
+		return
+	}
+	if time.Now().Before(c.asyncRetryAfter[key]) {
+		c.deliveryMu.Unlock()
+		return
+	}
+	c.asyncInFlight[key] = struct{}{}
+	c.deliveryMu.Unlock()
+	go func() {
+		delivered := c.SendNotification(groupID, desc, muteLabel, ext...)
+		if delivered && onSuccess != nil {
+			onSuccess()
+		}
+		c.deliveryMu.Lock()
+		if delivered {
+			delete(c.asyncRetryAfter, key)
+			delete(c.asyncRetryDelay, key)
+		} else {
+			if c.asyncRetryAfter == nil {
+				c.asyncRetryAfter = make(map[string]time.Time)
+				c.asyncRetryDelay = make(map[string]time.Duration)
+			}
+			delay := c.asyncRetryDelay[key]
+			if delay == 0 {
+				delay = 30 * time.Second
+			} else {
+				delay = min(delay*2, 15*time.Minute)
+			}
+			c.asyncRetryDelay[key] = delay
+			c.asyncRetryAfter[key] = time.Now().Add(delay)
+			c.pruneAsyncBackoffLocked()
+		}
+		delete(c.asyncInFlight, key)
+		c.deliveryMu.Unlock()
+	}()
+}
+
+const maxAsyncBackoffKeys = 4096
+
+// Bound memory when a failing webhook sees many short-lived event labels.
+// Evicted old event keys are no longer active after their phase changes.
+func (c *NotificationClass) pruneAsyncBackoffLocked() {
+	if len(c.asyncRetryAfter) <= maxAsyncBackoffKeys {
+		return
+	}
+	now := time.Now()
+	for key, until := range c.asyncRetryAfter {
+		if !now.Before(until) {
+			delete(c.asyncRetryAfter, key)
+			delete(c.asyncRetryDelay, key)
+		}
+	}
+	for key := range c.asyncRetryAfter {
+		if len(c.asyncRetryAfter) <= maxAsyncBackoffKeys {
+			break
+		}
+		delete(c.asyncRetryAfter, key)
+		delete(c.asyncRetryDelay, key)
+	}
+}
+
+type notificationKeyLock struct {
+	mu   sync.Mutex
+	refs int
+}
+
+func (c *NotificationClass) lockDeliveryKey(key string) func() {
+	if key == "" {
+		return func() {}
+	}
+	c.deliveryMu.Lock()
+	if c.deliveryKeys == nil {
+		c.deliveryKeys = make(map[string]*notificationKeyLock)
+	}
+	entry := c.deliveryKeys[key]
+	if entry == nil {
+		entry = new(notificationKeyLock)
+		c.deliveryKeys[key] = entry
+	}
+	entry.refs++
+	c.deliveryMu.Unlock()
+	entry.mu.Lock()
+	return func() {
+		entry.mu.Unlock()
+		c.deliveryMu.Lock()
+		entry.refs--
+		if entry.refs == 0 {
+			delete(c.deliveryKeys, key)
+		}
+		c.deliveryMu.Unlock()
+	}
 }
 
 func NewNotificationClass() *NotificationClass {
@@ -202,44 +422,29 @@ func (c *NotificationClass) sortList() {
 
 func (c *NotificationClass) UnMuteNotification(notificationGroupID uint64, muteLabel string) {
 	fullMuteLabel := NotificationMuteLabel.AppendNotificationGroupName(muteLabel, c.GetGroupName(notificationGroupID))
-	Cache.Delete(fullMuteLabel)
+	c.listMu.RLock()
+	for id := range c.groupToIDList[notificationGroupID] {
+		key := notificationMethodMuteLabel(fullMuteLabel, id)
+		c.deliveryMu.Lock()
+		if c.muteGeneration == nil {
+			c.muteGeneration = make(map[string]uint64)
+		}
+		c.muteGeneration[key]++
+		Cache.Delete(key)
+		c.deliveryMu.Unlock()
+	}
+	c.listMu.RUnlock()
+}
+
+func notificationMethodMuteLabel(label string, id uint64) string {
+	return fmt.Sprintf("%s:method-%d", label, id)
 }
 
 // SendNotification 向指定的通知方式组的所有通知方式发送通知
-func (c *NotificationClass) SendNotification(notificationGroupID uint64, desc string, muteLabel string, ext ...*model.Server) {
+func (c *NotificationClass) SendNotification(notificationGroupID uint64, desc string, muteLabel string, ext ...*model.Server) bool {
+	fullMuteLabel := ""
 	if muteLabel != "" {
-		// 将通知方式组名称加入静音标志
-		muteLabel := NotificationMuteLabel.AppendNotificationGroupName(muteLabel, c.GetGroupName(notificationGroupID))
-		// 通知防骚扰策略
-		var flag bool
-		if cacheN, has := Cache.Get(muteLabel); has {
-			nHistory := cacheN.(NotificationHistory)
-			// 每次提醒都增加一倍等待时间，最后每天最多提醒一次
-			if time.Now().After(nHistory.Until) {
-				flag = true
-				nHistory.Duration *= 2
-				if nHistory.Duration > time.Hour*24 {
-					nHistory.Duration = time.Hour * 24
-				}
-				nHistory.Until = time.Now().Add(nHistory.Duration)
-				// 缓存有效期加 10 分钟
-				Cache.Set(muteLabel, nHistory, nHistory.Duration+time.Minute*10)
-			}
-		} else {
-			// 新提醒直接通知
-			flag = true
-			Cache.Set(muteLabel, NotificationHistory{
-				Duration: firstNotificationDelay,
-				Until:    time.Now().Add(firstNotificationDelay),
-			}, firstNotificationDelay+time.Minute*10)
-		}
-
-		if !flag {
-			if Conf.Debug {
-				log.Println("NEZHA>> Muted repeated notification", desc, muteLabel)
-			}
-			return
-		}
+		fullMuteLabel = NotificationMuteLabel.AppendNotificationGroupName(muteLabel, c.GetGroupName(notificationGroupID))
 	}
 	// Copy the group under the lock. Webhook delivery can take minutes and must
 	// not block notification or notification-group updates while it is in flight.
@@ -250,24 +455,82 @@ func (c *NotificationClass) SendNotification(notificationGroupID uint64, desc st
 	}
 	c.listMu.RUnlock()
 
-	for _, n := range notifications {
-		log.Printf("NEZHA>> Try to notify %s", n.Name)
+	if len(notifications) == 0 {
+		log.Printf("NEZHA>> Notification group %d has no recipients", notificationGroupID)
+		return false
 	}
+	allDelivered := true
 	for _, n := range notifications {
-		ns := model.NotificationServerBundle{
-			Notification: n,
-			Server:       nil,
-			Loc:          Loc,
-		}
+		var server *model.Server
 		if len(ext) > 0 {
-			ns.Server = ext[0]
+			server = ext[0]
 		}
-		if err := ns.Send(desc); err != nil {
-			log.Printf("NEZHA>> Sending notification to %s failed: %v", n.Name, err)
-		} else {
-			log.Printf("NEZHA>> Sending notification to %s succeeded", n.Name)
+		if !c.sendRecipient(n, desc, fullMuteLabel, server) {
+			allDelivered = false
 		}
 	}
+	return allDelivered
+}
+
+func (c *NotificationClass) sendRecipient(n *model.Notification, desc, fullMuteLabel string, server *model.Server) bool {
+	if n == nil {
+		return false
+	}
+	cacheKey := ""
+	if fullMuteLabel != "" {
+		cacheKey = notificationMethodMuteLabel(fullMuteLabel, n.ID)
+	}
+	unlock := c.lockDeliveryKey(cacheKey)
+	defer unlock()
+	var generation uint64
+	if cacheKey != "" {
+		c.deliveryMu.Lock()
+		generation = c.muteGeneration[cacheKey]
+		c.deliveryMu.Unlock()
+	}
+	var history NotificationHistory
+	if cacheKey != "" {
+		if value, exists := Cache.Get(cacheKey); exists {
+			history = value.(NotificationHistory)
+			if time.Now().Before(history.Until) {
+				return true // this recipient already received this phase
+			}
+		}
+	}
+	log.Printf("NEZHA>> Try to notify %s", n.Name)
+	var err error
+	if c.sendForTest != nil {
+		err = c.sendForTest(n, desc, server)
+	} else {
+		err = (&model.NotificationServerBundle{Notification: n, Server: server, Loc: Loc}).Send(desc)
+	}
+	if err != nil {
+		// Webhook errors can contain a credential-bearing request URL. Keep
+		// failure type without ever writing the URL/token to dashboard logs.
+		log.Printf("NEZHA>> Sending notification to %s failed (%s)", n.Name, notificationFailureSummary(err))
+		return false
+	}
+	log.Printf("NEZHA>> Sending notification to %s succeeded", n.Name)
+	if cacheKey != "" {
+		if history.Duration == 0 {
+			history.Duration = firstNotificationDelay
+		} else {
+			history.Duration = min(history.Duration*2, 24*time.Hour)
+		}
+		history.Until = time.Now().Add(history.Duration)
+		c.deliveryMu.Lock()
+		if c.muteGeneration[cacheKey] == generation {
+			Cache.Set(cacheKey, history, history.Duration+10*time.Minute)
+		}
+		c.deliveryMu.Unlock()
+	}
+	return true
+}
+
+func notificationFailureSummary(err error) string {
+	// Even wrapped URL errors can include credentials in path/query. Log only
+	// the error type; the HTTP status path already returns a dedicated type.
+	return fmt.Sprintf("%T", err)
 }
 
 type _NotificationMuteLabel struct{}

service/singleton/notification_test.go

@@ -1,12 +1,405 @@
 package singleton
 
 import (
+	"errors"
+	"fmt"
+	"net/url"
+	"strings"
+	"sync/atomic"
 	"testing"
 	"time"
 
+	"github.com/patrickmn/go-cache"
+
 	"github.com/nezhahq/nezha/model"
 )
 
+func TestNotificationFailureDoesNotMuteRetryOrResendSuccessfulRecipient(t *testing.T) {
+	oldCache := Cache
+	Cache = cache.New(time.Minute, time.Minute)
+	t.Cleanup(func() { Cache = oldCache })
+	first := &model.Notification{Common: model.Common{ID: 301}, Name: "first"}
+	second := &model.Notification{Common: model.Common{ID: 302}, Name: "second"}
+	nc := newNotificationClassWithItems(first, second)
+	nc.UpdateGroup(&model.NotificationGroup{Common: model.Common{ID: 303}, Name: "retry-test"}, []uint64{first.ID, second.ID})
+	counts := map[uint64]int{}
+	nc.sendForTest = func(n *model.Notification, _ string, _ *model.Server) error {
+		counts[n.ID]++
+		if n.ID == second.ID && counts[n.ID] == 1 {
+			return errors.New("temporary failure")
+		}
+		return nil
+	}
+	label := "alert-retry-test"
+	nc.UnMuteNotification(303, label)
+	if nc.SendNotification(303, "incident", label) {
+		t.Fatal("partial delivery must report failure")
+	}
+	if !nc.SendNotification(303, "incident", label) {
+		t.Fatal("failed recipient should be retried successfully")
+	}
+	if counts[first.ID] != 1 || counts[second.ID] != 2 {
+		t.Fatalf("delivery attempts = %v, want first=1 second=2", counts)
+	}
+	nc.UnMuteNotification(303, label)
+}
+
+func TestNotificationConcurrentSameRecipientSendsOnce(t *testing.T) {
+	oldCache := Cache
+	Cache = cache.New(time.Minute, time.Minute)
+	t.Cleanup(func() { Cache = oldCache })
+	n := &model.Notification{Common: model.Common{ID: 401}, Name: "hook"}
+	nc := newNotificationClassWithItems(n)
+	nc.UpdateGroup(&model.NotificationGroup{Common: model.Common{ID: 402}, Name: "concurrent-test"}, []uint64{n.ID})
+	entered := make(chan struct{})
+	release := make(chan struct{})
+	var calls atomic.Int32
+	nc.sendForTest = func(*model.Notification, string, *model.Server) error {
+		if calls.Add(1) == 1 {
+			close(entered)
+			<-release
+		}
+		return nil
+	}
+	done := make(chan bool, 2)
+	go func() { done <- nc.SendNotification(402, "incident", "same-event") }()
+	<-entered
+	go func() { done <- nc.SendNotification(402, "incident", "same-event") }()
+	close(release)
+	if !<-done || !<-done {
+		t.Fatal("concurrent delivery returned failure")
+	}
+	if calls.Load() != 1 {
+		t.Fatalf("same event delivered %d times, want 1", calls.Load())
+	}
+}
+
+func TestNotificationAsyncCoalescesBlockedServiceReports(t *testing.T) {
+	oldCache := Cache
+	Cache = cache.New(time.Minute, time.Minute)
+	t.Cleanup(func() { Cache = oldCache })
+	n := &model.Notification{Common: model.Common{ID: 501}, Name: "hook"}
+	nc := newNotificationClassWithItems(n)
+	nc.UpdateGroup(&model.NotificationGroup{Common: model.Common{ID: 502}, Name: "service-test"}, []uint64{n.ID})
+	entered := make(chan struct{})
+	release := make(chan struct{})
+	var calls atomic.Int32
+	nc.sendForTest = func(*model.Notification, string, *model.Server) error {
+		if calls.Add(1) == 1 {
+			close(entered)
+			<-release
+		}
+		return nil
+	}
+	nc.SendNotificationAsync(502, "down", "service-down")
+	<-entered
+	for i := 0; i < 1000; i++ {
+		nc.SendNotificationAsync(502, "down", "service-down")
+	}
+	nc.deliveryMu.Lock()
+	inFlight := len(nc.asyncInFlight)
+	nc.deliveryMu.Unlock()
+	if inFlight != 1 || calls.Load() != 1 {
+		t.Fatalf("in-flight sends=%d attempts=%d, want one each", inFlight, calls.Load())
+	}
+	close(release)
+	deadline := time.After(time.Second)
+	for {
+		nc.deliveryMu.Lock()
+		remaining := len(nc.asyncInFlight)
+		nc.deliveryMu.Unlock()
+		if remaining == 0 {
+			break
+		}
+		select {
+		case <-deadline:
+			t.Fatal("async notification did not finish")
+		case <-time.After(time.Millisecond):
+		}
+	}
+}
+
+func TestNotificationFailureSummaryOmitsCredentialURL(t *testing.T) {
+	err := &url.Error{Op: "Post", URL: "https://example.test/bot-secret-token/send?key=secret-query", Err: errors.New("timeout")}
+	if summary := notificationFailureSummary(err); strings.Contains(summary, "secret") || strings.Contains(summary, "example.test") {
+		t.Fatalf("credential-bearing URL leaked: %s", summary)
+	}
+}
+
+func TestNotificationAsyncBacksOffFastFailures(t *testing.T) {
+	oldCache := Cache
+	Cache = cache.New(time.Minute, time.Minute)
+	t.Cleanup(func() { Cache = oldCache })
+	n := &model.Notification{Common: model.Common{ID: 511}, Name: "hook"}
+	nc := newNotificationClassWithItems(n)
+	nc.UpdateGroup(&model.NotificationGroup{Common: model.Common{ID: 512}, Name: "backoff-test"}, []uint64{n.ID})
+	var calls atomic.Int32
+	nc.sendForTest = func(*model.Notification, string, *model.Server) error {
+		calls.Add(1)
+		return errors.New("fast failure")
+	}
+	key := "512:service-down"
+	nc.SendNotificationAsync(512, "down", "service-down")
+	deadline := time.After(time.Second)
+	for {
+		nc.deliveryMu.Lock()
+		busy := len(nc.asyncInFlight) != 0
+		after := nc.asyncRetryAfter[key]
+		nc.deliveryMu.Unlock()
+		if !busy && !after.IsZero() {
+			break
+		}
+		select {
+		case <-deadline:
+			t.Fatal("first failed send did not record backoff")
+		case <-time.After(time.Millisecond):
+		}
+	}
+	for i := 0; i < 1000; i++ {
+		nc.SendNotificationAsync(512, "down", "service-down")
+	}
+	if calls.Load() != 1 {
+		t.Fatalf("fast failures caused %d attempts before retry-after", calls.Load())
+	}
+}
+
+func TestUnMuteDuringInflightSendCannotRemuteRecovery(t *testing.T) {
+	oldCache := Cache
+	Cache = cache.New(time.Minute, time.Minute)
+	t.Cleanup(func() { Cache = oldCache })
+	n := &model.Notification{Common: model.Common{ID: 521}, Name: "hook"}
+	nc := newNotificationClassWithItems(n)
+	nc.UpdateGroup(&model.NotificationGroup{Common: model.Common{ID: 522}, Name: "generation-test"}, []uint64{n.ID})
+	entered := make(chan struct{})
+	release := make(chan struct{})
+	var calls atomic.Int32
+	nc.sendForTest = func(*model.Notification, string, *model.Server) error {
+		if calls.Add(1) == 1 {
+			close(entered)
+			<-release
+		}
+		return nil
+	}
+	done := make(chan bool, 1)
+	go func() { done <- nc.SendNotification(522, "first failure", "static-label") }()
+	<-entered
+	nc.UnMuteNotification(522, "static-label")
+	close(release)
+	if !<-done {
+		t.Fatal("first send failed unexpectedly")
+	}
+	if !nc.SendNotification(522, "next failure", "static-label") || calls.Load() != 2 {
+		t.Fatalf("old in-flight send re-muted next failure: calls=%d", calls.Load())
+	}
+}
+
+func TestAsyncBackoffMemoryIsBounded(t *testing.T) {
+	nc := &NotificationClass{
+		asyncRetryAfter: make(map[string]time.Time),
+		asyncRetryDelay: make(map[string]time.Duration),
+	}
+	for i := 0; i < maxAsyncBackoffKeys+100; i++ {
+		key := fmt.Sprintf("old-event-%d", i)
+		nc.asyncRetryAfter[key] = time.Now().Add(time.Minute)
+		nc.asyncRetryDelay[key] = time.Minute
+	}
+	nc.pruneAsyncBackoffLocked()
+	if len(nc.asyncRetryAfter) > maxAsyncBackoffKeys || len(nc.asyncRetryDelay) != len(nc.asyncRetryAfter) {
+		t.Fatalf("unbounded backoff maps: after=%d delay=%d", len(nc.asyncRetryAfter), len(nc.asyncRetryDelay))
+	}
+}
+
+func TestServiceStateNotificationsKeepIncidentRecoveryOrder(t *testing.T) {
+	oldCache := Cache
+	Cache = cache.New(time.Minute, time.Minute)
+	t.Cleanup(func() { Cache = oldCache })
+	n := &model.Notification{Common: model.Common{ID: 531}, Name: "hook"}
+	nc := newNotificationClassWithItems(n)
+	nc.UpdateGroup(&model.NotificationGroup{Common: model.Common{ID: 532}, Name: "ordered-test"}, []uint64{n.ID})
+	entered := make(chan struct{})
+	release := make(chan struct{})
+	delivered := make(chan string, 3)
+	nc.sendForTest = func(_ *model.Notification, desc string, _ *model.Server) error {
+		if desc == "down" {
+			close(entered)
+			<-release
+		}
+		delivered <- desc
+		return nil
+	}
+	nc.SendServiceState(533, 532, "down", "state-down-1")
+	<-entered
+	nc.SendServiceState(533, 532, "good", "state-good-2")
+	select {
+	case message := <-delivered:
+		t.Fatalf("service transition overtook blocked incident: %s", message)
+	default:
+	}
+	close(release)
+	for _, want := range []string{"down", "good"} {
+		select {
+		case got := <-delivered:
+			if got != want {
+				t.Fatalf("delivery order got %s, want %s", got, want)
+			}
+		case <-time.After(time.Second):
+			t.Fatalf("missing %s delivery", want)
+		}
+	}
+	deadline := time.After(time.Second)
+	for {
+		nc.deliveryMu.Lock()
+		remaining := len(nc.serviceDelivery)
+		nc.deliveryMu.Unlock()
+		if remaining == 0 {
+			break
+		}
+		select {
+		case <-deadline:
+			t.Fatal("ordered delivery worker did not exit")
+		case <-time.After(time.Millisecond):
+		}
+	}
+}
+
+func TestServiceStateRetriesFailedIncidentBeforePendingRecovery(t *testing.T) {
+	oldCache := Cache
+	Cache = cache.New(time.Minute, time.Minute)
+	t.Cleanup(func() { Cache = oldCache })
+	n := &model.Notification{Common: model.Common{ID: 535}, Name: "hook"}
+	nc := newNotificationClassWithItems(n)
+	nc.UpdateGroup(&model.NotificationGroup{Common: model.Common{ID: 536}, Name: "retry-order-test"}, []uint64{n.ID})
+	t.Cleanup(func() { nc.CancelServiceState(537) })
+
+	attempts := make(chan string, 3)
+	releaseFirst := make(chan struct{})
+	var downAttempts atomic.Int32
+	nc.sendForTest = func(_ *model.Notification, desc string, _ *model.Server) error {
+		attempts <- desc
+		if desc == "down" && downAttempts.Add(1) == 1 {
+			<-releaseFirst
+			return errors.New("temporary incident failure")
+		}
+		return nil
+	}
+
+	nc.SendServiceState(537, 536, "down", "state-down-1")
+	nc.deliveryMu.Lock()
+	state := nc.serviceDelivery[537]
+	nc.deliveryMu.Unlock()
+	select {
+	case got := <-attempts:
+		if got != "down" {
+			t.Fatalf("first delivery = %q, want down", got)
+		}
+	case <-time.After(time.Second):
+		t.Fatal("incident delivery did not start")
+	}
+	nc.SendServiceState(537, 536, "good", "state-good-2")
+	close(releaseFirst)
+
+	for _, want := range []string{"down", "good"} {
+		select {
+		case got := <-attempts:
+			if got != want {
+				t.Fatalf("delivery after failed incident = %q, want %q", got, want)
+			}
+		case <-time.After(time.Second):
+			t.Fatalf("missing %s delivery", want)
+		}
+	}
+	select {
+	case <-state.done:
+	case <-time.After(time.Second):
+		t.Fatal("ordered delivery worker did not finish")
+	}
+}
+
+func TestServiceStateCancelStopsOldConfiguration(t *testing.T) {
+	oldCache := Cache
+	Cache = cache.New(time.Minute, time.Minute)
+	t.Cleanup(func() { Cache = oldCache })
+	n := &model.Notification{Common: model.Common{ID: 541}, Name: "hook"}
+	nc := newNotificationClassWithItems(n)
+	nc.UpdateGroup(&model.NotificationGroup{Common: model.Common{ID: 542}, Name: "old-group"}, []uint64{n.ID})
+	entered := make(chan struct{})
+	release := make(chan struct{})
+	var calls atomic.Int32
+	nc.sendForTest = func(*model.Notification, string, *model.Server) error {
+		if calls.Add(1) == 1 {
+			close(entered)
+			<-release
+		}
+		return errors.New("old webhook failed")
+	}
+	nc.SendServiceState(543, 542, "down", "old-event")
+	<-entered
+	nc.deliveryMu.Lock()
+	state := nc.serviceDelivery[543]
+	nc.deliveryMu.Unlock()
+	nc.CancelServiceState(543)
+	nc.deliveryMu.Lock()
+	remaining := len(nc.serviceDelivery)
+	nc.deliveryMu.Unlock()
+	if remaining != 0 {
+		t.Fatal("old service delivery worker remained registered after cancellation")
+	}
+	close(release)
+	select {
+	case <-state.done:
+	case <-time.After(time.Second):
+		t.Fatal("canceled old delivery worker did not exit")
+	}
+}
+
+func TestServiceUpdateCancelsOldRecipientRetry(t *testing.T) {
+	ss := newServiceMonitorSecurityHarness(t)
+	service := &model.Service{
+		Common:              model.Common{ID: 551, UserID: 1},
+		Name:                "generic-probe",
+		Type:                model.TaskTypeTCPPing,
+		Target:              "example.invalid:443",
+		Duration:            3600,
+		Notify:              true,
+		NotificationGroupID: 552,
+		Cover:               model.ServiceCoverIgnoreAll,
+	}
+	addServiceMonitorSecurityService(t, ss, service)
+	n := &model.Notification{Common: model.Common{ID: 553}, Name: "old-hook"}
+	NotificationShared.InsertForTest(n)
+	NotificationShared.UpdateGroup(&model.NotificationGroup{Common: model.Common{ID: 552}, Name: "old-group"}, []uint64{n.ID})
+	entered := make(chan struct{})
+	release := make(chan struct{})
+	NotificationShared.sendForTest = func(*model.Notification, string, *model.Server) error {
+		close(entered)
+		<-release
+		return errors.New("failed old recipient")
+	}
+	NotificationShared.SendServiceState(service.ID, 552, "old down", "old-state")
+	<-entered
+	NotificationShared.deliveryMu.Lock()
+	state := NotificationShared.serviceDelivery[service.ID]
+	NotificationShared.deliveryMu.Unlock()
+	updated := *service
+	updated.Notify = false
+	updated.NotificationGroupID = 554
+	if err := ss.Update(&updated); err != nil {
+		t.Fatal(err)
+	}
+	close(release)
+	select {
+	case <-state.done:
+	case <-time.After(time.Second):
+		t.Fatal("old recipient retry did not stop after service update")
+	}
+	NotificationShared.deliveryMu.Lock()
+	remaining := len(NotificationShared.serviceDelivery)
+	NotificationShared.deliveryMu.Unlock()
+	if remaining != 0 {
+		t.Fatal("old recipient retained after service update")
+	}
+}
+
 func newNotificationClassWithItems(items ...*model.Notification) *NotificationClass {
 	nc := NewEmptyNotificationClassForTest()
 	for _, item := range items {

service/singleton/server.go

@@ -117,6 +117,7 @@ func (c *ServerClass) Update(s *model.Server, uuid string) {
 func (c *ServerClass) Delete(idList []uint64) {
 	c.lockLifecycleWrite()
 	defer c.unlockLifecycleWrite()
+	cancelAlertDeliveriesForServers(idList)
 
 	c.listMu.Lock()
 

service/singleton/servicesentinel.go

@@ -2,6 +2,7 @@ package singleton
 
 import (
 	"cmp"
+	"crypto/sha256"
 	"fmt"
 	"iter"
 	"log"
@@ -9,6 +10,7 @@ import (
 	"slices"
 	"strings"
 	"sync"
+	"sync/atomic"
 	"time"
 
 	"github.com/gin-gonic/gin"
@@ -22,9 +24,11 @@ import (
 )
 
 const (
-	_CurrentStatusSize = 30 // 统计 15 分钟内的数据为当前状态
+	_CurrentStatusSize = 30 // initial capacity for the rolling service window
 )
 
+var serviceEventSequence atomic.Uint64
+
 type serviceResponseItem struct {
 	model.ServiceResponseItem
 
@@ -46,9 +50,101 @@ type _TodayStatsOfService struct {
 type serviceResponseData = _TodayStatsOfService
 
 type serviceTaskStatus struct {
-	lastStatus uint8
-	t          time.Time
-	result     []*pb.TaskResult
+	lastStatus    uint8
+	stateEpoch    uint64
+	lastHistoryAt time.Time
+	history       serviceResponseData
+	lastData      string
+	result        []serviceWindowBucket
+}
+
+type serviceProbeSample struct {
+	successful bool
+	failed     bool
+	delay      float64
+}
+
+type serviceWindowBucket struct {
+	at        time.Time
+	reporters map[uint64]serviceProbeSample
+}
+
+func (status *serviceTaskStatus) appendResult(now time.Time, reporter uint64, result *pb.TaskResult) {
+	bucketAt := now.Truncate(30 * time.Second)
+	if len(status.result) > 0 && bucketAt.Before(status.result[len(status.result)-1].at) {
+		status.result = nil // wall-clock moved backwards; old buckets are not comparable
+	}
+	if len(status.result) == 0 || !status.result[len(status.result)-1].at.Equal(bucketAt) {
+		status.result = append(status.result, serviceWindowBucket{at: bucketAt, reporters: make(map[uint64]serviceProbeSample)})
+	}
+	bucket := &status.result[len(status.result)-1]
+	sample := bucket.reporters[reporter]
+	if result.Successful {
+		sample.successful = true
+		sample.delay = float64(result.Delay)
+	} else {
+		sample.failed = true
+	}
+	bucket.reporters[reporter] = sample
+	cutoff := now.Add(-15 * time.Minute)
+	first := 0
+	for first < len(status.result) && status.result[first].at.Before(cutoff) {
+		first++
+	}
+	if first > 0 {
+		status.result = append([]serviceWindowBucket(nil), status.result[first:]...)
+	}
+	if result.Successful {
+		status.history.Up++
+		status.history.Delay = (status.history.Delay*float64(status.history.Up-1) + float64(result.Delay)) / float64(status.history.Up)
+	} else {
+		status.history.Down++
+	}
+	status.lastData = result.Data
+}
+
+// flushServiceHistory persists the previous non-overlapping reporting period
+// before the next observation is added, so long probe gaps cannot lose or
+// double-count samples in the 15-minute availability history.
+func (ss *ServiceSentinel) flushServiceHistory(serviceID uint64, status *serviceTaskStatus, now time.Time) {
+	if status.lastHistoryAt.IsZero() || now.Before(status.lastHistoryAt.Add(15*time.Minute)) {
+		return
+	}
+	if status.history.Up+status.history.Down == 0 {
+		status.lastHistoryAt = now
+		return
+	}
+	if !TSDBEnabled() {
+		if err := DB.Create(&model.ServiceHistory{
+			CreatedAt: status.lastHistoryAt,
+			ServiceID: serviceID,
+			AvgDelay:  status.history.Delay,
+			Data:      status.lastData,
+			Up:        status.history.Up,
+			Down:      status.history.Down,
+		}).Error; err != nil {
+			log.Printf("NEZHA>> Failed to save service monitor metrics: %v", err)
+			return
+		}
+	}
+	status.history = serviceResponseData{}
+	status.lastHistoryAt = now
+}
+
+func (status *serviceTaskStatus) responseData() serviceResponseData {
+	var data serviceResponseData
+	for _, bucket := range status.result {
+		for _, sample := range bucket.reporters {
+			if sample.successful {
+				data.Up++
+				data.Delay = (data.Delay*float64(data.Up-1) + sample.delay) / float64(data.Up)
+			}
+			if sample.failed {
+				data.Down++
+			}
+		}
+	}
+	return data
 }
 
 type pingStore struct {
@@ -228,7 +324,7 @@ func (ss *ServiceSentinel) loadServiceHistory() error {
 		}
 		ss.services[service.ID] = service
 		ss.serviceCurrentStatusData[service.ID] = new(serviceTaskStatus)
-		ss.serviceCurrentStatusData[service.ID].result = make([]*pb.TaskResult, 0, _CurrentStatusSize)
+		ss.serviceCurrentStatusData[service.ID].result = make([]serviceWindowBucket, 0, _CurrentStatusSize)
 		ss.serviceStatusToday[service.ID] = &_TodayStatsOfService{}
 		validServices = append(validServices, service)
 	}
@@ -371,6 +467,9 @@ func (ss *ServiceSentinel) Update(m *model.Service) error {
 	if err != nil {
 		return err
 	}
+	// A service edit may disable notifications or change their recipients.
+	// Cancel any retry worker holding the previous service configuration.
+	NotificationShared.CancelServiceState(m.ID)
 	if ss.services[m.ID] != nil {
 		// 停掉旧任务
 		CronShared.Remove(ss.services[m.ID].CronJobID)
@@ -387,7 +486,7 @@ func (ss *ServiceSentinel) Update(m *model.Service) error {
 		if ss.serviceCurrentStatusData[m.ID] == nil {
 			ss.serviceCurrentStatusData[m.ID] = new(serviceTaskStatus)
 		}
-		ss.serviceCurrentStatusData[m.ID].result = make([]*pb.TaskResult, 0, _CurrentStatusSize)
+		ss.serviceCurrentStatusData[m.ID].result = make([]serviceWindowBucket, 0, _CurrentStatusSize)
 		ss.serviceStatusToday[m.ID] = &_TodayStatsOfService{}
 	}
 	// 更新这个任务
@@ -404,6 +503,7 @@ func (ss *ServiceSentinel) Delete(ids []uint64) {
 	defer ss.servicesLock.Unlock()
 
 	for _, id := range ids {
+		NotificationShared.CancelServiceState(id)
 		delete(ss.serviceCurrentStatusData, id)
 		delete(ss.serviceResponseDataStore, id)
 		delete(ss.serviceResponsePing, id)
@@ -669,32 +769,16 @@ func (ss *ServiceSentinel) processReport(r ReportData, serverShared *ServerClass
 	}
 
 	currentTime := time.Now()
-	if serviceCurrentStatusData.t.IsZero() {
-		serviceCurrentStatusData.t = currentTime
-	}
-
-	// 写入当前数据
-	if serviceCurrentStatusData.t.Before(currentTime) {
-		serviceCurrentStatusData.t = currentTime.Add(30 * time.Second)
-		serviceCurrentStatusData.result = append(serviceCurrentStatusData.result, mh)
+	ss.flushServiceHistory(mh.GetId(), serviceCurrentStatusData, currentTime)
+	// Include every accepted reporter's result. A fixed count of samples is
+	// not a 15-minute window when probes have different intervals or reporters.
+	serviceCurrentStatusData.appendResult(currentTime, r.Reporter, mh)
+	if serviceCurrentStatusData.lastHistoryAt.IsZero() {
+		serviceCurrentStatusData.lastHistoryAt = currentTime
 	}
 
 	// 更新当前状态
-	ss.serviceResponseDataStore[mh.GetId()] = serviceResponseData{}
-
-	// 永远是最新的 30 个数据的状态 [01:00, 02:00, 03:00] -> [04:00, 02:00, 03: 00]
-	for _, cs := range serviceCurrentStatusData.result {
-		if cs.GetId() > 0 {
-			rd := ss.serviceResponseDataStore[mh.GetId()]
-			if cs.Successful {
-				rd.Up++
-				rd.Delay = (rd.Delay*float64(rd.Up-1) + float64(cs.Delay)) / float64(rd.Up)
-			} else {
-				rd.Down++
-			}
-			ss.serviceResponseDataStore[mh.GetId()] = rd
-		}
-	}
+	ss.serviceResponseDataStore[mh.GetId()] = serviceCurrentStatusData.responseData()
 
 	// 计算在线率，
 	var stateCode uint8
@@ -704,24 +788,11 @@ func (ss *ServiceSentinel) processReport(r ReportData, serverShared *ServerClass
 		if rd.Down+rd.Up > 0 {
 			upPercent = rd.Up * 100 / (rd.Down + rd.Up)
 		}
-		stateCode = GetStatusCode(upPercent)
-	}
-
-	if len(serviceCurrentStatusData.result) == _CurrentStatusSize {
-		serviceCurrentStatusData.t = currentTime
-		if !TSDBEnabled() {
-			rd := ss.serviceResponseDataStore[mh.GetId()]
-			if err := DB.Create(&model.ServiceHistory{
-				ServiceID: mh.GetId(),
-				AvgDelay:  rd.Delay,
-				Data:      mh.Data,
-				Up:        rd.Up,
-				Down:      rd.Down,
-			}).Error; err != nil {
-				log.Printf("NEZHA>> Failed to save service monitor metrics: %v", err)
-			}
+		if rd.Down+rd.Up == 0 {
+			stateCode = StatusNoData
+		} else {
+			stateCode = GetStatusCode(upPercent)
 		}
-		serviceCurrentStatusData.result = serviceCurrentStatusData.result[:0]
 	}
 
 	// 延迟报警
@@ -734,77 +805,100 @@ func (ss *ServiceSentinel) processReport(r ReportData, serverShared *ServerClass
 		lastStatus := serviceCurrentStatusData.lastStatus
 		// 存储新的状态值
 		serviceCurrentStatusData.lastStatus = stateCode
-
-		notifyCheck(&r, m, cs, mh, lastStatus, stateCode)
-	}
-
-	// TLS 证书报警
-	if ss.serviceReportBeforeTLSSideEffectsHook != nil {
-		ss.serviceReportBeforeTLSSideEffectsHook(mh.GetId())
-	}
-	var errMsg string
-	if strings.HasPrefix(mh.Data, "SSL证书错误：") {
-		// i/o timeout、connection timeout、EOF 错误
-		if !strings.HasSuffix(mh.Data, "timeout") &&
-			!strings.HasSuffix(mh.Data, "EOF") &&
-			!strings.HasSuffix(mh.Data, "timed out") {
-			errMsg = mh.Data
-			if cs.Notify {
-				muteLabel := NotificationMuteLabel.ServiceTLS(mh.GetId(), "network")
-				go NotificationShared.SendNotification(cs.NotificationGroupID, Localizer.Tf("[TLS] Fetch cert info failed, Reporter: %s, Error: %s", cs.Name, errMsg), muteLabel)
-			}
+		if stateCode != lastStatus {
+			serviceCurrentStatusData.stateEpoch = serviceEventSequence.Add(1)
 		}
-	} else {
-		// 清除网络错误静音缓存
-		NotificationShared.UnMuteNotification(cs.NotificationGroupID, NotificationMuteLabel.ServiceTLS(mh.GetId(), "network"))
 
-		var newCert = strings.Split(mh.Data, "|")
-		if len(newCert) > 1 {
-			enableNotify := cs.Notify
+		notifyCheck(&r, m, cs, mh, lastStatus, stateCode, serviceCurrentStatusData.stateEpoch)
+	}
 
-			// 首次获取证书信息时，缓存证书信息
-			if ss.tlsCertCache[mh.GetId()] == "" {
-				ss.tlsCertCache[mh.GetId()] = mh.Data
+	// TLS 证书报警仅适用于 HTTPS 探针。其它探针的任意响应文本
+	// 不得被当作证书资料解析。
+	if mh.Type == model.TaskTypeHTTPGet && strings.HasPrefix(strings.ToLower(cs.Target), "https://") {
+		if ss.serviceReportBeforeTLSSideEffectsHook != nil {
+			ss.serviceReportBeforeTLSSideEffectsHook(mh.GetId())
+		}
+		var errMsg string
+		if strings.HasPrefix(mh.Data, "SSL证书错误：") {
+			// i/o timeout、connection timeout、EOF 错误
+			if !strings.HasSuffix(mh.Data, "timeout") &&
+				!strings.HasSuffix(mh.Data, "EOF") &&
+				!strings.HasSuffix(mh.Data, "timed out") {
+				errMsg = mh.Data
+				if cs.Notify {
+					muteLabel := NotificationMuteLabel.ServiceTLS(mh.GetId(), "network")
+					NotificationShared.SendNotificationAsync(cs.NotificationGroupID, Localizer.Tf("[TLS] Fetch cert info failed, Reporter: %s, Error: %s", cs.Name, errMsg), muteLabel)
+				}
 			}
+		} else {
+			// 清除网络错误静音缓存
+			NotificationShared.UnMuteNotification(cs.NotificationGroupID, NotificationMuteLabel.ServiceTLS(mh.GetId(), "network"))
 
-			oldCert := strings.Split(ss.tlsCertCache[mh.GetId()], "|")
-			isCertChanged := false
-			expiresOld, _ := time.Parse("2006-01-02 15:04:05 -0700 MST", oldCert[1])
-			expiresNew, _ := time.Parse("2006-01-02 15:04:05 -0700 MST", newCert[1])
-
-			// 证书变更时，更新缓存
-			if oldCert[0] != newCert[0] && !expiresNew.Equal(expiresOld) {
-				isCertChanged = true
-				ss.tlsCertCache[mh.GetId()] = mh.Data
-			}
-
-			notificationGroupID := cs.NotificationGroupID
-			serviceName := cs.Name
-
-			// 需要发送提醒
-			if enableNotify {
-				// 证书过期提醒
-				if expiresNew.Before(time.Now().AddDate(0, 0, 7)) {
-					expiresTimeStr := expiresNew.Format("2006-01-02 15:04:05")
-					errMsg = Localizer.Tf(
-						"The TLS certificate will expire within seven days. Expiration time: %s",
-						expiresTimeStr,
-					)
-
-					// 静音规则： 服务id+证书过期时间
-					// 用于避免多个监测点对相同证书同时报警
-					muteLabel := NotificationMuteLabel.ServiceTLS(mh.GetId(), fmt.Sprintf("expire_%s", expiresTimeStr))
-					go NotificationShared.SendNotification(notificationGroupID, fmt.Sprintf("[TLS] %s %s", serviceName, errMsg), muteLabel)
+			var newCert = strings.Split(mh.Data, "|")
+			if len(newCert) > 1 {
+				enableNotify := cs.Notify
+				expiresNew, newErr := time.Parse("2006-01-02 15:04:05 -0700 MST", newCert[1])
+				if newErr != nil {
+					log.Printf("NEZHA>> Ignoring malformed TLS certificate expiry for service %d", mh.GetId())
+					return
 				}
 
-				// 证书变更提醒
-				if isCertChanged {
-					errMsg = Localizer.Tf(
-						"TLS certificate changed, old: issuer %s, expires at %s; new: issuer %s, expires at %s",
-						oldCert[0], expiresOld.Format("2006-01-02 15:04:05"), newCert[0], expiresNew.Format("2006-01-02 15:04:05"))
+				// 首次获取证书信息时，缓存证书信息
+				if ss.tlsCertCache[mh.GetId()] == "" {
+					ss.tlsCertCache[mh.GetId()] = mh.Data
+				}
 
-					// 证书变更后会自动更新缓存，所以不需要静音
-					go NotificationShared.SendNotification(notificationGroupID, fmt.Sprintf("[TLS] %s %s", serviceName, errMsg), "")
+				oldCert := strings.Split(ss.tlsCertCache[mh.GetId()], "|")
+				isCertChanged := false
+				expiresOld, oldErr := time.Parse("2006-01-02 15:04:05 -0700 MST", oldCert[1])
+				if oldErr != nil {
+					log.Printf("NEZHA>> Ignoring malformed TLS certificate expiry for service %d", mh.GetId())
+					return
+				}
+
+				// 只有通知成功后才推进基线；失败时下一次探针结果会重试。
+				if oldCert[0] != newCert[0] || !expiresNew.Equal(expiresOld) {
+					isCertChanged = true
+					if !enableNotify {
+						ss.tlsCertCache[mh.GetId()] = mh.Data
+					}
+				}
+
+				notificationGroupID := cs.NotificationGroupID
+				serviceName := cs.Name
+
+				// 需要发送提醒
+				if enableNotify {
+					// 证书过期提醒
+					if expiresNew.Before(time.Now().AddDate(0, 0, 7)) {
+						expiresTimeStr := expiresNew.Format("2006-01-02 15:04:05")
+						errMsg = Localizer.Tf(
+							"The TLS certificate will expire within seven days. Expiration time: %s",
+							expiresTimeStr,
+						)
+
+						// 静音规则： 服务id+证书过期时间
+						// 用于避免多个监测点对相同证书同时报警
+						muteLabel := NotificationMuteLabel.ServiceTLS(mh.GetId(), fmt.Sprintf("expire_%s", expiresTimeStr))
+						NotificationShared.SendNotificationAsync(notificationGroupID, fmt.Sprintf("[TLS] %s %s", serviceName, errMsg), muteLabel)
+					}
+
+					// 证书变更提醒
+					if isCertChanged {
+						errMsg = Localizer.Tf(
+							"TLS certificate changed, old: issuer %s, expires at %s; new: issuer %s, expires at %s",
+							oldCert[0], expiresOld.Format("2006-01-02 15:04:05"), newCert[0], expiresNew.Format("2006-01-02 15:04:05"))
+
+						oldRaw, newRaw, serviceID := ss.tlsCertCache[mh.GetId()], mh.Data, mh.GetId()
+						muteLabel := NotificationMuteLabel.ServiceTLS(serviceID, fmt.Sprintf("change_%x", sha256.Sum256([]byte(newRaw))))
+						NotificationShared.sendNotificationAsync(notificationGroupID, fmt.Sprintf("[TLS] %s %s", serviceName, errMsg), muteLabel, func() {
+							ss.serviceResponseDataStoreLock.Lock()
+							if ss.serviceCurrentStatusData[serviceID] != nil && ss.tlsCertCache[serviceID] == oldRaw {
+								ss.tlsCertCache[serviceID] = newRaw
+							}
+							ss.serviceResponseDataStoreLock.Unlock()
+						})
+					}
 				}
 			}
 		}
@@ -832,11 +926,11 @@ func delayCheck(r *ReportData, m map[uint64]*model.Server, ss *model.Service, mh
 	if mh.Delay > ss.MaxLatency {
 		// 延迟超过最大值
 		msg := Localizer.Tf("[Latency] %s %2f > %2f, Reporter: %s", ss.Name, mh.Delay, ss.MaxLatency, reporterServer.Name)
-		go NotificationShared.SendNotification(notificationGroupID, msg, minMuteLabel)
+		NotificationShared.SendNotificationAsync(notificationGroupID, msg, minMuteLabel)
 	} else if mh.Delay < ss.MinLatency {
 		// 延迟低于最小值
 		msg := Localizer.Tf("[Latency] %s %2f < %2f, Reporter: %s", ss.Name, mh.Delay, ss.MinLatency, reporterServer.Name)
-		go NotificationShared.SendNotification(notificationGroupID, msg, maxMuteLabel)
+		NotificationShared.SendNotificationAsync(notificationGroupID, msg, maxMuteLabel)
 	} else {
 		// 正常延迟， 清除静音缓存
 		NotificationShared.UnMuteNotification(notificationGroupID, minMuteLabel)
@@ -845,7 +939,7 @@ func delayCheck(r *ReportData, m map[uint64]*model.Server, ss *model.Service, mh
 }
 
 func notifyCheck(r *ReportData, m map[uint64]*model.Server,
-	ss *model.Service, mh *pb.TaskResult, lastStatus, stateCode uint8) {
+	ss *model.Service, mh *pb.TaskResult, lastStatus, stateCode uint8, stateEpoch uint64) {
 	// GHSA-jx78-55p5-rwv5: guard against concurrent server deletion (same TOCTOU
 	// class as the 2026-07-21 fix, a few dozen lines lower in the same worker).
 	// ServerShared has its own lock; m is a snapshot taken outside
@@ -858,14 +952,9 @@ func notifyCheck(r *ReportData, m map[uint64]*model.Server,
 	if isNeedSendNotification && reporterServer != nil {
 		notificationGroupID := ss.NotificationGroupID
 		notificationMsg := Localizer.Tf("[%s] %s Reporter: %s, Error: %s", StatusCodeToString(stateCode), ss.Name, reporterServer.Name, mh.Data)
-		muteLabel := NotificationMuteLabel.ServiceStateChanged(mh.GetId())
+		muteLabel := fmt.Sprintf("%s:%d:%d", NotificationMuteLabel.ServiceStateChanged(mh.GetId()), stateCode, stateEpoch)
 
-		// 状态变更时，清除静音缓存
-		if stateCode != lastStatus {
-			NotificationShared.UnMuteNotification(notificationGroupID, muteLabel)
-		}
-
-		go NotificationShared.SendNotification(notificationGroupID, notificationMsg, muteLabel)
+		NotificationShared.SendServiceState(mh.GetId(), notificationGroupID, notificationMsg, muteLabel)
 	}
 
 	// 判断是否需要触发任务
@@ -890,9 +979,6 @@ const (
 )
 
 func GetStatusCode[T constraints.Float | constraints.Integer](percent T) uint8 {
-	if percent == 0 {
-		return StatusNoData
-	}
 	if percent > 95 {
 		return StatusGood
 	}

service/singleton/servicesentinel_window_test.go

@@ -0,0 +1,63 @@
+package singleton
+
+import (
+	"testing"
+	"time"
+
+	pb "github.com/nezhahq/nezha/proto"
+)
+
+func TestServiceWindowCountsFirstFailureAsDown(t *testing.T) {
+	status := new(serviceTaskStatus)
+	status.appendResult(time.Unix(1000, 0), 1, &pb.TaskResult{Successful: false})
+	data := status.responseData()
+	if data.Down != 1 || data.Up != 0 || GetStatusCode(uint64(0)) != StatusDown {
+		t.Fatalf("first failed report = %+v, status %d; want Down", data, GetStatusCode(uint64(0)))
+	}
+}
+
+func TestServiceWindowExpiresOldReportsAndDoesNotResetAtThirty(t *testing.T) {
+	status := new(serviceTaskStatus)
+	start := time.Unix(1000, 0)
+	for i := 0; i < 30; i++ {
+		status.appendResult(start.Add(time.Duration(i)*30*time.Second), 1, &pb.TaskResult{Successful: true})
+	}
+	if data := status.responseData(); data.Up != 30 {
+		t.Fatalf("30th report reset window: %+v", data)
+	}
+	status.appendResult(start.Add(29*30*time.Second+time.Second), 2, &pb.TaskResult{Successful: false})
+	if data := status.responseData(); data.Up != 30 || data.Down != 1 {
+		t.Fatalf("31st report corrupted window: %+v", data)
+	}
+	status.appendResult(start.Add(31*time.Minute), 1, &pb.TaskResult{Successful: false})
+	if data := status.responseData(); data.Up != 0 || data.Down != 1 {
+		t.Fatalf("expired reports still influence status: %+v", data)
+	}
+}
+
+func TestServiceWindowDeduplicatesFloodWithinBucket(t *testing.T) {
+	status := new(serviceTaskStatus)
+	now := time.Unix(1000, 0)
+	for i := 0; i < 10000; i++ {
+		status.appendResult(now, 1, &pb.TaskResult{Successful: false, Data: "ignored payload"})
+	}
+	if len(status.result) != 1 || len(status.result[0].reporters) != 1 {
+		t.Fatalf("flood grew live window: buckets=%d", len(status.result))
+	}
+	if data := status.responseData(); data.Down != 1 {
+		t.Fatalf("duplicate reporter skewed status: %+v", data)
+	}
+	if status.history.Down != 10000 {
+		t.Fatalf("history lost raw reports: %+v", status.history)
+	}
+}
+
+func TestServiceWindowDoesNotEraseFailureWithLaterSuccess(t *testing.T) {
+	status := new(serviceTaskStatus)
+	now := time.Unix(1000, 0)
+	status.appendResult(now, 1, &pb.TaskResult{Successful: false})
+	status.appendResult(now.Add(time.Second), 1, &pb.TaskResult{Successful: true, Delay: 20})
+	if data := status.responseData(); data.Down != 1 || data.Up != 1 {
+		t.Fatalf("later success erased failure in same bucket: %+v", data)
+	}
+}

service/singleton/singleton.go

@@ -167,7 +167,7 @@ func CleanMonitorHistory() {
 	for _, alert := range alerts {
 		for _, rule := range alert.Rules {
 			// 是不是流量记录规则
-			if !rule.IsTransferDurationRule() {
+			if !rule.HasSafeCycleConfiguration() {
 				continue
 			}
 			dataCouldRemoveBefore := rule.GetTransferDurationStart().UTC()