Skip to content

Worker Lifecycle

The neops worker is a long-running process that discovers function blocks, registers them with the workflow engine, and continuously polls for jobs.

Startup Sequence

sequenceDiagram
    participant W as Worker
    participant FS as File System
    participant WE as Workflow Engine

    W->>W: Load environment (.env)
    W->>W: Check client/backend version compatibility
    W->>FS: Scan DIR_FUNCTION_BLOCKS directories
    FS-->>W: Discovered function block modules
    W->>W: Import modules (triggers @register_function_block)
    W->>WE: Register worker (name, IP)
    WE-->>W: worker_id
    W->>WE: Register function blocks (schemas, metadata)
    W->>W: Start heartbeat task
    W->>W: Start job polling task

1. Discovery

The worker scans directories listed in DIR_FUNCTION_BLOCKS (comma-separated paths). For each directory, it recursively imports all Python modules. Import triggers the @register_function_block decorator, which adds the class and its registration metadata to the global registry.

2. Registration

The worker registers itself with the workflow engine via the blackboard API, receiving a worker_id. It then sends all discovered function block registrations — including parameter/result JSON schemas, version information, and purity/idempotency flags.

3. Concurrent Tasks

Two async tasks run concurrently on the main event loop:

Task Interval Purpose
Heartbeat HEARTBEAT_INTERVAL (default 20s) Keeps the worker marked as online in the workflow engine
Job polling POLL_INTERVAL (default 10s) Fetches pending jobs from the blackboard

Job Execution

When the polling task receives jobs:

  1. Match job.function_block_id against the local registry.
  2. Submit the job to a ThreadPoolExecutor with max_workers=1.
  3. Inside the worker thread, call asyncio.run() to execute the function block (this gives each job its own event loop).
  4. Start blocking detection monitoring on the main loop.
  5. Collect the result and report it back to the blackboard.

Sequential Processing

Jobs are processed one at a time (max_workers=1). This avoids:

  • Connection exhaustion on shared devices.
  • Race conditions in context data.
  • Unpredictable resource usage.

If the workflow engine pushes multiple jobs, they are queued and executed in order.

Job Types

Type Method called Purpose
ACQUIRE acquire() Lock resources and expand context
EXECUTE execute_function_block() → run() Main function block logic (wrapped with error boundaries). The result carries sideEffects only when it is False, and the rollback data the block saved, also after an exception. A pure block whose result claims side effects is failed instead, with retryable=False. A failure before run() was entered gets side_effects=False and retryable=False
ROLLBACK rollback_function_block() → rollback() Undo this step’s changes when the run failed; the job mirrors the forward job’s parameters and context and carries forwardResult, which the worker parses into the block’s ForwardResult (data and rollback data by alias). A missing or unparseable forwardResult fails the job with retryable=False without calling rollback(). The default rollback() returns failure — override it in every block registered reversible=True

Graceful Shutdown

The worker handles SIGINT and SIGTERM signals:

  1. Set the shutdown event: no new polls. A job the last poll already returned still runs to its result, because it is claimed by this worker and the engine cannot release it.
  2. Keep the heartbeat running while the current job finishes, so the engine does not declare the worker offline mid-job.
  3. Wait at most SHUTDOWN_TIMEOUT seconds for the running job.
  4. Stop the heartbeat and deregister the worker from the workflow engine.
  5. Exit with code 0, or with code 1 when the timeout elapsed.

A job left behind is not re-queued

When SHUTDOWN_TIMEOUT elapses the worker logs that the job’s outcome is unknown, deregisters, and ends its process at once. A thread cannot be cancelled, so the job is stopped wherever it is, possibly with a device change half applied; ending the process is what keeps it from running on while the engine compensates it elsewhere. The engine never re-queues a claimed job: its cleanup fails the job as worker-offline and treats it as possibly executed, so a reversible block is rolled back and an irreversible one ends the run FAILED_UNSAFE for a human to resolve. Size the timeout for your longest job, and keep the orchestrator’s termination grace period above it.

Worker Expiry

If the worker fails to send heartbeats (e.g., network partition), the workflow engine marks it as expired.

  • The next heartbeat attempt receives a 404, causing WorkerExpiredError.
  • The worker logs the error and shuts down (exit code 1).
  • Jobs that were in-flight are failed by the engine as worker-offline (never re-queued) and compensated where the block is reversible.

The worker does not attempt to re-register automatically. Use your orchestrator’s restart policy (Docker restart: unless-stopped, Kubernetes restartPolicy: Always) to bring it back.

Upgrading Past Re-Versioned Base Blocks

A worker serves exactly the block versions its SDK registers, and the engine resolves a workflow step to the exact version it pins. The release that adds side_effects and rollback data moves seven base blocks from 0.1.0 to 0.1.1:

  • fb.base.neops.io/device_update_device, group_update_group, global_add_devices and global_add_groups are now registered pure, because declared db_updates are not side effects.
  • fb.base.neops.io/network_device_apply_configuration, network_device_parse_command_output and network_device_reload now report side_effects=False when they refuse before touching the device.

Symptom. A stored workflow definition that still pins one of these at 0.1.0 keeps starting, and it takes its CMS locks. But once no worker registers 0.1.0, its job stays pending. The engine fails it after BLACKBOARD_PENDING_JOB_TIMEOUT, 60 minutes by default.

Fix, before you replace the workers. Pick one:

  • Republish each affected definition with the pin moved to 0.1.1. Bump the workflow’s own version as well: published content is immutable.
  • Keep one worker on the previous SDK image until the runs pinned to 0.1.0 have drained.

Order. Upgrade the workflow engine first, then the workers; an older worker keeps working against the newer engine. To roll back, do it the other way round: workers first, then the engine.