fix(backend): fix memory leaks and resource issues in logs system
CI / build (push) Successful in 1m18s
CI / build (push) Successful in 1m18s
- 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
This commit is contained in:
@@ -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)
|
||||
|
||||
@@ -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()
|
||||
}
|
||||
|
||||
@@ -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()
|
||||
|
||||
@@ -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()
|
||||
|
||||
Reference in New Issue
Block a user