From 8bf163a931140b310ff130775560a29de61d2074 Mon Sep 17 00:00:00 2001 From: MrFastwind Date: Fri, 24 Jul 2026 16:25:09 +0200 Subject: [PATCH] fix(backend): fix memory leaks and resource issues in logs system - Fix Subscribe overwriting channel without closing old one (goroutine leak) - Add context.Context to Stream() for cancellation support - Add WebSocket read pump to detect dead clients (StreamLogs, StreamSteamCMDLogs, StreamRPTLogs) - Use defer f.Close() pattern in StreamRPTLogs file handling - Fix findLatestRPT nil dereference on failed os.Stat - Optimize findLatestRPT to collect stat results before sorting - Apply gofmt formatting to modlists.go, settings.go --- backend/internal/api/logs.go | 125 ++++++++++++++---- backend/internal/api/modlists.go | 2 +- backend/internal/api/settings.go | 84 ++++++++---- backend/internal/services/log_streamer.go | 18 ++- .../internal/services/log_streamer_test.go | 68 +++++++++- backend/internal/services/server_process.go | 4 +- backend/internal/services/steamcmd.go | 4 +- 7 files changed, 242 insertions(+), 63 deletions(-) diff --git a/backend/internal/api/logs.go b/backend/internal/api/logs.go index c3d240c..4dd111c 100644 --- a/backend/internal/api/logs.go +++ b/backend/internal/api/logs.go @@ -19,6 +19,25 @@ var upgrader = websocket.Upgrader{ CheckOrigin: func(r *http.Request) bool { return true }, } +const wsReadTimeout = 60 * time.Second + +func readPump(conn *websocket.Conn, done chan struct{}) { + defer close(done) + conn.SetReadLimit(4096) + conn.SetReadDeadline(time.Now().Add(wsReadTimeout)) + conn.SetPongHandler(func(string) error { + conn.SetReadDeadline(time.Now().Add(wsReadTimeout)) + return nil + }) + for { + _, _, err := conn.ReadMessage() + if err != nil { + return + } + conn.SetReadDeadline(time.Now().Add(wsReadTimeout)) + } +} + func (h *Handler) StreamLogs(c *gin.Context) { conn, err := upgrader.Upgrade(c.Writer, c.Request, nil) if err != nil { @@ -27,13 +46,24 @@ func (h *Handler) StreamLogs(c *gin.Context) { } defer conn.Close() + done := make(chan struct{}) + go readPump(conn, done) + clientID := conn.RemoteAddr().String() ch := h.streamer.Subscribe("server", clientID) defer h.streamer.Unsubscribe("server", clientID) - for line := range ch { - if err := conn.WriteMessage(websocket.TextMessage, []byte(line)); err != nil { - break + for { + select { + case line, ok := <-ch: + if !ok { + return + } + if err := conn.WriteMessage(websocket.TextMessage, []byte(line)); err != nil { + return + } + case <-done: + return } } } @@ -83,13 +113,24 @@ func (h *Handler) StreamSteamCMDLogs(c *gin.Context) { } defer conn.Close() + done := make(chan struct{}) + go readPump(conn, done) + clientID := conn.RemoteAddr().String() ch := h.streamer.Subscribe("steamcmd", clientID) defer h.streamer.Unsubscribe("steamcmd", clientID) - for line := range ch { - if err := conn.WriteMessage(websocket.TextMessage, []byte(line)); err != nil { - break + for { + select { + case line, ok := <-ch: + if !ok { + return + } + if err := conn.WriteMessage(websocket.TextMessage, []byte(line)); err != nil { + return + } + case <-done: + return } } } @@ -102,13 +143,22 @@ func (h *Handler) StreamRPTLogs(c *gin.Context) { } defer conn.Close() + done := make(chan struct{}) + go readPump(conn, done) + profilesDir := h.process.ProfilesDir() var currentPath string var currentOffset int64 ticker := time.NewTicker(250 * time.Millisecond) defer ticker.Stop() - for range ticker.C { + for { + select { + case <-done: + return + case <-ticker.C: + } + path := findLatestRPT(profilesDir) if path != currentPath { @@ -134,24 +184,32 @@ func (h *Handler) StreamRPTLogs(c *gin.Context) { continue } - f, err := os.Open(currentPath) - if err != nil { - currentPath = "" - continue - } - f.Seek(currentOffset, io.SeekStart) - buf := make([]byte, fi.Size()-currentOffset) - n, _ := io.ReadFull(f, buf) - f.Close() + if err := func() error { + f, err := os.Open(currentPath) + if err != nil { + currentPath = "" + return nil + } + defer f.Close() - currentOffset += int64(n) - for _, line := range strings.Split(string(buf[:n]), "\n") { - if line == "" { - continue + if _, err := f.Seek(currentOffset, io.SeekStart); err != nil { + return nil } - if err := conn.WriteMessage(websocket.TextMessage, []byte(line)); err != nil { - return + buf := make([]byte, fi.Size()-currentOffset) + n, _ := io.ReadFull(f, buf) + + currentOffset += int64(n) + for _, line := range strings.Split(string(buf[:n]), "\n") { + if line == "" { + continue + } + if err := conn.WriteMessage(websocket.TextMessage, []byte(line)); err != nil { + return err + } } + return nil + }(); err != nil { + return } } } @@ -162,12 +220,25 @@ func findLatestRPT(dir string) string { if err != nil || len(matches) == 0 { return "" } - sort.Slice(matches, func(i, j int) bool { - fi, _ := os.Stat(matches[i]) - fj, _ := os.Stat(matches[j]) - return fi.ModTime().After(fj.ModTime()) + type rptInfo struct { + path string + time time.Time + } + var infos []rptInfo + for _, m := range matches { + fi, err := os.Stat(m) + if err != nil { + continue + } + infos = append(infos, rptInfo{path: m, time: fi.ModTime()}) + } + if len(infos) == 0 { + return "" + } + sort.Slice(infos, func(i, j int) bool { + return infos[i].time.After(infos[j].time) }) - return matches[0] + return infos[0].path } func (h *Handler) GetLog(c *gin.Context) { diff --git a/backend/internal/api/modlists.go b/backend/internal/api/modlists.go index 99981a7..cad1e67 100644 --- a/backend/internal/api/modlists.go +++ b/backend/internal/api/modlists.go @@ -16,7 +16,7 @@ type createModlistInput struct { } type updateModlistInput struct { - Name string `json:"name" binding:"required"` + Name string `json:"name" binding:"required"` Mods []models.ModEntry `json:"mods"` } diff --git a/backend/internal/api/settings.go b/backend/internal/api/settings.go index 4b5cf97..1726921 100644 --- a/backend/internal/api/settings.go +++ b/backend/internal/api/settings.go @@ -7,20 +7,20 @@ import ( ) type updateSettingsInput struct { - IPPort *string `json:"ip_port"` - ServerParameters *string `json:"server_parameters"` - SteamBranch *string `json:"steam_branch"` - SteamUser *string `json:"steam_user"` - Platform *string `json:"platform"` - CBASettings *string `json:"cba_settings"` - AILevelPresets *string `json:"ai_level_presets"` - DifficultyPresets *string `json:"difficulty_presets"` - ActiveConfig *string `json:"active_config"` - ActiveModlist *string `json:"active_modlist"` - AutoUpdateOnStartup *bool `json:"auto_update_on_startup"` - AutoStartOnStartup *bool `json:"auto_start_on_startup"` - AutoUpdateModsOnStartup *bool `json:"auto_update_mods_on_startup"` - ScheduledUpdate *string `json:"scheduled_update"` + IPPort *string `json:"ip_port"` + ServerParameters *string `json:"server_parameters"` + SteamBranch *string `json:"steam_branch"` + SteamUser *string `json:"steam_user"` + Platform *string `json:"platform"` + CBASettings *string `json:"cba_settings"` + AILevelPresets *string `json:"ai_level_presets"` + DifficultyPresets *string `json:"difficulty_presets"` + ActiveConfig *string `json:"active_config"` + ActiveModlist *string `json:"active_modlist"` + AutoUpdateOnStartup *bool `json:"auto_update_on_startup"` + AutoStartOnStartup *bool `json:"auto_start_on_startup"` + AutoUpdateModsOnStartup *bool `json:"auto_update_mods_on_startup"` + ScheduledUpdate *string `json:"scheduled_update"` } func (h *Handler) GetSettings(c *gin.Context) { @@ -45,20 +45,48 @@ func (h *Handler) UpdateSettings(c *gin.Context) { return } - if input.IPPort != nil { s.IPPort = *input.IPPort } - if input.ServerParameters != nil { s.ServerParameters = *input.ServerParameters } - if input.SteamBranch != nil { s.SteamBranch = *input.SteamBranch } - if input.SteamUser != nil { s.SteamUser = *input.SteamUser } - if input.Platform != nil { s.Platform = *input.Platform } - if input.CBASettings != nil { s.CBASettings = *input.CBASettings } - if input.AILevelPresets != nil { s.AILevelPresets = *input.AILevelPresets } - if input.DifficultyPresets != nil { s.DifficultyPresets = *input.DifficultyPresets } - if input.ActiveConfig != nil { s.ActiveConfig = *input.ActiveConfig } - if input.ActiveModlist != nil { s.ActiveModlist = *input.ActiveModlist } - if input.AutoUpdateOnStartup != nil { s.AutoUpdateOnStartup = *input.AutoUpdateOnStartup } - if input.AutoStartOnStartup != nil { s.AutoStartOnStartup = *input.AutoStartOnStartup } - if input.AutoUpdateModsOnStartup != nil { s.AutoUpdateModsOnStartup = *input.AutoUpdateModsOnStartup } - if input.ScheduledUpdate != nil { s.ScheduledUpdate = *input.ScheduledUpdate } + if input.IPPort != nil { + s.IPPort = *input.IPPort + } + if input.ServerParameters != nil { + s.ServerParameters = *input.ServerParameters + } + if input.SteamBranch != nil { + s.SteamBranch = *input.SteamBranch + } + if input.SteamUser != nil { + s.SteamUser = *input.SteamUser + } + if input.Platform != nil { + s.Platform = *input.Platform + } + if input.CBASettings != nil { + s.CBASettings = *input.CBASettings + } + if input.AILevelPresets != nil { + s.AILevelPresets = *input.AILevelPresets + } + if input.DifficultyPresets != nil { + s.DifficultyPresets = *input.DifficultyPresets + } + if input.ActiveConfig != nil { + s.ActiveConfig = *input.ActiveConfig + } + if input.ActiveModlist != nil { + s.ActiveModlist = *input.ActiveModlist + } + if input.AutoUpdateOnStartup != nil { + s.AutoUpdateOnStartup = *input.AutoUpdateOnStartup + } + if input.AutoStartOnStartup != nil { + s.AutoStartOnStartup = *input.AutoStartOnStartup + } + if input.AutoUpdateModsOnStartup != nil { + s.AutoUpdateModsOnStartup = *input.AutoUpdateModsOnStartup + } + if input.ScheduledUpdate != nil { + s.ScheduledUpdate = *input.ScheduledUpdate + } if err := h.process.WriteUserconfigFiles(s); err != nil { c.JSON(http.StatusInternalServerError, gin.H{"error": "write userconfig: " + err.Error()}) diff --git a/backend/internal/services/log_streamer.go b/backend/internal/services/log_streamer.go index bb2d606..8ba2ba4 100644 --- a/backend/internal/services/log_streamer.go +++ b/backend/internal/services/log_streamer.go @@ -2,6 +2,7 @@ package services import ( "bufio" + "context" "io" "log" "sync" @@ -26,6 +27,10 @@ func (ls *LogStreamer) Subscribe(serverID, clientID string) chan string { ls.subs[serverID] = make(map[string]chan string) } + if old, ok := ls.subs[serverID][clientID]; ok { + close(old) + } + ch := make(chan string, 256) ls.subs[serverID][clientID] = ch return ch @@ -46,14 +51,23 @@ func (ls *LogStreamer) Unsubscribe(serverID, clientID string) { } } -func (ls *LogStreamer) Stream(serverID string, reader io.Reader, closeMsg string) { +func (ls *LogStreamer) Stream(ctx context.Context, serverID string, reader io.Reader, closeMsg string) { scanner := bufio.NewScanner(reader) for scanner.Scan() { + select { + case <-ctx.Done(): + return + default: + } line := scanner.Text() ls.broadcast(serverID, line) } if err := scanner.Err(); err != nil { - log.Printf("log stream error for server %s: %v", serverID, err) + select { + case <-ctx.Done(): + default: + log.Printf("log stream error for server %s: %v", serverID, err) + } } if closeMsg != "" { ls.broadcast(serverID, closeMsg) diff --git a/backend/internal/services/log_streamer_test.go b/backend/internal/services/log_streamer_test.go index 5151555..873fd4a 100644 --- a/backend/internal/services/log_streamer_test.go +++ b/backend/internal/services/log_streamer_test.go @@ -1,6 +1,7 @@ package services import ( + "context" "strings" "sync" "testing" @@ -95,7 +96,7 @@ func TestLogStreamer_Stream(t *testing.T) { ch := ls.Subscribe("server", "client1") reader := strings.NewReader("line1\nline2\nline3\n") - go ls.Stream("server", reader, "DONE") + go ls.Stream(context.Background(), "server", reader, "DONE") lines := []string{} for i := 0; i < 4; i++ { // 3 lines + DONE close message @@ -174,3 +175,68 @@ func TestLogStreamer_ConcurrentSubscribeUnsubscribe(t *testing.T) { wg.Wait() } + +func TestLogStreamer_SubscribeOverwriteClosesOld(t *testing.T) { + ls := NewLogStreamer() + + ch1 := ls.Subscribe("server", "client1") + ls.Broadcast("server", "first") + <-ch1 + + // Re-subscribe with same clientID — old channel should be closed + ch2 := ls.Subscribe("server", "client1") + + // Old channel should be closed + select { + case _, ok := <-ch1: + if ok { + t.Error("old channel should be closed after re-subscribe") + } + case <-time.After(100 * time.Millisecond): + t.Error("old channel not closed within timeout") + } + + // New channel should work + ls.Broadcast("server", "second") + select { + case line := <-ch2: + if line != "second" { + t.Errorf("new channel received %q, want %q", line, "second") + } + case <-time.After(100 * time.Millisecond): + t.Error("timeout waiting on new channel") + } +} + +func TestLogStreamer_StreamContextCancel(t *testing.T) { + ls := NewLogStreamer() + ch := ls.Subscribe("server", "client1") + + ctx, cancel := context.WithCancel(context.Background()) + reader := contextReader{ctx: ctx} + go ls.Stream(ctx, "server", reader, "") + + // Broadcast should still work while stream is running + ls.Broadcast("server", "live") + select { + case line := <-ch: + if line != "live" { + t.Errorf("received %q, want %q", line, "live") + } + case <-time.After(100 * time.Millisecond): + t.Error("timeout waiting for broadcast") + } + + // Cancel context — stream should stop + cancel() + time.Sleep(100 * time.Millisecond) +} + +type contextReader struct { + ctx context.Context +} + +func (r contextReader) Read(p []byte) (int, error) { + <-r.ctx.Done() + return 0, r.ctx.Err() +} diff --git a/backend/internal/services/server_process.go b/backend/internal/services/server_process.go index 66c19b8..58dc2b8 100644 --- a/backend/internal/services/server_process.go +++ b/backend/internal/services/server_process.go @@ -148,8 +148,8 @@ func (pm *ProcessManager) Start() error { pm.mu.Unlock() pm.state.Store(int32(procRunning)) - go pm.streamer.Stream("server", stdout, "") - go pm.streamer.Stream("server", stderr, "") + go pm.streamer.Stream(ctx, "server", stdout, "") + go pm.streamer.Stream(ctx, "server", stderr, "") go func() { cmd.Wait() diff --git a/backend/internal/services/steamcmd.go b/backend/internal/services/steamcmd.go index 8ab906e..37bebe1 100644 --- a/backend/internal/services/steamcmd.go +++ b/backend/internal/services/steamcmd.go @@ -117,8 +117,8 @@ func (s *SteamCmdManager) run(label string, args []string) error { s.cancel = cancel - go s.streamer.Stream("steamcmd", stdout, "") - go s.streamer.Stream("steamcmd", stderr, "") + go s.streamer.Stream(ctx, "steamcmd", stdout, "") + go s.streamer.Stream(ctx, "steamcmd", stderr, "") go func() { err := cmd.Wait()