checkpoint.h

Go to the documentation of this file.

 
#pragma once
 
#include <neograph/api.h>
#include <neograph/graph/configuration_failure.h>
#include <neograph/graph/types.h>
 
#include <asio/awaitable.hpp>
 
#include <cstdint>
 
#include <memory>
#include <optional>
#include <mutex>
#include <map>
#include <set>
#include <vector>
#include <unordered_set>
#include <unordered_map>
#include <string_view>
#include <chrono>
#include <functional>
 
namespace neograph::graph {
class NativeGraphCheckpoint;
struct NodeResult;
struct ChannelWrite;
namespace detail {
class ManagedBudgetJournalAccess;
NEOGRAPH_API std::uint64_t blocking_checkpoint_hops() noexcept;
NEOGRAPH_API asio::awaitable<void> run_on_blocking_pool(std::function<void()> work);
}  // namespace detail
 
class NEOGRAPH_API ManagedBudgetDeadlineClockUnavailable final : public std::runtime_error,
                                                                public ConfigurationFailure {
public:
    ManagedBudgetDeadlineClockUnavailable();
};
 
class NEOGRAPH_API SynchronousCheckpointStoreUnavailable final : public std::runtime_error,
                                                                 public ConfigurationFailure {
public:
    SynchronousCheckpointStoreUnavailable();
};
 
struct ManagedBudgetLeaseScope {
    std::string owner_scope;
    std::string thread_id;
    std::string storage_thread_id;
    std::string graph_identity;
    std::uint64_t original_ceiling = 0;
    std::optional<std::int64_t> original_deadline_ticks;
    std::string deadline_clock_identity;
};
 
class NEOGRAPH_API OwnedManagedBudgetLease final {
public:
    const ManagedBudgetLeaseScope& scope() const noexcept;
    const std::string& actor_id() const noexcept;
    const std::string& bank_generation() const noexcept;
    std::uint64_t revision() const noexcept;
    std::string head_checkpoint_id() const;
    std::string head_commitment() const;
    const std::string& execution_thread_id() const noexcept;
    const std::string& execution_storage_thread_id() const noexcept;
private:
    struct Impl;
    explicit OwnedManagedBudgetLease(std::shared_ptr<Impl> impl);
    std::shared_ptr<Impl> impl_;
    friend class detail::ManagedBudgetJournalAccess;
};
 
class NEOGRAPH_API ManagedBudgetEffectReceipt final {
public:
    ManagedBudgetEffectReceipt() = default;
    bool active() const noexcept;
    const std::string& effect_id() const;
    std::uint64_t claim_amount() const;
    const std::string& request_digest() const;
private:
    struct Impl;
    explicit ManagedBudgetEffectReceipt(std::shared_ptr<const Impl> impl);
    std::shared_ptr<const Impl> impl_;
    friend class detail::ManagedBudgetJournalAccess;
};
 
// 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.
// v3 (#91): pending-write records may carry a "mode" field ("overwrite").
// Absent means Reduce, so a v2 blob loads unchanged — same tolerant shape as the
// v1 -> v2 barrier_state addition. Nothing rejects an older version on read; the
// number records what the writer was capable of.
// v4 (#235): child checkpoints may carry the cumulative subgraph write journal
// in metadata. Standalone checkpoints remain readable as before; resuming a
// captured subgraph from an older checkpoint fails explicitly because an
// arbitrary custom reducer cannot reconstruct its historical deltas.
constexpr std::uint32_t CHECKPOINT_SCHEMA_VERSION = 4;
 
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;          
    std::shared_ptr<const NativeGraphCheckpoint> native_history;
    std::shared_ptr<const std::vector<ChannelWrite>> native_subgraph_writes;
    int64_t     step;              
    int64_t     timestamp;         
    std::uint32_t schema_version = CHECKPOINT_SCHEMA_VERSION;
 
    static NEOGRAPH_API std::string generate_id();
};
 
NEOGRAPH_API std::string managed_budget_checkpoint_commitment(const Checkpoint& checkpoint);
NEOGRAPH_API bool checkpoint_channel_blob_eligible(const json& channel);
NEOGRAPH_API json checkpoint_storage_shape(const Checkpoint& checkpoint);
NEOGRAPH_API void restore_checkpoint_storage_shape(Checkpoint& checkpoint, const json& shape,
    const std::map<std::string, json>& blobs);
 
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;      
    std::shared_ptr<const NodeResult> native_result;
};
 
NEOGRAPH_API json checkpoint_storage_metadata(const Checkpoint& checkpoint);
NEOGRAPH_API void restore_checkpoint_storage_envelope(Checkpoint& checkpoint);
 
class NEOGRAPH_API CheckpointStoreCore {
public:
    virtual ~CheckpointStoreCore() = default;
 
    virtual void save(const Checkpoint& cp) = 0;
    virtual std::optional<Checkpoint> load_latest(const std::string& thread_id) = 0;
    virtual std::optional<Checkpoint> load_by_id(const std::string& id) = 0;
    virtual std::vector<Checkpoint> list(const std::string& thread_id,
                                         int limit = 100) = 0;
    virtual void delete_thread(const std::string& thread_id) = 0;
 
    virtual bool requires_managed_budget(const std::string& thread_id);
    virtual std::shared_ptr<OwnedManagedBudgetLease> acquire_managed_budget_lease(
        const ManagedBudgetLeaseScope& scope, const std::string& expected_checkpoint_id,
        const std::string& expected_checkpoint_commitment);
    virtual ManagedBudgetEffectReceipt begin_managed_budget_effect(
        const std::shared_ptr<OwnedManagedBudgetLease>& lease, const std::string& effect_id,
        std::uint64_t exact_claim_amount, const std::string& prepared_request_digest);
    virtual void settle_managed_budget_effect(const std::shared_ptr<OwnedManagedBudgetLease>& lease,
        const ManagedBudgetEffectReceipt& effect, sp::runtime::Result genuine_outcome,
        const UsageAccumulator::AuthoritySnapshot& authority);
    virtual void publish_managed_budget_checkpoint(
        const std::shared_ptr<OwnedManagedBudgetLease>& lease, const Checkpoint& checkpoint);
    virtual void release_managed_budget_lease(const std::shared_ptr<OwnedManagedBudgetLease>& lease);
    virtual bool retains_native_checkpoint() const noexcept { return false; }
    virtual void publish_managed_budget_fork(
        const Checkpoint& authenticated_source, const Checkpoint& genuine_shared_bank_fork);
};
 
class NEOGRAPH_API AsyncCheckpointStore {
public:
    virtual ~AsyncCheckpointStore() = default;
 
    virtual asio::awaitable<void> save_async(const Checkpoint& cp) = 0;
    virtual asio::awaitable<std::optional<Checkpoint>>
    load_latest_async(const std::string& thread_id) = 0;
    virtual asio::awaitable<std::optional<Checkpoint>>
    load_by_id_async(const std::string& id) = 0;
    virtual asio::awaitable<std::vector<Checkpoint>>
    list_async(const std::string& thread_id, int limit = 100) = 0;
    virtual asio::awaitable<void>
    delete_thread_async(const std::string& thread_id) = 0;
 
    virtual asio::awaitable<bool> requires_managed_budget_async(std::string thread_id);
    virtual asio::awaitable<std::shared_ptr<OwnedManagedBudgetLease>> acquire_managed_budget_lease_async(
        ManagedBudgetLeaseScope scope, std::string expected_checkpoint_id,
        std::string expected_checkpoint_commitment);
    virtual asio::awaitable<ManagedBudgetEffectReceipt> begin_managed_budget_effect_async(
        std::shared_ptr<OwnedManagedBudgetLease> lease, std::string effect_id,
        std::uint64_t exact_claim_amount, std::string prepared_request_digest);
    virtual asio::awaitable<void> settle_managed_budget_effect_async(
        std::shared_ptr<OwnedManagedBudgetLease> lease, ManagedBudgetEffectReceipt effect,
        sp::runtime::Result genuine_outcome, UsageAccumulator::AuthoritySnapshot authority);
    virtual asio::awaitable<void> publish_managed_budget_checkpoint_async(
        std::shared_ptr<OwnedManagedBudgetLease> lease, Checkpoint checkpoint);
    virtual asio::awaitable<void> release_managed_budget_lease_async(
        std::shared_ptr<OwnedManagedBudgetLease> lease);
    virtual bool retains_native_checkpoint() const noexcept { return false; }
    virtual asio::awaitable<void> publish_managed_budget_fork_async(
        Checkpoint authenticated_source, Checkpoint genuine_shared_bank_fork);
};
 
class NEOGRAPH_API PendingWritesCheckpointStore {
public:
    virtual ~PendingWritesCheckpointStore() = default;
 
    virtual void put_writes(const std::string& thread_id,
                            const std::string& parent_checkpoint_id,
                            const PendingWrite& write) = 0;
    virtual std::vector<PendingWrite> get_writes(
        const std::string& thread_id,
        const std::string& parent_checkpoint_id) = 0;
    virtual void clear_writes(const std::string& thread_id,
                              const std::string& parent_checkpoint_id) = 0;
};
 
class NEOGRAPH_API CheckpointStore {
public:
    virtual ~CheckpointStore() = default;
 
    // ── Sync API ────────────────────────────────────────────────────────
    //
    // Legacy sync facade. Override sync operations or explicitly adapt a
    // native AsyncCheckpointStore. Missing sync operations fail explicitly;
    // async defaults offload sync-only backends to a bounded worker pool.
 
    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 defaults offload the matching synchronous operation instead of
    // blocking the caller's Asio executor. Async-native stores should override
    // these methods to avoid the blocking pool entirely.
 
    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);
 
    virtual bool requires_managed_budget(const std::string& thread_id);
    virtual asio::awaitable<bool> requires_managed_budget_async(std::string thread_id);
 
    virtual std::shared_ptr<OwnedManagedBudgetLease> acquire_managed_budget_lease(
        const ManagedBudgetLeaseScope& scope, const std::string& expected_checkpoint_id,
        const std::string& expected_checkpoint_commitment);
    virtual asio::awaitable<std::shared_ptr<OwnedManagedBudgetLease>> acquire_managed_budget_lease_async(
        ManagedBudgetLeaseScope scope, std::string expected_checkpoint_id,
        std::string expected_checkpoint_commitment);
    virtual ManagedBudgetEffectReceipt begin_managed_budget_effect(
        const std::shared_ptr<OwnedManagedBudgetLease>& lease, const std::string& effect_id,
        std::uint64_t exact_claim_amount, const std::string& prepared_request_digest);
    virtual asio::awaitable<ManagedBudgetEffectReceipt> begin_managed_budget_effect_async(
        std::shared_ptr<OwnedManagedBudgetLease> lease, std::string effect_id,
        std::uint64_t exact_claim_amount, std::string prepared_request_digest);
    virtual void settle_managed_budget_effect(const std::shared_ptr<OwnedManagedBudgetLease>& lease,
        const ManagedBudgetEffectReceipt& effect, sp::runtime::Result genuine_outcome,
        const UsageAccumulator::AuthoritySnapshot& authority);
    virtual asio::awaitable<void> settle_managed_budget_effect_async(
        std::shared_ptr<OwnedManagedBudgetLease> lease, ManagedBudgetEffectReceipt effect,
        sp::runtime::Result genuine_outcome, UsageAccumulator::AuthoritySnapshot authority);
    virtual void publish_managed_budget_checkpoint(
        const std::shared_ptr<OwnedManagedBudgetLease>& lease, const Checkpoint& checkpoint);
    virtual asio::awaitable<void> publish_managed_budget_checkpoint_async(
        std::shared_ptr<OwnedManagedBudgetLease> lease, Checkpoint checkpoint);
    virtual void release_managed_budget_lease(const std::shared_ptr<OwnedManagedBudgetLease>& lease);
    virtual asio::awaitable<void> release_managed_budget_lease_async(
        std::shared_ptr<OwnedManagedBudgetLease> lease);
    virtual bool retains_native_checkpoint() const noexcept { return false; }
    virtual void publish_managed_budget_fork(
        const Checkpoint& authenticated_source, const Checkpoint& genuine_shared_bank_fork);
    virtual asio::awaitable<void> publish_managed_budget_fork_async(
        Checkpoint authenticated_source, Checkpoint genuine_shared_bank_fork);
};
 
NEOGRAPH_API std::shared_ptr<CheckpointStore>
adapt_checkpoint_store(std::shared_ptr<CheckpointStoreCore> core);
 
NEOGRAPH_API std::shared_ptr<CheckpointStore>
adapt_async_checkpoint_store(std::shared_ptr<AsyncCheckpointStore> backend);
 
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;
 
    // The exact in-memory concrete type has no I/O boundary, so its async
    // operations stay on the caller executor instead of paying a legacy
    // blocking-pool handoff for each mutex-protected map operation. Derived
    // stores retain the base fallback unless they explicitly override async.
    asio::awaitable<void> save_async(const Checkpoint& cp) override;
    asio::awaitable<std::optional<Checkpoint>>
    load_latest_async(const std::string& thread_id) override;
    asio::awaitable<std::optional<Checkpoint>>
    load_by_id_async(const std::string& id) override;
    asio::awaitable<std::vector<Checkpoint>>
    list_async(const std::string& thread_id, int limit = 100) override;
    asio::awaitable<void>
    delete_thread_async(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;
 
    asio::awaitable<void> put_writes_async(
        const std::string& thread_id,
        const std::string& parent_checkpoint_id,
        const PendingWrite& write) override;
    asio::awaitable<std::vector<PendingWrite>> get_writes_async(
        const std::string& thread_id,
        const std::string& parent_checkpoint_id) override;
    asio::awaitable<void> clear_writes_async(
        const std::string& thread_id,
        const std::string& parent_checkpoint_id) override;
 
    bool requires_managed_budget(const std::string& thread_id) override;
    asio::awaitable<bool> requires_managed_budget_async(std::string thread_id) override;
    std::shared_ptr<OwnedManagedBudgetLease> acquire_managed_budget_lease(
        const ManagedBudgetLeaseScope& scope, const std::string& expected_checkpoint_id,
        const std::string& expected_checkpoint_commitment) override;
    ManagedBudgetEffectReceipt begin_managed_budget_effect(
        const std::shared_ptr<OwnedManagedBudgetLease>& lease, const std::string& effect_id,
        std::uint64_t exact_claim_amount, const std::string& prepared_request_digest) override;
    void settle_managed_budget_effect(const std::shared_ptr<OwnedManagedBudgetLease>& lease,
        const ManagedBudgetEffectReceipt& effect, sp::runtime::Result genuine_outcome,
        const UsageAccumulator::AuthoritySnapshot& authority) override;
    void publish_managed_budget_checkpoint(
        const std::shared_ptr<OwnedManagedBudgetLease>& lease, const Checkpoint& checkpoint) override;
    void release_managed_budget_lease(const std::shared_ptr<OwnedManagedBudgetLease>& lease) override;
    bool retains_native_checkpoint() const noexcept override { return true; }
    void publish_managed_budget_fork(
        const Checkpoint& authenticated_source, const Checkpoint& genuine_shared_bank_fork) override;
 
    // The managed-budget operations run on the calling thread, as the checkpoint operations above do: this store
    // is memory, so there is nothing to wait for and no worker thread to hand them to. A single-thread build has
    // none, and the engine runs a budgeted graph on this store there.
    asio::awaitable<std::shared_ptr<OwnedManagedBudgetLease>> acquire_managed_budget_lease_async(
        ManagedBudgetLeaseScope scope, std::string expected_checkpoint_id,
        std::string expected_checkpoint_commitment) override;
    asio::awaitable<ManagedBudgetEffectReceipt> begin_managed_budget_effect_async(
        std::shared_ptr<OwnedManagedBudgetLease> lease, std::string effect_id,
        std::uint64_t exact_claim_amount, std::string prepared_request_digest) override;
    asio::awaitable<void> settle_managed_budget_effect_async(
        std::shared_ptr<OwnedManagedBudgetLease> lease, ManagedBudgetEffectReceipt effect,
        sp::runtime::Result genuine_outcome, UsageAccumulator::AuthoritySnapshot authority) override;
    asio::awaitable<void> publish_managed_budget_checkpoint_async(
        std::shared_ptr<OwnedManagedBudgetLease> lease, Checkpoint checkpoint) override;
    asio::awaitable<void> release_managed_budget_lease_async(
        std::shared_ptr<OwnedManagedBudgetLease> lease) override;
    asio::awaitable<void> publish_managed_budget_fork_async(
        Checkpoint authenticated_source, Checkpoint genuine_shared_bank_fork) 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_;
    std::unordered_set<std::string> managed_budget_obligations_;
    std::unordered_map<std::string, json> managed_budget_journals_;
    std::map<std::tuple<std::string, std::string, std::string>, json> managed_budget_effects_;
    std::map<std::tuple<std::string, std::string, std::string>, sp::runtime::Result> managed_budget_results_;
    struct ManagedBudgetBranchHead {
        std::string financial_storage_thread_id;
        std::string execution_thread_id;
        std::string checkpoint_id;
        std::string commitment;
        std::shared_ptr<const NativeGraphCheckpoint> native_history;
        bool valid = true;
    };
    const ManagedBudgetBranchHead& authenticate_managed_budget_branch_locked(
        const Checkpoint& source, const std::string& commitment) const;
    void validate_managed_budget_execution_locked(
        const std::shared_ptr<OwnedManagedBudgetLease>& lease, const json& head) const;
    std::unordered_map<std::string, ManagedBudgetBranchHead> managed_budget_branch_heads_;
};
 
} // namespace neograph::graph