From e492baddac53fe875fef9ea96626774af7d036f9 Mon Sep 17 00:00:00 2001 From: m3ta-chiron Date: Sun, 18 Oct 2026 17:10:26 +0200 Subject: [PATCH] =?UTF-8?q?feat(queue):=20serialisierte=20Ansage-Queue=20m?= =?UTF-8?q?it=20Priorit=C3=A4ten,=20now-Sprung,=20Mute-Store=20und=20Rate-?= =?UTF-8?q?Limit-Struktur?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Worker-Loop: eine Ansage gleichzeitig; now > high > normal > low, FIFO je Level - Cancel bricht die laufende Ansage ab (ctx) und leert die Queue (Zahl als Rückgabe) - Close verwirft wartende Items, bricht die laufende Ansage ab und wartet auf den Worker - MuteStore: persistenter Mute-State als Datei mute im Config-Dir (atomar geschrieben) - RateLimiter: Struktur + Semantik für Task 05, bewusst noch nicht angebunden (Task-04-Default unbegrenzt, dokumentierte Seam in rate.go) --- internal/queue/doc.go | 7 +- internal/queue/mute.go | 84 ++++++++++ internal/queue/queue.go | 288 ++++++++++++++++++++++++++++++++ internal/queue/queue_test.go | 309 +++++++++++++++++++++++++++++++++++ internal/queue/rate.go | 57 +++++++ 5 files changed, 742 insertions(+), 3 deletions(-) create mode 100644 internal/queue/mute.go create mode 100644 internal/queue/queue.go create mode 100644 internal/queue/queue_test.go create mode 100644 internal/queue/rate.go diff --git a/internal/queue/doc.go b/internal/queue/doc.go index 5b75367..4dedffb 100644 --- a/internal/queue/doc.go +++ b/internal/queue/doc.go @@ -1,6 +1,7 @@ // 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, Rate-Limit mit prioritätsbewusstem -// Verwerfen. Implementierung folgt in Task 04 (Queue) und Task 05 -// (Rate-Limit, Ruhezeiten). +// 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. package queue diff --git a/internal/queue/mute.go b/internal/queue/mute.go new file mode 100644 index 0000000..098de56 --- /dev/null +++ b/internal/queue/mute.go @@ -0,0 +1,84 @@ +package queue + +import ( + "errors" + "os" + "path/filepath" + "strings" +) + +// MuteFileName ist der Name der Mute-Datei im Config-Verzeichnis (bindender +// Vertrag: „persistenten Mute-State toggeln (Datei mute im Config-Dir)"). +const MuteFileName = "mute" + +// ErrNoMutePath meldet einen Set-Versuch ohne konfigurierten Datei-Pfad +// (Config-Dir nicht bestimmbar — der Daemon deaktiviert dann Mute mit einer +// Start-Warnung statt zu scheitern). +var ErrNoMutePath = errors.New("mute: kein Datei-Pfad konfiguriert (Config-Dir nicht bestimmbar)") + +// MuteStore verwaltet den persistenten Mute-State als Datei im +// Config-Verzeichnis: Inhalt „true" = stumm, alles andere (oder keine Datei) +// = nicht stumm. +// +// Der Daemon liest die Datei bei jeder /speak-Anfrage neu — dadurch wirkt +// auch ein im Degraded Mode (Daemon tot) von der CLI direkt geschriebener +// State sofort, und der Zustand überlebt Daemon-Restarts. +type MuteStore struct { + // Path ist der vollständige Pfad der Mute-Datei. Leer (kein Config-Dir + // bestimmbar): Muted() ist immer false, Set ein Fehler. + Path string +} + +// Muted meldet den persistenten Mute-State. Eine fehlende oder unlesbare +// Datei heißt: nicht stumm (Fehlstart ohne Config-Dir wird nie „stumm aus +// Versehen"). +func (s MuteStore) Muted() bool { + if s.Path == "" { + return false + } + b, err := os.ReadFile(s.Path) + if err != nil { + return false + } + return strings.TrimSpace(string(b)) == "true" +} + +// Set schreibt den Mute-State atomar (temporäre Datei + Rename, damit +// Daemon und CLI nie eine halb geschriebene Datei lesen) und legt das +// Config-Verzeichnis bei Bedarf an. +func (s MuteStore) Set(muted bool) error { + if s.Path == "" { + return ErrNoMutePath + } + dir := filepath.Dir(s.Path) + if err := os.MkdirAll(dir, 0o755); err != nil { + return err + } + content := "false\n" + if muted { + content = "true\n" + } + tmp, err := os.CreateTemp(dir, ".mute-*") + if err != nil { + return err + } + name := tmp.Name() + if _, err := tmp.WriteString(content); err != nil { + tmp.Close() + os.Remove(name) + return err + } + if err := tmp.Close(); err != nil { + os.Remove(name) + return err + } + if err := os.Chmod(name, 0o644); err != nil { + os.Remove(name) + return err + } + if err := os.Rename(name, s.Path); err != nil { + os.Remove(name) + return err + } + return nil +} diff --git a/internal/queue/queue.go b/internal/queue/queue.go new file mode 100644 index 0000000..d77c91a --- /dev/null +++ b/internal/queue/queue.go @@ -0,0 +1,288 @@ +package queue + +import ( + "context" + "crypto/rand" + "encoding/hex" + "errors" + "fmt" + "io" + "sort" + "sync" + "time" +) + +// Priority eines Queue-Items — öffentlicher Vertrag der HTTP-API +// (low | normal | high; leer wird als normal behandelt). +type Priority string + +// Erlaubte Prioritäts-Werte (Spec: high > normal > low, FIFO je Level). +const ( + PriorityLow Priority = "low" + PriorityNormal Priority = "normal" + PriorityHigh Priority = "high" +) + +// Valid meldet, ob p ein erlaubter Prioritäts-Wert ist. +func (p Priority) Valid() bool { + switch p { + case PriorityLow, PriorityNormal, PriorityHigh: + return true + default: + return false + } +} + +// ErrClosed meldet einen Enqueue-Versuch nach Close (Daemon fährt herunter). +var ErrClosed = errors.New("queue: bereits geschlossen") + +// Item ist eine Ansage in der Warteschlange. Alle Felder kommen aus der +// HTTP-API (POST /speak); ID und Ablauf-Verwaltung liegen bei der Queue. +type Item struct { + // ID ist die kurze Ansage-Kennung (Antwort von POST /speak; Grundlage + // des --wait-Pollings der CLI). Wird von Enqueue vergeben, wenn leer. + ID string + + // Text ist der Ansage-Text. + Text string + + // Lang ist die Sprach-Pin (de | en); leer = automatische Erkennung + // durch den Daemon (langdetect). + Lang string + + // Provider pinnt die Provider-Kette auf eine Stufe (--provider); leer + // = konfigurierte Default-Kette. + Provider string + + // Voice pinnt eine Voice-ID (--voice); leer = Voice-Mapping/Defaults. + Voice string + + // Priority steuert die Abarbeitungs-Reihenfolge (high > normal > low, + // FIFO je Level). + Priority Priority + + // Now lässt die Ansage vor alle wartenden Items springen (auch vor + // high) — sie unterbricht aber nie die laufende Ansage. + Now bool + + seq uint64 // Enqueue-Reihenfolge: FIFO-Tiebrecher innerhalb eines Levels +} + +// rank liefert die Abarbeitungs-Klasse: now über high über normal über low. +func (it Item) rank() int { + if it.Now { + return 3 + } + switch it.Priority { + case PriorityHigh: + return 2 + case PriorityNormal: + return 1 + default: + return 0 + } +} + +// Speaker spielt ein Item ab. Im Daemon: Provider-Kette bauen und Speak +// aufrufen (Synthese mit Fallback + Wiedergabe über den Audio-Sink — die +// Kette aus internal/provider wird unverändert wiederverwendet). ctx wird +// abgebrochen, wenn die Ansage abgebrochen (POST /cancel) oder der Daemon +// heruntergefahren wird; die Kette bricht ihre Netz- und Prozess-Arbeit +// dann ab. Der Rückgabe-Fehler wird protokolliert, nie an die API gereicht. +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). +// +// 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. +type Queue struct { + speak Speaker + log io.Writer + + mu sync.Mutex + notEmpty *sync.Cond // Signal: neues Item; Broadcast: Close + items []Item + current *Item + cancelCur context.CancelFunc + seq uint64 + closed bool + 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{})} + 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) { + q.mu.Lock() + if q.closed { + q.mu.Unlock() + return "", ErrClosed + } + if item.Priority == "" { + item.Priority = PriorityNormal + } + q.seq++ + item.seq = q.seq + if item.ID == "" { + item.ID = newID() + } + q.items = append(q.items, item) + q.mu.Unlock() + q.notEmpty.Signal() + return item.ID, nil +} + +// Pending liefert eine Momentaufnahme der wartenden Items in +// Abarbeitungs-Reihenfolge (bestes Item zuerst) — Basis für GET /status. +func (q *Queue) Pending() []Item { + q.mu.Lock() + defer q.mu.Unlock() + out := make([]Item, len(q.items)) + copy(out, q.items) + sort.SliceStable(out, func(i, j int) bool { return itemBefore(out[i], out[j]) }) + return out +} + +// itemBefore meldet, ob a vor b abgearbeitet wird: höhere Klasse zuerst, +// innerhalb derselben Klasse die frühere Enqueue-Reihenfolge. +func itemBefore(a, b Item) bool { + if a.rank() != b.rank() { + return a.rank() > b.rank() + } + return a.seq < b.seq +} + +// Current liefert das gerade laufende Item (false: Queue ist idle). +func (q *Queue) Current() (Item, bool) { + q.mu.Lock() + defer q.mu.Unlock() + if q.current == nil { + return Item{}, false + } + return *q.current, true +} + +// Cancel bricht die laufende Ansage ab und leert die Queue. Rückgabe ist +// die Zahl der abgebrochenen Ansagen (laufende + wartende). +func (q *Queue) Cancel() int { + q.mu.Lock() + defer q.mu.Unlock() + n := len(q.items) + q.items = nil + if q.current != nil { + n++ + if q.cancelCur != nil { + q.cancelCur() + } + } + return n +} + +// Close fährt die Queue herunter: wartende Items werden verworfen, die +// laufende Ansage abgebrochen, der Worker-Loop beendet. Close wartet, bis +// der Loop sicher beendet ist (idempotent). +func (q *Queue) Close() { + q.mu.Lock() + if q.closed { + q.mu.Unlock() + <-q.done + return + } + q.closed = true + q.items = nil + if q.cancelCur != nil { + q.cancelCur() + } + 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. +func (q *Queue) loop() { + defer close(q.done) + for { + q.mu.Lock() + for len(q.items) == 0 && !q.closed { + 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 + 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)) + } + } +} + +// popLocked entnimmt das als nächstes fällige Item (mu muss gehalten werden). +func (q *Queue) popLocked() Item { + 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 +} + +// newID liefert eine kurze, kollisionsarme ID (6 Hex-Zeichen). IDs müssen +// nur pro Daemon-Lauf eindeutig sein (--wait-Polling); der Fallback ist +// praktisch unerreichbar, aber deterministisch sicher. +func newID() string { + var b [3]byte + if _, err := rand.Read(b[:]); err == nil { + return hex.EncodeToString(b[:]) + } + return fmt.Sprintf("id%d", time.Now().UnixNano()) +} + +// logf schreibt eine Protokoll-Zeile, wenn ein Log gesetzt ist. Wird nie +// unter dem Mutex gerufen (Aufrufer achten darauf). +func (q *Queue) logf(format string, args ...any) { + if q.log != nil { + fmt.Fprintf(q.log, format+"\n", args...) + } +} diff --git a/internal/queue/queue_test.go b/internal/queue/queue_test.go new file mode 100644 index 0000000..7e22e99 --- /dev/null +++ b/internal/queue/queue_test.go @@ -0,0 +1,309 @@ +package queue + +import ( + "context" + "os" + "path/filepath" + "slices" + "testing" + "time" +) + +// gesteuerteSpielerin ist ein testbarer Speaker: Jede Ansage meldet ihren +// Start auf started (Kanal-Puffer: der Worker blockiert nie) und wartet +// dann, bis der Test sie per step freigibt — so ist die Abarbeitung +// vollständig deterministisch steuerbar. Ein abgebrochener ctx beendet das +// Warten (Cancel-/Close-Pfade). Alle Felder werden nur vom Test gelesen; +// started-Reihenfolge == Start-Reihenfolge. +type gesteuerteSpielerin struct { + started chan Item + step chan struct{} + consumed int // von warteBis konsumierte Starts (nur Test-Goroutine) +} + +func neueSpielerin() *gesteuerteSpielerin { + return &gesteuerteSpielerin{started: make(chan Item, 16), step: make(chan struct{})} +} + +func (g *gesteuerteSpielerin) speak(ctx context.Context, item Item) error { + g.started <- item + select { + case <-g.step: + return nil + case <-ctx.Done(): + return ctx.Err() + } +} + +// warteBis wartet, bis insgesamt n Ansagen gestartet sind, und liefert die +// seit dem letzten warteBis-Aufruf neu gestarteten. +func (g *gesteuerteSpielerin) warteBis(t *testing.T, n int) []Item { + t.Helper() + var items []Item + deadline := time.After(2 * time.Second) + for g.consumed < n { + select { + case it := <-g.started: + g.consumed++ + items = append(items, it) + case <-deadline: + t.Fatalf("nur %d von %d Ansagen gestartet", g.consumed, n) + } + } + return items +} + +// keineWeitere prüft, dass keine weitere Ansage startet (Poll-Fenster). +func (g *gesteuerteSpielerin) keineWeitere(t *testing.T) { + t.Helper() + select { + case it := <-g.started: + g.consumed++ + t.Fatalf("unerwartete Ansage %q gestartet", it.Text) + case <-time.After(150 * time.Millisecond): + } +} + +// 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. +func TestAbarbeitungsReihenfolge(t *testing.T) { + g := neueSpielerin() + q := New(g.speak, nil) + defer q.Close() + + erster, err := q.Enqueue(Item{Text: "blocker", Priority: PriorityNormal}) + if err != nil { + t.Fatal(err) + } + if erster == "" { + t.Fatal("Enqueue vergibt keine ID") + } + g.warteBis(t, 1) // Blocker läuft, Worker gebunden. + + for _, it := range []Item{ + {Text: "niedrig", Priority: PriorityLow}, + {Text: "zweit-niedrig", Priority: PriorityLow}, + {Text: "mittel", Priority: PriorityNormal}, + {Text: "dringend", Priority: PriorityHigh}, + {Text: "sofort", Now: true}, + } { + if _, err := q.Enqueue(it); err != nil { + t.Fatal(err) + } + } + g.keineWeitere(t) // Serialisierung: Worker spielt noch den Blocker. + + // Freigeben: alles Übrige läuft in Klassen-Reihenfolge durch. + close(g.step) + rest := g.warteBis(t, 6) + + var texts []string + for _, it := range rest { + texts = append(texts, it.Text) + } + want := []string{"sofort", "dringend", "mittel", "niedrig", "zweit-niedrig"} + if !slices.Equal(texts, want) { + t.Errorf("Abarbeitung nach dem Blocker = %v, want %v", texts, want) + } +} + +// TestCurrentUndPending: Status-Sicht auf laufendes Item + Warteschlange. +func TestCurrentUndPending(t *testing.T) { + g := neueSpielerin() + q := New(g.speak, nil) + defer q.Close() + + if _, ok := q.Current(); ok { + t.Fatal("Current vor erster Ansage: want idle") + } + + q.Enqueue(Item{Text: "blocker"}) + g.warteBis(t, 1) + + q.Enqueue(Item{Text: "hoch", Priority: PriorityHigh}) + q.Enqueue(Item{Text: "niedrig", Priority: PriorityLow}) + + if cur, ok := q.Current(); !ok || cur.Text != "blocker" { + t.Errorf("Current = %+v (ok=%v), will Blocker", cur, ok) + } + var pendingTexts []string + for _, it := range q.Pending() { + pendingTexts = append(pendingTexts, it.Text) + } + if !slices.Equal(pendingTexts, []string{"hoch", "niedrig"}) { + t.Errorf("Pending = %v, want [hoch niedrig]", pendingTexts) + } + + close(g.step) + g.warteBis(t, 3) +} + +// TestCancel: laufende Ansage wird über ctx abgebrochen, Queue geleert, +// Gesamtzahl zurückgemeldet; danach ist die Queue wieder aufnahme- und +// spielbereit. +func TestCancel(t *testing.T) { + g := neueSpielerin() + q := New(g.speak, nil) + defer q.Close() + + q.Enqueue(Item{Text: "blocker"}) + g.warteBis(t, 1) + + q.Enqueue(Item{Text: "wartet-eins"}) + q.Enqueue(Item{Text: "wartet-zwei"}) + + if n := q.Cancel(); n != 3 { + t.Errorf("Cancel = %d, want 3 (laufende + 2 wartende)", n) + } + + g.keineWeitere(t) // Blocker durch ctx beendet, wartende Items sind weg. + waitIdle(t, q) + + if n := q.Cancel(); n != 0 { + t.Errorf("Cancel auf leerer Queue = %d, want 0", n) + } + + // Queue nimmt wieder an und spielt: + id, 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 { + t.Errorf("neue Ansage nicht gestartet: %+v", gestartet) + } +} + +// TestClose: wartende Items werden verworfen, die laufende Ansage über den +// abgebrochenen ctx beendet, Close wartet auf den Worker-Exit, danach ist +// Enqueue ein Fehler (und Close idempotent). +func TestClose(t *testing.T) { + g := neueSpielerin() + q := New(g.speak, nil) + + q.Enqueue(Item{Text: "blocker"}) + g.warteBis(t, 1) + q.Enqueue(Item{Text: "wartet"}) + + q.Close() + + g.keineWeitere(t) + if _, ok := q.Current(); ok { + t.Error("Current nach Close: want idle") + } + if _, err := q.Enqueue(Item{Text: "zu spät"}); err == nil { + t.Error("Enqueue nach Close: want Fehler") + } + q.Close() // idempotent +} + +// TestEnqueueVergibtIDs: IDs sind kurz und eindeutig. +func TestEnqueueVergibtIDs(t *testing.T) { + g := neueSpielerin() + q := New(g.speak, nil) + defer q.Close() + + seen := map[string]bool{} + for i := 0; i < 20; i++ { + id, 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) + } + seen[id] = true + } +} + +// waitIdle pollt, bis kein Item mehr läuft (Cancel-/Close-Pfade geben das +// nach dem Worker-Exit synchron frei; das Pollen deckt die Übergangsphase +// ab, ohne auf Timing zu trauen). +func waitIdle(t *testing.T, q *Queue) { + t.Helper() + deadline := time.Now().Add(time.Second) + for { + if _, ok := q.Current(); !ok { + return + } + if time.Now().After(deadline) { + t.Fatal("Queue wird nicht idle") + } + time.Sleep(5 * time.Millisecond) + } +} + +// TestMuteStore: Datei-Roundtrip, Fehlend-Heißt-Nicht-Stumm, atomares Set. +func TestMuteStore(t *testing.T) { + dir := t.TempDir() + path := filepath.Join(dir, MuteFileName) + + var s MuteStore + if s.Muted() { + t.Error("MuteStore ohne Pfad: Muted() = true, want false") + } + if err := s.Set(true); err == nil { + t.Error("Set ohne Pfad: want Fehler") + } + + s = MuteStore{Path: path} + if s.Muted() { + t.Error("ohne Datei: Muted() = true, want false (nie still aus Versehen)") + } + + if err := s.Set(true); err != nil { + t.Fatal(err) + } + if !s.Muted() { + t.Error("nach Set(true): Muted() = false") + } + if b, _ := os.ReadFile(path); string(b) != "true\n" { + t.Errorf("Datei-Inhalt = %q, want %q", b, "true\n") + } + + if err := s.Set(false); err != nil { + t.Fatal(err) + } + if s.Muted() { + t.Error("nach Set(false): Muted() = true") + } + + // Müll in der Datei heißt: nicht stumm (nur „true" ist stumm). + if err := os.WriteFile(path, []byte("vielleicht\n"), 0o644); err != nil { + t.Fatal(err) + } + if s.Muted() { + t.Error("Datei-Inhalt „vielleicht“: Muted() = true, want false") + } +} + +// TestRateLimiter: Struktur-Semantik (Task 05 verdrahtet sie in den Loop). +func TestRateLimiter(t *testing.T) { + // Unbegrenzt (Task-04-Default): Interval 0 erlaubt immer. + unlimited := &RateLimiter{} + for i := 0; i < 3; i++ { + if !unlimited.Allow() { + t.Fatal("Interval 0: Allow = false, want true (unbegrenzt)") + } + unlimited.Reserve() + } + + // Begrenzt: nach Reserve ist erst nach Interval wieder Allow. + l := &RateLimiter{Interval: 30 * time.Millisecond} + if !l.Allow() { + t.Fatal("frisches Limit: Allow = false") + } + l.Reserve() + if l.Allow() { + t.Fatal("sofort nach Reserve: Allow = true, want false") + } + if next := l.Next(); time.Now().After(next) { + t.Errorf("Next() = %v liegt in der Vergangenheit", next) + } + time.Sleep(40 * time.Millisecond) + if !l.Allow() { + t.Fatal("nach Interval: Allow = false, want true") + } +} diff --git a/internal/queue/rate.go b/internal/queue/rate.go new file mode 100644 index 0000000..ab947a6 --- /dev/null +++ b/internal/queue/rate.go @@ -0,0 +1,57 @@ +package queue + +import ( + "sync" + "time" +) + +// RateLimiter ist die Struktur für das Daemon-Gesetz gegen Ansage-Spam +// (Spec: Default ca. 1 Ansage / 30 s, konfigurierbar). +// +// 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. +type RateLimiter struct { + // Interval ist der Mindestabstand zwischen zwei Ansage-Starts. + // 0 oder negativ = unbegrenzt (Task-04-Default). + Interval time.Duration + + mu sync.Mutex + last time.Time // Start der zuletzt per Reserve gebuchten Ansage +} + +// Allow meldet, ob sofort eine Ansage starten dürfte. +func (l *RateLimiter) Allow() bool { + if l == nil || l.Interval <= 0 { + return true + } + l.mu.Lock() + defer l.mu.Unlock() + return time.Since(l.last) >= l.Interval +} + +// Reserve bucht einen Ansage-Slot (direkt nach Allow aufrufen). +func (l *RateLimiter) Reserve() { + if l == nil || l.Interval <= 0 { + return + } + l.mu.Lock() + defer l.mu.Unlock() + 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. +func (l *RateLimiter) Next() time.Time { + if l == nil || l.Interval <= 0 { + return time.Time{} + } + l.mu.Lock() + defer l.mu.Unlock() + return l.last.Add(l.Interval) +}