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 all `neograph_*` tables.

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, ordered by timestamp (newest first).

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 recent 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 detection.

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

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().

Definition at line 106 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 all neograph_* tables.

Test-only utility — destroys data. Useful in test fixtures that want a clean slate per test case.

◆ 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, ordered by timestamp (newest first).

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 recent checkpoint for a thread.

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 125 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 detection.

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

Definition at line 121 of file postgres_checkpoint.h.

◆ save()

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

overridevirtual

Save a checkpoint.

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

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()