diff --git a/backend/package/yuxi/storage/postgres/models_channels.py b/backend/package/yuxi/storage/postgres/models_channels.py index 5f65b326..ddd34fd5 100644 --- a/backend/package/yuxi/storage/postgres/models_channels.py +++ b/backend/package/yuxi/storage/postgres/models_channels.py @@ -123,6 +123,11 @@ class ChannelAccount(Base): Index("ix_channel_accounts_plugin_status", "plugin_status"), ) + # ORM 乐观锁：UPDATE 时自动附加 ``WHERE version = ?`` 并自增 version， + # 并发状态转换冲突时抛 ``StaleDataError``，由适配器层翻译为 ``ConflictError``。 + # 与 ``ChannelPairing`` / ``ChannelRouteBinding`` 保持一致的并发控制风格。 + __mapper_args__ = {"version_id_col": version} + def to_dict(self) -> dict[str, Any]: """序列化为字典；config 中的敏感字段由 persistence 层解密，to_dict 不返回明文。""" return { diff --git a/backend/package/yuxi/storage/postgres/models_external.py b/backend/package/yuxi/storage/postgres/models_external.py index a156f45f..9c3ce69f 100644 --- a/backend/package/yuxi/storage/postgres/models_external.py +++ b/backend/package/yuxi/storage/postgres/models_external.py @@ -7,7 +7,6 @@ created_at/updated_at/is_deleted/deleted_at），支持软删除与操作追溯 from __future__ import annotations -import base64 from typing import Any from sqlalchemy import ( @@ -171,7 +170,7 @@ class ExternalAdapterAsset(Base): nullable=False, default="active", index=True, - comment="资产状态：active/parse_failed/inactive", + comment="资产状态：active/invalid", ) status_message = Column(String(512), nullable=True, comment="状态说明，如解析失败原因") @@ -183,42 +182,6 @@ class ExternalAdapterAsset(Base): is_deleted = Column(Integer, nullable=False, default=0, index=True) deleted_at = Column(DateTime, nullable=True) - def to_dict(self) -> dict[str, Any]: - """序列化为字典；content 以 base64 编码返回，便于 JSON 传输。""" - content_b64 = base64.b64encode(self.content).decode("utf-8") if self.content else "" - return { - "id": self.id, - "adapter_type": self.adapter_type, - "asset_type": self.asset_type, - "name": self.name, - "content": content_b64, - "size": self.size, - "checksum": self.checksum, - "status": self.status, - "status_message": self.status_message, - "created_by": self.created_by, - "updated_by": self.updated_by, - "created_at": format_utc_datetime(self.created_at), - "updated_at": format_utc_datetime(self.updated_at), - } - - def to_list_dict(self) -> dict[str, Any]: - """序列化为列表项字典；不包含 content 字段，避免大体积响应。""" - return { - "id": self.id, - "adapter_type": self.adapter_type, - "asset_type": self.asset_type, - "name": self.name, - "size": self.size, - "checksum": self.checksum, - "status": self.status, - "status_message": self.status_message, - "created_by": self.created_by, - "updated_by": self.updated_by, - "created_at": format_utc_datetime(self.created_at), - "updated_at": format_utc_datetime(self.updated_at), - } - class ExternalSystem(Base): """外部系统一级实体。""" @@ -1033,6 +996,12 @@ class ExternalSystemQuotaUsage(Base): # 配额来源 source = Column(String(32), nullable=False, default="config", comment="配额来源：response_header/manual/config") source_header_name = Column(String(128), nullable=True, comment="响应头字段名，如 X-RateLimit-Remaining") + source_header_strategy = Column( + String(32), + nullable=False, + default="remaining", + comment="响应头解析策略：remaining/used/limit/fraction_used_limit", + ) # 告警阈值 warning_threshold = Column(Integer, nullable=False, default=80, comment="告警阈值百分比（0-100）") @@ -1076,6 +1045,7 @@ class ExternalSystemQuotaUsage(Base): "unit": self.unit, "source": self.source, "source_header_name": self.source_header_name, + "source_header_strategy": self.source_header_strategy, "warning_threshold": self.warning_threshold, "critical_threshold": self.critical_threshold, "last_updated_at": format_utc_datetime(self.last_updated_at) if self.last_updated_at else None, @@ -1115,7 +1085,7 @@ class ExternalSystemAlert(Base): String(64), nullable=False, index=True, - comment="告警类型：health_check_failed/circuit_open/token_refresh_failed/quota_near_limit/continuous_timeout/webhook_expired", + comment="告警类型：health_check_failed/circuit_open/token_refresh_failed/quota_near_limit/continuous_timeout/webhook_expired/webhook_event_failed", ) severity = Column(String(32), nullable=False, default="warning", comment="严重程度：info/warning/critical/fatal") status = Column( @@ -1256,6 +1226,12 @@ class ExternalSecretRotationPolicy(Base): last_rotation_error = Column(Text, nullable=True, comment="上次轮换错误信息") rotation_count = Column(Integer, nullable=False, default=0, comment="成功轮换次数") max_rotation_retries = Column(Integer, nullable=False, default=3, comment="最大重试次数") + current_retry_count = Column( + Integer, + nullable=False, + default=0, + comment="当前连续重试次数（失败时递增，成功/重置时清零）", + ) # 策略状态 status = Column( @@ -1304,6 +1280,7 @@ class ExternalSecretRotationPolicy(Base): "last_rotation_error": self.last_rotation_error, "rotation_count": self.rotation_count, "max_rotation_retries": self.max_rotation_retries, + "current_retry_count": self.current_retry_count, "status": self.status, "created_by": self.created_by, "updated_by": self.updated_by, diff --git a/backend/package/yuxi/storage/postgres/models_scheduler.py b/backend/package/yuxi/storage/postgres/models_scheduler.py index 817a32b8..4b0fdb48 100644 --- a/backend/package/yuxi/storage/postgres/models_scheduler.py +++ b/backend/package/yuxi/storage/postgres/models_scheduler.py @@ -47,7 +47,11 @@ class ScheduledTask(Base): 一条记录对应一次任务注册；`next_run_at` 由调度器在创建/执行后按 `schedule_kind`（cron / at）计算。`status` 表示任务状态机当前所处状态 - （active / paused / dead_letter），与 `enabled` 共同决定是否被 tick 扫描。 + （active / paused / dead_letter），`enabled` 与 `status` 通过应用层联动 + （见 ``scheduler_service.update_task`` 方案 E）：``enabled=False`` 且 + ``status=active`` 同步置 ``paused``；``enabled=True`` 且 ``status=paused`` + 同步置 ``active``。tick 扫描条件为 ``enabled=True AND status='active' + AND next_run_at <= now``，两者共同决定可执行性。 """ __tablename__ = "scheduled_tasks" @@ -59,7 +63,9 @@ class ScheduledTask(Base): task_id = Column( String(64), nullable=False, - comment="业务 UUID；system 任务使用约定前缀如 system:run_log_cleanup；唯一性由部分唯一索引 uq_scheduled_tasks_task_id_active 保障（is_deleted=0），软删除后允许重建", + comment="业务 UUID；system 任务使用约定前缀如 system:run_log_cleanup；" + "唯一性由部分唯一索引 uq_scheduled_tasks_task_id_active 保障（is_deleted=0），" + "软删除后允许重建", ) handler_name = Column( String(128), nullable=False, index=True, comment="引用已注册 handler（非外键，worker 执行时校验）" @@ -75,7 +81,8 @@ class ScheduledTask(Base): run_at = Column( DateTime, nullable=True, - comment="一次性执行时间（UTC，与 tz 无关）；at 类型必填且必须未来时间。at 类型创建时 next_run_at 同步设为此值，执行后 next_run_at 置 NULL", + comment="一次性执行时间（UTC，与 tz 无关）；at 类型必填且必须未来时间。" + "at 类型创建时 next_run_at 同步设为此值，执行后 next_run_at 置 NULL", ) tz = Column( String(64), @@ -86,12 +93,20 @@ class ScheduledTask(Base): payload = Column(JSON_VALUE, nullable=False, default=dict, comment="handler 入参（≤64KB）") # 运行时控制 - enabled = Column(Boolean, nullable=False, default=True, comment="是否启用；dead_letter 时为 False") + enabled = Column( + Boolean, + nullable=False, + default=True, + comment="是否启用；与 status 通过应用层联动（方案 E）：" + "enabled=False 且 status=active 同步置 paused；" + "dead_letter 任务需通过 resume_task 恢复，不能通过 enabled=True 直接启用", + ) delete_after_run = Column( Boolean, nullable=False, default=False, - comment="执行成功后是否删除任务记录；DB 默认 False，at 类型由应用层显式设置 True", + comment="执行成功后是否删除任务记录；仅对 at 类型有意义，" + "cron 类型应用层拒绝设置为 True；DB 默认 False，at 类型由应用层显式设置 True", ) block_strategy = Column( String(32), @@ -111,11 +126,19 @@ class ScheduledTask(Base): ) # 执行轨迹 - last_run_at = Column(DateTime, nullable=True, comment="上次执行开始时间") + last_run_at = Column( + DateTime, + nullable=True, + comment="上次执行开始时间（方案 B：语义统一为执行开始时间，" + "由 record_run_result 的 started_at 参数写入，非执行结束时间）", + ) next_run_at = Column( DateTime, nullable=True, - comment="下次执行时间（UTC）。cron 类型由调度器按 cron 表达式计算；at 类型创建时=run_at，执行后置 NULL；失败退避时按退避序列计算", + comment="下次执行时间（UTC）。cron 类型由调度器按 cron 表达式计算；" + "at 类型创建时=run_at，成功执行后置 NULL（方案 A：通过 " + "finalize_one_shot_task 置 NULL + status=paused 避免无限循环）；" + "失败退避时按退避序列计算", ) last_error = Column(Text, nullable=True, comment="最近一次错误信息") @@ -338,13 +361,15 @@ class ScheduledTaskIdempotency(Base): # 操作信息 operation = Column( - String(32), nullable=False, comment="操作类型：create / update / delete / pause / resume / trigger" + String(32), + nullable=False, + comment="操作类型：create / update / delete / pause / resume / trigger", ) response_body = Column(JSON_VALUE, nullable=True, comment="首次请求的响应体；用于重复请求重放") # 审计字段 - created_by = Column(String(64), nullable=True) - updated_by = Column(String(64), nullable=True) + created_by = Column(String(64), nullable=True, comment="发起幂等请求的操作人 UID") + updated_by = Column(String(64), nullable=True, comment="回写响应体时的操作人 UID") created_at = Column( DateTime, nullable=False, default=utc_now_naive, index=True, comment="创建时间；用于清理过期记录" ) @@ -352,6 +377,13 @@ class ScheduledTaskIdempotency(Base): is_deleted = Column(Integer, nullable=False, default=0, index=True) deleted_at = Column(DateTime, nullable=True) + __table_args__ = ( + CheckConstraint( + "operation IN ('create', 'update', 'delete', 'pause', 'resume', 'trigger')", + name="ck_scheduled_task_idempotency_operation", + ), + ) + def to_dict(self) -> dict[str, Any]: """序列化为 API 响应字典。""" return {