Files
talk/internal/queue/queue_test.go
T
m3ta-chironandClaude Fable 5.1 79314dfb0f fix(queue): fertige Ansagen nicht mehr als „abgebrochen" protokollieren
Der Worker rief cancel() auf seinem eigenen Kontext, BEVOR er ctx.Err()
prüfte — damit war jede normal beendete Ansage im Protokoll „abgebrochen",
auch bei Erfolg. Der Abbruch wird jetzt vor dem eigenen cancel() abgelesen.
Regressionstest: fertig heißt „fertig nach", ein echter Cancel heißt
„abgebrochen".

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
2026-10-24 09:50:08 +02:00

611 lines
19 KiB
Go

package queue
import (
"context"
"os"
"path/filepath"
"slices"
"strings"
"sync"
"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.
// (Ohne Gesetze: Null-Laws = unbegrenzt, keine Ruhezeit.)
func TestAbarbeitungsReihenfolge(t *testing.T) {
g := neueSpielerin()
q := New(Laws{}, g.speak, nil)
defer q.Close()
adm, err := q.Enqueue(Item{Text: "blocker", Priority: PriorityNormal})
if err != nil {
t.Fatal(err)
}
if adm.ID == "" || adm.Discarded != "" {
t.Fatalf("Enqueue = %+v, want Aufnahme mit ID", adm)
}
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(Laws{}, 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(Laws{}, 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:
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 != adm.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(Laws{}, 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(Laws{}, g.speak, nil)
defer q.Close()
seen := map[string]bool{}
for i := 0; i < 20; i++ {
adm, err := q.Enqueue(Item{Text: "x"})
if err != nil {
t.Fatal(err)
}
if len(adm.ID) < 4 || seen[adm.ID] {
t.Errorf("ID %q zu kurz oder doppelt", adm.ID)
}
seen[adm.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 (seit Task 05 in den Loop verwired).
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")
}
}
// 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()
// Das Ruhezeit-Fenster muss JETZT inaktiv sein, sonst würde die Ansage
// unten getoastet statt gesprochen — ein festes Fenster wäre eine
// Zeitbombe (rot zu bestimmten Tageszeiten). Deshalb dynamisch: eine
// Stunde, die zwei Stunden in der Zukunft beginnt.
window := inactiveQuietWindow(time.Now())
fenster := mustQuiet(t, window)
if fenster.Active(time.Now()) {
t.Fatalf("Testfenster %s ist aktiv — darf nicht sein", window)
}
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() != window || st2.QuietMode != QuietModeToast {
t.Errorf("Ruhezeit in LawState = %q/%q, want %q/toast", st2.Quiet, st2.QuietMode, window)
}
if st2.QuietActive || !st2.QuietUntil.IsZero() {
t.Errorf("Ruhezeit inaktiv erwartet, LawState meldet active=%v until=%v", st2.QuietActive, st2.QuietUntil)
}
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{}{}
}
// inactiveQuietWindow liefert ein "HH:MM-HH:MM"-Fenster von einer Stunde,
// das zwei Stunden nach now beginnt — es enthält now garantiert nicht, egal
// zu welcher Tageszeit der Test läuft (auch über Mitternacht hinweg).
func inactiveQuietWindow(now time.Time) string {
start := now.Add(2 * time.Hour)
end := now.Add(3 * time.Hour)
return start.Format("15:04") + "-" + end.Format("15:04")
}
// TestInactiveQuietWindow: Das Test-Hilfsfenster darf zu KEINER Tageszeit
// aktiv sein — sonst wäre TestLawState wieder eine Zeitbombe.
func TestInactiveQuietWindow(t *testing.T) {
base := time.Date(2026, 10, 24, 0, 0, 0, 0, time.Local)
for m := 0; m < 24*60; m += 7 {
now := base.Add(time.Duration(m) * time.Minute)
w := inactiveQuietWindow(now)
q, err := ParseQuietHours(w)
if err != nil {
t.Fatalf("%s: %v", now.Format("15:04"), err)
}
if q.Active(now) {
t.Errorf("um %s ist Fenster %s aktiv", now.Format("15:04"), w)
}
}
}
// syncLog ist ein nebenläufigkeitssicherer Log-Puffer für Worker-Zeilen.
type syncLog struct {
mu sync.Mutex
b strings.Builder
}
func (l *syncLog) Write(p []byte) (int, error) {
l.mu.Lock()
defer l.mu.Unlock()
return l.b.Write(p)
}
func (l *syncLog) String() string {
l.mu.Lock()
defer l.mu.Unlock()
return l.b.String()
}
// TestProtokollFertigNichtAbgebrochen: Eine normal beendete Ansage wird als
// „fertig" protokolliert, nicht als „abgebrochen" — der Worker darf seinen
// eigenen cancel() nicht als Abbruch deuten. Ein echter Cancel heißt
// dagegen „abgebrochen".
func TestProtokollFertigNichtAbgebrochen(t *testing.T) {
logb := &syncLog{}
g := neueSpielerin()
q := New(Laws{}, g.speak, logb)
defer q.Close()
q.Enqueue(Item{Text: "normal zu ende"})
g.warteBis(t, 1)
g.step <- struct{}{}
waitFor(t, func() bool { return strings.Contains(logb.String(), "fertig nach") })
if strings.Contains(logb.String(), "abgebrochen") {
t.Errorf("fertige Ansage als abgebrochen protokolliert:\n%s", logb.String())
}
q.Enqueue(Item{Text: "wird abgebrochen"})
g.warteBis(t, 2)
q.Cancel()
waitFor(t, func() bool { return strings.Contains(logb.String(), "abgebrochen") })
}
// waitFor pollt cond bis zu zwei Sekunden.
func waitFor(t *testing.T, cond func() bool) {
t.Helper()
deadline := time.Now().Add(2 * time.Second)
for !cond() {
if time.Now().After(deadline) {
t.Fatal("Bedingung nicht binnen 2 s erfüllt")
}
time.Sleep(5 * time.Millisecond)
}
}