fix: remove dead code, SSE error handling, bounded dedup maps

- Remove unused mustParseDuration from main.go
- Replace http.Error calls after SSE headers with plain returns;
  skip bad packets instead of killing the stream on marshal error
- Change processedPackets/seenPackets from boolean to timestamp values
  and prune entries older than 24h on each packet to prevent unbounded
  memory growth in long-running sessions

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
Daniel Pupius
2026-03-15 16:27:57 +00:00
parent 9e5fd5bcae
commit 1d61a89505
4 changed files with 29 additions and 24 deletions

View File

@@ -236,7 +236,6 @@ func (s *Server) handleStream(w http.ResponseWriter, r *http.Request) {
// Client disconnected, unsubscribe and return
logger.Info("Client disconnected, unsubscribing from broker")
s.config.Broker.Unsubscribe(packetChan)
http.Error(w, "Client disconnected", http.StatusGone)
return
case <-s.shutdown:
@@ -245,7 +244,6 @@ func (s *Server) handleStream(w http.ResponseWriter, r *http.Request) {
fmt.Fprintf(w, "event: info\ndata: Server shutting down, connection closed\n\n")
flusher.Flush()
s.config.Broker.Unsubscribe(packetChan)
http.Error(w, "Server is shutting down", http.StatusServiceUnavailable)
return
case <-heartbeatTicker.C:
@@ -259,7 +257,6 @@ func (s *Server) handleStream(w http.ResponseWriter, r *http.Request) {
if !ok {
// Channel closed, probably shutting down
logger.Info("Packet channel closed, ending stream")
http.Error(w, "Server is shutting down", http.StatusServiceUnavailable)
return
}
@@ -278,8 +275,7 @@ func (s *Server) handleStream(w http.ResponseWriter, r *http.Request) {
data, err := marshaler.Marshal(packet)
if err != nil {
logger.Errorw("Error marshaling packet to JSON", "error", err)
http.Error(w, "Error marshaling packet", http.StatusInternalServerError)
return
continue
}
// Send the event