Compute batching (#849)

### Compute Batch Submission
- Consolidate duplicated action submission logic in `httpcomputeservice` into a single `HandleSubmitAction` supporting both single-action and batch (actions array) payloads
- Group actions by queue in `RemoteHttpRunner` and submit as batches with configurable chunk size, falling back to individual submission on failure
- Extract shared helpers: `MakeErrorResult`, `ValidateQueueForEnqueue`, `ActivateActionInQueue`, `RemoveActionFromActiveMaps`

### Retracted Action State
- Add `Retracted` state to `RunnerAction` for retry-free rescheduling — an explicit request to pull an action back and reschedule it on a different runner without incrementing `RetryCount`
- Implement idempotent `RetractAction()` on `RunnerAction` and `ComputeServiceSession`
- Add `POST jobs/{lsn}/retract` and `queues/{queueref}/jobs/{lsn}/retract` HTTP endpoints
- Add state machine documentation and per-state comments to `RunnerAction`

### Compute Race Fixes
- Fix race in `HandleActionUpdates` where actions enqueued between session abandon and scheduler tick were never abandoned, causing `GetActionResult` to return 202 indefinitely
- Fix queue `ActiveCount` race where `NotifyQueueActionComplete` was called after releasing `m_ResultsLock`, allowing callers to observe stale counters immediately after `GetActionResult` returned OK

### Logging Optimization and ANSI improvements
- Improve `AnsiColorStdoutSink` write efficiency — single write call, dirty-flag flush, `RwLock` instead of `std::mutex`
- Move ANSI color emission from sink into formatters via `Formatter::SetColorEnabled()`; remove `ColorRangeStart`/`End` from `LogMessage`
- Extract color helpers (`AnsiColorForLevel`, `StripAnsiSgrSequences`) into `helpers.h`
- Strip upstream ANSI SGR escapes in non-color output mode. This enables colour in log messages without polluting log files with ANSI control sequences
- Move `RotatingFileSink`, `JsonFormatter`, and `FullFormatter` from header-only to pimpl with `.cpp` files

### CLI / Exec Refactoring
- Extract `ExecSessionRunner` class from ~920-line `ExecUsingSession` into focused methods and a `ExecSessionConfig` struct
- Replace monolithic `ExecCommand` with subcommand-based architecture (`http`, `inproc`, `beacon`, `dump`, `buildlog`)
- Allow parent options to appear after subcommand name by parsing subcommand args permissively and forwarding unmatched tokens to the parent parser

### Testing Improvements
- Fix `--test-suite` filter being ignored due to accumulation with default wildcard filter
- Add test suite banners to test listener output
- Made `function.session.abandon_pending` test more robust

### Startup / Reliability Fixes
- Fix silent exit when a second zenserver instance detects a port conflict — use `ZEN_CONSOLE_*` for log calls that precede `InitializeLogging()`
- Fix two potential SIGSEGV paths during early startup: guard `sentry_options_new()` returning nullptr, and throw on `ZenServerState::Register()` returning nullptr instead of dereferencing
- Fail on unrecognized zenserver `--mode` instead of silently defaulting to store

### Other
- Show host details (hostname, platform, CPU count, memory) when discovering new compute workers
- Move frontend `html.zip` from source tree into build directory
- Add format specifications for Compact Binary and Compressed Buffer wire formats
- Add `WriteCompactBinaryObject` to zencore
- Extended `ConsoleTui` with additional functionality
- Add `--vscode` option to `xmake sln` for clangd / `compile_commands.json` support
- Disable compute/horde/nomad in release builds (not yet production-ready)
- Disable unintended `ASIO_HAS_IO_URING` enablement
- Fix crashpad patch missing leading whitespace
- Clean up code triggering gcc false positives

9 files changed, 803 insertions, 553 deletions

diff --git a/src/zenutil/consoletui.cpp b/src/zenutil/consoletui.cpp
index 4410d463d..124132aed 100644
--- a/src/zenutil/consoletui.cpp
+++ b/src/zenutil/consoletui.cpp
@@ -480,4 +480,69 @@ TuiPollQuit()
 #endif
 }
 
+void
+TuiSetScrollRegion(uint32_t Top, uint32_t Bottom)
+{
+	printf("\033[%u;%ur", Top, Bottom);
+}
+
+void
+TuiResetScrollRegion()
+{
+	printf("\033[r");
+}
+
+void
+TuiMoveCursor(uint32_t Row, uint32_t Col)
+{
+	printf("\033[%u;%uH", Row, Col);
+}
+
+void
+TuiSaveCursor()
+{
+	printf(
+		"\033"
+		"7");
+}
+
+void
+TuiRestoreCursor()
+{
+	printf(
+		"\033"
+		"8");
+}
+
+void
+TuiEraseLine()
+{
+	printf("\033[2K");
+}
+
+void
+TuiWrite(std::string_view Text)
+{
+	fwrite(Text.data(), 1, Text.size(), stdout);
+}
+
+void
+TuiFlush()
+{
+	fflush(stdout);
+}
+
+void
+TuiShowCursor(bool Show)
+{
+	if (Show)
+	{
+		printf("\033[?25h");
+	}
+	else
+	{
+		printf("\033[?25l");
+	}
+}
+
 }  // namespace zen
diff --git a/src/zenutil/include/zenutil/consoletui.h b/src/zenutil/include/zenutil/consoletui.h
index 5f74fa82b..22737589b 100644
--- a/src/zenutil/include/zenutil/consoletui.h
+++ b/src/zenutil/include/zenutil/consoletui.h
@@ -57,4 +57,32 @@ uint32_t TuiConsoleRows(uint32_t Default = 40);
 // Should only be called while in alternate screen mode.
 bool TuiPollQuit();
 
+// Set the scrollable region of the terminal to rows [Top, Bottom] (1-based, inclusive).
+// Emits DECSTBM: ESC[<top>;<bottom>r
+void TuiSetScrollRegion(uint32_t Top, uint32_t Bottom);
+
+// Reset the scroll region to the full terminal. Emits ESC[r
+void TuiResetScrollRegion();
+
+// Move cursor to the given 1-based row and column. Emits ESC[<row>;<col>H
+void TuiMoveCursor(uint32_t Row, uint32_t Col);
+
+// Save cursor position. Emits ESC 7
+void TuiSaveCursor();
+
+// Restore cursor position. Emits ESC 8
+void TuiRestoreCursor();
+
+// Erase the entire current line. Emits ESC[2K
+void TuiEraseLine();
+
+// Write raw text to stdout without any formatting or newline.
+void TuiWrite(std::string_view Text);
+
+// Flush stdout.
+void TuiFlush();
+
+// Show or hide the cursor. Emits ESC[?25h or ESC[?25l
+void TuiShowCursor(bool Show);
+
 }  // namespace zen
diff --git a/src/zenutil/include/zenutil/logging/fullformatter.h b/src/zenutil/include/zenutil/logging/fullformatter.h
index 33cb94dae..0d026ed72 100644
--- a/src/zenutil/include/zenutil/logging/fullformatter.h
+++ b/src/zenutil/include/zenutil/logging/fullformatter.h
@@ -3,10 +3,8 @@
 #pragma once
 
 #include <zencore/logging/formatter.h>
-#include <zencore/logging/helpers.h>
-#include <zencore/memory/llm.h>
-#include <zencore/zencore.h>
 
+#include <memory>
 #include <string_view>
 
 namespace zen::logging {
@@ -14,195 +12,16 @@ namespace zen::logging {
 class FullFormatter final : public Formatter
 {
 public:
-	FullFormatter(std::string_view LogId, std::chrono::time_point<std::chrono::system_clock> Epoch)
-	: m_Epoch(Epoch)
-	, m_LogId(LogId)
-	, m_LinePrefix(128, ' ')
-	, m_UseFullDate(false)
-	{
-	}
+	FullFormatter(std::string_view LogId, std::chrono::time_point<std::chrono::system_clock> Epoch);
+	explicit FullFormatter(std::string_view LogId);
+	~FullFormatter() override;
 
-	FullFormatter(std::string_view LogId) : m_LogId(LogId), m_LinePrefix(128, ' '), m_UseFullDate(true) {}
-
-	virtual std::unique_ptr<Formatter> Clone() const override
-	{
-		ZEN_MEMSCOPE(ELLMTag::Logging);
-		if (m_UseFullDate)
-		{
-			return std::make_unique<FullFormatter>(m_LogId);
-		}
-		return std::make_unique<FullFormatter>(m_LogId, m_Epoch);
-	}
-
-	virtual void Format(const LogMessage& Msg, MemoryBuffer& OutBuffer) override
-	{
-		ZEN_MEMSCOPE(ELLMTag::Logging);
-
-		// Note that the sink is responsible for ensuring there is only ever a
-		// single caller in here
-
-		using namespace std::literals;
-
-		std::chrono::seconds TimestampSeconds;
-
-		std::chrono::milliseconds Millis;
-
-		if (m_UseFullDate)
-		{
-			TimestampSeconds = std::chrono::duration_cast<std::chrono::seconds>(Msg.GetTime().time_since_epoch());
-			if (TimestampSeconds != m_LastLogSecs)
-			{
-				RwLock::ExclusiveLockScope _(m_TimestampLock);
-				m_LastLogSecs = TimestampSeconds;
-
-				m_CachedLocalTm = helpers::SafeLocaltime(LogClock::to_time_t(Msg.GetTime()));
-				m_CachedDatetime.clear();
-				m_CachedDatetime.push_back('[');
-				helpers::Pad2(m_CachedLocalTm.tm_year % 100, m_CachedDatetime);
-				m_CachedDatetime.push_back('-');
-				helpers::Pad2(m_CachedLocalTm.tm_mon + 1, m_CachedDatetime);
-				m_CachedDatetime.push_back('-');
-				helpers::Pad2(m_CachedLocalTm.tm_mday, m_CachedDatetime);
-				m_CachedDatetime.push_back(' ');
-				helpers::Pad2(m_CachedLocalTm.tm_hour, m_CachedDatetime);
-				m_CachedDatetime.push_back(':');
-				helpers::Pad2(m_CachedLocalTm.tm_min, m_CachedDatetime);
-				m_CachedDatetime.push_back(':');
-				helpers::Pad2(m_CachedLocalTm.tm_sec, m_CachedDatetime);
-				m_CachedDatetime.push_back('.');
-			}
-
-			Millis = helpers::TimeFraction<std::chrono::milliseconds>(Msg.GetTime());
-		}
-		else
-		{
-			auto ElapsedTime = Msg.GetTime() - m_Epoch;
-			TimestampSeconds = std::chrono::duration_cast<std::chrono::seconds>(ElapsedTime);
-
-			if (m_CacheTimestamp.load() != TimestampSeconds)
-			{
-				RwLock::ExclusiveLockScope _(m_TimestampLock);
-
-				m_CacheTimestamp = TimestampSeconds;
-
-				int		  Count	  = int(TimestampSeconds.count());
-				const int LogSecs = Count % 60;
-				Count /= 60;
-				const int LogMins = Count % 60;
-				Count /= 60;
-				const int LogHours = Count;
-
-				m_CachedDatetime.clear();
-				m_CachedDatetime.push_back('[');
-				helpers::Pad2(LogHours, m_CachedDatetime);
-				m_CachedDatetime.push_back(':');
-				helpers::Pad2(LogMins, m_CachedDatetime);
-				m_CachedDatetime.push_back(':');
-				helpers::Pad2(LogSecs, m_CachedDatetime);
-				m_CachedDatetime.push_back('.');
-			}
-
-			Millis = std::chrono::duration_cast<std::chrono::milliseconds>(ElapsedTime - TimestampSeconds);
-		}
-
-		{
-			RwLock::SharedLockScope _(m_TimestampLock);
-			OutBuffer.append(m_CachedDatetime.begin(), m_CachedDatetime.end());
-		}
-
-		helpers::Pad3(static_cast<uint32_t>(Millis.count()), OutBuffer);
-		OutBuffer.push_back(']');
-		OutBuffer.push_back(' ');
-
-		if (!m_LogId.empty())
-		{
-			OutBuffer.push_back('[');
-			helpers::AppendStringView(m_LogId, OutBuffer);
-			OutBuffer.push_back(']');
-			OutBuffer.push_back(' ');
-		}
-
-		// append logger name if exists
-		if (Msg.GetLoggerName().size() > 0)
-		{
-			OutBuffer.push_back('[');
-			helpers::AppendStringView(Msg.GetLoggerName(), OutBuffer);
-			OutBuffer.push_back(']');
-			OutBuffer.push_back(' ');
-		}
-
-		OutBuffer.push_back('[');
-		// wrap the level name with color
-		Msg.ColorRangeStart = OutBuffer.size();
-		helpers::AppendStringView(helpers::LevelToShortString(Msg.GetLevel()), OutBuffer);
-		Msg.ColorRangeEnd = OutBuffer.size();
-		OutBuffer.push_back(']');
-		OutBuffer.push_back(' ');
-
-		// add source location if present
-		if (Msg.GetSource())
-		{
-			OutBuffer.push_back('[');
-			const char* Filename = helpers::ShortFilename(Msg.GetSource().Filename);
-			helpers::AppendStringView(Filename, OutBuffer);
-			OutBuffer.push_back(':');
-			helpers::AppendInt(Msg.GetSource().Line, OutBuffer);
-			OutBuffer.push_back(']');
-			OutBuffer.push_back(' ');
-		}
-
-		// Handle newlines in single log call by prefixing each additional line to make
-		// subsequent lines align with the first line in the message
-
-		const size_t LinePrefixCount = Min<size_t>(OutBuffer.size(), m_LinePrefix.size());
-
-		auto MsgPayload	  = Msg.GetPayload();
-		auto ItLineBegin  = MsgPayload.begin();
-		auto ItMessageEnd = MsgPayload.end();
-		bool IsFirstline  = true;
-
-		{
-			auto ItLineEnd = ItLineBegin;
-
-			auto EmitLine = [&] {
-				if (IsFirstline)
-				{
-					IsFirstline = false;
-				}
-				else
-				{
-					helpers::AppendStringView(std::string_view(m_LinePrefix.data(), LinePrefixCount), OutBuffer);
-				}
-				helpers::AppendStringView(std::string_view(&*ItLineBegin, ItLineEnd - ItLineBegin), OutBuffer);
-			};
-
-			while (ItLineEnd != ItMessageEnd)
-			{
-				if (*ItLineEnd++ == '\n')
-				{
-					EmitLine();
-					ItLineBegin = ItLineEnd;
-				}
-			}
-
-			if (ItLineBegin != ItMessageEnd)
-			{
-				EmitLine();
-				helpers::AppendStringView("\n"sv, OutBuffer);
-			}
-		}
-	}
+	std::unique_ptr<Formatter> Clone() const override;
+	void					   Format(const LogMessage& Msg, MemoryBuffer& OutBuffer) override;
 
 private:
-	std::chrono::time_point<std::chrono::system_clock> m_Epoch;
-	std::tm											   m_CachedLocalTm;
-	std::chrono::seconds							   m_LastLogSecs{std::chrono::seconds(87654321)};
-	std::atomic<std::chrono::seconds>				   m_CacheTimestamp{std::chrono::seconds(87654321)};
-	MemoryBuffer									   m_CachedDatetime;
-	std::string										   m_LogId;
-	std::string										   m_LinePrefix;
-	bool											   m_UseFullDate = true;
-	RwLock											   m_TimestampLock;
+	struct Impl;
+	std::unique_ptr<Impl> m_Impl;
 };
 
 }  // namespace zen::logging
diff --git a/src/zenutil/include/zenutil/logging/jsonformatter.h b/src/zenutil/include/zenutil/logging/jsonformatter.h
index 216b1b5e5..fb3193f3e 100644
--- a/src/zenutil/include/zenutil/logging/jsonformatter.h
+++ b/src/zenutil/include/zenutil/logging/jsonformatter.h
@@ -3,158 +3,24 @@
 #pragma once
 
 #include <zencore/logging/formatter.h>
-#include <zencore/logging/helpers.h>
-#include <zencore/memory/llm.h>
-#include <zencore/zencore.h>
 
+#include <memory>
 #include <string_view>
-#include <unordered_map>
 
 namespace zen::logging {
 
-using namespace std::literals;
-
 class JsonFormatter final : public Formatter
 {
 public:
-	JsonFormatter(std::string_view LogId) : m_LogId(LogId) {}
-
-	virtual std::unique_ptr<Formatter> Clone() const override { return std::make_unique<JsonFormatter>(m_LogId); }
-
-	virtual void Format(const LogMessage& Msg, MemoryBuffer& Dest) override
-	{
-		ZEN_MEMSCOPE(ELLMTag::Logging);
-
-		using std::chrono::duration_cast;
-		using std::chrono::milliseconds;
-		using std::chrono::seconds;
-
-		auto Secs = std::chrono::duration_cast<seconds>(Msg.GetTime().time_since_epoch());
-		if (Secs != m_LastLogSecs)
-		{
-			RwLock::ExclusiveLockScope _(m_TimestampLock);
-			m_CachedTm	  = helpers::SafeLocaltime(LogClock::to_time_t(Msg.GetTime()));
-			m_LastLogSecs = Secs;
-
-			// cache the date/time part for the next second.
-			m_CachedDatetime.clear();
-
-			helpers::AppendInt(m_CachedTm.tm_year + 1900, m_CachedDatetime);
-			m_CachedDatetime.push_back('-');
-
-			helpers::Pad2(m_CachedTm.tm_mon + 1, m_CachedDatetime);
-			m_CachedDatetime.push_back('-');
-
-			helpers::Pad2(m_CachedTm.tm_mday, m_CachedDatetime);
-			m_CachedDatetime.push_back(' ');
-
-			helpers::Pad2(m_CachedTm.tm_hour, m_CachedDatetime);
-			m_CachedDatetime.push_back(':');
-
-			helpers::Pad2(m_CachedTm.tm_min, m_CachedDatetime);
-			m_CachedDatetime.push_back(':');
-
-			helpers::Pad2(m_CachedTm.tm_sec, m_CachedDatetime);
+	explicit JsonFormatter(std::string_view LogId);
+	~JsonFormatter() override;
 
-			m_CachedDatetime.push_back('.');
-		}
-		helpers::AppendStringView("{"sv, Dest);
-		helpers::AppendStringView("\"time\": \""sv, Dest);
-		{
-			RwLock::SharedLockScope _(m_TimestampLock);
-			Dest.append(m_CachedDatetime.begin(), m_CachedDatetime.end());
-		}
-		auto Millis = helpers::TimeFraction<milliseconds>(Msg.GetTime());
-		helpers::Pad3(static_cast<uint32_t>(Millis.count()), Dest);
-		helpers::AppendStringView("\", "sv, Dest);
-
-		helpers::AppendStringView("\"status\": \""sv, Dest);
-		helpers::AppendStringView(helpers::LevelToShortString(Msg.GetLevel()), Dest);
-		helpers::AppendStringView("\", "sv, Dest);
-
-		helpers::AppendStringView("\"source\": \""sv, Dest);
-		helpers::AppendStringView("zenserver"sv, Dest);
-		helpers::AppendStringView("\", "sv, Dest);
-
-		helpers::AppendStringView("\"service\": \""sv, Dest);
-		helpers::AppendStringView("zencache"sv, Dest);
-		helpers::AppendStringView("\", "sv, Dest);
-
-		if (!m_LogId.empty())
-		{
-			helpers::AppendStringView("\"id\": \""sv, Dest);
-			helpers::AppendStringView(m_LogId, Dest);
-			helpers::AppendStringView("\", "sv, Dest);
-		}
-
-		if (Msg.GetLoggerName().size() > 0)
-		{
-			helpers::AppendStringView("\"logger.name\": \""sv, Dest);
-			helpers::AppendStringView(Msg.GetLoggerName(), Dest);
-			helpers::AppendStringView("\", "sv, Dest);
-		}
-
-		if (Msg.GetThreadId() != 0)
-		{
-			helpers::AppendStringView("\"logger.thread_name\": \""sv, Dest);
-			helpers::PadUint(Msg.GetThreadId(), 0, Dest);
-			helpers::AppendStringView("\", "sv, Dest);
-		}
-
-		if (Msg.GetSource())
-		{
-			helpers::AppendStringView("\"file\": \""sv, Dest);
-			WriteEscapedString(Dest, helpers::ShortFilename(Msg.GetSource().Filename));
-			helpers::AppendStringView("\","sv, Dest);
-
-			helpers::AppendStringView("\"line\": \""sv, Dest);
-			helpers::AppendInt(Msg.GetSource().Line, Dest);
-			helpers::AppendStringView("\","sv, Dest);
-		}
-
-		helpers::AppendStringView("\"message\": \""sv, Dest);
-		WriteEscapedString(Dest, Msg.GetPayload());
-		helpers::AppendStringView("\""sv, Dest);
-
-		helpers::AppendStringView("}\n"sv, Dest);
-	}
+	std::unique_ptr<Formatter> Clone() const override;
+	void					   Format(const LogMessage& Msg, MemoryBuffer& Dest) override;
 
 private:
-	static inline const std::unordered_map<char, std::string_view> s_SpecialCharacterMap{{'\b', "\\b"sv},
-																						 {'\f', "\\f"sv},
-																						 {'\n', "\\n"sv},
-																						 {'\r', "\\r"sv},
-																						 {'\t', "\\t"sv},
-																						 {'"', "\\\""sv},
-																						 {'\\', "\\\\"sv}};
-
-	static void WriteEscapedString(MemoryBuffer& Dest, const std::string_view& Text)
-	{
-		const char* RangeStart = Text.data();
-		const char* End		   = Text.data() + Text.size();
-		for (const char* It = RangeStart; It != End; ++It)
-		{
-			if (auto SpecialIt = s_SpecialCharacterMap.find(*It); SpecialIt != s_SpecialCharacterMap.end())
-			{
-				if (RangeStart != It)
-				{
-					Dest.append(RangeStart, It);
-				}
-				helpers::AppendStringView(SpecialIt->second, Dest);
-				RangeStart = It + 1;
-			}
-		}
-		if (RangeStart != End)
-		{
-			Dest.append(RangeStart, End);
-		}
-	};
-
-	std::tm				 m_CachedTm{0, 0, 0, 0, 0, 0, 0, 0, 0};
-	std::chrono::seconds m_LastLogSecs{0};
-	MemoryBuffer		 m_CachedDatetime;
-	std::string			 m_LogId;
-	RwLock				 m_TimestampLock;
+	struct Impl;
+	std::unique_ptr<Impl> m_Impl;
 };
 
 }  // namespace zen::logging
diff --git a/src/zenutil/include/zenutil/logging/rotatingfilesink.h b/src/zenutil/include/zenutil/logging/rotatingfilesink.h
index cebc5b110..e0ff7eca1 100644
--- a/src/zenutil/include/zenutil/logging/rotatingfilesink.h
+++ b/src/zenutil/include/zenutil/logging/rotatingfilesink.h
@@ -2,14 +2,11 @@
 
 #pragma once
 
-#include <zencore/basicfile.h>
-#include <zencore/logging/formatter.h>
-#include <zencore/logging/messageonlyformatter.h>
 #include <zencore/logging/sink.h>
-#include <zencore/memory/llm.h>
 
-#include <atomic>
+#include <cstddef>
 #include <filesystem>
+#include <memory>
 
 namespace zen::logging {
 
@@ -19,230 +16,21 @@ namespace zen::logging {
 class RotatingFileSink : public Sink
 {
 public:
-	RotatingFileSink(const std::filesystem::path& BaseFilename, std::size_t MaxSize, std::size_t MaxFiles, bool RotateOnOpen = false)
-	: m_BaseFilename(BaseFilename)
-	, m_MaxSize(MaxSize)
-	, m_MaxFiles(MaxFiles)
-	, m_Formatter(std::make_unique<MessageOnlyFormatter>())
-	{
-		ZEN_MEMSCOPE(ELLMTag::Logging);
-
-		std::error_code Ec;
-		if (RotateOnOpen)
-		{
-			RwLock::ExclusiveLockScope RotateLock(m_Lock);
-			Rotate(RotateLock, Ec);
-		}
-		else
-		{
-			m_CurrentFile.Open(m_BaseFilename, BasicFile::Mode::kWrite, Ec);
-			if (!Ec)
-			{
-				m_CurrentSize = m_CurrentFile.FileSize(Ec);
-			}
-			if (!Ec)
-			{
-				if (m_CurrentSize > m_MaxSize)
-				{
-					RwLock::ExclusiveLockScope RotateLock(m_Lock);
-					Rotate(RotateLock, Ec);
-				}
-			}
-		}
-
-		if (Ec)
-		{
-			throw std::system_error(Ec, fmt::format("Failed to open log file '{}'", m_BaseFilename.string()));
-		}
-	}
-
-	virtual ~RotatingFileSink()
-	{
-		try
-		{
-			RwLock::ExclusiveLockScope RotateLock(m_Lock);
-			m_CurrentFile.Close();
-		}
-		catch (const std::exception&)
-		{
-		}
-	}
+	RotatingFileSink(const std::filesystem::path& BaseFilename, std::size_t MaxSize, std::size_t MaxFiles, bool RotateOnOpen = false);
+	~RotatingFileSink() override;
 
 	RotatingFileSink(const RotatingFileSink&) = delete;
 	RotatingFileSink(RotatingFileSink&&)	  = delete;
-
 	RotatingFileSink& operator=(const RotatingFileSink&) = delete;
 	RotatingFileSink& operator=(RotatingFileSink&&) = delete;
 
-	virtual void Log(const LogMessage& Msg) override
-	{
-		ZEN_MEMSCOPE(ELLMTag::Logging);
-
-		try
-		{
-			MemoryBuffer Formatted;
-			if (TrySinkIt(Msg, Formatted))
-			{
-				return;
-			}
-
-			// This intentionally has no limit on the number of retries, see
-			// comment above.
-			for (;;)
-			{
-				{
-					RwLock::ExclusiveLockScope RotateLock(m_Lock);
-					// Only rotate if no-one else has rotated before us
-					if (m_CurrentSize > m_MaxSize || !m_CurrentFile.IsOpen())
-					{
-						std::error_code Ec;
-						Rotate(RotateLock, Ec);
-						if (Ec)
-						{
-							return;
-						}
-					}
-				}
-				if (TrySinkIt(Formatted))
-				{
-					return;
-				}
-			}
-		}
-		catch (const std::exception&)
-		{
-			// Silently eat errors
-		}
-	}
-	virtual void Flush() override
-	{
-		if (!m_NeedFlush)
-		{
-			return;
-		}
-
-		ZEN_MEMSCOPE(ELLMTag::Logging);
-
-		try
-		{
-			RwLock::SharedLockScope Lock(m_Lock);
-			if (m_CurrentFile.IsOpen())
-			{
-				m_CurrentFile.Flush();
-			}
-		}
-		catch (const std::exception&)
-		{
-			// Silently eat errors
-		}
-
-		m_NeedFlush = false;
-	}
-
-	virtual void SetFormatter(std::unique_ptr<Formatter> InFormatter) override
-	{
-		ZEN_MEMSCOPE(ELLMTag::Logging);
-
-		try
-		{
-			RwLock::ExclusiveLockScope _(m_Lock);
-			m_Formatter = std::move(InFormatter);
-		}
-		catch (const std::exception&)
-		{
-			// Silently eat errors
-		}
-	}
+	void Log(const LogMessage& Msg) override;
+	void Flush() override;
+	void SetFormatter(std::unique_ptr<Formatter> InFormatter) override;
 
 private:
-	void Rotate(RwLock::ExclusiveLockScope&, std::error_code& OutEc)
-	{
-		ZEN_MEMSCOPE(ELLMTag::Logging);
-
-		m_CurrentFile.Close();
-
-		OutEc = RotateFiles(m_BaseFilename, m_MaxFiles);
-		if (OutEc)
-		{
-			return;
-		}
-
-		m_CurrentFile.Open(m_BaseFilename, BasicFile::Mode::kWrite, OutEc);
-		if (OutEc)
-		{
-			return;
-		}
-
-		m_CurrentSize = m_CurrentFile.FileSize(OutEc);
-		if (OutEc)
-		{
-			// FileSize failed but we have an open file — reset to 0
-			// so we can at least attempt writes from the start
-			m_CurrentSize = 0;
-			OutEc.clear();
-		}
-	}
-
-	bool TrySinkIt(const LogMessage& Msg, MemoryBuffer& OutFormatted)
-	{
-		ZEN_MEMSCOPE(ELLMTag::Logging);
-
-		RwLock::SharedLockScope Lock(m_Lock);
-		if (!m_CurrentFile.IsOpen())
-		{
-			return false;
-		}
-		m_Formatter->Format(Msg, OutFormatted);
-		size_t AddSize	= OutFormatted.size();
-		size_t WritePos = m_CurrentSize.fetch_add(AddSize);
-		if (WritePos + AddSize > m_MaxSize)
-		{
-			return false;
-		}
-		std::error_code Ec;
-		m_CurrentFile.Write(OutFormatted.data(), OutFormatted.size(), WritePos, Ec);
-		if (Ec)
-		{
-			return false;
-		}
-		m_NeedFlush = true;
-		return true;
-	}
-
-	bool TrySinkIt(const MemoryBuffer& Formatted)
-	{
-		ZEN_MEMSCOPE(ELLMTag::Logging);
-
-		RwLock::SharedLockScope Lock(m_Lock);
-		if (!m_CurrentFile.IsOpen())
-		{
-			return false;
-		}
-		size_t AddSize	= Formatted.size();
-		size_t WritePos = m_CurrentSize.fetch_add(AddSize);
-		if (WritePos + AddSize > m_MaxSize)
-		{
-			return false;
-		}
-
-		std::error_code Ec;
-		m_CurrentFile.Write(Formatted.data(), Formatted.size(), WritePos, Ec);
-		if (Ec)
-		{
-			return false;
-		}
-		m_NeedFlush = true;
-		return true;
-	}
-
-	RwLock						m_Lock;
-	const std::filesystem::path m_BaseFilename;
-	const std::size_t			m_MaxSize;
-	const std::size_t			m_MaxFiles;
-	std::unique_ptr<Formatter>	m_Formatter;
-	std::atomic_size_t			m_CurrentSize;
-	BasicFile					m_CurrentFile;
-	std::atomic<bool>			m_NeedFlush = false;
+	struct Impl;
+	std::unique_ptr<Impl> m_Impl;
 };
 
 }  // namespace zen::logging
diff --git a/src/zenutil/logging/fullformatter.cpp b/src/zenutil/logging/fullformatter.cpp
new file mode 100644
index 000000000..2a4840241
--- /dev/null
+++ b/src/zenutil/logging/fullformatter.cpp
@@ -0,0 +1,235 @@
+// Copyright Epic Games, Inc. All Rights Reserved.
+
+#include <zenutil/logging/fullformatter.h>
+
+#include <zencore/intmath.h>
+#include <zencore/logging/helpers.h>
+#include <zencore/logging/memorybuffer.h>
+#include <zencore/memory/llm.h>
+#include <zencore/thread.h>
+#include <zencore/zencore.h>
+
+#include <atomic>
+#include <chrono>
+#include <string>
+
+namespace zen::logging {
+
+struct FullFormatter::Impl
+{
+	Impl(std::string_view LogId, std::chrono::time_point<std::chrono::system_clock> Epoch)
+	: m_Epoch(Epoch)
+	, m_LogId(LogId)
+	, m_LinePrefix(128, ' ')
+	, m_UseFullDate(false)
+	{
+	}
+
+	explicit Impl(std::string_view LogId) : m_LogId(LogId), m_LinePrefix(128, ' '), m_UseFullDate(true) {}
+
+	std::chrono::time_point<std::chrono::system_clock> m_Epoch;
+	std::tm											   m_CachedLocalTm{};
+	std::chrono::seconds							   m_LastLogSecs{std::chrono::seconds(87654321)};
+	std::atomic<std::chrono::seconds>				   m_CacheTimestamp{std::chrono::seconds(87654321)};
+	MemoryBuffer									   m_CachedDatetime;
+	std::string										   m_LogId;
+	std::string										   m_LinePrefix;
+	bool											   m_UseFullDate = true;
+	RwLock											   m_TimestampLock;
+};
+
+FullFormatter::FullFormatter(std::string_view LogId, std::chrono::time_point<std::chrono::system_clock> Epoch)
+: m_Impl(std::make_unique<Impl>(LogId, Epoch))
+{
+}
+
+FullFormatter::FullFormatter(std::string_view LogId) : m_Impl(std::make_unique<Impl>(LogId))
+{
+}
+
+FullFormatter::~FullFormatter() = default;
+
+std::unique_ptr<Formatter>
+FullFormatter::Clone() const
+{
+	ZEN_MEMSCOPE(ELLMTag::Logging);
+	std::unique_ptr<FullFormatter> Copy;
+	if (m_Impl->m_UseFullDate)
+	{
+		Copy = std::make_unique<FullFormatter>(m_Impl->m_LogId);
+	}
+	else
+	{
+		Copy = std::make_unique<FullFormatter>(m_Impl->m_LogId, m_Impl->m_Epoch);
+	}
+	Copy->SetColorEnabled(IsColorEnabled());
+	return Copy;
+}
+
+void
+FullFormatter::Format(const LogMessage& Msg, MemoryBuffer& OutBuffer)
+{
+	ZEN_MEMSCOPE(ELLMTag::Logging);
+
+	// Note that the sink is responsible for ensuring there is only ever a
+	// single caller in here
+
+	using namespace std::literals;
+
+	std::chrono::seconds TimestampSeconds;
+
+	std::chrono::milliseconds Millis;
+
+	if (m_Impl->m_UseFullDate)
+	{
+		TimestampSeconds = std::chrono::duration_cast<std::chrono::seconds>(Msg.GetTime().time_since_epoch());
+		if (TimestampSeconds != m_Impl->m_LastLogSecs)
+		{
+			RwLock::ExclusiveLockScope _(m_Impl->m_TimestampLock);
+			m_Impl->m_LastLogSecs = TimestampSeconds;
+
+			m_Impl->m_CachedLocalTm = helpers::SafeLocaltime(LogClock::to_time_t(Msg.GetTime()));
+			m_Impl->m_CachedDatetime.clear();
+			m_Impl->m_CachedDatetime.push_back('[');
+			helpers::Pad2(m_Impl->m_CachedLocalTm.tm_year % 100, m_Impl->m_CachedDatetime);
+			m_Impl->m_CachedDatetime.push_back('-');
+			helpers::Pad2(m_Impl->m_CachedLocalTm.tm_mon + 1, m_Impl->m_CachedDatetime);
+			m_Impl->m_CachedDatetime.push_back('-');
+			helpers::Pad2(m_Impl->m_CachedLocalTm.tm_mday, m_Impl->m_CachedDatetime);
+			m_Impl->m_CachedDatetime.push_back(' ');
+			helpers::Pad2(m_Impl->m_CachedLocalTm.tm_hour, m_Impl->m_CachedDatetime);
+			m_Impl->m_CachedDatetime.push_back(':');
+			helpers::Pad2(m_Impl->m_CachedLocalTm.tm_min, m_Impl->m_CachedDatetime);
+			m_Impl->m_CachedDatetime.push_back(':');
+			helpers::Pad2(m_Impl->m_CachedLocalTm.tm_sec, m_Impl->m_CachedDatetime);
+			m_Impl->m_CachedDatetime.push_back('.');
+		}
+
+		Millis = helpers::TimeFraction<std::chrono::milliseconds>(Msg.GetTime());
+	}
+	else
+	{
+		auto ElapsedTime = Msg.GetTime() - m_Impl->m_Epoch;
+		TimestampSeconds = std::chrono::duration_cast<std::chrono::seconds>(ElapsedTime);
+
+		if (m_Impl->m_CacheTimestamp.load() != TimestampSeconds)
+		{
+			RwLock::ExclusiveLockScope _(m_Impl->m_TimestampLock);
+
+			m_Impl->m_CacheTimestamp = TimestampSeconds;
+
+			int		  Count	  = int(TimestampSeconds.count());
+			const int LogSecs = Count % 60;
+			Count /= 60;
+			const int LogMins = Count % 60;
+			Count /= 60;
+			const int LogHours = Count;
+
+			m_Impl->m_CachedDatetime.clear();
+			m_Impl->m_CachedDatetime.push_back('[');
+			helpers::Pad2(LogHours, m_Impl->m_CachedDatetime);
+			m_Impl->m_CachedDatetime.push_back(':');
+			helpers::Pad2(LogMins, m_Impl->m_CachedDatetime);
+			m_Impl->m_CachedDatetime.push_back(':');
+			helpers::Pad2(LogSecs, m_Impl->m_CachedDatetime);
+			m_Impl->m_CachedDatetime.push_back('.');
+		}
+
+		Millis = std::chrono::duration_cast<std::chrono::milliseconds>(ElapsedTime - TimestampSeconds);
+	}
+
+	{
+		RwLock::SharedLockScope _(m_Impl->m_TimestampLock);
+		OutBuffer.append(m_Impl->m_CachedDatetime.begin(), m_Impl->m_CachedDatetime.end());
+	}
+
+	helpers::Pad3(static_cast<uint32_t>(Millis.count()), OutBuffer);
+	OutBuffer.push_back(']');
+	OutBuffer.push_back(' ');
+
+	if (!m_Impl->m_LogId.empty())
+	{
+		OutBuffer.push_back('[');
+		helpers::AppendStringView(m_Impl->m_LogId, OutBuffer);
+		OutBuffer.push_back(']');
+		OutBuffer.push_back(' ');
+	}
+
+	// append logger name if exists
+	if (Msg.GetLoggerName().size() > 0)
+	{
+		OutBuffer.push_back('[');
+		helpers::AppendStringView(Msg.GetLoggerName(), OutBuffer);
+		OutBuffer.push_back(']');
+		OutBuffer.push_back(' ');
+	}
+
+	OutBuffer.push_back('[');
+	if (IsColorEnabled())
+	{
+		helpers::AppendAnsiColor(Msg.GetLevel(), OutBuffer);
+	}
+	helpers::AppendStringView(helpers::LevelToShortString(Msg.GetLevel()), OutBuffer);
+	if (IsColorEnabled())
+	{
+		helpers::AppendAnsiReset(OutBuffer);
+	}
+	OutBuffer.push_back(']');
+	OutBuffer.push_back(' ');
+
+	// add source location if present
+	if (Msg.GetSource())
+	{
+		OutBuffer.push_back('[');
+		const char* Filename = helpers::ShortFilename(Msg.GetSource().Filename);
+		helpers::AppendStringView(Filename, OutBuffer);
+		OutBuffer.push_back(':');
+		helpers::AppendInt(Msg.GetSource().Line, OutBuffer);
+		OutBuffer.push_back(']');
+		OutBuffer.push_back(' ');
+	}
+
+	// Handle newlines in single log call by prefixing each additional line to make
+	// subsequent lines align with the first line in the message
+
+	size_t		 AnsiBytes		 = IsColorEnabled() ? (helpers::AnsiColorForLevel(Msg.GetLevel()).size() + helpers::kAnsiReset.size()) : 0;
+	const size_t LinePrefixCount = Min<size_t>(OutBuffer.size() - AnsiBytes, m_Impl->m_LinePrefix.size());
+
+	auto MsgPayload	  = Msg.GetPayload();
+	auto ItLineBegin  = MsgPayload.begin();
+	auto ItMessageEnd = MsgPayload.end();
+	bool IsFirstline  = true;
+
+	{
+		auto ItLineEnd = ItLineBegin;
+
+		auto EmitLine = [&] {
+			if (IsFirstline)
+			{
+				IsFirstline = false;
+			}
+			else
+			{
+				helpers::AppendStringView(std::string_view(m_Impl->m_LinePrefix.data(), LinePrefixCount), OutBuffer);
+			}
+			helpers::AppendStringView(std::string_view(&*ItLineBegin, ItLineEnd - ItLineBegin), OutBuffer);
+		};
+
+		while (ItLineEnd != ItMessageEnd)
+		{
+			if (*ItLineEnd++ == '\n')
+			{
+				EmitLine();
+				ItLineBegin = ItLineEnd;
+			}
+		}
+
+		if (ItLineBegin != ItMessageEnd)
+		{
+			EmitLine();
+			helpers::AppendStringView("\n"sv, OutBuffer);
+		}
+	}
+}
+
+}  // namespace zen::logging
diff --git a/src/zenutil/logging/jsonformatter.cpp b/src/zenutil/logging/jsonformatter.cpp
new file mode 100644
index 000000000..673a03c94
--- /dev/null
+++ b/src/zenutil/logging/jsonformatter.cpp
@@ -0,0 +1,198 @@
+// Copyright Epic Games, Inc. All Rights Reserved.
+
+#include <zenutil/logging/jsonformatter.h>
+
+#include <zencore/logging/helpers.h>
+#include <zencore/memory/llm.h>
+#include <zencore/thread.h>
+#include <zencore/zencore.h>
+
+#include <chrono>
+#include <string>
+#include <unordered_map>
+
+namespace zen::logging {
+
+using namespace std::literals;
+
+static void
+WriteEscapedString(MemoryBuffer& Dest, std::string_view Text)
+{
+	// Strip ANSI SGR sequences before escaping so they don't appear in JSON output
+	static const auto IsEscapeStart = [](char C) { return C == '\033'; };
+
+	const char* RangeStart = Text.data();
+	const char* End		   = Text.data() + Text.size();
+
+	static const std::unordered_map<char, std::string_view> SpecialCharacterMap{
+		{'\b', "\\b"sv},
+		{'\f', "\\f"sv},
+		{'\n', "\\n"sv},
+		{'\r', "\\r"sv},
+		{'\t', "\\t"sv},
+		{'"', "\\\""sv},
+		{'\\', "\\\\"sv},
+	};
+
+	for (const char* It = RangeStart; It != End; ++It)
+	{
+		// Skip ANSI SGR escape sequences (\033[...m)
+		if (*It == '\033' && (It + 1) < End && *(It + 1) == '[')
+		{
+			if (RangeStart != It)
+			{
+				Dest.append(RangeStart, It);
+			}
+			const char* Seq = It + 2;
+			while (Seq < End && *Seq != 'm')
+			{
+				++Seq;
+			}
+			if (Seq < End)
+			{
+				++Seq;	// skip 'm'
+			}
+			It		   = Seq - 1;  // -1 because the for loop will ++It
+			RangeStart = Seq;
+			continue;
+		}
+
+		if (auto SpecialIt = SpecialCharacterMap.find(*It); SpecialIt != SpecialCharacterMap.end())
+		{
+			if (RangeStart != It)
+			{
+				Dest.append(RangeStart, It);
+			}
+			helpers::AppendStringView(SpecialIt->second, Dest);
+			RangeStart = It + 1;
+		}
+	}
+	if (RangeStart != End)
+	{
+		Dest.append(RangeStart, End);
+	}
+}
+
+struct JsonFormatter::Impl
+{
+	explicit Impl(std::string_view LogId) : m_LogId(LogId) {}
+
+	std::tm				 m_CachedTm{0, 0, 0, 0, 0, 0, 0, 0, 0};
+	std::chrono::seconds m_LastLogSecs{0};
+	MemoryBuffer		 m_CachedDatetime;
+	std::string			 m_LogId;
+	RwLock				 m_TimestampLock;
+};
+
+JsonFormatter::JsonFormatter(std::string_view LogId) : m_Impl(std::make_unique<Impl>(LogId))
+{
+}
+
+JsonFormatter::~JsonFormatter() = default;
+
+std::unique_ptr<Formatter>
+JsonFormatter::Clone() const
+{
+	return std::make_unique<JsonFormatter>(m_Impl->m_LogId);
+}
+
+void
+JsonFormatter::Format(const LogMessage& Msg, MemoryBuffer& Dest)
+{
+	ZEN_MEMSCOPE(ELLMTag::Logging);
+
+	using std::chrono::duration_cast;
+	using std::chrono::milliseconds;
+	using std::chrono::seconds;
+
+	auto Secs = duration_cast<seconds>(Msg.GetTime().time_since_epoch());
+	if (Secs != m_Impl->m_LastLogSecs)
+	{
+		RwLock::ExclusiveLockScope _(m_Impl->m_TimestampLock);
+		m_Impl->m_CachedTm	  = helpers::SafeLocaltime(LogClock::to_time_t(Msg.GetTime()));
+		m_Impl->m_LastLogSecs = Secs;
+
+		// cache the date/time part for the next second.
+		m_Impl->m_CachedDatetime.clear();
+
+		helpers::AppendInt(m_Impl->m_CachedTm.tm_year + 1900, m_Impl->m_CachedDatetime);
+		m_Impl->m_CachedDatetime.push_back('-');
+
+		helpers::Pad2(m_Impl->m_CachedTm.tm_mon + 1, m_Impl->m_CachedDatetime);
+		m_Impl->m_CachedDatetime.push_back('-');
+
+		helpers::Pad2(m_Impl->m_CachedTm.tm_mday, m_Impl->m_CachedDatetime);
+		m_Impl->m_CachedDatetime.push_back(' ');
+
+		helpers::Pad2(m_Impl->m_CachedTm.tm_hour, m_Impl->m_CachedDatetime);
+		m_Impl->m_CachedDatetime.push_back(':');
+
+		helpers::Pad2(m_Impl->m_CachedTm.tm_min, m_Impl->m_CachedDatetime);
+		m_Impl->m_CachedDatetime.push_back(':');
+
+		helpers::Pad2(m_Impl->m_CachedTm.tm_sec, m_Impl->m_CachedDatetime);
+
+		m_Impl->m_CachedDatetime.push_back('.');
+	}
+	helpers::AppendStringView("{"sv, Dest);
+	helpers::AppendStringView("\"time\": \""sv, Dest);
+	{
+		RwLock::SharedLockScope _(m_Impl->m_TimestampLock);
+		Dest.append(m_Impl->m_CachedDatetime.begin(), m_Impl->m_CachedDatetime.end());
+	}
+	auto Millis = helpers::TimeFraction<milliseconds>(Msg.GetTime());
+	helpers::Pad3(static_cast<uint32_t>(Millis.count()), Dest);
+	helpers::AppendStringView("\", "sv, Dest);
+
+	helpers::AppendStringView("\"status\": \""sv, Dest);
+	helpers::AppendStringView(helpers::LevelToShortString(Msg.GetLevel()), Dest);
+	helpers::AppendStringView("\", "sv, Dest);
+
+	helpers::AppendStringView("\"source\": \""sv, Dest);
+	helpers::AppendStringView("zenserver"sv, Dest);
+	helpers::AppendStringView("\", "sv, Dest);
+
+	helpers::AppendStringView("\"service\": \""sv, Dest);
+	helpers::AppendStringView("zencache"sv, Dest);
+	helpers::AppendStringView("\", "sv, Dest);
+
+	if (!m_Impl->m_LogId.empty())
+	{
+		helpers::AppendStringView("\"id\": \""sv, Dest);
+		helpers::AppendStringView(m_Impl->m_LogId, Dest);
+		helpers::AppendStringView("\", "sv, Dest);
+	}
+
+	if (Msg.GetLoggerName().size() > 0)
+	{
+		helpers::AppendStringView("\"logger.name\": \""sv, Dest);
+		helpers::AppendStringView(Msg.GetLoggerName(), Dest);
+		helpers::AppendStringView("\", "sv, Dest);
+	}
+
+	if (Msg.GetThreadId() != 0)
+	{
+		helpers::AppendStringView("\"logger.thread_name\": \""sv, Dest);
+		helpers::PadUint(Msg.GetThreadId(), 0, Dest);
+		helpers::AppendStringView("\", "sv, Dest);
+	}
+
+	if (Msg.GetSource())
+	{
+		helpers::AppendStringView("\"file\": \""sv, Dest);
+		WriteEscapedString(Dest, helpers::ShortFilename(Msg.GetSource().Filename));
+		helpers::AppendStringView("\","sv, Dest);
+
+		helpers::AppendStringView("\"line\": \""sv, Dest);
+		helpers::AppendInt(Msg.GetSource().Line, Dest);
+		helpers::AppendStringView("\","sv, Dest);
+	}
+
+	helpers::AppendStringView("\"message\": \""sv, Dest);
+	WriteEscapedString(Dest, Msg.GetPayload());
+	helpers::AppendStringView("\""sv, Dest);
+
+	helpers::AppendStringView("}\n"sv, Dest);
+}
+
+}  // namespace zen::logging
diff --git a/src/zenutil/logging.cpp b/src/zenutil/logging/logging.cpp
index 1258ca155..ea2448a42 100644
--- a/src/zenutil/logging.cpp
+++ b/src/zenutil/logging/logging.cpp
@@ -238,8 +238,10 @@ FinishInitializeLogging(const LoggingOptions& LogOptions)
 
 		const std::string StartLogTime = zen::DateTime::Now().ToIso8601();
 
-		static constinit logging::LogPoint LogStartPoint{{}, logging::Info, "log starting at {}"};
-		logging::Registry::Instance().ApplyAll([&](auto Logger) { Logger->Log(LogStartPoint, fmt::make_format_args(StartLogTime)); });
+		logging::Registry::Instance().ApplyAll([&](auto Logger) {
+			static constinit logging::LogPoint LogStartPoint{{}, logging::Info, "log starting at {}"};
+			Logger->Log(LogStartPoint, fmt::make_format_args(StartLogTime));
+		});
 	}
 
 	g_IsLoggingInitialized = true;
diff --git a/src/zenutil/logging/rotatingfilesink.cpp b/src/zenutil/logging/rotatingfilesink.cpp
new file mode 100644
index 000000000..23cf60d16
--- /dev/null
+++ b/src/zenutil/logging/rotatingfilesink.cpp
@@ -0,0 +1,249 @@
+// Copyright Epic Games, Inc. All Rights Reserved.
+
+#include <zenutil/logging/rotatingfilesink.h>
+
+#include <zencore/basicfile.h>
+#include <zencore/filesystem.h>
+#include <zencore/logging/helpers.h>
+#include <zencore/logging/messageonlyformatter.h>
+#include <zencore/memory/llm.h>
+#include <zencore/thread.h>
+
+#include <atomic>
+
+namespace zen::logging {
+
+struct RotatingFileSink::Impl
+{
+	Impl(const std::filesystem::path& BaseFilename, std::size_t MaxSize, std::size_t MaxFiles, bool RotateOnOpen)
+	: m_BaseFilename(BaseFilename)
+	, m_MaxSize(MaxSize)
+	, m_MaxFiles(MaxFiles)
+	, m_Formatter(std::make_unique<MessageOnlyFormatter>())
+	{
+		ZEN_MEMSCOPE(ELLMTag::Logging);
+
+		std::error_code Ec;
+		if (RotateOnOpen)
+		{
+			RwLock::ExclusiveLockScope RotateLock(m_Lock);
+			Rotate(RotateLock, Ec);
+		}
+		else
+		{
+			m_CurrentFile.Open(m_BaseFilename, BasicFile::Mode::kWrite, Ec);
+			if (!Ec)
+			{
+				m_CurrentSize = m_CurrentFile.FileSize(Ec);
+			}
+			if (!Ec)
+			{
+				if (m_CurrentSize > m_MaxSize)
+				{
+					RwLock::ExclusiveLockScope RotateLock(m_Lock);
+					Rotate(RotateLock, Ec);
+				}
+			}
+		}
+
+		if (Ec)
+		{
+			throw std::system_error(Ec, fmt::format("Failed to open log file '{}'", m_BaseFilename.string()));
+		}
+	}
+
+	~Impl()
+	{
+		try
+		{
+			RwLock::ExclusiveLockScope RotateLock(m_Lock);
+			m_CurrentFile.Close();
+		}
+		catch (const std::exception&)
+		{
+		}
+	}
+
+	void Rotate(RwLock::ExclusiveLockScope&, std::error_code& OutEc)
+	{
+		ZEN_MEMSCOPE(ELLMTag::Logging);
+
+		m_CurrentFile.Close();
+
+		OutEc = RotateFiles(m_BaseFilename, m_MaxFiles);
+		if (OutEc)
+		{
+			return;
+		}
+
+		m_CurrentFile.Open(m_BaseFilename, BasicFile::Mode::kWrite, OutEc);
+		if (OutEc)
+		{
+			return;
+		}
+
+		m_CurrentSize = m_CurrentFile.FileSize(OutEc);
+		if (OutEc)
+		{
+			// FileSize failed but we have an open file — reset to 0
+			// so we can at least attempt writes from the start
+			m_CurrentSize = 0;
+			OutEc.clear();
+		}
+	}
+
+	bool TrySinkIt(const LogMessage& Msg, MemoryBuffer& OutFormatted)
+	{
+		ZEN_MEMSCOPE(ELLMTag::Logging);
+
+		RwLock::SharedLockScope Lock(m_Lock);
+		if (!m_CurrentFile.IsOpen())
+		{
+			return false;
+		}
+		m_Formatter->Format(Msg, OutFormatted);
+		helpers::StripAnsiSgrSequences(OutFormatted);
+		size_t AddSize	= OutFormatted.size();
+		size_t WritePos = m_CurrentSize.fetch_add(AddSize);
+		if (WritePos + AddSize > m_MaxSize)
+		{
+			return false;
+		}
+		std::error_code Ec;
+		m_CurrentFile.Write(OutFormatted.data(), OutFormatted.size(), WritePos, Ec);
+		if (Ec)
+		{
+			return false;
+		}
+		m_NeedFlush = true;
+		return true;
+	}
+
+	bool TrySinkIt(const MemoryBuffer& Formatted)
+	{
+		ZEN_MEMSCOPE(ELLMTag::Logging);
+
+		RwLock::SharedLockScope Lock(m_Lock);
+		if (!m_CurrentFile.IsOpen())
+		{
+			return false;
+		}
+		size_t AddSize	= Formatted.size();
+		size_t WritePos = m_CurrentSize.fetch_add(AddSize);
+		if (WritePos + AddSize > m_MaxSize)
+		{
+			return false;
+		}
+
+		std::error_code Ec;
+		m_CurrentFile.Write(Formatted.data(), Formatted.size(), WritePos, Ec);
+		if (Ec)
+		{
+			return false;
+		}
+		m_NeedFlush = true;
+		return true;
+	}
+
+	RwLock						m_Lock;
+	const std::filesystem::path m_BaseFilename;
+	const std::size_t			m_MaxSize;
+	const std::size_t			m_MaxFiles;
+	std::unique_ptr<Formatter>	m_Formatter;
+	std::atomic_size_t			m_CurrentSize;
+	BasicFile					m_CurrentFile;
+	std::atomic<bool>			m_NeedFlush = false;
+};
+
+RotatingFileSink::RotatingFileSink(const std::filesystem::path& BaseFilename, std::size_t MaxSize, std::size_t MaxFiles, bool RotateOnOpen)
+: m_Impl(std::make_unique<Impl>(BaseFilename, MaxSize, MaxFiles, RotateOnOpen))
+{
+}
+
+RotatingFileSink::~RotatingFileSink() = default;
+
+void
+RotatingFileSink::Log(const LogMessage& Msg)
+{
+	ZEN_MEMSCOPE(ELLMTag::Logging);
+
+	try
+	{
+		MemoryBuffer Formatted;
+		if (m_Impl->TrySinkIt(Msg, Formatted))
+		{
+			return;
+		}
+
+		// This intentionally has no limit on the number of retries, see
+		// comment above.
+		for (;;)
+		{
+			{
+				RwLock::ExclusiveLockScope RotateLock(m_Impl->m_Lock);
+				// Only rotate if no-one else has rotated before us
+				if (m_Impl->m_CurrentSize > m_Impl->m_MaxSize || !m_Impl->m_CurrentFile.IsOpen())
+				{
+					std::error_code Ec;
+					m_Impl->Rotate(RotateLock, Ec);
+					if (Ec)
+					{
+						return;
+					}
+				}
+			}
+			if (m_Impl->TrySinkIt(Formatted))
+			{
+				return;
+			}
+		}
+	}
+	catch (const std::exception&)
+	{
+		// Silently eat errors
+	}
+}
+
+void
+RotatingFileSink::Flush()
+{
+	if (!m_Impl->m_NeedFlush)
+	{
+		return;
+	}
+
+	ZEN_MEMSCOPE(ELLMTag::Logging);
+
+	try
+	{
+		RwLock::SharedLockScope Lock(m_Impl->m_Lock);
+		if (m_Impl->m_CurrentFile.IsOpen())
+		{
+			m_Impl->m_CurrentFile.Flush();
+		}
+	}
+	catch (const std::exception&)
+	{
+		// Silently eat errors
+	}
+
+	m_Impl->m_NeedFlush = false;
+}
+
+void
+RotatingFileSink::SetFormatter(std::unique_ptr<Formatter> InFormatter)
+{
+	ZEN_MEMSCOPE(ELLMTag::Logging);
+
+	try
+	{
+		RwLock::ExclusiveLockScope _(m_Impl->m_Lock);
+		m_Impl->m_Formatter = std::move(InFormatter);
+	}
+	catch (const std::exception&)
+	{
+		// Silently eat errors
+	}
+}
+
+}  // namespace zen::logging