Skip to content

sgns::crdt::GlobalDB

#include <globaldb.hpp>

Public Classes

Name
struct BackupOptions

Public Types

Name
enum class uint8_t Error { ROCKSDB_IO = 0, IPFS_DB_NOT_CREATED, DAG_SYNCHER_NOT_LISTENING, CRDT_DATASTORE_NOT_CREATED, PUBSUB_BROADCASTER_NOT_CREATED, INVALID_PARAMETERS, GLOBALDB_NOT_STARTED, POST_COMMIT_ERASE_ANOMALY}
Enumeration of error codes used in the proof classes.
using base::Buffer Buffer
using CrdtDatastore::QueryResult QueryResult
using storage::rocksdb RocksDB
using CrdtHeads::CRDTListResult CRDTHeadListResult
using std::pair< HierarchicalKey, Buffer > DataPair
Pair of key and value to be stored in CRDT.
using CrdtDatastore::CRDTElementFilterCallback GlobalDBFilterCallback
CRDT Filter callback type.
using CrdtDatastore::CRDTNewElementCallback GlobalDBNewElementCallback
using CrdtDatastore::CRDTDeletedElementCallback GlobalDBDeletedElementCallback

Public Functions

Name
outcome::result< std::shared_ptr< GlobalDB > > New(std::shared_ptr< boost::asio::io_context > context, std::string databasePath, std::shared_ptr< sgns::ipfs_pubsub::GossipPubSub > pubsub, std::shared_ptr< CrdtOptions > crdtOptions, std::shared_ptr< sgns::ipfs_lite::ipfs::graphsync::Network > graphsyncnetwork, std::shared_ptr< libp2p::basic::Scheduler > scheduler, std::shared_ptr< sgns::ipfs_lite::ipfs::graphsync::RequestIdGenerator > generator, std::shared_ptr< RocksDB > datastore =nullptr, BackupOptions backup_options =BackupOptions)
Factory method to create a GlobalDB instance.
void SetEraseInterruptHookForTest(std::function< bool(int step)> hook)
Test-only interruption seam for EraseEntry's compose window.
~GlobalDB()
Destructor or GlobalDB.
outcome::result< CID > Put(const HierarchicalKey & key, const Buffer & value, const std::unordered_set< std::string > & topics)
Puts key-value pair to the CRDT store, optionally specifying a broadcast topic.
outcome::result< CID > PutConvergentImmutable(const HierarchicalKey & key, const Buffer & value, const std::unordered_set< std::string > & topics)
outcome::result< CID > Put(const std::vector< DataPair > & data_vector, const std::unordered_set< std::string > & topics)
Writes a batch of CRDT data all at once.
outcome::result< void > PutLocal(const HierarchicalKey & key, const Buffer & value, const std::string & id)
Puts a key-value pair directly into local storage, bypassing DAG broadcast. See CrdtDatastore::PutKeyLocal for the full rationale/contract.
outcome::result< Buffer > Get(const HierarchicalKey & key)
outcome::result< CID > Remove(const HierarchicalKey & key, const std::unordered_set< std::string > & topics)
outcome::result< QueryResult > QueryKeyValues(std::string_view keyPrefix)
outcome::result< QueryResult > QueryKeyValues(const std::string & prefix_base, const std::string & middle_part, const std::string & remainder_prefix)
Queries with a middle part that can be a wildcard, negated string or normal string.
outcome::result< std::string > KeyToString(const Buffer & key) const
std::shared_ptr< AtomicTransaction > BeginTransaction()
outcome::result< void > AddBroadcastTopic(const std::string & topicName)
void AddTopicName(std::string topicName)
void AddListenTopic(std::string topicName)
void PrintDataStore()
std::shared_ptr< RocksDB > GetDataStore()
outcome::result< Buffer > GetRaw(const Buffer & key) const
Reads a node-local key straight from RocksDB, bypassing the CRDT layer.
outcome::result< void > PutRaw(const Buffer & key, const Buffer & value)
Writes a node-local key straight to RocksDB, bypassing the CRDT layer.
outcome::result< QueryResult > QueryRaw(const Buffer & key_prefix) const
Prefix-scans node-local keys straight from RocksDB, bypassing the CRDT layer.
outcome::result< void > EraseEntry(const std::string & key, const CID & cid, const std::optional< std::pair< Buffer, Buffer > > & extra_entry =std::nullopt)
Generic node-local physical erase: removes, in ONE atomic RocksDB WriteBatch, the CRDT set entries for key (via the CrdtSet owner), the raw block bytes for cid (via the DAGSyncer owner), and the node-local raw bookkeeping entry at key itself.
std::shared_ptr< sgns::crdt::PubSubBroadcasterExt > GetBroadcaster()
std::shared_ptr< CRDTWorkJournal > GetWorkJournal() const
bool RegisterElementFilter(const std::string & pattern, GlobalDBFilterCallback filter)
bool RegisterNewElementCallback(const std::string & pattern, GlobalDBNewElementCallback callback)
bool RegisterDeletedElementCallback(const std::string & pattern, GlobalDBDeletedElementCallback callback)
void UnregisterElementFilter(const std::string & pattern)
void UnregisterNewElementCallback(const std::string & pattern)
Unregisters the new element callback for a pattern.
void UnregisterDeletedElementCallback(const std::string & pattern)
Unregisters the deleted element callback for a pattern.
void Start()
Starts the GlobalDB instance.
void ShutdownNow()
Immediately quiesce and shut down CRDT intake and workers. Safe to call multiple times.
void StartCIDReceiving()
Starts receiving CIDs.
void StartCICSync()
Starts CIC synchronization.
void StartRebroadcastHeads()
Starts rebroadcasting heads.
outcome::result< CRDTHeadListResult > GetCRDTHeadList()
outcome::result< uint64_t > GetKeyPriority(const std::string & key) const
outcome::result< uint64_t > GetCRDTHeadHeight(const CID & aCid, const std::string & topic)
outcome::result< void > CRDTHeadRemove(const CID & aCid, const std::string & topic)
outcome::result< void > CRDTHeadAdd(const CID & aCid, const std::string & topic, uint64_t priority)
outcome::result< crdt::CrdtDatastore::JobStatus > GetCIDJobStatus(const CID & cid) const
outcome::result< void > RequestHeadBroadcast(const std::set< std::string > & topics)
Request head broadcast for specified topics.
outcome::result< std::unordered_set< std::string > > GetMonitoredTopics() const
Get the topics that are being listened to.
std::shared_ptr< crdt::CrdtDatastore > GetCRDTDataStore()
outcome::result< std::vector< std::pair< std::string, base::Buffer > > > GetLocalDeltaKeyValues(const std::string & cid_string)
outcome::result< std::vector< std::string > > GetLocalDeltaKeys(const std::string & cid_string) const
outcome::result< bool > HasLocalBlock(const std::string & cid_string) const

Public Types Documentation

enum Error

Enumerator Value Description
ROCKSDB_IO 0 RocksDB wasn't opened.
IPFS_DB_NOT_CREATED IPFS datastore not created.
DAG_SYNCHER_NOT_LISTENING DAG Syncher listen error.
CRDT_DATASTORE_NOT_CREATED CRDT DataStore not created.
PUBSUB_BROADCASTER_NOT_CREATED CRDT DataStore not created.
INVALID_PARAMETERS Invalid parameters.
GLOBALDB_NOT_STARTED Start wasn't called.
POST_COMMIT_ERASE_ANOMALY Erase already committed; post-commit seal re-check failed.

Enumeration of error codes used in the proof classes.

using Buffer

using sgns::crdt::GlobalDB::Buffer =  base::Buffer;

using QueryResult

using sgns::crdt::GlobalDB::QueryResult =  CrdtDatastore::QueryResult;

using RocksDB

using sgns::crdt::GlobalDB::RocksDB =  storage::rocksdb;

using CRDTHeadListResult

using sgns::crdt::GlobalDB::CRDTHeadListResult =  CrdtHeads::CRDTListResult;

using DataPair

using sgns::crdt::GlobalDB::DataPair =  std::pair<HierarchicalKey, Buffer>;

Pair of key and value to be stored in CRDT.

using GlobalDBFilterCallback

using sgns::crdt::GlobalDB::GlobalDBFilterCallback =  CrdtDatastore::CRDTElementFilterCallback;

CRDT Filter callback type.

using GlobalDBNewElementCallback

using sgns::crdt::GlobalDB::GlobalDBNewElementCallback =  CrdtDatastore::CRDTNewElementCallback;

using GlobalDBDeletedElementCallback

using sgns::crdt::GlobalDB::GlobalDBDeletedElementCallback =  CrdtDatastore::CRDTDeletedElementCallback;

Public Functions Documentation

function New

static outcome::result< std::shared_ptr< GlobalDB > > New(
    std::shared_ptr< boost::asio::io_context > context,
    std::string databasePath,
    std::shared_ptr< sgns::ipfs_pubsub::GossipPubSub > pubsub,
    std::shared_ptr< CrdtOptions > crdtOptions,
    std::shared_ptr< sgns::ipfs_lite::ipfs::graphsync::Network > graphsyncnetwork,
    std::shared_ptr< libp2p::basic::Scheduler > scheduler,
    std::shared_ptr< sgns::ipfs_lite::ipfs::graphsync::RequestIdGenerator > generator,
    std::shared_ptr< RocksDB > datastore =nullptr,
    BackupOptions backup_options =BackupOptions{ false, 15, 12, true }
)

Factory method to create a GlobalDB instance.

Parameters:

  • context The io context used to run its inner methods
  • databasePath Local system's path where data will be stored, not used if datastore is not nullptr
  • pubsub The pubsub instance used to communicate
  • crdtOptions CRDT options
  • graphsyncnetwork The graphsync networks used
  • scheduler libp2p scheduler
  • generator The request ID generator from graphsync
  • datastore datastore to be used. If not defined, created using databasePath
  • backup_options configuration for automatic backups of the CRDT data

Return: Instance of the GlobalDB initialized or Error

function SetEraseInterruptHookForTest

static void SetEraseInterruptHookForTest(
    std::function< bool(int step)> hook
)

Test-only interruption seam for EraseEntry's compose window.

Parameters:

  • hook invoked after each staging step (1 = after CRDT-entry staging, 2 = after block staging, 3 = after raw-map staging); a true return aborts the erase WITHOUT committing, so the dying batch proves all-or-nothing. Pass nullptr to clear.

Note: No production caller may invoke this method.

function ~GlobalDB

~GlobalDB()

Destructor or GlobalDB.

function Put

outcome::result< CID > Put(
    const HierarchicalKey & key,
    const Buffer & value,
    const std::unordered_set< std::string > & topics
)

Puts key-value pair to the CRDT store, optionally specifying a broadcast topic.

Parameters:

  • key The hierarchical key where the value should be stored.
  • value The value to store.
  • topics Topics to publish to.

Return: outcome::success on success, or outcome::failure otherwise.

function PutConvergentImmutable

outcome::result< CID > PutConvergentImmutable(
    const HierarchicalKey & key,
    const Buffer & value,
    const std::unordered_set< std::string > & topics
)

Writes an authoritative immutable record that converges by SHA-256 content ordering.

function Put

outcome::result< CID > Put(
    const std::vector< DataPair > & data_vector,
    const std::unordered_set< std::string > & topics
)

Writes a batch of CRDT data all at once.

Parameters:

  • data_vector A set of crdt to be written in a single transaction
  • topics Topics to publish to.

Return: outcome::failure on error or success otherwise

function PutLocal

outcome::result< void > PutLocal(
    const HierarchicalKey & key,
    const Buffer & value,
    const std::string & id
)

Puts a key-value pair directly into local storage, bypassing DAG broadcast. See CrdtDatastore::PutKeyLocal for the full rationale/contract.

Parameters:

  • key The hierarchical key where the value should be stored.
  • value The value to store.
  • id Provenance/tie-break identifier for the local write.

Return: outcome::success on success, or outcome::failure otherwise.

function Get

outcome::result< Buffer > Get(
    const HierarchicalKey & key
)

Parameters:

  • key - value key

Return: value as a Buffer

Gets a value that corresponds to specified key.

function Remove

outcome::result< CID > Remove(
    const HierarchicalKey & key,
    const std::unordered_set< std::string > & topics
)

Parameters:

  • key to remove from storage
  • topics Topics to publish to

Return: outcome::failure on error or success otherwise

Removes value for a given key.

function QueryKeyValues

outcome::result< QueryResult > QueryKeyValues(
    std::string_view keyPrefix
)

Parameters:

  • keyPrefix - keys prefix to match. An empty prefix matches any key.

Return: list of key-value pairs matches prefix

Queries CRDT key-value pairs by prefix. If the prefix is empty returns all elements that were not tombstoned

function QueryKeyValues

outcome::result< QueryResult > QueryKeyValues(
    const std::string & prefix_base,
    const std::string & middle_part,
    const std::string & remainder_prefix
)

Queries with a middle part that can be a wildcard, negated string or normal string.

Parameters:

  • prefix_base The base prefix to query
  • middle_part Either a string (normal query), '*' or !string
  • remainder_prefix The remainder part of the query prefix

Return: A list of key value pairs

function KeyToString

outcome::result< std::string > KeyToString(
    const Buffer & key
) const

Parameters:

  • key - binary key to convert

Return: string represenation of a unique key part

Converts a unique key part to a string representation

function BeginTransaction

std::shared_ptr< AtomicTransaction > BeginTransaction()

Return: new transaction

Create a transaction object

function AddBroadcastTopic

outcome::result< void > AddBroadcastTopic(
    const std::string & topicName
)

function AddTopicName

void AddTopicName(
    std::string topicName
)

function AddListenTopic

void AddListenTopic(
    std::string topicName
)

function PrintDataStore

void PrintDataStore()

function GetDataStore

std::shared_ptr< RocksDB > GetDataStore()

function GetRaw

outcome::result< Buffer > GetRaw(
    const Buffer & key
) const

Reads a node-local key straight from RocksDB, bypassing the CRDT layer.

Parameters:

Return: The stored value, or operation_canceled once shutdown released the store.

function PutRaw

outcome::result< void > PutRaw(
    const Buffer & key,
    const Buffer & value
)

Writes a node-local key straight to RocksDB, bypassing the CRDT layer.

Parameters:

  • key Raw key, not a HierarchicalKey; nothing here is replicated.
  • value Value to store.

Return: Failure on I/O error, or operation_canceled once shutdown released the store.

function QueryRaw

outcome::result< QueryResult > QueryRaw(
    const Buffer & key_prefix
) const

Prefix-scans node-local keys straight from RocksDB, bypassing the CRDT layer.

Parameters:

  • key_prefix Raw key prefix.

Return: Matching key/value pairs, or operation_canceled once shutdown released the store.

function EraseEntry

outcome::result< void > EraseEntry(
    const std::string & key,
    const CID & cid,
    const std::optional< std::pair< Buffer, Buffer > > & extra_entry =std::nullopt
)

Generic node-local physical erase: removes, in ONE atomic RocksDB WriteBatch, the CRDT set entries for key (via the CrdtSet owner), the raw block bytes for cid (via the DAGSyncer owner), and the node-local raw bookkeeping entry at key itself.

Parameters:

  • key CRDT element key == raw bookkeeping key, built by the domain owner; never concatenated here.
  • cid CID whose raw block bytes are erased alongside the entries for key.
  • extra_entry OPTIONAL caller-built key+value buffer pair staged into the SAME batch after the raw-map removal and before the single commit (default: none – existing two-arg callers are unchanged). The buffers are OPAQUE: GlobalDB stays generic and never inspects or builds key shapes. The value-XOR-marker atomicity contract: when the caller stages the prune marker here, one commit establishes "value absent AND marker present" together – an interrupted batch leaves BOTH sides in their pre-erase state, never either alone.

Return:

  • ).
  • outcome::success on success, or outcome::failure on a guard rejection, a staging error, or a commit error. Error::POST_COMMIT_ERASE_ANOMALY reports a COMMITTED erase: the batch landed, then the post-commit seal re-check failed. For that value the data IS erased – callers must not treat it as "nothing was erased" (guard rejections, Error::INVALID_PARAMETERS, erase nothing).

This is the ONLY entry to physical erasure. Guards run inside the API and reject fail-closed (Error::INVALID_PARAMETERS) BEFORE any staging: a CID that is the head of ANY topic, the root of a pending/active sync job, LRU-cached, or unresolved is not erased.

Compose critical section (D-01): the ENTIRE compose – guard evaluation, staging, the single commit, and the post-commit seal re-assert – runs inside one owner-vended CrdtSet critical section (CrdtDatastore::MakeComposeLock), the same discipline PutElems already uses across build+commit. The former CHECK-THEN-ACT window (WR-02 – a concurrent PutElems re-creating the erased entries between the under-lock staging and an outside-the-lock commit) is CLOSED by construction: every CrdtSet-mediated write takes the same recursive mutex, so a concurrent PutElems can never interleave between a guard check and the commit. The hold is scoped to this single per-tx compose (acquired as the first act, released after the commit and the post-commit seal re-assert) and is NEVER held across evaluations or multiple composes – D-01 bounded holds, as documented on CrdtSet::MakeComposeLock. The seal marker of cid is never addressed and is re-asserted after the commit – a vanished seal fails loudly with Error::POST_COMMIT_ERASE_ANOMALY (a COMMITTED erase – see Fires NO CRDT hooks and publishes NO tombstones: this is a local prune operation, not a CRDT-logical delete (the network-wide tombstone the CRDT delete path publishes must never fire here).

key addresses both the CRDT element key and the raw map entry; the caller (TransactionManager::EraseTransactionEntry / the future TxPruner) builds domain keys – this facade stays generic.

function GetBroadcaster

std::shared_ptr< sgns::crdt::PubSubBroadcasterExt > GetBroadcaster()

function GetWorkJournal

std::shared_ptr< CRDTWorkJournal > GetWorkJournal() const

function RegisterElementFilter

bool RegisterElementFilter(
    const std::string & pattern,
    GlobalDBFilterCallback filter
)

Parameters:

  • pattern The pattern to match elements against.
  • filter The callback to invoke for matching elements.

Return: true if the filter was successfully registered, false otherwise.

Registers a filter callback for elements matching a pattern.

function RegisterNewElementCallback

bool RegisterNewElementCallback(
    const std::string & pattern,
    GlobalDBNewElementCallback callback
)

Parameters:

  • pattern The pattern to match new elements against.
  • callback The callback to invoke for matching new elements.

Return: true if the callback was successfully registered, false otherwise.

Registers a callback for new elements matching a pattern.

function RegisterDeletedElementCallback

bool RegisterDeletedElementCallback(
    const std::string & pattern,
    GlobalDBDeletedElementCallback callback
)

Parameters:

  • pattern The pattern to match deleted elements against.
  • callback The callback to invoke for matching deleted elements.

Return: true if the callback was successfully registered, false otherwise.

Registers a callback for deleted elements matching a pattern.

function UnregisterElementFilter

void UnregisterElementFilter(
    const std::string & pattern
)

Parameters:

  • pattern The pattern to unregister the filter for.

Unregisters the filter callback for a pattern.

function UnregisterNewElementCallback

void UnregisterNewElementCallback(
    const std::string & pattern
)

Unregisters the new element callback for a pattern.

Parameters:

  • pattern The pattern to unregister the new element callback for.

function UnregisterDeletedElementCallback

void UnregisterDeletedElementCallback(
    const std::string & pattern
)

Unregisters the deleted element callback for a pattern.

Parameters:

  • pattern The pattern to unregister the deleted element callback for.

function Start

void Start()

Starts the GlobalDB instance.

function ShutdownNow

void ShutdownNow()

Immediately quiesce and shut down CRDT intake and workers. Safe to call multiple times.

function StartCIDReceiving

void StartCIDReceiving()

Starts receiving CIDs.

function StartCICSync

void StartCICSync()

Starts CIC synchronization.

function StartRebroadcastHeads

void StartRebroadcastHeads()

Starts rebroadcasting heads.

function GetCRDTHeadList

outcome::result< CRDTHeadListResult > GetCRDTHeadList()

function GetKeyPriority

outcome::result< uint64_t > GetKeyPriority(
    const std::string & key
) const

Parameters:

  • key key string

Return: priority of the key or outcome::failure on error

Read a key's per-key CRDT priority through the CRDT datastore (read-only). Contract mirrors CrdtDatastore::GetPriority / CrdtSet::GetPriority: returns the stored priority minus 1, and a key with NO priority record reads as success with value 0 (the NOT_FOUND case is tolerated) – 0 never implies existence; callers must establish key existence separately.

function GetCRDTHeadHeight

outcome::result< uint64_t > GetCRDTHeadHeight(
    const CID & aCid,
    const std::string & topic
)

function CRDTHeadRemove

outcome::result< void > CRDTHeadRemove(
    const CID & aCid,
    const std::string & topic
)

function CRDTHeadAdd

outcome::result< void > CRDTHeadAdd(
    const CID & aCid,
    const std::string & topic,
    uint64_t priority
)

function GetCIDJobStatus

outcome::result< crdt::CrdtDatastore::JobStatus > GetCIDJobStatus(
    const CID & cid
) const

function RequestHeadBroadcast

outcome::result< void > RequestHeadBroadcast(
    const std::set< std::string > & topics
)

Request head broadcast for specified topics.

Parameters:

  • topics Vector of topic names to broadcast heads for

Return: outcome::success on success, or outcome::failure on error

function GetMonitoredTopics

outcome::result< std::unordered_set< std::string > > GetMonitoredTopics() const

Get the topics that are being listened to.

Return: A set of the monitored topic names

function GetCRDTDataStore

std::shared_ptr< crdt::CrdtDatastore > GetCRDTDataStore()

function GetLocalDeltaKeyValues

outcome::result< std::vector< std::pair< std::string, base::Buffer > > > GetLocalDeltaKeyValues(
    const std::string & cid_string
)

function GetLocalDeltaKeys

outcome::result< std::vector< std::string > > GetLocalDeltaKeys(
    const std::string & cid_string
) const

Parameters:

  • cid_string CID string of the delta block

Return: carried keys (element keys then tombstone keys) or outcome::failure

Enumerate the keys a delta block carries – elements AND tombstones – keys-only, through the CRDT datastore (read-only, parse-at-call-time, strictly local: never fetches from the network, unknown CID fails). Elements-only precedent GetLocalDeltaKeyValues is left untouched.

function HasLocalBlock

outcome::result< bool > HasLocalBlock(
    const std::string & cid_string
) const

Parameters:

  • cid_string CID string of the delta block

Return: true when bytes exist; false when absent; failure on a malformed CID string, an uninitialized DAG syncer, or a store read error (callers fail closed on error)

Byte-level local block presence (read-only): true iff raw block BYTES exist for cid_string in the block store, with NO parse attempted. Unlike the parse-based read paths (and HasBlock), a PRESENT-but-corrupt block answers TRUE here – the erase-side block gate uses this to tell an ABSENT block (already erased – erase-completing evidence) from a corrupt one (fail closed). Delegates to GraphsyncDAGSyncer::HasRawBlockBytes.

Updated on 2026-10-23 at 16:46:18 +0000