Struct Node 

pub struct Node {
    pub metrics: Arc<IpfrsMetrics>,
    /* private fields */
}

IPFRS unified node combining all layers

Fields

metrics: Arc<IpfrsMetrics>

Prometheus metrics for this node instance.

All block operations, DHT calls, inference sessions, GossipSub events, storage stats, and GC runs are recorded here.

Implementations

impl Node

pub fn create_user( &self, username: String, email: Option<String>, roles: HashSet<Role>, ) -> Result<User>

Create a new user (requires auth enabled)

pub fn verify_token(&self, token: &str) -> Result<AuthToken>

Verify an authentication token

pub fn check_permission( &self, token: &AuthToken, permission: Permission, ) -> Result<()>

Check if a token has a specific permission

impl Node

pub async fn add_file(&mut self, path: impl AsRef<Path>) -> Result<Cid>

Add a file from the filesystem

Reads the file, stores it as a block, and announces the CID to the DHT (best-effort — a network error will not cause this method to fail).

pub async fn add_bytes(&mut self, data: impl Into<Bytes>) -> Result<Cid>

Add bytes directly to storage

Stores the data as a block and announces the CID to the DHT (best-effort — a network error will not cause this method to fail).

Uses write-time deduplication: if the block already exists the write is skipped and a debug message is emitted, saving I/O and disk space.

pub async fn add_reader<R>(&mut self, reader: R) -> Result<Cid>

where R: AsyncRead + Unpin,

Add content from an async reader

Reads all data from the provided reader, stores it as a block, and returns the CID. This is useful for streaming data from files, network streams, or other async sources.

Example

use ipfrs::{Node, NodeConfig};
use tokio::fs::File;

let mut node = Node::new(NodeConfig::default())?;
node.start().await?;

let file = File::open("data.bin").await?;
let cid = node.add_reader(file).await?;
println!("Stored with CID: {}", cid);

pub async fn get(&mut self, cid: &Cid) -> Result<Option<Bytes>>

Get content by CID

First checks local storage. If not found locally and the network is available, queries the DHT for providers and attempts to fetch the block from a remote peer (best-effort).

pub async fn get_range( &self, cid: &Cid, offset: usize, length: Option<usize>, ) -> Result<Option<Bytes>>

Get a byte range from content

Retrieves a specific byte range from a block, similar to HTTP 206 Partial Content. This is useful for streaming large files or implementing range requests.

Parameters

• cid: The content identifier
• offset: Starting byte position (0-indexed)
• length: Number of bytes to read (None for all remaining bytes)

Returns

• Ok(Some(bytes)) - The requested byte range
• Ok(None) - Block not found
• Err(_) - Invalid range or other error

Example

use ipfrs::{Node, NodeConfig};

let mut node = Node::new(NodeConfig::default())?;
node.start().await?;

// Get bytes 100-199 (100 bytes starting at offset 100)
if let Some(data) = node.get_range(&cid, 100, Some(100)).await? {
    println!("Retrieved {} bytes", data.len());
}

pub async fn get_to_file( &mut self, cid: &Cid, path: impl AsRef<Path>, ) -> Result<()>

Get content and write to file

pub async fn add_directory(&mut self, dir_path: impl AsRef<Path>) -> Result<Cid>

Add a directory recursively

Traverses a directory tree, stores all files as blocks, and creates a directory structure using IPLD. Returns the root CID.

Directory Structure

Directories are stored as IPLD maps where:

• Keys are file/directory names
• Values are either:
  • Links to file blocks (for files)
  • Nested maps (for subdirectories)

Example

use ipfrs::{Node, NodeConfig};

let mut node = Node::new(NodeConfig::default())?;
node.start().await?;

let root_cid = node.add_directory("/path/to/directory").await?;
println!("Directory stored with root CID: {}", root_cid);

pub async fn get_directory( &mut self, cid: &Cid, output_path: impl AsRef<Path>, ) -> Result<()>

Get a directory and write all files to the filesystem

Retrieves a directory DAG from storage and recreates the directory structure on the filesystem.

Example

use ipfrs::{Node, NodeConfig};

let mut node = Node::new(NodeConfig::default())?;
node.start().await?;

node.get_directory(&dir_cid, "/path/to/output").await?;

pub async fn put_block(&self, block: &Block) -> Result<()>

Store a raw block

pub async fn put_blocks(&self, blocks: &[Block]) -> Result<()>

Store multiple blocks atomically

pub async fn get_block(&self, cid: &Cid) -> Result<Option<Block>>

Retrieve a block by CID

pub async fn get_blocks(&self, cids: &[Cid]) -> Result<Vec<Option<Block>>>

Retrieve multiple blocks

pub async fn has_block(&self, cid: &Cid) -> Result<bool>

Check if a block exists

pub async fn has_blocks(&self, cids: &[Cid]) -> Result<Vec<bool>>

Check if multiple blocks exist

pub async fn delete_block(&self, cid: &Cid) -> Result<()>

Delete a block

pub async fn delete_blocks(&self, cids: &[Cid]) -> Result<()>

Delete multiple blocks

pub async fn block_stat(&self, cid: &Cid) -> Result<Option<BlockStat>>

Get detailed statistics about a block

Returns comprehensive information about a block including its size, CID details, and storage metadata.

Example

use ipfrs::{Node, NodeConfig};

let mut node = Node::new(NodeConfig::default())?;
node.start().await?;

if let Some(stat) = node.block_stat(&cid).await? {
    println!("Block size: {} bytes", stat.size);
    println!("CID: {}", stat.cid);
}

pub async fn block_rm(&self, cid: &Cid) -> Result<()>

Remove a block from storage

Removes a block if it’s safe to do so. This method checks pinning status and refuses to remove pinned blocks to prevent accidental data loss.

Safety

This operation is irreversible. The block will be permanently deleted from storage. Pinned blocks are protected and cannot be removed until unpinned.

Example

use ipfrs::{Node, NodeConfig};

let mut node = Node::new(NodeConfig::default())?;
node.start().await?;

node.block_rm(&cid).await?;
println!("Block removed");

pub fn list_blocks(&self) -> Result<Vec<Cid>>

List all CIDs in storage

pub fn storage_stats(&self) -> Result<StorageStats>

Get storage statistics including deduplication counters

pub fn cache_stats(&self) -> Result<CacheStatsSnapshot>

Return a point-in-time snapshot of the L1 block cache statistics.

This exposes hits, misses, evictions and the derived hit-rate for the in-process LRU cache that wraps the underlying Sled block store.

pub async fn flush(&self) -> Result<()>

Flush pending writes to disk

pub async fn get_block_raw(&self, cid: &Cid) -> Result<Option<Vec<u8>>>

Retrieve the raw bytes of a block by CID.

Returns None when the block is not found locally (no network fallback). This is a low-level counterpart to get() that skips DHT lookup.

pub async fn put_block_raw(&mut self, data: Vec<u8>) -> Result<Cid>

Store raw bytes as a block and return the resulting CID.

The CID is derived from the content using the raw codec and SHA2-256 hash.

impl Node

pub fn new(config: NodeConfig) -> Result<Self>

Create a new IPFRS node

pub async fn start(&mut self) -> Result<()>

Start the IPFRS node

pub async fn stop(&mut self) -> Result<()>

Stop the IPFRS node

pub fn warmup(&self) -> Result<()>

Pre-initialize lazy components for faster first access

By default, semantic router and TensorLogic store are initialized lazily on first use. This method forces their initialization upfront, which can be useful for:

• Warmup scenarios where you want predictable latency
• Load testing where you want to measure steady-state performance
• Detecting configuration errors early at startup

Example

use ipfrs::{Node, NodeConfig};

let mut node = Node::new(NodeConfig::default())?;
node.start().await?;

// Pre-initialize all components for faster first access
node.warmup()?;

// Now semantic and tensorlogic calls will be instant

pub fn status(&self) -> NodeStatus

Get node status

pub async fn diagnostics(&self) -> Result<NodeDiagnostics>

Get comprehensive node diagnostics

Collects detailed diagnostic information about node health, resource usage, and performance. This is useful for monitoring, troubleshooting, and optimization.

Example

use ipfrs::{Node, NodeConfig, DiagnosticAnalyzer};

let mut node = Node::new(NodeConfig::default())?;
node.start().await?;

// Get diagnostics
let diagnostics = node.diagnostics().await?;

// Analyze and get recommendations
let recommendations = DiagnosticAnalyzer::analyze(&diagnostics);
for rec in recommendations {
    println!("{:?}: {}", rec.severity, rec.message);
}

// Or get a human-readable report
let report = DiagnosticAnalyzer::health_report(&diagnostics);
println!("{}", report);

pub fn is_semantic_enabled(&self) -> bool

Check if semantic routing is enabled

Returns true if the semantic router is configured to be enabled. Note: The router will be lazily initialized on first use.

Example

use ipfrs::{Node, NodeConfig};

let mut node = Node::new(NodeConfig::default())?;
node.start().await?;

if node.is_semantic_enabled() {
    println!("Semantic search is available");
}

pub fn is_tensorlogic_enabled(&self) -> bool

Check if TensorLogic is enabled

Returns true if the TensorLogic store is configured to be enabled. Note: The store will be lazily initialized on first use.

Example

use ipfrs::{Node, NodeConfig};

let mut node = Node::new(NodeConfig::default())?;
node.start().await?;

if node.is_tensorlogic_enabled() {
    println!("Logic programming is available");
}

pub fn is_running(&self) -> bool

Check if the node is running

Returns true if the node has been started and the network component is active.

Example

use ipfrs::{Node, NodeConfig};

let mut node = Node::new(NodeConfig::default())?;

assert!(!node.is_running());

node.start().await?;
assert!(node.is_running());

pub fn is_semantic_initialized(&self) -> bool

Check if semantic router has been initialized

Returns true if the semantic router has been lazily initialized. This is different from is_semantic_enabled() which checks if it’s configured.

pub fn is_tensorlogic_initialized(&self) -> bool

Check if TensorLogic store has been initialized

Returns true if the TensorLogic store has been lazily initialized. This is different from is_tensorlogic_enabled() which checks if it’s configured.

pub fn auth_manager(&self) -> Result<Arc<AuthManager>>

Get the authentication manager if enabled

pub fn is_auth_enabled(&self) -> bool

Check if authentication is enabled

pub fn tls_manager(&self) -> Result<Arc<TlsManager>>

Get the TLS manager if enabled

pub fn is_tls_enabled(&self) -> bool

Check if TLS is enabled

impl Node

pub async fn dag_put(&self, data: Ipld) -> Result<Cid>

Store an IPLD DAG node

Serializes the IPLD data structure, stores it as a block, and returns the CID. This is useful for storing structured data with links to other blocks.

Example

use ipfrs::{Node, NodeConfig};
use ipfrs_core::Ipld;
use std::collections::BTreeMap;

let mut node = Node::new(NodeConfig::default())?;
node.start().await?;

let mut map = BTreeMap::new();
map.insert("name".to_string(), Ipld::String("Alice".to_string()));
map.insert("age".to_string(), Ipld::Integer(30));

let cid = node.dag_put(Ipld::Map(map)).await?;

pub async fn dag_get(&self, cid: &Cid) -> Result<Option<Ipld>>

Retrieve an IPLD DAG node

Fetches a block by CID and deserializes it as IPLD.

Example

use ipfrs::{Node, NodeConfig};

let mut node = Node::new(NodeConfig::default())?;
node.start().await?;

if let Some(data) = node.dag_get(&cid).await? {
    println!("Retrieved DAG node: {:?}", data);
}

pub async fn dag_resolve(&self, root: &Cid, path: &str) -> Result<Option<Cid>>

Resolve an IPLD path

Navigates through IPLD structures following a path like “/key1/key2/0”. Returns the CID if the path leads to a link.

Path Format

• Map keys: “/key”
• List indices: “/0”, “/1”, etc.
• Nested: “/map_key/0/nested_key”

Example

use ipfrs::{Node, NodeConfig};

let mut node = Node::new(NodeConfig::default())?;
node.start().await?;

// Resolve path: /users/0/profile
if let Some(cid) = node.dag_resolve(&root_cid, "/users/0/profile").await? {
    println!("Resolved to CID: {}", cid);
}

pub async fn dag_traverse( &self, root: &Cid, max_depth: Option<usize>, ) -> Result<Vec<Cid>>

Traverse a DAG and collect all reachable CIDs

Performs a breadth-first traversal starting from the root CID, following all links and collecting all reachable CIDs.

Parameters

• root: The starting CID
• max_depth: Maximum depth to traverse (None for unlimited)

Example

use ipfrs::{Node, NodeConfig};

let mut node = Node::new(NodeConfig::default())?;
node.start().await?;

// Traverse up to depth 10
let cids = node.dag_traverse(&root_cid, Some(10)).await?;
println!("Found {} reachable blocks", cids.len());

pub async fn dag_export( &self, root: &Cid, path: impl AsRef<Path>, ) -> Result<DagExportStats>

Export a DAG to CAR (Content Addressable aRchive) format

pub async fn dag_import(&self, path: impl AsRef<Path>) -> Result<DagImportStats>

Import blocks from a CAR file

impl Node

pub fn peer_id(&self) -> Result<String>

Get local peer ID

pub async fn peers(&self) -> Result<Vec<String>>

Get connected peers

pub async fn connect(&mut self, addr: &str) -> Result<()>

Connect to a peer

pub async fn disconnect(&mut self, peer_id: &str) -> Result<()>

Disconnect from a peer

pub async fn provide(&mut self, cid: &Cid) -> Result<()>

Announce content to DHT (provide)

pub async fn find_providers(&mut self, cid: &Cid) -> Result<Vec<PeerId>>

Find providers for content in DHT.

Queries the Kademlia DHT and waits up to 30 seconds for provider results. Returns the peer IDs of nodes that have announced themselves as providers for the given CID.

pub async fn find_providers_timeout( &mut self, cid: &Cid, timeout: Duration, ) -> Result<Vec<PeerId>>

Find providers for content with a custom timeout.

Like find_providers but lets the caller specify how long to wait.

pub fn network_stats(&self) -> Result<NetworkStats>

Get network statistics

pub fn bitswap_stats(&self) -> Result<BitswapStats>

Get bitswap statistics

pub async fn ping(&mut self, _peer_id: &str) -> Result<()>

Ping a peer (placeholder implementation)

pub async fn find_peer(&mut self, peer_id: &str) -> Result<Vec<String>>

Find a peer’s addresses in the DHT (placeholder implementation)

pub fn bootstrap_peers(&self) -> Result<Vec<String>>

Get bootstrap peers

pub async fn add_bootstrap_peer(&mut self, addr: &str) -> Result<()>

Add a bootstrap peer

pub async fn remove_bootstrap_peer(&mut self, _addr: &str) -> Result<()>

Remove a bootstrap peer

impl Node

pub async fn pin_add( &self, cid: &Cid, recursive: bool, name: Option<String>, ) -> Result<()>

Pin a block (prevent it from being garbage collected)

pub async fn pin_rm(&self, cid: &Cid, recursive: bool) -> Result<()>

Unpin a block

pub fn pin_ls(&self) -> Result<Vec<PinInfo>>

List all pinned blocks

pub async fn pin_verify(&self) -> Result<Vec<(Cid, bool)>>

Verify all pins are available

pub async fn pin_save(&self, path: impl AsRef<Path>) -> Result<()>

Save pin index to disk

pub async fn pin_load(&self, path: impl AsRef<Path>) -> Result<()>

Load pin index from disk

impl Node

pub async fn repo_gc(&self, dry_run: bool) -> Result<GcResult>

Run garbage collection

pub async fn gc_blocks( &self, dry_run: bool, min_age_secs: u64, ) -> Result<OrphanGcResult>

Run orphan block garbage collection using OrphanGarbageCollector.

Collects blocks that are present in the store but are not referenced by any pinned CID. When dry_run is true the orphans are reported but not actually deleted. The min_age_secs parameter is stored in the config for future use when per-block timestamps become available.

pub async fn repo_gc_with_options( &self, dry_run: bool, min_age_secs: u64, ) -> Result<GcResult>

Run garbage collection with explicit age filter.

min_age_secs – blocks younger than this many seconds are spared (passed through to the underlying GC config).

pub async fn repo_fsck(&self) -> Result<FsckResult>

Verify repository integrity

pub async fn repo_fsck_quick(&self) -> Result<FsckResult>

Run quick filesystem check (only verify CIDs match content)

pub fn gc_stats(&self) -> Result<(usize, usize)>

Get garbage collection statistics (without running GC)

pub async fn repo_stat(&self) -> Result<RepoStats>

Get comprehensive repository statistics

pub async fn block_distribution(&self) -> Result<BlockDistribution>

Get block size distribution

pub async fn find_duplicates(&self) -> Result<Vec<Vec<Cid>>>

Find duplicate blocks (same content, different CIDs)

pub async fn largest_blocks(&self, limit: usize) -> Result<Vec<(Cid, u64)>>

Get largest blocks in the repository

pub async fn find_orphaned(&self) -> Result<Vec<Cid>>

Find orphaned blocks (not pinned and not referenced)

impl Node

pub fn semantic_stats(&self) -> Result<SemanticStats>

Get semantic router statistics

Returns comprehensive statistics about the semantic index including vector count, dimension, distance metric, and cache performance.

Returns

Statistics about the semantic router

Errors

Returns error if semantic routing is not enabled

Example

use ipfrs::{Node, NodeConfig};

let mut node = Node::new(NodeConfig::default())?;
node.start().await?;

let stats = node.semantic_stats()?;
println!("Indexed vectors: {}", stats.num_vectors);
println!("Vector dimension: {}", stats.dimension);
println!("Cache size: {}/{}", stats.cache_size, stats.cache_capacity);

pub async fn index_content(&self, cid: &Cid, embedding: &[f32]) -> Result<()>

Index content with its semantic embedding

Adds content to the semantic index for similarity search. The embedding should be a vector representation of the content (e.g., from a sentence transformer model).

Arguments

• cid - Content identifier to index
• embedding - Vector embedding of the content

Example

use ipfrs::{Node, NodeConfig};

let mut node = Node::new(NodeConfig::default())?;
node.start().await?;

// Index content with 768-dimensional embedding (e.g., from BERT)
let embedding = vec![0.5; 768];
node.index_content(&cid, &embedding).await?;

pub async fn search_similar( &self, query_embedding: &[f32], k: usize, ) -> Result<Vec<SearchResult>>

Search for similar content by semantic similarity

Performs k-nearest neighbor search over indexed content using vector similarity. Returns the top k most similar items.

Arguments

• query_embedding - Query vector to search for
• k - Number of results to return

Returns

Vector of search results ordered by similarity (highest first)

Example

use ipfrs::{Node, NodeConfig};

let mut node = Node::new(NodeConfig::default())?;
node.start().await?;

// Search for top 10 similar documents
let query_embedding = vec![0.3; 768];
let results = node.search_similar(&query_embedding, 10).await?;

for result in results {
    println!("CID: {}, Score: {}", result.cid, result.score);
}

pub async fn search_hybrid( &self, query_embedding: &[f32], k: usize, filter: QueryFilter, ) -> Result<Vec<SearchResult>>

Search with advanced filtering options

Performs semantic search with additional filters like minimum score threshold, CID prefix matching, and result limits.

Arguments

• query_embedding - Query vector to search for
• k - Number of results to return
• filter - Query filter options

Returns

Vector of filtered search results ordered by similarity

Example

use ipfrs::{Node, NodeConfig, QueryFilter};

let mut node = Node::new(NodeConfig::default())?;
node.start().await?;

// Search with filters
let query_embedding = vec![0.3; 768];
let filter = QueryFilter {
    min_score: Some(0.8),  // Only results with score >= 0.8
    max_score: None,        // No max score filter
    max_results: Some(5),   // Limit to 5 results
    cid_prefix: None,       // No CID filtering
};

let results = node.search_hybrid(&query_embedding, 20, filter).await?;

for result in results {
    println!("High-confidence match: {} ({})", result.cid, result.score);
}

pub async fn save_semantic_index(&self, path: impl AsRef<Path>) -> Result<()>

Save the semantic index to disk

Persists the entire HNSW index including all vectors and CID mappings to a file for later loading.

Arguments

• path - Path to save the index file

Example

use ipfrs::{Node, NodeConfig};

let mut node = Node::new(NodeConfig::default())?;
node.start().await?;

// Save the semantic index
node.save_semantic_index("semantic.index").await?;
println!("Semantic index saved");

pub async fn load_semantic_index(&self, path: impl AsRef<Path>) -> Result<()>

Load a semantic index from disk

Loads a previously saved HNSW index from disk, replacing the current index.

Arguments

• path - Path to the saved index file

Example

use ipfrs::{Node, NodeConfig};

let mut node = Node::new(NodeConfig::default())?;
node.start().await?;

// Load the semantic index
node.load_semantic_index("semantic.index").await?;
println!("Semantic index loaded");

impl Node

pub fn tensorlogic_stats(&self) -> Result<TensorLogicStats>

Get TensorLogic statistics

Returns information about the TensorLogic store including counts of stored terms, predicates, and rules.

Returns

Statistics about TensorLogic storage

Errors

Returns error if TensorLogic is not enabled

Example

use ipfrs::{Node, NodeConfig};

let mut node = Node::new(NodeConfig::default())?;
node.start().await?;

let stats = node.tensorlogic_stats()?;
println!("TensorLogic enabled: {}", stats.enabled);

pub async fn put_term(&self, term: &Term) -> Result<Cid>

Store a logical term

Serializes and stores a TensorLogic term as a content-addressed block. Terms can be constants, variables, functions, or references to other CIDs.

Arguments

• term - The logical term to store

Returns

CID of the stored term

Example

use ipfrs::{Node, NodeConfig, Term, Constant};

let mut node = Node::new(NodeConfig::default())?;
node.start().await?;

// Store a constant term
let term = Term::Const(Constant::String("Alice".to_string()));
let cid = node.put_term(&term).await?;
println!("Stored term with CID: {}", cid);

pub async fn get_term(&self, cid: &Cid) -> Result<Option<Term>>

Retrieve a logical term by CID

Fetches and deserializes a TensorLogic term from storage.

Arguments

• cid - Content identifier of the term

Returns

The term if found, None otherwise

Example

use ipfrs::{Node, NodeConfig};

let mut node = Node::new(NodeConfig::default())?;
node.start().await?;

if let Some(term) = node.get_term(&cid).await? {
    println!("Retrieved term: {}", term);
}

pub async fn store_predicate(&self, predicate: &Predicate) -> Result<Cid>

Store a logical predicate

Stores a predicate (named relation with arguments) as a content-addressed block.

Arguments

• predicate - The predicate to store

Returns

CID of the stored predicate

Example

use ipfrs::{Node, NodeConfig, Predicate, Term, Constant};

let mut node = Node::new(NodeConfig::default())?;
node.start().await?;

// Store a predicate: parent("Alice", "Bob")
let predicate = Predicate::new(
    "parent".to_string(),
    vec![
        Term::Const(Constant::String("Alice".to_string())),
        Term::Const(Constant::String("Bob".to_string())),
    ],
);

let cid = node.store_predicate(&predicate).await?;
println!("Stored predicate with CID: {}", cid);

pub async fn get_predicate(&self, cid: &Cid) -> Result<Option<Predicate>>

Retrieve a logical predicate by CID

Fetches and deserializes a predicate from storage.

Arguments

• cid - Content identifier of the predicate

Returns

The predicate if found, None otherwise

Example

use ipfrs::{Node, NodeConfig};

let mut node = Node::new(NodeConfig::default())?;
node.start().await?;

if let Some(predicate) = node.get_predicate(&cid).await? {
    println!("Retrieved predicate: {}", predicate);
}

pub async fn store_rule(&self, rule: &Rule) -> Result<Cid>

Store a logical rule

Stores a Horn clause (head :- body) as a content-addressed block.

Arguments

• rule - The rule to store

Returns

CID of the stored rule

Example

use ipfrs::{Node, NodeConfig, Rule, Predicate, Term, Constant};

let mut node = Node::new(NodeConfig::default())?;
node.start().await?;

// Store a fact: parent("Alice", "Bob")
let fact = Rule::fact(Predicate::new(
    "parent".to_string(),
    vec![
        Term::Const(Constant::String("Alice".to_string())),
        Term::Const(Constant::String("Bob".to_string())),
    ],
));

let cid = node.store_rule(&fact).await?;
println!("Stored rule with CID: {}", cid);

pub async fn get_rule(&self, cid: &Cid) -> Result<Option<Rule>>

Retrieve a logical rule by CID

Fetches and deserializes a rule from storage.

Arguments

• cid - Content identifier of the rule

Returns

The rule if found, None otherwise

Example

use ipfrs::{Node, NodeConfig};

let mut node = Node::new(NodeConfig::default())?;
node.start().await?;

if let Some(rule) = node.get_rule(&cid).await? {
    println!("Retrieved rule: head={}, body_len={}", rule.head, rule.body.len());
}

pub fn add_fact(&self, fact: Predicate) -> Result<()>

Add a fact to the knowledge base

Adds a logical fact (predicate with no body) to the in-memory knowledge base. Facts are used during inference queries.

Arguments

• fact - The predicate to add as a fact

Example

use ipfrs::{Node, NodeConfig, Predicate, Term, Constant};

let mut node = Node::new(NodeConfig::default())?;
node.start().await?;

// Add fact: parent("Alice", "Bob")
let fact = Predicate::new(
    "parent".to_string(),
    vec![
        Term::Const(Constant::String("Alice".to_string())),
        Term::Const(Constant::String("Bob".to_string())),
    ],
);

node.add_fact(fact)?;

pub fn add_rule(&self, rule: Rule) -> Result<()>

Add a rule to the knowledge base

Adds a logical rule (Horn clause) to the in-memory knowledge base. Rules are used during inference queries.

Arguments

• rule - The rule to add

Example

use ipfrs::{Node, NodeConfig, Rule, Predicate, Term};

let mut node = Node::new(NodeConfig::default())?;
node.start().await?;

// Add rule: grandparent(X, Z) :- parent(X, Y), parent(Y, Z)
let head = Predicate::new("grandparent".to_string(), vec![
    Term::Var("X".to_string()),
    Term::Var("Z".to_string()),
]);
let body = vec![
    Predicate::new("parent".to_string(), vec![
        Term::Var("X".to_string()),
        Term::Var("Y".to_string()),
    ]),
    Predicate::new("parent".to_string(), vec![
        Term::Var("Y".to_string()),
        Term::Var("Z".to_string()),
    ]),
];

node.add_rule(Rule::new(head, body))?;

pub fn infer(&self, goal: &Predicate) -> Result<Vec<Substitution>>

Run inference query

Executes a logical query using backward chaining inference. Returns all variable substitutions that satisfy the goal.

Arguments

• goal - The query predicate to prove

Returns

Vector of variable substitutions (bindings) that satisfy the goal

Example

use ipfrs::{Node, NodeConfig, Predicate, Term, Constant};

let mut node = Node::new(NodeConfig::default())?;
node.start().await?;

// Add some facts
node.add_fact(Predicate::new("parent".to_string(), vec![
    Term::Const(Constant::String("Alice".to_string())),
    Term::Const(Constant::String("Bob".to_string())),
]))?;

// Query: parent("Alice", X)?
let goal = Predicate::new("parent".to_string(), vec![
    Term::Const(Constant::String("Alice".to_string())),
    Term::Var("X".to_string()),
]);

let solutions = node.infer(&goal)?;
for solution in solutions {
    println!("Solution: {:?}", solution);
}

pub fn prove(&self, goal: &Predicate) -> Result<Option<Proof>>

Generate proof tree

Constructs a formal proof for a given goal predicate using backward chaining. Returns the proof if one can be found, None otherwise.

Arguments

• goal - The goal to prove

Returns

Proof object if goal can be proven, None otherwise

Example

use ipfrs::{Node, NodeConfig, Predicate, Term, Constant};

let mut node = Node::new(NodeConfig::default())?;
node.start().await?;

// Add some facts
node.add_fact(Predicate::new("parent".to_string(), vec![
    Term::Const(Constant::String("Alice".to_string())),
    Term::Const(Constant::String("Bob".to_string())),
]))?;

// Generate proof
let goal = Predicate::new("parent".to_string(), vec![
    Term::Const(Constant::String("Alice".to_string())),
    Term::Const(Constant::String("Bob".to_string())),
]);

if let Some(proof) = node.prove(&goal)? {
    println!("Proof found: {:?}", proof);
}

pub async fn store_proof(&self, proof: &Proof) -> Result<Cid>

Store a proof and return its CID

Serializes and stores a proof tree as a content-addressed block.

Arguments

• proof - The proof to store

Returns

CID of the stored proof

Example

use ipfrs::{Node, NodeConfig, Predicate, Term, Constant};

let mut node = Node::new(NodeConfig::default())?;
node.start().await?;

// Add fact and generate proof
node.add_fact(Predicate::new("parent".to_string(), vec![
    Term::Const(Constant::String("Alice".to_string())),
    Term::Const(Constant::String("Bob".to_string())),
]))?;

let goal = Predicate::new("parent".to_string(), vec![
    Term::Const(Constant::String("Alice".to_string())),
    Term::Const(Constant::String("Bob".to_string())),
]);

if let Some(proof) = node.prove(&goal)? {
    let cid = node.store_proof(&proof).await?;
    println!("Proof stored with CID: {}", cid);
}

pub async fn get_proof(&self, cid: &Cid) -> Result<Option<Proof>>

Retrieve a proof by CID

Fetches and deserializes a proof from storage.

Arguments

• cid - Content identifier of the proof

Returns

The proof if found, None otherwise

Example

use ipfrs::{Node, NodeConfig};

let mut node = Node::new(NodeConfig::default())?;
node.start().await?;

if let Some(proof) = node.get_proof(&cid).await? {
    println!("Retrieved proof for goal: {}", proof.goal);
}

pub fn verify_proof(&self, proof: &Proof) -> Result<bool>

Verify a proof against the current knowledge base

Checks if a proof tree is valid by verifying that:

• All facts exist in the knowledge base
• All rules exist and are correctly applied
• All subproofs are valid

Arguments

• proof - The proof to verify

Returns

true if the proof is valid, false otherwise

Example

use ipfrs::{Node, NodeConfig, Predicate, Term, Constant};

let mut node = Node::new(NodeConfig::default())?;
node.start().await?;

// Add fact
node.add_fact(Predicate::new("parent".to_string(), vec![
    Term::Const(Constant::String("Alice".to_string())),
    Term::Const(Constant::String("Bob".to_string())),
]))?;

// Generate and verify proof
let goal = Predicate::new("parent".to_string(), vec![
    Term::Const(Constant::String("Alice".to_string())),
    Term::Const(Constant::String("Bob".to_string())),
]);

if let Some(proof) = node.prove(&goal)? {
    let is_valid = node.verify_proof(&proof)?;
    println!("Proof is valid: {}", is_valid);
}

pub async fn distributed_infer( &mut self, goal: &str, max_depth: usize, timeout: Duration, ) -> Result<DistributedInferResult>

Run distributed inference across connected peers.

Executes a two-phase query:

1. Local fast-path – runs backward-chaining inference against the node’s own knowledge base. If at least one solution is found the result is returned immediately without contacting any peers.

2. Distributed session – when local inference yields no results and the node has an active network, an InferenceRequest is serialised as JSON and published to the INFERENCE_REQUEST GossipSub topic. A DistributedReasonerV2 session is opened and the node waits up to timeout collecting InferenceResponse messages. Each received response is folded into the session as a RemoteResult.

Arguments

• goal – Datalog goal string, e.g. "parent(alice, X)".
• max_depth – Maximum backward-chaining depth for both local and remote engines.
• timeout – Maximum wall-clock time to wait for remote responses.

Errors

Returns an error only when TensorLogic is not enabled in the node configuration. Network errors are swallowed and result in an empty remote_bindings list.

pub fn make_inference_request( &self, goal: &str, max_depth: usize, ) -> InferenceRequest

Construct an InferenceRequest ready to be serialised and sent to a remote peer.

The request_id is a freshly generated UUID v4.

pub async fn publish_rule(&self, rule: &Rule) -> Result<Cid>

Store a rule as a content-addressed block and return its CID.

The rule is encoded using the DAG-CBOR IPLD codec and stored in the local block store. Identical rules always yield the same CID, enabling deduplication across the network. After storing, the CID is provided to the DHT automatically (best-effort) via the existing put_block path.

Arguments

• rule – The logical rule to publish.

Returns

The content-addressed CID of the stored rule block.

pub async fn fetch_rule(&self, cid: &Cid) -> Result<Rule>

Fetch a rule by its CID from local storage or the network.

First checks local block storage, then attempts DHT provider discovery for remote fetching (best-effort). The block bytes are decoded back into an IR Rule using the IPLD codec.

Arguments

• cid – Content identifier of the rule block.

Returns

The decoded rule.

Errors

Returns an error if the block cannot be found locally or on the network, or if the block bytes cannot be decoded as a rule.

pub async fn import_rules_from_cids(&self, cids: &[Cid]) -> Result<usize>

Import all rules from a CID list into the local knowledge base.

Iterates over the provided CIDs, fetching each rule (from local storage or the network) and asserting it into the in-memory knowledge base. Rules that fail to fetch are logged as warnings and skipped; the count of successfully imported rules is returned.

Arguments

• cids – Slice of CIDs referencing rule blocks to import.

Returns

Number of rules successfully fetched and added to the knowledge base.

pub fn kb_stats(&self) -> Result<KnowledgeBaseStats>

Get knowledge base statistics

Returns statistics about the in-memory knowledge base including counts of facts and rules.

Returns

Knowledge base statistics

Example

use ipfrs::{Node, NodeConfig};

let mut node = Node::new(NodeConfig::default())?;
node.start().await?;

let stats = node.kb_stats()?;
println!("Facts: {}, Rules: {}", stats.num_facts, stats.num_rules);

pub async fn save_knowledge_base(&self, path: impl AsRef<Path>) -> Result<()>

Save the knowledge base to disk

Persists the entire knowledge base (facts and rules) to a file for later loading.

Arguments

• path - Path to save the knowledge base file

Example

use ipfrs::{Node, NodeConfig};

let mut node = Node::new(NodeConfig::default())?;
node.start().await?;

// Save the knowledge base
node.save_knowledge_base("knowledge.kb").await?;
println!("Knowledge base saved");

pub async fn load_knowledge_base(&self, path: impl AsRef<Path>) -> Result<()>

Load a knowledge base from disk

Loads a previously saved knowledge base from disk, replacing the current KB.

Arguments

• path - Path to the saved knowledge base file

Example

use ipfrs::{Node, NodeConfig};

let mut node = Node::new(NodeConfig::default())?;
node.start().await?;

// Load the knowledge base
node.load_knowledge_base("knowledge.kb").await?;
println!("Knowledge base loaded");

pub async fn infer_streaming( &self, goal_str: &str, timeout_secs: u64, ) -> Result<InferenceResultStream>

Start a distributed inference session and return a stream of partial results arriving incrementally as peers respond.

This is the streaming counterpart of distributed_infer. Instead of waiting for all peers to reply before returning, it opens a channel and delivers each peer’s contribution as a PartialResult the moment it arrives.

Arguments

• goal_str – Datalog goal string, e.g. "parent(alice, X)".
• timeout_secs – Hard deadline in seconds; no more results will be delivered after this deadline.

Returns

An InferenceResultStream the caller can poll with InferenceResultStream::next_partial.

Errors

Returns an error when TensorLogic is not enabled.

pub async fn prove_distributed( &self, goal_str: &str, max_depth: usize, ) -> Result<ProofTree>

Prove a goal using distributed backward chaining across DHT peers.

Parses goal_str, snapshots the local knowledge base, then runs DistributedBackwardChainer::prove_with_tree. When the node has an active network the chainer may delegate unresolved sub-goals to DHT peers; otherwise it degrades to local-only backward chaining.

Arguments

• goal_str - Datalog goal string, e.g. "parent(alice, X)".
• max_depth - Maximum backward-chaining depth.

Returns

A ProofTree with is_complete = true when the goal was fully proved.

pub async fn accumulate_gradients( &self, local_grad: Vec<f32>, min_peers: usize, timeout_secs: u64, ) -> Result<Vec<f32>>

Run a distributed gradient accumulation round.

Serialises local_grad as an Arrow IPC block, stores it in the local content-addressed store, and broadcasts its CID to all peers via the GRADIENT_SYNC GossipSub topic.

After broadcasting, the method waits up to timeout_secs seconds for peer gradient CIDs to be delivered through the caller-supplied channel receiver peer_cid_rx. Each CID is fetched from the block store, decoded from Arrow IPC, and accumulated. Once at least min_peers peer gradients have been collected, FedAvg is applied.

If storage is not available the function falls back to returning the local gradient unchanged, so callers never get a hard error during unit tests without a running network.

Arguments

• local_grad — The gradient computed by this node
• peer_cid_rx — Receiver for (peer_id, cid_string) pairs delivered by the network event loop when GRADIENT_SYNC messages arrive. Pass None to skip peer collection.
• min_peers — Minimum number of peer responses needed before aggregating
• timeout_secs — Maximum wall-clock seconds to wait for peers

Returns

The FedAvg of local + peer gradients, or just the local gradient when no peers responded within the timeout.

pub async fn stream_gradient_to_peer( &self, peer_id: &str, gradient: Vec<f32>, chunk_size: usize, ) -> Result<usize>

Stream a gradient to a specific peer via TensorSwap.

Encodes gradient into Arrow IPC chunks of chunk_size elements each, then streams them over the TensorSwap protocol to peer_id.

Stubbing note

Full peer-to-peer transport requires a live network connection managed by the ipfrs-network layer. When no live connection to peer_id is available the method still performs the complete encoding path (chunking, Arrow IPC serialisation, CRC-32 computation) and returns the chunk count as Ok(n_chunks). This ensures the encoding/validation logic is always exercised, even in integration tests that run without a real network.

pub async fn receive_gradient_from_peer( &self, peer_id: &str, timeout_secs: u64, ) -> Result<Vec<f32>>

Receive a gradient stream from a peer.

Awaits a gradient streamed by peer_id over TensorSwap and reassembles the chunks back into a flat Vec<f32>.

Stubbing note

When no live connection to peer_id is available (e.g., in unit tests) the method logs the attempt and returns Ok(vec![]) immediately after timeout_secs seconds without blocking. In production the method would block until the full stream is received or the timeout expires.

pub async fn import_remote_kb(&self, cid_str: &str) -> Result<KbMergeDiff>

Fetch a remote knowledge base snapshot identified by cid_str and merge it into the local knowledge base using content-hash deduplication.

Arguments

• cid_str – CID string of the root KnowledgeBaseIpld block.

Returns

A KbMergeDiff summarising how many facts/rules were added, skipped, or conflicting.

Errors

Returns an error if the CID is malformed, any block is missing, decoding fails, or TensorLogic / storage is not enabled.

pub async fn export_kb(&self) -> Result<Cid>

Serialize the local knowledge base as an IPLD block DAG and return its root CID.

The returned CID can be shared with peers who will use import_remote_kb to fetch and merge the KB.

Errors

Returns an error if TensorLogic / storage is not enabled or if block encoding fails.