Function HandleBeginBatch

src/common/rpc_handlers.cpp:1422–1493

BeginAppUploadBatch

// BeginAppUploadBatch
RpcResult HandleBeginBatch(uint32_t appId, const std::vector<PB::Field>& reqBody) {
    uint64_t batchId = BatchTracker_NextId();
    uint32_t accountId = 0;
    if (!RequireAccountId("BeginAppUploadBatch", appId, accountId)) {
        // Fail early: error skips CompleteBatch, preventing orphaned local blobs.
        return RpcResult(PB::Writer(), kEResultFail);
    }

    // If a previous batch's cloud publish is still in-flight, wait for it so
    // FetchCloudStateForServe sees the fresh CN. Typically a no-op (single batch)
    // or instant (publish already landed between batches).
    CloudStorage::WaitForPendingPublish(accountId, appId);

    // The CN returned here is what Steam records as synced and what we must publish
    // at CompleteBatch -- they have to match, or Steam re-downloads what it just
    // uploaded. Assign max(local, cloud)+1, strictly above whatever the cloud holds.
    uint64_t localCN = LocalStorage::GetChangeNumber(accountId, appId);
    uint64_t cloudCN = 0;
    if (CloudStorage::IsCloudActive()) {
        // Runs on Steam's BeginAppUploadBatch thread; time it (serve-cached, usually
        // fast, but a cache miss does a live fetch that could stall the main loop).
        auto tbb = std::chrono::steady_clock::now();
        auto cloud = CloudStorage::FetchCloudStateForServe(accountId, appId);
        auto bbMs = std::chrono::duration_cast<std::chrono::milliseconds>(
            std::chrono::steady_clock::now() - tbb).count();
        if (bbMs > 500)
            LOG("[NS] BeginBatch app=%u: FetchCloudStateForServe took %lldms "
                "(on Steam's thread)", appId, (long long)bbMs);
        if (cloud.status == CloudStorage::StateFetchStatus::Ok) {
            cloudCN = cloud.state.cn;
        } else if (cloud.status != CloudStorage::StateFetchStatus::NotFound) {
            // Cloud CN unknown: assigning localCN+1 could regress the cloud CN.
            // Fail the batch; Steam retries once the fetch resolves.
            LOG("[NS] BeginBatch app=%u: cloud CN unverifiable (status=%d) -- failing batch "
                "to avoid regressive CN; Steam will retry", appId,
                static_cast<int>(cloud.status));
            return RpcResult(PB::Writer(), kEResultFail);
        }
    }
    uint64_t assignedCN = (std::max)(localCN, cloudCN) + 1;
    uint64_t appBuildId = 0;
    PrepareBatchCanonicalTokens(accountId, appId);
    PendingOpsJournal::RecordUploadBatchStart(accountId, appId);

    int uploadCount = 0, deleteCount = 0;
    for (auto& f : reqBody) {
        if (f.fieldNum == 3 && f.wireType == PB::LengthDelimited) {
            std::string name(reinterpret_cast<const char*>(f.data), f.dataLen);
            LOG("[NS-BATCH]   upload: %s", SanitizeForLog(name).c_str());
            TryCaptureRootToken(accountId, appId,
                CanonicalizeUploadRootToken(accountId, appId, StripRootToken(name), ExtractRootToken(name)));
            ++uploadCount;
        }
        if (f.fieldNum == 4 && f.wireType == PB::LengthDelimited) {
            std::string name(reinterpret_cast<const char*>(f.data), f.dataLen);
            LOG("[NS-BATCH]   delete: %s", SanitizeForLog(name).c_str());
            TryCaptureRootToken(accountId, appId,
                CanonicalizeUploadRootToken(accountId, appId, StripRootToken(name), ExtractRootToken(name)));