removed unused default effector actions, renamed secure to secure manager, restructured sync manager and catch up call in lifecycle, added more docstrings

Author: lukvmil

src/koi_net/assembler.py

@@ -1,10 +1,10 @@
-from enum import StrEnum
 import inspect
+from enum import StrEnum
 from typing import Any, Protocol
 from dataclasses import make_dataclass
 
-from pydantic import BaseModel
 import structlog
+from pydantic import BaseModel
 
 from .entrypoints.base import EntryPoint
 
@@ -38,6 +38,7 @@ def __new__(cls, name: str, bases: tuple, dct: dict[str]):
         return cls
 
 class NodeContainer(Protocol):
+    """Dummy 'shape' for node containers built by assembler."""
     entrypoint = EntryPoint
 
 class NodeAssembler(metaclass=BuildOrderer):    
@@ -79,8 +80,6 @@ def _build_deps(
 
             dep_graph[comp_name] = (comp_type, dep_names)
 
-            # print(f"{comp_name} ({comp_type}) -> {dep_names}")
-            
         return dep_graph
 
     @classmethod

src/koi_net/core.py

@@ -20,7 +20,7 @@
 from .network.event_buffer import EventBuffer
 from .processor.pipeline import KnowledgePipeline
 from .processor.kobj_queue import KobjQueue
-from .secure import Secure
+from .secure_manager import SecureManager
 from .entrypoints import NodeServer, NodePoller
 from .processor.knowledge_handlers import (
     basic_manifest_handler, 
@@ -54,7 +54,7 @@ class BaseNode(NodeAssembler):
         directory_path=config.koi_net.cache_directory_path)
     identity = NodeIdentity
     graph = NetworkGraph
-    secure = Secure
+    secure_manager = SecureManager
     handshaker = Handshaker
     error_handler = ErrorHandler
     request_handler = RequestHandler

src/koi_net/default_actions.py

@@ -32,8 +32,8 @@ class Effector:
 
     cache: Cache
     resolver: NetworkResolver
-    kobj_queue: KobjQueue | None
-    action_context: ActionContext | None
+    kobj_queue: KobjQueue
+    action_context: ActionContext
     _action_table: dict[
         type[RID], 
         Callable[

src/koi_net/effector.py

@@ -9,6 +9,7 @@
 
 
 class Handshaker:
+    """Handles handshaking with other nodes."""
     def __init__(
         self, 
         cache: Cache, 
@@ -21,9 +22,11 @@ def __init__(
 
     def handshake_with(self, target: KoiNetNode):
         """Initiates a handshake with target node.
+        
         Pushes successive `FORGET` and `NEW` events to target node to
         reset the target's cache in case it already knew this node. 
         """
+        
         log.debug(f"Initiating handshake with {target}")
         self.event_queue.push(
             Event.from_rid(

src/koi_net/handshaker.py

@@ -9,7 +9,7 @@
 class NodeIdentity:
     """Represents a node's identity (RID, profile)."""
 
-    config: NodeConfig    
+    config: NodeConfig
 
     def __init__(self, config: NodeConfig):
         self.config = config
@@ -20,4 +20,4 @@ def rid(self) -> KoiNetNode:
 
     @property
     def profile(self) -> NodeProfile:
-        return self.config.koi_net.node_profile
+        return self.config.koi_net.node_profile

src/koi_net/identity.py

@@ -2,6 +2,7 @@
 from contextlib import contextmanager, asynccontextmanager
 
 from rid_lib.ext import Bundle
+from rid_lib.types import KoiNetNode
 
 from .sync_manager import SyncManager
 from .handshaker import Handshaker
@@ -90,31 +91,29 @@ def start(self):
         self.event_worker.thread.start()
         self.graph.generate()
 
-        # refresh to reflect changes (if any) in config.yaml
-        
+        # refresh to reflect changes (if any) in config.yaml node profile
         self.kobj_queue.push(bundle=Bundle.generate(
             rid=self.identity.rid,
             contents=self.identity.profile.model_dump()
         ))
 
-        log.debug("Waiting for kobj queue to empty")
         self.kobj_queue.q.join()
 
-        if self.sync_manager.catch_up_with_coordinators():
-            pass
+        node_providers = self.graph.get_neighbors(
+            direction="in",
+            allowed_type=[KoiNetNode]
+        )
+        
+        if node_providers:
+            log.debug(f"Catching up with `orn:koi-net.node` providers: {node_providers}")
+            self.sync_manager.catch_up_with(node_providers, [KoiNetNode])
 
         elif self.config.koi_net.first_contact.rid:
-            log.debug(f"I don't have any edges with coordinators, reaching out to first contact {self.config.koi_net.first_contact.rid!r}")
-            
+            log.debug(f"No edges with `orn:koi-net.node` providers, reaching out to first contact {self.config.koi_net.first_contact.rid!r}")
             self.handshaker.handshake_with(self.config.koi_net.first_contact.rid)
-        
-
+            
     def stop(self):
-        """Stops a node.
-        
-        Finishes processing knowledge object queue.
-        """        
-        log.info(f"Waiting for kobj queue to empty ({self.kobj_queue.q.unfinished_tasks} tasks remaining)")
+        """Stops a node, send stop signals to workers."""
 
         self.kobj_queue.q.put(STOP_WORKER)
         self.event_queue.q.put(STOP_WORKER)

src/koi_net/lifecycle.py

@@ -87,6 +87,8 @@
 )
 
 class LogSystem:
+    """Handles initializing the logging system."""
+    
     def __init__(self):
         file_handler = RotatingFileHandler(
             filename="log.ndjson",

src/koi_net/log_system.py

@@ -27,7 +27,7 @@
 )
 from ..protocol.node import NodeProfile, NodeType
 from ..protocol.model_map import API_MODEL_MAP
-from ..secure import Secure
+from ..secure_manager import SecureManager
 from .error_handler import ErrorHandler
 
 log = structlog.stdlib.get_logger()
@@ -58,19 +58,19 @@ class RequestHandler:
 
     cache: Cache
     identity: NodeIdentity
-    secure: Secure
+    secure_manager: SecureManager
     error_handler: ErrorHandler
 
     def __init__(
         self, 
         cache: Cache,
         identity: NodeIdentity,
-        secure: Secure,
+        secure_manager: SecureManager,
         error_handler: ErrorHandler
     ):
         self.cache = cache
         self.identity = identity
-        self.secure = secure
+        self.secure_manager = secure_manager
         self.error_handler = error_handler
 
     def get_base_url(self, node_rid: KoiNetNode) -> str:
@@ -105,7 +105,7 @@ def make_request(
         url = self.get_base_url(node) + path
         log.info(f"Making request to {url}")
 
-        signed_envelope = self.secure.create_envelope(
+        signed_envelope = self.secure_manager.create_envelope(
             payload=request,
             target=node
         )
@@ -131,7 +131,7 @@ def make_request(
             return
 
         resp_envelope = resp_env_model.model_validate_json(result.text)
-        self.secure.validate_envelope(resp_envelope)
+        self.secure_manager.validate_envelope(resp_envelope)
 
         return resp_envelope.payload
 

src/koi_net/network/request_handler.py

@@ -8,7 +8,7 @@
 from koi_net.processor.kobj_queue import KobjQueue
 from koi_net.protocol.consts import BROADCAST_EVENTS_PATH, FETCH_BUNDLES_PATH, FETCH_MANIFESTS_PATH, FETCH_RIDS_PATH, POLL_EVENTS_PATH
 from koi_net.protocol.envelope import SignedEnvelope
-from koi_net.secure import Secure
+from koi_net.secure_manager import SecureManager
 
 from ..protocol.api_models import (
     EventsPayload,
@@ -36,15 +36,15 @@ def __init__(
         cache: Cache,
         kobj_queue: KobjQueue,
         poll_event_buf: EventBuffer,
-        secure: Secure
+        secure_manager: SecureManager
     ):
         self.cache = cache
         self.kobj_queue = kobj_queue
         self.poll_event_buf = poll_event_buf
-        self.secure = secure
+        self.secure_manager = secure_manager
 
     def handle_response(self, path: str, req: SignedEnvelope):
-        self.secure.validate_envelope(req)
+        self.secure_manager.validate_envelope(req)
 
         response_map = {
             BROADCAST_EVENTS_PATH: self.broadcast_events_handler,
@@ -59,7 +59,7 @@ def handle_response(self, path: str, req: SignedEnvelope):
         if response is None:
             return
 
-        return self.secure.create_envelope(
+        return self.secure_manager.create_envelope(
             payload=response,
             target=req.source_node
         )