feat(queue): Rate-Limit und Ruhezeiten als Gesetz im Worker-Loop

Das Anti-Spam-Gesetz wird an den Loop verwired (die Rate-Limiter-Struktur aus Task 04): Jede gestartete Ansage bucht das Fenster — low über dem Budget verfällt (beim Eintritt, Admission nennt den Grund für die 202-Antwort), normal parkt bis zum Slot (Slot-Wecker broadcastet die Cond, geparkte bleiben in der Queue und in /status sichtbar), high darf mit Burst-Recht verstoßen, now bricht bewusst aus Rate-Limit UND Ruhezeit aus und bucht trotzdem ein neues Fenster. Ruhezeiten (quiet.go: lokales Fenster HH:MM-HH:MM über Mitternacht) verwerfen still oder ersetzen durch Toast — gilt auch für high. Rangordnung: mute (API) > now > Ruhezeit > Rate-Limit. Enqueue liefert Admission statt nur ID; LawState für /status. Lock-Ordnung Queue → RateLimiter bleibt gewahrt.
2026-10-18 17:40:41 +02:00
1 parent b20e6dd319
commit 5385a4b87d
@@ -92,50 +92,91 @@ func (it Item) rank() int {
 type Speaker func(ctx context.Context, item Item) error

 // Queue ist die serialisierte Ansage-Queue des Daemons: genau eine Ansage
-// gleichzeitig, Reihenfolge now > high > normal > low (FIFO je Level).
+// gleichzeitig, Reihenfolge now > high > normal > low (FIFO je Level) —
+// seit Task 05 unter dem Anti-Spam-Gesetz (Laws): normal parkt bis zum
+// Rate-Limit-Slot, low wird über dem Budget verworfen, high darf mit
+// Burst-Recht verstoßen, now bricht bewusst aus, und Ruhezeiten verwerfen
+// bzw. toasten statt Audio.
 //
 // Die Synchronisation ist bewusst simpel und offensichtlich korrekt statt
 // clever: EIN Mutex + EIN Cond schützen alles (Items, aktuelles Item,
-// Closed-Flag, cancelCur). Der Worker ist der einzige, der Items entnimmt;
-// pop und das Setzen von current/cancelCur geschehen unter demselben Lock,
-// damit ein Item nie zwischen „wartet" und „läuft" der Cancelierung
-// entwischen kann. Cancel/Close rufen cancelCur unter dem Lock ab — der
-// Worker bemerkt die Abbrechung in seiner speak-Funktion über ctx.
+// Closed-Flag, cancelCur, Slot-Wecker). Der Worker ist der einzige, der
+// Items entnimmt; take und das Setzen von current/cancelCur geschehen unter
+// demselben Lock, damit ein Item nie zwischen „wartet" und „läuft" der
+// Cancelierung entwischen kann. Cancel/Close rufen cancelCur unter dem Lock
+// ab — der Worker bemerkt die Abbrechung in seiner speak-Funktion über ctx.
+// Geparkte Items bleiben in der Warteschlange liegen; ein Slot-Wecker
+// (armWakeLocked) broadcastet zum Fenster-Ablauf, damit der Worker auch ohne
+// neuen Enqueue aufwacht — die Wait-Bedingung prüft jedes Mal die Gesetze
+// neu, verlorene Signale sind dadurch harmlos.
 type Queue struct {
+	laws  Laws
+	limit *RateLimiter
 	speak Speaker
 	log   io.Writer

 	mu        sync.Mutex
-	notEmpty  *sync.Cond // Signal: neues Item; Broadcast: Close
+	notEmpty  *sync.Cond // Signal: neues Item; Broadcast: Close/Slot frei
 	items     []Item
 	current   *Item
 	cancelCur context.CancelFunc
 	seq       uint64
 	closed    bool
+	wake      *time.Timer   // weckt den geparkten Worker zum Slot-Termin
+	wakeAt    time.Time     // Termin des laufenden Weckers
 	done      chan struct{} // geschlossen, wenn der Worker-Loop endet
 }

 // New startet die Queue mit ihrem Worker-Loop (eine Ansage gleichzeitig).
-// log ist optional (nil = still) und empfängt die Lebenslauf-Zeilen der
-// Items. Close beendet den Loop und wartet auf sein Ende.
-func New(speak Speaker, log io.Writer) *Queue {
-	q := &Queue{speak: speak, log: log, done: make(chan struct{})}
+// laws trägt die Anti-Spam-Gesetze (Nullwert = alles erlaubt; der Daemon
+// liefert die Config-Gesetze). log ist optional (nil = still) und empfängt
+// die Lebenslauf-Zeilen der Items. Close beendet den Loop und wartet auf
+// sein Ende.
+func New(laws Laws, speak Speaker, log io.Writer) *Queue {
+	q := &Queue{
+		laws:  laws,
+		limit: &RateLimiter{Interval: laws.RateLimit},
+		speak: speak,
+		log:   log,
+		done:  make(chan struct{}),
+	}
 	q.notEmpty = sync.NewCond(&q.mu)
 	go q.loop()
 	return q
 }

-// Enqueue stellt item in die Queue und liefert dessen ID (bei leerer ID neu
-// vergeben: kurz und zufällig). Nach Close ist Enqueue ein Fehler.
-func (q *Queue) Enqueue(item Item) (string, error) {
+// Enqueue stellt item in die Warteschlange und liefert die Aufnahme-
+// Entscheidung: Aufgenommene tragen ihre ID (bei leerer ID neu vergeben:
+// kurz und zufällig). Ruhezeit-Verwurf/Toast und low über dem Fenster-Budget
+// werden schon beim Eintritt entschieden (die 202-Antwort von POST /speak
+// nennt den Grund); alles andere entscheidet der Worker, wenn das Item
+// fällig ist. Nach Close ist Enqueue ein Fehler.
+func (q *Queue) Enqueue(item Item) (Admission, error) {
 	q.mu.Lock()
 	if q.closed {
 		q.mu.Unlock()
-		return "", ErrClosed
+		return Admission{}, ErrClosed
 	}
 	if item.Priority == "" {
 		item.Priority = PriorityNormal
 	}
+	switch q.laws.decide(item, time.Now(), q.limit.Allow()) {
+	case verdictDiscardQuiet:
+		q.mu.Unlock()
+		q.logf("queue: ansage (%q, priorität %s) verworfen — ruhezeit %s aktiv (modus discard)",
+			item.Text, item.Priority, q.laws.Quiet)
+		return Admission{Discarded: DiscardQuietHours}, nil
+	case verdictToast:
+		q.mu.Unlock()
+		q.logf("queue: ansage (%q, priorität %s) durch toast ersetzt — ruhezeit %s aktiv (modus toast)",
+			item.Text, item.Priority, q.laws.Quiet)
+		return Admission{Discarded: DiscardQuietHours, Toast: true}, nil
+	case verdictDiscardRate:
+		q.mu.Unlock()
+		q.logf("queue: ansage (%q, priorität low) verworfen — rate-limit %s",
+			item.Text, q.laws.RateLimit)
+		return Admission{Discarded: DiscardRateLimit}, nil
+	}
 	q.seq++
 	item.seq = q.seq
 	if item.ID == "" {
@@ -144,7 +185,7 @@ func (q *Queue) Enqueue(item Item) (string, error) {
 	q.items = append(q.items, item)
 	q.mu.Unlock()
 	q.notEmpty.Signal()
-	return item.ID, nil
+	return Admission{ID: item.ID}, nil
 }

 // Pending liefert eine Momentaufnahme der wartenden Items in
@@ -208,64 +249,171 @@ func (q *Queue) Close() {
 	if q.cancelCur != nil {
 		q.cancelCur()
 	}
+	if q.wake != nil {
+		q.wake.Stop()
+		q.wake = nil
+	}
 	q.mu.Unlock()
 	q.notEmpty.Broadcast()
 	<-q.done
 }

-// loop ist der Worker: wartet auf Items und spielt sie einzeln ab. Nach
-// Close (oder wenn Close während einer Ansage zuschlägt) beendet er sich,
-// sobald die laufende Ansage zurückgekehrt ist.
+// LawState liefert die /status-Sicht auf die Anti-Spam-Gesetze: das
+// konfigurierte Rate-Limit mit dem nächsten freien Slot (Nullzeit = frei
+// oder unbegrenzt) und das Ruhezeit-Fenster mit seinem Ende, wenn es
+// gerade aktiv ist.
+func (q *Queue) LawState() LawState {
+	now := time.Now()
+	return LawState{
+		RateLimit:      q.laws.RateLimit,
+		AllowHighBurst: q.laws.AllowHighBurst,
+		NextSlot:       q.limit.Next(),
+		Quiet:          q.laws.Quiet,
+		QuietActive:    q.laws.Quiet.Active(now),
+		QuietMode:      q.laws.QuietMode,
+		QuietUntil:     q.laws.Quiet.Until(now),
+	}
+}
+
+// loop ist der Worker: wartet auf fällige Items und spielt sie einzeln ab.
+// „Fällig" heißt seit Task 05 auch gesetzkonform: Geparkte schlafen bis zum
+// Rate-Limit-Slot (armWakeLocked weckt), low- und Ruhezeit-Verwürfe laufen
+// ohne Audio durchs Protokoll, now und high-burst spielen sofort.
+// Nach Close (oder wenn Close während einer Ansage zuschlägt) beendet er
+// sich, sobald die laufende Ansage zurückgekehrt ist.
 func (q *Queue) loop() {
 	defer close(q.done)
 	for {
 		q.mu.Lock()
-		for len(q.items) == 0 && !q.closed {
+		for !q.closed && !q.readyLocked() {
+			if len(q.items) > 0 {
+				// Bestes Item parkt auf den Slot: Wecker zum Termin armieren.
+				q.armWakeLocked(q.limit.Next())
+			}
 			q.notEmpty.Wait()
 		}
 		if q.closed {
 			q.mu.Unlock()
 			return
 		}
-		item := q.popLocked()
-		ctx, cancel := context.WithCancel(context.Background())
-		q.current = &item
-		q.cancelCur = cancel
+		item, v := q.takeLocked()
+		if v == verdictWait {
+			// Kurz ohne Fortschritt geweckt (z. B. verlorenes Signal):
+			// wieder schlafen legen. Kann nach readyLocked eigentlich nie
+			// eintreten — sicherheitshalber geparkt statt gedreht.
+			q.mu.Unlock()
+			continue
+		}
+		if v == verdictPlay {
+			// Fenster buchen, BEVOR gesprochen wird: Sobald das Item als
+			// laufend gilt, steht der nächste Slot-Termin fest.
+			q.limit.Reserve()
+			ctx, cancel := context.WithCancel(context.Background())
+			q.current = &item
+			q.cancelCur = cancel
+			q.mu.Unlock()
+
+			q.logf("queue: ansage %s (%q, Priorität %s, now=%v) startet", item.ID, item.Text, item.Priority, item.Now)
+			start := time.Now()
+			err := q.speak(ctx, item)
+			dur := time.Since(start)
+			cancel()
+
+			q.mu.Lock()
+			q.current = nil
+			q.cancelCur = nil
+			q.mu.Unlock()
+
+			switch {
+			case ctx.Err() != nil:
+				q.logf("queue: ansage %s abgebrochen", item.ID)
+			case err != nil:
+				q.logf("queue: ansage %s fehlgeschlagen nach %s: %v", item.ID, dur.Round(time.Millisecond), err)
+			default:
+				q.logf("queue: ansage %s fertig nach %s", item.ID, dur.Round(time.Millisecond))
+			}
+			continue
+		}
 		q.mu.Unlock()
-
-		q.logf("queue: ansage %s (%q, Priorität %s, now=%v) startet", item.ID, item.Text, item.Priority, item.Now)
-		start := time.Now()
-		err := q.speak(ctx, item)
-		dur := time.Since(start)
-		cancel()
-
-		q.mu.Lock()
-		q.current = nil
-		q.cancelCur = nil
-		q.mu.Unlock()
-
-		switch {
-		case ctx.Err() != nil:
-			q.logf("queue: ansage %s abgebrochen", item.ID)
-		case err != nil:
-			q.logf("queue: ansage %s fehlgeschlagen nach %s: %v", item.ID, dur.Round(time.Millisecond), err)
-		default:
-			q.logf("queue: ansage %s fertig nach %s", item.ID, dur.Round(time.Millisecond))
+		switch v {
+		case verdictDiscardRate:
+			q.logf("queue: ansage %s (%q, priorität low) verworfen — rate-limit %s",
+				item.ID, item.Text, q.laws.RateLimit)
+		case verdictDiscardQuiet:
+			q.logf("queue: ansage %s (%q) verworfen — ruhezeit %s aktiv (modus discard)",
+				item.ID, item.Text, q.laws.Quiet)
+		case verdictToast:
+			q.logf("queue: ansage %s (%q) durch toast ersetzt — ruhezeit %s aktiv (modus toast)",
+				item.ID, item.Text, q.laws.Quiet)
 		}
 	}
 }

-// popLocked entnimmt das als nächstes fällige Item (mu muss gehalten werden).
-func (q *Queue) popLocked() Item {
+// readyLocked meldet, ob der Worker arbeiten kann (mu muss gehalten
+// werden): geschlossen immer; sonst nur, wenn das beste Item nicht bloß
+// auf seinen Rate-Limit-Slot parkt. Verwürfe zählen als Arbeit — sie
+// laufen ohne Audio, aber mit Protokoll durch den Worker.
+func (q *Queue) readyLocked() bool {
+	if q.closed || len(q.items) == 0 {
+		return q.closed
+	}
+	return q.verdictLocked(q.items[q.bestIndexLocked()]) != verdictWait
+}
+
+// takeLocked entnimmt das als nächstes fällige Item, wenn es spielbar oder
+// entscheidbar ist (mu muss gehalten werden). Bei verdictWait wird NICHT
+// entnommen — das Item bleibt geparkt, bis der Slot frei ist (oder ein
+// besseres Item vorgezogen wird).
+func (q *Queue) takeLocked() (Item, verdict) {
+	best := q.bestIndexLocked()
+	item := q.items[best]
+	v := q.verdictLocked(item)
+	if v != verdictWait {
+		q.items = append(q.items[:best], q.items[best+1:]...)
+	}
+	return item, v
+}
+
+// bestIndexLocked liefert den Index des als nächstes fälligen Items
+// (mu muss gehalten werden).
+func (q *Queue) bestIndexLocked() int {
 	best := 0
 	for i := 1; i < len(q.items); i++ {
 		if itemBefore(q.items[i], q.items[best]) {
 			best = i
 		}
 	}
-	item := q.items[best]
-	q.items = append(q.items[:best], q.items[best+1:]...)
-	return item
+	return best
+}
+
+// verdictLocked wendet die Gesetze auf ein Item an (mu muss gehalten
+// werden — die Lock-Ordnung Queue → RateLimiter bleibt gewahrt).
+func (q *Queue) verdictLocked(item Item) verdict {
+	return q.laws.decide(item, time.Now(), q.limit.Allow())
+}
+
+// armWakeLocked armiert den Slot-Wecker auf until (mu muss gehalten
+// werden): Er broadcastet die Cond zum Termin, damit der geparkte Worker
+// auch ohne neuen Enqueue aufwacht. Ein bereits laufender Wecker mit
+// früherem (oder gleichem) Termin reicht; sonst wird er ersetzt.
+func (q *Queue) armWakeLocked(until time.Time) {
+	if q.wake != nil && !q.wakeAt.After(until) {
+		return
+	}
+	if q.wake != nil {
+		q.wake.Stop()
+	}
+	q.wakeAt = until
+	d := time.Until(until)
+	if d < 0 {
+		d = 0
+	}
+	q.wake = time.AfterFunc(d, func() {
+		q.mu.Lock()
+		q.wake = nil
+		q.mu.Unlock()
+		q.notEmpty.Broadcast()
+	})
 }

 // newID liefert eine kurze, kollisionsarme ID (6 Hex-Zeichen). IDs müssen