diff --git a/mooncake-store/include/ha/leadership/master_service_supervisor.h b/mooncake-store/include/ha/leadership/master_service_supervisor.h index dfdac3d67a..99378c017d 100644 --- a/mooncake-store/include/ha/leadership/master_service_supervisor.h +++ b/mooncake-store/include/ha/leadership/master_service_supervisor.h @@ -1,10 +1,13 @@ #pragma once #include +#include +#include #include #include #include "master_config.h" +#include "shutdown_request.h" namespace mooncake { namespace ha { @@ -46,15 +49,25 @@ class ServingStateGate { } // namespace detail +// Bound on an HA shutdown, from the first observation of the request to +// process exit. main() arms the shutdown watchdog with it. +inline constexpr std::chrono::seconds kHaShutdownDeadline{15}; + class MasterServiceSupervisor { public: + // shutdown_requested defaults to the process signal flag; tests inject + // their own predicate. The predicate can run concurrently on the + // supervisor and trigger threads, so it must be thread-safe and must not + // throw. explicit MasterServiceSupervisor( - const MasterServiceSupervisorConfig& config); + const MasterServiceSupervisorConfig& config, + std::function shutdown_requested = ShutdownRequested); int Start(); private: MasterServiceSupervisorConfig config_; + std::function shutdown_requested_; }; } // namespace ha diff --git a/mooncake-store/include/shutdown_request.h b/mooncake-store/include/shutdown_request.h new file mode 100644 index 0000000000..602d56eef6 --- /dev/null +++ b/mooncake-store/include/shutdown_request.h @@ -0,0 +1,53 @@ +#pragma once + +#include + +#include + +namespace mooncake { + +// Process-wide SIGINT/SIGTERM intake for a process that drains before it +// exits. The first signal sets a sticky flag. A second signal ends the +// process at once with _exit(128 + signo). + +// Installs the SIGINT and SIGTERM handlers with sigaction and SA_ONSTACK, +// which the Go c-shared HA backends in the master process require. Returns +// false with errno set if a sigaction call fails. A handler installed before +// the failure stays installed. +bool InstallShutdownSignalHandlers(); + +// True once SIGINT or SIGTERM was received. Safe from any thread. +bool ShutdownRequested(); + +// Records a shutdown request that code other than the signal handler +// observed, such as an injected predicate. Sticky and idempotent. Call it +// before any shutdown work: it starts the watchdog deadline. +void LatchShutdownRequest(); + +// True once LatchShutdownRequest or ExitAfterShutdown ran. +bool ShutdownLatched(); + +// Starts a detached watchdog thread. Once a signal was received or a request +// was latched, the watchdog ends the process with _exit(EXIT_FAILURE) if it +// is still alive `deadline` later. The deadline starts once and is never +// reset. The thread takes no locks and does no I/O. Call it once, serially, +// before the work it bounds; a repeated serial call returns true without a +// second thread. Concurrent calls are not supported. Returns false if the +// thread cannot be created. +bool StartShutdownWatchdog(std::chrono::milliseconds deadline); + +// Ends the process after a completed shutdown. Latches the request, flushes +// glog only when the watchdog thread exists, so a blocked flush still ends +// at the deadline, then calls _exit(status); no destructors run. +[[noreturn]] void ExitAfterShutdown(int status); + +namespace detail { + +using SigactionFn = int (*)(int, const struct sigaction*, struct sigaction*); + +// Test seam: InstallShutdownSignalHandlers() calls this with ::sigaction. +bool InstallShutdownSignalHandlersWith(SigactionFn install); + +} // namespace detail + +} // namespace mooncake diff --git a/mooncake-store/src/CMakeLists.txt b/mooncake-store/src/CMakeLists.txt index 1706e72728..91fa180b66 100644 --- a/mooncake-store/src/CMakeLists.txt +++ b/mooncake-store/src/CMakeLists.txt @@ -65,6 +65,7 @@ set(MOONCAKE_STORE_MASTER_SOURCES ha/common/redis/redis_connection.cpp ha/leadership/backends/redis/redis_leader_coordinator.cpp ha/leadership/master_service_supervisor.cpp + shutdown_request.cpp ha/kv/etcd_ha_kv_backend.cpp ha/kv/redis_ha_kv_backend.cpp ha/kv/ha_kv_backend_factory.cpp diff --git a/mooncake-store/src/ha/leadership/master_service_supervisor.cpp b/mooncake-store/src/ha/leadership/master_service_supervisor.cpp index 7a4e1a8c0c..2e4f67e7c7 100644 --- a/mooncake-store/src/ha/leadership/master_service_supervisor.cpp +++ b/mooncake-store/src/ha/leadership/master_service_supervisor.cpp @@ -4,6 +4,7 @@ #include #include #include +#include #include #include #include @@ -18,11 +19,13 @@ #include "ha/leadership/leader_label_reconciler.h" #include "ha/kv/ha_kv_backend_factory.h" #include "ha/oplog/oplog_batch_storage.h" +#include "ha/oplog/oplog_test_failpoint.h" #include "ha/standby_controller.h" #include "k8s_lease_helper.h" #include "master_admin_service.h" #include "rpc_service.h" #include "rpc_worker_pool.h" +#include "shutdown_request.h" namespace mooncake { namespace ha { @@ -33,6 +36,7 @@ constexpr auto kAcquireRetryInterval = std::chrono::seconds(1); constexpr auto kRenewCheckInterval = std::chrono::seconds(1); constexpr auto kSupervisorRetryInterval = std::chrono::seconds(1); constexpr auto kLabelReconcileRetryInterval = std::chrono::seconds(1); +constexpr auto kShutdownPollInterval = std::chrono::milliseconds(100); constexpr char kLeaderLabelKey[] = "mooncake.io/store-role"; constexpr char kLeaderLabelValue[] = "leader"; @@ -154,6 +158,104 @@ class LeadershipRenewalGuard { std::atomic lost_{false}; }; +// Every shutdown check goes through here. The first true observation +// latches the request, which starts the watchdog deadline before any +// shutdown work, and keeps later checks true even if an injected predicate +// turns false. +bool ShutdownObserved(const std::function& shutdown_requested) { + if (ShutdownLatched()) { + return true; + } + if (!shutdown_requested()) { + return false; + } + LatchShutdownRequest(); + return true; +} + +// What the supervisor holds when a shutdown ends the process. +struct ShutdownLease { + // Set while this process holds the lease; it is left to expire by TTL. + const LeadershipSession* session = nullptr; + // Result of an ordinary release that already ran for the session. + std::optional prior_release{}; +}; + +// Last step before the terminal exit. Returns the exit status. +int FinishShutdown(const ShutdownLease& lease) { + if (lease.prior_release.has_value()) { + LOG_IF(INFO, *lease.prior_release == ErrorCode::OK) + << "Shutdown complete after leadership release, result=OK"; + LOG_IF(WARNING, *lease.prior_release != ErrorCode::OK) + << "Shutdown complete after leadership release, result=" + << toString(*lease.prior_release); + return 0; + } + if (lease.session != nullptr) { + LOG(INFO) << "Shutdown complete, no release attempted; the lease is " + "left to TTL expiry, view_version=" + << lease.session->view.view_version; + return 0; + } + LOG(INFO) << "Shutdown complete, no lease held"; + return 0; +} + +// Polls the shutdown request while the listener may run and fences the +// serving session through the gate, like a leadership loss. +class ServingShutdownTrigger { + public: + ServingShutdownTrigger(const std::function& shutdown_requested, + std::function on_shutdown) + : shutdown_requested_(shutdown_requested), + on_shutdown_(std::move(on_shutdown)) { + worker_ = std::thread([this] { Run(); }); + } + + ~ServingShutdownTrigger() { Stop(); } + + ServingShutdownTrigger(const ServingShutdownTrigger&) = delete; + ServingShutdownTrigger& operator=(const ServingShutdownTrigger&) = delete; + + void Stop() { + { + std::lock_guard lock(mutex_); + stop_.store(true, std::memory_order_release); + } + cv_.notify_all(); + if (worker_.joinable()) { + worker_.join(); + } + } + + private: + void Run() { + std::unique_lock lock(mutex_); + while (!stop_.load(std::memory_order_acquire)) { + if (ShutdownObserved(shutdown_requested_)) { + // on_shutdown_ runs with the mutex unlocked on purpose: + // Stop() must be able to signal and join while the callback + // is inside server.stop(). If that callback blocks, the + // join in Stop() waits, and the watchdog is what bounds the + // whole process, not this interlock. + lock.unlock(); + on_shutdown_(); + return; + } + cv_.wait_for(lock, kShutdownPollInterval, [this] { + return stop_.load(std::memory_order_acquire); + }); + } + } + + const std::function& shutdown_requested_; + std::function on_shutdown_; + std::thread worker_; + std::mutex mutex_; + std::condition_variable cv_; + std::atomic stop_{false}; +}; + void LogLeadershipReleaseWarning(std::string_view context, ErrorCode err) { if (err == ErrorCode::OK) { return; @@ -178,23 +280,25 @@ bool HandleSupervisorError(std::string_view action, ErrorCode err, } bool HandleLeadershipPhaseError(std::string_view release_context, - std::string_view action, - LeaderCoordinator& coordinator, - const LeadershipSession& session, ErrorCode err, - HABackendType backend_type) { - LogLeadershipReleaseWarning(release_context, - coordinator.ReleaseLeadership(session)); + std::string_view action, ErrorCode release_err, + ErrorCode err, HABackendType backend_type) { + LogLeadershipReleaseWarning(release_context, release_err); return HandleSupervisorError(action, err, backend_type); } tl::expected WarmupLeadership( - LeaderCoordinator& coordinator, const LeadershipSession& session) { + LeaderCoordinator& coordinator, const LeadershipSession& session, + const std::function& shutdown_requested) { const auto deadline = std::chrono::steady_clock::now() + session.lease_ttl; const auto max_sleep_interval = std::chrono::duration_cast( kRenewCheckInterval); while (true) { + // The caller checks the request first and exits on a shutdown. + if (ShutdownObserved(shutdown_requested)) { + return true; + } auto renewed = coordinator.RenewLeadership(session); if (!renewed) { return tl::make_unexpected(renewed.error()); @@ -311,7 +415,8 @@ void EnterStandbyMode(MasterAdminServer& admin_server, int RunSupervisorLoop(const HABackendSpec& spec, const MasterServiceSupervisorConfig& config, MasterAdminServer& admin_server, - std::unique_ptr standby_controller) { + std::unique_ptr standby_controller, + const std::function& shutdown_requested) { auto label_reconciler = MakeLeaderLabelReconciler(config); label_reconciler.SetLeader(false); SetRuntimeState(admin_server, MasterRuntimeState::kStarting); @@ -325,10 +430,33 @@ int RunSupervisorLoop(const HABackendSpec& spec, SetRuntimeState(admin_server, state); }); + // Every supervisor-selected shutdown ends here without unwinding. + // Returning would destroy the per-iteration coordinator, and the etcd + // coordinator's destructor revokes a held lease. + auto shutdown_exit = [&](std::string_view phase, + const ShutdownLease& lease) -> int { + LOG(INFO) << "Shutdown requested, finishing from phase=" << phase; + admin_server.Stop(); + accept_standby_runtime_updates.store(false, std::memory_order_release); + standby_controller->StopStandby(); + ExitAfterShutdown(FinishShutdown(lease)); + }; + // Result of the ordinary release that ended the previous term, for a + // request observed before the next standby check. + std::optional released_last_term; + + if (ShutdownObserved(shutdown_requested)) { + return shutdown_exit("starting", ShutdownLease{}); + } EnterStandbyMode(admin_server, *standby_controller, accept_standby_runtime_updates, std::nullopt); while (true) { + if (ShutdownObserved(shutdown_requested)) { + return shutdown_exit( + "standby", ShutdownLease{.prior_release = released_last_term}); + } + released_last_term.reset(); auto coordinator = CreateLeaderCoordinator(spec); if (!coordinator) { if (HandleSupervisorError("create leader coordinator", @@ -342,9 +470,15 @@ int RunSupervisorLoop(const HABackendSpec& spec, std::optional leadership_session; while (!leadership_session.has_value()) { + if (ShutdownObserved(shutdown_requested)) { + return shutdown_exit("candidate", ShutdownLease{}); + } SetRuntimeState(admin_server, MasterRuntimeState::kCandidate); auto current_view = leader_coordinator.ReadCurrentView(); + if (ShutdownObserved(shutdown_requested)) { + return shutdown_exit("candidate", ShutdownLease{}); + } if (!current_view) { EnterStandbyMode(admin_server, *standby_controller, accept_standby_runtime_updates, std::nullopt); @@ -358,9 +492,15 @@ int RunSupervisorLoop(const HABackendSpec& spec, UpdateObservedLeader(admin_server, *standby_controller, current_view.value()); if (!current_view.value().has_value()) { + if (ShutdownObserved(shutdown_requested)) { + return shutdown_exit("candidate", ShutdownLease{}); + } auto acquire = leader_coordinator.TryAcquireLeadership( config.local_hostname); if (!acquire) { + if (ShutdownObserved(shutdown_requested)) { + return shutdown_exit("candidate", ShutdownLease{}); + } EnterStandbyMode(admin_server, *standby_controller, accept_standby_runtime_updates, std::nullopt); @@ -383,6 +523,9 @@ int RunSupervisorLoop(const HABackendSpec& spec, break; } + if (ShutdownObserved(shutdown_requested)) { + return shutdown_exit("candidate", ShutdownLease{}); + } EnterStandbyMode(admin_server, *standby_controller, accept_standby_runtime_updates, acquire->observed_view); @@ -423,6 +566,28 @@ int RunSupervisorLoop(const HABackendSpec& spec, continue; } + ShutdownLease term_lease{.session = &*leadership_session}; + if (ShutdownObserved(shutdown_requested)) { + return shutdown_exit("acquired", term_lease); + } + // The ordinary release that ends a failed or lost term. A request + // observed before it skips the release; one observed after it keeps + // the result, so shutdown neither repeats it nor reports no lease. + auto release_term = [&](std::string_view phase) -> ErrorCode { + if (ShutdownObserved(shutdown_requested)) { + shutdown_exit(phase, term_lease); // Does not return. + } + const ErrorCode err = + leader_coordinator.ReleaseLeadership(*leadership_session); + term_lease = ShutdownLease{.prior_release = err}; + released_last_term = err; + TestFailPoint::Wait("supervisor_after_term_release"); + if (ShutdownObserved(shutdown_requested)) { + shutdown_exit(phase, term_lease); // Does not return. + } + return err; + }; + // Start the lease keepalive before catch-up. The grant happens in // TryAcquireLeadership, but etcd KeepAlive previously started only in // the warmup phase, so a large Oplog replay expired the lease first. @@ -430,14 +595,16 @@ int RunSupervisorLoop(const HABackendSpec& spec, auto claim_error = ClaimProducerViewForServing( spec, config, leadership_session->view.view_version); + if (ShutdownObserved(shutdown_requested)) { + return shutdown_exit("claim", term_lease); + } if (claim_error != ErrorCode::OK) { EnterStandbyMode(admin_server, *standby_controller, accept_standby_runtime_updates, leadership_session->view); if (HandleLeadershipPhaseError( "producer view claim failure", "claim producer view", - leader_coordinator, *leadership_session, claim_error, - spec.type)) { + release_term("claim"), claim_error, spec.type)) { return -1; } continue; @@ -445,14 +612,17 @@ int RunSupervisorLoop(const HABackendSpec& spec, accept_standby_runtime_updates.store(false, std::memory_order_release); auto promotion_ctx = standby_controller->PromoteStandbyAndExport(); + if (ShutdownObserved(shutdown_requested)) { + return shutdown_exit("promotion", term_lease); + } if (!promotion_ctx) { EnterStandbyMode(admin_server, *standby_controller, accept_standby_runtime_updates, leadership_session->view); - if (HandleLeadershipPhaseError( - "standby promotion failure", "promote standby for serve", - leader_coordinator, *leadership_session, - promotion_ctx.error(), spec.type)) { + if (HandleLeadershipPhaseError("standby promotion failure", + "promote standby for serve", + release_term("promotion"), + promotion_ctx.error(), spec.type)) { return -1; } continue; @@ -460,6 +630,10 @@ int RunSupervisorLoop(const HABackendSpec& spec, auto return_to_standby = [&](const std::optional& leader_view) { + if (ShutdownObserved(shutdown_requested)) { + shutdown_exit("standby_reentry", + term_lease); // Does not return. + } if (promotion_ctx->metadata_store && promotion_ctx->applied_cursor.last_seq != 0) { const uint64_t applied_seq = @@ -485,32 +659,32 @@ int RunSupervisorLoop(const HABackendSpec& spec, << "applied_seq=" << promotion_ctx->applied_cursor.last_seq; return_to_standby(std::nullopt); - LogLeadershipReleaseWarning( - "catch-up lease loss", - leader_coordinator.ReleaseLeadership(*leadership_session)); + LogLeadershipReleaseWarning("catch-up lease loss", + release_term("catch_up_loss")); admin_server.SetObservedLeader(std::nullopt); continue; } LOG(INFO) << "Entering warmup phase..."; SetRuntimeState(admin_server, MasterRuntimeState::kLeaderWarmup); - auto warmup_result = - WarmupLeadership(leader_coordinator, *leadership_session); + auto warmup_result = WarmupLeadership( + leader_coordinator, *leadership_session, shutdown_requested); + if (ShutdownObserved(shutdown_requested)) { + return shutdown_exit("warmup", term_lease); + } if (!warmup_result) { return_to_standby(leadership_session->view); if (HandleLeadershipPhaseError( "renewal startup failure", "start leadership renewal", - leader_coordinator, *leadership_session, - warmup_result.error(), spec.type)) { + release_term("warmup"), warmup_result.error(), spec.type)) { return -1; } continue; } if (!warmup_result.value()) { return_to_standby(std::nullopt); - LogLeadershipReleaseWarning( - "warmup expiration", - leader_coordinator.ReleaseLeadership(*leadership_session)); + LogLeadershipReleaseWarning("warmup expiration", + release_term("warmup_expiry")); LOG(WARNING) << "Leadership expired during warmup phase"; admin_server.SetObservedLeader(std::nullopt); continue; @@ -542,7 +716,26 @@ int RunSupervisorLoop(const HABackendSpec& spec, // Runs the term's handlers once they are registered. It drains before // the service goes, so release the service through release_service. std::unique_ptr worker_pool; + // Stops the term's resources that exist, in the serving-exit order, + // and keeps the service, the index and the gate alive through the + // exit. Each step is a no-op on a resource not yet created or + // already stopped. + auto stop_term_resources = + [&](std::unique_ptr& monitor) { + worker_pool.reset(); + StopLeadershipMonitor(monitor); + wrapped_master_service->StopBatchOpLogWriter(); + DeactivateServingState(admin_server, label_reconciler); + LOG(INFO) << "Shutdown stopped the term's workers, monitor " + "and OpLog writer"; + }; auto release_service = [&]() { + if (ShutdownObserved(shutdown_requested)) { + std::unique_ptr no_monitor; + stop_term_resources(no_monitor); + shutdown_exit("service_release", + term_lease); // Does not return. + } worker_pool.reset(); wrapped_master_service.reset(); }; @@ -572,6 +765,11 @@ int RunSupervisorLoop(const HABackendSpec& spec, : wrapped_master_service->RestoreFromStandby( promotion_ctx->objects, promotion_ctx->applied_seq_id, promotion_ctx->segments, promotion_ctx->weight_metadata); + if (ShutdownObserved(shutdown_requested)) { + std::unique_ptr no_monitor; + stop_term_resources(no_monitor); + return shutdown_exit("restore", term_lease); + } if (!restore_result) { LOG(ERROR) << "Standby restore failed: " << toString(restore_result.error()); @@ -579,10 +777,10 @@ int RunSupervisorLoop(const HABackendSpec& spec, DeactivateServingState(admin_server, label_reconciler); return_to_standby(leadership_session->view); SetRuntimeState(admin_server, MasterRuntimeState::kRecovering); - if (HandleLeadershipPhaseError( - "standby restore failure", "restore standby state", - leader_coordinator, *leadership_session, - restore_result.error(), spec.type)) { + if (HandleLeadershipPhaseError("standby restore failure", + "restore standby state", + release_term("restore"), + restore_result.error(), spec.type)) { return -1; } continue; @@ -597,9 +795,8 @@ int RunSupervisorLoop(const HABackendSpec& spec, release_service(); DeactivateServingState(admin_server, label_reconciler); return_to_standby(std::nullopt); - LogLeadershipReleaseWarning( - "restore lease loss", - leader_coordinator.ReleaseLeadership(*leadership_session)); + LogLeadershipReleaseWarning("restore lease loss", + release_term("restore_loss")); admin_server.SetObservedLeader(std::nullopt); continue; } @@ -608,14 +805,20 @@ int RunSupervisorLoop(const HABackendSpec& spec, auto serve_preflight = leader_coordinator.RenewLeadership(*leadership_session); + if (ShutdownObserved(shutdown_requested)) { + std::unique_ptr no_monitor; + stop_term_resources(no_monitor); + return shutdown_exit("preflight", term_lease); + } if (!serve_preflight) { release_service(); DeactivateServingState(admin_server, label_reconciler); return_to_standby(leadership_session->view); - if (HandleLeadershipPhaseError( - "serve preflight failure", - "validate leadership before serving", leader_coordinator, - *leadership_session, serve_preflight.error(), spec.type)) { + if (HandleLeadershipPhaseError("serve preflight failure", + "validate leadership before serving", + release_term("preflight"), + serve_preflight.error(), + spec.type)) { return -1; } continue; @@ -624,9 +827,8 @@ int RunSupervisorLoop(const HABackendSpec& spec, release_service(); DeactivateServingState(admin_server, label_reconciler); return_to_standby(std::nullopt); - LogLeadershipReleaseWarning( - "serve preflight expiration", - leader_coordinator.ReleaseLeadership(*leadership_session)); + LogLeadershipReleaseWarning("serve preflight expiration", + release_term("preflight_expiry")); LOG(WARNING) << "Leadership expired before entering serve phase"; admin_server.SetObservedLeader(std::nullopt); continue; @@ -651,41 +853,72 @@ int RunSupervisorLoop(const HABackendSpec& spec, server.stop(); }); }); + if (ShutdownObserved(shutdown_requested)) { + std::unique_ptr started_monitor; + if (leadership_monitor) { + started_monitor = std::move(leadership_monitor.value()); + } + stop_term_resources(started_monitor); + return shutdown_exit("monitor_start", term_lease); + } if (!leadership_monitor) { release_service(); DeactivateServingState(admin_server, label_reconciler); return_to_standby(leadership_session->view); if (HandleLeadershipPhaseError( "serve monitor startup failure", "start leadership monitor", - leader_coordinator, *leadership_session, - leadership_monitor.error(), spec.type)) { + release_term("monitor_start"), leadership_monitor.error(), + spec.type)) { return -1; } continue; } auto leadership_monitor_handle = std::move(leadership_monitor.value()); + ServingShutdownTrigger shutdown_trigger(shutdown_requested, [&]() { + serving_state.RequestShutdown([&]() { + admin_server.SetServiceAvailable(false); + label_reconciler.SetLeader(false); + SetRuntimeState(admin_server, MasterRuntimeState::kStandby); + LOG(INFO) << "Trying to stop server, reason=shutdown requested"; + server.stop(); + }); + }); + std::optional> ec; - serving_state.RunIfActive([&]() { ec.emplace(server.async_start()); }); + TestFailPoint::Wait("supervisor_before_serve_start"); + serving_state.RunIfActive([&]() { + if (!ShutdownObserved(shutdown_requested)) { + ec.emplace(server.async_start()); + } + }); if (!ec.has_value()) { + shutdown_trigger.Stop(); + if (ShutdownObserved(shutdown_requested)) { + stop_term_resources(leadership_monitor_handle); + return shutdown_exit("serve_start", term_lease); + } StopLeadershipMonitor(leadership_monitor_handle); release_service(); DeactivateServingState(admin_server, label_reconciler); return_to_standby(std::nullopt); - LogLeadershipReleaseWarning( - "serve startup cancellation", - leader_coordinator.ReleaseLeadership(*leadership_session)); + LogLeadershipReleaseWarning("serve startup cancellation", + release_term("serve_start_cancel")); continue; } if (ec->hasResult()) { + shutdown_trigger.Stop(); LOG(ERROR) << "Failed to start master service: " << ec->result().value(); + if (ShutdownObserved(shutdown_requested)) { + stop_term_resources(leadership_monitor_handle); + return shutdown_exit("serve_start_failure", term_lease); + } StopLeadershipMonitor(leadership_monitor_handle); release_service(); DeactivateServingState(admin_server, label_reconciler); return_to_standby(leadership_session->view); - auto err = - leader_coordinator.ReleaseLeadership(*leadership_session); + auto err = release_term("serve_start_failure"); if (err != ErrorCode::OK) { LOG(ERROR) << "Failed to release leadership: " << toString(err); } @@ -693,14 +926,18 @@ int RunSupervisorLoop(const HABackendSpec& spec, } ErrorCode publish_ready_err = ErrorCode::OK; - if (serving_state.IsActive()) { + if (serving_state.IsActive() && !ShutdownObserved(shutdown_requested)) { // The backend call can take several seconds. Keep it outside the // transition gate so leadership loss can stop the listener // promptly. publish_ready_err = leader_coordinator.PublishServiceReady(*leadership_session); if (publish_ready_err == ErrorCode::OK) { + TestFailPoint::Wait("supervisor_before_serve_activation"); serving_state.RunIfActive([&]() { + if (ShutdownObserved(shutdown_requested)) { + return; + } ActivateServingState(admin_server, wrapped_master_service, label_reconciler); // Serving owns the restored index now; drop the handoff @@ -722,20 +959,27 @@ int RunSupervisorLoop(const HABackendSpec& spec, << toString(publish_ready_err); StopLeadershipMonitor(leadership_monitor_handle); auto server_err = std::move(ec.value()).get(); + shutdown_trigger.Stop(); LOG(ERROR) << "Master service stopped after readiness failure: " << server_err; + if (ShutdownObserved(shutdown_requested)) { + stop_term_resources(leadership_monitor_handle); + return shutdown_exit("readiness_failure", term_lease); + } release_service(); return_to_standby(leadership_session->view); if (HandleLeadershipPhaseError( "service readiness publication failure", - "publish master service readiness", leader_coordinator, - *leadership_session, publish_ready_err, spec.type)) { + "publish master service readiness", + release_term("readiness_failure"), publish_ready_err, + spec.type)) { return -1; } continue; } auto server_err = std::move(ec.value()).get(); + shutdown_trigger.Stop(); // Finish the calls the workers still hold before the term's OpLog // writer and leadership go, as inline handlers would have. worker_pool.reset(); @@ -744,10 +988,13 @@ int RunSupervisorLoop(const HABackendSpec& spec, StopLeadershipMonitor(leadership_monitor_handle); wrapped_master_service->StopBatchOpLogWriter(); DeactivateServingState(admin_server, label_reconciler); - auto err = leader_coordinator.ReleaseLeadership(*leadership_session); + auto err = release_term("serving"); LOG(INFO) << "Release leadership: " << toString(err); auto current_view = leader_coordinator.ReadCurrentView(); release_service(); + if (ShutdownObserved(shutdown_requested)) { + return shutdown_exit("standby_reentry", term_lease); + } // Serving drops the handoff in ActivateServingState. If the listener // stopped before that, preserve the still-populated catch-up cursor. if (promotion_ctx->metadata_store) { @@ -769,8 +1016,9 @@ int RunSupervisorLoop(const HABackendSpec& spec, } // namespace MasterServiceSupervisor::MasterServiceSupervisor( - const MasterServiceSupervisorConfig& config) - : config_(config) {} + const MasterServiceSupervisorConfig& config, + std::function shutdown_requested) + : config_(config), shutdown_requested_(std::move(shutdown_requested)) {} int MasterServiceSupervisor::Start() { auto spec = BuildHABackendSpec(config_); @@ -799,7 +1047,8 @@ int MasterServiceSupervisor::Start() { return -1; } return RunSupervisorLoop(*spec, config_, admin_server, - std::move(standby_controller)); + std::move(standby_controller), + shutdown_requested_); } } // namespace ha diff --git a/mooncake-store/src/master.cpp b/mooncake-store/src/master.cpp index 8f2446fc8b..3db2691b73 100644 --- a/mooncake-store/src/master.cpp +++ b/mooncake-store/src/master.cpp @@ -27,6 +27,7 @@ #include "master_admin_service.h" #include "rpc_worker_pool.h" #include "rpc_service.h" +#include "shutdown_request.h" #include "types.h" #include "common/network.h" #include "glog_compat.h" @@ -1639,12 +1640,34 @@ int main(int argc, char* argv[]) { } if (master_config.enable_ha) { + // Arm the shutdown bound and the handlers before any lease can be + // acquired; without them a SIGTERM kills a leader mid-term. This + // call is serial and one-shot from main(): the check-then-store in + // StartShutdownWatchdog assumes no concurrent caller. + if (!mooncake::StartShutdownWatchdog( + mooncake::ha::kHaShutdownDeadline)) { + LOG(ERROR) << "Failed to start the shutdown watchdog, " + "refusing to start"; + return 1; + } + if (!mooncake::InstallShutdownSignalHandlers()) { + PLOG(ERROR) << "Failed to install SIGINT/SIGTERM handlers, " + "refusing to start"; + return 1; + } mooncake::MasterServiceSupervisorConfig supervisor_config{ master_config}; supervisor_config.http_metadata_server = metadata_server_ptr; supervisor_config.http_metadata_remote_url = http_metadata_remote_url; mooncake::ha::MasterServiceSupervisor supervisor(supervisor_config); - return supervisor.Start(); + const int result = supervisor.Start(); + // Finish a requested process exit without further main/static + // teardown; supervisor-owned cleanup may already have run on this + // error path. + if (mooncake::ShutdownRequested() || mooncake::ShutdownLatched()) { + mooncake::ExitAfterShutdown(result); + } + return result; } else { // version is not used in non-HA mode, just pass a dummy value mooncake::ViewVersionId version = 0; diff --git a/mooncake-store/src/rpc_service.cpp b/mooncake-store/src/rpc_service.cpp index de169b404e..174cc65296 100644 --- a/mooncake-store/src/rpc_service.cpp +++ b/mooncake-store/src/rpc_service.cpp @@ -10,6 +10,7 @@ #include "master_metric_manager.h" #include "master_service.h" +#include "ha/oplog/oplog_test_failpoint.h" #include "rpc_worker_pool.h" #include "rpc_helper.h" #include "types.h" @@ -993,6 +994,7 @@ std::vector> WrappedMasterService::BatchRemove( tl::expected WrappedMasterService::MountSegment( const Segment& segment, const UUID& client_id) { + TestFailPoint::Wait("rpc_mount_segment_before_service"); return execute_rpc( "MountSegment", [&] { return master_service_.MountSegment(segment, client_id); }, diff --git a/mooncake-store/src/shutdown_request.cpp b/mooncake-store/src/shutdown_request.cpp new file mode 100644 index 0000000000..5d5ea5db42 --- /dev/null +++ b/mooncake-store/src/shutdown_request.cpp @@ -0,0 +1,112 @@ +#include "shutdown_request.h" + +#include +#include + +#include +#include +#include +#include +#include +#include + +namespace mooncake { +namespace { + +constexpr auto kWatchdogPollInterval = std::chrono::milliseconds(50); + +// 0 until the first SIGINT/SIGTERM, then that signal number. +std::atomic g_shutdown_signal{0}; +static_assert(std::atomic::is_always_lock_free, + "the signal handler needs a lock-free flag"); + +// A shutdown request seen outside the signal handler. The watchdog polls it. +std::atomic g_shutdown_latched{false}; +static_assert(std::atomic::is_always_lock_free, + "the watchdog polls a lock-free flag"); + +// Set only after the watchdog thread exists; gates the exit flush. +std::atomic g_watchdog_running{false}; + +void HandleShutdownSignal(int signo) { + // Async-signal-safe: one lock-free exchange, or _exit on a repeat. No + // logging, locking or allocation. + if (g_shutdown_signal.exchange(signo, std::memory_order_acq_rel) != 0) { + _exit(128 + signo); + } +} + +void RunWatchdog(std::chrono::milliseconds deadline) { + // Sleep-polls two atomics only, so a stalled cleanup, a held glog mutex + // or a full stderr pipe cannot stop it. + while (!ShutdownRequested() && !ShutdownLatched()) { + std::this_thread::sleep_for(kWatchdogPollInterval); + } + const auto expires_at = std::chrono::steady_clock::now() + deadline; + for (auto now = std::chrono::steady_clock::now(); now < expires_at; + now = std::chrono::steady_clock::now()) { + std::this_thread::sleep_for( + std::min( + expires_at - now, kWatchdogPollInterval)); + } + _exit(EXIT_FAILURE); +} + +} // namespace + +namespace detail { + +bool InstallShutdownSignalHandlersWith(SigactionFn install) { + struct sigaction action{}; + action.sa_handler = HandleShutdownSignal; + sigemptyset(&action.sa_mask); + action.sa_flags = SA_ONSTACK | SA_RESTART; + for (int signo : {SIGINT, SIGTERM}) { + if (install(signo, &action, nullptr) != 0) { + return false; + } + } + return true; +} + +} // namespace detail + +bool InstallShutdownSignalHandlers() { + return detail::InstallShutdownSignalHandlersWith(&::sigaction); +} + +bool ShutdownRequested() { + return g_shutdown_signal.load(std::memory_order_acquire) != 0; +} + +void LatchShutdownRequest() { + g_shutdown_latched.store(true, std::memory_order_release); +} + +bool ShutdownLatched() { + return g_shutdown_latched.load(std::memory_order_acquire); +} + +bool StartShutdownWatchdog(std::chrono::milliseconds deadline) { + // Serial callers only: the check and the store are separate steps. + if (g_watchdog_running.load(std::memory_order_acquire)) { + return true; + } + try { + std::thread(RunWatchdog, deadline).detach(); + } catch (const std::system_error&) { + return false; + } + g_watchdog_running.store(true, std::memory_order_release); + return true; +} + +void ExitAfterShutdown(int status) { + LatchShutdownRequest(); + if (g_watchdog_running.load(std::memory_order_acquire)) { + google::FlushLogFiles(google::GLOG_INFO); + } + _exit(status); +} + +} // namespace mooncake diff --git a/mooncake-store/tests/CMakeLists.txt b/mooncake-store/tests/CMakeLists.txt index 00fb1a2a7a..7064ee17b9 100644 --- a/mooncake-store/tests/CMakeLists.txt +++ b/mooncake-store/tests/CMakeLists.txt @@ -501,6 +501,13 @@ target_compile_definitions(oplog_test_failpoint_test target_link_libraries(oplog_test_failpoint_test PRIVATE glog::glog gtest gtest_main pthread) add_test(NAME oplog_test_failpoint_test COMMAND oplog_test_failpoint_test) +add_executable(shutdown_request_test shutdown_request_test.cpp + ../src/shutdown_request.cpp) +target_include_directories(shutdown_request_test + PRIVATE ${MOONCAKE_STORE_INCLUDE_ROOT}) +target_link_libraries(shutdown_request_test PRIVATE glog::glog gtest gtest_main + pthread) +add_test(NAME shutdown_request_test COMMAND shutdown_request_test) add_ha_test(master_service_ha_test ha/master_service_ha_test.cpp) foreach( diff --git a/mooncake-store/tests/ha/leadership/high_availability_test.cpp b/mooncake-store/tests/ha/leadership/high_availability_test.cpp index 9b03e221d0..1c61a22b33 100644 --- a/mooncake-store/tests/ha/leadership/high_availability_test.cpp +++ b/mooncake-store/tests/ha/leadership/high_availability_test.cpp @@ -1,3 +1,4 @@ +#include #include #include @@ -8,20 +9,29 @@ #include #include #include +#include +#include #include #include #include #include +#include #include +#include #ifdef STORE_USE_ETCD #include "etcd_helper.h" #endif #include "common/network.h" +#include "ha/kv/ha_kv_backend_factory.h" #include "ha/leadership/leader_coordinator_factory.h" #include "ha/leadership/high_availability_test_fixture.h" #include "ha/leadership/master_service_supervisor.h" +#include "ha/oplog/oplog_batch_storage.h" +#include "ha/oplog/oplog_types.h" +#include "master_client.h" #include "master_service.h" +#include "shutdown_request.h" #include "types.h" namespace mooncake { @@ -35,6 +45,15 @@ DEFINE_bool(ha_supervisor_child, false, DEFINE_int32(ha_supervisor_rpc_port, 0, "Child supervisor RPC port"); DEFINE_int32(ha_supervisor_admin_port, 0, "Child supervisor admin port"); DEFINE_string(ha_supervisor_cluster_id, "", "Child supervisor cluster ID"); +DEFINE_bool(ha_supervisor_enable_oplog, false, + "Child supervisor enables the batch OpLog"); +DEFINE_string(ha_supervisor_shutdown_file, "", + "Child supervisor also treats this file's existence as a " + "shutdown request"); +DEFINE_int64(ha_supervisor_client_live_ttl_sec, DEFAULT_CLIENT_LIVE_TTL_SEC, + "Child client active TTL"); +DEFINE_int64(ha_supervisor_client_suspicion_ttl_sec, + DEFAULT_CLIENT_SUSPICION_TTL_SEC, "Child client suspicion TTL"); void HighAvailabilityTest::SetUpTestSuite() { // Initialize glog @@ -178,31 +197,71 @@ class FakeLeaderCoordinator : public ha::LeaderCoordinator { class SupervisorChildProcess { public: - SupervisorChildProcess(int rpc_port, int admin_port, std::string cluster_id) + SupervisorChildProcess( + int rpc_port, int admin_port, std::string cluster_id, + bool enable_oplog = false, std::string log_path = "", + std::string shutdown_file = "", + int64_t client_live_ttl_sec = DEFAULT_CLIENT_LIVE_TTL_SEC, + int64_t client_suspicion_ttl_sec = DEFAULT_CLIENT_SUSPICION_TTL_SEC) : rpc_port_(rpc_port), admin_port_(admin_port), - cluster_id_(std::move(cluster_id)) {} + cluster_id_(std::move(cluster_id)), + enable_oplog_(enable_oplog), + log_path_(std::move(log_path)), + shutdown_file_(std::move(shutdown_file)), + client_live_ttl_sec_(client_live_ttl_sec), + client_suspicion_ttl_sec_(client_suspicion_ttl_sec) {} ~SupervisorChildProcess() { Stop(); } bool Start() { - pid_ = fork(); - if (pid_ != 0) { - return pid_ > 0; + // The parent may run client threads, so build everything the child + // needs before fork(); between fork and exec the child only calls + // dup2, execv and _exit. + std::vector args = { + "/proc/self/exe", + "--ha_supervisor_child=true", + "--etcd_endpoints=" + FLAGS_etcd_endpoints, + "--ha_supervisor_rpc_port=" + std::to_string(rpc_port_), + "--ha_supervisor_admin_port=" + std::to_string(admin_port_), + "--ha_supervisor_cluster_id=" + cluster_id_, + std::string("--ha_supervisor_enable_oplog=") + + (enable_oplog_ ? "true" : "false"), + "--ha_supervisor_client_live_ttl_sec=" + + std::to_string(client_live_ttl_sec_), + "--ha_supervisor_client_suspicion_ttl_sec=" + + std::to_string(client_suspicion_ttl_sec_), + }; + if (!shutdown_file_.empty()) { + args.push_back("--ha_supervisor_shutdown_file=" + shutdown_file_); + } + std::vector argv; + for (auto& arg : args) { + argv.push_back(arg.data()); + } + argv.push_back(nullptr); + + int log_fd = -1; + if (!log_path_.empty()) { + log_fd = open(log_path_.c_str(), + O_WRONLY | O_CREAT | O_TRUNC | O_CLOEXEC, 0644); + if (log_fd < 0) { + return false; + } } - const std::string endpoints_arg = - "--etcd_endpoints=" + FLAGS_etcd_endpoints; - const std::string rpc_port_arg = - "--ha_supervisor_rpc_port=" + std::to_string(rpc_port_); - const std::string admin_port_arg = - "--ha_supervisor_admin_port=" + std::to_string(admin_port_); - const std::string cluster_arg = - "--ha_supervisor_cluster_id=" + cluster_id_; - execl("/proc/self/exe", "/proc/self/exe", "--ha_supervisor_child=true", - endpoints_arg.c_str(), rpc_port_arg.c_str(), - admin_port_arg.c_str(), cluster_arg.c_str(), nullptr); - _exit(127); + pid_ = fork(); + if (pid_ == 0) { + if (log_fd >= 0) { + dup2(log_fd, STDERR_FILENO); + } + execv("/proc/self/exe", argv.data()); + _exit(127); + } + if (log_fd >= 0) { + close(log_fd); + } + return pid_ > 0; } bool IsRunning() { @@ -213,10 +272,25 @@ class SupervisorChildProcess { if (waitpid(pid_, &status, WNOHANG) == 0) { return true; } + exit_status_ = status; pid_ = 0; return false; } + bool Signal(int signo) { return pid_ > 0 && kill(pid_, signo) == 0; } + + // Raw wait status, or nullopt if the child is still running at timeout. + std::optional WaitForExit(std::chrono::milliseconds timeout) { + const auto deadline = std::chrono::steady_clock::now() + timeout; + while (IsRunning()) { + if (std::chrono::steady_clock::now() >= deadline) { + return std::nullopt; + } + std::this_thread::sleep_for(std::chrono::milliseconds(10)); + } + return exit_status_; + } + void Stop() { if (pid_ <= 0) { return; @@ -228,9 +302,15 @@ class SupervisorChildProcess { private: pid_t pid_{0}; + std::optional exit_status_; int rpc_port_; int admin_port_; std::string cluster_id_; + bool enable_oplog_; + std::string log_path_; + std::string shutdown_file_; + int64_t client_live_ttl_sec_; + int64_t client_suspicion_ttl_sec_; }; struct HttpResponse { @@ -277,8 +357,9 @@ int RunSupervisorChild() { config.nof_eviction_ratio = DEFAULT_NOF_EVICTION_RATIO; config.nof_eviction_high_watermark_ratio = DEFAULT_NOF_EVICTION_HIGH_WATERMARK_RATIO; - config.client_active_ttl_sec = DEFAULT_CLIENT_LIVE_TTL_SEC; - config.client_suspicion_ttl_sec = DEFAULT_CLIENT_SUSPICION_TTL_SEC; + config.client_active_ttl_sec = FLAGS_ha_supervisor_client_live_ttl_sec; + config.client_suspicion_ttl_sec = + FLAGS_ha_supervisor_client_suspicion_ttl_sec; config.nof_heartbeat_interval_sec = DEFAULT_NOF_HEARTBEAT_INTERVAL_SEC; config.nof_heartbeat_probe_timeout_ms = DEFAULT_NOF_HEARTBEAT_PROBE_TIMEOUT_MS; @@ -294,10 +375,26 @@ int RunSupervisorChild() { config.ha_backend_connstring = FLAGS_etcd_endpoints; config.etcd_endpoints = FLAGS_etcd_endpoints; config.cluster_id = FLAGS_ha_supervisor_cluster_id; - config.enable_oplog = false; + config.enable_oplog = FLAGS_ha_supervisor_enable_oplog; - ha::MasterServiceSupervisor supervisor(config); - return supervisor.Start(); + // Arm the shutdown bound and the handlers like main() does in HA mode. + if (!StartShutdownWatchdog(ha::kHaShutdownDeadline) || + !InstallShutdownSignalHandlers()) { + return 1; + } + std::function shutdown_requested = ShutdownRequested; + if (!FLAGS_ha_supervisor_shutdown_file.empty()) { + shutdown_requested = [path = FLAGS_ha_supervisor_shutdown_file] { + std::error_code ec; + return ShutdownRequested() || std::filesystem::exists(path, ec); + }; + } + ha::MasterServiceSupervisor supervisor(config, shutdown_requested); + const int result = supervisor.Start(); + if (ShutdownRequested() || ShutdownLatched()) { + ExitAfterShutdown(result); + } + return result; } TEST_F(HighAvailabilityTest, AcquiredViewFlowsIntoServingMasterService) { @@ -467,6 +564,768 @@ TEST_F(HighAvailabilityTest, HaWithoutOplogRestoresEmptyContextAndServes) { EXPECT_EQ(HttpGet(ports[1], "/get_all_keys").status, 200); } +namespace { + +using namespace std::chrono_literals; + +constexpr std::string_view kWarmingPrefix = "__mooncake_service_warming__:"; + +#ifdef MOONCAKE_ENABLE_TEST_FAILPOINTS +constexpr bool kFailPointsEnabled = true; +#else +constexpr bool kFailPointsEnabled = false; +#endif + +// File-handshake fail points (see TestFailPoint::Wait): the test arms a point, +// waits for the process under test to hit it, and releases it. Without +// MOONCAKE_ENABLE_TEST_FAILPOINTS the points are no-ops and every method here +// does nothing, so cases that need them skip first. +class ScopedFailPointDir { + public: + ScopedFailPointDir() { + if (!kFailPointsEnabled) { + return; + } + std::string pattern = + ::testing::TempDir() + "mooncake-shutdown-fp-XXXXXX"; + if (mkdtemp(pattern.data()) == nullptr) { + ADD_FAILURE() << "mkdtemp failed for " << pattern; + return; + } + dir_ = pattern; + setenv("MOONCAKE_TEST_FAILPOINT_DIR", dir_.c_str(), 1); + setenv("MOONCAKE_TEST_FAILPOINT_TIMEOUT_SEC", "20", 1); + } + + ~ScopedFailPointDir() { + if (dir_.empty()) { + return; + } + unsetenv("MOONCAKE_TEST_FAILPOINT_DIR"); + unsetenv("MOONCAKE_TEST_FAILPOINT_TIMEOUT_SEC"); + std::error_code ec; + std::filesystem::remove_all(dir_, ec); + } + + ScopedFailPointDir(const ScopedFailPointDir&) = delete; + ScopedFailPointDir& operator=(const ScopedFailPointDir&) = delete; + + void Arm(const std::string& name) { Touch(name + ".arm"); } + + bool WaitForHit(const std::string& name, + std::chrono::milliseconds timeout) { + const auto deadline = std::chrono::steady_clock::now() + timeout; + while (std::chrono::steady_clock::now() < deadline) { + if (std::filesystem::exists(dir_ + "/" + name + ".hit")) { + return true; + } + std::this_thread::sleep_for(10ms); + } + return false; + } + + void Release(const std::string& name) { Touch(name + ".release"); } + + private: + void Touch(const std::string& file) { + if (!dir_.empty()) { + std::ofstream(dir_ + "/" + file).put('\n'); + } + } + + std::string dir_; +}; + +std::string TempPath(const std::string& name, const std::string& suffix) { + return ::testing::TempDir() + name + "-" + std::to_string(getpid()) + + suffix; +} + +std::string MasterViewKey(const std::string& cluster_id) { + return "mooncake-store/" + cluster_id + "/master_view"; +} + +std::optional ReadEtcdKey(const std::string& key, + EtcdRevisionId* revision = nullptr) { + std::string value; + EtcdRevisionId rev = 0; + const ErrorCode err = EtcdHelper::Get(key.c_str(), key.size(), value, rev); + if (err == ErrorCode::ETCD_KEY_NOT_EXIST) { + return std::nullopt; + } + if (err != ErrorCode::OK) { + ADD_FAILURE() << "etcd Get(" << key << ") failed: " << toString(err); + return std::nullopt; + } + if (revision != nullptr) { + *revision = rev; + } + return value; +} + +bool WaitForEtcdKeyGone(const std::string& key, + std::chrono::milliseconds timeout) { + const auto deadline = std::chrono::steady_clock::now() + timeout; + while (std::chrono::steady_clock::now() < deadline) { + if (!ReadEtcdKey(key).has_value()) { + return true; + } + std::this_thread::sleep_for(100ms); + } + return false; +} + +bool WaitForHealthSubstring(SupervisorChildProcess& child, int admin_port, + const std::string& needle, + std::chrono::milliseconds timeout) { + const auto deadline = std::chrono::steady_clock::now() + timeout; + while (std::chrono::steady_clock::now() < deadline && child.IsRunning()) { + const auto health = HttpGet(admin_port, "/health"); + if (health.status == 200 && + health.body.find(needle) != std::string::npos) { + return true; + } + std::this_thread::sleep_for(100ms); + } + return false; +} + +std::string ReadTextFile(const std::string& path) { + std::ifstream input(path); + return std::string(std::istreambuf_iterator(input), + std::istreambuf_iterator()); +} + +bool WaitForLogLine(const std::string& path, const std::string& needle, + std::chrono::milliseconds timeout) { + const auto deadline = std::chrono::steady_clock::now() + timeout; + while (std::chrono::steady_clock::now() < deadline) { + if (ReadTextFile(path).find(needle) != std::string::npos) { + return true; + } + std::this_thread::sleep_for(20ms); + } + return false; +} + +// True if the child logged "finishing from phase=" exactly, so that +// "serve_start" does not match "serve_start_failure". +bool LogHasPhase(const std::string& log, std::string_view phase) { + return log.find("finishing from phase=" + std::string(phase) + "\n") != + std::string::npos; +} + +bool LogContains(const std::string& log, std::string_view needle) { + return log.find(needle) != std::string::npos; +} + +void ExpectExitedWith(const std::optional& status, int code) { + ASSERT_TRUE(status.has_value()) << "child did not exit in time"; + ASSERT_TRUE(WIFEXITED(*status)) + << "child was terminated by signal " + << (WIFSIGNALED(*status) ? WTERMSIG(*status) : 0); + EXPECT_EQ(WEXITSTATUS(*status), code); +} + +int64_t MillisSince(std::chrono::steady_clock::time_point start) { + return std::chrono::duration_cast( + std::chrono::steady_clock::now() - start) + .count(); +} + +std::optional FindDurableSegmentMount( + const std::string& cluster_id, const std::string& segment_name) { + auto backend = + CreateHaKvBackend(ha::HABackendSpec{.type = ha::HABackendType::ETCD, + .connstring = FLAGS_etcd_endpoints, + .cluster_namespace = cluster_id}); + if (!backend) { + ADD_FAILURE() << "CreateHaKvBackend failed: " + << toString(backend.error()); + return std::nullopt; + } + OpLogBatchStorage storage(cluster_id, *backend.value()); + DurablePrefix prefix; + if (const auto err = storage.ReadDurablePrefix(prefix); + err != ErrorCode::OK) { + ADD_FAILURE() << "ReadDurablePrefix failed: " << toString(err); + return std::nullopt; + } + for (uint64_t batch_id = 1; batch_id <= prefix.batch_id; ++batch_id) { + OpLogBatchRecord batch; + if (const auto err = storage.ReadBatch(batch_id, batch); + err != ErrorCode::OK) { + ADD_FAILURE() << "ReadBatch(" << batch_id + << ") failed: " << toString(err); + return std::nullopt; + } + for (const auto& entry : batch.entries) { + if (entry.op_type != OpType::SEGMENT_MOUNT) { + continue; + } + SegmentMountOp op; + if (struct_pack::deserialize_to(op, entry.payload) != + struct_pack::errc::ok || + op.segment_name != segment_name) { + continue; + } + return op; + } + } + return std::nullopt; +} + +constexpr size_t kTestSegmentSize = 16 * 1024 * 1024; + +tl::expected MountTestSegment(MasterClient& client, + const std::string& name) { + Segment segment; + segment.id = generate_uuid(); + segment.name = name; + segment.base = 0x300000000; + segment.size = kTestSegmentSize; + segment.te_endpoint = "127.0.0.1:1"; + return client.MountSegment(segment); +} + +// Reads the lease ID that the child publishes in the warming value of +// master_view, which is the decimal owner token. +std::optional WaitForWarmingLeaseId( + SupervisorChildProcess& child, const std::string& key, + std::chrono::milliseconds timeout) { + const auto deadline = std::chrono::steady_clock::now() + timeout; + while (std::chrono::steady_clock::now() < deadline && child.IsRunning()) { + const auto value = ReadEtcdKey(key); + if (value.has_value() && value->starts_with(kWarmingPrefix)) { + try { + return std::stoll(value->substr(kWarmingPrefix.size())); + } catch (const std::exception&) { + ADD_FAILURE() << "unparsable warming value: " << *value; + return std::nullopt; + } + } + std::this_thread::sleep_for(20ms); + } + return std::nullopt; +} + +struct PortPair { + int rpc; + int admin; +}; + +std::vector FreePortPairs(size_t count) { + const auto ports = getFreeTcpPorts(static_cast(count * 2)); + std::vector pairs; + for (size_t i = 0; i + 1 < ports.size(); i += 2) { + pairs.push_back({ports[i], ports[i + 1]}); + } + return pairs; +} + +std::string ClusterIdFor(const std::string& name) { + return name + "-" + std::to_string(getpid()); +} + +} // namespace + +TEST_F(HighAvailabilityTest, ServingLeaderSigtermDrainsAndKeepsLease) { + if (auto skip_reason = GetEtcdSkipReason(); skip_reason.has_value()) { + GTEST_SKIP() << *skip_reason; + } + + const auto ports = FreePortPairs(1); + ASSERT_EQ(ports.size(), 1u); + const std::string cluster_id = ClusterIdFor("sigterm-serving"); + const std::string log_path = TempPath("sigterm-serving", ".log"); + SupervisorChildProcess child(ports[0].rpc, ports[0].admin, cluster_id, + /*enable_oplog=*/false, log_path); + ASSERT_TRUE(child.Start()); + ASSERT_TRUE(WaitForServing(child, ports[0].admin, 20s)); + const std::string key = MasterViewKey(cluster_id); + ASSERT_TRUE(ReadEtcdKey(key).has_value()); + + const auto sent = std::chrono::steady_clock::now(); + ASSERT_TRUE(child.Signal(SIGTERM)); + const auto status = child.WaitForExit(10s); + LOG(INFO) << "SIGTERM to exit: " << MillisSince(sent) << " ms"; + ExpectExitedWith(status, 0); + + EXPECT_TRUE(ReadEtcdKey(key).has_value()) + << "no deliberate release: the key is left to TTL expiry"; + const std::string log = ReadTextFile(log_path); + EXPECT_TRUE(LogContains(log, "reason=shutdown requested")) << log; + EXPECT_TRUE(LogHasPhase(log, "serving")) << log; + EXPECT_TRUE(LogContains(log, "no release attempted")) << log; + EXPECT_TRUE(WaitForEtcdKeyGone(key, 8s)) << "the lease must expire by TTL"; +} + +TEST_F(HighAvailabilityTest, StandbySigtermLeavesLeaderUntouched) { + if (auto skip_reason = GetEtcdSkipReason(); skip_reason.has_value()) { + GTEST_SKIP() << *skip_reason; + } + + const auto ports = FreePortPairs(2); + ASSERT_EQ(ports.size(), 2u); + const std::string cluster_id = ClusterIdFor("sigterm-standby"); + const std::string log_path_a = TempPath("sigterm-standby-a", ".log"); + const std::string log_path_b = TempPath("sigterm-standby-b", ".log"); + SupervisorChildProcess leader(ports[0].rpc, ports[0].admin, cluster_id, + /*enable_oplog=*/false, log_path_a); + ASSERT_TRUE(leader.Start()); + ASSERT_TRUE(WaitForServing(leader, ports[0].admin, 20s)); + SupervisorChildProcess standby(ports[1].rpc, ports[1].admin, cluster_id, + /*enable_oplog=*/false, log_path_b); + ASSERT_TRUE(standby.Start()); + ASSERT_TRUE(WaitForHealthSubstring( + standby, ports[1].admin, + "\"leader_address\":\"127.0.0.1:" + std::to_string(ports[0].rpc) + "\"", + 20s)); + + const std::string key = MasterViewKey(cluster_id); + EtcdRevisionId revision_before = 0; + const auto value_before = ReadEtcdKey(key, &revision_before); + ASSERT_TRUE(value_before.has_value()); + + const auto sent = std::chrono::steady_clock::now(); + ASSERT_TRUE(standby.Signal(SIGTERM)); + const auto status = standby.WaitForExit(10s); + LOG(INFO) << "standby SIGTERM to exit: " << MillisSince(sent) << " ms"; + ExpectExitedWith(status, 0); + + EtcdRevisionId revision_after = 0; + const auto value_after = ReadEtcdKey(key, &revision_after); + EXPECT_EQ(value_after, value_before); + EXPECT_EQ(revision_after, revision_before); + EXPECT_TRUE(WaitForServing(leader, ports[0].admin, 2s)); + EXPECT_TRUE(LogContains(ReadTextFile(log_path_b), "no lease held")); +} + +TEST_F(HighAvailabilityTest, WarmupSigtermKeepsLease) { + if (auto skip_reason = GetEtcdSkipReason(); skip_reason.has_value()) { + GTEST_SKIP() << *skip_reason; + } + + const auto ports = FreePortPairs(1); + ASSERT_EQ(ports.size(), 1u); + const std::string cluster_id = ClusterIdFor("sigterm-warmup"); + const std::string log_path = TempPath("sigterm-warmup", ".log"); + SupervisorChildProcess child(ports[0].rpc, ports[0].admin, cluster_id, + /*enable_oplog=*/false, log_path); + ASSERT_TRUE(child.Start()); + // The initial warmup lasts the lease TTL, so the signal lands inside it. + ASSERT_TRUE(WaitForLogLine(log_path, "Entering warmup phase...", 20s)); + + const auto sent = std::chrono::steady_clock::now(); + ASSERT_TRUE(child.Signal(SIGTERM)); + const auto status = child.WaitForExit(10s); + LOG(INFO) << "warmup SIGTERM to exit: " << MillisSince(sent) << " ms"; + ExpectExitedWith(status, 0); + + const auto value = ReadEtcdKey(MasterViewKey(cluster_id)); + ASSERT_TRUE(value.has_value()); + EXPECT_TRUE(value->starts_with(kWarmingPrefix)) << *value; + const std::string log = ReadTextFile(log_path); + EXPECT_TRUE(LogContains(log, "no release attempted")) << log; + EXPECT_TRUE(LogHasPhase(log, "warmup")) << log; + EXPECT_FALSE(LogContains(log, "Master runtime state -> recovering")) + << "the request must be seen in the initial warmup, not after restore"; +} + +TEST_F(HighAvailabilityTest, OplogLeaderCommitsQueuedRecordsBeforeExit) { + if (auto skip_reason = GetEtcdSkipReason(); skip_reason.has_value()) { + GTEST_SKIP() << *skip_reason; + } + + ScopedFailPointDir failpoints; + const auto ports = FreePortPairs(2); + ASSERT_EQ(ports.size(), 2u); + const std::string cluster_id = ClusterIdFor("sigterm-oplog"); + const std::string log_path = TempPath("sigterm-oplog", ".log"); + const std::string segment_name = + "ha-shutdown-d-" + std::to_string(getpid()); + constexpr int64_t kLiveTtlSec = 120; + constexpr int64_t kSuspicionTtlSec = 240; + { + SupervisorChildProcess child(ports[0].rpc, ports[0].admin, cluster_id, + /*enable_oplog=*/true, log_path, "", + kLiveTtlSec, kSuspicionTtlSec); + ASSERT_TRUE(child.Start()); + ASSERT_TRUE(WaitForServing(child, ports[0].admin, 30s)); + if (kFailPointsEnabled) { + failpoints.Arm("batch_before_txn"); + } + + MasterClient client(generate_uuid()); + ASSERT_EQ(client.Connect("127.0.0.1:" + std::to_string(ports[0].rpc)), + ErrorCode::OK); + auto mounted = MountTestSegment(client, segment_name); + ASSERT_TRUE(mounted.has_value()) << toString(mounted.error()); + + const auto sent = std::chrono::steady_clock::now(); + if (kFailPointsEnabled) { + // Hold the writer before its etcd transaction so the entry is + // queued but not durable while the drain starts. + ASSERT_TRUE(failpoints.WaitForHit("batch_before_txn", 5s)); + ASSERT_TRUE(child.Signal(SIGTERM)); + ASSERT_TRUE( + WaitForLogLine(log_path, "reason=shutdown requested", 5s)); + std::this_thread::sleep_for(1s); + EXPECT_TRUE(child.IsRunning()) + << "the drain must wait for the writer"; + EXPECT_FALSE( + LogContains(ReadTextFile(log_path), "Shutdown complete")); + failpoints.Release("batch_before_txn"); + } else { + ASSERT_TRUE(child.Signal(SIGTERM)); + } + const auto status = child.WaitForExit(10s); + LOG(INFO) << "OpLog SIGTERM to exit: " << MillisSince(sent) << " ms"; + ExpectExitedWith(status, 0); + } + + EXPECT_TRUE(ReadEtcdKey(MasterViewKey(cluster_id)).has_value()); + const auto durable = FindDurableSegmentMount(cluster_id, segment_name); + ASSERT_TRUE(durable.has_value()) + << "the mounted segment is missing from the durable OpLog"; + EXPECT_EQ(durable->capacity, kTestSegmentSize); + EXPECT_EQ(durable->transport_endpoint, "127.0.0.1:1"); + + // A successor waits out the lease TTL and replays the log. A restored + // segment is a placeholder until its client remounts, so the admin + // segment list cannot show it; the restore log line counts it instead. + const std::string successor_log = + TempPath("sigterm-oplog-successor", ".log"); + SupervisorChildProcess successor(ports[1].rpc, ports[1].admin, cluster_id, + /*enable_oplog=*/true, successor_log, "", + kLiveTtlSec, kSuspicionTtlSec); + ASSERT_TRUE(successor.Start()); + ASSERT_TRUE(WaitForServing(successor, ports[1].admin, 40s)); + const std::string successor_text = ReadTextFile(successor_log); + EXPECT_TRUE(LogContains(successor_text, + "Restored from standby: 0 objects, " + "1 segments")) + << successor_text; +} + +// Revokes the leader's lease and signals it back to back, in either order. +void RunSigtermRacingLeaseLoss(bool revoke_first) { + const auto ports = FreePortPairs(1); + ASSERT_EQ(ports.size(), 1u); + const std::string name = + revoke_first ? "sigterm-revoke-first" : "sigterm-signal-first"; + const std::string cluster_id = ClusterIdFor(name); + const std::string log_path = TempPath(name, ".log"); + SupervisorChildProcess child(ports[0].rpc, ports[0].admin, cluster_id, + /*enable_oplog=*/false, log_path); + ASSERT_TRUE(child.Start()); + const std::string key = MasterViewKey(cluster_id); + const auto lease_id = WaitForWarmingLeaseId(child, key, 20s); + ASSERT_TRUE(lease_id.has_value()); + ASSERT_TRUE(WaitForServing(child, ports[0].admin, 20s)); + + if (revoke_first) { + ASSERT_EQ(EtcdHelper::RevokeLease(*lease_id), ErrorCode::OK); + ASSERT_TRUE(child.Signal(SIGTERM)); + } else { + ASSERT_TRUE(child.Signal(SIGTERM)); + ASSERT_EQ(EtcdHelper::RevokeLease(*lease_id), ErrorCode::OK); + } + ExpectExitedWith(child.WaitForExit(10s), 0); + + EXPECT_TRUE(LogContains(ReadTextFile(log_path), "Shutdown complete")); + EXPECT_FALSE(ReadEtcdKey(key).has_value()) << "the lease was revoked"; +} + +TEST_F(HighAvailabilityTest, SigtermRacingLeaseLossExitsZeroRevokeFirst) { + if (auto skip_reason = GetEtcdSkipReason(); skip_reason.has_value()) { + GTEST_SKIP() << *skip_reason; + } + RunSigtermRacingLeaseLoss(/*revoke_first=*/true); +} + +TEST_F(HighAvailabilityTest, SigtermRacingLeaseLossExitsZeroSignalFirst) { + if (auto skip_reason = GetEtcdSkipReason(); skip_reason.has_value()) { + GTEST_SKIP() << *skip_reason; + } + RunSigtermRacingLeaseLoss(/*revoke_first=*/false); +} + +TEST_F(HighAvailabilityTest, SigtermBeforeServeStartStopsWriter) { + if (auto skip_reason = GetEtcdSkipReason(); skip_reason.has_value()) { + GTEST_SKIP() << *skip_reason; + } + if (!kFailPointsEnabled) { + GTEST_SKIP() << "needs MOONCAKE_ENABLE_TEST_FAILPOINTS"; + } + + ScopedFailPointDir failpoints; + failpoints.Arm("supervisor_before_serve_start"); + const auto ports = FreePortPairs(1); + ASSERT_EQ(ports.size(), 1u); + const std::string cluster_id = ClusterIdFor("sigterm-before-serve"); + const std::string log_path = TempPath("sigterm-before-serve", ".log"); + SupervisorChildProcess child(ports[0].rpc, ports[0].admin, cluster_id, + /*enable_oplog=*/true, log_path); + ASSERT_TRUE(child.Start()); + ASSERT_TRUE(failpoints.WaitForHit("supervisor_before_serve_start", 30s)); + + const auto sent = std::chrono::steady_clock::now(); + ASSERT_TRUE(child.Signal(SIGTERM)); + ASSERT_TRUE(WaitForLogLine(log_path, "reason=shutdown requested", 5s)); + EXPECT_TRUE(WaitForHealthSubstring(child, ports[0].admin, + "\"service_ready\":false", 5s)); + failpoints.Release("supervisor_before_serve_start"); + ExpectExitedWith(child.WaitForExit(10s), 0); + LOG(INFO) << "pre-listener SIGTERM to exit: " << MillisSince(sent) << " ms"; + + const std::string log = ReadTextFile(log_path); + EXPECT_TRUE(LogHasPhase(log, "serve_start")) << log; + const auto stopped = log.find( + "Shutdown stopped the term's workers, monitor and OpLog " + "writer"); + ASSERT_NE(stopped, std::string::npos) << log; + EXPECT_LT(stopped, log.find("Shutdown complete")) << log; + const auto value = ReadEtcdKey(MasterViewKey(cluster_id)); + ASSERT_TRUE(value.has_value()); + EXPECT_TRUE(value->starts_with(kWarmingPrefix)) << *value; +} + +TEST_F(HighAvailabilityTest, SigtermBeforeServeActivationNeverServes) { + if (auto skip_reason = GetEtcdSkipReason(); skip_reason.has_value()) { + GTEST_SKIP() << *skip_reason; + } + if (!kFailPointsEnabled) { + GTEST_SKIP() << "needs MOONCAKE_ENABLE_TEST_FAILPOINTS"; + } + + ScopedFailPointDir failpoints; + failpoints.Arm("supervisor_before_serve_activation"); + const auto ports = FreePortPairs(1); + ASSERT_EQ(ports.size(), 1u); + const std::string cluster_id = ClusterIdFor("sigterm-before-activation"); + const std::string log_path = TempPath("sigterm-before-activation", ".log"); + SupervisorChildProcess child(ports[0].rpc, ports[0].admin, cluster_id, + /*enable_oplog=*/false, log_path); + ASSERT_TRUE(child.Start()); + ASSERT_TRUE( + failpoints.WaitForHit("supervisor_before_serve_activation", 30s)); + + ASSERT_TRUE(child.Signal(SIGTERM)); + ASSERT_TRUE(WaitForLogLine(log_path, "reason=shutdown requested", 5s)); + failpoints.Release("supervisor_before_serve_activation"); + ExpectExitedWith(child.WaitForExit(10s), 0); + + const std::string log = ReadTextFile(log_path); + EXPECT_FALSE(LogContains(log, "Master runtime state -> serving")) << log; + EXPECT_TRUE(LogHasPhase(log, "serving")) << log; + const auto value = ReadEtcdKey(MasterViewKey(cluster_id)); + ASSERT_TRUE(value.has_value()); + EXPECT_EQ(*value, "127.0.0.1:" + std::to_string(ports[0].rpc)) + << "published, then left to TTL"; +} + +TEST_F(HighAvailabilityTest, InjectedPredicateDrivesServingShutdown) { + if (auto skip_reason = GetEtcdSkipReason(); skip_reason.has_value()) { + GTEST_SKIP() << *skip_reason; + } + + const auto ports = FreePortPairs(1); + ASSERT_EQ(ports.size(), 1u); + const std::string cluster_id = ClusterIdFor("injected-predicate"); + const std::string log_path = TempPath("injected-predicate", ".log"); + const std::string shutdown_file = + TempPath("injected-predicate", ".shutdown"); + std::filesystem::remove(shutdown_file); + SupervisorChildProcess child(ports[0].rpc, ports[0].admin, cluster_id, + /*enable_oplog=*/false, log_path, + shutdown_file); + ASSERT_TRUE(child.Start()); + ASSERT_TRUE(WaitForServing(child, ports[0].admin, 20s)); + + const auto sent = std::chrono::steady_clock::now(); + std::ofstream(shutdown_file).put('\n'); + const auto status = child.WaitForExit(10s); + LOG(INFO) << "injected request to exit: " << MillisSince(sent) << " ms"; + std::filesystem::remove(shutdown_file); + ExpectExitedWith(status, 0); + + const std::string log = ReadTextFile(log_path); + EXPECT_TRUE(LogContains(log, "reason=shutdown requested")) << log; + EXPECT_TRUE(LogHasPhase(log, "serving")) << log; + EXPECT_TRUE(ReadEtcdKey(MasterViewKey(cluster_id)).has_value()); +} + +TEST_F(HighAvailabilityTest, SecondSigtermDuringDrainExits143) { + if (auto skip_reason = GetEtcdSkipReason(); skip_reason.has_value()) { + GTEST_SKIP() << *skip_reason; + } + if (!kFailPointsEnabled) { + GTEST_SKIP() << "needs MOONCAKE_ENABLE_TEST_FAILPOINTS"; + } + + ScopedFailPointDir failpoints; + failpoints.Arm("supervisor_before_serve_start"); + const auto ports = FreePortPairs(1); + ASSERT_EQ(ports.size(), 1u); + const std::string cluster_id = ClusterIdFor("second-sigterm"); + const std::string log_path = TempPath("second-sigterm", ".log"); + SupervisorChildProcess child(ports[0].rpc, ports[0].admin, cluster_id, + /*enable_oplog=*/true, log_path); + ASSERT_TRUE(child.Start()); + ASSERT_TRUE(failpoints.WaitForHit("supervisor_before_serve_start", 30s)); + ASSERT_TRUE(child.Signal(SIGTERM)); + ASSERT_TRUE(WaitForLogLine(log_path, "reason=shutdown requested", 5s)); + + const auto sent = std::chrono::steady_clock::now(); + ASSERT_TRUE(child.Signal(SIGTERM)); + ExpectExitedWith(child.WaitForExit(2s), 143); + LOG(INFO) << "second SIGTERM to exit: " << MillisSince(sent) << " ms"; + EXPECT_TRUE(ReadEtcdKey(MasterViewKey(cluster_id)).has_value()); +} + +TEST_F(HighAvailabilityTest, SigtermWaitsForAdmittedHandler) { + if (auto skip_reason = GetEtcdSkipReason(); skip_reason.has_value()) { + GTEST_SKIP() << *skip_reason; + } + if (!kFailPointsEnabled) { + GTEST_SKIP() << "needs MOONCAKE_ENABLE_TEST_FAILPOINTS"; + } + + ScopedFailPointDir failpoints; + const auto ports = FreePortPairs(1); + ASSERT_EQ(ports.size(), 1u); + const std::string cluster_id = ClusterIdFor("sigterm-admitted"); + const std::string log_path = TempPath("sigterm-admitted", ".log"); + const std::string segment_name = + "ha-shutdown-k-" + std::to_string(getpid()); + SupervisorChildProcess child(ports[0].rpc, ports[0].admin, cluster_id, + /*enable_oplog=*/true, log_path); + ASSERT_TRUE(child.Start()); + ASSERT_TRUE(WaitForServing(child, ports[0].admin, 30s)); + + failpoints.Arm("rpc_mount_segment_before_service"); + MasterClient client(generate_uuid()); + ASSERT_EQ(client.Connect("127.0.0.1:" + std::to_string(ports[0].rpc)), + ErrorCode::OK); + // The response may be lost when the server closes the socket, so the + // result is ignored. + std::jthread mount([&] { (void)MountTestSegment(client, segment_name); }); + ASSERT_TRUE(failpoints.WaitForHit("rpc_mount_segment_before_service", 5s)) + << "no handler was admitted"; + + ASSERT_TRUE(child.Signal(SIGTERM)); + ASSERT_TRUE(WaitForLogLine(log_path, "reason=shutdown requested", 5s)); + std::this_thread::sleep_for(1s); + EXPECT_TRUE(child.IsRunning()) << "the drain must wait for the handler"; + EXPECT_FALSE(LogContains(ReadTextFile(log_path), "finishing from phase=")); + + const auto released = std::chrono::steady_clock::now(); + failpoints.Release("rpc_mount_segment_before_service"); + const auto status = child.WaitForExit(10s); + LOG(INFO) << "release to exit: " << MillisSince(released) << " ms"; + mount.join(); + ExpectExitedWith(status, 0); + + EXPECT_TRUE(LogHasPhase(ReadTextFile(log_path), "serving")); + EXPECT_TRUE(FindDurableSegmentMount(cluster_id, segment_name).has_value()) + << "the admitted handler's entry must be durable before exit"; +} + +TEST_F(HighAvailabilityTest, InjectedRequestBlockedInDrainEndsByWatchdog) { + if (auto skip_reason = GetEtcdSkipReason(); skip_reason.has_value()) { + GTEST_SKIP() << *skip_reason; + } + if (!kFailPointsEnabled) { + GTEST_SKIP() << "needs MOONCAKE_ENABLE_TEST_FAILPOINTS"; + } + + ScopedFailPointDir failpoints; + const auto ports = FreePortPairs(1); + ASSERT_EQ(ports.size(), 1u); + const std::string cluster_id = ClusterIdFor("blocked-drain"); + const std::string log_path = TempPath("blocked-drain", ".log"); + const std::string shutdown_file = TempPath("blocked-drain", ".shutdown"); + std::filesystem::remove(shutdown_file); + SupervisorChildProcess child(ports[0].rpc, ports[0].admin, cluster_id, + /*enable_oplog=*/false, log_path, + shutdown_file); + ASSERT_TRUE(child.Start()); + ASSERT_TRUE(WaitForServing(child, ports[0].admin, 20s)); + + failpoints.Arm("rpc_mount_segment_before_service"); + MasterClient client(generate_uuid()); + ASSERT_EQ(client.Connect("127.0.0.1:" + std::to_string(ports[0].rpc)), + ErrorCode::OK); + std::jthread mount([&] { + (void)MountTestSegment(client, + "ha-shutdown-l-" + std::to_string(getpid())); + }); + ASSERT_TRUE(failpoints.WaitForHit("rpc_mount_segment_before_service", 5s)); + + // No signal: only the latched request starts the watchdog deadline. The + // handler is never released, so the drain cannot finish by itself. + const auto requested = std::chrono::steady_clock::now(); + std::ofstream(shutdown_file).put('\n'); + const auto status = child.WaitForExit(25s); + const auto elapsed_ms = MillisSince(requested); + LOG(INFO) << "injected request to watchdog exit: " << elapsed_ms << " ms"; + mount.join(); + std::filesystem::remove(shutdown_file); + ExpectExitedWith(status, 1); + EXPECT_GE(elapsed_ms, 14000); + EXPECT_LE(elapsed_ms, 20000); + + const std::string log = ReadTextFile(log_path); + EXPECT_TRUE(LogContains(log, "reason=shutdown requested")) << log; + EXPECT_FALSE(LogContains(log, "Shutdown complete")) << log; +} + +TEST_F(HighAvailabilityTest, LateRequestAfterOrdinaryReleaseKeepsResult) { + if (auto skip_reason = GetEtcdSkipReason(); skip_reason.has_value()) { + GTEST_SKIP() << *skip_reason; + } + if (!kFailPointsEnabled) { + GTEST_SKIP() << "needs MOONCAKE_ENABLE_TEST_FAILPOINTS"; + } + + ScopedFailPointDir failpoints; + const auto ports = FreePortPairs(1); + ASSERT_EQ(ports.size(), 1u); + const std::string cluster_id = ClusterIdFor("late-request"); + const std::string log_path = TempPath("late-request", ".log"); + const std::string shutdown_file = TempPath("late-request", ".shutdown"); + std::filesystem::remove(shutdown_file); + SupervisorChildProcess child(ports[0].rpc, ports[0].admin, cluster_id, + /*enable_oplog=*/false, log_path, + shutdown_file); + ASSERT_TRUE(child.Start()); + const std::string key = MasterViewKey(cluster_id); + const auto lease_id = WaitForWarmingLeaseId(child, key, 20s); + ASSERT_TRUE(lease_id.has_value()); + ASSERT_TRUE(WaitForServing(child, ports[0].admin, 20s)); + + // Losing the lease stops the listener, and the serving exit releases + // through the supervisor's ordinary path; hold it right after that release. + failpoints.Arm("supervisor_after_term_release"); + ASSERT_EQ(EtcdHelper::RevokeLease(*lease_id), ErrorCode::OK); + ASSERT_TRUE(failpoints.WaitForHit("supervisor_after_term_release", 10s)); + std::ofstream(shutdown_file).put('\n'); + failpoints.Release("supervisor_after_term_release"); + ExpectExitedWith(child.WaitForExit(10s), 0); + std::filesystem::remove(shutdown_file); + + const std::string log = ReadTextFile(log_path); + EXPECT_TRUE(LogHasPhase(log, "serving")) << log; + EXPECT_TRUE( + LogContains(log, "Shutdown complete after leadership release, result=")) + << log; + EXPECT_FALSE(LogContains(log, "no lease held")) << log; + EXPECT_FALSE(LogContains(log, "no release attempted")) << log; + EXPECT_FALSE(LogContains(log, "Release leadership: ")) + << "the exit must come from the check right after the release"; +} + TEST_F(HighAvailabilityTest, EtcdBasicOperations) { if (auto skip_reason = GetEtcdSkipReason(); skip_reason.has_value()) { GTEST_SKIP() << *skip_reason; diff --git a/mooncake-store/tests/shutdown_request_test.cpp b/mooncake-store/tests/shutdown_request_test.cpp new file mode 100644 index 0000000000..a9a0a9ae19 --- /dev/null +++ b/mooncake-store/tests/shutdown_request_test.cpp @@ -0,0 +1,417 @@ +#include "shutdown_request.h" + +#include +#include +#include +#include +#include +#include +#include + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +namespace mooncake { +namespace { + +using namespace std::chrono_literals; +using Clock = std::chrono::steady_clock; + +// Every case that installs handlers, latches or starts the watchdog runs in a +// forked child, because the flags and the dispositions are process-wide. The +// gtest parent never installs handlers, never latches and never starts the +// watchdog; the in-process cases start no threads, so the parent stays +// single-threaded and a child may run arbitrary code after fork. + +void SignalReady(int ready_fd) { + const char byte = 'r'; + ssize_t written; + do { + written = ::write(ready_fd, &byte, 1); + } while (written < 0 && errno == EINTR); +} + +// Forks; the child runs body(ready_fd) with stderr on a pipe and calls +// _exit(0) if body returns. The parent reads stderr only after the child +// exits, so a child that fills the pipe blocks, as case 8 needs. +class Child { + public: + explicit Child(const std::function& body) { + int ready[2]; + int err[2]; + if (::pipe2(ready, O_CLOEXEC) != 0 || ::pipe2(err, O_CLOEXEC) != 0) { + ADD_FAILURE() << "pipe2: " << std::strerror(errno); + return; + } + pid_ = ::fork(); + if (pid_ == 0) { + ::close(ready[0]); + ::close(err[0]); + ::dup2(err[1], STDERR_FILENO); + body(ready[1]); + _exit(0); + } + ready_read_ = ready[0]; + err_read_ = err[0]; + ::close(ready[1]); + ::close(err[1]); + if (pid_ < 0) { + ADD_FAILURE() << "fork: " << std::strerror(errno); + return; + } + ::fcntl(err_read_, F_SETFL, O_NONBLOCK); + } + + ~Child() { + if (pid_ > 0 && IsRunning()) { + ::kill(pid_, SIGKILL); + ::waitpid(pid_, &status_, 0); + } + if (ready_read_ >= 0) ::close(ready_read_); + if (err_read_ >= 0) ::close(err_read_); + } + + Child(const Child&) = delete; + Child& operator=(const Child&) = delete; + + pid_t pid() const { return pid_; } + + // Waits for one byte on the ready pipe. + bool WaitReady(std::chrono::milliseconds timeout) { + struct pollfd pfd{ready_read_, POLLIN, 0}; + if (::poll(&pfd, 1, static_cast(timeout.count())) <= 0) { + return false; + } + char byte; + return ::read(ready_read_, &byte, 1) == 1; + } + + bool IsRunning() { + if (reaped_) return false; + const pid_t result = ::waitpid(pid_, &status_, WNOHANG); + if (result == pid_) { + reaped_ = true; + return false; + } + return true; + } + + // Raw wait status, or nullopt on timeout (then SIGKILL + reap). + std::optional Wait(std::chrono::milliseconds timeout) { + const auto deadline = Clock::now() + timeout; + while (IsRunning()) { + if (Clock::now() >= deadline) { + ::kill(pid_, SIGKILL); + ::waitpid(pid_, &status_, 0); + reaped_ = true; + return std::nullopt; + } + std::this_thread::sleep_for(5ms); + } + return status_; + } + + // Non-blocking read of what the child wrote to stderr. + std::string ReadStderr() { + std::string text; + char buffer[4096]; + ssize_t n; + while ((n = ::read(err_read_, buffer, sizeof(buffer))) > 0) { + text.append(buffer, static_cast(n)); + } + return text; + } + + private: + pid_t pid_ = -1; + int ready_read_ = -1; + int err_read_ = -1; + int status_ = 0; + bool reaped_ = false; +}; + +std::chrono::milliseconds Since(Clock::time_point start) { + return std::chrono::duration_cast(Clock::now() - + start); +} + +void ExpectExitedWith(const std::optional& status, int code) { + ASSERT_TRUE(status.has_value()) << "child did not exit in time"; + ASSERT_TRUE(WIFEXITED(*status)) + << "child was terminated by signal " + << (WIFSIGNALED(*status) ? WTERMSIG(*status) : 0); + EXPECT_EQ(WEXITSTATUS(*status), code); +} + +// Blocks inside the sink to hold glog's mutex. glog calls sinks with its log +// mutex held, so FlushLogFiles on another thread blocks on that mutex. +#pragma GCC diagnostic push +#pragma GCC diagnostic ignored "-Wdeprecated-declarations" +class BlockingSink : public google::LogSink { + public: + using google::LogSink::send; + void send(google::LogSeverity, const char*, const char*, int, + const struct ::tm*, const char*, size_t) override { + entered.store(true, std::memory_order_release); + for (;;) { + pause(); + } + } + std::atomic entered{false}; +}; +#pragma GCC diagnostic pop + +// Parks a thread inside BlockingSink::send, or ends the child with status 3. +void BlockLogMutex(BlockingSink& sink) { + google::AddLogSink(&sink); + std::thread([] { LOG(INFO) << "block"; }).detach(); + const auto give_up = Clock::now() + 2s; + while (!sink.entered.load(std::memory_order_acquire)) { + if (Clock::now() > give_up) { + _exit(3); + } + std::this_thread::sleep_for(1ms); + } +} + +class FirstSignalTest : public ::testing::TestWithParam {}; + +TEST_P(FirstSignalTest, FirstSignalSetsFlagAndKeepsRunning) { + Child child([](int ready_fd) { + if (!InstallShutdownSignalHandlers()) _exit(5); + SignalReady(ready_fd); + const auto give_up = Clock::now() + 5s; + while (!ShutdownRequested()) { + if (Clock::now() > give_up) _exit(2); + std::this_thread::sleep_for(1ms); + } + _exit(ShutdownLatched() ? 4 : 0); + }); + ASSERT_TRUE(child.WaitReady(5s)); + ASSERT_EQ(::kill(child.pid(), GetParam()), 0); + ExpectExitedWith(child.Wait(7s), 0); + EXPECT_EQ(child.ReadStderr(), "") << "the handler must not do I/O"; +} + +INSTANTIATE_TEST_SUITE_P(Signals, FirstSignalTest, + ::testing::Values(SIGTERM, SIGINT)); + +struct SignalPair { + int first; + int second; + int expected_status; +}; + +class SecondSignalTest : public ::testing::TestWithParam {}; + +TEST_P(SecondSignalTest, SecondSignalExits128PlusSigno) { + const SignalPair pair = GetParam(); + Child child([](int ready_fd) { + if (!InstallShutdownSignalHandlers()) _exit(5); + SignalReady(ready_fd); + bool reported = false; + const auto give_up = Clock::now() + 10s; + while (Clock::now() < give_up) { + if (!reported && ShutdownRequested()) { + reported = true; + SignalReady(ready_fd); + } + std::this_thread::sleep_for(1ms); + } + _exit(2); + }); + ASSERT_TRUE(child.WaitReady(5s)) << "child never became ready"; + ASSERT_EQ(::kill(child.pid(), pair.first), 0); + ASSERT_TRUE(child.WaitReady(5s)) << "first signal was not handled"; + ASSERT_EQ(::kill(child.pid(), pair.second), 0); + ExpectExitedWith(child.Wait(2s), pair.expected_status); + EXPECT_EQ(child.ReadStderr(), ""); +} + +INSTANTIATE_TEST_SUITE_P(Pairs, SecondSignalTest, + ::testing::Values(SignalPair{SIGTERM, SIGTERM, 143}, + SignalPair{SIGTERM, SIGINT, 130}, + SignalPair{SIGINT, SIGINT, 130})); + +struct RecordedCall { + int signo; + int flags; + void (*handler)(int); +}; +std::vector& Calls() { + static std::vector calls; + return calls; +} + +int RecordingSigaction(int signo, const struct sigaction* act, + struct sigaction*) { + Calls().push_back({signo, act->sa_flags, act->sa_handler}); + return 0; +} + +int FailingSigaction(int signo, const struct sigaction* act, + struct sigaction*) { + Calls().push_back({signo, act->sa_flags, act->sa_handler}); + errno = EINVAL; + return -1; +} + +TEST(ShutdownRequestTest, InstallUsesSaOnstackForBothSignals) { + Calls().clear(); + ASSERT_TRUE(detail::InstallShutdownSignalHandlersWith(&RecordingSigaction)); + ASSERT_EQ(Calls().size(), 2u); + EXPECT_EQ(Calls()[0].signo, SIGINT); + EXPECT_EQ(Calls()[1].signo, SIGTERM); + for (const auto& call : Calls()) { + EXPECT_TRUE(call.flags & SA_ONSTACK); + EXPECT_TRUE(call.flags & SA_RESTART); + EXPECT_FALSE(call.flags & SA_NODEFER); + EXPECT_NE(call.handler, SIG_DFL); + EXPECT_NE(call.handler, SIG_IGN); + } +} + +TEST(ShutdownRequestTest, InstallFailureReturnsFalse) { + Calls().clear(); + errno = 0; + EXPECT_FALSE(detail::InstallShutdownSignalHandlersWith(&FailingSigaction)); + EXPECT_EQ(errno, EINVAL); + EXPECT_EQ(Calls().size(), 1u); +} + +TEST(ShutdownRequestTest, RealInstallSetsSaOnstack) { + Child child([](int) { + if (!InstallShutdownSignalHandlers()) _exit(5); + struct sigaction old{}; + ::sigaction(SIGTERM, nullptr, &old); + _exit((old.sa_flags & SA_ONSTACK) ? 0 : 3); + }); + ExpectExitedWith(child.Wait(5s), 0); +} + +TEST(ShutdownRequestTest, WatchdogWithoutRequestNeverFires) { + Child child([](int) { + if (!StartShutdownWatchdog(200ms)) _exit(5); + std::this_thread::sleep_for(1s); + _exit(0); + }); + ExpectExitedWith(child.Wait(5s), 0); +} + +TEST(ShutdownRequestTest, WatchdogEndsBlockedCleanup) { + const auto start = Clock::now(); + Child child([](int) { + if (!InstallShutdownSignalHandlers()) _exit(5); + if (!StartShutdownWatchdog(500ms)) _exit(5); + ::raise(SIGTERM); + for (;;) pause(); + }); + ExpectExitedWith(child.Wait(5s), 1); + const auto elapsed = Since(start); + EXPECT_GE(elapsed.count(), 500); + EXPECT_LE(elapsed.count(), 2000); +} + +TEST(ShutdownRequestTest, WatchdogEndsCleanupBlockedOnFullStderr) { + const auto start = Clock::now(); + Child child([](int) { + if (!InstallShutdownSignalHandlers()) _exit(5); + if (!StartShutdownWatchdog(500ms)) _exit(5); + const int flags = ::fcntl(STDERR_FILENO, F_GETFL); + ::fcntl(STDERR_FILENO, F_SETFL, flags | O_NONBLOCK); + char chunk[4096]; + std::memset(chunk, 'x', sizeof(chunk)); + while (::write(STDERR_FILENO, chunk, sizeof(chunk)) > 0) { + } + ::fcntl(STDERR_FILENO, F_SETFL, flags); + ::raise(SIGTERM); + (void)!::write(STDERR_FILENO, "x", 1); // Blocks: the pipe is full. + _exit(0); + }); + ExpectExitedWith(child.Wait(5s), 1); + EXPECT_LE(Since(start).count(), 2000); +} + +TEST(ShutdownRequestTest, WatchdogEndsExitBlockedInLogFlush) { + const auto start = Clock::now(); + Child child([](int) { + if (!InstallShutdownSignalHandlers()) _exit(5); + if (!StartShutdownWatchdog(500ms)) _exit(5); + BlockingSink sink; + BlockLogMutex(sink); + ::raise(SIGTERM); + ExitAfterShutdown(0); + }); + auto status = child.Wait(5s); + ASSERT_TRUE(status.has_value()); + ASSERT_TRUE(WIFEXITED(*status)); + EXPECT_NE(WEXITSTATUS(*status), 3) << "sink never entered"; + EXPECT_NE(WEXITSTATUS(*status), 0) << "fixture did not block FlushLogFiles"; + EXPECT_EQ(WEXITSTATUS(*status), 1); + EXPECT_LE(Since(start).count(), 2000); +} + +TEST(ShutdownRequestTest, WatchdogBoundsExitWithoutSignal) { + const auto start = Clock::now(); + Child child([](int) { + if (!StartShutdownWatchdog(500ms)) _exit(5); + BlockingSink sink; + BlockLogMutex(sink); + ExitAfterShutdown(0); + }); + auto status = child.Wait(5s); + ASSERT_TRUE(status.has_value()); + ASSERT_TRUE(WIFEXITED(*status)); + EXPECT_NE(WEXITSTATUS(*status), 3) << "sink never entered"; + EXPECT_NE(WEXITSTATUS(*status), 0) << "fixture did not block FlushLogFiles"; + EXPECT_EQ(WEXITSTATUS(*status), 1); + EXPECT_LE(Since(start).count(), 2000); +} + +TEST(ShutdownRequestTest, LatchedRequestStartsDeadlineWithoutSignal) { + const auto start = Clock::now(); + Child child([](int) { + if (!StartShutdownWatchdog(500ms)) _exit(5); + LatchShutdownRequest(); + if (!ShutdownLatched() || ShutdownRequested()) _exit(4); + for (;;) pause(); + }); + ExpectExitedWith(child.Wait(5s), 1); + const auto elapsed = Since(start); + EXPECT_GE(elapsed.count(), 500); + EXPECT_LE(elapsed.count(), 2000); +} + +TEST(ShutdownRequestTest, LateBlockedFlushGetsNoNewDeadline) { + Child child([](int ready_fd) { + if (!InstallShutdownSignalHandlers()) _exit(5); + if (!StartShutdownWatchdog(1000ms)) _exit(5); + BlockingSink sink; + BlockLogMutex(sink); + SignalReady(ready_fd); + ::raise(SIGTERM); + std::this_thread::sleep_for(700ms); + ExitAfterShutdown(0); + }); + ASSERT_TRUE(child.WaitReady(5s)); + const auto ready_at = Clock::now(); + auto status = child.Wait(5s); + const auto elapsed = Since(ready_at); + ASSERT_TRUE(status.has_value()); + ASSERT_TRUE(WIFEXITED(*status)); + EXPECT_EQ(WEXITSTATUS(*status), 1); + // A deadline restarted at the flush would end at 1700 ms or later. + EXPECT_GE(elapsed.count(), 900); + EXPECT_LE(elapsed.count(), 1500); +} + +} // namespace +} // namespace mooncake