bisheng LDAP 组织定时校对与 SSO 换型 Relink基于 Celery 的 6h 自动同步与冲突治理实战【免费下载链接】bishengBISHENG is an open LLM devops platform for next generation Enterprise AI applications. Powerful and comprehensive features include: GenAI workflow, RAG, Agent, Unified model management, Evaluation, SFT, Dataset Management, Enterprise-level System Management, Observability and more.项目地址: https://gitcode.com/GitHub_Trending/bi/bisheng导读本文围绕 bisheng 开源 LLM DevOps 平台在 v2.5.1 中落地的F015 特性LDAP 组织定时校对 SSO 换型 relink系统讲解如何用 Celery Beat 每 6 小时强制校对 SSO/LDAP 部门树与 bisheng 内部组织结构自动修复漂移并在 SSO 更换 HR 系统后通过 relink 接口重建 external_id 映射。读完本文你将掌握org_sync_log事件化改造与复合索引设计、基于 Redis SETNX 的并发幂等锁、last_sync_ts时间戳仲裁的冲突决策表INV-T12、ts 冲突周报与每日升级告警链路以及 external_id_map / path_plus_name 两种 relink 匹配策略的 HTTP 调用方式。文中所有结论均有仓库源码与测试用例佐证可直接对照实现进行二次开发或运维排障。1. 背景为什么需要“定时校对 换型 relink”1.1 集团 IT 的用户故事F015 对应的用户故事来自多租户需求文档spec.md §1作为集团 IT我希望系统每 6 小时自动校对 SSO 端部门树与 bisheng 内部结构发现漂移时自动修复SSO 换 HR 系统时提供 relink 接口重建映射以便部门归属长期保持一致换型不丢失用户归属。在真实的企业环境里SSO 侧的部门树会不断变化新增、改名、删除、跨部门移动而 Gateway 实时同步只能覆盖“在线发生”的变更一旦出现断网、Provider 短暂不可用或批量历史数据导入bisheng 内部结构与 SSO 就会产生漂移。F015 的价值在于兜底用 Celery Beat 每 6 小时做一次全量拉取式校对spec.md AD-01 决策为 6h非 1h/24h自动修复新增/命名变更自动 upsert删除进入 orphaned孤儿子部门并标记跨租户移动自动触发用户租户重算换型逃生门SSO 从旧 HR 系统迁到新系统时 external_id 全部变化通过 relink 接口用路径名称或显式映射重建关联避免用户归属丢失。1.2 与既有特性的关系F015 不是从零开发而是站在 v2.5.1 一系列前置特性之上tasks.md「前置」节前置能力来源F015 中的用途OrgSyncTsGuardF014 T04每 op 的 ts 决策INV-T12DeptUpsertServiceF014 T07upsert 落库_OrgSyncLogBufferflush_logF014 T12批摘要日志verify_hmacF014 T05relink HTTP 签名校验DepartmentDeletionHandler.on_deleted(dept_id, deletion_source)F011归档后孤儿租户处理UserTenantSyncService.sync_user(user_id, trigger)F012跨租户移动后的用户租户重算DepartmentDao.aget_by_source_external_id / aupsert_by_external_id / aarchive_by_external_idF014 T03部门 DAO 基础操作AuditLogDao.ainsert_v2F011冲突/relink 审计留痕DeletionSource.CELERY_RECONCILE/UserTenantSyncTrigger.CELERY_RECONCILEF011 constants标注操作来源这些前置实现的当前仓库位置分别为 org_sync/domain/services/ts_guard.py、sso_sync/domain/services/dept_upsert_service.py、sso_sync/domain/services/org_sync_log_writer.py 等。2. 总体架构与任务拆解2.1 决策锁定D1–D8tasks.md「决策锁定」明确了 8 项架构决策是理解代码布局的钥匙D1Service 放bisheng/org_sync/domain/services/API endpoint 放bisheng/org_sync/api/endpoints/relink.pyF009 扩展D26h Beat 用独立reconcile_all_organizations不复用 F009 的check_org_sync_schedules后者响应用户 cronF015 是系统强制D3org_sync_log表加 4 列 复合索引不拆事件表——摘要行event_type事件行非空D4relink 多候选冲突存 Redis Hashrelink_conflict:{dept_id}TTL 7 天D5前端 UI 不纳入归 F019 admin-scope consoleD6细粒度 15 个任务对齐 F014 的 Test-Alongside 模式D7独立 worktree 开发D8错误码 19314–19318 写入errcode/sso_sync.pyMMM193 与 F014 共享。从当前仓库的实际文件布局看这些决策已全部落地org_sync/domain/services/下存在reconcile_service.py、remote_dept_differ.py、relink_service.py、relink_conflict_store.py、ts_guard.py、reconciler.pyorg_sync/api/endpoints/relink.py存在错误码定义于 common/errcode/sso_sync.py。2.2 依赖图与并行建议tasks.md 给出了完整依赖图关键路径如下T01 (errcode ReconcileConf) ├─→ T02 (Alembic 迁移) → T03 (OrgSyncLog ORMDAO) → T04 (aget_all_active) → T06 (主编排) │ └─→ T10 (TsConflictReporter) → T13 (weekly/daily beats) ├─→ T05 (RemoteDeptDiffer 纯函数) → T06 └─→ T07 (Relink schemasservice) → T08 (Relink API) ─┐ └─→ T12 (RelinkConflictStore) T09 (Celery 6h fan-out) / T11 (event 持久化) / T14 (SETNX 并发幂等) T01-T14 ──→ T15 (AC 对照 e2e 压测占位)并行建议T01 后三路并行 — (T02→T03→T04→T06) / (T05→T07→T08→T12) / (T10)T09/T11/T13/T14 收尾T15 统一验收。这种拆法让纯函数T05 diff、无 IO 的决策T01 常量可以最早启动数据层与编排层则串行推进。2.3 开发模式约定Test-Alongside单任务 实现代码 单测/集成测共 2–4 文件Celery 任务用直接调.apply()同步触发或 mockapply_asyncRedis SETNX/ZSET/Hash 用mock_redisfixtureProviderfetch_departments用MagicMock()返回 DTO 列表迁移通过 MySQL 手工alembic upgrade head/downgrade -1往返 SQLitetable_definitions.py同步冲突告警复用 F011send_inbox_noticelist_global_super_admin_ids。3. 配置与错误码T013.1 ReconcileConf 配置模型ReconcileConf定义在 core/config/reconcile.pytasks.md 中字段更全仓库当前实现保留了锁与冲突 TTL 两项其余字段按设计在 settings 注册class ReconcileConf(BaseModel): beat_cron_reconcile: str 0 */6 * * * # 6h 校对 beat_cron_weekly_report: str 0 9 * * MON # 周一 09:00 周报 beat_cron_daily_escalation: str 0 9 * * * # 每日 09:00 升级 redis_lock_ttl_seconds: int 1800 # 校对锁 TTL30min relink_conflict_ttl_seconds: int 604800 # relink 冲突 7d weekly_conflict_threshold: int 3 # 周报冲突阈值 daily_escalation_days: int 5 # 5 天未解决升级 task_time_limit: int 1800 # Celery 硬超时 task_soft_time_limit: int 1500 # Celery 软超时仓库实现中redis_lock_ttl_seconds的注释解释了设计意图30 分钟与 Celery 软超时对齐保证卡死的 worker 不会永久持有锁relink_conflict_ttl_seconds要求管理员必须在 7 天内通过resolve-conflict接口解决冲突。这些配置经Settings注册为reconcile子配置运行时通过settings.reconcile.xxx访问见 reconcile_service.py 中settings.reconcile.redis_lock_ttl_seconds的用法。3.2 错误码 19314–19318F015 的错误码全部追加在 common/errcode/sso_sync.pyMMM193 与 F014 共享模块19310–19313 已被 F014 占用Code错误类含义19314SsoReconcileLockBusyError同一 config 的校对正在执行Redis SETNX 锁忙跳过本次触发19315SsoRelinkStrategyUnsupportedErrorrelinkmatching_strategy不是external_id_map/path_plus_name19316SsoRelinkConflictUnresolvedError冲突候选列表为空或所选new_external_id不在已存候选内19317SsoSameTsRemoveAppliedWarnError同 ts 的 upsert/remove 冲突已按 remove 为准应用告警类不抛出19318SsoReconcileReservedError预留仓库源码与 tasks.md 的编号完全一致。特别值得注意的是 19317 在 reconcile_service.py 中只记日志不抛出——管理员通过org_sync_log事件行和audit_log观察而不是走 HTTP 错误面。4. 数据层改造T02/T03/T044.1 Alembic 迁移org_sync_log 加 4 列 复合索引T02 是数据层的地基迁移文件v2_5_1_f015_reconcile_log_fields.py复用 F014 的_column_exists/_index_exists幂等 helper为org_sync_log表追加列类型默认值用途event_typeString(32)事件类型空批摘要行 /ts_conflict/stale_ts/conflict_weekly_sent/conflict_daily_escalation_sentlevelString(16)info日志级别info / warn / errorexternal_idString(128)NULL事件行关联的部门 external_idsource_tsBigIntegerNULLINV-T12 捕获的 incoming ts审计用同时创建复合索引CREATE INDEX idx_conflict_lookup ON org_sync_log (level, event_type, external_id, create_time);这个索引直接服务于 5.5.3 节的冲突计数查询levelwarn AND event_typets_conflict AND external_id? AND create_time now-7d。spec.md §5.5.3 还给出了兼容补丁 SQLALTER TABLE ... ADD INDEX / ADD COLUMN用于 v2.5.0/F009 已建表但缺字段/索引的升级场景。SQLite 侧的test/fixtures/table_definitions.py同步更新T03 的 SQLite 测试绿灯即为同步佐证MySQL 侧做alembic upgrade head downgrade -1 upgrade head往返验证。4.2 OrgSyncLog ORM 与 DAO 扩展OrgSyncLogORM 追加上述 4 列后OrgSyncLogDao新增 3 个 classmethodtasks.md T03acreate_event(event_type, level, external_id, source_ts, config_id, error_detailsNone, tenant_id1)— 事件行快捷构造摘要计数器全 0acount_recent_conflicts(external_id, days7) - int— 利用idx_conflict_lookup统计冲突次数aget_conflicts_since(since, event_typets_conflict, levelwarn) - list[OrgSyncLog]— 聚合输入Service 层按 external_id group。T03 的 6 条测试覆盖了事件行持久化、摘要行计数器保持为 0、按 external_id 与时间窗过滤、窗口外返回 0、按序返回以及摘要行event_type与事件行event_typets_conflict共存可分别查出。4.3 OrgSyncConfigDao.aget_all_activeT04 为 6h fan-out 提供入口aget_all_active扫描所有租户的 active 配置。与 F009aget_active_cron_configs的关键区别是不按 schedule_type 过滤——F015 是系统强制校对与用户配置的 cron 无关。调用方负责过滤provider sso_realtimeF014 seed id9999仅用于 HMAC 实时日志不经过fetch_departments。5. Diff 引擎RemoteDeptDifferT055.1 输出 DTORemoteDeptDiffer定义于 org_sync/domain/services/remote_dept_differ.py输出四类 dataclassdataclass class UpsertOp: # CREATE / UPDATE 语义 external_id: str; name: str; parent_external_id: Optional[str] sort_order: int; incoming_ts: int; is_new: bool existing_dept_id: Optional[int] None # UPDATE 时带上已有 id dataclass class ArchiveOp: # 远端不再列出 → 软归档 external_id: str; dept_id: int; mounted_tenant_id: Optional[int]; incoming_ts: int dataclass class MoveOp: # 父节点变更 叶子租户影响标志 external_id: str; dept_id: int; new_parent_external_id: Optional[str] crosses_tenant: bool; incoming_ts: int dataclass class ReconcileDiff: upserts: list[UpsertOp]; archives: list[ArchiveOp]; moves: list[MoveOp]5.2 纯函数设计diff(remote_depts, local_depts, source, ts)是无 IO 的纯函数它包装 F009 的reconcile_departments见 org_sync/domain/services/reconciler.py并把拓扑排序保证upsert 父先于子、archive 子先于父继承下来同时给每个 op 注入incoming_ts与crosses_tenant。crosses_tenant的计算是叶子租户推导沿 local 父链向上找最近的is_tenant_root1节点取其mounted_tenant_id作为叶子租户移动前后叶子租户不同则crosses_tenantTrue。仓库实现中_derive_leaf_tenant_id还带环路防御visited 集合与“找不到挂载点视为 Root 租户”的兜底属于对 pathological 数据的防御性处理。6. 主编排服务OrgReconcileServiceT066.1 11 步主流程org_sync/domain/services/reconcile_service.py 的reconcile_config(config_id)是 F015 的心脏完整流程如下1) 加载 configprovidersso_realtime 或 status!active → skipped 2) 获取 Redis SETNX 锁 org_reconcile:{config_id}TTLredis_lock_ttl_seconds 3) Provider.authenticate() fetch_departments(sync_scope.root_dept_ids) 4) local DepartmentDao.aget_active_by_tenant(config.tenant_id) 5) diff RemoteDeptDiffer.diff(remote, local, source, tsnow) 6) Upsert 循环逐 op 走 OrgSyncTsGuardSKIP_TS → stale_ts 事件行 warnAPPLY → DeptUpsertService 7) Archive 循环逐 op 走 Guard同 ts 冲突remove wins→ ts_conflict 事件行 auditarchive DeletionHandler 8) 跨租户移动 → 主部门成员逐个 UserTenantSyncService.sync_user(uid, CELERY_RECONCILE) 9) flush_log 批摘要 10) 逐条持久化 event_rowsacreate_event单条失败不影响其他 11) 返回 ReconcileResult 供 Celery 记录仓库实现比 tasks.md 骨架更完善的地方包括租户上下文显式管理——Celery 任务入口没有 HTTP 中间件服务在bypass_tenant_filter()下设置current_tenant_id避免 SQLAlchemy tenant_filter 事件报 20004 缺租户上下文逐 op try/except——单个部门 upsert/archive 失败不中断整轮校对失败进入result.errors逐事件行容错——event row 持久化单条失败只记日志。6.2 三个关键子流程UpsertAC-02对每个UpsertOp查DepartmentDao.aget_by_source_external_idOrgSyncTsGuard.check_and_update(existing, incoming_ts, upsert)决策。APPLY 时通过DeptUpsertService.upsert_from_sync_payload落库。is_new决定 buffer 计数是dept_created还是dept_updated与 F009 批摘要语义对齐保证管理端历史面板数字符合预期。mount 标记is_tenant_root在 upsert 中不被修改——这正是 AC-02 “新部门 upsert 不动挂载标记”的落点。ArchiveAC-03 / AC-11对每个ArchiveOp同样走 Guard。同 ts 冲突的检测条件是last_sync_ts incoming_ts且is_deleted 0——说明本批 upsert 已应用但随后又收到同 ts 的 remove此时写ts_conflict事件行error_details.resolutionremove_wins写audit_log.actiondept.sync_conflict审计失败不中断操作记 19317 告警日志DepartmentDao.aarchive_by_external_id落删除DepartmentArchiveCleanupService.arun_for_archived_department清理DepartmentDeletionHandler.on_deleted(dept_id, DeletionSource.CELERY_RECONCILE)统一处理孤儿租户INV-T8mount 部门归档后Tenant.statusorphaned并告警。跨租户移动AC-04 / INV-T2crosses_tenantTrue的移动对部门所有主部门成员is_primaryTrue逐个调用UserTenantSyncService.sync_user(uid, CELERY_RECONCILE)触发token_version 1使 JWT 的tenant_id失效刷新。6.3 Redis SETNX 锁AC-13redis await get_redis_client() key forg_reconcile:{config_id} ok await redis.async_connection.set(key, b1, nxTrue, exex) if not ok: raise SsoReconcileLockBusyError.http_exception()锁在finally中释放删除失败靠 TTL 兜底。tasks.md 的 T14 专项测试覆盖同 config 并发一成一 19314、不同 config 锁相互独立、异常时锁仍释放、SETNX 调用 ex1800。7. Relink 子系统T07/T08/T127.1 使用场景SSO 换型更换 HR 系统时新系统为部门生成了全新的external_id旧 id 全部失效。如果不处理下次实时同步会把所有旧部门判定为“已删除”并归档用户归属随之丢失。relink 的作用就是在迁移窗口内把旧 external_id 重新指向新 id。7.2 两种匹配策略org_sync/domain/services/relink_service.py 实现两种策略external_id_map运维提供old_ext - new_ext显式映射字典逐条查 deptby sourceold_ext后重写external_idpath_plus_name对每个 old_ext在同 source 且未被占用的 active 部门中按(path, name)精确匹配找候选单候选自动 apply多候选存入RelinkConflictStore并返回 conflicts 列表由管理员人工确认spec.md AD-02 决策人工确认避免误操作。策略不识别时抛SsoRelinkStrategyUnsupportedError19315。每次实际重写都写审计actiondept.relink_applied自动路径或dept.relink_resolved人工解决路径metadata 含{old_ext, new_ext, strategy}满足 INV-T7 的“挂载/解绑强制 audit”。7.3 dry_run 预演dry_runTrue时只收集would_apply清单不写 DB、不写冲突存储用于迁移前验证匹配结果。这是 AC-07 的核心也是运维上线前必做的安全检查。7.4 多候选冲突存储org_sync/domain/services/relink_conflict_store.py 用 Redis Hash 存储relink_conflict:{dept_id}的每个 field 是候选new_external_idvalue 是 JSON含 path/name/score。TTL 7 天超时自动过期忘掉的冲突重跑 relink 即可重建。仓库实现特意在save前先delete保证干净替换并在get中对 bytes/str 双客户端形态做兼容、对损坏 JSON 容错。模块 docstring 解释了“为什么用 Redis 不用表”冲突是短生命周期的临时数据、丢失可接受、避免引入迁移/模型/清理 cron。7.5 resolve-conflict管理员从候选列表中选择一个new_external_id提交若所选不在候选内或候选已过期抛SsoRelinkConflictUnresolvedError19316且保留存储合法则重写 external_id 写dept.relink_resolved审计 删除存储条目。8. HTTP API 层T08org_sync/api/endpoints/relink.py 提供两个 HMAC 签名的内部端点挂在router.py并在utils/http_middleware.py的TENANT_CHECK_EXEMPT_PATHS中登记以绕过租户上下文中间件POST /api/v1/internal/departments/relink Body: { old_external_ids: [abc, ...], matching_strategy: external_id_map | path_plus_name, external_id_map: {abc: new_abc, ...}, // external_id_map 策略必填 source: sso, dry_run: false } Returns: {applied: [...], would_apply: [...], conflicts: [...]} POST /api/v1/internal/departments/relink/resolve-conflict Body: {dept_id: 123, chosen_new_external_id: new_abc} Returns: {dept_id: 123, old_external_id: abc, new_external_id: new_abc}两个端点都依赖 F014 的verify_hmac做请求签名校验无签名 401Service 层在 ROOT_TENANT_ID bypass_tenant_filter下运行与 F014 LoginSyncService 的模式一致。9. Celery 任务6h 校对 冲突告警T09/T139.1 6h 校对 fan-outworker/org_sync/reconcile_tasks.py 定义两个任务bisheng_celery.task(acks_lateTrue) def reconcile_all_organizations(): 6h Beat entry: fan out reconcile per active OrgSyncConfig. loop asyncio.new_event_loop() try: loop.run_until_complete(_fan_out_all()) finally: loop.close() async def _fan_out_all() - None: configs await OrgSyncConfigDao.aget_all_active() for c in configs: if c.provider sso_realtime: continue reconcile_single_config.apply_async(args[c.id], queueknowledge_celery) bisheng_celery.task(acks_lateTrue, time_limit1800, soft_time_limit1500) def reconcile_single_config(config_id: int): Execute one reconcile run; swallow lock-busy without retry. loop asyncio.new_event_loop() try: loop.run_until_complete(OrgReconcileService.reconcile_config(config_id)) except SsoReconcileLockBusyError: logger.warning(freconcile_single_config {config_id} skipped: lock busy) except Exception: logger.exception(freconcile_single_config {config_id} failed) finally: loop.close()两个关键设计锁忙不重试SsoReconcileLockBusyError只记 warning避免重试风暴叠加硬超时 1800s 与 Redis 锁 TTL 一致卡死 worker 也不会永久持锁。子任务进knowledge_celery队列。Beat 注册在CeleryConf.validate中追加if reconcile_all_organizations not in self.beat_schedule: self.beat_schedule[reconcile_all_organizations] { task: bisheng.worker.org_sync.reconcile_tasks.reconcile_all_organizations, schedule: crontab.from_string(0 */6 * * *), # every 6h }9.2 冲突周报与每日升级TsConflictReporterorg_sync/domain/services/ts_conflict_reporter.py提供两个告警方法由两条 Beat 独立触发任务cron逻辑report_ts_conflicts_weekly0 9 * * MON周一 09:00聚合过去 7 天ts_conflict事件按 external_id 计数count weekly_conflict_threshold(3)的条目发全局超管站内消息payload 含冲突详情 “是否需要 relink”建议 本周总冲突数并写conflict_weekly_sent标记行report_ts_conflicts_daily_escalation0 9 * * *每日 09:00若最近一次周报标记已超过daily_escalation_days(5)天且冲突仍存在升级为每日告警全部解决则reasonresolved不升级升级状态机的三种退出路径值得注意无周报标记no_weekly_marker、仍在宽限期内within_grace、冲突已解决resolved。告警复用 F011 的send_inbox_noticelist_global_super_admin_ids对应的conflict_weekly_sent/conflict_daily_escalation_sent标记行也写入org_sync_log使告警行为本身可审计。10. 冲突决策核心OrgSyncTsGuard 与 INV-T1210.1 决策表org_sync/domain/services/ts_guard.py 是 F014Gateway 实时与 F015Celery 校对共享的纯决策函数把 spec.md §5.5.2 的决策表落地incoming ts vs last_sync_ts动作incoming_ts last_sync_ts应用变更 更新last_sync_tsAC-10incoming_ts last_sync_ts同 source 同方向幂等跳过已应用incoming_ts last_sync_tsupsert remove 双向以 remove 为准AC-11从严避免幽灵部门remove 已应用时后续 upsert 观察is_deleted1而 SKIP_TSincoming_ts last_sync_ts跳过 写org_sync_loglevelwarn陈旧消息丢弃AC-0910.2 纯函数实现Guard 不执行任何 I/O只有两个输出APPLY/SKIP_TS。对从未见过的 external_idupsert 放行、remove 静默丢弃没有历史可保护。决策与写入分离的设计让 8 种组合无需 DB 往返即可单测。“同 ts remove wins”不变量的成立依赖一个关键事实第一个写入方把is_deleted1落库后续同 ts 的 upsert 在 Guard 中读到该标记即判 SKIP_TS——这与 F015 同批内“先 upsert 后 archive”的冲突由reconcile_service的is_same_ts_conflict检测并审计形成互补两层共同保证 INV-T12。10.3 不变量映射小结INV-T12ts 最大为准 同 ts remove 优先T05 diff 注入incoming_ts→ T06 逐 op 走 Guard → 同 ts 冲突走 AC-11remove wins auditdept.sync_conflict 19317→ T11 持久化 → T14 并发兜底INV-T8孤儿 TenantT06 归档 mount 部门后调DepartmentDeletionHandler.on_deleted(dept_id, CELERY_RECONCILE)→ F011 将Tenant.statusorphaned并告警INV-T7挂载/解绑强制 auditT07resolve_conflict写dept.relink_resolvedT06 同 ts 冲突写dept.sync_conflictINV-T2用户唯一叶子T06crosses_tenantTrue主动sync_user(uid, CELERY_RECONCILE)→token_version 1。11. 验收标准矩阵AC-01 ~ AC-13AC验收点关键测试仓库中已存在AC-01每 6h 执行 SSO 全量校对test_beat_schedule_registers_reconcile_all_every_6h、test_reconcile_all_dispatches_single_config_per_active_configtest/celery/test_reconcile_celery_tasks.pyAC-02新部门自动 upsert 且不动挂载标记test_reconcile_new_dept_upserts_preserves_mountAC-03删除部门标记 is_deleted、挂载点 orphanedtest_reconcile_removed_dept_triggers_department_deletion_handlerAC-04主部门跨 Tenant 变更触发 UserTenantSyncServicetest_reconcile_primary_dept_change_triggers_user_tenant_syncAC-05external_id_map 策略应用成功test_relink_external_id_map_strategy_appliestest/department/test_department_relink_service.pyAC-06path_plus_name 单候选自动 apply、多候选 conflictstest_relink_path_plus_name_single_candidate_auto_apply、test_relink_path_plus_name_multi_candidate_returns_conflictsAC-07dry_run 返回 would_apply 不写入test_relink_dry_run_returns_would_apply_no_db_write HTTP dry_runtest/department/test_relink_api_integration.pyAC-0810 万部门校对 30 minlocust 压测占位scripts/performance/locust_ldap_reconcile_100k.py发版前专项不在 CIAC-09陈旧 ts 跳过 warn 事件行test_reconcile_stale_ts_skipped_writes_warn_eventAC-10更新 ts 应用并更新 last_sync_tstest_reconcile_newer_ts_applies_and_updates_last_sync_tsAC-11同 ts upsert/remove 以 remove 为准 audit 19317test_reconcile_same_ts_upsert_then_remove_prefers_removeAC-12周报 ≥3 次冲突告警 5 天升级每日test_weekly_report_aggregates_conflicts_above_threshold、test_daily_escalation_triggers_after_5_days_unresolvedAC-13并发同 external_id 同 ts 去重幂等test_concurrent_same_ts_same_config_deduped_by_setnxtest/org_sync/test_org_reconcile_service.py上表的关键测试均已实际存在于当前仓库的测试目录中如 test/org_sync/test_org_reconcile_service.py 包含全部 16 条 reconcile 相关用例涵盖 AC-02/03/04/09/10/11/13 与 event 持久化、锁独立性、异常释放、TTL 断言等可用pytest test/ -k org_sync or tenant or sso or reconcile or relink一键回归。12. 开发与运维命令速查# 单任务跑测 .venv/bin/pytest test/test_org_reconcile_service.py -v .venv/bin/pytest test/test_reconcile_celery_tasks.py -v .venv/bin/pytest test/test_ts_conflict_reporter.py -v .venv/bin/pytest test/test_department_relink_service.py -v .venv/bin/pytest test/test_relink_conflict_store.py -v .venv/bin/pytest test/test_remote_dept_differ.py -v .venv/bin/pytest test/test_org_sync_log_dao_f015.py -v # 迁移往返验证T02 .venv/bin/alembic upgrade head .venv/bin/alembic downgrade -1 .venv/bin/alembic upgrade head # 全 feature 回归 .venv/bin/pytest test/ -k org_sync or tenant or sso or reconcile or relink -v # Worker Beat 手工 QA .venv/bin/celery -A bisheng.worker.main:bisheng_celery worker -Q knowledge_celery -l info .venv/bin/celery -A bisheng.worker.main:bisheng_celery beat -l info # 手动触发 6h 校对不等待 Beat .venv/bin/python -c from bisheng.worker.org_sync.reconcile_tasks import reconcile_all_organizations; reconcile_all_organizations.delay() # relink HTTP 调用HMAC 签名X-Signature 用共享 secret 对 # POST\n/api/v1/internal/departments/relink\nbody 做 SHA256 得到 curl -X POST http://localhost:7860/api/v1/internal/departments/relink \ -H X-Signature: sha256 -H Content-Type: application/json \ -d {old_external_ids:[abc], matching_strategy:path_plus_name, dry_run:true}注意以上命令中的相对路径均相对src/backend目录执行reconcile_all_organizations.delay()是运维验证 6h 校对链路的快捷入口正常生产环境依赖 Celery Beat 的0 */6 * * *调度。13. 边界情况与运维要点校对期间部门变更Redis SETNX 锁避免并发同步AC-13。若锁忙Celery 任务只记 warning 不重试下一轮 6h 自动补偿。relink 冲突长期未解决候选在 Redis 中 7 天过期若 5 天内未通过resolve-conflict解决周报升级为每日告警直到冲突消失或人工处理。SSO 端暂时不可用authenticate或fetch_departments失败时整轮跳过result.errors记录不写任何变更下轮重试审计失败、事件行持久化失败均为单点容错不中断整轮校对。陈旧消息任何incoming_ts last_sync_ts的消息无论来自 Gateway 还是 Celery一律丢弃并写 warn 事件行不覆盖 bisheng 当前状态AC-09。性能专项AC-08 的 10 万部门压测wall time 30 min、MySQL/Redis 压力基线是发版前 2 周的专项工作不在 CI 范围基线数据记录于features/v2.5.1/015-ldap-reconcile-celery/ac-verification.md。14. 扩展阅读特性规格与验收标准features/v2.5.1/015-ldap-reconcile-celery/spec.md任务拆解与依赖图本文骨架来源features/v2.5.1/015-ldap-reconcile-celery/tasks.md主编排服务src/backend/bisheng/org_sync/domain/services/reconcile_service.pyts 决策守卫src/backend/bisheng/org_sync/domain/services/ts_guard.pyDiff 引擎src/backend/bisheng/org_sync/domain/services/remote_dept_differ.pyRelink 服务与冲突存储src/backend/bisheng/org_sync/domain/services/relink_service.py、src/backend/bisheng/org_sync/domain/services/relink_conflict_store.pyrelink HTTP 端点src/backend/bisheng/org_sync/api/endpoints/relink.py错误码定义src/backend/bisheng/common/errcode/sso_sync.py集成测试src/backend/test/org_sync/test_org_reconcile_service.py、src/backend/test/celery/test_reconcile_celery_tasks.py【免费下载链接】bishengBISHENG is an open LLM devops platform for next generation Enterprise AI applications. Powerful and comprehensive features include: GenAI workflow, RAG, Agent, Unified model management, Evaluation, SFT, Dataset Management, Enterprise-level System Management, Observability and more.项目地址: https://gitcode.com/GitHub_Trending/bi/bisheng创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考