storage: keep segment_set formatting bounded under fmt

Since the fmt 9 upgrade, formatter resolution for segment_set selects
fmt/ranges.h's range formatter over the ostream operator<<, printing
every segment and silently bypassing the 8-segment truncation added in
2023 for this same problem. On partitions with thousands of segments,
formatting a log handle (e.g. "Removing: {}" in log_manager::remove)
then builds a multi-megabyte string in one contiguous fmt buffer, which
can fail allocation on a fragmented shard and abort the process; the
allocation failure report is itself suppressed by the logger's
re-entrancy silencer, so such crashes appear as bare SIGABRTs. Move the
truncation into a format_to member and add an explicit full
fmt::formatter specialization, which takes precedence over any partial
specialization, so no future fmt upgrade or include change can route
around the bound; operator<< now delegates to it. The new gtest covers
the fmt dispatch path (the one that regressed) and asserts ostream
output stays identical.

Author: pgellert

src/v/storage/segment_set.cc

@@ -24,6 +24,9 @@
 #include <seastar/core/seastar.hh>
 #include <seastar/core/thread.hh>
 
+#include <fmt/format.h>
+#include <fmt/ostream.h>
+
 #include <algorithm>
 #include <exception>
 
@@ -137,25 +140,29 @@ segment_set::upper_bound(model::term_id term) const {
       _handles.cbegin(), _handles.cend(), term, segment_ordering{});
 }
 
-std::ostream& operator<<(std::ostream& o, const segment_set& s) {
-    o << "{size: " << s.size() << ", [";
+fmt::iterator segment_set::format_to(fmt::iterator out) const {
+    out = fmt::format_to(out, "{{size: {}, [", size());
     static constexpr size_t max_to_log = 8;
     static constexpr size_t halved = max_to_log / 2;
-    if (s.size() <= max_to_log) {
-        for (auto& p : s) {
-            o << p;
+    if (size() <= max_to_log) {
+        for (const auto& p : *this) {
+            out = fmt::format_to(out, "{}", p);
         }
     } else {
-        for (auto it = s.begin(); it != std::next(s.begin(), halved); ++it) {
-            o << *it;
+        for (auto it = begin(); it != std::next(begin(), halved); ++it) {
+            out = fmt::format_to(out, "{}", *it);
         }
-        o << "...";
-        for (auto it = std::next(s.begin(), s.size() - halved); it != s.end();
-             ++it) {
-            o << *it;
+        out = fmt::format_to(out, "...");
+        for (auto it = std::next(begin(), size() - halved); it != end(); ++it) {
+            out = fmt::format_to(out, "{}", *it);
         }
     }
-    return o << "]}";
+    return fmt::format_to(out, "]}}");
+}
+
+std::ostream& operator<<(std::ostream& o, const segment_set& s) {
+    fmt::print(o, "{}", s);
+    return o;
 }
 
 static bool

src/v/storage/segment_set.h

@@ -11,6 +11,7 @@
 
 #pragma once
 
+#include "base/format_to.h"
 #include "features/fwd.h"
 #include "storage/batch_cache.h"
 #include "storage/file_sanitizer_types.h"
@@ -90,6 +91,9 @@ class segment_set {
 
     segment_set copy() const noexcept { return *this; }
 
+    /// Formats a bounded summary of the segments, as the set can be large.
+    fmt::iterator format_to(fmt::iterator out) const;
+
 private:
     segment_set(const segment_set&) noexcept = default;
 
@@ -128,3 +132,25 @@ ss::future<std::optional<segment_set>>
 maybe_create_contiguous_segment_set(segment_set::underlying_t segs);
 
 } // namespace storage
+
+/// Explicit full specialization so that formatter resolution always uses the
+/// bounded segment_set::format_to. segment_set is a range, and without this a
+/// matching partial specialization (e.g. fmt/ranges.h's range formatter, which
+/// outranks the ostream operator<< fallback since fmt 9) would print every
+/// segment unbounded.
+template<>
+struct fmt::formatter<storage::segment_set> {
+    constexpr fmt::format_parse_context::iterator
+    parse(fmt::format_parse_context& ctx) const {
+        auto it = ctx.begin();
+        if (it != ctx.end() && *it != '}') {
+            throw fmt::format_error("invalid format specifier for this type");
+        }
+        return it;
+    }
+
+    fmt::iterator
+    format(const storage::segment_set& s, fmt::format_context& ctx) const {
+        return s.format_to(ctx.out());
+    }
+};

src/v/storage/tests/BUILD

@@ -669,6 +669,7 @@ redpanda_cc_gtest(
         "//src/v/storage:resources",
         "//src/v/test_utils:gtest",
         "//src/v/utils:directory_walker",
+        "@fmt",
         "@googletest//:gtest",
         "@seastar",
     ],

src/v/storage/tests/segment_set_test.cc

@@ -17,10 +17,13 @@
 
 #include <seastar/core/seastar.hh>
 
+#include <fmt/format.h>
+#include <gmock/gmock.h>
 #include <gtest/gtest.h>
 
 #include <filesystem>
 #include <optional>
+#include <sstream>
 
 static ss::logger segment_set_test_log("segment_set_test");
 
@@ -265,4 +268,52 @@ TEST_F(SegmentSetFixtureTest, recovery) {
     }
 }
 
+namespace {
+size_t count_occurrences(std::string_view haystack, std::string_view needle) {
+    size_t count = 0;
+    for (auto pos = haystack.find(needle); pos != std::string_view::npos;
+         pos = haystack.find(needle, pos + needle.size())) {
+        ++count;
+    }
+    return count;
+}
+} // anonymous namespace
+
+// Formatting a segment_set must stay bounded regardless of its size: it
+// prints at most 8 segments. The fmt path is asserted separately from
+// operator<< because fmt resolves formatters independently of the ostream
+// operator (e.g. fmt/ranges.h matches segment_set as a range) and has
+// silently printed every segment in the past, OOM-aborting shards
+// mid-log-statement on partitions with thousands of segments.
+TEST_F(SegmentSetFixtureTest, format_is_bounded) {
+    using o = model::offset;
+    size_t dir_idx = 100;
+    auto make_set = [&](int num_segs) {
+        ss::make_directory(ss::format("{}", dir_idx)).get();
+        segment_set::underlying_t segs;
+        for (int i = 0; i < num_segs; ++i) {
+            segs.push_back(
+              make_segment(
+                dir_idx, test_case::segment_spec(o{2 * i}, o{2 * i + 1}))
+                .get());
+        }
+        ++dir_idx;
+        return segment_set{std::move(segs)};
+    };
+
+    auto large = make_set(10);
+    auto via_fmt = fmt::format("{}", large);
+    EXPECT_EQ(count_occurrences(via_fmt, "offset_tracker"), 8);
+    EXPECT_THAT(via_fmt, testing::HasSubstr("{size: 10, ["));
+    EXPECT_THAT(via_fmt, testing::HasSubstr("..."));
+
+    std::ostringstream os;
+    os << large;
+    EXPECT_EQ(os.str(), via_fmt);
+
+    auto small_fmt = fmt::format("{}", make_set(3));
+    EXPECT_EQ(count_occurrences(small_fmt, "offset_tracker"), 3);
+    EXPECT_THAT(small_fmt, testing::Not(testing::HasSubstr("...")));
+}
+
 } // namespace storage