diff --git a/src/backend/bisheng/database/models/tenant.py b/src/backend/bisheng/database/models/tenant.py index f728d8f9f..2af8c5348 100644 --- a/src/backend/bisheng/database/models/tenant.py +++ b/src/backend/bisheng/database/models/tenant.py @@ -373,6 +373,21 @@ class TenantDao: result = await session.exec(stmt) return [row for row in result.all()] + @classmethod + def get_children_ids_active(cls, root_id: int = ROOT_TENANT_ID) -> list[int]: + """Sync counterpart of :meth:`aget_children_ids_active`. + + Used by Celery Beat tasks that run in a synchronous worker context and + cannot await the async DAO (e.g. ``sync_information_article``). + """ + with bypass_tenant_filter(): + with get_sync_db_session() as session: + stmt = select(Tenant.id).where( + Tenant.parent_tenant_id == root_id, + Tenant.status == "active", + ) + return [row for row in session.exec(stmt).all()] + @classmethod async def aget_non_active_ids(cls) -> list[int]: """Return ids of tenants in disabled/archived/orphaned status. diff --git a/src/backend/bisheng/worker/information/article.py b/src/backend/bisheng/worker/information/article.py index 8575d6f51..bb6d0b634 100644 --- a/src/backend/bisheng/worker/information/article.py +++ b/src/backend/bisheng/worker/information/article.py @@ -1,19 +1,26 @@ -from bisheng.worker._asyncio_utils import run_async_task +import asyncio from datetime import datetime -from typing import Dict, List, Optional, Tuple from loguru import logger -from sqlmodel import select, func +from sqlmodel import select +from bisheng.core.context.tenant import ( + DEFAULT_TENANT_ID, + current_tenant_id, + set_current_tenant_id, +) from bisheng.core.database.dialect_helpers import json_array_contains +from bisheng.worker._asyncio_utils import run_async_task def _db_dialect() -> str: try: from bisheng.core.database.manager import sync_get_database_connection + return sync_get_database_connection().engine.dialect.name except Exception: - return 'mysql' + return "mysql" + from bisheng.channel.domain.models.channel import Channel from bisheng.channel.domain.models.channel_info_source import ChannelInfoSource @@ -21,8 +28,9 @@ from bisheng.channel.domain.models.channel_knowledge_sync import ( ChannelKnowledgeSync, ChannelKnowledgeSyncDao, ) -from bisheng.channel.domain.repositories.implementations.channel_info_source_repository_impl import \ - ChannelInfoSourceRepositoryImpl +from bisheng.channel.domain.repositories.implementations.channel_info_source_repository_impl import ( + ChannelInfoSourceRepositoryImpl, +) from bisheng.channel.domain.schemas.article_schema import ArticleDocument from bisheng.channel.domain.schemas.channel_manager_schema import ( AddArticlesToKnowledgeSpaceRequest, @@ -30,8 +38,9 @@ from bisheng.channel.domain.schemas.channel_manager_schema import ( from bisheng.channel.domain.services.article_es_service import ArticleEsService from bisheng.channel.domain.services.channel_service import ChannelService from bisheng.core.database import get_sync_db_session -from bisheng.core.external.bisheng_information_client.bisheng_information_manager import \ - get_bisheng_information_client_sync +from bisheng.core.external.bisheng_information_client.bisheng_information_manager import ( + get_bisheng_information_client_sync, +) from bisheng.core.logger import trace_id_var from bisheng.utils import generate_uuid from bisheng.worker.main import bisheng_celery @@ -40,13 +49,46 @@ from bisheng.worker.main import bisheng_celery @bisheng_celery.task def sync_information_article(information_id: str = None): trace_id_var.set(f"sync_all_information_articles_{generate_uuid()}") + # Celery Beat fires this once with no request/tenant context. The + # information-source tables are tenant-aware, so iterate every active tenant + # and run the sync under that tenant's context (mirrors reconcile_all_tenants); + # otherwise the first SELECT on channel_info_source raises NoTenantContextError + # and nothing ever syncs on a multi-tenant deploy. + for tenant_id in _active_tenant_ids_sync(): + token = set_current_tenant_id(tenant_id) + try: + _sync_information_article_for_tenant(information_id) + except Exception: + # Isolate per-tenant failures so one bad tenant never blocks the rest. + logger.exception(f"sync_information_article failed for tenant={tenant_id}") + finally: + current_tenant_id.reset(token) + + +def _active_tenant_ids_sync() -> list[int]: + """Active tenant ids to sync — sync counterpart of reconcile._active_tenant_ids. + + Single-tenant deployments behave as tenant_id=1. + """ + from bisheng.common.services.config_service import settings + + if not settings.multi_tenant.enabled: + return [DEFAULT_TENANT_ID] + + from bisheng.database.models.tenant import ROOT_TENANT_ID, TenantDao + + child_ids = TenantDao.get_children_ids_active() + return [ROOT_TENANT_ID, *child_ids] + + +def _sync_information_article_for_tenant(information_id: str = None): logger.debug(f"Starting to sync information articles for {information_id}.") article_service = ArticleEsService() article_service.ensure_index_sync() need_update_informations = [] # v2.5 Module D: record article ids indexed during THIS worker run so the # knowledge-space sync hook below can push just the fresh ones. - indexed_by_source: Dict[str, List[str]] = {} + indexed_by_source: dict[str, list[str]] = {} with get_sync_db_session() as session: channel_info_repository = ChannelInfoSourceRepositoryImpl(session) page, page_size = 1, 1000 @@ -57,10 +99,12 @@ def sync_information_article(information_id: str = None): for one in information_list: try: logger.debug(f"Syncing information for {one.id} - {one.source_name}") - if (one.update_time.strftime("%Y-%m-%d") == datetime.now().strftime("%Y-%m-%d") - ) and (one.update_time.strftime("%Y-%m-%d %H:%M") != one.create_time.strftime("%Y-%m-%d %H:%M")): + if (one.update_time.strftime("%Y-%m-%d") == datetime.now().strftime("%Y-%m-%d")) and ( + one.update_time.strftime("%Y-%m-%d %H:%M") != one.create_time.strftime("%Y-%m-%d %H:%M") + ): logger.debug( - f"Skip information for {one.id} - {one.source_name}, because it has already been updated today.") + f"Skip information for {one.id} - {one.source_name}, because it has already been updated today." + ) continue need_update_informations.append(one.id) @@ -72,7 +116,6 @@ def sync_information_article(information_id: str = None): except Exception as e: logger.exception(f"Failed to sync information article for source {one.id}: {e}") - page += 1 logger.debug("Finished syncing information articles") @@ -91,7 +134,7 @@ def sync_information_article(information_id: str = None): ) -def _update_channels_by_source_id(source_ids: List[str]): +def _update_channels_by_source_id(source_ids: list[str]): """Update latest_article_update_time for channels that use the specified source_id.""" with get_sync_db_session() as session: dialect = session.bind.dialect.name if session.bind else _db_dialect() @@ -117,31 +160,32 @@ def _sync_one_information_article(information: ChannelInfoSource, article_servic information_client = get_bisheng_information_client_sync() page, page_size, current = 1, 10, 0 - all_new_ids: List[str] = [] + all_new_ids: list[str] = [] while True: - resp = information_client.get_information_articles(information.id, False, - min_create_time=latest_create_time, - page=page, - page_size=page_size) + resp = information_client.get_information_articles( + information.id, False, min_create_time=latest_create_time, page=page, page_size=page_size + ) articles = [] doc_ids = [] for article in resp.articles: - articles.append(ArticleDocument( - source_type=0 if information.source_type == "wechat" else 1, - source_id=information.id, - title=article.title, - content=article.markdown_content, - content_html=article.html_content, - cover_image=article.icon, - publish_time=datetime.fromisoformat(article.publish_date), - source_url=article.original_url, - create_time=datetime.fromisoformat(article.create_time), - update_time=datetime.fromisoformat(article.update_time), - )) + articles.append( + ArticleDocument( + source_type=0 if information.source_type == "wechat" else 1, + source_id=information.id, + title=article.title, + content=article.markdown_content, + content_html=article.html_content, + cover_image=article.icon, + publish_time=datetime.fromisoformat(article.publish_date), + source_url=article.original_url, + create_time=datetime.fromisoformat(article.create_time), + update_time=datetime.fromisoformat(article.update_time), + ) + ) doc_ids.append(article.id) try: article_service.bulk_index_articles_sync(articles, doc_ids) - except Exception as e: + except Exception: # if es timeout or over memory change to one by one for tmp_index, tmp_one in enumerate(articles): article_service.index_article(tmp_one, doc_ids[tmp_index]) @@ -175,10 +219,11 @@ def _sync_one_information_article(information: ChannelInfoSource, article_servic def _find_sub_channel_filter_group( - channel: Channel, sub_channel_name: str, -) -> Optional[Dict]: + channel: Channel, + sub_channel_name: str, +) -> dict | None: """Return the ChannelFilterRules group for a sub-channel, or None.""" - for g in (channel.filter_rules or []): + for g in channel.filter_rules or []: if not isinstance(g, dict): continue if g.get("channel_type") == "sub" and g.get("name") == sub_channel_name: @@ -189,12 +234,12 @@ def _find_sub_channel_filter_group( def _resolve_article_ids_for_config( config: ChannelKnowledgeSync, channel: Channel, - indexed_by_source: Dict[str, List[str]], + indexed_by_source: dict[str, list[str]], article_service: ArticleEsService, -) -> List[str]: +) -> list[str]: """Compute the article ids this config should receive this run.""" channel_sources = [str(s) for s in (channel.source_list or [])] - new_ids: List[str] = [] + new_ids: list[str] = [] for sid in channel_sources: new_ids.extend(indexed_by_source.get(sid, [])) if not new_ids: @@ -219,24 +264,18 @@ def _resolve_article_ids_for_config( filter_rules=[group], ) except Exception as exc: - logger.exception( - f"Filter evaluation failed for config {config.id}: {exc}" - ) + logger.exception(f"Filter evaluation failed for config {config.id}: {exc}") return [] def _drop_dead_target_configs( - configs: List[ChannelKnowledgeSync], -) -> List[ChannelKnowledgeSync]: + configs: list[ChannelKnowledgeSync], +) -> list[ChannelKnowledgeSync]: """Filter out configs whose knowledge_space or folder has been deleted.""" space_ids = { - int(c.knowledge_space_id) for c in configs - if c.knowledge_space_id and str(c.knowledge_space_id).isdigit() - } - folder_ids = { - int(c.folder_id) for c in configs - if c.folder_id and str(c.folder_id).isdigit() + int(c.knowledge_space_id) for c in configs if c.knowledge_space_id and str(c.knowledge_space_id).isdigit() } + folder_ids = {int(c.folder_id) for c in configs if c.folder_id and str(c.folder_id).isdigit()} if not space_ids and not folder_ids: return list(configs) @@ -247,42 +286,32 @@ def _drop_dead_target_configs( existing_folders = set() with get_sync_db_session() as session: if space_ids: - for row in session.exec( - select(Knowledge.id).where(Knowledge.id.in_(space_ids)) - ).all(): + for row in session.exec(select(Knowledge.id).where(Knowledge.id.in_(space_ids))).all(): kid = row[0] if isinstance(row, tuple) else row existing_spaces.add(int(kid)) if folder_ids: - for row in session.exec( - select(KnowledgeFile.id).where(KnowledgeFile.id.in_(folder_ids)) - ).all(): + for row in session.exec(select(KnowledgeFile.id).where(KnowledgeFile.id.in_(folder_ids))).all(): fid = row[0] if isinstance(row, tuple) else row existing_folders.add(int(fid)) - survivors: List[ChannelKnowledgeSync] = [] + survivors: list[ChannelKnowledgeSync] = [] for c in configs: sid = c.knowledge_space_id if sid and str(sid).isdigit() and int(sid) not in existing_spaces: - logger.warning( - f"Sync config {c.id}: knowledge_space {sid} no longer exists; " - f"skipping." - ) + logger.warning(f"Sync config {c.id}: knowledge_space {sid} no longer exists; skipping.") continue fid_val = c.folder_id if fid_val and str(fid_val).isdigit() and int(fid_val) not in existing_folders: - logger.warning( - f"Sync config {c.id}: folder {fid_val} no longer exists; " - f"skipping." - ) + logger.warning(f"Sync config {c.id}: folder {fid_val} no longer exists; skipping.") continue survivors.append(c) return survivors def _sync_new_articles_to_knowledge_spaces( - indexed_by_source: Dict[str, List[str]], + indexed_by_source: dict[str, list[str]], information_id: str = None, - article_service: Optional[ArticleEsService] = None, + article_service: ArticleEsService | None = None, ) -> None: if not indexed_by_source: logger.debug("No freshly-indexed articles; skipping knowledge-space sync hook.") @@ -294,9 +323,7 @@ def _sync_new_articles_to_knowledge_spaces( if information_id: dialect = session.bind.dialect.name if session.bind else _db_dialect() channels = session.exec( - select(Channel).where( - json_array_contains(Channel.source_list, f'"{information_id}"', dialect) - ) + select(Channel).where(json_array_contains(Channel.source_list, f'"{information_id}"', dialect)) ).all() else: channels = session.exec(select(Channel)).all() @@ -320,13 +347,10 @@ def _sync_new_articles_to_knowledge_spaces( return channel_by_id = {c.id: c for c in channels} - logger.info( - f"Knowledge-space sync hook: {len(sync_configs)} enabled configs " - f"across {len(channel_ids)} channels." - ) + logger.info(f"Knowledge-space sync hook: {len(sync_configs)} enabled configs across {len(channel_ids)} channels.") # Build a list of dispatch tasks, skipping configs that have nothing to send. - pending: List[Tuple[ChannelKnowledgeSync, AddArticlesToKnowledgeSpaceRequest]] = [] + pending: list[tuple[ChannelKnowledgeSync, AddArticlesToKnowledgeSpaceRequest]] = [] for config in sync_configs: try: channel = channel_by_id.get(config.channel_id) @@ -334,7 +358,10 @@ def _sync_new_articles_to_knowledge_spaces( continue article_ids = _resolve_article_ids_for_config( - config, channel, indexed_by_source, article_service, + config, + channel, + indexed_by_source, + article_service, ) if not article_ids: continue @@ -343,29 +370,21 @@ def _sync_new_articles_to_knowledge_spaces( kid = int(config.knowledge_space_id) except (TypeError, ValueError): logger.warning( - f"Config {config.id}: non-int knowledge_space_id " - f"{config.knowledge_space_id!r}, skipping." + f"Config {config.id}: non-int knowledge_space_id {config.knowledge_space_id!r}, skipping." ) continue req = AddArticlesToKnowledgeSpaceRequest( knowledge_id=kid, article_ids=article_ids, - parent_id=( - int(config.folder_id) - if config.folder_id and str(config.folder_id).isdigit() - else None - ), + parent_id=(int(config.folder_id) if config.folder_id and str(config.folder_id).isdigit() else None), # Background sync is best-effort: missing ES docs and duplicate # file names in the target space must not abort the batch. skip_missing_and_duplicates=True, ) pending.append((config, req)) except Exception as exc: - logger.exception( - f"Failed to prepare sync for config {config.id} " - f"(channel {config.channel_id}): {exc}" - ) + logger.exception(f"Failed to prepare sync for config {config.id} (channel {config.channel_id}): {exc}") if not pending: return @@ -380,7 +399,7 @@ def _sync_new_articles_to_knowledge_spaces( async def _dispatch_all( - pending: List[Tuple[ChannelKnowledgeSync, AddArticlesToKnowledgeSpaceRequest]], + pending: list[tuple[ChannelKnowledgeSync, AddArticlesToKnowledgeSpaceRequest]], ) -> None: async def _one(config: ChannelKnowledgeSync, req: AddArticlesToKnowledgeSpaceRequest): try: @@ -392,16 +411,14 @@ async def _dispatch_all( f"(config {config.id})" ) except Exception as exc: - logger.exception( - f"Failed to sync config {config.id} " - f"(channel {config.channel_id}): {exc}" - ) + logger.exception(f"Failed to sync config {config.id} (channel {config.channel_id}): {exc}") await asyncio.gather(*(_one(c, r) for c, r in pending), return_exceptions=True) async def _async_add_articles_to_knowledge( - req: AddArticlesToKnowledgeSpaceRequest, user_id: int, + req: AddArticlesToKnowledgeSpaceRequest, + user_id: int, ): """Run add_articles_to_knowledge_space in an async context. diff --git a/src/backend/test/channel/test_sync_information_article_tenant.py b/src/backend/test/channel/test_sync_information_article_tenant.py new file mode 100644 index 000000000..6324c64b3 --- /dev/null +++ b/src/backend/test/channel/test_sync_information_article_tenant.py @@ -0,0 +1,78 @@ +"""Regression: sync_information_article must run under per-tenant context. + +The information-source tables (channel_info_source / channel / ...) are +tenant-aware. Celery Beat fires the task with no request context, so the task +body must iterate every active tenant and set its context before querying — +otherwise the first SELECT raises NoTenantContextError and nothing syncs on a +multi-tenant deploy. Mirrors test_information_reconcile_worker. +""" + +from __future__ import annotations + +from types import SimpleNamespace +from unittest.mock import MagicMock, patch + +import bisheng.worker.information.article as article_mod + + +def test_sync_iterates_each_active_tenant_under_context(): + """Each active tenant's context is set and the per-tenant sync runs once.""" + tenant_seen: list[int] = [] + synced_under: list[int] = [] + + with ( + patch.object(article_mod, "_active_tenant_ids_sync", return_value=[1, 2, 3]), + patch.object(article_mod, "set_current_tenant_id", side_effect=lambda t: tenant_seen.append(t) or f"tok-{t}"), + patch.object(article_mod, "current_tenant_id", MagicMock()), + patch.object( + article_mod, + "_sync_information_article_for_tenant", + side_effect=lambda info: synced_under.append(tenant_seen[-1]), + ), + ): + article_mod.sync_information_article.run() + + assert tenant_seen == [1, 2, 3] + assert synced_under == [1, 2, 3] + + +def test_sync_isolates_per_tenant_failure_and_resets_context(): + """One tenant failing does not stop the rest; context is reset every time.""" + tenant_seen: list[int] = [] + reset_tokens: list = [] + + ctx_mock = MagicMock() + ctx_mock.reset.side_effect = lambda tok: reset_tokens.append(tok) + + def _body(info): + if tenant_seen[-1] == 2: + raise RuntimeError("boom") + + with ( + patch.object(article_mod, "_active_tenant_ids_sync", return_value=[1, 2, 3]), + patch.object(article_mod, "set_current_tenant_id", side_effect=lambda t: tenant_seen.append(t) or f"tok-{t}"), + patch.object(article_mod, "current_tenant_id", ctx_mock), + patch.object(article_mod, "_sync_information_article_for_tenant", side_effect=_body), + ): + article_mod.sync_information_article.run() + + assert tenant_seen == [1, 2, 3] + # finally-block reset fires once per tenant, even for the one that raised. + assert reset_tokens == ["tok-1", "tok-2", "tok-3"] + + +def test_active_tenant_ids_single_tenant_fallback(): + """Single-tenant deploy syncs only the default tenant.""" + fake_settings = SimpleNamespace(multi_tenant=SimpleNamespace(enabled=False)) + with patch("bisheng.common.services.config_service.settings", fake_settings): + assert article_mod._active_tenant_ids_sync() == [article_mod.DEFAULT_TENANT_ID] + + +def test_active_tenant_ids_multi_tenant_includes_root_and_children(): + """Multi-tenant deploy syncs root + every active child tenant.""" + fake_settings = SimpleNamespace(multi_tenant=SimpleNamespace(enabled=True)) + with ( + patch("bisheng.common.services.config_service.settings", fake_settings), + patch("bisheng.database.models.tenant.TenantDao.get_children_ids_active", return_value=[2, 3]), + ): + assert article_mod._active_tenant_ids_sync() == [1, 2, 3]