Merge branch 'main' into docs

Author: lukvmil

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "koi-net"
-version = "1.3.0-beta.9"
+version = "1.3.2"
 description = "Implementation of KOI-net protocol in Python"
 maintainers = [
     { name = "Luke Miller", email = "luke@block.science" }

src/koi_net/components/cache.py

@@ -3,6 +3,7 @@
 from pathlib import Path
 from dataclasses import dataclass
 
+from pydantic import ValidationError
 from rid_lib.core import RID, RIDType
 from rid_lib.ext import Bundle
 from rid_lib.ext.utils import b64_encode, b64_decode
@@ -50,7 +51,16 @@ def read(self, rid: RID) -> Bundle | None:
                 mode="r",
                 encoding="utf-8"
             ) as f:
-                return Bundle.model_validate_json(f.read())
+                file_content = f.read()
+            
+            if file_content == "":
+                return None
+
+            try:
+                return Bundle.model_validate_json(file_content)
+            except ValidationError:
+                return None
+            
         except FileNotFoundError:
             return None
 

src/koi_net/components/knowledge_handlers/forget_node_handler.py

@@ -1,20 +1,18 @@
 from dataclasses import dataclass
 
-from rid_lib.types import KoiNetNode
+from rid_lib.types import KoiNetEdge, KoiNetNode
 
 from koi_net.protocol.edge import EdgeProfile
 from koi_net.protocol.knowledge_object import KnowledgeObject
 from koi_net.protocol.event import EventType
 from ..interfaces import KnowledgeHandler, HandlerType
 from ..kobj_queue import KobjQueue
-from ..graph import NetworkGraph
 from ..cache import Cache
 
 
 @dataclass
 class ForgetNodeHandler(KnowledgeHandler):
     cache: Cache
-    graph: NetworkGraph
     kobj_queue: KobjQueue
 
     handler_type = HandlerType.Final
@@ -26,10 +24,11 @@ def handle(self, kobj: KnowledgeObject):
         if kobj.normalized_event_type != EventType.FORGET:
             return
 
-        for edge_rid in self.graph.get_edges():
+        for edge_rid in self.cache.list_rids(rid_types=[KoiNetEdge]):
             edge_bundle = self.cache.read(edge_rid)
             if not edge_bundle:
                 continue
+            
             edge_profile = edge_bundle.validate_contents(EdgeProfile)
 
             if kobj.rid in (edge_profile.source, edge_profile.target):

src/koi_net/components/knowledge_handlers/node_contact_handler.py

@@ -5,6 +5,7 @@
 
 from koi_net.exceptions import RequestError
 from koi_net.config.base import BaseNodeConfig
+from koi_net.infra import depends_on
 from koi_net.protocol.node import NodeProfile, NodeType
 from koi_net.protocol.edge import EdgeProfile, EdgeStatus, EdgeType, generate_edge_bundle
 from koi_net.protocol.knowledge_object import KnowledgeObject
@@ -28,19 +29,12 @@ class NodeContactHandler(KnowledgeHandler):
     handler_type = HandlerType.Network
     rid_types = (KoiNetNode,)
 
-    def handle(self, kobj: KnowledgeObject):
-        """Makes contact with providers of RID types of interest.
-        
-        When an incoming node knowledge object is identified as a provider
-        of an RID type of interest, this handler will propose a new edge 
-        subscribing to future node events, and fetch existing nodes to catch 
-        up to the current state.
-        """
+    def process_node(self, node_rid: KoiNetNode, node_bundle: Bundle):
         # prevents nodes from attempting to form a self loop
-        if kobj.rid == self.identity.rid:
+        if node_rid == self.identity.rid:
             return
 
-        node_profile = kobj.bundle.validate_contents(NodeProfile)
+        node_profile = node_bundle.validate_contents(NodeProfile)
 
         available_rid_types = list(
             set(self.config.koi_net.rid_types_of_interest) & 
@@ -51,7 +45,7 @@ def handle(self, kobj: KnowledgeObject):
             return
 
         edge_rid = self.graph.get_edge(
-            source=kobj.rid,
+            source=node_rid,
             target=self.identity.rid,
         )
 
@@ -74,7 +68,7 @@ def handle(self, kobj: KnowledgeObject):
         else:
             self.log.info(f"Proposing new edge with node provider {available_rid_types}")
             edge_bundle = generate_edge_bundle(
-                source=kobj.rid,
+                source=node_rid,
                 target=self.identity.rid,
                 rid_types=available_rid_types,
                 edge_type=(
@@ -90,7 +84,7 @@ def handle(self, kobj: KnowledgeObject):
         self.log.info("Catching up on network state")
         try:
             payload = self.request_handler.fetch_rids(
-                node=kobj.rid, 
+                node=node_rid, 
                 rid_types=available_rid_types
             )
         except RequestError:
@@ -107,5 +101,25 @@ def handle(self, kobj: KnowledgeObject):
 
             # marked as external since we are handling RIDs from another node
             # will fetch remotely instead of checking local cache
-            self.kobj_queue.push(rid=rid, source=kobj.rid)
+            self.kobj_queue.push(rid=rid, source=node_rid)
         self.log.info("Done")
+    
+    def handle(self, kobj: KnowledgeObject):
+        """Makes contact with providers of RID types of interest.
+        
+        When an incoming node knowledge object is identified as a provider
+        of an RID type of interest, this handler will propose a new edge 
+        subscribing to future node events, and fetch existing nodes to catch 
+        up to the current state.
+        """
+        self.process_node(kobj.rid, kobj.bundle)
+    
+    @depends_on("graph", "kobj_worker")
+    def start(self):
+        self.log.info("Starting node contact analysis on cached profiles...")
+        for rid in self.cache.list_rids(rid_types=(KoiNetNode,)):
+            bundle = self.cache.read(rid)
+            if not bundle:
+                continue
+            
+            self.process_node(rid, bundle)

src/koi_net/components/sync_manager.py

@@ -1,8 +1,10 @@
 from dataclasses import dataclass
 from logging import Logger
+from rid_lib import RIDType
 from rid_lib.ext import Cache
 from rid_lib.types import KoiNetNode
 
+from ..config.base import BaseNodeConfig
 from ..infra import depends_on
 from ..exceptions import RequestError
 from .graph import NetworkGraph
@@ -18,25 +20,30 @@ class SyncManager:
     log: Logger
     graph: NetworkGraph
     cache: Cache
+    config: BaseNodeConfig
     request_handler: RequestHandler
     kobj_queue: KobjQueue
 
     @depends_on("graph", "kobj_worker")
     def start(self):
-        """Catches up with node providers on startup."""
-        
-        node_providers = self.graph.get_neighbors(
-            direction="in",
-            allowed_type=KoiNetNode
-        )
-        
-        if not node_providers:
-            return
-        
-        self.log.debug(f"Catching up with `orn:koi-net.node` providers: {node_providers}")
-        self.catch_up_with(node_providers, [KoiNetNode])
+        """Catches up with providers on startup."""
+        self.catch_up_with_all(self.config.koi_net.rid_types_of_interest)
 
-    def catch_up_with(self, nodes, rid_types):
+    def catch_up_with_all(self, rid_types: list[RIDType]):
+        node_providers = []
+        for rid_type in rid_types:
+            providers = self.graph.get_neighbors(
+                direction="in",
+                allowed_type=rid_type
+            )
+            
+            if not node_providers:
+                continue
+            
+            self.log.debug(f"Catching up with {rid_type} providers: {node_providers}")
+            self.catch_up_with(node_providers, [rid_type])
+    
+    def catch_up_with(self, nodes: list[KoiNetNode], rid_types: list[RIDType]):
         """Catches up with the state of RID types within other nodes."""
 
         for node in nodes:

src/koi_net/core.py

@@ -1,3 +1,4 @@
+import sys
 from pathlib import Path
 
 from .infra import LogSystem
@@ -72,7 +73,14 @@ class BaseNode(BaseAssembly):
     edge_negotiation_handler: EdgeNegotiationHandler = EdgeNegotiationHandler
     secure_profile_handler: SecureProfileHandler = SecureProfileHandler
 
-    def __new__(cls, *args, root_dir: Path = Path.cwd(), **kwargs):
+    def __new__(cls, *args, root_dir: Path | None = None, **kwargs):
+        if root_dir is None:
+            if len(sys.argv) > 1:
+                root_dir = Path(sys.argv[1])
+                print(f"Root dir was set by CLI to '{root_dir}'")
+            else:
+                root_dir = Path.cwd()
+        
         cls._log_system()
         return super().__new__(cls, *args, root_dir=root_dir, **kwargs)
 

src/koi_net/infra/container.py

@@ -1,7 +1,7 @@
-from queue import Queue
 import time
-from typing import TYPE_CHECKING, Any
 import threading
+from queue import Queue
+from typing import TYPE_CHECKING, Any
 from pathlib import Path
 from logging import Logger