Skip to content

Architecture

The WorkManager is an event-driven worker system built on Dapr that enables you to deploy and manage workers written in different programming languages. Workers subscribe to pub/sub topics, process incoming CloudEvents, and optionally publish responses.

Note: "Core Components" and "Data Flow" below describe the original, single-service design. As of the WorkManager/Runtime split, every worker's Create/Start/Stop/Delete/LoadCode and message processing actually run in a separate Virtufin.WorkManager.Runtime service, via Dapr actors that each own one IEngine instance. See WorkManager/Runtime Split: Engines vs. Actors for how the two relate; the sections below still apply to the non-execution RPCs (GetTag(s)/SetTag(s)/GetWorkerHistory/startup recovery), which have not yet moved.

System Overview

flowchart TB
    subgraph WM["WorkManager Service"]
        W[WorkerManager<br/>Create/Start/Stop/<br/>Delete/Recover]
        ER[EngineRegistry<br/>text/x-python<br/>text/x-csharp<br/>...]
        WR[WorkerRegistry<br/>W1 W2 ...]
        W <--> ER
        ER <--> WR
        W --> DC[Dapr Client<br/>State / Pub/Sub]
        DC <--> DS[Dapr Subscriptions]
        DS --> WR
    end
    DC --> DR[Dapr Runtime]
    DR --> SS[(State Store<br/>statestore)]
    DR --> PS[(Pub/Sub<br/>pubsub)]

Core Components

WorkManager (Components/WorkManager.cs)

The central orchestrator that manages the entire worker lifecycle. It:

  • Creates, starts, stops, and deletes workers
  • Manages Dapr subscriptions to pub/sub topics
  • Persists worker configurations to the state store for recovery
  • Handles message processing and publishes output
  • Publishes worker output back to pub/sub topics

Worker (Runtime/Worker.cs)

Represents a single worker instance with:

  • Id: Unique identifier (GUID)
  • CodeSource: Where the code comes from (URL or inline content)
  • MimeType: The content type identifying the language/engine
  • Topics: The pub/sub topics to subscribe to (one or more)
  • Status: Either Stopped or Running
  • History: Track of all code changes over time

EngineRegistry (Runtime/EngineRegistry.cs)

Maps MIME types to engine implementations. Each engine factory creates new engine instances per worker.

Engine Implementations (Engines/)

All engines implement the IEngine interface:

public interface IEngine
{
    Task LoadCodeAsync(byte[] code, CancellationToken cancellationToken = default);
    Task<CloudEvent?> ProcessAsync(CloudEvent input, CancellationToken cancellationToken = default);
}

Available Engines:

Engine MIME Type Description
PythonEngine text/x-python, application/x-python Executes Python source or a packaged wheel via a python3 subprocess (one engine, format auto-detected)
CSharpSourceEngine text/x-csharp Compiles and executes C# source code via a subprocess
DotNetDllEngine application/x-dotnet-dll Loads pre-compiled .NET assemblies in-process (via an embedded CoreCLR)
NativeDllEngine application/x-native-dll Loads per-architecture native shared libraries in-process

Engine Process Model

Engines fall into two groups by how isolated a worker's code is from the host process:

flowchart TB
    subgraph Host["WorkManager / WorkManager.Runtime process (Native AOT)"]
        DDE[DotNetDllEngine]
        NDE[NativeDllEngine]
        SEB["SubprocessEngineBase&lt;br/&gt;(shared lifecycle + Unix-socket IPC)"]
        PE["PythonEngine&lt;br/&gt;(source or wheel, auto-detected)"]
        CSE[CSharpSourceEngine]
        SEB --- PE
        SEB --- CSE
    end
    DDE -->|hostfxr: embedded CoreCLR,<br/>same OS process| BR[DotNetDll.Bridge]
    PE -->|spawns| PY1[python3 subprocess]
    CSE -->|spawns| CSB[CSharpSource.Bridge subprocess<br/>Roslyn compile + AssemblyLoadContext]
    PY1 <-->|Unix socket:<br/>length-prefixed JSON frames| PE
    CSB <-->|Unix socket:<br/>length-prefixed JSON frames| CSE

DotNetDllEngine/NativeDllEngine run worker code in-process — a crash there can take the whole host process down with it. PythonEngine/CSharpSourceEngine both run worker code in a genuinely separate child OS process instead, via the shared SubprocessEngineBase (process lifecycle, health-pinging, auto-restart-on-crash, and the wire protocol are all implemented once, in the base class): a 4-byte length-prefixed JSON frame protocol over a Unix domain socket, with ready/process/result/ping/pong/shutdown/error frame types. CSharpSourceEngine's own child process (CSharpSourceEngine.Bridge) is a real, separately published, JIT-capable (non-AOT) executable — required because compiling C# source with Roslyn and then dynamically loading/executing the result needs a JIT, which a Native-AOT-published process does not have.

WorkerRegistry (Runtime/WorkerRegistry.cs)

In-memory storage for active worker instances. Workers are recovered from persistent state on restart.

WorkManager/Runtime Split: Engines vs. Actors

An IEngine implementation (PythonEngine, CSharpSourceEngine, ...) only knows how to load code and process one CloudEvent at a time — it has no idea what a "worker" is, doesn't persist anything, and doesn't know about topics. That's deliberate: an engine is a pure execution mechanism, reused identically whether it's invoked in-process or hosted remotely.

Everything else a worker needs — a stable identity, persisted configuration, topic subscriptions, Running/Stopped status, and being reachable across pod restarts and reassignments — is the job of a Dapr actor. As of the WorkManager/Runtime split, each execution mechanism has its own dedicated actor type running in a separate Virtufin.WorkManager.Runtime service/pod, never a shared generic "Worker" actor. text/x-python and application/x-python share one actor type (PythonWorker) since PythonEngine handles both formats itself -- they aren't distinct execution mechanisms, just two ways of delivering Python code (see the Engine Process Model diagram above):

MIME Type Engine Runtime Actor Type
text/x-python, application/x-python PythonEngine PythonWorker
text/x-csharp CSharpSourceEngine CSharpSourceWorker
application/x-dotnet-dll DotNetDllEngine DotnetDllWorker
application/x-native-dll NativeDllEngine NativeDllWorker

Each actor instance (keyed by worker id) owns exactly one engine instance for its lifetime, and delegates to it:

flowchart LR
    subgraph Actor["One actor instance, e.g. PythonWorkerActor (per worker id)"]
        Eng["PythonEngine<br/>(owned 1:1, does the actual work)"]
        Store["RuntimeWorkerStore<br/>(persisted RuntimeWorkerRecord)"]
        Pool["RuntimeTopicSubscriptionPool<br/>(topic subscribe/unsubscribe)"]
    end
    Actor -->|"LoadCodeAsync / ProcessAsync"| Eng
    Actor -->|"Save/Load on Create/Start/Stop/Delete"| Store
    Actor -->|"Subscribe/Unsubscribe on Start/Stop"| Pool

The gateway (Virtufin.WorkManager) never hosts these actors — it only proxies to them, resolving the actor type from the worker's mime type (RuntimeActorTypeNames.ForMimeType) for the execution-affecting RPCs: CreateWorker, StartWorker, StopWorker, DeleteWorker, LoadCode. Because the Runtime service is a separate process, a crash inside one worker's engine (a bad C# compile, a Python subprocess dying, a native library segfault) can only take down that Runtime pod — never the gateway's own query API (ListWorkers, GetWorkerHistory, etc. keep serving).

sequenceDiagram
    participant C as Client
    participant GW as WorkManager (gateway)
    participant RT as WorkManager.Runtime
    participant Actor as PythonWorkerActor
    participant Eng as PythonEngine (child process)

    C->>GW: CreateWorker(mime_type: text/x-python, code, topics)
    GW->>GW: RuntimeActorTypeNames.ForMimeType(mime_type) -> PythonWorker
    GW->>RT: Dapr actor invoke: IWorkerRuntimeActor.CreateAsync(payload)
    RT->>Actor: activate (mint new instance for this worker id)
    Actor->>Eng: LoadCodeAsync(code)
    Eng-->>Actor: ready
    Actor->>Actor: persist RuntimeWorkerRecord (status: Stopped)
    Actor-->>GW: ack
    GW-->>C: worker id

What stays on the old path, by design: GetTag(s)/SetTag(s)/GetWorkerHistory still go through the original in-process WorkerActor path described earlier in this document, for every mime type -- but only as a thin, persisted-config accessor now. Tags and history never touch execution, so they're meant to stay gateway-local permanently, not a temporary gap. Startup/on-demand recovery's engine construction and topic subscription, however, are gated off entirely for any mime type ForMimeType recognizes (a real bug, fixed after the split: recovering a Runtime-routed worker in-process too was duplicating its engine and subscription) -- recovery only does real in-process work for a mime type actor routing doesn't recognize.

Because a Dapr actor activates only when something invokes it, a Runtime worker whose topics go quiet has nothing to reactivate it (no subscription -> no messages -> no invocations): its subscriptions die with its pod restart or with the Runtime's 24h ActorIdleTimeout, permanently. The gateway closes that gap with RuntimeWorkerPokeService: it sweeps the persisted catalog (default every 10 minutes, RUNTIME_WORKER_POKE_INTERVAL_SECONDS to override, plus once on startup) and calls TouchAsync on each running Runtime-routed worker's actor -- a live actor gets a no-op, a deactivated one reactivates and its self-heal re-subscribes its topics. The re-subscription gap after a Runtime pod restart or an idle timeout is thereby bounded by the poke interval instead of being unbounded.

Data Flow

Worker Creation Flow

  1. Client calls CreateWorkerAsync() via gRPC or direct injection
  2. WorkManager validates the MIME type has a registered engine
  3. Code is fetched (from URL or extracted from content)
  4. Engine instance is created and code is loaded
  5. Worker configuration is persisted to Dapr state store
  6. Worker is added to the in-memory registry in Stopped status
  7. Client calls StartWorkerAsync() to subscribe to Dapr topic and set status to Running

Message Processing Flow

  1. Dapr delivers a CloudEvent to the subscription handler
  2. Worker engine processes the input CloudEvent
  3. If engine returns a response CloudEvent, publish to output topic

Worker Recovery Flow

On service restart:

  1. The WorkManagerRecoveryHostedService (BackgroundService) automatically calls RecoverWorkersAsync() during startup — no operator action or gRPC call required. The hosted service acquires a leader lease via the Dapr state store so concurrent instances elect a single recovery coordinator (see Worker Lifecycle: Recovery for the lease semantics).
  2. All worker configurations are loaded from state store
  3. For each worker:
  4. Engine is created and code is re-loaded
  5. Dapr subscription is re-established
  6. Worker status is set to Running

Persistence

Worker configurations are stored in the Dapr state store with these keys:

  • sys.workmanager.worker.index - List of all worker IDs
  • sys.workmanager.worker.<worker_id> - Individual worker configuration
  • sys.workmanager-runtime.worker.<worker_id> - The Runtime's worker record
  • sys.workmanager.recovery.leader - The recovery leader lease

All of them come from naming/virtufin.service.workmanager.yaml (WorkManagerNames); see Naming.

Code Source Types

Workers support two code source types:

CodeSourceUrl: Code fetched from an HTTP(S) URL at worker creation time 

{ "url": "https://example.com/worker.py" }

CodeSourceContent: Base64-encoded inline code 

{ "content": "aW1wb3J0IGpzb24..." }

Extension Points

Adding a New Engine

  1. Create a class implementing IEngine
  2. Register it in Program.cs:
engineRegistry.Register(
    new ContentType("text/x-my-language"),
    "MyLanguage",
    () => new MyLanguageEngine()
);

Custom Worker Logic

Workers must implement a ProcessAsync(CloudEvent input) function:

Python: 

def Process(event):
    return {
        "type": "com.example.output",
        "source": "my-worker",
        "data": { "result": event["data"] }
    }

C#: 

public static CloudEvent ProcessAsync(CloudEvent input)
{
    return new CloudEvent
    {
        Type = "com.example.output",
        Source = new Uri("urn:my-worker"),
        Data = input.Data
    };
}