降低状态持久化写盘频率

README.md

@@ -97,7 +97,7 @@ Docker 镜像中的默认路径为：
 | 环境变量 | 默认值 | 说明 |
 | --- | --- | --- |
 | `CONFIG_PATH` | `/app/config/config.json` | 主配置文件 |
-| `STATS_PATH` | `/app/data/stats.json` | 月流量与状态持久化文件 |
+| `STATS_PATH` | `/app/data/stats.json` | 月流量与状态持久化文件；每 60 秒写入，关键操作与正常退出时立即写入 |
 | `WEB_DIR` | `/app/web` | WebUI 静态文件目录 |
 | `HTTP_ADDR` | `:80` | WebUI 与 HTTP API 监听地址 |
 | `AGENT_ADDR` | `:35601` | 客户端 TCP 上报监听地址 |

server/app.go

@@ -60,16 +60,17 @@ type App struct {
 	connectionID  atomic.Uint64
 	generation    atomic.Uint64
 	agentRunning  atomic.Bool
-	reloadWrites atomic.Int32
+	reloadPending atomic.Bool
 
 	certMu sync.RWMutex
 	certs  map[string]*CertState
 
-	statsWake chan struct{}
 	persistMu sync.Mutex
 	logger    *log.Logger
 }
 
+const statsFlushInterval = time.Minute
+
 func NewApp(opts Options) (*App, error) {
 	doc, runtime, err := readConfig(opts.ConfigPath)
 	if err != nil {
@@ -83,7 +84,6 @@ func NewApp(opts Options) (*App, error) {
 		cancel:    cancel,
 		nodes:     make(map[string]*NodeState),
 		certs:     make(map[string]*CertState),
-		statsWake: make(chan struct{}, 1),
 		logger:    log.New(os.Stdout, "serverstatus ", log.LstdFlags|log.Lmicroseconds),
 	}
 	app.applyValidatedConfig(doc, runtime, false)
@@ -94,7 +94,6 @@ func NewApp(opts Options) (*App, error) {
 func (a *App) StartBackground() {
 	go a.statsLoop()
 	go a.sslLoop()
-	a.wakeStatsWriter()
 }
 
 func (a *App) Close() {
@@ -216,8 +215,12 @@ func (a *App) applyValidatedConfig(doc ConfigDocument, runtime RuntimeConfig, di
 			_ = conn.Close()
 		}
 	}
-	a.reloadWrites.Store(2)
-	a.wakeStatsWriter()
+	if disconnect {
+		a.reloadPending.Store(true)
+		if err := a.PersistStats(); err != nil {
+			a.logger.Printf("write stats after config change: %v", err)
+		}
+	}
 }
 
 func sameServerIdentity(left, right ServerConfig) bool {
@@ -246,14 +249,13 @@ func (a *App) disconnectAll(reason string) {
 }
 
 func (a *App) statsLoop() {
-	ticker := time.NewTicker(time.Second)
+	ticker := time.NewTicker(statsFlushInterval)
 	defer ticker.Stop()
 	for {
 		select {
 		case <-a.ctx.Done():
 			return
 		case <-ticker.C:
-		case <-a.statsWake:
 		}
 		if err := a.PersistStats(); err != nil {
 			a.logger.Printf("write stats: %v", err)
@@ -261,18 +263,15 @@ func (a *App) statsLoop() {
 	}
 }
 
-func (a *App) wakeStatsWriter() {
-	select {
-	case a.statsWake <- struct{}{}:
-	default:
-	}
-}
-
 func (a *App) SnapshotStats() map[string]any {
-	return a.snapshotStats(false)
+	result := a.snapshotStats()
+	if a.reloadPending.Swap(false) {
+		result["reload"] = true
+	}
+	return result
 }
 
-func (a *App) snapshotStats(consumeReload bool) map[string]any {
+func (a *App) snapshotStats() map[string]any {
 	runtime := a.RuntimeSnapshot()
 	now := time.Now()
 	servers := make([]any, 0, len(runtime.Servers))
@@ -325,19 +324,13 @@ func (a *App) snapshotStats(consumeReload bool) map[string]any {
 		"sslcerts": a.sslSnapshot(runtime.SSLCerts, now),
 		"updated":  strconv.FormatInt(now.Unix(), 10),
 	}
-	if a.reloadWrites.Load() > 0 {
-		result["reload"] = true
-		if consumeReload {
-			a.reloadWrites.Add(-1)
-		}
-	}
 	return result
 }
 
 func (a *App) PersistStats() error {
 	a.persistMu.Lock()
 	defer a.persistMu.Unlock()
-	return writeStatsFile(a.opts.StatsPath, a.snapshotStats(true))
+	return writeStatsFile(a.opts.StatsPath, a.snapshotStats())
 }
 
 func monthResetWindow(now time.Time, monthStart int) bool {
@@ -452,7 +445,9 @@ func (a *App) ResetTraffic(username string) (map[string]any, *APIError) {
 	node.LastNetworkIn, node.LastNetworkOut = networkIn, networkOut
 	server := node.Config
 	a.nodeMu.Unlock()
-	a.wakeStatsWriter()
+	if err := a.PersistStats(); err != nil {
+		a.logger.Printf("write stats after traffic reset for %q: %v", username, err)
+	}
 	return map[string]any{
 		"server": server,
 		"stats": map[string]any{

server/app_test.go

@@ -1,10 +1,34 @@
 package main
 
 import (
+	"encoding/json"
 	"net"
+	"os"
 	"testing"
+	"time"
 )
 
+func readPersistedServer(t *testing.T, app *App) map[string]any {
+	t.Helper()
+	data, err := os.ReadFile(app.opts.StatsPath)
+	if err != nil {
+		t.Fatal(err)
+	}
+	var stats map[string]any
+	if err := json.Unmarshal(data, &stats); err != nil {
+		t.Fatal(err)
+	}
+	servers, ok := stats["servers"].([]any)
+	if !ok || len(servers) != 1 {
+		t.Fatalf("unexpected persisted servers: %#v", stats["servers"])
+	}
+	server, ok := servers[0].(map[string]any)
+	if !ok {
+		t.Fatalf("unexpected persisted server: %#v", servers[0])
+	}
+	return server
+}
+
 func TestTrafficBaselinesResetIndependently(t *testing.T) {
 	node := &NodeState{LastNetworkIn: 100, LastNetworkOut: 0}
 	updateTrafficBaselines(node, 150, 500, false)
@@ -48,3 +72,110 @@ func TestDisconnectPreservesOfflineDisplayMetadata(t *testing.T) {
 		t.Fatalf("offline display metadata was discarded: %#v", serverStats)
 	}
 }
+
+func TestStatsPersistenceUsesLiveMemoryBetweenFlushes(t *testing.T) {
+	if statsFlushInterval != time.Minute {
+		t.Fatalf("unexpected stats flush interval: %s", statsFlushInterval)
+	}
+
+	app := newTestApp(t, minimalTestConfig())
+	if err := app.PersistStats(); err != nil {
+		t.Fatal(err)
+	}
+	initialInfo, err := os.Stat(app.opts.StatsPath)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	app.nodeMu.Lock()
+	node := app.nodes["s01"]
+	node.Connected = true
+	node.ConnectionID = 7
+	app.nodeMu.Unlock()
+	if !app.updateAgent("s01", 7, AgentStats{CPU: 42, NetworkIn: 1000, NetworkOut: 2000}) {
+		t.Fatal("agent update was rejected")
+	}
+
+	live := app.SnapshotStats()["servers"].([]any)[0].(map[string]any)
+	if live["cpu"] != 42 {
+		t.Fatalf("live snapshot was not updated: %#v", live)
+	}
+	beforeFlushInfo, err := os.Stat(app.opts.StatsPath)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if !os.SameFile(initialInfo, beforeFlushInfo) {
+		t.Fatal("reading the live snapshot unexpectedly rewrote stats.json")
+	}
+	if _, exists := readPersistedServer(t, app)["cpu"]; exists {
+		t.Fatal("agent update reached disk before the next persistence run")
+	}
+
+	if err := app.PersistStats(); err != nil {
+		t.Fatal(err)
+	}
+	afterFlushInfo, err := os.Stat(app.opts.StatsPath)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if os.SameFile(beforeFlushInfo, afterFlushInfo) {
+		t.Fatal("updated stats were not persisted")
+	}
+	if cpu := readPersistedServer(t, app)["cpu"]; cpu != float64(42) {
+		t.Fatalf("unexpected persisted CPU value: %#v", cpu)
+	}
+
+	if err := app.PersistStats(); err != nil {
+		t.Fatal(err)
+	}
+	cleanInfo, err := os.Stat(app.opts.StatsPath)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if os.SameFile(afterFlushInfo, cleanInfo) {
+		t.Fatal("fixed-interval persistence unexpectedly skipped a write")
+	}
+}
+
+func TestReloadNoticeIsIndependentFromPersistence(t *testing.T) {
+	app := newTestApp(t, minimalTestConfig())
+	app.reloadPending.Store(true)
+	if err := app.PersistStats(); err != nil {
+		t.Fatal(err)
+	}
+	data, err := os.ReadFile(app.opts.StatsPath)
+	if err != nil {
+		t.Fatal(err)
+	}
+	var persisted map[string]any
+	if err := json.Unmarshal(data, &persisted); err != nil {
+		t.Fatal(err)
+	}
+	if _, exists := persisted["reload"]; exists {
+		t.Fatal("reload notice must not be persisted to stats.json")
+	}
+	if reloaded, _ := app.SnapshotStats()["reload"].(bool); !reloaded {
+		t.Fatal("pending reload notice was not returned by the live endpoint")
+	}
+	if _, exists := app.SnapshotStats()["reload"]; exists {
+		t.Fatal("reload notice was returned more than once")
+	}
+}
+
+func TestResetTrafficPersistsImmediately(t *testing.T) {
+	app := newTestApp(t, minimalTestConfig())
+	app.nodeMu.Lock()
+	node := app.nodes["s01"]
+	node.Connected = true
+	node.HasUpdate = true
+	node.Stats = AgentStats{NetworkIn: 1234, NetworkOut: 5678}
+	app.nodeMu.Unlock()
+
+	if _, apiErr := app.ResetTraffic("s01"); apiErr != nil {
+		t.Fatal(apiErr)
+	}
+	persisted := readPersistedServer(t, app)
+	if persisted["last_network_in"] != float64(1234) || persisted["last_network_out"] != float64(5678) {
+		t.Fatalf("traffic reset was not persisted immediately: %#v", persisted)
+	}
+}

server/ssl.go

@@ -98,7 +98,6 @@ func (a *App) executeSSLCheck(key string, config SSLCertConfig) {
 	if err != nil {
 		state.LastError = err.Error()
 		a.certMu.Unlock()
-		a.wakeStatsWriter()
 		return
 	}
 	state.ExpireTS = expireTS
@@ -126,7 +125,6 @@ func (a *App) executeSSLCheck(key string, config SSLCertConfig) {
 		}
 	}
 	a.certMu.Unlock()
-	a.wakeStatsWriter()
 
 	for _, alert := range alerts {
 		if err := a.sendCallback(config.Callback, alert.message, "ServerStatusSSL"); err != nil {

server/tcp_server.go

@@ -176,7 +176,6 @@ func (a *App) connectAgent(username, password string, conn net.Conn, family int)
 	node.Pong = false
 	node.Online4 = family == 4
 	node.Online6 = family == 6
-	a.wakeStatsWriter()
 	return id, append([]MonitorConfig(nil), a.runtime.Monitors...), nil
 }
 
@@ -194,7 +193,6 @@ func (a *App) disconnectAgent(username string, conn net.Conn, connectionID uint6
 	node.HasUpdate = false
 	node.Pong = false
 	a.nodeMu.Unlock()
-	a.wakeStatsWriter()
 	time.AfterFunc(25*time.Second, func() {
 		if a.ctx.Err() != nil {
 			return
@@ -226,7 +224,6 @@ func (a *App) updateAgent(username string, connectionID uint64, update AgentStat
 	node.HasUpdate = true
 	node.LastUpdate = time.Now()
 	a.nodeMu.Unlock()
-	a.wakeStatsWriter()
 	a.evaluateWatchdogs(username, false)
 	return true
 }