subprocess tracking using Jobs on Windows/hub (#796)HEADmain

This change introduces job object support on Windows to be able to more accurately track and limit resource usage on storage instances created by the hub service. It also ensures that all child instances can be torn down reliably on exit.

Also made it so hub tests no longer pop up console windows while running.

6 files changed, 199 insertions, 8 deletions

diff --git a/src/zencore/include/zencore/process.h b/src/zencore/include/zencore/process.h
index c51163a68..809312c7b 100644
--- a/src/zencore/include/zencore/process.h
+++ b/src/zencore/include/zencore/process.h
@@ -9,6 +9,10 @@
 
 namespace zen {
 
+#if ZEN_PLATFORM_WINDOWS
+class JobObject;
+#endif
+
 /** Basic process abstraction
  */
 class ProcessHandle
@@ -46,6 +50,7 @@ private:
 
 /** Basic process creation
  */
+
 struct CreateProcOptions
 {
 	enum
@@ -63,6 +68,9 @@ struct CreateProcOptions
 	const std::filesystem::path* WorkingDirectory = nullptr;
 	uint32_t					 Flags			  = 0;
 	std::filesystem::path		 StdoutFile;
+#if ZEN_PLATFORM_WINDOWS
+	JobObject* AssignToJob = nullptr;  // When set, the process is created suspended, assigned to the job, then resumed
+#endif
 };
 
 #if ZEN_PLATFORM_WINDOWS
@@ -99,6 +107,31 @@ private:
 	std::vector<HandleType> m_ProcessHandles;
 };
 
+#if ZEN_PLATFORM_WINDOWS
+/** Windows Job Object wrapper
+ *
+ * When configured with JOB_OBJECT_LIMIT_KILL_ON_JOB_CLOSE, the OS will
+ * terminate all assigned child processes when the job handle is closed
+ * (including abnormal termination of the owning process). This provides
+ * an OS-level guarantee against orphaned child processes.
+ */
+class JobObject
+{
+public:
+	JobObject();
+	~JobObject();
+	JobObject(const JobObject&) = delete;
+	JobObject& operator=(const JobObject&) = delete;
+
+	void			   Initialize();
+	bool			   AssignProcess(void* ProcessHandle);
+	[[nodiscard]] bool IsValid() const;
+
+private:
+	void* m_JobHandle = nullptr;
+};
+#endif	// ZEN_PLATFORM_WINDOWS
+
 bool IsProcessRunning(int pid);
 bool IsProcessRunning(int pid, std::error_code& OutEc);
 int	 GetCurrentProcessId();
diff --git a/src/zencore/process.cpp b/src/zencore/process.cpp
index 4a2668912..226a94050 100644
--- a/src/zencore/process.cpp
+++ b/src/zencore/process.cpp
@@ -490,6 +490,8 @@ CreateProcNormal(const std::filesystem::path& Executable, std::string_view Comma
 	LPSECURITY_ATTRIBUTES ProcessAttributes = nullptr;
 	LPSECURITY_ATTRIBUTES ThreadAttributes	= nullptr;
 
+	const bool AssignToJob = Options.AssignToJob && Options.AssignToJob->IsValid();
+
 	DWORD CreationFlags = 0;
 	if (Options.Flags & CreateProcOptions::Flag_NewConsole)
 	{
@@ -503,6 +505,10 @@ CreateProcNormal(const std::filesystem::path& Executable, std::string_view Comma
 	{
 		CreationFlags |= CREATE_NEW_PROCESS_GROUP;
 	}
+	if (AssignToJob)
+	{
+		CreationFlags |= CREATE_SUSPENDED;
+	}
 
 	const wchar_t* WorkingDir = nullptr;
 	if (Options.WorkingDirectory != nullptr)
@@ -571,6 +577,15 @@ CreateProcNormal(const std::filesystem::path& Executable, std::string_view Comma
 		return nullptr;
 	}
 
+	if (AssignToJob)
+	{
+		if (!Options.AssignToJob->AssignProcess(ProcessInfo.hProcess))
+		{
+			ZEN_WARN("Failed to assign newly created process to job object");
+		}
+		ResumeThread(ProcessInfo.hThread);
+	}
+
 	CloseHandle(ProcessInfo.hThread);
 	return ProcessInfo.hProcess;
 }
@@ -644,6 +659,8 @@ CreateProcUnelevated(const std::filesystem::path& Executable, std::string_view C
 	};
 	PROCESS_INFORMATION ProcessInfo = {};
 
+	const bool AssignToJob = Options.AssignToJob && Options.AssignToJob->IsValid();
+
 	if (Options.Flags & CreateProcOptions::Flag_NewConsole)
 	{
 		CreateProcFlags |= CREATE_NEW_CONSOLE;
@@ -652,6 +669,10 @@ CreateProcUnelevated(const std::filesystem::path& Executable, std::string_view C
 	{
 		CreateProcFlags |= CREATE_NO_WINDOW;
 	}
+	if (AssignToJob)
+	{
+		CreateProcFlags |= CREATE_SUSPENDED;
+	}
 
 	ExtendableWideStringBuilder<256> CommandLineZ;
 	CommandLineZ << CommandLine;
@@ -679,6 +700,15 @@ CreateProcUnelevated(const std::filesystem::path& Executable, std::string_view C
 		return nullptr;
 	}
 
+	if (AssignToJob)
+	{
+		if (!Options.AssignToJob->AssignProcess(ProcessInfo.hProcess))
+		{
+			ZEN_WARN("Failed to assign newly created process to job object");
+		}
+		ResumeThread(ProcessInfo.hThread);
+	}
+
 	CloseHandle(ProcessInfo.hThread);
 	return ProcessInfo.hProcess;
 }
@@ -845,6 +875,65 @@ ProcessMonitor::IsActive() const
 
 //////////////////////////////////////////////////////////////////////////
 
+#if ZEN_PLATFORM_WINDOWS
+JobObject::JobObject() = default;
+
+JobObject::~JobObject()
+{
+	if (m_JobHandle)
+	{
+		CloseHandle(m_JobHandle);
+		m_JobHandle = nullptr;
+	}
+}
+
+void
+JobObject::Initialize()
+{
+	ZEN_ASSERT(m_JobHandle == nullptr, "JobObject already initialized");
+
+	m_JobHandle = CreateJobObjectW(nullptr, nullptr);
+	if (!m_JobHandle)
+	{
+		ZEN_WARN("Failed to create job object: {}", zen::GetLastError());
+		return;
+	}
+
+	JOBOBJECT_EXTENDED_LIMIT_INFORMATION LimitInfo = {};
+	LimitInfo.BasicLimitInformation.LimitFlags	   = JOB_OBJECT_LIMIT_KILL_ON_JOB_CLOSE;
+
+	if (!SetInformationJobObject(m_JobHandle, JobObjectExtendedLimitInformation, &LimitInfo, sizeof(LimitInfo)))
+	{
+		ZEN_WARN("Failed to set job object limits: {}", zen::GetLastError());
+		CloseHandle(m_JobHandle);
+		m_JobHandle = nullptr;
+	}
+}
+
+bool
+JobObject::AssignProcess(void* ProcessHandle)
+{
+	ZEN_ASSERT(m_JobHandle != nullptr, "JobObject not initialized");
+	ZEN_ASSERT(ProcessHandle != nullptr, "ProcessHandle is null");
+
+	if (!AssignProcessToJobObject(m_JobHandle, ProcessHandle))
+	{
+		ZEN_WARN("Failed to assign process to job object: {}", zen::GetLastError());
+		return false;
+	}
+
+	return true;
+}
+
+bool
+JobObject::IsValid() const
+{
+	return m_JobHandle != nullptr;
+}
+#endif	// ZEN_PLATFORM_WINDOWS
+
+//////////////////////////////////////////////////////////////////////////
+
 bool
 IsProcessRunning(int pid, std::error_code& OutEc)
 {
diff --git a/src/zenserver/hub/hubservice.cpp b/src/zenserver/hub/hubservice.cpp
index a00446a75..bf0e294c5 100644
--- a/src/zenserver/hub/hubservice.cpp
+++ b/src/zenserver/hub/hubservice.cpp
@@ -8,6 +8,7 @@
 #include <zencore/filesystem.h>
 #include <zencore/fmtutils.h>
 #include <zencore/logging.h>
+#include <zencore/process.h>
 #include <zencore/scopeguard.h>
 #include <zencore/system.h>
 #include <zenutil/zenserverprocess.h>
@@ -150,6 +151,10 @@ struct StorageServerInstance
 
 	inline uint16_t GetBasePort() const { return m_ServerInstance.GetBasePort(); }
 
+#if ZEN_PLATFORM_WINDOWS
+	void SetJobObject(JobObject* InJobObject) { m_JobObject = InJobObject; }
+#endif
+
 private:
 	void				  WakeLocked();
 	RwLock				  m_Lock;
@@ -161,6 +166,9 @@ private:
 	std::filesystem::path m_TempDir;
 	std::filesystem::path m_HydrationPath;
 	ResourceMetrics		  m_ResourceMetrics;
+#if ZEN_PLATFORM_WINDOWS
+	JobObject* m_JobObject = nullptr;
+#endif
 
 	void SpawnServerProcess();
 
@@ -191,6 +199,9 @@ StorageServerInstance::SpawnServerProcess()
 
 	m_ServerInstance.SetServerExecutablePath(GetRunningExecutablePath());
 	m_ServerInstance.SetDataDir(m_BaseDir);
+#if ZEN_PLATFORM_WINDOWS
+	m_ServerInstance.SetJobObject(m_JobObject);
+#endif
 	const uint16_t BasePort = m_ServerInstance.SpawnServerAndWaitUntilReady();
 
 	ZEN_DEBUG("Storage server instance for module '{}' started, listening on port {}", m_ModuleId, BasePort);
@@ -380,6 +391,21 @@ struct HttpHubService::Impl
 		// flexibility, and to allow running multiple hubs on the same host if
 		// necessary.
 		m_RunEnvironment.SetNextPortNumber(21000);
+
+#if ZEN_PLATFORM_WINDOWS
+		if (m_UseJobObject)
+		{
+			m_JobObject.Initialize();
+			if (m_JobObject.IsValid())
+			{
+				ZEN_INFO("Job object initialized for hub service child process management");
+			}
+			else
+			{
+				ZEN_WARN("Failed to initialize job object; child processes will not be auto-terminated on hub crash");
+			}
+		}
+#endif
 	}
 
 	void Cleanup()
@@ -422,6 +448,12 @@ struct HttpHubService::Impl
 				IsNewInstance = true;
 				auto NewInstance =
 					std::make_unique<StorageServerInstance>(m_RunEnvironment, ModuleId, m_FileHydrationPath, m_HydrationTempPath);
+#if ZEN_PLATFORM_WINDOWS
+				if (m_JobObject.IsValid())
+				{
+					NewInstance->SetJobObject(&m_JobObject);
+				}
+#endif
 				Instance = NewInstance.get();
 				m_Instances.emplace(std::string(ModuleId), std::move(NewInstance));
 
@@ -579,10 +611,15 @@ struct HttpHubService::Impl
 	inline int GetInstanceLimit() { return m_InstanceLimit; }
 	inline int GetMaxInstanceCount() { return m_MaxInstanceCount; }
 
+	bool m_UseJobObject = true;
+
 private:
-	ZenServerEnvironment													m_RunEnvironment;
-	std::filesystem::path													m_FileHydrationPath;
-	std::filesystem::path													m_HydrationTempPath;
+	ZenServerEnvironment  m_RunEnvironment;
+	std::filesystem::path m_FileHydrationPath;
+	std::filesystem::path m_HydrationTempPath;
+#if ZEN_PLATFORM_WINDOWS
+	JobObject m_JobObject;
+#endif
 	RwLock																	m_Lock;
 	std::unordered_map<std::string, std::unique_ptr<StorageServerInstance>> m_Instances;
 	std::unordered_set<std::string>											m_DeprovisioningModules;
@@ -817,6 +854,12 @@ HttpHubService::~HttpHubService()
 {
 }
 
+void
+HttpHubService::SetUseJobObject(bool Enable)
+{
+	m_Impl->m_UseJobObject = Enable;
+}
+
 const char*
 HttpHubService::BaseUri() const
 {
diff --git a/src/zenserver/hub/hubservice.h b/src/zenserver/hub/hubservice.h
index 1a5a8c57c..ef24bba69 100644
--- a/src/zenserver/hub/hubservice.h
+++ b/src/zenserver/hub/hubservice.h
@@ -28,6 +28,13 @@ public:
 
 	void SetNotificationEndpoint(std::string_view UpstreamNotificationEndpoint, std::string_view InstanceId);
 
+	/** Enable or disable the use of a Windows Job Object for child process management.
+	 *  When enabled, all spawned child processes are assigned to a job object with
+	 *  JOB_OBJECT_LIMIT_KILL_ON_JOB_CLOSE, ensuring children are terminated if the hub
+	 *  crashes or is force-killed. Must be called before Initialize(). No-op on non-Windows.
+	 */
+	void SetUseJobObject(bool Enable);
+
 private:
 	HttpRequestRouter m_Router;
 
diff --git a/src/zenutil/include/zenutil/zenserverprocess.h b/src/zenutil/include/zenutil/zenserverprocess.h
index b781a03a9..954916fe2 100644
--- a/src/zenutil/include/zenutil/zenserverprocess.h
+++ b/src/zenutil/include/zenutil/zenserverprocess.h
@@ -97,9 +97,12 @@ struct ZenServerInstance
 	inline int		   GetPid() const { return m_Process.Pid(); }
 	inline void		   SetOwnerPid(int Pid) { m_OwnerPid = Pid; }
 	void*			   GetProcessHandle() const { return m_Process.Handle(); }
-	bool			   IsRunning();
-	bool			   Terminate();
-	std::string		   GetLogOutput() const;
+#if ZEN_PLATFORM_WINDOWS
+	void SetJobObject(JobObject* Job) { m_JobObject = Job; }
+#endif
+	bool		IsRunning();
+	bool		Terminate();
+	std::string GetLogOutput() const;
 
 	inline ServerMode GetServerMode() const { return m_ServerMode; }
 
@@ -148,6 +151,9 @@ private:
 	std::string					m_Name;
 	std::filesystem::path		m_OutputCapturePath;
 	std::filesystem::path		m_ServerExecutablePath;
+#if ZEN_PLATFORM_WINDOWS
+	JobObject* m_JobObject = nullptr;
+#endif
 
 	void CreateShutdownEvent(int BasePort);
 	void SpawnServer(int BasePort, std::string_view AdditionalServerArgs, int WaitTimeoutMs);
diff --git a/src/zenutil/zenserverprocess.cpp b/src/zenutil/zenserverprocess.cpp
index 579ba450a..0f8ab223d 100644
--- a/src/zenutil/zenserverprocess.cpp
+++ b/src/zenutil/zenserverprocess.cpp
@@ -831,8 +831,15 @@ ZenServerInstance::SpawnServerInternal(int ChildId, std::string_view ServerArgs,
 		m_ServerExecutablePath.empty() ? (BaseDir / "zenserver" ZEN_EXE_SUFFIX_LITERAL) : m_ServerExecutablePath;
 	const std::filesystem::path OutputPath =
 		OpenConsole ? std::filesystem::path{} : std::filesystem::temp_directory_path() / ("zenserver_" + m_Name + ".log");
-	CreateProcOptions CreateOptions = {.WorkingDirectory = &CurrentDirectory, .Flags = CreationFlags, .StdoutFile = OutputPath};
-	CreateProcResult  ChildPid		= CreateProc(Executable, CommandLine.ToView(), CreateOptions);
+	CreateProcOptions CreateOptions = {
+		.WorkingDirectory = &CurrentDirectory,
+		.Flags			  = CreationFlags,
+		.StdoutFile		  = OutputPath,
+#if ZEN_PLATFORM_WINDOWS
+		.AssignToJob = m_JobObject,
+#endif
+	};
+	CreateProcResult ChildPid = CreateProc(Executable, CommandLine.ToView(), CreateOptions);
 #if ZEN_PLATFORM_WINDOWS
 	if (!ChildPid)
 	{
@@ -841,6 +848,12 @@ ZenServerInstance::SpawnServerInternal(int ChildId, std::string_view ServerArgs,
 		{
 			ZEN_DEBUG("Regular spawn failed - spawning elevated server");
 			CreateOptions.Flags |= CreateProcOptions::Flag_Elevated;
+			// ShellExecuteEx (used by the elevated path) does not support job object assignment
+			if (CreateOptions.AssignToJob)
+			{
+				ZEN_WARN("Elevated process spawn does not support job object assignment; child will not be auto-terminated on parent exit");
+				CreateOptions.AssignToJob = nullptr;
+			}
 			ChildPid = CreateProc(Executable, CommandLine.ToView(), CreateOptions);
 		}
 		else