f-29: stop: add a proper stop

Signed-off-by: GnomeZworc <nicolas.boufidjeline@g3e.fr>

pkg/db/kv/admin_server.go

@@ -1,6 +1,8 @@
 package kv
 
 import (
+	"context"
+	"errors"
 	"fmt"
 	"log/slog"
 	"net/http"
@@ -12,21 +14,28 @@ import (
 type AdminServer struct {
 	db     *badger.DB
 	logger *slog.Logger
+	srv    *http.Server
 }
 
-func NewAdminServer(db *badger.DB, logger *slog.Logger) *AdminServer {
-	return &AdminServer{db: db, logger: logger}
-}
-
-func (s *AdminServer) Start(address string) {
+func NewAdminServer(db *badger.DB, logger *slog.Logger, address string) *AdminServer {
+	s := &AdminServer{db: db, logger: logger}
 	mux := http.NewServeMux()
 	mux.HandleFunc("/db", s.dbHandler)
-	s.logger.Info("admin server listening", "address", address)
-	if err := http.ListenAndServe(address, mux); err != nil {
+	s.srv = &http.Server{Addr: address, Handler: mux}
+	return s
+}
+
+func (s *AdminServer) Start() {
+	s.logger.Info("admin server listening", "address", s.srv.Addr)
+	if err := s.srv.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) {
 		s.logger.Error("admin server stopped", "error", err)
 	}
 }
 
+func (s *AdminServer) Shutdown(ctx context.Context) error {
+	return s.srv.Shutdown(ctx)
+}
+
 func (s *AdminServer) dbHandler(w http.ResponseWriter, r *http.Request) {
 	if r.Method != http.MethodGet {
 		http.Error(w, "method not allowed", http.StatusMethodNotAllowed)

pkg/prometheus/server.go

@@ -1,6 +1,8 @@
 package promserver
 
 import (
+	"context"
+	"errors"
 	"log"
 	"net/http"
 
@@ -8,13 +10,30 @@ import (
 	"github.com/prometheus/client_golang/prometheus/promhttp"
 )
 
-// Start launches the Prometheus metrics HTTP server on the given address.
-// The provided registry is used to expose metrics at /metrics.
-func Start(address string, registry *prometheus.Registry) {
+// Server exposes the Prometheus metrics endpoint.
+type Server struct {
+	srv *http.Server
+}
+
+// New builds the metrics server for the given address and registry.
+func New(address string, registry *prometheus.Registry) *Server {
 	mux := http.NewServeMux()
 	mux.Handle("/metrics", promhttp.HandlerFor(registry, promhttp.HandlerOpts{
 		EnableOpenMetrics: true,
 	}))
-	log.Printf("Prometheus server listening on %s", address)
-	log.Fatal(http.ListenAndServe(address, mux))
+	return &Server{srv: &http.Server{Addr: address, Handler: mux}}
+}
+
+// Start blocks until the server stops. A failure is logged, never fatal: an
+// unavailable metrics endpoint must not bring the whole agent down.
+func (s *Server) Start() {
+	log.Printf("Prometheus server listening on %s", s.srv.Addr)
+	if err := s.srv.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) {
+		log.Printf("Prometheus server stopped: %v", err)
+	}
+}
+
+// Shutdown stops the server, waiting for in-flight requests until ctx expires.
+func (s *Server) Shutdown(ctx context.Context) error {
+	return s.srv.Shutdown(ctx)
 }

pkg/worker/queue.go

@@ -1,13 +1,19 @@
 package worker
 
-import "log"
+import (
+	"log"
+	"sync"
+)
 
 // Task is a function to be executed asynchronously by a worker.
 type Task func()
 
 // Queue is a FIFO channel-backed task queue consumed by worker goroutines.
 type Queue struct {
-	tasks chan Task
+	tasks   chan Task
+	wg      sync.WaitGroup
+	mu      sync.RWMutex
+	stopped bool
 }
 
 // New creates a Queue with the given channel buffer size.
@@ -15,8 +21,16 @@ func New(bufferSize int) *Queue {
 	return &Queue{tasks: make(chan Task, bufferSize)}
 }
 
-// Submit enqueues a task. Blocks if the queue is full.
+// Submit enqueues a task. Blocks if the queue is full. Tasks submitted after
+// Stop are rejected and logged rather than enqueued.
 func (q *Queue) Submit(t Task) {
+	q.mu.RLock()
+	defer q.mu.RUnlock()
+
+	if q.stopped {
+		log.Print("worker: queue stopped, task rejected")
+		return
+	}
 	q.tasks <- t
 }
 
@@ -24,10 +38,28 @@ func (q *Queue) Submit(t Task) {
 func (q *Queue) Start(n int) {
 	log.Printf("worker: starting %d workers", n)
 	for i := range n {
+		q.wg.Add(1)
 		go func(id int) {
+			defer q.wg.Done()
 			for task := range q.tasks {
 				task()
 			}
 		}(i)
 	}
 }
+
+// Stop rejects new tasks, then waits for the queued and in-flight ones to
+// finish. The write lock is what makes closing the channel safe: it is only
+// taken once every in-flight Submit has released its read lock.
+func (q *Queue) Stop() {
+	q.mu.Lock()
+	if q.stopped {
+		q.mu.Unlock()
+		return
+	}
+	q.stopped = true
+	close(q.tasks)
+	q.mu.Unlock()
+
+	q.wg.Wait()
+}

pkg/worker/queue_test.go

@@ -101,3 +101,79 @@ func TestQueue_SubmitBlocksWhenFull(t *testing.T) {
 		t.Error("Submit aurait dû se débloquer après démarrage d'un worker")
 	}
 }
+
+func TestStop_AttendLesTachesEnCours(t *testing.T) {
+	q := New(10)
+	q.Start(2)
+
+	var mu sync.Mutex
+	done := 0
+	for range 5 {
+		q.Submit(func() {
+			time.Sleep(20 * time.Millisecond)
+			mu.Lock()
+			done++
+			mu.Unlock()
+		})
+	}
+
+	q.Stop()
+
+	mu.Lock()
+	defer mu.Unlock()
+	if done != 5 {
+		t.Errorf("Stop devrait attendre les 5 tâches, %d terminées", done)
+	}
+}
+
+func TestStop_RejetteLesTachesSuivantes(t *testing.T) {
+	q := New(10)
+	q.Start(1)
+	q.Stop()
+
+	executed := make(chan struct{}, 1)
+	q.Submit(func() { executed <- struct{}{} })
+
+	select {
+	case <-executed:
+		t.Error("une tâche soumise après Stop ne doit pas être exécutée")
+	case <-time.After(100 * time.Millisecond):
+	}
+}
+
+func TestStop_Idempotent(t *testing.T) {
+	q := New(10)
+	q.Start(1)
+	q.Stop()
+	q.Stop()
+}
+
+func TestStop_SansTacheEnCours(t *testing.T) {
+	q := New(10)
+	q.Start(3)
+
+	done := make(chan struct{})
+	go func() { q.Stop(); close(done) }()
+
+	select {
+	case <-done:
+	case <-time.After(2 * time.Second):
+		t.Fatal("Stop n'a pas rendu la main sur une file vide")
+	}
+}
+
+func TestSubmit_ConcurrentAvecStop(t *testing.T) {
+	q := New(100)
+	q.Start(4)
+
+	var wg sync.WaitGroup
+	for range 20 {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			q.Submit(func() {})
+		}()
+	}
+	q.Stop()
+	wg.Wait()
+}