fix: fix deduplication manager (call to CleanupExpiredEntries)

web/apistream.go

@@ -664,7 +664,7 @@ func (w *WebAPIServer) PutRecords(c *fiber.Ctx) error {
 		return apiErr.HTTPResponse(c)
 	}
 
-	// check batch id for deduplication
+	// check batch id for batch deduplication
 	batch_id := c.Get("x-ministream-batch-id", "")
 	dedup_id := fmt.Sprintf("%s:%s", streamPtr.GetUUID().String(), batch_id)
 	if batch_id != "" {

web/requestdeduplicator .go

@@ -1,6 +1,7 @@
 package web
 
 import (
+	"errors"
 	"sync"
 	"time"
 )
@@ -14,6 +15,7 @@ type RequestDeduplicatorManager struct {
 }
 
 func (rdm *RequestDeduplicatorManager) Init() {
+	// Start the cleanup goroutine to remove expired entries periodically
 	rdm.stopChan = make(chan struct{})
 	rdm.wg.Add(1)
 	go func() {
@@ -31,10 +33,15 @@ func (rdm *RequestDeduplicatorManager) Init() {
 	}()
 }
 
-func (rdm *RequestDeduplicatorManager) Stop() {
+func (rdm *RequestDeduplicatorManager) Stop() error {
+	if rdm.stopChan == nil {
+		// the manager has not been initialized
+		return errors.New("RequestDeduplicatorManager has not been initialized")
+	}
 	close(rdm.stopChan)
 	rdm.wg.Wait()
 	rdm.Clear()
+	return nil
 }
 
 func (rdm *RequestDeduplicatorManager) Exists(requestID string) bool {
@@ -46,6 +53,9 @@ func (rdm *RequestDeduplicatorManager) Exists(requestID string) bool {
 }
 
 func (rdm *RequestDeduplicatorManager) Add(requestID string) {
+	// Add the requestID to the map
+	// a requestID is considered as processed when it is added to the map
+	// a requestID is composed of "<stream UUID>:<request batch ID>"
 	rdm.mu.Lock()
 	rdm.processedIDs[requestID] = time.Now()
 	rdm.mu.Unlock()
@@ -59,6 +69,11 @@ func (rdm *RequestDeduplicatorManager) Remove(requestID string) {
 
 // cleanupExpiredEntries periodically removes expired entries
 func (rdm *RequestDeduplicatorManager) CleanupExpiredEntries() {
+	// In order to keep the map small in memory, we remove expired entries from the map periodically.
+	// Keep the last n minutes of requestIDs in the map.
+	// It means that if a new request with a same requestID is received after n minutes,
+	// it will not be considered as a duplicate.
+	// This is a trade-off between memory and performance.
 	rdm.mu.Lock()
 	for requestID, timestamp := range rdm.processedIDs {
 		if time.Since(timestamp) > rdm.ttl {
@@ -69,6 +84,7 @@ func (rdm *RequestDeduplicatorManager) CleanupExpiredEntries() {
 }
 
 func (rdm *RequestDeduplicatorManager) Clear() {
+	// Clear the map (remove all requestIDs from the map)
 	rdm.mu.Lock()
 	rdm.processedIDs = make(map[string]time.Time)
 	rdm.mu.Unlock()

web/webapiserver.go

@@ -122,12 +122,16 @@ func (w *WebAPIServer) AddRoutes(app *fiber.App) {
 }
 
 func (w *WebAPIServer) ShutdownServer() {
-	w.reqDedupManager.Stop()
+	_ = w.reqDedupManager.Stop()
 	w.funcShutdownServer()
 }
 
+func (w *WebAPIServer) StartServer() {
+	w.reqDedupManager.Init()
+}
+
 func (w *WebAPIServer) RestartServer() {
-	w.reqDedupManager.Stop()
+	_ = w.reqDedupManager.Stop()
 	w.reqDedupManager.Init()
 	w.funcRestartServer()
 }

web/webserver/webserver.go

@@ -79,9 +79,11 @@ func (s *Server) Start() error {
 	if s.status != ServerStatusInitialized {
 		return fmt.Errorf("cannot start server: invalid status code %d", s.status)
 	}
+	s.webAPIServer.StartServer()
 	s.SetStatus(ServerStatusRunning)
 	s.Listen()
 	err := s.HandleSignals()
+	// server is stopped
 	s.SetStatus(ServerStatusInitialized)
 	return err
 }