futu_server/
identity.rs

1//! Dynamic daemon startup readiness and identity ownership.
2
3use std::sync::Arc;
4use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
5
6use parking_lot::RwLock;
7
8#[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
9#[serde(rename_all = "snake_case")]
10pub enum StartupState {
11    PendingAuth,
12    Authenticating,
13    Ready,
14    RetryPending,
15    ShuttingDown,
16}
17
18#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
19pub struct IdentitySnapshot {
20    pub generation: u64,
21    pub user_id: Option<u64>,
22    pub attribution: Option<i32>,
23    pub state: StartupState,
24}
25
26#[derive(Debug, Clone, Copy, PartialEq, Eq)]
27pub enum StartupEvent {
28    BeginAuthentication,
29    Authenticated {
30        user_id: u64,
31        attribution: Option<i32>,
32    },
33    AuthFailed,
34    RetryRequired,
35    Shutdown,
36}
37
38#[derive(Debug, Clone, Copy, PartialEq, Eq)]
39pub enum StartupTransitionError {
40    StaleGeneration,
41    InvalidTransition,
42}
43
44#[derive(Clone)]
45pub struct StartupReadiness {
46    inner: Arc<StartupReadinessInner>,
47}
48
49struct StartupReadinessInner {
50    snapshot: RwLock<IdentitySnapshot>,
51    changes: tokio::sync::watch::Sender<IdentitySnapshot>,
52    pending_init_connects: AtomicUsize,
53    pending_init_drained: tokio::sync::Notify,
54    /// GitHub issue #55: 是否曾经进入过 `Ready`。Platform 重连窗口
55    /// (`Ready → RetryPending → Authenticating`) 与首登期共用同一组状态，
56    /// 只有 "曾经 Ready" 的窗口才放行订阅账本 proto（见 [`StartupReadiness::allows_proto`]）。
57    has_been_ready: AtomicBool,
58    /// Reviewer M-8: the user_id that last reached `Ready` (0 = none). The
59    /// reconnect-window admission only applies while the snapshot still carries
60    /// no identity or the same identity; a different account must complete its
61    /// own login first. `Shutdown` clears `has_been_ready`.
62    last_ready_user_id: std::sync::atomic::AtomicU64,
63}
64
65pub struct PendingInitConnectGuard {
66    inner: Arc<StartupReadinessInner>,
67}
68
69impl StartupReadiness {
70    pub fn new_pending() -> Self {
71        Self::from_snapshot(IdentitySnapshot {
72            generation: 0,
73            user_id: None,
74            attribution: None,
75            state: StartupState::PendingAuth,
76        })
77    }
78
79    pub fn ready(user_id: u64, attribution: Option<i32>) -> Self {
80        Self::from_snapshot(IdentitySnapshot {
81            generation: 0,
82            user_id: Some(user_id),
83            attribution,
84            state: StartupState::Ready,
85        })
86    }
87
88    fn from_snapshot(snapshot: IdentitySnapshot) -> Self {
89        let (changes, _) = tokio::sync::watch::channel(snapshot.clone());
90        let has_been_ready = snapshot.state == StartupState::Ready;
91        let last_ready_user_id = snapshot.user_id.filter(|_| has_been_ready).unwrap_or(0);
92        Self {
93            inner: Arc::new(StartupReadinessInner {
94                snapshot: RwLock::new(snapshot),
95                changes,
96                pending_init_connects: AtomicUsize::new(0),
97                pending_init_drained: tokio::sync::Notify::new(),
98                has_been_ready: AtomicBool::new(has_been_ready),
99                last_ready_user_id: std::sync::atomic::AtomicU64::new(last_ready_user_id),
100            }),
101        }
102    }
103
104    /// Reviewer M-8: the reconnect window belongs to the identity that was last
105    /// `Ready`. While reconnecting the snapshot carries no user_id (admit); if a
106    /// different user_id shows up before `Ready`, this is a new login (reject).
107    fn reconnect_window_identity_matches(&self, snapshot: &IdentitySnapshot) -> bool {
108        let last = self.inner.last_ready_user_id.load(Ordering::Acquire);
109        snapshot.user_id.is_none_or(|uid| uid == last)
110    }
111
112    /// 是否曾经进入过 `Ready`（首登成功过）。重连窗口据此区分于首登期。
113    #[must_use]
114    pub fn has_been_ready(&self) -> bool {
115        self.inner.has_been_ready.load(Ordering::Acquire)
116    }
117
118    #[must_use]
119    pub fn snapshot(&self) -> IdentitySnapshot {
120        self.inner.snapshot.read().clone()
121    }
122
123    pub fn subscribe(&self) -> tokio::sync::watch::Receiver<IdentitySnapshot> {
124        self.inner.changes.subscribe()
125    }
126
127    pub async fn await_init_identity(&self) -> Result<IdentitySnapshot, StartupState> {
128        let mut changes = self.subscribe();
129        loop {
130            let snapshot = changes.borrow_and_update().clone();
131            match snapshot.state {
132                StartupState::Ready if snapshot.user_id.is_some() => return Ok(snapshot),
133                StartupState::ShuttingDown => return Err(StartupState::ShuttingDown),
134                StartupState::PendingAuth
135                | StartupState::Authenticating
136                | StartupState::RetryPending
137                | StartupState::Ready => {}
138            }
139            if changes.changed().await.is_err() {
140                return Err(StartupState::ShuttingDown);
141            }
142        }
143    }
144
145    pub async fn await_init_identity_or_cancel<F>(
146        &self,
147        cancelled: F,
148    ) -> Result<Option<IdentitySnapshot>, StartupState>
149    where
150        F: std::future::Future<Output = ()>,
151    {
152        tokio::select! {
153            biased;
154            identity = self.await_init_identity() => identity.map(Some),
155            () = cancelled => Ok(None),
156        }
157    }
158
159    pub(crate) fn track_pending_init_connect(&self) -> PendingInitConnectGuard {
160        self.inner
161            .pending_init_connects
162            .fetch_add(1, Ordering::AcqRel);
163        PendingInitConnectGuard {
164            inner: Arc::clone(&self.inner),
165        }
166    }
167
168    pub async fn await_pending_init_connects_drained(&self) {
169        loop {
170            let drained = self.inner.pending_init_drained.notified();
171            if self.inner.pending_init_connects.load(Ordering::Acquire) == 0 {
172                return;
173            }
174            drained.await;
175        }
176    }
177
178    pub fn pending_init_connect_count(&self) -> usize {
179        self.inner.pending_init_connects.load(Ordering::Acquire)
180    }
181
182    pub fn transition(
183        &self,
184        expected_generation: u64,
185        event: StartupEvent,
186    ) -> Result<IdentitySnapshot, StartupTransitionError> {
187        let mut snapshot = self.inner.snapshot.write();
188        if snapshot.generation != expected_generation {
189            return Err(StartupTransitionError::StaleGeneration);
190        }
191        let (state, user_id, attribution) = match (snapshot.state, event) {
192            (
193                StartupState::PendingAuth | StartupState::RetryPending,
194                StartupEvent::BeginAuthentication,
195            ) => (StartupState::Authenticating, None, snapshot.attribution),
196            (
197                StartupState::Authenticating,
198                StartupEvent::Authenticated {
199                    user_id,
200                    attribution,
201                },
202            ) => (StartupState::Ready, Some(user_id), attribution),
203            (StartupState::Authenticating, StartupEvent::AuthFailed)
204            | (StartupState::Ready, StartupEvent::RetryRequired) => {
205                (StartupState::RetryPending, None, snapshot.attribution)
206            }
207            (
208                StartupState::PendingAuth
209                | StartupState::Authenticating
210                | StartupState::Ready
211                | StartupState::RetryPending,
212                StartupEvent::Shutdown,
213            ) => (
214                StartupState::ShuttingDown,
215                snapshot.user_id,
216                snapshot.attribution,
217            ),
218            _ => return Err(StartupTransitionError::InvalidTransition),
219        };
220        snapshot.generation = snapshot.generation.saturating_add(1);
221        snapshot.state = state;
222        snapshot.user_id = user_id;
223        snapshot.attribution = attribution;
224        if state == StartupState::Ready {
225            if let Some(uid) = user_id {
226                self.inner.last_ready_user_id.store(uid, Ordering::Release);
227            }
228            self.inner.has_been_ready.store(true, Ordering::Release);
229        } else if state == StartupState::ShuttingDown {
230            // A shutting-down daemon never admits subscription-ledger writes.
231            self.inner.has_been_ready.store(false, Ordering::Release);
232        }
233        let published = snapshot.clone();
234        self.inner.changes.send_replace(published.clone());
235        Ok(published)
236    }
237
238    #[must_use]
239    pub fn allows_proto(&self, proto_id: u32) -> bool {
240        let snapshot = self.inner.snapshot.read().clone();
241        match snapshot.state {
242            StartupState::Ready => true,
243            StartupState::PendingAuth
244            | StartupState::Authenticating
245            | StartupState::RetryPending => {
246                Self::is_prelogin_proto(proto_id)
247                    || (self.has_been_ready()
248                        && self.reconnect_window_identity_matches(&snapshot)
249                        && Self::is_subscription_ledger_proto(proto_id))
250            }
251            StartupState::ShuttingDown => false,
252        }
253    }
254
255    /// GitHub issue #55 (BUG-v1.7.3-005): Platform 重连窗口内仍放行的订阅账本
256    /// proto。
257    ///
258    /// C++ 参照：`APIServer/APIServerCS_Core.cpp:108-131 OnRecvClientReq` 对客户端
259    /// 请求只做频控，没有 "gateway 未就绪" gate；`APIServer/APIServer_Qot_Sub.cpp` →
260    /// `NNProtoCenter/Quote/MktQotSub.cpp:98-128 SubOrUnSub` 在 Platform 断连期间
261    /// 只写本地 `m_listRecord` + 起定时器，`Timer_SendSubReq` (`:426-437`) 未登录
262    /// 不发，重连后 `GTWCmdAndPushReply.cpp:342-362 → ReSubAll → MktQotSub::ReSub`
263    /// (`:217-280`) 从账本重放。Rust 的等价排队语义已在 `Qot_Sub` handler 内
264    /// （route=None 仍本地 commit，重连后 `quote_replay` 从账本重建 desired set），
265    /// 因此这三个 proto 在重连窗口不能被 router 挡在门外，否则 SDK
266    /// `OpenQuoteContext.on_api_socket_reconnected → _reconnect_subscribe` 的自动
267    /// 重订阅会丢，新连接账本缺项 → `GetRT` / `GetKL` 本地 IsSub gate 返 "未订阅"。
268    ///
269    /// - `QOT_SUB`(3001)：订阅 / 退订都走同一 proto（`is_sub_or_un_sub`），退订同样
270    ///   只改账本，重连后重放更小的集合。
271    /// - `QOT_REG_QOT_PUSH`(3002)：纯本地 push 注册。
272    /// - `QOT_GET_SUB_INFO`(3003)：只读账本。
273    ///
274    /// 行情拉取 / 交易类 proto 仍被拒（它们需要活的 backend route）；首登期（从未
275    /// Ready）仍拒全部业务 proto，避免没有可重放连接语义时 silent 记账。
276    pub const fn is_subscription_ledger_proto(proto_id: u32) -> bool {
277        matches!(
278            proto_id,
279            futu_core::proto_id::QOT_SUB
280                | futu_core::proto_id::QOT_REG_QOT_PUSH
281                | futu_core::proto_id::QOT_GET_SUB_INFO
282        )
283    }
284
285    pub const fn is_prelogin_proto(proto_id: u32) -> bool {
286        matches!(
287            proto_id,
288            futu_core::proto_id::INIT_CONNECT
289                | futu_core::proto_id::GET_GLOBAL_STATE
290                | futu_core::proto_id::KEEP_ALIVE
291                | futu_core::proto_id::VERIFICATION
292        )
293    }
294}
295
296impl Drop for PendingInitConnectGuard {
297    fn drop(&mut self) {
298        let previous = self
299            .inner
300            .pending_init_connects
301            .fetch_sub(1, Ordering::AcqRel);
302        debug_assert!(previous > 0);
303        if previous <= 1 {
304            self.inner.pending_init_drained.notify_waiters();
305        }
306    }
307}
308
309impl Default for StartupReadiness {
310    fn default() -> Self {
311        Self::ready(0, None)
312    }
313}
314
315#[cfg(test)]
316#[path = "identity/tests.rs"]
317mod tests;
futu_server/identity.rs

futu_server/
identity.rs
