Run .NET workers

Workers execute your code. The Pipelogiq API authenticates applications and bootstraps worker sessions; the scheduler decides when stages should run. Both .NET transports use the same handler contracts and stage executor.

Choose a transport

HTTP fits networks that allow only outbound HTTPS or applications that cannot carry a broker client. AMQP fits deployments that can connect directly to the broker. Neither choice changes handler code or makes delivery exactly once. The HTTP package removes the broker dependency from the worker process; it does not remove the server's broker.

For an HTTP program you can run immediately, use the quickstart.

Start an AMQP worker

Use the package-feed configuration from the quickstart and reference PipelogiqSDK version 0.5.0 plus Microsoft.Extensions.Hosting version 8.0.1. This complete Program.cs starts a worker and remains running until Ctrl+C:

using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Hosting;
using PipelogiqSDK.Abstractions;
using PipelogiqSDK.Api;
using PipelogiqSDK.Configuration;
using PipelogiqSDK.Runner;
using PipelogiqSDK.StageHelper;

var key = Environment.GetEnvironmentVariable("PIPELOGIQ_API_KEY");
if (string.IsNullOrWhiteSpace(key))
    throw new InvalidOperationException("Set PIPELOGIQ_API_KEY.");

var builder = Host.CreateApplicationBuilder(args);
builder.Services.AddPipelogiq(new PipelogiqRunnerOptions
{
    ApiKey = key,
    ApiUrl = Environment.GetEnvironmentVariable("PIPELOGIQ_API_URL")
        ?? "http://localhost:8081",
    WorkerName = "amqp-worker",
    DrainGracePeriod = TimeSpan.FromSeconds(30)
});
builder.Services.AddTransient<HelloHandler>();
using var host = builder.Build();
var lifetime = host.Services.GetRequiredService<IHostApplicationLifetime>();
var runner = host.Services.GetRequiredService<PipelineRunner>();
runner.RegisterHandler("HelloHandler", typeof(HelloHandler));
await host.StartAsync();
try
{
    await runner.StartAsync(lifetime.ApplicationStopping);
}
catch (OperationCanceledException) when (lifetime.ApplicationStopping.IsCancellationRequested)
{
}
finally
{
    using var shutdown = new CancellationTokenSource(TimeSpan.FromSeconds(40));
    await host.StopAsync(shutdown.Token);
}

internal sealed record HelloInput(string Name);
internal sealed class HelloHandler : IStageHandler<HelloInput>
{
    public Task<IStageResult> ExecuteAsync(HelloInput input, IStageContext? context = null)
        => Task.FromResult<IStageResult>(StageResult.Success($"Hello {input.Name}"));
}

The bootstrap response supplies the application ID, session token, queue names, and broker connection. The worker must be able to reach the advertised broker host. PIPELOGIQ_RABBITMQ_URL provides an explicit local connection-string override when a local network layout needs one. Keep it in deployment configuration, not source.

Handler names and dependency injection

Registration is explicit and must finish before StartAsync:

builder.Services.AddScoped<OrderRepository>();
builder.Services.AddTransient<ProcessOrderHandler>();
// After building the host:
runner.RegisterHandler("ProcessOrderHandler", typeof(ProcessOrderHandler));

This is a wiring pattern for your own repository and handler classes. Type registration creates and disposes a DI scope per stage attempt. A scoped database context can therefore be injected into the handler normally. The handler must also be registered with DI; naming its type in the runner does not register its dependencies.

The instance overload, runner.RegisterHandler("Name", instance), reuses that object. Its lifetime and thread safety belong to your application. Use it for intentionally reusable instances; prefer type registration for handlers with scoped dependencies. Rejecting a late or duplicate registration is preferable to changing routing while a worker is running.

StartAsync runs the worker loop until cancellation. It is not a fire-and-forget readiness notification. Keep and await its task, observe faults, and stop it before disposing its service provider. Starting the Generic Host and starting the runner are separate operations; both matter when your application has other hosted services, such as Telegram.

What the runtime owns

An attempt proceeds through session bootstrap, delivery, lease acquisition, handler execution, result reporting, and acknowledgement. Leases are renewed during execution and result reporting. A stale or superseded execution must not overwrite newer state.

HTTP reports the result to the API and acknowledges according to the server response. AMQP publishes a result with broker confirmation and handles mandatory returns. Acceptance of the HTTP request or broker publication means the result was queued; it does not synchronously prove the scheduler has applied it. Read pipeline status for the resulting state.

The runtime renews sessions through heartbeats, reports lifecycle state, and recovers transport sessions. AMQP also restores cancelled consumers and handles missing/recreated queues. Queue provisioning defaults to AssertOnly; ordinarily the control plane owns topology. Review configuration before changing provisioning behavior.

Cancellation, timeouts, and shutdown

var token = context.GetCancellationToken();
var response = await httpClient.GetAsync(requestUri, token);

This pattern belongs inside an async handler with a configured HttpClient and request URI. Passing the token lets timeout, cancellation, or lease loss stop work cooperatively. An uncooperative task can keep running; ignoring its late result does not undo its external effects.

On normal shutdown, intake stops first and active work receives DrainGracePeriod to finish. The runtime then cancels remaining work and returns unfinished deliveries as appropriate. Configure your container/service termination grace period to allow for draining and final network cleanup. Abrupt process termination cannot perform a graceful drain.

Producers without workers

A web API that only creates workflows needs PipelogiqSDK.Core. It can use PipelineBuilder.Create(name, explicitOptions) directly or register AddPipelogiqCore(options) for the API client and shared services. It needs no running PipelineRunner and no RabbitMQ client.

Always pass explicit PipelogiqRunnerOptions to builders, especially when one process serves multiple applications. Ambient builder configuration and AddPipelogiqToken are obsolete. Treat API keys as application credentials; end-user identity and business permissions remain the consumer's responsibility.

Logs and traces

Use context.LogInfo, LogWarning, and LogError for stage logs. PipelineLogger.Logs is a snapshot; changing the returned list does not append logs.

Builders propagate W3C traceparent and tracestate from the current activity, respecting supplied values. The executor creates stage activities and exposes execution ID, attempt, idempotency key, timeout, cancellation, and trace IDs through IStageExecutionContext and extension methods. Register PipelogiqSDK as an activity source in your OpenTelemetry application. Exporter choice, endpoint configuration, and collector operation belong to that application.

The SDK can preserve trace context without a listener, but this does not export a trace by itself. Similarly, registering an agent metrics observer does not create a Prometheus HTTP endpoint.