fix(service): serialize reports with service lifecycle

Co-authored-by: naiba/CloudCode <hi+cloudcode@nai.ba>

service/singleton/servicesentinel.go

@@ -74,8 +74,11 @@ type ServiceSentinel struct {
 	serviceCurrentStatusData     map[uint64]*serviceTaskStatus    // 当前任务结果缓存
 	serviceResponseDataStore     map[uint64]serviceResponseData   // 当前数据
 
-	serviceResponsePing map[uint64]map[uint64]*pingStore // [service_id] -> ClientID -> delay
-	tlsCertCache        map[uint64]string
+	serviceResponsePing                   map[uint64]map[uint64]*pingStore // guarded by serviceResponseDataStoreLock; [service_id] -> ClientID -> delay
+	tlsCertCache                          map[uint64]string                // guarded by serviceResponseDataStoreLock
+	serviceReportValidatedHook            func(uint64)
+	loadStatsResponseLockedHook           func()
+	serviceReportBeforeTLSSideEffectsHook func(uint64)
 
 	servicesLock    sync.RWMutex
 	serviceListLock sync.RWMutex
@@ -386,6 +389,7 @@ func (ss *ServiceSentinel) Delete(ids []uint64) {
 	for _, id := range ids {
 		delete(ss.serviceCurrentStatusData, id)
 		delete(ss.serviceResponseDataStore, id)
+		delete(ss.serviceResponsePing, id)
 		delete(ss.tlsCertCache, id)
 		delete(ss.serviceStatusToday, id)
 
@@ -398,12 +402,15 @@ func (ss *ServiceSentinel) Delete(ids []uint64) {
 }
 
 func (ss *ServiceSentinel) LoadStats() map[uint64]*serviceResponseItem {
-	ss.servicesLock.RLock()
-	defer ss.servicesLock.RUnlock()
 	ss.serviceResponseDataStoreLock.RLock()
 	defer ss.serviceResponseDataStoreLock.RUnlock()
+	if ss.loadStatsResponseLockedHook != nil {
+		ss.loadStatsResponseLockedHook()
+	}
 	ss.monthlyStatusLock.Lock()
 	defer ss.monthlyStatusLock.Unlock()
+	ss.servicesLock.RLock()
+	defer ss.servicesLock.RUnlock()
 
 	// 刷新最新一天的数据
 	for k := range ss.services {
@@ -536,8 +543,23 @@ func (ss *ServiceSentinel) worker() {
 			log.Printf("NEZHA>> Incorrect service monitor report %+v", r)
 			continue
 		}
+		if ss.serviceReportValidatedHook != nil {
+			ss.serviceReportValidatedHook(r.Data.GetId())
+		}
 
 		mh := r.Data
+		// Serialize Delete and Update before this accepted report causes any side effect.
+		ss.serviceResponseDataStoreLock.Lock()
+		serviceStatusToday := ss.serviceStatusToday[mh.GetId()]
+		serviceCurrentStatusData := ss.serviceCurrentStatusData[mh.GetId()]
+		currentService, serviceExists := ss.Get(mh.GetId())
+		if serviceStatusToday == nil || serviceCurrentStatusData == nil || !serviceExists ||
+			!canReportServiceResult(currentService, reporter, mh.GetType()) {
+			ss.serviceResponseDataStoreLock.Unlock()
+			continue
+		}
+		cs = currentService
+
 		if mh.Type == model.TaskTypeTCPPing || mh.Type == model.TaskTypeICMPPing {
 			// TCP/ICMP Ping 使用平均值计算后再写入
 			serviceTcpMap, ok := ss.serviceResponsePing[mh.GetId()]
@@ -594,33 +616,31 @@ func (ss *ServiceSentinel) worker() {
 			}
 		}
 
-		ss.serviceResponseDataStoreLock.Lock()
 		// 写入当天状态
 		if mh.Successful {
-			ss.serviceStatusToday[mh.GetId()].Delay = (ss.serviceStatusToday[mh.
-				GetId()].Delay*float64(ss.serviceStatusToday[mh.GetId()].Up) +
-				float64(mh.Delay)) / float64(ss.serviceStatusToday[mh.GetId()].Up+1)
-			ss.serviceStatusToday[mh.GetId()].Up++
+			serviceStatusToday.Delay = (serviceStatusToday.Delay*float64(serviceStatusToday.Up) +
+				float64(mh.Delay)) / float64(serviceStatusToday.Up+1)
+			serviceStatusToday.Up++
 		} else {
-			ss.serviceStatusToday[mh.GetId()].Down++
+			serviceStatusToday.Down++
 		}
 
 		currentTime := time.Now()
-		if ss.serviceCurrentStatusData[mh.GetId()].t.IsZero() {
-			ss.serviceCurrentStatusData[mh.GetId()].t = currentTime
+		if serviceCurrentStatusData.t.IsZero() {
+			serviceCurrentStatusData.t = currentTime
 		}
 
 		// 写入当前数据
-		if ss.serviceCurrentStatusData[mh.GetId()].t.Before(currentTime) {
-			ss.serviceCurrentStatusData[mh.GetId()].t = currentTime.Add(30 * time.Second)
-			ss.serviceCurrentStatusData[mh.GetId()].result = append(ss.serviceCurrentStatusData[mh.GetId()].result, mh)
+		if serviceCurrentStatusData.t.Before(currentTime) {
+			serviceCurrentStatusData.t = currentTime.Add(30 * time.Second)
+			serviceCurrentStatusData.result = append(serviceCurrentStatusData.result, mh)
 		}
 
 		// 更新当前状态
 		ss.serviceResponseDataStore[mh.GetId()] = serviceResponseData{}
 
 		// 永远是最新的 30 个数据的状态 [01:00, 02:00, 03:00] -> [04:00, 02:00, 03: 00]
-		for _, cs := range ss.serviceCurrentStatusData[mh.GetId()].result {
+		for _, cs := range serviceCurrentStatusData.result {
 			if cs.GetId() > 0 {
 				rd := ss.serviceResponseDataStore[mh.GetId()]
 				if cs.Successful {
@@ -644,8 +664,8 @@ func (ss *ServiceSentinel) worker() {
 			stateCode = GetStatusCode(upPercent)
 		}
 
-		if len(ss.serviceCurrentStatusData[mh.GetId()].result) == _CurrentStatusSize {
-			ss.serviceCurrentStatusData[mh.GetId()].t = currentTime
+		if len(serviceCurrentStatusData.result) == _CurrentStatusSize {
+			serviceCurrentStatusData.t = currentTime
 			if !TSDBEnabled() {
 				rd := ss.serviceResponseDataStore[mh.GetId()]
 				if err := DB.Create(&model.ServiceHistory{
@@ -658,10 +678,9 @@ func (ss *ServiceSentinel) worker() {
 					log.Printf("NEZHA>> Failed to save service monitor metrics: %v", err)
 				}
 			}
-			ss.serviceCurrentStatusData[mh.GetId()].result = ss.serviceCurrentStatusData[mh.GetId()].result[:0]
+			serviceCurrentStatusData.result = serviceCurrentStatusData.result[:0]
 		}
 
-		cs, _ = ss.Get(mh.GetId())
 		m := ServerShared.GetList()
 		// 延迟报警
 		if mh.Delay > 0 {
@@ -669,16 +688,18 @@ func (ss *ServiceSentinel) worker() {
 		}
 
 		// 状态变更报警+触发任务执行
-		if stateCode == StatusDown || stateCode != ss.serviceCurrentStatusData[mh.GetId()].lastStatus {
-			lastStatus := ss.serviceCurrentStatusData[mh.GetId()].lastStatus
+		if stateCode == StatusDown || stateCode != serviceCurrentStatusData.lastStatus {
+			lastStatus := serviceCurrentStatusData.lastStatus
 			// 存储新的状态值
-			ss.serviceCurrentStatusData[mh.GetId()].lastStatus = stateCode
+			serviceCurrentStatusData.lastStatus = stateCode
 
 			notifyCheck(&r, m, cs, mh, lastStatus, stateCode)
 		}
-		ss.serviceResponseDataStoreLock.Unlock()
 
 		// TLS 证书报警
+		if ss.serviceReportBeforeTLSSideEffectsHook != nil {
+			ss.serviceReportBeforeTLSSideEffectsHook(mh.GetId())
+		}
 		var errMsg string
 		if strings.HasPrefix(mh.Data, "SSL证书错误：") {
 			// i/o timeout、connection timeout、EOF 错误
@@ -746,6 +767,7 @@ func (ss *ServiceSentinel) worker() {
 				}
 			}
 		}
+		ss.serviceResponseDataStoreLock.Unlock()
 	}
 }
 

service/singleton/servicesentinel_lifecycle_test.go

@@ -0,0 +1,301 @@
+package singleton
+
+import (
+	"context"
+	"fmt"
+	"os"
+	"os/exec"
+	"strings"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/nezhahq/nezha/model"
+)
+
+const serviceSentinelLifecycleSuccessMarker = "service-sentinel-stale-report-lifecycle-success"
+
+func TestServiceSentinelWorkerIgnoresStaleReportAfterDeletion(t *testing.T) {
+	if os.Getenv("NEZHA_SERVICE_SENTINEL_LIFECYCLE_CHILD") == "1" {
+		testServiceSentinelWorkerIgnoresStaleReportAfterDeletionChild(t)
+		return
+	}
+
+	// Given
+	ctx, cancel := context.WithTimeout(t.Context(), 5*time.Second)
+	defer cancel()
+	child := exec.CommandContext(ctx, os.Args[0], "-test.run=^TestServiceSentinelWorkerIgnoresStaleReportAfterDeletion$")
+	child.Env = append(os.Environ(), "NEZHA_SERVICE_SENTINEL_LIFECYCLE_CHILD=1")
+
+	// When
+	output, err := child.CombinedOutput()
+
+	// Then
+	if ctx.Err() != nil {
+		t.Fatalf("service sentinel lifecycle child timed out: %v\n%s", ctx.Err(), output)
+	}
+	if err != nil {
+		t.Fatalf("service sentinel lifecycle child failed: %v\n%s", err, output)
+	}
+	if !strings.Contains(string(output), serviceSentinelLifecycleSuccessMarker) {
+		t.Fatalf("service sentinel lifecycle child did not report success:\n%s", output)
+	}
+}
+
+func testServiceSentinelWorkerIgnoresStaleReportAfterDeletionChild(t *testing.T) {
+	// Given
+	ctx, cancel := context.WithTimeout(t.Context(), time.Second)
+	defer cancel()
+	ss := newServiceMonitorSecurityHarness(t,
+		&model.Server{Common: model.Common{ID: 1, UserID: 1}, Name: "reporter"},
+	)
+	for _, service := range []*model.Service{
+		{
+			Common:      model.Common{ID: 10, UserID: 1},
+			Name:        "stale-service",
+			Type:        model.TaskTypeTCPPing,
+			Target:      "stale.example.invalid:443",
+			Duration:    3600,
+			Cover:       model.ServiceCoverIgnoreAll,
+			SkipServers: map[uint64]bool{1: true},
+		},
+		{
+			Common:      model.Common{ID: 20, UserID: 1},
+			Name:        "valid-service",
+			Type:        model.TaskTypeTCPPing,
+			Target:      "valid.example.invalid:443",
+			Duration:    3600,
+			Cover:       model.ServiceCoverIgnoreAll,
+			SkipServers: map[uint64]bool{1: true},
+		},
+	} {
+		addServiceMonitorSecurityService(t, ss, service)
+	}
+	acceptedStaleReport := make(chan struct{})
+	releaseWorker := make(chan struct{})
+	var releaseOnce sync.Once
+	releaseWorkerHook := func() {
+		releaseOnce.Do(func() { close(releaseWorker) })
+	}
+	ss.serviceReportValidatedHook = func(serviceID uint64) {
+		if serviceID == 10 {
+			close(acceptedStaleReport)
+			<-releaseWorker
+		}
+	}
+	t.Cleanup(func() {
+		releaseWorkerHook()
+		ss.Close()
+	})
+
+	// When
+	ss.Dispatch(serviceMonitorResult(1, 10, model.TaskTypeTCPPing, true))
+	select {
+	case <-acceptedStaleReport:
+	case <-ctx.Done():
+		t.Fatal(ctx.Err())
+	}
+	ss.Delete([]uint64{10})
+	releaseWorkerHook()
+	ss.Dispatch(serviceMonitorResult(1, 20, model.TaskTypeTCPPing, true))
+	ss.Close()
+
+	// Then
+	var staleHistoryCount int64
+	if err := DB.Model(&model.ServiceHistory{}).
+		Where("service_id = ? AND server_id = ?", 10, 1).
+		Count(&staleHistoryCount).Error; err != nil {
+		t.Fatal(err)
+	}
+	if staleHistoryCount != 0 {
+		t.Fatalf("expected stale service to write zero per-reporter history rows, got %d", staleHistoryCount)
+	}
+	var validHistoryCount int64
+	if err := DB.Model(&model.ServiceHistory{}).
+		Where("service_id = ? AND server_id = ?", 20, 1).
+		Count(&validHistoryCount).Error; err != nil {
+		t.Fatal(err)
+	}
+	if validHistoryCount != 1 {
+		t.Fatalf("expected exactly one valid service history row, got %d", validHistoryCount)
+	}
+	ss.serviceResponseDataStoreLock.RLock()
+	_, stalePingCached := ss.serviceResponsePing[10]
+	validStats := ss.serviceStatusToday[20]
+	ss.serviceResponseDataStoreLock.RUnlock()
+	if stalePingCached {
+		t.Fatal("expected stale service ping cache to be deleted")
+	}
+	if validStats == nil || validStats.Up != 1 || validStats.Down != 0 {
+		t.Fatalf("expected valid service stats up=1 down=0, got %+v", validStats)
+	}
+	if _, err := fmt.Fprintln(os.Stdout, serviceSentinelLifecycleSuccessMarker); err != nil {
+		t.Fatal(err)
+	}
+}
+
+func TestServiceSentinelWorkerRevalidatesReportAfterUpdate(t *testing.T) {
+	// Given
+	ctx, cancel := context.WithTimeout(t.Context(), time.Second)
+	defer cancel()
+	ss := newServiceMonitorSecurityHarness(t,
+		&model.Server{Common: model.Common{ID: 1, UserID: 1}, Name: "reporter"},
+	)
+	service := &model.Service{
+		Common:      model.Common{ID: 10, UserID: 1},
+		Name:        "updatable-service",
+		Type:        model.TaskTypeTCPPing,
+		Target:      "updatable.example.invalid:443",
+		Duration:    3600,
+		Cover:       model.ServiceCoverIgnoreAll,
+		SkipServers: map[uint64]bool{1: true},
+	}
+	addServiceMonitorSecurityService(t, ss, service)
+	ss.serviceResponseDataStoreLock.Lock()
+	ss.serviceStatusToday[service.ID] = &_TodayStatsOfService{Up: 7, Down: 3, Delay: 12.5}
+	ss.serviceResponseDataStoreLock.Unlock()
+	acceptedReport := make(chan struct{})
+	releaseWorker := make(chan struct{})
+	var releaseOnce sync.Once
+	releaseWorkerHook := func() {
+		releaseOnce.Do(func() { close(releaseWorker) })
+	}
+	ss.serviceReportValidatedHook = func(serviceID uint64) {
+		if serviceID == service.ID {
+			close(acceptedReport)
+			<-releaseWorker
+		}
+	}
+	t.Cleanup(func() {
+		releaseWorkerHook()
+		ss.Close()
+	})
+
+	// When
+	ss.Dispatch(serviceMonitorResult(1, service.ID, model.TaskTypeTCPPing, true))
+	select {
+	case <-acceptedReport:
+	case <-ctx.Done():
+		t.Fatal(ctx.Err())
+	}
+	updatedService := *service
+	updatedService.Name = "updated-service"
+	updatedService.SkipServers = map[uint64]bool{}
+	if err := ss.Update(&updatedService); err != nil {
+		t.Fatal(err)
+	}
+	releaseWorkerHook()
+	ss.Close()
+
+	// Then
+	var historyCount int64
+	if err := DB.Model(&model.ServiceHistory{}).
+		Where("service_id = ? AND server_id = ?", service.ID, 1).
+		Count(&historyCount).Error; err != nil {
+		t.Fatal(err)
+	}
+	if historyCount != 0 {
+		t.Fatalf("expected updated service to write zero per-reporter history rows, got %d", historyCount)
+	}
+	ss.serviceResponseDataStoreLock.RLock()
+	_, pingCached := ss.serviceResponsePing[service.ID]
+	stats := ss.serviceStatusToday[service.ID]
+	ss.serviceResponseDataStoreLock.RUnlock()
+	if pingCached {
+		t.Fatal("expected updated service report to leave no ping cache entry")
+	}
+	if stats == nil || stats.Up != 7 || stats.Down != 3 || stats.Delay != 12.5 {
+		t.Fatalf("expected existing service stats to remain unchanged, got %+v", stats)
+	}
+	currentService, ok := ss.Get(service.ID)
+	if !ok || currentService.Name != updatedService.Name || currentService.SkipServers[1] {
+		t.Fatalf("expected updated service configuration, got %+v", currentService)
+	}
+}
+
+func TestServiceSentinelLoadStatsFollowsLifecycleLockOrder(t *testing.T) {
+	// Given
+	ss := &ServiceSentinel{
+		serviceStatusToday:       make(map[uint64]*_TodayStatsOfService),
+		serviceResponseDataStore: make(map[uint64]serviceResponseData),
+		services:                 make(map[uint64]*model.Service),
+		monthlyStatus:            make(map[uint64]*serviceResponseItem),
+	}
+	ss.loadStatsResponseLockedHook = func() {
+		if ss.serviceResponseDataStoreLock.TryLock() {
+			ss.serviceResponseDataStoreLock.Unlock()
+			t.Fatal("LoadStats invoked the hook before acquiring the response read lock")
+		}
+		if !ss.monthlyStatusLock.TryLock() {
+			t.Fatal("LoadStats acquired monthlyStatusLock before the response lock hook")
+		}
+		ss.monthlyStatusLock.Unlock()
+		if !ss.servicesLock.TryLock() {
+			t.Fatal("LoadStats acquired servicesLock before the response lock hook")
+		}
+		ss.servicesLock.Unlock()
+	}
+
+	// When / Then
+	ss.LoadStats()
+}
+
+func TestServiceSentinelWorkerHoldsResponseLockDuringTLSSideEffects(t *testing.T) {
+	// Given
+	ctx, cancel := context.WithTimeout(t.Context(), time.Second)
+	defer cancel()
+	ss := newServiceMonitorSecurityHarness(t,
+		&model.Server{Common: model.Common{ID: 1, UserID: 1}, Name: "reporter"},
+	)
+	service := &model.Service{
+		Common:      model.Common{ID: 10, UserID: 1},
+		Name:        "tls-service",
+		Type:        model.TaskTypeHTTPGet,
+		Target:      "https://tls.example.invalid",
+		Duration:    3600,
+		Cover:       model.ServiceCoverIgnoreAll,
+		SkipServers: map[uint64]bool{1: true},
+	}
+	addServiceMonitorSecurityService(t, ss, service)
+	tlsSideEffectsReady := make(chan struct{})
+	releaseTLSSideEffects := make(chan struct{})
+	var releaseOnce sync.Once
+	releaseTLSSideEffectsHook := func() {
+		releaseOnce.Do(func() { close(releaseTLSSideEffects) })
+	}
+	ss.serviceReportBeforeTLSSideEffectsHook = func(serviceID uint64) {
+		if serviceID == service.ID {
+			close(tlsSideEffectsReady)
+			<-releaseTLSSideEffects
+		}
+	}
+	t.Cleanup(func() {
+		releaseTLSSideEffectsHook()
+		ss.Close()
+	})
+	report := serviceMonitorResult(1, service.ID, model.TaskTypeHTTPGet, true)
+	report.Data.Data = "issuer|2030-01-02 15:04:05 +0000 UTC"
+
+	// When
+	ss.Dispatch(report)
+	select {
+	case <-tlsSideEffectsReady:
+	case <-ctx.Done():
+		t.Fatal(ctx.Err())
+	}
+	responseLockAcquired := ss.serviceResponseDataStoreLock.TryLock()
+	if responseLockAcquired {
+		ss.serviceResponseDataStoreLock.Unlock()
+		t.Fatal("worker released the response lock before TLS side effects")
+	}
+	releaseTLSSideEffectsHook()
+	ss.Close()
+
+	// Then
+	ss.serviceResponseDataStoreLock.RLock()
+	cachedCertificate := ss.tlsCertCache[service.ID]
+	ss.serviceResponseDataStoreLock.RUnlock()
+	if cachedCertificate != report.Data.Data {
+		t.Fatalf("expected TLS cache %q, got %q", report.Data.Data, cachedCertificate)
+	}
+}