archival: Deduplicate archival STM commands

The archival STM commands are not idempotent. The STM implementation
depends on the right order of the log replay. At the same time the STM
maintains its own copy of the insync-offset. This value is stored in the
manifest and in the local snapshot.

This commit adds simple mechanism that checks the offset of every new
command before this command can be applied. If the offset is below the
insync offset of the STM the command is not applied. The implementation
assumes that archival STM batches are either applied or not. The batch
can't be applied partially. This allows us to check only the last offset
of the batch.

Signed-off-by: Evgeny Lazin <[email protected]>

src/v/cluster/archival/archival_metadata_stm.cc

@@ -1050,6 +1050,22 @@ ss::future<> archival_metadata_stm::do_apply(const model::record_batch& b) {
         _manifest->advance_insync_offset(b.last_offset());
         co_return;
     }
+    if (b.last_offset() < _manifest->get_insync_offset()) {
+        // Archival STM commands are not idempotent. Applying same command twice
+        // may cause a problem. This branch is used to deduplicate the commands.
+        // It's only applied to non-data batches that can affect the state of
+        // this STM.
+        // We're assuming that commands are applied in all or nothing
+        // fashion and there is no need to track in-sync offset within the
+        // record batch.
+        vlog(
+          _logger.warn,
+          "Command with offset {} (type: {}) is already applied to the "
+          "archival_metadata_stm",
+          b.base_offset(),
+          b.header().type);
+        co_return;
+    }
 
     if (b.header().type == model::record_batch_type::prefix_truncate) {
         // Special case handling for prefix_truncate batches: these