coordinator.h

Go to the documentation of this file.

 
#pragma once
 
#include <neograph/api.h>
#include <neograph/graph/types.h>
#include <neograph/graph/checkpoint.h>
#include <neograph/graph/scheduler.h>
#include <memory>
#include <optional>
#include <string>
#include <unordered_map>
#include <vector>
 
namespace neograph { class HookRuntime; }
namespace sp { class NativeArchive; }
 
namespace neograph::graph {
 
class GraphState;
class CancelToken;
namespace detail { struct SubgraphWriteJournal; }
 
struct CheckpointHookContext {
    std::string owner_scope;
    std::string run_id;
    std::string trace_id;
    std::shared_ptr<CancelToken> cancellation;
    std::optional<std::chrono::system_clock::time_point> parent_deadline;
};
 
struct ResumeContext {
    bool have_cp = false;
 
    std::string checkpoint_id;
    json channel_values;         
    std::shared_ptr<const NativeGraphCheckpoint> native_history;
    json metadata;               
    int start_step = 0;          
    CheckpointPhase phase = CheckpointPhase::Completed;
    std::string current_node;
    std::vector<std::string> next_nodes;
 
    std::unordered_map<std::string, NodeResult> replay_results;
 
    BarrierState barrier_state;
    std::shared_ptr<const Checkpoint> managed_budget_source;
};
 
NEOGRAPH_API std::vector<PendingWrite> resume_pending_writes(CheckpointStore& store,
                                                             const Checkpoint& checkpoint);
 
NEOGRAPH_API void copy_resume_pending_writes(CheckpointStore& store, const Checkpoint& source,
                                             const Checkpoint& target);
 
class NEOGRAPH_API CheckpointCoordinator {
public:
    CheckpointCoordinator(std::shared_ptr<CheckpointStore> store,
                          std::string thread_id,
                          std::shared_ptr<::neograph::HookRuntime> hook_runtime = {},
                          CheckpointHookContext hook_context = {},
                          std::shared_ptr<sp::NativeArchive> native_archive = {},
                          std::shared_ptr<detail::SubgraphWriteJournal> native_journal = {});
 
    void set_managed_budget_lease(std::shared_ptr<OwnedManagedBudgetLease> lease);
 
    bool enabled() const noexcept { return store_ != nullptr && !thread_id_.empty(); }
 
    bool records_pending_writes(const std::string& parent_cp_id) const noexcept {
        return enabled() && !parent_cp_id.empty();
    }
 
    const std::shared_ptr<CheckpointStore>& store() const noexcept { return store_; }
    const std::string& thread_id() const noexcept { return thread_id_; }
 
    std::string save_super_step(const GraphState& state,
                                const std::string& current_node,
                                const std::vector<std::string>& next_nodes,
                                CheckpointPhase phase,
                                int step,
                                const std::string& parent_id,
                                const BarrierState& barrier_state) const;
 
    ResumeContext load_for_resume() const;
 
    asio::awaitable<ResumeContext> load_for_resume_async() const;
    asio::awaitable<ResumeContext> load_for_resume_by_id_async(
        std::string checkpoint_id) const;
 
    void record_pending_write(const std::string& parent_cp_id,
                              const std::string& task_id,
                              const std::string& task_path,
                              const std::string& node_name,
                              const NodeResult& nr,
                              int step) const;
 
    void clear_pending_writes(const std::string& parent_cp_id) const;
 
    // ── Async peers (Stage 3 / Sem 3.6 incremental) ─────────────────────
    //
    // Each routes to the matching CheckpointStore::*_async so the
    // coroutine engine path doesn't block the io_context on
    // checkpoint I/O. Behaviour identical to the sync versions.
 
    asio::awaitable<std::string> save_super_step_async(
        const GraphState& state,
        const std::string& current_node,
        const std::vector<std::string>& next_nodes,
        CheckpointPhase phase,
        int step,
        const std::string& parent_id,
        const BarrierState& barrier_state) const;
 
    asio::awaitable<std::string> save_super_step_async(
        const GraphState& state,
        const std::string& current_node,
        const std::vector<std::string>& next_nodes,
        CheckpointPhase phase,
        int step,
        const std::string& parent_id,
        const BarrierState& barrier_state,
        const json& metadata) const;
 
    asio::awaitable<Checkpoint> commit_super_step_async(
        const GraphState& state,
        const std::string& current_node,
        const std::vector<std::string>& next_nodes,
        int step,
        const std::string& parent_id,
        const BarrierState& barrier_state,
        const json& metadata = json()) const;
 
    asio::awaitable<void> record_pending_write_async(
        const std::string& parent_cp_id,
        const std::string& task_id,
        const std::string& task_path,
        const std::string& node_name,
        const NodeResult& nr,
        int step) const;
 
    asio::awaitable<void> clear_pending_writes_async(
        const std::string& parent_cp_id) const;
 
private:
    asio::awaitable<void> publish_checkpoint_async(const Checkpoint& checkpoint) const;
    void capture_state(Checkpoint& checkpoint, const GraphState& state) const;
 
    std::shared_ptr<CheckpointStore> store_;
    std::string thread_id_;
    std::shared_ptr<::neograph::HookRuntime> hook_runtime_;
    CheckpointHookContext hook_context_;
    std::shared_ptr<sp::NativeArchive> native_archive_;
    std::shared_ptr<detail::SubgraphWriteJournal> native_journal_;
    std::shared_ptr<OwnedManagedBudgetLease> managed_budget_lease_;
    bool native_memory_ = false;
};
 
} // namespace neograph::graph