[v25.3.x] iobuf: fix operator<=> and operator== for string_view comparisons

src/v/bytes/iobuf.cc

@@ -182,43 +182,27 @@ std::strong_ordering iobuf::operator<=>(const iobuf& o) const {
 }
 
 bool iobuf::operator==(std::string_view o) const {
-    if (_size != o.size()) {
-        return false;
-    }
-    bool are_equal = true;
-    std::string_view::size_type n = 0;
-    auto in = iobuf::iterator_consumer(cbegin(), cend());
-    std::ignore = in.consume(
-      size_bytes(), [&are_equal, &o, &n](const char* src, size_t fg_sz) {
-          /// Both strings are equiv in total size, so its safe to assume
-          /// the next chunk to compare is the remaining to cmp or the
-          /// fragment size
-          const auto size = std::min((o.size() - n), fg_sz);
-          std::string_view a_view(src, size);
-          std::string_view b_view(o.data() + n, size);
-          n += size;
-          are_equal &= (a_view == b_view);
-          return !are_equal ? ss::stop_iteration::yes : ss::stop_iteration::no;
-      });
-    return are_equal;
+    return size_bytes() == o.size()
+           && (*this <=> o) == std::strong_ordering::equal;
 }
 
 std::strong_ordering iobuf::operator<=>(std::string_view o) const {
-    std::strong_ordering cmp = std::strong_ordering::equal;
-    auto in = iobuf::iterator_consumer(cbegin(), cend());
-    std::string_view other = o;
-    std::ignore = in.consume(
-      std::min(size_bytes(), o.size()),
-      [&cmp, &other](const char* src, size_t fg_sz) {
-          cmp = std::string_view(src, fg_sz) <=> other;
-          other.remove_prefix(std::min(fg_sz, other.size()));
-          return cmp == std::strong_ordering::equal ? ss::stop_iteration::yes
-                                                    : ss::stop_iteration::no;
-      });
-    if (cmp == std::strong_ordering::equal) {
-        cmp = size_bytes() <=> o.size();
+    auto other = o;
+    // first compare the common length prefix
+    for (const auto& frag : *this) {
+        auto compare_len = std::min(frag.size(), other.size());
+        std::strong_ordering cmp = std::string_view(frag.get(), compare_len)
+                                   <=> other.substr(0, compare_len);
+        if (cmp != std::strong_ordering::equal) {
+            return cmp;
+        }
+        other.remove_prefix(compare_len);
+        if (other.empty()) {
+            break;
+        }
     }
-    return cmp;
+    // prefix was equal, only size matters now
+    return size_bytes() <=> o.size();
 }
 
 /**

src/v/bytes/tests/iobuf_bench.cc

@@ -49,21 +49,42 @@ size_t move_bench() {
     return inner_iters * 2;
 }
 
-template<size_t Size, typename cmp_fn, bool same>
+// Overloaded make_rhs: convert iobuf to the desired RHS type
+template<typename T>
+T make_rhs(iobuf&& src);
+
+template<>
+iobuf make_rhs<iobuf>(iobuf&& src) {
+    return std::move(src);
+}
+
+template<>
+ss::sstring make_rhs<ss::sstring>(iobuf&& src) {
+    return src.linearize_to_string();
+}
+
+// Get the view type for comparison (iobuf& or string_view)
+const iobuf& as_cmp_arg(const iobuf& b) { return b; }
+std::string_view as_cmp_arg(const ss::sstring& b) { return b; }
+
+template<size_t Size, typename cmp_fn, bool same, typename rhs_type = iobuf>
 size_t cmp_bench() {
-    iobuf a = make_iobuf(Size);
-    iobuf a_copy = a.copy();
-    iobuf b = make_iobuf(Size, !same);
-    iobuf b_copy;
-    for (const auto& frag : b) {
+    // LHS: iobuf (single fragment and fragmented variants)
+    iobuf lhs = make_iobuf(Size);
+    iobuf lhs_fragmented;
+    for (const auto& frag : lhs) {
         for (char c : std::string_view(frag.get(), frag.size())) {
-            b_copy.append(&c, 1);
+            lhs_fragmented.append(&c, 1);
         }
     }
+
+    // RHS: converted to target type (iobuf or ss::sstring)
+    auto rhs = make_rhs<rhs_type>(make_iobuf(Size, !same));
+
     perf_tests::start_measuring_time();
     for (auto i = inner_iters; i--;) {
-        perf_tests::do_not_optimize(cmp_fn{}(a, b));
-        perf_tests::do_not_optimize(cmp_fn{}(a_copy, b_copy));
+        perf_tests::do_not_optimize(cmp_fn{}(lhs, as_cmp_arg(rhs)));
+        perf_tests::do_not_optimize(cmp_fn{}(lhs_fragmented, as_cmp_arg(rhs)));
     }
     perf_tests::stop_measuring_time();
     return inner_iters * 2;
@@ -147,3 +168,20 @@ PERF_TEST(iobuf, eq_bench_large_same) {
 PERF_TEST(iobuf, append_bench_small) { return append_bench<1'000, 4>(); }
 PERF_TEST(iobuf, append_bench_medium) { return append_bench<1'000, 40_KiB>(); }
 PERF_TEST(iobuf, append_bench_large) { return append_bench<1'000, 400_KiB>(); }
+
+// iobuf vs string_view comparisons
+// clang-format off
+PERF_TEST(iobuf, sv_eq_0000)      { return cmp_bench<   0, std::equal_to<>, false, ss::sstring>(); }
+PERF_TEST(iobuf, sv_eq_0001)      { return cmp_bench<   1, std::equal_to<>, false, ss::sstring>(); }
+PERF_TEST(iobuf, sv_eq_1024)      { return cmp_bench<1024, std::equal_to<>, false, ss::sstring>(); }
+PERF_TEST(iobuf, sv_eq_0000_same) { return cmp_bench<   0, std::equal_to<>, true,  ss::sstring>(); }
+PERF_TEST(iobuf, sv_eq_0001_same) { return cmp_bench<   1, std::equal_to<>, true,  ss::sstring>(); }
+PERF_TEST(iobuf, sv_eq_1024_same) { return cmp_bench<1024, std::equal_to<>, true,  ss::sstring>(); }
+
+PERF_TEST(iobuf, sv_cmp_0000)      { return cmp_bench<   0, std::less<>, false, ss::sstring>(); }
+PERF_TEST(iobuf, sv_cmp_0001)      { return cmp_bench<   1, std::less<>, false, ss::sstring>(); }
+PERF_TEST(iobuf, sv_cmp_1024)      { return cmp_bench<1024, std::less<>, false, ss::sstring>(); }
+PERF_TEST(iobuf, sv_cmp_0000_same) { return cmp_bench<   0, std::less<>, true,  ss::sstring>(); }
+PERF_TEST(iobuf, sv_cmp_0001_same) { return cmp_bench<   1, std::less<>, true,  ss::sstring>(); }
+PERF_TEST(iobuf, sv_cmp_1024_same) { return cmp_bench<1024, std::less<>, true,  ss::sstring>(); }
+// clang-format on

src/v/bytes/tests/iobuf_tests.cc

@@ -856,3 +856,68 @@ SEASTAR_THREAD_TEST_CASE(iobuf_linearize) {
     iobuf large = iobuf::from(std::string(128_KiB + 1, 'a'));
     BOOST_CHECK_THROW(large.linearize_to_string(), std::runtime_error);
 }
+
+SEASTAR_THREAD_TEST_CASE(iobuf_spaceship_string_view) {
+    // Multi-fragment iobuf compared against string_view, these are potential
+    // sources of bugs as the comparison is done fragment by fragment.
+    iobuf buf = iobuf::from("ab");
+    buf.append_fragments(iobuf::from("cd"));
+
+    // Equal comparison
+    BOOST_CHECK(
+      (buf <=> std::string_view("abcd")) == std::strong_ordering::equal);
+
+    // Less than - iobuf is less
+    BOOST_CHECK(
+      (buf <=> std::string_view("abce")) == std::strong_ordering::less);
+
+    // Greater than - iobuf is greater
+    BOOST_CHECK(
+      (buf <=> std::string_view("abcc")) == std::strong_ordering::greater);
+
+    // Size difference - iobuf shorter
+    BOOST_CHECK(
+      (buf <=> std::string_view("abcde")) == std::strong_ordering::less);
+
+    // Size difference - iobuf longer
+    BOOST_CHECK(
+      (buf <=> std::string_view("abc")) == std::strong_ordering::greater);
+
+    // Test with many small fragments - difference in second fragment
+    iobuf buf2;
+    buf2.append_fragments(iobuf::from("a"));
+    buf2.append_fragments(iobuf::from("b"));
+    buf2.append_fragments(iobuf::from("c"));
+
+    BOOST_CHECK(
+      (buf2 <=> std::string_view("abc")) == std::strong_ordering::equal);
+    BOOST_CHECK(
+      (buf2 <=> std::string_view("abd")) == std::strong_ordering::less);
+    BOOST_CHECK(
+      (buf2 <=> std::string_view("abb")) == std::strong_ordering::greater);
+
+    // Test mismatch in middle of second fragment
+    iobuf buf3;
+    buf3.append_fragments(iobuf::from("xx"));
+    buf3.append_fragments(iobuf::from("yy"));
+
+    BOOST_CHECK(
+      (buf3 <=> std::string_view("xxyy")) == std::strong_ordering::equal);
+    BOOST_CHECK(
+      (buf3 <=> std::string_view("xxyz")) == std::strong_ordering::less);
+    BOOST_CHECK(
+      (buf3 <=> std::string_view("xxyx")) == std::strong_ordering::greater);
+
+    // Past bug reproducers
+    iobuf buf4;
+    buf4.append_fragments(iobuf::from("b"));
+    buf4.append_fragments(iobuf::from("a"));
+    BOOST_CHECK(
+      (buf4 <=> std::string_view("ax")) == std::strong_ordering::greater);
+
+    iobuf buf5;
+    buf5.append_fragments(iobuf::from("a"));
+    buf5.append_fragments(iobuf::from("z"));
+    BOOST_CHECK(
+      (buf5 <=> std::string_view("bx")) == std::strong_ordering::less);
+}