Skip to content

Running workers locally

The local WorkManager runs workers in-process, in a test or a notebook. Workers subscribe to topics on a local bus, their outputs are routed like pubsub, and every event goes through the same per-event rules as on the cluster. A local run is a rehearsal of a deployment: the same worker, the same configuration, the same routing, without a cluster.

It ships in two packages with the same semantics:

  • C#: LocalWorkManager and WorkerObservables.Run in Virtufin.Behavior.DevKit;
  • Python: virtufin.worker_devkit.local in virtufin-worker-devkit, which needs the [local] extra (reactivex).

Local and on the cluster

Locally, everything happens in one process. A feedback loop, such as a strategy whose orders come back to it as positions, is a set of workers wired by topics, exactly as it would be deployed.

flowchart LR
    NB["Notebook or test"] -->|"PublishAsync(topic, IObservable)"| BUS
    subgraph LWM["LocalWorkManager (one process)"]
        BUS["Local bus<br/>one subject per topic<br/>one global FIFO"]
        RULES["WorkerEventRules<br/>prepare, process, finish, route"]
        subgraph W["Workers, in deploy order"]
            S["Strategy"]
            E["SimulatedExecutor"]
            P["FillPositions"]
        end
        BUS -->|"codec bytes"| RULES
        RULES <--> W
        RULES -->|"outputs by publishtopic or ce-type"| BUS
        RULES -->|"worker throws"| DLQ["dlq.#lt;topic#gt;"]
    end
    S -. "order submitted" .-> E
    E -. "order filled" .-> P
    P -. "position" .-> S
    BUS -->|"Observe(topic)"| OBS["Observers: tables, charts, asserts"]
    DLQ -->|"ObserveDeadLetters(topic)"| OBS

On the cluster, the same worker runs in a WorkManager Runtime actor. The per-event rules are the same code: WorkerEventRules lives in Virtufin.Worker.Host, which both the Runtime and Virtufin.Behavior.DevKit use.

flowchart LR
    NB["Notebook"] -->|"gateway CreateWorker<br/>text/x-csharp or nupkg"| WM["WorkManager"]
    WM --> ACT
    subgraph RT["WorkManager Runtime"]
        POOL["Redis-stream<br/>subscription pool"]
        ACT["Runtime actor<br/>engine: C# source bridge, DLL or Python"]
        RULES["WorkerEventRules<br/>prepare, process, finish, route"]
        OUT["Outbox"]
        POOL -->|"codec bytes"| ACT
        ACT <--> RULES
        RULES --> OUT
    end
    TOPICS["Pubsub topics"] --> POOL
    OUT -->|"outputs by publishtopic or ce-type"| TOPICS
    RULES -->|"worker throws"| DLQ["dlq.#lt;topic#gt;"]

One event, side by side

Each event a worker receives goes through the same steps locally and in the Runtime.

sequenceDiagram
    autonumber
    participant Pub as Publisher
    participant Bus as Bus (local) or pubsub (cluster)
    participant Rules as WorkerEventRules
    participant W as Worker
    participant Sub as Subscribers
    Note over Bus,Rules: Same code locally and in the Runtime
    Pub->>Bus: publish(topic, event)
    Bus->>Rules: event as CloudEvents protobuf bytes
    Rules->>Rules: PrepareInput: datacontenttype default,<br/>config fills missing attributes, secrets win
    Rules->>W: ProcessAllAsync(input)
    alt worker returns outputs
        W-->>Rules: outputs, in order
        Rules->>Rules: FinishOutputs: id and source defaults,<br/>secret attributes stripped
        Rules->>Rules: Route: publishtopic (stripped), else ce-type
        Rules->>Sub: each output to its topic
        Note right of Rules: an output with neither is logged and skipped
    else worker throws
        W--xRules: exception
        Rules->>Sub: DeadLetter: the original event with dlq* attributes,<br/>once, to dlq.#lt;topic#gt;
        Note right of Rules: processing continues with the next event
    end

API

C

using System.Reactive.Linq;
using Virtufin.Behavior.DevKit;

using var wm = new LocalWorkManager();

// Consumers first, as on the cluster: the bus keeps no history.
// executor, positions and strategy are IWorker instances.
wm.Deploy(executor, [submittedTopic], config: executorConfig);
wm.Deploy(positions, [submittedTopic, filledTopic], config: positionsConfig);
wm.Deploy(strategy, [candleTopic], config: strategyConfig, name: "breakout");

var orders = new List<CloudEvent>();
using var o = wm.Observe(submittedTopic).Subscribe(orders.Add);
using var d = wm.ObserveDeadLetters(candleTopic).Subscribe(dl => Console.WriteLine(dl["dlqerrormessage"]));

await wm.PublishAsync(candleTopic, candles.ToObservable());  // each candle drained before the next
Member What it does
new LocalWorkManager(options?) A local WorkManager. LocalWorkManagerOptions: MaxCascade (default 10 000), ActorType (the dlqactortype, default LocalWorkManager), Logger, TimeProvider (for dlqfailedtime).
Deploy(worker, topics, config?, secrets?, name?) Subscribes an IWorker to topics; returns a LocalWorker (Id, Name, Topics, Config, SecretNames, Processed, DeadLettered).
Remove(localWorker) Stops delivering to it.
Workers The deployed workers, in deploy order.
Observe(topic) Every event published to the topic from now on (hot, no history).
ObserveDeadLetters(topic) The dead letters of the topic (dlq.<topic>).
PublishAsync(topic, CloudEvent) Publishes one event; completes when everything it caused is drained.
PublishAsync(topic, IObservable<CloudEvent>) Publishes a sequence, each event drained before the next; completes with the sequence and fails with its error.
Complete() Completes every observer; later publishes fail. Dispose() completes and releases the bus.

Without topics, worker.Run(inputs, config?, secrets?) (WorkerObservables) runs one worker over an IObservable<CloudEvent>. It processes one event at a time, in order, under the same rules, and emits every output in wire form with its publishtopic kept. Nothing is routed, nothing loops back, and a throw ends the sequence with OnError.

Python

from virtufin.worker_devkit.local import LocalWorkManager

with LocalWorkManager() as wm:
    wm.deploy(executor, [submitted_topic], config=executor_config)
    wm.deploy(positions, [submitted_topic, filled_topic], config=positions_config)
    wm.deploy(strategy_module, [candle_topic], config=strategy_config)

    orders = []
    wm.observe(submitted_topic).subscribe(orders.append)
    wm.observe_dead_letters(candle_topic).subscribe(print)

    wm.publish(candle_topic, candles)   # synchronous; a list, an Observable or one dict

The Python API mirrors the C# one in snake case: deploy, remove, workers, observe, observe_dead_letters, publish, complete, close, and a context manager. deploy accepts a WorkerBase, a Process(event) function or a module that defines Process, which is the same entry point the cluster's Python engine calls. publish is synchronous: it returns once the cascade is drained. Events are the dicts a Python worker sees on the cluster. run(worker, inputs, config, secrets) returns an Observable of output dicts. Install it with pip install 'virtufin-worker-devkit[local]'. Deployed workers do not need reactivex.

The package also exposes the pieces underneath: wire, the protobuf codec of the Python engine, and event_rules, the per-event rules as a Python worker sees them.

Semantics

These are the Runtime's rules, applied by the same code (C#) or pinned by the same conformance vectors (Python):

  • Codec bytes on the bus. Every event crosses the bus as CloudEvents protobuf bytes and is decoded for each receiver. A worker sees what the codec carries, and one receiver's changes never reach another.
  • Input rules. A missing datacontenttype becomes application/json. The worker's config fills the attributes the event leaves out, and the event's own attributes win. The worker's secrets are set over everything.
  • Output rules. An output without an id gets a fresh one. In C#, an output without a source gets the input's. Secret attributes are stripped.
  • Routing. An output goes to its publishtopic (removed from the published event), else to its ce-type. In C#, an output with neither is logged and skipped. A Python output without a type goes to the topic unknown, and one without a source gets urn:unknown, as on the cluster's Python engine.
  • Failures. A throw dead-letters the original event, as it was on the bus, once to dlq.<topic>. The dead letter carries dlqworkerid, dlqactortype, dlqerrortype, dlqerrormessage (at most 1024 characters) and dlqfailedtime. The worker carries on with the next event. A Python worker's exception is reported as System.InvalidOperationException with str(exception), as the Python engine reports it.
  • Workers. C# workers are called through ProcessAllAsync, so IMultiOutputWorker outputs are published in order. Python workers return None, a dict, or a list of dicts.

The vectors are conformance/worker-event-rules.v1.json. workmanager owns them and ships them in the Virtufin.Worker.Host package. worker-devkit keeps a byte-identical copy, and a test fails when the copy drifts from the pinned package. The C# and the Python test suites both run every vector.

Ordering

The local bus is deterministic. There is one global FIFO. Each external publish is drained completely, breadth-first, before the next one starts. On each event, the observers of its topic see it first, then the subscribed workers in deploy order. A publish from inside the cascade (from a worker or an observer) is queued, never run recursively. MaxCascade (max_cascade) caps the events one publish may cause. Beyond the cap the publish fails, which catches a worker whose output reaches its own input topic.

sequenceDiagram
    autonumber
    participant NB as Notebook
    participant Bus as Local bus (FIFO)
    participant S as Strategy
    participant E as SimulatedExecutor
    participant P as FillPositions
    NB->>Bus: candle 1
    Bus->>S: candle 1
    S-->>Bus: order submitted
    Bus->>E: order submitted
    Bus->>P: order submitted
    E-->>Bus: order filled
    Bus->>P: order filled
    P-->>Bus: position
    Bus->>S: position
    Note over NB,P: candle 1's cascade is drained before candle 2
    NB->>Bus: candle 2
    Bus->>S: candle 2 (the strategy already knows its position)
    Note over NB,P: On the cluster nothing waits: a replay pauses after<br/>each candle that ordered until its position arrives

On the cluster, workers run concurrently and pubsub does not wait for a cascade. A replay that is compared with a local run has to pace itself: after each candle on which the local run ordered, wait for that candle's position event before publishing the next candle.

What it does not emulate

The local WorkManager is a rehearsal, not a simulation of the infrastructure. These differences are deliberate:

  • Redelivery. Locally, every event is delivered exactly once. Pubsub can redeliver after a timeout or a restart.
  • Batching. The Runtime publishes a worker's outputs in batches. Locally, each output is queued on its own, in the same order.
  • Backpressure. There is no bounded inbox, and a slow worker never falls behind. A cascade runs to the end before the next external event.
  • The tail race on deploy. A cluster worker subscribes from the tail, so events published while it is being created can be missed. Locally, Deploy takes effect at once. On the cluster, deploy consumers first and wait for the subscription before publishing.
  • Reload state resets. The Runtime reloads a worker when a newer library is released, and the worker's in-memory state starts fresh. A local worker keeps its state for as long as it is deployed.
  • The dead letter's error type. Locally, dlqerrortype is the worker's own exception type. On the cluster, the failure crosses the Dapr actor call first, so the deployed Runtime often reports the transport's exception type instead (see dead letters). dlqactortype is LocalWorkManager instead of the engine's actor type.

See the worker workflow for where a local run fits between unit tests and a deployment.