checkpoint.h

Go to the documentation of this file.

 
#pragma once
 
#include <neograph/api.h>
#include <neograph/graph/types.h>
 
#include <asio/awaitable.hpp>
 
#include <cstdint>
 
#include <optional>
#include <mutex>
#include <map>
#include <set>
#include <vector>
#include <string_view>
#include <chrono>
 
namespace neograph::graph {
 
// Unsigned + fixed-width: schema versions are non-negative and a
// platform-variable `int` width is wrong for a value persisted to
// disk and round-tripped through JSON. Old `int` left the door open
// to a `-1` sentinel value that wasn't documented anywhere.
constexpr std::uint32_t CHECKPOINT_SCHEMA_VERSION = 2;
 
enum class CheckpointPhase {
    Before,         
    After,          
    Completed,      
    NodeInterrupt,  
    Updated         
};
 
NEOGRAPH_API const char* to_string(CheckpointPhase phase);
 
NEOGRAPH_API CheckpointPhase parse_checkpoint_phase(std::string_view s);
 
struct Checkpoint {
    std::string id;                
    std::string thread_id;         
    json        channel_values;    
    json        channel_versions;  
    std::string parent_id;         
    std::string current_node;      
    std::vector<std::string> next_nodes;
    CheckpointPhase interrupt_phase = CheckpointPhase::Completed;  
    std::map<std::string, std::set<std::string>> barrier_state;
    json        metadata;          
    int64_t     step;              
    int64_t     timestamp;         
    std::uint32_t schema_version = CHECKPOINT_SCHEMA_VERSION;
 
    static std::string generate_id();
};
 
struct PendingWrite {
    std::string task_id;        
    std::string task_path;      
    std::string node_name;      
    json        writes;         
    json        command;        
    json        sends;          
    int64_t     step;           
    int64_t     timestamp;      
};
 
class NEOGRAPH_API CheckpointStore {
public:
    virtual ~CheckpointStore() = default;
 
    // ── Sync API ────────────────────────────────────────────────────────
    //
    // Stage 3 / Semester 3.1: each sync method is now non-pure with a
    // default implementation that bridges to its async peer via
    // `neograph::async::run_sync`. The legacy stores (InMemory, SQLite,
    // Postgres) still override these directly. Async-native stores can
    // override only the *_async() variants and inherit these as facades.
    // Override at least one side per method — overriding neither yields
    // infinite mutual recursion at call time.
 
    virtual void save(const Checkpoint& cp);
 
    virtual std::optional<Checkpoint> load_latest(const std::string& thread_id);
 
    virtual std::optional<Checkpoint> load_by_id(const std::string& id);
 
    virtual std::vector<Checkpoint> list(const std::string& thread_id,
                                          int limit = 100);
 
    virtual void delete_thread(const std::string& thread_id);
 
    // ── Async API ───────────────────────────────────────────────────────
    //
    // Each method's default body co_returns the matching sync call. Real
    // async stores (libpq pipeline in Sem 3.3) override these to perform
    // non-blocking I/O.
 
    virtual asio::awaitable<void> save_async(const Checkpoint& cp);
    virtual asio::awaitable<std::optional<Checkpoint>>
    load_latest_async(const std::string& thread_id);
    virtual asio::awaitable<std::optional<Checkpoint>>
    load_by_id_async(const std::string& id);
    virtual asio::awaitable<std::vector<Checkpoint>>
    list_async(const std::string& thread_id, int limit = 100);
    virtual asio::awaitable<void>
    delete_thread_async(const std::string& thread_id);
 
    // ── Pending writes (fine-grained progress log) ──────────────────────
 
    virtual void put_writes(const std::string& /*thread_id*/,
                            const std::string& /*parent_checkpoint_id*/,
                            const PendingWrite& /*write*/) {}
 
    virtual std::vector<PendingWrite> get_writes(
        const std::string& /*thread_id*/,
        const std::string& /*parent_checkpoint_id*/) { return {}; }
 
    virtual void clear_writes(const std::string& /*thread_id*/,
                              const std::string& /*parent_checkpoint_id*/) {}
 
    // Async peers for the pending-writes API. Each defaults to calling
    // the matching sync method (which itself is a no-op for stores that
    // don't override it), so existing custom stores keep working.
 
    virtual asio::awaitable<void> put_writes_async(
        const std::string& thread_id,
        const std::string& parent_checkpoint_id,
        const PendingWrite& write);
    virtual asio::awaitable<std::vector<PendingWrite>> get_writes_async(
        const std::string& thread_id,
        const std::string& parent_checkpoint_id);
    virtual asio::awaitable<void> clear_writes_async(
        const std::string& thread_id,
        const std::string& parent_checkpoint_id);
};
 
class NEOGRAPH_API InMemoryCheckpointStore : public CheckpointStore {
public:
    void save(const Checkpoint& cp) override;
    std::optional<Checkpoint> load_latest(const std::string& thread_id) override;
    std::optional<Checkpoint> load_by_id(const std::string& id) override;
    std::vector<Checkpoint> list(const std::string& thread_id,
                                  int limit = 100) override;
    void delete_thread(const std::string& thread_id) override;
 
    void put_writes(const std::string& thread_id,
                    const std::string& parent_checkpoint_id,
                    const PendingWrite& write) override;
    std::vector<PendingWrite> get_writes(
        const std::string& thread_id,
        const std::string& parent_checkpoint_id) override;
    void clear_writes(const std::string& thread_id,
                      const std::string& parent_checkpoint_id) override;
 
    size_t size() const;
 
    size_t blob_count() const;
 
    size_t pending_writes_count(const std::string& thread_id,
                                const std::string& parent_checkpoint_id) const;
 
private:
    Checkpoint split_blobs_locked(Checkpoint cp);
 
    Checkpoint join_blobs_locked(Checkpoint cp) const;
 
    mutable std::mutex mutex_;
    std::map<std::string, std::vector<Checkpoint>> by_thread_;  
    std::map<std::string, Checkpoint> by_id_;                   
    std::map<std::tuple<std::string, std::string, uint64_t>, json> blobs_;
    // Keyed by (thread_id, parent_checkpoint_id) → ordered list of pending writes
    std::map<std::pair<std::string, std::string>, std::vector<PendingWrite>> pending_;
};
 
} // namespace neograph::graph