utils: add managed_bytes_basic_view::byte_iterator

bytes-wise iterator which works both as bidirectional-iterator and as
output-iterator (for mutable views). Allows using managed_bytes_view in
algorithms which are iterator based.

Added unit tests for covering the iterator functionality.

test/boost/managed_bytes_test.cc

@@ -12,6 +12,8 @@
 #include "utils/serialization.hh"
 #include "test/lib/random_utils.hh"
 #include <boost/test/unit_test.hpp>
+#include <iterator>
+#include <memory>
 #include <unordered_set>
 
 struct fragmenting_allocation_strategy : standard_allocation_strategy {
@@ -412,3 +414,156 @@ static constexpr bool constexpr_managed_bytes() {
 }
 
 static_assert(constexpr_managed_bytes());
+
+// Factory for managed_bytes instances whose backing memory is allocated
+// through a fragmenting_allocation_strategy.  The factory must outlive
+// all managed_bytes it produces.  Returned pointers use a custom deleter
+// that frees through the factory's allocator.
+class managed_bytes_factory {
+    fragmenting_allocation_strategy _alloc;
+
+    struct deleter {
+        fragmenting_allocation_strategy* alloc;
+        void operator()(managed_bytes* mb) const {
+            with_allocator(*alloc, [&] {
+                delete mb;
+            });
+        }
+    };
+public:
+    using pointer = std::unique_ptr<managed_bytes, deleter>;
+
+    // frag_size is the desired *data payload* per fragment.  The allocator
+    // limit must also account for the per-fragment metadata overhead so that
+    // managed_bytes' internal max_seg() computes back to frag_size.
+    managed_bytes_factory(size_t frag_size)
+        : _alloc(frag_size + std::max(sizeof(multi_chunk_blob_storage), sizeof(single_chunk_blob_storage)))
+    {}
+
+    // Build a managed_bytes holding `data`, fragmented so that each
+    // fragment is exactly the configured frag_size bytes (the last
+    // fragment may be shorter).
+    pointer make(bytes_view data) {
+        managed_bytes* mb;
+        with_allocator(_alloc, [&] {
+            mb = new managed_bytes(data);
+        });
+        return pointer(mb, deleter{&_alloc});
+    }
+};
+
+// Fragment size used by byte_iterator tests: small enough to guarantee many
+// chunk boundaries for 100-byte inputs (4-byte payload → 25 fragments, 24
+// cross-chunk transitions).
+static constexpr size_t iter_frag_size = 4;
+
+// Convenience: cast a managed_bytes value_type (int8_t) to uint8_t for
+// comparison against bytes literals without sign-extension surprises.
+static uint8_t as_u8(managed_bytes_view::value_type v) {
+    return static_cast<uint8_t>(v);
+}
+
+BOOST_AUTO_TEST_CASE(test_byte_iterator_empty) {
+    // An empty managed_bytes view has begin() == end() and yields no elements.
+    managed_bytes mb;
+    auto v = managed_bytes_view(mb);
+    BOOST_CHECK(v.begin() == v.end());
+    BOOST_CHECK(!(v.begin() != v.end()));
+}
+
+BOOST_AUTO_TEST_CASE(test_byte_iterator_forward_traversal) {
+    // Forward iteration with pre-increment visits every byte in order,
+    // correctly crossing chunk boundaries.
+    managed_bytes_factory factory(iter_frag_size);
+    auto b = tests::random::get_bytes(100);
+    auto mb = factory.make(b);
+    auto v = managed_bytes_view(*mb);
+
+    size_t idx = 0;
+    for (auto it = v.begin(); it != v.end(); ++it, ++idx) {
+        BOOST_CHECK_EQUAL(as_u8(*it), as_u8(b[idx]));
+    }
+    BOOST_CHECK_EQUAL(idx, b.size());
+}
+
+BOOST_AUTO_TEST_CASE(test_byte_iterator_post_increment) {
+    // Post-increment returns a copy of the iterator before advancing, while
+    // the original moves forward.
+    managed_bytes_factory factory(iter_frag_size);
+    auto b = tests::random::get_bytes(100);
+    auto mb = factory.make(b);
+    auto v = managed_bytes_view(*mb);
+
+    auto it = v.begin();
+    for (size_t i = 0; i < b.size(); ++i) {
+        auto prev = it++;
+        BOOST_CHECK_EQUAL(as_u8(*prev), as_u8(b[i]));
+    }
+    BOOST_CHECK(it == v.end());
+}
+
+BOOST_AUTO_TEST_CASE(test_byte_iterator_equality) {
+    managed_bytes_factory factory(iter_frag_size);
+    auto b = tests::random::get_bytes(100);
+    auto mb = factory.make(b);
+    auto v = managed_bytes_view(*mb);
+
+    // begin() == begin(), end() == end()
+    BOOST_CHECK(v.begin() == v.begin());
+    BOOST_CHECK(v.end()   == v.end());
+    // begin() != end() for non-empty view
+    BOOST_CHECK(v.begin() != v.end());
+
+    // Two iterators advanced the same number of steps compare equal.
+    auto it1 = v.begin();
+    auto it2 = v.begin();
+    for (size_t i = 0; i < 13; ++i) { ++it1; ++it2; }
+    BOOST_CHECK(it1 == it2);
+    ++it1;
+    BOOST_CHECK(it1 != it2);
+}
+
+BOOST_AUTO_TEST_CASE(test_byte_iterator_distance) {
+    // operator- returns the signed distance between two iterators.
+    managed_bytes_factory factory(iter_frag_size);
+    auto b = tests::random::get_bytes(100);
+    auto mb = factory.make(b);
+    auto v = managed_bytes_view(*mb);
+
+    auto begin = v.begin();
+    auto end   = v.end();
+    // Distance from end to begin equals -(size). Note the subtraction is
+    // defined as (lhs._pos - rhs._pos) cast to ptrdiff_t — see the
+    // implementation — so end - begin == size.
+    BOOST_CHECK_EQUAL(end - begin, static_cast<std::ptrdiff_t>(b.size()));
+    BOOST_CHECK_EQUAL(begin - end, -static_cast<std::ptrdiff_t>(b.size()));
+
+    // Advance to a mid-point and verify partial distances.
+    auto mid = v.begin();
+    for (size_t i = 0; i < 37; ++i) {
+        ++mid;
+    }
+    BOOST_CHECK_EQUAL(mid - begin,  37);
+    BOOST_CHECK_EQUAL(end  - mid,   static_cast<std::ptrdiff_t>(b.size()) - 37);
+}
+
+BOOST_AUTO_TEST_CASE(test_byte_iterator_mutable) {
+    // The mutable iterator (iterator, not const_iterator) allows writing
+    // through operator* and must traverse correctly too.
+    managed_bytes_factory factory(iter_frag_size);
+    auto b = tests::random::get_bytes(100);
+    auto mb = factory.make(b);
+    auto mv = managed_bytes_mutable_view(*mb);
+
+    // Increment every byte by 1 using the mutable forward iterator.
+    for (auto it = mv.begin(); it != mv.end(); ++it) {
+        ++(*it);
+    }
+
+    // Verify via the immutable view.
+    auto v = managed_bytes_view(*mb);
+    size_t idx = 0;
+    for (auto it = v.begin(); it != v.end(); ++it, ++idx) {
+        BOOST_CHECK_EQUAL(as_u8(*it), static_cast<uint8_t>(as_u8(b[idx]) + 1));
+    }
+}

utils/managed_bytes.hh

@@ -16,6 +16,8 @@
 #include <seastar/util/alloc_failure_injector.hh>
 #include <type_traits>
 #include <utility>
+#include <iterator>
+#include <cstddef>
 
 class bytes_ostream;
 
@@ -481,6 +483,62 @@ public:
         return func(bv);
     }
 
+    template <typename CharT>
+    requires std::is_same_v<CharT, value_type> || std::is_same_v<CharT, value_type_maybe_const> || std::is_same_v<CharT, char> || std::is_same_v<CharT, const char>
+    class byte_iterator {
+        managed_bytes_basic_view _view;
+
+    public:
+        using iterator_category = std::forward_iterator_tag;
+        using value_type        = CharT;
+        using difference_type   = std::ptrdiff_t;
+        using pointer           = value_type*;
+        using reference         = value_type&;
+
+        byte_iterator() = default;
+
+        explicit byte_iterator(managed_bytes_basic_view view)
+            : _view(view)
+        {}
+
+        reference operator*() const {
+            // value_type might be unsigned, but the underlying data is always signed, so we need to cast it.
+            return reinterpret_cast<reference>(_view.front());
+        }
+
+        byte_iterator& operator++() {
+            _view.remove_prefix(1);
+            return *this;
+        }
+        byte_iterator operator++(int) {
+            auto copy = *this;
+            ++*this;
+            return copy;
+        }
+
+        difference_type operator-(const byte_iterator& o) const {
+            return static_cast<difference_type>(o._view.size_bytes()) - static_cast<difference_type>(_view.size_bytes());
+        }
+
+        bool operator==(const byte_iterator& o) const noexcept {
+            return _view.size_bytes() == o._view.size_bytes();
+        }
+        bool operator!=(const byte_iterator& o) const noexcept {
+            return !(*this == o);
+        }
+    };
+
+    using const_iterator = byte_iterator<value_type_maybe_const>;
+    // For immutable views, iterator and const_iterator are the same (like std::string_view).
+    // For mutable views, iterator allows non-const access.
+    using iterator = std::conditional_t<is_mutable == mutable_view::yes, byte_iterator<value_type>, const_iterator>;
+
+    const_iterator begin() const { return const_iterator(*this); }
+    const_iterator end() const { return const_iterator(); }
+
+    iterator begin() { return iterator(*this); }
+    iterator end() { return iterator(); }
+
     friend managed_bytes_basic_view<mutable_view::no> build_managed_bytes_view_from_internals(bytes_view current_fragment, multi_chunk_blob_storage* next_fragment, size_t size);
 };
 static_assert(FragmentedView<managed_bytes_view>);