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. Multi-instance coordination uses Dapr competing consumers (shared app-id); see the horizontal scalability requirement 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 the worker SHALL be persisted to the state store, 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

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

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. The output CloudEvent type field SHALL determine the publish topic.

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 matching their type field

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: Cross-Instance Coordination

For topics with multiple WorkManager instances, Dapr competing consumers SHALL be used as the coordination primitive. When multiple replicas share a Dapr app-id, Dapr's consumer group semantics deliver each published message to exactly one instance — the instances compete, they do not load-share. Messages delivered to an instance that has no matching worker registered for the topic 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 model replaces the prior distributed-lock approach (removed in LIBRARY_VERSION 0.2.5). The lock-based group coordination (GroupLockOptions, LockContentionException, AcquireGroupLockAsync, ReleaseGroupLockAsync, VIRTUFIN_WORKER_GROUP) is no longer supported. Operators MUST NOT deploy multiple WM replicas expecting throughput scaling on a single topic — the competing-consumer model ensures exactly-once delivery at the process level, not thread-pool parallelism.

Scenario: Message lands on a replica with no matching worker

  • WHEN Dapr's competing-consumer delivery routes a message to a WorkManager instance 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 automatic re-routing to a replica that does have a matching worker SHALL occur

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 SHALL run with restricted interpreter flags and a package allowlist. All non-allowed imports SHALL be blocked.

Scenario: Allowed import

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

Scenario: Blocked import

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

Requirement: Worker Contract (IWorker)

The IWorker interface SHALL define an asynchronous ProcessAsync method returning Task<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: 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 across different topics, not across the same topic. Multiple WorkManager replicas deployed against the same Dapr pubsub component SHALL compete for messages on each topic, not share load. This is a deliberate design choice that prioritizes hot-swap of workers within a topic over throughput scaling for that topic. Operators MUST NOT deploy multiple WM replicas expecting throughput gain on a single topic; they SHALL deploy multiple workers on different topics on the same WM replica instead.

Scenario: Multiple WM replicas on the same topic

  • WHEN two WorkManager replicas both run a worker for topic orders
  • THEN each message published to orders is delivered to exactly one of the two replicas (competing consumers at the Dapr layer)
  • AND the worker that wins the message processes it; the worker on the other replica receives nothing for that message
  • AND there is NO automatic re-routing — a message that lands on a replica with no matching local worker is dropped at the WM layer, acked with no redelivery (an explicit business-logic skip, not a processing failure)

Scenario: Hot-swap within a topic

  • WHEN a worker on orders stops and a new worker on orders starts on the same WM replica
  • THEN the new worker inherits the subscription via the LatestWorkerId dispatch mechanism
  • AND the previous worker stops receiving messages on orders

Scenario: Scaling throughput

  • WHEN operators want to scale throughput for a high-volume topic
  • THEN they SHALL NOT deploy multiple WM replicas expecting load sharing — that will cause competing and dropped messages
  • AND they SHALL instead run multiple workers on different topics on the same WM replica (the typical Virtufin design), or scale within a topic by optimizing the worker code
  • AND the Pub/Sub Topics spec §Horizontal Scalability section is the authoritative reference

Scenario: Observability for competing-consumer behavior

  • WHEN an operator sees a worker not receiving messages on a topic
  • THEN they SHALL check whether another WM replica is winning the Dapr competition for that topic
  • 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