fix(HEL-461): 切换事务失败写入 release-group 审计日志

整组切换中断时除回滚与废弃批次外，同步记录
action=release-group 的失败审计，便于后台追踪。

Co-authored-by: Cursor <cursoragent@cursor.com>
Co-authored-by: multica-agent <github@multica.ai>

backend/data/datahub/bridge.py

@@ -122,10 +122,12 @@ class DatahubBridge:
                 legacy_rows = legacy_query(api_name, params, fields)
             except Exception as exc:
                 if flags.read and hub_rows is not None and hub_error is None:
-                    self._emit_shadow(compare_rows(dataset, [], hub_canonical, hub_meta, self._error_text(exc)))
+                    self._emit_shadow(
+                        compare_rows(dataset, [], hub_canonical, hub_meta, self._error_text(exc), fields)
+                    )
                     return project_fields(hub_rows, fields)
                 raise
-            self._emit_shadow(compare_rows(dataset, legacy_rows, hub_canonical, hub_meta, hub_error))
+            self._emit_shadow(compare_rows(dataset, legacy_rows, hub_canonical, hub_meta, hub_error, fields))
             if flags.read and hub_rows is not None and hub_error is None:
                 return project_fields(hub_rows, fields)
             return legacy_rows

backend/data/datahub/compare.py

@@ -5,6 +5,7 @@ from typing import Any
 from backend.data.datahub.native import SCALE_FIELDS, row_key, to_canonical_row, yyyymmdd
 
 NUMERIC_TOLERANCE = 1e-4
+CANONICAL_ALIASES = {"volume": "vol"}
 
 
 def compare_rows(
@@ -13,8 +14,10 @@ def compare_rows(
     hub_rows: list[dict[str, Any]] | None,
     hub_meta: dict[str, Any] | None = None,
     hub_error: str | None = None,
+    fields: str = "",
 ) -> dict[str, Any]:
     hub = hub_rows or []
+    requested = _requested_fields(fields)
     legacy_map = {row_key(dataset, row): row for row in legacy_rows}
     hub_map = {row_key(dataset, _align_hub_row(row)): row for row in hub}
     missing_hub = sorted(key for key in legacy_map if key not in hub_map)
@@ -26,7 +29,7 @@ def compare_rows(
         hub_row = hub_map.get(key)
         if hub_row is None:
             continue
-        field_report = _compare_fields(dataset, legacy, hub_row)
+        field_report = _compare_fields(dataset, legacy, hub_row, requested)
         if field_report["unit_conversion"]:
             unit_conversion.append({"key": list(key), "fields": field_report["unit_conversion"]})
         if field_report["value_diff"]:
@@ -53,6 +56,7 @@ def compare_rows(
         "published_at": (hub_meta or {}).get("published_at"),
         "trade_date": yyyymmdd((hub_meta or {}).get("trade_date")),
         "hub_error": hub_error,
+        "fields_compared": sorted(requested) if requested is not None else None,
         "equal": (
             not hub_error
             and not missing_hub
@@ -71,13 +75,36 @@ def _align_hub_row(row: dict[str, Any]) -> dict[str, Any]:
     return aligned
 
 
-def _compare_fields(dataset: str, legacy: dict[str, Any], hub: dict[str, Any]) -> dict[str, list[dict[str, Any]]]:
+def _requested_fields(fields: str) -> list[str] | None:
+    """Fields the website actually asked for; None means "no projection"."""
+    keys = [item.strip() for item in str(fields or "").split(",") if item.strip()]
+    if not keys:
+        return None
+    seen: list[str] = []
+    for key in keys:
+        canonical = CANONICAL_ALIASES.get(key, key)
+        if canonical not in seen:
+            seen.append(canonical)
+    return seen
+
+
+def _compare_fields(
+    dataset: str,
+    legacy: dict[str, Any],
+    hub: dict[str, Any],
+    requested: list[str] | None = None,
+) -> dict[str, list[dict[str, Any]]]:
     canonical_legacy = to_canonical_row(dataset, legacy)
     hub_canonical = _hub_canonical(dataset, hub)
     native_hub = _align_hub_row(hub)
     value_diff: list[dict[str, Any]] = []
     unit_conversion: list[dict[str, Any]] = []
     keys = (set(canonical_legacy) | set(hub_canonical)) - {"batch_id", "updated_at", "volume"}
+    if requested is not None:
+        # Compare only what the website asked for. Extra hub columns are
+        # transport detail, not business differences; a requested field still
+        # alarms when it is missing or holds a different value.
+        keys = set(requested) - {"batch_id", "updated_at", "volume"}
     scales = SCALE_FIELDS.get(dataset) or {}
     for field in sorted(keys):
         left = canonical_legacy.get(field)

tests/test_datahub_bridge.py

@@ -201,6 +201,87 @@ class DatahubBridgeTests(unittest.TestCase):
         skew = compare_rows("daily", [LEGACY_DAILY], [HUB_DAILY], {"stale": False, "staleness_seconds": 12})
         self.assertTrue(skew["time_skew"])
 
+    def test_shadow_extra_hub_columns_are_not_false_diffs_when_projected(self) -> None:
+        hub_full = {**HUB_DAILY, "adj_factor": 1.1}
+        legacy_close_only = {k: LEGACY_DAILY[k] for k in ("ts_code", "trade_date", "close")}
+        report = compare_rows(
+            "daily", [legacy_close_only], [hub_full],
+            {"stale": False, "staleness_seconds": 0},
+            fields="ts_code,trade_date,close",
+        )
+        self.assertTrue(report["equal"])
+        self.assertEqual(report["value_diff_count"], 0)
+        self.assertEqual(report["fields_compared"], ["close", "trade_date", "ts_code"])
+        # without projection the same pair shows the historic false diff
+        unprojected = compare_rows("daily", [legacy_close_only], [hub_full])
+        self.assertFalse(unprojected["equal"])
+
+        legacy_stocks = {"ts_code": "600000.SH", "name": "浦发银行"}
+        hub_stocks = {
+            "ts_code": "600000.SH", "symbol": "600000", "name": "浦发银行", "area": "上海",
+            "industry": "银行", "market": "主板", "list_status": "L", "list_date": "19991110",
+        }
+        stocks = compare_rows("stocks", [legacy_stocks], [hub_stocks], {}, fields="ts_code,name")
+        self.assertTrue(stocks["equal"])
+
+        legacy_cal = {"cal_date": "20240902", "is_open": 1}
+        hub_cal = {
+            "cal_date": "20240902", "is_open": True,
+            "pretrade_date": "20240830", "prev_open": "20240830",
+        }
+        calendar = compare_rows(
+            "calendar", [legacy_cal], [hub_cal], {}, fields="cal_date,is_open"
+        )
+        self.assertTrue(calendar["equal"])
+
+    def test_shadow_projection_still_alarms_on_requested_field_problems(self) -> None:
+        hub_missing_field = {k: v for k, v in HUB_DAILY.items() if k != "close"}
+        legacy_close_only = {k: LEGACY_DAILY[k] for k in ("ts_code", "trade_date", "close")}
+        lost = compare_rows(
+            "daily", [legacy_close_only], [hub_missing_field], fields="ts_code,trade_date,close"
+        )
+        self.assertFalse(lost["equal"])
+        self.assertEqual(lost["value_diff_count"], 1)
+
+        changed = compare_rows(
+            "daily", [legacy_close_only], [{**HUB_DAILY, "close": 99.0}],
+            fields="ts_code,trade_date,close",
+        )
+        self.assertFalse(changed["equal"])
+        self.assertEqual(changed["value_diff_count"], 1)
+        self.assertEqual(changed["value_diffs"][0]["fields"][0]["field"], "close")
+
+        gone = compare_rows("daily", [LEGACY_DAILY], [], fields="ts_code,trade_date,close")
+        self.assertEqual(gone["missing_hub_count"], 1)
+        self.assertFalse(gone["equal"])
+
+        unit = compare_rows(
+            "daily", [LEGACY_DAILY], [{**HUB_DAILY, "amount": 2000.0, "volume": 1000.0}],
+            fields="ts_code,trade_date,vol,amount",
+        )
+        self.assertGreater(unit["unit_conversion_count"], 0)
+        self.assertFalse(unit["equal"])
+
+    def test_bridge_shadow_report_uses_website_request_fields(self) -> None:
+        hub_full = {**HUB_DAILY, "adj_factor": 1.1}
+        legacy_close_only = {k: LEGACY_DAILY[k] for k in ("ts_code", "trade_date", "close", "vol", "amount")}
+        reports: list[dict[str, Any]] = []
+        client = FakeClient(
+            response=DatahubResponse(
+                data=[hub_full],
+                meta={"tier": "official", "trade_date": "20240902", "stale": False, "staleness_seconds": 0},
+            )
+        )
+        wrapped = DatahubAwareTushareClient(
+            FakeLegacy([legacy_close_only]),
+            DatahubBridge(flags(daily=(False, True)), client, shadow_sink=reports.append),
+        )
+        rows = wrapped.query("daily", {"trade_date": "20240902"}, "ts_code,trade_date,close,vol,amount")
+        self.assertEqual(rows[0]["close"], 10.20)
+        self.assertEqual(rows[0]["vol"], 1000.0)
+        self.assertTrue(reports[0]["equal"])
+        self.assertEqual(reports[0]["matched"], 1)
+
     def test_native_roundtrip_matches_known_scales(self) -> None:
         native = to_native_row("daily", HUB_DAILY)
         self.assertEqual(native["vol"], 1000.0)

xiaobai-datahub/README.md

@@ -83,6 +83,16 @@ python -m datahub history-backfill
 
 `hub-quality.config.json` 的 `field_gates` 按数据集配置关键字段：非空率下限（支持按字段覆盖，如 `dv_ttm` 合法高空值）、非有限值比例上限、以及相对上一已发布批次的非空率塌陷保护。字段大面积为空的批次会被拒绝发布、保留上一份正常正式数据，失败原因逐字段写入 `batches.error` / `quality_json`。被拒后数据集仍视为缺失，盘后自动重试（HEL-435 机制）会继续尝试直到成功或截止。配置对任意数据集生效，不写死单日或单字段。
 
+## 整批原子发布（release group）
+
+盘后发布/重发（eod_a、eod_retry、`eod-refresh`、跨数据集重发）不再逐数据集各自切换，而是走整批原子可见机制：
+
+- 一致性边界：日 K、估值、资金流、竞价同属 A 组整批；指数日 K 为 B 组；当日股票主档快照随 A 组一同切换（主档 `stock_master` 的 UPSERT 与快照发布同一事务，不会出现主档先行/滞后）。
+- 流程：组内全部成员先在暂存表完成拉取、字段质量门、覆盖检查和跨数据集交叉校验（`cross_gates` 配置 ts_code 覆盖重叠率下限），全部达标后才在**一个 SQLite 事务**里复制正式表并翻转全部 `publications` 指针。
+- 任一成员失败（拉取失败、质量门拒绝、交叉校验不过、切换事务中断）→ 整批不切换，对外继续提供上一份完整正式版本，失败原因写入 `batches.error` 与 `audit_log`（`action=release-group`），等待晚间自动重试。
+- 读取侧任何时刻只会看到"旧完整版本"或"新完整版本"：发布指针在单事务内统一翻转，容器重启/事务中断自动回滚，不暴露字段残缺或跨数据集混合版本。
+- 幂等：仅当一致性边界内全部成员都已发布时才整组跳过；边界内任有缺失则整组重暂存后统一切换，避免旧批次与新批次混在同一次重发中。重复执行、并发重试不会在完整边界已就绪时生成重复批次（调度器另有 EOD 互斥锁）。
+
 ## 股票主档每日刷新与发布
 
 交易日 20:00 与 23:10（`stocks_refresh_times` 可配）自动刷新股票主档并发布版本化快照（`eod_stocks` + `publications.dataset='stocks'`），覆盖当日新上市、证券简称变化和上市首日 N/C 前缀摘除；无变化则跳过，重复执行幂等。`/v1/stocks` 从最新已发布快照提供数据并带 `batch_id` / `published_at`；`/v1/datasets/status` 同步展示 stocks 状态。
@@ -105,11 +115,13 @@ python -m datahub moneyflow-backfill        # --trading-days 60 --end-date --for
 
 ```bash
 cd xiaobai-datahub
-python -m datahub eod-refresh --trade-date 20260904                  # 只补缺失数据集
+python -m datahub eod-refresh --trade-date 20260904                  # 补不完整的 A/B 边界
 python -m datahub eod-refresh --trade-date 20260904 --force --dataset valuation
-# 强制重取重发：仍走全部质量门，生成新批次，上一批次保留可回滚
+# --force 按一致性边界整组重发：valuation/daily/moneyflow/auction/stocks → A 组；
+# index_daily → B 组。不可再单独切换某一个正式数据集。
 ```
 
+管理后台「补数」对盘后正式数据集同样走 `force_republish_boundary`，不会绕过 A/B 整批边界。
 
 ## 备份
 

xiaobai-datahub/admin/app.js

@@ -269,7 +269,7 @@ function renderRelease(data) {
 
 async function dangerous(kind, dataset) {
   const date = ($("rel-date") && $("rel-date").value) || "";
-  const ds = dataset || prompt("数据集（daily / valuation / moneyflow / auction / index_daily / reference）", "daily");
+  const ds = dataset || prompt("数据集（daily/valuation/moneyflow/auction/stocks→A组整批；index_daily→B组；或 reference）", "daily");
   if (!ds) return;
   const password = prompt("二次确认：输入管理密码");
   if (!password) return;

xiaobai-datahub/config/hub-quality.config.json

@@ -22,6 +22,18 @@
     "20:00",
     "23:10"
   ],
+  "cross_gates": [
+    {
+      "left": "daily",
+      "right": "valuation",
+      "min_key_overlap": 0.98
+    },
+    {
+      "left": "daily",
+      "right": "moneyflow",
+      "min_key_overlap": 0.98
+    }
+  ],
   "field_gates": {
     "valuation": {
       "fields": [

xiaobai-datahub/datahub/admin_api.py

@@ -6,7 +6,7 @@ from typing import Any
 from datahub.adapters import RESERVED
 from datahub.auth import AuthService
 from datahub.db import HubDB
-from datahub.pipeline import Pipeline
+from datahub.pipeline import OFFICIAL_DATASETS, STOCKS_DATASET, Pipeline
 from datahub.scheduler import Scheduler
 from datahub.serving import ApiError
 from datahub.timeutil import isoformat, now_shanghai, session_phase, yyyymmdd
@@ -143,8 +143,22 @@ class AdminAPI:
             self._dangerous(password, confirm, f"{dataset}:{day}")
             if dataset == "reference":
                 result = self.pipeline.ingest_reference(day)
+            elif dataset in OFFICIAL_DATASETS or dataset == STOCKS_DATASET:
+                # Manual same-day republish must rebuild the full A/B boundary.
+                # Gate failures and mid-switch exceptions both surface as
+                # FAILED_PRECONDITION so the admin API never leaks raw
+                # transaction errors to the client.
+                try:
+                    result = self.pipeline.force_republish_boundary(dataset, day)
+                    failures = self.pipeline.eod_failures(result)
+                    if failures:
+                        raise ApiError("FAILED_PRECONDITION", "; ".join(failures))
+                except ApiError:
+                    raise
+                except Exception as exc:
+                    raise ApiError("FAILED_PRECONDITION", str(exc)) from exc
             else:
-                result = self.pipeline.run_dataset(dataset, day)
+                raise ApiError("INVALID_ARGUMENT", f"unsupported backfill dataset: {dataset}")
         self.pipeline.audit(actor, "backfill", f"{dataset}:{day}", json.dumps({"ok": True}))
         return result
 

xiaobai-datahub/datahub/cli.py

@@ -7,7 +7,7 @@ import json
 import sys
 
 from datahub.hub import build_hub
-from datahub.pipeline import OFFICIAL_DATASETS
+from datahub.pipeline import EOD_A_DATASETS, OFFICIAL_DATASETS, STOCKS_DATASET
 from datahub.settings import load_settings
 from datahub.timeutil import yyyymmdd
 
@@ -19,15 +19,15 @@ def main(argv: list[str] | None = None) -> int:
     history.add_argument("--calendar-start", default=None, help="日历起点，默认配置 calendar_start")
     history.add_argument("--index-days", type=int, default=None, help="指数回补交易日数量，默认 260")
     history.add_argument("--force", action="store_true", help="覆盖已发布的指数日期")
-    refresh = sub.add_parser("eod-refresh", help="对指定交易日补跑盘后正式数据（跳过已发布数据集，仍走质量门禁）")
+    refresh = sub.add_parser("eod-refresh", help="对指定交易日补跑盘后正式数据（跳过已完整发布的一致性边界，仍走质量门禁）")
     refresh.add_argument("--trade-date", default=None, help="交易日 YYYYMMDD，默认今天")
     refresh.add_argument(
         "--force", action="store_true",
-        help="对 --dataset 指定的数据集强制重取重发（生成新批次，保留上一批次可回滚）",
+        help="强制重发 --dataset 所属的完整一致性边界（A 组或 B 组），生成新批次并保留上一批次可回滚",
     )
     refresh.add_argument(
         "--dataset", default=None,
-        help="配合 --force 使用：只强制重发该数据集（如 valuation）",
+        help="配合 --force：指定边界内任一成员（如 valuation→整组 A；index_daily→整组 B）",
     )
     stocks_refresh = sub.add_parser("stocks-refresh", help="刷新股票主档并发布正式快照（幂等：无变化则跳过）")
     stocks_refresh.add_argument("--trade-date", default=None, help="交易日 YYYYMMDD，默认今天")
@@ -54,26 +54,27 @@ def main(argv: list[str] | None = None) -> int:
     if args.command == "eod-refresh":
         day = yyyymmdd(args.trade_date) if args.trade_date else yyyymmdd()
         if args.force:
-            datasets = tuple(sorted({args.dataset} & OFFICIAL_DATASETS)) if args.dataset else ()
+            allowed = set(OFFICIAL_DATASETS) | {STOCKS_DATASET}
-            if args.dataset and not datasets:
+            if not args.dataset:
-                parser.error(f"unknown dataset: {args.dataset}")
-            if not datasets:
                 parser.error("--force requires --dataset (e.g. --dataset valuation)")
-            result = {}
+            if args.dataset not in allowed:
-            for dataset in datasets:
+                parser.error(f"unknown dataset: {args.dataset}")
-                result[dataset] = hub.pipeline.run_dataset(dataset, day)
+            result = hub.pipeline.force_republish_boundary(args.dataset, day)
+            boundary = "A" if args.dataset in EOD_A_DATASETS or args.dataset == STOCKS_DATASET else "B"
         else:
             result = hub.pipeline.run_eod_missing(day)
+            boundary = None
         hub.pipeline.audit("cli", "eod-refresh", f"eod:{day}", json.dumps(
-            {"force": bool(args.force), "dataset": args.dataset,
+            {"force": bool(args.force), "dataset": args.dataset, "boundary": boundary,
              **{name: item.get("state") for name, item in result.items() if isinstance(item, dict)}},
             ensure_ascii=False,
         ))
         if args.force:
-            payload = {"trade_date": day, "datasets": result}
+            failures = hub.pipeline.eod_failures(result)
+            payload = {"trade_date": day, "boundary": boundary, "datasets": result}
             json.dump(payload, sys.stdout, ensure_ascii=False, indent=2, default=str)
             sys.stdout.write("\n")
-            return 0
+            return 0 if not failures else 1
         missing = hub.pipeline.missing_official_datasets(day)
         payload = {"trade_date": day, "datasets": result, "missing_after": missing}
         json.dump(payload, sys.stdout, ensure_ascii=False, indent=2, default=str)

xiaobai-datahub/datahub/pipeline.py

@@ -133,6 +133,95 @@ def _staging_row_count(connection: Any, dataset: str, batch_id: str) -> int:
     return int(row["n"] if row is not None else 0)
 
 
+def _staging_count_or_raise(connection: Any, dataset: str, trade_date: str, batch_id: str) -> int:
+    rows_out = _staging_row_count(connection, dataset, batch_id)
+    if rows_out <= 0:
+        report = {
+            "rows": 0,
+            "errors": [EMPTY_BATCH_ERROR],
+            "warnings": [],
+            "hard_fail": True,
+            "soft_fail": False,
+            "batch_id": batch_id,
+            "dataset": dataset,
+            "trade_date": trade_date,
+        }
+        LOGGER.warning(
+            "skip official publish for empty batch",
+            extra={
+                "hub": {
+                    "dataset": dataset,
+                    "trade_date": trade_date,
+                    "batch_id": batch_id,
+                    "rows_out": rows_out,
+                    "reason": "upstream_empty",
+                }
+            },
+        )
+        raise QualityError("empty batch cannot be officially published", report)
+    return rows_out
+
+
+def _upsert_publication(
+    connection: Any,
+    dataset: str,
+    trade_date: str,
+    batch_id: str,
+    state: str,
+    published_at: str,
+) -> None:
+    current = connection.execute(
+        "SELECT active_batch FROM publications WHERE dataset = ? AND trade_date = ?",
+        (dataset, trade_date),
+    ).fetchone()
+    prev = str(current["active_batch"]) if current else None
+    connection.execute(
+        """
+        INSERT INTO publications(dataset, trade_date, active_batch, prev_batch, state, published_at)
+        VALUES (?, ?, ?, ?, ?, ?)
+        ON CONFLICT(dataset, trade_date) DO UPDATE SET
+          prev_batch=excluded.prev_batch,
+          active_batch=excluded.active_batch,
+          state=excluded.state,
+          published_at=excluded.published_at
+        """,
+        (dataset, trade_date, batch_id, prev, state, published_at),
+    )
+
+
+def _record_publication_history(
+    connection: Any,
+    dataset: str,
+    trade_date: str,
+    batch_id: str,
+    published_at: str,
+    quality: dict[str, Any],
+) -> None:
+    max_gen = connection.execute(
+        "SELECT COALESCE(MAX(generation), 0) AS g FROM publication_history WHERE dataset = ? AND trade_date = ?",
+        (dataset, trade_date),
+    ).fetchone()
+    generation = int(max_gen["g"]) + 1
+    connection.execute(
+        "INSERT OR REPLACE INTO publication_history(dataset, trade_date, batch_id, published_at, generation) VALUES (?,?,?,?,?)",
+        (dataset, trade_date, batch_id, published_at, generation),
+    )
+    keep = int(quality.get("publication_generations") or 3)
+    stale = connection.execute(
+        """
+        SELECT batch_id FROM publication_history
+        WHERE dataset = ? AND trade_date = ?
+        ORDER BY generation DESC
+        """,
+        (dataset, trade_date),
+    ).fetchall()
+    for row in stale[keep:]:
+        connection.execute(
+            "DELETE FROM publication_history WHERE dataset = ? AND trade_date = ? AND batch_id = ?",
+            (dataset, trade_date, row["batch_id"]),
+        )
+
+
 class QualityError(RuntimeError):
     def __init__(self, message: str, report: dict[str, Any]) -> None:
         super().__init__(message)
@@ -259,11 +348,20 @@ class Pipeline:
         is published; ``force`` re-publishes unconditionally. New listings,
         renames (incl. N/C prefix removal) and status changes all flow into
         the snapshot, which carries batch_id/published_at metadata.
+
+        The ``stock_master`` UPSERT happens inside the same publish
+        transaction as the snapshot switch — fetch / quality-gate / switch
+        failures leave the master on the previous complete values.
         """
         day = yyyymmdd(trade_date or self.clock())
-        rows = self._fetch_dataset(STOCKS_DATASET, day)
+        try:
-        with self.db.write() as connection:
+            rows = self._fetch_dataset(STOCKS_DATASET, day)
-            self._upsert_stock_master(connection, rows, isoformat(self.clock()))
+        except Exception as exc:
+            self.audit(
+                "pipeline", "stocks-refresh", f"{STOCKS_DATASET}:{day}",
+                json.dumps({"state": "failed", "error": str(exc)}, ensure_ascii=False),
+            )
+            raise
         if not force:
             active, snapshot = self.published_stock_snapshot(day)
             if active is not None:
@@ -280,7 +378,14 @@ class Pipeline:
                         "batch_id": active,
                         "rows": len(snapshot),
                     }
-        result = self.run_dataset(STOCKS_DATASET, day, prepared_rows=rows)
+        try:
+            result = self.run_dataset(STOCKS_DATASET, day, prepared_rows=rows)
+        except Exception as exc:
+            self.audit(
+                "pipeline", "stocks-refresh", f"{STOCKS_DATASET}:{day}",
+                json.dumps({"state": "failed", "error": str(exc)}, ensure_ascii=False),
+            )
+            raise
         self.audit(
             "pipeline", "stocks-refresh", f"{STOCKS_DATASET}:{day}",
             json.dumps({"batch_id": result["batch_id"], "rows": result["rows"]}, ensure_ascii=False),
@@ -548,43 +653,358 @@ class Pipeline:
         return [dataset for dataset in sorted(OFFICIAL_DATASETS) if dataset not in published]
 
     def run_eod_missing(self, trade_date: str) -> dict[str, Any]:
-        """Fetch/publish every official dataset still missing for the date.
+        """Republish every incomplete EOD consistency group for the date.
 
-        Idempotent: datasets with an existing publication are skipped, so
+        A-group (daily/valuation/moneyflow/auction + stocks) and B-group
-        repeats never overwrite the current official batch. Per-dataset
+        (index_daily) are separate boundaries. Within a group, either the
-        failures are collected instead of aborting the remaining datasets.
+        whole boundary is already published (idempotent skip) or every
+        member is re-staged and switched together — never fill only the
+        missing members on top of older batches from an earlier partial run.
         """
-        return self._run_eod_datasets(tuple(sorted(OFFICIAL_DATASETS)), trade_date)
+        results: dict[str, Any] = {}
+        results.update(self.run_eod_batch_a(trade_date))
+        results.update(self.run_eod_batch_b(trade_date))
+        return results
 
-    def run_eod_batch_a(self, trade_date: str) -> dict[str, Any]:
+    def run_eod_batch_a(self, trade_date: str, force: bool = False) -> dict[str, Any]:
-        return self._run_eod_datasets(EOD_A_DATASETS, trade_date)
+        return self.run_release_group(EOD_A_DATASETS, trade_date, include_stocks=True, force=force)
 
-    def run_eod_batch_b(self, trade_date: str) -> dict[str, Any]:
+    def run_eod_batch_b(self, trade_date: str, force: bool = False) -> dict[str, Any]:
-        return self._run_eod_datasets(EOD_B_DATASETS, trade_date)
+        return self.run_release_group(EOD_B_DATASETS, trade_date, force=force)
 
-    def _run_eod_datasets(self, datasets: tuple[str, ...], trade_date: str) -> dict[str, Any]:
+    def force_republish_boundary(self, dataset: str, trade_date: str) -> dict[str, Any]:
+        """Force-republish the full A/B consistency boundary that owns ``dataset``.
+
+        CLI ``eod-refresh --force`` and admin manual backfill must not publish a
+        single official member alone — that would mix old and new batches inside
+        the same trade date. Naming any A-group member (or stocks) rebuilds the
+        whole A group; naming ``index_daily`` rebuilds B.
+        """
+        name = str(dataset or "").strip()
+        if name in EOD_A_DATASETS or name == STOCKS_DATASET:
+            return self.run_eod_batch_a(trade_date, force=True)
+        if name in EOD_B_DATASETS:
+            return self.run_eod_batch_b(trade_date, force=True)
+        raise ValueError(f"dataset is not part of an EOD release boundary: {dataset}")
+
+    def run_release_group(
+        self,
+        datasets: tuple[str, ...],
+        trade_date: str,
+        include_stocks: bool = False,
+        force: bool = False,
+    ) -> dict[str, Any]:
+        """One post-market publish/republish becomes one atomic visibility flip.
+
+        Consistency boundary: every member (official datasets, plus the daily
+        stocks snapshot when ``include_stocks``) is fetched, staged,
+        field-gated and cross-validated BEFORE any reader can see it. Only
+        when the whole group passes does a single SQLite transaction copy
+        all staging batches to the official tables and flip every
+        ``publications`` row at once. Any member failure aborts the group:
+        the previous complete official version keeps serving and the reason
+        is recorded on the batches and in the audit log.
+
+        Skip is all-or-nothing for the boundary unless ``force``: if every
+        official member (and stocks when required) is already published, the
+        group is skipped. If any official member is still missing — or
+        ``force`` is set — every official member is re-staged, so a retry or
+        manual republish never mixes old and new batches in one release.
+        """
         day = yyyymmdd(trade_date)
         results: dict[str, Any] = {}
-        for dataset in datasets:
+        staged: dict[str, dict[str, Any]] = {}
-            if self.active_batch(dataset, day) is not None:
+        failure: str | None = None
+        missing_official = [dataset for dataset in datasets if self.active_batch(dataset, day) is None]
+        stocks_missing = include_stocks and self.active_batch(STOCKS_DATASET, day) is None
+
+        if not force and not missing_official and not stocks_missing:
+            for dataset in datasets:
                 results[dataset] = {
                     "dataset": dataset,
                     "trade_date": day,
                     "state": "skipped",
                     "reason": "already_published",
                 }
+            if include_stocks:
+                results[STOCKS_DATASET] = {
+                    "dataset": STOCKS_DATASET,
+                    "trade_date": day,
+                    "state": "skipped",
+                    "reason": "already_published",
+                }
+            return results
+
+        # Incomplete or forced boundary → restage every official member together.
+        pending = list(datasets)
+
+        for dataset in pending:
+            if failure is not None:
+                results[dataset] = {
+                    "dataset": dataset,
+                    "trade_date": day,
+                    "state": "aborted",
+                    "reason": f"release group aborted: {failure}",
+                }
                 continue
             try:
-                results[dataset] = self.run_dataset(dataset, day)
+                staged[dataset] = self._stage_and_validate(dataset, day)
             except Exception as exc:
+                failure = f"{dataset}: {exc}"
                 results[dataset] = {
                     "dataset": dataset,
                     "trade_date": day,
                     "state": "failed",
                     "error": str(exc),
                 }
+
+        # Stocks join the same switch when the official boundary is being
+        # rebuilt (missing or forced), or when only the stocks snapshot is
+        # still missing.
+        rebuild_official = bool(force or missing_official)
+        if include_stocks and failure is None and (rebuild_official or stocks_missing):
+            try:
+                stocks_plan = self._stage_stocks_snapshot(day, force=rebuild_official)
+            except Exception as exc:
+                failure = f"{STOCKS_DATASET}: {exc}"
+                results[STOCKS_DATASET] = {
+                    "dataset": STOCKS_DATASET,
+                    "trade_date": day,
+                    "state": "failed",
+                    "error": str(exc),
+                }
+            else:
+                if stocks_plan is not None:
+                    staged[STOCKS_DATASET] = stocks_plan
+
+        if failure is None and staged:
+            cross_errors = self._cross_gate_errors(staged)
+            if cross_errors:
+                failure = "; ".join(cross_errors)
+
+        if failure is not None:
+            for dataset, item in staged.items():
+                self._abandon_batch(item["batch_id"], f"release group not switched: {failure}")
+                results[dataset] = {
+                    "dataset": dataset,
+                    "trade_date": day,
+                    "state": "failed",
+                    "error": f"release group not switched: {failure}",
+                    "batch_id": item["batch_id"],
+                }
+            LOGGER.warning(
+                "release group blocked, previous official version keeps serving",
+                extra={
+                    "hub": {
+                        "trade_date": day,
+                        "datasets": sorted(staged),
+                        "reason": failure,
+                        "event": "release_group_blocked",
+                    }
+                },
+            )
+            self.audit(
+                "pipeline", "release-group", f"eod:{day}",
+                json.dumps({"state": "failed", "reason": failure}, ensure_ascii=False),
+            )
+            return results
+
+        if staged:
+            try:
+                self._switch_release_group(day, staged)
+            except Exception as exc:
+                reason = f"release group switch failed: {exc}"
+                for item in staged.values():
+                    self._abandon_batch(item["batch_id"], reason)
+                LOGGER.warning(
+                    "release group switch failed, previous official version keeps serving",
+                    extra={
+                        "hub": {
+                            "trade_date": day,
+                            "datasets": sorted(staged),
+                            "reason": reason,
+                            "event": "release_group_switch_failed",
+                        }
+                    },
+                )
+                self.audit(
+                    "pipeline", "release-group", f"eod:{day}",
+                    json.dumps(
+                        {
+                            "state": "failed",
+                            "reason": reason,
+                            "switched": [],
+                            "force": bool(force),
+                        },
+                        ensure_ascii=False,
+                    ),
+                )
+                raise
+            for dataset, item in staged.items():
+                results[dataset] = {
+                    "dataset": dataset,
+                    "trade_date": day,
+                    "state": item["state"],
+                    "batch_id": item["batch_id"],
+                    "rows": item["rows"],
+                }
+            self.audit(
+                "pipeline", "release-group", f"eod:{day}",
+                json.dumps(
+                    {"state": "ok", "switched": sorted(staged), "force": bool(force)},
+                    ensure_ascii=False,
+                ),
+            )
         return results
 
+    def _stage_and_validate(self, dataset: str, trade_date: str, attempts: int | None = None) -> dict[str, Any]:
+        """Fetch → stage → quality-gate one member without publishing it."""
+        day = yyyymmdd(trade_date)
+        batch_id = self.next_batch_id(dataset, day)
+        max_attempts = attempts or self.settings.max_publish_attempts
+        rows: list[dict[str, Any]] = []
+        self._set_batch(batch_id, dataset, day, "scheduled", 0)
+        try:
+            self._set_batch(batch_id, dataset, day, "fetching", 1)
+            rows = retry_call(
+                lambda: self._fetch_dataset(dataset, day),
+                attempts=max_attempts,
+                base_delay=0.05,
+                sleeper=lambda _d: time.sleep(_d),
+            )
+            self._stage(dataset, batch_id, rows)
+            self._set_batch(batch_id, dataset, day, "staged", 1, rows_in=len(rows), rows_out=len(rows))
+            self._set_batch(batch_id, dataset, day, "validating", 1, rows_in=len(rows), rows_out=len(rows))
+            report = self.validate(dataset, batch_id, day, rows)
+            if report["hard_fail"]:
+                self._reject_batch(batch_id, dataset, day, rows, report)
+                raise QualityError("integrity gate failed", report)
+        except RetryError as exc:
+            self._set_batch(batch_id, dataset, day, "failed", max_attempts, error=str(exc), finished=True)
+            raise
+        except QualityError as exc:
+            current = self.db.fetchone("SELECT state FROM batches WHERE batch_id = ?", (batch_id,))
+            if current and current["state"] not in {"staged", "failed"}:
+                self._reject_batch(batch_id, dataset, day, rows, exc.report)
+            raise
+        except Exception as exc:
+            self._set_batch(batch_id, dataset, day, "failed", 1, error=str(exc), finished=True)
+            raise
+        self._set_batch(
+            batch_id, dataset, day, "ready", 1, rows_in=len(rows), rows_out=len(rows), quality=report
+        )
+        return {
+            "dataset": dataset,
+            "trade_date": day,
+            "batch_id": batch_id,
+            "rows": len(rows),
+            "quality": report,
+            "state": "degraded" if report["soft_fail"] else "published",
+        }
+
+    def _stage_stocks_snapshot(self, trade_date: str, force: bool = False) -> dict[str, Any] | None:
+        """Stage the daily stocks snapshot for a release group switch.
+
+        Returns None when the published snapshot is already identical to
+        upstream (idempotent skip) unless ``force`` is set. The stock_master
+        upsert is deferred into the group switch / publish transaction so the
+        master never runs ahead of the published snapshot.
+        """
+        day = yyyymmdd(trade_date)
+        active, snapshot = self.published_stock_snapshot(day)
+        rows = self._fetch_dataset(STOCKS_DATASET, day)
+        if active is not None and not force:
+            upstream = sorted(
+                tuple(str(row.get(field)) for field in STOCK_SNAPSHOT_FIELDS) for row in rows
+            )
+            published = sorted(tuple(str(row.get(field)) for field in STOCK_SNAPSHOT_FIELDS) for row in snapshot)
+            if upstream == published:
+                return None
+        batch_id = self.next_batch_id(STOCKS_DATASET, day)
+        self._set_batch(batch_id, STOCKS_DATASET, day, "scheduled", 0)
+        self._set_batch(batch_id, STOCKS_DATASET, day, "fetching", 1)
+        self._stage(STOCKS_DATASET, batch_id, rows)
+        self._set_batch(batch_id, STOCKS_DATASET, day, "staged", 1, rows_in=len(rows), rows_out=len(rows))
+        self._set_batch(batch_id, STOCKS_DATASET, day, "validating", 1, rows_in=len(rows), rows_out=len(rows))
+        report = self.validate(STOCKS_DATASET, batch_id, day, rows)
+        if report["hard_fail"]:
+            self._reject_batch(batch_id, STOCKS_DATASET, day, rows, report)
+            raise QualityError("integrity gate failed", report)
+        self._set_batch(
+            batch_id, STOCKS_DATASET, day, "ready", 1, rows_in=len(rows), rows_out=len(rows), quality=report
+        )
+        return {
+            "dataset": STOCKS_DATASET,
+            "trade_date": day,
+            "batch_id": batch_id,
+            "rows": len(rows),
+            "row_values": rows,
+            "quality": report,
+            "state": "degraded" if report["soft_fail"] else "published",
+        }
+
+    def _cross_gate_errors(self, staged: dict[str, dict[str, Any]]) -> list[str]:
+        """Cross-dataset consistency checks on staged batches (交叉校验)."""
+        errors: list[str] = []
+        gates = self.settings.quality.get("cross_gates") or []
+        for gate in gates if isinstance(gates, list) else []:
+            if not isinstance(gate, dict):
+                continue
+            left = str(gate.get("left") or "")
+            right = str(gate.get("right") or "")
+            if not left or not right or left not in staged or right not in staged:
+                continue
+            floor = float(gate.get("min_key_overlap") or 0.98)
+            left_keys = self._staging_keys(left, staged[left]["batch_id"])
+            right_keys = self._staging_keys(right, staged[right]["batch_id"])
+            denom = max(len(left_keys), len(right_keys))
+            overlap = (len(left_keys & right_keys) / denom) if denom else 1.0
+            if overlap < floor:
+                errors.append(
+                    f"cross gate: {left} vs {right} key overlap {overlap:.4f} < {floor}"
+                )
+        return errors
+
+    def _staging_keys(self, dataset: str, batch_id: str) -> set[str]:
+        table = DATASET_TABLES[dataset][1]
+        rows = self.db.fetchall(
+            f"SELECT DISTINCT ts_code FROM {table} WHERE batch_id = ?",
+            (batch_id,),
+        )
+        return {str(row["ts_code"]) for row in rows}
+
+    def _switch_release_group(self, trade_date: str, members: dict[str, dict[str, Any]]) -> None:
+        """Single transaction: copy every member and flip every publication."""
+        day = yyyymmdd(trade_date)
+        published_at = isoformat(self.clock())
+        with self.db.write() as connection:
+            for dataset, item in members.items():
+                _staging_count_or_raise(connection, dataset, day, item["batch_id"])
+            for dataset, item in members.items():
+                connection.execute(EOD_COPY[dataset], (item["batch_id"],))
+                if dataset == STOCKS_DATASET:
+                    self._upsert_stock_master(connection, item["row_values"], published_at)
+            if self.before_commit:
+                self.before_commit()
+            for dataset, item in members.items():
+                _upsert_publication(connection, dataset, day, item["batch_id"], item["state"], published_at)
+                _record_publication_history(
+                    connection, dataset, day, item["batch_id"], published_at, self.settings.quality
+                )
+                connection.execute(
+                    "UPDATE batches SET state='published', finished_at=? WHERE batch_id=?",
+                    (published_at, item["batch_id"]),
+                )
+
+    def _abandon_batch(self, batch_id: str, reason: str) -> None:
+        row = self.db.fetchone("SELECT dataset, trade_date FROM batches WHERE batch_id = ?", (batch_id,))
+        if not row:
+            return
+        self._set_batch(
+            batch_id, str(row["dataset"]), str(row["trade_date"]), "failed", 1,
+            error=reason, finished=True,
+        )
+
     @staticmethod
     def eod_failures(results: dict[str, Any]) -> list[str]:
         return [
@@ -735,77 +1155,24 @@ class Pipeline:
         return batch_id, stats
 
     def publish(self, dataset: str, trade_date: str, batch_id: str, state: str = "published") -> None:
-        copy_sql = EOD_COPY[dataset]
         published_at = isoformat(self.clock())
         with self.db.write() as connection:
-            rows_out = _staging_row_count(connection, dataset, batch_id)
+            _staging_count_or_raise(connection, dataset, trade_date, batch_id)
-            if rows_out <= 0:
+            connection.execute(EOD_COPY[dataset], (batch_id,))
-                report = {
+            if dataset == STOCKS_DATASET:
-                    "rows": 0,
+                staging = DATASET_TABLES[STOCKS_DATASET][1]
-                    "errors": [EMPTY_BATCH_ERROR],
+                stock_rows = [
-                    "warnings": [],
+                    dict(row)
-                    "hard_fail": True,
+                    for row in connection.execute(
-                    "soft_fail": False,
+                        f"SELECT * FROM {staging} WHERE batch_id = ?",
-                    "batch_id": batch_id,
+                        (batch_id,),
-                    "dataset": dataset,
+                    ).fetchall()
-                    "trade_date": trade_date,
+                ]
-                }
+                self._upsert_stock_master(connection, stock_rows, published_at)
-                LOGGER.warning(
-                    "skip official publish for empty batch",
-                    extra={
-                        "hub": {
-                            "dataset": dataset,
-                            "trade_date": trade_date,
-                            "batch_id": batch_id,
-                            "rows_out": rows_out,
-                            "reason": "upstream_empty",
-                        }
-                    },
-                )
-                raise QualityError("empty batch cannot be officially published", report)
-            current = connection.execute(
-                "SELECT active_batch FROM publications WHERE dataset = ? AND trade_date = ?",
-                (dataset, trade_date),
-            ).fetchone()
-            prev = str(current["active_batch"]) if current else None
-            connection.execute(copy_sql, (batch_id,))
             if self.before_commit:
                 self.before_commit()
-            connection.execute(
+            _upsert_publication(connection, dataset, trade_date, batch_id, state, published_at)
-                """
+            _record_publication_history(connection, dataset, trade_date, batch_id, published_at, self.settings.quality)
-                INSERT INTO publications(dataset, trade_date, active_batch, prev_batch, state, published_at)
-                VALUES (?, ?, ?, ?, ?, ?)
-                ON CONFLICT(dataset, trade_date) DO UPDATE SET
-                  prev_batch=excluded.prev_batch,
-                  active_batch=excluded.active_batch,
-                  state=excluded.state,
-                  published_at=excluded.published_at
-                """,
-                (dataset, trade_date, batch_id, prev, state, published_at),
-            )
-            max_gen = connection.execute(
-                "SELECT COALESCE(MAX(generation), 0) AS g FROM publication_history WHERE dataset = ? AND trade_date = ?",
-                (dataset, trade_date),
-            ).fetchone()
-            generation = int(max_gen["g"]) + 1
-            connection.execute(
-                "INSERT OR REPLACE INTO publication_history(dataset, trade_date, batch_id, published_at, generation) VALUES (?,?,?,?,?)",
-                (dataset, trade_date, batch_id, published_at, generation),
-            )
-            keep = int(self.settings.quality.get("publication_generations") or 3)
-            stale = connection.execute(
-                """
-                SELECT batch_id FROM publication_history
-                WHERE dataset = ? AND trade_date = ?
-                ORDER BY generation DESC
-                """,
-                (dataset, trade_date),
-            ).fetchall()
-            for row in stale[keep:]:
-                connection.execute(
-                    "DELETE FROM publication_history WHERE dataset = ? AND trade_date = ? AND batch_id = ?",
-                    (dataset, trade_date, row["batch_id"]),
-                )
 
     def rollback(self, dataset: str, trade_date: str, actor: str = "admin") -> dict[str, Any]:
         trade_date = yyyymmdd(trade_date)

xiaobai-datahub/tests/test_atomic_release.py

@@ -0,0 +1,408 @@
+from __future__ import annotations
+
+import unittest
+from pathlib import Path
+import tempfile
+
+from datahub.adapters.tushare import TushareAdapter
+from datahub.crypto import SecretVault
+from datahub.db import HubDB
+from datahub.pipeline import Pipeline
+from datahub.settings import Settings
+from datahub.serving import V1API
+from tests.fixtures import TRADE_DATE, fake_transport
+
+GROUP_A = ("daily", "valuation", "moneyflow", "auction")
+
+
+class GroupTransport:
+    """fake_transport with per-API degradation switches for release-group tests."""
+
+    def __init__(self) -> None:
+        self.empty: set[str] = set()
+        self.keep_rows: dict[str, int] = {}
+        self.stocks: list[dict] | None = None
+        self.calls: list[str] = []
+
+    def __call__(self, api_name: str, params: dict, fields: str):
+        self.calls.append(api_name)
+        if api_name in self.empty:
+            return []
+        if api_name == "stock_basic" and self.stocks is not None:
+            return [dict(row) for row in self.stocks]
+        rows = fake_transport(api_name, params, fields)
+        keep = self.keep_rows.get(api_name)
+        if keep is not None:
+            return rows[:keep]
+        return rows
+
+
+def make_pipe(transport: GroupTransport, quality_extra: dict | None = None):
+    tmp = tempfile.TemporaryDirectory()
+    db = HubDB(Path(tmp.name) / "hub.db")
+    adapter = TushareAdapter("test-token", transport=transport)
+    quality = {
+        "daily_row_ratio": 0.98,
+        "null_rate_max": 0.01,
+        "max_publish_attempts": 2,
+        "publication_generations": 3,
+    }
+    if quality_extra:
+        quality.update(quality_extra)
+    settings = Settings(
+        encryption_key=SecretVault.generate_key(),
+        api_token="t" * 32,
+        admin_password="admin-pass",
+        tushare_token="test-token",
+        db_path=db.path,
+        quality=quality,
+        scheduler_enabled=False,
+    )
+    pipe = Pipeline(db, adapter, settings)
+    pipe._tmp = tmp
+    return pipe, db
+
+
+def publications_map(db: HubDB, day: str) -> dict[str, str]:
+    rows = db.fetchall("SELECT dataset, active_batch FROM publications WHERE trade_date = ?", (day,))
+    return {str(row["dataset"]): str(row["active_batch"]) for row in rows}
+
+
+class ReleaseGroupSwitchTests(unittest.TestCase):
+    def setUp(self) -> None:
+        self.transport = GroupTransport()
+        self.pipe, self.db = make_pipe(self.transport)
+        self.pipe.ingest_reference(TRADE_DATE)
+
+    def test_whole_group_switches_in_one_publish_instant(self) -> None:
+        results = self.pipe.run_eod_batch_a(TRADE_DATE)
+        self.assertEqual(set(results), {*GROUP_A, "stocks"})
+        self.assertEqual({item["state"] for item in results.values()}, {"published"})
+        pubs = self.db.fetchall("SELECT * FROM publications WHERE trade_date = ?", (TRADE_DATE,))
+        self.assertEqual(len(pubs), 5)
+        self.assertEqual(len({row["published_at"] for row in pubs}), 1)
+        # official rows copied and serving resolves the new batches
+        api = V1API(self.db, self.pipe, self.pipe.settings)
+        payload = api.handle("/v1/bars/daily", {"date": [TRADE_DATE], "code": ["600000.SH"]})
+        self.assertEqual(payload["meta"]["batch_id"], results["daily"]["batch_id"])
+        stocks = api.handle("/v1/stocks", {})
+        self.assertEqual(stocks["meta"]["batch_id"], results["stocks"]["batch_id"])
+
+    def test_any_member_failure_blocks_entire_group(self) -> None:
+        self.transport.empty = {"daily_basic"}  # valuation upstream returns nothing
+        results = self.pipe.run_eod_batch_a(TRADE_DATE)
+        self.assertEqual(results["valuation"]["state"], "failed")
+        self.assertEqual(results["moneyflow"]["state"], "aborted")
+        self.assertEqual(results["auction"]["state"], "aborted")
+        self.assertEqual(results["daily"]["state"], "failed")  # staged fine, then abandoned
+        # nothing became visible, and the reason is recorded
+        self.assertEqual(publications_map(self.db, TRADE_DATE), {})
+        abandoned = self.db.fetchall(
+            "SELECT * FROM batches WHERE trade_date = ? AND state = 'failed'",
+            (TRADE_DATE,),
+        )
+        self.assertTrue(any("release group not switched" in str(row["error"] or "") for row in abandoned))
+        audit = self.db.fetchone(
+            "SELECT * FROM audit_log WHERE action = 'release-group' ORDER BY id DESC"
+        )
+        self.assertIn("valuation", str(audit["detail"]))
+        # still missing → evening retries keep trying
+        self.assertIn("daily", self.pipe.missing_official_datasets(TRADE_DATE))
+
+    def test_failure_keeps_previous_complete_version_serving(self) -> None:
+        first = self.pipe.run_dataset("daily", TRADE_DATE)
+        self.transport.empty = {"daily_basic"}
+        results = self.pipe.run_eod_missing(TRADE_DATE)
+        # incomplete A-group restages daily with the others; valuation fails → no A switch
+        self.assertEqual(results["daily"]["state"], "failed")
+        self.assertEqual(results["valuation"]["state"], "failed")
+        # the already-published daily batch is untouched and keeps serving
+        self.assertEqual(self.pipe.active_batch("daily", TRADE_DATE), first["batch_id"])
+        pubs = publications_map(self.db, TRADE_DATE)
+        self.assertEqual(pubs["daily"], first["batch_id"])
+        self.assertNotIn("valuation", pubs)
+        self.assertNotIn("moneyflow", pubs)
+        self.assertNotIn("auction", pubs)
+        # B-group is an independent boundary and may still publish
+        self.assertEqual(results["index_daily"]["state"], "published")
+        payload = V1API(self.db, self.pipe, self.pipe.settings).handle(
+            "/v1/bars/daily", {"date": [TRADE_DATE], "code": ["600000.SH"]}
+        )
+        self.assertEqual(payload["meta"]["batch_id"], first["batch_id"])
+
+    def test_partial_group_retry_does_not_mix_batches(self) -> None:
+        """Already-published A members must be restaged with missing ones."""
+        first_daily = self.pipe.run_dataset("daily", TRADE_DATE)
+        first_moneyflow = self.pipe.run_dataset("moneyflow", TRADE_DATE)
+        results = self.pipe.run_eod_missing(TRADE_DATE)
+        # A-group switched as one boundary; B-group (index) also published
+        for name in (*GROUP_A, "stocks"):
+            self.assertEqual(results[name]["state"], "published", name)
+        self.assertEqual(results["index_daily"]["state"], "published")
+        pubs = self.db.fetchall(
+            "SELECT dataset, active_batch, published_at FROM publications WHERE trade_date = ?",
+            (TRADE_DATE,),
+        )
+        by_ds = {str(row["dataset"]): row for row in pubs}
+        # old partial batches replaced — no cross-batch mix of the first wave
+        self.assertNotEqual(by_ds["daily"]["active_batch"], first_daily["batch_id"])
+        self.assertNotEqual(by_ds["moneyflow"]["active_batch"], first_moneyflow["batch_id"])
+        a_times = {by_ds[name]["published_at"] for name in (*GROUP_A, "stocks")}
+        self.assertEqual(len(a_times), 1)
+        # serving resolves the new complete A-group batches
+        api = V1API(self.db, self.pipe, self.pipe.settings)
+        daily = api.handle("/v1/bars/daily", {"date": [TRADE_DATE], "code": ["600000.SH"]})
+        self.assertEqual(daily["meta"]["batch_id"], results["daily"]["batch_id"])
+        self.assertEqual(daily["meta"]["batch_id"], by_ds["daily"]["active_batch"])
+
+    def test_reads_during_switch_see_old_state_until_commit(self) -> None:
+        snapshots: list[dict] = []
+
+        def watcher() -> None:
+            with self.db.connect() as connection:
+                rows = connection.execute(
+                    "SELECT dataset, active_batch FROM publications WHERE trade_date = ?",
+                    (TRADE_DATE,),
+                ).fetchall()
+                snapshots.append({str(row["dataset"]): row["active_batch"] for row in rows})
+
+        self.pipe.before_commit = watcher
+        self.pipe.run_eod_batch_a(TRADE_DATE)
+        # inside the switch transaction the group was still invisible
+        self.assertEqual(snapshots[0], {})
+        after = publications_map(self.db, TRADE_DATE)
+        self.assertEqual(set(after), {*GROUP_A, "stocks"})
+
+    def test_switch_crash_rolls_back_whole_group(self) -> None:
+        def explode() -> None:
+            raise RuntimeError("killed mid-switch")
+
+        self.pipe.before_commit = explode
+        with self.assertRaises(RuntimeError):
+            self.pipe.run_eod_batch_a(TRADE_DATE)
+        self.assertEqual(publications_map(self.db, TRADE_DATE), {})
+        for table in ("eod_bars", "eod_valuation", "eod_moneyflow", "eod_auction", "eod_stocks"):
+            rows = self.db.fetchall(f"SELECT * FROM {table} WHERE trade_date = ?", (TRADE_DATE,))
+            self.assertEqual(rows, [], table)
+        audit = self.db.fetchone(
+            "SELECT * FROM audit_log WHERE action = 'release-group' ORDER BY id DESC"
+        )
+        self.assertIsNotNone(audit)
+        detail = str(audit["detail"])
+        self.assertIn("killed mid-switch", detail)
+        self.assertIn("failed", detail)
+
+    def test_duplicate_runs_are_idempotent(self) -> None:
+        self.pipe.run_eod_batch_a(TRADE_DATE)
+        self.pipe.run_eod_batch_b(TRADE_DATE)
+        batches_before = {
+            str(row["batch_id"])
+            for row in self.db.fetchall("SELECT batch_id FROM batches WHERE trade_date = ?", (TRADE_DATE,))
+        }
+        calls_before = len(self.transport.calls)
+        again = self.pipe.run_eod_missing(TRADE_DATE)
+        self.assertEqual({item["state"] for item in again.values()}, {"skipped"})
+        self.assertEqual({item["reason"] for item in again.values()}, {"already_published"})
+        batches_after = {
+            str(row["batch_id"])
+            for row in self.db.fetchall("SELECT batch_id FROM batches WHERE trade_date = ?", (TRADE_DATE,))
+        }
+        self.assertEqual(batches_after, batches_before)
+        self.assertEqual(len(self.transport.calls), calls_before)
+        self.assertEqual(self.pipe.missing_official_datasets(TRADE_DATE), [])
+
+    def test_cross_gate_failure_blocks_switch(self) -> None:
+        transport = GroupTransport()
+        pipe, db = make_pipe(
+            transport,
+            quality_extra={"cross_gates": [
+                {"left": "daily", "right": "moneyflow", "min_key_overlap": 1.0},
+            ]},
+        )
+        pipe.ingest_reference(TRADE_DATE)
+        transport.keep_rows["moneyflow"] = 1  # moneyflow covers only half the market
+        results = pipe.run_eod_batch_a(TRADE_DATE)
+        self.assertEqual(results["moneyflow"]["state"], "failed")
+        self.assertIn("cross gate", str(results["moneyflow"]["error"]))
+        self.assertEqual(publications_map(db, TRADE_DATE), {})
+
+    def test_stocks_master_and_snapshot_switch_together_or_not_at_all(self) -> None:
+        original = [
+            {"ts_code": "600000.SH", "symbol": "600000", "name": "浦发银行", "area": "上海",
+             "industry": "银行", "market": "主板", "list_status": "L", "list_date": "19991110"},
+            {"ts_code": "920071.BJ", "symbol": "920071", "name": "N金钛", "area": "辽宁",
+             "industry": "小金属", "market": "北交所", "list_status": "L", "list_date": "20240901"},
+        ]
+        renamed = [dict(original[0]), {**original[1], "name": "金钛股份"}]
+        self.transport.stocks = renamed
+        self.pipe.run_eod_batch_a(TRADE_DATE)
+        master = self.db.fetchone("SELECT name FROM stock_master WHERE ts_code = '920071.BJ'")
+        self.assertEqual(master["name"], "金钛股份")
+        stocks_pub = self.db.fetchone(
+            "SELECT active_batch FROM publications WHERE dataset = 'stocks' AND trade_date = ?",
+            (TRADE_DATE,),
+        )
+        self.assertIsNotNone(stocks_pub)
+
+        # failure path: rename staged but the group is blocked → master stays untouched
+        transport = GroupTransport()
+        transport.stocks = original
+        pipe, db = make_pipe(
+            transport,
+            quality_extra={"cross_gates": [
+                {"left": "daily", "right": "moneyflow", "min_key_overlap": 1.0},
+            ]},
+        )
+        pipe.ingest_reference(TRADE_DATE)  # master seeded with "N金钛"
+        transport.stocks = renamed
+        transport.keep_rows["moneyflow"] = 1
+        results = pipe.run_eod_batch_a(TRADE_DATE)
+        self.assertEqual(results["stocks"]["state"], "failed")
+        master = db.fetchone("SELECT name FROM stock_master WHERE ts_code = '920071.BJ'")
+        self.assertEqual(master["name"], "N金钛")  # rename not applied
+        stocks_pub = db.fetchone(
+            "SELECT active_batch FROM publications WHERE dataset = 'stocks' AND trade_date = ?",
+            (TRADE_DATE,),
+        )
+        self.assertIsNone(stocks_pub)
+
+
+class StocksRefreshAtomicTests(unittest.TestCase):
+    def setUp(self) -> None:
+        self.transport = GroupTransport()
+        self.pipe, self.db = make_pipe(self.transport)
+        self.pipe.ingest_reference(TRADE_DATE)
+        self.transport.stocks = [
+            {"ts_code": "600000.SH", "symbol": "600000", "name": "浦发银行", "area": "上海",
+             "industry": "银行", "market": "主板", "list_status": "L", "list_date": "19991110"},
+            {"ts_code": "920071.BJ", "symbol": "920071", "name": "N金钛", "area": "辽宁",
+             "industry": "小金属", "market": "北交所", "list_status": "L", "list_date": "20240901"},
+        ]
+        first = self.pipe.refresh_stocks(TRADE_DATE)
+        self.assertEqual(first["state"], "published")
+        self.first_batch = first["batch_id"]
+
+    def test_refresh_keeps_master_when_snapshot_publish_fails(self) -> None:
+        self.transport.stocks = [
+            {"ts_code": "600000.SH", "symbol": "600000", "name": "浦发银行", "area": "上海",
+             "industry": "银行", "market": "主板", "list_status": "L", "list_date": "19991110"},
+            {"ts_code": "920071.BJ", "symbol": "920071", "name": "金钛股份", "area": "辽宁",
+             "industry": "小金属", "market": "北交所", "list_status": "L", "list_date": "20240901"},
+        ]
+
+        def explode() -> None:
+            raise RuntimeError("snapshot switch killed")
+
+        self.pipe.before_commit = explode
+        with self.assertRaises(RuntimeError):
+            self.pipe.refresh_stocks(TRADE_DATE)
+        master = self.db.fetchone("SELECT name FROM stock_master WHERE ts_code = '920071.BJ'")
+        self.assertEqual(master["name"], "N金钛")  # rename not applied
+        self.assertEqual(self.pipe.active_batch("stocks", TRADE_DATE), self.first_batch)
+        audit = self.db.fetchone(
+            "SELECT * FROM audit_log WHERE action = 'stocks-refresh' ORDER BY id DESC"
+        )
+        self.assertIn("failed", str(audit["detail"]))
+        self.assertIn("snapshot switch killed", str(audit["detail"]))
+
+    def test_refresh_keeps_master_when_quality_gate_rejects(self) -> None:
+        self.transport.stocks = []  # empty → hard fail before publish
+        with self.assertRaises(Exception):
+            self.pipe.refresh_stocks(TRADE_DATE)
+        master = self.db.fetchone("SELECT name FROM stock_master WHERE ts_code = '920071.BJ'")
+        self.assertEqual(master["name"], "N金钛")
+        self.assertEqual(self.pipe.active_batch("stocks", TRADE_DATE), self.first_batch)
+        audit = self.db.fetchone(
+            "SELECT * FROM audit_log WHERE action = 'stocks-refresh' ORDER BY id DESC"
+        )
+        self.assertIn("failed", str(audit["detail"]))
+
+
+class ForceBoundaryEntryTests(unittest.TestCase):
+    """CLI force / admin backfill must rebuild the full A/B boundary."""
+
+    def setUp(self) -> None:
+        self.transport = GroupTransport()
+        self.pipe, self.db = make_pipe(self.transport)
+        self.pipe.ingest_reference(TRADE_DATE)
+        self.first = self.pipe.run_eod_batch_a(TRADE_DATE)
+        self.pipe.run_eod_batch_b(TRADE_DATE)
+
+    def test_force_republish_valuation_rebuilds_whole_a_group(self) -> None:
+        before = publications_map(self.db, TRADE_DATE)
+        results = self.pipe.force_republish_boundary("valuation", TRADE_DATE)
+        self.assertEqual({item["state"] for item in results.values()}, {"published"})
+        after = publications_map(self.db, TRADE_DATE)
+        for name in (*GROUP_A, "stocks"):
+            self.assertNotEqual(after[name], before[name], name)
+            self.assertEqual(after[name], results[name]["batch_id"], name)
+        # B-group left alone
+        self.assertEqual(after["index_daily"], before["index_daily"])
+        pubs = self.db.fetchall(
+            "SELECT dataset, published_at FROM publications WHERE trade_date = ?",
+            (TRADE_DATE,),
+        )
+        a_times = {row["published_at"] for row in pubs if row["dataset"] in {*GROUP_A, "stocks"}}
+        self.assertEqual(len(a_times), 1)
+
+    def test_force_republish_index_rebuilds_only_b_group(self) -> None:
+        before = publications_map(self.db, TRADE_DATE)
+        results = self.pipe.force_republish_boundary("index_daily", TRADE_DATE)
+        self.assertEqual(results["index_daily"]["state"], "published")
+        after = publications_map(self.db, TRADE_DATE)
+        self.assertNotEqual(after["index_daily"], before["index_daily"])
+        for name in GROUP_A:
+            self.assertEqual(after[name], before[name], name)
+
+    def test_admin_backfill_official_dataset_uses_boundary(self) -> None:
+        from datahub.admin_api import AdminAPI
+        from datahub.auth import AuthService
+        from datahub.crypto import SecretVault
+        from datahub.scheduler import Scheduler
+        from datahub.serving import ApiError
+
+        vault = SecretVault(self.pipe.settings.encryption_key)
+        auth = AuthService(self.db, vault, self.pipe.settings.api_token, "StartPass1")
+        admin = AdminAPI(self.db, self.pipe, Scheduler(self.db, self.pipe), auth)
+        before = publications_map(self.db, TRADE_DATE)
+        result = admin.backfill("moneyflow", TRADE_DATE, "StartPass1", f"moneyflow:{TRADE_DATE}", "tester")
+        self.assertEqual(result["moneyflow"]["state"], "published")
+        after = publications_map(self.db, TRADE_DATE)
+        for name in (*GROUP_A, "stocks"):
+            self.assertNotEqual(after[name], before[name], name)
+        # bad password / wrong confirm still rejected
+        with self.assertRaises(ApiError):
+            admin.backfill("daily", TRADE_DATE, "wrong", f"daily:{TRADE_DATE}", "tester")
+
+    def test_admin_backfill_switch_crash_is_failed_precondition(self) -> None:
+        from datahub.admin_api import AdminAPI
+        from datahub.auth import AuthService
+        from datahub.crypto import SecretVault
+        from datahub.scheduler import Scheduler
+        from datahub.serving import ApiError
+
+        vault = SecretVault(self.pipe.settings.encryption_key)
+        auth = AuthService(self.db, vault, self.pipe.settings.api_token, "StartPass1")
+        admin = AdminAPI(self.db, self.pipe, Scheduler(self.db, self.pipe), auth)
+        before = publications_map(self.db, TRADE_DATE)
+
+        def explode() -> None:
+            raise RuntimeError("killed mid-switch")
+
+        self.pipe.before_commit = explode
+        with self.assertRaises(ApiError) as ctx:
+            admin.backfill("valuation", TRADE_DATE, "StartPass1", f"valuation:{TRADE_DATE}", "tester")
+        self.assertEqual(ctx.exception.code, "FAILED_PRECONDITION")
+        self.assertIn("killed mid-switch", ctx.exception.message)
+        # previous complete A/B versions keep serving
+        self.assertEqual(publications_map(self.db, TRADE_DATE), before)
+        audit = self.db.fetchone(
+            "SELECT * FROM audit_log WHERE action = 'release-group' ORDER BY id DESC"
+        )
+        self.assertIsNotNone(audit)
+        self.assertIn("failed", str(audit["detail"]))
+        self.assertIn("killed mid-switch", str(audit["detail"]))
+
+
+if __name__ == "__main__":
+    unittest.main()

xiaobai-datahub/tests/test_quality_gates.py

@@ -228,25 +228,34 @@ class GateRetryInterplayTests(unittest.TestCase):
 
 
 class ForceRepublishTests(unittest.TestCase):
-    def test_run_dataset_over_published_keeps_prev_for_rollback(self) -> None:
+    def test_force_boundary_republish_keeps_prev_for_rollback(self) -> None:
         transport = ValuationTransport()
         pipe, db = make_pipe(transport)
         pipe.ingest_reference(TRADE_DATE)
-        first = pipe.run_dataset("valuation", TRADE_DATE)
+        first = pipe.run_eod_batch_a(TRADE_DATE)
+        first_val = first["valuation"]["batch_id"]
+        first_daily = first["daily"]["batch_id"]
         transport.mode = "vr_all_null"
-        with self.assertRaises(QualityError):
+        blocked = pipe.force_republish_boundary("valuation", TRADE_DATE)
-            pipe.run_dataset("valuation", TRADE_DATE)  # gate holds: bad re-publish refused
+        self.assertEqual(blocked["valuation"]["state"], "failed")
+        self.assertEqual(pipe.active_batch("valuation", TRADE_DATE), first_val)
+        self.assertEqual(pipe.active_batch("daily", TRADE_DATE), first_daily)
         transport.mode = "ok"
-        second = pipe.run_dataset("valuation", TRADE_DATE)  # CLI --force path
+        second = pipe.force_republish_boundary("valuation", TRADE_DATE)
-        self.assertNotEqual(first["batch_id"], second["batch_id"])
+        self.assertEqual(second["valuation"]["state"], "published")
-        pub = db.fetchone(
+        self.assertNotEqual(second["valuation"]["batch_id"], first_val)
-            "SELECT * FROM publications WHERE dataset='valuation' AND trade_date=?",
+        self.assertNotEqual(second["daily"]["batch_id"], first_daily)
+        pubs = db.fetchall(
+            "SELECT dataset, active_batch, prev_batch, published_at FROM publications WHERE trade_date=?",
             (TRADE_DATE,),
         )
-        self.assertEqual(pub["active_batch"], second["batch_id"])
+        by_ds = {str(row["dataset"]): row for row in pubs}
-        self.assertEqual(pub["prev_batch"], first["batch_id"])
+        a_times = {by_ds[name]["published_at"] for name in ("daily", "valuation", "moneyflow", "auction", "stocks")}
+        self.assertEqual(len(a_times), 1)
+        self.assertEqual(by_ds["valuation"]["active_batch"], second["valuation"]["batch_id"])
+        self.assertEqual(by_ds["valuation"]["prev_batch"], first_val)
         rolled = pipe.rollback("valuation", TRADE_DATE, actor="cli")
-        self.assertEqual(rolled["active_batch"], first["batch_id"])
+        self.assertEqual(rolled["active_batch"], first_val)
 
 
 if __name__ == "__main__":