archival: Fence spillover command

Signed-off-by: Evgeny Lazin <4lazin@gmail.com>

Author: Lazin

src/v/cluster/archival/ntp_archiver_service.cc

@@ -2996,6 +2996,11 @@ ss::future<> ntp_archiver::apply_spillover() {
     if (!may_begin_uploads()) {
         co_return;
     }
+    archival_stm_fence fence = {
+      .read_write_fence
+      = _parent.archival_meta_stm()->manifest().get_applied_offset(),
+      .emit_rw_fence_cmd = emit_read_write_fence(_feature_table),
+    };
 
     const auto manifest_upload_timeout = _conf->manifest_upload_timeout();
     const auto manifest_upload_backoff = _conf->cloud_storage_initial_backoff();
@@ -3034,6 +3039,12 @@ ss::future<> ntp_archiver::apply_spillover() {
         auto tail = [&]() {
             cloud_storage::spillover_manifest tail(_ntp, _rev);
             for (const auto& meta : manifest()) {
+                vlog(
+                  _rtclog.trace,
+                  "Adding segment {} to the spillover manifest that starts at "
+                  "{}",
+                  meta,
+                  tail.get_start_offset().value_or(model::offset{}));
                 tail.add(meta);
                 // No performance impact since all writes here are
                 // sequential.
@@ -3090,6 +3101,13 @@ ss::future<> ntp_archiver::apply_spillover() {
         auto deadline = ss::lowres_clock::now() + sync_timeout;
 
         auto batch = _parent.archival_meta_stm()->batch_start(deadline, _as);
+        if (fence.emit_rw_fence_cmd) {
+            vlog(
+              _rtclog.debug,
+              "spillover, read-write fence: {}",
+              fence.read_write_fence);
+            batch.read_write_fence(fence.read_write_fence);
+        }
         batch.spillover(spillover_meta);
         if (manifest().get_archive_start_offset() == model::offset{}) {
             vlog(