Function DrainQueueForApp

src/common/cloud_work_queue.cpp:584–645

}

bool DrainQueueForApp(uint32_t accountId, uint32_t appId) {
    if (!g_provider) return true;

    std::string prefix = std::to_string(accountId) + "/" + std::to_string(appId) + "/";

    constexpr int POLL_MS = 100;
    constexpr int BASE_TIMEOUT_MS = 30000;
    constexpr int PER_ITEM_MS = 3000;
    constexpr int MAX_TIMEOUT_MS = 180000;

    // Scale timeout with pending work count.
    int pendingCount = 0;
    {
        std::lock_guard<std::mutex> countLock(g_queueMutex);
        for (const auto& item : g_workQueue)
            if (!item.bestEffort && item.cloudPath.rfind(prefix, 0) == 0)
                ++pendingCount;
    }
    int timeoutMs = (std::min)(MAX_TIMEOUT_MS, BASE_TIMEOUT_MS + pendingCount * PER_ITEM_MS);

    auto start = std::chrono::steady_clock::now();
    auto deadline = start + std::chrono::milliseconds(timeoutMs);

    std::unique_lock<std::mutex> lock(g_queueMutex);
    RequeueFailedWorkForPrefixLocked(prefix);
    g_queueCV.notify_all();

    // Best-effort items continue in background.
    bool completed = !HasPendingCommitWorkForPrefixLocked(prefix);
    bool failed = HasFailedCommitWorkForPrefix(prefix);
    if (completed && !failed) {
        LOG("[CloudStorage] DrainQueueForApp: no pending commit work for %s", prefix.c_str());
        return true;
    }

    LOG("[CloudStorage] DrainQueueForApp: waiting for %s", prefix.c_str());
    while (std::chrono::steady_clock::now() < deadline) {
        completed = !HasPendingCommitWorkForPrefixLocked(prefix);
        failed = HasFailedCommitWorkForPrefix(prefix);
        if (completed || failed) break;
        if (g_shuttingDown.load(std::memory_order_seq_cst)) break;
        g_drainCV.wait_for(lock, std::chrono::milliseconds(POLL_MS));
    }

    auto elapsed = std::chrono::duration_cast<std::chrono::milliseconds>(
        std::chrono::steady_clock::now() - start).count();

    if (completed && !failed) {
        LOG("[CloudStorage] DrainQueueForApp: done for %s (%lld ms)",
            prefix.c_str(), (long long)elapsed);
    } else if (failed) {
        LOG("[CloudStorage] DrainQueueForApp: failed work for %s after %lld ms",
            prefix.c_str(), (long long)elapsed);
    } else if (g_shuttingDown.load(std::memory_order_seq_cst)) {
        LOG("[CloudStorage] DrainQueueForApp: shutdown for %s (%lld ms)",
            prefix.c_str(), (long long)elapsed);
    } else {
        LOG("[CloudStorage] DrainQueueForApp: TIMEOUT for %s after %lld ms",