sr/seq_writer: fix retry crash in soft delete

do_delete_subject_impermanent had a latent ordering bug:
get_versions(include_deleted::no) was called BEFORE
is_subject_deleted(), so on retry after a write collision
where the subject was already soft-deleted (by the winning
writer), get_versions would throw subject_not_found — which
propagated as HTTP 404 to the client.

Fix by checking is_subject_deleted first. A cached version
list (via lw_shared_ptr) preserves the pre-delete version
list across retries, since after a subject-level soft delete
all versions are marked deleted and the store can no longer
distinguish individually-deleted versions from those deleted
as part of the subject-level operation.

This bug was never observed on the kafka::client transport
because the full Kafka protocol round-trip (SASL, request
queuing, acks) acts as a natural pacer between different
nodes' writes, making write collisions rare. The RPC
transport's lower latency tightens the timing window enough
to trigger collisions routinely in integration tests.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
Signed-off-by: Oren Leiman <oren.leiman@redpanda.com>

Author: oleiman

src/v/pandaproxy/schema_registry/seq_writer.cc

@@ -586,13 +586,31 @@ seq_writer::do_delete_subject_impermanent(
   context_subject sub, model::offset write_at) {
     co_await check_mutable(sub.ctx, sub.sub);
 
+    // On retry after a write collision, the subject may already be
+    // soft-deleted (by the collided write or a concurrent writer). Return
+    // the cached version list from the first attempt, since the store no
+    // longer distinguishes individually-deleted versions after a
+    // subject-level soft delete.
+    if (co_await _store.is_subject_deleted(sub)) {
+        if (
+          _delete_versions_cache.has_value()
+          && _delete_versions_cache->sub == sub) {
+            auto versions = std::move(_delete_versions_cache->versions);
+            _delete_versions_cache.reset();
+            co_return versions;
+        }
+        // Subject was already deleted before our first attempt.
+        co_return co_await _store.get_versions(sub, include_deleted::yes);
+    }
+
     // Grab the versions before they're gone.
     auto versions = co_await _store.get_versions(sub, include_deleted::no);
 
-    // Inspect the subject to see if its already deleted
-    if (co_await _store.is_subject_deleted(sub)) {
-        co_return std::make_optional(std::move(versions));
-    }
+    // Cache versions for potential retry — after a subject-level soft
+    // delete all versions are marked deleted and the pre-delete list
+    // cannot be reconstructed from the store. Tagged with subject so
+    // stale entries from a prior delete of a different subject are ignored.
+    _delete_versions_cache.emplace(delete_version_cache{sub, versions.copy()});
 
     // Check that the subject is not referenced
     if (co_await _store.is_referenced(sub, std::nullopt)) {
@@ -624,6 +642,7 @@ seq_writer::do_delete_subject_impermanent(
     }
 
     if (co_await produce_and_apply(write_at, std::move(rb).build())) {
+        _delete_versions_cache.reset();
         co_return versions;
     } else {
         // Pass up a None, our caller's cue to retry

src/v/pandaproxy/schema_registry/seq_writer.h

@@ -225,6 +225,15 @@ class seq_writer final : public ss::peering_sharded_service<seq_writer> {
     /// Shard 0 only: Serialize write operations.
     ssx::semaphore _write_sem{1, "pproxy/schema-write"};
 
+    /// Shard 0 only: cached version list for delete_subject_impermanent
+    /// retries. Protected by _write_sem. Tagged with subject so stale
+    /// entries from a previous delete of a different subject are ignored.
+    struct delete_version_cache {
+        context_subject sub;
+        chunked_vector<schema_version> versions;
+    };
+    std::optional<delete_version_cache> _delete_versions_cache;
+
     // ======================
     // End of Shard 0 state
 };

src/v/pandaproxy/schema_registry/test/BUILD

@@ -216,6 +216,7 @@ redpanda_cc_btest(
         ":utils",
         "//src/v/model",
         "//src/v/pandaproxy/schema_registry:core",
+        "//src/v/storage:record_batch_builder",
         "//src/v/test_utils:seastar_boost",
         "@boost//:test",
         "@seastar",

src/v/pandaproxy/schema_registry/test/consume_to_store.cc

@@ -299,3 +299,137 @@ SEASTAR_THREAD_TEST_CASE(test_writes_disabled) {
           return e.code() == pps::error_code::writes_disabled;
       });
 }
+
+/// Transport that simulates a write collision during soft delete.
+///
+/// Every write to the _schemas topic is tagged with the offset the writer
+/// expects to land at. If another node writes first, the offset doesn't
+/// match — a "write collision" — and the operation retries. Before
+/// retrying, read_sync() catches up on the topic, consuming the winning
+/// writer's record into the store.
+///
+/// This transport drives that sequence:
+///   1. Initial read_sync(): HWM=1, _loaded_offset already 0 → no-op
+///   2. First produce (the delete): returns wrong offset → collision
+///   3. Retry read_sync(): HWM=2, consumes the competing delete batch
+///   4. Retry attempt: subject already deleted → returns cached versions
+class colliding_transport final : public pps::transport {
+public:
+    explicit colliding_transport(
+      pps::context_subject sub, model::offset collision_offset)
+      : _collision_offset(collision_offset) {
+        // Build the competing writer's delete batch at the collision offset.
+        storage::record_batch_builder rb{
+          model::record_batch_type::raft_data, collision_offset};
+        rb.add_raw_kv(
+          to_json_iobuf(
+            pps::delete_subject_key{
+              .seq{collision_offset}, .node{model::node_id{99}}, .sub{sub}}),
+          to_json_iobuf(pps::delete_subject_value{.sub{sub}}));
+        _competing_batch = std::move(rb).build();
+    }
+
+    ss::future<> stop() final { return ss::now(); }
+    ss::future<cluster::errc> create_topic(
+      model::topic_namespace_view,
+      int32_t,
+      cluster::topic_properties,
+      std::optional<int16_t>) final {
+        throw std::runtime_error(
+          "colliding_transport::create_topic not implemented");
+    }
+    bool has_ephemeral_credentials() const final { return false; }
+
+    ss::future<pps::produce_result> produce(model::record_batch) override {
+        ++_produce_calls;
+        if (_produce_calls == 1) {
+            // Collision: return an offset that doesn't match write_at.
+            co_return pps::produce_result{
+              .base_offset = _collision_offset + model::offset{1}};
+        }
+        // The retry should never produce — it finds the subject
+        // already deleted and returns the cached version list.
+        throw std::runtime_error("unexpected second produce call");
+    }
+
+    ss::future<model::offset> get_high_watermark() override {
+        if (_produce_calls == 0) {
+            // Before collision: only the schema record at offset 0.
+            co_return model::offset{1};
+        }
+        // After collision: schema at 0, competing delete at 1.
+        co_return model::offset{2};
+    }
+
+    ss::future<> consume_range(
+      model::offset start,
+      model::offset end,
+      ss::noncopyable_function<ss::future<>(model::record_batch)> consumer)
+      override {
+        if (
+          _competing_batch.has_value() && start <= _collision_offset
+          && _collision_offset < end) {
+            co_await consumer(std::move(*_competing_batch));
+            _competing_batch.reset();
+        }
+    }
+
+    int produce_calls() const { return _produce_calls; }
+
+private:
+    std::optional<model::record_batch> _competing_batch;
+    model::offset _collision_offset;
+    int _produce_calls{0};
+};
+
+SEASTAR_THREAD_TEST_CASE(test_delete_subject_write_collision_retry) {
+    pps::enable_qualified_subjects::set_local(true);
+    auto reset_flag = ss::defer(
+      [] { pps::enable_qualified_subjects::reset_local(); });
+
+    // Store setup: insert a schema so there's something to delete.
+    pps::sharded_store store;
+    store.start(pps::is_mutable::yes, ss::default_smp_service_group()).get();
+    auto stop_store = ss::defer([&store]() { store.stop().get(); });
+
+    const auto version = pps::schema_version{1};
+    store
+      .upsert(
+        pps::seq_marker{
+          .seq = model::offset{0},
+          .node = model::node_id{0},
+          .version = version,
+          .key_type = pps::seq_marker_key_type::schema},
+        pps::subject_schema{subject0, int_def0.share()},
+        id0,
+        version,
+        pps::is_deleted::no)
+      .get();
+
+    // The competing delete will land at offset 1 (the next available).
+    colliding_transport transport(subject0, model::offset{1});
+
+    ss::sharded<pps::seq_writer> seq;
+    seq
+      .start(
+        model::node_id{0},
+        ss::default_smp_service_group(),
+        &transport,
+        std::reference_wrapper(store),
+        ss::sharded_parameter(
+          [] { return std::make_unique<sequence_state_checker_test>(); }))
+      .get();
+    auto stop_seq = ss::defer([&seq]() { seq.stop().get(); });
+
+    // Advance _loaded_offset past the schema record.
+    seq.local().advance_offset(model::offset{0}).get();
+
+    // First produce collides. The retry's read_sync consumes the
+    // competing delete batch, finds the subject already soft-deleted,
+    // and returns the cached version list from the first attempt.
+    auto versions = seq.local().delete_subject_impermanent(subject0).get();
+
+    BOOST_REQUIRE_EQUAL(versions.size(), 1);
+    BOOST_CHECK_EQUAL(versions[0], version);
+    BOOST_CHECK_EQUAL(transport.produce_calls(), 1);
+}