Add storage_e2e_single_threaded test in bazel

.clang-tidy

@@ -1,7 +1,7 @@
 ---
 Checks:          'clang-diagnostic-*,clang-analyzer-*,cert-*,cppcoreguidelines-*,hicpp-*,modernize-*,performance-*,misc-*,bugprone-*,-modernize-use-trailing-return-type,-modernize-use-nodiscard,-hicpp-named-parameter,-misc-include-cleaner,-clang-analyzer-optin.core.EnumCastOutOfRange'
 WarningsAsErrors: ''
-HeaderFilterRegex: '.*'
+HeaderFilterRegex: '^(?!external/.*).*'
 FormatStyle:    file
 CheckOptions:
   - key:             cert-dcl16-c.NewSuffixes

MODULE.bazel

@@ -13,9 +13,9 @@ git_override(
 bazel_dep(name = "rules_boost", repo_name = "com_github_nelhage_rules_boost")
 archive_override(
     module_name = "rules_boost",
-    integrity = "sha256-Sk6Rjh7r4pUyRYQ4QO2qhgZYvdkWbyrLJTCl9oY6Z7I=",
-    strip_prefix = "rules_boost-21cd44506bd15e06db89aa2866b9be90002d8eec",
-    urls = "https://github.com/nelhage/rules_boost/archive/21cd44506bd15e06db89aa2866b9be90002d8eec.tar.gz",
+    integrity = "sha256-ICuTlyfxD3u1tr12YyeT5qPcA/ZkEUvtMDA2ZRzz1zc=",
+    strip_prefix = "rules_boost-e23cc59f87d5049618472d6ce0ca0ed5ef0c23dc",
+    urls = "https://github.com/nelhage/rules_boost/archive/e23cc59f87d5049618472d6ce0ca0ed5ef0c23dc.tar.gz",
 )
 
 non_module_boost_repositories = use_extension("@com_github_nelhage_rules_boost//:boost/repositories.bzl", "non_module_dependencies")

bazel/build.bzl

@@ -19,7 +19,8 @@ def redpanda_cc_library(
         visibility = None,
         include_prefix = None,
         copts = [],
-        deps = []):
+        deps = [],
+        tags = []):
     """
     Define a Redpanda C++ library.
     """
@@ -36,6 +37,7 @@ def redpanda_cc_library(
         implementation_deps = implementation_deps,
         deps = deps,
         copts = redpanda_copts() + copts,
+        tags = tags,
         features = [
             "layering_check",
         ],

bazel/repositories.bzl

@@ -167,19 +167,19 @@ def data_dependency():
     http_archive(
         name = "seastar",
         build_file = "//bazel/thirdparty:seastar.BUILD",
-        sha256 = "eefe8af689aac287971738e4d97ed495c7a285a7fe18b1aa8bbbfe025b528e6c",
-        strip_prefix = "seastar-c21f48b4aced07a80308215f7d801503a223d9db",
-        url = "https://github.com/redpanda-data/seastar/archive/c21f48b4aced07a80308215f7d801503a223d9db.tar.gz",
+        sha256 = "948e5f8ad769114a4721c81d5829f675ac51482e4a9a106d57ab313104308666",
+        strip_prefix = "seastar-4350d7960ec5ac8a1cf9274316c8f1ab1896d24e",
+        url = "https://github.com/redpanda-data/seastar/archive/4350d7960ec5ac8a1cf9274316c8f1ab1896d24e.tar.gz",
         patches = ["//bazel/thirdparty:seastar-fortify-source.patch"],
         patch_args = ["-p1"],
     )
 
     http_archive(
         name = "unordered_dense",
         build_file = "//bazel/thirdparty:unordered_dense.BUILD",
-        sha256 = "98c9d02ff8761d50a2cb6ebd53f78f7d311f6980aef509efdcdaa5f3868ca06c",
-        strip_prefix = "unordered_dense-9338f301522a965309ecec58ce61f54a52fb5c22",
-        url = "https://github.com/redpanda-data/unordered_dense/archive/9338f301522a965309ecec58ce61f54a52fb5c22.tar.gz",
+        sha256 = "8393d08b2a41949c70345926515036df55643e80118b608bcec6f4202d4a3026",
+        strip_prefix = "unordered_dense-f30ed41b58af8c79788e8581fe57a6faf856258e",
+        url = "https://github.com/martinus/unordered_dense/archive/f30ed41b58af8c79788e8581fe57a6faf856258e.tar.gz",
     )
 
     http_archive(

cmake/dependencies.cmake

@@ -46,7 +46,7 @@ set(CMAKE_CXX_STANDARD
   "C++ standard to build with.")
 fetch_dep(seastar
   REPO https://github.com/redpanda-data/seastar.git
-  TAG v24.2.x
+  TAG v24.3.x
   PATCH_COMMAND sed -i "s/add_subdirectory (tests/# add_subdirectory (tests/g" CMakeLists.txt)
 
 fetch_dep(avro

src/v/cloud_io/BUILD

@@ -128,3 +128,19 @@ redpanda_cc_library(
         "@seastar",
     ],
 )
+
+redpanda_cc_library(
+    name = "basic_cache_service_api",
+    srcs = [
+        "basic_cache_service_api.cc",
+    ],
+    hdrs = [
+        "basic_cache_service_api.h",
+    ],
+    include_prefix = "cloud_io",
+    visibility = ["//visibility:public"],
+    deps = [
+        "//src/v/base",
+        "@seastar",
+    ],
+)

src/v/cloud_io/CMakeLists.txt

@@ -2,6 +2,7 @@ v_cc_library(
   NAME cloud_io
   SRCS
     auth_refresh_bg_op.cc
+    basic_cache_service_api.cc
     io_resources.cc
     io_result.cc
     remote.cc

src/v/cloud_io/basic_cache_service_api.cc

@@ -0,0 +1,59 @@
+/*
+ * Copyright 2021 Redpanda Data, Inc.
+ *
+ * Licensed as a Redpanda Enterprise file under the Redpanda Community
+ * License (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * https://github.com/redpanda-data/redpanda/blob/master/licenses/rcl.md
+ */
+
+#include "cloud_io/basic_cache_service_api.h"
+
+#include <seastar/core/lowres_clock.hh>
+#include <seastar/core/manual_clock.hh>
+
+namespace cloud_io {
+
+std::ostream& operator<<(std::ostream& o, cache_element_status s) {
+    switch (s) {
+    case cache_element_status::available:
+        o << "cache_element_available";
+        break;
+    case cache_element_status::not_available:
+        o << "cache_element_not_available";
+        break;
+    case cache_element_status::in_progress:
+        o << "cache_element_in_progress";
+        break;
+    }
+    return o;
+}
+
+template<class Clock>
+void basic_space_reservation_guard<Clock>::wrote_data(
+  uint64_t written_bytes, size_t written_objects) {
+    // Release the reservation, and update usage stats for how much we actually
+    // wrote.
+    _cache.reserve_space_release(
+      _bytes, _objects, written_bytes, written_objects);
+
+    // This reservation is now used up.
+    _bytes = 0;
+    _objects = 0;
+}
+
+template<class Clock>
+basic_space_reservation_guard<Clock>::~basic_space_reservation_guard() {
+    if (_bytes || _objects) {
+        // This is the case of a failed write, where wrote_data was never
+        // called: release the reservation and do not acquire any space
+        // usage for the written data (there should be none).
+        _cache.reserve_space_release(_bytes, _objects, 0, 0);
+    }
+}
+
+template class basic_space_reservation_guard<ss::lowres_clock>;
+template class basic_space_reservation_guard<ss::manual_clock>;
+
+} // namespace cloud_io

src/v/cloud_io/basic_cache_service_api.h

@@ -0,0 +1,159 @@
+/*
+ * Copyright 2021 Redpanda Data, Inc.
+ *
+ * Licensed as a Redpanda Enterprise file under the Redpanda Community
+ * License (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * https://github.com/redpanda-data/redpanda/blob/master/licenses/rcl.md
+ */
+
+#pragma once
+
+#include "base/seastarx.h"
+#include "base/units.h"
+
+#include <seastar/core/file.hh>
+#include <seastar/core/future.hh>
+#include <seastar/core/io_priority_class.hh>
+#include <seastar/core/iostream.hh>
+#include <seastar/core/lowres_clock.hh>
+
+#include <filesystem>
+#include <iterator>
+#include <optional>
+#include <set>
+#include <string_view>
+
+namespace cloud_io {
+
+static constexpr size_t default_write_buffer_size = 128_KiB;
+static constexpr unsigned default_write_behind = 10;
+static constexpr size_t default_read_buffer_size = 128_KiB;
+static constexpr unsigned default_read_ahead = 10;
+
+struct [[nodiscard]] cache_item {
+    ss::file body;
+    size_t size;
+};
+
+struct [[nodiscard]] cache_item_str {
+    ss::input_stream<char> body;
+    size_t size;
+};
+
+enum class [[nodiscard]] cache_element_status {
+    available,
+    not_available,
+    in_progress,
+};
+
+std::ostream& operator<<(std::ostream& o, cache_element_status);
+
+template<class Clock>
+class basic_cache_service_api;
+
+/// RAII guard for bytes reserved in the cache: constructed prior to a call
+/// to cache::put, and may be destroyed afterwards.
+template<class Clock>
+class basic_space_reservation_guard {
+public:
+    basic_space_reservation_guard(
+      basic_cache_service_api<Clock>& cache,
+      uint64_t bytes,
+      size_t objects) noexcept
+      : _cache(cache)
+      , _bytes(bytes)
+      , _objects(objects) {}
+
+    basic_space_reservation_guard(const basic_space_reservation_guard&)
+      = delete;
+    basic_space_reservation_guard() = delete;
+    basic_space_reservation_guard(basic_space_reservation_guard&& rhs) noexcept
+      : _cache(rhs._cache)
+      , _bytes(rhs._bytes)
+      , _objects(rhs._objects) {
+        rhs._bytes = 0;
+        rhs._objects = 0;
+    }
+
+    ~basic_space_reservation_guard();
+
+    /// After completing the write operation that this space reservation
+    /// protected, indicate how many bytes were really written: this is used to
+    /// atomically update cache usage stats to free the reservation and update
+    /// the bytes used stats together.
+    ///
+    /// May only be called once per reservation.
+    void wrote_data(uint64_t, size_t);
+
+private:
+    basic_cache_service_api<Clock>& _cache;
+
+    // Size acquired at time of reservation
+    uint64_t _bytes{0};
+    size_t _objects{0};
+};
+
+template<class Clock = ss::lowres_clock>
+class basic_cache_service_api {
+public:
+    basic_cache_service_api() = default;
+    basic_cache_service_api(const basic_cache_service_api&) = default;
+    basic_cache_service_api& operator=(const basic_cache_service_api&)
+      = default;
+    basic_cache_service_api(basic_cache_service_api&&) = default;
+    basic_cache_service_api& operator=(basic_cache_service_api&&) = default;
+    virtual ~basic_cache_service_api() = default;
+
+    /// Get cached value as a stream if it exists
+    ///
+    /// \param key is a cache key
+    /// \param read_buffer_size is a read buffer size for the iostream
+    /// \param readahead number of pages that can be read asynchronously
+    virtual ss::future<std::optional<cache_item_str>> get(
+      std::filesystem::path key,
+      ss::io_priority_class io_priority,
+      size_t read_buffer_size = default_read_buffer_size,
+      unsigned int read_ahead = default_read_ahead)
+      = 0;
+
+    /// Add new value to the cache, overwrite if it's already exist
+    ///
+    /// \param key is a cache key
+    /// \param io_priority is an io priority of disk write operation
+    /// \param data is an input stream containing data
+    /// \param write_buffer_size is a write buffer size for disk write
+    /// \param write_behind number of pages that can be written asynchronously
+    /// \param reservation caller must have reserved cache space before
+    /// proceeding with put
+    virtual ss::future<> put(
+      std::filesystem::path key,
+      ss::input_stream<char>& data,
+      basic_space_reservation_guard<Clock>& reservation,
+      ss::io_priority_class io_priority,
+      size_t write_buffer_size = default_write_buffer_size,
+      unsigned int write_behind = default_write_behind)
+      = 0;
+
+    /// \brief Checks if the value is cached
+    ///
+    /// \note returned value \c cache_element_status::in_progress is
+    /// shard-local, it indicates that the object is being written by this
+    /// shard. If the value is being written by another shard, this function
+    /// returns only committed result. The value
+    /// \c cache_element_status::in_progress can be used as a hint since ntp are
+    /// stored on the same shard most of the time.
+    virtual ss::future<cache_element_status>
+    is_cached(const std::filesystem::path& key) = 0;
+
+    // Call this before starting a download, to trim the cache if necessary
+    // and wait until enough free space is available.
+    virtual ss::future<basic_space_reservation_guard<Clock>>
+      reserve_space(uint64_t, size_t) = 0;
+
+    // Release capacity acquired via `reserve_space`.  This spawns
+    // a background fiber in order to be callable from the guard destructor.
+    virtual void reserve_space_release(uint64_t, size_t, uint64_t, size_t) = 0;
+};
+} // namespace cloud_io

src/v/cloud_storage/BUILD

@@ -125,6 +125,7 @@ redpanda_cc_library(
         "//src/v/bytes:iostream",
         "//src/v/bytes:streambuf",
         "//src/v/cloud_io:auth_refresh_bg_op",
+        "//src/v/cloud_io:basic_cache_service_api",
         "//src/v/cloud_io:io_resources",
         "//src/v/cloud_io:io_result",
         "//src/v/cloud_io:remote",