1635 lines
62 KiB
Python
1635 lines
62 KiB
Python
import json
|
||
import logging
|
||
import time
|
||
import hashlib
|
||
import uuid
|
||
from datetime import date, datetime, timezone
|
||
from typing import Any, Dict, List, Optional
|
||
|
||
from sqlalchemy import and_, case, exists, func, not_, or_, select
|
||
from sqlalchemy.exc import IntegrityError, OperationalError
|
||
from sqlalchemy.orm import class_mapper
|
||
|
||
from app import db
|
||
from app.timeutil import utc_now_iso, utc_now_naive
|
||
from app.models import (
|
||
WESP_SUPPRESS_SYNC_ENQUEUE,
|
||
Component,
|
||
ComponentLoadingTime,
|
||
FeedDispenser,
|
||
FeedMixer,
|
||
FeedingLocation,
|
||
FeedingPeriod,
|
||
FeedingPoint,
|
||
Ingredient,
|
||
LoadingReport,
|
||
LoadingReportComponent,
|
||
PeriodRecipe,
|
||
DailyTripSkip,
|
||
DailyIngredientSkip,
|
||
DailyUnloadingGroupSkip,
|
||
DailyIngredientReplacement,
|
||
DailyComponentNormAdjustment,
|
||
Recipe,
|
||
SyncClient,
|
||
SyncConflict,
|
||
SyncDelivery,
|
||
SyncEngineState,
|
||
SyncQueue,
|
||
Trip,
|
||
UnloadingGroup,
|
||
UnloadingReport,
|
||
UnloadingReportGroup,
|
||
)
|
||
from app.services.sync_record_data import get_record_data_for_sync
|
||
from config import Config
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
ALLOWED_SYNC_ACTIONS = {"create", "update", "delete", "delete_confirmed"}
|
||
REPORT_TABLES = {
|
||
"loading_report",
|
||
"unloading_report",
|
||
"loading_report_component",
|
||
"component_loading_time",
|
||
"unloading_report_group",
|
||
}
|
||
|
||
# Порядок: родительские отчёты до строк с FK на них (снапшот для новых узлов).
|
||
REPORT_SNAPSHOT_MODELS: tuple[type, ...] = (
|
||
LoadingReport,
|
||
LoadingReportComponent,
|
||
ComponentLoadingTime,
|
||
UnloadingReport,
|
||
UnloadingReportGroup,
|
||
)
|
||
# Порядок: FK — дочерние после родителей (FeedDispenser до FeedingPeriod; период до FeedingPoint).
|
||
SNAPSHOT_MODELS: tuple[type, ...] = (
|
||
Component,
|
||
Recipe,
|
||
Ingredient,
|
||
UnloadingGroup,
|
||
FeedMixer,
|
||
FeedDispenser,
|
||
FeedingLocation,
|
||
FeedingPeriod,
|
||
FeedingPoint,
|
||
PeriodRecipe,
|
||
DailyTripSkip,
|
||
DailyIngredientSkip,
|
||
DailyUnloadingGroupSkip,
|
||
DailyIngredientReplacement,
|
||
DailyComponentNormAdjustment,
|
||
Trip,
|
||
)
|
||
|
||
SNAPSHOT_MODELS = (
|
||
*SNAPSHOT_MODELS,
|
||
*REPORT_SNAPSHOT_MODELS,
|
||
)
|
||
SNAPSHOT_TABLE_ORDER: Dict[str, int] = {
|
||
m.__tablename__: idx for idx, m in enumerate(SNAPSHOT_MODELS)
|
||
}
|
||
SERVER_MASTER_TABLES = {
|
||
"component",
|
||
"recipe",
|
||
"ingredient",
|
||
"unloading_group",
|
||
"feed_mixer",
|
||
"feeding_location",
|
||
"feeding_period",
|
||
"feeding_point",
|
||
"feed_dispenser",
|
||
"period_recipes",
|
||
"daily_trip_skip",
|
||
"daily_ingredient_skip",
|
||
"daily_unloading_group_skip",
|
||
"daily_ingredient_replacement",
|
||
"daily_component_norm_adjustment",
|
||
"trip",
|
||
}
|
||
|
||
RECIPE_REFERENCE_TABLES = {"loading_report", "unloading_report"}
|
||
# Follow-up roadmap (phase 3): keep current queue-based pull in this iteration.
|
||
# Outbox/CDC or watermark delta sync should be implemented as a separate project phase.
|
||
|
||
|
||
def _is_sqlite_lock_message(exc: BaseException) -> bool:
|
||
parts: list[str] = [str(exc).lower()]
|
||
orig = getattr(exc, "orig", None)
|
||
if orig is not None:
|
||
parts.append(str(orig).lower())
|
||
m = " ".join(parts)
|
||
return bool(
|
||
"database is locked" in m
|
||
or "busy" in m
|
||
or ("locked" in m and ("sqlite" in m or "operational" in m))
|
||
)
|
||
|
||
|
||
def db_commit_with_retry(
|
||
session=None, attempts: int = 8, base_delay: float = 0.2
|
||
) -> None:
|
||
import time as _time
|
||
|
||
sess = session or db.session
|
||
for i in range(attempts):
|
||
try:
|
||
sess.commit()
|
||
return
|
||
except Exception as e: # pragma: no cover - retry path
|
||
if _is_sqlite_lock_message(e) and i < attempts - 1:
|
||
logger.warning(
|
||
"[SYNC-DB] commit locked, попытка %s/%s",
|
||
i + 1,
|
||
attempts,
|
||
)
|
||
_time.sleep(base_delay * (2**i))
|
||
continue
|
||
raise
|
||
|
||
|
||
def db_flush_with_retry(attempts: int = 12, base_delay: float = 0.04) -> None:
|
||
"""SQLite под нагрузкой sync pull + админка: flush иногда ловит database is locked."""
|
||
import time as _time
|
||
|
||
for i in range(attempts):
|
||
try:
|
||
db.session.flush()
|
||
return
|
||
except OperationalError as e:
|
||
orig = getattr(e, "orig", None)
|
||
msg = str(orig).lower() if orig is not None else str(e).lower()
|
||
if ("locked" in msg or "busy" in msg) and i < attempts - 1:
|
||
_time.sleep(base_delay * (1.45**i))
|
||
continue
|
||
try:
|
||
db.session.rollback()
|
||
except Exception:
|
||
pass
|
||
raise
|
||
|
||
|
||
def _ensure_sync_client_for_pull(
|
||
client_id: str,
|
||
client_name: Optional[str],
|
||
client_ip: Optional[str],
|
||
now,
|
||
) -> SyncClient:
|
||
"""Строка sync_clients по node_id: без гонки INSERT (SQLite UPSERT), re-open после soft-delete."""
|
||
name = (client_name or f"client-{client_id[:8]}")[:100]
|
||
ip_val = (client_ip or None)[:45] if client_ip else None
|
||
|
||
bind = db.session.get_bind()
|
||
tbl = SyncClient.__table__
|
||
|
||
if bind.dialect.name == "sqlite":
|
||
from sqlalchemy.dialects.sqlite import insert as sqlite_insert
|
||
|
||
ins = sqlite_insert(tbl).values(
|
||
id=str(uuid.uuid4()),
|
||
node_id=client_id,
|
||
client_name=name,
|
||
ip_address=ip_val,
|
||
port=80,
|
||
status="active",
|
||
last_seen=now,
|
||
total_syncs=0,
|
||
last_error=None,
|
||
is_enabled=True,
|
||
personal_snapshot_completed_at=None,
|
||
personal_snapshot_cursor=0,
|
||
created_at=now,
|
||
updated_at=now,
|
||
is_deleted=False,
|
||
deleted_at=None,
|
||
deleted_by=None,
|
||
)
|
||
ins = ins.on_conflict_do_update(
|
||
index_elements=[tbl.c.node_id],
|
||
set_={
|
||
"last_seen": ins.excluded.last_seen,
|
||
"updated_at": ins.excluded.updated_at,
|
||
"client_name": ins.excluded.client_name,
|
||
"ip_address": ins.excluded.ip_address,
|
||
"status": "active",
|
||
"is_enabled": True,
|
||
"is_deleted": False,
|
||
"deleted_at": None,
|
||
"deleted_by": None,
|
||
},
|
||
)
|
||
db.session.execute(ins)
|
||
db_flush_with_retry()
|
||
else:
|
||
client: Optional[SyncClient] = db.session.execute(
|
||
select(SyncClient).where(SyncClient.node_id == client_id)
|
||
).scalar_one_or_none()
|
||
if not client:
|
||
for _ in range(8):
|
||
try:
|
||
client = SyncClient(
|
||
node_id=client_id,
|
||
client_name=name,
|
||
ip_address=ip_val,
|
||
status="active",
|
||
is_enabled=True,
|
||
last_seen=now,
|
||
)
|
||
db.session.add(client)
|
||
db_flush_with_retry()
|
||
break
|
||
except IntegrityError:
|
||
db.session.rollback()
|
||
client = db.session.execute(
|
||
select(SyncClient).where(SyncClient.node_id == client_id)
|
||
).scalar_one_or_none()
|
||
if client is not None:
|
||
client.last_seen = now
|
||
if client_name:
|
||
client.client_name = name
|
||
if client_ip:
|
||
client.ip_address = ip_val
|
||
client.is_deleted = False
|
||
client.deleted_at = None
|
||
client.deleted_by = None
|
||
client.status = "active"
|
||
client.is_enabled = True
|
||
db_flush_with_retry()
|
||
break
|
||
if client is None:
|
||
raise RuntimeError(
|
||
"Не удалось зарегистрировать sync_clients по node_id после повторов"
|
||
)
|
||
else:
|
||
client.last_seen = now
|
||
if client_name:
|
||
client.client_name = name
|
||
if client_ip:
|
||
client.ip_address = ip_val
|
||
client.is_deleted = False
|
||
client.deleted_at = None
|
||
client.deleted_by = None
|
||
client.status = "active"
|
||
client.is_enabled = True
|
||
db_flush_with_retry()
|
||
|
||
row = db.session.execute(
|
||
select(SyncClient).where(SyncClient.node_id == client_id)
|
||
).scalar_one_or_none()
|
||
if row is None:
|
||
raise RuntimeError("sync_clients: строка не найдена после upsert")
|
||
return row
|
||
|
||
|
||
def _log_changes_statistics(changes: List[Dict[str, Any]], context: str) -> None:
|
||
if not changes:
|
||
return
|
||
stats: Dict[str, int] = {}
|
||
for change in changes:
|
||
key = f"{change.get('table_name', 'unknown')}.{change.get('action', 'unknown')}"
|
||
stats[key] = stats.get(key, 0) + 1
|
||
logger.info("[SYNC] %s stats: %s", context, stats)
|
||
|
||
|
||
def _sync_pull_changes_stats(changes: List[Dict[str, Any]]) -> str:
|
||
if not changes:
|
||
return "{}"
|
||
stats: Dict[str, int] = {}
|
||
for c in changes:
|
||
if not isinstance(c, dict):
|
||
continue
|
||
key = f"{c.get('table_name', '?')}.{c.get('action', '?')}"
|
||
stats[key] = stats.get(key, 0) + 1
|
||
s = str(stats)
|
||
return s[:400] + ("…" if len(s) > 400 else "")
|
||
|
||
|
||
def _sync_pull_preview_task_ids(changes: Any, limit: int = 5) -> str:
|
||
if not isinstance(changes, list) or not changes:
|
||
return "—"
|
||
parts: List[str] = []
|
||
for c in changes[:limit]:
|
||
if isinstance(c, dict):
|
||
tid = str(c.get("task_id") or c.get("id") or "")[:8]
|
||
parts.append(tid or "?")
|
||
suffix = "…" if len(changes) > limit else ""
|
||
return ",".join(parts) + suffix
|
||
|
||
|
||
def format_sync_pull_out_line(client_id: str, status_code: int, payload: Dict[str, Any]) -> str:
|
||
"""Одна строка [SYNC-PULL-OUT] — вызывать из app.routes.sync (видно в консоли рядом с [SYNC-PULL-START])."""
|
||
_cid = (client_id or "")[:12]
|
||
ch = payload.get("changes")
|
||
n = len(ch) if isinstance(ch, list) else 0
|
||
stats = _sync_pull_changes_stats(ch) if isinstance(ch, list) else "{}"
|
||
bp = payload.get("bootstrap_progress")
|
||
bp_s = "—"
|
||
if isinstance(bp, dict):
|
||
bp_s = f"{bp.get('phase')}:{bp.get('cursor')}/{bp.get('total_models')}"
|
||
return (
|
||
"[SYNC-PULL-OUT] node=%s… http=%s batch_id=%s total=%s has_more=%s remaining=%s "
|
||
"initial_sync=%s retry_after=%s bootstrap=%s task_ids=%s stats=%s"
|
||
% (
|
||
_cid,
|
||
status_code,
|
||
str(payload.get("batch_id") or "")[:32],
|
||
payload.get("total", n),
|
||
payload.get("has_more"),
|
||
payload.get("remaining_hint"),
|
||
payload.get("initial_sync_active"),
|
||
payload.get("retry_after_sec"),
|
||
bp_s,
|
||
_sync_pull_preview_task_ids(ch),
|
||
stats,
|
||
)
|
||
)
|
||
|
||
|
||
def requeue_stuck_processing(timeout_minutes: int = 15) -> int:
|
||
from datetime import timedelta
|
||
from sqlalchemy import text
|
||
|
||
# processed_at пишется через utc_now_naive(); сравнение с datetime.now() (локальное)
|
||
# давало ложный requeue (например UTC+3: «зависшие» через 5 мин вместо 5 ч).
|
||
cutoff = utc_now_naive() - timedelta(minutes=timeout_minutes)
|
||
stuck_filter = """
|
||
status = 'processing'
|
||
AND processed_at IS NOT NULL
|
||
AND processed_at < :cutoff
|
||
"""
|
||
with db.engine.connect() as conn:
|
||
count_row = conn.execute(
|
||
text(f"SELECT COUNT(*) FROM sync_queue WHERE {stuck_filter}"),
|
||
{"cutoff": cutoff},
|
||
).fetchone()
|
||
n = int(count_row[0] if count_row else 0)
|
||
if n <= 0:
|
||
return 0
|
||
if n > 5000:
|
||
logger.warning(
|
||
"[SYNC] requeue_stuck_processing: зависших задач %s (timeout %s мин) — массовый requeue",
|
||
n,
|
||
timeout_minutes,
|
||
)
|
||
# Универсальные задачи: без сброса delivery клиент не увидит задачу снова.
|
||
# Подзапрос вместо IN (:t0..:t39000) — лимит SQLite на число bind-параметров.
|
||
conn.execute(
|
||
text(
|
||
f"""
|
||
DELETE FROM sync_delivery
|
||
WHERE task_id IN (
|
||
SELECT id FROM sync_queue WHERE {stuck_filter}
|
||
)
|
||
"""
|
||
),
|
||
{"cutoff": cutoff},
|
||
)
|
||
result = conn.execute(
|
||
text(
|
||
f"""
|
||
UPDATE sync_queue
|
||
SET status = 'pending',
|
||
processed_at = NULL,
|
||
retry_count = MIN((COALESCE(retry_count, 0) + 1), COALESCE(max_retries, 3))
|
||
WHERE {stuck_filter}
|
||
"""
|
||
),
|
||
{"cutoff": cutoff},
|
||
)
|
||
conn.commit()
|
||
rc = getattr(result, "rowcount", -1)
|
||
try:
|
||
updated = int(rc)
|
||
except (TypeError, ValueError):
|
||
updated = -1
|
||
# SQLite часто отдаёт rowcount=-1/0 при массовом UPDATE — тогда ориентируемся на COUNT.
|
||
if updated < 0:
|
||
updated = n
|
||
elif updated == 0 and n > 0:
|
||
updated = n
|
||
logger.info(
|
||
"[SYNC] requeue_stuck_processing: найдено %s, processing→pending %s (timeout %s мин)",
|
||
n,
|
||
updated,
|
||
timeout_minutes,
|
||
)
|
||
return updated
|
||
|
||
|
||
def _clear_sync_delivery_for_tasks(task_ids: List[str]) -> None:
|
||
"""Сброс delivery для повторной выдачи универсальных задач (см. requeue_stuck_processing)."""
|
||
if not task_ids:
|
||
return
|
||
with db.session.no_autoflush:
|
||
rows = list(
|
||
db.session.execute(
|
||
select(SyncDelivery).where(SyncDelivery.task_id.in_(task_ids))
|
||
).scalars().all()
|
||
)
|
||
for row in rows:
|
||
db.session.delete(row)
|
||
|
||
|
||
def enqueue_sync_queue_task(
|
||
table_name: str,
|
||
record_id: str,
|
||
action: str,
|
||
*,
|
||
priority: int = 3,
|
||
target_node_id: Optional[str] = None,
|
||
) -> None:
|
||
"""
|
||
Поставить задачу в sync_queue (как create_sync_task в legacy/proga_monolith).
|
||
|
||
Идемпотентно: при совпадении (table_name, record_id, action, target_node_id)
|
||
существующая запись переводится в pending и при необходимости снимается soft-delete.
|
||
"""
|
||
if action not in ALLOWED_SYNC_ACTIONS:
|
||
logger.warning(
|
||
"[SYNC] enqueue: неизвестное action=%s для %s.%s",
|
||
action,
|
||
table_name,
|
||
record_id,
|
||
)
|
||
|
||
now = utc_now_naive()
|
||
filters = [
|
||
SyncQueue.table_name == table_name,
|
||
SyncQueue.record_id == record_id,
|
||
SyncQueue.action == action,
|
||
]
|
||
if target_node_id is None:
|
||
filters.append(SyncQueue.target_node_id.is_(None))
|
||
else:
|
||
filters.append(SyncQueue.target_node_id == target_node_id)
|
||
|
||
# after_insert/after_update вызывают эту функцию во время flush; execute() иначе
|
||
# включает autoflush и даёт SAWarning «add() внутри flush».
|
||
with db.session.no_autoflush:
|
||
# SQLite (и др.): в UNIQUE несколько NULL — допускаются дубли с target_node_id IS NULL;
|
||
# scalar_one_or_none() падает с MultipleResultsFound.
|
||
_prefer_active = case(
|
||
(or_(SyncQueue.is_deleted.is_(False), SyncQueue.is_deleted.is_(None)), 0),
|
||
else_=1,
|
||
)
|
||
rows = list(
|
||
db.session.execute(
|
||
select(SyncQueue)
|
||
.where(and_(*filters))
|
||
.order_by(_prefer_active, SyncQueue.created_at.desc())
|
||
).scalars().all()
|
||
)
|
||
if rows:
|
||
existing = rows[0]
|
||
for dup in rows[1:]:
|
||
if dup.is_deleted is True:
|
||
continue
|
||
dup.soft_delete(deleted_by_user="sync_dedup")
|
||
was_redeliverable = existing.status in ("processing", "completed")
|
||
existing.status = "pending"
|
||
existing.priority = max(int(existing.priority or 1), int(priority))
|
||
existing.error_message = None
|
||
existing.processed_at = None
|
||
existing.completed_at = None
|
||
if getattr(existing, "is_deleted", None):
|
||
existing.is_deleted = False
|
||
existing.deleted_at = None
|
||
existing.deleted_by = None
|
||
if was_redeliverable and existing.target_node_id is None:
|
||
_clear_sync_delivery_for_tasks([existing.id])
|
||
return
|
||
|
||
db.session.add(
|
||
SyncQueue(
|
||
table_name=table_name,
|
||
record_id=record_id,
|
||
action=action,
|
||
status="pending",
|
||
target_node_id=target_node_id,
|
||
source_node_id=None,
|
||
priority=priority,
|
||
retry_count=0,
|
||
max_retries=3,
|
||
created_at=now,
|
||
)
|
||
)
|
||
|
||
|
||
def _rows_for_snapshot(model) -> List[Any]:
|
||
"""Строки таблицы для снапшота. Нужен .unique(): у FeedingPeriod.recipes lazy='joined' (JOIN дублирует parent)."""
|
||
query = select(model)
|
||
if hasattr(model, "is_deleted"):
|
||
query = query.where(model.is_deleted.is_(False))
|
||
return list(db.session.execute(query).unique().scalars().all())
|
||
|
||
|
||
def bootstrap_universal_sync() -> int:
|
||
"""Универсальные задачи create для всех строк (как legacy bootstrap_universal_sync)."""
|
||
models = list(SNAPSHOT_MODELS)
|
||
created = 0
|
||
for model in models:
|
||
for obj in _rows_for_snapshot(model):
|
||
if getattr(model, "__tablename__", "") == "period_recipes":
|
||
record_id = f"{obj.period_id}:{obj.recipe_id}"
|
||
else:
|
||
record_id = obj.id
|
||
exists = db.session.execute(
|
||
select(SyncQueue)
|
||
.where(
|
||
SyncQueue.table_name == model.__tablename__,
|
||
SyncQueue.record_id == record_id,
|
||
SyncQueue.action == "create",
|
||
SyncQueue.target_node_id.is_(None),
|
||
SyncQueue.is_deleted.is_(False),
|
||
)
|
||
.order_by(SyncQueue.created_at.desc())
|
||
.limit(1)
|
||
).scalars().first()
|
||
if exists:
|
||
continue
|
||
enqueue_sync_queue_task(
|
||
model.__tablename__, record_id, "create", priority=2, target_node_id=None
|
||
)
|
||
created += 1
|
||
return created
|
||
|
||
|
||
def enqueue_personal_snapshot(client_id: str) -> int:
|
||
"""Персональный снапшот для нового узла (legacy MIN_SNAPSHOT_TASKS)."""
|
||
models = list(SNAPSHOT_MODELS)
|
||
created = 0
|
||
for model in models:
|
||
for obj in _rows_for_snapshot(model):
|
||
if getattr(model, "__tablename__", "") == "period_recipes":
|
||
record_id = f"{obj.period_id}:{obj.recipe_id}"
|
||
else:
|
||
record_id = obj.id
|
||
enqueue_sync_queue_task(
|
||
model.__tablename__, record_id, "create", priority=2, target_node_id=client_id
|
||
)
|
||
created += 1
|
||
return created
|
||
|
||
|
||
def _personal_enqueue_done(client: SyncClient) -> bool:
|
||
return int(client.personal_snapshot_cursor or 0) >= len(SNAPSHOT_MODELS)
|
||
|
||
|
||
def _personal_tasks_in_flight(client_id: str) -> int:
|
||
return int(
|
||
db.session.scalar(
|
||
select(func.count()).select_from(SyncQueue).where(
|
||
SyncQueue.target_node_id == client_id,
|
||
SyncQueue.status.in_(("pending", "processing")),
|
||
SyncQueue.is_deleted.is_(False),
|
||
)
|
||
)
|
||
or 0
|
||
)
|
||
|
||
|
||
def _client_initial_sync_active(client_id: str, client: SyncClient) -> bool:
|
||
if not _personal_enqueue_done(client):
|
||
return True
|
||
return _personal_tasks_in_flight(client_id) > 0
|
||
|
||
|
||
def _maybe_mark_personal_snapshot_completed(client: SyncClient, client_id: str) -> None:
|
||
"""Фиксирует завершение первичной синхронизации только когда enqueue done и все personal задачи confirmed."""
|
||
if client.personal_snapshot_completed_at is not None:
|
||
return
|
||
if not _personal_enqueue_done(client):
|
||
return
|
||
if _personal_tasks_in_flight(client_id) > 0:
|
||
return
|
||
client.personal_snapshot_completed_at = utc_now_naive()
|
||
|
||
|
||
def _sort_tasks_snapshot_order(tasks: List[SyncQueue]) -> List[SyncQueue]:
|
||
def _key(task: SyncQueue) -> tuple:
|
||
order = SNAPSHOT_TABLE_ORDER.get(task.table_name or "", 999)
|
||
created = task.created_at or datetime.min
|
||
return (order, created)
|
||
|
||
return sorted(tasks, key=_key)
|
||
|
||
|
||
def _pull_response_meta(
|
||
client_id: str, client: SyncClient, *, retry_after: int
|
||
) -> Dict[str, Any]:
|
||
initial_active = _client_initial_sync_active(client_id, client)
|
||
personal_remaining = _personal_tasks_in_flight(client_id)
|
||
return {
|
||
"server_now": utc_now_iso(),
|
||
"timestamp": utc_now_iso(),
|
||
"retry_after_sec": retry_after,
|
||
"initial_sync_active": initial_active,
|
||
"remaining_hint": personal_remaining,
|
||
"has_more": initial_active or personal_remaining > 0,
|
||
}
|
||
|
||
|
||
def _universal_queue_seeded() -> bool:
|
||
"""В очереди уже есть универсальные задачи (любой статус).
|
||
|
||
Раньше проверяли только pending — после первого pull всё уходит в processing,
|
||
pending=0 → повторный bootstrap_universal_sync() на каждом /pull (минуты, SQLite lock).
|
||
"""
|
||
cnt = db.session.scalar(
|
||
select(func.count()).select_from(SyncQueue).where(
|
||
SyncQueue.target_node_id.is_(None),
|
||
SyncQueue.is_deleted.is_(False),
|
||
)
|
||
)
|
||
return (cnt or 0) > 0
|
||
|
||
|
||
def _get_engine_state() -> SyncEngineState:
|
||
state = db.session.get(SyncEngineState, 1)
|
||
if state is None:
|
||
state = SyncEngineState(id=1)
|
||
db.session.add(state)
|
||
db_flush_with_retry()
|
||
return state
|
||
|
||
|
||
def _model_record_id(model: Any, obj: Any) -> str:
|
||
if getattr(model, "__tablename__", "") == "period_recipes":
|
||
return f"{obj.period_id}:{obj.recipe_id}"
|
||
return str(obj.id)
|
||
|
||
|
||
def _enqueue_snapshot_model(model: Any, target_node_id: Optional[str]) -> int:
|
||
created = 0
|
||
for obj in _rows_for_snapshot(model):
|
||
enqueue_sync_queue_task(
|
||
model.__tablename__,
|
||
_model_record_id(model, obj),
|
||
"create",
|
||
priority=2,
|
||
target_node_id=target_node_id,
|
||
)
|
||
created += 1
|
||
return created
|
||
|
||
|
||
def _bootstrap_universal_step(state: SyncEngineState, *, max_models: int) -> Dict[str, Any]:
|
||
cursor = int(state.universal_bootstrap_cursor or 0)
|
||
total = len(SNAPSHOT_MODELS)
|
||
created = 0
|
||
processed = 0
|
||
while cursor < total and processed < max(1, max_models):
|
||
model = SNAPSHOT_MODELS[cursor]
|
||
created += _enqueue_snapshot_model(model, None)
|
||
cursor += 1
|
||
processed += 1
|
||
done = cursor >= total
|
||
state.universal_bootstrap_cursor = 0 if done else cursor
|
||
if done and state.universal_bootstrap_completed_at is None:
|
||
state.universal_bootstrap_completed_at = utc_now_naive()
|
||
state.universal_bootstrap_last_error = None
|
||
return {"done": done, "created": created, "cursor": cursor, "total": total}
|
||
|
||
|
||
def _enqueue_personal_snapshot_step(
|
||
client: SyncClient, *, max_models: int
|
||
) -> Dict[str, Any]:
|
||
cursor = int(client.personal_snapshot_cursor or 0)
|
||
total = len(SNAPSHOT_MODELS)
|
||
created = 0
|
||
processed = 0
|
||
while cursor < total and processed < max(1, max_models):
|
||
model = SNAPSHOT_MODELS[cursor]
|
||
created += _enqueue_snapshot_model(model, client.node_id)
|
||
cursor += 1
|
||
processed += 1
|
||
done = cursor >= total
|
||
client.personal_snapshot_cursor = total if done else cursor
|
||
return {"done": done, "created": created, "cursor": cursor, "total": total}
|
||
|
||
|
||
def _batch_id(client_id: str, task_ids: List[str]) -> str:
|
||
payload = f"{client_id}:{','.join(task_ids)}".encode("utf-8")
|
||
return hashlib.sha1(payload).hexdigest()
|
||
|
||
|
||
def create_sync_conflict(
|
||
table_name: str,
|
||
record_id: str,
|
||
conflict_type: str,
|
||
local_data: Optional[Dict[str, Any]],
|
||
remote_data: Optional[Dict[str, Any]],
|
||
resolution: str = "pending",
|
||
) -> Optional[SyncConflict]:
|
||
try:
|
||
conflict = SyncConflict(
|
||
table_name=table_name,
|
||
record_id=record_id,
|
||
conflict_type=conflict_type,
|
||
local_data=json.dumps(local_data, ensure_ascii=False) if local_data else None,
|
||
remote_data=json.dumps(remote_data, ensure_ascii=False) if remote_data else None,
|
||
resolution=resolution,
|
||
)
|
||
db.session.add(conflict)
|
||
db.session.flush()
|
||
return conflict
|
||
except Exception:
|
||
logger.exception("Ошибка создания конфликта синхронизации")
|
||
return None
|
||
|
||
|
||
def _validate_sync_changes_payload(changes: Any) -> Optional[str]:
|
||
if not isinstance(changes, list):
|
||
return "changes должен быть списком"
|
||
|
||
for idx, change in enumerate(changes):
|
||
if not isinstance(change, dict):
|
||
return f"changes[{idx}] должен быть объектом"
|
||
missing = [f for f in ("table_name", "record_id", "action", "data") if f not in change]
|
||
if missing:
|
||
return f"changes[{idx}] отсутствуют поля: {', '.join(missing)}"
|
||
if not isinstance(change["table_name"], str) or not change["table_name"].strip():
|
||
return f"changes[{idx}].table_name должен быть непустой строкой"
|
||
if not isinstance(change["record_id"], str) or not change["record_id"].strip():
|
||
return f"changes[{idx}].record_id должен быть непустой строкой"
|
||
if change["action"] not in ALLOWED_SYNC_ACTIONS:
|
||
return (
|
||
f"changes[{idx}].action должен быть одним из: "
|
||
f"{', '.join(sorted(ALLOWED_SYNC_ACTIONS))}"
|
||
)
|
||
if not isinstance(change["data"], dict):
|
||
return f"changes[{idx}].data должен быть объектом"
|
||
return None
|
||
|
||
|
||
def _check_for_sync_conflict(
|
||
table_name: str, record_id: str, client_data: Dict[str, Any]
|
||
) -> Optional[Dict[str, Any]]:
|
||
try:
|
||
server_record = get_record_data_for_sync(table_name, record_id)
|
||
if not server_record:
|
||
return None
|
||
server_version = server_record.get("version", 1)
|
||
client_version = client_data.get("version", 1)
|
||
if server_version == client_version:
|
||
return None
|
||
return {
|
||
"table_name": table_name,
|
||
"record_id": record_id,
|
||
"server_version": server_version,
|
||
"client_version": client_version,
|
||
"server_data": server_record,
|
||
"client_data": client_data,
|
||
"conflict_type": "version_mismatch",
|
||
}
|
||
except Exception:
|
||
logger.exception("Ошибка проверки конфликта")
|
||
return None
|
||
|
||
|
||
def _is_uuid_text(value: Any) -> bool:
|
||
try:
|
||
uuid.UUID(str(value))
|
||
return True
|
||
except Exception:
|
||
return False
|
||
|
||
|
||
def _column_keys_for_model(model: type) -> set[str]:
|
||
"""Только колонки ORM: payload из get_object_data может содержать лишние ключи (legacy soft-delete)."""
|
||
return {p.key for p in class_mapper(model).column_attrs}
|
||
|
||
|
||
def _parse_sync_iso_datetime(s: str) -> datetime:
|
||
"""JSON pull/push: даты с сервера — ISO-строки; SQLite DateTime в ORM ждёт datetime."""
|
||
t = (s or "").strip()
|
||
if not t:
|
||
raise ValueError("empty datetime string")
|
||
if t.endswith("Z"):
|
||
t = t[:-1] + "+00:00"
|
||
if "T" not in t and len(t) >= 10 and t[4] == "-" and t[7] == "-":
|
||
return datetime.strptime(t[:10], "%Y-%m-%d")
|
||
t_norm = t.replace(" ", "T", 1)
|
||
try:
|
||
dt = datetime.fromisoformat(t_norm)
|
||
except ValueError:
|
||
dt = datetime.strptime(t_norm[:19], "%Y-%m-%dT%H:%M:%S")
|
||
if dt.tzinfo is not None:
|
||
dt = dt.astimezone(timezone.utc).replace(tzinfo=None)
|
||
return dt
|
||
|
||
|
||
def _parse_sync_iso_date(s: str) -> date:
|
||
t = (s or "").strip()
|
||
if not t:
|
||
raise ValueError("empty date string")
|
||
if "T" in t or (len(t) > 10 and t[10] in " T"):
|
||
return _parse_sync_iso_datetime(t).date()
|
||
return datetime.strptime(t[:10], "%Y-%m-%d").date()
|
||
|
||
|
||
def _coerce_sync_payload_datetimes(model: type, payload: Dict[str, Any]) -> None:
|
||
mp = class_mapper(model)
|
||
for key in list(payload.keys()):
|
||
val = payload[key]
|
||
if val is None or not isinstance(val, str):
|
||
continue
|
||
try:
|
||
attr = mp.attrs.get(key)
|
||
if attr is None or not getattr(attr, "columns", None):
|
||
continue
|
||
py_t = attr.columns[0].type.python_type
|
||
except (NotImplementedError, AttributeError, KeyError):
|
||
continue
|
||
if not isinstance(py_t, type):
|
||
continue
|
||
try:
|
||
if issubclass(py_t, datetime):
|
||
payload[key] = _parse_sync_iso_datetime(val)
|
||
elif issubclass(py_t, date):
|
||
payload[key] = _parse_sync_iso_date(val)
|
||
except Exception:
|
||
continue
|
||
|
||
|
||
def _payload_for_model_apply(
|
||
model: type, table_name: str, record_id: str, data: Dict[str, Any]
|
||
) -> Dict[str, Any]:
|
||
keys = _column_keys_for_model(model)
|
||
out = {k: v for k, v in data.items() if k in keys}
|
||
if table_name == "period_recipes" and ":" in record_id:
|
||
pid, rid = record_id.split(":", 1)
|
||
out.setdefault("period_id", pid)
|
||
out.setdefault("recipe_id", rid)
|
||
_coerce_sync_payload_datetimes(model, out)
|
||
return out
|
||
|
||
|
||
def _apply_sync_change(table_name: str, record_id: str, action: str, data: Dict[str, Any]) -> Dict[str, Any]:
|
||
model_map = {
|
||
"component": Component,
|
||
"recipe": Recipe,
|
||
"ingredient": Ingredient,
|
||
"unloading_group": UnloadingGroup,
|
||
"loading_report": LoadingReport,
|
||
"loading_report_component": LoadingReportComponent,
|
||
"component_loading_time": ComponentLoadingTime,
|
||
"unloading_report": UnloadingReport,
|
||
"unloading_report_group": UnloadingReportGroup,
|
||
"feed_dispenser": FeedDispenser,
|
||
"feeding_period": FeedingPeriod,
|
||
"period_recipes": PeriodRecipe,
|
||
"daily_trip_skip": DailyTripSkip,
|
||
"daily_ingredient_skip": DailyIngredientSkip,
|
||
"daily_unloading_group_skip": DailyUnloadingGroupSkip,
|
||
"daily_ingredient_replacement": DailyIngredientReplacement,
|
||
"daily_component_norm_adjustment": DailyComponentNormAdjustment,
|
||
"feed_mixer": FeedMixer,
|
||
"feeding_location": FeedingLocation,
|
||
"feeding_point": FeedingPoint,
|
||
"trip": Trip,
|
||
}
|
||
model = model_map.get(table_name)
|
||
if not model:
|
||
return {"success": False, "error": "Неизвестная таблица"}
|
||
payload = _payload_for_model_apply(model, table_name, record_id, data)
|
||
if action in ("create", "update") and table_name in RECIPE_REFERENCE_TABLES:
|
||
rid = payload.get("recipe_id") or data.get("recipe_id")
|
||
if not _is_uuid_text(rid):
|
||
return {"success": False, "error": "recipe_id must be UUID"}
|
||
|
||
try:
|
||
# SAVEPOINT: при ошибке flush сессия остаётся пригодной для следующих строк батча
|
||
# (клиент: apply_batch; сервер: process_push). Без rollback вложенной транзакции
|
||
# внешняя транзакция часто в «rollback-only», и остальные apply ломают финальный commit.
|
||
is_remote_pull = bool(db.session.info.get(WESP_SUPPRESS_SYNC_ENQUEUE))
|
||
with db.session.begin_nested():
|
||
if table_name == "period_recipes":
|
||
period_id, recipe_id = record_id.split(":", 1)
|
||
existing_record = db.session.execute(
|
||
select(model).where(model.period_id == period_id, model.recipe_id == recipe_id)
|
||
).scalar_one_or_none()
|
||
else:
|
||
search_id = payload.get("id", record_id)
|
||
existing_record = db.session.get(model, search_id) or db.session.get(model, record_id)
|
||
|
||
if existing_record:
|
||
if action in ("create", "update"):
|
||
for key, value in payload.items():
|
||
setattr(existing_record, key, value)
|
||
if hasattr(existing_record, "updated_at") and not is_remote_pull:
|
||
existing_record.updated_at = utc_now_naive()
|
||
if hasattr(existing_record, "version") and "version" not in payload:
|
||
existing_record.version += 1
|
||
elif action == "delete":
|
||
if hasattr(existing_record, "is_deleted") and getattr(existing_record, "is_deleted"):
|
||
pass
|
||
elif hasattr(existing_record, "soft_delete"):
|
||
existing_record.soft_delete(deleted_by_user="sync", reason="sync", request=None)
|
||
else:
|
||
db.session.delete(existing_record)
|
||
else:
|
||
if action in ("create", "update"):
|
||
record = model(**payload)
|
||
if hasattr(record, "id"):
|
||
record.id = record_id
|
||
if hasattr(record, "updated_at") and not is_remote_pull:
|
||
record.updated_at = utc_now_naive()
|
||
db.session.add(record)
|
||
|
||
# Каскад soft-delete — через after_update listeners (models/__init__.py),
|
||
# чтобы не дублировать enqueue/SQL при soft_delete() и setattr(is_deleted=True).
|
||
|
||
db.session.flush()
|
||
return {"success": True}
|
||
except Exception as e:
|
||
logger.exception("Ошибка apply_sync_change")
|
||
return {"success": False, "error": str(e)}
|
||
|
||
|
||
def build_consistency_snapshot() -> Dict[str, Any]:
|
||
model_map = {
|
||
"component": Component,
|
||
"recipe": Recipe,
|
||
"ingredient": Ingredient,
|
||
"feeding_period": FeedingPeriod,
|
||
"period_recipes": PeriodRecipe,
|
||
}
|
||
out: Dict[str, Any] = {"master_tables": {}, "generated_at": utc_now_iso()}
|
||
for table_name, model in model_map.items():
|
||
rows = _rows_for_snapshot(model)
|
||
serialized: List[str] = []
|
||
for row in rows:
|
||
record_id = _model_record_id(model, row)
|
||
version = int(getattr(row, "version", 1) or 1)
|
||
content_hash = str(getattr(row, "content_hash", "") or "")
|
||
serialized.append(f"{record_id}:{version}:{content_hash}")
|
||
serialized.sort()
|
||
digest = hashlib.sha256(("|".join(serialized)).encode("utf-8")).hexdigest()
|
||
out["master_tables"][table_name] = {"count": len(serialized), "digest": digest}
|
||
|
||
invalid_recipe_refs = int(
|
||
db.session.scalar(
|
||
select(func.count()).select_from(LoadingReport).where(
|
||
or_(
|
||
LoadingReport.recipe_id.is_(None),
|
||
func.length(LoadingReport.recipe_id) != 36,
|
||
)
|
||
)
|
||
)
|
||
or 0
|
||
)
|
||
invalid_recipe_refs += int(
|
||
db.session.scalar(
|
||
select(func.count()).select_from(UnloadingReport).where(
|
||
or_(
|
||
UnloadingReport.recipe_id.is_(None),
|
||
func.length(UnloadingReport.recipe_id) != 36,
|
||
)
|
||
)
|
||
)
|
||
or 0
|
||
)
|
||
out["recipe_uuid_reference_issues"] = invalid_recipe_refs
|
||
out["guarantees"] = {
|
||
"master_recipes_db_single_writer": True,
|
||
"reports_aggregated_on_server": True,
|
||
"sqlite_files_not_guaranteed_byte_identical": True,
|
||
"verification": "Диагностика: GET /api/sync/consistency (auth).",
|
||
}
|
||
return out
|
||
|
||
|
||
def apply_sync_change(table_name: str, record_id: str, action: str, data: Dict[str, Any]) -> Dict[str, Any]:
|
||
"""Публичная обёртка для применения одного изменения (resolve конфликтов, тесты)."""
|
||
return _apply_sync_change(table_name, record_id, action, data)
|
||
|
||
|
||
class SyncManager:
|
||
@staticmethod
|
||
def process_pull(
|
||
client_id: str,
|
||
limit: Optional[int] = None,
|
||
*,
|
||
client_name: Optional[str] = None,
|
||
client_ip: Optional[str] = None,
|
||
initial_sync_header: bool = False,
|
||
) -> Dict[str, Any]:
|
||
now = utc_now_naive()
|
||
_cid = (client_id or "")[:12]
|
||
logger.info(
|
||
"[SYNC-PULL] старт node_id=%s… limit=%s initial_hdr=%s",
|
||
_cid,
|
||
limit,
|
||
initial_sync_header,
|
||
)
|
||
|
||
try:
|
||
requeue_stuck_processing(
|
||
timeout_minutes=getattr(Config, "SYNC_REQUEUE_TIMEOUT_MINUTES", 15)
|
||
)
|
||
except Exception:
|
||
logger.exception("[SYNC-PULL] Ошибка requeue_stuck_processing")
|
||
|
||
client = _ensure_sync_client_for_pull(
|
||
client_id, client_name, client_ip, now
|
||
)
|
||
|
||
logger.info("[SYNC-PULL] клиент в реестре id=%s…", str(getattr(client, "id", ""))[:8])
|
||
|
||
state = _get_engine_state()
|
||
if state.universal_bootstrap_completed_at is None and _universal_queue_seeded():
|
||
state.universal_bootstrap_completed_at = now
|
||
state.universal_bootstrap_last_error = None
|
||
|
||
initial_sync_active = _client_initial_sync_active(client_id, client)
|
||
needs_personal_enqueue = not _personal_enqueue_done(client)
|
||
retry_after = int(getattr(Config, "SYNC_PULL_RETRY_AFTER_SEC", 2))
|
||
bootstrap_models_per_pull = int(
|
||
getattr(Config, "SYNC_PULL_BOOTSTRAP_MODELS_PER_PULL", 1)
|
||
)
|
||
personal_models_per_pull = int(
|
||
getattr(Config, "SYNC_PULL_SNAPSHOT_MODELS_PER_PULL", 1)
|
||
)
|
||
if initial_sync_header:
|
||
bootstrap_models_per_pull = int(
|
||
getattr(Config, "SYNC_PULL_BOOTSTRAP_MODELS_PER_PULL_INITIAL", 5)
|
||
)
|
||
personal_models_per_pull = int(
|
||
getattr(Config, "SYNC_PULL_SNAPSHOT_MODELS_PER_PULL_INITIAL", 5)
|
||
)
|
||
|
||
pull_limit = int(limit or getattr(Config, "SYNC_PULL_LIMIT", 100))
|
||
if pull_limit <= 0:
|
||
pull_limit = 100
|
||
|
||
def _select_tasks(*, personal_only: bool) -> List[SyncQueue]:
|
||
delivery_subq = (
|
||
select(SyncDelivery.id)
|
||
.filter(SyncDelivery.client_id == client.id, SyncDelivery.task_id == SyncQueue.id)
|
||
.correlate(SyncQueue)
|
||
)
|
||
filters = [SyncQueue.is_deleted.is_(False)]
|
||
if personal_only:
|
||
filters.extend(
|
||
[
|
||
SyncQueue.status == "pending",
|
||
SyncQueue.target_node_id == client_id,
|
||
]
|
||
)
|
||
else:
|
||
# Универсальные задачи: выдаём и в processing, если этому клиенту ещё не доставляли
|
||
# (иначе первый pull блокирует остальных до requeue).
|
||
filters.append(
|
||
or_(
|
||
and_(
|
||
SyncQueue.target_node_id == client_id,
|
||
SyncQueue.status == "pending",
|
||
),
|
||
and_(
|
||
SyncQueue.target_node_id.is_(None),
|
||
not_(exists(delivery_subq)),
|
||
SyncQueue.status.in_(("pending", "processing")),
|
||
),
|
||
)
|
||
)
|
||
rows = list(
|
||
db.session.execute(
|
||
select(SyncQueue)
|
||
.where(and_(*filters))
|
||
.order_by(SyncQueue.priority.desc(), SyncQueue.created_at.asc())
|
||
.limit(pull_limit)
|
||
).scalars().all()
|
||
)
|
||
if personal_only:
|
||
return _sort_tasks_snapshot_order(rows)
|
||
return rows
|
||
|
||
personal_only_pull = initial_sync_active
|
||
tasks = _select_tasks(personal_only=personal_only_pull)
|
||
|
||
# Универсальный bootstrap не должен блокировать выдачу уже накопленных задач
|
||
# (в t.ч. живых insert из UI), иначе клиент долго получает только 202 с пустым телом.
|
||
if (
|
||
needs_personal_enqueue
|
||
and state.universal_bootstrap_completed_at is None
|
||
and len(tasks) == 0
|
||
):
|
||
try:
|
||
step = _bootstrap_universal_step(
|
||
state, max_models=bootstrap_models_per_pull
|
||
)
|
||
db_commit_with_retry(
|
||
attempts=int(getattr(Config, "SYNC_SNAPSHOT_COMMIT_ATTEMPTS", 25)),
|
||
base_delay=float(
|
||
getattr(Config, "SYNC_SNAPSHOT_COMMIT_BASE_DELAY", 0.15)
|
||
),
|
||
)
|
||
tasks = _select_tasks(personal_only=personal_only_pull)
|
||
if not step["done"] and len(tasks) == 0:
|
||
meta = _pull_response_meta(client_id, client, retry_after=retry_after)
|
||
pl202: Dict[str, Any] = {
|
||
"success": True,
|
||
"changes": [],
|
||
"total": 0,
|
||
"batch_id": "",
|
||
"bootstrap_progress": {
|
||
"phase": "universal",
|
||
"cursor": step["cursor"],
|
||
"total_models": step["total"],
|
||
},
|
||
**meta,
|
||
}
|
||
return {"status_code": 202, "payload": pl202}
|
||
except Exception as boot_exc:
|
||
state.universal_bootstrap_last_error = str(boot_exc)[:500]
|
||
logger.exception("[SYNC-PULL-BOOTSTRAP]")
|
||
try:
|
||
db.session.rollback()
|
||
except Exception:
|
||
pass
|
||
return {
|
||
"status_code": 500,
|
||
"payload": {"error": True, "message": "Bootstrap failed"},
|
||
}
|
||
|
||
min_snap = int(getattr(Config, "SYNC_MIN_SNAPSHOT_TASKS", 20))
|
||
logger.info(
|
||
"[SYNC-PULL] needs_enqueue=%s initial_active=%s задач=%s min_snap=%s",
|
||
needs_personal_enqueue,
|
||
initial_sync_active,
|
||
len(tasks),
|
||
min_snap,
|
||
)
|
||
if needs_personal_enqueue and len(tasks) < min_snap:
|
||
try:
|
||
step = _enqueue_personal_snapshot_step(
|
||
client, max_models=personal_models_per_pull
|
||
)
|
||
logger.info(
|
||
"[SYNC-PULL-SNAPSHOT] поставлено в очередь задач: %s (node %s…), commit…",
|
||
step["created"],
|
||
_cid,
|
||
)
|
||
db_commit_with_retry(
|
||
attempts=int(getattr(Config, "SYNC_SNAPSHOT_COMMIT_ATTEMPTS", 25)),
|
||
base_delay=float(getattr(Config, "SYNC_SNAPSHOT_COMMIT_BASE_DELAY", 0.15)),
|
||
)
|
||
logger.info("[SYNC-PULL-SNAPSHOT] commit OK node %s…", _cid)
|
||
initial_sync_active = _client_initial_sync_active(client_id, client)
|
||
personal_only_pull = initial_sync_active
|
||
tasks = _select_tasks(personal_only=personal_only_pull)
|
||
logger.info("[SYNC-PULL-SNAPSHOT] после снапшота задач=%s", len(tasks))
|
||
if not step["done"] and len(tasks) == 0:
|
||
meta = _pull_response_meta(client_id, client, retry_after=retry_after)
|
||
pl202b: Dict[str, Any] = {
|
||
"success": True,
|
||
"changes": [],
|
||
"total": 0,
|
||
"batch_id": "",
|
||
"bootstrap_progress": {
|
||
"phase": "personal",
|
||
"cursor": step["cursor"],
|
||
"total_models": step["total"],
|
||
},
|
||
**meta,
|
||
}
|
||
return {"status_code": 202, "payload": pl202b}
|
||
except Exception as snap_exc:
|
||
logger.error(
|
||
"[SYNC-PULL-SNAPSHOT] ошибка снапшота node %s…: %s",
|
||
_cid,
|
||
snap_exc,
|
||
exc_info=True,
|
||
)
|
||
try:
|
||
db.session.rollback()
|
||
except Exception:
|
||
pass
|
||
return {
|
||
"status_code": 500,
|
||
"payload": {
|
||
"error": True,
|
||
"message": f"Snapshot failed: {snap_exc!s}"[:500],
|
||
},
|
||
}
|
||
|
||
changes: List[Dict[str, Any]] = []
|
||
task_ids: List[str] = []
|
||
work_items: List[tuple] = []
|
||
with db.session.no_autoflush:
|
||
for task in tasks:
|
||
record_data = get_record_data_for_sync(task.table_name, task.record_id)
|
||
action = str(task.action or "update").strip()
|
||
ts = task.created_at.isoformat() if task.created_at else utc_now_iso()
|
||
if not record_data:
|
||
if action == "delete":
|
||
record_data = {}
|
||
else:
|
||
logger.warning(
|
||
"[SYNC-PULL] Данные записи не получены, задача не выдана (pending): %s.%s action=%s",
|
||
task.table_name,
|
||
(task.record_id or "")[:48],
|
||
action,
|
||
)
|
||
continue
|
||
work_items.append((task, record_data, action, ts))
|
||
|
||
for task, record_data, action, ts in work_items:
|
||
task.status = "processing"
|
||
task.processed_at = now
|
||
if task.target_node_id is None:
|
||
db.session.add(SyncDelivery(client_id=client.id, task_id=task.id, delivered_at=now))
|
||
task_ids.append(task.id)
|
||
changes.append(
|
||
{
|
||
"id": task.id,
|
||
"task_id": task.id,
|
||
"table_name": task.table_name,
|
||
"record_id": task.record_id,
|
||
"action": task.action,
|
||
"version": record_data.get("version", 1),
|
||
"content_hash": record_data.get("content_hash", ""),
|
||
"data": record_data,
|
||
"timestamp": ts,
|
||
"priority": task.priority,
|
||
}
|
||
)
|
||
|
||
try:
|
||
db_commit_with_retry()
|
||
except Exception as c_exc:
|
||
logger.error(
|
||
"[SYNC-PULL-COMMIT] ошибка финального commit node %s…: %s",
|
||
_cid,
|
||
c_exc,
|
||
exc_info=True,
|
||
)
|
||
db.session.rollback()
|
||
return {
|
||
"status_code": 500,
|
||
"payload": {
|
||
"error": True,
|
||
"message": f"Commit failed: {c_exc!s}"[:500],
|
||
},
|
||
}
|
||
|
||
_log_changes_statistics(changes, "выдано клиенту")
|
||
initial_sync_active = _client_initial_sync_active(client_id, client)
|
||
personal_remaining = _personal_tasks_in_flight(client_id)
|
||
if initial_sync_active:
|
||
pending_remaining = personal_remaining
|
||
has_more = True
|
||
else:
|
||
pending_remaining = int(
|
||
db.session.scalar(
|
||
select(func.count()).select_from(SyncQueue).where(
|
||
SyncQueue.status == "pending",
|
||
or_(
|
||
SyncQueue.target_node_id == client_id,
|
||
SyncQueue.target_node_id.is_(None),
|
||
),
|
||
SyncQueue.is_deleted.is_(False),
|
||
)
|
||
)
|
||
or 0
|
||
)
|
||
has_more = pending_remaining > 0
|
||
pl200: Dict[str, Any] = {
|
||
"success": True,
|
||
"changes": changes,
|
||
"total": len(changes),
|
||
"has_more": has_more,
|
||
"remaining_hint": personal_remaining if initial_sync_active else pending_remaining,
|
||
"batch_id": _batch_id(client_id, task_ids) if task_ids else "",
|
||
"server_now": utc_now_iso(),
|
||
"timestamp": utc_now_iso(),
|
||
"initial_sync_active": initial_sync_active,
|
||
}
|
||
return {"status_code": 200, "payload": pl200}
|
||
|
||
@staticmethod
|
||
def process_confirm(client_id: str, task_ids: List[str]) -> Dict[str, Any]:
|
||
now = utc_now_naive()
|
||
client: Optional[SyncClient] = db.session.execute(
|
||
select(SyncClient).where(SyncClient.node_id == client_id)
|
||
).scalar_one_or_none()
|
||
if not client:
|
||
return {
|
||
"status_code": 404,
|
||
"payload": {"error": True, "message": "Клиент синхронизации не зарегистрирован"},
|
||
}
|
||
|
||
tasks = db.session.execute(
|
||
select(SyncQueue).where(
|
||
SyncQueue.id.in_(task_ids), SyncQueue.status == "processing"
|
||
)
|
||
).scalars().all()
|
||
active_clients = db.session.execute(
|
||
select(SyncClient.id).where(
|
||
SyncClient.is_enabled.is_(True),
|
||
SyncClient.status == "active",
|
||
or_(SyncClient.is_deleted.is_(False), SyncClient.is_deleted.is_(None)),
|
||
)
|
||
).all()
|
||
active_client_ids = {row[0] for row in active_clients}
|
||
|
||
updated = 0
|
||
ignored = 0
|
||
completed_tasks: List[SyncQueue] = []
|
||
for task in tasks:
|
||
is_personal_task = task.target_node_id == client_id
|
||
is_delivered_universal = False
|
||
if task.target_node_id is None:
|
||
is_delivered_universal = (
|
||
db.session.execute(
|
||
select(SyncDelivery).where(
|
||
SyncDelivery.client_id == client.id,
|
||
SyncDelivery.task_id == task.id,
|
||
)
|
||
).scalar_one_or_none()
|
||
is not None
|
||
)
|
||
if not (is_personal_task or is_delivered_universal):
|
||
ignored += 1
|
||
continue
|
||
|
||
if is_personal_task:
|
||
task.status = "completed"
|
||
task.completed_at = now
|
||
updated += 1
|
||
completed_tasks.append(task)
|
||
continue
|
||
|
||
delivered_rows = db.session.execute(
|
||
select(SyncDelivery.client_id)
|
||
.where(SyncDelivery.task_id == task.id)
|
||
.where(SyncDelivery.client_id.in_(active_client_ids))
|
||
.distinct()
|
||
).all()
|
||
delivered_client_ids = {row[0] for row in delivered_rows}
|
||
if active_client_ids and active_client_ids.issubset(delivered_client_ids):
|
||
task.status = "completed"
|
||
task.completed_at = now
|
||
updated += 1
|
||
completed_tasks.append(task)
|
||
|
||
try:
|
||
if updated:
|
||
db_commit_with_retry()
|
||
_maybe_mark_personal_snapshot_completed(client, client_id)
|
||
if _personal_tasks_in_flight(client_id) == 0 and _personal_enqueue_done(client):
|
||
db_commit_with_retry()
|
||
try:
|
||
from app.services.sync_notification_hooks import (
|
||
notify_tasks_delivered_to_client,
|
||
)
|
||
|
||
notify_tasks_delivered_to_client(client_id, completed_tasks)
|
||
except Exception:
|
||
logger.exception("[SYNC-CONFIRM] notify delivered failed")
|
||
except Exception as e:
|
||
db.session.rollback()
|
||
return {"status_code": 500, "payload": {"error": True, "message": str(e)}}
|
||
|
||
initial_sync_active = _client_initial_sync_active(client_id, client)
|
||
return {
|
||
"status_code": 200,
|
||
"payload": {
|
||
"success": True,
|
||
"updated": updated,
|
||
"ignored": ignored,
|
||
"initial_sync_active": initial_sync_active,
|
||
"server_now": utc_now_iso(),
|
||
"timestamp": utc_now_iso(),
|
||
},
|
||
}
|
||
|
||
@staticmethod
|
||
def process_push(client_id: str, changes: Any) -> Dict[str, Any]:
|
||
payload_error = _validate_sync_changes_payload(changes)
|
||
if payload_error:
|
||
return {"status_code": 400, "payload": {"error": True, "message": payload_error}}
|
||
|
||
conflicts: List[Dict[str, Any]] = []
|
||
applied_changes: List[Dict[str, Any]] = []
|
||
conflict_entries_to_persist: List[Dict[str, Any]] = []
|
||
|
||
batch_size = int(getattr(Config, "SYNC_BATCH_SIZE", 30))
|
||
all_changes = list(changes)
|
||
total_batches = (len(all_changes) + batch_size - 1) // batch_size if all_changes else 0
|
||
|
||
for i in range(0, len(all_changes), batch_size):
|
||
batch_start = time.time()
|
||
batch = all_changes[i : i + batch_size]
|
||
batch_applied = 0
|
||
batch_conflicts = 0
|
||
for change in batch:
|
||
table_name = change["table_name"]
|
||
record_id = change["record_id"]
|
||
action = change["action"]
|
||
client_data = change["data"]
|
||
|
||
if table_name in REPORT_TABLES:
|
||
result = _apply_sync_change(table_name, record_id, action, client_data)
|
||
if not result["success"]:
|
||
conflicts.append(
|
||
{
|
||
"table_name": table_name,
|
||
"record_id": record_id,
|
||
"action": action,
|
||
"error": result["error"],
|
||
"conflict_type": "application_error",
|
||
}
|
||
)
|
||
conflict_entries_to_persist.append(
|
||
{
|
||
"table_name": table_name,
|
||
"record_id": record_id,
|
||
"conflict_type": "application_error",
|
||
"local_data": client_data,
|
||
"remote_data": None,
|
||
"resolution": "pending",
|
||
}
|
||
)
|
||
batch_conflicts += 1
|
||
continue
|
||
applied_changes.append(
|
||
{"table_name": table_name, "record_id": record_id, "action": action}
|
||
)
|
||
batch_applied += 1
|
||
continue
|
||
|
||
if table_name in SERVER_MASTER_TABLES:
|
||
conflict = _check_for_sync_conflict(table_name, record_id, client_data)
|
||
if conflict:
|
||
logger.info(
|
||
"[SYNC-PUSH] server master: пропуск устаревшей версии %s %s",
|
||
table_name,
|
||
record_id,
|
||
)
|
||
continue
|
||
result = _apply_sync_change(table_name, record_id, action, client_data)
|
||
if not result["success"]:
|
||
conflicts.append(
|
||
{
|
||
"table_name": table_name,
|
||
"record_id": record_id,
|
||
"action": action,
|
||
"error": result["error"],
|
||
"conflict_type": "application_error",
|
||
}
|
||
)
|
||
conflict_entries_to_persist.append(
|
||
{
|
||
"table_name": table_name,
|
||
"record_id": record_id,
|
||
"conflict_type": "application_error",
|
||
"local_data": client_data,
|
||
"remote_data": None,
|
||
"resolution": "pending",
|
||
}
|
||
)
|
||
batch_conflicts += 1
|
||
continue
|
||
applied_changes.append(
|
||
{"table_name": table_name, "record_id": record_id, "action": action}
|
||
)
|
||
batch_applied += 1
|
||
continue
|
||
|
||
conflict = _check_for_sync_conflict(table_name, record_id, client_data)
|
||
if conflict:
|
||
conflicts.append(conflict)
|
||
conflict_entries_to_persist.append(
|
||
{
|
||
"table_name": table_name,
|
||
"record_id": record_id,
|
||
"conflict_type": "version_mismatch",
|
||
"local_data": client_data,
|
||
"remote_data": conflict.get("server_data"),
|
||
"resolution": "pending",
|
||
}
|
||
)
|
||
batch_conflicts += 1
|
||
continue
|
||
|
||
result = _apply_sync_change(table_name, record_id, action, client_data)
|
||
if not result["success"]:
|
||
conflicts.append(
|
||
{
|
||
"table_name": table_name,
|
||
"record_id": record_id,
|
||
"action": action,
|
||
"error": result["error"],
|
||
"conflict_type": "application_error",
|
||
}
|
||
)
|
||
conflict_entries_to_persist.append(
|
||
{
|
||
"table_name": table_name,
|
||
"record_id": record_id,
|
||
"conflict_type": "application_error",
|
||
"local_data": client_data,
|
||
"remote_data": None,
|
||
"resolution": "pending",
|
||
}
|
||
)
|
||
batch_conflicts += 1
|
||
continue
|
||
|
||
applied_changes.append(
|
||
{"table_name": table_name, "record_id": record_id, "action": action}
|
||
)
|
||
batch_applied += 1
|
||
|
||
logger.info(
|
||
"[SYNC-PUSH-BATCH] Батч %s/%s обработан: применено %s, конфликтов %s, время %.3fs",
|
||
(i // batch_size) + 1,
|
||
total_batches,
|
||
batch_applied,
|
||
batch_conflicts,
|
||
time.time() - batch_start,
|
||
)
|
||
|
||
if conflicts:
|
||
db.session.rollback()
|
||
persisted_conflicts: List[Dict[str, Any]] = []
|
||
try:
|
||
for c in conflict_entries_to_persist:
|
||
obj = create_sync_conflict(
|
||
table_name=c["table_name"],
|
||
record_id=c["record_id"],
|
||
conflict_type=c["conflict_type"],
|
||
local_data=c["local_data"],
|
||
remote_data=c["remote_data"],
|
||
resolution=c["resolution"],
|
||
)
|
||
if obj:
|
||
persisted_conflicts.append(
|
||
{
|
||
"id": obj.id,
|
||
"table_name": c["table_name"],
|
||
"record_id": c["record_id"],
|
||
"conflict_type": c["conflict_type"],
|
||
}
|
||
)
|
||
if persisted_conflicts:
|
||
db_commit_with_retry()
|
||
except Exception:
|
||
logger.exception("[SYNC-PUSH-CONFLICT-PERSIST] Ошибка сохранения конфликтов")
|
||
db.session.rollback()
|
||
|
||
return {
|
||
"status_code": 409,
|
||
"payload": {
|
||
"success": False,
|
||
"error": True,
|
||
"message": "Обнаружены конфликты синхронизации. Изменения не применены.",
|
||
"conflicts": conflicts,
|
||
"persisted_conflicts": persisted_conflicts,
|
||
"total_applied": 0,
|
||
"total_conflicts": len(conflicts),
|
||
},
|
||
}
|
||
|
||
try:
|
||
db_commit_with_retry()
|
||
except Exception:
|
||
db.session.rollback()
|
||
return {"status_code": 500, "payload": {"error": True, "message": "Commit failed"}}
|
||
|
||
_log_changes_statistics(applied_changes, "применено на сервере")
|
||
_pcid = (client_id or "")[:12]
|
||
logger.info(
|
||
"[SYNC-PUSH-DB] node=%s… применено_записей=%s stats=%s",
|
||
_pcid,
|
||
len(applied_changes),
|
||
_sync_pull_changes_stats(applied_changes),
|
||
)
|
||
try:
|
||
from app.services.sync_notification_hooks import notify_reports_pushed_from_client
|
||
|
||
notify_reports_pushed_from_client(client_id, applied_changes)
|
||
except Exception:
|
||
logger.exception("[SYNC-PUSH] notify reports failed")
|
||
try:
|
||
from app.services.feed_quality.evaluator import evaluate_reports_pushed_from_client
|
||
|
||
evaluate_reports_pushed_from_client(applied_changes, send_notifications=False)
|
||
except Exception:
|
||
logger.exception("[SYNC-PUSH] feed_quality evaluate failed")
|
||
return {
|
||
"status_code": 200,
|
||
"payload": {
|
||
"success": True,
|
||
"applied_changes": applied_changes,
|
||
"conflicts": [],
|
||
"total_applied": len(applied_changes),
|
||
"total_conflicts": 0,
|
||
"server_now": utc_now_iso(),
|
||
"timestamp": utc_now_iso(),
|
||
},
|
||
}
|