CORE-16766 Schema Migration: integrate client and test in ducktape end to end

proto/redpanda/core/admin/v2/shadow_link.proto

@@ -721,11 +721,11 @@ message SchemaRegistrySyncStatus {
 
     // Sync currently running. This is unset when no Schema Registry sync is
     // running.
-    SchemaRegistryCurrentSync current_sync = 2;
+    optional SchemaRegistryCurrentSync current_sync = 2;
 
     // Counters from the most recently completed full sync. This is unset until
     // a full sync completes on the current task instance.
-    SchemaRegistrySyncSummary last_full_sync = 3;
+    optional SchemaRegistrySyncSummary last_full_sync = 3;
 
     // Cumulative counters since the Schema Registry task started. These
     // counters reset after task restart or leadership transfer.
@@ -765,11 +765,11 @@ message SchemaRegistryInventory {
 message SchemaRegistrySyncSummary {
     // Time when the sync started. For cumulative counters, this is the time
     // when the Schema Registry task started.
-    google.protobuf.Timestamp start_time = 1;
+    optional google.protobuf.Timestamp start_time = 1;
 
     // Time when the sync finished. This is unset for a sync that is still
     // running and for cumulative counters.
-    google.protobuf.Timestamp finish_time = 2;
+    optional google.protobuf.Timestamp finish_time = 2;
 
     // Number of selected subject versions created, updated, or deleted on the
     // destination.

src/v/cluster_link/BUILD

@@ -51,6 +51,7 @@ redpanda_cc_library(
         "//src/v/kafka/client:cluster",
         "//src/v/kafka/client:configuration",
         "//src/v/kafka/protocol",
+        "//src/v/net:tls",
         "//src/v/ssx:sformat",
         "@seastar",
     ],
@@ -259,6 +260,7 @@ redpanda_cc_library(
         "//src/v/cluster:plugin_backend",
         "//src/v/cluster:topic_metrics_watcher",
         "//src/v/cluster_link/model",
+        "//src/v/cluster_link/schema_registry_sync:http_source_reader",
         "//src/v/cluster_link/schema_registry_sync:mirroring_task",
         "//src/v/cluster_link/schema_registry_sync:source_reader",
         "//src/v/config",

src/v/cluster_link/schema_registry_sync/BUILD

@@ -12,12 +12,46 @@ redpanda_cc_library(
     visibility = ["//visibility:public"],
     deps = [
         "//src/v/base",
+        "//src/v/cluster_link/model",
         "//src/v/container:chunked_vector",
         "//src/v/pandaproxy/schema_registry:types",
         "@seastar",
     ],
 )
 
+redpanda_cc_library(
+    name = "http_source_reader",
+    srcs = [
+        "http_source_reader.cc",
+    ],
+    hdrs = [
+        "http_source_reader.h",
+    ],
+    implementation_deps = [
+        "//src/v/base",
+        "//src/v/cluster_link:utils",
+        "//src/v/config",
+        "//src/v/http",
+        "//src/v/net",
+        "//src/v/pandaproxy/schema_registry/rest_client:error",
+        "//src/v/utils:retry_chain_node",
+        "@ada",
+    ],
+    visibility = ["//visibility:public"],
+    deps = [
+        ":source_reader",
+        "//src/v/cluster_link/model",
+        "//src/v/container:chunked_vector",
+        "//src/v/net:tls",
+        "//src/v/pandaproxy/schema_registry:types",
+        "//src/v/pandaproxy/schema_registry/rest_client:client",
+        "//src/v/pandaproxy/schema_registry/rest_client:credentials",
+        "//src/v/ssx:semaphore",
+        "//src/v/utils:unresolved_address",
+        "@seastar",
+    ],
+)
+
 redpanda_cc_library(
     name = "mirroring_task",
     srcs = [

src/v/cluster_link/schema_registry_sync/http_source_reader.cc

@@ -0,0 +1,295 @@
+/*
+ * Copyright 2026 Redpanda Data, Inc.
+ *
+ * Use of this software is governed by the Business Source License
+ * included in the file licenses/BSL.md
+ *
+ * As of the Change Date specified in that file, in accordance with
+ * the Business Source License, use of this software will be governed
+ * by the Apache License, Version 2.0
+ */
+
+#include "cluster_link/schema_registry_sync/http_source_reader.h"
+
+#include "cluster_link/schema_registry_sync/unavailable_source_reader.h"
+#include "cluster_link/utils.h"
+#include "config/configuration.h"
+#include "config/tls_config.h"
+#include "http/client.h"
+#include "net/tls.h"
+#include "net/tls_certificate_probe.h"
+#include "net/transport.h"
+#include "pandaproxy/schema_registry/rest_client/error.h"
+#include "utils/retry_chain_node.h"
+
+#include <seastar/core/coroutine.hh>
+#include <seastar/core/semaphore.hh>
+
+#include <ada.h>
+#include <charconv>
+#include <chrono>
+#include <system_error>
+#include <variant>
+
+using namespace std::chrono_literals;
+
+namespace cluster_link::schema_registry_sync {
+
+namespace {
+
+// Per-call retry budget for source requests. The rest_client retries
+// network/timeout/5xx within this window; an exhausted budget surfaces as
+// retries_exhausted, which the reader maps to source_unavailable.
+constexpr auto request_timeout = 30s;
+constexpr auto request_backoff = 100ms;
+
+// Map a rest_client failure onto a source_error. A source that is unreachable
+// or rejecting every request is source_unavailable and parks the link; a 404 is
+// subject_not_found; any other terminal status is a per-item operation_failed.
+source_error to_source_error(rc::domain_error err) {
+    using enum boost::beast::http::status;
+    auto kind = ss::visit(
+      err,
+      [](const rc::http_call_error& call) {
+          return ss::visit(
+            call,
+            [](const rc::http_status_error& s) {
+                // Auth failures are link-wide and deterministic, so back off
+                // rather than re-fail every subject in turn.
+                return s.status == unauthorized || s.status == forbidden
+                         ? source_error_kind::source_unavailable
+                         : source_error_kind::operation_failed;
+            },
+            // A permanent exception with no HTTP response: source unreachable.
+            [](const ss::sstring&) {
+                return source_error_kind::source_unavailable;
+            });
+      },
+      [](const rc::retries_exhausted&) {
+          return source_error_kind::source_unavailable;
+      },
+      [](const rc::aborted_error&) {
+          return source_error_kind::source_unavailable;
+      },
+      [](const rc::subject_not_found&) {
+          return source_error_kind::subject_not_found;
+      },
+      [](const auto&) { return source_error_kind::operation_failed; });
+    return source_error{.kind = kind, .message = fmt::format("{}", err)};
+}
+
+} // namespace
+
+std::optional<net::unresolved_address>
+parse_source_address(std::string_view url) {
+    auto parsed = ada::parse<ada::url_aggregator>(url);
+    if (!parsed || parsed->get_hostname().empty()) {
+        return std::nullopt;
+    }
+    // Only host:port is carried into the transport; a path prefix, query, or
+    // fragment would be silently dropped (requests always hit the SR API at the
+    // root). Reject rather than connect to the wrong place. ada normalizes an
+    // empty path to "/" for special schemes, so "/" means "no prefix".
+    if (auto path = parsed->get_pathname(); path != "/" && !path.empty()) {
+        return std::nullopt;
+    }
+    if (!parsed->get_search().empty() || !parsed->get_hash().empty()) {
+        return std::nullopt;
+    }
+    // ada normalizes away a port equal to the scheme default (per the WHATWG
+    // URL spec), so get_port() is empty for e.g. https://host:443. Fall back to
+    // the scheme's default port (443 for https, 80 for http) rather than a
+    // fixed one, so a source behind a standard port is reachable; an explicit
+    // non-default port (e.g. :8081) is still honored.
+    uint16_t port = parsed->scheme_default_port();
+    if (auto port_sv = parsed->get_port(); !port_sv.empty()) {
+        auto [ptr, ec] = std::from_chars(
+          port_sv.data(), port_sv.data() + port_sv.size(), port);
+        if (ec != std::errc{} || ptr != port_sv.data() + port_sv.size()) {
+            return std::nullopt;
+        }
+    }
+    // A non-special scheme has no default port; reject rather than connect to
+    // 0.
+    if (port == 0) {
+        return std::nullopt;
+    }
+    return net::unresolved_address{ss::sstring{parsed->get_hostname()}, port};
+}
+
+http_source_reader::http_source_reader(http_source_connection conn)
+  : _conn(std::move(conn)) {}
+
+http_source_reader::http_source_reader(std::unique_ptr<rc::client> client)
+  : _client(std::move(client)) {}
+
+ss::future<source_result<rc::client*>>
+http_source_reader::ensure_client(ss::abort_source&) {
+    if (_client) {
+        co_return _client.get();
+    }
+    try {
+        net::base_transport::configuration cfg{.server_addr = _conn->address};
+        if (_conn->tls_enabled) {
+            auto builder = co_await net::get_credentials_builder({
+              .truststore = _conn->truststore,
+              .k_store = _conn->client_key,
+              .min_tls_version = config::from_config(
+                config::shard_local_cfg().tls_min_version()),
+              .enable_renegotiation
+              = config::shard_local_cfg().tls_enable_renegotiation(),
+              .require_client_auth = false,
+            });
+            cfg.credentials
+              = co_await net::build_reloadable_credentials_with_probe<
+                ss::tls::certificate_credentials>(
+                std::move(builder), "cluster_link/sr_source", "source");
+            if (_conn->provide_sni) {
+                cfg.tls_sni_hostname = _conn->address.host();
+            }
+        }
+        _client = std::make_unique<rc::client>(
+          std::make_unique<http::client>(cfg),
+          _conn->endpoint,
+          _conn->auth,
+          ppsr::qualified_subjects_enabled::yes);
+        co_return _client.get();
+    } catch (...) {
+        co_return std::unexpected(
+          source_error{
+            .kind = source_error_kind::source_unavailable,
+            .message = fmt::format(
+              "failed to build source Schema Registry client: {}",
+              std::current_exception())});
+    }
+}
+
+ss::future<source_result<chunked_vector<ppsr::context>>>
+http_source_reader::list_contexts(ss::abort_source& as) {
+    auto units = co_await ss::get_units(_inflight, 1, as);
+    auto client = co_await ensure_client(as);
+    if (!client.has_value()) {
+        co_return std::unexpected(std::move(client.error()));
+    }
+    retry_chain_node rtc(as, request_timeout, request_backoff);
+    // No prefix filter: enumerate every context so subjects in non-default
+    // contexts are discovered. The default context is always present (".").
+    auto res = co_await client.value()->list_contexts(rtc);
+    if (!res.has_value()) {
+        co_return std::unexpected(to_source_error(std::move(res.error())));
+    }
+    co_return std::move(res.value());
+}
+
+ss::future<source_result<chunked_vector<ppsr::context_subject>>>
+http_source_reader::list_subjects(ppsr::context ctx, ss::abort_source& as) {
+    auto units = co_await ss::get_units(_inflight, 1, as);
+    auto client = co_await ensure_client(as);
+    if (!client.has_value()) {
+        co_return std::unexpected(std::move(client.error()));
+    }
+    retry_chain_node rtc(as, request_timeout, request_backoff);
+    // Scope discovery to the requested context: the rest_client sends a
+    // subjectPrefix hint and filters the response to that context. Include
+    // soft-deleted subjects so a subject whose every version is deleted is
+    // still discovered; the reconcile classifies per-version deleted state (via
+    // two list_subject_versions calls) and propagates the soft-deletes.
+    auto res = co_await client.value()->list_subjects(
+      rtc, ppsr::include_deleted::yes, ctx);
+    if (!res.has_value()) {
+        co_return std::unexpected(to_source_error(std::move(res.error())));
+    }
+    co_return std::move(res.value());
+}
+
+ss::future<source_result<chunked_vector<ppsr::schema_version>>>
+http_source_reader::list_subject_versions(
+  ppsr::context_subject sub,
+  ppsr::include_deleted include_deleted,
+  ss::abort_source& as) {
+    auto units = co_await ss::get_units(_inflight, 1, as);
+    auto client = co_await ensure_client(as);
+    if (!client.has_value()) {
+        co_return std::unexpected(std::move(client.error()));
+    }
+    retry_chain_node rtc(as, request_timeout, request_backoff);
+    auto res = co_await client.value()->list_subject_versions(
+      sub, rtc, include_deleted);
+    if (!res.has_value()) {
+        co_return std::unexpected(to_source_error(std::move(res.error())));
+    }
+    co_return std::move(res.value());
+}
+
+ss::future<source_result<ppsr::stored_schema>>
+http_source_reader::read_subject_version(
+  ppsr::context_subject sub,
+  ppsr::schema_version version,
+  ss::abort_source& as) {
+    auto units = co_await ss::get_units(_inflight, 1, as);
+    auto client = co_await ensure_client(as);
+    if (!client.has_value()) {
+        co_return std::unexpected(std::move(client.error()));
+    }
+    retry_chain_node rtc(as, request_timeout, request_backoff);
+    // Include deleted: the reconcile fetches soft-deleted source versions to
+    // propagate the soft-delete, so a deleted version must still resolve here
+    // (an active version is returned just the same).
+    auto res = co_await client.value()->get_schema_by_version(
+      sub, version, rtc, ppsr::include_deleted::yes);
+    if (!res.has_value()) {
+        co_return std::unexpected(to_source_error(std::move(res.error())));
+    }
+    // Unmodeled response fields (parsed_schema::unknown_fields) are ignored;
+    // honoring the unsupported-feature policy is future work.
+    co_return std::move(res.value().schema);
+}
+
+ss::future<> http_source_reader::stop() {
+    // Idempotent: the reader can be stopped more than once (e.g. an in-flight
+    // reconciler stopping the task before link teardown stops it again). The
+    // rest_client's gate must be closed exactly once, so release the client
+    // after shutting it down and skip it on a repeat call.
+    if (auto client = std::move(_client); client) {
+        co_await client->shutdown();
+    }
+}
+
+std::unique_ptr<source_reader> http_source_reader_factory::create(
+  const model::schema_registry_sync_config::shadow_schema_registry_api*
+    api_cfg) {
+    if (api_cfg == nullptr) {
+        return std::make_unique<unavailable_source_reader>();
+    }
+    auto address = parse_source_address(api_cfg->source_url);
+    if (!address.has_value()) {
+        return std::make_unique<unavailable_source_reader>(fmt::format(
+          "invalid source Schema Registry URL: '{}'", api_cfg->source_url));
+    }
+
+    http_source_connection conn{
+      .address = std::move(*address),
+      .endpoint = api_cfg->source_url,
+      .tls_enabled = bool(api_cfg->tls_enabled),
+      .provide_sni = bool(api_cfg->tls_provide_sni)};
+
+    if (api_cfg->ca.has_value()) {
+        conn.truststore = to_certificate(api_cfg->ca.value());
+    }
+    if (api_cfg->cert.has_value() && api_cfg->key.has_value()) {
+        conn.client_key = to_key_store(
+          api_cfg->key.value(), api_cfg->cert.value());
+    }
+    if (api_cfg->auth_config.has_value()) {
+        conn.auth = ss::visit(
+          api_cfg->auth_config.value(),
+          [](const model::schema_registry_sync_config::basic_auth& basic) {
+              return rc::basic_auth_credentials{
+                .username = basic.username, .password = basic.password};
+          });
+    }
+
+    return std::make_unique<http_source_reader>(std::move(conn));
+}
+
+} // namespace cluster_link::schema_registry_sync