opensource/bisheng
3
0
Fork 0
mirror of https://github.com/dataelement/bisheng.git synced 2026-08-30 17:58:00 +08:00
Code Issues Packages Projects Releases Wiki Activity

fix(channel): sync information articles under per-tenant context

Browse Source 
sync_information_article iterated channel_info_source (a tenant-aware table)
with no tenant context set, so on a multi-tenant deploy the first SELECT raised
NoTenantContextError and the daily article sync never ran — subscribed sources'
content never updated.

Mirror reconcile_all_tenants: enumerate active tenants (root + active children,
or the default tenant when multi-tenancy is off) and run the sync under each
tenant's context, isolating per-tenant failures. Add a sync counterpart
TenantDao.get_children_ids_active for the sync worker path, and fix a latent
missing `import asyncio` in the knowledge-space dispatch hook.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
LineWalker
2026-07-11 21:41:36 +08:00
parent c42a3edc08
commit 715ebaf991
3 changed files with 202 additions and 92 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/backend/bisheng/database/models/tenant.py
+15
View File
@@ -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.
src/backend/bisheng/worker/information/article.py
+109 -92
View File
@@ -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.
src/backend/test/channel/test_sync_information_article_tenant.py
+78
View File
@@ -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]