minor zenstore/blockstore fixes (#821)

- Fix clang-format error accidentally introduced by recent PR
- Fix `FileSize()` CAS race that repeatedly invalidated the cache when concurrent callers both missed; remove `store(0)` on CAS failure
- Fix `WriteChunks` not accounting for initial alignment padding in `m_TotalSize`, causing drift vs `WriteChunk`'s correct accounting
- Fix Create retry sleep computing negative values (100 - N*100 instead of 100 + N*100), matching the Open retry pattern
- Fix `~BlockStore` error log missing format placeholder for `Ex.what()`
- Fix `GetFreeBlockIndex` infinite loop when all indexes have orphan files on disk but aren't in `m_ChunkBlocks`; bound probe to `m_MaxBlockCount`
- Fix `IterateBlock` ignoring `SmallSizeCallback` return value for single out-of-bounds chunks, preventing early termination
- Fix `BlockStoreCompactState::IterateBlocks` iterating map by value instead of const reference

1 files changed, 44 insertions, 64 deletions

diff --git a/src/zen/cmds/bench_cmd.cpp b/src/zen/cmds/bench_cmd.cpp
index 658b42da6..908d62257 100644
--- a/src/zen/cmds/bench_cmd.cpp
+++ b/src/zen/cmds/bench_cmd.cpp
@@ -28,12 +28,9 @@ namespace zen {
 //////////////////////////////////////////////////////////////////////////
 // BenchPurgeSubCmd
 
-BenchPurgeSubCmd::BenchPurgeSubCmd()
-: ZenSubCmdBase("purge", "Purge standby memory (system cache)")
+BenchPurgeSubCmd::BenchPurgeSubCmd() : ZenSubCmdBase("purge", "Purge standby memory (system cache)")
 {
-	SubOptions().add_options()("single",
-							   "Do not spawn child processes",
-							   cxxopts::value<bool>(m_SingleProcess)->default_value("false"));
+	SubOptions().add_options()("single", "Do not spawn child processes", cxxopts::value<bool>(m_SingleProcess)->default_value("false"));
 }
 
 void
@@ -104,17 +101,10 @@ BenchPurgeSubCmd::Run(const ZenCliOptions& GlobalOptions)
 //////////////////////////////////////////////////////////////////////////
 // BenchHttpSubCmd
 
-
-BenchHttpSubCmd::BenchHttpSubCmd()
-: ZenSubCmdBase("http", "Benchmark an HTTP server")
+BenchHttpSubCmd::BenchHttpSubCmd() : ZenSubCmdBase("http", "Benchmark an HTTP server")
 {
 	SubOptions().add_option("", "u", "url", "URL to benchmark", cxxopts::value<std::string>(m_Url), "<url>");
-	SubOptions().add_option("",
-							"n",
-							"count",
-							"Number of requests to send",
-							cxxopts::value<int>(m_Count)->default_value("100"),
-							"<count>");
+	SubOptions().add_option("", "n", "count", "Number of requests to send", cxxopts::value<int>(m_Count)->default_value("100"), "<count>");
 	SubOptions().add_option("",
 							"c",
 							"concurrency",
@@ -127,12 +117,8 @@ BenchHttpSubCmd::BenchHttpSubCmd()
 							"HTTP method to use (GET, HEAD)",
 							cxxopts::value<std::string>(m_Method)->default_value("GET"),
 							"<method>");
-	SubOptions().add_option("",
-							"",
-							"unix-socket",
-							"Unix domain socket path (overrides TCP)",
-							cxxopts::value<std::string>(m_SocketPath),
-							"<path>");
+	SubOptions()
+		.add_option("", "", "unix-socket", "Unix domain socket path (overrides TCP)", cxxopts::value<std::string>(m_SocketPath), "<path>");
 	SubOptions().add_options()("no-keepalive",
 							   "Close connection after each request (disables keep-alive)",
 							   cxxopts::value<bool>(m_NoKeepAlive)->default_value("false"));
@@ -179,8 +165,7 @@ BenchHttpSubCmd::Run(const ZenCliOptions& GlobalOptions)
 
 	if (m_Method != "GET" && m_Method != "HEAD")
 	{
-		throw OptionParseException(fmt::format("Unsupported HTTP method '{}'. Supported: GET, HEAD", m_Method),
-								   SubOptions().help());
+		throw OptionParseException(fmt::format("Unsupported HTTP method '{}'. Supported: GET, HEAD", m_Method), SubOptions().help());
 	}
 
 	auto [BaseUri, Path] = SplitUrl(m_Url);
@@ -201,16 +186,16 @@ BenchHttpSubCmd::Run(const ZenCliOptions& GlobalOptions)
 	// (non-zenserver, timeout, unreachable) is silently ignored.
 	try
 	{
-		HttpClientSettings ProbeSettings{.ConnectTimeout = std::chrono::milliseconds(2000),
-										 .Timeout		 = std::chrono::milliseconds(2000),
-										 .UnixSocketPath = m_SocketPath};
-		HttpClient			ProbeHttp(BaseUri, ProbeSettings);
+		HttpClientSettings	 ProbeSettings{.ConnectTimeout = std::chrono::milliseconds(2000),
+										   .Timeout		   = std::chrono::milliseconds(2000),
+										   .UnixSocketPath = m_SocketPath};
+		HttpClient			 ProbeHttp(BaseUri, ProbeSettings);
 		HttpClient::Response ProbeResp = ProbeHttp.Get("/health/info");
 
 		if (ProbeResp.IsSuccess())
 		{
-			CbObject		 Info		   = ProbeResp.AsObject();
-			std::string_view BuildVersion  = Info["BuildVersion"].AsString();
+			CbObject		 Info		  = ProbeResp.AsObject();
+			std::string_view BuildVersion = Info["BuildVersion"].AsString();
 
 			if (!BuildVersion.empty())
 			{
@@ -222,15 +207,15 @@ BenchHttpSubCmd::Run(const ZenCliOptions& GlobalOptions)
 				std::string_view OS	  = Info["OS"].AsString();
 				std::string_view Arch = Info["Arch"].AsString();
 
-				CbObjectView System		  = Info["System"].AsObjectView();
-				int64_t		 LpCount	  = System["lp_count"].AsInt64();
-				int64_t		 TotalMemMiB  = System["total_memory_mb"].AsInt64();
+				CbObjectView System		 = Info["System"].AsObjectView();
+				int64_t		 LpCount	 = System["lp_count"].AsInt64();
+				int64_t		 TotalMemMiB = System["total_memory_mb"].AsInt64();
 
 				ZEN_CONSOLE("             : {}, {}, {} logical processors, {} RAM",
-							 OS,
-							 Arch,
-							 LpCount,
-							 NiceBytes(static_cast<uint64_t>(TotalMemMiB) * 1024 * 1024));
+							OS,
+							Arch,
+							LpCount,
+							NiceBytes(static_cast<uint64_t>(TotalMemMiB) * 1024 * 1024));
 			}
 		}
 	}
@@ -251,8 +236,8 @@ BenchHttpSubCmd::Run(const ZenCliOptions& GlobalOptions)
 void
 BenchHttpSubCmd::RunFixedCount(const std::string& BaseUri, const std::string& Path)
 {
-	std::atomic<int>	 NextRequest{0};
-	std::vector<double>  AllLatencies;
+	std::atomic<int>	NextRequest{0};
+	std::vector<double> AllLatencies;
 	AllLatencies.reserve(m_Count);
 	std::mutex			 LatencyMutex;
 	std::atomic<int>	 ErrorCount{0};
@@ -264,8 +249,7 @@ BenchHttpSubCmd::RunFixedCount(const std::string& BaseUri, const std::string& Pa
 	auto WorkerFn = [&]() {
 		std::vector<double> LocalLatencies;
 
-		HttpClientSettings Settings{.UnixSocketPath		   = m_SocketPath,
-									.ForbidReuseConnection = m_NoKeepAlive};
+		HttpClientSettings Settings{.UnixSocketPath = m_SocketPath, .ForbidReuseConnection = m_NoKeepAlive};
 		HttpClient		   Http(BaseUri, Settings);
 
 		while (true)
@@ -346,17 +330,17 @@ BenchHttpSubCmd::RunFixedCount(const std::string& BaseUri, const std::string& Pa
 
 	ZEN_CONSOLE("  Requests  : {:L} total, {:L} success, {:L} errors", TotalCount, SuccessCount, ErrorCount.load());
 	ZEN_CONSOLE("  Latency   : min={:.1f}ms  mean={:.1f}ms  p50={:.1f}ms  p95={:.1f}ms  p99={:.1f}ms  max={:.1f}ms",
-				 PercentileMs(0),
-				 MeanMs,
-				 PercentileMs(50),
-				 PercentileMs(95),
-				 PercentileMs(99),
-				 PercentileMs(100));
+				PercentileMs(0),
+				MeanMs,
+				PercentileMs(50),
+				PercentileMs(95),
+				PercentileMs(99),
+				PercentileMs(100));
 	ZEN_CONSOLE("  Throughput: {:.1f} req/s  down: {}/s  up: {}/s  (elapsed: {:.2f}s)",
-				 Rps,
-				 NiceBytes(DownBytesPerSec),
-				 NiceBytes(UpBytesPerSec),
-				 TotalSeconds);
+				Rps,
+				NiceBytes(DownBytesPerSec),
+				NiceBytes(UpBytesPerSec),
+				TotalSeconds);
 }
 
 void
@@ -380,8 +364,7 @@ BenchHttpSubCmd::RunContinuous(const std::string& BaseUri, const std::string& Pa
 	Stopwatch RunTimer;
 
 	auto WorkerFn = [&]() {
-		HttpClientSettings Settings{.UnixSocketPath		   = m_SocketPath,
-									.ForbidReuseConnection = m_NoKeepAlive};
+		HttpClientSettings Settings{.UnixSocketPath = m_SocketPath, .ForbidReuseConnection = m_NoKeepAlive};
 		HttpClient		   Http(BaseUri, Settings);
 
 		while (!s_BenchAbort.load(std::memory_order_relaxed))
@@ -435,11 +418,11 @@ BenchHttpSubCmd::RunContinuous(const std::string& BaseUri, const std::string& Pa
 			int64_t UpBytes	  = IntervalUploadBytes.exchange(0);
 
 			// Snapshot and reset latency histogram
-			uint64_t			   HistCount = LatencyHistogram.Count();
-			int64_t				   HistMin	 = LatencyHistogram.Min();
-			int64_t				   HistMax	 = LatencyHistogram.Max();
-			double				   HistMean	 = LatencyHistogram.Mean();
-			metrics::SampleSnapshot Snap	 = LatencyHistogram.Snapshot();
+			uint64_t				HistCount = LatencyHistogram.Count();
+			int64_t					HistMin	  = LatencyHistogram.Min();
+			int64_t					HistMax	  = LatencyHistogram.Max();
+			double					HistMean  = LatencyHistogram.Mean();
+			metrics::SampleSnapshot Snap	  = LatencyHistogram.Snapshot();
 			LatencyHistogram.Clear();
 
 			// Format elapsed as HH:MM:SS
@@ -451,7 +434,8 @@ BenchHttpSubCmd::RunContinuous(const std::string& BaseUri, const std::string& Pa
 			if (HistCount > 0)
 			{
 				ZEN_CONSOLE(
-					"[{:02d}:{:02d}:{:02d}]  req/s: {:L}  errors: {:L}  lat(ms): min={:.1f} mean={:.1f} p95={:.1f} p99={:.1f} max={:.1f}  down: {}/s  up: {}/s",
+					"[{:02d}:{:02d}:{:02d}]  req/s: {:L}  errors: {:L}  lat(ms): min={:.1f} mean={:.1f} p95={:.1f} p99={:.1f} max={:.1f}  "
+					"down: {}/s  up: {}/s",
 					Hours,
 					Minutes,
 					Secs,
@@ -467,11 +451,7 @@ BenchHttpSubCmd::RunContinuous(const std::string& BaseUri, const std::string& Pa
 			}
 			else
 			{
-				ZEN_CONSOLE("[{:02d}:{:02d}:{:02d}]  req/s: 0  errors: {:L}  (no successful requests)",
-							Hours,
-							Minutes,
-							Secs,
-							Errors);
+				ZEN_CONSOLE("[{:02d}:{:02d}:{:02d}]  req/s: 0  errors: {:L}  (no successful requests)", Hours, Minutes, Secs, Errors);
 			}
 		}
 	};
@@ -493,8 +473,8 @@ BenchHttpSubCmd::RunContinuous(const std::string& BaseUri, const std::string& Pa
 	std::signal(SIGINT, PrevSigInt);
 	std::signal(SIGTERM, PrevSigTerm);
 
-	double   TotalSeconds = RunTimer.GetElapsedTimeMs() / 1000.0;
-	int64_t  TotalCount	  = TotalSuccessCount.load() + TotalErrorCount.load();
+	double	 TotalSeconds = RunTimer.GetElapsedTimeMs() / 1000.0;
+	int64_t	 TotalCount	  = TotalSuccessCount.load() + TotalErrorCount.load();
 	uint64_t DownPerSec	  = TotalSeconds > 0.0 ? static_cast<uint64_t>(TotalDownloadBytes.load() / TotalSeconds) : 0;
 	uint64_t UpPerSec	  = TotalSeconds > 0.0 ? static_cast<uint64_t>(TotalUploadBytes.load() / TotalSeconds) : 0;