Идемпотентное изменение остатков

Оглавление серии

Повтор задачи должен возвращать прежний результат, а не прибавлять поступление ещё раз. Это свойство называют идемпотентностью: несколько выполнений одного логического действия имеют тот же предметный эффект, что и одно. Добавим в «Импортёр» отметку batchId и запишем её вместе с новыми остатками и поколением каталога. Малый учебный пакет применится одним Lua-скриптом.

Состояние stages/08 заменяет неидемпотентный цикл предыдущих шагов. Код имеет статус not_run: Lua, повтор и количества не проверялись запуском. Атомарность означает отсутствие вмешательства других команд во время исполнения скрипта; она не обещает автоматический откат при произвольной ошибке Lua и не заменяет долговечность Redis. Ниже объясним условия, в которых наш договор малого пакета применим.

Отметка хранит результат

Для пакета используем ключ importer:done:{batchId}. Если он существует, скрипт возвращает сохранённый JSON и не меняет продукты. Если нет, вычисляет новые количества, записывает их, увеличивает generation и сохраняет результат. Важно хранить сам результат, а не только true: повтор должен вернуть imported и batchId того же действия, чтобы worker мог завершиться с прежним ответом независимо от номера попытки.

Полный файл apply.js содержит скрипт и обёртку вызова. productsKey и generationKey импортируются из catalog.js; rows уже проверены CSV-парсером, а batchId — договором задачи.

import { productsKey, generationKey } from './catalog.js';

const script = `
local previous = redis.call('GET', KEYS[3])
if previous then return previous end
local rows = cjson.decode(ARGV[1])
local updates = {}
for _, row in ipairs(rows) do
  local raw = redis.call('HGET', KEYS[1], row.sku)
  local quantity = 0
  if raw then quantity = cjson.decode(raw).quantity end
  local nextQuantity = quantity + row.quantity
  if nextQuantity > 1000000 then return redis.error_reply('quantity_limit') end
  table.insert(updates, {sku=row.sku, title=row.title, quantity=nextQuantity})
end
for _, row in ipairs(updates) do
  redis.call('HSET', KEYS[1], row.sku, cjson.encode(row))
end
redis.call('INCR', KEYS[2])
local result = cjson.encode({imported=#updates, batchId=ARGV[2]})
redis.call('SET', KEYS[3], result)
return result
`;

export async function applyShipment(redis, batchId, rows) {
  const result = await redis.eval(script, 3,
    productsKey, generationKey, `importer:done:${batchId}`,
    JSON.stringify(rows), batchId);
  return JSON.parse(result);
}

Сначала собираем updates, не меняя hash. Каждая строка читает прежнее количество и вычисляет следующее. Если предел количества превышен, скрипт возвращает ошибку до записи любого товара. Этот порядок предотвращает частичное изменение по ожидаемому предметному отказу. Если начать HSET сразу внутри первого цикла, ошибочная третья позиция могла бы оставить уже изменённые первые две, несмотря на атомарное исполнение относительно других клиентов.

Все части результата связаны

Остатки, generation и отметка относятся к одному смысловому результату. Если записать только отметку до остатков, повтор решит, что пакет уже применён, и пропустит реальное поступление. Если записать остатки, а отметить пакет отдельной поздней командой, сбой между ними приведёт к повторной прибавке. Поэтому маркер «сделано» нельзя хранить как случайный внешний кеш вокруг неатомарной бизнес-операции.

Worker после проверки CSV теперь вызывает applyShipment(connection, batchId, rows). Возвращённый result становится результатом задачи BullMQ. Обработка одинакового batchId в двух воркерах последовательно проходит через Redis-скрипт: первый меняет данные и сохраняет отметку, второй находит результат. Внешние процессы могут быть конкурентными, а точка принятия пакета остаётся одной согласованной операцией на выбранном Redis.

Ожидаемая первая обработка даёт лампу 20, блокнот 8 и карандаш 4. Повтор того же batchId возвращает imported 3, но количества не меняются. Другой batchId означает новое поступление и прибавляет те же количества ещё раз. Идемпотентность относится к идентичности действия, а не к имени файла или совпадению текста CSV. Это ожидаемая модель, не результат выполненных повторных запусков.

batch A впервые → остатки 20, 8, 4
batch A повторно → остатки 20, 8, 4
batch B с тем же CSV → остатки 32, 13, 8

Идентификатор должен обозначать неизменное действие

В нашем договоре dataset всегда shipment-a, а файл снимка считается неизменным. Поэтому batchId однозначно относится к конкретному содержимому. Если в production пользователь может загрузить другой файл под тем же ID, нужно сохранить и сравнить хеш содержимого или отдельную версию входа. Иначе исправленный CSV получит старый result, хотя человек ожидает новую обработку. Повторный ключ с другим содержимым должен иметь явный отказ или иной предусмотренный смысл.

Отметка не имеет TTL. Это сознательное различие с кешем: удаление cache-ключа лишь приводит к повторному чтению, а удаление done разрешает снова прибавить остатки. Позднее retention BullMQ сможет удалить завершённую задачу, однако бизнес-отметка останется. Срок её хранения требуется выбирать по правилу допустимого повторного поступления и долговечному журналу, а не копировать короткий TTL, удобный для каталога.

Если один и тот же batchId принадлежит разным организациям, общий ключ склеит несвязанные действия. Поэтому реальное приложение должно включать владельца в область идемпотентности и проверять права до постановки и чтения результата. Локальная серия имеет один вымышленный каталог и не реализует аутентификацию. Политика именования облегчает объяснение, но не является готовым многопользовательским контрактом SaaS.

Атомарность Lua имеет границы

Redis не допускает вмешательства обычной другой команды во время выполнения скрипта, но уже выполненные записи не откатываются автоматически при поздней runtime-ошибке. Наш скрипт предварительно проверяет вычисляемые количества, а namespace должен содержать ключи ожидаемых типов. Неправильный тип generation, повреждённый JSON или отказ памяти могут нарушить предположения. Для важных данных нужно заранее контролировать формат и ошибки либо использовать транзакционную бизнес-базу с подходящими ограничениями.

Скрипт блокирует обработку других команд на время вычисления. Поэтому rows ограничен сотней небольших строк, а количество — миллионом. Для большого CSV нельзя просто убрать предел и отправить весь файл в EVAL. Потребуется разбить работу на согласованные части с отдельными ключами идемпотентности, учесть прогресс и определить, считается ли завершением весь импорт или каждый пакет. Увеличение размера меняет договор, а не только время ожидания.

Все ключи скрипта в этом курсе принадлежат одному обычному Redis. Redis Cluster требует учитывать размещение нескольких ключей в одном hash slot и допустимость многоключевого скрипта. Пример не заявляет совместимость с кластером и не подключается к нему. При переносе следует заново рассмотреть ключи, область атомарности и стратегию масштабирования, вместо механической замены адреса соединения.

Долговечность остаётся обязанностью конфигурации Redis и восстановления. Если после подтверждения потерять и продукты, и отметки из-за восстановления старого состояния, повтор может изменить другой исходный каталог. Если восстановить их несогласованно, можно получить потерю или повтор поступления. Поэтому резервирование должно учитывать обе категории вместе. Идемпотентная функция защищает от повторного исполнения при сохранённых данных, а не от любой возможной потери хранилища.

Паттерн описан в BullMQ Idempotent jobs, а исполнение скриптов — в Redis EVAL. После восьмого состояния повтор одного сохранённого пакета возвращает прежний результат. Теперь можно предметно рассмотреть потерю блокировки worker и конкуренцию, сохраняя правильное изменение малого каталога.