在本站中

    tokio::spawn(async {
      let mut ticker = tokio::time::interval(std::time::Duration::from_secs(600));
      loop {
        ticker.tick().await;
        match module_search::engine::reindex_incremental().await {
          Ok(r) => tracing::info!(
            upserts = r.upserts,
            deletes = r.deletes,
            elapsed_ms = r.elapsed_ms,
            "search: index refreshed"
          ),
          Err(e) => tracing::warn!(error = %e, "search: index refresh failed"),
        }
      }
    });

      tokio::spawn(async move {
        let mut ticker = tokio::time::interval(std::time::Duration::from_secs(interval_secs));
        // interval 的首个 tick 立即就绪：先吃掉，避免启动即打网关。
        ticker.tick().await;
        loop {
          ticker.tick().await;
          // 回填/关单明细已在 target=pay_audit 日志留痕。
          let _ = module_course::server::reconcile_pending_orders_once().await;
        }
      });

use tokio::sync::OnceCell;

static POOL: OnceCell<DatabaseConnection> = OnceCell::const_new();

pub async fn init_pool(url: &str) -> Result<(), DbErr> {
  POOL.get_or_try_init(|| async { Database::connect(build_connect_options(url)).await }).await?;
  Ok(())
}

pub async fn get_or_try_init<E, F, Fut>(&self, f: F) -> Result<&T, E>
where
    F: FnOnce() -> Fut,
    Fut: Future<Output = Result<T, E>>,
{
    crate::trace::async_trace_leaf().await;

    if self.initialized() {
        // SAFETY: The OnceCell has been fully initialized.
        unsafe { Ok(self.get_unchecked()) }
    } else {
        // Here we try to acquire the semaphore permit. Holding the permit
        // will allow us to set the value of the OnceCell, and prevents
        // other tasks from initializing the OnceCell while we are holding
        // it.
        match self.semaphore.acquire().await {
            Ok(permit) => {
                debug_assert!(!self.initialized());

                // If `f()` panics or `select!` is called, this
                // `get_or_try_init` call is aborted and the semaphore
                // permit is dropped.
                let value = f().await;

                match value {
                    Ok(value) => Ok(self.set_value(value, permit)),
                    Err(e) => Err(e),
                }
            }
            Err(_) => {
                debug_assert!(self.initialized());

                // SAFETY: The semaphore has been closed. This only happens
                // when the OnceCell is fully initialized.
                unsafe { Ok(self.get_unchecked()) }
            }
        }
    }
}

static POOL: OnceCell<String> = OnceCell::const_new();
static ATTEMPTS: AtomicUsize = AtomicUsize::new(0);

async fn get_pool(caller: usize, start: Instant) -> Result<&'static String, String> {
  POOL
    .get_or_try_init(|| async move {
      let n = ATTEMPTS.fetch_add(1, Ordering::SeqCst) + 1;
      println!("[{:>3}ms] caller {caller}: init attempt {n} starts", start.elapsed().as_millis());
      tokio::time::sleep(Duration::from_millis(50)).await;
      if n == 1 {
        println!("[{:>3}ms] caller {caller}: attempt {n} fails", start.elapsed().as_millis());
        Err(format!("connect refused (attempt {n})"))
      } else {
        Ok(format!("pool#{n}"))
      }
    })
    .await
}

#[tokio::main(flavor = "multi_thread", worker_threads = 4)]
async fn main() {
  let start = Instant::now();
  let handles: Vec<_> = (1..=5)
    .map(|caller| {
      tokio::spawn(async move {
        let r = get_pool(caller, start).await;
        println!("[{:>3}ms] caller {caller}: {r:?}", start.elapsed().as_millis());
      })
    })
    .collect();
  for h in handles {
    h.await.unwrap();
  }
  println!("init attempts in total: {}", ATTEMPTS.load(Ordering::SeqCst));
}

[  0ms] caller 1: init attempt 1 starts
[ 53ms] caller 1: attempt 1 fails
[ 53ms] caller 1: Err("connect refused (attempt 1)")
[ 53ms] caller 3: init attempt 2 starts
[106ms] caller 3: Ok("pool#2")
[106ms] caller 2: Ok("pool#2")
[106ms] caller 4: Ok("pool#2")
[106ms] caller 5: Ok("pool#2")
init attempts in total: 2

static ENGINE: OnceLock<Mutex<Option<Arc<SearchEngine>>>> = OnceLock::new();

static INIT_LOCK: OnceLock<tokio::sync::Mutex<()>> = OnceLock::new();

pub async fn get_or_build() -> Result<Arc<SearchEngine>, String> {
  // 快路径：引擎已就绪，无需加初始化锁。
  {
    let guard = engine_slot().lock().map_err(|e| format!("search lock poisoned: {}", e))?;
    if let Some(e) = guard.as_ref() {
      return Ok(e.clone());
    }
  }

  // 慢路径：串行化初始化。
  let _init = init_lock().lock().await;
  // 双检：等待锁期间可能已被其他调用者完成初始化。
  {
    let guard = engine_slot().lock().map_err(|e| format!("search lock poisoned: {}", e))?;
    if let Some(e) = guard.as_ref() {
      return Ok(e.clone());
    }
  }
  init_or_load(&resolve_index_dir()).await
}