encryption: Ensure stopping timers in provider cache objects

utils::loading cache has a timer that can, if we're unlucky, be runnnig
while the encryption context/extensions referencing the various host
objects containing them are destroyed in the case of unit testing.

Add a stop phase in encryption context shutdown closing the caches.

(cherry picked from commit ee98f5d361)

ent/encryption/encryption.cc

@@ -475,6 +475,14 @@ public:
             for (auto&& [id, h] : _per_thread_kmip_host_cache[this_shard_id()]) {
                 co_await h->disconnect();
             }
+            static auto stop_all = [](auto&& cache) -> future<> {
+                for (auto& [k, host] : cache) {
+                    co_await host->stop();
+                }
+            };
+            co_await stop_all(_per_thread_kms_host_cache[this_shard_id()]);
+            co_await stop_all(_per_thread_gcp_host_cache[this_shard_id()]);
+
             _per_thread_provider_cache[this_shard_id()].clear();
             _per_thread_system_key_cache[this_shard_id()].clear();
             _per_thread_kmip_host_cache[this_shard_id()].clear();

ent/encryption/gcp_host.cc

@@ -101,6 +101,7 @@ public:
     ~impl() = default;
 
     future<> init();
+    future<> stop();
     const host_options& options() const {
         return _options;
     }
@@ -840,6 +841,11 @@ future<> encryption::gcp_host::impl::init() {
     _initialized = true;
 }
 
+future<> encryption::gcp_host::impl::stop() {
+    co_await _attr_cache.stop();
+    co_await _id_cache.stop();
+}
+
 std::tuple<std::string, std::string> encryption::gcp_host::impl::parse_key(std::string_view spec) {
     auto i = spec.find_last_of('/');
     if (i == std::string_view::npos) {
@@ -1002,6 +1008,10 @@ future<> encryption::gcp_host::init() {
     return _impl->init();
 }
 
+future<> encryption::gcp_host::stop() {
+    return _impl->stop();
+}
+
 const encryption::gcp_host::host_options& encryption::gcp_host::options() const {
     return _impl->options();
 }

ent/encryption/gcp_host.hh

@@ -65,6 +65,8 @@ public:
     ~gcp_host();
 
     future<> init();
+    future<> stop();
+
     const host_options& options() const;
 
     struct option_override : public t_credentials_source<std::optional<std::string>> {

ent/encryption/kmip_host.cc

@@ -724,9 +724,11 @@ future<> kmip_host::impl::connect() {
 }
 
 future<> kmip_host::impl::disconnect() {
-    return do_for_each(_options.hosts, [this](const sstring& host) {
+    co_await do_for_each(_options.hosts, [this](const sstring& host) {
         return clear_connections(host);
     });
+    co_await _attr_cache.stop();
+    co_await _id_cache.stop();
 }
 
 static unsigned from_str(unsigned (*f)(char*, int, int*), const sstring& s, const sstring& what) {

ent/encryption/kms_host.cc

@@ -160,6 +160,8 @@ public:
     ~impl() = default;
 
     future<> init();
+    future<> stop();
+
     const host_options& options() const {
         return _options;
     }
@@ -988,6 +990,11 @@ future<> encryption::kms_host::impl::init() {
     _initialized = true;
 }
 
+future<> encryption::kms_host::impl::stop() {
+    co_await _attr_cache.stop();
+    co_await _id_cache.stop();
+}
+
 future<encryption::kms_host::impl::key_and_id_type> encryption::kms_host::impl::create_key(const attr_cache_key& k) {
     auto& master_key = k.master_key;
     auto& aws_assume_role_arn = k.aws_assume_role_arn;
@@ -1150,6 +1157,10 @@ future<> encryption::kms_host::init() {
     return _impl->init();
 }
 
+future<> encryption::kms_host::stop() {
+    return _impl->stop();
+}
+
 const encryption::kms_host::host_options& encryption::kms_host::options() const {
     return _impl->options();
 }

ent/encryption/kms_host.hh

@@ -63,6 +63,8 @@ public:
     ~kms_host();
 
     future<> init();
+    future<> stop();
+
     const host_options& options() const;
 
     struct option_override {