futu_backend/auth/commconfig/

runner.rs

//! CommConfig background transaction actor.

use std::sync::Arc;
use std::time::Duration;

use async_trait::async_trait;
use futu_domain_auth::{
    COMMCONFIG_RETRY_DELAY_MS_LIKE_CPP, CommConfigPageAction, plan_commconfig_page_action_like_cpp,
    plan_commconfig_runtime_failure_like_cpp,
};

use super::accessors::delay_until_next_refresh;
use super::clock::server_now_ts;
use super::fetch_page::{CommConfigPageFetchError, fetch_page};
use super::projection::project_common_config;
use super::snapshot::SharedCommConfig;
use super::store::{CommConfigStoreError, save_last_good};
use super::transaction::{CommConfigTransaction, CommConfigTransactionStep, CommonConfigDocument};
use super::types::CommConfigSource;
use super::wire::decode_commconfig_page;

/// C++ login waits up to 30 seconds for pre-login CommonConfig and allows
/// login to continue when that request cannot be replayed.
///
/// Ref: `FutuOpenD/Src/FTGateway/FTGateway.cpp:288-291` passes `30000` with
/// `bIgnoreWhenUnReplay=true`, and `:798-799` makes pull failure non-fatal.
/// This fixed value is protocol-parity evidence, not server configuration;
/// revise it if the upstream pre-login wait contract changes.
pub const COMMCONFIG_PRE_LOGIN_TIMEOUT: Duration = Duration::from_secs(30);

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum CommConfigPersistence {
    Persisted,
    MemoryOnly,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum CommConfigBootstrapOutcome {
    Published { persistence: CommConfigPersistence },
    Dropped,
    TimedOut,
}

#[async_trait]
pub(super) trait CommConfigPageSource: Send + Sync {
    async fn fetch_page(
        &self,
        begin_id: i32,
    ) -> Result<serde_json::Value, CommConfigPageFetchError>;
}

struct ReqwestCommConfigPageSource {
    http: reqwest::Client,
    client_type: u8,
    device_id: String,
    user_id: u64,
    svr_time_offset: i64,
}

#[async_trait]
impl CommConfigPageSource for ReqwestCommConfigPageSource {
    async fn fetch_page(
        &self,
        begin_id: i32,
    ) -> Result<serde_json::Value, CommConfigPageFetchError> {
        fetch_page(
            &self.http,
            self.client_type,
            &self.device_id,
            self.user_id,
            begin_id,
            self.svr_time_offset,
        )
        .await
    }
}

#[derive(Debug, PartialEq)]
pub(super) enum CommConfigRunOutcome {
    Dropped,
    Committed {
        document: CommonConfigDocument,
        next_refresh_ts: i64,
    },
}

pub(super) async fn run_transaction<S>(source: &S, svr_time_offset: i64) -> CommConfigRunOutcome
where
    S: CommConfigPageSource + ?Sized,
{
    let mut transaction = CommConfigTransaction::new();
    loop {
        let begin_id = transaction.current_begin_id();
        let json = match source.fetch_page(begin_id).await {
            Ok(json) => json,
            Err(error) => {
                let action = plan_commconfig_runtime_failure_like_cpp(error.runtime_failure());
                let CommConfigPageAction::RetrySamePage { delay_ms, reason } = action else {
                    unreachable!("runtime failure policy must retry the same page");
                };
                tracing::warn!(
                    error = %error,
                    begin_id,
                    ?reason,
                    delay_ms,
                    "commconfig: page fetch failed; retrying same page"
                );
                tokio::time::sleep(Duration::from_millis(delay_ms)).await;
                continue;
            }
        };
        let page = decode_commconfig_page(&json);
        let ret_code = page.facts.ret_code;
        let ret_msg = page.ret_msg.clone();
        let action = plan_commconfig_page_action_like_cpp(page.facts);
        let step = transaction.apply_page(page, action, server_now_ts(svr_time_offset));
        match step {
            CommConfigTransactionStep::RetainedForRetry { begin_id } => {
                let CommConfigPageAction::RetrySamePage { delay_ms, reason } = action else {
                    unreachable!("retained transaction requires retry action");
                };
                tracing::warn!(
                    ?ret_code,
                    ?ret_msg,
                    begin_id,
                    ?reason,
                    delay_ms,
                    "commconfig: retryable business response; retrying same page"
                );
                tokio::time::sleep(Duration::from_millis(delay_ms)).await;
            }
            CommConfigTransactionStep::Dropped => {
                tracing::warn!(
                    ?ret_code,
                    ?ret_msg,
                    ?action,
                    "commconfig: dropping incomplete transaction"
                );
                return CommConfigRunOutcome::Dropped;
            }
            CommConfigTransactionStep::FetchNext { .. } => {}
            CommConfigTransactionStep::Committed {
                document,
                next_refresh_ts,
            } => {
                return CommConfigRunOutcome::Committed {
                    document,
                    next_refresh_ts,
                };
            }
        }
    }
}

pub(super) fn publish_live_snapshot(
    shared: &SharedCommConfig,
    document: &CommonConfigDocument,
    next_refresh_ts: i64,
) {
    let generation = shared.load().generation.saturating_add(1);
    let snapshot = project_common_config(
        document,
        next_refresh_ts,
        CommConfigSource::Live,
        generation,
    );
    shared.store(Arc::new(snapshot));
}

fn persist_and_publish_complete_transaction(
    shared: &SharedCommConfig,
    document: &CommonConfigDocument,
    next_refresh_ts: i64,
    persist: impl FnOnce(&CommonConfigDocument) -> Result<(), CommConfigStoreError>,
) -> CommConfigPersistence {
    let persistence = match persist(document) {
        Ok(()) => CommConfigPersistence::Persisted,
        Err(error) => {
            tracing::warn!(
                error = %error,
                "commconfig: failed to persist complete last-good; publishing in memory"
            );
            CommConfigPersistence::MemoryOnly
        }
    };
    publish_live_snapshot(shared, document, next_refresh_ts);
    let current = shared.load();
    tracing::info!(
        generation = current.generation,
        platform_pools = current.guaranteed_ip.len(),
        broker_pools = current.guaranteed_ip_broker.len(),
        web_pools = current.guaranteed_ip_web.len(),
        next_refresh_ts,
        ?persistence,
        "commconfig: fetched complete transaction and published"
    );
    persistence
}

pub(super) async fn bootstrap_from_source_with_persistence<S>(
    shared: &SharedCommConfig,
    source: &S,
    svr_time_offset: i64,
    deadline: Duration,
    persist: impl FnOnce(&CommonConfigDocument) -> Result<(), CommConfigStoreError>,
) -> CommConfigBootstrapOutcome
where
    S: CommConfigPageSource + ?Sized,
{
    match tokio::time::timeout(deadline, run_transaction(source, svr_time_offset)).await {
        Err(_) => {
            tracing::warn!(
                deadline_secs = deadline.as_secs(),
                "commconfig: pre-login transaction timed out; retaining last-good"
            );
            CommConfigBootstrapOutcome::TimedOut
        }
        Ok(CommConfigRunOutcome::Dropped) => CommConfigBootstrapOutcome::Dropped,
        Ok(CommConfigRunOutcome::Committed {
            document,
            next_refresh_ts,
        }) => CommConfigBootstrapOutcome::Published {
            persistence: persist_and_publish_complete_transaction(
                shared,
                &document,
                next_refresh_ts,
                persist,
            ),
        },
    }
}

pub async fn bootstrap_before_login(
    shared: &SharedCommConfig,
    http: reqwest::Client,
    client_type: u8,
    device_id: String,
    user_id: u64,
    svr_time_offset: i64,
) -> CommConfigBootstrapOutcome {
    let source = ReqwestCommConfigPageSource {
        http,
        client_type,
        device_id,
        user_id,
        svr_time_offset,
    };
    bootstrap_from_source_with_persistence(
        shared,
        &source,
        svr_time_offset,
        COMMCONFIG_PRE_LOGIN_TIMEOUT,
        save_last_good,
    )
    .await
}

pub fn spawn_actor(
    shared: SharedCommConfig,
    http: reqwest::Client,
    client_type: u8,
    device_id: String,
    user_id: u64,
    svr_time_offset: i64,
) -> tokio::task::JoinHandle<()> {
    tokio::spawn(async move {
        let source = ReqwestCommConfigPageSource {
            http,
            client_type,
            device_id,
            user_id,
            svr_time_offset,
        };
        tokio::time::sleep(Duration::from_millis(COMMCONFIG_RETRY_DELAY_MS_LIKE_CPP)).await;

        loop {
            let sleep_secs = match run_transaction(&source, svr_time_offset).await {
                CommConfigRunOutcome::Committed {
                    document,
                    next_refresh_ts,
                } => {
                    persist_and_publish_complete_transaction(
                        &shared,
                        &document,
                        next_refresh_ts,
                        save_last_good,
                    );
                    delay_until_next_refresh(next_refresh_ts, server_now_ts(svr_time_offset))
                }
                CommConfigRunOutcome::Dropped => {
                    delay_until_next_refresh(0, server_now_ts(svr_time_offset))
                }
            };
            tokio::time::sleep(Duration::from_secs(sleep_secs)).await;
        }
    })
}