Merge remote-tracking branch 'origin/main' into de/cache-with-block-store

1 files changed, 19 insertions, 1302 deletions

diff --git a/zenserver/upstream/upstreamapply.cpp b/zenserver/upstream/upstreamapply.cpp
index 17a6bb3cf..9758e7565 100644
--- a/zenserver/upstream/upstreamapply.cpp
+++ b/zenserver/upstream/upstreamapply.cpp
@@ -4,1303 +4,26 @@
 
 #if ZEN_WITH_COMPUTE_SERVICES
 
-#	include "jupiter.h"
-#	include "zen.h"
-
 #	include <zencore/compactbinary.h>
 #	include <zencore/compactbinarybuilder.h>
-#	include <zencore/compactbinarypackage.h>
-#	include <zencore/compactbinaryvalidation.h>
-#	include <zencore/compress.h>
 #	include <zencore/fmtutils.h>
-#	include <zencore/session.h>
-#	include <zencore/stats.h>
 #	include <zencore/stream.h>
-#	include <zencore/thread.h>
 #	include <zencore/timer.h>
 #	include <zencore/workthreadpool.h>
 
 #	include <zenstore/cas.h>
 #	include <zenstore/cidstore.h>
 
-#	include <auth/authmgr.h>
-#	include <upstream/upstreamcache.h>
-
-#	include "cache/structuredcachestore.h"
 #	include "diag/logging.h"
 
 #	include <fmt/format.h>
 
-#	include <algorithm>
 #	include <atomic>
-#	include <set>
-#	include <stack>
-#	include <thread>
-#	include <unordered_map>
 
 namespace zen {
 
 using namespace std::literals;
 
-static const IoBuffer EmptyBuffer;
-static const IoHash	  EmptyBufferId = IoHash::HashBuffer(EmptyBuffer);
-
-namespace detail {
-
-	class HordeUpstreamApplyEndpoint final : public UpstreamApplyEndpoint
-	{
-	public:
-		HordeUpstreamApplyEndpoint(const CloudCacheClientOptions& ComputeOptions,
-								   const UpstreamAuthConfig&	  ComputeAuthConfig,
-								   const CloudCacheClientOptions& StorageOptions,
-								   const UpstreamAuthConfig&	  StorageAuthConfig,
-								   CasStore&					  CasStore,
-								   CidStore&					  CidStore,
-								   AuthMgr&						  Mgr)
-		: m_Log(logging::Get("upstream-apply"))
-		, m_CasStore(CasStore)
-		, m_CidStore(CidStore)
-		, m_AuthMgr(Mgr)
-		{
-			m_DisplayName = fmt::format("{} - '{}'+'{}'", ComputeOptions.Name, ComputeOptions.ServiceUrl, StorageOptions.ServiceUrl);
-			m_ChannelId	  = fmt::format("zen-{}", zen::GetSessionIdString());
-
-			{
-				std::unique_ptr<CloudCacheTokenProvider> TokenProvider;
-
-				if (ComputeAuthConfig.OAuthUrl.empty() == false)
-				{
-					TokenProvider =
-						CloudCacheTokenProvider::CreateFromOAuthClientCredentials({.Url			 = ComputeAuthConfig.OAuthUrl,
-																				   .ClientId	 = ComputeAuthConfig.OAuthClientId,
-																				   .ClientSecret = ComputeAuthConfig.OAuthClientSecret});
-				}
-				else if (ComputeAuthConfig.OpenIdProvider.empty() == false)
-				{
-					TokenProvider =
-						CloudCacheTokenProvider::CreateFromCallback([this, ProviderName = std::string(ComputeAuthConfig.OpenIdProvider)]() {
-							AuthMgr::OpenIdAccessToken Token = m_AuthMgr.GetOpenIdAccessToken(ProviderName);
-							return CloudCacheAccessToken{.Value = Token.AccessToken, .ExpireTime = Token.ExpireTime};
-						});
-				}
-				else
-				{
-					CloudCacheAccessToken AccessToken{.Value	  = std::string(ComputeAuthConfig.AccessToken),
-													  .ExpireTime = CloudCacheAccessToken::TimePoint::max()};
-					TokenProvider = CloudCacheTokenProvider::CreateFromStaticToken(AccessToken);
-				}
-
-				m_Client = new CloudCacheClient(ComputeOptions, std::move(TokenProvider));
-			}
-
-			{
-				std::unique_ptr<CloudCacheTokenProvider> TokenProvider;
-
-				if (StorageAuthConfig.OAuthUrl.empty() == false)
-				{
-					TokenProvider =
-						CloudCacheTokenProvider::CreateFromOAuthClientCredentials({.Url			 = StorageAuthConfig.OAuthUrl,
-																				   .ClientId	 = StorageAuthConfig.OAuthClientId,
-																				   .ClientSecret = StorageAuthConfig.OAuthClientSecret});
-				}
-				else if (StorageAuthConfig.OpenIdProvider.empty() == false)
-				{
-					TokenProvider =
-						CloudCacheTokenProvider::CreateFromCallback([this, ProviderName = std::string(StorageAuthConfig.OpenIdProvider)]() {
-							AuthMgr::OpenIdAccessToken Token = m_AuthMgr.GetOpenIdAccessToken(ProviderName);
-							return CloudCacheAccessToken{.Value = Token.AccessToken, .ExpireTime = Token.ExpireTime};
-						});
-				}
-				else
-				{
-					CloudCacheAccessToken AccessToken{.Value	  = std::string(StorageAuthConfig.AccessToken),
-													  .ExpireTime = CloudCacheAccessToken::TimePoint::max()};
-					TokenProvider = CloudCacheTokenProvider::CreateFromStaticToken(AccessToken);
-				}
-
-				m_StorageClient = new CloudCacheClient(StorageOptions, std::move(TokenProvider));
-			}
-		}
-
-		virtual ~HordeUpstreamApplyEndpoint() = default;
-
-		virtual UpstreamEndpointHealth Initialize() override { return CheckHealth(); }
-
-		virtual bool IsHealthy() const override { return m_HealthOk.load(); }
-
-		virtual UpstreamEndpointHealth CheckHealth() override
-		{
-			try
-			{
-				CloudCacheSession Session(m_Client);
-				CloudCacheResult  Result = Session.Authenticate();
-
-				m_HealthOk = Result.ErrorCode == 0;
-
-				return {.Reason = std::move(Result.Reason), .Ok = Result.Success};
-			}
-			catch (std::exception& Err)
-			{
-				return {.Reason = Err.what(), .Ok = false};
-			}
-		}
-
-		virtual std::string_view DisplayName() const override { return m_DisplayName; }
-
-		virtual PostUpstreamApplyResult PostApply(UpstreamApplyRecord ApplyRecord) override
-		{
-			int64_t Bytes{};
-			double	ElapsedSeconds{};
-
-			try
-			{
-				UpstreamData UpstreamData;
-				if (!ProcessApplyKey(ApplyRecord, UpstreamData))
-				{
-					return {.Error{.ErrorCode = -1, .Reason = "Failed to generate task data"}};
-				}
-
-				{
-					std::scoped_lock Lock(m_TaskMutex);
-					if (m_PendingTasks.contains(UpstreamData.TaskId))
-					{
-						// Pending task is already queued, return success
-						return {.Bytes = Bytes, .ElapsedSeconds = ElapsedSeconds, .Success = true};
-					}
-					m_PendingTasks[UpstreamData.TaskId] = std::move(ApplyRecord);
-				}
-
-				CloudCacheSession ComputeSession(m_Client);
-				CloudCacheSession StorageSession(m_StorageClient);
-
-				{
-					CloudCacheResult Result = BatchPutBlobsIfMissing(StorageSession, UpstreamData.Blobs);
-					Bytes += Result.Bytes;
-					ElapsedSeconds += Result.ElapsedSeconds;
-					if (!Result.Success)
-					{
-						return {.Error{.ErrorCode = Result.ErrorCode ? Result.ErrorCode : -1,
-									   .Reason	  = !Result.Reason.empty() ? std::move(Result.Reason) : "Failed to upload blobs"},
-								.Bytes			= Bytes,
-								.ElapsedSeconds = ElapsedSeconds};
-					}
-					UpstreamData.Blobs.clear();
-				}
-
-				{
-					CloudCacheResult Result = BatchPutObjectsIfMissing(StorageSession, UpstreamData.Objects);
-					Bytes += Result.Bytes;
-					ElapsedSeconds += Result.ElapsedSeconds;
-					if (!Result.Success)
-					{
-						return {.Error{.ErrorCode = Result.ErrorCode ? Result.ErrorCode : -1,
-									   .Reason	  = !Result.Reason.empty() ? std::move(Result.Reason) : "Failed to upload objects"},
-								.Bytes			= Bytes,
-								.ElapsedSeconds = ElapsedSeconds};
-					}
-
-					PutRefResult RefResult = StorageSession.PutRef("requests"sv,
-																   UpstreamData.TaskId,
-																   UpstreamData.Objects[UpstreamData.TaskId].GetBuffer().AsIoBuffer(),
-																   ZenContentType::kCbObject);
-					Log().debug("Put ref {} Need={} Bytes={} Duration={}s Result={}",
-								UpstreamData.TaskId,
-								RefResult.Needs.size(),
-								RefResult.Bytes,
-								RefResult.ElapsedSeconds,
-								RefResult.Success);
-
-					Bytes += RefResult.Bytes;
-					ElapsedSeconds += RefResult.ElapsedSeconds;
-					if (!RefResult.Success)
-					{
-						return {.Error{.ErrorCode = RefResult.ErrorCode ? RefResult.ErrorCode : -1,
-									   .Reason	  = !RefResult.Reason.empty() ? std::move(RefResult.Reason) : "Failed to add task ref"},
-								.Bytes			= Bytes,
-								.ElapsedSeconds = ElapsedSeconds};
-					}
-					UpstreamData.Objects.clear();
-				}
-
-				CbObjectWriter Writer;
-				Writer.AddString("c"sv, m_ChannelId);
-				Writer.AddObjectAttachment("r"sv, UpstreamData.RequirementsId);
-				Writer.BeginArray("t"sv);
-				Writer.AddObjectAttachment(UpstreamData.TaskId);
-				Writer.EndArray();
-				CbObject TasksObject = Writer.Save();
-				IoBuffer TasksData	 = TasksObject.GetBuffer().AsIoBuffer();
-
-				CloudCacheResult Result = ComputeSession.PostComputeTasks(TasksData);
-				Log().debug("Post compute task {} Bytes={} Duration={}s Result={}",
-							TasksObject.GetHash(),
-							Result.Bytes,
-							Result.ElapsedSeconds,
-							Result.Success);
-				Bytes += Result.Bytes;
-				ElapsedSeconds += Result.ElapsedSeconds;
-				if (!Result.Success)
-				{
-					{
-						std::scoped_lock Lock(m_TaskMutex);
-						m_PendingTasks.erase(UpstreamData.TaskId);
-					}
-
-					return {.Error{.ErrorCode = Result.ErrorCode ? Result.ErrorCode : -1,
-								   .Reason	  = !Result.Reason.empty() ? std::move(Result.Reason) : "Failed to post compute task"},
-							.Bytes			= Bytes,
-							.ElapsedSeconds = ElapsedSeconds};
-				}
-
-				Log().info("Task posted {}", UpstreamData.TaskId);
-				return {.Bytes = Bytes, .ElapsedSeconds = ElapsedSeconds, .Success = true};
-			}
-			catch (std::exception& Err)
-			{
-				m_HealthOk = false;
-				return {.Error{.ErrorCode = -1, .Reason = Err.what()}, .Bytes = Bytes, .ElapsedSeconds = ElapsedSeconds};
-			}
-		}
-
-		[[nodiscard]] CloudCacheResult BatchPutBlobsIfMissing(CloudCacheSession& Session, const std::map<IoHash, IoBuffer>& Blobs)
-		{
-			if (Blobs.size() == 0)
-			{
-				return {.Success = true};
-			}
-
-			int64_t Bytes{};
-			double	ElapsedSeconds{};
-
-			// Batch check for missing blobs
-			std::set<IoHash> Keys;
-			std::transform(Blobs.begin(), Blobs.end(), std::inserter(Keys, Keys.end()), [](const auto& It) { return It.first; });
-
-			CloudCacheExistsResult ExistsResult = Session.BlobExists(Keys);
-			Log().debug("Queried {} missing blobs Need={} Duration={}s Result={}",
-						Keys.size(),
-						ExistsResult.Needs.size(),
-						ExistsResult.ElapsedSeconds,
-						ExistsResult.Success);
-			ElapsedSeconds += ExistsResult.ElapsedSeconds;
-			if (!ExistsResult.Success)
-			{
-				return {.Bytes			= Bytes,
-						.ElapsedSeconds = ElapsedSeconds,
-						.ErrorCode		= ExistsResult.ErrorCode ? ExistsResult.ErrorCode : -1,
-						.Reason			= !ExistsResult.Reason.empty() ? std::move(ExistsResult.Reason) : "Failed to check if blobs exist"};
-			}
-
-			// TODO: Batch upload missing blobs
-
-			for (const auto& Hash : ExistsResult.Needs)
-			{
-				CloudCacheResult Result = Session.PutBlob(Hash, Blobs.at(Hash));
-				Log().debug("Put blob {} Bytes={} Duration={}s Result={}", Hash, Result.Bytes, Result.ElapsedSeconds, Result.Success);
-				Bytes += Result.Bytes;
-				ElapsedSeconds += Result.ElapsedSeconds;
-				if (!Result.Success)
-				{
-					return {.Bytes			= Bytes,
-							.ElapsedSeconds = ElapsedSeconds,
-							.ErrorCode		= Result.ErrorCode ? Result.ErrorCode : -1,
-							.Reason			= !Result.Reason.empty() ? std::move(Result.Reason) : "Failed to put blobs"};
-				}
-			}
-
-			return {.Bytes = Bytes, .ElapsedSeconds = ElapsedSeconds, .Success = true};
-		}
-
-		[[nodiscard]] CloudCacheResult BatchPutObjectsIfMissing(CloudCacheSession& Session, const std::map<IoHash, CbObject>& Objects)
-		{
-			if (Objects.size() == 0)
-			{
-				return {.Success = true};
-			}
-
-			int64_t Bytes{};
-			double	ElapsedSeconds{};
-
-			// Batch check for missing objects
-			std::set<IoHash> Keys;
-			std::transform(Objects.begin(), Objects.end(), std::inserter(Keys, Keys.end()), [](const auto& It) { return It.first; });
-
-			// Todo: Endpoint doesn't exist for objects
-			CloudCacheExistsResult ExistsResult = Session.ObjectExists(Keys);
-			Log().debug("Queried {} missing objects Need={} Duration={}s Result={}",
-						Keys.size(),
-						ExistsResult.Needs.size(),
-						ExistsResult.ElapsedSeconds,
-						ExistsResult.Success);
-			ElapsedSeconds += ExistsResult.ElapsedSeconds;
-			if (!ExistsResult.Success)
-			{
-				return {.Bytes			= Bytes,
-						.ElapsedSeconds = ElapsedSeconds,
-						.ErrorCode		= ExistsResult.ErrorCode ? ExistsResult.ErrorCode : -1,
-						.Reason = !ExistsResult.Reason.empty() ? std::move(ExistsResult.Reason) : "Failed to check if objects exist"};
-			}
-
-			// TODO: Batch upload missing objects
-
-			for (const auto& Hash : ExistsResult.Needs)
-			{
-				CloudCacheResult Result = Session.PutObject(Hash, Objects.at(Hash).GetBuffer().AsIoBuffer());
-				Log().debug("Put object {} Bytes={} Duration={}s Result={}", Hash, Result.Bytes, Result.ElapsedSeconds, Result.Success);
-				Bytes += Result.Bytes;
-				ElapsedSeconds += Result.ElapsedSeconds;
-				if (!Result.Success)
-				{
-					return {.Bytes			= Bytes,
-							.ElapsedSeconds = ElapsedSeconds,
-							.ErrorCode		= Result.ErrorCode ? Result.ErrorCode : -1,
-							.Reason			= !Result.Reason.empty() ? std::move(Result.Reason) : "Failed to put objects"};
-				}
-			}
-
-			return {.Bytes = Bytes, .ElapsedSeconds = ElapsedSeconds, .Success = true};
-		}
-
-		enum class ComputeTaskState : int32_t
-		{
-			Queued	  = 0,
-			Executing = 1,
-			Complete  = 2,
-		};
-
-		enum class ComputeTaskOutcome : int32_t
-		{
-			Success		 = 0,
-			Failed		 = 1,
-			Cancelled	 = 2,
-			NoResult	 = 3,
-			Exipred		 = 4,
-			BlobNotFound = 5,
-			Exception	 = 6,
-		};
-
-		[[nodiscard]] static std::string_view ComputeTaskStateToString(const ComputeTaskState Outcome)
-		{
-			switch (Outcome)
-			{
-				case ComputeTaskState::Queued:
-					return "Queued"sv;
-				case ComputeTaskState::Executing:
-					return "Executing"sv;
-				case ComputeTaskState::Complete:
-					return "Complete"sv;
-			};
-			return "Unknown"sv;
-		}
-
-		[[nodiscard]] static std::string_view ComputeTaskOutcomeToString(const ComputeTaskOutcome Outcome)
-		{
-			switch (Outcome)
-			{
-				case ComputeTaskOutcome::Success:
-					return "Success"sv;
-				case ComputeTaskOutcome::Failed:
-					return "Failed"sv;
-				case ComputeTaskOutcome::Cancelled:
-					return "Cancelled"sv;
-				case ComputeTaskOutcome::NoResult:
-					return "NoResult"sv;
-				case ComputeTaskOutcome::Exipred:
-					return "Exipred"sv;
-				case ComputeTaskOutcome::BlobNotFound:
-					return "BlobNotFound"sv;
-				case ComputeTaskOutcome::Exception:
-					return "Exception"sv;
-			};
-			return "Unknown"sv;
-		}
-
-		virtual GetUpstreamApplyUpdatesResult GetUpdates(WorkerThreadPool& ThreadPool) override
-		{
-			int64_t Bytes{};
-			double	ElapsedSeconds{};
-
-			{
-				std::scoped_lock Lock(m_TaskMutex);
-				if (m_PendingTasks.empty())
-				{
-					if (m_CompletedTasks.empty())
-					{
-						// Nothing to do.
-						return {.Success = true};
-					}
-
-					UpstreamApplyCompleted CompletedTasks;
-					std::swap(CompletedTasks, m_CompletedTasks);
-					return {.Bytes = Bytes, .ElapsedSeconds = ElapsedSeconds, .Completed = std::move(CompletedTasks), .Success = true};
-				}
-			}
-
-			try
-			{
-				CloudCacheSession ComputeSession(m_Client);
-
-				CloudCacheResult UpdatesResult = ComputeSession.GetComputeUpdates(m_ChannelId);
-				Log().debug("Get compute updates Bytes={} Duration={}s Result={}",
-							UpdatesResult.Bytes,
-							UpdatesResult.ElapsedSeconds,
-							UpdatesResult.Success);
-				Bytes += UpdatesResult.Bytes;
-				ElapsedSeconds += UpdatesResult.ElapsedSeconds;
-				if (!UpdatesResult.Success)
-				{
-					return {.Error{.ErrorCode = UpdatesResult.ErrorCode, .Reason = std::move(UpdatesResult.Reason)},
-							.Bytes			= Bytes,
-							.ElapsedSeconds = ElapsedSeconds};
-				}
-
-				if (!UpdatesResult.Success)
-				{
-					return {.Error{.ErrorCode = -1, .Reason = "Failed get task updates"}, .Bytes = Bytes, .ElapsedSeconds = ElapsedSeconds};
-				}
-
-				CbObject TaskStatus = LoadCompactBinaryObject(std::move(UpdatesResult.Response));
-
-				for (auto& It : TaskStatus["u"sv])
-				{
-					CbObjectView			 Status	 = It.AsObjectView();
-					IoHash					 TaskId	 = Status["h"sv].AsHash();
-					const ComputeTaskState	 State	 = (ComputeTaskState)Status["s"sv].AsInt32();
-					const ComputeTaskOutcome Outcome = (ComputeTaskOutcome)Status["o"sv].AsInt32();
-
-					Log().info("Task {} State={}", TaskId, ComputeTaskStateToString(State));
-
-					// Only completed tasks need to be processed
-					if (State != ComputeTaskState::Complete)
-					{
-						continue;
-					}
-
-					IoHash			  WorkerId{};
-					IoHash			  ActionId{};
-					UpstreamApplyType ApplyType{};
-
-					{
-						std::scoped_lock Lock(m_TaskMutex);
-						auto			 TaskIt = m_PendingTasks.find(TaskId);
-						if (TaskIt != m_PendingTasks.end())
-						{
-							WorkerId  = TaskIt->second.WorkerDescriptor.GetHash();
-							ActionId  = TaskIt->second.Action.GetHash();
-							ApplyType = TaskIt->second.Type;
-							m_PendingTasks.erase(TaskIt);
-						}
-					}
-
-					if (WorkerId == IoHash::Zero)
-					{
-						Log().warn("Task {} missing from pending tasks", TaskId);
-						continue;
-					}
-
-					if (Outcome != ComputeTaskOutcome::Success)
-					{
-						const std::string_view Detail = Status["d"sv].AsString();
-						{
-							std::scoped_lock Lock(m_TaskMutex);
-							m_CompletedTasks[WorkerId][ActionId] = {
-								.Error{.ErrorCode = -1, .Reason = fmt::format("Task {} {}", ComputeTaskOutcomeToString(Outcome), Detail)}};
-						}
-						continue;
-					}
-
-					ThreadPool.ScheduleWork([this,
-											 ApplyType,
-											 ResultHash = Status["r"sv].AsHash(),
-											 TaskId		= std::move(TaskId),
-											 WorkerId	= std::move(WorkerId),
-											 ActionId	= std::move(ActionId)]() {
-						GetUpstreamApplyResult Result = ProcessTaskStatus(ApplyType, ResultHash);
-						Log().debug("Task Processed {} Files={} Attachments={} ExitCode={}",
-									TaskId,
-									Result.OutputFiles.size(),
-									Result.OutputPackage.GetAttachments().size(),
-									Result.Error.ErrorCode);
-						{
-							std::scoped_lock Lock(m_TaskMutex);
-							m_CompletedTasks[WorkerId][ActionId] = std::move(Result);
-						}
-					});
-				}
-
-				{
-					std::scoped_lock Lock(m_TaskMutex);
-					if (m_CompletedTasks.empty())
-					{
-						// Nothing to do.
-						return {.Bytes = Bytes, .ElapsedSeconds = ElapsedSeconds, .Success = true};
-					}
-					UpstreamApplyCompleted CompletedTasks;
-					std::swap(CompletedTasks, m_CompletedTasks);
-					return {.Bytes = Bytes, .ElapsedSeconds = ElapsedSeconds, .Completed = std::move(CompletedTasks), .Success = true};
-				}
-			}
-			catch (std::exception& Err)
-			{
-				m_HealthOk = false;
-				return {
-					.Error{.ErrorCode = -1, .Reason = Err.what()},
-					.Bytes			= Bytes,
-					.ElapsedSeconds = ElapsedSeconds,
-				};
-			}
-		}
-
-		virtual UpstreamApplyEndpointStats& Stats() override { return m_Stats; }
-
-	private:
-		spdlog::logger& Log() { return m_Log; }
-
-		spdlog::logger&			   m_Log;
-		CasStore&				   m_CasStore;
-		CidStore&				   m_CidStore;
-		AuthMgr&				   m_AuthMgr;
-		std::string				   m_DisplayName;
-		RefPtr<CloudCacheClient>   m_Client;
-		RefPtr<CloudCacheClient>   m_StorageClient;
-		UpstreamApplyEndpointStats m_Stats;
-		std::atomic_bool		   m_HealthOk{false};
-		std::string				   m_ChannelId;
-
-		std::mutex										m_TaskMutex;
-		std::unordered_map<IoHash, UpstreamApplyRecord> m_PendingTasks;
-		UpstreamApplyCompleted							m_CompletedTasks;
-
-		struct UpstreamData
-		{
-			std::map<IoHash, IoBuffer> Blobs;
-			std::map<IoHash, CbObject> Objects;
-			IoHash					   TaskId;
-			IoHash					   RequirementsId;
-		};
-
-		struct UpstreamDirectory
-		{
-			std::filesystem::path					 Path;
-			std::map<std::string, UpstreamDirectory> Directories;
-			std::set<std::string>					 Files;
-		};
-
-		[[nodiscard]] GetUpstreamApplyResult ProcessTaskStatus(const UpstreamApplyType ApplyType, const IoHash& ResultHash)
-		{
-			try
-			{
-				CloudCacheSession Session(m_StorageClient);
-
-				int64_t Bytes{};
-				double	ElapsedSeconds{};
-
-				// Get Result object and all Object Attachments + Binary Attachment IDs
-				CloudCacheResult ObjectRefResult = Session.GetRef("responses"sv, ResultHash, ZenContentType::kCbObject);
-				Log().debug("Get ref {} Bytes={} Duration={}s Result={}",
-							ResultHash,
-							ObjectRefResult.Bytes,
-							ObjectRefResult.ElapsedSeconds,
-							ObjectRefResult.Success);
-				Bytes += ObjectRefResult.Bytes;
-				ElapsedSeconds += ObjectRefResult.ElapsedSeconds;
-
-				if (!ObjectRefResult.Success)
-				{
-					return {.Error{.ErrorCode = -1, .Reason = "Failed to get result object data"},
-							.Bytes			= Bytes,
-							.ElapsedSeconds = ElapsedSeconds};
-				}
-
-				std::vector<IoHash>		   ObjectsToIterate;
-				std::map<IoHash, IoBuffer> ObjectData;
-				std::map<IoHash, IoBuffer> BinaryData;
-
-				ObjectData[ResultHash] = ObjectRefResult.Response;
-				CbObject Object		   = LoadCompactBinaryObject(ObjectData[ResultHash]);
-				Object.IterateAttachments([&](CbFieldView Field) {
-					if (Field.IsObjectAttachment())
-					{
-						const IoHash AttachmentHash = Field.AsObjectAttachment();
-						if (!ObjectData.contains(AttachmentHash))
-						{
-							ObjectsToIterate.push_back(AttachmentHash);
-						}
-					}
-					else if (Field.IsBinaryAttachment())
-					{
-						const IoHash AttachmentHash = Field.AsBinaryAttachment();
-						BinaryData[AttachmentHash]	= {};
-					}
-				});
-
-				while (!ObjectsToIterate.empty())
-				{
-					const IoHash Hash = ObjectsToIterate.back();
-					ObjectsToIterate.pop_back();
-
-					CloudCacheResult ObjectResult = Session.GetObject(Hash);
-					Log().debug("Get object {} Bytes={} Duration={}s Result={}",
-								Hash,
-								ObjectResult.Bytes,
-								ObjectResult.ElapsedSeconds,
-								ObjectResult.Success);
-					Bytes += ObjectRefResult.Bytes;
-					ElapsedSeconds += ObjectRefResult.ElapsedSeconds;
-					if (!ObjectResult.Success)
-					{
-						return {.Error{.ErrorCode = ObjectResult.ErrorCode, .Reason = std::move(ObjectResult.Reason)},
-								.Bytes			= Bytes,
-								.ElapsedSeconds = ElapsedSeconds};
-					}
-					ObjectData[Hash] = std::move(ObjectResult.Response);
-
-					CbObject IterateObject = LoadCompactBinaryObject(ObjectData[Hash]);
-					IterateObject.IterateAttachments([&](CbFieldView Field) {
-						if (Field.IsObjectAttachment())
-						{
-							const IoHash AttachmentHash = Field.AsObjectAttachment();
-							if (!ObjectData.contains(AttachmentHash))
-							{
-								ObjectsToIterate.push_back(AttachmentHash);
-							}
-						}
-						else if (Field.IsBinaryAttachment())
-						{
-							const IoHash AttachmentHash = Field.AsBinaryAttachment();
-							BinaryData[AttachmentHash]	= {};
-						}
-					});
-				}
-
-				// Batch load all binary data
-				for (auto& It : BinaryData)
-				{
-					CloudCacheResult BlobResult = Session.GetBlob(It.first);
-					Log().debug("Get blob {} Bytes={} Duration={}s Result={}",
-								It.first,
-								BlobResult.Bytes,
-								BlobResult.ElapsedSeconds,
-								BlobResult.Success);
-					Bytes += ObjectRefResult.Bytes;
-					ElapsedSeconds += ObjectRefResult.ElapsedSeconds;
-					if (!BlobResult.Success)
-					{
-						return {.Error{.ErrorCode = BlobResult.ErrorCode, .Reason = std::move(BlobResult.Reason)},
-								.Bytes			= Bytes,
-								.ElapsedSeconds = ElapsedSeconds};
-					}
-					It.second = std::move(BlobResult.Response);
-				}
-
-				CbObject ResultObject = LoadCompactBinaryObject(ObjectData[ResultHash]);
-				int32_t	 ExitCode	  = ResultObject["e"sv].AsInt32();
-				IoHash	 StdOutHash	  = ResultObject["so"sv].AsBinaryAttachment();
-				IoHash	 StdErrHash	  = ResultObject["se"sv].AsBinaryAttachment();
-				IoHash	 OutputHash	  = ResultObject["o"sv].AsObjectAttachment();
-
-				std::string StdOut = std::string((const char*)BinaryData[StdOutHash].GetData(), BinaryData[StdOutHash].GetSize());
-				std::string StdErr = std::string((const char*)BinaryData[StdErrHash].GetData(), BinaryData[StdErrHash].GetSize());
-
-				if (OutputHash == IoHash::Zero)
-				{
-					return {.Error{.ErrorCode = ExitCode, .Reason = "Task completed with no output object"},
-							.Bytes			= Bytes,
-							.ElapsedSeconds = ElapsedSeconds,
-							.StdOut			= std::move(StdOut),
-							.StdErr			= std::move(StdErr)};
-				}
-
-				CbObject OutputObject = LoadCompactBinaryObject(ObjectData[OutputHash]);
-
-				switch (ApplyType)
-				{
-					case UpstreamApplyType::Simple:
-						{
-							std::map<std::filesystem::path, IoHash> OutputFiles;
-
-							ResolveMerkleTreeDirectory(""sv, OutputHash, ObjectData, OutputFiles);
-
-							return {.OutputFiles	= std::move(OutputFiles),
-									.FileData		= std::move(BinaryData),
-									.Bytes			= Bytes,
-									.ElapsedSeconds = ElapsedSeconds,
-									.StdOut			= std::move(StdOut),
-									.StdErr			= std::move(StdErr),
-									.Success		= true};
-						}
-						break;
-					case UpstreamApplyType::Asset:
-						{
-							if (ExitCode != 0)
-							{
-								return {.Error{.ErrorCode = ExitCode, .Reason = "Task completed with errors"},
-										.Bytes			= Bytes,
-										.ElapsedSeconds = ElapsedSeconds,
-										.StdOut			= std::move(StdOut),
-										.StdErr			= std::move(StdErr)};
-							}
-
-							// Get build.output
-							IoHash	 BuildOutputId;
-							IoBuffer BuildOutput;
-							for (auto& It : OutputObject["f"sv])
-							{
-								const CbObjectView FileObject = It.AsObjectView();
-								if (FileObject["n"sv].AsString() == "Build.output"sv)
-								{
-									BuildOutputId = FileObject["h"sv].AsBinaryAttachment();
-									BuildOutput	  = BinaryData[BuildOutputId];
-									break;
-								}
-							}
-
-							if (BuildOutput.GetSize() == 0)
-							{
-								return {.Error{.ErrorCode = ExitCode, .Reason = "Build.output file not found in task results"},
-										.Bytes			= Bytes,
-										.ElapsedSeconds = ElapsedSeconds,
-										.StdOut			= std::move(StdOut),
-										.StdErr			= std::move(StdErr)};
-							}
-
-							// Get Output directory node
-							IoBuffer OutputDirectoryTree;
-							for (auto& It : OutputObject["d"sv])
-							{
-								const CbObjectView DirectoryObject = It.AsObjectView();
-								if (DirectoryObject["n"sv].AsString() == "Outputs"sv)
-								{
-									OutputDirectoryTree = ObjectData[DirectoryObject["h"sv].AsObjectAttachment()];
-									break;
-								}
-							}
-
-							if (OutputDirectoryTree.GetSize() == 0)
-							{
-								return {.Error{.ErrorCode = ExitCode, .Reason = "Outputs directory not found in task results"},
-										.Bytes			= Bytes,
-										.ElapsedSeconds = ElapsedSeconds,
-										.StdOut			= std::move(StdOut),
-										.StdErr			= std::move(StdErr)};
-							}
-
-							// load build.output as CbObject
-
-							// Move Outputs from Horde to CbPackage
-
-							std::unordered_map<IoHash, IoHash> CidToCompressedId;
-							CbPackage						   OutputPackage;
-							CbObject						   OutputDirectoryTreeObject = LoadCompactBinaryObject(OutputDirectoryTree);
-							int64_t							   TotalAttachmentBytes		 = 0;
-							int64_t							   TotalRawAttachmentBytes	 = 0;
-
-							for (auto& It : OutputDirectoryTreeObject["f"sv])
-							{
-								CbObjectView FileObject = It.AsObjectView();
-								// Name is the uncompressed hash
-								IoHash DecompressedId = IoHash::FromHexString(FileObject["n"sv].AsString());
-								// Hash is the compressed data hash, and how it is stored in Horde
-								IoHash CompressedId = FileObject["h"sv].AsBinaryAttachment();
-
-								if (!BinaryData.contains(CompressedId))
-								{
-									Log().warn("Object attachment chunk not retrieved from Horde {}", CompressedId);
-									return {.Error{.ErrorCode = -1, .Reason = "Object attachment chunk not retrieved from Horde"},
-											.Bytes			= Bytes,
-											.ElapsedSeconds = ElapsedSeconds};
-								}
-								CidToCompressedId[DecompressedId] = CompressedId;
-							}
-
-							// Iterate attachments, verify all chunks exist, and add to CbPackage
-							bool	 AnyErrors		   = false;
-							CbObject BuildOutputObject = LoadCompactBinaryObject(BuildOutput);
-							BuildOutputObject.IterateAttachments([&](CbFieldView Field) {
-								const IoHash DecompressedId = Field.AsHash();
-								if (!CidToCompressedId.contains(DecompressedId))
-								{
-									Log().warn("Attachment not found {}", DecompressedId);
-									AnyErrors = true;
-									return;
-								}
-								const IoHash& CompressedId = CidToCompressedId.at(DecompressedId);
-
-								if (!BinaryData.contains(CompressedId))
-								{
-									Log().warn("Missing output {} compressed {} uncompressed", CompressedId, DecompressedId);
-									AnyErrors = true;
-									return;
-								}
-
-								CompressedBuffer AttachmentBuffer =
-									CompressedBuffer::FromCompressed(SharedBuffer(BinaryData[CompressedId]));
-
-								if (!AttachmentBuffer)
-								{
-									Log().warn(
-										"Invalid output encountered (not valid CompressedBuffer format) {} compressed {} uncompressed",
-										CompressedId,
-										DecompressedId);
-									AnyErrors = true;
-									return;
-								}
-
-								TotalAttachmentBytes += AttachmentBuffer.GetCompressedSize();
-								TotalRawAttachmentBytes += AttachmentBuffer.GetRawSize();
-
-								CbAttachment Attachment(AttachmentBuffer);
-								OutputPackage.AddAttachment(Attachment);
-							});
-
-							if (AnyErrors)
-							{
-								return {.Error{.ErrorCode = -1, .Reason = "Failed to get result object attachment data"},
-										.Bytes			= Bytes,
-										.ElapsedSeconds = ElapsedSeconds,
-										.StdOut			= std::move(StdOut),
-										.StdErr			= std::move(StdErr)};
-							}
-
-							OutputPackage.SetObject(BuildOutputObject);
-
-							return {.OutputPackage			 = std::move(OutputPackage),
-									.TotalAttachmentBytes	 = TotalAttachmentBytes,
-									.TotalRawAttachmentBytes = TotalRawAttachmentBytes,
-									.Bytes					 = Bytes,
-									.ElapsedSeconds			 = ElapsedSeconds,
-									.StdOut					 = std::move(StdOut),
-									.StdErr					 = std::move(StdErr),
-									.Success				 = true};
-						}
-						break;
-				}
-
-				return {.Error{.ErrorCode = ExitCode, .Reason = "Unknown apply type"},
-						.Bytes			= Bytes,
-						.ElapsedSeconds = ElapsedSeconds,
-						.StdOut			= std::move(StdOut),
-						.StdErr			= std::move(StdErr)};
-			}
-			catch (std::exception& Err)
-			{
-				return {.Error{.ErrorCode = -1, .Reason = Err.what()}};
-			}
-		}
-
-		[[nodiscard]] bool ProcessApplyKey(const UpstreamApplyRecord& ApplyRecord, UpstreamData& Data)
-		{
-			std::string								ExecutablePath;
-			std::string								WorkingDirectory;
-			std::vector<std::string>				Arguments;
-			std::map<std::string, std::string>		Environment;
-			std::set<std::filesystem::path>			InputFiles;
-			std::set<std::string>					Outputs;
-			std::map<std::filesystem::path, IoHash> InputFileHashes;
-
-			ExecutablePath = ApplyRecord.WorkerDescriptor["path"sv].AsString();
-			if (ExecutablePath.empty())
-			{
-				Log().warn("process apply upstream FAILED, '{}', path missing from worker descriptor",
-						   ApplyRecord.WorkerDescriptor.GetHash());
-				return false;
-			}
-
-			WorkingDirectory = ApplyRecord.WorkerDescriptor["workdir"sv].AsString();
-
-			for (auto& It : ApplyRecord.WorkerDescriptor["executables"sv])
-			{
-				CbObjectView FileEntry = It.AsObjectView();
-				if (!ProcessFileEntry(FileEntry, InputFiles, InputFileHashes, Data.Blobs))
-				{
-					return false;
-				}
-			}
-
-			for (auto& It : ApplyRecord.WorkerDescriptor["files"sv])
-			{
-				CbObjectView FileEntry = It.AsObjectView();
-				if (!ProcessFileEntry(FileEntry, InputFiles, InputFileHashes, Data.Blobs))
-				{
-					return false;
-				}
-			}
-
-			for (auto& It : ApplyRecord.WorkerDescriptor["dirs"sv])
-			{
-				std::string_view Directory = It.AsString();
-				std::string		 DummyFile = fmt::format("{}/.zen_empty_file", Directory);
-				InputFiles.insert(DummyFile);
-				Data.Blobs[EmptyBufferId]  = EmptyBuffer;
-				InputFileHashes[DummyFile] = EmptyBufferId;
-			}
-
-			for (auto& It : ApplyRecord.WorkerDescriptor["environment"sv])
-			{
-				std::string_view Env   = It.AsString();
-				auto			 Index = Env.find('=');
-				if (Index == std::string_view::npos)
-				{
-					Log().warn("process apply upstream FAILED, environment '{}' malformed", Env);
-					return false;
-				}
-
-				Environment[std::string(Env.substr(0, Index))] = Env.substr(Index + 1);
-			}
-
-			switch (ApplyRecord.Type)
-			{
-				case UpstreamApplyType::Simple:
-					{
-						for (auto& It : ApplyRecord.WorkerDescriptor["arguments"sv])
-						{
-							Arguments.push_back(std::string(It.AsString()));
-						}
-
-						for (auto& It : ApplyRecord.WorkerDescriptor["outputs"sv])
-						{
-							Outputs.insert(std::string(It.AsString()));
-						}
-					}
-					break;
-				case UpstreamApplyType::Asset:
-					{
-						static const std::filesystem::path BuildActionPath = "Build.action"sv;
-						static const std::filesystem::path InputPath	   = "Inputs"sv;
-						const IoHash					   ActionId		   = ApplyRecord.Action.GetHash();
-
-						Arguments.push_back("-Build=build.action");
-						Outputs.insert("Build.output");
-						Outputs.insert("Outputs");
-
-						InputFiles.insert(BuildActionPath);
-						InputFileHashes[BuildActionPath] = ActionId;
-						Data.Blobs[ActionId]			 = IoBufferBuilder::MakeCloneFromMemory(ApplyRecord.Action.GetBuffer().GetData(),
-																					ApplyRecord.Action.GetBuffer().GetSize());
-
-						bool AnyErrors = false;
-						ApplyRecord.Action.IterateAttachments([&](CbFieldView Field) {
-							const IoHash				Cid		   = Field.AsHash();
-							const std::filesystem::path FilePath   = {InputPath / Cid.ToHexString()};
-							IoBuffer					DataBuffer = m_CidStore.FindChunkByCid(Cid);
-
-							if (!DataBuffer)
-							{
-								Log().warn("process apply upstream FAILED, input CID chunk '{}' missing", Cid);
-								AnyErrors = true;
-								return;
-							}
-
-							if (InputFiles.contains(FilePath))
-							{
-								return;
-							}
-
-							const IoHash CompressedId = IoHash::HashBuffer(DataBuffer.GetData(), DataBuffer.GetSize());
-
-							InputFiles.insert(FilePath);
-							InputFileHashes[FilePath] = CompressedId;
-							Data.Blobs[CompressedId]  = std::move(DataBuffer);
-						});
-
-						if (AnyErrors)
-						{
-							return false;
-						}
-					}
-					break;
-			}
-
-			const UpstreamDirectory RootDirectory = BuildDirectoryTree(InputFiles);
-
-			CbObject	 Sandbox	  = BuildMerkleTreeDirectory(RootDirectory, InputFileHashes, Data.Blobs, Data.Objects);
-			const IoHash SandboxHash  = Sandbox.GetHash();
-			Data.Objects[SandboxHash] = std::move(Sandbox);
-
-			{
-				std::string_view HostPlatform = ApplyRecord.WorkerDescriptor["host"sv].AsString();
-				if (HostPlatform.empty())
-				{
-					Log().warn("process apply upstream FAILED, 'host' platform not provided");
-					return false;
-				}
-
-				int32_t LogicalCores = ApplyRecord.WorkerDescriptor["cores"sv].AsInt32();
-				int64_t Memory		 = ApplyRecord.WorkerDescriptor["memory"sv].AsInt64();
-				bool	Exclusive	 = ApplyRecord.WorkerDescriptor["exclusive"sv].AsBool();
-
-				std::string Condition = fmt::format("Platform == '{}'", HostPlatform);
-				if (HostPlatform == "Win64")
-				{
-					// TODO
-					// Condition += " && Pool == 'Win-RemoteExec'";
-				}
-
-				std::map<std::string_view, int64_t> Resources;
-				if (LogicalCores > 0)
-				{
-					Resources["LogicalCores"sv] = LogicalCores;
-				}
-				if (Memory > 0)
-				{
-					Resources["RAM"sv] = std::max(Memory / 1024LL / 1024LL / 1024LL, 1LL);
-				}
-
-				CbObject	 Requirements	 = BuildRequirements(Condition, Resources, Exclusive);
-				const IoHash RequirementsId	 = Requirements.GetHash();
-				Data.Objects[RequirementsId] = std::move(Requirements);
-				Data.RequirementsId			 = RequirementsId;
-			}
-
-			CbObject Task = BuildTask(ExecutablePath, Arguments, Environment, WorkingDirectory, SandboxHash, Data.RequirementsId, Outputs);
-
-			const IoHash TaskId	 = Task.GetHash();
-			Data.Objects[TaskId] = std::move(Task);
-			Data.TaskId			 = TaskId;
-
-			return true;
-		}
-
-		[[nodiscard]] bool ProcessFileEntry(const CbObjectView&						 FileEntry,
-											std::set<std::filesystem::path>&		 InputFiles,
-											std::map<std::filesystem::path, IoHash>& InputFileHashes,
-											std::map<IoHash, IoBuffer>&				 Blobs)
-		{
-			const std::filesystem::path FilePath   = FileEntry["name"sv].AsString();
-			const IoHash				ChunkId	   = FileEntry["hash"sv].AsHash();
-			const uint64_t				Size	   = FileEntry["size"sv].AsUInt64();
-			IoBuffer					DataBuffer = m_CasStore.FindChunk(ChunkId);
-
-			if (!DataBuffer)
-			{
-				Log().warn("process apply upstream FAILED, worker CAS chunk '{}' missing", ChunkId);
-				return false;
-			}
-
-			if (DataBuffer.Size() != Size)
-			{
-				Log().warn("process apply upstream FAILED, worker CAS chunk '{}' size: {}, action spec expected {}",
-						   ChunkId,
-						   DataBuffer.Size(),
-						   Size);
-				return false;
-			}
-
-			if (InputFiles.contains(FilePath))
-			{
-				Log().warn("process apply upstream FAILED, worker CAS chunk '{}' size: {} duplicate filename {}", ChunkId, Size, FilePath);
-				return false;
-			}
-
-			InputFiles.insert(FilePath);
-			InputFileHashes[FilePath] = ChunkId;
-			Blobs[ChunkId]			  = std::move(DataBuffer);
-			return true;
-		}
-
-		[[nodiscard]] UpstreamDirectory BuildDirectoryTree(const std::set<std::filesystem::path>& InputFiles)
-		{
-			static const std::filesystem::path					RootPath;
-			std::map<std::filesystem::path, UpstreamDirectory*> AllDirectories;
-			UpstreamDirectory									RootDirectory = {.Path = RootPath};
-
-			AllDirectories[RootPath] = &RootDirectory;
-
-			// Build tree from flat list
-			for (const auto& Path : InputFiles)
-			{
-				if (Path.has_parent_path())
-				{
-					if (!AllDirectories.contains(Path.parent_path()))
-					{
-						std::stack<std::string> PathSplit;
-						{
-							std::filesystem::path ParentPath = Path.parent_path();
-							PathSplit.push(ParentPath.filename().string());
-							while (ParentPath.has_parent_path())
-							{
-								ParentPath = ParentPath.parent_path();
-								PathSplit.push(ParentPath.filename().string());
-							}
-						}
-						UpstreamDirectory* ParentPtr = &RootDirectory;
-						while (!PathSplit.empty())
-						{
-							if (!ParentPtr->Directories.contains(PathSplit.top()))
-							{
-								std::filesystem::path NewParentPath		= {ParentPtr->Path / PathSplit.top()};
-								ParentPtr->Directories[PathSplit.top()] = {.Path = NewParentPath};
-								AllDirectories[NewParentPath]			= &ParentPtr->Directories[PathSplit.top()];
-							}
-							ParentPtr = &ParentPtr->Directories[PathSplit.top()];
-							PathSplit.pop();
-						}
-					}
-
-					AllDirectories[Path.parent_path()]->Files.insert(Path.filename().string());
-				}
-				else
-				{
-					RootDirectory.Files.insert(Path.filename().string());
-				}
-			}
-
-			return RootDirectory;
-		}
-
-		[[nodiscard]] CbObject BuildMerkleTreeDirectory(const UpstreamDirectory&					   RootDirectory,
-														const std::map<std::filesystem::path, IoHash>& InputFileHashes,
-														const std::map<IoHash, IoBuffer>&			   Blobs,
-														std::map<IoHash, CbObject>&					   Objects)
-		{
-			CbObjectWriter DirectoryTreeWriter;
-
-			if (!RootDirectory.Files.empty())
-			{
-				DirectoryTreeWriter.BeginArray("f"sv);
-				for (const auto& File : RootDirectory.Files)
-				{
-					const std::filesystem::path FilePath = {RootDirectory.Path / File};
-					const IoHash&				FileHash = InputFileHashes.at(FilePath);
-					const uint64_t				FileSize = Blobs.at(FileHash).Size();
-					DirectoryTreeWriter.BeginObject();
-					DirectoryTreeWriter.AddString("n"sv, File);
-					DirectoryTreeWriter.AddBinaryAttachment("h"sv, FileHash);
-					DirectoryTreeWriter.AddInteger("s"sv, FileSize);  // Size
-					// DirectoryTreeWriter.AddInteger("a"sv, 0);		  // Attributes Currently unneeded
-					DirectoryTreeWriter.EndObject();
-				}
-				DirectoryTreeWriter.EndArray();
-			}
-
-			if (!RootDirectory.Directories.empty())
-			{
-				DirectoryTreeWriter.BeginArray("d"sv);
-				for (const auto& Item : RootDirectory.Directories)
-				{
-					CbObject	 Directory	   = BuildMerkleTreeDirectory(Item.second, InputFileHashes, Blobs, Objects);
-					const IoHash DirectoryHash = Directory.GetHash();
-					Objects[DirectoryHash]	   = std::move(Directory);
-
-					DirectoryTreeWriter.BeginObject();
-					DirectoryTreeWriter.AddString("n"sv, Item.first);
-					DirectoryTreeWriter.AddObjectAttachment("h"sv, DirectoryHash);
-					DirectoryTreeWriter.EndObject();
-				}
-				DirectoryTreeWriter.EndArray();
-			}
-
-			return DirectoryTreeWriter.Save();
-		}
-
-		void ResolveMerkleTreeDirectory(const std::filesystem::path&			 ParentDirectory,
-										const IoHash&							 DirectoryHash,
-										const std::map<IoHash, IoBuffer>&		 Objects,
-										std::map<std::filesystem::path, IoHash>& OutputFiles)
-		{
-			CbObject Directory = LoadCompactBinaryObject(Objects.at(DirectoryHash));
-
-			for (auto& It : Directory["f"sv])
-			{
-				const CbObjectView			FileObject = It.AsObjectView();
-				const std::filesystem::path Path	   = ParentDirectory / FileObject["n"sv].AsString();
-
-				OutputFiles[Path] = FileObject["h"sv].AsBinaryAttachment();
-			}
-
-			for (auto& It : Directory["d"sv])
-			{
-				const CbObjectView DirectoryObject = It.AsObjectView();
-
-				ResolveMerkleTreeDirectory(ParentDirectory / DirectoryObject["n"sv].AsString(),
-										   DirectoryObject["h"sv].AsObjectAttachment(),
-										   Objects,
-										   OutputFiles);
-			}
-		}
-
-		[[nodiscard]] CbObject BuildRequirements(const std::string_view						Condition,
-												 const std::map<std::string_view, int64_t>& Resources,
-												 const bool									Exclusive)
-		{
-			CbObjectWriter Writer;
-			Writer.AddString("c", Condition);
-			if (!Resources.empty())
-			{
-				Writer.BeginArray("r");
-				for (const auto& Resource : Resources)
-				{
-					Writer.BeginArray();
-					Writer.AddString(Resource.first);
-					Writer.AddInteger(Resource.second);
-					Writer.EndArray();
-				}
-				Writer.EndArray();
-			}
-			Writer.AddBool("e", Exclusive);
-			return Writer.Save();
-		}
-
-		[[nodiscard]] CbObject BuildTask(const std::string_view					   Executable,
-										 const std::vector<std::string>&		   Arguments,
-										 const std::map<std::string, std::string>& Environment,
-										 const std::string_view					   WorkingDirectory,
-										 const IoHash&							   SandboxHash,
-										 const IoHash&							   RequirementsId,
-										 const std::set<std::string>&			   Outputs)
-		{
-			CbObjectWriter TaskWriter;
-			TaskWriter.AddString("e"sv, Executable);
-
-			if (!Arguments.empty())
-			{
-				TaskWriter.BeginArray("a"sv);
-				for (const auto& Argument : Arguments)
-				{
-					TaskWriter.AddString(Argument);
-				}
-				TaskWriter.EndArray();
-			}
-
-			if (!Environment.empty())
-			{
-				TaskWriter.BeginArray("v"sv);
-				for (const auto& Env : Environment)
-				{
-					TaskWriter.BeginArray();
-					TaskWriter.AddString(Env.first);
-					TaskWriter.AddString(Env.second);
-					TaskWriter.EndArray();
-				}
-				TaskWriter.EndArray();
-			}
-
-			if (!WorkingDirectory.empty())
-			{
-				TaskWriter.AddString("w"sv, WorkingDirectory);
-			}
-
-			TaskWriter.AddObjectAttachment("s"sv, SandboxHash);
-			TaskWriter.AddObjectAttachment("r"sv, RequirementsId);
-
-			// Outputs
-			if (!Outputs.empty())
-			{
-				TaskWriter.BeginArray("o"sv);
-				for (const auto& Output : Outputs)
-				{
-					TaskWriter.AddString(Output);
-				}
-				TaskWriter.EndArray();
-			}
-
-			return TaskWriter.Save();
-		}
-	};
-}  // namespace detail
-
-//////////////////////////////////////////////////////////////////////////
-
 struct UpstreamApplyStats
 {
 	static constexpr uint64_t MaxSampleCount = 1000ull;
@@ -1360,7 +83,8 @@ public:
 	, m_CasStore(CasStore)
 	, m_CidStore(CidStore)
 	, m_Stats(Options.StatsEnabled)
-	, m_AsyncWorkPool(Options.ThreadCount)
+	, m_UpstreamAsyncWorkPool(Options.UpstreamThreadCount)
+	, m_DownstreamAsyncWorkPool(Options.DownstreamThreadCount)
 	{
 	}
 
@@ -1423,7 +147,9 @@ public:
 				m_ApplyTasks[WorkerId][ActionId] = {.State = UpstreamApplyState::Queued, .Result{}, .ExpireTime = std::move(ExpireTime)};
 			}
 
-			m_AsyncWorkPool.ScheduleWork([this, ApplyRecord = std::move(ApplyRecord)]() { ProcessApplyRecord(std::move(ApplyRecord)); });
+			ApplyRecord.Timepoints["zen-queue-added"] = DateTime::NowTicks();
+			m_UpstreamAsyncWorkPool.ScheduleWork(
+				[this, ApplyRecord = std::move(ApplyRecord)]() { ProcessApplyRecord(std::move(ApplyRecord)); });
 
 			return {.ApplyId = ActionId, .Success = true};
 		}
@@ -1447,8 +173,10 @@ public:
 
 	virtual void GetStatus(CbObjectWriter& Status) override
 	{
-		Status << "worker_threads" << m_Options.ThreadCount;
-		Status << "queue_count" << m_AsyncWorkPool.PendingWork();
+		Status << "upstream_worker_threads" << m_Options.UpstreamThreadCount;
+		Status << "upstream_queue_count" << m_UpstreamAsyncWorkPool.PendingWork();
+		Status << "downstream_worker_threads" << m_Options.DownstreamThreadCount;
+		Status << "downstream_queue_count" << m_DownstreamAsyncWorkPool.PendingWork();
 
 		Status.BeginArray("endpoints");
 		for (const auto& Ep : m_Endpoints)
@@ -1501,11 +229,14 @@ private:
 			{
 				if (Endpoint->IsHealthy())
 				{
-					PostUpstreamApplyResult Result = Endpoint->PostApply(std::move(ApplyRecord));
+					ApplyRecord.Timepoints["zen-queue-dispatched"] = DateTime::NowTicks();
+					PostUpstreamApplyResult Result				   = Endpoint->PostApply(std::move(ApplyRecord));
 					{
 						std::scoped_lock Lock(m_ApplyTasksMutex);
 						if (auto Status = FindStatus(WorkerId, ActionId); Status != nullptr)
 						{
+							Status->Timepoints.merge(Result.Timepoints);
+
 							if (Result.Success)
 							{
 								Status->State = UpstreamApplyState::Executing;
@@ -1553,7 +284,7 @@ private:
 		{
 			if (Endpoint->IsHealthy())
 			{
-				GetUpstreamApplyUpdatesResult Result = Endpoint->GetUpdates(m_AsyncWorkPool);
+				GetUpstreamApplyUpdatesResult Result = Endpoint->GetUpdates(m_DownstreamAsyncWorkPool);
 				m_Stats.Add(*Endpoint, Result);
 
 				if (!Result.Success)
@@ -1572,6 +303,9 @@ private:
 							{
 								Status->State  = UpstreamApplyState::Complete;
 								Status->Result = std::move(It2.second);
+								Status->Result.Timepoints.merge(Status->Timepoints);
+								Status->Result.Timepoints["zen-queue-complete"] = DateTime::NowTicks();
+								Status->Timepoints.clear();
 							}
 						}
 					}
@@ -1686,7 +420,8 @@ private:
 	std::mutex											m_ApplyTasksMutex;
 	std::vector<std::unique_ptr<UpstreamApplyEndpoint>> m_Endpoints;
 	Event												m_ShutdownEvent;
-	WorkerThreadPool									m_AsyncWorkPool;
+	WorkerThreadPool									m_UpstreamAsyncWorkPool;
+	WorkerThreadPool									m_DownstreamAsyncWorkPool;
 	std::thread											m_UpstreamUpdatesThread;
 	std::thread											m_EndpointMonitorThread;
 	RunState											m_RunState;
@@ -1700,24 +435,6 @@ UpstreamApply::Create(const UpstreamApplyOptions& Options, CasStore& CasStore, C
 	return std::make_unique<UpstreamApplyImpl>(Options, CasStore, CidStore);
 }
 
-std::unique_ptr<UpstreamApplyEndpoint>
-UpstreamApplyEndpoint::CreateHordeEndpoint(const CloudCacheClientOptions& ComputeOptions,
-										   const UpstreamAuthConfig&	  ComputeAuthConfig,
-										   const CloudCacheClientOptions& StorageOptions,
-										   const UpstreamAuthConfig&	  StorageAuthConfig,
-										   CasStore&					  CasStore,
-										   CidStore&					  CidStore,
-										   AuthMgr&						  Mgr)
-{
-	return std::make_unique<detail::HordeUpstreamApplyEndpoint>(ComputeOptions,
-																ComputeAuthConfig,
-																StorageOptions,
-																StorageAuthConfig,
-																CasStore,
-																CidStore,
-																Mgr);
-}
-
 }  // namespace zen
 
 #endif	// ZEN_WITH_COMPUTE_SERVICES