Приём пачки: память вердиктов и отказ по каждой операции

Что именно сервер обязан ответить на повтор, договорено ещё в части про оффлайн: повтор принятой операции получает duplicate, повтор отвергнутой — тот же rejected с той же причиной. Помнить это надо не про пачку, а про каждую операцию по отдельности. Пачка собирается заново при каждой отправке — сегодня в ней три операции, завтра тысяча, — а имя операции живёт с ней от рождения: opId появился на устройстве в тот же миг, что и сама запись.

Память вердиктов ввела статья про потерянный ответ, и там же записан сеанс psql, в котором повтор внутри пачки оборвал транзакцию целиком. С этого места мы и продолжим — на Node, с настоящим драйвером и с таблицей, у которой ключ шире, чем было обещано.

Шире он ровно по той причине, по которой в этой главе вообще есть статья про владение: opId придумывает клиент. Приставка «A» в примерах — сокращение ради читаемости: настоящее имя устройства — случайный UUID, и случайно два телефона на одном имени не встретятся. Но имя приезжает в теле запроса, и проверить его сервер не может: прислать чужое — не взлом, а обычный POST. А законно повторяться этому имени положено: тот же opId приходит с двух устройств одного человека, и разбирается это ниже. Ключом из одного op_id сервер не отличил бы одно от другого — человек с чужим именем операции получил бы чужой вердикт и своей записи не дождался. Поэтому первичный ключ — пара (user_id, op_id), и дальше в статье «память» всегда значит «память этого человека».

Соберём память на своей таблице из главы про данные — той, у которой есть уникальный индекс и настоящий 23505. Вердикты пачки здесь уже вынесены, и всё, что делает remember, — запоминает их и отвечает на повтор. Прогноз: что вернут три вызова — первая отправка пачки, та же пачка второй раз и та же пачка от другого человека, — и сколько строк окажется в памяти?

// Таблица с индексами и уникальностью — на подложке; здесь только её типы и новое
type Row = Record<string, unknown> & { id: number };
declare class DbError extends Error {
  code: string;
  constraint: string;
}
declare class Table {
  scanned: number;
  index(name: string, column: string, options?: { unique?: boolean }): Table;
  insert(values: Record<string, unknown>): Row;
  update(id: number, patch: Record<string, unknown>): Row | null;
  remove(id: number): boolean;
  find(indexName: string, value: unknown): Row[];
  where(match: (row: Row) => boolean): Row[];
}

type Verdict = { opId: string; status: "applied" | "duplicate" | "rejected"; reason?: string };

// PRIMARY KEY (user_id, op_id). У модельной таблицы ключ один, поэтому пара
// склеена в строку — в базе это две колонки
const verdicts = new Table().index("sync_verdicts_pkey", "key", { unique: true });
const key = (userId: string, opId: string): string => userId + "|" + opId;

// Память вердиктов: вердикт уже вынесен, здесь его только запоминают
function remember(userId: string, computed: Verdict): Verdict {
  try {
    verdicts.insert({ key: key(userId, computed.opId), ...computed });
    return computed;
  } catch (error) {
    if (!(error instanceof DbError) || error.code !== "23505") throw error;
    const saved = verdicts.find("sync_verdicts_pkey", key(userId, computed.opId))[0];
    return saved.status === "applied"
      ? { opId: computed.opId, status: "duplicate" }
      : { opId: computed.opId, status: "rejected", reason: saved.reason as string };
  }
}

// вердикты этой пачки уже вынесены — чем именно, разбирает следующий раздел
const batch: Verdict[] = [
  { opId: "A:1", status: "applied" },
  { opId: "A:2", status: "rejected", reason: "повторов больше сотни" },
  { opId: "A:3", status: "applied" },
];
const show = (label: string, list: Verdict[]): void => {
  console.log(label, list.map((v) => v.opId + " " + v.status + (v.reason ? " · " + v.reason : "")).join(", "));
};

show("первая отправка:", batch.map((op) => remember("аня", op)));
show("та же пачка второй раз:", batch.map((op) => remember("аня", op)));
show("та же пачка от Бори:", batch.map((op) => remember("боря", op)));
console.log("строк в памяти:", verdicts.where(() => true).length);

Первая отправка разобрана обычным порядком. На второй ни одна операция заново не разбиралась: A:1 и A:3 вернулись как duplicate, а A:2 — тем же rejected и с той же причиной, слово в слово. Это важнее, чем кажется: причина отказа — единственное, что телефон может показать человеку, и меняться она между повторами не должна. А пачка Бори разобрана целиком, хотя имена операций в ней те же: ключ — пара, и строки в памяти легли рядом, не спорив друг с другом.

В описании схемы это выглядит так. Составной первичный ключ у Drizzle объявляется третьим аргументом pgTable — тем, где живут индексы и проверки:

import { pgTable, primaryKey, text, timestamp, uuid } from 'drizzle-orm/pg-core';

export const syncVerdicts = pgTable(
  'sync_verdicts',
  {
    userId: uuid('user_id')
      .notNull()
      .references(() => users.id, { onDelete: 'cascade' }),
    // имя операции родилось на устройстве человека, поэтому ключом оно работает
    // только в паре с владельцем
    opId: text('op_id').notNull(),
    status: text('status').notNull(),
    reason: text('reason'),
    decidedAt: timestamp('decided_at', { withTimezone: true, mode: 'date' }).notNull().defaultNow(),
  },
  (table) => [primaryKey({ columns: [table.userId, table.opId] })],
);

Отдельного индекса под эту память не нужно: единственный запрос к ней — «что я ответил на эту операцию этому человеку», и на него отвечает сам первичный ключ. А decided_at здесь не для чистки, а для разбора инцидентов: по нему видно, когда сервер вынес вердикт, если через месяц придётся объяснять, почему у телефона строка одна, а у планшета другая.

Потому что чистки не будет. Память вердиктов не чистится — ни от applied, ни от rejected, никогда. Это прямое следствие обещания, которое часть V проверяла прогоном: файл дневника можно открыть на двух новых устройствах, и ни одна операция из него не станет второй записью — тот же вердикт по тому же opId сколько угодно раз. Слово «сколько угодно» стоит понимать буквально: файл могут открыть и через год.

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

Память вердиктов — единственное место, где сервер помнит про операцию. Всё остальное он помнит про строку.

В модели remember ловил DbError с кодом 23505 и читал сохранённый вердикт — и это работало. Приём знакомый: глава про свою таблицу учила ловить именно код, а не текст, потому что на вопрос «занято ли» может ответить только тот, кто вставляет. Перенесём его на настоящий Postgres — и окажется, что у модели не было одной вещи, которая всё меняет: транзакции.

Пачка приходит одной транзакцией. Первая её операция — повтор уже принятой A:1, вторая — новая A:2, с которой всё в порядке. Прогноз: что напечатает драйвер на первой вставке, что — на второй, чем ответит COMMIT и что останется в памяти вердиктов, когда пачка кончится?

import pg from 'pg';

// адрес базы — первой строкой и без пароля: видно, куда ушли запросы
const url = new URL(process.env.DATABASE_URL);
url.password = '';
console.log('база:', url.href);

const c = new pg.Client({ connectionString: process.env.DATABASE_URL });
await c.connect();

await c.query(`CREATE TABLE sync_verdicts (
  user_id text NOT NULL,
  op_id text NOT NULL,
  status text NOT NULL,
  reason text,
  PRIMARY KEY (user_id, op_id)
)`);
// A:1 уже принята прошлой пачкой: ответ на неё телефон не получил
await c.query(`INSERT INTO sync_verdicts VALUES ('u1', 'A:1', 'applied', NULL)`);

const memory = async () =>
  (await c.query(`SELECT op_id, status FROM sync_verdicts WHERE user_id = 'u1' ORDER BY op_id`)).rows
    .map((row) => `${row.op_id}:${row.status}`)
    .join(',') || 'пусто';

// Сцена 1. Пачка — одна транзакция, а память вердиктов пишется обычной вставкой:
// дубль ловится по коду ошибки, как учит глава про свою таблицу
console.log('\n-- пачка [A:1 повтор, A:2 новая], память пишется обычной вставкой');
await c.query('BEGIN');
for (const [opId, status] of [['A:1', 'applied'], ['A:2', 'applied']]) {
  try {
    const put = await c.query(`INSERT INTO sync_verdicts VALUES ('u1', $1, $2, NULL)`, [opId, status]);
    console.log(`  ${opId}: вставлено строк ${put.rowCount}`);
  } catch (err) {
    const where = err.constraint ? ` (${err.constraint})` : '';
    console.log(`  ${opId}: ${err.code}${where} ${err.message}`);
  }
}
const ended = await c.query('COMMIT');
console.log('  COMMIT ответил:', ended.command);
console.log('  в памяти:', await memory());

// Сцена 2. Та же пачка, но вставка помечена ON CONFLICT DO NOTHING RETURNING
console.log('\n-- та же пачка через ON CONFLICT (user_id, op_id) DO NOTHING RETURNING');
await c.query('BEGIN');
for (const [opId, status] of [['A:1', 'applied'], ['A:2', 'applied']]) {
  const put = await c.query(
    `INSERT INTO sync_verdicts VALUES ('u1', $1, $2, NULL)
     ON CONFLICT (user_id, op_id) DO NOTHING RETURNING status`,
    [opId, status],
  );
  console.log(`  ${opId}: вставлено строк ${put.rowCount} | вернулось: ${put.rows[0]?.status ?? '—'}`);
}
await c.query('COMMIT');
console.log('  в памяти:', await memory());

// Сцена 3. Отметки [at, opId] сравниваются в JavaScript, а не в SQL: при равном
// времени порядок решает opId — обычная строка, а у строк есть правило сравнения
console.log('\n-- порядок двух имён операций: B:1 и a:1');
const collation = (await c.query(`SELECT datcollate FROM pg_database WHERE datname = current_database()`)).rows[0].datcollate;
console.log('  правило сравнения строк этой базы:', collation);
console.log('  в JavaScript "B:1" < "a:1" →', 'B:1' < 'a:1');
const compare = async (suffix) =>
  (await c.query(`SELECT 'B:1'${suffix} < 'a:1'${suffix} AS less`)).rows[0].less;
console.log('  в базе как есть →', await compare(''));
console.log('  в базе COLLATE "C" →', await compare(' COLLATE "C"'));
const other = (await c.query(
  `SELECT collname FROM pg_collation WHERE collname IN ('en_US.UTF-8', 'en_US.utf8') ORDER BY collname LIMIT 1`,
)).rows[0]?.collname;
console.log(`  в базе COLLATE "${other}" →`, await compare(` COLLATE "${other}"`));

// Сцена 4. Время отметки: в jsonb рядом со строкой и отдельной колонкой bigint
console.log('\n-- откуда приезжает at отметки');
await c.query(`CREATE TABLE sync_rows (row_id text PRIMARY KEY, marks jsonb NOT NULL, at_column bigint NOT NULL)`);
await c.query(`INSERT INTO sync_rows VALUES ('A-1', $1, 1757577000123)`, [
  JSON.stringify({ weightKg: { at: 1757577000123, opId: 'A:3' } }),
]);
const { rows: [stored] } = await c.query(`SELECT marks -> 'weightKg' -> 'at' AS from_jsonb, at_column FROM sync_rows`);
console.log('  из jsonb:', stored.from_jsonb, '| тип значения от драйвера:', typeof stored.from_jsonb);
console.log('  из колонки bigint:', stored.at_column, '| тип значения от драйвера:', typeof stored.at_column);

await c.end();

база: postgres://postgres@127.0.0.1:<порт-базы>/app

-- пачка [A:1 повтор, A:2 новая], память пишется обычной вставкой
  A:1: 23505 (sync_verdicts_pkey) duplicate key value violates unique constraint "sync_verdicts_pkey"
  A:2: 25P02 current transaction is aborted, commands ignored until end of transaction block
  COMMIT ответил: ROLLBACK
  в памяти: A:1:applied

-- та же пачка через ON CONFLICT (user_id, op_id) DO NOTHING RETURNING
  A:1: вставлено строк 0 | вернулось: —
  A:2: вставлено строк 1 | вернулось: applied
  в памяти: A:1:applied,A:2:applied

-- порядок двух имён операций: B:1 и a:1
  правило сравнения строк этой базы: C
  в JavaScript "B:1" < "a:1" → true
  в базе как есть → true
  в базе COLLATE "C" → true
  в базе COLLATE "en_US.UTF-8" → false

-- откуда приезжает at отметки
  из jsonb: 1757577000123 | тип значения от драйвера: number
  из колонки bigint: 1757577000123 | тип значения от драйвера: string

Первая вставка дала ровно то, что обещала глава про таблицу: код 23505 и имя нарушенного ограничения. А вторая не дала ничего — 25P02, «current transaction is aborted, commands ignored until end of transaction block». После ошибки Postgres не выполняет в этом блоке ничего: команды принимаются и отбрасываются до конца транзакции. И COMMIT в такой транзакции коммитом не является — драйвер получил на него ROLLBACK. Пачка из двух операций, где сломана была первая, целиком не состоялась.

Вот почему повтор — не «редкий случай, который можно обработать ошибкой». Повтор на телефоне — это норма: потерянный ответ случается каждый раз, когда сеть пропала между запросом и ответом, а первой в повторной пачке стоит как раз та операция, чей ответ пропал. Сервер, который ловит дубль исключением, превращает обычный день в очередь, которая не проходит никогда.

Вторая половина прогона — та же пачка, но каждая вставка помечена ON CONFLICT (user_id, op_id) DO NOTHING RETURNING. Повтор вставил ноль строк и ничего не вернул, соседняя операция прошла, в памяти стало две записи. Ноль возвращённых строк — это не ошибка и не отказ, это ответ: «такая запись уже есть». Ровно тот же приём, что ноль изменённых строк у обновления с условием, и читается он так же — не как сбой, а как развилка.

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

Ноль изменённых строк — это ответ, а не ошибка. Там, где ответ пытаются получить исключением, транзакции не выживают.

Память устроена, форма записи выбрана. Остался порядок: что сервер делает с операцией и в какой последовательности. Шагов три, и все три идут внутри транзакции пачки, под уже занятой строкой-счётчиком.

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

У шага есть уточнение, которое легко пропустить, а стоит оно целого правила. Если строка уже есть и ни одна её отметка не сдвинулась, вердикт — applied без всякой проверки: строка осталась той же, что уже проходила проверку, и гонять её второй раз не нужно. Номера такая операция тоже не получает — менять в строке нечего.

Соберём этот шаг вместе с третьим, чтобы видеть сразу и вердикты, и номера. Слияние по полям и полный порядок правок берутся с подложки — это тот самый код части V, общий у телефона и сервера. Пачка идёт длинная: вставка, чужая правка веса с планшета (она позже по времени), своя правка веса (она раньше), правка строки, которой нет, опечатка в повторах и обычная правка после неё. Прогноз: какой вердикт получит каждая операция, кому достанется номер и что в конце окажется в строке?

type Fields = Record<string, string | number | boolean>;
type Op = { opId: string; rowId: string; fields: Fields; at: number };
type Mark = { at: number; opId: string };
type Marks = Record<string, Mark>;
type Row = Fields & { id: string };
type Verdict = { opId: string; status: "applied" | "duplicate" | "rejected"; reason?: string };

// слияние по полям и полный порядок правок — из части V, целиком на подложке
declare function merge(row: Fields | undefined, op: Op, mark: Marks): Row;

// строка итога: поля, отметки [at, opId] по полям и номер последней правки
type Stored = { fields: Fields; marks: Marks; seq: number };
const rows = new Map<string, Stored>();
let head = 0;

// проверяется СТРОКА после применения операции, а не поля самой операции
function checkRow(row: Row): string | null {
  const missing = ["exercise", "weightKg", "reps"].filter((field) => !(field in row));
  if (missing.length > 0) return "нет полей: " + missing.join(", ");
  return Number(row.reps) > 100 ? "повторов больше сотни" : null;
}

// Шаг первый: вердикт считается в JavaScript, в базу не пишется ничего
function decide(op: Op): { verdict: Verdict; moved: boolean; row: Row; marks: Marks } {
  const stored = rows.get(op.rowId);
  const before = stored?.marks ?? {};
  const marks: Marks = { ...before };
  const row = merge(stored?.fields, op, marks);
  let moved = JSON.stringify(marks) !== JSON.stringify(before);
  // у существующей строки ни одна отметка не сдвинулась: эта строка уже проверена
  const reason = stored && !moved ? null : checkRow(row);
  if (reason) moved = false;
  const verdict: Verdict = reason
    ? { opId: op.opId, status: "rejected", reason }
    : { opId: op.opId, status: "applied" };
  return { verdict, moved, row, marks };
}

// Шаг третий: строка итога пишется, и номер выдаётся только сдвинувшей отметку
function accept(op: Op): Verdict {
  const { verdict, moved, row, marks } = decide(op);
  if (!moved) return verdict;
  head += 1;
  const { id, ...fields } = row;
  rows.set(op.rowId, { fields, marks, seq: head });
  return verdict;
}

const batch: Op[] = [
  { opId: "A:1", rowId: "A-1", fields: { exercise: "жим", weightKg: 80, reps: 5 }, at: 1 },
  // вес той же строки правили с планшета, и правка эта позже
  { opId: "B:1", rowId: "A-1", fields: { weightKg: 90 }, at: 5 },
  { opId: "A:2", rowId: "A-1", fields: { weightKg: 85 }, at: 3 },
  { opId: "A:3", rowId: "A-2", fields: { weightKg: 60 }, at: 6 },
  { opId: "A:4", rowId: "A-1", fields: { reps: 800 }, at: 7 },
  { opId: "A:5", rowId: "A-1", fields: { reps: 8 }, at: 8 },
];

for (const op of batch) {
  const verdict = accept(op);
  console.log(op.opId, verdict.status + (verdict.reason ? " · " + verdict.reason : ""), "| номер счётчика:", head);
}
const kept = rows.get("A-1");
if (kept) {
  console.log("строка A-1:", JSON.stringify(kept.fields), "| её номер:", kept.seq);
  console.log("отметка веса:", JSON.stringify(kept.marks.weightKg));
}
console.log("строк у сервера:", rows.size);

Разберём по строкам. A:1 завела строку — номер один. B:1 с планшета подняла вес до девяноста — номер два, отметка веса теперь её. A:2 правит то же поле — вес восемьдесят пять, — но пришла с меньшим временем и проиграла: ни одна отметка не сдвинулась, вердикт applied, номер не выдан. A:3 правит строку A-2, которой у сервера нет, — результат наложения неполон, и отказ называет отсутствующие поля. A:4 с восемьюстами повторами отвергнута по строке после применения, а строка при этом осталась целой: у неё по-прежнему пять повторов. И A:5, пришедшая сразу за ядовитой, разобрана как обычно — номер три.

Шаг второй — память. Вердикт вычислен, но ещё не записан; записывается он той самой вставкой с ON CONFLICT DO NOTHING RETURNING. Шаг третий — строка. Вернулась строка — операция новая: сервер берёт номер у счётчика (и только если отметка сдвинулась) и пишет строку итога. Ноль строк — операция уже была: сервер читает сохранённый вердикт, отдаёт applied как duplicate, rejected — тем же rejected, и строку не пишет.

На Drizzle весь приём операции — вот столько:

// внутри транзакции пачки, под взятой строкой-счётчиком

// 1. Вердикт вычислен в JavaScript: merge на копии отметок и проверка строки
//    после применения. Ни одной записи в базу на этом шаге нет
const { verdict, moved, fields, marks } = decide(stored, op);

// 2. Память вердиктов — тот же вызов, что в листинге части V
const [saved] = await tx
  .insert(syncVerdicts)
  .values({ userId, opId: op.opId, status: verdict.status, reason: verdict.reason })
  .onConflictDoNothing()
  .returning({ status: syncVerdicts.status });

// 3a. Строк не вернулось — операция уже была: отдаём сохранённый ответ и ничего
//     не пишем. readVerdict заменяет applied на duplicate
if (!saved) return readVerdict(tx, userId, op.opId);

// 3b. Номер получает только операция, сдвинувшая хоть одну отметку
if (moved) {
  const [head] = await tx
    .update(syncHeads)
    .set({ seq: sql`${syncHeads.seq} + 1` })
    .where(eq(syncHeads.userId, userId))
    .returning({ seq: syncHeads.seq });
  const seq = Number(head.seq);
  await tx
    .insert(syncRows)
    .values({ userId, rowId: op.rowId, fields, marks, seq })
    .onConflictDoUpdate({ target: [syncRows.userId, syncRows.rowId], set: { fields, marks, seq } });
}
return verdict;

Точек сохранения здесь нет, и это решение, а не упущение. SAVEPOINT нужен, когда часть транзакции надо откатить, — а откатывать нечего: до вставки памяти сервер не записал ни байта. Заводить точку сохранения на каждую операцию к тому же дорого: на пачке из тысячи операций переполняется кеш подтранзакций (он держит 64 номера), и та же работа занимает вдвое больше времени.

Теперь ловушка, ради которой этот порядок и выбран. Поменяем два последних шага местами: сначала записать строку, потом вставить память. Пока правило проверки не меняется, разницы не видно — оба порядка дают одни и те же вердикты. Разница появляется после выката, который правило смягчил: вчера подход с полутора сотнями повторов отвергали, сегодня принимают. Отвергнутая вчера операция, чей ответ потерялся, приезжает снова.

Прогноз к прогону: две копии одной пачки, ушедшие одна за другой, — какие вердикты получит каждая и каким станет номер счётчика? И вторая сцена: повтор отвергнутой операции после смягчения правила — что ответит сервер и останется ли у него строка при каждом из двух порядков?

import pg from 'pg';

// адрес базы — первой строкой и без пароля: видно, куда ушли запросы
const url = new URL(process.env.DATABASE_URL);
url.password = '';
console.log('база:', url.href);

const cfg = { connectionString: process.env.DATABASE_URL };
const phone = new pg.Client(cfg);
const web = new pg.Client(cfg);
const reader = new pg.Client(cfg);
await Promise.all([phone.connect(), web.connect(), reader.connect()]);
console.log('уровень изоляции:', (await phone.query('SHOW transaction_isolation')).rows[0].transaction_isolation);

await reader.query(`CREATE TABLE sync_heads (user_id text PRIMARY KEY, seq bigint NOT NULL DEFAULT 0);
  CREATE TABLE sync_rows (user_id text, row_id text, fields jsonb NOT NULL, marks jsonb NOT NULL, seq bigint,
    PRIMARY KEY (user_id, row_id));
  CREATE TABLE sync_verdicts (user_id text, op_id text, status text NOT NULL, reason text,
    PRIMARY KEY (user_id, op_id))`);

// порядок правок полный: сначала время, потом opId — правило телефона, дословно
const newer = (a, b) => (a.at !== b.at ? a.at > b.at : a.opId > b.opId);
// слияние по полям: поле меняется, только если правка новее его отметки
const merge = (row, op, mark) => {
  const next = { ...row };
  for (const field of Object.keys(op.fields)) {
    const was = mark[field];
    if (was && !newer(op, was)) continue;
    next[field] = op.fields[field];
    mark[field] = { at: op.at, opId: op.opId };
  }
  return next;
};
// проверка строки ПОСЛЕ применения операции, а не полей самой операции
const strict = (row) => {
  const missing = ['exercise', 'weightKg', 'reps'].filter((field) => !(field in row));
  if (missing.length) return 'нет полей: ' + missing.join(', ');
  return Number(row.reps) > 100 ? 'повторов больше сотни' : null;
};
// выкат, смягчивший правило: до тысячи повторов теперь можно
const relaxed = (row) => {
  const reason = strict(row);
  return reason === 'повторов больше сотни' && Number(row.reps) <= 1000 ? null : reason;
};

const pidOf = async (client) => (await client.query('SELECT pg_backend_pid() AS pid')).rows[0].pid;
// номера обслуживающих процессов спрашиваем заранее: запросы одного соединения
// драйвер выстраивает в очередь, и у занятого ожиданием клиента спросить нечего
const pids = { phone: await pidOf(phone), web: await pidOf(web) };
const waitsForLock = async (pid) => {
  for (let i = 0; i < 400; i += 1) {
    const { rows } = await reader.query('SELECT wait_event_type FROM pg_stat_activity WHERE pid = $1', [pid]);
    if (rows[0]?.wait_event_type === 'Lock') return true;
    await new Promise((resolve) => setTimeout(resolve, 25));
  }
  return false;
};

/*
 * Приём пачки. Порядок шагов — предмет опыта:
 *   'память' — вердикт вычислен в JS без записи, потом вставка памяти, и строка
 *              итога пишется, только если вставка вернула строку;
 *   'строка' — строка итога пишется до вставки памяти.
 */
async function push(client, user, batch, options = {}) {
  const { order = 'память', check = strict, afterLock, beforeCommit } = options;
  const verdicts = [];
  await client.query('BEGIN');
  await client.query('INSERT INTO sync_heads (user_id) VALUES ($1) ON CONFLICT DO NOTHING', [user]);
  await client.query('SELECT seq FROM sync_heads WHERE user_id = $1 FOR UPDATE', [user]);
  if (afterLock) afterLock();

  const saved = async (opId) => {
    const { rows: [v] } = await client.query(
      'SELECT status, reason FROM sync_verdicts WHERE user_id = $1 AND op_id = $2', [user, opId]);
    return v.status === 'applied' ? { opId, status: 'duplicate' } : { opId, status: v.status, reason: v.reason };
  };

  for (const op of batch.slice(0, 1000)) {
    const { rows: [stored] } = await client.query(
      'SELECT fields, marks FROM sync_rows WHERE user_id = $1 AND row_id = $2', [user, op.rowId]);
    const before = stored?.marks ?? {};
    const marks = { ...before };
    const next = merge(stored?.fields, op, marks);
    let moved = JSON.stringify(marks) !== JSON.stringify(before);
    // у существующей строки ни одна отметка не сдвинулась — строка та же, что уже проверена
    const reason = stored && !moved ? null : check(next);
    if (reason) moved = false;
    const computed = reason ? { opId: op.opId, status: 'rejected', reason } : { opId: op.opId, status: 'applied' };

    const writeRow = async () => {
      if (!moved) return;
      const { rows: [head] } = await client.query(
        'UPDATE sync_heads SET seq = seq + 1 WHERE user_id = $1 RETURNING seq', [user]);
      await client.query(
        `INSERT INTO sync_rows (user_id, row_id, fields, marks, seq) VALUES ($1, $2, $3, $4, $5)
         ON CONFLICT (user_id, row_id) DO UPDATE SET fields = excluded.fields, marks = excluded.marks, seq = excluded.seq`,
        [user, op.rowId, next, marks, Number(head.seq)]);
    };
    const remember = () => client.query(
      `INSERT INTO sync_verdicts (user_id, op_id, status, reason) VALUES ($1, $2, $3, $4)
       ON CONFLICT DO NOTHING RETURNING status`,
      [user, op.opId, computed.status, computed.reason ?? null]);

    if (order === 'память') {
      const put = await remember();
      if (put.rowCount) { await writeRow(); verdicts.push(computed); } else verdicts.push(await saved(op.opId));
    } else {
      await writeRow();
      const put = await remember();
      verdicts.push(put.rowCount ? computed : await saved(op.opId));
    }
  }
  if (beforeCommit) await beforeCommit();
  await client.query('COMMIT');
  return verdicts;
}

const short = (list) => list.map((v) => v.status).join('/');
const withReason = (v) => v.status + (v.reason ? ' · ' + v.reason : '');
const sorted = (fields) => JSON.stringify(Object.fromEntries(Object.keys(fields).sort().map((k) => [k, fields[k]])));
const serverRows = async (user) =>
  (await reader.query('SELECT row_id, fields, seq FROM sync_rows WHERE user_id = $1 ORDER BY seq', [user])).rows
    .map((row) => `${row.row_id}#${row.seq} ${sorted(row.fields)}`)
    .join(' ') || '[]';
const head = async (user) =>
  (await reader.query('SELECT seq FROM sync_heads WHERE user_id = $1', [user])).rows[0]?.seq ?? 'строки нет';
const remembered = async (user, opId) =>
  (await reader.query('SELECT status FROM sync_verdicts WHERE user_id = $1 AND op_id = $2', [user, opId])).rows[0].status;

// Сцена 1. Две копии одной пачки одновременно: ответ на первую потерялся, и
// телефон отправил её заново, не дождавшись. Строку-счётчик завела регистрация
await reader.query(`INSERT INTO sync_heads (user_id) VALUES ('u1')`);
const batch = [
  { opId: 'A:1', rowId: 'A-1', fields: { exercise: 'жим', weightKg: 80, reps: 5 }, at: 1 },
  { opId: 'A:2', rowId: 'A-1', fields: { reps: 800 }, at: 2 },
  { opId: 'A:3', rowId: 'A-1', fields: { weightKg: 85 }, at: 3 },
];
let waited = false;
let letWebIn = () => {};
const webTurn = new Promise((resolve) => { letWebIn = resolve; });
// вторая копия отправляется следом за первой, а не наперегонки с ней: порядок
// двух отправок — предмет урока, и жребию его отдавать нельзя
const first = push(phone, 'u1', batch, {
  afterLock: () => letWebIn(),
  beforeCommit: async () => { waited = await waitsForLock(pids.web); },
});
const second = webTurn.then(() => push(web, 'u1', batch));
const [one, two] = await Promise.all([first, second]);
console.log('\n-- две копии пачки [A:1 вставка, A:2 опечатка, A:3 правка веса] одновременно');
console.log('  первая копия:', short(one), '| вторая копия:', short(two));
console.log('  причина отказа:', one[1].reason, '| та же у второй копии:', two[1].reason === one[1].reason);
console.log('  вторая ждала первую:', waited);
console.log('  номер счётчика:', await head('u1'), '| строки сервера:', await serverRows('u1'));

// Сцена 2. Отвергнутая операция, ответ на которую потерялся, а правило проверки
// успели смягчить выкатом. Сцена проходит дважды: порядком «вычислить → память →
// строка» и обратным — «вычислить → строка → память»
const late = { opId: 'B:1', rowId: 'B-1', fields: { exercise: 'присед', weightKg: 100, reps: 150 }, at: 1 };
for (const [user, order] of [['u2', 'память'], ['u3', 'строка']]) {
  const before = await push(phone, user, [late], { order, check: strict });
  const after = await push(phone, user, [late], { order, check: relaxed });
  console.log(`\n-- порядок «${order === 'память' ? 'вычислить → память → строка' : 'вычислить → строка → память'}»`);
  console.log('  при строгом правиле:', withReason(before[0]));
  console.log('  повтор после смягчения:', withReason(after[0]), '| в памяти:', await remembered(user, 'B:1'));
  console.log('  номер счётчика:', await head(user), '| строки сервера:', await serverRows(user));
}

await Promise.all([phone.end(), web.end(), reader.end()]);

база: postgres://postgres@127.0.0.1:<порт-базы>/app
уровень изоляции: read committed

-- две копии пачки [A:1 вставка, A:2 опечатка, A:3 правка веса] одновременно
  первая копия: applied/rejected/applied | вторая копия: duplicate/rejected/duplicate
  причина отказа: повторов больше сотни | та же у второй копии: true
  вторая ждала первую: true
  номер счётчика: 2 | строки сервера: A-1#2 {"exercise":"жим","reps":5,"weightKg":85}

-- порядок «вычислить → память → строка»
  при строгом правиле: rejected · повторов больше сотни
  повтор после смягчения: rejected · повторов больше сотни | в памяти: rejected
  номер счётчика: 0 | строки сервера: []

-- порядок «вычислить → строка → память»
  при строгом правиле: rejected · повторов больше сотни
  повтор после смягчения: rejected · повторов больше сотни | в памяти: rejected
  номер счётчика: 1 | строки сервера: B-1#1 {"exercise":"присед","reps":150,"weightKg":100}

Первая сцена вышла скучной, и это лучшая новость статьи: первая копия получила applied/rejected/applied, вторая — duplicate/rejected/duplicate с той же причиной, номер счётчика два, строка одна. Вторая копия при этом честно дождалась первой — строку-счётчик держала транзакция, и это видно по самой базе. Ни дублей, ни потерянных правок, ни взаимных блокировок.

А вторая сцена показывает разницу порядков. При «память, потом строка» повтор получил тот же rejected с той же причиной, номер счётчика остался нулём, строк у сервера нет — ровно то, что есть у телефона: он выбросил эту операцию из очереди и пересобрал строку без неё. При «строка, потом память» повтор тоже получил rejected — память-то не менялась, — но строку сервер успел записать до того, как узнал, что вердикт уже сохранён. У сервера подход есть, у телефона его нет. Обратно телефону он и не приедет: свои операции клиент узнаёт по префиксу имени и со страницы пропускает. А вот второе устройство того же человека заберёт подход с курсора как чужую правку и покажет — с этого дня у двух телефонов одного человека разные дневники. Один и тот же вердикт, два разных состояния.

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

Сначала запомнить ответ, потом менять данные. Обратный порядок работает ровно до того выката, который поменяет правило проверки.

Форма ответа не меняется с части V: PushInput { batch } на входе, PushOutput { verdicts } на выходе, и каждый вердикт — про свою операцию. Код ошибки HTTP при этом остаётся про запрос целиком: нет токена, тело не разобралось, тело больше предела. Почему отказ по одной операции не 4xx, часть V показала прогоном: строгий сервер запирает две нормальные записи третьей, и очередь не проходит никогда.

Одну строку той таблицы вердиктов пора уточнить. Там сказано, что applied значит «операция легла в журнал сервера и получила номер», — и первое верно, а второе не всегда. Операция, все поля которой проиграли более новым отметкам, принята и никакого номера не получает: в строке от неё ничего не изменилось, и ни одному устройству эта операция не нужна. Для телефона разницы нет — он убирает из очереди и applied, и duplicate, — но для сервера разница есть, и именно она не даёт номерам расти на пустом месте.

У rejected в части VI появляется ещё одно поле — строка целиком, записями по отметкам. Зачем она там и почему без неё телефон теряет чужие правки, разбирает статья про итог строк и надгробия: поле необязательное, и старый клиент его просто не видит.

Остаётся длина пачки. Телефон, месяц проживший без сети, отправит очередь целиком — и на повтор отправит её же. Сервер разбирает за один запрос не больше тысячи операций и отдаёт вердикты только им; остаток телефон дошлёт сам, потому что операция без вердикта остаётся в очереди неотправленной. Проверим это на настоящем клиенте части V — он для такого случая не правился ни на строку. Прогноз: сколько операций уйдёт в первом заходе, сколько вердиктов придёт и что останется в очереди после второго?

type Fields = Record<string, string | number>;
type Op = { opId: string; rowId: string; fields: Fields; at: number };
type Entry = Op & { seq: number };
type Verdict = { opId: string; status: "applied" | "duplicate" | "rejected"; reason?: string };
type Page = { changes: Entry[]; cursor: number; hasMore: boolean };
interface Connection {
  push(batch: Op[]): Promise<{ verdicts: Verdict[] }>;
  pull(cursor: number, limit: number): Promise<Page>;
}
interface Server {
  as(user: string): Connection;
}
interface Client {
  journal: { newId(): string; change(rowId: string, fields: Fields): void };
  queue(): Op[];
  sync(): Promise<Verdict[]>;
  readonly cursor: number;
}
declare function createServer(): Server;
declare function createClient(device: string, server: Connection): Client;

const MAX_OPS = 1000;
let offered = 0;
let judged = 0;

// Сервер разбирает не больше MAX_OPS операций за запрос и отдаёт вердикты
// только им. Клиент части V при этом не меняется — на нём и проверим
function limited(connection: Connection): Connection {
  return {
    async push(batch: Op[]): Promise<{ verdicts: Verdict[] }> {
      offered = batch.length;
      const answer = await connection.push(batch.slice(0, MAX_OPS));
      judged = answer.verdicts.length;
      return answer;
    },
    pull: (cursor: number, limit: number) => connection.pull(cursor, limit),
  };
}

async function main(): Promise<void> {
  const server = createServer();
  const phone = createClient("A", limited(server.as("u1")));
  const rowId = phone.journal.newId();
  phone.journal.change(rowId, { exercise: "жим", weightKg: 80, reps: 5 });
  // месяц без сети: очередь телефона длиннее, чем сервер берёт за один раз
  for (let n = 1; n < 1200; n += 1) phone.journal.change(rowId, { weightKg: 80 + (n % 20) });
  console.log("в очереди до отправки:", phone.queue().length);

  await phone.sync();
  console.log("первый заход: отправлено", offered, "| вердиктов", judged, "| в очереди", phone.queue().length, "| курсор", phone.cursor);
  await phone.sync();
  console.log("второй заход: отправлено", offered, "| вердиктов", judged, "| в очереди", phone.queue().length, "| курсор", phone.cursor);
}

main();

Тысяча двести ушло, тысяча разобрана, двести осталось; второй заход забрал остаток, и очередь пуста. Курсор при этом двигался обычным порядком: свои операции клиент узнаёт по префиксу имени и на экран их второй раз не кладёт.

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

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

Успех с пустым списком вердиктов клиент читает как приглашение зайти снова. Если разбирать нечего — это ошибка запроса, а не успех.

В первом прогоне статьи есть третья сцена, к памяти вердиктов отношения не имеющая, но объясняющая, почему вердикт считается в JavaScript, а не запросом. Правило порядка правок из части V — полное: сначала время at, а при равном времени имя операции. Имя — обычная строка, и сравнивают строки не все одинаково.

Прогон печатает четыре ответа на один вопрос: «B:1» меньше «a:1» или нет? В JavaScript — да, строки сравниваются по кодам знаков, и заглавная «B» идёт раньше строчной «a». В базе, поднятой с правилом сравнения C, — тоже да, по той же причине. С правилом en_US.UTF-8 — нет: обычные локали сравнивают буквы по алфавиту, а регистр учитывают в последнюю очередь. Один и тот же порядок, посчитанный двумя местами, выходит разным.

Отсюда правило: сравнение отметок живёт в JavaScript целиком. Сервер читает строку и её отметки, накладывает операцию тем же кодом, что телефон, и решает всё сам; база при этом отбирает строки и сортирует их по номеру — а номер целое число, и правило сравнения строк к нему не относится. Соблазн переложить слияние в SQL (jsonb, case, «взять самую новую отметку одним запросом») стоит пережить: тогда правило окажется зависящим от локали кластера, и сервер, поднятый в другой стране, начнёт расходиться с телефоном на операциях с одинаковым временем.

Четвёртая сцена того же прогона — про at. В отметке он лежит числом внутри jsonb, и оттуда приезжает числом: jsonb у драйвера разбирается как JSON, а в JSON число — это число. Но если тому же at отвести отдельную колонку bigint, приедет он строкой — так же, как номер в прошлой статье. И сломается на этом не равенство, а порядок: правило сравнит «10» с «9», получит ложь, и правка, сделанная позже, проиграет более ранней. Поэтому at и держат в отметках, а если он всё-таки оказался колонкой — приводят к числу на границе чтения, один раз.

Второе, что стоит сказать про границу: fields у операции — не «любой объект, приехавший от клиента». Пачка приходит с провода, и разбирает её тот же валидатор на границе каркаса, что и остальные входы: имя операции — строка, at — число, поля — известного вида. Всё, что описано в схеме, доезжает до приёма; всё, что не описано, не доезжает вовсе. Сервер, который получает at строкой от доверчивого разбора, ломается ровно так же, как на колонке bigint, — только искать причину придётся дольше.

Если сравнение решает, чья правка новее, сравнивать надо там, где правило сравнения выбрали вы, а не установщик кластера.