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?, 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? stateStoreName = null, 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
stateStoreName string
allowedCodeSourceHosts string

Fields

CodeFetcherHttpClientName 

public const string CodeFetcherHttpClientName = "CodeFetcher"

Field Value

string

DeadLetterTopicSuffix

Suffix appended to the topic name for dead-letter messages that have exhausted retries.

public const string DeadLetterTopicSuffix = "-dead"

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 const string ProbeEventType = "com.virtufin.workmanager.topic-probe"

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(CreateWorkerRequest, SourceCredential?, CancellationToken)

Creates a new worker and starts processing.

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

Parameters

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.

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.

GetTag(Guid, string)

Gets the worker's current value for a single tag.

public string? GetTag(Guid id, string key)

Parameters

id Guid
key string

Returns

string

GetTags(Guid)

Returns a snapshot of the worker's full tag map.

public IReadOnlyDictionary<string, string> GetTags(Guid id)

Parameters

id Guid

Returns

IReadOnlyDictionary<string, string>

GetWorkerHistory(Guid)

Gets the code change history for a worker.

public IReadOnlyList<HistoryEntry> GetWorkerHistory(Guid id)

Parameters

id Guid

The worker ID.

Returns

IReadOnlyList<HistoryEntry>

A read-only list of history entries.

Exceptions

WorkerNotFoundException

Thrown if the worker is not found.

ListWorkers()

Lists all workers with their current status.

public IReadOnlyList<WorkerInfo> ListWorkers()

Returns

IReadOnlyList<WorkerInfo>

A read-only list of worker information.

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(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.

SetTag(Guid, string, string)

Sets a single tag on the worker. Tags are resource metadata and are NOT injected into the engine process.

public void SetTag(Guid id, string key, string value)

Parameters

id Guid
key string
value string

SetTags(Guid, IDictionary<string, string>)

Replaces the worker's entire tag map with the supplied dictionary.

public void SetTags(Guid id, IDictionary<string, string> tags)

Parameters

id Guid
tags IDictionary<string, string>

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.