Отмена доходит до внешнего собеседника, а токен не покидает единой точки

- контекст проложен от воркера и обоих входов до внешних вызовов: ffmpeg и
  ffprobe заводятся через exec.CommandContext, SpeechKit и Object Storage
  принимают ctx вместо context.Background, скачивание записи идёт запросом с
  контекстом. Прежде остановка сервиса не доходила до чужой работы вовсе
- прерванный шаг приговора не выносит: убитый по контексту ffmpeg отдаёт
  «signal: killed», от настоящего отказа неотличимо ни типом, ни errors.Is, и
  различает их только ctx.Err(). Задача остаётся на повтор, попытку не тратит и
  отправителю о несуществующем сбое не сообщает; воркер не считает остановку
  отказом, а задача не забирается вовсе, если нас уже остановили
- клиента Bot API заводит единая точка internal/adapter/telegram: токен стоит в
  пути каждого обращения, а http.Client кладёт адрес в *url.Error целиком.
  Чистка на месте употребления закрывала один вызов из пяти — теперь свой Do
  чистит отказ, подменённый логгер вычищает токен из строк самой библиотеки, а
  транспорт бота токена не получает вовсе
- принятие операции распознавания защищено от отмены своим пределом: SpeechKit
  мог её принять и начать считать деньги, а потерянный идентификатор заставил
  бы повтор оплатить ту же запись второй раз
- приём по HTTP доводит запись до задачи независимо от отправителя: на
  контексте запроса один обрыв соединения терял полностью загруженную запись
- ответ Telegram с не-2xx кодом больше не становится записью: прежде тело
  отказа доезжало до хранилища и умирало на ffprobe, уводя диагностику
This commit is contained in:
av
2026-08-13 10:27:54 +03:00
parent d8d6bcc193
commit f494dcb83e
22 changed files with 855 additions and 137 deletions
+71 -21
View File
@@ -1,6 +1,7 @@
package service
import (
"context"
"errors"
"fmt"
"io"
@@ -75,7 +76,7 @@ func NewTranscribeService(
}
}
func (s *TranscribeService) CreateJobFromTelegram(file io.Reader, fileName string, chatId int64, replyMsgId int) (*entity.TranscribeJob, error) {
func (s *TranscribeService) CreateJobFromTelegram(ctx context.Context, file io.Reader, fileName string, chatId int64, replyMsgId int) (*entity.TranscribeJob, error) {
job := &entity.TranscribeJob{
State: entity.StateCreated,
Source: entity.SourceTelegram,
@@ -83,19 +84,19 @@ func (s *TranscribeService) CreateJobFromTelegram(file io.Reader, fileName strin
TgReplyMessageId: &replyMsgId,
}
return s.createTranscribeJob(job, file, fileName)
return s.createTranscribeJob(ctx, job, file, fileName)
}
func (s *TranscribeService) CreateJobFromApi(file io.Reader, fileName string) (*entity.TranscribeJob, error) {
func (s *TranscribeService) CreateJobFromApi(ctx context.Context, file io.Reader, fileName string) (*entity.TranscribeJob, error) {
job := &entity.TranscribeJob{
State: entity.StateCreated,
Source: entity.SourceApi,
}
return s.createTranscribeJob(job, file, fileName)
return s.createTranscribeJob(ctx, job, file, fileName)
}
func (s *TranscribeService) createTranscribeJob(job *entity.TranscribeJob, file io.Reader, fileName string) (*entity.TranscribeJob, error) {
func (s *TranscribeService) createTranscribeJob(ctx context.Context, job *entity.TranscribeJob, file io.Reader, fileName string) (*entity.TranscribeJob, error) {
// Определяем расширение файла
ext := filepath.Ext(fileName)
if ext == "" {
@@ -122,7 +123,7 @@ func (s *TranscribeService) createTranscribeJob(job *entity.TranscribeJob, file
// строка журнала вместе с идентификатором записи собрала бы её целиком.
s.logger.Info("Creating transcribe job", "file_ext", ext)
info, err := s.metaviewer.GetInfo(work.Path())
info, err := s.metaviewer.GetInfo(ctx, work.Path())
if err != nil {
s.logger.Error("Failed to get file info", "error", err, "file_ext", ext)
return nil, err
@@ -160,28 +161,41 @@ func (s *TranscribeService) createTranscribeJob(job *entity.TranscribeJob, file
return job, nil
}
func (s *TranscribeService) FindAndRunConversionJob() error {
return s.runStep(entity.StateCreated, conversionAcquireTimeout, s.convertJob)
func (s *TranscribeService) FindAndRunConversionJob(ctx context.Context) error {
return s.runStep(ctx, entity.StateCreated, conversionAcquireTimeout, s.convertJob)
}
func (s *TranscribeService) FindAndRunTranscribeJob() error {
return s.runStep(entity.StateConverted, transcribeAcquireTimeout, s.transcribeJob)
func (s *TranscribeService) FindAndRunTranscribeJob(ctx context.Context) error {
return s.runStep(ctx, entity.StateConverted, transcribeAcquireTimeout, s.transcribeJob)
}
func (s *TranscribeService) FindAndRunTranscribeCheckJob() error {
return s.runStep(entity.StateTranscribe, checkAcquireTimeout, s.checkTranscribeJob)
func (s *TranscribeService) FindAndRunTranscribeCheckJob(ctx context.Context) error {
return s.runStep(ctx, entity.StateTranscribe, checkAcquireTimeout, s.checkTranscribeJob)
}
// runStep забирает задачу и отдаёт её шагу. Отказ шага не оставляет задачу
// захваченной до конца срока: захват снимается, и задача ждёт нарастающую паузу
// — иначе повтор наступал бы через восемь часов, а не через секунду.
func (s *TranscribeService) runStep(state string, expiration time.Duration, step func(job *entity.TranscribeJob, holder string) error) error {
//
// Контекст доходит до шага, а через него — до внешнего собеседника: остановка
// сервиса убивает `ffmpeg` и обрывает запрос к распознаванию. Прерванный шаг
// приговора не выносит: задача остаётся пригодной к повтору, попытки не тратит
// и отправителю о несуществующем сбое не сообщает — исход остановки отличается
// от исхода отказа на каждом шаге.
func (s *TranscribeService) runStep(ctx context.Context, state string, expiration time.Duration, step func(ctx context.Context, job *entity.TranscribeJob, holder string) error) error {
// Нас уже остановили — задачу не забираем: захват стоил бы ей попытки, а
// работы всё равно не будет. Исход «шаг не сделал ничего» — это `NoopJobError`
// по смыслу, и он же не поднимает уровень и не считается в метрику.
if ctx.Err() != nil {
return &contract.NoopJobError{State: state}
}
job, holder, err := s.findJob(state, expiration)
if err != nil {
return err
}
if err := step(job, holder); err != nil {
if err := step(ctx, job, holder); err != nil {
s.scheduleRetry(job, holder, err)
return err
}
@@ -189,7 +203,7 @@ func (s *TranscribeService) runStep(state string, expiration time.Duration, step
return nil
}
func (s *TranscribeService) convertJob(job *entity.TranscribeJob, holder string) error {
func (s *TranscribeService) convertJob(ctx context.Context, job *entity.TranscribeJob, holder string) error {
s.logger.Info("Starting conversion job", "job_id", job.Id)
if job.FileID == nil {
@@ -227,13 +241,27 @@ func (s *TranscribeService) convertJob(job *entity.TranscribeJob, holder string)
// Измеряем время конвертации
startTime := clock.Start()
err = s.converter.Convert(src.Path(), dest.Path())
err = s.converter.Convert(ctx, src.Path(), dest.Path())
conversionDuration := time.Since(startTime)
// Записываем метрику времени конвертации
metrics.ObserveConversionDuration(srcExt, "ogg", err != nil, conversionDuration.Seconds())
if err != nil {
// Остановка сервиса — не приговор записи. Убитый по контексту `ffmpeg`
// отдаёт `signal: killed`, и от настоящего отказа конвертации
// (`exit status N`) эта ошибка неотличима ни типом, ни `errors.Is`:
// различает их только контекст шага. Без этой развилки каждый деплой
// хоронил бы конвертируемую запись в `failed` — состояние терминальное,
// и вернуть её оттуда может только владелец правкой в панели, — да ещё
// и сообщал бы отправителю о сбое, которого не было.
if ctxErr := ctx.Err(); ctxErr != nil {
s.logger.Info("File conversion interrupted by shutdown",
"job_id", job.Id,
"duration", conversionDuration)
return fmt.Errorf("conversion interrupted: %w", ctxErr)
}
s.logger.Error("File conversion failed",
"error", err,
"job_id", job.Id,
@@ -276,7 +304,7 @@ func (s *TranscribeService) convertJob(job *entity.TranscribeJob, holder string)
return nil
}
func (s *TranscribeService) transcribeJob(job *entity.TranscribeJob, holder string) error {
func (s *TranscribeService) transcribeJob(ctx context.Context, job *entity.TranscribeJob, holder string) error {
s.logger.Info("Starting transcribe job", "job_id", job.Id)
if job.FileID == nil {
@@ -304,8 +332,12 @@ func (s *TranscribeService) transcribeJob(job *entity.TranscribeJob, holder stri
s.logger.Info("Starting recognition", "job_id", job.Id, "file_id", *job.FileID)
// Запускаем асинхронное распознавание
operationID, err := s.recognizer.Recognize(content, fileRecord.FileName)
operationID, err := s.recognizer.Recognize(ctx, content, fileRecord.FileName)
if err != nil {
if ctxErr := ctx.Err(); ctxErr != nil {
s.logger.Info("Recognition interrupted by shutdown", "job_id", job.Id)
return fmt.Errorf("recognition interrupted: %w", ctxErr)
}
s.logger.Error("Failed to start recognition", "error", err, "job_id", job.Id)
return err
}
@@ -335,7 +367,7 @@ func (s *TranscribeService) transcribeJob(job *entity.TranscribeJob, holder stri
return nil
}
func (s *TranscribeService) checkTranscribeJob(job *entity.TranscribeJob, holder string) error {
func (s *TranscribeService) checkTranscribeJob(ctx context.Context, job *entity.TranscribeJob, holder string) error {
if job.RecognitionOpID == nil {
s.logger.Error("Recognition operation ID not found", "job_id", job.Id)
return fmt.Errorf("recognition opId not found for job: %s", job.Id)
@@ -345,8 +377,12 @@ func (s *TranscribeService) checkTranscribeJob(job *entity.TranscribeJob, holder
// Проверяем статус операции
s.logger.Info("Checking operation status", "job_id", job.Id, "operation_id", opId)
recResult, err := s.recognizer.CheckRecognitionStatus(opId)
recResult, err := s.recognizer.CheckRecognitionStatus(ctx, opId)
if err != nil {
if ctxErr := ctx.Err(); ctxErr != nil {
s.logger.Info("Status check interrupted by shutdown", "job_id", job.Id)
return fmt.Errorf("status check interrupted: %w", ctxErr)
}
s.logger.Error("Failed to check recognition status", "error", err, "operation_id", opId)
return err
}
@@ -375,8 +411,12 @@ func (s *TranscribeService) checkTranscribeJob(job *entity.TranscribeJob, holder
}
// Операция завершена, получаем результат
transcriptionText, err := s.recognizer.GetRecognitionText(opId)
transcriptionText, err := s.recognizer.GetRecognitionText(ctx, opId)
if err != nil {
if ctxErr := ctx.Err(); ctxErr != nil {
s.logger.Info("Text fetch interrupted by shutdown", "job_id", job.Id)
return fmt.Errorf("text fetch interrupted: %w", ctxErr)
}
s.logger.Error("Failed to get recognition text", "error", err, "operation_id", opId)
return err
}
@@ -450,6 +490,16 @@ func (s *TranscribeService) scheduleRetry(job *entity.TranscribeJob, holder stri
return
}
// Остановка попытки не тратит: задача не виновата в том, что нас
// перезапустили. Счётчик растёт при захвате, поэтому здесь его возвращают
// назад — иначе пять выкладок подряд уводят живую запись в «мертва» с
// приговором «попытки исчерпаны».
if errors.Is(stepErr, context.Canceled) || errors.Is(stepErr, context.DeadlineExceeded) {
if job.Attempts > 0 {
job.Attempts--
}
}
job.RetryAfter(clock.Now().Add(retryDelay(job.Attempts)))
if err := s.jobRepo.Save(job, holder); err != nil {