[CORE-16776] cluster_link: prefer the source-reported deleted flag for SR sync

src/v/cluster_link/schema_registry_sync/http_source_reader.cc

@@ -285,10 +285,11 @@ http_source_reader::read_subject_version(
     if (!res.has_value()) {
         co_return std::unexpected(to_source_error(std::move(res.error())));
     }
-    // Carry the read through unchanged: the schema plus any unsupported fields
-    // the source served but Redpanda cannot store. The reconciler applies the
-    // configured unsupported-feature policy to
-    // `source_schema_read::unsupported`.
+    // Carry the read through unchanged: the schema, any unsupported fields the
+    // source served but Redpanda cannot store, and whether the source reported
+    // the `deleted` flag. The reconciler applies the configured
+    // unsupported-feature policy to `source_schema_read::unsupported` and
+    // prefers the reported `deleted` over its listing-derived fallback.
     co_return std::move(res.value());
 }
 

src/v/cluster_link/schema_registry_sync/mirroring_task.cc

@@ -602,10 +602,9 @@ ss::future<task::state_transition> mirroring_task::full_source_sync(
                                     node);
         if (!dest_deleted) {
             work.upserts.push_back(node);
-            // Authoritative deleted-state from the listing partition: the
-            // reconciler imports these soft-deleted regardless of the
-            // per-version source body, which a standard source (e.g. Confluent)
-            // does not populate with a `deleted` flag by default.
+            // Listing-derived soft-delete, recorded as the fallback signal: the
+            // reconciler prefers the version body's own `deleted` flag and
+            // consults this set only when the source omits it from the body.
             work.soft_deleted.insert(node);
         }
     }

src/v/cluster_link/schema_registry_sync/reconciler.cc

@@ -45,10 +45,10 @@ constexpr size_t reconcile_reserve_bytes = 100_KiB;
 // on them. Unqualified references resolve against the referring schema's own
 // context.
 chunked_vector<ppsr::subject_version>
-resolve_refs(const ppsr::stored_schema& stored) {
-    const auto& parent_ctx = stored.schema.sub().ctx;
+resolve_refs(const ppsr::subject_schema& schema) {
+    const auto& parent_ctx = schema.sub().ctx;
     chunked_vector<ppsr::subject_version> out;
-    for (const auto& ref : stored.schema.def().refs()) {
+    for (const auto& ref : schema.def().refs()) {
         out.push_back(
           ppsr::subject_version{ref.sub.resolve(parent_ctx), ref.version});
     }
@@ -57,8 +57,8 @@ resolve_refs(const ppsr::stored_schema& stored) {
 
 // Byte-size proxy for a schema body: the canonical definition's length. This
 // is what the byte-semaphore budgets, and what tests control via the fake.
-size_t body_size(const ppsr::stored_schema& s) {
-    return s.schema.def().raw()().size_bytes();
+size_t body_size(const ppsr::subject_schema& s) {
+    return s.def().raw()().size_bytes();
 }
 
 // Rewrites a schema's contexts source->destination for import: the subject's
@@ -439,16 +439,21 @@ ss::future<bool> reconciler::import_body(
     if (fail_if_contains_unsupported(n, read.unsupported)) {
         co_return false;
     }
-    // Deleted-state is authoritative from the caller's active-vs-deleted
-    // listing partition, not the per-version source body: a standard source
-    // (e.g. Confluent) omits the `deleted` flag from that body by default, so
-    // the fetched flag cannot be trusted to propagate a soft-delete.
-    read.schema.deleted = _soft_deleted.contains(n) ? ppsr::is_deleted::yes
-                                                    : ppsr::is_deleted::no;
     data(n).state = node_state::importing;
+    // into_stored() consumes `read`, so lift out the unsupported-feature list
+    // first: it is accounted for only after a successful import below.
+    const auto unsupported = std::move(read.unsupported);
+    // Materialize the stored schema, preferring the source's own reported
+    // deleted flag: it is fresher (read on a later call than the listing) and
+    // avoids the listing set-difference, which can race. Fall back to the
+    // caller's active-vs-deleted listing partition only when the source omitted
+    // the flag, as a source that does not report it cannot otherwise convey a
+    // soft-delete.
+    auto stored = std::move(read).into_stored(
+      _soft_deleted.contains(n) ? ppsr::is_deleted::yes : ppsr::is_deleted::no);
     // The graph key `n` stays in the source namespace; only the schema written
     // to the destination is remapped.
-    auto remapped = remap_for_import(*_mapper, std::move(read.schema));
+    auto remapped = remap_for_import(*_mapper, std::move(stored));
     if (!remapped.has_value()) {
         vlog(
           cllog.warn,
@@ -484,8 +489,8 @@ ss::future<bool> reconciler::import_body(
     ++_stats->versions_changed;
     // Account for REMOVE-stripped features only after the projection actually
     // lands on the destination, so a failed import does not report features as
-    // removed. Only read.schema was moved above; read.unsupported is intact.
-    count_if_contains_unsupported_removed(n, read.unsupported);
+    // removed.
+    count_if_contains_unsupported_removed(n, unsupported);
     wake(n);
     co_return true;
 }

src/v/cluster_link/schema_registry_sync/tests/http_source_reader_test.cc

@@ -173,15 +173,9 @@ TEST(http_source_reader, read_subject_version_returns_schema) {
     EXPECT_THAT(
       res,
       Optional(AllOf(
+        Field("id", &pps::source_schema_read::id, pps::schema_id{100001}),
         Field(
-          "schema",
-          &pps::source_schema_read::schema,
-          AllOf(
-            Field("id", &pps::stored_schema::id, pps::schema_id{100001}),
-            Field(
-              "version",
-              &pps::stored_schema::version,
-              pps::schema_version{3}))),
+          "version", &pps::source_schema_read::version, pps::schema_version{3}),
         Field(
           "unsupported", &pps::source_schema_read::unsupported, IsEmpty()))));
 }

src/v/cluster_link/schema_registry_sync/tests/reconciler_test.cc

@@ -21,6 +21,8 @@
 #include <seastar/core/abort_source.hh>
 #include <seastar/core/when_all.hh>
 
+#include <gmock/gmock.h>
+
 #include <limits>
 #include <string>
 #include <tuple>
@@ -364,20 +366,29 @@ TEST(reconciler, seed_replicated_upsert_is_still_imported) {
     EXPECT_GE(h.source.reads(a, 1), 1);
 }
 
-// A soft-deleted source version imports preserving its deleted state: the
-// caller declares it in work_set.soft_deleted and import_schema stores it
-// soft-deleted. Underpins the task syncing soft-deleted source versions
-// preserving their deleted state.
-TEST(reconciler, imports_soft_deleted_as_deleted) {
+// Soft-delete propagation onto an active destination version via the
+// listing-derived fallback: with a source that omits the `deleted` flag from
+// the body, re-importing the version with the caller's declared deleted-state
+// transitions an existing active version to soft-deleted. The node is in the
+// seed (a destination version is in `all`), so this also guards that a seeded
+// upsert is still imported. Underpins the task propagating a source soft-delete
+// onto a live destination version.
+TEST(reconciler, propagates_soft_delete_over_active_destination) {
     reconcile_harness h;
     auto a = ppsr::context_subject::unqualified("a");
+    // Destination already holds a:v1 active; the source has soft-deleted it but
+    // does not echo the flag, so only the listing-derived set conveys it.
+    h.destination.import_schema(make_schema(a, 1, R"({"v":1})")).get();
+    h.source.reports_deleted_flag = false;
     h.source.add(a, 1, ppsr::is_deleted::yes);
 
+    chunked_hash_set<ppsr::subject_version> seed;
+    seed.insert(key(a, 1));
     srs::work_set work;
     work.upserts.push_back(key(a, 1));
     work.soft_deleted.insert(key(a, 1));
 
-    auto stats = h.run(std::move(work)).get();
+    auto stats = h.run(std::move(work), std::move(seed)).get();
     ASSERT_TRUE(stats.has_value());
     EXPECT_EQ(stats->versions_changed, 1);
     EXPECT_EQ(stats->errors, 0);
@@ -388,26 +399,24 @@ TEST(reconciler, imports_soft_deleted_as_deleted) {
     EXPECT_EQ(all[ia].deleted, ppsr::is_deleted::yes);
 }
 
-// Soft-delete propagation onto an active destination version: re-importing the
-// version with the caller's declared deleted-state transitions an existing
-// active version to soft-deleted. The node is in the seed (a destination
-// version is in `all`), so this also guards that a seeded upsert is still
-// imported. Underpins the task propagating a source soft-delete onto a live
-// destination version.
-TEST(reconciler, propagates_soft_delete_over_active_destination) {
+// When the source omits the `deleted` flag from the per-version body, the
+// caller's listing-derived soft-delete set is the fallback:
+// read_subject_version reports deleted == nullopt, so declaring the node in
+// work_set.soft_deleted still lands it soft-deleted on the destination.
+// Guards the fallback path for a source that does not echo the flag.
+TEST(reconciler, listing_soft_delete_used_when_source_omits_flag) {
     reconcile_harness h;
     auto a = ppsr::context_subject::unqualified("a");
-    // Destination already holds a:v1 active; the source has soft-deleted it.
-    h.destination.import_schema(make_schema(a, 1, R"({"v":1})")).get();
-    h.source.add(a, 1, ppsr::is_deleted::yes);
+    // Source does not echo a deleted flag, so the body cannot convey the
+    // soft-delete; only the listing-derived set can.
+    h.source.reports_deleted_flag = false;
+    h.source.add(a, 1, ppsr::is_deleted::no);
 
-    chunked_hash_set<ppsr::subject_version> seed;
-    seed.insert(key(a, 1));
     srs::work_set work;
     work.upserts.push_back(key(a, 1));
     work.soft_deleted.insert(key(a, 1));
 
-    auto stats = h.run(std::move(work), std::move(seed)).get();
+    auto stats = h.run(std::move(work)).get();
     ASSERT_TRUE(stats.has_value());
     EXPECT_EQ(stats->versions_changed, 1);
     EXPECT_EQ(stats->errors, 0);
@@ -418,32 +427,61 @@ TEST(reconciler, propagates_soft_delete_over_active_destination) {
     EXPECT_EQ(all[ia].deleted, ppsr::is_deleted::yes);
 }
 
-// The caller's declared soft-delete wins over the per-version source body: a
-// standard (e.g. Confluent) source omits the `deleted` flag from that body by
-// default, so read_subject_version reports the version active, yet declaring it
-// in work_set.soft_deleted still lands it soft-deleted on the destination.
-// Guards the cross-vendor soft-delete propagation the HTTP source reader
-// relies on.
-TEST(reconciler, caller_soft_delete_wins_over_active_source_body) {
+// The source's reported deleted flag wins over the listing-derived set when the
+// two disagree: a version the source reports soft-deleted lands deleted even
+// though it is absent from work_set.soft_deleted. The body is authoritative
+// when present.
+TEST(reconciler, reported_deleted_wins_over_listing_absent) {
+    reconcile_harness h;
+    auto a = ppsr::context_subject::unqualified("a");
+    // Source reports the version soft-deleted; the listing set does NOT list
+    // it.
+    h.source.add(a, 1, ppsr::is_deleted::yes);
+
+    srs::work_set work;
+    work.upserts.push_back(key(a, 1));
+    // work.soft_deleted intentionally left empty.
+
+    EXPECT_THAT(
+      h.run(std::move(work)).get(),
+      testing::Optional(
+        testing::AllOf(
+          testing::Field(
+            "versions_changed", &srs::reconcile_stats::versions_changed, 1),
+          testing::Field("errors", &srs::reconcile_stats::errors, 0))));
+
+    const auto& all = h.destination.get_all();
+    auto ia = index_of(all, "a");
+    ASSERT_GE(ia, 0);
+    EXPECT_EQ(all[ia].deleted, ppsr::is_deleted::yes);
+}
+
+// The mirror case, and the point of preferring the body: a fresh body reporting
+// the version active wins over a stale listing set that still lists it as
+// soft-deleted (the listing was read earlier and may have raced). The version
+// lands active despite being in work_set.soft_deleted.
+TEST(reconciler, reported_active_wins_over_stale_listing_soft_delete) {
     reconcile_harness h;
     auto a = ppsr::context_subject::unqualified("a");
-    // Source body reports the version ACTIVE, like a standard SR that does not
-    // echo a deleted flag.
+    // Source reports the version active; the stale listing set still lists it.
     h.source.add(a, 1, ppsr::is_deleted::no);
 
     srs::work_set work;
     work.upserts.push_back(key(a, 1));
     work.soft_deleted.insert(key(a, 1));
 
-    auto stats = h.run(std::move(work)).get();
-    ASSERT_TRUE(stats.has_value());
-    EXPECT_EQ(stats->versions_changed, 1);
-    EXPECT_EQ(stats->errors, 0);
+    EXPECT_THAT(
+      h.run(std::move(work)).get(),
+      testing::Optional(
+        testing::AllOf(
+          testing::Field(
+            "versions_changed", &srs::reconcile_stats::versions_changed, 1),
+          testing::Field("errors", &srs::reconcile_stats::errors, 0))));
 
     const auto& all = h.destination.get_all();
     auto ia = index_of(all, "a");
     ASSERT_GE(ia, 0);
-    EXPECT_EQ(all[ia].deleted, ppsr::is_deleted::yes);
+    EXPECT_EQ(all[ia].deleted, ppsr::is_deleted::no);
 }
 
 TEST(reconciler, cyclic_source_does_not_hang) {

src/v/cluster_link/schema_registry_sync/tests/sr_sync_test_fixtures.h

@@ -107,6 +107,12 @@ key(const ppsr::context_subject& sub, int32_t version) {
 struct fake_source_state {
     chunked_vector<ppsr::context> contexts{ppsr::default_context};
     chunked_vector<ppsr::stored_schema> schemas;
+    // Whether the source echoes the `deleted` flag in per-version bodies. True
+    // (Redpanda, and Confluent queried with the extended-fields header) makes
+    // read_subject_version report source_schema_read::deleted; false models a
+    // source that omits it entirely, so deleted is nullopt and the caller must
+    // fall back to the listing-derived soft-delete set.
+    bool reports_deleted_flag = true;
     std::optional<srs::source_error> list_contexts_error;
     std::optional<srs::source_error> list_subjects_error;
     // Forces list_subject_versions to fail for specific subjects, letting a
@@ -280,8 +286,18 @@ class fake_source_reader final : public srs::source_reader {
                   it != _state->unsupported_features.end()) {
                     unsupported = it->second.copy();
                 }
+                // A source that reports the flag echoes its state; one that
+                // omits it yields nullopt, and the reconciler's fallback (via
+                // into_stored) supplies the deleted state instead.
+                auto shared = s.share();
                 co_return ppsr::source_schema_read{
-                  .schema = s.share(), .unsupported = std::move(unsupported)};
+                  .schema = std::move(shared.schema),
+                  .version = shared.version,
+                  .id = shared.id,
+                  .deleted = _state->reports_deleted_flag
+                               ? std::optional<ppsr::is_deleted>{s.deleted}
+                               : std::nullopt,
+                  .unsupported = std::move(unsupported)};
             }
         }
         co_return std::unexpected(

src/v/pandaproxy/schema_registry/rest_client/parse.cc

@@ -682,7 +682,7 @@ parse_subject_version(iobuf body, qualified_subjects_enabled qualified) {
         std::optional<iobuf> schema;
         schema_type type{schema_type::avro};
         schema_definition::references refs;
-        is_deleted deleted{false};
+        std::optional<is_deleted> deleted;
         std::optional<schema_metadata> metadata;
         chunked_vector<unsupported_feature> unsupported;
 
@@ -698,20 +698,22 @@ parse_subject_version(iobuf body, qualified_subjects_enabled qualified) {
                 }
                 // Absent fields fall back to defaults/sentinels; completeness
                 // is a higher-layer concern. Unmodeled fields were recorded in
-                // `unsupported` above for the caller to act on.
+                // `unsupported` above for the caller to act on. `deleted` stays
+                // nullopt when the body omitted the flag, so a caller can tell
+                // an absent flag from an explicit false; into_stored() resolves
+                // it to a concrete value with the caller's fallback.
                 co_return source_schema_read{
-                  .schema = stored_schema{
-                    .schema = subject_schema{
-                      subject.value_or(invalid_subject),
-                      schema_definition{
-                        schema_definition::raw_string{
-                          std::move(schema).value_or(iobuf{})},
-                        type,
-                        std::move(refs),
-                        std::move(metadata)}},
-                    .version = version.value_or(invalid_schema_version),
-                    .id = id.value_or(invalid_schema_id),
-                    .deleted = deleted},
+                  .schema = subject_schema{
+                    subject.value_or(invalid_subject),
+                    schema_definition{
+                      schema_definition::raw_string{
+                        std::move(schema).value_or(iobuf{})},
+                      type,
+                      std::move(refs),
+                      std::move(metadata)}},
+                  .version = version.value_or(invalid_schema_version),
+                  .id = id.value_or(invalid_schema_id),
+                  .deleted = deleted,
                   .unsupported = std::move(unsupported)};
             }
             if (p.token() != token::key) {