Persistent CheckpointStore backed by PostgreSQL via libpq. More...

#include <postgres_checkpoint.h>

Inheritance diagram for neograph::graph::PostgresCheckpointStore:

Collaboration diagram for neograph::graph::PostgresCheckpointStore:

Public Member Functions
size_t	blob_count ()
	Test helper: count rows in `neograph_checkpoint_blobs`.

void	clear_writes (const std::string &thread_id, const std::string &parent_checkpoint_id) override
	Discard pending writes for a parent checkpoint once its successor super-step has been fully committed.

void	delete_thread (const std::string &thread_id) override
	Delete all checkpoints for a thread.

void	drop_schema ()
	Drop checkpoint data tables, retaining obligations, migration markers, currency heads and write-ahead effects.

std::vector< PendingWrite >	get_writes (const std::string &thread_id, const std::string &parent_checkpoint_id) override
	Load all pending writes attached to a parent checkpoint.

std::vector< Checkpoint >	list (const std::string &thread_id, int limit=100) override
	List checkpoints for a thread, most recently saved first, one entry per ID; the first entry is load_latest().

std::optional< Checkpoint >	load_by_id (const std::string &id) override
	Load a checkpoint by its unique ID.

std::optional< Checkpoint >	load_latest (const std::string &thread_id) override
	Load the most recently saved checkpoint for a thread.

size_t	pool_size () const
	Number of connections in the pool.

	PostgresCheckpointStore (const std::string &conn_str, size_t pool_size=8)

void	put_writes (const std::string &thread_id, const std::string &parent_checkpoint_id, const PendingWrite &write) override
	Record a successful node execution within an in-progress super-step.

size_t	reconnect_count () const
	Number of times a pool slot's connection was replaced after a broken connection or an interrupted async exchange.

bool	requires_managed_budget (const std::string &thread_id) override
	Indexed denial obligation; does not authenticate or authorize a bank.

void	save (const Checkpoint &cp) override
	Save a checkpoint.

Public Member Functions inherited from neograph::graph::CheckpointStore
virtual bool	retains_native_checkpoint () const noexcept
	Describes real local C++ custody, not a financial admission/grant.

Detailed Description

Persistent CheckpointStore backed by PostgreSQL via libpq.

Construct with a libpq connection string (e.g. "postgresql://user:pass@host:5432/dbname"). The constructor opens the connection eagerly and runs ensure_schema() so callers get an immediate failure if credentials or DDL permissions are wrong, rather than a delayed surprise on first save().

Async connection deadline

Async initial/replacement connection attempts use one NeoGraph-wide deadline for the complete attempt, including every host and resolved IP. A positive connect_timeout written directly in conn_str supplies the deadline in seconds, with libpq's documented two-second minimum. Values from PGCONNECT_TIMEOUT or a service file are not available before the initial nonblocking connection step and therefore use NeoGraph's 30-second default. If the explicit value is absent, zero, or negative, NeoGraph also applies that production-safety default. This intentionally differs from libpq's synchronous per-host timeout semantics: an async multi-host attempt receives one budget rather than one budget per host. Synchronous construction and synchronous replacement retain native libpq behavior and are unchanged by this policy.

Cancelling an in-flight async query quarantines its pool connection and sends PostgreSQL's out-of-band cancel request on NeoGraph's bounded worker pool before closing that connection. The cancelled operation is never retried; the empty slot is rebuilt lazily by the next operation.

Definition at line 143 of file postgres_checkpoint.h.

Constructor & Destructor Documentation

◆ PostgresCheckpointStore()

neograph::graph::PostgresCheckpointStore::PostgresCheckpointStore	(	const std::string &	conn_str,
		size_t	pool_size = `8`
	)

explicit

Parameters

conn_str	libpq connection string. Anything libpq accepts.
pool_size	Number of connections to open eagerly. Defaults to 8 — a sensible match for typical small server worker pools. Set to 1 for embedded / single-thread use to save one PG backend per store. Must be >= 1.

Exceptions

std::runtime_error on connection or DDL failure.

Member Function Documentation

◆ blob_count()

size_t neograph::graph::PostgresCheckpointStore::blob_count ( )

Test helper: count rows in neograph_checkpoint_blobs.

Lets integration tests verify blob dedup across saves the same way InMemoryCheckpointStore::blob_count() does.

◆ clear_writes()

void neograph::graph::PostgresCheckpointStore::clear_writes	(	const std::string &	,
		const std::string &
	)

overridevirtual

Discard pending writes for a parent checkpoint once its successor super-step has been fully committed.

Called by the engine after the new super-step checkpoint has been durably saved, so pending writes are never cleared while still being the only record of a node's output.

Parameters

thread_id	Thread identifier.
parent_checkpoint_id	Checkpoint whose pending writes to clear.

Reimplemented from neograph::graph::CheckpointStore.

◆ delete_thread()

void neograph::graph::PostgresCheckpointStore::delete_thread ( const std::string & thread_id )

overridevirtual

Delete all checkpoints for a thread.

Parameters

thread_id Thread identifier to delete.

Reimplemented from neograph::graph::CheckpointStore.

◆ drop_schema()

void neograph::graph::PostgresCheckpointStore::drop_schema ( )

Drop checkpoint data tables, retaining obligations, migration markers, currency heads and write-ahead effects.

No bank is reset by this utility.

◆ get_writes()

std::vector< PendingWrite > neograph::graph::PostgresCheckpointStore::get_writes	(	const std::string &	,
		const std::string &
	)

overridevirtual

Load all pending writes attached to a parent checkpoint.

Called by the engine on resume to skip already-completed tasks. Default implementation returns an empty vector.

Parameters

thread_id	Thread identifier.
parent_checkpoint_id	Checkpoint whose pending writes to load.

Returns: Vector of pending writes, in insertion order.

Reimplemented from neograph::graph::CheckpointStore.

◆ list()

std::vector< Checkpoint > neograph::graph::PostgresCheckpointStore::list	(	const std::string &	thread_id,
		int	limit = `100`
	)

overridevirtual

List checkpoints for a thread, most recently saved first, one entry per ID; the first entry is load_latest().

Parameters

thread_id	Thread identifier.
limit	Maximum number of checkpoints to return (default: 100).

Returns: Vector of checkpoints.

Reimplemented from neograph::graph::CheckpointStore.

◆ load_by_id()

std::optional< Checkpoint > neograph::graph::PostgresCheckpointStore::load_by_id ( const std::string & id )

overridevirtual

Load a checkpoint by its unique ID.

Parameters

id	Checkpoint UUID.

Returns: The checkpoint, or std::nullopt if not found.

Reimplemented from neograph::graph::CheckpointStore.

◆ load_latest()

std::optional< Checkpoint > neograph::graph::PostgresCheckpointStore::load_latest ( const std::string & thread_id )

overridevirtual

Load the most recently saved checkpoint for a thread.

"Latest" is save order, not timestamp or step: a later run on a thread restarts at step 0, and saves can share a millisecond.

Parameters

thread_id Thread identifier.

Returns: The latest checkpoint, or std::nullopt if none exists.

Reimplemented from neograph::graph::CheckpointStore.

◆ pool_size()

size_t neograph::graph::PostgresCheckpointStore::pool_size ( ) const

inline

Number of connections in the pool.

Useful for benchmarks and for confirming the pool was sized as expected.

Definition at line 161 of file postgres_checkpoint.h.

◆ put_writes()

void neograph::graph::PostgresCheckpointStore::put_writes	(	const std::string &	,
		const std::string &	,
		const PendingWrite &
	)

overridevirtual

Record a successful node execution within an in-progress super-step.

Called by the engine immediately after a node returns successfully and before its writes are applied to the shared GraphState. The parent_checkpoint_id anchors the pending write to the super-step boundary it was produced under.

Default implementation is a no-op so custom stores keep working; such stores fall back to "full super-step replay" on resume.

Parameters

thread_id	Thread identifier.
parent_checkpoint_id	Checkpoint marking the start of the in-progress super-step.
write	The pending write record to persist.

Reimplemented from neograph::graph::CheckpointStore.

◆ reconnect_count()

size_t neograph::graph::PostgresCheckpointStore::reconnect_count ( ) const

inline

Number of times a pool slot's connection was replaced after a broken connection or an interrupted async exchange.

Cumulative; useful for monitoring (e.g. Prometheus gauge) and for tests that want to assert the retry path fired.

Definition at line 157 of file postgres_checkpoint.h.

◆ save()

void neograph::graph::PostgresCheckpointStore::save ( const Checkpoint & cp )

overridevirtual

Save a checkpoint.

Saving an ID the thread already holds replaces that checkpoint and makes it the thread's latest, as if it were saved for the first time.

Parameters

cp	The checkpoint to persist.

Reimplemented from neograph::graph::CheckpointStore.

The documentation for this class was generated from the following file:

include/neograph/graph/postgres_checkpoint.h

Public Member Functions

Detailed Description

Async connection deadline

Constructor & Destructor Documentation

◆ PostgresCheckpointStore()

Member Function Documentation

◆ blob_count()

◆ clear_writes()

◆ delete_thread()

◆ drop_schema()

◆ get_writes()

◆ list()

◆ load_by_id()

◆ load_latest()

◆ pool_size()

◆ put_writes()

◆ reconnect_count()

◆ save()