Client

ClientJobDatabase(db_path: str, jobs_compress: bool = True)

Lightweight client-side job and events store.

Source code in norfab\core\client.py

def __init__(self, db_path: str, jobs_compress: bool = True):
    self.db_path = db_path
    self.jobs_compress = jobs_compress
    self._local = threading.local()
    self._lock = threading.Lock()
    self._initialize_database()

fetch_jobs(statuses: List[str] = None, limit: int = 10, min_poll_age: float = 0, service: str = None, task: str = None, workers_completed: List[str] = None, last: int = None) -> List[dict]

Fetch jobs with flexible filtering and complete job attributes.

Parameters:

Name	Type	Description	Default
statuses	List[str]	List of job statuses to filter by (default: all statuses)	None
limit	int	Maximum number of jobs to return (used when last is not specified)	10
min_poll_age	float	Minimum seconds since last poll (for rate limiting GET requests)	0
service	str	Service name to filter by (optional)	None
task	str	Task name to filter by (optional)	None
workers_completed	List[str]	List of worker names that completed the job (optional)	None
last	int	Return only the last x number of jobs (newest first), overrides limit (optional)	None

Returns:

Type	Description
List[dict]	List of job dictionaries with complete attributes including:
List[dict]	uuid, service, task, args, kwargs, workers_*, status, result_data,
List[dict]	errors, timestamps, etc.

Source code in norfab\core\client.py

def fetch_jobs(
    self,
    statuses: List[str] = None,
    limit: int = 10,
    min_poll_age: float = 0,
    service: str = None,
    task: str = None,
    workers_completed: List[str] = None,
    last: int = None,
) -> List[dict]:
    """Fetch jobs with flexible filtering and complete job attributes.

    Args:
        statuses: List of job statuses to filter by (default: all statuses)
        limit: Maximum number of jobs to return (used when last is not specified)
        min_poll_age: Minimum seconds since last poll (for rate limiting GET requests)
        service: Service name to filter by (optional)
        task: Task name to filter by (optional)
        workers_completed: List of worker names that completed the job (optional)
        last: Return only the last x number of jobs (newest first), overrides limit (optional)

    Returns:
        List of job dictionaries with complete attributes including:
        uuid, service, task, args, kwargs, workers_*, status, result_data,
        errors, timestamps, etc.
    """
    conditions = []
    params = []

    # Filter by statuses
    if statuses:
        placeholders = ",".join(["?"] * len(statuses))
        conditions.append(f"status IN ({placeholders})")
        params.extend(statuses)

    # Filter by poll age (for dispatcher throttling)
    if min_poll_age > 0:
        poll_threshold = time.time() - min_poll_age
        conditions.append(
            "(last_poll_timestamp IS NULL OR last_poll_timestamp <= ?)"
        )
        params.append(poll_threshold)

    # Filter by service
    if service:
        conditions.append("service = ?")
        params.append(service)

    # Filter by task
    if task:
        conditions.append("task = ?")
        params.append(task)

    # Filter by workers_completed (JSON contains check)
    if workers_completed:
        # For each worker, check if it's in the JSON array
        worker_conditions = []
        for worker in workers_completed:
            # SQLite JSON string contains check
            worker_conditions.append("workers_completed LIKE ?")
            params.append(f'%"{worker}"%')
        conditions.append(f"({' OR '.join(worker_conditions)})")

    # Build WHERE clause
    where_clause = " AND ".join(conditions) if conditions else "1=1"

    # Determine order and limit based on 'last' parameter
    order_direction = "DESC" if last is not None else "ASC"
    result_limit = last if last is not None else limit

    with self._transaction(write=False) as conn:
        cur = conn.execute(
            f"""
            SELECT uuid, service, task, args, kwargs, workers_requested, timeout, deadline,
                   workers_dispatched, workers_started, workers_completed, status,
                   result_data, errors, received_timestamp, started_timestamp,
                   completed_timestamp, created_at, last_poll_timestamp
            FROM jobs
            WHERE {where_clause}
            ORDER BY created_at {order_direction}
            LIMIT ?
            """,
            (*params, result_limit),
        )
        rows = cur.fetchall()
    return [self._hydrate(row) for row in rows]

NFPClient(inventory: NorFabInventory, broker: str, name: str, exit_event: Optional[threading.Event] = None, event_queue: Optional[queue.Queue] = None)

Bases: object

NFPClient is a client class for interacting with a broker using ZeroMQ for messaging. It handles sending and receiving messages, managing connections, and performing tasks.

Attributes:

Name	Type	Description
broker	str	The broker address.
ctx	Context	The ZeroMQ context.
broker_socket	Socket	The ZeroMQ socket for communication with the broker.
poller	Poller	The ZeroMQ poller for managing socket events.
name	str	The name of the client.
stats_send_to_broker	int	Counter for messages sent to the broker.
stats_recv_from_broker	int	Counter for messages received from the broker.
stats_reconnect_to_broker	int	Counter for reconnections to the broker.
stats_recv_event_from_broker	int	Counter for events received from the broker.
client_private_key_file	str	Path to the client's private key file.
broker_public_key_file	str	Path to the broker's public key file.

Methods:

Name	Description
ensure_bytes	Helper function to convert workers target to bytes.
reconnect_to_broker	Connects or reconnects to the broker.
send_to_broker	Sends a message to the broker.
rcv_from_broker	Waits for a response from the broker.
post	Sends a job request to the broker and returns the result.
get	Sends a job reply message to the broker requesting job results.
get_iter	Sends a job reply message to the broker requesting job results and yields results iteratively.
fetch_file	Downloads a file from the Broker File Sharing Service.
run_job	Runs a job and returns results produced by workers.
run_job_iter	Runs a job and yields results produced by workers iteratively.
destroy	Cleans up and destroys the client instance.

Parameters:

Name	Type	Description	Default
inventory	NorFabInventory	The inventory object containing base directory information.	required
broker	str	The broker object for communication.	required
name	str	The name of the client.	required
exit_event	Event	An event to signal client exit. Defaults to None.	None
event_queue	Queue	A queue for handling events. Defaults to None.	None

Source code in norfab\core\client.py

def __init__(
    self,
    inventory: NorFabInventory,
    broker: str,
    name: str,
    exit_event: Optional[threading.Event] = None,
    event_queue: Optional[queue.Queue] = None,
):
    self.inventory = inventory
    self.name = name
    self.zmq_name = f"{self.name}-{uuid4().hex}"
    self.broker = broker
    self.base_dir = os.path.join(
        self.inventory.base_dir, "__norfab__", "files", "client", self.name
    )
    self.file_transfers = {}  # file transfers tracker
    self.zmq_auth = self.inventory.broker.get("zmq_auth", True)
    self.socket_lock = threading.Lock()  # used to protect socket object
    self.build_message = NFP.MessageBuilder()

    # create base directories
    os.makedirs(self.base_dir, exist_ok=True)

    self.job_db = ClientJobDatabase(
        os.path.join(self.base_dir, f"{self.name}.db"),
        jobs_compress=True,
    )

    # generate certificates and create directories
    if self.zmq_auth is not False:
        generate_certificates(
            self.base_dir,
            cert_name=self.name,
            broker_keys_dir=os.path.join(
                self.inventory.base_dir,
                "__norfab__",
                "files",
                "broker",
                "public_keys",
            ),
            inventory=self.inventory,
        )
        self.public_keys_dir = os.path.join(self.base_dir, "public_keys")
        self.private_keys_dir = os.path.join(self.base_dir, "private_keys")

    self.ctx = zmq.Context()
    self.poller = zmq.Poller()
    self.reconnect_to_broker()

    self.exit_event = threading.Event() if exit_event is None else exit_event
    self.destroy_event = (
        threading.Event()
    )  # destroy event, used by worker to stop its client
    self.mmi_queue = queue.Queue(maxsize=0)
    self.event_queue = event_queue or queue.Queue(maxsize=1000)

    # Configuration for dispatcher
    self.poll_interval = 0.5  # Seconds between GET polls for same job (throttling)
    self.dispatch_batch_size = 10  # Max jobs to process per dispatch cycle

    # start receiver thread - handles all incoming messages
    self.recv_thread = threading.Thread(
        target=recv, daemon=True, name=f"{self.name}_recv_thread", args=(self,)
    )
    self.recv_thread.start()

    # start dispatcher thread - sends POST/GET requests asynchronously
    self.dispatcher_thread = threading.Thread(
        target=dispatcher,
        daemon=True,
        name=f"{self.name}_dispatcher",
        args=(self,),
    )
    self.dispatcher_thread.start()

ensure_bytes(value: Any) -> bytes

Helper function to convert value to bytes.

Source code in norfab\core\client.py

def ensure_bytes(self, value: Any) -> bytes:
    """
    Helper function to convert value to bytes.
    """
    if isinstance(value, bytes):
        return value
    # transform string to bytes
    if isinstance(value, str):
        return value.encode("utf-8")
    # convert value to json string
    else:
        return json.dumps(value).encode("utf-8")

reconnect_to_broker()

Connect or reconnect to the broker.

This method handles the connection or reconnection to the broker by:

• Closing the existing broker socket if it exists.
• Creating a new DEALER socket.
• Setting the socket options including the identity and linger.
• Loading the client's private and public keys for CURVE encryption.
• Loading the broker's public key for CURVE encryption.
• Connecting the socket to the broker.
• Registering the socket with the poller for incoming messages.
• Logging the connection status.
• Incrementing the reconnect statistics counter.

Source code in norfab\core\client.py

def reconnect_to_broker(self):
    """
    Connect or reconnect to the broker.

    This method handles the connection or reconnection to the broker by:

    - Closing the existing broker socket if it exists.
    - Creating a new DEALER socket.
    - Setting the socket options including the identity and linger.
    - Loading the client's private and public keys for CURVE encryption.
    - Loading the broker's public key for CURVE encryption.
    - Connecting the socket to the broker.
    - Registering the socket with the poller for incoming messages.
    - Logging the connection status.
    - Incrementing the reconnect statistics counter.
    """
    if self.broker_socket:
        self.poller.unregister(self.broker_socket)
        self.broker_socket.close()

    self.broker_socket = self.ctx.socket(zmq.DEALER)
    self.broker_socket.setsockopt_unicode(zmq.IDENTITY, self.zmq_name, "utf8")
    self.broker_socket.linger = 0

    if self.zmq_auth is not False:
        # We need two certificates, one for the client and one for
        # the server. The client must know the server's public key
        # to make a CURVE connection.
        self.client_private_key_file = os.path.join(
            self.private_keys_dir, f"{self.name}.key_secret"
        )
        client_public, client_secret = zmq.auth.load_certificate(
            self.client_private_key_file
        )
        self.broker_socket.curve_secretkey = client_secret
        self.broker_socket.curve_publickey = client_public

        # The client must know the server's public key to make a CURVE connection.
        self.broker_public_key_file = os.path.join(
            self.public_keys_dir, "broker.key"
        )
        server_public, _ = zmq.auth.load_certificate(self.broker_public_key_file)
        self.broker_socket.curve_serverkey = server_public

    self.broker_socket.connect(self.broker)
    self.poller.register(self.broker_socket, zmq.POLLIN)
    log.debug(f"{self.name} - client connected to broker at '{self.broker}'")
    self.stats_reconnect_to_broker += 1

send_to_broker(command, service, workers, uuid, request)

Sends a command to the broker.

Parameters:

Name	Type	Description	Default
command	str	The command to send (e.g., NFP.POST, NFP.GET).	required
service	str	The service to which the command is related.	required
workers	str	The workers involved in the command.	required
uuid	str	The unique identifier for the request.	required
request	str	The request payload to be sent.	required

Source code in norfab\core\client.py

def send_to_broker(self, command, service, workers, uuid, request):
    """
    Sends a command to the broker.

    Args:
        command (str): The command to send (e.g., NFP.POST, NFP.GET).
        service (str): The service to which the command is related.
        workers (str): The workers involved in the command.
        uuid (str): The unique identifier for the request.
        request (str): The request payload to be sent.
    """
    if command == NFP.POST:
        msg = self.build_message.client_to_broker_post(
            command=command,
            service=service,
            workers=workers,
            uuid=uuid,
            request=request,
        )
    elif command == NFP.GET:
        msg = self.build_message.client_to_broker_get(
            command=command,
            service=service,
            workers=workers,
            uuid=uuid,
            request=request,
        )
    elif command == NFP.PUT:
        msg = self.build_message.client_to_broker_put(
            command=command,
            service=service,
            workers=workers,
            uuid=uuid,
            request=request,
        )
    elif command == NFP.MMI:
        msg = self.build_message.client_to_broker_mmi(
            command=command,
            service=service,
            workers=workers,
            uuid=uuid,
            request=request,
        )
    else:
        log.error(
            f"{self.name} - cannot send '{command}' to broker, command unsupported"
        )
        return

    log.debug(f"{self.name} - sending '{msg}'")

    with self.socket_lock:
        self.broker_socket.send_multipart(msg)
        self.stats_send_to_broker += 1

mmi(service: str, task: str = None, args: list = None, kwargs: dict = None, workers: Union[str, list] = 'all', uuid: hex = None, timeout: int = 30) -> dict

Send an MMI (management interface) request to a service via the broker.

MMI requests are intended for lightweight, broker-routed management or introspection operations (e.g., service metadata, health, inventory-like queries) that return a single aggregated response payload.

Parameters:

Name	Type	Description	Default
service	str	Target service name.	required
task	str	Service task name to execute.	None
args	list	Positional arguments for the task.	None
kwargs	dict	Keyword arguments for the task.	None
workers	Union[str, list]	Workers selector. Can be "all", "any", or a list of names.	'all'
uuid	hex	Optional request UUID. If not provided, a random UUID is generated.	None
timeout	int	Maximum time (seconds) to wait for the MMI reply.	30

Returns:

Type	Description
dict	Dictionary containing status, results, and errors keys: status: HTTP-like status code as a string (e.g., "200", "408"). results: Decoded JSON payload from the broker/service. errors: List of error strings.

Source code in norfab\core\client.py

def mmi(
    self,
    service: str,
    task: str = None,
    args: list = None,
    kwargs: dict = None,
    workers: Union[str, list] = "all",
    uuid: hex = None,
    timeout: int = 30,
) -> dict:
    """
    Send an MMI (management interface) request to a service via the broker.

    MMI requests are intended for lightweight, broker-routed management or
    introspection operations (e.g., service metadata, health, inventory-like
    queries) that return a single aggregated response payload.

    Args:
        service: Target service name.
        task: Service task name to execute.
        args: Positional arguments for the task.
        kwargs: Keyword arguments for the task.
        workers: Workers selector. Can be ``"all"``, ``"any"``, or a list of names.
        uuid: Optional request UUID. If not provided, a random UUID is generated.
        timeout: Maximum time (seconds) to wait for the MMI reply.

    Returns:
        Dictionary containing ``status``, ``results``, and ``errors`` keys:

            - ``status``: HTTP-like status code as a string (e.g., ``"200"``, ``"408"``).
            - ``results``: Decoded JSON payload from the broker/service.
            - ``errors``: List of error strings.
    """
    service_str = service
    uuid_str = uuid or uuid4().hex
    args = args or []
    kwargs = kwargs or {}
    ret = {"status": "200", "results": {}, "errors": []}

    service = self.ensure_bytes(service_str)
    uuid = self.ensure_bytes(uuid_str)
    workers = self.ensure_bytes(workers)

    request = self.ensure_bytes(
        {"task": task, "kwargs": kwargs or {}, "args": args or []}
    )

    self.send_to_broker(NFP.MMI, service, workers, uuid, request)

    deadline = time.time() + timeout
    while time.time() < deadline:
        # check if need to stop
        if self.exit_event.is_set() or self.destroy_event.is_set():
            ret["errors"].append(
                f"{self.name} - '{uuid_str}:{service_str}' MMI interrupted (client stopping)"
            )
            ret["status"] = "499"
            break

        try:
            msg = self.mmi_queue.get(block=True, timeout=0.5)
            self.mmi_queue.task_done()
        except queue.Empty:
            continue

        (
            empty,
            reply_header,
            reply_command,
            reply_service,
            reply_uuid,
            reply_status,
            reply_task_result,
        ) = msg

        # Defer unrelated replies and continue scanning.
        if reply_uuid != uuid:
            self.mmi_queue.put(m)
            continue

        if reply_header != NFP.BROKER and reply_command != NFP.MMI:
            ret["errors"].append(
                f"{self.name} - '{uuid_str}:{service_str}' MMI unexpected reply header/command"
            )
            ret["status"] = reply_status.decode("utf-8")
            break

        try:
            ret["results"] = json.loads(reply_task_result.decode("utf-8"))
            ret["status"] = reply_status.decode("utf-8")
        except Exception as e:
            ret["errors"].append(
                f"{self.name} - '{uuid_str}:{service_str}' MMI failed to decode reply payload: {e}"
            )
            ret["results"] = {"status": "Invalid MMI response payload"}
            ret["status"] = "500"
        break
    else:
        msg = f"{self.name} - '{uuid_str}:{service_str}' MMI request {timeout}s timeout exceeded."
        log.error(msg)
        ret["errors"].append(msg)
        ret["results"] = {"status": "MMI Request Timeout"}
        ret["status"] = "408"

    return ret

delete_fetched_files(filepath: str = '*') -> dict

Delete files and folders matching the filepath glob pattern.

Parameters:

Name	Type	Description	Default
filepath	str	Glob pattern to match files/folders. Default is "*" (all files).	'*'

Returns:

Name	Type	Description
dict	dict	Dictionary with 'deleted' list of deleted paths and 'errors' list of error messages.

Source code in norfab\core\client.py

def delete_fetched_files(self, filepath: str = "*") -> dict:
    """
    Delete files and folders matching the filepath glob pattern.

    Args:
        filepath (str): Glob pattern to match files/folders. Default is "*" (all files).

    Returns:
        dict: Dictionary with 'deleted' list of deleted paths and 'errors' list of error messages.
    """
    files_folder = os.path.join(self.base_dir, "fetchedfiles")

    result = {"deleted": [], "errors": []}

    # Build full pattern path
    pattern = os.path.join(files_folder, filepath)

    # Find all matching files and folders
    matches = glob.glob(pattern, recursive=True)

    # Sort by depth (deepest first) to avoid deleting parent before children
    matches.sort(key=lambda x: x.count(os.sep), reverse=True)

    for match in matches:
        try:
            if os.path.isfile(match):
                os.remove(match)
                result["deleted"].append(match)
                log.debug(f"{self.name} - deleted file: {match}")
            elif os.path.isdir(match):
                shutil.rmtree(match)
                result["deleted"].append(match)
                log.debug(f"{self.name} - deleted folder: {match}")
        except Exception as e:
            error_msg = f"Failed to delete {match}: {str(e)}"
            result["errors"].append(error_msg)
            log.error(f"{self.name} - {error_msg}")

    return result

fetch_file(url: str, chunk_size: int = 256000, pipeline: int = 10, timeout: int = 600, read: bool = False) -> Tuple[str, Any]

Fetches a file from a given URL and saves it to a specified destination.

Parameters:

Name	Type	Description	Default
url	str	The URL of the file to be fetched.	required
chunk_size	int	The size of each chunk to be fetched. Default is 250000 bytes.	256000
pipeline	int	The number of chunks to be fetched in transit. Default is 10.	10
timeout	int	The maximum time (in seconds) to wait for the file to be fetched. Default is 600 seconds.	600
read	bool	If True, the file content is read and returned. If False, the file path is returned. Default is False.	False

Returns:

Name	Type	Description
tuple	Tuple[str, Any]	A tuple containing the status code (str) and the reply (str). The reply can be the file content, file path, or an error message.

Raises:

Type	Description
Exception	If there is an error in fetching the file or if the file's MD5 hash does not match the expected hash.

Source code in norfab\core\client.py

def fetch_file(
    self,
    url: str,
    chunk_size: int = 256000,
    pipeline: int = 10,
    timeout: int = 600,
    read: bool = False,
) -> Tuple[str, Any]:
    """
    Fetches a file from a given URL and saves it to a specified destination.

    Parameters:
        url (str): The URL of the file to be fetched.
        chunk_size (int, optional): The size of each chunk to be fetched. Default is 250000 bytes.
        pipeline (int, optional): The number of chunks to be fetched in transit. Default is 10.
        timeout (int, optional): The maximum time (in seconds) to wait for the file to be fetched. Default is 600 seconds.
        read (bool, optional): If True, the file content is read and returned. If False, the file path is returned. Default is False.

    Returns:
        tuple: A tuple containing the status code (str) and the reply (str). The reply can be the file content, file path, or an error message.

    Raises:
        Exception: If there is an error in fetching the file or if the file's MD5 hash does not match the expected hash.
    """

    # round up digit e.g. if 2.0 -> 2 if 2.1 -> 3 if 0.01 -> 1
    def round_up(num):
        return max(1, (int(num) + (not num.is_integer())))

    uuid = uuid4().hex
    result = {"status": "200", "content": None, "error": None}
    downloaded = False

    # run sanity checks
    if not url.startswith("nf://"):
        result["status"] = "500"
        result["error"] = "Invalid url format"
        return result

    # prevent path traversal / absolute paths
    url_path = url.replace("nf://", "")
    url_path = url_path.lstrip("/\\")
    destination = os.path.abspath(
        os.path.join(self.base_dir, "fetchedfiles", *url_path.split("/"))
    )
    fetched_root = os.path.abspath(os.path.join(self.base_dir, "fetchedfiles"))
    if os.path.commonpath([fetched_root, destination]) != fetched_root:
        result["status"] = "500"
        result["error"] = "Invalid url path"
        return result

    os.makedirs(os.path.split(destination)[0], exist_ok=True)

    self.file_transfers[uuid] = {
        "total_bytes_received": 0,  # Total bytes received
        "offset": 0,  # Offset of next chunk request
        "credit": pipeline,  # Up to PIPELINE chunks in transit
        "chunk_size": chunk_size,
        "file_hash": hashlib.md5(),
    }

    # get file details
    file_details = self.run_job(
        service="filesharing",
        workers="all",
        task="file_details",
        kwargs={"url": url},
        timeout=timeout,
    )
    for w_name, w_res in file_details.items():
        if not w_res["failed"]:
            file_details = w_res["result"]
            self.file_transfers[uuid].update(file_details)
            self.file_transfers[uuid]["w_name"] = w_name
            self.file_transfers[uuid]["chunk_requests_remaining"] = round_up(
                file_details["size_bytes"] / chunk_size
            )
            break
    else:
        result["status"] = "404"
        result["error"] = "File download failed - file not found"
        _ = self.file_transfers.pop(uuid)
        return result

    log.debug(f"{self.name}:fetch_file - retrieved file details - {file_details}")

    # check if file already downloaded
    if os.path.isfile(destination):
        file_hash = hashlib.md5()
        with open(destination, "rb") as f:
            chunk = f.read(8192)
            while chunk:
                file_hash.update(chunk)
                chunk = f.read(8192)
        md5hash = file_hash.hexdigest()
        downloaded = md5hash == file_details["md5hash"]

    if file_details["exists"] and not downloaded:
        self.file_transfers[uuid]["destination"] = open(destination, "wb")
        # decrement by 1 because calling fetch_job sends first chunk
        self.file_transfers[uuid]["chunk_requests_remaining"] -= 1
        # run fetch file job
        file_fetch_job = self.run_job(
            uuid=uuid,
            service="filesharing",
            workers=[w_name],
            task="fetch_file",
            kwargs={"url": url, "offset": 0, "chunk_size": chunk_size},
            timeout=timeout,
        )
        file_fetch_job = file_fetch_job[w_name]

        if file_fetch_job["failed"]:
            result["status"] = "404"
            result["error"] = file_fetch_job["errors"]
            downloaded = False
        else:
            downloaded = True

    if downloaded:
        # Verify streaming did not mark job failed (e.g., MD5 mismatch)
        download_job = self.job_db.get_job(uuid)
        if download_job and download_job.get("status") == JobStatus.FAILED:
            result["error"] = (
                f"File download job {uuid} failed: {download_job.get('errors', [])}"
            )
            result["status"] = "400"
        elif read:
            with open(destination, "r", encoding="utf-8") as f:
                result["content"] = f.read()
        else:
            result["content"] = destination

    _ = self.file_transfers.pop(uuid)

    return result

run_job(service: str, task: str, uuid: str = None, args: list = None, kwargs: dict = None, workers: Union[str, list] = 'all', timeout: int = 600, markdown: bool = False, nowait: bool = False) -> Any

Run a job on the specified service and task, with optional arguments and timeout settings.

This method submits a job to the database and waits for the dispatcher and receiver threads to process it asynchronously. The job progresses through states: NEW -> SUBMITTING -> DISPATCHED -> STARTED -> COMPLETED (or FAILED/STALE)

Parameters:

Name	Type	Description	Default
service	str	The name of the service to run the job on.	required
task	str	The task to be executed.	required
uuid	str	A unique identifier for the job. If not provided, a new UUID will be generated. Defaults to None.	None
args	list	A list of positional arguments to pass to the task. Defaults to None.	None
kwargs	dict	A dictionary of keyword arguments to pass to the task. Defaults to None.	None
workers	str	The workers to run the job on. Defaults to "all".	'all'
timeout	int	The maximum time in seconds to wait for the job to complete. Defaults to 600.	600
markdown	bool	Convert results to markdown representation	False
nowait	bool	If false, wait for job to complete for timeout, return job details otherwise	False

Returns:

Name	Type	Description
Any	Any	The result of the job if successful, or None if the job failed, timed out, or became stale.

Source code in norfab\core\client.py

def run_job(
    self,
    service: str,
    task: str,
    uuid: str = None,
    args: list = None,
    kwargs: dict = None,
    workers: Union[str, list] = "all",
    timeout: int = 600,
    markdown: bool = False,
    nowait: bool = False,
) -> Any:
    """
    Run a job on the specified service and task, with optional arguments and timeout settings.

    This method submits a job to the database and waits for the dispatcher and receiver
    threads to process it asynchronously. The job progresses through states:
    NEW -> SUBMITTING -> DISPATCHED -> STARTED -> COMPLETED (or FAILED/STALE)

    Args:
        service (str): The name of the service to run the job on.
        task (str): The task to be executed.
        uuid (str, optional): A unique identifier for the job. If not provided, a new UUID will be generated. Defaults to None.
        args (list, optional): A list of positional arguments to pass to the task. Defaults to None.
        kwargs (dict, optional): A dictionary of keyword arguments to pass to the task. Defaults to None.
        workers (str, optional): The workers to run the job on. Defaults to "all".
        timeout (int, optional): The maximum time in seconds to wait for the job to complete. Defaults to 600.
        markdown (bool, optional): Convert results to markdown representation
        nowait (bool, optional): If false, wait for job to complete for timeout, return job details otherwise

    Returns:
        Any: The result of the job if successful, or None if the job failed, timed out, or became stale.
    """
    uuid = uuid or uuid4().hex
    args = args or []
    kwargs = kwargs or {}
    result = None
    job = None
    deadline = time.time() + timeout

    self.job_db.add_job(
        uuid, service, task, workers, args, kwargs, timeout, deadline
    )

    if nowait:
        return {
            "uuid": uuid,
            "service": service,
        }
    else:
        while time.time() < deadline:
            if self.exit_event.is_set() or self.destroy_event.is_set():
                break
            job = self.job_db.get_job(uuid)
            if not job:
                break
            if job["status"] == JobStatus.COMPLETED:
                result = job.get("result_data")
                break
            if job["status"] == JobStatus.FAILED:
                log.warning(
                    f"{self.name} - job {uuid} failed: {job.get('errors', [])}"
                )
                break
            if job["status"] == JobStatus.STALE:
                log.warning(
                    f"{self.name} - job {uuid} became stale: {job.get('errors', [])}"
                )
                break
            time.sleep(0.2)

        return markdown_results(job, service, task, kwargs) if markdown else result

destroy()

Gracefully shuts down the client.

This method logs an interrupt message, sets the destroy event, and destroys the client context to ensure a clean shutdown.

Source code in norfab\core\client.py

def destroy(self):
    """
    Gracefully shuts down the client.

    This method logs an interrupt message, sets the destroy event, and
    destroys the client context to ensure a clean shutdown.
    """
    log.info(f"{self.name} - client interrupt received, killing client")
    self.destroy_event.set()
    self.job_db.close()
    self.ctx.destroy()
    # close all file transfer files
    for file_transfer in self.file_transfers.values():
        file_transfer["destination"].close()

recv(client)

Receiver thread: processes all incoming messages from the broker and updates the database.

This function continuously polls the client's broker socket for messages until the client's exit event is set. It handles: - EVENT messages: stored in the events table - RESPONSE messages: updates job status in the database based on response type

The receiver thread is the ONLY thread that reads from the socket, eliminating contention issues. All job state changes are persisted to the database.

Parameters:

Name	Type	Description	Default
client	object	The client instance containing the broker socket, poller, job_db, and configuration.	required

Source code in norfab\core\client.py

def recv(client):
    """
    Receiver thread: processes all incoming messages from the broker and updates the database.

    This function continuously polls the client's broker socket for messages
    until the client's exit event is set. It handles:
    - EVENT messages: stored in the events table
    - RESPONSE messages: updates job status in the database based on response type

    The receiver thread is the ONLY thread that reads from the socket, eliminating
    contention issues. All job state changes are persisted to the database.

    Args:
        client (object): The client instance containing the broker socket,
                         poller, job_db, and configuration.
    """
    while not client.exit_event.is_set() and not client.destroy_event.is_set():
        # Poll socket for messages every 500ms interval
        try:
            items = client.poller.poll(500)
        except KeyboardInterrupt:
            break
        except Exception:
            continue

        if not items:
            continue

        with client.socket_lock:
            try:
                msg = client.broker_socket.recv_multipart(zmq.NOBLOCK)
            except zmq.Again:
                continue

        client.stats_recv_from_broker += 1

        # Message format: [empty, header, command, service, uuid, status, payload]
        if len(msg) < 7:
            log.error(f"{client.name} - received malformed message: {msg}")
            continue

        command = msg[2]
        juuid = msg[4].decode("utf-8")
        status = msg[5].decode("utf-8")

        log.debug(
            f"{client.name} - received '{command}' message from broker, juuid {juuid}, status {status}"
        )

        if command == NFP.STREAM:
            payload = msg[6]  # payload is a chunk of bytes
            handle_stream(client, juuid, status, payload)
            continue

        try:
            payload = json.loads(msg[6].decode("utf-8"))
        except Exception as e:
            log.error(
                f"{client.name} - failed to parse message, error '{e}'", exc_info=True
            )
            continue

        # Handle EVENT messages
        if command == NFP.EVENT:
            handle_event(client, juuid, payload, msg)

        # Handle RESPONSE messages
        if command == NFP.RESPONSE:
            handle_response(client, juuid, status, payload)

        # handle MMI messages
        if command == NFP.MMI:
            client.mmi_queue.put(msg)

handle_event(client: object, juuid: str, payload: dict, msg: list)

Handle EVENT messages and update job database accordingly.

Parameters:

Name	Type	Description	Default
client	object	The client instance	required
juuid	str	Job UUID	required
payload	dict	Event payload dictionary	required
msg	list	Original message multipart for queue	required

Source code in norfab\core\client.py

def handle_event(client: object, juuid: str, payload: dict, msg: list):
    """
    Handle EVENT messages and update job database accordingly.

    Args:
        client: The client instance
        juuid: Job UUID
        payload: Event payload dictionary
        msg: Original message multipart for queue
    """
    client.event_queue.put(msg)
    client.stats_recv_event_from_broker += 1
    client.job_db.add_event(
        job_uuid=juuid,
        message=payload.get("message", ""),
        severity=payload.get("severity", "INFO"),
        task=payload.get("task"),
        event_data=payload,
    )

handle_response(client, juuid: str, status: str, payload: dict)

Handle RESPONSE messages and update job database accordingly.

Uses job status to determine context: - SUBMITTING: expecting broker 202 with workers list - DISPATCHED/STARTED: expecting worker ACKs (202), results (200), or pending (300)

Status codes: - 202: Accepted (POST acknowledged by broker or worker) - 200: OK (GET completed with results) - 300: Pending (job still in progress) - 4xx: Client errors - 5xx: Server errors

Source code in norfab\core\client.py

def handle_response(client, juuid: str, status: str, payload: dict):
    """
    Handle RESPONSE messages and update job database accordingly.

    Uses job status to determine context:
    - SUBMITTING: expecting broker 202 with workers list
    - DISPATCHED/STARTED: expecting worker ACKs (202), results (200), or pending (300)

    Status codes:
    - 202: Accepted (POST acknowledged by broker or worker)
    - 200: OK (GET completed with results)
    - 300: Pending (job still in progress)
    - 4xx: Client errors
    - 5xx: Server errors
    """
    job = client.job_db.get_job(juuid)
    if not job:
        log.debug(f"{client.name} - received response for unknown job {juuid}")
        return

    # Broker accepted POST - contains dispatched workers list
    if status == "202":  # ACCEPTED
        workers_list = payload["workers"]
        client.job_db.update_job(
            juuid,
            status=JobStatus.DISPATCHED,
            workers_dispatched=workers_list,
            started_ts=time.ctime(),
        )
        log.debug(f"{client.name} - job {juuid} dispatched to workers: {workers_list}")
        return

    # Worker created the job
    if status == "201":  # JOB CREATED
        worker_single = payload["worker"]
        started = set(job.get("workers_started", []))
        started.add(worker_single)
        client.job_db.update_job(
            juuid,
            status=JobStatus.STARTED,
            workers_started=list(started),
        )
        log.debug(
            f"{client.name} - job {juuid} acknowledged by worker: {worker_single}"
        )
        return

        # GET dispatched to workers (broker 202 response to GET)
        if workers_list:
            log.debug(
                f"{client.name} - job {juuid} GET dispatched to workers: {workers_list}"
            )
        return

    # Handle 200 OK - GET completed with results
    if status == "200":
        dispatched = set(job.get("workers_dispatched", []))
        completed = set(job.get("workers_completed", []))
        existing_results = job.get("result_data") or {}

        # Merge new results with existing (results keyed by worker name)
        if isinstance(payload, dict):
            for worker_name in payload.keys():
                completed.add(worker_name)
            existing_results.update(payload)

        is_complete = completed == dispatched and len(dispatched) > 0

        client.job_db.update_job(
            juuid,
            status=JobStatus.COMPLETED if is_complete else JobStatus.STARTED,
            workers_completed=list(completed),
            result_data=existing_results,
            completed_ts=time.ctime() if is_complete else None,
        )

        if is_complete:
            log.debug(f"{client.name} - job {juuid} completed")
        return

    # Handle 300 Pending - job still in progress
    if status == "300":
        worker = payload.get("worker")
        if worker and worker not in job["workers_started"]:
            job["workers_started"].append(worker)
            client.job_db.update_job(
                juuid,
                status=JobStatus.STARTED,
                workers_started=job["workers_started"],
            )
        return

    # Handle error statuses (4xx, 5xx)
    if status.startswith("4") or status.startswith("5"):
        error_msg = payload.get("error", payload.get("status", f"Error {status}"))
        client.job_db.update_job(
            juuid,
            status=JobStatus.FAILED,
            append_errors=[error_msg],
            completed_ts=time.ctime(),
        )
        log.error(f"{client.name} - job {juuid} failed: {error_msg}")
        return

dispatch_new_jobs(client)

Find NEW jobs and send POST requests to broker. Non-blocking: sends request and updates status to SUBMITTING.

Source code in norfab\core\client.py

def dispatch_new_jobs(client):
    """
    Find NEW jobs and send POST requests to broker.
    Non-blocking: sends request and updates status to SUBMITTING.
    """
    for job in client.job_db.fetch_jobs(
        [JobStatus.NEW], limit=client.dispatch_batch_size
    ):
        juuid = job["uuid"]

        try:
            # Send POST request (non-blocking)
            service = client.ensure_bytes(job["service"])
            uuid_bytes = client.ensure_bytes(juuid)
            workers = client.ensure_bytes(job["workers_requested"])
            request = client.ensure_bytes(
                {
                    "task": job["task"],
                    "kwargs": job["kwargs"] or {},
                    "args": job["args"] or [],
                }
            )

            client.send_to_broker(NFP.POST, service, workers, uuid_bytes, request)

            # Update status - receiver will handle the response
            client.job_db.update_job(
                juuid,
                status=JobStatus.SUBMITTING,
                last_poll_ts=time.time(),
            )
            log.debug(f"{client.name} - dispatched POST for job {juuid}")

        except Exception as e:
            msg = f"{client.name} - failed to dispatch job {juuid}: {e}"
            log.error(msg, exc_info=True)
            client.job_db.update_job(
                juuid,
                status=JobStatus.FAILED,
                errors=[msg],
                completed_ts=time.ctime(),
            )

poll_active_jobs(client)

Find active jobs and send GET requests to poll for results. Non-blocking: sends request with 5-second throttling via last_poll_timestamp.

Source code in norfab\core\client.py

def poll_active_jobs(client):
    """
    Find active jobs and send GET requests to poll for results.
    Non-blocking: sends request with 5-second throttling via last_poll_timestamp.
    """
    # Jobs that are ready for GET polling (dispatched or started)
    active_statuses = [JobStatus.DISPATCHED, JobStatus.STARTED]

    # fetch_jobs filters by min_poll_age to enforce polling throttle
    for job in client.job_db.fetch_jobs(
        active_statuses,
        limit=client.dispatch_batch_size,
        min_poll_age=client.poll_interval,
    ):
        juuid = job["uuid"]
        deadline = job["deadline"]
        now = time.time()

        # Check if job has exceeded deadline
        if now >= deadline:
            client.job_db.update_job(
                juuid,
                status=JobStatus.STALE,
                errors=["Job deadline reached without completion"],
                completed_ts=time.ctime(),
            )
            continue

        try:
            # Send GET request (non-blocking)
            service = client.ensure_bytes(job["service"])
            uuid_bytes = client.ensure_bytes(juuid)
            workers = client.ensure_bytes(job["workers_dispatched"])
            request = client.ensure_bytes(
                {
                    "task": job["task"],
                    "kwargs": job["kwargs"] or {},
                    "args": job["args"] or [],
                }
            )

            client.send_to_broker(NFP.GET, service, workers, uuid_bytes, request)

            # Update last_poll_ts to enforce 5-second throttle
            client.job_db.update_job(
                juuid,
                last_poll_ts=time.time(),
            )
            log.debug(f"{client.name} - sent GET poll for job {juuid}")

        except Exception as e:
            log.error(f"{client.name} - failed to poll job {juuid}: {e}", exc_info=True)

dispatcher(client)

Dispatcher thread: sends POST and GET requests asynchronously.

This thread: 1. Finds NEW jobs and sends POST requests to broker 2. Finds DISPATCHED/STARTED jobs and sends GET requests to poll for results

It does NOT wait for responses - the receiver thread handles all incoming messages and updates the database.

Parameters:

Name	Type	Description	Default
client	object	The client instance containing job_db, exit_event, and configuration.	required

Source code in norfab\core\client.py

def dispatcher(client):
    """
    Dispatcher thread: sends POST and GET requests asynchronously.

    This thread:
    1. Finds NEW jobs and sends POST requests to broker
    2. Finds DISPATCHED/STARTED jobs and sends GET requests to poll for results

    It does NOT wait for responses - the receiver thread handles all incoming
    messages and updates the database.

    Args:
        client (object): The client instance containing job_db, exit_event, and configuration.
    """
    while not client.exit_event.is_set() and not client.destroy_event.is_set():
        try:
            dispatch_new_jobs(client)
            poll_active_jobs(client)
        except Exception as e:
            log.error(f"{client.name} - dispatcher error: {e}", exc_info=True)
        time.sleep(0.1)