futu_gateway_core/bridge/

dispatcher.rs

//! bridge push dispatcher (v1.4.89 P2-B extracted from bridge/mod.rs)
//!
//! 2 `impl GatewayBridge` 方法：
//! - `create_push_dispatcher`: 把 ApiServer + external_sinks 连起来构造
//!   `PushDispatcher`
//! - `start_push_dispatcher`: spawn push-channel consumer task，
//!   跑 staleness-detector / orphan-scanner / main dispatch loop
//!
//! 本 module 负责 push event 到 client 的分发链路；TradeReQuery 触发的交易数据刷新
//! 会按账户 real/sim 路由到 broker / Platform backend。

use std::{sync::Arc, time::Duration};

use futu_server::listener::ApiServer;
use futu_server::push::PushDispatcher;

use super::{
    GatewayBridge, PushEvent, build_order_fill_update_push, build_order_update_push,
    keep_subscribe_quotes, resolve_push_header, resubscribe_quotes, sync_current_quote_desired_set,
};

#[cfg(test)]
mod tests;
mod trade_requery;
mod trade_requery_worker;

use trade_requery_worker::{
    TRADE_REQUERY_WORKER_QUEUE_CAPACITY, collect_cached_order_backend_ids,
    enqueue_trade_requery_event, handle_trade_requery_worker_event,
    record_place_order_update_push_after_dispatch, spawn_account_asset_refresh_after_trade_notice,
};

// Ref: FutuOpenD/Src/NNProtoCenter/Trade/NNProto_Trd_OnPush.cpp:125-136
// handles NOTICE_TYPE_ORDER_NTF as the direct-order fast path.
const TRADE_NOTICE_ORDER_NTF: u32 = 100;
// Ref: FutuOpenD/Src/NNProtoCenter/Trade/NNProto_Trd_OnPush.cpp:162-175
// handles NOTICE_TYPE_ORDER_FILL_NTF as the direct-fill fast path.
const TRADE_NOTICE_ORDER_FILL_UPDATE: u32 = 6;
impl GatewayBridge {
    /// 创建推送分发器
    pub fn create_push_dispatcher(
        &self,
        server: &ApiServer,
        external_sinks: Vec<Arc<dyn futu_server::push::ExternalPushSink>>,
    ) -> PushDispatcher {
        let mut dispatcher = PushDispatcher::new(
            Arc::clone(server.connections()),
            self.subscription_runtime().manager(),
        )
        .with_metrics(Arc::clone(self.push_runtime().metrics()));
        for sink in external_sinks {
            dispatcher = dispatcher.with_external_sink(sink);
        }
        dispatcher
    }

    /// 启动推送分发任务: 消费 push_rx channel，处理行情推送和交易重查
    ///
    /// `external_sinks`: 外部推送接收器列表 (REST WebSocket, gRPC 等)
    pub fn start_push_dispatcher(
        &self,
        server: &ApiServer,
        mut push_rx: tokio::sync::mpsc::Receiver<PushEvent>,
        external_sinks: Vec<Arc<dyn futu_server::push::ExternalPushSink>>,
    ) {
        let dispatcher = Arc::new(self.create_push_dispatcher(server, external_sinks));
        let caches = self.caches();
        let broker_runtime_for_dispatch = self.broker_runtime().clone();
        let trd_cache = Arc::clone(&caches.trd_cache);
        // v1.4.83 §9 F4: push_health 供 dispatcher 检查 circuit breaker
        let push_health_for_dispatch = Arc::clone(self.push_runtime().push_health());
        let background_task_tracker_for_dispatch = self.background_runtime().tracker();
        let (trade_requery_tx, mut trade_requery_rx) =
            tokio::sync::mpsc::channel::<PushEvent>(TRADE_REQUERY_WORKER_QUEUE_CAPACITY);
        {
            let dispatcher_for_trade_requery = Arc::clone(&dispatcher);
            let broker_runtime_for_trade_requery = broker_runtime_for_dispatch.clone();
            let trd_cache_for_trade_requery = Arc::clone(&trd_cache);
            let push_health_for_trade_requery = Arc::clone(&push_health_for_dispatch);
            self.background_runtime().track(tokio::spawn(async move {
                tracing::info!(
                    capacity = TRADE_REQUERY_WORKER_QUEUE_CAPACITY,
                    "trade requery worker task started"
                );
                while let Some(event) = trade_requery_rx.recv().await {
                    handle_trade_requery_worker_event(
                        Arc::clone(&dispatcher_for_trade_requery),
                        broker_runtime_for_trade_requery.clone(),
                        Arc::clone(&trd_cache_for_trade_requery),
                        Arc::clone(&push_health_for_trade_requery),
                        event,
                    )
                    .await;
                }
                tracing::warn!("trade requery worker task ended");
            }));
        }
        // v1.4.83 §9 F3: push-stream staleness detector → auto re-subscribe
        {
            let push_health_for_staleness = Arc::clone(self.push_runtime().push_health());
            let subscriptions_for_staleness = self.subscription_runtime().manager();
            let static_cache_for_staleness = Arc::clone(&caches.static_cache);
            let qot_right_cache_for_staleness = Arc::clone(&caches.qot_right_cache);
            let broker_runtime_for_staleness = self.broker_runtime().clone();
            // v1.4.106 codex 0631 F5: dual counter — staleness detector 触发的
            // resubscribe 也走 metrics_for_staleness.
            let metrics_for_staleness = Arc::clone(self.push_runtime().metrics());
            self.background_runtime().track(tokio::spawn(async move {
                tracing::info!(
                    "v1.4.83 §9 F3/F4: push-stream staleness detector started \
                     (interval=30s, stale_threshold=60s, F4 trip if re-sub fails within 60s)"
                );
                let mut interval = tokio::time::interval(std::time::Duration::from_secs(30));
                // skip first immediate tick
                interval.tick().await;
                // F4 state: 最近一次 F3 re-subscribe 触发的 Unix ms
                // 0 = 从未触发过; 非 0 = 记录上次触发时间
                let mut last_resub_at_ms: i64 = 0;
                loop {
                    interval.tick().await;
                    let snap = push_health_for_staleness.snapshot();
                    // 未启动 / 尚无 push → 跳过 (还没订阅过不是 staleness)
                    if snap.last_push_received_at_ms == 0 {
                        continue;
                    }
                    let now_ms = crate::bridge::clock::wall_clock_since_unix_epoch_or_zero(
                        "push staleness detector",
                    )
                    .as_millis() as i64;
                    let stale_ms = now_ms.saturating_sub(snap.last_push_received_at_ms);
                    if stale_ms < 60_000 {
                        // 健康: 清 F4 resub 触发跟踪
                        last_resub_at_ms = 0;
                        continue;
                    }
                    // v1.4.106 codex 0932 F5 [P2]: 检查 quote + trade 两类活跃订阅.
                    //
                    // 旧实装只看 qot_conn_ids → trade-only subscriber 的场景下 trade
                    // push stale (CMD 4716/14716 通道异常) 被 silent 忽略.
                    // 修法: 拆 quote vs trade 两套语义 — quote stale 触发 re-subscribe;
                    // trade stale 走 warn-only (没有等价 trade 重订阅协议, trade push
                    // 是 backend 无条件下发, daemon 端不主动订阅).
                    let qot_conn_ids = subscriptions_for_staleness.get_all_qot_conn_ids();
                    let trd_conn_ids = subscriptions_for_staleness.get_all_trd_conn_ids();
                    if qot_conn_ids.is_empty() && trd_conn_ids.is_empty() {
                        // 完全无订阅者, stale 是自然的
                        continue;
                    }
                    // v1.4.83 §9 F4: 若前次 re-sub <60s 但 push 仍 stale → 真故障
                    if last_resub_at_ms != 0 && now_ms.saturating_sub(last_resub_at_ms) < 60_000 {
                        push_health_for_staleness.trip_circuit();
                        tracing::error!(
                            stale_ms,
                            since_last_resub_ms = now_ms - last_resub_at_ms,
                            qot_conns = qot_conn_ids.len(),
                            trd_conns = trd_conn_ids.len(),
                            "v1.4.83 §9 F4: push stream still stale after recent re-subscribe, \
                             circuit tripped (30s cooldown, dispatcher skips)"
                        );
                        continue;
                    }
                    // v1.4.106 F5: trade-only stale (没有 qot subscriber, 但有 trade
                    // subscriber) → trade push 通道可能异常. 没有"重订阅"动作可做
                    // (trade push 是 backend 无条件下发), 走 warn 让 ops 注意 + 通过
                    // F4 staleness 累积自然 trip circuit (回 backend / 重连).
                    if qot_conn_ids.is_empty() {
                        tracing::warn!(
                            stale_ms,
                            trd_conns = trd_conn_ids.len(),
                            "v1.4.106 audit 0932 F5: trade-only push stream stale >60s — \
                             no trade re-subscribe protocol exists; recommend manual reconnect \
                             if persists (will accumulate to F4 circuit trip on next cycle)"
                        );
                        // 走 stale watermark, 不调 resubscribe_quotes (那只 cover qot)
                        continue;
                    }
                    // qot 有 subscriber → 触发 re-subscribe (resubscribe_quotes 走 qot)
                    let Some(be) = broker_runtime_for_staleness.platform_backend() else {
                        tracing::warn!(
                            "v1.4.83 §9 F3: stale detected but backend unavailable, \
                             retry next cycle"
                        );
                        continue;
                    };
                    push_health_for_staleness.record_resubscribe();
                    last_resub_at_ms = now_ms;
                    let outcome = resubscribe_quotes(
                        &be,
                        &subscriptions_for_staleness,
                        &static_cache_for_staleness,
                        &qot_right_cache_for_staleness.get(),
                    )
                    .await;
                    // v1.4.106 codex 0631 F4: partial 时调 record_resubscribe_partial
                    // 让 PushHealth 看到 degraded 状态.
                    if outcome.is_partial {
                        push_health_for_staleness
                            .record_resubscribe_partial(outcome.missed_keys.len() as u32);
                    }
                    // v1.4.106 codex 0631 F5: dual counter —
                    // resubscribe_attempts 触发数 (本 staleness loop 触发了 1 次).
                    // resubscribe_applied_keys 真应用 keys 数.
                    metrics_for_staleness
                        .resubscribe_attempts
                        .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
                    metrics_for_staleness.resubscribe_applied_keys.fetch_add(
                        outcome.applied_keys as u64,
                        std::sync::atomic::Ordering::Relaxed,
                    );
                    // legacy resubscribe_ops_total 仍 bump (向后兼容 v1.4.106 前
                    // 的监控 dashboard).
                    metrics_for_staleness.resubscribe_ops.fetch_add(
                        outcome.applied_keys as u64,
                        std::sync::atomic::Ordering::Relaxed,
                    );
                    tracing::warn!(
                        stale_ms,
                        attempted = outcome.attempted_keys,
                        applied = outcome.applied_keys,
                        missed = outcome.missed_keys.len(),
                        is_partial = outcome.is_partial,
                        active_qot_conns = qot_conn_ids.len(),
                        active_trd_conns = trd_conn_ids.len(),
                        "v1.4.83 §9 F3 + v1.4.106 audit 0631 F4: push stream stale \
                         >60s with active subscribers — auto re-subscribed quotes"
                    );
                }
            }));
        }
        // v1.4.112: C++ MktQotSub periodically replays the current subscribe set
        // with timer_sub=1 every 2 minutes. This is independent from push-stream
        // staleness: mixed QUOTE+ORDER_BOOK can keep global push health fresh while
        // backend orderbook status still needs keep-sub maintenance.
        {
            let subscriptions_for_keep_sub = self.subscription_runtime().manager();
            let static_cache_for_keep_sub = Arc::clone(&caches.static_cache);
            let qot_right_cache_for_keep_sub = Arc::clone(&caches.qot_right_cache);
            let broker_runtime_for_keep_sub = self.broker_runtime().clone();
            self.background_runtime().track(tokio::spawn(async move {
                tracing::info!(
                    "quote keep-sub status task started \
                     (interval=120s, C++ MktQotSub KeepSubStatusPause_S)"
                );
                let mut interval = tokio::time::interval(Duration::from_secs(120));
                // Avoid catch-up bursts if a large multi-market keep-sub replay
                // takes longer than the interval.
                interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
                // skip first immediate tick: C++ first keep-sub happens after the
                // last normal subscribe request ages past KeepSubStatusPause_S.
                interval.tick().await;
                loop {
                    interval.tick().await;
                    let active_keys = subscriptions_for_keep_sub.qot_global_desired_keys().len();
                    if active_keys == 0 {
                        continue;
                    }
                    let Some(be) = broker_runtime_for_keep_sub.platform_backend() else {
                        tracing::warn!(
                            active_keys,
                            "v1.4.112: quote keep-sub skipped because platform backend is unavailable"
                        );
                        continue;
                    };
                    let outcome = keep_subscribe_quotes(
                        &be,
                        &subscriptions_for_keep_sub,
                        &static_cache_for_keep_sub,
                        &qot_right_cache_for_keep_sub.get(),
                    )
                    .await;
                    if outcome.is_partial {
                        tracing::warn!(
                            active_keys,
                            attempted = outcome.attempted_keys,
                            applied = outcome.applied_keys,
                            missed = outcome.missed_keys.len(),
                            error = ?outcome.error_reason,
                            "v1.4.112: quote keep-sub completed with partial outcome"
                        );
                    } else {
                        tracing::debug!(
                            active_keys,
                            attempted = outcome.attempted_keys,
                            applied = outcome.applied_keys,
                            "v1.4.112: quote keep-sub refreshed current desired set"
                        );
                    }
                }
            }));
        }
        // QOT 连接断开清理同步任务。
        //
        // C++ `QotSubscribe::ClearConnSubInfo` 对断开的 conn 不会立刻清
        // `m_setSub`：只有满足最短退订窗口后才移除并触发 backend UnSub。
        // Rust 的断线事件在 futu-server 层产生，那里没有 backend 句柄，所以
        // SubscriptionManager 只标记 generation；gateway 在这里把 due cleanup
        // 后的当前 desired set 同步给 CMD6211。
        {
            let subscriptions_for_disconnect_sync = self.subscription_runtime().manager();
            let static_cache_for_disconnect_sync = Arc::clone(&caches.static_cache);
            let qot_right_cache_for_disconnect_sync = Arc::clone(&caches.qot_right_cache);
            let broker_runtime_for_disconnect_sync = self.broker_runtime().clone();
            self.background_runtime().track(tokio::spawn(async move {
                let mut interval = tokio::time::interval(std::time::Duration::from_secs(5));
                interval.tick().await;
                let mut last_synced_generation =
                    subscriptions_for_disconnect_sync.qot_disconnect_sync_generation();
                loop {
                    interval.tick().await;
                    let due = subscriptions_for_disconnect_sync.cleanup_due_disconnected_qot();
                    if !due.is_empty() {
                        tracing::info!(
                            due = ?due,
                            "v1.4.107: QOT disconnected conn cleanup became due; syncing backend desired set"
                        );
                    }

                    let generation =
                        subscriptions_for_disconnect_sync.qot_disconnect_sync_generation();
                    if generation == last_synced_generation {
                        continue;
                    }

                    let Some(be) = broker_runtime_for_disconnect_sync.platform_backend() else {
                        tracing::warn!(
                            generation,
                            last_synced_generation,
                            "v1.4.107: QOT disconnect cleanup pending but backend unavailable; retry next tick"
                        );
                        continue;
                    };

                    let outcome = sync_current_quote_desired_set(
                        &be,
                        &subscriptions_for_disconnect_sync,
                        &static_cache_for_disconnect_sync,
                        &qot_right_cache_for_disconnect_sync.get(),
                        &due,
                    )
                    .await;
                    if outcome.is_partial || outcome.error_reason.is_some() {
                        tracing::warn!(
                            generation,
                            error = ?outcome.error_reason,
                            missed = ?outcome.missed_keys,
                            "v1.4.107: QOT disconnect cleanup desired-set sync did not fully apply; retry next tick"
                        );
                        continue;
                    }
                    last_synced_generation = generation;
                }
            }));
        }
        // v1.4.83 §9 F6: spawn orphan order 定期扫描任务 (独立 tokio task,
        // 不阻塞 push dispatch 主循环).
        {
            let trd_cache_for_scan = Arc::clone(&caches.trd_cache);
            let push_health_for_scan = Arc::clone(self.push_runtime().push_health());
            self.background_runtime().track(tokio::spawn(async move {
                tracing::info!(
                    "v1.4.83 §9 F6: orphan-order scan task started (interval=30s, threshold=300s)"
                );
                let mut interval = tokio::time::interval(std::time::Duration::from_secs(30));
                // skip first immediate tick (task 刚 spawn 还没订单可扫)
                interval.tick().await;
                loop {
                    interval.tick().await;
                    let now_secs = crate::bridge::clock::wall_clock_since_unix_epoch_or_zero(
                        "orphan-order scan",
                    )
                    .as_secs_f64();
                    let orphans = trd_cache_for_scan.scan_orphan_orders(now_secs, 300.0);
                    push_health_for_scan.record_orphan_scan(orphans.len() as u64);
                    if !orphans.is_empty() {
                        // 每个 orphan 单独 warn log, 方便 ops grep
                        for o in &orphans {
                            tracing::warn!(
                                acc_id = o.acc_id,
                                order_id = o.order_id,
                                order_id_ex = %o.order_id_ex,
                                code = %o.code,
                                age_secs = o.age_secs,
                                "v1.4.83 §9 F6 (v1.4.103 BUG-WUZONG-001 expanded): \
                                 orphan order detected (status in 0/1/2/4 stub stuck, \
                                 create_timestamp > 5min 前 — push 通道可能断流 / order 卡住)"
                            );
                        }
                    }
                }
            }));
        }
        self.background_runtime().track(tokio::spawn(async move {
            tracing::info!("push dispatcher task started");
            while let Some(event) = push_rx.recv().await {
                // v1.4.83 §9 F4: circuit breaker — tripped 且 cooldown 内
                // (30s) 跳过 dispatch 避免持续错误堆积. cooldown 过了
                // `should_skip_dispatch` 内部 auto-reset 返 false.
                if push_health_for_dispatch.should_skip_dispatch() {
                    tracing::warn!(
                        "v1.4.83 §9 F4: circuit tripped — skipping push event during cooldown"
                    );
                    continue;
                }
                match event {
                    PushEvent::QuotePush {
                        sec_key,
                        sub_type,
                        rehab_type,
                        proto_id,
                        body,
                    } => {
                        push_health_for_dispatch.record_qot_push_sub_type(sub_type);
                        // v1.4.106 codex 1131 F4 [P1]: rehab_type plumb 到
                        // PushDispatcher.push_qot — 内部用 (sec_key, sub_type,
                        // rehab_type) 三元 key 查 push_regs 决定路由 conn 集合.
                        dispatcher
                            .push_qot(&sec_key, sub_type, rehab_type, proto_id, body)
                            .await;
                    }
                    PushEvent::QuotePushToConn {
                        conn_id,
                        proto_id,
                        body,
                    } => {
                        dispatcher.push_qot_to_conn(conn_id, proto_id, body).await;
                    }
                    event @ (PushEvent::TradeReQuery { .. }
                    | PushEvent::CryptoOrderReQuery { .. }
                    | PushEvent::CryptoAssetReQuery { .. }) => {
                        enqueue_trade_requery_event(
                            &trade_requery_tx,
                            &push_health_for_dispatch,
                            event,
                        );
                    }
                    PushEvent::TradeOrderDirect { acc_id, orders } => {
                        // C++ `NOTICE_TYPE_ORDER_NTF` 高速通道已携带完整订单内容:
                        // `NNProto_Trd_OnPush.cpp:125-136` 直接 UnPackOrderList +
                        // UpdateAndNotifyOneOrder，不再发 4707/14707 或 4708/14708 重查。
                        let Some(acc) = trd_cache.lookup_account(acc_id) else {
                            tracing::warn!(
                                acc_id,
                                count = orders.len(),
                                "skip direct order notification — account cache miss, cannot derive C++ trd_env"
                            );
                            continue;
                        };
                        let trd_env = acc.trd_env;
                        let mut cached_orders = Vec::with_capacity(orders.len());
                        for raw_order in &orders {
                            match futu_backend::trade_query::order_proto_to_cached_like_cpp(
                                trd_env, raw_order,
                            ) {
                                Some(order) => {
                                    if trd_cache.upsert_order(acc_id, order.clone()) {
                                        cached_orders.push(order);
                                    } else {
                                        tracing::debug!(
                                            acc_id,
                                            order_id_ex = %order.order_id_ex,
                                            order_version = order.order_version,
                                            "skip direct order notification row — cache has same/newer order state"
                                        );
                                    }
                                }
                                None => {
                                    tracing::warn!(
                                        acc_id,
                                        order_id = ?raw_order.order_id,
                                        "skip direct order notification row — C++ UnPackOrderItem required fields missing or market invalid"
                                    );
                                }
                            }
                        }
                        let order_ids_for_clear = collect_cached_order_backend_ids(&cached_orders);
                        if !order_ids_for_clear.is_empty() {
                            let cleared = trd_cache
                                .clear_pending_confirm_for_orders(acc_id, &order_ids_for_clear);
                            if cleared > 0 {
                                tracing::info!(
                                    acc_id,
                                    cleared,
                                    order_ids_count = order_ids_for_clear.len(),
                                    "direct order notify → cleared pending broker confirm flags"
                                );
                            }
                        }
                        for order in &cached_orders {
                            let Some((trd_env, trd_market)) =
                                resolve_push_header(&trd_cache, acc_id, order.trd_market)
                            else {
                                tracing::warn!(
                                    acc_id,
                                    order_id_ex = %order.order_id_ex,
                                    order_market = ?order.trd_market,
                                    "skip direct Trd_UpdateOrder push — account cache miss"
                                );
                                continue;
                            };
                            let push_body =
                                build_order_update_push(acc_id, trd_env, trd_market, order);
                            dispatcher
                                .push_trd_acc(
                                    acc_id,
                                    futu_core::proto_id::TRD_UPDATE_ORDER,
                                    push_body,
                                )
                                .await;
                            record_place_order_update_push_after_dispatch(
                                order, trd_env, trd_market,
                            );
                        }
                        tracing::debug!(
                            acc_id,
                            count = cached_orders.len(),
                            notice_type = TRADE_NOTICE_ORDER_NTF,
                            "direct order notification applied"
                        );
                    }
                    PushEvent::TradeOrderCached { acc_id, orders } => {
                        for order in &orders {
                            let Some((trd_env, trd_market)) =
                                resolve_push_header(&trd_cache, acc_id, order.trd_market)
                            else {
                                tracing::warn!(
                                    acc_id,
                                    order_id_ex = %order.order_id_ex,
                                    order_market = ?order.trd_market,
                                    "skip cached Trd_UpdateOrder push — account cache miss"
                                );
                                continue;
                            };
                            let push_body =
                                build_order_update_push(acc_id, trd_env, trd_market, order);
                            dispatcher
                                .push_trd_acc(
                                    acc_id,
                                    futu_core::proto_id::TRD_UPDATE_ORDER,
                                    push_body,
                                )
                                .await;
                            record_place_order_update_push_after_dispatch(
                                order, trd_env, trd_market,
                            );
                        }
                    }
                    PushEvent::TradeFillDirect { acc_id, fills } => {
                        // C++ `NOTICE_TYPE_ORDER_FILL_NTF` 高速通道已携带完整成交内容:
                        // `NNProto_Trd_OnPush.cpp:162-175` 直接 UnPackDealList +
                        // UpdateAndNotifyOneDeal，不再发 4710/14710 重查。
                        let cleared = trd_cache.clear_pending_confirm_for_acc(acc_id);
                        if cleared > 0 {
                            tracing::info!(
                                acc_id,
                                cleared,
                                "direct fill notify → cleared pending broker confirm flags"
                            );
                        }
                        for fill in &fills {
                            let Some((trd_env, trd_market)) =
                                resolve_push_header(&trd_cache, acc_id, fill.trd_market)
                            else {
                                tracing::warn!(
                                    acc_id,
                                    fill_id_ex = %fill.fill_id_ex,
                                    fill_market = ?fill.trd_market,
                                    "skip direct Trd_UpdateOrderFill push — account cache miss"
                                );
                                continue;
                            };
                            let push_body =
                                build_order_fill_update_push(acc_id, trd_env, trd_market, fill);
                            dispatcher
                                .push_trd_acc(
                                    acc_id,
                                    futu_core::proto_id::TRD_UPDATE_ORDER_FILL,
                                    push_body,
                                )
                                .await;
                        }
                        spawn_account_asset_refresh_after_trade_notice(
                            &background_task_tracker_for_dispatch,
                            broker_runtime_for_dispatch.clone(),
                            Arc::clone(&trd_cache),
                            acc_id,
                            TRADE_NOTICE_ORDER_FILL_UPDATE,
                            true,
                        );
                    }
                    PushEvent::BroadcastPush { proto_id, body } => {
                        // 广播推送: 发送给所有已注册推送通知的客户端
                        dispatcher.push_broadcast(proto_id, body).await;
                    }
                }
            }
            tracing::warn!("push dispatcher task ended");
        }));
    }
}