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

[mnajda-redpanda]

Schema Registry API-mode sync propagates a source version's soft-delete state
to the destination. Until now the imported deleted-state was taken solely from
the listing set-difference (all − active): the per-version body could not be
trusted because a Confluent source omitted the deleted flag from it by
default.

Now that the SR REST client opts into Confluent's extended schema fields, the
per-version body carries deleted reliably for both Redpanda and Confluent
sources. This PR prefers that source-reported flag over the listing-derived
set: it is fresher (read on a later call than the listing) and avoids the
listing set-difference, which can race with a concurrent (un)delete on the
source. The listing-derived set is kept as a fallback for a source that still
omits the flag from the body.

Backports Required

• none - not a bug fix
• none - this is a backport
• none - issue does not exist in previous branches
• none - papercut/not impactful enough to backport
• v26.1.x
• v25.3.x
• v25.2.x

Release Notes

• none

[Copilot]

Pull request overview

This PR updates schema-registry sync behavior to distinguish an explicitly reported deleted flag from an omitted one, enabling the reconciler to prefer the per-version body’s deleted state when present and fall back to the listing-derived soft-delete set only when the source omits the flag.

Changes:

• Add source_schema_read::reported_deleted (std::optional<is_deleted>) to capture whether deleted was present in the source response body.
• Update JSON parsing to preserve “absent vs explicit false” and keep schema.deleted defaulting to no when absent.
• Adjust schema-registry sync reconciler logic and tests to prefer reported_deleted over the listing-derived soft-delete set, with explicit fallback behavior.

Reviewed changes

Copilot reviewed 8 out of 8 changed files in this pull request and generated 1 comment.

Show a summary per file

File	Description
src/v/pandaproxy/schema_registry/types.h	Adds reported_deleted to represent presence/absence of the deleted flag.
src/v/pandaproxy/schema_registry/rest_client/parse.cc	Parses deleted into an optional and wires it into source_schema_read.
src/v/pandaproxy/schema_registry/rest_client/tests/parse_test.cc	Adds/updates tests validating reported_deleted semantics for present/absent cases.
src/v/cluster_link/schema_registry_sync/reconciler.cc	Prefers reported_deleted when present; otherwise falls back to listing-derived soft-deletes.
src/v/cluster_link/schema_registry_sync/mirroring_task.cc	Updates comments to reflect the new “fallback” role of the listing-derived set.
src/v/cluster_link/schema_registry_sync/http_source_reader.cc	Updates comments clarifying that reported_deleted is carried through to the reconciler.
src/v/cluster_link/schema_registry_sync/tests/sr_sync_test_fixtures.h	Extends the fake source to model sources that omit the deleted flag.
src/v/cluster_link/schema_registry_sync/tests/reconciler_test.cc	Adds/adjusts tests for precedence rules between body-reported vs listing-derived deleted state.

[pgellert]

What do you think about an alternative type definition that looks something like this, which would avoid the potential inconsistency between schema.deleted and reported_deleted altogether?

struct source_schema_read {
    subject_schema schema;
    schema_version version{invalid_schema_version};
    schema_id id{invalid_schema_id};
    /// As reported by the source; nullopt when the body omitted `deleted`.
    std::optional<is_deleted> deleted;
    chunked_vector<unsupported_feature> unsupported;

    stored_schema into_stored(is_deleted fallback = is_deleted::no) && {
        return {std::move(schema), version, id, deleted.value_or(fallback)};
    }
};

[mnajda-redpanda]

That's a much better way to do it, implemented :)

[vbotbuildovich]

Retry command for Build#87003

please wait until all jobs are finished before running the slash command

/ci-repeat 1
skip-redpanda-build
skip-units
skip-rebase
tests/rptest/tests/cloud_topics/e2e_test.py::EndToEndCloudTopicsStorageModeToggleTest.test_toggle_storage_mode

[mnajda-redpanda]

force-push 1: implement @pgellert suggestion

• source_schema_read: flatten schema/version/id to top level; drop the
  reported_deleted sidecar + eager schema.deleted write in favor of a
  single deleted optional resolved by a new into_stored(fallback).
• reconciler import_body: materialize via std::move(read).into_stored(...)
  instead of mutating read.schema.deleted; resolve_refs/body_size now take
  subject_schema.

force-push 2: clean up a comment

force-push 3: formatting

force-push 4: align missed tests

force-push 5: clang-tidy fix

[vbotbuildovich]

CI test results

test results on build#87013

test_status	test_class	test_method	test_arguments	test_kind	job_url	passed	reason	test_history
FLAKY(PASS)	ShutdownTest	test_timely_shutdown_with_failures	null	integration	https://buildkite.com/redpanda/redpanda/builds/87013#019f5c48-00cf-4c58-b293-63b29e6227b7	10/11	Test PASSES after retries.No significant increase in flaky rate(baseline=0.0159, p0=1.0000, reject_threshold=0.0100. adj_baseline=0.1000, p1=0.3487, trust_threshold=0.5000)	https://redpanda.metabaseapp.com/dashboard/87-tests?tab=142-dt-individual-test-history&test_class=ShutdownTest&test_method=test_timely_shutdown_with_failures