diff options
Diffstat (limited to 'src/zenremotestore/builds/buildstorageutil.cpp')
| -rw-r--r-- | src/zenremotestore/builds/buildstorageutil.cpp | 1792 |
1 files changed, 1537 insertions, 255 deletions
diff --git a/src/zenremotestore/builds/buildstorageutil.cpp b/src/zenremotestore/builds/buildstorageutil.cpp index b249d7d52..dc8f79a47 100644 --- a/src/zenremotestore/builds/buildstorageutil.cpp +++ b/src/zenremotestore/builds/buildstorageutil.cpp @@ -2,293 +2,116 @@ #include <zenremotestore/builds/buildstorageutil.h> +#include <zencore/basicfile.h> +#include <zencore/compactbinary.h> +#include <zencore/compactbinaryutil.h> +#include <zencore/compactbinaryvalue.h> +#include <zencore/except.h> #include <zencore/fmtutils.h> +#include <zencore/logging/broadcastsink.h> +#include <zencore/parallelwork.h> #include <zencore/timer.h> +#include <zencore/trace.h> +#include <zenremotestore/builds/buildcontent.h> +#include <zenremotestore/builds/buildmanifest.h> +#include <zenremotestore/builds/buildprimecache.h> #include <zenremotestore/builds/buildstorage.h> #include <zenremotestore/builds/buildstoragecache.h> +#include <zenremotestore/builds/buildupdatefolder.h> +#include <zenremotestore/builds/builduploadfolder.h> +#include <zenremotestore/builds/buildvalidatebuildpart.h> #include <zenremotestore/builds/jupiterbuildstorage.h> #include <zenremotestore/chunking/chunkblock.h> +#include <zenremotestore/chunking/chunkingcache.h> +#include <zenremotestore/chunking/chunkingcontroller.h> #include <zenremotestore/jupiter/jupiterhost.h> -#include <zenremotestore/operationlogoutput.h> +#include <zenutil/filesystemutils.h> +#include <zenutil/logging.h> +#include <zenutil/progress.h> +#include <zenutil/wildcard.h> #include <zenutil/zenserverprocess.h> -namespace zen { -namespace { - std::string ConnectionSettingsToString(const HttpClientSettings& ClientSettings) - { - ExtendableStringBuilder<128> SB; - SB << "\n LogCategory: " << ClientSettings.LogCategory; - SB << "\n ConnectTimeout: " << ClientSettings.ConnectTimeout.count() << " ms"; - SB << "\n Timeout: " << ClientSettings.Timeout.count() << " ms"; - SB << "\n AccessTokenProvider: " << ClientSettings.AccessTokenProvider.has_value(); - SB << "\n AssumeHttp2: " << ClientSettings.AssumeHttp2; - SB << "\n AllowResume: " << ClientSettings.AllowResume; - SB << "\n RetryCount: " << ClientSettings.RetryCount; - SB << "\n SessionId: " << ClientSettings.SessionId.ToString(); - SB << "\n Verbose: " << ClientSettings.Verbose; - SB << "\n MaximumInMemoryDownloadSize: " << ClientSettings.MaximumInMemoryDownloadSize; - return SB.ToString(); - } -} // namespace - -BuildStorageResolveResult -ResolveBuildStorage(OperationLogOutput& Output, - const HttpClientSettings& ClientSettings, - std::string_view Host, - std::string_view OverrideHost, - std::string_view ZenCacheHost, - ZenCacheResolveMode ZenResolveMode, - bool Verbose) -{ - bool AllowZenCacheDiscovery = ZenResolveMode == ZenCacheResolveMode::Discovery || ZenResolveMode == ZenCacheResolveMode::All; - bool AllowLocalZenCache = ZenResolveMode == ZenCacheResolveMode::LocalHost || ZenResolveMode == ZenCacheResolveMode::All; - - auto GetHostNameFromUrl = [](std::string_view Url) -> std::string_view { - std::string::size_type HostnameStart = 0; - std::string::size_type HostnameLength = std::string::npos; - if (auto StartPos = Url.find("//"); StartPos != std::string::npos) - { - HostnameStart = StartPos + 2; - } - if (auto EndPos = Url.find("/", HostnameStart); EndPos != std::string::npos) - { - HostnameLength = EndPos - HostnameStart; - } - if (auto EndPos = Url.find(":", HostnameStart); EndPos != std::string::npos) - { - HostnameLength = EndPos - HostnameStart; - } - return Url.substr(HostnameStart, HostnameLength); - }; +#include <numeric> - std::string HostUrl; - std::string HostName; - double HostLatencySec = -1.0; +#if ZEN_WITH_TESTS +# include <zencore/testing.h> +# include <zencore/testutils.h> +# include <zenhttp/httpclientauth.h> +# include <zenremotestore/builds/filebuildstorage.h> +#endif // ZEN_WITH_TESTS - std::string CacheUrl; - std::string CacheName; - bool HostAssumeHttp2 = ClientSettings.AssumeHttp2; - bool CacheAssumeHttp2 = ClientSettings.AssumeHttp2; - double CacheLatencySec = -1.0; - - JupiterServerDiscovery DiscoveryResponse; - const std::string_view DiscoveryHost = Host.empty() ? OverrideHost : Host; +namespace zen { - if (OverrideHost.empty() || (ZenCacheHost.empty() && AllowZenCacheDiscovery)) +StorageInstance::~StorageInstance() +{ + if (CacheLogSink) { - if (Verbose) + if (Ref<logging::BroadcastSink> Broadcast = GetDefaultBroadcastSink()) { - ZEN_OPERATION_LOG_INFO(Output, - "Querying servers at '{}/api/v1/status/servers'\n Connection settings:{}", - DiscoveryHost, - ConnectionSettingsToString(ClientSettings)); + Broadcast->RemoveSink(CacheLogSink); } - - DiscoveryResponse = DiscoverJupiterEndpoints(DiscoveryHost, ClientSettings); } +} - if (!OverrideHost.empty()) - { - if (Verbose) - { - ZEN_OPERATION_LOG_INFO(Output, "Testing server endpoint at '{}/health/live'. Assume http2: {}", OverrideHost, HostAssumeHttp2); - } - if (JupiterEndpointTestResult TestResult = TestJupiterEndpoint(OverrideHost, HostAssumeHttp2, ClientSettings.Verbose); - TestResult.Success) - { - if (Verbose) - { - ZEN_OPERATION_LOG_INFO(Output, "Server endpoint at '{}/api/v1/status/servers' succeeded", OverrideHost); - } - HostUrl = OverrideHost; - HostName = GetHostNameFromUrl(OverrideHost); - HostLatencySec = TestResult.LatencySeconds; - } - else - { - throw std::runtime_error(fmt::format("Host {} could not be reached. Reason: {}", OverrideHost, TestResult.FailureReason)); - } - } - else - { - if (DiscoveryResponse.ServerEndPoints.empty()) - { - throw std::runtime_error(fmt::format("Failed to find any builds hosts at {}", DiscoveryHost)); - } - - for (const JupiterServerDiscovery::EndPoint& ServerEndpoint : DiscoveryResponse.ServerEndPoints) - { - if (!ServerEndpoint.BaseUrl.empty()) - { - if (Verbose) - { - ZEN_OPERATION_LOG_INFO(Output, - "Testing server endpoint at '{}/health/live'. Assume http2: {}", - ServerEndpoint.BaseUrl, - ServerEndpoint.AssumeHttp2); - } - - if (JupiterEndpointTestResult TestResult = - TestJupiterEndpoint(ServerEndpoint.BaseUrl, ServerEndpoint.AssumeHttp2, ClientSettings.Verbose); - TestResult.Success) - { - if (Verbose) - { - ZEN_OPERATION_LOG_INFO(Output, "Server endpoint at '{}/api/v1/status/servers' succeeded", ServerEndpoint.BaseUrl); - } - - HostUrl = ServerEndpoint.BaseUrl; - HostAssumeHttp2 = ServerEndpoint.AssumeHttp2; - HostName = ServerEndpoint.Name; - HostLatencySec = TestResult.LatencySeconds; - break; - } - else - { - ZEN_OPERATION_LOG_DEBUG(Output, - "Unable to reach host {}. Reason: {}", - ServerEndpoint.BaseUrl, - TestResult.FailureReason); - } - } - } - if (HostUrl.empty()) - { - throw std::runtime_error(fmt::format("Failed to find any usable builds hosts out of {} using {}", - DiscoveryResponse.ServerEndPoints.size(), - DiscoveryHost)); - } - } - if (ZenCacheHost.empty()) - { - if (AllowZenCacheDiscovery) - { - for (const JupiterServerDiscovery::EndPoint& CacheEndpoint : DiscoveryResponse.CacheEndPoints) - { - if (!CacheEndpoint.BaseUrl.empty()) - { - if (Verbose) - { - ZEN_OPERATION_LOG_INFO(Output, - "Testing cache endpoint at '{}/status/builds'. Assume http2: {}", - CacheEndpoint.BaseUrl, - CacheEndpoint.AssumeHttp2); - } +void +StorageInstance::SetupCacheSession(std::string_view TargetUrl, std::string_view Mode, const Oid& SessionId) +{ + CacheSession = std::make_unique<SessionsServiceClient>(SessionsServiceClient::Options{ + .TargetUrl = std::string(TargetUrl), + .AppName = "zen", + .Mode = std::string(Mode), + .SessionId = SessionId, + }); + CacheSession->Announce(); + CacheLogSink = CacheSession->CreateLogSink(); + GetDefaultBroadcastSink()->AddSink(CacheLogSink); +} - if (ZenCacheEndpointTestResult TestResult = - TestZenCacheEndpoint(CacheEndpoint.BaseUrl, CacheEndpoint.AssumeHttp2, ClientSettings.Verbose); - TestResult.Success) - { - if (Verbose) - { - ZEN_OPERATION_LOG_INFO(Output, "Cache endpoint at '{}/status/builds' succeeded", CacheEndpoint.BaseUrl); - } +using namespace std::literals; - CacheUrl = CacheEndpoint.BaseUrl; - CacheAssumeHttp2 = CacheEndpoint.AssumeHttp2; - CacheName = CacheEndpoint.Name; - CacheLatencySec = TestResult.LatencySeconds; - break; - } - } - } - } - if (CacheUrl.empty() && AllowLocalZenCache) - { - ZenServerState State; - if (State.InitializeReadOnly()) - { - State.Snapshot([&](const ZenServerState::ZenServerEntry& Entry) { - if (CacheUrl.empty()) - { - std::string ZenServerLocalHostUrl = fmt::format("http://127.0.0.1:{}", Entry.EffectiveListenPort.load()); - if (ZenCacheEndpointTestResult TestResult = - TestZenCacheEndpoint(ZenServerLocalHostUrl, /*AssumeHttp2*/ false, ClientSettings.Verbose); - TestResult.Success) - { - CacheUrl = ZenServerLocalHostUrl; - CacheAssumeHttp2 = false; - CacheName = "localhost"; - CacheLatencySec = TestResult.LatencySeconds; - } - } - }); - } - } - } - else +std::vector<ChunkBlockDescription> +ParseBlockMetadatas(std::span<const CbObject> BlockMetadatas) +{ + std::vector<ChunkBlockDescription> UnorderedList; + UnorderedList.reserve(BlockMetadatas.size()); + for (size_t CacheBlockMetadataIndex = 0; CacheBlockMetadataIndex < BlockMetadatas.size(); CacheBlockMetadataIndex++) { - if (Verbose) - { - ZEN_OPERATION_LOG_INFO(Output, "Testing cache endpoint at '{}/status/builds'. Assume http2: {}", ZenCacheHost, false); - } - if (ZenCacheEndpointTestResult TestResult = TestZenCacheEndpoint(ZenCacheHost, /*AssumeHttp2*/ false, ClientSettings.Verbose); - TestResult.Success) + const CbObject& CacheBlockMetadata = BlockMetadatas[CacheBlockMetadataIndex]; + ChunkBlockDescription Description = ParseChunkBlockDescription(CacheBlockMetadata); + if (Description.BlockHash != IoHash::Zero) { - CacheUrl = ZenCacheHost; - CacheName = GetHostNameFromUrl(ZenCacheHost); - CacheLatencySec = TestResult.LatencySeconds; - } - else - { - ZEN_WARN("Unable to reach cache host {}. Reason: {}", ZenCacheHost, TestResult.FailureReason); + UnorderedList.emplace_back(std::move(Description)); } } - - return BuildStorageResolveResult{.HostUrl = HostUrl, - .HostName = HostName, - .HostAssumeHttp2 = HostAssumeHttp2, - .HostLatencySec = HostLatencySec, - - .CacheUrl = CacheUrl, - .CacheName = CacheName, - .CacheAssumeHttp2 = CacheAssumeHttp2, - .CacheLatencySec = CacheLatencySec}; + return UnorderedList; } std::vector<ChunkBlockDescription> -GetBlockDescriptions(OperationLogOutput& Output, +GetBlockDescriptions(LoggerRef InLog, BuildStorageBase& Storage, BuildStorageCache* OptionalCacheStorage, const Oid& BuildId, - const Oid& BuildPartId, std::span<const IoHash> BlockRawHashes, bool AttemptFallback, bool IsQuiet, bool IsVerbose) { using namespace std::literals; - - if (!IsQuiet) - { - ZEN_OPERATION_LOG_INFO(Output, "Fetching metadata for {} blocks", BlockRawHashes.size()); - } - - Stopwatch GetBlockMetadataTimer; + ZEN_SCOPED_LOG(InLog); std::vector<ChunkBlockDescription> UnorderedList; tsl::robin_map<IoHash, size_t, IoHash::Hasher> BlockDescriptionLookup; if (OptionalCacheStorage && !BlockRawHashes.empty()) { std::vector<CbObject> CacheBlockMetadatas = OptionalCacheStorage->GetBlobMetadatas(BuildId, BlockRawHashes); - UnorderedList.reserve(CacheBlockMetadatas.size()); - for (size_t CacheBlockMetadataIndex = 0; CacheBlockMetadataIndex < CacheBlockMetadatas.size(); CacheBlockMetadataIndex++) + if (!CacheBlockMetadatas.empty()) { - const CbObject& CacheBlockMetadata = CacheBlockMetadatas[CacheBlockMetadataIndex]; - ChunkBlockDescription Description = ParseChunkBlockDescription(CacheBlockMetadata); - if (Description.BlockHash == IoHash::Zero) + UnorderedList = ParseBlockMetadatas(CacheBlockMetadatas); + for (size_t DescriptionIndex = 0; DescriptionIndex < UnorderedList.size(); DescriptionIndex++) { - ZEN_OPERATION_LOG_WARN(Output, "Unexpected/invalid block metadata received from remote cache, skipping block"); + const ChunkBlockDescription& Description = UnorderedList[DescriptionIndex]; + BlockDescriptionLookup.insert_or_assign(Description.BlockHash, DescriptionIndex); } - else - { - UnorderedList.emplace_back(std::move(Description)); - } - } - for (size_t DescriptionIndex = 0; DescriptionIndex < UnorderedList.size(); DescriptionIndex++) - { - const ChunkBlockDescription& Description = UnorderedList[DescriptionIndex]; - BlockDescriptionLookup.insert_or_assign(Description.BlockHash, DescriptionIndex); } } @@ -315,7 +138,7 @@ GetBlockDescriptions(OperationLogOutput& Output, if (Description.BlockHash == IoHash::Zero) { - ZEN_OPERATION_LOG_WARN(Output, "Unexpected/invalid block metadata received from remote store, skipping block"); + ZEN_WARN("Unexpected/invalid block metadata received from remote store, skipping block"); } else { @@ -355,15 +178,6 @@ GetBlockDescriptions(OperationLogOutput& Output, } } - if (!IsQuiet) - { - ZEN_OPERATION_LOG_INFO(Output, - "GetBlockMetadata for {} took {}. Found {} blocks", - BuildPartId, - NiceTimeSpanMs(GetBlockMetadataTimer.GetElapsedTimeMs()), - Result.size()); - } - if (Result.size() != BlockRawHashes.size()) { std::string ErrorDescription = @@ -385,7 +199,7 @@ GetBlockDescriptions(OperationLogOutput& Output, } if (AttemptFallback) { - ZEN_OPERATION_LOG_WARN(Output, "{} Attemping fallback options.", ErrorDescription); + ZEN_WARN("{} Attemping fallback options.", ErrorDescription); std::vector<ChunkBlockDescription> AugmentedBlockDescriptions; AugmentedBlockDescriptions.reserve(BlockRawHashes.size()); std::vector<ChunkBlockDescription> FoundBlocks = ParseChunkBlockDescriptionList(Storage.FindBlocks(BuildId, (uint64_t)-1)); @@ -408,7 +222,7 @@ GetBlockDescriptions(OperationLogOutput& Output, { if (!IsQuiet) { - ZEN_OPERATION_LOG_INFO(Output, "Found block {} via context find successfully", BlockHash); + ZEN_INFO("Found block {} via context find successfully", BlockHash); } AugmentedBlockDescriptions.emplace_back(std::move(*ListBlocksIt)); } @@ -453,4 +267,1472 @@ GetBlockDescriptions(OperationLogOutput& Output, return Result; } +////////////////////// Shared helpers + +std::filesystem::path +ZenStateFilePath(const std::filesystem::path& ZenFolderPath) +{ + return ZenFolderPath / "current_state.cbo"; +} +std::filesystem::path +ZenTempFolderPath(const std::filesystem::path& ZenFolderPath) +{ + return ZenFolderPath / "tmp"; +} + +CbObject +GetBuild(BuildStorageBase& Storage, const Oid& BuildId, bool IsQuiet) +{ + Stopwatch GetBuildTimer; + CbObject BuildObject = Storage.GetBuild(BuildId); + if (!IsQuiet) + { + ZEN_CONSOLE("GetBuild took {}. Name: '{}', Payload size: {}", + NiceTimeSpanMs(GetBuildTimer.GetElapsedTimeMs()), + BuildObject["name"sv].AsString(), + NiceBytes(BuildObject.GetSize())); + + ZEN_CONSOLE("{}", GetCbObjectAsNiceString(BuildObject, " "sv, "\n"sv)); + } + return BuildObject; +} + +uint64_t +GetMaxMemoryBufferSize(size_t MaxBlockSize, bool BoostWorkerMemory) +{ + return BoostWorkerMemory ? (MaxBlockSize + 16u * 1024u) : 1024u * 1024u; +} + +void +DownloadLargeBlob(BuildStorageBase& Storage, + const std::filesystem::path& DownloadFolder, + const Oid& BuildId, + const IoHash& ChunkHash, + const std::uint64_t PreferredMultipartChunkSize, + ParallelWork& Work, + WorkerThreadPool& NetworkPool, + std::atomic<uint64_t>& DownloadedChunkByteCount, + std::atomic<uint64_t>& MultipartAttachmentCount, + std::function<void(IoBuffer&& Payload)>&& OnDownloadComplete) +{ + ZEN_TRACE_CPU("DownloadLargeBlob"); + + struct WorkloadData + { + TemporaryFile TempFile; + }; + std::shared_ptr<WorkloadData> Workload(std::make_shared<WorkloadData>()); + + std::error_code Ec; + Workload->TempFile.CreateTemporary(DownloadFolder, Ec); + if (Ec) + { + throw std::runtime_error( + fmt::format("Failed opening temporary file '{}', reason: ({}) {}", Workload->TempFile.GetPath(), Ec.message(), Ec.value())); + } + std::vector<std::function<void()>> WorkItems = Storage.GetLargeBuildBlob( + BuildId, + ChunkHash, + PreferredMultipartChunkSize, + [&Work, Workload, &DownloadedChunkByteCount](uint64_t Offset, const IoBuffer& Chunk) { + DownloadedChunkByteCount += Chunk.GetSize(); + + if (!Work.IsAborted()) + { + ZEN_TRACE_CPU("Async_DownloadLargeBlob_OnReceive"); + Workload->TempFile.Write(Chunk.GetView(), Offset); + } + }, + [&Work, Workload, OnDownloadComplete = std::move(OnDownloadComplete)]() { + if (!Work.IsAborted()) + { + ZEN_TRACE_CPU("Async_DownloadLargeBlob_OnComplete"); + + uint64_t PayloadSize = Workload->TempFile.FileSize(); + void* FileHandle = Workload->TempFile.Detach(); + ZEN_ASSERT(FileHandle != nullptr); + IoBuffer Payload(IoBuffer::File, FileHandle, 0, PayloadSize, true); + Payload.SetDeleteOnClose(true); + OnDownloadComplete(std::move(Payload)); + } + }); + if (!WorkItems.empty()) + { + MultipartAttachmentCount++; + } + for (auto& WorkItem : WorkItems) + { + Work.ScheduleWork(NetworkPool, [WorkItem = std::move(WorkItem)](std::atomic<bool>& AbortFlag) { + if (!AbortFlag) + { + ZEN_TRACE_CPU("Async_DownloadLargeBlob_Work"); + + WorkItem(); + } + }); + } +} + +CompositeBuffer +ValidateBlob(std::atomic<bool>& AbortFlag, + IoBuffer&& Payload, + const IoHash& BlobHash, + uint64_t& OutCompressedSize, + uint64_t& OutDecompressedSize) +{ + ZEN_TRACE_CPU("ValidateBlob"); + + if (Payload.GetContentType() != ZenContentType::kCompressedBinary) + { + throw std::runtime_error(fmt::format("Blob {} ({} bytes) has unexpected content type '{}'", + BlobHash, + Payload.GetSize(), + ToString(Payload.GetContentType()))); + } + IoHash RawHash; + uint64_t RawSize; + CompressedBuffer Compressed = CompressedBuffer::FromCompressed(SharedBuffer(Payload), RawHash, RawSize); + if (!Compressed) + { + throw std::runtime_error(fmt::format("Blob {} ({} bytes) compressed header is invalid", BlobHash, Payload.GetSize())); + } + if (RawHash != BlobHash) + { + throw std::runtime_error( + fmt::format("Blob {} ({} bytes) compressed header has a mismatching raw hash {}", BlobHash, Payload.GetSize(), RawHash)); + } + + IoHashStream Hash; + bool CouldDecompress = Compressed.DecompressToStream( + 0, + RawSize, + [&AbortFlag, &Hash](uint64_t SourceOffset, uint64_t SourceSize, uint64_t Offset, const CompositeBuffer& RangeBuffer) { + ZEN_UNUSED(SourceOffset, SourceSize, Offset); + if (!AbortFlag) + { + for (const SharedBuffer& Segment : RangeBuffer.GetSegments()) + { + Hash.Append(Segment.GetView()); + } + return true; + } + return false; + }); + + if (AbortFlag) + { + return CompositeBuffer{}; + } + + if (!CouldDecompress) + { + throw std::runtime_error( + fmt::format("Blob {} ({} bytes) failed to decompress - header information mismatch", BlobHash, Payload.GetSize())); + } + IoHash ValidateRawHash = Hash.GetHash(); + if (ValidateRawHash != BlobHash) + { + throw std::runtime_error(fmt::format("Blob {} ({} bytes) decompressed hash {} does not match header information", + BlobHash, + Payload.GetSize(), + ValidateRawHash)); + } + OodleCompressor Compressor; + OodleCompressionLevel CompressionLevel; + uint64_t BlockSize; + if (!Compressed.TryGetCompressParameters(Compressor, CompressionLevel, BlockSize)) + { + throw std::runtime_error(fmt::format("Blob {} ({} bytes) failed to get compression details", BlobHash, Payload.GetSize())); + } + OutCompressedSize = Payload.GetSize(); + OutDecompressedSize = RawSize; + if (CompressionLevel == OodleCompressionLevel::None) + { + // Only decompress to composite if we need it for block verification + CompositeBuffer DecompressedComposite = Compressed.DecompressToComposite(); + if (!DecompressedComposite) + { + throw std::runtime_error(fmt::format("Blob {} ({} bytes) failed to decompress to composite", BlobHash, Payload.GetSize())); + } + return DecompressedComposite; + } + return CompositeBuffer{}; +} + +CompositeBuffer +ValidateBlob(std::atomic<bool>& AbortFlag, + BuildStorageBase& Storage, + const Oid& BuildId, + const IoHash& BlobHash, + uint64_t& OutCompressedSize, + uint64_t& OutDecompressedSize) +{ + ZEN_TRACE_CPU("ValidateBlob"); + IoBuffer Payload = Storage.GetBuildBlob(BuildId, BlobHash); + if (!Payload) + { + throw std::runtime_error(fmt::format("Blob {} could not be found", BlobHash)); + } + return ValidateBlob(AbortFlag, std::move(Payload), BlobHash, OutCompressedSize, OutDecompressedSize); +} + +std::vector<std::pair<Oid, std::string>> +ResolveBuildPartNames(CbObjectView BuildObject, + const Oid& BuildId, + const std::vector<Oid>& BuildPartIds, + std::span<const std::string> BuildPartNames, + std::uint64_t& OutPreferredMultipartChunkSize) +{ + std::vector<std::pair<Oid, std::string>> Result; + { + CbObjectView PartsObject = BuildObject["parts"sv].AsObjectView(); + if (!PartsObject) + { + throw std::runtime_error("Build object does not have a 'parts' object"); + } + + OutPreferredMultipartChunkSize = BuildObject["chunkSize"sv].AsUInt64(OutPreferredMultipartChunkSize); + + std::vector<std::pair<Oid, std::string>> AvailableParts; + + for (CbFieldView PartView : PartsObject) + { + const std::string BuildPartName = std::string(PartView.GetName()); + const Oid BuildPartId = PartView.AsObjectId(); + if (BuildPartId == Oid::Zero) + { + ExtendableStringBuilder<128> SB; + for (CbFieldView ScanPartView : PartsObject) + { + SB.Append(fmt::format("\n {}: {}", ScanPartView.GetName(), ScanPartView.AsObjectId())); + } + throw std::runtime_error(fmt::format("Build object parts does not have a '{}' object id{}", BuildPartName, SB.ToView())); + } + AvailableParts.push_back({BuildPartId, BuildPartName}); + } + + if (BuildPartIds.empty() && BuildPartNames.empty()) + { + Result = AvailableParts; + } + else + { + for (const std::string& BuildPartName : BuildPartNames) + { + if (auto It = std::find_if(AvailableParts.begin(), + AvailableParts.end(), + [&BuildPartName](const auto& Part) { return Part.second == BuildPartName; }); + It != AvailableParts.end()) + { + Result.push_back(*It); + } + else + { + throw std::runtime_error(fmt::format("Build {} object does not have a part named '{}'", BuildId, BuildPartName)); + } + } + for (const Oid& BuildPartId : BuildPartIds) + { + if (auto It = std::find_if(AvailableParts.begin(), + AvailableParts.end(), + [&BuildPartId](const auto& Part) { return Part.first == BuildPartId; }); + It != AvailableParts.end()) + { + Result.push_back(*It); + } + else + { + throw std::runtime_error(fmt::format("Build {} object does not have a part with id '{}'", BuildId, BuildPartId)); + } + } + } + + if (Result.empty()) + { + throw std::runtime_error(fmt::format("Build object does not have any parts", BuildId)); + } + } + return Result; +} + +void +NormalizePartSelection(std::vector<Oid>& BuildPartIds, std::vector<std::string>& BuildPartNames, std::string_view HelpText) +{ + const bool HasWildcard = std::find(BuildPartNames.begin(), BuildPartNames.end(), "*") != BuildPartNames.end(); + if (HasWildcard) + { + if (BuildPartNames.size() != 1 || !BuildPartIds.empty()) + { + throw OptionParseException("'*' cannot be combined with other part names or ids", std::string(HelpText)); + } + BuildPartNames.clear(); + return; + } + + if (BuildPartIds.empty() && BuildPartNames.empty()) + { + BuildPartNames.push_back("default"); + } +} + +ChunkedFolderContent +GetRemoteContent(LoggerRef InLog, + StorageInstance& Storage, + const Oid& BuildId, + const std::vector<std::pair<Oid, std::string>>& BuildParts, + const BuildManifest& Manifest, + std::span<const std::string> IncludeWildcards, + std::span<const std::string> ExcludeWildcards, + std::unique_ptr<ChunkingController>& OutChunkController, + std::vector<ChunkedFolderContent>& OutPartContents, + std::vector<ChunkBlockDescription>& OutBlockDescriptions, + std::vector<IoHash>& OutLooseChunkHashes, + bool IsQuiet, + bool IsVerbose, + bool DoExtraContentVerify) +{ + ZEN_TRACE_CPU("GetRemoteContent"); + ZEN_SCOPED_LOG(InLog); + + Stopwatch GetBuildPartTimer; + const Oid BuildPartId = BuildParts[0].first; + const std::string_view BuildPartName = BuildParts[0].second; + CbObject BuildPartManifest = Storage.BuildStorage->GetBuildPart(BuildId, BuildPartId); + if (!IsQuiet) + { + ZEN_INFO("GetBuildPart {} ('{}') took {}. Payload size: {}", + BuildPartId, + BuildPartName, + NiceTimeSpanMs(GetBuildPartTimer.GetElapsedTimeMs()), + NiceBytes(BuildPartManifest.GetSize())); + ZEN_INFO("{}", GetCbObjectAsNiceString(BuildPartManifest, " "sv, "\n"sv)); + } + + { + CbObjectView Chunker = BuildPartManifest["chunker"sv].AsObjectView(); + std::string_view ChunkerName = Chunker["name"sv].AsString(); + CbObjectView Parameters = Chunker["parameters"sv].AsObjectView(); + OutChunkController = CreateChunkingController(ChunkerName, Parameters); + } + + auto ParseBuildPartManifest = [&Log, IsQuiet, IsVerbose, DoExtraContentVerify](StorageInstance& Storage, + const Oid& BuildId, + const Oid& BuildPartId, + CbObject BuildPartManifest, + std::span<const std::string> IncludeWildcards, + std::span<const std::string> ExcludeWildcards, + const BuildManifest::Part* OptionalManifest, + ChunkedFolderContent& OutRemoteContent, + std::vector<ChunkBlockDescription>& OutBlockDescriptions, + std::vector<IoHash>& OutLooseChunkHashes) { + std::vector<uint32_t> AbsoluteChunkOrders; + std::vector<uint64_t> LooseChunkRawSizes; + std::vector<IoHash> BlockRawHashes; + + ReadBuildContentFromCompactBinary(BuildPartManifest, + OutRemoteContent.Platform, + OutRemoteContent.Paths, + OutRemoteContent.RawHashes, + OutRemoteContent.RawSizes, + OutRemoteContent.Attributes, + OutRemoteContent.ChunkedContent.SequenceRawHashes, + OutRemoteContent.ChunkedContent.ChunkCounts, + AbsoluteChunkOrders, + OutLooseChunkHashes, + LooseChunkRawSizes, + BlockRawHashes); + + // TODO: GetBlockDescriptions for all BlockRawHashes in one go - check for local block descriptions when we cache them + + { + if (!IsQuiet) + { + ZEN_INFO("Fetching metadata for {} blocks", BlockRawHashes.size()); + } + + Stopwatch GetBlockMetadataTimer; + + bool AttemptFallback = false; + OutBlockDescriptions = GetBlockDescriptions(Log(), + *Storage.BuildStorage, + Storage.CacheStorage.get(), + BuildId, + BlockRawHashes, + AttemptFallback, + IsQuiet, + IsVerbose); + + if (!IsQuiet) + { + ZEN_INFO("GetBlockMetadata for {} took {}. Found {} blocks", + BuildPartId, + NiceTimeSpanMs(GetBlockMetadataTimer.GetElapsedTimeMs()), + OutBlockDescriptions.size()); + } + } + + CalculateLocalChunkOrders(AbsoluteChunkOrders, + OutLooseChunkHashes, + LooseChunkRawSizes, + OutBlockDescriptions, + OutRemoteContent.ChunkedContent.ChunkHashes, + OutRemoteContent.ChunkedContent.ChunkRawSizes, + OutRemoteContent.ChunkedContent.ChunkOrders, + DoExtraContentVerify); + + std::vector<std::filesystem::path> DeletedPaths; + + if (OptionalManifest) + { + tsl::robin_set<std::string> PathsInManifest; + PathsInManifest.reserve(OptionalManifest->Files.size()); + for (const std::filesystem::path& ManifestPath : OptionalManifest->Files) + { + PathsInManifest.insert(ToLower(ManifestPath.generic_string())); + } + for (const std::filesystem::path& RemotePath : OutRemoteContent.Paths) + { + if (!PathsInManifest.contains(ToLower(RemotePath.generic_string()))) + { + DeletedPaths.push_back(RemotePath); + } + } + } + + if (!IncludeWildcards.empty() || !ExcludeWildcards.empty()) + { + for (const std::filesystem::path& RemotePath : OutRemoteContent.Paths) + { + if (!IncludePath(IncludeWildcards, ExcludeWildcards, ToLower(RemotePath.generic_string()), /*CaseSensitive*/ true)) + { + DeletedPaths.push_back(RemotePath); + } + } + } + + if (!DeletedPaths.empty()) + { + OutRemoteContent = DeletePathsFromChunkedContent(OutRemoteContent, DeletedPaths); + InlineRemoveUnusedHashes(OutLooseChunkHashes, OutRemoteContent.ChunkedContent.ChunkHashes); + } + +#if ZEN_BUILD_DEBUG + ValidateChunkedFolderContent(OutRemoteContent, OutBlockDescriptions, OutLooseChunkHashes, IncludeWildcards, ExcludeWildcards); +#endif // ZEN_BUILD_DEBUG + }; + + auto FindManifest = [&Manifest](const Oid& BuildPartId, std::string_view BuildPartName) -> const BuildManifest::Part* { + if (Manifest.Parts.empty()) + { + return nullptr; + } + if (Manifest.Parts.size() == 1) + { + if (Manifest.Parts[0].PartId == Oid::Zero && Manifest.Parts[0].PartName.empty()) + { + return &Manifest.Parts[0]; + } + } + + auto It = std::find_if(Manifest.Parts.begin(), Manifest.Parts.end(), [BuildPartId, BuildPartName](const BuildManifest::Part& Part) { + if (Part.PartId != Oid::Zero) + { + return Part.PartId == BuildPartId; + } + if (!Part.PartName.empty()) + { + return Part.PartName == BuildPartName; + } + return false; + }); + if (It != Manifest.Parts.end()) + { + return &(*It); + } + return nullptr; + }; + + OutPartContents.resize(1); + ParseBuildPartManifest(Storage, + BuildId, + BuildPartId, + BuildPartManifest, + IncludeWildcards, + ExcludeWildcards, + FindManifest(BuildPartId, BuildPartName), + OutPartContents[0], + OutBlockDescriptions, + OutLooseChunkHashes); + ChunkedFolderContent RemoteContent; + if (BuildParts.size() > 1) + { + std::vector<ChunkBlockDescription> OverlayBlockDescriptions; + std::vector<IoHash> OverlayLooseChunkHashes; + for (size_t PartIndex = 1; PartIndex < BuildParts.size(); PartIndex++) + { + const Oid& OverlayBuildPartId = BuildParts[PartIndex].first; + const std::string& OverlayBuildPartName = BuildParts[PartIndex].second; + Stopwatch GetOverlayBuildPartTimer; + CbObject OverlayBuildPartManifest = Storage.BuildStorage->GetBuildPart(BuildId, OverlayBuildPartId); + if (!IsQuiet) + { + ZEN_INFO("GetBuildPart {} ('{}') took {}. Payload size: {}", + OverlayBuildPartId, + OverlayBuildPartName, + NiceTimeSpanMs(GetOverlayBuildPartTimer.GetElapsedTimeMs()), + NiceBytes(OverlayBuildPartManifest.GetSize())); + } + + ChunkedFolderContent OverlayPartContent; + std::vector<ChunkBlockDescription> OverlayPartBlockDescriptions; + std::vector<IoHash> OverlayPartLooseChunkHashes; + + ParseBuildPartManifest(Storage, + BuildId, + OverlayBuildPartId, + OverlayBuildPartManifest, + IncludeWildcards, + ExcludeWildcards, + FindManifest(OverlayBuildPartId, OverlayBuildPartName), + OverlayPartContent, + OverlayPartBlockDescriptions, + OverlayPartLooseChunkHashes); + OutPartContents.push_back(OverlayPartContent); + OverlayBlockDescriptions.insert(OverlayBlockDescriptions.end(), + OverlayPartBlockDescriptions.begin(), + OverlayPartBlockDescriptions.end()); + OverlayLooseChunkHashes.insert(OverlayLooseChunkHashes.end(), + OverlayPartLooseChunkHashes.begin(), + OverlayPartLooseChunkHashes.end()); + } + + RemoteContent = MergeChunkedFolderContents(OutPartContents[0], std::span<const ChunkedFolderContent>(OutPartContents).subspan(1)); + { + tsl::robin_set<IoHash> AllBlockHashes; + for (const ChunkBlockDescription& Description : OutBlockDescriptions) + { + AllBlockHashes.insert(Description.BlockHash); + } + for (const ChunkBlockDescription& Description : OverlayBlockDescriptions) + { + if (!AllBlockHashes.contains(Description.BlockHash)) + { + AllBlockHashes.insert(Description.BlockHash); + OutBlockDescriptions.push_back(Description); + } + } + } + { + tsl::robin_set<IoHash> AllLooseChunkHashes(OutLooseChunkHashes.begin(), OutLooseChunkHashes.end()); + for (const IoHash& OverlayLooseChunkHash : OverlayLooseChunkHashes) + { + if (!AllLooseChunkHashes.contains(OverlayLooseChunkHash)) + { + AllLooseChunkHashes.insert(OverlayLooseChunkHash); + OutLooseChunkHashes.push_back(OverlayLooseChunkHash); + } + } + } + } + else + { + RemoteContent = OutPartContents[0]; + } + return RemoteContent; +} +std::string +GetCbObjectAsNiceString(CbObjectView Object, std::string_view Prefix, std::string_view Suffix) +{ + ExtendableStringBuilder<512> SB; + std::vector<std::pair<std::string, std::string>> NameStringValuePairs; + for (CbFieldView Field : Object) + { + std::string_view Name = Field.GetName(); + switch (CbValue Accessor = Field.GetValue(); Accessor.GetType()) + { + case CbFieldType::String: + NameStringValuePairs.push_back({std::string(Name), std::string(Accessor.AsString())}); + break; + case CbFieldType::IntegerPositive: + NameStringValuePairs.push_back({std::string(Name), fmt::format("{}", Accessor.AsIntegerPositive())}); + break; + case CbFieldType::IntegerNegative: + NameStringValuePairs.push_back({std::string(Name), fmt::format("{}", Accessor.AsIntegerNegative())}); + break; + case CbFieldType::Float32: + { + const float Value = Accessor.AsFloat32(); + if (std::isfinite(Value)) + { + NameStringValuePairs.push_back({std::string(Name), fmt::format("{:.9g}", Value)}); + } + else + { + NameStringValuePairs.push_back({std::string(Name), "null"}); + } + } + break; + case CbFieldType::Float64: + { + const double Value = Accessor.AsFloat64(); + if (std::isfinite(Value)) + { + NameStringValuePairs.push_back({std::string(Name), fmt::format("{:.17g}", Value)}); + } + else + { + NameStringValuePairs.push_back({std::string(Name), "null"}); + } + } + break; + case CbFieldType::BoolFalse: + NameStringValuePairs.push_back({std::string(Name), "false"}); + break; + case CbFieldType::BoolTrue: + NameStringValuePairs.push_back({std::string(Name), "true"}); + break; + case CbFieldType::Hash: + { + NameStringValuePairs.push_back({std::string(Name), Accessor.AsHash().ToHexString()}); + } + break; + case CbFieldType::Uuid: + { + StringBuilder<Oid::StringLength + 1> Builder; + Accessor.AsUuid().ToString(Builder); + NameStringValuePairs.push_back({std::string(Name), Builder.ToString()}); + } + break; + case CbFieldType::DateTime: + { + ExtendableStringBuilder<64> Builder; + Builder << DateTime(Accessor.AsDateTimeTicks()).ToIso8601(); + NameStringValuePairs.push_back({std::string(Name), Builder.ToString()}); + } + break; + case CbFieldType::TimeSpan: + { + ExtendableStringBuilder<64> Builder; + const TimeSpan Span(Accessor.AsTimeSpanTicks()); + if (Span.GetDays() == 0) + { + Builder << Span.ToString("%h:%m:%s.%n"); + } + else + { + Builder << Span.ToString("%d.%h:%m:%s.%n"); + } + NameStringValuePairs.push_back({std::string(Name), Builder.ToString()}); + break; + } + case CbFieldType::ObjectId: + NameStringValuePairs.push_back({std::string(Name), Accessor.AsObjectId().ToString()}); + break; + } + } + std::string::size_type LongestKey = 0; + for (const std::pair<std::string, std::string>& KeyValue : NameStringValuePairs) + { + LongestKey = Max(KeyValue.first.length(), LongestKey); + } + for (const std::pair<std::string, std::string>& KeyValue : NameStringValuePairs) + { + SB.Append(fmt::format("{}{:<{}}: {}{}", Prefix, KeyValue.first, LongestKey, KeyValue.second, Suffix)); + } + return SB.ToString(); +} + +#if ZEN_WITH_TESTS + +namespace buildstorageoperations_testutils { + struct TestState + { + TestState(const std::filesystem::path& InRootPath) + : RootPath(InRootPath) + , LogOutput(CreateStandardProgress(Log)) + , ChunkController(CreateStandardChunkingController(StandardChunkingControllerSettings{})) + , ChunkCache(CreateMemoryChunkingCache()) + , WorkerPool(2) + , NetworkPool(2) + { + } + + void Initialize() + { + StoragePath = RootPath / "storage"; + TempPath = RootPath / "temp"; + SystemRootDir = RootPath / "sysroot"; + ZenFolderPath = RootPath / ".zen"; + + CreateDirectories(TempPath); + CreateDirectories(StoragePath); + + Storage.BuildStorage = CreateFileBuildStorage(StoragePath, StorageStats, false); + } + + void CreateSourceData(const std::filesystem::path& Source, std::span<const std::string> Paths, std::span<const uint64_t> Sizes) + { + const std::filesystem::path SourcePath = RootPath / Source; + CreateDirectories(SourcePath); + for (size_t FileIndex = 0; FileIndex < Paths.size(); FileIndex++) + { + const std::string& FilePath = Paths[FileIndex]; + const uint64_t FileSize = Sizes[FileIndex]; + IoBuffer FileData = FileSize > 0 ? CreateSemiRandomBlob(FileSize) : IoBuffer{}; + WriteFile(SourcePath / FilePath, FileData); + } + } + + std::vector<std::pair<Oid, std::string>> Upload(const Oid& BuildId, + const Oid& BuildPartId, + const std::string_view BuildPartName, + const std::filesystem::path& Source, + const std::filesystem::path& ManifestPath) + { + const std::filesystem::path SourcePath = RootPath / Source; + CbObject MetaData; + BuildsOperationUploadFolder Upload(Log, + *LogOutput, + Storage, + AbortFlag, + PauseFlag, + WorkerPool, + NetworkPool, + BuildId, + SourcePath, + true, + MetaData, + BuildsOperationUploadFolder::Options{.TempDir = TempPath}); + return Upload.Execute(BuildPartId, BuildPartName, ManifestPath, *ChunkController, *ChunkCache); + } + + void ValidateUpload(const Oid& BuildId, const std::vector<std::pair<Oid, std::string>>& Parts) + { + for (auto Part : Parts) + { + BuildsOperationValidateBuildPart Validate(Log, + *LogOutput, + *Storage.BuildStorage, + AbortFlag, + PauseFlag, + WorkerPool, + NetworkPool, + BuildId, + Part.first, + Part.second, + BuildsOperationValidateBuildPart::Options{.TempFolder = TempPath / "validate"}); + Validate.Execute(); + } + } + + FolderContent Download(const Oid& BuildId, + const Oid& BuildPartId, + const std::string_view BuildPartName, + const std::filesystem::path& Target, + bool Append) + { + const std::filesystem::path TargetPath = RootPath / Target; + + CreateDirectories(TargetPath); + + uint64_t PreferredMultipartChunkSize = 32u * 1024u * 1024u; + CbObject BuildObject = Storage.BuildStorage->GetBuild(BuildId); + std::vector<Oid> PartIds; + if (BuildPartId != Oid::Zero) + { + PartIds.push_back(BuildPartId); + } + std::vector<std::string> PartNames; + if (!BuildPartName.empty()) + { + PartNames.push_back(std::string(BuildPartName)); + } + std::vector<std::pair<Oid, std::string>> AllBuildParts = + ResolveBuildPartNames(BuildObject, BuildId, PartIds, PartNames, PreferredMultipartChunkSize); + + std::vector<ChunkedFolderContent> PartContents; + + std::vector<ChunkBlockDescription> BlockDescriptions; + std::vector<IoHash> LooseChunkHashes; + + ChunkedFolderContent RemoteContent = GetRemoteContent(Log, + Storage, + BuildId, + AllBuildParts, + {}, + {}, + {}, + ChunkController, + PartContents, + BlockDescriptions, + LooseChunkHashes, + /*IsQuiet*/ false, + /*IsVerbose*/ false, + /*DoExtraContentVerify*/ true); + + GetFolderContentStatistics LocalFolderScanStats; + + struct ContentVisitor : public GetDirectoryContentVisitor + { + virtual void AsyncVisitDirectory(const std::filesystem::path& RelativeRoot, DirectoryContent&& Content) + { + RwLock::ExclusiveLockScope _(ExistingPathsLock); + for (const std::filesystem::path& FileName : Content.FileNames) + { + if (RelativeRoot.empty()) + { + ExistingPaths.push_back(FileName); + } + else + { + ExistingPaths.push_back(RelativeRoot / FileName); + } + } + } + + RwLock ExistingPathsLock; + std::vector<std::filesystem::path> ExistingPaths; + } Visitor; + + Latch PendingWorkCount(1); + + GetDirectoryContent(TargetPath, + DirectoryContentFlags::IncludeFiles | DirectoryContentFlags::Recursive, + Visitor, + WorkerPool, + PendingWorkCount); + + PendingWorkCount.CountDown(); + PendingWorkCount.Wait(); + + FolderContent CurrentLocalFolderState = GetValidFolderContent( + WorkerPool, + LocalFolderScanStats, + TargetPath, + Visitor.ExistingPaths, + [](uint64_t PathCount, uint64_t CompletedPathCount) { ZEN_UNUSED(PathCount, CompletedPathCount); }, + 1000, + AbortFlag, + PauseFlag); + + ChunkingStatistics LocalChunkingStats; + ChunkedFolderContent LocalContent = ChunkFolderContent( + LocalChunkingStats, + WorkerPool, + TargetPath, + CurrentLocalFolderState, + *ChunkController, + *ChunkCache, + 1000, + [&](bool IsAborted, bool IsPaused, std::ptrdiff_t) { ZEN_UNUSED(IsAborted, IsPaused); }, + AbortFlag, + PauseFlag); + + if (Append) + { + RemoteContent = ApplyChunkedContentOverlay(LocalContent, RemoteContent, {}, {}); + } + + const ChunkedContentLookup LocalLookup = BuildChunkedContentLookup(LocalContent); + const ChunkedContentLookup RemoteLookup = BuildChunkedContentLookup(RemoteContent); + + BuildsOperationUpdateFolder Download(Log, + *LogOutput, + Storage, + AbortFlag, + PauseFlag, + WorkerPool, + NetworkPool, + BuildId, + TargetPath, + LocalContent, + LocalLookup, + RemoteContent, + RemoteLookup, + BlockDescriptions, + LooseChunkHashes, + BuildsOperationUpdateFolder::Options{.SystemRootDir = SystemRootDir, + .ZenFolderPath = ZenFolderPath, + .ValidateCompletedSequences = true}); + FolderContent ResultingState; + Download.Execute(ResultingState); + + return ResultingState; + } + + void ValidateDownload(std::span<const std::string> Paths, + std::span<const uint64_t> Sizes, + const std::filesystem::path& Source, + const std::filesystem::path& Target, + const FolderContent& DownloadContent) + { + const std::filesystem::path SourcePath = RootPath / Source; + const std::filesystem::path TargetPath = RootPath / Target; + + CHECK_EQ(Paths.size(), DownloadContent.Paths.size()); + tsl::robin_map<std::string, uint64_t> ExpectedSizes; + tsl::robin_map<std::string, IoHash> ExpectedHashes; + for (size_t Index = 0; Index < Paths.size(); Index++) + { + const std::string LookupString = std::filesystem::path(Paths[Index]).generic_string(); + ExpectedSizes.insert_or_assign(LookupString, Sizes[Index]); + std::filesystem::path FilePath = SourcePath / Paths[Index]; + const IoHash SourceHash = IoHash::HashBuffer(IoBufferBuilder::MakeFromFile(FilePath.make_preferred())); + ExpectedHashes.insert_or_assign(LookupString, SourceHash); + } + for (size_t Index = 0; Index < DownloadContent.Paths.size(); Index++) + { + const std::string LookupString = std::filesystem::path(DownloadContent.Paths[Index]).generic_string(); + auto SizeIt = ExpectedSizes.find(LookupString); + CHECK_NE(SizeIt, ExpectedSizes.end()); + CHECK_EQ(SizeIt->second, DownloadContent.RawSizes[Index]); + std::filesystem::path FilePath = TargetPath / DownloadContent.Paths[Index]; + const IoHash DownloadedHash = IoHash::HashBuffer(IoBufferBuilder::MakeFromFile(FilePath.make_preferred())); + auto HashIt = ExpectedHashes.find(LookupString); + CHECK_NE(HashIt, ExpectedHashes.end()); + CHECK_EQ(HashIt->second, DownloadedHash); + } + } + + const std::filesystem::path RootPath; + std::filesystem::path StoragePath; + std::filesystem::path TempPath; + std::filesystem::path SystemRootDir; + std::filesystem::path ZenFolderPath; + + LoggerRef Log = ConsoleLog(); + std::unique_ptr<ProgressBase> LogOutput; + + std::unique_ptr<ChunkingController> ChunkController; + std::unique_ptr<ChunkingCache> ChunkCache; + + StorageInstance Storage; + BuildStorageBase::Statistics StorageStats; + + WorkerThreadPool WorkerPool; + WorkerThreadPool NetworkPool; + + std::atomic<bool> AbortFlag; + std::atomic<bool> PauseFlag; + }; + +} // namespace buildstorageoperations_testutils + +TEST_SUITE_BEGIN("remotestore.buildstorageutil"); + +TEST_CASE("normalizepartselection.empty_defaults_to_default") +{ + std::vector<Oid> Ids; + std::vector<std::string> Names; + NormalizePartSelection(Ids, Names, {}); + CHECK(Ids.empty()); + REQUIRE_EQ(Names.size(), 1u); + CHECK_EQ(Names[0], "default"); +} + +TEST_CASE("normalizepartselection.wildcard_alone_clears_names") +{ + std::vector<Oid> Ids; + std::vector<std::string> Names = {"*"}; + NormalizePartSelection(Ids, Names, {}); + CHECK(Ids.empty()); + CHECK(Names.empty()); +} + +TEST_CASE("normalizepartselection.wildcard_with_other_name_throws") +{ + std::vector<Oid> Ids; + std::vector<std::string> Names = {"*", "foo"}; + CHECK_THROWS_AS(NormalizePartSelection(Ids, Names, {}), OptionParseException); +} + +TEST_CASE("normalizepartselection.wildcard_with_ids_throws") +{ + std::vector<Oid> Ids = {Oid::NewOid()}; + std::vector<std::string> Names = {"*"}; + CHECK_THROWS_AS(NormalizePartSelection(Ids, Names, {}), OptionParseException); +} + +TEST_CASE("normalizepartselection.explicit_name_unchanged") +{ + std::vector<Oid> Ids; + std::vector<std::string> Names = {"foo"}; + NormalizePartSelection(Ids, Names, {}); + CHECK(Ids.empty()); + REQUIRE_EQ(Names.size(), 1u); + CHECK_EQ(Names[0], "foo"); +} + +TEST_CASE("normalizepartselection.ids_only_unchanged") +{ + const Oid Id = Oid::NewOid(); + std::vector<Oid> Ids = {Id}; + std::vector<std::string> Names; + NormalizePartSelection(Ids, Names, {}); + REQUIRE_EQ(Ids.size(), 1u); + CHECK_EQ(Ids[0], Id); + CHECK(Names.empty()); +} + +TEST_CASE("buildstorageoperations.upload.folder") +{ + using namespace buildstorageoperations_testutils; + + FastRandom BaseRandom; + + const size_t FileCount = 11; + + const std::string Paths[FileCount] = {{"file_1"}, + {"file_2.exe"}, + {"file_3.txt"}, + {"dir_1/dir1_file_1.exe"}, + {"dir_1/dir1_file_2.pdb"}, + {"dir_1/dir1_file_3.txt"}, + {"dir_2/dir2_dir1/dir2_dir1_file_1.exe"}, + {"dir_2/dir2_dir1/dir2_dir1_file_2.pdb"}, + {"dir_2/dir2_dir1/dir2_dir1_file_3.dll"}, + {"dir_2/dir2_dir2/dir2_dir2_file_1.txt"}, + {"dir_2/dir2_dir2/dir2_dir2_file_2.json"}}; + const uint64_t Sizes[FileCount] = + {6u * 1024u, 0, 798, 19u * 1024u, 7u * 1024u, 93, 31u * 1024u, 17u * 1024u, 13u * 1024u, 2u * 1024u, 3u * 1024u}; + + ScopedTemporaryDirectory SourceFolder; + TestState State(SourceFolder.Path()); + State.Initialize(); + State.CreateSourceData("source", Paths, Sizes); + + const Oid BuildId = Oid::NewOid(); + const Oid BuildPartId = Oid::NewOid(); + const std::string BuildPartName = "default"; + + auto Result = State.Upload(BuildId, BuildPartId, BuildPartName, "source", {}); + + CHECK_EQ(Result.size(), 1u); + CHECK_EQ(Result[0].first, BuildPartId); + CHECK_EQ(Result[0].second, BuildPartName); + State.ValidateUpload(BuildId, Result); + + FolderContent DownloadContent = State.Download(BuildId, Oid::Zero, {}, "download", /* Append */ false); + CHECK_EQ(DownloadContent.Paths.size(), FileCount); + State.ValidateDownload(Paths, Sizes, "source", "download", DownloadContent); +} + +TEST_CASE("buildstorageoperations.upload.manifest") +{ + using namespace buildstorageoperations_testutils; + + FastRandom BaseRandom; + + const size_t FileCount = 11; + + const std::string Paths[FileCount] = {{"file_1"}, + {"file_2.exe"}, + {"file_3.txt"}, + {"dir_1/dir1_file_1.exe"}, + {"dir_1/dir1_file_2.pdb"}, + {"dir_1/dir1_file_3.txt"}, + {"dir_2/dir2_dir1/dir2_dir1_file_1.exe"}, + {"dir_2/dir2_dir1/dir2_dir1_file_2.pdb"}, + {"dir_2/dir2_dir1/dir2_dir1_file_3.dll"}, + {"dir_2/dir2_dir2/dir2_dir2_file_1.txt"}, + {"dir_2/dir2_dir2/dir2_dir2_file_2.json"}}; + const uint64_t Sizes[FileCount] = + {6u * 1024u, 0, 798, 19u * 1024u, 7u * 1024u, 93, 31u * 1024u, 17u * 1024u, 13u * 1024u, 2u * 1024u, 3u * 1024u}; + + ScopedTemporaryDirectory SourceFolder; + TestState State(SourceFolder.Path()); + State.Initialize(); + State.CreateSourceData("source", Paths, Sizes); + + std::span<const std::string> ManifestFiles(Paths); + ManifestFiles = ManifestFiles.subspan(0, FileCount / 2); + + std::span<const uint64_t> ManifestSizes(Sizes); + ManifestSizes = ManifestSizes.subspan(0, FileCount / 2); + + ExtendableStringBuilder<1024> Manifest; + for (const std::string& FilePath : ManifestFiles) + { + Manifest << FilePath << "\n"; + } + + WriteFile(State.RootPath / "manifest.txt", IoBuffer(IoBuffer::Wrap, Manifest.Data(), Manifest.Size())); + + const Oid BuildId = Oid::NewOid(); + const Oid BuildPartId = Oid::NewOid(); + const std::string BuildPartName = "default"; + + auto Result = State.Upload(BuildId, BuildPartId, BuildPartName, "source", State.RootPath / "manifest.txt"); + + CHECK_EQ(Result.size(), 1u); + CHECK_EQ(Result[0].first, BuildPartId); + CHECK_EQ(Result[0].second, BuildPartName); + State.ValidateUpload(BuildId, Result); + + FolderContent DownloadContent = State.Download(BuildId, Oid::Zero, {}, "download", /* Append */ false); + State.ValidateDownload(ManifestFiles, ManifestSizes, "source", "download", DownloadContent); +} + +TEST_CASE("buildstorageoperations.memorychunkingcache") +{ + using namespace buildstorageoperations_testutils; + + FastRandom BaseRandom; + + const size_t FileCount = 11; + + const std::string Paths[FileCount] = {{"file_1"}, + {"file_2.exe"}, + {"file_3.txt"}, + {"dir_1/dir1_file_1.exe"}, + {"dir_1/dir1_file_2.pdb"}, + {"dir_1/dir1_file_3.txt"}, + {"dir_2/dir2_dir1/dir2_dir1_file_1.exe"}, + {"dir_2/dir2_dir1/dir2_dir1_file_2.pdb"}, + {"dir_2/dir2_dir1/dir2_dir1_file_3.dll"}, + {"dir_2/dir2_dir2/dir2_dir2_file_1.txt"}, + {"dir_2/dir2_dir2/dir2_dir2_file_2.json"}}; + const uint64_t Sizes[FileCount] = + {6u * 1024u, 0, 798, 19u * 1024u, 7u * 1024u, 93, 31u * 1024u, 17u * 1024u, 13u * 1024u, 2u * 1024u, 3u * 1024u}; + + ScopedTemporaryDirectory SourceFolder; + TestState State(SourceFolder.Path()); + State.Initialize(); + State.CreateSourceData("source", Paths, Sizes); + + const Oid BuildId = Oid::NewOid(); + const Oid BuildPartId = Oid::NewOid(); + const std::string BuildPartName = "default"; + + { + const std::filesystem::path SourcePath = SourceFolder.Path() / "source"; + CbObject MetaData; + BuildsOperationUploadFolder Upload(State.Log, + *State.LogOutput, + State.Storage, + State.AbortFlag, + State.PauseFlag, + State.WorkerPool, + State.NetworkPool, + BuildId, + SourcePath, + true, + MetaData, + BuildsOperationUploadFolder::Options{.TempDir = State.TempPath}); + auto Result = Upload.Execute(BuildPartId, BuildPartName, {}, *State.ChunkController, *State.ChunkCache); + + CHECK_EQ(Upload.m_ChunkingStats.FilesStoredInCache.load(), FileCount - 1); // Zero size files are not stored in cache + CHECK_EQ(Upload.m_ChunkingStats.BytesStoredInCache.load(), std::accumulate(&Sizes[0], &Sizes[FileCount], uint64_t(0))); + CHECK(Upload.m_ChunkingStats.ChunksStoredInCache.load() >= FileCount - 1); // Zero size files are not stored in cache + + CHECK_EQ(Result.size(), 1u); + CHECK_EQ(Result[0].first, BuildPartId); + CHECK_EQ(Result[0].second, BuildPartName); + } + + auto Result = State.Upload(BuildId, BuildPartId, BuildPartName, "source", {}); + + const Oid BuildId2 = Oid::NewOid(); + const Oid BuildPartId2 = Oid::NewOid(); + + { + const std::filesystem::path SourcePath = SourceFolder.Path() / "source"; + CbObject MetaData; + BuildsOperationUploadFolder Upload(State.Log, + *State.LogOutput, + State.Storage, + State.AbortFlag, + State.PauseFlag, + State.WorkerPool, + State.NetworkPool, + BuildId2, + SourcePath, + true, + MetaData, + BuildsOperationUploadFolder::Options{.TempDir = State.TempPath}); + Upload.Execute(BuildPartId2, BuildPartName, {}, *State.ChunkController, *State.ChunkCache); + + CHECK_EQ(Upload.m_ChunkingStats.FilesFoundInCache.load(), FileCount - 1); // Zero size files are not stored in cache + CHECK_EQ(Upload.m_ChunkingStats.BytesFoundInCache.load(), std::accumulate(&Sizes[0], &Sizes[FileCount], uint64_t(0))); + CHECK(Upload.m_ChunkingStats.ChunksFoundInCache.load() >= FileCount - 1); // Zero size files are not stored in cache + } + + FolderContent DownloadContent = State.Download(BuildId2, BuildPartId2, {}, "download", /* Append */ false); + State.ValidateDownload(Paths, Sizes, "source", "download", DownloadContent); +} + +TEST_CASE("buildstorageoperations.upload.multipart") +{ + // Disabled since it relies on authentication and specific block being present in cloud storage + if (false) + { + using namespace buildstorageoperations_testutils; + + FastRandom BaseRandom; + + const size_t FileCount = 11; + + const std::string Paths[FileCount] = {{"file_1"}, + {"file_2.exe"}, + {"file_3.txt"}, + {"dir_1/dir1_file_1.exe"}, + {"dir_1/dir1_file_2.pdb"}, + {"dir_1/dir1_file_3.txt"}, + {"dir_2/dir2_dir1/dir2_dir1_file_1.exe"}, + {"dir_2/dir2_dir1/dir2_dir1_file_2.pdb"}, + {"dir_2/dir2_dir1/dir2_dir1_file_3.dll"}, + {"dir_2/dir2_dir2/dir2_dir2_file_1.txt"}, + {"dir_2/dir2_dir2/dir2_dir2_file_2.json"}}; + const uint64_t Sizes[FileCount] = + {6u * 1024u, 0, 798, 19u * 1024u, 7u * 1024u, 93, 31u * 1024u, 17u * 1024u, 13u * 1024u, 2u * 1024u, 3u * 1024u}; + + ScopedTemporaryDirectory SourceFolder; + TestState State(SourceFolder.Path()); + State.Initialize(); + State.CreateSourceData("source", Paths, Sizes); + + std::span<const std::string> ManifestFiles1(Paths); + ManifestFiles1 = ManifestFiles1.subspan(0, FileCount / 2); + + std::span<const uint64_t> ManifestSizes1(Sizes); + ManifestSizes1 = ManifestSizes1.subspan(0, FileCount / 2); + + std::span<const std::string> ManifestFiles2(Paths); + ManifestFiles2 = ManifestFiles2.subspan(FileCount / 2 - 1); + + std::span<const uint64_t> ManifestSizes2(Sizes); + ManifestSizes2 = ManifestSizes2.subspan(FileCount / 2 - 1); + + const Oid BuildPart1Id = Oid::NewOid(); + const std::string BuildPart1Name = "part1"; + const Oid BuildPart2Id = Oid::NewOid(); + const std::string BuildPart2Name = "part2"; + { + CbObjectWriter Writer; + Writer.BeginObject("parts"sv); + { + Writer.BeginObject(BuildPart1Name); + { + Writer.AddObjectId("partId"sv, BuildPart1Id); + Writer.BeginArray("files"sv); + for (const std::string& ManifestFile : ManifestFiles1) + { + Writer.AddString(ManifestFile); + } + Writer.EndArray(); // files + } + Writer.EndObject(); // part1 + + Writer.BeginObject(BuildPart2Name); + { + Writer.AddObjectId("partId"sv, BuildPart2Id); + Writer.BeginArray("files"sv); + for (const std::string& ManifestFile : ManifestFiles2) + { + Writer.AddString(ManifestFile); + } + Writer.EndArray(); // files + } + Writer.EndObject(); // part2 + } + Writer.EndObject(); // parts + + ExtendableStringBuilder<1024> Manifest; + CompactBinaryToJson(Writer.Save(), Manifest); + WriteFile(State.RootPath / "manifest.json", IoBuffer(IoBuffer::Wrap, Manifest.Data(), Manifest.Size())); + } + + const Oid BuildId = Oid::NewOid(); + + auto Result = State.Upload(BuildId, {}, {}, "source", State.RootPath / "manifest.json"); + + CHECK_EQ(Result.size(), 2u); + CHECK_EQ(Result[0].first, BuildPart1Id); + CHECK_EQ(Result[0].second, BuildPart1Name); + CHECK_EQ(Result[1].first, BuildPart2Id); + CHECK_EQ(Result[1].second, BuildPart2Name); + State.ValidateUpload(BuildId, Result); + + FolderContent DownloadContent = State.Download(BuildId, Oid::Zero, {}, "download", /* Append */ false); + State.ValidateDownload(Paths, Sizes, "source", "download", DownloadContent); + + FolderContent Part1DownloadContent = State.Download(BuildId, BuildPart1Id, {}, "download_part1", /* Append */ false); + State.ValidateDownload(ManifestFiles1, ManifestSizes1, "source", "download_part1", Part1DownloadContent); + + FolderContent Part2DownloadContent = State.Download(BuildId, Oid::Zero, BuildPart2Name, "download_part2", /* Append */ false); + State.ValidateDownload(ManifestFiles2, ManifestSizes2, "source", "download_part2", Part2DownloadContent); + + (void)State.Download(BuildId, BuildPart1Id, BuildPart1Name, "download_part1+2", /* Append */ false); + FolderContent Part1And2DownloadContent = State.Download(BuildId, BuildPart2Id, {}, "download_part1+2", /* Append */ true); + State.ValidateDownload(Paths, Sizes, "source", "download_part1+2", Part1And2DownloadContent); + } +} + +TEST_CASE("buildstorageoperations.partial.block.download" * doctest::skip(true)) +{ + const std::string OidcExecutableName = "OidcToken" ZEN_EXE_SUFFIX_LITERAL; + std::filesystem::path OidcTokenExePath = (GetRunningExecutablePath().parent_path() / OidcExecutableName).make_preferred(); + + HttpClientSettings ClientSettings{ + .LogCategory = "httpbuildsclient", + .AccessTokenProvider = + httpclientauth::CreateFromOidcTokenExecutable(OidcTokenExePath, "https://jupiter.devtools.epicgames.com", true, false, false), + .AssumeHttp2 = false, + .AllowResume = true, + .RetryCount = 0, + .Verbose = false}; + + HttpClient HttpClient("https://euc.jupiter.devtools.epicgames.com", ClientSettings); + + const std::string_view Namespace = "fortnite.oplog"; + const std::string_view Bucket = "fortnitegame.staged-build.fortnite-main.ps4-client"; + const Oid BuildId = Oid::FromHexString("09a76ea92ad301d4724fafad"); + + { + HttpClient::Response Response = HttpClient.Get(fmt::format("/api/v2/builds/{}/{}/{}", Namespace, Bucket, BuildId), + HttpClient::Accept(ZenContentType::kCbObject)); + CbValidateError ValidateResult = CbValidateError::None; + CbObject Object = ValidateAndReadCompactBinaryObject(IoBuffer(Response.ResponsePayload), ValidateResult); + REQUIRE(ValidateResult == CbValidateError::None); + } + + std::vector<ChunkBlockDescription> BlockDescriptions; + { + CbObjectWriter Request; + + Request.BeginArray("blocks"sv); + { + Request.AddHash(IoHash::FromHexString("7c353ed782675a5e8f968e61e51fc797ecdc2882")); + } + Request.EndArray(); + + IoBuffer Payload = Request.Save().GetBuffer().AsIoBuffer(); + Payload.SetContentType(ZenContentType::kCbObject); + + HttpClient::Response BlockDescriptionsResponse = + HttpClient.Post(fmt::format("/api/v2/builds/{}/{}/{}/blocks/getBlockMetadata", Namespace, Bucket, BuildId), + Payload, + HttpClient::Accept(ZenContentType::kCbObject)); + REQUIRE(BlockDescriptionsResponse.IsSuccess()); + + CbValidateError ValidateResult = CbValidateError::None; + CbObject Object = ValidateAndReadCompactBinaryObject(IoBuffer(BlockDescriptionsResponse.ResponsePayload), ValidateResult); + REQUIRE(ValidateResult == CbValidateError::None); + + { + CbArrayView BlocksArray = Object["blocks"sv].AsArrayView(); + for (CbFieldView Block : BlocksArray) + { + ChunkBlockDescription Description = ParseChunkBlockDescription(Block.AsObjectView()); + BlockDescriptions.emplace_back(std::move(Description)); + } + } + } + + REQUIRE(!BlockDescriptions.empty()); + + const IoHash BlockHash = BlockDescriptions.back().BlockHash; + + const ChunkBlockDescription& BlockDescription = BlockDescriptions.front(); + REQUIRE(!BlockDescription.ChunkRawHashes.empty()); + REQUIRE(!BlockDescription.ChunkCompressedLengths.empty()); + + std::vector<std::pair<uint64_t, uint64_t>> ChunkOffsetAndSizes; + uint64_t Offset = gsl::narrow<uint32_t>(CompressedBuffer::GetHeaderSizeForNoneEncoder() + BlockDescription.HeaderSize); + + for (uint32_t ChunkCompressedSize : BlockDescription.ChunkCompressedLengths) + { + ChunkOffsetAndSizes.push_back(std::make_pair(Offset, ChunkCompressedSize)); + Offset += ChunkCompressedSize; + } + + ScopedTemporaryDirectory SourceFolder; + + auto Validate = [&](std::span<const uint32_t> ChunkIndexesToFetch) { + std::vector<std::pair<uint64_t, uint64_t>> Ranges; + for (uint32_t ChunkIndex : ChunkIndexesToFetch) + { + Ranges.push_back(ChunkOffsetAndSizes[ChunkIndex]); + } + + HttpClient::KeyValueMap Headers; + if (!Ranges.empty()) + { + ExtendableStringBuilder<512> SB; + for (const std::pair<uint64_t, uint64_t>& R : Ranges) + { + if (SB.Size() > 0) + { + SB << ", "; + } + SB << R.first << "-" << R.first + R.second - 1; + } + Headers.Entries.insert({"Range", fmt::format("bytes={}", SB.ToView())}); + } + + HttpClient::Response GetBlobRangesResponse = HttpClient.Download( + fmt::format("/api/v2/builds/{}/{}/{}/blobs/{}?supportsRedirect=false", Namespace, Bucket, BuildId, BlockHash), + SourceFolder.Path(), + Headers); + + REQUIRE(GetBlobRangesResponse.IsSuccess()); + [[maybe_unused]] MemoryView RangesMemoryView = GetBlobRangesResponse.ResponsePayload.GetView(); + + std::vector<std::pair<uint64_t, uint64_t>> PayloadRanges = GetBlobRangesResponse.GetRanges(Ranges); + if (PayloadRanges.empty()) + { + // We got the whole blob, use the ranges as is + PayloadRanges = Ranges; + } + + REQUIRE(PayloadRanges.size() == Ranges.size()); + + for (uint32_t RangeIndex = 0; RangeIndex < PayloadRanges.size(); RangeIndex++) + { + const std::pair<uint64_t, uint64_t>& PayloadRange = PayloadRanges[RangeIndex]; + + CHECK_EQ(PayloadRange.second, Ranges[RangeIndex].second); + + IoBuffer ChunkPayload(GetBlobRangesResponse.ResponsePayload, PayloadRange.first, PayloadRange.second); + IoHash RawHash; + uint64_t RawSize; + CompressedBuffer CompressedChunk = CompressedBuffer::FromCompressed(SharedBuffer(ChunkPayload), RawHash, RawSize); + CHECK(CompressedChunk); + CHECK_EQ(RawHash, BlockDescription.ChunkRawHashes[ChunkIndexesToFetch[RangeIndex]]); + CHECK_EQ(RawSize, BlockDescription.ChunkRawLengths[ChunkIndexesToFetch[RangeIndex]]); + } + }; + + { + // Single + std::vector<uint32_t> ChunkIndexesToFetch{uint32_t(BlockDescription.ChunkCompressedLengths.size() / 2)}; + Validate(ChunkIndexesToFetch); + } + { + // Many + std::vector<uint32_t> ChunkIndexesToFetch; + for (uint32_t Index = 0; Index < BlockDescription.ChunkCompressedLengths.size() / 16; Index++) + { + ChunkIndexesToFetch.push_back(uint32_t(BlockDescription.ChunkCompressedLengths.size() / 6 + Index * 7)); + ChunkIndexesToFetch.push_back(uint32_t(BlockDescription.ChunkCompressedLengths.size() / 6 + Index * 7 + 1)); + ChunkIndexesToFetch.push_back(uint32_t(BlockDescription.ChunkCompressedLengths.size() / 6 + Index * 7 + 3)); + } + Validate(ChunkIndexesToFetch); + } + + { + // First and last + std::vector<uint32_t> ChunkIndexesToFetch{0, uint32_t(BlockDescription.ChunkCompressedLengths.size() - 1)}; + Validate(ChunkIndexesToFetch); + } +} +TEST_SUITE_END(); + +void +buildstorageutil_forcelink() +{ +} + +#endif // ZEN_WITH_TESTS + } // namespace zen |