From 8d4fe8187b06a792cbe84d0b076260dc8d83dba1 Mon Sep 17 00:00:00 2001
From: Evgeny Lazin <evgeny@vectorized.io>
Date: Wed, 30 Aug 2023 19:27:05 +0000
Subject: [PATCH] cloud_storage: Make partition reader span ...

... multiple manifests. If the reader starts in one spillover manifest
it should be able to proceed to the next if needed.

(cherry picked from commit 2ba2f33879befaf521d40ba7e920ec3d136419e3)
---
 src/v/cloud_storage/remote_partition.cc | 35 +++++++++++++++++++++++++
 1 file changed, 35 insertions(+)

diff --git a/src/v/cloud_storage/remote_partition.cc b/src/v/cloud_storage/remote_partition.cc
index fbaa0ebec978..b946e2041aaf 100644
--- a/src/v/cloud_storage/remote_partition.cc
+++ b/src/v/cloud_storage/remote_partition.cc
@@ -625,6 +625,41 @@ class partition_record_batch_reader_impl final
               = co_await _partition->materialized().get_segment_reader_units();
 
             auto maybe_manifest = _view_cursor->manifest();
+            if (
+              maybe_manifest.has_value()
+              && _next_segment_base_offset == model::offset{}
+              && _view_cursor->get_status()
+                   == async_manifest_view_cursor_status::
+                     materialized_spillover) {
+                // End of the manifest is reached, but the cursor is pointing at
+                // the spillover manifest. We need to reset the cursor to the
+                // next manifest and try to find the next segment there.
+                vlog(
+                  _ctxlog.debug,
+                  "maybe_reset_reader, end of the manifest is reached, "
+                  "resetting cursor to the next manifest");
+                auto stop = co_await _view_cursor->next_iter();
+                if (stop == ss::stop_iteration::yes) {
+                    vlog(_ctxlog.debug, "maybe_reset_reader, last manifest");
+                    co_return false;
+                }
+                try {
+                    _next_segment_base_offset
+                      = co_await _view_cursor->with_manifest(
+                        [&](const partition_manifest& m) {
+                            return m.get_start_offset().value_or(
+                              model::offset{});
+                        });
+                } catch (...) {
+                    vlog(
+                      _ctxlog.debug,
+                      "maybe_reset_reader, failed to get next manifest: {}",
+                      std::current_exception());
+                    co_return false;
+                }
+                // NOTE: maybe_manifest is invalidated at this point
+                maybe_manifest = _view_cursor->manifest();
+            }
             if (
               maybe_manifest.has_value()
               && _next_segment_base_offset != model::offset{}) {