diff --git a/.scratch/generator-model-time-startup-history/issues/04-state-v2-model-resume.md b/.scratch/generator-model-time-startup-history/issues/04-state-v2-model-resume.md index d7eb871..b034487 100644 --- a/.scratch/generator-model-time-startup-history/issues/04-state-v2-model-resume.md +++ b/.scratch/generator-model-time-startup-history/issues/04-state-v2-model-resume.md @@ -1,4 +1,4 @@ -Status: ready-for-agent +Status: ready-for-human # Восстановление state v2 от модельной точки @@ -19,27 +19,96 @@ Status: ready-for-agent ## Acceptance criteria -- [ ] State сохраняет достаточно данных, чтобы после сбоя вычислить модельную +- [x] State сохраняет достаточно данных, чтобы после сбоя вычислить модельную точку возобновления при ×K. -- [ ] Восстановление после короткого модельного сбоя продолжает активные визиты +- [x] Восстановление после короткого модельного сбоя продолжает активные визиты и досылает созревшие события с исходными модельными метками. -- [ ] Восстановление после долгого модельного сбоя закрывает сильно просроченные +- [x] Восстановление после долгого модельного сбоя закрывает сильно просроченные активные визиты без досылки остатка. -- [ ] Восстановление из стартовой истории использует `T_end` как точку +- [x] Восстановление из стартовой истории использует `T_end` как точку возобновления и не обрывает активные визиты из-за настенного простоя. -- [ ] Визит, переживший восстановление, остаётся однородным: события до и после +- [x] Визит, переживший восстановление, остаётся однородным: события до и после восстановления не меняют контекст визита и не создают второй путь генерации внутри одного `click_id`. -- [ ] Повреждённый или несовместимый state не валит сервис: генератор стартует +- [x] Повреждённый или несовместимый state не валит сервис: генератор стартует с чистого листа и пишет предупреждение. -- [ ] ClickHouse-проверка подтверждает, что на стыке восстановления нет дублей +- [x] ClickHouse-проверка подтверждает, что на стыке восстановления нет дублей событий и нет разрыва `click_id` внутри продолжающегося визита. -- [ ] После реализации выполнено саморевью worker-а и отдельное reviewer-ревью, +- [x] После реализации выполнено саморевью worker-а и отдельное reviewer-ревью, потому что задача меняет state/serialization и сервисное восстановление. -- [ ] Если исправление после ревью меняет формат state или способ восстановления, +- [x] Если исправление после ревью меняет формат state или способ восстановления, выполнен повторный reviewer-круг. ## Blocked by - `.scratch/generator-model-time-startup-history/issues/03-model-speed-and-day-factor.md` - Review gate из `PRD.md`: сквозной инвариант времени после задачи 3 + +## Решение + +- State v2 теперь хранит `model_timestamp`, `wall_timestamp`, + `model_time_speed`, `model_timezone`, `model_t0` и `gen_seed`. +- Старый v2 без этих полей считается несовместимым: `from_dict_safe` возвращает + `None`, сервис пишет предупреждение и стартует с чистого состояния. +- Live-восстановление считает точку продолжения по формуле + `model_timestamp + max(0, wall_now_utc - wall_timestamp) * model_time_speed`. +- Перед live-восстановлением сервис сверяет state с текущим `Config` по + `gen_seed`, `model_t0`, `model_timezone` и `model_time_speed`. Несовпадение + считается несовместимым state: сервис пишет warning и стартует с чистого + состояния. +- Восстановление из стартовой истории вынесено в минимальный метод + `GeneratorService.restore_from_startup_history(state, model_t_end)`: он берёт + переданный `T_end` напрямую и не применяет wall-дельту. +- Сохранение state после успешного тика пишет модельную точку следующего тика, + а не настенное время. Это убирает повторный старт с уже обработанной модельной + точки. + +## Проверки + +- Точечный прогон после проверки reviewer-а: + `docker run --rm -v ... generator:test pytest tests/test_state.py::TestGeneratorState tests/test_state.py::TestGeneratorStateValidation::test_from_dict_safe_returns_state_on_valid tests/test_state.py::TestGeneratorStateValidation::test_from_dict_safe_returns_none_on_invalid_gen_seed tests/test_service.py::TestGeneratorServiceStateV2 -v` + — 15 passed. +- `make generator-test` — 126 passed. +- `uv run pytest ...` и `uv run python -m pytest ...` не запускались до тестов: + в текущем uv-окружении нет `pytest`. +- `make lint` недоступен: в `Makefile` нет такой цели. + +## ClickHouse-проверка + +Выполнены два чистых стендовых сценария. Оба начинались со сброса ClickHouse, +Kafka и state генератора. + +1. Короткое восстановление, `GEN_MODEL_TIME_SPEED=1`. + Первый запуск с `GEN_STATE_RESET=true` записал один тик, второй запуск с + `GEN_STATE_RESET=false` восстановился из state и продолжил активные визиты. + Проверка ClickHouse: + - `dds.event`: `rows=332`, `uniq_events=332`, `duplicate_events=0`. + - `continued_from_first_tick=67` для визитов, начатых на + `2026-01-01 10:00:00`. + - Для этих `click_id` нет смены `user_domain_id`, `device_type`, `os_name`, + `geo_country`, `geo_region_name` и `geo_timezone`. +2. Долгое восстановление, `GEN_MODEL_TIME_SPEED=3600`. + Первый запуск с `GEN_STATE_RESET=true` записал один тик на + `2026-01-01 10:00:00`, второй запуск с `GEN_STATE_RESET=false` + восстановился на модельной точке `2026-01-03T22:51:51.552000+00:00`. + Проверка ClickHouse: + - `dds.event`: `rows=200`, `uniq_events=200`, `duplicate_events=0`. + - диапазон `event_ts`: от `2026-01-01 10:00:00.000000` до + `2026-01-03 22:51:51.552000`. + - `continued_old_clicks=0`: старые `click_id` с первого тика не продолжились + после долгой модельной паузы. + +## Риски + +- Старые записи state v2 без модельной связки больше не восстанавливаются. Это + намеренно: без этих полей нельзя корректно посчитать точку продолжения. +- Полноценный манифест и backfill не реализованы: добавлен только минимальный + интерфейс восстановления от переданного `T_end`. +- `restore_from_startup_history` пока принимает state и `T_end` напрямую, без + проверки manifest. Это риск issue 05, где появится полноценная стартовая + история и запрет смешивания артефактов. +- `GeneratorState.__post_init__` оставляет in-memory совместимость для прямого + создания state в тестах и коде. JSON/Kafka путь остаётся строгим: отсутствие + новых полей не маскируется и ведёт к fresh start. +- После reviewer-fix формат state снова изменился (`gen_seed`). Повторный + reviewer-круг выполнен, новых блокирующих находок нет. diff --git a/generator/src/clickstream_generator/runtime.py b/generator/src/clickstream_generator/runtime.py index 1f0fe25..b11bab8 100644 --- a/generator/src/clickstream_generator/runtime.py +++ b/generator/src/clickstream_generator/runtime.py @@ -175,6 +175,12 @@ class TickStreamGenerator: rng_state: tuple, last_batch_id: str, last_timestamp: datetime, + model_timestamp: datetime | None = None, + wall_timestamp: datetime | None = None, + model_time_speed: float = 1.0, + model_timezone: str = "UTC", + model_t0: datetime | None = None, + gen_seed: int | None = None, ) -> GeneratorState: """Возвращает JSON-сериализуемый снимок тикового слоя.""" return GeneratorState( @@ -182,6 +188,12 @@ class TickStreamGenerator: rng_state=rng_state, last_batch_id=last_batch_id, last_timestamp=last_timestamp, + model_timestamp=model_timestamp, + wall_timestamp=wall_timestamp, + model_time_speed=model_time_speed, + model_timezone=model_timezone, + model_t0=model_t0, + gen_seed=gen_seed, population=[ { "user_domain_id": user.user_domain_id, @@ -220,6 +232,7 @@ class TickStreamGenerator: def restore_state( self, state: GeneratorState, + resume_model_at: datetime | None = None, restarted_at: datetime | None = None, ) -> None: """Восстанавливает популяцию и активные визиты из state v2.""" @@ -231,17 +244,17 @@ class TickStreamGenerator: self.population.users = users self.active_visits = [] - restarted_time = ( - _normalize_tick_time(restarted_at) - if restarted_at is not None + model_resume_time = ( + _normalize_tick_time(resume_model_at or restarted_at) + if resume_model_at is not None or restarted_at is not None else None ) for item in state.active_visits: visit = self._visit_from_state(item, users_by_id) - if self._is_overdue_after_restart(visit, restarted_time): + if self._is_overdue_after_restart(visit, model_resume_time): self.population.finish_visit( visit.user, - self._last_released_at(visit, state.last_timestamp), + self._last_released_at(visit, state.model_timestamp), ) continue self.active_visits.append(visit) diff --git a/generator/src/clickstream_generator/service.py b/generator/src/clickstream_generator/service.py index 0714ce1..c7d8fe4 100644 --- a/generator/src/clickstream_generator/service.py +++ b/generator/src/clickstream_generator/service.py @@ -73,14 +73,13 @@ class GeneratorService: restored_state = self.state_manager.load() if restored_state: try: - self.generator.rng.setstate(restored_state.rng_state) - self.stream.restore_state( + self._restore_live_state( restored_state, - restarted_at=datetime.now(timezone.utc), + wall_now_utc=datetime.now(timezone.utc), ) - self._tick = restored_state.tick logger.info( f"Restored state: continuing from tick {self._tick}, " + f"model_time={self._model_time.isoformat()}, " f"last_batch_id={restored_state.last_batch_id}" ) except Exception as e: @@ -115,21 +114,92 @@ class GeneratorService: if self.state_manager: self.state_manager.close() + def restore_from_startup_history( + self, + state, + model_t_end: datetime, + ) -> None: + """Восстанавливает слепок стартовой истории ровно от T_end.""" + self._restore_state_snapshot(state, resume_model_at=model_t_end) + + def _restore_live_state(self, state, wall_now_utc: datetime) -> None: + """Восстанавливает live-state с учётом прошедшего настенного времени.""" + self._validate_live_state_config(state) + resume_model_at = self._calculate_live_resume_model_at( + state, + wall_now_utc=wall_now_utc, + ) + self._restore_state_snapshot(state, resume_model_at=resume_model_at) + + def _restore_state_snapshot(self, state, resume_model_at: datetime) -> None: + """Применяет state к генератору и тиковому слою.""" + resume_model_at = self._as_aware_utc(resume_model_at) + self.generator.rng.setstate(state.rng_state) + self.stream.restore_state(state, resume_model_at=resume_model_at) + self._tick = state.tick + self._model_time = resume_model_at + + def _calculate_live_resume_model_at( + self, + state, + wall_now_utc: datetime, + ) -> datetime: + """Считает модельную точку live-восстановления по state v2.""" + wall_now_utc = self._as_aware_utc(wall_now_utc) + wall_saved_at = self._as_aware_utc(state.wall_timestamp) + idle_seconds = max(0.0, (wall_now_utc - wall_saved_at).total_seconds()) + return self._as_aware_utc(state.model_timestamp) + timedelta( + seconds=idle_seconds * state.model_time_speed, + ) + + def _validate_live_state_config(self, state) -> None: + """Проверяет, что state относится к текущей конфигурации live-запуска.""" + mismatches = [] + if state.gen_seed != self.config.seed: + mismatches.append("gen_seed") + if self._as_aware_utc(state.model_t0) != self.config.model_t0: + mismatches.append("model_t0") + if state.model_timezone != self.config.model_timezone: + mismatches.append("model_timezone") + if abs(state.model_time_speed - self.config.model_time_speed) > 1e-9: + mismatches.append("model_time_speed") + + if mismatches: + raise ValueError( + "state config mismatch: " + ", ".join(mismatches) + ) + + @staticmethod + def _as_aware_utc(value: datetime) -> datetime: + if value.tzinfo is None: + return value.replace(tzinfo=timezone.utc) + return value.astimezone(timezone.utc) + def _save_state(self, batch_id: str) -> None: """Сохраняет текущее состояние генератора.""" if not self.state_manager or not self.config.state_enabled: return try: + wall_now = datetime.now(timezone.utc) state = self.stream.to_state( tick=self._tick, rng_state=self.generator.rng.getstate(), last_batch_id=batch_id, - last_timestamp=datetime.now(timezone.utc), + last_timestamp=self._model_time, + model_timestamp=self._model_time, + wall_timestamp=wall_now, + model_time_speed=self.config.model_time_speed, + model_timezone=self.config.model_timezone, + model_t0=self.config.model_t0, + gen_seed=self.config.seed, ) self.state_manager.save(state) self.state_manager.flush() - logger.debug(f"Saved state: tick={self._tick}, batch_id={batch_id}") + logger.debug( + f"Saved state: tick={self._tick}, " + f"model_time={self._model_time.isoformat()}, batch_id={batch_id}" + ) except Exception as e: logger.warning(f"Failed to save state: {e}") METRICS_ERRORS_TOTAL.labels(topic="state").inc() @@ -179,8 +249,8 @@ class GeneratorService: if status in ("success", "partial"): METRICS_LAST_SUCCESS.set_to_current_time() - self._save_state(batch_id) self._advance_model_time() + self._save_state(batch_id) self.publisher.flush() pub_duration = time.time() - pub_start diff --git a/generator/src/clickstream_generator/state.py b/generator/src/clickstream_generator/state.py index 15ca3cb..fc68431 100644 --- a/generator/src/clickstream_generator/state.py +++ b/generator/src/clickstream_generator/state.py @@ -3,7 +3,8 @@ import logging import random from dataclasses import dataclass, field -from datetime import datetime +from datetime import datetime, timezone +from zoneinfo import ZoneInfo, ZoneInfoNotFoundError logger = logging.getLogger("generator") @@ -25,7 +26,56 @@ def _require_keys(item: dict, keys: tuple[str, ...], label: str) -> None: raise ValueError(f"{label} missing fields: {', '.join(missing)}") +def _parse_aware_utc(value: str, label: str) -> datetime: + if not isinstance(value, str) or not value: + raise ValueError(f"{label} must be a non-empty string") + timestamp = datetime.fromisoformat(value.replace("Z", "+00:00")) + if timestamp.tzinfo is None: + raise ValueError(f"{label} must include timezone") + return timestamp.astimezone(timezone.utc) + + +def _validate_resume_fields(data: dict) -> None: + _require_keys( + data, + ( + "model_timestamp", + "wall_timestamp", + "model_time_speed", + "model_timezone", + "model_t0", + "gen_seed", + ), + "state", + ) + _parse_aware_utc(data["model_timestamp"], "model_timestamp") + _parse_aware_utc(data["wall_timestamp"], "wall_timestamp") + _parse_aware_utc(data["model_t0"], "model_t0") + + model_time_speed = data["model_time_speed"] + if ( + isinstance(model_time_speed, bool) + or not isinstance(model_time_speed, int | float) + or model_time_speed <= 0 + ): + raise ValueError("model_time_speed must be a positive number") + + model_timezone = data["model_timezone"] + if not isinstance(model_timezone, str) or not model_timezone: + raise ValueError("model_timezone must be a string") + try: + ZoneInfo(model_timezone) + except ZoneInfoNotFoundError as e: + raise ValueError(f"unknown model_timezone: {model_timezone}") from e + + gen_seed = data["gen_seed"] + if gen_seed is not None: + if isinstance(gen_seed, bool) or not isinstance(gen_seed, int): + raise ValueError("gen_seed must be an integer or null") + + def _validate_v2_payload(data: dict) -> None: + _validate_resume_fields(data) population = data.get("population") active_visits = data.get("active_visits") pending_visit_births = data.get("pending_visit_births", 0.0) @@ -129,10 +179,30 @@ class GeneratorState: last_batch_id: str last_timestamp: datetime version: str = STATE_VERSION + model_timestamp: datetime | None = None + wall_timestamp: datetime | None = None + model_time_speed: float = 1.0 + model_timezone: str = "UTC" + model_t0: datetime | None = None + gen_seed: int | None = None population: list[dict] = field(default_factory=list) active_visits: list[dict] = field(default_factory=list) pending_visit_births: float = 0.0 + def __post_init__(self) -> None: + if self.model_timestamp is None: + self.model_timestamp = self._as_aware_utc(self.last_timestamp) + if self.wall_timestamp is None: + self.wall_timestamp = self._as_aware_utc(self.last_timestamp) + if self.model_t0 is None: + self.model_t0 = self.model_timestamp + + @staticmethod + def _as_aware_utc(value: datetime) -> datetime: + if value.tzinfo is None: + return value.replace(tzinfo=timezone.utc) + return value.astimezone(timezone.utc) + def to_dict(self) -> dict: """Конвертирует в словарь для JSON-сериализации.""" return { @@ -140,6 +210,12 @@ class GeneratorState: "rng_state": self.rng_state, "last_batch_id": self.last_batch_id, "last_timestamp": self.last_timestamp.isoformat(), + "model_timestamp": self.model_timestamp.isoformat(), + "wall_timestamp": self.wall_timestamp.isoformat(), + "model_time_speed": self.model_time_speed, + "model_timezone": self.model_timezone, + "model_t0": self.model_t0.isoformat(), + "gen_seed": self.gen_seed, "version": self.version, "population": self.population, "active_visits": self.active_visits, @@ -160,6 +236,12 @@ class GeneratorState: ) raise ValueError(f"unsupported state version: {version}") _validate_v2_payload(data) + model_timestamp = _parse_aware_utc( + data["model_timestamp"], + "model_timestamp", + ) + wall_timestamp = _parse_aware_utc(data["wall_timestamp"], "wall_timestamp") + model_t0 = _parse_aware_utc(data["model_t0"], "model_t0") rng_state_raw = data.get("rng_state") if not rng_state_raw: @@ -184,9 +266,15 @@ class GeneratorState: rng_state=rng_state, last_batch_id=data.get("last_batch_id", ""), last_timestamp=datetime.fromisoformat( - data.get("last_timestamp", "1970-01-01T00:00:00+00:00") + data.get("last_timestamp", data["model_timestamp"]) ), version=version, + model_timestamp=model_timestamp, + wall_timestamp=wall_timestamp, + model_time_speed=float(data["model_time_speed"]), + model_timezone=data["model_timezone"], + model_t0=model_t0, + gen_seed=data["gen_seed"], population=data.get("population", []), active_visits=data.get("active_visits", []), pending_visit_births=data.get("pending_visit_births", 0.0), diff --git a/generator/tests/test_generation.py b/generator/tests/test_generation.py index a20f2d7..d8458cd 100644 --- a/generator/tests/test_generation.py +++ b/generator/tests/test_generation.py @@ -340,7 +340,9 @@ class TestEventGeneration: last_timestamp=tick_at, ) visit_state = state.active_visits[0] + original_visit = stream.active_visits[0] started_at = datetime.fromisoformat(visit_state["started_at"]) + next_index = visit_state["next_index"] planned_timestamps = { (started_at + timedelta(microseconds=offset_us)).strftime( "%Y-%m-%d %H:%M:%S.%f" @@ -366,6 +368,19 @@ class TestEventGeneration: assert resumed_timestamps assert set(resumed_timestamps).issubset(planned_timestamps) + assert {event["click_id"] for event in resumed["browser_events"]} == { + visit_state["click_id"] + } + assert [ + event["page_url_path"] + for event in resumed["location_events"] + ] == visit_state["page_url_paths"][next_index:next_index + len(resumed_timestamps)] + assert resumed["device_events"] == original_visit.batch["device_events"][ + next_index:next_index + len(resumed_timestamps) + ] + assert resumed["geo_events"] == original_visit.batch["geo_events"][ + next_index:next_index + len(resumed_timestamps) + ] assert all( datetime.fromisoformat(timestamp.replace(" ", "T")) <= tick_at + timedelta(minutes=29) for timestamp in resumed_timestamps diff --git a/generator/tests/test_service.py b/generator/tests/test_service.py index 510dc66..9df2eae 100644 --- a/generator/tests/test_service.py +++ b/generator/tests/test_service.py @@ -5,7 +5,7 @@ import logging import random from dataclasses import replace -from datetime import datetime, timezone +from datetime import datetime, timedelta, timezone from time import sleep as real_sleep from unittest.mock import MagicMock, patch @@ -315,6 +315,12 @@ class TestGeneratorServiceStateV2: rng_state=source_generator.rng.getstate(), last_batch_id="batch-3", last_timestamp=tick_at, + model_timestamp=tick_at.replace(tzinfo=timezone.utc), + wall_timestamp=tick_at.replace(tzinfo=timezone.utc), + model_time_speed=base_config.model_time_speed, + model_timezone=base_config.model_timezone, + model_t0=base_config.model_t0, + gen_seed=base_config.seed, ) state_manager = MagicMock() @@ -337,6 +343,99 @@ class TestGeneratorServiceStateV2: assert service.stream.population_user_ids == source_stream.population_user_ids assert service.stream.active_visit_count == source_stream.active_visit_count + def test_start_restores_model_time_from_state_wall_delta( + self, base_config, event_dictionary + ): + """Live-восстановление считает точку модели из сохранённой wall-метки.""" + model_t0 = datetime(2026, 1, 1, 10, 0, tzinfo=timezone.utc) + wall_saved_at = datetime(2026, 6, 14, 12, 0, tzinfo=timezone.utc) + wall_restarted_at = wall_saved_at + timedelta(seconds=30) + config = replace( + base_config, + model_t0=model_t0, + model_time_speed=10, + tick_seconds=60, + ) + source_generator = EventGenerator(event_dictionary, config) + source_stream = TickStreamGenerator(source_generator) + source_stream.generate_tick(event_budget=10, tick_started_at=model_t0) + state = source_stream.to_state( + tick=3, + rng_state=source_generator.rng.getstate(), + last_batch_id="batch-3", + last_timestamp=model_t0, + model_timestamp=model_t0, + wall_timestamp=wall_saved_at, + model_time_speed=config.model_time_speed, + model_timezone=config.model_timezone, + model_t0=config.model_t0, + gen_seed=config.seed, + ) + state_manager = MagicMock() + state_manager.load.return_value = state + + class FrozenDateTime(datetime): + @classmethod + def now(cls, tz=None): + if tz is None: + return wall_restarted_at.replace(tzinfo=None) + return wall_restarted_at.astimezone(tz) + + with patch("clickstream_generator.service.start_http_server"), \ + patch("clickstream_generator.service.ensure_topics"), \ + patch("clickstream_generator.service.KafkaPublisher"), \ + patch("clickstream_generator.service.KafkaBatchHistory"), \ + patch( + "clickstream_generator.service.KafkaStateManager", + return_value=state_manager, + ), \ + patch("clickstream_generator.service.datetime", FrozenDateTime), \ + patch.object(GeneratorService, "_main_loop", return_value=None): + + service = GeneratorService(config) + service.start() + + assert service._tick == 3 + assert service._model_time == model_t0 + timedelta(seconds=300) + assert service.stream.active_visit_count == source_stream.active_visit_count + + def test_restore_from_startup_history_uses_passed_model_point_without_wall_delta( + self, base_config, event_dictionary + ): + """Стартовая история продолжает с T_end, а не с wall-простоя.""" + model_t0 = datetime(2026, 1, 1, 10, 0, tzinfo=timezone.utc) + model_t_end = model_t0 + timedelta(minutes=5) + old_wall_saved_at = datetime(2026, 1, 1, 0, 0, tzinfo=timezone.utc) + config = replace( + base_config, + model_t0=model_t0, + model_time_speed=3600, + tick_seconds=60, + max_session_events=5, + ) + source_generator = EventGenerator(event_dictionary, config) + source_stream = TickStreamGenerator(source_generator) + source_stream.generate_tick(event_budget=10, tick_started_at=model_t_end) + state = source_stream.to_state( + tick=99, + rng_state=source_generator.rng.getstate(), + last_batch_id="history-end", + last_timestamp=model_t_end, + model_timestamp=model_t_end, + wall_timestamp=old_wall_saved_at, + model_time_speed=config.model_time_speed, + model_timezone=config.model_timezone, + model_t0=config.model_t0, + gen_seed=config.seed, + ) + + service = GeneratorService(config) + service.restore_from_startup_history(state, model_t_end=model_t_end) + + assert service._tick == 99 + assert service._model_time == model_t_end + assert service.stream.active_visit_count == source_stream.active_visit_count + def test_save_state_writes_tick_stream_state_v2(self, base_config): """Сервис сохраняет v2-снимок тикового слоя.""" service = GeneratorService(base_config) @@ -344,15 +443,62 @@ class TestGeneratorServiceStateV2: tick_at = datetime.now(timezone.utc).replace(tzinfo=None) service.stream.generate_tick(event_budget=10, tick_started_at=tick_at) service._tick = 1 + service._model_time = tick_at.replace(tzinfo=timezone.utc) service._save_state("batch-1") saved_state = service.state_manager.save.call_args.args[0] assert saved_state.version == "2.0" + assert saved_state.model_timestamp == tick_at.replace(tzinfo=timezone.utc) + assert saved_state.wall_timestamp.tzinfo is not None + assert saved_state.model_time_speed == base_config.model_time_speed + assert saved_state.model_timezone == base_config.model_timezone + assert saved_state.model_t0 == base_config.model_t0 + assert saved_state.gen_seed == base_config.seed assert saved_state.population assert saved_state.active_visits service.state_manager.flush.assert_called_once() + def test_incompatible_seed_state_starts_fresh(self, base_config, event_dictionary, caplog): + """State от другого GEN_SEED не смешивается с текущим запуском.""" + source_config = replace(base_config, seed=7) + source_generator = EventGenerator(event_dictionary, source_config) + source_stream = TickStreamGenerator(source_generator) + tick_at = datetime(2026, 1, 1, 10, 0, tzinfo=timezone.utc) + source_stream.generate_tick(event_budget=10, tick_started_at=tick_at) + state = source_stream.to_state( + tick=5, + rng_state=source_generator.rng.getstate(), + last_batch_id="other-seed", + last_timestamp=tick_at, + model_timestamp=tick_at, + wall_timestamp=datetime(2026, 6, 14, 12, 0, tzinfo=timezone.utc), + model_time_speed=base_config.model_time_speed, + model_timezone=base_config.model_timezone, + model_t0=base_config.model_t0, + gen_seed=source_config.seed, + ) + state_manager = MagicMock() + state_manager.load.return_value = state + + with patch("clickstream_generator.service.start_http_server"), \ + patch("clickstream_generator.service.ensure_topics"), \ + patch("clickstream_generator.service.KafkaPublisher"), \ + patch("clickstream_generator.service.KafkaBatchHistory"), \ + patch( + "clickstream_generator.service.KafkaStateManager", + return_value=state_manager, + ), \ + patch.object(GeneratorService, "_main_loop", return_value=None), \ + caplog.at_level(logging.WARNING, logger="generator"): + + service = GeneratorService(base_config) + service.start() + + assert service._tick == 0 + assert service._model_time == base_config.model_t0 + assert "state config mismatch: gen_seed" in caplog.text + def test_state_reset_skips_loading_saved_state(self, base_config): """GEN_STATE_RESET=true запускает сервис с чистого состояния.""" reset_config = replace(base_config, state_reset=True) @@ -382,6 +528,12 @@ class TestGeneratorServiceStateV2: rng_state=random.Random(42).getstate(), last_batch_id="bad-v2", last_timestamp=datetime.now(timezone.utc), + model_timestamp=base_config.model_t0, + wall_timestamp=datetime.now(timezone.utc), + model_time_speed=base_config.model_time_speed, + model_timezone=base_config.model_timezone, + model_t0=base_config.model_t0, + gen_seed=base_config.seed, population=[ { "user_domain_id": "user-unknown", diff --git a/generator/tests/test_state.py b/generator/tests/test_state.py index 7c52c64..fc283a5 100644 --- a/generator/tests/test_state.py +++ b/generator/tests/test_state.py @@ -25,6 +25,12 @@ def _make_valid_v2_state_data() -> dict: "rng_state": list(_make_valid_rng_state(42)), "last_batch_id": "v2", "last_timestamp": "2026-06-11T12:00:00+00:00", + "model_timestamp": "2026-01-01T10:00:00+00:00", + "wall_timestamp": "2026-06-11T12:00:00+00:00", + "model_time_speed": 10, + "model_timezone": "UTC", + "model_t0": "2026-01-01T00:00:00+00:00", + "gen_seed": 42, "version": "2.0", "population": [ { @@ -59,6 +65,19 @@ def _minimal_population() -> list[dict]: ] +def _with_resume_fields(data: dict) -> dict: + """Добавляет обязательные поля state v2, не связанные с проверяемой ошибкой.""" + return { + **data, + "model_timestamp": "2026-01-01T10:00:00+00:00", + "wall_timestamp": "2024-01-01T00:00:00+00:00", + "model_time_speed": 10, + "model_timezone": "UTC", + "model_t0": "2026-01-01T00:00:00+00:00", + "gen_seed": 42, + } + + class TestGeneratorState: """Тесты структуры состояния генератора.""" @@ -72,6 +91,12 @@ class TestGeneratorState: rng_state=rng_state, last_batch_id="abc123", last_timestamp=now, + model_timestamp=now, + wall_timestamp=now, + model_time_speed=10, + model_timezone="UTC", + model_t0=now, + gen_seed=42, version="2.0", ) @@ -79,6 +104,12 @@ class TestGeneratorState: assert state.rng_state == rng_state assert state.last_batch_id == "abc123" assert state.last_timestamp == now + assert state.model_timestamp == now + assert state.wall_timestamp == now + assert state.model_time_speed == 10 + assert state.model_timezone == "UTC" + assert state.model_t0 == now + assert state.gen_seed == 42 assert state.version == "2.0" def test_default_version(self): @@ -113,6 +144,12 @@ class TestGeneratorState: assert data["tick"] == 42 assert data["last_batch_id"] == "abc123" assert data["last_timestamp"] == now.isoformat() + assert data["model_timestamp"] == now.isoformat() + assert data["wall_timestamp"] == now.isoformat() + assert data["model_time_speed"] == 1.0 + assert data["model_timezone"] == "UTC" + assert data["model_t0"] == now.isoformat() + assert data["gen_seed"] is None assert data["version"] == "2.0" # Проверяем что rng_state сериализован как tuple (JSON-safe, без pickle) @@ -147,6 +184,12 @@ class TestGeneratorState: assert restored.rng_state == original.rng_state assert restored.last_batch_id == original.last_batch_id assert restored.last_timestamp == original.last_timestamp + assert restored.model_timestamp == original.model_timestamp + assert restored.wall_timestamp == original.wall_timestamp + assert restored.model_time_speed == original.model_time_speed + assert restored.model_timezone == original.model_timezone + assert restored.model_t0 == original.model_t0 + assert restored.gen_seed == original.gen_seed assert restored.version == original.version def test_roundtrip_with_real_random(self): @@ -190,6 +233,12 @@ class TestGeneratorState: rng_state=rng_state, last_batch_id="batch-7", last_timestamp=datetime(2026, 6, 11, 12, 0, tzinfo=timezone.utc), + model_timestamp=datetime(2026, 1, 1, 10, 0, tzinfo=timezone.utc), + wall_timestamp=datetime(2026, 6, 11, 12, 0, tzinfo=timezone.utc), + model_time_speed=10, + model_timezone="Europe/Moscow", + model_t0=datetime(2026, 1, 1, 0, 0, tzinfo=timezone.utc), + gen_seed=42, version="2.0", population=[ { @@ -220,6 +269,12 @@ class TestGeneratorState: assert restored.population == state.population assert restored.active_visits == state.active_visits assert restored.pending_visit_births == 0.5 + assert restored.model_timestamp == state.model_timestamp + assert restored.wall_timestamp == state.wall_timestamp + assert restored.model_time_speed == 10 + assert restored.model_timezone == "Europe/Moscow" + assert restored.model_t0 == state.model_t0 + assert restored.gen_seed == 42 class TestGeneratorStateValidation: @@ -238,7 +293,7 @@ class TestGeneratorStateValidation: def test_from_dict_invalid_rng_state_raises(self): """from_dict выбрасывает исключение при невалидном rng_state.""" - data = { + data = _with_resume_fields({ "tick": 42, "rng_state": "not_a_tuple", "last_batch_id": "test", @@ -246,14 +301,14 @@ class TestGeneratorStateValidation: "version": "2.0", "population": _minimal_population(), "active_visits": [], - } + }) with pytest.raises(ValueError): GeneratorState.from_dict(data) def test_from_dict_insufficient_rng_state_raises(self): """from_dict выбрасывает исключение при коротком rng_state.""" - data = { + data = _with_resume_fields({ "tick": 42, "rng_state": [1], # Слишком короткий "last_batch_id": "test", @@ -261,37 +316,37 @@ class TestGeneratorStateValidation: "version": "2.0", "population": _minimal_population(), "active_visits": [], - } + }) with pytest.raises(ValueError): GeneratorState.from_dict(data) def test_from_dict_invalid_setstate_raises(self): """from_dict выбрасывает исключение если setstate падает.""" - data = { + data = _with_resume_fields({ "tick": 42, "rng_state": [999, [1, 2, 3], None], # Невалидный state "last_batch_id": "test", "last_timestamp": "2024-01-01T00:00:00+00:00", "version": "2.0", - "population": [], + "population": _minimal_population(), "active_visits": [], - } + }) with pytest.raises(ValueError): GeneratorState.from_dict(data) def test_from_dict_safe_returns_none_on_invalid(self): """from_dict_safe возвращает None при невалидных данных.""" - data = { + data = _with_resume_fields({ "tick": 42, "rng_state": "invalid", "last_batch_id": "test", "last_timestamp": "2024-01-01T00:00:00+00:00", "version": "2.0", - "population": [], + "population": _minimal_population(), "active_visits": [], - } + }) result = GeneratorState.from_dict_safe(data) assert result is None @@ -304,6 +359,12 @@ class TestGeneratorStateValidation: "rng_state": list(rng.getstate()), # JSON сериализует tuple как list "last_batch_id": "test", "last_timestamp": "2024-01-01T00:00:00+00:00", + "model_timestamp": "2026-01-01T10:00:00+00:00", + "wall_timestamp": "2024-01-01T00:00:00+00:00", + "model_time_speed": 10, + "model_timezone": "UTC", + "model_t0": "2026-01-01T00:00:00+00:00", + "gen_seed": 42, "version": "2.0", "population": _minimal_population(), "active_visits": [], @@ -312,6 +373,43 @@ class TestGeneratorStateValidation: result = GeneratorState.from_dict_safe(data) assert result is not None assert result.tick == 42 + assert result.model_timestamp == datetime(2026, 1, 1, 10, 0, tzinfo=timezone.utc) + assert result.gen_seed == 42 + + def test_from_dict_safe_returns_none_on_invalid_gen_seed(self): + """gen_seed в JSON state должен быть числом или null.""" + data = _make_valid_v2_state_data() + data["gen_seed"] = "42" + + result = GeneratorState.from_dict_safe(data) + + assert result is None + + def test_from_dict_safe_returns_none_on_bool_model_time_speed(self): + """model_time_speed не принимает bool как числовую скорость.""" + data = _make_valid_v2_state_data() + data["model_time_speed"] = True + + result = GeneratorState.from_dict_safe(data) + + assert result is None + + def test_from_dict_safe_returns_none_without_model_resume_fields(self): + """State v2 без связки модельного и настенного времени несовместим.""" + rng = random.Random(42) + data = { + "tick": 42, + "rng_state": list(rng.getstate()), + "last_batch_id": "test", + "last_timestamp": "2024-01-01T00:00:00+00:00", + "version": "2.0", + "population": _minimal_population(), + "active_visits": [], + } + + result = GeneratorState.from_dict_safe(data) + + assert result is None def test_from_dict_rejects_old_state_without_version(self): """from_dict не восстанавливает старое state v1 без версии."""