diff --git a/data/world-inventory.json b/data/world-inventory.json index 38d08dd..b283651 100644 --- a/data/world-inventory.json +++ b/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" } ] } diff --git a/docs/architecture/orders/fate.md b/docs/architecture/orders/fate.md index 419f891..0e21839 100644 --- a/docs/architecture/orders/fate.md +++ b/docs/architecture/orders/fate.md @@ -125,20 +125,22 @@ **Опоздание — заказ прячется от ранних слепков.** `created_at` не подделывается — строка создана, когда заказ родился, — но в слепках дней -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 — -сломалась бы та самая сверка, ради которой всё строится. +а с 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 — сломалась бы та +самая сверка, ради которой всё строится. **Пересечения.** Броски независимы, пересечения выходят арифметикой, приоритет мастер-спеки работает по-настоящему. Исключений два, и оба названы выше: дубль diff --git a/docs/architecture/orders/snapshot.md b/docs/architecture/orders/snapshot.md index 7ef6407..4eef6a7 100644 --- a/docs/architecture/orders/snapshot.md +++ b/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 — ночная выгрузка diff --git a/generator/src/clickstream_generator/commerce.py b/generator/src/clickstream_generator/commerce.py index 59ef50f..a95795c 100644 --- a/generator/src/clickstream_generator/commerce.py +++ b/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: diff --git a/generator/src/clickstream_generator/day.py b/generator/src/clickstream_generator/day.py index 754a064..3d8fe86 100644 --- a/generator/src/clickstream_generator/day.py +++ b/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, ) diff --git a/generator/src/clickstream_generator/orders.py b/generator/src/clickstream_generator/orders.py index 849112d..da9acc0 100644 --- a/generator/src/clickstream_generator/orders.py +++ b/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]: """Момент внутри окна: час по таблице весов и равномерная секунда в нём. diff --git a/generator/src/clickstream_generator/serialize.py b/generator/src/clickstream_generator/serialize.py index 01bdff7..3e94b1d 100644 --- a/generator/src/clickstream_generator/serialize.py +++ b/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( { diff --git a/generator/src/clickstream_generator/world.py b/generator/src/clickstream_generator/world.py index 2b5b664..e2f5993 100644 --- a/generator/src/clickstream_generator/world.py +++ b/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) diff --git a/generator/tests/test_day.py b/generator/tests/test_day.py index 58f23e8..aeedb7d 100644 --- a/generator/tests/test_day.py +++ b/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)) ] diff --git a/generator/tests/test_orders.py b/generator/tests/test_orders.py index 1eae821..c41fec0 100644 --- a/generator/tests/test_orders.py +++ b/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)): diff --git a/generator/tests/test_snapshot.py b/generator/tests/test_snapshot.py index 53acdfe..fb52077 100644 --- a/generator/tests/test_snapshot.py +++ b/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`.