6. Тестирование goroutines

                      Жизненный цикл горутины в рантайме Go

      +------------------+
      |      _Gidle      | (Аллокация структуры g в памяти)
      +--------+---------+
               | newproc (go func())
               v
      +------------------+     Планировщик (M/P)     +------------------+
      |    _Grunnable    | ------------------------> |     _Grunning    |
      +--------+---------+ <------------------------ +--------+---------+
               ^               sysmon / прерывание            |
               |                                              | gopark (канал, мьютекс)
               | goready                                      v
               |                                     +------------------+
               +------------------------------------ |     _Gwaiting    |
                                                     +------------------+
                                                              |
                                             Завершение работы|
                                                              v
      +------------------+      Очистка стека        +------------------+
      |   Пул gFree      | <------------------------ |      _Gdead      |
      +------------------+                           +------------------+
        (g готова к reuse)                             (Уменьшает NumGoroutine)

// АНТИПАТТЕРН: Хрупкая проверка утечки через глобальный счетчик
func TestWorker_BadLeakCheck(t *testing.T) {
    t.Parallel() // <-- ФАТАЛЬНАЯ ОШИБКА: Разрушение изоляции!
    startGoroutines := runtime.NumGoroutine()

    runWorker()

    // Считать, что разница точно равна нулю — опасная иллюзия
    require.Equal(t, startGoroutines, runtime.NumGoroutine()) 
}

package pool_test

import (
    "context"
    "runtime"
    "sync/atomic"
    "testing"
    "time"

    "github.com/stretchr/testify/require"
    "yourproject/internal/pool"
)

func TestWorkerPool_ConcurrencyLimit(t *testing.T) {
    // ВАЖНО: Запуск строго последовательный (без t.Parallel())
    // для обеспечения чистоты метрик runtime.NumGoroutine()

    const maxWorkers = 5
    const totalTasks = 50

    // Фиксируем исходное состояние горутин до старта пула
    baseGoroutines := runtime.NumGoroutine()

    p := pool.NewWorkerPool(maxWorkers)
    p.Start(context.Background())

    var activeTasks int32

    // Act: Направляем массированный шквал задач в очередь
    for i := 0; i < totalTasks; i++ {
        p.Submit(func() {
            atomic.AddInt32(&activeTasks, 1)
            time.Sleep(50 * time.Millisecond) // Имитируем тяжелый I/O или вычисления
            atomic.AddInt32(&activeTasks, -1)
        })
    }

    // Даем планировщику Go время на развертывание пула
    time.Sleep(10 * time.Millisecond) 

    // Assert 1: Проверяем физический лимит горутин рантайма.
    // Количество горутин не должно превышать: база + maxWorkers + диспетчер
    currentGoroutines := runtime.NumGoroutine()
    require.LessOrEqual(t, currentGoroutines, baseGoroutines+maxWorkers+2, 
        "Worker Pool превысил физический лимит выделенных горутин!")

    // Assert 2: Проверяем логический лимит одновременно выполняемых задач
    require.LessOrEqual(t, int(atomic.LoadInt32(&activeTasks)), maxWorkers,
        "Превышен лимит параллельно обрабатываемых задач!")

    p.Stop() // Корректно останавливаем пул и дожидаемся завершения
}

func TestService_GracefulShutdown(t *testing.T) {
    t.Parallel()

    ctx, cancel := context.WithCancel(context.Background())
    svc := service.NewBackgroundService()

    // Запускаем фоновый сервис. Метод Start обязан быть неблокирующим:
    // он поднимает рабочие горутины и возвращает управление.
    svc.Start(ctx)

    // Даем сервису время перейти в устойчивое рабочее состояние
    time.Sleep(10 * time.Millisecond)

    // Act: Инициируем остановку через контекстную отмену
    cancel()

    // Assert: сервис обязан полностью завершить работу за отведенный бюджет времени.
    // Оборачиваем синхронное ожидание svc.Wait() в отдельную горутину с сигнальным каналом
    done := make(chan struct{})
    go func() {
        svc.Wait() // Внутри ожидает завершения внутренней sync.WaitGroup
        close(done)
    }()

    select {
    case <-done:
        // Превосходно: все воркеры штатно отреагировали на cancel() и завершились
    case <-time.After(500 * time.Millisecond):
        t.Fatal("Graceful Shutdown провалился: фоновые горутины зависли и не завершились за 500мс")
    }
}

func TestDataPipeline_FailFast(t *testing.T) {
    t.Parallel()

    // Конструируем моки зависимостей
    processor := mocks.NewMockProcessor(t)

    // Воркер 1: Имитирует медленную операцию, которая зависнет навсегда без отмены контекста
    processor.EXPECT().Process(gomock.Any(), "task_1").DoAndReturn(func(ctx context.Context, _ string) error {
        <-ctx.Done() // Ожидаем сигнала прерывания от соседней упавшей ветки
        return ctx.Err()
    })

    // Воркер 2: Моментально терпит аварию и возвращает критическую ошибку
    expectedErr := errors.New("critical DB connection lost")
    processor.EXPECT().Process(gomock.Any(), "task_2").Return(expectedErr)

    pipeline := worker.NewPipeline(processor)

    // Act: Фиксируем время выполнения пайплайна
    start := time.Now()
    err := pipeline.RunAll(context.Background(), []string{"task_1", "task_2"})

    // Assert: Проверяем каскадный эффект отмены
    require.ErrorIs(t, err, expectedErr, "Пайплайн обязан вернуть исходную ошибку от task_2")

    // Если бы механизм fail-fast не сработал, воркер task_1 выполнялся бы до глобального таймаута.
    // Мы доказываем, что отмена произошла практически мгновенно:
    require.Less(t, time.Since(start), 100*time.Millisecond, 
        "Пайплайн обязан был прекратить работу немедленно после сбоя task_2")
}