Kris
b10cb080de
本次提交包含多项代码改进与功能增强： 1. 新增服务账号禁用错误类型与状态映射 2. 优化出站管道阶段顺序与上下文字段 3. 完善会话与事务管理逻辑，修复并发冲突处理 4. 调整钉钉与微信插件的适配逻辑 5. 优化内容校验与限流降级策略 6. 新增渠道降级投递事件与围栏回滚机制 7. 修正配置schema与权限校验逻辑 8. 完善审计与监控相关的日志与钩子处理
161 lines
8.3 KiB
Python
161 lines
8.3 KiB
Python
"""channels 模块的定时任务 handler 注册入口。
|
||
|
||
本模块是 channels 限界上下文接入 scheduler 限界上下文的唯一登记点，
|
||
对齐 scheduler 上下文设计方案 §14.3 的"业务域 container 主动注册"模式。
|
||
|
||
与 ``infrastructure/channel_use_cases.py`` 同级，遵循 channels 的分层约定：
|
||
装配逻辑集中在 ``infrastructure/``，包根不出现装配代码。
|
||
|
||
对外暴露：
|
||
|
||
- ``register_scheduler_handlers(registry)``：业务模块注册函数，由 worker
|
||
启动钩子（``services/run_worker.py`` 的 ``_worker_startup``）统一调用。
|
||
内部实例化本模块的 handler 并调用 ``registry.register(handler)``。
|
||
|
||
边界规范：
|
||
- 本模块是 channels 接入 scheduler 的**唯一装配点**
|
||
- handler 的依赖（如 ``session_factory``）在函数内部构造，不暴露给装配层
|
||
- 新增 handler 时，在本函数内追加 ``registry.register(...)`` 即可，无需改
|
||
``run_worker.py``（前提：本模块已登记在 ``_BUSINESS_HANDLER_REGISTRARS``）
|
||
- handler 禁止依赖 channels api 进程内存态（``PluginRegistry`` /
|
||
``DegradationManager`` / ``EventBus`` 等），只能通过 ``session_factory``
|
||
+ DB 查询，确保在 worker 进程独立可运行
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
from yuxi.channels.application.extension.scheduler_handlers.channel_audit_log_retention_handler import (
|
||
ChannelAuditLogRetentionHandler,
|
||
)
|
||
from yuxi.channels.application.extension.scheduler_handlers.channel_content_review_retention_handler import (
|
||
ChannelContentReviewRetentionHandler,
|
||
)
|
||
from yuxi.channels.application.extension.scheduler_handlers.channel_idempotency_cleanup_handler import (
|
||
ChannelIdempotencyCleanupHandler,
|
||
)
|
||
from yuxi.channels.application.extension.scheduler_handlers.channel_outbox_recovery_handler import (
|
||
ChannelOutboxRecoveryHandler,
|
||
)
|
||
from yuxi.channels.application.extension.scheduler_handlers.channel_outbox_terminal_cleanup_handler import (
|
||
ChannelOutboxTerminalCleanupHandler,
|
||
)
|
||
from yuxi.channels.application.extension.scheduler_handlers.channel_pairing_expiration_handler import (
|
||
ChannelPairingExpirationHandler,
|
||
)
|
||
from yuxi.channels.application.extension.scheduler_handlers.channel_pairing_terminal_cleanup_handler import (
|
||
ChannelPairingTerminalCleanupHandler,
|
||
)
|
||
from yuxi.channels.application.extension.scheduler_handlers.channel_session_inactive_cleanup_handler import (
|
||
ChannelSessionInactiveCleanupHandler,
|
||
)
|
||
from yuxi.channels.infrastructure.channel_use_cases import (
|
||
create_channel_audit_log_retention_handler_dependencies,
|
||
create_channel_content_review_retention_handler_dependencies,
|
||
create_channel_idempotency_cleanup_handler_dependencies,
|
||
create_channel_outbox_recovery_handler_dependencies,
|
||
create_channel_outbox_terminal_cleanup_handler_dependencies,
|
||
create_channel_pairing_expiration_handler_dependencies,
|
||
create_channel_pairing_terminal_cleanup_handler_dependencies,
|
||
create_channel_session_inactive_cleanup_handler_dependencies,
|
||
)
|
||
from yuxi.scheduler.framework.runtime import HandlerRegistry
|
||
from yuxi.services.run_queue_service import get_arq_pool
|
||
from yuxi.storage.postgres.manager import pg_manager
|
||
|
||
__all__ = ["register_scheduler_handlers"]
|
||
|
||
|
||
async def register_scheduler_handlers(registry: HandlerRegistry) -> None:
|
||
"""向 scheduler 注册 channels 模块的定时任务 handler。
|
||
|
||
由 worker 启动钩子在装配阶段统一调用。新增 handler 时在此函数内追加
|
||
``registry.register(...)`` 即可。
|
||
|
||
注册的 handler 分两类：
|
||
|
||
1. **运维型周期清理 handler**（FR-04 ~ FR-08）：由 ``schedule_kind="interval"``
|
||
周期触发，清理超期终态记录与回收卡死任务，避免表无限增长。worker 进程
|
||
不注入 api 进程内存态（``EventBus`` / ``PluginRegistry`` 等），通过
|
||
``session_factory`` 创建独立 DB 会话执行清理。
|
||
2. **业务状态恢复/过期扫描 handler**（FR-22 / FR-33 / FR-34）：替代原
|
||
api 进程后台扫描器，由 scheduler worker 周期触发，解决共享
|
||
``AsyncSession`` 并发使用与锁 TTL 不匹配问题。
|
||
|
||
所有 handler 通过分布式锁串行化多 worker 并发执行（lock_key=
|
||
``scheduler:{class_name}``, TTL=300s），锁获取失败时 skip（不 fail-closed，
|
||
下一周期重试）。
|
||
|
||
Args:
|
||
registry: worker 进程级 handler 注册表。
|
||
"""
|
||
from yuxi.channels.infrastructure.channel_use_cases import (
|
||
create_worker_cache_port,
|
||
)
|
||
|
||
session_factory = pg_manager.get_async_session_context
|
||
arq_pool = await get_arq_pool()
|
||
cache_port = await create_worker_cache_port()
|
||
|
||
# ─────────────────────────────────────────────────────────────────────
|
||
# P1 运维型周期清理 handler（FR-04 ~ FR-08）
|
||
# ─────────────────────────────────────────────────────────────────────
|
||
|
||
# FR-04：清理长期 inactive 的临时会话（is_temporary=true 且 last_message_at 超时）
|
||
session_cleanup_deps = create_channel_session_inactive_cleanup_handler_dependencies(
|
||
session_factory, cache_port=cache_port
|
||
)
|
||
registry.register(ChannelSessionInactiveCleanupHandler(**session_cleanup_deps))
|
||
|
||
# FR-05：物理清理超期 DEAD / SENT 终态 Outbox 条目，best-effort 发布
|
||
# OutboxEntryPurgedEvent（event_publisher=None，worker 进程未注入）
|
||
outbox_cleanup_deps = create_channel_outbox_terminal_cleanup_handler_dependencies(
|
||
session_factory, cache_port=cache_port
|
||
)
|
||
registry.register(ChannelOutboxTerminalCleanupHandler(**outbox_cleanup_deps))
|
||
|
||
# FR-06：物理清理超期终态配对记录（EXPIRED / REJECTED / REVOKED）
|
||
pairing_cleanup_deps = create_channel_pairing_terminal_cleanup_handler_dependencies(
|
||
session_factory, cache_port=cache_port
|
||
)
|
||
registry.register(ChannelPairingTerminalCleanupHandler(**pairing_cleanup_deps))
|
||
|
||
# ─────────────────────────────────────────────────────────────────────
|
||
# FR-22 / FR-33 / FR-34：业务恢复/过期扫描 handler
|
||
# 由 scheduler worker 触发，替代原 api 进程后台扫描器
|
||
# ─────────────────────────────────────────────────────────────────────
|
||
|
||
# FR-22：Outbox 恢复扫描（PENDING / SENT_UNCONFIRMED 入队重试 + 死信标记）
|
||
outbox_recovery_deps = create_channel_outbox_recovery_handler_dependencies(
|
||
session_factory,
|
||
arq_pool=arq_pool,
|
||
cache_port=cache_port,
|
||
)
|
||
registry.register(ChannelOutboxRecoveryHandler(**outbox_recovery_deps))
|
||
|
||
# FR-33：配对过期扫描（PENDING → EXPIRED）
|
||
pairing_expiration_deps = create_channel_pairing_expiration_handler_dependencies(
|
||
session_factory, cache_port=cache_port
|
||
)
|
||
registry.register(ChannelPairingExpirationHandler(**pairing_expiration_deps))
|
||
|
||
# FR-34：审计日志保留期清理
|
||
audit_log_retention_deps = create_channel_audit_log_retention_handler_dependencies(
|
||
session_factory, cache_port=cache_port
|
||
)
|
||
registry.register(ChannelAuditLogRetentionHandler(**audit_log_retention_deps))
|
||
|
||
# 内容审核记录保留期清理（CR-STATS-01 配套 retention）
|
||
# 物理删除超过 retention_days 的审核记录，避免表无限增长。默认 180 天，
|
||
# 比审计日志 90 天更长，因其用于运营分析与趋势统计。
|
||
content_review_retention_deps = create_channel_content_review_retention_handler_dependencies(
|
||
session_factory, cache_port=cache_port
|
||
)
|
||
registry.register(ChannelContentReviewRetentionHandler(**content_review_retention_deps))
|
||
|
||
# FR-19：幂等记录过期清理
|
||
# 物理删除 expires_at 已过期的幂等记录，避免表无限增长。默认 TTL 24h。
|
||
idempotency_cleanup_deps = create_channel_idempotency_cleanup_handler_dependencies(
|
||
session_factory, cache_port=cache_port
|
||
)
|
||
registry.register(ChannelIdempotencyCleanupHandler(**idempotency_cleanup_deps))
|