Background jobs & workflows, code-first

What you get

Quickstart

Install the core plus the adapters you need. While Klassd.Workflows is in beta the packages are prerelease:

dotnet add package Klassd.Workflows.Core --prerelease
dotnet add package Klassd.Workflows.Storage.Postgres --prerelease   # durable store (or .Storage.MongoDb)
dotnet add package Klassd.Workflows.Kubernetes --prerelease         # K8s executor (omit for local only)
dotnet add package Klassd.Workflows.Artifacts.S3 --prerelease       # artifact store (or .Artifacts.Gcs)

1. Define a job

Any class implementing IJob is a unit of work. Use the IJobContext to log, report progress, and read arguments:

public sealed class MyJob : IJob
{
    public async Task RunAsync(IJobContext ctx)
    {
        ctx.Log("starting");
        ctx.ReportProgress(50, "halfway");
        await Task.Delay(1000, ctx.CancellationToken);
        ctx.Log("done");
    }
}

2. Wire it up in Program.cs

AddKlassdWorkflowsCore() returns a builder you use to pick a durable store; pick an executor separately:

var workflows = builder.Services.AddKlassdWorkflowsCore();
workflows.UsePostgres("Host=…;Database=…;Username=…;Password=…");  // or .UseMongo(...) / in-memory

builder.Services.AddKubernetesExecutor(builder.Configuration);     // or AddLocalExecutor(workerDll)

3. Run

Enqueue jobs from code, or open the dashboard to start/stop them and watch live console output.

Scheduling

Fire a job now, or register a recurring job with a cron expression (parsed by Cronos):

scheduler.AddOrUpdateRecurring<MyJob>("nightly", "0 2 * * *");   // cron
await scheduler.EnqueueAsync<MyJob>();                            // fire now

Recurring workflows are registered the same way with AddOrUpdateRecurringWorkflow(id, name, cron). Pod resources (CPU/memory requests & limits) are set per job with a [JobResources] attribute and can be retuned from config without a recompile.

Workflows (DAGs)

Jobs compose into a directed acyclic graph that fans out, waits on dependencies and passes data between nodes. The orchestrator runs in the scheduler; each node runs as a normal worker pod, so every node has its own live console.

registry.Register(new WorkflowBuilder("catalog-integration")
    .Add<MarketFinderJob>("markets")                       // root: emits "market_ids"
    .Add<DataProxyJob>("data-proxy")                       // parallel root: writes an artifact
    .Add<IntegrationJob>("integration", n => n
        .DependsOn("markets", "data-proxy")
        .FanOutOver("markets", "market_ids", itemArgument: "market"))   // one pod per market
    .Add<PublishJob>("publish", n => n.DependsOn("integration").WithRetries(2))
    .Add<FinalizerJob>("finalizer", n => n
        .DependsOn("publish", "data-proxy")
        .BindInput("dataset_ref", "data-proxy", "dataset_ref"))          // reads the artifact
    .Build());

• Dependencies — a node starts once all its dependencies are satisfied; a failed dependency skips dependents.
• Fan-out — read an upstream output as a JSON array and start one execution per element.
• Conditions — run a node only when a predicate over upstream outputs holds (otherwise it's benignly omitted).
• Retries — re-run a failed execution up to n times, per fan-out item.
• Artifacts — large payloads pass through an IArtifactStore; a node saves an artifact and publishes the small reference downstream.

Executors: local & Kubernetes

The same worker runs locally and in the cluster — only the executor that launches it differs. Communication is a line protocol on stdout, so Kubernetes pod logs are the transport for free.

• Local — AddLocalExecutor(...) launches the worker as a child process per job. No cluster required; ideal for dev.
• Kubernetes — AddKubernetesExecutor(...) creates a batch/v1 Job (one pod, restartPolicy: Never) per execution, tails its logs, and cleans up via ttlSecondsAfterFinished. Stopping a job deletes the Job; SIGTERM cancels the worker's token.

Storage & artifacts

Two pluggable seams, both with built-in adapters and open for your own:

• Job store (IJobStore) — holds executions, recurring entries and workflow runs. In-memory by default; UsePostgres(...) or UseMongo(...) for durability.
• Artifact store (IArtifactStore) — holds large payloads between nodes. The worker selects a provider by name at runtime (file, s3, gcs), so the choice is per-deployment, not compiled in.

Add your own by implementing the interface (plus an IArtifactStoreProvider for artifacts) — exactly how the Postgres/Mongo and S3/GCS packages do it.

Packages

Install the core plus the adapters you need — each keeps its SDK isolated, so you only pull in what you wire up. While in beta, add --prerelease.

View on GitHubAll NuGet packages