Итоговый проект: система рассылок

Продукт простой и узнаваемый: человек оставляет почту — получает welcome-письмо, а раз в неделю — дайджест новинок. За простотой стоит конвейер из задач, который обязан не терять письма, не дублировать их и не душить почтовый шлюз. Два типа задач: welcome и digest. Welcome ставится из HTTP-хендлера с отложкой в пять минут — за это окно человек успевает передумать, и письмо честно не уходит. Дайджест — периодическая задача, а уведомления внутри неё собираются в пачки агрегатором, чтобы один пользователь получил одно письмо, а не пятьдесят. Под письмами — два слоя инфраструктуры: очереди critical и default с разными приоритетами и middleware с логами и метриками, через который проходит каждая задача. Сборка идёт по швам, которые уже разобраны: постановка из веба, идемпотентность по ID, группы, ретраи, тесты, остановка. Каждое решение помечено номером урока-источника, так что непонятный стык открывается заново за минуту, а весь каркас переносится в свой проект почти без правок. Порядок разбора: сначала карта проекта и общая архитектура, затем welcome-конвейер с постановкой из веба и двойной защитой отписки, дальше дайджест с периодикой и агрегацией, и в конце — надёжность: ретраи, тесты, наблюдаемость и остановка. Строки кода сокращены до существенного: декларации типов, обвязка логирования и импорты опущены, логика сохранена целиком.

Архитектура: два типа задач и две очереди#

Проект живёт в трёх частях: HTTP-сервис принимает подписки и отписки, клиент ставит задачи, сервер их исполняет — все три могут быть одним процессом на старте и разными сервисами в проде. Типы задач задаются константами: TypeWelcomeEmail и TypeDigestEmail регистрируются в mux из восьмого урока. Очереди две: welcome-письма едут в default, дайджест — в critical, потому что утренняя рассылка не должна ждать за потоком одиночных писем; приоритеты и ловушки нескольких очередей — тема седьмого урока. Вокруг каждой задачи — middleware из двадцать третьего урока: лог с типом, ID и длительностью, счётчики обработок и ошибок; без него рассылка — чёрный ящик, с ним — наблюдаемый процесс. Отдельный закон проекта: письма шлёт только воркер, никакой хендлер не отправляет почту синхронно. HTTP-ответ возвращает 202 Accepted сразу после постановки задачи — пользователь не ждёт SMTP, а очередь гарантирует доставку даже при перезапуске сервиса. Клиент и сервер строятся на одном RedisClientOpt и живут в одном репозитории, но пакеты раздельные: api, tasks, mailer — при разделе на два сервиса граница уже проведена. Регистрация хендлеров идёт через srv.Handle с константой типа: опечатка в имени ловится компиляцией, а не ночным инцидентом. Гонки между подпиской и отпиской разруливают два механизма: детерминированный ID задачи и проверка состояния подписки в самом хендлере.

Кирпич проекта — урок-источник#

Кирпич	Урок	Роль в проекте
ProcessIn(5 * time.Minute)	05	окно на отмену: письмо уходит не сразу
TaskID("welcome:" + email)	04, 12	детерминированный ID, дубль постановки ловится
Очереди critical и default	07	дайджест не ждёт за одиночными письмами
GroupAggregator + GroupGracePeriod	10	уведомления одного пользователя — одной пачкой
Ретраи с backoff, SkipRetry	06, 13	SMTP деградирует — письма не теряются, битые не зацикливаются
Тесты хендлеров без Redis	22	логика писем держится под замком
Graceful shutdown	19	деплой не роняет письма на полуслове

Проверка: состояние после ProcessIn

Welcome-письмо поставлено через asynq.ProcessIn(5 * time.Minute). Где задача через минуту?

Welcome: постановка из HTTP-хендлера#

Хендлер подписки делает четыре шага. Первый — декод тела и нормализация почты: TrimSpace и нижний регистр, иначе ada@example.com и ADA@example.com станут двумя подписчиками. Второй — сохранение подписчика в хранилище: если запись уже есть, это доменная ситуация, отвечаем 409, а не ставим второе письмо. Третий — payload: json.Marshal структуры WelcomePayload с одним полем Email; payloads из четвёртого урока держат маленькими и плоскими. Четвёртый — постановка: asynq.NewTask(TypeWelcomeEmail, payload) и client.Enqueue с тремя опциями. ProcessIn(5 * time.Minute) даёт окно на отмену — тарифицированное письмо не уходит мгновенно. TaskID("welcome:" + email) делает постановку идемпотентной: повторная подписка на ту же почту вернёт ошибку дубликата вместо второго письма — приём из уроков про ID и уникальность. Queue("default") и MaxRetry(5) завершают контракт. Ошибку дубликата хендлер распознаёт через errors.Is(asynq.ErrDuplicateTask) и трактует спокойно: подписка уже активна, клиенту достаточно 202 — но лог запишет попытку, потому что молчаливый дубль однажды окажется не дублем, а симптомом. Для постановки с наследованием контекста запроса есть EnqueueContext: пока подписка ставится синхронно и быстро, обычного Enqueue хватает, но при постановке из под веба при таймаутах клиента контекстная версия честнее.

POST /subscribe: постановка welcome

func (s *Server) handleSubscribe(w http.ResponseWriter, r *http.Request) {
    var req SubscribeRequest
    if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
        http.Error(w, "битое тело", http.StatusBadRequest)
        return
    }
    email := strings.ToLower(strings.TrimSpace(req.Email))
    if err := s.store.CreateSubscriber(r.Context(), email); err != nil {
        http.Error(w, "уже подписан", http.StatusConflict)
        return
    }
    payload, err := json.Marshal(WelcomePayload{Email: email})
    if err != nil {
        http.Error(w, "marshal", http.StatusInternalServerError)
        return
    }
    _, err = s.client.Enqueue(asynq.NewTask(TypeWelcomeEmail, payload),
        asynq.ProcessIn(5*time.Minute),
        asynq.TaskID("welcome:"+email),
        asynq.Queue("default"),
        asynq.MaxRetry(5),
    )
    if err != nil {
        log.Printf("enqueue welcome %s: %v", email, err)
    }
    w.WriteHeader(http.StatusAccepted)
}

Хендлер не отправляет почту — он фиксирует намерение в очереди и отвечает 202. ProcessIn откладывает письмо на пять минут, TaskID строит детерминированный идентификатор из адреса: повторный POST на ту же почту упадёт с ошибкой дубликата, не создав вторую задачу. Ошибка постановки не роняет ответ — письмо уже сохранено в хранилище, а провал логируется для ручного повтора.

Welcome-хендлер: идемпотентность и отписка#

Обратная сторона отложки — за пять минут человек может передумать, и письмо обязано это пережить. Отписка работает в два слоя. Первый — быстрая отмена: хендлер POST /unsubscribe помечает подписку неактивной в хранилище и просит Inspector удалить задачу DeleteTask("default", id) — для scheduled-задачи это штатная операция, а состояние подписки в базе меняется всегда, даже если задачи в очереди уже нет. Второй слой — проверка в самом welcome-хендлере: перед отправкой он перечитывает подписчика и молча выходит с nil, если подписка неактивна или не подтверждена. Двойная защита не паранойя, а норма: удаление из очереди и чтение из хранилища — две разные системы, и гонка «письмо уже в active, отписка приходит в этот момент» разрешается только проверкой на месте. Хендлер идемпотентен и по-другому: он не держит в голове, сколько раз его вызовут, — повторный вызов после ретрая упадёт на том же фильтре и не отправит второе письмо. Таймауты из одиннадцатого урока заданы опцией Timeout при постановке: письмо обязано уйти за две минуты, иначе попытка проваливается и очередь забирает её снова — висячий SMTP-коннект не удержит воркер навсегда. Ошибки классифицируются по четырнадцатому уроку: битый payload — SkipRetry, недоступность хранилища и SMTP — обычный возврат ошибки с wrap, дальше работает backoff из шестого.

Хендлер welcome: проверка перед отправкой

func (s *Server) handleWelcome(ctx context.Context, t *asynq.Task) error {
    var p WelcomePayload
    if err := json.Unmarshal(t.Payload(), &p); err != nil {
        return fmt.Errorf("unmarshal welcome: %v: %w", err, asynq.SkipRetry)
    }
    sub, err := s.store.Subscriber(ctx, p.Email)
    if err != nil {
        return fmt.Errorf("load subscriber %s: %w", p.Email, err)
    }
    if sub == nil || !sub.Confirmed {
        log.Printf("welcome skip: %s отписался", p.Email)
        return nil
    }
    if err := s.mailer.SendWelcome(ctx, p.Email); err != nil {
        return fmt.Errorf("send welcome %s: %w", p.Email, err)
    }
    log.Printf("welcome sent: %s", p.Email)
    return nil
}

Хендлер перечитывает состояние подписки непосредственно перед отправкой — это второй слой защиты, который закрывает гонку с отпиской. Битый payload получает SkipRetry: ретраить нечего, данные не исправятся. Хранилище и SMTP упавшими ошибками с wrap уходят в обычный ретрай по политике, а сам хендлер остаётся чистой функцией — с fake-хранилищем и fake-mailer он тестируется без Redis.

Живое место: гонка отписки и письма

Самый скользкий стык проекта. Отписка удалила задачу — а письмо уже в active на другом воркере: DeleteTask не трогает исполняемые задачи, и единственный барьер остаётся в хендлере. Обратная гонка тоже возможна: проверка подписки прошла, письмо уходит — и в этот же миг человек жмёт отписаться. Идеала нет, есть договор: письмо, отправленное в окне пары секунд вокруг отписки, считается допустимым, а вот второе welcome-письмо — нет. Поэтому TaskID в постановке обязателен: он отсекает дубль на входе, а не на выходе, когда письмо уже ушло.

Проверка: отмена отложенного письма

Человек отписался через две минуты после подписки, welcome-задача ещё в scheduled. Что делает отписка?

Дайджест: периодика и группы#

Периодический дайджест ставится двумя способами, и оба уже были в курсе. Способ первый — cron-постановка из девятого урока: планировщик раз в сутки парсит cron-выражение и вызывает Enqueue с ProcessAt на следующий слот. Способ второй — PeriodicTaskManager: конструктор NewPeriodicTaskManager получает RedisConnOpt и SchedulerOpts, провайдер отдаёт срез PeriodicTaskConfig с CronSpec и задачей, Run блокируется до сигнала — менеджер удобен, когда конфигурация приезжает из хранилища и меняется без деплоя. Вторая половина дайджеста — агрегация из десятого урока: каждое событие продукта ставит крошечную задачу с опцией Group и ключом-почтой, сервер копит их в группе, а GroupAggregatorFunc собирает пачку в один TypeDigestEmail с адресами внутри. Настройки Config управляют терпением: GroupGracePeriod — минимальное окно ожидания хвоста (меньше секунды запрещено), GroupMaxDelay — жёсткий потолок ожидания, GroupMaxSize — размер пачки, по достижении которого сборка происходит немедленно. Дайджест-хендлер получает payload со срезом адресов, шлёт одно письмо на человека и помечает события доставленными — идемпотентность и здесь держится на детерминированных ID и проверках состояния: повторный запуск пачки после падения воркера не задвоит письма, потому что отметка о доставке пишется тем же хранилищем, которое читает фильтр.

Конфиг сервера: очереди и агрегатор

srv := asynq.NewServer(redisOpt, asynq.Config{
    Concurrency: 10,
    Queues: map[string]int{
        "critical": 6,
        "default":  3,
    },
    GroupGracePeriod: 2 * time.Minute,
    GroupMaxDelay:    15 * time.Minute,
    GroupMaxSize:     50,
    GroupAggregator: asynq.GroupAggregatorFunc(
        func(group string, tasks []*asynq.Task) *asynq.Task {
            var emails []string
            for _, t := range tasks {
                var p EventPayload
                if json.Unmarshal(t.Payload(), &p) == nil {
                    emails = append(emails, p.Email)
                }
            }
            payload, _ := json.Marshal(DigestPayload{Emails: emails})
            return asynq.NewTask(TypeDigestEmail, payload)
        },
    ),
})

Queues задаёт веса очередей: critical с дайджестом получает вдвое больше внимания воркеров, чем default с welcome-письмами. GroupGracePeriod держит группу открытой две минуты в надежде на хвост событий, GroupMaxDelay не даст ждать дольше пятнадцати, GroupMaxSize собирает пачку сразу при пятидесяти задачах. Агрегатор склеивает payload в один дайджест-задачу — это и есть то самое одно-письмо-на-пользователя.

Пачка собралась раньше срока

Прочитайте код и выберите, что он выведет:

// GroupGracePeriod: 2 * time.Minute
// GroupMaxSize: 50
// в группу прилетела 50-я задача,
// grace period ещё не истёк

Надёжность: ретраи, тесты, наблюдаемость#

Рассылка прощает задержки, но не потерю письма, и каждый урок курса здесь превращается в строку конфига или тест. Ретраи: SMTP деградирует редко, но метко — дефолтный экспоненциальный backoff из шестого урока растянет пять попыток на часы, а кастомный RetryDelayFunc из тринадцатого умеет уважать Retry-After почтового провайдера. Тесты: хендлеры — обычные функции, поэтому подход двадцать второго урока ложится без адаптации — fake-хранилище на карте, fake-mailer, который пишет в срез, и проверки на трёх сценариях: подписан, отписался, битый payload. Middleware из двадцать третьего урока пишет тип, ID, очередь и длительность каждой задачи — эти строки и метрики отвечают на вопрос «почему письмо ушло дважды» быстрее любой догадки. Дашборд asynqmon из шестнадцатого урока ставится рядом одной командой и показывает scheduled-окна, группы и латентность очередей вживую. Graceful shutdown из девятнадцатого закрывает деплой: SIGTSTP останавливает забор новых задач, SIGTERM даёт активным письмам дописаться. Восстановитель из глубин курса довершает картину: воркер, умерший посреди письма, отдаст задачу обратно, IsOrphaned подскажет такие случаи в списке активных. Ошибки хендлеров классифицированы, payload валидируется до отправки — грабли двадцать четвёртого урока сняты заранее.

Про песочницу: письмо без SMTP

Песочница сайта исполняет программу-имитацию на чистом Go: fake-mailer печатает письма в stdout, очередь эмулируется срезом с задержками — смысл постановки, отмены и агрегации виден без Redis и без Docker. Локально всё по-настоящему: Go, Redis в контейнере, а рядом — asynqmon для живого дашборда: scheduled-окна welcome, группы дайджеста и латентность очередей видны в тот же момент, когда вы жмёте кнопку подписки.

Чек-лист перед запуском#

Постановка welcome идемпотентна: детерминированный TaskID из email, дубль ловится ошибкой.
У welcome есть окно отмены: ProcessIn, удаление из очереди при отписке, проверка в хендлере.
Дайджест собирается группами: GracePeriod, MaxDelay и MaxSize подобраны под трафик.
Очереди разведены: дайджест в critical, одиночные письма в default, веса проверены под нагрузкой.
Ошибки классифицированы: битые payload — SkipRetry, SMTP и хранилище — ретрай с backoff.
Хендлеры протестированы без Redis: подписан, отписался, битый payload.
Middleware пишет тип, ID, очередь и длительность; метрики уходят в мониторинг.
Деплой переживёт SIGTERM: graceful shutdown включён и отрепетирован.
Периодика дайджеста описана cron-выражением и проверена на локальном расписании.

Соберите постановку welcome

Допишите опции: отложка на пять минут и детерминированный идентификатор задачи.

_, err = s.client.Enqueue(asynq.NewTask(TypeWelcomeEmail, payload), asynq.(5*time.Minute), asynq.("welcome:"+email), )

Соберите конвейер подписки

Расставьте шаги хендлера подписки: от декода тела до ответа клиенту.

Строки

Ваша программа

Нажимайте на строки слева, чтобы собрать программу

Карточки: итоговый проект

Термин

ProcessIn(5 * time.Minute)

нажмите, чтобы перевернуть

Определение

Опция постановки: задача уходит в scheduled и исполнится через пять минут. Окно, за которое успевает прийти отписка.

1 / 6

Проект собран, и чек-лист курса закрыт целиком: постановка из веба, отложки и окно отмены, идемпотентные ID, очереди с приоритетами, группы и агрегация, ретраи с backoff, тесты без Redis, middleware, дашборд и graceful shutdown — двадцать четыре урока превратились в один работающий сервис. Мини-эксперимент: поднимите Redis, поставьте welcome через ProcessIn на пять минут, откройте asynqmon, найдите задачу в scheduled, отставьте её RunTask'ом — и сравните поведение с отпиской, которая удаляет письмо DeleteTask'ом. Дальше три дороги. Первая — курс по Redis: очередь asynq — тонкий слой над структурами данных Redis, и понимание списков, отсортированных множеств и pubsub объяснит, куда уходит память под тысячи scheduled-задач. Вторая — курс по cron: cron-выражения из девятого урока станут родным языком. Третья — регулярная практика: возьмите каркас этого проекта, добавьте подтверждение почты по токену, отписочную ссылку с одноразовым ключом и шаблоны писем — и у вас будет продакшн-рассылка, а не учебная.

Песочница урока

Редактируйте код и запускайте его прямо здесь — интерпретатор работает в браузере, ничего устанавливать не нужно. За каждый запуск — +2 XP.

Вывод

Запустите код, чтобы увидеть вывод.

Коротко: что запомнить

Итоговый проект — система email-рассылок: два типа задач, две очереди, HTTP-сервис плюс воркеры.
Welcome ставится из хендлера с ProcessIn(5 * time.Minute) — окно на отмену для отписки.
TaskID("welcome:" + email) делает постановку идемпотентной: дубль ловится ошибкой дубликата.
Отмена письма работает в два слоя: DeleteTask из очереди плюс проверка подписки в хендлере.
Дайджест — периодическая задача; уведомления собираются группами через GroupAggregatorFunc.
GracePeriod, MaxDelay и MaxSize управляют терпением группы; MaxSize собирает пачку немедленно.
Ретраи с backoff спасают от деградации SMTP, SkipRetry отсекает битые payload.
Хендлеры тестируются без Redis, деплой закрывает graceful shutdown, asynqmon показывает всё вживую.

Домашнее задание

Соберите проект в трёх пакетах — api, tasks, mailer — и запустите целиком: HTTP-хендлер подписки с идемпотентной постановкой welcome, хендлер письма с проверкой подписки, PeriodicTaskManager или cron-постановку дайджеста и агрегатор групп. Снимите три сценария тестами без Redis: повторная подписка не создаёт дубль, отписка в окне пяти минут оставляет письмо неотправленным, битый payload не должен ретраиться. Поднимите asynqmon и посмотрите, как welcome живёт в scheduled и исчезает при отписке — этот сервис станет вашим стартовым шаблоном.

Вопросы и ответы по уроку

Почему окно отмены именно пять минут?

Пять минут — компромисс из проекта: человек успевает передумать, а бизнес не копит ненужных ожиданий. Значение задаётся ProcessIn в одном месте и проверяется нагрузкой: чем длиннее окно, тем больше задач копится в scheduled и тем дольше жить записи о подписке до подтверждения.

Почему проверка подписки в хендлере, если задача уже удалена при отписке?

DeleteTask не трогает исполняемые задачи, а гонка «письмо уже в active» вполне реальна. Хендлер читает актуальное состояние непосредственно перед отправкой — это второй слой, который гарантирует: неактуальное письмо не уйдёт, даже если отменить из очереди не удалось.

Что брать для периодики — cron-постановку или PeriodicTaskManager?

Cron-постановка проще и достаточно, когда расписание зашито в код и не меняется. PeriodicTaskManager берут, когда конфигурация приезжает из хранилища или конфига и должна применяться без перезапуска: провайдер опрашивается периодически, и новые PeriodicTaskConfig подхватываются живым менеджером.

Термины урока

Ключевые понятия из этого урока — короткие определения в глоссарии. Наведите на термин, чтобы вспомнить, откройте карточку для подробностей и ссылки на практику.