Table of Contents

Class WorkManager

Namespace
Virtufin.WorkManager.Components
Assembly
Virtufin.WorkManager.dll

Manages worker lifecycle, subscriptions, and processing.

public sealed class WorkManager : IAsyncDisposable, IWorkerRecoveryExecutor
Inheritance
WorkManager
Implements
Inherited Members

Constructors

WorkManager(PubsubClient, StateClient, IHttpClientFactory, IEngineRegistry, ILogger<WorkManager>, ILoggerFactory, DaprResiliencePipeline, WorkerProcessingRetryPipeline, ResilientDaprPublisher, ApiLifecyclePublisher, IInstanceIdProvider, string?)

Creates a new WorkManager instance.

public WorkManager(Pubsub.PubsubClient pubsubClient, State.StateClient stateClient, IHttpClientFactory httpClientFactory, IEngineRegistry engineRegistry, ILogger<WorkManager> logger, ILoggerFactory loggerFactory, DaprResiliencePipeline resilience, WorkerProcessingRetryPipeline workerRetry, ResilientDaprPublisher publisher, ApiLifecyclePublisher lifecyclePublisher, IInstanceIdProvider instanceIdProvider, string? allowedCodeSourceHosts = null)

Parameters

pubsubClient Pubsub.PubsubClient
stateClient State.StateClient
httpClientFactory IHttpClientFactory
engineRegistry IEngineRegistry
logger ILogger<WorkManager>
loggerFactory ILoggerFactory
resilience DaprResiliencePipeline
workerRetry WorkerProcessingRetryPipeline
publisher ResilientDaprPublisher
lifecyclePublisher ApiLifecyclePublisher
instanceIdProvider IInstanceIdProvider
allowedCodeSourceHosts string

Fields

CodeFetcherHttpClientName 

public const string CodeFetcherHttpClientName = "CodeFetcher"

Field Value

string

DefaultCodeFetchTimeoutSeconds 

public const int DefaultCodeFetchTimeoutSeconds = 30

Field Value

int

ProbeEventType

CloudEvent type marking a topic liveness probe. Probes carry the publishing instance id in the ProbeInstanceExtension CloudEvent extension attribute; the handler echoes probes carrying its own id and drops all probes without dispatching them to workers. The marker must travel as CloudEvent attributes inside the payload — NOT as Dapr publish metadata — because the streamer populates TopicMessage.Type/Extensions from the CloudEvent envelope parsed out of the stored payload, not from the broker-level metadata.

public static readonly string ProbeEventType

Field Value

string

ProbeInstanceExtension 

public const string ProbeInstanceExtension = "wmtopicprobe"

Field Value

string

PublishTopicAttribute

Extension attribute a worker's response CloudEvent can carry to publish to a topic other than its own ce-type. Most workers don't need this -- their response's Type serves as both the wire ce-type and the publish topic, which is why that's still the default. It exists for workers whose ce-type must stay a stable identifier independent of a per-scenario topic (e.g. the pubsub-topics spec's trading events: topic sc.<scenarioId>.trading.order.submitted, ce-type com.virtufin.trading.order.submitted). Stripped from the outgoing envelope before publish -- it's WorkManager-internal routing, not part of the event a subscriber should see.

public const string PublishTopicAttribute = "publishtopic"

Field Value

string

UnknownCloudEventSourceUri 

public static readonly Uri UnknownCloudEventSourceUri

Field Value

Uri

UnknownCloudEventType 

public const string UnknownCloudEventType = "unknown"

Field Value

string

Methods

CreateWorkerAsync(Guid, CreateWorkerRequest, SourceCredential?, CancellationToken)

Creates a new worker and starts processing.

public Task<Guid> CreateWorkerAsync(Guid workerId, CreateWorkerRequest request, SourceCredential? credential = null, CancellationToken cancellationToken = default)

Parameters

workerId Guid

The worker's id, minted by the caller (WorkManagerGrpcService.CreateWorker) before this call -- routed through CreateAsync(CreateWorkerPayload, CancellationToken) so whichever instance Dapr's placement picks for that id's first activation is where the worker actually gets materialized, closing a gap where a plain (non-actor-routed) create could leave an orphaned local copy on a different instance than the one the actor later activates on -- confirmed live to resurrect a genuinely deleted worker in ListWorkersAsync(CancellationToken) forever (virtufin/common#9, Phase 2).

request CreateWorkerRequest

The worker creation request.

credential SourceCredential

Optional, transient per-request fetch credential (see Virtufin.WorkManager.Protos.SourceCredential). Never persisted -- used only for this creation's own fetch. A named reference is instead captured onto NamedCredential by the caller (the gRPC service layer) before request reaches here, so it survives into persisted state for recovery.

cancellationToken CancellationToken

Cancellation token.

Returns

Task<Guid>

The ID of the created worker (the same workerId passed in).

Exceptions

EngineNotFoundException

Thrown if no engine is registered for the requested MIME type.

DeleteWorkerAsync(Guid, CancellationToken)

Deletes a worker and its subscriptions.

public Task DeleteWorkerAsync(Guid id, CancellationToken cancellationToken = default)

Parameters

id Guid

The worker ID.

cancellationToken CancellationToken

Cancellation token.

Returns

Task

Exceptions

WorkerNotFoundException

Thrown if the worker is not found.

DisposeAsync()

Performs application-defined tasks associated with freeing, releasing, or resetting unmanaged resources asynchronously.

public ValueTask DisposeAsync()

Returns

ValueTask

A task that represents the asynchronous dispose operation.

GetRuntimeRoutedWorkersAsync(CancellationToken) 

public Task<List<(Guid WorkerId, string RuntimeActorType)>> GetRuntimeRoutedWorkersAsync(CancellationToken cancellationToken)

Parameters

cancellationToken CancellationToken

Returns

Task<List<(Guid WorkerId, string RuntimeActorType)>>

GetTagAsync(Guid, string, CancellationToken)

Gets the worker's current value for a single tag. Falls back to the persisted config -- everything this needs (Tags) is already there, so unlike Stop/Delete/etc. this never needs peer-forwarding at all.

public Task<string?> GetTagAsync(Guid id, string key, CancellationToken cancellationToken)

Parameters

id Guid
key string
cancellationToken CancellationToken

Returns

Task<string>

GetTagsAsync(Guid, CancellationToken)

Returns a snapshot of the worker's full tag map. See GetTagAsync(Guid, string, CancellationToken).

public Task<IReadOnlyDictionary<string, string>> GetTagsAsync(Guid id, CancellationToken cancellationToken)

Parameters

id Guid
cancellationToken CancellationToken

Returns

Task<IReadOnlyDictionary<string, string>>

GetWorkerHistoryAsync(Guid, CancellationToken)

Gets the code change history for a worker. Falls back to the persisted config -- see GetTagAsync(Guid, string, CancellationToken) for why this never needs peer-forwarding.

public Task<IReadOnlyList<HistoryEntry>> GetWorkerHistoryAsync(Guid id, CancellationToken cancellationToken)

Parameters

id Guid

The worker ID.

cancellationToken CancellationToken

Cancellation token.

Returns

Task<IReadOnlyList<HistoryEntry>>

A read-only list of history entries.

Exceptions

WorkerNotFoundException

Thrown if the worker is not found, locally or persisted.

ListWorkersAsync(CancellationToken) 

public Task<IReadOnlyList<WorkerInfo>> ListWorkersAsync(CancellationToken cancellationToken = default)

Parameters

cancellationToken CancellationToken

Returns

Task<IReadOnlyList<WorkerInfo>>

LoadCode(Guid, CodeSource, SourceCredential?, CancellationToken)

Loads new code into an existing worker. The codeSource carries either an in-memory byte payload or a URL to fetch — callers construct the appropriate subtype via CodeSourceContent(byte[]) or CodeSourceUrl(Uri, string?) on the wire.

public Task LoadCode(Guid id, CodeSource codeSource, SourceCredential? credential = null, CancellationToken cancellationToken = default)

Parameters

id Guid

The worker ID.

codeSource CodeSource

Either content bytes or a URL.

credential SourceCredential

Optional, transient per-request fetch credential -- see CreateWorkerAsync(Guid, CreateWorkerRequest, SourceCredential?, CancellationToken).

cancellationToken CancellationToken

Cancellation token.

Returns

Task

Exceptions

WorkerNotFoundException

Thrown if the worker is not found.

ArgumentException

Thrown on validation failure (empty content, non-HTTP URL).

RecoverWorkersAsync(CancellationToken)

Recovers all workers from persistent state and resumes processing.

public Task RecoverWorkersAsync(CancellationToken cancellationToken = default)

Parameters

cancellationToken CancellationToken

Cancellation token.

Returns

Task

Exceptions

EngineNotFoundException

Thrown if an engine is missing for a recovered worker.

RegisterEngine(ContentType, string, Func<IEngine>)

Registers an engine to handle code execution for a specific content type.

public void RegisterEngine(ContentType contentType, string languageName, Func<IEngine> engineFactory)

Parameters

contentType ContentType

The content type handled by the engine (e.g., "text/x-python").

languageName string

The programming language name (e.g., "Python").

engineFactory Func<IEngine>

Factory function that creates new engine instances.

Exceptions

InvalidOperationException

Thrown if an engine is already registered for the content type.

SetTagAsync(Guid, string, string, CancellationToken)

Sets a single tag on the worker. Tags are resource metadata and are NOT injected into the engine process. Falls back to a persisted read-modify-write when the worker has no local in-memory Worker -- always true for a Runtime-actor-routed worker (see RecoverSingleWorkerAsync(WorkerConfig, bool, CancellationToken)), whose engine and topic subscription deliberately never get materialized here.

public Task SetTagAsync(Guid id, string key, string value, CancellationToken cancellationToken)

Parameters

id Guid
key string
value string
cancellationToken CancellationToken

Returns

Task

SetTagsAsync(Guid, IDictionary<string, string>, CancellationToken)

Replaces the worker's entire tag map with the supplied dictionary. See SetTagAsync(Guid, string, string, CancellationToken) for the persisted-only fallback.

public Task SetTagsAsync(Guid id, IDictionary<string, string> tags, CancellationToken cancellationToken)

Parameters

id Guid
tags IDictionary<string, string>
cancellationToken CancellationToken

Returns

Task

StartWorkerAsync(Guid, CancellationToken)

Starts a stopped worker, resuming topic subscriptions and processing.

public Task StartWorkerAsync(Guid id, CancellationToken cancellationToken = default)

Parameters

id Guid

The worker ID.

cancellationToken CancellationToken

Cancellation token.

Returns

Task

Exceptions

WorkerNotFoundException

Thrown if the worker is not found.

StopWorkerAsync(Guid, CancellationToken)

Stops a running worker, pausing topic subscriptions and processing.

public Task StopWorkerAsync(Guid id, CancellationToken cancellationToken = default)

Parameters

id Guid

The worker ID.

cancellationToken CancellationToken

Cancellation token.

Returns

Task

Exceptions

WorkerNotFoundException

Thrown if the worker is not found.