[Store] Add batch OpLog snapshot coordinator

mooncake-store/include/ha/snapshot/batch_oplog/batch_oplog_snapshot_coordinator.h

@@ -0,0 +1,126 @@
+#pragma once
+
+#include <chrono>
+#include <condition_variable>
+#include <cstddef>
+#include <cstdint>
+#include <functional>
+#include <memory>
+#include <mutex>
+#include <optional>
+#include <string>
+#include <thread>
+
+#include "ha/oplog/oplog_batch_types.h"
+#include "types.h"
+
+namespace mooncake {
+
+inline constexpr uint64_t kDefaultBatchOpLogSnapshotIntervalSeconds = 600;
+
+class HaKvBackend;
+class HotStandbyService;
+class SnapshotMaintenanceLease;
+class SnapshotObjectStore;
+
+struct BatchOpLogSnapshotCoordinatorConfig {
+    uint64_t snapshot_interval_seconds{
+        kDefaultBatchOpLogSnapshotIntervalSeconds};
+    size_t chunk_object_count{1000000};
+    std::string snapshot_root;
+    std::function<std::chrono::steady_clock::time_point()> clock;
+};
+
+struct BatchOpLogSnapshotCoordinatorStatus {
+    bool running{false};
+    bool attempt_in_flight{false};
+    bool promotion_requested{false};
+    uint64_t attempts{0};
+    ErrorCode last_error{ErrorCode::OK};
+    std::optional<DurablePrefix> catch_up_target;
+};
+
+// Coordinates the opt-in batch-OpLog snapshot path. Construction alone does
+// not start a worker or change HotStandbyService behavior.
+class BatchOpLogSnapshotCoordinator final {
+   public:
+    using LeaseFactory =
+        std::function<std::unique_ptr<SnapshotMaintenanceLease>()>;
+
+    BatchOpLogSnapshotCoordinator(HotStandbyService& standby,
+                                  HaKvBackend& backend,
+                                  SnapshotObjectStore& object_store,
+                                  std::string cluster_id,
+                                  BatchOpLogSnapshotCoordinatorConfig config,
+                                  LeaseFactory lease_factory = {});
+    BatchOpLogSnapshotCoordinator(HotStandbyService& standby,
+                                  HaKvBackend& backend,
+                                  SnapshotObjectStore& object_store,
+                                  std::string cluster_id,
+                                  std::string snapshot_root,
+                                  uint64_t snapshot_interval_seconds =
+                                      kDefaultBatchOpLogSnapshotIntervalSeconds,
+                                  size_t chunk_object_count = 1000000);
+    ~BatchOpLogSnapshotCoordinator();
+
+    BatchOpLogSnapshotCoordinator(const BatchOpLogSnapshotCoordinator&) =
+        delete;
+    BatchOpLogSnapshotCoordinator& operator=(
+        const BatchOpLogSnapshotCoordinator&) = delete;
+
+    // Starts periodic scheduling. RunOnce() remains available for
+    // deterministic tests and callers that own the scheduling loop.
+    void Start();
+    void Stop();
+
+    // Executes at most one attempt. An ineligible cycle returns OK and leaves
+    // the standby OpLog apply loop untouched.
+    ErrorCode RunOnce();
+    ErrorCode PollOnce() { return RunOnce(); }
+
+    // Called by HotStandbyService before promotion/stop. Promotion keeps a
+    // fully uploaded candidate eligible for the background publish step.
+    void NotifyPromotion();
+    void OnPromotion() { NotifyPromotion(); }
+
+    BatchOpLogSnapshotCoordinatorStatus GetStatus() const;
+    bool IsRunning() const;
+    bool IsAttemptInFlight() const;
+    ErrorCode last_error() const;
+
+   private:
+    using Clock = std::chrono::steady_clock;
+
+    void SchedulerLoop();
+    ErrorCode RunAttempt();
+    std::optional<uint64_t> ReadLatestBatchId(ErrorCode& error) const;
+    bool CatchUpComplete(const DurablePrefix& target) const;
+    void FinishAttempt(ErrorCode error, bool count_attempt);
+    Clock::time_point Now() const;
+    void OnCaptureReleased();
+    std::optional<DurablePrefix> ReadDurablePrefix() const;
+    void RequestStop();
+
+    HotStandbyService& standby_;
+    HaKvBackend& backend_;
+    SnapshotObjectStore& object_store_;
+    std::string cluster_id_;
+    BatchOpLogSnapshotCoordinatorConfig config_;
+    LeaseFactory lease_factory_;
+
+    mutable std::mutex mutex_;
+    std::condition_variable cv_;
+    std::thread worker_;
+    bool running_{false};
+    bool stop_requested_{false};
+    bool attempt_in_flight_{false};
+    bool capture_active_{false};
+    bool promotion_requested_{false};
+    uint64_t attempts_{0};
+    ErrorCode last_error_{ErrorCode::OK};
+    std::optional<Clock::time_point> last_attempt_complete_;
+    std::optional<DurablePrefix> capture_cursor_;
+    std::optional<DurablePrefix> catch_up_target_;
+};
+
+}  // namespace mooncake

mooncake-store/include/ha/snapshot/batch_oplog_snapshot_coordinator.h

@@ -0,0 +1,5 @@
+#pragma once
+
+// Compatibility include; the batch-OpLog implementation lives in the
+// batch_oplog-specific directory.
+#include "ha/snapshot/batch_oplog/batch_oplog_snapshot_coordinator.h"

mooncake-store/include/hot_standby_service.h

@@ -178,6 +178,17 @@ class HotStandbyService {
                                          std::vector<StandbyObjectEntry>& out);
     void EndBatchOpLogSnapshotCapture(BatchOpLogSnapshotCapture& capture);
 
+    // N06 coordinator seams. These stay inert unless a coordinator is
+    // explicitly constructed by the caller.
+    std::optional<DurablePrefix> GetLastAppliedBatchOpLogSnapshotPrefix() const;
+    void CancelBatchOpLogSnapshotCapture();
+    using SnapshotLifecycleCallback = std::function<void()>;
+    void SetBatchOpLogSnapshotCaptureReleasedCallback(
+        SnapshotLifecycleCallback callback);
+    void SetBatchOpLogSnapshotPromotionCallback(
+        SnapshotLifecycleCallback callback);
+    void SetBatchOpLogSnapshotStopCallback(SnapshotLifecycleCallback callback);
+
     // Inject a snapshot provider (from external snapshot implementation).
     void SetSnapshotProvider(std::unique_ptr<SnapshotProvider> provider);
 
@@ -222,6 +233,8 @@ class HotStandbyService {
     void HandleSnapshotCaptureRequest(
         const OpLogBatchStandbyPollResult& result);
     void CancelSnapshotCapture();
+    void NotifySnapshotPromotion();
+    void NotifySnapshotStop();
 
     // Shared body for Promote() and PromoteAndExportSnapshot(): runs the
     // promotion sequence machine transitions + gap resolution + final
@@ -278,6 +291,8 @@ class HotStandbyService {
     std::atomic<bool> replication_loop_running_{false};
     std::mutex replication_loop_mutex_;
     std::condition_variable replication_loop_cv_;
+    mutable std::mutex batch_snapshot_cursor_mutex_;
+    std::optional<DurablePrefix> last_applied_batch_snapshot_prefix_;
 
     std::shared_ptr<BatchOpLogSnapshotCapture::LeaseState>
         snapshot_capture_state_{
@@ -287,7 +302,11 @@ class HotStandbyService {
     // Synchronization
     mutable std::mutex mutex_;
     mutable std::mutex sync_status_callback_mutex_;
+    mutable std::mutex snapshot_lifecycle_callback_mutex_;
     SyncStatusCallback sync_status_callback_;
+    SnapshotLifecycleCallback snapshot_capture_released_callback_;
+    SnapshotLifecycleCallback snapshot_promotion_callback_;
+    SnapshotLifecycleCallback snapshot_stop_callback_;
 };
 
 }  // namespace mooncake

mooncake-store/src/CMakeLists.txt

@@ -43,6 +43,7 @@ set(MOONCAKE_STORE_MASTER_SOURCES
     ha/snapshot/batch_oplog/batch_oplog_snapshot_provider.cpp
     ha/snapshot/batch_oplog/writer.cpp
     ha/snapshot/batch_oplog/batch_oplog_snapshot_publisher.cpp
+    ha/snapshot/batch_oplog/batch_oplog_snapshot_coordinator.cpp
     ha/snapshot/snapshot_maintenance_lease.cpp
     ha/snapshot/master_snapshot_codec.cpp
     ha/snapshot/local_ssd_codec.cpp