zen hub command (#877)

- Feature: Added `zen hub` command for managing a hub server and its provisioned module instances:
  - `zen hub up` - Start a hub server (equivalent to `zen up` in hub mode)
  - `zen hub down` - Shut down a hub server
  - `zen hub provision <moduleid>` - Provision a storage server instance for a module
  - `zen hub deprovision <moduleid>` - Deprovision a storage server instance
  - `zen hub hibernate <moduleid>` - Hibernate a provisioned instance (shut down, data preserved)
  - `zen hub wake <moduleid>` - Wake a hibernated instance
  - `zen hub status [moduleid]` - Show state of all instances or a specific module
- Feature: Added new hub HTTP endpoints for instance lifecycle management:
  - `POST /hub/modules/{moduleid}/hibernate` - Hibernate the instance for the given module
  - `POST /hub/modules/{moduleid}/wake` - Wake a hibernated instance for the given module
- Improvement: `zen up` refactored to use shared `StartupZenServer`/`ShutdownZenServer` helpers (also used by `zen hub up`/`zen hub down`)
- Bugfix: Fixed shutdown event not being cleared after the server process exits in `ZenServerInstance::Shutdown()`, which could cause stale state on reuse

1 files changed, 440 insertions, 0 deletions

diff --git a/src/zen/cmds/hub_cmd.cpp b/src/zen/cmds/hub_cmd.cpp
new file mode 100644
index 000000000..5bdd3a922
--- /dev/null
+++ b/src/zen/cmds/hub_cmd.cpp
@@ -0,0 +1,440 @@
+// Copyright Epic Games, Inc. All Rights Reserved.
+
+#include "hub_cmd.h"
+
+#include <zencore/compactbinary.h>
+#include <zencore/compactbinaryutil.h>
+#include <zencore/filesystem.h>
+#include <zencore/fmtutils.h>
+#include <zencore/iobuffer.h>
+#include <zencore/logging.h>
+#include <zencore/process.h>
+#include <zenhttp/httpclient.h>
+#include <zenutil/zenserverprocess.h>
+
+#include <vector>
+
+namespace zen {
+
+//////////////////////////////////////////////////////////////////////////
+// HubUpSubCmd
+
+HubUpSubCmd::HubUpSubCmd() : ZenSubCmdBase("up", "Bring hub server up")
+{
+	SubOptions().add_option("", "p", "port", "Host port", cxxopts::value(m_Port)->default_value("0"), "<hostport>");
+	SubOptions().add_option("", "b", "base-dir", "Parent folder of server executable", cxxopts::value(m_ProgramBaseDir), "<directory>");
+	SubOptions().add_option("", "c", "show-console", "Open a console window for the zenserver process", cxxopts::value(m_ShowConsole), "");
+	SubOptions().add_option("",
+							"l",
+							"show-log",
+							"Show the output log of the zenserver process after successful start",
+							cxxopts::value(m_ShowLog),
+							"");
+}
+
+void
+HubUpSubCmd::Run(const ZenCliOptions& GlobalOptions)
+{
+	if (m_ShowConsole && m_ShowLog)
+	{
+		throw OptionParseException("'--show-console' conflicts with '--show-log'", SubOptions().help());
+	}
+
+	std::optional<int> StartResult = StartupZenServer(ConsoleLog(),
+													  {.ProgramBaseDir = m_ProgramBaseDir,
+													   .Port		   = m_Port,
+													   .OpenConsole	   = m_ShowConsole,
+													   .ShowLog		   = m_ShowLog,
+													   .ExtraArgs	   = GlobalOptions.PassthroughCommandLine,
+													   .Mode		   = ZenServerInstance::ServerMode::kHubServer});
+	if (!StartResult.has_value())
+	{
+		ZEN_CONSOLE("Zen server already running");
+		return;
+	}
+	if (*StartResult != 0)
+	{
+		throw ErrorWithReturnCode("Zen server failed to start", *StartResult);
+	}
+
+	ZEN_CONSOLE("Zen server up");
+}
+
+//////////////////////////////////////////////////////////////////////////
+// HubDownSubCmd
+
+HubDownSubCmd::HubDownSubCmd() : ZenSubCmdBase("down", "Bring hub server down")
+{
+	SubOptions().add_option("", "p", "port", "Host port", cxxopts::value(m_Port)->default_value("0"), "<hostport>");
+	SubOptions().add_option("", "a", "all", "Shut down all running zen server instances", cxxopts::value(m_All), "");
+	SubOptions().add_option("", "f", "force", "Force terminate if graceful shutdown fails", cxxopts::value(m_ForceTerminate), "<force>");
+	SubOptions().add_option("", "b", "base-dir", "Parent folder of server executable", cxxopts::value(m_ProgramBaseDir), "<directory>");
+	SubOptions()
+		.add_option("", "", "data-dir", "Path to data directory to inspect for running server", cxxopts::value(m_DataDir), "<file>");
+}
+
+void
+HubDownSubCmd::Run(const ZenCliOptions& GlobalOptions)
+{
+	ZEN_UNUSED(GlobalOptions);
+
+	if (m_ProgramBaseDir.empty())
+	{
+		m_ProgramBaseDir = GetRunningExecutablePath().parent_path();
+	}
+
+	ZenServerState Instance;
+	Instance.Initialize();
+
+	if (m_All)
+	{
+		struct EntryInfo
+		{
+			uint16_t Port = 0;
+			uint32_t Pid  = 0;
+		};
+		std::vector<EntryInfo> Entries;
+		Instance.Snapshot([&Entries](const ZenServerState::ZenServerEntry& Entry) {
+			uint16_t Port = Entry.DesiredListenPort.load();
+			uint32_t Pid  = Entry.Pid.load();
+			if (Port != 0 && Pid != 0)
+			{
+				Entries.push_back({Port, Pid});
+			}
+		});
+
+		if (Entries.empty())
+		{
+			ZEN_CONSOLE("No zen server instances to bring down");
+			return;
+		}
+
+		int FailCount = 0;
+		for (const EntryInfo& Info : Entries)
+		{
+			Instance.Sweep();
+			ZenServerState::ZenServerEntry* Entry = Instance.Lookup(Info.Port);
+			if (Entry && Entry->Pid.load() == Info.Pid)
+			{
+				if (!ShutdownZenServer(ConsoleLog(), Instance, Entry, m_ProgramBaseDir))
+				{
+					ZEN_CONSOLE_WARN("Failed to shutdown server on port {} (pid {})", Info.Port, Info.Pid);
+					++FailCount;
+				}
+			}
+		}
+
+		if (FailCount > 0 && !m_ForceTerminate)
+		{
+			throw std::runtime_error(fmt::format("Failed to shutdown {} instance(s), use --force to hard terminate", FailCount));
+		}
+		return;
+	}
+
+	ZenServerState::ZenServerEntry* Entry = Instance.Lookup(m_Port);
+
+	if (!m_DataDir.empty())
+	{
+		if (!IsFile(m_DataDir / ".lock"))
+		{
+			throw std::runtime_error(fmt::format("Lock file does not exist in directory '{}'", m_DataDir));
+		}
+		CbValidateError ValidateResult = CbValidateError::None;
+		if (CbObject LockFileObject =
+				ValidateAndReadCompactBinaryObject(IoBufferBuilder::MakeFromFile(m_DataDir / ".lock"), ValidateResult);
+			ValidateResult == CbValidateError::None && LockFileObject)
+		{
+			LockFileInfo Info = ReadLockFilePayload(LockFileObject);
+			std::string	 Reason;
+			if (!ValidateLockFileInfo(Info, Reason))
+			{
+				throw std::runtime_error(fmt::format("Lock file in directory '{}' is not valid. Reason: '{}'", m_DataDir, Reason));
+			}
+			Entry = Instance.LookupByEffectivePort(Info.EffectiveListenPort);
+		}
+		else
+		{
+			throw std::runtime_error(
+				fmt::format("Lock file in directory '{}' is malformed. Reason: '{}'", m_DataDir, ToString(ValidateResult)));
+		}
+	}
+
+	if (Entry)
+	{
+		if (ShutdownZenServer(ConsoleLog(), Instance, Entry, m_ProgramBaseDir))
+		{
+			return;
+		}
+	}
+
+	if (m_ForceTerminate)
+	{
+		std::filesystem::path ServerExePath = m_ProgramBaseDir / "zenserver" ZEN_EXE_SUFFIX_LITERAL;
+		ProcessHandle		  RunningProcess;
+		if (std::error_code Ec = FindProcess(ServerExePath, RunningProcess, /*IncludeSelf*/ false); !Ec)
+		{
+			ZEN_CONSOLE_WARN("Attempting hard terminate of zen process with pid ({})", RunningProcess.Pid());
+			if (RunningProcess.Terminate(0))
+			{
+				ZEN_CONSOLE("Terminate complete");
+				return;
+			}
+			throw std::runtime_error("Failed to terminate server, still running");
+		}
+		else
+		{
+			ZEN_CONSOLE_WARN("Failed to find process '{}', reason: {}", ServerExePath.string(), Ec.message());
+		}
+	}
+	else if (Entry)
+	{
+		throw std::runtime_error(
+			fmt::format("Failed to shutdown server on port {}, use --force to hard terminate process", Entry->DesiredListenPort.load()));
+	}
+
+	ZEN_CONSOLE("No zen server to bring down");
+}
+
+//////////////////////////////////////////////////////////////////////////
+// HubProvisionSubCmd
+
+HubProvisionSubCmd::HubProvisionSubCmd() : ZenSubCmdBase("provision", "Provision a hub module instance")
+{
+	SubOptions().add_option("", "u", "hosturl", ZenCmdBase::kHostUrlHelp, cxxopts::value(m_HostName)->default_value(""), "<hosturl>");
+	SubOptions().add_option("", "", "moduleid", "Module ID to provision", cxxopts::value(m_ModuleId)->default_value(""), "<moduleid>");
+	SubOptions().parse_positional({"moduleid"});
+}
+
+void
+HubProvisionSubCmd::Run(const ZenCliOptions& GlobalOptions)
+{
+	ZEN_UNUSED(GlobalOptions);
+
+	m_HostName = ZenCmdBase::ResolveTargetHostSpec(m_HostName);
+	if (m_HostName.empty())
+	{
+		throw OptionParseException("Unable to resolve hub host specification", SubOptions().help());
+	}
+	if (m_ModuleId.empty())
+	{
+		throw OptionParseException("moduleid is required", SubOptions().help());
+	}
+
+	HttpClient Http = ZenCmdBase::CreateHttpClient(m_HostName);
+	if (HttpClient::Response Resp =
+			Http.Post(fmt::format("/hub/modules/{}/provision", m_ModuleId), HttpClient::KeyValueMap{}, HttpClient::KeyValueMap{}))
+	{
+		CbObject		 Obj  = Resp.AsObject();
+		std::string_view Id	  = Obj["moduleId"].AsString();
+		std::string_view Uri  = Obj["baseUri"].AsString();
+		uint16_t		 Port = Obj["port"].AsUInt16();
+		ZEN_CONSOLE("module '{}' provisioned: {} (port {})", Id, Uri, Port);
+	}
+	else
+	{
+		Resp.ThrowError("Provision failed");
+	}
+}
+
+//////////////////////////////////////////////////////////////////////////
+// HubDeprovisionSubCmd
+
+HubDeprovisionSubCmd::HubDeprovisionSubCmd() : ZenSubCmdBase("deprovision", "Deprovision a hub module instance")
+{
+	SubOptions().add_option("", "u", "hosturl", ZenCmdBase::kHostUrlHelp, cxxopts::value(m_HostName)->default_value(""), "<hosturl>");
+	SubOptions().add_option("", "", "moduleid", "Module ID to deprovision", cxxopts::value(m_ModuleId)->default_value(""), "<moduleid>");
+	SubOptions().parse_positional({"moduleid"});
+}
+
+void
+HubDeprovisionSubCmd::Run(const ZenCliOptions& GlobalOptions)
+{
+	ZEN_UNUSED(GlobalOptions);
+
+	m_HostName = ZenCmdBase::ResolveTargetHostSpec(m_HostName);
+	if (m_HostName.empty())
+	{
+		throw OptionParseException("Unable to resolve hub host specification", SubOptions().help());
+	}
+	if (m_ModuleId.empty())
+	{
+		throw OptionParseException("moduleid is required", SubOptions().help());
+	}
+
+	HttpClient Http = ZenCmdBase::CreateHttpClient(m_HostName);
+	if (HttpClient::Response Resp =
+			Http.Post(fmt::format("/hub/modules/{}/deprovision", m_ModuleId), HttpClient::KeyValueMap{}, HttpClient::KeyValueMap{}))
+	{
+		CbObject		 Obj = Resp.AsObject();
+		std::string_view Id	 = Obj["moduleId"].AsString();
+		ZEN_CONSOLE("module '{}' deprovisioned", Id);
+	}
+	else
+	{
+		Resp.ThrowError("Deprovision failed");
+	}
+}
+
+//////////////////////////////////////////////////////////////////////////
+// HubHibernateSubCmd
+
+HubHibernateSubCmd::HubHibernateSubCmd() : ZenSubCmdBase("hibernate", "Hibernate a hub module instance")
+{
+	SubOptions().add_option("", "u", "hosturl", ZenCmdBase::kHostUrlHelp, cxxopts::value(m_HostName)->default_value(""), "<hosturl>");
+	SubOptions().add_option("", "", "moduleid", "Module ID to hibernate", cxxopts::value(m_ModuleId)->default_value(""), "<moduleid>");
+	SubOptions().parse_positional({"moduleid"});
+}
+
+void
+HubHibernateSubCmd::Run(const ZenCliOptions& GlobalOptions)
+{
+	ZEN_UNUSED(GlobalOptions);
+
+	m_HostName = ZenCmdBase::ResolveTargetHostSpec(m_HostName);
+	if (m_HostName.empty())
+	{
+		throw OptionParseException("Unable to resolve hub host specification", SubOptions().help());
+	}
+	if (m_ModuleId.empty())
+	{
+		throw OptionParseException("moduleid is required", SubOptions().help());
+	}
+
+	HttpClient Http = ZenCmdBase::CreateHttpClient(m_HostName);
+	if (HttpClient::Response Resp =
+			Http.Post(fmt::format("/hub/modules/{}/hibernate", m_ModuleId), HttpClient::KeyValueMap{}, HttpClient::KeyValueMap{}))
+	{
+		CbObject		 Obj = Resp.AsObject();
+		std::string_view Id	 = Obj["moduleId"].AsString();
+		ZEN_CONSOLE("module '{}' hibernated", Id);
+	}
+	else
+	{
+		Resp.ThrowError("Hibernate failed");
+	}
+}
+
+//////////////////////////////////////////////////////////////////////////
+// HubWakeSubCmd
+
+HubWakeSubCmd::HubWakeSubCmd() : ZenSubCmdBase("wake", "Wake a hibernated hub module instance")
+{
+	SubOptions().add_option("", "u", "hosturl", ZenCmdBase::kHostUrlHelp, cxxopts::value(m_HostName)->default_value(""), "<hosturl>");
+	SubOptions().add_option("", "", "moduleid", "Module ID to wake", cxxopts::value(m_ModuleId)->default_value(""), "<moduleid>");
+	SubOptions().parse_positional({"moduleid"});
+}
+
+void
+HubWakeSubCmd::Run(const ZenCliOptions& GlobalOptions)
+{
+	ZEN_UNUSED(GlobalOptions);
+
+	m_HostName = ZenCmdBase::ResolveTargetHostSpec(m_HostName);
+	if (m_HostName.empty())
+	{
+		throw OptionParseException("Unable to resolve hub host specification", SubOptions().help());
+	}
+	if (m_ModuleId.empty())
+	{
+		throw OptionParseException("moduleid is required", SubOptions().help());
+	}
+
+	HttpClient Http = ZenCmdBase::CreateHttpClient(m_HostName);
+	if (HttpClient::Response Resp =
+			Http.Post(fmt::format("/hub/modules/{}/wake", m_ModuleId), HttpClient::KeyValueMap{}, HttpClient::KeyValueMap{}))
+	{
+		CbObject		 Obj = Resp.AsObject();
+		std::string_view Id	 = Obj["moduleId"].AsString();
+		ZEN_CONSOLE("module '{}' woken", Id);
+	}
+	else
+	{
+		Resp.ThrowError("Wake failed");
+	}
+}
+
+//////////////////////////////////////////////////////////////////////////
+// HubStatusSubCmd
+
+HubStatusSubCmd::HubStatusSubCmd() : ZenSubCmdBase("status", "Show status of hub module instances")
+{
+	SubOptions().add_option("", "u", "hosturl", ZenCmdBase::kHostUrlHelp, cxxopts::value(m_HostName)->default_value(""), "<hosturl>");
+	SubOptions()
+		.add_option("", "", "moduleid", "Module ID (omit to list all)", cxxopts::value(m_ModuleId)->default_value(""), "<moduleid>");
+	SubOptions().parse_positional({"moduleid"});
+}
+
+void
+HubStatusSubCmd::Run(const ZenCliOptions& GlobalOptions)
+{
+	ZEN_UNUSED(GlobalOptions);
+
+	m_HostName = ZenCmdBase::ResolveTargetHostSpec(m_HostName);
+	if (m_HostName.empty())
+	{
+		throw OptionParseException("Unable to resolve hub host specification", SubOptions().help());
+	}
+
+	HttpClient Http = ZenCmdBase::CreateHttpClient(m_HostName);
+
+	if (!m_ModuleId.empty())
+	{
+		if (HttpClient::Response Resp = Http.Get(fmt::format("/hub/modules/{}", m_ModuleId)))
+		{
+			CbObject		 Obj   = Resp.AsObject();
+			std::string_view Id	   = Obj["moduleId"].AsString();
+			std::string_view State = Obj["state"].AsString();
+			ZEN_CONSOLE("module '{}': {}", Id, State);
+		}
+		else
+		{
+			Resp.ThrowError("Status query failed");
+		}
+	}
+	else
+	{
+		if (HttpClient::Response Resp = Http.Get("/hub/status"))
+		{
+			CbObject	Obj		= Resp.AsObject();
+			CbArrayView Modules = Obj["modules"].AsArrayView();
+			if (Modules.Num() == 0)
+			{
+				ZEN_CONSOLE("No modules");
+			}
+			else
+			{
+				for (CbFieldView Module : Modules)
+				{
+					CbObjectView ModObj = Module.AsObjectView();
+					ZEN_CONSOLE("module '{}': {}", ModObj["moduleId"].AsString(), ModObj["state"].AsString());
+				}
+			}
+		}
+		else
+		{
+			Resp.ThrowError("Status query failed");
+		}
+	}
+}
+
+//////////////////////////////////////////////////////////////////////////
+// HubCommand
+
+HubCommand::HubCommand()
+{
+	m_Options.add_options()("h,help", "Print help");
+	m_Options.add_option("__hidden__", "", "subcommand", "", cxxopts::value<std::string>(m_SubCommand)->default_value(""), "");
+	m_Options.parse_positional({"subcommand"});
+
+	AddSubCommand(m_UpSubCmd);
+	AddSubCommand(m_DownSubCmd);
+	AddSubCommand(m_ProvisionSubCmd);
+	AddSubCommand(m_DeprovisionSubCmd);
+	AddSubCommand(m_HibernateSubCmd);
+	AddSubCommand(m_WakeSubCmd);
+	AddSubCommand(m_StatusSubCmd);
+}
+
+HubCommand::~HubCommand() = default;
+
+}  // namespace zen