From e1442d3b1deaaf4798bb5b6f787b4af15179e364 Mon Sep 17 00:00:00 2001 From: Light1YT Date: Sat, 13 Jun 2026 16:24:03 +0500 Subject: [PATCH] feat(tradein-scheduler): nightly newbuilding-enrichment schedule (#973) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Регистрирует отдельный in-app scheduler-источник для cian_newbuilding enrichment-backfill (houses_price_dynamics / house_reliability_checks / house_reviews), который раньше бежал только инлайн в Cian full-sweep. - run_newbuilding_enrich() wrapper (heartbeat → backfill → mark_done/failed), делегирует backfill_newbuilding_enrichment (#972, уже в main). - trigger_newbuilding_enrich_run() + dispatch-ветка в scheduler_loop (зеркалит trigger_sber_index_pull_run): _claim_run guard, async task, zombie-reap наследуется; idempotency наследуется (skip уже обогащённых, per-house SAVEPOINT, bounded per-fire limit дренирует backlog 306 домов). - seed-миграция 103: source='newbuilding_enrich', окно 00:00-01:00 UTC (03:00-04:00 МСК, до 01:00+ UTC sweep-блока), limit=25, next_run_at на завтра. Засеяно enabled=FALSE (dormant): на prod весь external-HTTP scraping намеренно на паузе. Расписание+триггер готовы, но не стартуют до намеренного возобновления (UPDATE ... SET enabled=true). Не активирую одиночный источник при общей паузе. Closes #973 --- tradein-mvp/backend/app/services/scheduler.py | 53 +++++ .../app/tasks/newbuilding_enrich_backfill.py | 79 +++++++ ...rape_schedules_seed_newbuilding_enrich.sql | 90 ++++++++ .../tasks/test_newbuilding_enrich_backfill.py | 201 ++++++++++++++++++ 4 files changed, 423 insertions(+) create mode 100644 tradein-mvp/backend/data/sql/103_scrape_schedules_seed_newbuilding_enrich.sql diff --git a/tradein-mvp/backend/app/services/scheduler.py b/tradein-mvp/backend/app/services/scheduler.py index 201cd4fe..0ae23c35 100644 --- a/tradein-mvp/backend/app/services/scheduler.py +++ b/tradein-mvp/backend/app/services/scheduler.py @@ -37,6 +37,14 @@ Sources: Rosreestr open-data portal for new quarter availability — pure HTTP detect+alert, NO download/shell-out; window 07:00-08:00 UTC, ~28-day cadence) + - newbuilding_enrich → run_newbuilding_enrich + (tasks/newbuilding_enrich_backfill.py, #973; nightly slice of the + cian_newbuilding houses enrichment — resolves cian_zhk_url, fetches + the ЖК page, persists houses_price_dynamics / house_reliability_checks + / house_reviews. Cian HTTP (curl_cffi chrome120) + anti-bot delay. + Idempotent: skips houses already enriched, per-house SAVEPOINT, so + each fire drains the next `limit` pending houses; window 00:00-01:00 + UTC = 03:00-04:00 МСК, before the 01:00+ UTC sweep block) """ from __future__ import annotations @@ -679,6 +687,49 @@ async def trigger_rosreestr_quarter_poll_run( return run_id +async def trigger_newbuilding_enrich_run(db: Session, schedule_row: dict[str, Any]) -> int | None: + """Создать scrape_runs + launch run_newbuilding_enrich в asyncio.create_task (#973). + + Nightly slice of the cian_newbuilding enrichment backfill: resolves cian_zhk_url from + house_sources.ext_id when missing, fetches the ЖК page (curl_cffi chrome120) and persists + houses_price_dynamics / house_reliability_checks / house_reviews. This is the standalone + schedule the inline cian full-sweep path never had — 306 fetchable cian houses on prod, + only ~3 enriched so far, so the backlog drains over successive nights. + + Mirrors trigger_sber_index_pull_run: claim run → create_task → mark_done/failed delegated + to tasks/newbuilding_enrich_backfill.run_newbuilding_enrich (which owns the run lifecycle). + + Idempotency: run_newbuilding_enrich → backfill_newbuilding_enrichment skips houses that + already have BOTH price_dynamics AND reliability rows (force=False) and uses a per-house + SAVEPOINT, so a re-fire never re-enriches done houses or double-appends on a partial run. + `limit` (default_params) bounds how many houses one fire fetches — anti-bot politeness; + the nightly cadence drains the rest. zombie-reap is inherited from the shared loop. + + Returns run_id или None (skip — already running). + """ + run_id = _claim_run(db, schedule_row) + if run_id is None: + return None + + params = schedule_row.get("default_params") or {} + + async def _run() -> None: + run_db = SessionLocal() + try: + from app.tasks.newbuilding_enrich_backfill import run_newbuilding_enrich + + await run_newbuilding_enrich(run_db, run_id=run_id, params=params) + except Exception: + logger.exception("scheduler: run_newbuilding_enrich crashed run_id=%d", run_id) + finally: + run_db.close() + + task = asyncio.create_task(_run()) + task.add_done_callback(lambda t: t.exception() if not t.cancelled() else None) + logger.info("scheduler: triggered newbuilding_enrich run_id=%d", run_id) + return run_id + + async def trigger_asking_to_sold_ratio_run(db: Session, schedule_row: dict[str, Any]) -> int | None: """Создать scrape_runs + launch recompute_asking_to_sold_ratios в executor (sync DB-only task). @@ -1003,6 +1054,8 @@ async def scheduler_loop() -> None: await trigger_sber_index_pull_run(db, sch) elif source == "rosreestr_quarter_poll": await trigger_rosreestr_quarter_poll_run(db, sch) + elif source == "newbuilding_enrich": + await trigger_newbuilding_enrich_run(db, sch) else: logger.warning("scheduler: unknown source=%s, skip", source) finally: diff --git a/tradein-mvp/backend/app/tasks/newbuilding_enrich_backfill.py b/tradein-mvp/backend/app/tasks/newbuilding_enrich_backfill.py index c83e9343..e244fc9d 100644 --- a/tradein-mvp/backend/app/tasks/newbuilding_enrich_backfill.py +++ b/tradein-mvp/backend/app/tasks/newbuilding_enrich_backfill.py @@ -71,6 +71,7 @@ from dataclasses import dataclass, field, fields from sqlalchemy import text from sqlalchemy.orm import Session +from app.services import scrape_runs as runs_mod from app.services.scraper_settings import get_scraper_delay logger = logging.getLogger(__name__) @@ -79,6 +80,7 @@ __all__ = [ "NewbuildingEnrichBackfillResult", "backfill_newbuilding_enrichment", "count_cian_newbuilding_houses", + "run_newbuilding_enrich", ] # Source tags written by the Cian newbuilding enrichment (kept in sync with @@ -642,3 +644,80 @@ async def _sleep_with_jitter(delay: float, idx: int, total: int, *, force: bool if not force and idx >= total - 1: return await asyncio.sleep(delay * random.uniform(0.8, 1.2)) + + +async def run_newbuilding_enrich( + db: Session, + *, + run_id: int, + params: dict, # type: ignore[type-arg] +) -> NewbuildingEnrichBackfillResult: + """Execute the newbuilding-enrichment backfill with run lifecycle management (#973). + + Thin scheduler wrapper around backfill_newbuilding_enrichment() — mirrors + tasks/yandex_address_backfill.run_yandex_address_backfill: emit a heartbeat before the + batch, delegate to the proven backfill, then mark the scrape_run done/failed. + + Idempotency is inherited from backfill_newbuilding_enrichment(): with force=False its + SELECT excludes houses that already have BOTH price_dynamics AND reliability rows, and + a per-house SAVEPOINT makes a partial/aborted run resume cleanly on the next tick. So + each nightly fire processes only the next slice of still-pending cian_newbuilding houses + (306 fetchable on prod, ~3 enriched so far — many nights to drain at a polite cadence). + + Per-tick bound: `limit` caps how many houses one fire fetches (anti-bot politeness; a + full 306-house sweep in one tick would hammer Cian). The scheduler re-fires nightly, so + the backlog drains over successive nights without operator intervention. + + Params (from default_params jsonb in scrape_schedules): + limit: int — max houses to process this fire (default 25). + force: bool — re-process already-enriched houses (UPSERT/dedup-guarded; default False). + request_delay_sec: float | None — seconds between fetches; None → get_scraper_delay('cian'). + """ + limit = int(params.get("limit", 25)) + force = bool(params.get("force", False)) + raw_delay = params.get("request_delay_sec") + request_delay_sec = float(raw_delay) if raw_delay is not None else None + + counters: dict[str, int] = { + "processed": 0, + "succeeded": 0, + "skipped_already_enriched": 0, + "failed_resolve": 0, + "failed_fetch": 0, + "failed_save": 0, + } + + try: + runs_mod.update_heartbeat(db, run_id, counters) + + result = await backfill_newbuilding_enrichment( + db, + limit=limit, + force=force, + request_delay_sec=request_delay_sec, + ) + + counters = result.to_dict() + runs_mod.mark_done(db, run_id, counters) + logger.info( + "scheduler: newbuilding_enrich run_id=%d done — processed=%d ok=%d skip=%d " + "resolve_fail=%d fetch_fail=%d save_fail=%d | rows pd=%d reliability=%d reviews=%d " + "| pending=%d %.1fs", + run_id, + result.processed, + result.succeeded, + result.skipped_already_enriched, + result.failed_resolve, + result.failed_fetch, + result.failed_save, + result.price_dynamics_rows, + result.reliability_rows, + result.review_rows, + result.cian_houses_pending, + result.duration_sec, + ) + return result + except Exception as exc: + logger.exception("scheduler: newbuilding_enrich run_id=%d failed", run_id) + runs_mod.mark_failed(db, run_id, str(exc)[:1000], counters) + raise diff --git a/tradein-mvp/backend/data/sql/103_scrape_schedules_seed_newbuilding_enrich.sql b/tradein-mvp/backend/data/sql/103_scrape_schedules_seed_newbuilding_enrich.sql new file mode 100644 index 00000000..f7c36e63 --- /dev/null +++ b/tradein-mvp/backend/data/sql/103_scrape_schedules_seed_newbuilding_enrich.sql @@ -0,0 +1,90 @@ +-- 103_scrape_schedules_seed_newbuilding_enrich.sql +-- Issue #973 (GG-форсайт, 950-E2) — seed scrape_schedules row for the nightly +-- newbuilding-enrichment backfill. +-- +-- Context: +-- The three newbuilding-enrichment tables (houses_price_dynamics, +-- house_reliability_checks, house_reviews) are populated only INLINE during the cian +-- full-sweep — there was never a standalone schedule for them. On prod 306 houses are +-- linked to ext_source='cian_newbuilding' (all fetchable) but only ~18 have price +-- dynamics, ~3 have reliability, 0 have reviews. This schedule drains that backlog +-- nightly, one bounded slice at a time. +-- +-- Components deployed together (same release as this migration): +-- 1. tasks/newbuilding_enrich_backfill.py — run_newbuilding_enrich() scheduler wrapper +-- (heartbeat → backfill_newbuilding_enrichment → mark_done/mark_failed). The +-- backfill itself (backfill_newbuilding_enrichment) shipped with #972. +-- 2. scheduler.py — trigger_newbuilding_enrich_run() + dispatch branch +-- `elif source == "newbuilding_enrich"` in scheduler_loop(). +-- 3. This migration — seeds the scrape_schedules row. +-- +-- Idempotency (inherited from backfill_newbuilding_enrichment, #972): +-- With force=False the backfill's SELECT excludes houses that already have BOTH +-- price_dynamics AND reliability rows, and each house is processed under a SAVEPOINT, +-- so a re-fire never re-enriches a done house nor double-appends on a partial run. +-- default_params.limit bounds how many houses one fire fetches (anti-bot politeness); +-- the nightly cadence drains the remaining backlog over successive nights. +-- +-- Schedule window 00:00-01:00 UTC (= 03:00-04:00 МСК): +-- Deep-night МСК slot, placed BEFORE the 01:00+ UTC sweep block (listing_source_snapshot +-- 01:00-02:00, avito/cian/yandex sweeps 02:00-05:00 UTC) so the Cian HTTP fetches of this +-- task do not overlap the daytime/early-morning SERP-sweeps and compete for the Cian +-- anti-bot budget. Per-house polite delay = get_scraper_delay('cian') (~5s) with ±20% +-- jitter; at limit=25 a fire is ~2-3 min of fetches. +-- +-- enabled = FALSE (seeded dormant, deliberately). As of 2026-06-13 every external-HTTP +-- scrape source on prod is enabled=false (scraping paused on purpose) except +-- deactivate_stale_avito (pure-internal). Seeding this dormant preserves that posture — +-- the schedule + trigger code ship ready, but do NOT begin fetching Cian until scraping is +-- resumed deliberately (avoids a lone external-HTTP source firing while the rest are paused). +-- TO ACTIVATE when scraping resumes: +-- UPDATE scrape_schedules SET enabled=true WHERE source='newbuilding_enrich'; +-- (ON CONFLICT DO NOTHING below will NOT change an already-existing row's enabled flag.) +-- +-- next_run_at bootstrapped to tomorrow 00:00 UTC — the scheduler will not fire immediately +-- on deploy (same pattern as 078/079/082/088/091/093/096). +-- +-- Idempotent: ON CONFLICT (source) DO NOTHING — safe to re-apply. +-- +-- Dependencies: +-- 052_scrape_schedules.sql (table + UNIQUE(source)) +-- scheduler.py change (trigger_newbuilding_enrich_run must be deployed) +-- tasks/newbuilding_enrich_backfill.py run_newbuilding_enrich() must be deployed (#973), +-- which delegates to backfill_newbuilding_enrichment() shipped with #972. +-- +-- Deploy order: +-- Apply after deploying the scheduler.py + task changes so the scheduler can dispatch +-- 'newbuilding_enrich' correctly on first fire. + +BEGIN; + +INSERT INTO scrape_schedules ( + source, + enabled, + window_start_hour, + window_end_hour, + next_run_at, + default_params +) +VALUES +( + 'newbuilding_enrich', + false, -- DORMANT: prod scraping paused — flip true to activate (see note above) + 0, -- window 00:00-01:00 UTC = 03:00-04:00 МСК + 1, + ((CURRENT_DATE + INTERVAL '1 day') + make_interval(hours => 0)) AT TIME ZONE 'UTC', + '{"limit": 25, "force": false}'::jsonb -- bounded per-fire slice; request_delay_sec omitted → get_scraper_delay('cian') +) +ON CONFLICT (source) DO NOTHING; + +COMMENT ON TABLE scrape_schedules IS + 'In-app scheduler config (заменяет cron-script setup). ' + 'Sources: avito_city_sweep, yandex_city_sweep (dormant, #561), ' + 'cian_history_backfill, rosreestr_dkp_import, listing_source_snapshot (#570), ' + 'asking_to_sold_ratio_refresh (#648), refresh_search_matview (#769), ' + 'yandex_address_backfill (#855, EKB pilot), ' + 'sber_index_pull (#887, monthly СберИндекс city-level price index), ' + 'rosreestr_quarter_poll (#888, monthly Rosreestr new-quarter availability check), ' + 'newbuilding_enrich (#973, nightly cian_newbuilding enrichment backfill slice).'; + +COMMIT; diff --git a/tradein-mvp/backend/tests/tasks/test_newbuilding_enrich_backfill.py b/tradein-mvp/backend/tests/tasks/test_newbuilding_enrich_backfill.py index 947d92b3..6f8f1bcc 100644 --- a/tradein-mvp/backend/tests/tasks/test_newbuilding_enrich_backfill.py +++ b/tradein-mvp/backend/tests/tasks/test_newbuilding_enrich_backfill.py @@ -24,6 +24,7 @@ from app.tasks.newbuilding_enrich_backfill import ( # noqa: E402 NewbuildingEnrichBackfillResult, _save_cian_reviews, backfill_newbuilding_enrichment, + run_newbuilding_enrich, ) # --------------------------------------------------------------------------- @@ -516,3 +517,203 @@ async def test_backfill_resolve_then_idempotent_rerun_no_reresolve() -> None: assert r2.succeeded == 0 assert len(db.price_dynamics) == 1 # no duplicate assert len(db.reliability) == 1 + + +# --------------------------------------------------------------------------- +# #973 (950-E2): nightly schedule — run_newbuilding_enrich wrapper, scheduler +# trigger/dispatch wiring, and the seed migration 103. +# --------------------------------------------------------------------------- + +import inspect # noqa: E402 +import re # noqa: E402 +from pathlib import Path # noqa: E402 + +from app.services import scheduler # noqa: E402 +from app.tasks import newbuilding_enrich_backfill as task_mod # noqa: E402 + +_SQL_DIR = Path(__file__).resolve().parents[2] / "data" / "sql" +_MIGRATION_103 = _SQL_DIR / "103_scrape_schedules_seed_newbuilding_enrich.sql" + + +# ── run_newbuilding_enrich wrapper: lifecycle + param plumbing ──────────────── + + +@pytest.mark.asyncio +async def test_run_wrapper_marks_done_and_passes_params(monkeypatch: pytest.MonkeyPatch) -> None: + """run_newbuilding_enrich emits heartbeat → delegates with parsed params → mark_done.""" + seen: dict = {} + + async def _fake_backfill(_db, *, limit, force, request_delay_sec): + seen.update(limit=limit, force=force, request_delay_sec=request_delay_sec) + return NewbuildingEnrichBackfillResult(processed=3, succeeded=2, price_dynamics_rows=2) + + monkeypatch.setattr(task_mod, "backfill_newbuilding_enrichment", _fake_backfill) + monkeypatch.setattr(task_mod.runs_mod, "update_heartbeat", lambda *a, **k: None) + marked: dict = {} + monkeypatch.setattr( + task_mod.runs_mod, + "mark_done", + lambda _db, run_id, counters: marked.update(run_id=run_id, counters=dict(counters)), + ) + monkeypatch.setattr(task_mod.runs_mod, "mark_failed", lambda *a, **k: None) + + out = await run_newbuilding_enrich( + object(), # type: ignore[arg-type] + run_id=55, + params={"limit": 7, "force": True, "request_delay_sec": 0}, + ) + + assert out.processed == 3 + # default_params plumbed through to the backfill. + assert seen == {"limit": 7, "force": True, "request_delay_sec": 0.0} + # run finalised via mark_done with the result counters. + assert marked["run_id"] == 55 + assert marked["counters"]["processed"] == 3 + assert marked["counters"]["succeeded"] == 2 + + +@pytest.mark.asyncio +async def test_run_wrapper_defaults_when_params_empty(monkeypatch: pytest.MonkeyPatch) -> None: + """Empty default_params → limit=25, force=False, request_delay_sec=None (→ scraper delay).""" + seen: dict = {} + + async def _fake_backfill(_db, *, limit, force, request_delay_sec): + seen.update(limit=limit, force=force, request_delay_sec=request_delay_sec) + return NewbuildingEnrichBackfillResult() + + monkeypatch.setattr(task_mod, "backfill_newbuilding_enrichment", _fake_backfill) + monkeypatch.setattr(task_mod.runs_mod, "update_heartbeat", lambda *a, **k: None) + monkeypatch.setattr(task_mod.runs_mod, "mark_done", lambda *a, **k: None) + monkeypatch.setattr(task_mod.runs_mod, "mark_failed", lambda *a, **k: None) + + await run_newbuilding_enrich(object(), run_id=1, params={}) # type: ignore[arg-type] + + assert seen == {"limit": 25, "force": False, "request_delay_sec": None} + + +@pytest.mark.asyncio +async def test_run_wrapper_marks_failed_and_reraises(monkeypatch: pytest.MonkeyPatch) -> None: + """Backfill raising → mark_failed + re-raise (no silent swallow).""" + + async def _boom(_db, **kwargs): + raise RuntimeError("cian exploded") + + monkeypatch.setattr(task_mod, "backfill_newbuilding_enrichment", _boom) + monkeypatch.setattr(task_mod.runs_mod, "update_heartbeat", lambda *a, **k: None) + monkeypatch.setattr(task_mod.runs_mod, "mark_done", lambda *a, **k: None) + failed: dict = {} + monkeypatch.setattr( + task_mod.runs_mod, + "mark_failed", + lambda _db, run_id, err, counters: failed.update(run_id=run_id, err=err), + ) + + with pytest.raises(RuntimeError, match="cian exploded"): + await run_newbuilding_enrich(object(), run_id=9, params={}) # type: ignore[arg-type] + + assert failed["run_id"] == 9 + assert "cian exploded" in failed["err"] + + +# ── Scheduler wiring: trigger fn claims + dispatches; skip when already running ── + + +def test_trigger_fn_exists_and_is_async() -> None: + fn = getattr(scheduler, "trigger_newbuilding_enrich_run", None) + assert fn is not None, "trigger_newbuilding_enrich_run missing from scheduler" + assert inspect.iscoroutinefunction(fn) + + +def test_dispatch_branch_wired() -> None: + """scheduler_loop dispatches source='newbuilding_enrich' to the trigger.""" + loop_src = inspect.getsource(scheduler.scheduler_loop) + assert 'source == "newbuilding_enrich"' in loop_src + assert "trigger_newbuilding_enrich_run(db, sch)" in loop_src + + +def test_trigger_claims_run_and_delegates_to_task() -> None: + """Trigger claims via _claim_run and delegates to run_newbuilding_enrich (async task).""" + trig_src = inspect.getsource(scheduler.trigger_newbuilding_enrich_run) + assert "_claim_run(db, schedule_row)" in trig_src + assert "run_newbuilding_enrich" in trig_src + assert "asyncio.create_task(" in trig_src + + +@pytest.mark.asyncio +async def test_trigger_claims_and_dispatches(monkeypatch: pytest.MonkeyPatch) -> None: + """Happy path: _claim_run returns a run_id → an asyncio task is spawned.""" + monkeypatch.setattr(scheduler, "_claim_run", lambda _db, _row: 4242) + spawned: list = [] + real_create_task = scheduler.asyncio.create_task + + def _spy_create_task(coro): + spawned.append(coro) + # Wrap the real coro so it actually runs (and closes the SessionLocal). + return real_create_task(coro) + + monkeypatch.setattr(scheduler.asyncio, "create_task", _spy_create_task) + # Stop the spawned _run() from touching a real DB / importing the heavy task. + monkeypatch.setattr(scheduler, "SessionLocal", lambda: MagicMock()) + + run_id = await scheduler.trigger_newbuilding_enrich_run( + MagicMock(), {"source": "newbuilding_enrich", "default_params": {}} + ) + + assert run_id == 4242 + assert len(spawned) == 1 + + +@pytest.mark.asyncio +async def test_trigger_skips_when_already_running(monkeypatch: pytest.MonkeyPatch) -> None: + """_claim_run None (already running / lock held) → trigger returns None, no task.""" + monkeypatch.setattr(scheduler, "_claim_run", lambda _db, _row: None) + spawned: list = [] + monkeypatch.setattr(scheduler.asyncio, "create_task", lambda coro: spawned.append(coro)) + + run_id = await scheduler.trigger_newbuilding_enrich_run( + MagicMock(), {"source": "newbuilding_enrich", "default_params": {}} + ) + + assert run_id is None + assert spawned == [] + + +# ── Migration 103 ───────────────────────────────────────────────────────────── + + +def test_migration_103_exists() -> None: + assert _MIGRATION_103.is_file(), f"missing migration: {_MIGRATION_103}" + + +def test_migration_103_seeds_correct_source_and_window() -> None: + sql = _MIGRATION_103.read_text("utf-8") + assert "'newbuilding_enrich'" in sql + insert_block = sql.split("INSERT INTO scrape_schedules")[1] + # Window 00:00-01:00 UTC = 03:00-04:00 МСК (start=0, end=1). + # Values may carry a trailing inline comment, so match "," at line start only. + assert re.search(r"^\s*0,", insert_block, re.MULTILINE), "window_start_hour 0 missing" + assert re.search(r"^\s*1,", insert_block, re.MULTILINE), "window_end_hour 1 missing" + + +def test_migration_103_is_idempotent() -> None: + sql = _MIGRATION_103.read_text("utf-8") + assert "ON CONFLICT (source) DO NOTHING" in sql + + +def test_migration_103_is_transactional() -> None: + sql = _MIGRATION_103.read_text("utf-8") + assert "BEGIN;" in sql + assert "COMMIT;" in sql + + +def test_migration_103_no_psycopg_trap() -> None: + """Plain seed (no bind params), but guard the :x::type trap anyway (psycopg v3).""" + sql = _MIGRATION_103.read_text("utf-8") + assert not re.search(r":\w+::", sql) + + +def test_migration_103_bootstraps_next_run_at() -> None: + """next_run_at bootstrapped to tomorrow so the scheduler does not fire on deploy.""" + sql = _MIGRATION_103.read_text("utf-8") + assert "next_run_at" in sql + assert "CURRENT_DATE + INTERVAL '1 day'" in sql -- 2.45.3