Ray повторил задачу. Как сделать так, чтобы запись осталась одна

Повторный запуск помогает конвейеру ИИ пережить падение рабочего процесса, но может ещё раз выполнить запись во внешнюю систему. В Ray полезно заранее разделить вычисление и подтверждение результата: тогда восстановление задачи не превращается в незаметное удвоение действия.

Что именно повторяет Ray

В документации Ray 2.59.0 описаны разные причины отказа. Если рабочий процесс неожиданно погиб, обычная задача по умолчанию получает до трёх повторов. max_retries меняет этот предел; ноль отключает такие повторы, минус один разрешает неограниченные.

Python-исключение из прикладного кода по умолчанию не вызывает повтор. Для него есть отдельный параметр retry_exceptions: можно разрешить все исключения или перечислить конкретные типы. Поэтому политика восстановления должна учитывать причину ошибки, а не просто наличие неудачного результата.

Есть менее очевидный случай: восстановление потерянного объекта может потребовать повторного выполнения уже завершившейся задачи. Следовательно, сохранённый где-то побочный эффект нельзя считать автоматически защищённым тем, что вычисление однажды закончилось.

Учебный пример: один документ, один результат

Представим обработчик документов: модель извлекает поля, а задача записывает их в таблицу. Рабочий процесс может оборваться после записи, но до передачи результата. Это условный сценарий, а не сообщение о конкретном инциденте. Повтор с новым случайным идентификатором создаст вторую строку, если приложение не распознаёт одну логическую операцию.

Для небольшого локального эксперимента возьмите SQLite и задайте ключу документа ограничение PRIMARY KEY. Выполните следующий фрагмент дважды в одной тестовой базе; он демонстрирует защиту записи, а не работу кластера Ray:

import sqlite3

with sqlite3.connect("retry-demo.sqlite") as db:
    db.execute("CREATE TABLE IF NOT EXISTS results "
               "(operation_key TEXT PRIMARY KEY, payload TEXT)")
    db.execute("INSERT INTO results VALUES (?, ?) "
               "ON CONFLICT(operation_key) DO NOTHING",
               ("document-17:extract-v1", "demo-result"))
    rows = db.execute("SELECT COUNT(*) FROM results").fetchone()[0]
    print(rows)

На новой тестовой базе оба запуска должны вывести 1. Транзакции SQLite определяют границу фиксации; уникальный ключ позволяет распознать повтор. В рабочем конвейере сохраняйте результат операции и возвращайте его повторному вызову, проверяя соответствие входных данных ключу. Одинаковый ключ для разных документов скроет настоящую работу.

Граница этого приёма

Пример защищает только одну запись в этой базе. Он не объединяет транзакцией запись и обращение к внешнему API. Для такого действия проверьте поддержку постоянного ключа идемпотентности у получателя и возможность прочитать итог после обрыва. Неизвестный исход сначала сверяйте с системой, которая выполнила действие.

Критерий приёмки — намеренно прервать тест после фиксации, повторить ту же логическую операцию и увидеть один эффект с тем же результатом. Число попыток может вырасти; число выполненных действий должно остаться ожидаемым.