diff --git a/internal/queue/doc.go b/internal/queue/doc.go index 4dedffb..00d5969 100644 --- a/internal/queue/doc.go +++ b/internal/queue/doc.go @@ -1,7 +1,9 @@ // Package queue implementiert die Ansage-Queue des Daemons: serialisierte // Abarbeitung (eine Ansage gleichzeitig), Prioritäten (high > normal > low, // FIFO je Level), now-Items springen vor, dazu den persistenten Mute-State -// (Datei mute im Config-Dir) und die Rate-Limit-Struktur. Das Rate-Limit -// ist als Struktur vorhanden, aber noch nicht angebunden — Wiring und -// Ruhezeiten folgen in Task 05. +// (Datei mute im Config-Dir). Seit Task 05 unter dem Anti-Spam-Gesetz +// (rate.go/quiet.go): Rate-Limit — jede gestartete Ansage bucht das Fenster; +// low über dem Budget verfällt, normal parkt bis zum Slot, high darf mit +// Burst-Recht verstoßen — und Ruhezeiten, die still verwerfen oder durch +// einen Toast ersetzen. now:true bricht bewusst aus beiden aus. package queue diff --git a/internal/queue/queue.go b/internal/queue/queue.go index d77c91a..9072016 100644 --- a/internal/queue/queue.go +++ b/internal/queue/queue.go @@ -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 diff --git a/internal/queue/queue_test.go b/internal/queue/queue_test.go index 7e22e99..14794b3 100644 --- a/internal/queue/queue_test.go +++ b/internal/queue/queue_test.go @@ -67,17 +67,18 @@ func (g *gesteuerteSpielerin) keineWeitere(t *testing.T) { // TestAbarbeitungsReihenfolge: now vor high vor normal vor low, FIFO je // Level. Der erste Blocker besetzt den Worker; danach werden alle Level // gleichzeitig eingereiht — die Abarbeitung muss der Klasse folgen. +// (Ohne Gesetze: Null-Laws = unbegrenzt, keine Ruhezeit.) func TestAbarbeitungsReihenfolge(t *testing.T) { g := neueSpielerin() - q := New(g.speak, nil) + q := New(Laws{}, g.speak, nil) defer q.Close() - erster, err := q.Enqueue(Item{Text: "blocker", Priority: PriorityNormal}) + adm, err := q.Enqueue(Item{Text: "blocker", Priority: PriorityNormal}) if err != nil { t.Fatal(err) } - if erster == "" { - t.Fatal("Enqueue vergibt keine ID") + if adm.ID == "" || adm.Discarded != "" { + t.Fatalf("Enqueue = %+v, want Aufnahme mit ID", adm) } g.warteBis(t, 1) // Blocker läuft, Worker gebunden. @@ -111,7 +112,7 @@ func TestAbarbeitungsReihenfolge(t *testing.T) { // TestCurrentUndPending: Status-Sicht auf laufendes Item + Warteschlange. func TestCurrentUndPending(t *testing.T) { g := neueSpielerin() - q := New(g.speak, nil) + q := New(Laws{}, g.speak, nil) defer q.Close() if _, ok := q.Current(); ok { @@ -144,7 +145,7 @@ func TestCurrentUndPending(t *testing.T) { // spielbereit. func TestCancel(t *testing.T) { g := neueSpielerin() - q := New(g.speak, nil) + q := New(Laws{}, g.speak, nil) defer q.Close() q.Enqueue(Item{Text: "blocker"}) @@ -165,13 +166,13 @@ func TestCancel(t *testing.T) { } // Queue nimmt wieder an und spielt: - id, err := q.Enqueue(Item{Text: "danach"}) + adm, err := q.Enqueue(Item{Text: "danach"}) if err != nil { t.Fatal(err) } g.step <- struct{}{} // gibt die wartende Ansage im Speaker frei gestartet := g.warteBis(t, 2) - if len(gestartet) != 1 || gestartet[0].ID != id { + if len(gestartet) != 1 || gestartet[0].ID != adm.ID { t.Errorf("neue Ansage nicht gestartet: %+v", gestartet) } } @@ -181,7 +182,7 @@ func TestCancel(t *testing.T) { // Enqueue ein Fehler (und Close idempotent). func TestClose(t *testing.T) { g := neueSpielerin() - q := New(g.speak, nil) + q := New(Laws{}, g.speak, nil) q.Enqueue(Item{Text: "blocker"}) g.warteBis(t, 1) @@ -202,19 +203,19 @@ func TestClose(t *testing.T) { // TestEnqueueVergibtIDs: IDs sind kurz und eindeutig. func TestEnqueueVergibtIDs(t *testing.T) { g := neueSpielerin() - q := New(g.speak, nil) + q := New(Laws{}, g.speak, nil) defer q.Close() seen := map[string]bool{} for i := 0; i < 20; i++ { - id, err := q.Enqueue(Item{Text: "x"}) + adm, err := q.Enqueue(Item{Text: "x"}) if err != nil { t.Fatal(err) } - if len(id) < 4 || seen[id] { - t.Errorf("ID %q zu kurz oder doppelt", id) + if len(adm.ID) < 4 || seen[adm.ID] { + t.Errorf("ID %q zu kurz oder doppelt", adm.ID) } - seen[id] = true + seen[adm.ID] = true } } @@ -279,7 +280,7 @@ func TestMuteStore(t *testing.T) { } } -// TestRateLimiter: Struktur-Semantik (Task 05 verdrahtet sie in den Loop). +// TestRateLimiter: Struktur-Semantik (seit Task 05 in den Loop verwired). func TestRateLimiter(t *testing.T) { // Unbegrenzt (Task-04-Default): Interval 0 erlaubt immer. unlimited := &RateLimiter{} @@ -307,3 +308,210 @@ func TestRateLimiter(t *testing.T) { t.Fatal("nach Interval: Allow = false, want true") } } + +// fensterUm baut ein "HH:MM-HH:MM"-Ruhefenster, das t sicher einschließt +// (eine Minute davor bis eine Minute danach; Minute-Auflösung reicht). +func fensterUm(t time.Time) string { + return t.Add(-time.Minute).Format("15:04") + "-" + t.Add(time.Minute).Format("15:04") +} + +// warteAuf verifies mit Pollen eine Bedingung (Zeit-Gesetze sind real, der +// gesteuerte Speaker allein sieht sie nicht). +func warteAuf(t *testing.T, timeout time.Duration, cond func() bool, was string) { + t.Helper() + deadline := time.Now().Add(timeout) + for !cond() { + if time.Now().After(deadline) { + t.Fatalf("%s innerhalb %s nicht eingetreten", was, timeout) + } + time.Sleep(5 * time.Millisecond) + } +} + +// TestRateLimitLowVerworfenNormalParkt: mit kurzem Fenster spielt die +// erste Ansage sofort; low über dem Budget wird schon beim Eintritt +// verworfen (Admission nennt den Grund), normal parkt und spielt nach +// Fenster-Ablauf. +func TestRateLimitLowVerworfenNormalParkt(t *testing.T) { + g := neueSpielerin() + q := New(Laws{RateLimit: 400 * time.Millisecond}, g.speak, nil) + defer q.Close() + + if adm, err := q.Enqueue(Item{Text: "erste"}); err != nil || adm.ID == "" || adm.Discarded != "" { + t.Fatalf("erste Enqueue = %+v (%v), want Aufnahme", adm, err) + } + g.warteBis(t, 1) // erste läuft — Fenster ist gebucht. + + adm, err := q.Enqueue(Item{Text: "low-eins", Priority: PriorityLow}) + if err != nil { + t.Fatal(err) + } + if adm.ID != "" || adm.Discarded != DiscardRateLimit || adm.Toast { + t.Errorf("low über Budget = %+v, want Verwurf %q", adm, DiscardRateLimit) + } + if adm, err = q.Enqueue(Item{Text: "low-zwei", Priority: PriorityLow}); err != nil || adm.Discarded != DiscardRateLimit { + t.Errorf("zweiter low-Verwurf = %+v (%v)", adm, err) + } + + adm, err = q.Enqueue(Item{Text: "zweite"}) + if err != nil || adm.ID == "" || adm.Discarded != "" { + t.Fatalf("normal = %+v (%v), want Aufnahme (parkt bis zum Slot)", adm, err) + } + + // Status-Sicht während der erste läuft: zweite wartet (normal, geparkt). + warteAuf(t, time.Second, func() bool { + for _, it := range q.Pending() { + if it.Text == "zweite" { + return true + } + } + return false + }, "geparkte normale Ansage in Pending") + + close(g.step) // erste beenden — zweite muss weiter parken … + g.keineWeitere(t) + // … und nach Fenster-Ablauf starten (Deadline deckt Jitter locker). + gestartet := g.warteBis(t, 2) + if len(gestartet) != 1 || gestartet[0].Text != "zweite" { + t.Errorf("nach Fenster-Ablauf = %+v, will zweite", gestartet) + } + // low-eins/low-zwei spielen nie: genau 2 Starts insgesamt. + g.keineWeitere(t) +} + +// TestHighBurstEinUndAus: mit Burst-Recht startet high sofort nach der +// laufenden Ansage, obwohl das Fenster voll ist; ohne Recht parkt high +// wie normal (kein Verstoß). +func TestHighBurstEinUndAus(t *testing.T) { + g := neueSpielerin() + q := New(Laws{RateLimit: 10 * time.Second, AllowHighBurst: true}, g.speak, nil) + + q.Enqueue(Item{Text: "erste"}) + g.warteBis(t, 1) + if adm, err := q.Enqueue(Item{Text: "dringend", Priority: PriorityHigh}); err != nil || adm.ID == "" { + t.Fatalf("high mit Burst-Recht = %+v (%v), want Aufnahme", adm, err) + } + g.step <- struct{}{} // erste beenden + gestartet := g.warteBis(t, 2) // high startet trotz vollem Fenster + if len(gestartet) != 1 || gestartet[0].Text != "dringend" { + t.Errorf("burst = %+v, will dringend sofort", gestartet) + } + q.Close() + + g2 := neueSpielerin() + q2 := New(Laws{RateLimit: 10 * time.Second, AllowHighBurst: false}, g2.speak, nil) + defer q2.Close() + + q2.Enqueue(Item{Text: "erste"}) + g2.warteBis(t, 1) // erste läuft — Fenster gebucht + if adm, err := q2.Enqueue(Item{Text: "dringend", Priority: PriorityHigh}); err != nil || adm.ID == "" { + t.Fatalf("high ohne Burst-Recht = %+v (%v), want Aufnahme (parkt)", adm, err) + } + g2.step <- struct{}{} // erste beenden + g2.keineWeitere(t) // high wartet den Slot ab — kein Verstoß ohne Recht +} + +// TestNowBypassesRateLimit: now startet trotz vollem Fensters sofort — +// der dokumentierte Ausbruch (er startet trotzdem ein neues Fenster). +func TestNowBypassesRateLimit(t *testing.T) { + g := neueSpielerin() + q := New(Laws{RateLimit: 10 * time.Second}, g.speak, nil) + defer q.Close() + + q.Enqueue(Item{Text: "erste"}) + g.warteBis(t, 1) // Fenster gebucht (10 s) + + if adm, err := q.Enqueue(Item{Text: "sofort", Now: true}); err != nil || adm.ID == "" || adm.Discarded != "" { + t.Fatalf("now über vollem Fenster = %+v (%v), want Aufnahme", adm, err) + } + g.step <- struct{}{} + gestartet := g.warteBis(t, 2) // lange vor Fenster-Ablauf (10 s) — sonst Timeout + if len(gestartet) != 1 || gestartet[0].Text != "sofort" { + t.Errorf("now = %+v, will sofort", gestartet) + } + + // Der now-Bruch hat ein neues Fenster gebucht: normal parkt jetzt. + if adm, err := q.Enqueue(Item{Text: "danach"}); err != nil || adm.ID == "" { + t.Fatalf("normal nach now = %+v (%v)", adm, err) + } + close(g.step) + g.keineWeitere(t) // parkt bis zum Slot — nicht sofort +} + +// TestRuhezeitDiscardUndToast: aktives Fenster verwirft normal und high +// (Burst ist ein Rate-Recht, kein Ruhe-Recht) — toast ersetzt stattdessen; +// now bricht bewusst aus. +func TestRuhezeitDiscardUndToast(t *testing.T) { + fenster := fensterUm(time.Now()) + + g := neueSpielerin() + q := New(Laws{Quiet: mustQuiet(t, fenster), QuietMode: QuietModeDiscard}, g.speak, nil) + defer q.Close() + + adm, err := q.Enqueue(Item{Text: "normal"}) + if err != nil || adm.ID != "" || adm.Discarded != DiscardQuietHours || adm.Toast { + t.Errorf("normal in Ruhezeit = %+v (%v), want stiller Verwurf", adm, err) + } + if adm, err = q.Enqueue(Item{Text: "dringend", Priority: PriorityHigh}); err != nil || adm.Discarded != DiscardQuietHours { + t.Errorf("high in Ruhezeit = %+v (%v), want auch Verwurf", adm, err) + } + adm, err = q.Enqueue(Item{Text: "sofort", Now: true}) + if err != nil || adm.ID == "" || adm.Discarded != "" { + t.Fatalf("now in Ruhezeit = %+v (%v), want Ausbruch", adm, err) + } + gestartet := g.warteBis(t, 1) + if len(gestartet) != 1 || gestartet[0].Text != "sofort" { + t.Errorf("in Ruhezeit = %+v, will nur sofort", gestartet) + } + g.keineWeitere(t) + + g2 := neueSpielerin() + q2 := New(Laws{Quiet: mustQuiet(t, fenster), QuietMode: QuietModeToast}, g2.speak, nil) + defer q2.Close() + + adm, err = q2.Enqueue(Item{Text: "normal"}) + if err != nil || adm.Discarded != DiscardQuietHours || !adm.Toast { + t.Errorf("normal in Ruhezeit (toast) = %+v (%v), want Toast-Ersetzung", adm, err) + } + g2.keineWeitere(t) +} + +// mustQuiet parst ein Ruhefenster oder bricht den Test ab. +func mustQuiet(t *testing.T, fenster string) QuietHours { + t.Helper() + q, err := ParseQuietHours(fenster) + if err != nil { + t.Fatal(err) + } + return q +} + +// TestLawState: unbegrenzt liefert freien Slot; nach gebuchtem Fenster +// liegt der nächste Slot in der Zukunft; Ruhezeit-Fenster ist sichtbar. +func TestLawState(t *testing.T) { + g := neueSpielerin() + q := New(Laws{}, g.speak, nil) + st := q.LawState() + if !st.NextSlot.IsZero() || st.RateLimit != 0 { + t.Errorf("unbegrenzt = %+v, want Nullzeit-Slot", st) + } + q.Close() + + fenster := mustQuiet(t, "18:00-08:00") + q2 := New(Laws{RateLimit: time.Hour, AllowHighBurst: true, Quiet: fenster, QuietMode: QuietModeToast}, g.speak, nil) + defer q2.Close() + st2 := q2.LawState() + if st2.RateLimit != time.Hour || !st2.AllowHighBurst { + t.Errorf("LawState = %+v, want 1h-Fenster mit Burst-Recht", st2) + } + if st2.Quiet.String() != "18:00-08:00" || st2.QuietMode != QuietModeToast { + t.Errorf("Ruhezeit in LawState = %q/%q", st2.Quiet, st2.QuietMode) + } + + q2.Enqueue(Item{Text: "erste"}) + g.warteBis(t, 1) + if next := q2.LawState().NextSlot; !next.After(time.Now()) { + t.Errorf("nach Start = %v, want Slot in der Zukunft", next) + } + g.step <- struct{}{} +} diff --git a/internal/queue/quiet.go b/internal/queue/quiet.go new file mode 100644 index 0000000..1a19887 --- /dev/null +++ b/internal/queue/quiet.go @@ -0,0 +1,109 @@ +package queue + +import ( + "fmt" + "strconv" + "strings" + "time" +) + +// Erlaubte Werte für quiet_mode (öffentlicher Config-Wert). +const ( + QuietModeDiscard = "discard" + QuietModeToast = "toast" +) + +// clockTime ist eine Tageszeit in Minuten-Auflösung (lokale Zeit, 24 h). +type clockTime struct{ hour, minute int } + +// minutes liefert Minuten seit Mitternacht. +func (c clockTime) minutes() int { return c.hour*60 + c.minute } + +// String rendert die Uhrzeit als "HH:MM". +func (c clockTime) String() string { return fmt.Sprintf("%02d:%02d", c.hour, c.minute) } + +// QuietHours ist ein Ruhezeit-Fenster in lokaler Zeit, z. B. "18:00-08:00" +// (überspannt Mitternacht). Der Nullwert (Start == End) ist deaktiviert — +// rund-um-die-Uhr-Stumm ist mute, nicht quiet_hours. +type QuietHours struct { + Start, End clockTime +} + +// String rendert das Fenster als "HH:MM-HH:MM" (Nullwert: ""). +func (q QuietHours) String() string { + if q.Start == q.End { + return "" + } + return q.Start.String() + "-" + q.End.String() +} + +// ParseQuietHours parst ein Ruhezeit-Fenster "HH:MM-HH:MM" (lokale Zeit, +// 24 h; Stunden ein- oder zweistellig). Fenster über Mitternacht +// (Start > End) sind erlaubt; Start == End deaktiviert das Fenster. Der +// Leerstring ist das deaktivierte Fenster (kein Fehler). +func ParseQuietHours(s string) (QuietHours, error) { + if s == "" { + return QuietHours{}, nil + } + startS, endS, ok := strings.Cut(s, "-") + if !ok { + return QuietHours{}, fmt.Errorf("quiet_hours: %q ist nicht HH:MM-HH:MM", s) + } + start, err := parseClock(startS) + if err != nil { + return QuietHours{}, fmt.Errorf("quiet_hours: %v", err) + } + end, err := parseClock(endS) + if err != nil { + return QuietHours{}, fmt.Errorf("quiet_hours: %v", err) + } + return QuietHours{Start: start, End: end}, nil +} + +// parseClock parst "H[H]:MM" in Uhrzeit-Bestandteile. +func parseClock(s string) (clockTime, error) { + hourS, minuteS, ok := strings.Cut(s, ":") + if !ok || len(hourS) < 1 || len(hourS) > 2 || len(minuteS) != 2 { + return clockTime{}, fmt.Errorf("%q ist nicht HH:MM", s) + } + hour, err := strconv.Atoi(hourS) + if err != nil { + return clockTime{}, fmt.Errorf("%q ist nicht HH:MM", s) + } + minute, err := strconv.Atoi(minuteS) + if err != nil { + return clockTime{}, fmt.Errorf("%q ist nicht HH:MM", s) + } + if hour < 0 || hour > 23 || minute < 0 || minute > 59 { + return clockTime{}, fmt.Errorf("%q ist keine gültige Uhrzeit (24 h)", s) + } + return clockTime{hour: hour, minute: minute}, nil +} + +// Active meldet, ob der lokale Zeitpunkt t im Ruhefenster liegt. +// Tagfenster (Start < End) gelten ab Start bis ausschließlich End; +// Nachtfenster (Start > End) überspannen Mitternacht. +func (q QuietHours) Active(t time.Time) bool { + if q.Start == q.End { + return false + } + m := t.Hour()*60 + t.Minute() + s, e := q.Start.minutes(), q.End.minutes() + if s < e { + return m >= s && m < e + } + return m >= s || m < e +} + +// Until liefert das lokale Ende des gerade aktiven Fensters — Nullzeit, +// wenn gerade kein Ruhefenster aktiv ist. Basis für GET /status. +func (q QuietHours) Until(t time.Time) time.Time { + if !q.Active(t) { + return time.Time{} + } + end := time.Date(t.Year(), t.Month(), t.Day(), q.End.hour, q.End.minute, 0, 0, t.Location()) + if q.Start.minutes() > q.End.minutes() && t.Hour()*60+t.Minute() >= q.Start.minutes() { + end = end.AddDate(0, 0, 1) // Nachtfenster: das Ende ist am Folgetag + } + return end +} diff --git a/internal/queue/quiet_test.go b/internal/queue/quiet_test.go new file mode 100644 index 0000000..54944c6 --- /dev/null +++ b/internal/queue/quiet_test.go @@ -0,0 +1,131 @@ +package queue + +import ( + "testing" + "time" +) + +// lokal baut eine lokale Test-Zeit (Stunde/Minute reichen — Ruhefenster +// arbeiten auf Minuten-Auflösung). +func lokal(hour, minute int) time.Time { + return time.Date(2026, 10, 18, hour, minute, 0, 0, time.Local) +} + +// TestParseQuietHours: Format HH:MM-HH:MM (überspannt Mitternacht), +// klare Fehler für Müll. +func TestParseQuietHours(t *testing.T) { + cases := []struct { + in string + start clockTime + end clockTime + bad bool + }{ + {in: "18:00-08:00", start: clockTime{18, 0}, end: clockTime{8, 0}}, + {in: "9:30-17:45", start: clockTime{9, 30}, end: clockTime{17, 45}}, + {in: "22:00-06:00", start: clockTime{22, 0}, end: clockTime{6, 0}}, + {in: "08:00-08:00", start: clockTime{8, 0}, end: clockTime{8, 0}}, // deaktiviert, aber gültig + {in: ""}, // Leerstring = deaktiviert, kein Fehler + {in: "18:00", bad: true}, + {in: "18:00-", bad: true}, + {in: "-08:00", bad: true}, + {in: "24:00-08:00", bad: true}, + {in: "18:60-08:00", bad: true}, + {in: "1800-0800", bad: true}, + {in: "ab:00-08:00", bad: true}, + } + for _, tc := range cases { + q, err := ParseQuietHours(tc.in) + if tc.bad { + if err == nil { + t.Errorf("ParseQuietHours(%q) = %+v, want Fehler", tc.in, q) + } + continue + } + if err != nil { + t.Errorf("ParseQuietHours(%q): unerwarteter Fehler %v", tc.in, err) + continue + } + if q.Start != tc.start || q.End != tc.end { + t.Errorf("ParseQuietHours(%q) = %v-%v, want %v-%v", tc.in, q.Start, q.End, tc.start, tc.end) + } + } +} + +// TestQuietHoursActive: Tagfenster, Nachtfenster über Mitternacht und der +// deaktivierte Nullwert über alle Grenzen (Start inklusiv, End exklusiv). +func TestQuietHoursActive(t *testing.T) { + cases := []struct { + fenster string + hour, minute int + want bool + }{ + // Tagfenster 09:00-17:00 + {"09:00-17:00", 8, 59, false}, + {"09:00-17:00", 9, 0, true}, // Start inklusiv + {"09:00-17:00", 12, 30, true}, + {"09:00-17:00", 16, 59, true}, + {"09:00-17:00", 17, 0, false}, // End exklusiv + {"09:00-17:00", 23, 0, false}, + // Nachtfenster 18:00-08:00 (überspannt Mitternacht) + {"18:00-08:00", 17, 59, false}, + {"18:00-08:00", 18, 0, true}, + {"18:00-08:00", 23, 59, true}, + {"18:00-08:00", 0, 0, true}, // über Mitternacht + {"18:00-08:00", 7, 59, true}, + {"18:00-08:00", 8, 0, false}, // End exklusiv + {"18:00-08:00", 12, 0, false}, + // Zweites Nachtfenster + {"22:00-06:00", 3, 0, true}, + {"22:00-06:00", 21, 59, false}, + // Start == End: deaktiviert + {"08:00-08:00", 8, 0, false}, + {"08:00-08:00", 20, 0, false}, + } + for _, tc := range cases { + q := mustQuiet(t, tc.fenster) + if got := q.Active(lokal(tc.hour, tc.minute)); got != tc.want { + t.Errorf("%s um %02d:%02d: Active = %v, want %v", tc.fenster, tc.hour, tc.minute, got, tc.want) + } + } +} + +// TestQuietHoursUntil: Fenster-Ende (heute bzw. morgen bei Nachtfenstern), +// Nullzeit wenn inaktiv. +func TestQuietHoursUntil(t *testing.T) { + // Tagfenster: Ende heute. + q := mustQuiet(t, "09:00-17:00") + if want := lokal(17, 0); !q.Until(lokal(12, 0)).Equal(want) { + t.Errorf("Until(12:00) = %v, want %v (heute 17:00)", q.Until(lokal(12, 0)), want) + } + if !q.Until(lokal(3, 0)).IsZero() { + t.Error("Until außerhalb des Fensters: want Nullzeit") + } + + // Nachtfenster: ab 18:00 ist das Ende morgen früh. + n := mustQuiet(t, "18:00-08:00") + if want := lokal(0, 0).Add(32 * time.Hour); !n.Until(lokal(20, 0)).Equal(want) { + t.Errorf("Until(20:00) = %v, want %v (morgen 08:00)", n.Until(lokal(20, 0)), want) + } + // Nach Mitternacht (00:30) ist das Ende heute 08:00. + if want := lokal(8, 0); !n.Until(lokal(0, 30)).Equal(want) { + t.Errorf("Until(00:30) = %v, want %v (heute 08:00)", n.Until(lokal(0, 30)), want) + } +} + +// TestQuietHoursStringRoundtrip: String liefert die konfigurierte Form +// zurück und parst identisch; der Nullwert rendert leer. +func TestQuietHoursStringRoundtrip(t *testing.T) { + for _, s := range []string{"18:00-08:00", "09:00-17:00", "22:00-06:00"} { + q := mustQuiet(t, s) + if q.String() != s { + t.Errorf("String() = %q, want %q", q.String(), s) + } + q2 := mustQuiet(t, q.String()) + if q2 != q { + t.Errorf("Roundtrip von %q: %+v ≠ %+v", s, q2, q) + } + } + if (QuietHours{}).String() != "" { + t.Error("Nullwert-String nicht leer") + } +} diff --git a/internal/queue/rate.go b/internal/queue/rate.go index ab947a6..d7a638c 100644 --- a/internal/queue/rate.go +++ b/internal/queue/rate.go @@ -5,19 +5,24 @@ import ( "time" ) -// RateLimiter ist die Struktur für das Daemon-Gesetz gegen Ansage-Spam -// (Spec: Default ca. 1 Ansage / 30 s, konfigurierbar). +// Verwurfs-Gründe im HTTP-Vertrag von POST /speak +// (202 {"id":"","discarded":"…"}). +const ( + DiscardRateLimit = "rate_limit" // priority=low über dem Fenster-Budget + DiscardQuietHours = "quiet_hours" // Ruhezeit aktiv (discard oder toast) +) + +// RateLimiter ist das Zeitfenster des Daemon-Gesetzes gegen Ansage-Spam: +// Spec-Default ca. 1 Ansage / 30 s (queue.rate_limit), 0 = unbegrenzt. +// Jede gestartete Ansage bucht das Fenster neu (Reserve) — auch eine per +// now oder high-burst ausgebrochene: Das Fenster zählt Abspiel-Starts, +// nicht deren Legitimität. // -// SEAM FÜR TASK 05: Struktur und Semantik stehen und sind getestet, sind -// aber noch NICHT an den Worker-Loop angebunden — Task 04 erzeugt keinen -// Limiter (unbegrenzt, so verlangt es der Task-04-Brief: „Default unendlich -// in T04-Tests, dokumentiert"). Task 05 verdrahtet den Limiter (config -// queue.rate_limit) in den Loop und ergänzt die prioritätsbewusste -// Entscheidung: low verwerfen, normal parken, high konfigurierbar verstoßen -// (queue.allow_high_burst) — dafür ist Next() bereits Teil der Struktur. +// Lock-Ordnung: RateLimiter.mu ist immer die innere Sperre — Aufrufer +// halten beim Allow/Reserve/Next die Queue-sperre (niemals umgekehrt). type RateLimiter struct { - // Interval ist der Mindestabstand zwischen zwei Ansage-Starts. - // 0 oder negativ = unbegrenzt (Task-04-Default). + // Interval ist der Mindestabstand zwischen zwei Ansage-Starten. + // 0 oder negativ = unbegrenzt. Interval time.Duration mu sync.Mutex @@ -34,7 +39,10 @@ func (l *RateLimiter) Allow() bool { return time.Since(l.last) >= l.Interval } -// Reserve bucht einen Ansage-Slot (direkt nach Allow aufrufen). +// Reserve bucht einen Ansage-Slot (direkt nach der Play-Entscheidung +// aufrufen, vor dem eigentlichen Abspielen — auch fehlgeschlagene Ansagen +// verbrauchen das Fenster: Ein kaputter Provider darf nicht zum Maschinengewehr +// werden). func (l *RateLimiter) Reserve() { if l == nil || l.Interval <= 0 { return @@ -44,14 +52,112 @@ func (l *RateLimiter) Reserve() { l.last = time.Now() } -// Next liefert, wann der nächste Ansage-Slot frei wird (Nullzeit, wenn -// Allow sofort true liefern würde) — Basis für das Parken wartender Items -// in Task 05. +// Next liefert, wann der nächste Ansage-Slot frei wird — Nullzeit, wenn +// Allow sofort true liefern würde (unbegrenzt oder Fenster abgelaufen). +// Basis für das Parken wartender Items und den Status-Endpoint. func (l *RateLimiter) Next() time.Time { if l == nil || l.Interval <= 0 { return time.Time{} } l.mu.Lock() defer l.mu.Unlock() + if l.last.IsZero() { + return time.Time{} + } return l.last.Add(l.Interval) } + +// Laws sind die Anti-Spam-Gesetze des Daemons (Task 05): Rate-Limit mit +// verkonfigurierbarem Verstoß-Recht für high, plus Ruhezeiten. Der Nullwert +// erlaubt alles (unbegrenzt, kein Ruhefenster) — der Daemon liefert die +// Gesetze aus der Config; Default siehe internal/config.Defaults. +type Laws struct { + // RateLimit ist der Mindestabstand zwischen zwei Ansage-Starten + // (0 = unbegrenzt). + RateLimit time.Duration + + // AllowHighBurst erlaubt priority=high, das Rate-Limit bewusst zu + // verstoßen (Default true). Ohne Burst-Recht wartet high wie normal. + // Die Ruhezeit gilt davon unberührt auch für high. + AllowHighBurst bool + + // Quiet ist das Ruhezeit-Fenster (lokale Zeit; Nullwert = deaktiviert). + Quiet QuietHours + + // QuietMode bestimmt das Verhalten in der Ruhezeit: "discard" verwirft + // still, "toast" ersetzt Audio durch eine Benachrichtigung — aktuell + // eine Protokoll-Zeile über den Notification-Pfad des Loggers; der + // echte OS-Toast folgt mit der Notification-Anbindung (dokumentierte + // Einschränkung, siehe Task-05-Report). + QuietMode string +} + +// verdict ist die Gesetzes-Entscheidung über ein fälliges Item. +type verdict int + +const ( + verdictPlay verdict = iota // abspielen + verdictWait // parken bis zum freien Slot + verdictDiscardRate // verwerfen: low über dem Budget + verdictDiscardQuiet // verwerfen: Ruhezeit (discard) + verdictToast // Toast statt Audio (Ruhezeit) +) + +// decide wendet die Gesetze auf ein fälliges Item an (rein, ohne Zustand): +// slotFree sagt, ob der Rate-Limit-Slot gerade frei ist. Rangordnung: +// +// 1. now:true bricht bewusst aus — überspringt Rate-Limit UND Ruhezeit +// (dokumentierter Ausbruch für Dringendes; startet trotzdem ein neues +// Rate-Fenster). +// 2. Ruhezeit: discard verwirft, toast ersetzt — gilt auch für high, +// denn Burst ist ein Rate-Recht, kein Ruhe-Recht. +// 3. Rate-Limit: freier Slot spielt; high spielt mit Burst-Recht; low +// wird verworfen; normal (und high ohne Recht) parken bis zum Slot. +func (l Laws) decide(item Item, now time.Time, slotFree bool) verdict { + if item.Now { + return verdictPlay + } + if l.Quiet.Active(now) { + if l.QuietMode == QuietModeToast { + return verdictToast + } + return verdictDiscardQuiet + } + if slotFree { + return verdictPlay + } + if item.Priority == PriorityHigh && l.AllowHighBurst { + return verdictPlay + } + if item.Priority == PriorityLow { + return verdictDiscardRate + } + return verdictWait +} + +// Admission ist das Ergebnis von Enqueue: Aufgenommene tragen ihre ID, +// verworfene nennen den Grund (die 202-Antwort von POST /speak führt ihn). +type Admission struct { + // ID der aufgenommenen Ansage (leer, wenn verworfen). + ID string + + // Discarded ist der Verwurfs-Grund (DiscardRateLimit bzw. + // DiscardQuietHours); leer, wenn die Ansage aufgenommen wurde. + Discarded string + + // Toast meldet, dass die Ansage in der Ruhezeit durch eine + // Benachrichtigung ersetzt wurde (quiet_mode = "toast"). + Toast bool +} + +// LawState ist die /status-Sicht auf die Anti-Spam-Gesetze: konfiguriertes +// Rate-Limit mit dem nächsten freien Slot sowie das Ruhezeit-Fenster. +type LawState struct { + RateLimit time.Duration + AllowHighBurst bool + NextSlot time.Time // Nullzeit = Slot frei (oder unbegrenzt) + Quiet QuietHours + QuietActive bool + QuietMode string + QuietUntil time.Time // Nullzeit = Ruhezeit inaktiv +}