data/world-inventory.json

@@ -63,37 +63,37 @@
     {
       "day": 0,
       "date": "2026-06-01",
-      "sha256": "6c3e3c1c8689ec23f53f0953db1b941e6f010e9e56a74d8d017f40195d3dec17"
+      "sha256": "93b39f261cbbe894402ebe0525460fed1809e5712639be4b73b79dac2b39f5bd"
     },
     {
       "day": 1,
       "date": "2026-06-02",
-      "sha256": "9acb523cfd47f3d97d55122025852982ba5a1d8362036308ce72a4d9fb767f43"
+      "sha256": "4fef382cd84868ccfb61a9d8610cdd26f9a749bbae8b20800c45e520e9ce414e"
     },
     {
       "day": 2,
       "date": "2026-06-03",
-      "sha256": "b81482491c4c0b874ea84d5f3f05144100c906dacc059c068b331df258b97172"
+      "sha256": "caf9ed2e77d2fcc99066a548dcba60f61f1ff0d4402de31457a06f23fd2d8306"
     },
     {
       "day": 3,
       "date": "2026-06-04",
-      "sha256": "9d1ab326c4bf8b78c9fcaf7015d8029f9ad4cff5558b89c730eb428d175cc9ee"
+      "sha256": "cdb9e4e00b57c04985866b5a851a8e2f23b3e9205ea131c7a5076672ee8b5669"
     },
     {
       "day": 4,
       "date": "2026-06-05",
-      "sha256": "0e30d0e192e121121003c0cc9aaa55e24924afb983caff15ab105705fbf2bd00"
+      "sha256": "c58d2c20fe3b4cee2973f34c322090dbb66d3793bfaab0337c22b53fb5a3ec7f"
     },
     {
       "day": 5,
       "date": "2026-06-06",
-      "sha256": "9a2ad812fcff9a40a988f24f8aa4fc7a379873d173a95ddacd438592aed0bfb5"
+      "sha256": "bbf28b1d375e4815695c72a5915a22a3c9b36195b97ec4cc7c739466dbe70b53"
     },
     {
       "day": 6,
       "date": "2026-06-07",
-      "sha256": "e1706a27ff55de489a505489148c7a959329d95ad6597d92d3f78505f67bb221"
+      "sha256": "8ca077576bcf014c0d492c7d7190e38c0e46ec95951ad840df93eb75ee5a00d3"
     }
   ]
 }

docs/architecture/orders/fate.md

@@ -127,18 +127,20 @@
 подделывается — строка создана, когда заказ родился, — но в слепках дней
 d…d+δ−1 её нет, а с d+δ она появляется в том состоянии, до которого заказ
 дожил: отменённый на второй день и опоздавший на третий приедет в первом же
-своём слепке как `cancelled` — «выгрузка догоняет жизнь». Задержка — таблица
-весов из трёх строк: 0 на подавляющем весе, 1 и 2 — это и есть «D+1/D+2»
-мастер-спеки. Меряется она в днях снятия слепка, а не отправки: иначе сдвиг
-отправки удвоился бы, и обещанные D+1/D+2 стали бы D+2/D+3. Дальше таблица не
-идёт: заказ с δ = 6 приехал бы ровно в одном слепке, и обещание «пропущенный
-день ничего не ломает» на нём перестало бы быть верным; при δ ≤ 2 у всякого
-заказа слепков не меньше пяти. Легенда: заказ ушёл в ручную обработку и попал
-в выгрузку позже. Отклонено: *сдвиг `created_at`* — подделка аудита источника:
-день создания строки разошёлся бы с днём покупки, чьё равенство держит
-синхронная модель ([откуда берётся заказ](snapshot.md)), заказ уехал бы в
-чужую партицию, и «выручка дня D» перестала бы отвечать покупкам дня D —
-сломалась бы та самая сверка, ради которой всё строится.
+своём слепке как `cancelled` — «выгрузка догоняет жизнь». Таблица задержки
+состоит из трёх строк: δ = 0/1/2 с весами 90/5/5; 1 и 2 — это D+1 и D+2
+мастер-спеки. Бросок идёт последним в подпотоке `ORDERS`; затем δ назначенного
+планом заказа обнуляется. Меряется задержка в днях снятия слепка, а не
+отправки: иначе сдвиг отправки удвоился бы, и обещанные D+1/D+2 стали бы
+D+2/D+3. Дальше таблица не идёт: заказ с δ = 6 приехал бы ровно в одном
+слепке, и обещание «пропущенный день ничего не ломает» на нём перестало бы
+быть верным; при δ ≤ 2 у всякого заказа слепков не меньше пяти.
+Легенда: заказ ушёл в ручную обработку и попал в выгрузку позже. Отклонено:
+*сдвиг `created_at`* — подделка аудита источника: день создания строки
+разошёлся бы с днём покупки, чьё равенство держит синхронная модель
+([откуда берётся заказ](snapshot.md)), заказ уехал бы в чужую партицию,
+и «выручка дня D» перестала бы отвечать покупкам дня D — сломалась бы та
+самая сверка, ради которой всё строится.
 
 **Пересечения.** Броски независимы, пересечения выходят арифметикой, приоритет
 мастер-спеки работает по-настоящему. Исключений два, и оба названы выше: дубль

docs/architecture/orders/snapshot.md

@@ -60,12 +60,14 @@
 топик льёт даг* — второй путь доставки и второй сериализатор; *слепок едет
 топиком `hits`* — убивает два режима приёма.
 
-**Сборка окна.** Слепок дня D несёт заказы, рождённые в дни D−6…D, и
-собирается переигровкой этих семи дней: заказы дня — производная всей воронки
-дня, дешёвого пути к ним нет. Цена — семь проигрышей дня (~14 с) на слепок; у
-начала оси окно усекается само. Отклонено: *кэш заказов на томе* — состояние
-между прогонами; *окно держит хранилище* — топик перестаёт нести слепок;
-*K = 1* — это страховочный срез 1 мастер-спеки, он в резерве.
+**Сборка окна.** Слепок дня D берет заказы, рожденные в дни D−6…D, а несет из
+них те, что уже попали в выгрузку: механика опоздания живет в
+[судьбе заказа](fate.md). Окно собирается переигровкой этих семи дней: заказы
+дня — производная всей воронки дня, дешёвого пути к ним нет. Цена — семь
+проигрышей дня (~14 с) на слепок; у начала оси окно усекается само. Отклонено:
+*кэш заказов на томе* — состояние между прогонами; *окно держит хранилище* —
+топик перестаёт нести слепок; *K = 1* — это страховочный срез 1 мастер-спеки,
+он в резерве.
 
 **Отправка.** Слепок **снимается на границе суток, а отправляется следующим
 прогоном**: даг, играющий день D, отправляет слепок дня D−1 — ночная выгрузка

generator/src/clickstream_generator/commerce.py

@@ -186,14 +186,18 @@ class _Events:
     basket: NDArray[np.int64]
 
 
-# Что торговая половина отдает дню: поток, покупки и их событийные исходы.
-Woven = tuple[
-    dict[str, NDArray[Any]],
-    NDArray[np.uint8],
-    NDArray[np.int64],
-    Purchases,
-    NDArray[np.int8],
-]
+@dataclass(frozen=True, slots=True)
+class Woven:
+    """Что торговая половина отдает дню после вплетения событий."""
+
+    columns: dict[str, NDArray[Any]]
+    page: NDArray[np.uint8]
+    product: NDArray[np.int64]
+    purchases: Purchases
+    event_outcome: NDArray[np.int8]
+    # Для каждой покупки: назначен ли её заказ планом.
+    assigned_order: NDArray[np.bool_]
+
 
 # День без единой корзины: покупок в нём нет, а форма у рядов есть.
 _NO_PURCHASES = Purchases(
@@ -206,6 +210,7 @@ _NO_PURCHASES = Purchases(
     moment=np.empty(0, dtype="datetime64[s]"),
 )
 _NO_EVENT_OUTCOMES = np.empty(0, dtype=np.int8)
+_NO_ASSIGNED_ORDERS = np.empty(0, dtype=np.bool_)
 
 
 def weave(
@@ -230,7 +235,14 @@ def weave(
     rng = day_stream(seed, day, Component.COMMERCE)
     baskets = _baskets(rng, page, product, columns["VisitID"])
     if not len(baskets):
-        return columns, page, product, _NO_PURCHASES, _NO_EVENT_OUTCOMES
+        return Woven(
+            columns=columns,
+            page=page,
+            product=product,
+            purchases=_NO_PURCHASES,
+            event_outcome=_NO_EVENT_OUTCOMES,
+            assigned_order=_NO_ASSIGNED_ORDERS,
+        )
 
     goods = catalog.catalog()
     draws = _draws(rng, baskets)
@@ -247,7 +259,14 @@ def weave(
     columns, page, product, event_outcome = _corrupt_events(
         seed, day, columns, page, product, purchases, assigned
     )
-    return columns, page, product, purchases, event_outcome
+    return Woven(
+        columns=columns,
+        page=page,
+        product=product,
+        purchases=purchases,
+        event_outcome=event_outcome,
+        assigned_order=assigned,
+    )
 
 
 def _draws(rng: np.random.Generator, baskets: _Baskets) -> _Draws:

generator/src/clickstream_generator/day.py

@@ -165,7 +165,7 @@ def stream(seed: int, day: int) -> Day:
     # покупки дня — из них заказная сторона собирает заказы.
     person = np.repeat(audience.person_id[visits.cookie], visits.pages)
     assigned_order = np.repeat(visits.ordering, visits.pages)
-    columns, page, product, purchases, purchase_outcome = commerce.weave(
+    woven = commerce.weave(
         seed,
         day,
         {name: value[order] for name, value in columns.items()},
@@ -176,11 +176,11 @@ def stream(seed: int, day: int) -> Day:
     )
     return Day(
         day=day,
-        columns=columns,
-        page=page,
-        product=product,
-        orders=orders.of_day(seed, day, purchases),
-        purchase_outcome=purchase_outcome,
+        columns=woven.columns,
+        page=woven.page,
+        product=woven.product,
+        orders=orders.of_day(seed, day, woven.purchases, woven.assigned_order),
+        purchase_outcome=woven.event_outcome,
     )
 
 

generator/src/clickstream_generator/orders.py

@@ -24,11 +24,13 @@
 одной доли перебрасывала бы весь подпоток после себя
 (docs/architecture/orders/fate.md).
 
-**Судьба заказа — исход и его моменты**: из окна изменяемости заказ выходит
-оплаченным или отменённым, а когда именно это случилось, сказано смещением в
-секундах от рождения заказа. Момента, которого у исхода нет, нет и в данных:
-его место занимает −1, а не ноль, — иначе «оплатили в секунду рождения» было
-бы не отличить от «не оплатили вовсе».
+**Судьба заказа — исход, его моменты и задержка выгрузки**. Из окна
+изменяемости заказ выходит оплаченным или отменённым, а когда именно это
+случилось, сказано смещением в секундах от рождения заказа. Момента, которого
+у исхода нет, нет и в данных: его место занимает −1, а не ноль, — иначе
+«оплатили в секунду рождения» было бы не отличить от «не оплатили вовсе».
+Задержка говорит, сколько ранних слепков заказ пропустит; назначенный планом
+приезжает сразу.
 
 **Состояние на границе суток** — чтение готовой судьбы, а не накопление:
 слепок дня D учитывает моменты не позже границы D|D+1 и по ним называет
@@ -62,6 +64,7 @@ _DELIVERY_CUMULATIVE = np.cumsum(
 )
 _OUTCOME_CUMULATIVE = np.cumsum(world.ORDER_OUTCOME_WEIGHTS)
 _MOMENT_HOUR_CUMULATIVE = np.cumsum(world.ORDER_MOMENT_HOUR_WEIGHTS)
+_SNAPSHOT_DELAY_CUMULATIVE = np.cumsum(world.ORDER_SNAPSHOT_DELAY_WEIGHTS)
 
 
 class OrderOutcome(IntEnum):
@@ -103,12 +106,18 @@ class Orders:
     outcome: NDArray[np.int64]
     paid_after: NDArray[np.int64]
     cancelled_after: NDArray[np.int64]
+    snapshot_delay: NDArray[np.int64]
 
     def __len__(self) -> int:
         return self.user_id.size
 
 
-def of_day(seed: int, day: int, purchases: Purchases) -> Orders:
+def of_day(
+    seed: int,
+    day: int,
+    purchases: Purchases,
+    assigned_order: NDArray[np.bool_],
+) -> Orders:
     """Заказы дня `day`: его покупки, к которым бэкенд добавил свои деньги."""
     rng = day_stream(seed, day, Component.ORDERS)
     delivery = _delivery(rng, len(purchases))
@@ -116,6 +125,7 @@ def of_day(seed: int, day: int, purchases: Purchases) -> Orders:
     product, quantity, items_total = _delta(rng, purchases)
     created_at = _created_at(rng, purchases)
     discount = _discount(purchases, items_total)
+    snapshot_delay = _snapshot_delay(rng, assigned_order)
     return Orders(
         day=day,
         order_id=purchases.order_id,
@@ -130,6 +140,7 @@ def of_day(seed: int, day: int, purchases: Purchases) -> Orders:
         outcome=outcome,
         paid_after=paid_after,
         cancelled_after=cancelled_after,
+        snapshot_delay=snapshot_delay,
     )
 
 
@@ -166,6 +177,11 @@ def at_boundary(rows: Orders, day: int) -> tuple[list[str], NDArray[np.datetime6
     return status.tolist(), updated
 
 
+def arrived(rows: Orders, day: int) -> NDArray[np.bool_]:
+    """Какие заказы уже попали в выгрузку к слепку дня `day`."""
+    return rows.day + rows.snapshot_delay <= day
+
+
 def _boundary(day: int) -> np.datetime64:
     """Граница суток `day`|`day+1` абсолютной меткой: полночь пояса счётчика.
 
@@ -218,6 +234,15 @@ def _created_at(
     )
 
 
+def _snapshot_delay(
+    rng: np.random.Generator,
+    assigned_order: NDArray[np.bool_],
+) -> NDArray[np.int64]:
+    """Сколько ранних слепков пропустит заказ; назначенный приезжает сразу."""
+    delay = pick(rng, _SNAPSHOT_DELAY_CUMULATIVE, assigned_order.size)
+    return np.where(assigned_order, 0, delay)
+
+
 def _moment(rng: np.random.Generator, orders: int) -> NDArray[np.int64]:
     """Момент внутри окна: час по таблице весов и равномерная секунда в нём.
 

generator/src/clickstream_generator/serialize.py

@@ -45,7 +45,7 @@ from numpy.typing import NDArray
 
 from clickstream_generator import catalog, schema, world
 from clickstream_generator.day import Day
-from clickstream_generator.orders import Orders, at_boundary
+from clickstream_generator.orders import Orders, arrived, at_boundary
 
 _ARRAY_PREFIX = "Array("
 
@@ -72,9 +72,10 @@ def events(day: Day, limit: int | None = None) -> list[bytes]:
 def orders(window: Sequence[Orders], day: int) -> list[bytes]:
     """Канонические байты слепка дня `day`: по документу JSON на заказ.
 
-    `window` — заказы дней окна, от раннего дня к позднему: слепок несёт их
-    подряд, и порядок строк выходит порядком рождения заказов, он же
-    возрастание `order_id`. Какие это дни, решает `orders.window`.
+    `window` — заказы дней окна, от раннего дня к позднему. Уже приехавшие
+    заказы слепок несёт подряд, поэтому порядок строк остаётся порядком
+    рождения, он же возрастание `order_id`. Какие это дни, решает
+    `orders.window`.
 
     Деньги уезжают строками с ровно двумя знаками, а не числами: у заказа они
     станут `Decimal`, и дробь двоичного числа была бы потерей точности до
@@ -97,7 +98,8 @@ def orders(window: Sequence[Orders], day: int) -> list[bytes]:
         delivery = rows.delivery.tolist()
         total = rows.total.tolist()
 
-        for number, order_id in enumerate(rows.order_id):
+        for number in np.flatnonzero(arrived(rows, day)).tolist():
+            order_id = rows.order_id[number]
             payloads.append(
                 orjson.dumps(
                     {

generator/src/clickstream_generator/world.py

@@ -283,3 +283,8 @@ ORDER_MOMENT_HOUR_WEIGHTS = tuple(
 # выходит вдвое-втрое ниже броска, около ориентира мастер-спеки в 1–2%. Число
 # черновое: калибровка — пересборка эталонного мира, этап 7.
 ORDER_DELTA_PERCENT = 4
+
+# На сколько первых дневных слепков заказ опоздает: веса δ = 0, 1, 2.
+# Назначенные планом заказы после броска получают 0, чтобы мост склейки
+# появился в стартовом мире. Числа черновые: калибровка — этап 7.
+ORDER_SNAPSHOT_DELAY_WEIGHTS = (90, 5, 5)

generator/tests/test_day.py

@@ -68,6 +68,7 @@ def orders(events: day.Day) -> list[list[Any]]:
             int(theirs.outcome[number]),
             int(theirs.paid_after[number]),
             int(theirs.cancelled_after[number]),
+            int(theirs.snapshot_delay[number]),
         ]
         for number in range(len(theirs))
     ]

generator/tests/test_orders.py

@@ -87,6 +87,25 @@ def test_every_order_of_the_day_is_a_purchase_of_the_day(weekday: day.Day):
     )
 
 
+def test_an_assigned_order_never_waits_for_a_later_snapshot(
+    monkeypatch: pytest.MonkeyPatch,
+):
+    """Гарантия плана сильнее опоздания: мост склейки приезжает сразу."""
+    # Все неназначенные заказы опаздывают на два дня. Нулевую задержку после
+    # такого броска могут получить только заказы, защищённые планом.
+    monkeypatch.setattr(
+        orders, "_SNAPSHOT_DELAY_CUMULATIVE", np.array([0, 0, 1], dtype=np.int64)
+    )
+    events = day.stream(CANONICAL_SEED, WEEKDAY)
+    audience = plan.audience(CANONICAL_SEED, WEEKDAY)
+
+    promised = set(audience.person_id[audience.assigned_order].tolist())
+    immediate = set(events.orders.user_id[events.orders.snapshot_delay == 0].tolist())
+    assert promised
+    assert immediate == promised
+    assert (events.orders.snapshot_delay == 2).any()
+
+
 def dropped_line(client: list[tuple[Any, int]], order: list[tuple[Any, int]]):
     """Единственный индекс, удалением которого корзина клиента даёт заказ."""
     for gone in range(len(client)):

generator/tests/test_snapshot.py

@@ -176,6 +176,63 @@ def test_the_rows_go_in_the_order_of_birth(days: list[day_module.Day]):
     assert len(set(numbers)) == len(numbers)
 
 
+def test_lateness_changes_only_the_rows_present_in_a_snapshot(
+    days: list[day_module.Day],
+):
+    """Неопоздавшие строки остаются прежними байтами и в прежнем порядке."""
+    number = SHORT_WINDOW_DAY
+    window = [today.orders for today in days[: number + 1]]
+    eager = serialize.orders(
+        [
+            replace(rows, snapshot_delay=np.zeros_like(rows.snapshot_delay))
+            for rows in window
+        ],
+        number,
+    )
+    actual = serialize.orders(window, number)
+    actual_ids = {json.loads(payload)["order_id"] for payload in actual}
+    expected = [
+        payload for payload in eager if json.loads(payload)["order_id"] in actual_ids
+    ]
+
+    assert len(actual) < len(eager)
+    assert actual == expected
+
+
+def test_a_two_day_late_order_first_arrives_in_its_lived_state(
+    days: list[day_module.Day],
+):
+    """Заказ δ = 2 пропускает D и D+1 и догоняет жизнь, не меняя `created_at`."""
+    honest = days[0].orders
+    target = 0
+    delay = np.zeros_like(honest.snapshot_delay)
+    delay[target] = 2
+    cancelled_after = honest.cancelled_after.copy()
+    cancelled_after[target] = 30 * 3600
+    late = replace(
+        honest,
+        snapshot_delay=delay,
+        cancelled_after=cancelled_after,
+    )
+    order_id = late.order_id[target]
+
+    for number in (0, 1):
+        assert order_id not in {
+            json.loads(payload)["order_id"]
+            for payload in serialize.orders([late], number)
+        }
+
+    arrived = {
+        record["order_id"]: record
+        for record in map(json.loads, serialize.orders([late], 2))
+    }[order_id]
+    created_at = np.datetime_as_string(
+        late.created_at[target], unit="ms", timezone="UTC"
+    )
+    assert arrived["status"] == "cancelled"
+    assert arrived["created_at"] == created_at
+
+
 def test_the_state_is_read_at_the_boundary_of_the_day(days: list[day_module.Day]):
     """Дыхание окна: оплаченный назавтра заказ в сегодняшнем слепке `created`.