Skip to content

Worker Management

Purpose

Event-driven worker management system enabling deployment and execution of workers written in multiple languages (Python, C#, .NET DLL, NativeAOT) that subscribe to Dapr pub/sub topics and process CloudEvents. Every process subscribed to a topic receives every message on it (one consumer group per sidecar), and each worker runs in exactly one process; see the Cross-Instance Coordination and horizontal scalability requirements below.

Topic and state-key naming for worker lifecycle events is defined in pubsub-topics/spec.md. The lifecycle topic is workmanager.lifecycle; finer-grained infra events (heartbeats, instance state changes) live under workmanager.infra.*.

Requirements

Requirement: Worker Lifecycle Management

The WorkManager SHALL support creating, starting, stopping, and deleting workers. Workers SHALL persist their configuration and restore automatically on service restart.

Scenario: Create and start a worker

  • WHEN a client creates a worker with a topic subscription and code payload, then starts it
  • THEN the worker SHALL be persisted to the state store, and once its code is loaded it SHALL be subscribed to the topic and set to Running status

Scenario: Hot-reload worker code

  • WHEN a client updates a running worker's code
  • THEN the worker SHALL switch to the new code without restarting the service

Scenario: Auto-recovery on restart

  • WHEN the WorkManager service restarts
  • THEN all previously persisted workers SHALL be restored to their saved state before the health check passes

Scenario: Worker version history

  • WHEN a worker's code is updated
  • THEN the previous code version SHALL be preserved in a versioned history

Scenario: Create returns before the code is loaded

  • WHEN a client creates a worker
  • THEN the WorkManager SHALL persist it and return with status Loading without waiting for the code fetch, dependency resolution or engine load
  • AND it SHALL load the code in the background and set the status to Stopped (or Running, see deferred start) once loaded, or LoadFailed on failure

Scenario: Start a worker that is still loading

  • WHEN a client starts a worker whose status is Loading
  • THEN the WorkManager SHALL accept the request and start the worker (subscribe it and set it to Running) as soon as loading completes

Scenario: Start a worker whose load failed

  • WHEN a client starts a worker whose status is LoadFailed
  • THEN the WorkManager SHALL reject the request with FAILED_PRECONDITION

Requirement: Service Architecture

The WorkManager SHALL use the virtufin-api gRPC for all state operations. The WorkManager SHALL NOT call DaprClient.GetStateAsync, SaveStateAsync, DeleteStateAsync, or any other Dapr state API directly. The Dapr sidecar SHALL remain in use for service-invocation mTLS, distributed tracing, and metrics.

Scenario: Service persists worker state

  • WHEN the WorkManager saves or retrieves worker configuration
  • THEN it SHALL call the API's State gRPC service
  • AND it SHALL NOT call any Dapr state API directly

Scenario: Service subscribes to a worker topic

  • WHEN the WorkManager subscribes to a worker's input topic
  • THEN it SHALL call Pubsub.Subscribe on the virtufin-api
  • AND it SHALL NOT call DaprPublishSubscribeClient.SubscribeAsync directly

Requirement: Worker Lifecycle Events

The WorkManager SHALL publish worker lifecycle events to the workmanager.lifecycle topic on every state transition. Topic, ce-type, and state-key conventions are defined in pubsub-topics/spec.md. Events SHALL be CloudEvents v1.0 envelopes with ce-type = com.virtufin.workmanager.lifecycle.<state>.

Scenario: Worker created

  • WHEN a worker is created
  • THEN the WorkManager SHALL publish a worker.created event with worker_id and topic in the data payload

Scenario: Worker started

  • WHEN a worker is started
  • THEN the WorkManager SHALL publish a worker.started event

Scenario: Worker stopped

  • WHEN a worker is stopped
  • THEN the WorkManager SHALL publish a worker.stopped event

Scenario: Worker error

  • WHEN a worker process fails or throws
  • THEN the WorkManager SHALL publish a worker.error event with error_type and error_message in the data payload

Scenario: Worker loaded

  • WHEN a worker's code and dependencies finish loading
  • THEN the WorkManager SHALL publish a worker.loaded event with worker_id and the resolved Virtufin.* library versions in the data payload

Scenario: Worker load failed

  • WHEN loading a worker's code or dependencies fails
  • THEN the WorkManager SHALL publish a worker.load_failed event with error_type and error_message in the data payload

Requirement: Polyglot Worker Engines

The WorkManager SHALL support executing worker code in multiple languages via a pluggable engine architecture. Built-in engines SHALL include Python (subprocess), C# (Roslyn runtime compilation), pre-compiled .NET DLL (in-process, the default DotNetDllEngine via hostfxr + AssemblyLoadContext), and a native DLL engine (in-process, NativeDllEngine, see the Native DLL Engine / Native Worker ABI requirements below).

Scenario: Python worker execution

  • WHEN a worker with a Python engine processes a message
  • THEN the worker code SHALL execute in a dedicated Python subprocess with a configurable timeout

Scenario: C# source worker execution

  • WHEN a worker with a C# source engine processes a message
  • THEN the code SHALL be compiled at runtime via Roslyn and executed in-process

Scenario: DLL worker execution (in-process, default)

  • WHEN a worker with MIME type application/x-dotnet-dll processes a message
  • THEN the worker DLL SHALL be loaded into an isolated AssemblyLoadContext and executed in-process in the WorkManager, with sub-microsecond per-call latency

Scenario: Engine extension

  • WHEN a new engine implementation conforms to the IEngine interface
  • THEN it SHALL be registrable in the engine registry without modifying core WorkManager code

Requirement: Event-Driven Message Processing

Workers SHALL subscribe to Dapr pub/sub topics and process incoming CloudEvents. Each output CloudEvent's publishtopic extension, when present, SHALL determine its publish topic (the extension is removed before publishing); otherwise its type field SHALL.

Scenario: Message processing

  • WHEN a CloudEvent is published to a topic a worker subscribes to
  • THEN the worker's code SHALL be invoked with the event data and any output events SHALL be published to the topic named by their publishtopic extension, or else their type field

Scenario: Several outputs from one input

  • WHEN a multi-output worker returns several CloudEvents for one input
  • THEN every one of them SHALL be published, in the order returned, each to its own topic by the rule above
  • AND an empty result SHALL publish nothing

Scenario: Concurrent message handling

  • WHEN multiple messages arrive for the same worker
  • THEN they SHALL be processed concurrently (the engine's ProcessAsync is re-entrant; the worker author is responsible for internal synchronization)

Requirement: Worker-Addressed Topics

A CreateWorker topic MAY contain the placeholder {workerId}. The WorkManager SHALL replace every occurrence with the id it mints for the worker, before the worker is registered anywhere, and SHALL persist and report the expanded topic. A topic names what its data is, not who happens to consume it: commands meant for one worker instance SHALL be addressed by that instance's id, never by a caller-side run or session tag.

Scenario: Command topic addressed by the worker

  • WHEN a client creates a worker with topic command.httpgetworker.{workerId}
  • THEN the worker SHALL be subscribed to command.httpgetworker.<its id>, and GetWorker/ListWorkers SHALL report that expanded topic
  • AND the client SHALL derive the topic from the id CreateWorker returns (to target a trigger or publish commands)

Scenario: Orphans stay isolated

  • WHEN a client dies before deleting its workers and triggers, and a later client creates workers of the same kind
  • THEN the orphaned triggers SHALL only reach the orphaned workers, because each worker's command topic is its own

Requirement: Trigger Ownership

A trigger MAY name the worker it exists for (owner_worker_id on Trigger.ScheduleTrigger). When the WorkManager deletes a worker it SHALL delete every trigger owned by that worker, after the worker itself is deleted, so a trigger never outlives the worker it drives. A failure to delete owned triggers SHALL be logged and SHALL NOT fail the worker deletion. Triggers without an owner SHALL be unaffected.

Scenario: Deleting a worker deletes its triggers

  • WHEN a worker that owns triggers is deleted
  • THEN its owned triggers SHALL be deleted and SHALL stop firing
  • AND triggers owned by other workers, and unowned triggers, SHALL remain

Requirement: Cross-Instance Coordination

Messages SHALL reach WorkManager processes by per-sidecar delivery, not competing consumers: the pub/sub component gives every sidecar its own consumer group (consumerID: {podName}), so every process that holds a subscription for a topic receives every message on it (Pub/Sub Topics spec, "Subscription Implementation Pattern"). A process subscribes to a topic only while it hosts a worker for it, and each worker is hosted by exactly one process (its owning instance or Runtime actor placement), so each worker processes each message exactly once. A message that reaches a process with no worker registered for the topic (e.g. one that left between delivery and dispatch) SHALL be dropped, acked with no redelivery (an explicit business-logic skip per the Pub/Sub Topics spec's "Retry Policy on Dapr Subscribe Exceptions" requirement, not a processing failure).

This replaces the competing-consumer model (shared app-id, each message to exactly one replica) and the earlier distributed-lock approach before it (removed in LIBRARY_VERSION 0.2.5): the lock-based group coordination (GroupLockOptions, LockContentionException, AcquireGroupLockAsync, ReleaseGroupLockAsync, VIRTUFIN_WORKER_GROUP) is no longer supported.

Scenario: Message lands on a replica with no matching worker

  • WHEN a message reaches a WorkManager process that has no worker registered for that topic
  • THEN the message SHALL be dropped and acked with no redelivery (an explicit business-logic skip, not a processing failure)
  • AND no re-routing SHALL occur: every process hosting a worker for the topic receives the message itself

Scenario: Every subscribed process receives every message

  • WHEN two processes (WorkManager or Runtime pods) each host a worker subscribed to topic orders
  • THEN each process SHALL receive every message published to orders, and each worker SHALL process it once
  • AND no message SHALL be load-balanced between the processes

Requirement: Worker Code Sources

Workers SHALL accept code from inline Base64-encoded content or from HTTP(S) URLs. URL-based sources SHALL be validated against SSRF and host allowlist constraints as defined in the cross-cutting security requirements. URL-based sources MAY carry a per-request credential (see Worker Source Authentication) used only for that fetch.

Scenario: Inline code worker

  • WHEN a worker is created with a Base64-encoded code string
  • THEN the decoded code SHALL be passed to the engine for execution

Scenario: URL code worker

  • WHEN a worker is created with a URL pointing to code
  • THEN the code SHALL be fetched, SHA-256 verified, and passed to the engine

Scenario: Blocked URL

  • WHEN a worker's code URL targets a private IP or a host not in the allowlist
  • THEN the worker creation SHALL fail with a security rejection

Requirement: Python Sandboxing

Python worker subprocesses running raw source code (text/x-python) SHALL run with restricted interpreter flags and a package allowlist; all non-allowed imports SHALL be blocked. A Python worker packaged as a wheel (application/x-python) is NOT subject to this blocklist/allowlist: its dependencies are declared and vetted via its build-time Requires-Dist manifest instead of being restricted at runtime.

Scenario: Allowed import

  • WHEN a raw-source Python worker imports a package listed in PYTHON_ALLOWED_PACKAGES
  • THEN the import SHALL succeed

Scenario: Blocked import

  • WHEN a raw-source Python worker imports a package not in the allowlist
  • THEN the import SHALL fail with an error

Scenario: Wheel-packaged worker import

  • WHEN a Python worker packaged as a wheel imports any module
  • THEN the import SHALL succeed regardless of the blocklist/allowlist, since its dependencies were vetted at publish time via the wheel's declared Requires-Dist manifest, not restricted at runtime

Requirement: Worker Contract (IWorker)

The IWorker interface SHALL define an asynchronous ProcessAsync method returning Task<CloudEvent?>. A worker that can produce several outputs per input SHALL additionally implement IMultiOutputWorker, whose ProcessManyAsync returns Task<IReadOnlyList<CloudEvent>>. WorkerBase SHALL accept constructor parameters (responseType, source) instead of abstract property overrides for CloudEvent routing.

Scenario: Async worker process

  • WHEN the WorkManager invokes a worker's ProcessAsync method
  • THEN the call SHALL be awaited asynchronously without blocking threads

Scenario: Multi-output worker

  • WHEN the engine hosts a worker that implements IMultiOutputWorker
  • THEN it SHALL call ProcessManyAsync and publish every returned CloudEvent in order
  • AND a worker implementing only IWorker SHALL keep its single-output behavior unchanged

Scenario: Multi-output worker on a single-output host

  • WHEN a host that predates IMultiOutputWorker calls a multi-output worker's ProcessAsync
  • THEN the worker SHALL return its only output when it has one, and otherwise an error response naming the host's single-output limit, rather than silently dropping outputs

Scenario: Native ABI stays single-output

  • WHEN a worker runs through the native (FlatBuffers) ABI
  • THEN it SHALL return at most one CloudEvent per input; multiple outputs are a managed (.NET, Python) worker capability

Scenario: Constructor-based configuration

  • WHEN a worker extends WorkerBase or CommandWorker
  • THEN it SHALL pass the response CloudEvent type and source URI via the constructor, not via abstract property overrides

Scenario: ApiWorker with pre-configured ApiClient

  • WHEN a worker extends ApiWorker with apiHost, apiPort, responseEventType, and source
  • THEN the base class SHALL create an ApiClient at the configured host/port and provide it to the abstract HandleAsync(CloudEvent, JsonNode, ApiClient) method (ApiWorker's ProcessAsync(CloudEvent, JsonNode) override dispatches to it)

Scenario: ApiCommandWorker with pre-configured ApiClient

  • WHEN a worker extends ApiCommandWorker with apiHost, apiPort, responseEventType, and source
  • THEN the base class SHALL create an ApiClient at the configured host/port and make it available during HandleCommandAsync

Requirement: CommandWorker Base Class

A CommandWorker abstract class SHALL extend WorkerBase to simplify command-based workers. It SHALL parse the CloudEvent data as JSON, extract a "command" field, and dispatch to the abstract HandleCommandAsync method. Unknown commands SHALL produce automatic error responses.

Scenario: Command dispatch

  • WHEN a CloudEvent arrives with {"command": "hello", ...} as data
  • THEN CommandWorker SHALL parse the JSON, extract "hello", and call HandleCommandAsync(input, "hello", node)

Scenario: Unknown command

  • WHEN CommandWorker cannot match the command to any handler
  • THEN it SHALL return an error response with "Unknown command: <command>"

Scenario: Correlation ID propagation

  • WHEN CommandWorker creates a response via Response(input, ...) or Error(input, ...)
  • THEN the correlationid from the input CloudEvent SHALL be propagated to the output CloudEvent via the WithCorrelationId extension method

Requirement: Correlation ID Propagation

Worker responses SHALL propagate the correlationid extension attribute from the input CloudEvent to the output CloudEvent. The CloudEvent.WithCorrelationId(input) extension method SHALL handle this as a no-op when absent.

Scenario: Response with correlation ID

  • WHEN a worker creates a Response() or Error() via WorkerBase helpers
  • THEN the output CloudEvent SHALL carry the same correlationid as the input CloudEvent if present

Scenario: Input without correlation ID

  • WHEN a worker processes an input CloudEvent without a correlationid extension
  • THEN the output CloudEvent SHALL NOT have a correlationid extension

Requirement: Generic CommandWorker Base Class

The Virtufin.Worker.DevKit package SHALL provide a generic CommandWorker<T> where T : struct, Enum and a generic ApiCommandWorker<T> where T : struct, Enum as the canonical base classes for command-based workers. The wire protocol of the CloudEvent data.command field SHALL be the lowercase identifier of the enum value (e.g. enum WebSocketManagerCommand.create corresponds to the wire string "create"). The base class SHALL parse the wire string with Enum.TryParse<T>(command, ignoreCase: true, out _) and dispatch to an abstract HandleAsync(CloudEvent, T, JsonNode) method. An unknown wire command SHALL produce an error response. The non-generic CommandWorker / ApiCommandWorker types are removed (alpha, no backward compatibility).

Scenario: Generic dispatch on valid command

  • WHEN a CloudEvent arrives with {"command": "create", ...} and the worker is CommandWorker<MyEnum> where MyEnum.create exists
  • THEN the base SHALL parse "create" into MyEnum.create and invoke HandleAsync(input, MyEnum.create, node)

Scenario: Wire name == identifier (lowercase)

  • WHEN a worker defines an enum value WebSocketManagerCommand.create
  • THEN the corresponding wire command string SHALL be the lowercase identifier "create"

Scenario: Unknown command

  • WHEN the wire command string does not match any enum value
  • THEN the base SHALL return an error response with message "Unknown command: <command>"

Scenario: Missing command field

  • WHEN the CloudEvent data has no "command" field
  • THEN the base SHALL return an error response with message "Missing 'command' field"

Scenario: Malformed JSON

  • WHEN the CloudEvent data cannot be parsed as JSON or parses to null
  • THEN the base SHALL return an error response with message "Failed to parse command JSON" (or wrap the parser exception in "Internal error: …")

Scenario: Generic ApiCommandWorker

  • WHEN a worker extends ApiCommandWorker<T> and the api_host field is present in the command JSON
  • THEN the base SHALL create or reconfigure the ApiClient for the parsed host/port (default port 5002) and make it available via the Api property during HandleAsync

Scenario: Backend invocation via dynamic gateway

  • WHEN a worker invokes a backend service via dynamic gateway = Api.Gateway; await gateway.<service>.<method>(requestData);
  • THEN the ServiceClient returned by <service> SHALL resolve to the named service and <method> SHALL invoke the corresponding method via the API Gateway

Scenario: New enum value triggers compiler warning on the switch

  • WHEN a new value is added to the command enum and the worker's HandleAsync switch expression does not handle it
  • THEN the C# compiler SHALL emit a warning (CS8509 or equivalent) for the unhandled enum value

Requirement: In-Process DotNet DLL Engine

The WorkManager SHALL support an in-process DotNetDllEngine (default for MIME type application/x-dotnet-dll) that loads pre-compiled .NET worker DLLs directly into the AOT-compiled WorkManager process. The engine SHALL embed the .NET runtime (CoreCLR) into the host process on first use via the hostfxr C API and SHALL load the worker DLL into a per-worker, collectible AssemblyLoadContext. The engine SHALL be AOT-compatible: the engine code itself is compiled with <IsAotCompatible>true</IsAotCompatible> and SHALL NOT use reflection on its own types; the only reflection SHALL be on the JIT-loaded worker assembly.

Scenario: First LoadCodeAsync initializes CoreCLR

  • WHEN the first LoadCodeAsync is called for MIME type application/x-dotnet-dll
  • THEN the engine SHALL load libhostfxr via NativeLibrary.Load and call hostfxr_initialize_for_runtime_config to initialize CoreCLR in the host process; the cost SHALL be paid once per WorkManager process

Scenario: Subsequent LoadCodeAsync in the same process

  • WHEN a second LoadCodeAsync is called after CoreCLR is already initialized
  • THEN the engine SHALL skip the hostfxr initialization and load the new worker DLL directly

Scenario: Per-worker AssemblyLoadContext

  • WHEN a LoadCodeAsync is called with a worker nupkg
  • THEN the engine SHALL extract the worker DLL (and its sibling dependencies) from the nupkg's lib/<tfm>/ entries, create a new collectible AssemblyLoadContext, and load the worker assembly into it

Scenario: DevKit resolution

  • WHEN the JIT-loaded worker assembly references Virtufin.Worker.DevKit
  • THEN the engine SHALL resolve the reference to the AOT-compiled Virtufin.Worker.DevKit instance already loaded in the WorkManager process (so the worker and the engine see the same type identity)

Scenario: Direct in-process dispatch

  • WHEN ProcessAsync is called on the engine with a CloudEvent
  • THEN the engine SHALL call the worker's IWorker.ProcessAsync directly, with no socket, no JSON serialization, and no subprocess scheduling; the per-call latency SHALL be sub-microsecond

Scenario: Runtime discovery on target

  • WHEN the WorkManager runs on a target that does not have a compatible .NET runtime installed
  • THEN the first LoadCodeAsync SHALL fail with a clear error message that names the missing runtime and the discovery paths (DOTNET_ROOT, system install, side-by-side host/fxr/<version>/)

Scenario: Roll-forward to a newer runtime

  • WHEN the WorkManager is built for .NET 10 and the target has a .NET 11+ runtime installed
  • THEN the engine SHALL use the newer runtime; the generated runtimeconfig.json SHALL declare "rollForward": "Major"

Requirement: Native DLL Engine

The WorkManager SHALL support a NativeDllEngine for executing workers shipped as a per-architecture native shared library (.so / .dll / .dylib). The engine SHALL run in-process inside the (NativeAOT-compiled) WorkManager and SHALL NOT require a separate worker-host subprocess or a JIT build stage. The engine SHALL be registered for MIME type application/x-native-dll. Per-architecture entries SHALL live under runtimes/<rid>/native/ inside the worker zip, matching the NuGet convention for native packages.

Scenario: Native worker execution

  • WHEN a worker with MIME type application/x-native-dll processes a CloudEvent
  • THEN the engine SHALL NativeLibrary.Load the matching runtimes/<rid>/native/<file> entry from the worker zip, resolve the entry_point and free_result exports declared in manifest.json, marshal the input CloudEvent to a FlatBuffer, invoke the native Process function, and deserialize the returned FlatBuffer into a WorkerResponse.

Scenario: Per-architecture entry selection

  • WHEN a worker zip contains entries under runtimes/<rid>/native/ for at least one supported RID (linux-x64, linux-arm64)
  • THEN the engine SHALL select the entry whose <rid> directory matches RuntimeInformation.RuntimeIdentifier of the running process and SHALL load the corresponding lib<library>.so, <library>.dll, or lib<library>.dylib per the platform (see Native Worker Library Naming).

Scenario: Unsupported architecture

  • WHEN the worker zip does not contain a runtimes/<rid>/native/ entry for the running architecture
  • THEN the engine SHALL throw NotSupportedException listing the supported archs (linux-x64, linux-arm64).

Scenario: Missing manifest

  • WHEN the worker zip does not contain manifest.json at the root
  • THEN the engine SHALL refuse to load the worker and throw a clear error naming the missing file.

Scenario: ABI version mismatch

  • WHEN the manifest.json declares an abi_version that is not in the engine's supported set
  • THEN the engine SHALL refuse to load the worker and throw a clear error naming the supported versions.

Requirement: Native Worker ABI

The NativeDllEngine SHALL define and enforce a stable C ABI between the managed engine and the native worker. The ABI SHALL consist of: (a) a FlatBuffers schema for the CloudEvent wire format; (b) two C function exports Process and FreeResult; (c) a VirtufinHost struct carrying host callbacks; (d) a manifest.json declaring the ABI version, the library basename, and the export names. The engine SHALL vendor a virtufin_worker_api.h header for distribution to worker authors.

Scenario: ABI struct layout

  • WHEN a native worker is loaded
  • THEN the engine SHALL pass a VirtufinHost* whose first field is an opaque engine_ptr (a GCHandle to the engine instance), whose second field is abi_version (currently 1), and whose remaining fields are C function pointers log, gateway_call, and free_response.

Scenario: CloudEvent FlatBuffer input

  • WHEN the engine invokes Process(host, in_buf, in_len, &out_buf, &out_len)
  • THEN in_buf SHALL point to a FlatBuffer-encoded CloudEvent matching the vendored worker_api.fbs schema. The extensions[correlationid] field SHALL carry the correlation id from the input CloudEvent when present.

Scenario: WorkerResponse FlatBuffer output

  • WHEN Process returns 0
  • THEN *out_buf SHALL point to a FlatBuffer-encoded WorkerResponse in which exactly one of result_event or error_message is set. If result_event is set, the engine SHALL publish it on the reply topic derived from the input CloudEvent. If error_message is set, the engine SHALL surface a worker.error lifecycle event and publish an error response if the worker did not specify one.

Scenario: FreeResult pairing

  • WHEN Process returns a non-null *out_buf
  • THEN the engine SHALL call FreeResult(out_buf) exactly once, including on the error path. The worker SHALL use the same allocator for the *out_buf allocation that FreeResult releases.

Scenario: Logging callback

  • WHEN the native worker calls host->log(level, message)
  • THEN the engine SHALL map level (0=trace, 1=debug, 2=info, 3=warn, 4=error) to the corresponding Microsoft.Extensions.Logging.LogLevel and forward message to the engine's structured logger. The lifetime of message SHALL be limited to the call.

Scenario: Gateway call indirection

  • WHEN the native worker calls host->gateway_call(service, method, request, request_len, &response, &response_len)
  • THEN the engine SHALL deserialize request as JSON into a Dictionary<string, object?>, invoke ApiClient.InvokeAsync(service, method, dict), JSON-serialize the returned dictionary into *response (allocated with Marshal.AllocCoTaskMem), and return 0 on success or non-zero on internal failure. The worker SHALL free *response via host->free_response(response).

Scenario: Manifest format

  • WHEN the engine parses the worker's manifest.json
  • THEN the JSON object SHALL contain abi_version (integer) and library (non-empty string matching [A-Za-z0-9_.-]+), and MAY contain entry_point (string, default "Process") and free_result (string, default "FreeResult"). Additional fields SHALL be ignored by v1. A missing or empty library field SHALL cause the engine to refuse the worker with a clear error.

Requirement: Native Worker Identity and Lifecycle

The NativeDllEngine SHALL communicate worker identity to the native worker via process-level environment variables set at LoadCodeAsync time, and SHALL release the loaded library and all engine-owned resources on worker stop.

Scenario: Worker identity env vars

  • WHEN the engine loads a native worker
  • THEN it SHALL set VIRTUFIN_WORKER_ID to the worker GUID and VIRTUFIN_WORKER_TOPIC to the worker's input topic. The worker SHALL read these once at initialization.

Scenario: Correlation id from input

  • WHEN the engine marshals the input CloudEvent for a Process call
  • THEN the FlatBuffer's extensions[correlationid] attribute SHALL carry the input CloudEvent's correlationid extension attribute if present. Workers SHALL read it from the FlatBuffer, not from env vars.

Scenario: Concurrent Process calls

  • WHEN multiple Process invocations are made against the same loaded library
  • THEN the engine SHALL allow them to proceed concurrently. The native worker SHALL be reentrant; any per-worker mutable state SHALL be protected by the worker's own synchronization primitives.

Scenario: Reload under load

  • WHEN LoadCodeAsync is called while a ProcessAsync is in flight
  • THEN the engine SHALL block the reload until the in-flight call completes. The new code SHALL take effect for subsequent calls only.

Scenario: Cleanup on stop

  • WHEN the engine is disposed or the worker is stopped
  • THEN the engine SHALL call NativeLibrary.Free on the loaded library, free the GCHandle for the engine pointer, delete the temp file holding the DLL, and restore the prior values of VIRTUFIN_WORKER_ID and VIRTUFIN_WORKER_TOPIC if they were set.

Requirement: Native Worker Caveats

The NativeDllEngine SHALL document its isolation and timeout characteristics in the engine README and the worker author guide.

Scenario: In-process crash isolation caveat

  • WHEN a native worker raises an unrecoverable error (segmentation fault, illegal instruction, stack overflow from native code)
  • THEN the WorkManager process SHALL terminate because the error cannot be caught in a way that leaves the .NET runtime in a defined state. Workers SHALL be authored under the assumption that a crash terminates the entire WorkManager. This caveat is unique to the NativeDllEngine; the Python and DotNetDll engines isolate their workers in subprocesses.

Scenario: Unmanaged timeout caveat

  • WHEN the MessageHandlingTimeoutSeconds elapses while a native Process call is in flight
  • THEN the CancellationToken on the managed side SHALL fire and the engine SHALL surface a worker.error lifecycle event, but the native call SHALL continue to run until it returns or the process crashes. Workers SHALL be authored to honor the timeout cooperatively (e.g. via periodic checks of a flag set from a watchdog thread, if at all) or SHALL be guaranteed to return within the timeout.

Scenario: ABI version policy

  • WHEN the WorkManager ships a new major ABI version
  • THEN the engine SHALL continue to load workers declaring the previous ABI version for at least one release cycle, and SHALL refuse workers declaring ABIs older than that.

Requirement: Native Worker Library Naming

The NativeDllEngine SHALL derive the per-platform shared library filename from the manifest's library field by appending an OS-specific prefix and suffix, matching the convention used by NuGet-distributed native packages (runtimes/<rid>/native/). The engine SHALL pick the filename based on RuntimeInformation.IsOSPlatform.

Scenario: Linux filename

  • WHEN the running platform reports OSPlatform.Linux
  • THEN the engine SHALL load lib<library>.so from runtimes/<rid>/native/.

Scenario: Windows filename

  • WHEN the running platform reports OSPlatform.Windows
  • THEN the engine SHALL load <library>.dll from runtimes/<rid>/native/.

Scenario: macOS filename

  • WHEN the running platform reports OSPlatform.OSX
  • THEN the engine SHALL load lib<library>.dylib from runtimes/<rid>/native/.

Scenario: Library name validation

  • WHEN the manifest's library field is empty or contains characters outside [A-Za-z0-9_.-]
  • THEN the engine SHALL refuse the worker and throw a clear error naming the offending value.

Requirement: Worker Tags (Resource Metadata)

The WorkManager SHALL let clients attach arbitrary key/value tags to a worker via CreateWorkerRequest.tags, SetTag, and SetTags, retrievable via GetTag/GetTags/ListWorkers (WorkerInfo.tags). Tags SHALL be resource metadata only — used for filtering (ListWorkersRequest.filters: repeated TagFilter) and identification — and SHALL NOT be injected into the worker's process environment or otherwise exposed to the worker's code.

Scenario: Tags filter ListWorkers

  • WHEN a client calls ListWorkers with one or more TagFilter{key, value} entries
  • THEN only workers whose tags match every filter SHALL be returned

Scenario: Tags are not process-visible

  • WHEN a worker is loaded by any engine
  • THEN its tags SHALL NOT be set as environment variables and SHALL NOT be passed to the worker's ProcessAsync — a worker has no supported way to read its own tags

Scenario: SetTag/SetTags update stored metadata only

  • WHEN a client calls SetTag or SetTags on a running worker
  • THEN the WorkManager SHALL update the persisted tag metadata immediately; no engine reload is required since tags are never applied to the process

Requirement: Worker Creation-Time Config

The WorkManager SHALL let clients set worker-instance defaults via CreateWorkerRequest.config (map<string, string>), retrievable via WorkerInfo.config. Unlike the removed worker environment-variable feature (which mutated process-wide Environment state and let co-loaded workers silently stomp on each other's values), config SHALL be delivered per-call, per-worker-instance: Worker.ProcessAsync SHALL stamp every config entry onto the incoming CloudEvent as an extension attribute before dispatching to the engine, using the same generic extension mechanism already used for correlationid. Config SHALL be immutable after creation — there SHALL be no SetConfig/GetConfig RPCs.

Scenario: Config delivered as CloudEvent extension attributes

  • WHEN a worker created with config: {"url": "https://example.com"} receives a triggering CloudEvent
  • THEN the CloudEvent SHALL carry a url extension attribute set to "https://example.com" by the time it reaches the worker's engine

Scenario: Config does not leak between co-loaded workers

  • WHEN two workers, each created with a different config value for the same key, are loaded in the same WorkManager process
  • THEN each worker's incoming CloudEvents SHALL only ever carry that worker's own config value — never the other worker's

Scenario: Config is immutable after creation

  • WHEN a client wants to change a worker's config
  • THEN the WorkManager SHALL NOT provide an RPC to do so; the worker MUST be deleted and recreated with the new config

Scenario: Config survives WorkManager restart

  • WHEN a WorkManager instance restarts and recovers a worker via RecoverWorkers
  • THEN the recovered worker SHALL retain its original config

Requirement: Worker Manager Horizontal Scalability

The WorkManager SHALL scale horizontally by spreading workers across processes, not by sharing one worker's load: every worker subscribed to a topic receives every message on it, wherever it runs (Cross-Instance Coordination), and a worker processes its topic's messages in order, one at a time. Throughput for a high-volume stream SHALL therefore come from splitting it across topics (e.g. one topic per currency pair) and workers, not from adding replicas of one worker.

Scenario: Multiple WM replicas on the same topic

  • WHEN two WorkManager replicas (or Runtime pods) each run a worker for topic orders
  • THEN each message published to orders SHALL be delivered to both replicas, and both workers SHALL process it
  • AND this SHALL hold for distinct workers and for two deployments of the same package alike: deploying a second copy of a worker duplicates its processing (and its outputs), it does not share load

Scenario: Several workers on one topic

  • WHEN several workers on the same WM replica (or Runtime pod) subscribe to the same topic, e.g. several indicator workers on one candle topic
  • THEN every message on the topic SHALL be delivered to each of them, and each SHALL process it with its own config and state
  • AND one worker failing on a message SHALL NOT prevent its delivery to the others
  • AND each worker SHALL receive the topic's messages in publish order

Scenario: Hot-swap within a topic

  • WHEN a worker on orders is replaced by a new worker on orders
  • THEN the replacement SHALL be stop-then-start: stopping the old worker unsubscribes it, and from then on only the new worker receives messages on orders
  • AND while both are subscribed, both SHALL receive every message (there is no "latest worker wins" dispatch)

Scenario: Scaling throughput

  • WHEN operators want to scale throughput for a high-volume stream
  • THEN they SHALL NOT deploy more copies of the same worker on the same topic expecting load sharing -- every copy processes every message
  • AND they SHALL instead split the stream across topics (e.g. per-pair topics) with a worker per topic, which the WorkManager spreads across processes, or speed up the worker itself
  • AND the Pub/Sub Topics spec "Subscription Implementation Pattern" requirement is the authoritative reference for delivery

Scenario: Observability for competing-consumer behavior

  • WHEN an operator sees a worker not receiving messages on a topic
  • THEN they SHALL check that the worker is RUNNING and subscribed to that exact topic, and that its process's subscription is open (messages published before a subscription opens are not delivered: Pub/Sub Topics spec, "New subscription starts at the stream tail")
  • AND there is no competition between replicas to rule out: every subscribed process receives every message
  • AND the workmanager.lifecycle topic SHALL include worker_id, replica_id, and topic fields so operators can correlate

Requirement: Per-worker routing via heartbeat-based forwarding

Each worker SHALL be owned by the instance whose IInstanceIdProvider.GetInstanceId() matches the worker's WorkerConfig.InstanceId field in the Dapr state store (key: workers/{workerId}). Worker lifecycle RPCs (StartWorker, StopWorker, DeleteWorker, LoadCode, SetTag) SHALL be routed to the owning instance via direct gRPC forwarding when they land on a different replica. Peer endpoints SHALL be resolved from the heartbeat-driven PeerRegistry.

Scenario: StopWorker reaches the owning replica via forward

  • WHEN a StopWorker(id: 21) RPC lands on replica A
  • AND worker 21 was created on replica B
  • THEN replica A's handler SHALL catch WorkerNotFoundException
  • AND SHALL read WorkerConfig.InstanceId from the state store
  • AND SHALL resolve replica B's gRPC endpoint from the PeerRegistry
  • AND SHALL forward the StopWorker call via PeerChannelCache<WorkManagerClient>
  • AND replica B SHALL stop the worker and update its state
  • AND on forward failure (Unavailable), replica A SHALL reclaim the worker (re-stamp InstanceId, re-create locally) and retry

Requirement: Worker Source Authentication

URL-based code sources SHALL support an optional per-request credential, carried on the same request as the code source (CreateWorkerRequest/LoadCodeRequest), never persisted beyond the fetch it authorizes. A credential SHALL be one of: a directly-supplied Basic or Bearer secret, or a named reference resolved server-side. SSRF and host-allowlist validation SHALL apply regardless of whether a credential is present or which kind it is.

Scenario: Directly-supplied credential

  • WHEN a code source URL is fetched with a Basic or Bearer credential attached to the request
  • THEN the fetch SHALL use that credential's Authorization header, and no other

Scenario: Named credential resolution

  • WHEN a code source URL is fetched with a named credential reference
  • THEN the service SHALL resolve the name to a server-held credential and validate the target host is permitted for that named credential before injecting it

Scenario: Unresolvable named credential

  • WHEN a request references a named credential the service does not recognize, or whose target host is not permitted for that name
  • THEN the fetch SHALL fail with a clear authorization error rather than proceeding unauthenticated

Scenario: No credential supplied

  • WHEN a code source URL is fetched with no credential on the request
  • THEN the fetch SHALL proceed without an Authorization header

Scenario: Credentials are not persisted

  • WHEN a code source fetch completes, successfully or not
  • THEN the supplied credential SHALL NOT be logged or retained beyond that fetch

Requirement: Multi-Topic Worker Subscription

A worker MAY subscribe to more than one topic. CreateWorkerRequest SHALL accept a list of topics (minimum one); the WorkManager SHALL open a subscription for each distinct topic in the list and dispatch any triggering CloudEvent on any of them to the worker.

Scenario: Worker created with multiple topics

  • WHEN a client creates a worker with more than one topic in CreateWorkerRequest
  • THEN the WorkManager SHALL subscribe to every listed topic
  • AND a CloudEvent arriving on any of the worker's topics SHALL be dispatched to that worker's ProcessAsync

Scenario: Worker created with one topic

  • WHEN a client creates a worker with exactly one topic
  • THEN behavior SHALL be identical to the single-topic case that predates this requirement — this is a superset capability, not a new required pattern

Scenario: Shared topic across workers unaffected

  • WHEN multiple workers (with possibly-overlapping topic lists) are registered
  • THEN the existing per-topic dispatch rule (all local workers registered for a topic receive a matching message; see the Worker Lifecycle Events and Cross-Instance Coordination requirements) SHALL apply independently per topic, unchanged by a worker having additional topics

Scenario: Worker deletion unsubscribes all its topics

  • WHEN a worker with multiple topics is deleted
  • THEN the WorkManager SHALL unsubscribe/tear down the worker's registration on every one of its topics
  • AND a topic SHALL only have its Dapr subscription fully torn down once no worker (in this or any other topic-sharing registration) still references it