package worker import ( "context" "errors" "fmt" "log/slog" "strings" "sync" "testing" "time" "git.vakhrushev.me/av/transcriber/internal/contract" "github.com/prometheus/client_golang/prometheus" ) // Проверки этого файла судят одну развилку воркера: пустой прогон против // отказа. Инвариант проекта — «NoopJobError не ошибка» — стоит ровно на ней, а // цена срабатывания отложенная: три воркера опрашивают базу раз в секунду, и // пустой прогон, принятый за отказ, даёт три записи в секунду и столько же // засчитанных сбоев, которых не было. // journalBuffer собирает журнал прогона. Пишут в него из горутины воркера, а // читает проверка — отсюда мьютекс. type journalBuffer struct { mu sync.Mutex text strings.Builder } func (b *journalBuffer) Write(p []byte) (int, error) { b.mu.Lock() defer b.mu.Unlock() return b.text.Write(p) } func (b *journalBuffer) String() string { b.mu.Lock() defer b.mu.Unlock() return b.text.String() } // runOnce прогоняет воркер ровно один раз и возвращает журнал этого прогона. // // Воркер останавливает **второй** прогон, а не первый: отменённый контекст // теперь и значит «нас остановили», и отмена изнутри первого шага сделала бы // его исход неотличимым от остановки — проверка судила бы не то, что заявляет. // Пауза между прогонами на время проверки укорочена до миллисекунды: ждать // секунду ради второго вызова незачем. func runOnce(t *testing.T, name string, work func(ctx context.Context) error) string { t.Helper() journal := &journalBuffer{} logger := slog.New(slog.NewTextHandler(journal, nil)) ctx, cancel := context.WithCancel(context.Background()) defer cancel() var once sync.Once done := make(chan struct{}) calls := 0 w := NewCallbackWorker(name, func(ctx context.Context) error { calls++ if calls > 1 { // Первый прогон уже рассужен: журнал написан, счётчик сдвинут. once.Do(func() { cancel() close(done) }) return &contract.NoopJobError{State: "stopping"} } return work(ctx) }, logger) w.interval = time.Millisecond finished := make(chan struct{}) go func() { w.Start(ctx) close(finished) }() select { case <-done: case <-time.After(5 * time.Second): t.Fatal("работа воркера не была вызвана") } select { case <-finished: case <-time.After(5 * time.Second): t.Fatal("воркер не остановился по отмене контекста") } return journal.String() } // runRecords оставляет от журнала только записи об исходе прогона. Жизненный // цикл самого воркера — старт и остановка — по конвенции идёт на INFO и к // прогону не относится; требование говорит о том, что воркер пишет про свой // прогон, а не о том, что он молчит вообще. func runRecords(journal string) string { var kept []string for _, line := range strings.Split(strings.TrimSpace(journal), "\n") { if line == "" { continue } if strings.Contains(line, "msg=\"Worker started\"") || strings.Contains(line, "msg=\"Worker received shutdown signal") { continue } kept = append(kept, line) } return strings.Join(kept, "\n") } // jobCount читает счётчик работы воркера из общего реестра процесса. Судит // реестр, а не переменную пакета: метка, потерянная в точке употребления, // переменную не ломает, а на странице метрик видна. func jobCount(t *testing.T, worker, errLabel string) float64 { t.Helper() families, err := prometheus.DefaultGatherer.Gather() if err != nil { t.Fatalf("не удалось собрать метрики: %v", err) } for _, mf := range families { if mf.GetName() != "transcriber_worker_job_count" { continue } for _, m := range mf.GetMetric() { var gotWorker, gotErr string for _, label := range m.GetLabel() { switch label.GetName() { case "name": gotWorker = label.GetValue() case "error": gotErr = label.GetValue() } } if gotWorker == worker && gotErr == errLabel { return m.GetCounter().GetValue() } } } return 0 } // Обёртка `%w` объявлена конвенцией проекта умолчанием, и до этой задачи первая // же обёртка на пути сломала бы распознавание молча. Оракул держит именно // обёрнутое значение: на голом признак узнавался и приведением типа, то есть // проверка прошла бы и на починенном, и на сломанном коде. func TestWrappedNoopIsNotAFailure(t *testing.T) { const name = "wrapped_noop_worker" before := jobCount(t, name, "false") beforeErr := jobCount(t, name, "true") journal := runOnce(t, name, func(context.Context) error { return fmt.Errorf("find and acquire job: %w", &contract.NoopJobError{State: "created"}) }) // Записи о старте и остановке воркера законны и к прогону не относятся — // проверяется отсутствие записи об исходе прогона. if got := runRecords(journal); got != "" { t.Errorf("пустой прогон попал в журнал: %q", got) } if got := jobCount(t, name, "false"); got != before { t.Errorf("счётчик успешных прогонов вырос на пустом прогоне: было %v, стало %v", before, got) } if got := jobCount(t, name, "true"); got != beforeErr { t.Errorf("пустой прогон засчитан отказом: было %v, стало %v", beforeErr, got) } } // Без этой проверки оракул был бы зелен и на коде, который не считает отказом // вообще ничего. func TestFailureIsLoggedAndCounted(t *testing.T) { const name = "failing_worker" before := jobCount(t, name, "true") journal := runOnce(t, name, func(context.Context) error { return errors.New("database is gone") }) if !strings.Contains(journal, "database is gone") { t.Errorf("отказ не виден владельцу: журнал %q", journal) } if got := jobCount(t, name, "true"); got != before+1 { t.Errorf("отказ не засчитан: было %v, стало %v", before, got) } } // Счёт успешных прогонов — знаменатель доли отказов. Реализация, снявшая его, // проходит обе проверки выше, а владелец теряет способность отличить «три // прогона в секунду, все отказали» от «три отказа среди тысячи прогонов». func TestSuccessIsCounted(t *testing.T) { const name = "successful_worker" before := jobCount(t, name, "false") journal := runOnce(t, name, func(context.Context) error { return nil }) if got := jobCount(t, name, "false"); got != before+1 { t.Errorf("успешный прогон не засчитан: было %v, стало %v", before, got) } if strings.Contains(journal, "Worker error") { t.Errorf("успешный прогон записан отказом: журнал %q", journal) } } // Остановка сервиса — не отказ шага: контекст отменили мы сами. Без этой // развилки каждая выкладка красит журнал владельца отказами и накручивает // счётчик сбоев, которых не было, — тот же довод, по которому не считается // `NoopJobError`. Судит контекст, а не текст ошибки: убитый по контексту // процесс отдаёт «signal: killed», и `errors.Is` его с отменой не свяжет. func TestShutdownIsNotAFailure(t *testing.T) { const name = "stopped_worker" beforeErr := jobCount(t, name, "true") beforeOk := jobCount(t, name, "false") journal := &journalBuffer{} logger := slog.New(slog.NewTextHandler(journal, nil)) ctx, cancel := context.WithCancel(context.Background()) defer cancel() w := NewCallbackWorker(name, func(context.Context) error { // Так выглядит шаг, которого застала остановка. cancel() return errors.New("ffmpeg conversion failed: signal: killed") }, logger) w.interval = time.Millisecond finished := make(chan struct{}) go func() { w.Start(ctx) close(finished) }() select { case <-finished: case <-time.After(5 * time.Second): t.Fatal("воркер не остановился по отмене контекста") } if got := journal.String(); strings.Contains(got, "Worker error") { t.Errorf("остановка записана отказом: журнал %q", got) } if got := jobCount(t, name, "true"); got != beforeErr { t.Errorf("остановка засчитана отказом: было %v, стало %v", beforeErr, got) } if got := jobCount(t, name, "false"); got != beforeOk { t.Errorf("остановка засчитана успешным прогоном: было %v, стало %v", beforeOk, got) } }