Jattac.Libs.Tempo 1.2.0

Jattac.Libs.Tempo

A high-performance, multi-tenant, multi-priority throttled queue for .NET 8. Tempo is designed for production systems that need to protect downstream resources from traffic floods while ensuring fairness and reliability across tenants.

---

Table of Contents

• Overview
• Features
• Installation
• Quick Start
• Core Concepts
• Configuration
• Enqueue Modes
• Lifecycle Callbacks
• Deduplication
• Circuit Breaker
• Elastic Overflow
• Observability
  • Queue Snapshot — in-flight and upcoming items (v1.1.0)
  • Aggregate Snapshot
• Scheduler
• ASP.NET Core Integration
• Standalone Usage (No DI)
• License

---

Overview

Tempo sits between your application and any rate-sensitive downstream resource — an external API, a database write path, a message broker, or a third-party service. Instead of calling downstream directly, you enqueue work items. Tempo dispatches them at a controlled rate using a token-bucket algorithm, maintains per-tenant fairness via round-robin scheduling, and exposes hooks for every significant event in the lifecycle.

Typical use cases:

• Rate-limiting outbound HTTP calls to a third-party API that enforces a quota
• Fair allocation of processing capacity across multiple tenants in a SaaS application
• Protecting a write-heavy database table from burst traffic
• Scheduled background job execution with overlap and missed-run control

---

Features

• Token-bucket rate limiting — Smooth, configurable global throughput with burst support. The bucket starts full and refills continuously.
• Multi-tenant fair scheduling — Round-robin dispatch across tenants prevents any single tenant from monopolising the queue.
• Multi-priority lanes — Configurable number of priority levels. Lane 0 is highest priority. The dispatcher drains higher-priority lanes before lower ones within each tenant cycle.
• Deduplication — Per-tenant, zero-configuration deduplication. Return a key from your work item and duplicates are silently dropped while the original is pending or in-flight.
• Circuit breaker — Automated per-tenant isolation that trips on persistent consecutive failures and flushes the tenant backlog to the overflow path.
• Elastic overflow — Async hooks to redirect excess or failed traffic to external storage for deferred replay, preventing item loss under quota pressure.
• Bounded channels — Global channel capacity and per-tenant depth limits protect process memory from unbounded growth.
• Processor timeout — Optional per-item timeout that treats slow processors as failures, preventing one slow tenant from blocking all others.
• Callback isolation — All user-supplied callbacks are wrapped with an optional timeout and run in a fire-and-forget task so that a hanging callback cannot stall the dispatch loop.
• Observability — Full metrics via System.Diagnostics.Metrics, health checks via Microsoft.Extensions.Diagnostics.HealthChecks, periodic depth and state snapshots, and a pluggable sink interface.
• Scheduler — A cron/interval/one-shot scheduler built on top of the queue, with configurable missed-run and overlap policies.
• ASP.NET Core integration — First-class DI registration, IHostedService lifecycle, and health check integration via extension methods.

---

Installation

dotnet add package Jattac.Libs.Tempo

Target framework: .NET 8.0

---

Quick Start

1. Define a Work Item and Processor

// The work item carries the data your processor needs.
public record NotificationWork(Guid TenantId, string RecipientEmail, int Priority = 0)
    : ITempoWork;

// The processor performs the actual operation for each item.
public class NotificationProcessor : ITempoProcessor<NotificationWork>
{
    public async Task<bool> ProcessAsync(NotificationWork work, CancellationToken ct)
    {
        await emailClient.SendAsync(work.RecipientEmail, ct);
        return true; // Return false to signal a recoverable failure without throwing.
    }
}

2. Register in the DI Container

builder.Services
    .AddTempoQueue<NotificationWork, NotificationProcessor>(options =>
    {
        options.MessagesPerSecond = 10;
        options.BurstCapacity = 20;
        options.MaxPendingItemsPerTenant = 200;
    })
    .OnProcessed((id, work) =>
    {
        logger.LogInformation("Sent {Email}, correlation {Id}", work.RecipientEmail, id);
        return Task.CompletedTask;
    })
    .OnFailed((id, work, error) =>
    {
        logger.LogError("Failed {Email}: {Error}", work.RecipientEmail, error);
        return Task.CompletedTask;
    });

3. Enqueue Work

public class NotificationService(TempoQueue<NotificationWork> queue)
{
    // Await-result mode: suspends until the item is processed and returns success/failure.
    public async Task<bool> SendAsync(NotificationWork item)
    {
        var (success, correlationId) = await queue.EnqueueAsync(item);
        return success;
    }

    // Fire-and-forget mode: returns as soon as the item is accepted by the queue.
    public async Task SendFireAndForgetAsync(NotificationWork item)
    {
        await queue.EnqueueFireAndForgetAsync(item);
    }
}

---

Core Concepts

ITempoWork

Every work item must implement ITempoWork:

public interface ITempoWork
{
    Guid TenantId { get; }
    int Priority { get; }
    string? DeduplicationKey { get; } // Return null to disable deduplication.
}

TenantId determines which tenant's round-robin slot is used. Priority selects the lane (0 is highest). DeduplicationKey opts the item into per-tenant deduplication.

ITempoProcessor<TWork>

public interface ITempoProcessor<TWork> where TWork : ITempoWork
{
    Task<bool> ProcessAsync(TWork work, CancellationToken cancellationToken);
}

Return true for success, false for a recoverable failure. Throw for an unexpected error. Both false and exceptions count as failures for circuit-breaker and health-check purposes.

---

Configuration

TempoQueueSettings can be bound from IConfiguration or set inline via an Action<TempoQueueSettings> delegate.

Throughput

Property	Type	Default	Description
MessagesPerSecond	int	5	Sustained global dispatch rate. Must be greater than zero.
BurstCapacity	int	10	Token-bucket ceiling. Allows short bursts above the sustained rate. Must be greater than zero.

Queue Depth

Property	Type	Default	Description
ChannelCapacity	int	1000	Global limit on buffered items. Back-pressures writers when full.
MaxPendingItemsPerTenant	int	100	Per-tenant depth limit across all priority lanes. Triggers OnQuotaExceeded when reached.
ResumeThresholdFactor	double	0.7	Fraction of MaxPendingItemsPerTenant at which OnTenantBacklogDrained fires. Must be in (0.0, 1.0).
PriorityLaneCount	int	2	Number of priority lanes. Lane 0 is highest priority.

Health Thresholds

Property	Type	Default	Description
UnhealthyConsecutiveFailureLimit	int	5	Consecutive failures for a single tenant before health is Unhealthy and the circuit breaker callback is invoked.
DegradedBacklogGrowingFor	TimeSpan	60s	Duration the backlog must be growing before health transitions to Degraded.
DegradedFailureRatePerSecond	double	1.0	Rolling 60-second failure rate above which health is Degraded.
DegradedNoSuccessFor	TimeSpan	5m	Window with no successful dispatch after which health is Degraded.

Timeouts and Isolation

Property	Type	Default	Description
ProcessorTimeout	TimeSpan?	null	If set, wraps each ProcessAsync call in a timeout. Items that exceed the limit are treated as failures.
CallbackTimeout	TimeSpan?	null	If set, bounds all user-supplied callbacks. A callback that exceeds the limit is abandoned and logged; the library continues normally. The abandoned task continues running in the background until it completes.

Reporting Intervals

Property	Type	Default	Description
DepthReportInterval	TimeSpan	30s	How often OnDepthReport fires.
ObservabilityReportInterval	TimeSpan	30s	How often OnObservabilityReport fires and registered sinks are written.

Binding from IConfiguration

builder.Services.AddTempoQueue<MyWork, MyProcessor>(
    builder.Configuration.GetSection("Tempo"));

{
  "Tempo": {
    "MessagesPerSecond": 20,
    "BurstCapacity": 40,
    "MaxPendingItemsPerTenant": 500,
    "ProcessorTimeout": "00:00:30",
    "CallbackTimeout": "00:00:05"
  }
}

---

Enqueue Modes

Await-Result Mode

The caller suspends until the item has been processed (or failed). Returns a (bool Success, Guid CorrelationId) tuple.

var (success, correlationId) = await queue.EnqueueAsync(work);

An optional correlation ID can be supplied to correlate the enqueue with an upstream trace:

var id = Guid.NewGuid();
var (success, _) = await queue.EnqueueAsync(work, correlationId: id);

Fire-and-Forget Mode

The caller returns as soon as the item is accepted by the queue. The correlation ID is returned for tracing.

Guid correlationId = await queue.EnqueueFireAndForgetAsync(work);

---

Lifecycle Callbacks

All callbacks are registered on the builder returned by AddTempoQueue. Each callback has an overload that receives the DI IServiceProvider, allowing resolution of scoped services without capturing them in a closure.

OnProcessed

Fires after the processor returns true.

.OnProcessed(async (correlationId, work) =>
{
    await audit.RecordSuccessAsync(correlationId);
})

// DI-scoped overload:
.OnProcessed(async (correlationId, work, sp) =>
{
    await sp.GetRequiredService<IAuditService>().RecordSuccessAsync(correlationId);
})

OnFailed

Fires when the processor returns false or throws.

.OnFailed(async (correlationId, work, errorMessage) =>
{
    await deadLetter.WriteAsync(work, errorMessage);
})

OnQuotaExceeded

Fires when MaxPendingItemsPerTenant is reached for the work item's tenant. Return true to signal that the item was handled externally (no exception is thrown to the caller). Return false to throw TempoQuotaExceededException.

.OnQuotaExceeded(async (tenantId, work) =>
{
    await overflowStore.SaveAsync(work);
    return true;
})

OnTenantFailureCircuitBroken

Fires when a tenant's consecutive failure count reaches UnhealthyConsecutiveFailureLimit. Return true to trip the circuit — new items for this tenant are redirected to OnQuotaExceeded and the existing backlog is flushed there. Return false to leave the circuit closed.

.OnTenantFailureCircuitBroken(async (tenantId, lastError) =>
{
    await alerts.NotifyAsync($"Circuit open for tenant {tenantId}: {lastError}");
    return true;
})

To close the circuit after remediation:

queue.ResetTenantCircuit(tenantId);

OnTenantBacklogDrained

Fires when a tenant's pending count drops below MaxPendingItemsPerTenant * ResumeThresholdFactor. Use this to trigger replay from an overflow store.

.OnTenantBacklogDrained(async (tenantId) =>
{
    await overflowStore.ReplayAsync(tenantId);
})

OnDepthReport

Fires periodically with a lightweight TempoDepthSnapshot containing pending counts per tenant and per lane.

.OnDepthReport(async (snapshot) =>
{
    metrics.RecordDepth(snapshot.TotalPending);
})

OnObservabilityReport

Fires periodically with a full TempoObservabilitySnapshot. See the Observability section for the complete property reference.

.OnObservabilityReport(async (snapshot) =>
{
    logger.LogInformation("Throughput: {Rate}/s, P95 wait: {P95}",
        snapshot.ThroughputPerSecond, snapshot.P95WaitTime);
})

---

Deduplication

Per-tenant deduplication drops duplicate enqueues while the original is pending or in-flight. No configuration is required — implement DeduplicationKey on your work type:

public record OrderWork(Guid TenantId, string OrderId, int Priority = 0) : ITempoWork
{
    // Items with the same TenantId + OrderId combination are deduplicated.
    // A duplicate enqueue returns (false, id) rather than throwing.
    public string? DeduplicationKey => OrderId;
}

The key is scoped per tenant, so tenant-A/order-42 and tenant-B/order-42 are treated as distinct items. Once processing completes (success or failure), the key is released and the same key can be re-enqueued.

Return null from DeduplicationKey to disable deduplication for that work type. No changes are required to existing code that does not use the interface default.

---

Circuit Breaker

The circuit breaker automatically isolates a tenant that is consistently failing, preventing a broken downstream from consuming all dispatcher capacity.

builder.Services
    .AddTempoQueue<MyWork, MyProcessor>(opts =>
    {
        opts.UnhealthyConsecutiveFailureLimit = 5;
    })
    .OnTenantFailureCircuitBroken(async (tenantId, lastError) =>
    {
        await notificationService.AlertOnCallAsync(tenantId, lastError);
        return true;
    })
    .OnQuotaExceeded(async (tenantId, work) =>
    {
        // Also receives the flushed backlog items when the circuit trips.
        await overflowStore.SaveAsync(work);
        return true;
    });

When the circuit trips:

1. The tenant's existing backlog is flushed through OnQuotaExceeded.
2. All subsequent enqueues for that tenant are redirected to OnQuotaExceeded without entering the queue.
3. The circuit remains open until queue.ResetTenantCircuit(tenantId) is called.

---

Elastic Overflow

Elastic overflow handles quota pressure without dropping items. Redirect excess traffic to an external store and replay it when the tenant's backlog drains.

builder.Services
    .AddTempoQueue<MyWork, MyProcessor>(opts =>
    {
        opts.MaxPendingItemsPerTenant = 100;
        opts.ResumeThresholdFactor = 0.6; // Replay signal fires at 60 items (60%).
    })
    .OnQuotaExceeded(async (tenantId, work) =>
    {
        await db.SaveToOverflowTableAsync(work);
        return true;
    })
    .OnTenantBacklogDrained(async (tenantId) =>
    {
        await db.ReplayOverflowForTenantAsync(tenantId);
    });

---

Observability

Queue Snapshot — in-flight and upcoming items (v1.1.0)

Two new pull-only read paths let you answer "what is happening right now?" without any callbacks or polling intervals inside the library itself.

queue.Inflight — the item currently being processed

TempoInflightSnapshot<TWork>? snapshot = queue.Inflight;

if (snapshot is null)
{
    Console.WriteLine("Queue is idle.");
}
else
{
    Console.WriteLine($"Processing tenant {snapshot.TenantId}");
    Console.WriteLine($"  Priority       : {snapshot.Priority}");
    Console.WriteLine($"  Correlation ID : {snapshot.CorrelationId}");
    Console.WriteLine($"  Enqueued at    : {snapshot.EnqueuedAt:u}");
    Console.WriteLine($"  Started at     : {snapshot.StartedAt:u}");
    Console.WriteLine($"  Wait time      : {snapshot.WaitTime.TotalMilliseconds:F0} ms");
    Console.WriteLine($"  Elapsed        : {snapshot.ElapsedProcessingTime.TotalMilliseconds:F0} ms");
    // ElapsedProcessingTime is recomputed on every access — never store the value.
}

Inflight is O(1) and lock-free. Safe to read from any thread at any time, including inside an HTTP request handler that runs concurrently with the dispatch loop.

The snapshot is null in three situations: before the first item is enqueued, when the queue is idle between dispatches, and after the queue is stopped.

queue.GetQueueSnapshot(maxDepthPerTenant) — items still waiting

// Top 10 items per (tenant, priority-lane) pair (default).
IReadOnlyList<TempoQueueItemSnapshot<TWork>> pending = queue.GetQueueSnapshot(maxDepthPerTenant: 10);

// Sort globally by enqueue time for a "what comes next overall" view:
foreach (var item in pending.OrderBy(x => x.EnqueuedAt))
{
    Console.WriteLine(
        $"[lane {item.Priority}] tenant={item.TenantId}  " +
        $"pos={item.PositionWithinTenant}  " +
        $"waited={item.WaitTime.TotalMilliseconds:F0} ms  " +
        $"corr={item.CorrelationId}");
}

The in-flight item is not included — it was dequeued before ProcessAsync was called. Use queue.Inflight alongside GetQueueSnapshot to get the complete real-time picture.

maxDepthPerTenant is clamped to [1, 100] regardless of what you pass.

Cross-tenant ordering is not guaranteed. Items are ordered within each tenant (FIFO), but tenant ordering across the list is non-deterministic. Sort by EnqueuedAt after receiving the list if you need a global order.

Building a live-feed endpoint

[ApiController]
[Route("api/v1/queue")]
public class QueueInspectorController(TempoQueue<NotificationWork> queue) : ControllerBase
{
    [HttpGet("live")]
    public IActionResult GetLive([FromQuery] int depth = 10)
    {
        var inflight = queue.Inflight;
        var upcoming = queue.GetQueueSnapshot(depth);

        return Ok(new
        {
            inflight = inflight is null ? null : new
            {
                tenantId      = inflight.TenantId,
                correlationId = inflight.CorrelationId,
                enqueuedAt    = inflight.EnqueuedAt,
                startedAt     = inflight.StartedAt,   // use this on the frontend for the live timer
                waitMs        = (long) inflight.WaitTime.TotalMilliseconds,
                elapsedMs     = (long) inflight.ElapsedProcessingTime.TotalMilliseconds,
            },
            upcoming = upcoming.Select(x => new
            {
                tenantId             = x.TenantId,
                priority             = x.Priority,
                positionWithinTenant = x.PositionWithinTenant,
                correlationId        = x.CorrelationId,
                enqueuedAt           = x.EnqueuedAt,
                waitMs               = (long) x.WaitTime.TotalMilliseconds,
            }),
            snapshotTakenAtUtc = DateTimeOffset.UtcNow,
        });
    }
}

Example response when one message is being sent and two more are waiting:

{
  "inflight": {
    "tenantId": "3fa85f64-5717-4562-b3fc-2c963f66afa6",
    "correlationId": "d2c0e31e-8a14-4711-ba82-3e1e6e9e8f1c",
    "enqueuedAt": "2026-06-24T08:01:10.000Z",
    "startedAt":  "2026-06-24T08:01:10.312Z",
    "waitMs": 312,
    "elapsedMs": 189
  },
  "upcoming": [
    {
      "tenantId": "3fa85f64-5717-4562-b3fc-2c963f66afa6",
      "priority": 0,
      "positionWithinTenant": 1,
      "correlationId": "a1b2c3d4-0000-0000-0000-000000000001",
      "enqueuedAt": "2026-06-24T08:01:10.800Z",
      "waitMs": 812
    },
    {
      "tenantId": "9b9d1bca-93a1-4d5e-b9e3-1dc12e6dfa99",
      "priority": 1,
      "positionWithinTenant": 1,
      "correlationId": "a1b2c3d4-0000-0000-0000-000000000002",
      "enqueuedAt": "2026-06-24T08:01:11.100Z",
      "waitMs": 512
    }
  ],
  "snapshotTakenAtUtc": "2026-06-24T08:01:11.612Z"
}

Frontend tip: elapsedMs is stale the moment it is serialised. On the frontend, use startedAt to drive a local setInterval timer instead:

const [elapsedMs, setElapsedMs] = useState(() =>
  Math.max(0, Date.now() - new Date(item.startedAt).getTime())
);
useEffect(() => {
  const id = setInterval(() =>
    setElapsedMs(Math.max(0, Date.now() - new Date(item.startedAt).getTime())), 1_000);
  return () => clearInterval(id);
}, [item.startedAt]);

Thread safety guarantee

Reader	Writer	Safe?
HTTP thread reading queue.Inflight	Dispatch loop updating the in-flight field	Yes — Volatile.Read / Volatile.Write; reference assignment is atomic on .NET
HTTP thread calling GetQueueSnapshot	Dispatch loop dequeuing items concurrently	Yes — ConcurrentQueue<T> enumeration is non-destructive under concurrent mutation
Multiple concurrent HTTP threads reading Inflight	—	Yes — reads are independent and lock-free

---

Aggregate Snapshot

Call GetSnapshot() at any time to obtain a point-in-time TempoObservabilitySnapshot:

var snapshot = queue.Observability.GetSnapshot();

Console.WriteLine($"Total pending:      {snapshot.TotalPending}");
Console.WriteLine($"Throughput:         {snapshot.ThroughputPerSecond:F1}/s");
Console.WriteLine($"Enqueue rate:       {snapshot.EnqueueRatePerSecond:F1}/s");
Console.WriteLine($"P50 wait:           {snapshot.MedianWaitTime.TotalMilliseconds:F0} ms");
Console.WriteLine($"P95 wait:           {snapshot.P95WaitTime.TotalMilliseconds:F0} ms");
Console.WriteLine($"P99 wait:           {snapshot.P99WaitTime.TotalMilliseconds:F0} ms");
Console.WriteLine($"P50 processing:     {snapshot.MedianProcessingTime.TotalMilliseconds:F0} ms");
Console.WriteLine($"P95 processing:     {snapshot.P95ProcessingTime.TotalMilliseconds:F0} ms");
Console.WriteLine($"Token utilisation:  {snapshot.TokenBucketUtilization:P0}");
Console.WriteLine($"Estimated drain:    {snapshot.EstimatedGlobalDrainTime}");
Console.WriteLine($"Failure rate:       {snapshot.FailureRatePerSecond:F2}/s");
Console.WriteLine($"Circuit-broken:     {string.Join(", ", snapshot.CircuitBrokenTenants)}");
Console.WriteLine($"Health:             {snapshot.HealthStatus}");

Full Snapshot Property Reference

Property	Description
TotalPending	Items waiting across all tenants and lanes.
PendingByTenant	Per-tenant pending counts.
PendingByPriority	Per-lane pending counts.
ThroughputPerSecond	Successful dispatches per second (rolling 60 s).
EnqueueRatePerSecond	Enqueues per second (rolling 60 s).
TotalProcessedSinceStart	Cumulative successful dispatches since queue start.
TotalEnqueuedSinceStart	Cumulative enqueues since queue start.
TokenBucketLevel	Current token count. 0 = throttling active, capacity = full.
TokenBucketCapacity	Configured burst ceiling.
TokenBucketUtilization	0.0 = idle (full bucket), 1.0 = saturated (empty).
EstimatedGlobalDrainTime	Time to drain the current backlog at the configured rate.
TotalFailuresSinceStart	Cumulative failures since queue start.
FailureRatePerSecond	Failures per second (rolling 60 s).
LastFailureAt	Timestamp of the most recent failure; null if none.
LastFailureError	Error message from the most recent failure; null if none.
ConsecutiveFailuresByTenant	Per-tenant consecutive failure counts.
CircuitBrokenTenants	Set of tenant IDs currently circuit-broken.
MedianWaitTime	P50 enqueue-to-dispatch latency (includes token-bucket wait).
P95WaitTime	P95 enqueue-to-dispatch latency.
P99WaitTime	P99 enqueue-to-dispatch latency.
MedianProcessingTime	P50 time spent inside ProcessAsync.
P95ProcessingTime	P95 time spent inside ProcessAsync.
IsDispatcherRunning	true while the background dispatch loop is active.
IsBacklogGrowing	true when enqueue rate exceeds dispatch rate.
LastSuccessAt	Timestamp of the most recent successful dispatch; null if none.
HealthStatus	Healthy, Degraded, or Unhealthy.
SnapshotTakenAtUtc	UTC time at which this snapshot was captured.

Health Status

Value	Meaning
Healthy	Normal operation. Backlog is stable; failure rate is within thresholds.
Degraded	Backlog is growing, failure rate is elevated, or no successful dispatch within DegradedNoSuccessFor.
Unhealthy	Dispatcher is stopped, or a tenant has exceeded UnhealthyConsecutiveFailureLimit.

Health Checks

builder.Services
    .AddHealthChecks()
    .AddTempoCheck<MyWork>(name: "tempo-notifications");

The health check maps TempoHealthStatus to the standard HealthStatus values and includes the full snapshot in the health check data dictionary for inspection via the health endpoint.

System.Diagnostics.Metrics

Tempo publishes metrics under the meter name Jattac.Libs.Tempo. These are consumable by any OpenTelemetry-compatible collector. No additional configuration is required; the meter is registered automatically when the queue starts.

Observability Sinks

Implement ITempoObservabilitySink to receive periodic snapshots and forward them to any backend:

public class DatadogTempoSink : ITempoObservabilitySink
{
    public Task OnSnapshotAsync(TempoObservabilitySnapshot snapshot, CancellationToken ct)
    {
        datadogClient.Gauge("tempo.pending", snapshot.TotalPending);
        datadogClient.Gauge("tempo.throughput", snapshot.ThroughputPerSecond);
        return Task.CompletedTask;
    }
}

// Register an instance directly:
builder.Services
    .AddTempoQueue<MyWork, MyProcessor>(opts => { ... })
    .WithObservabilitySink(new DatadogTempoSink());

// Or resolve from DI:
builder.Services
    .AddSingleton<DatadogTempoSink>()
    .AddTempoQueue<MyWork, MyProcessor>(opts => { ... })
    .WithObservabilitySink<DatadogTempoSink>();

---

Scheduler

TempoScheduler<TJob> is a cron, interval, and one-shot job scheduler built on top of TempoQueue. The scheduler evaluates registered jobs on each tick and enqueues due runs into an internal queue, inheriting all of Tempo's rate-limiting, fairness, and observability guarantees.

Defining a Job

public class ReportJob : ITempoJob
{
    public Guid TenantId { get; init; }

    public async Task<bool> ProcessAsync(TempoScheduledWork<ReportJob> work, CancellationToken ct)
    {
        await reportService.GenerateAsync(work.Job.TenantId, work.ScheduledAt, ct);
        return true;
    }
}

Schedule Types

TempoSchedule is a sealed discriminated union. Pattern-match against the three variants:

Variant	Constructor	Description
OnceAt	new TempoSchedule.OnceAt(DateTimeOffset fireAt)	Fires exactly once at the specified instant.
Every	new TempoSchedule.Every(TimeSpan interval, DateTimeOffset? startAt = null)	Fires repeatedly at a fixed interval. Starts immediately if startAt is null.
Cron	new TempoSchedule.Cron(string expression)	Standard 5-field cron expression (Cronos). Seconds and years fields are not supported.

Missed Run Policy

Controls behaviour when scheduled runs are missed due to process restart or a delayed tick:

Value	Behaviour
MissedRunPolicy.Skip	Missed runs are discarded. LastFiredAt advances to now.
MissedRunPolicy.RunOnce	Exactly one run is enqueued for the earliest missed slot.
MissedRunPolicy.CatchUp	One run is enqueued per missed slot in chronological order, up to MaxCatchUpSlots.

Overlap Policy

Controls behaviour when a new run is due but a previous instance is still in-flight:

Value	Behaviour
OverlapPolicy.Skip	The new run is skipped. Use when overlapping runs would cause double-processing.
OverlapPolicy.Queue	The new run is always enqueued regardless of in-flight state.

Registering the Scheduler

builder.Services.AddTempoScheduler<ReportJob>(builder =>
{
    builder
        .AddJob(
            new ReportJob { TenantId = tenantA },
            new TempoSchedule.Cron("0 9 * * MON-FRI"),
            opts =>
            {
                opts.MissedRunPolicy = MissedRunPolicy.RunOnce;
                opts.OverlapPolicy = OverlapPolicy.Skip;
                opts.MaxDuration = TimeSpan.FromMinutes(30);
            })
        .AddJob(
            new ReportJob { TenantId = tenantB },
            new TempoSchedule.Every(TimeSpan.FromHours(1)))
        .WithSettings(s =>
        {
            s.TickInterval = TimeSpan.FromSeconds(5);
            s.MaxCatchUpSlots = 50;
        })
        .WithQueueSettings(q =>
        {
            q.MessagesPerSecond = 5;
            q.BurstCapacity = 10;
        });
});

Scheduler Settings

Property	Type	Default	Description
TickInterval	TimeSpan	1s	How often the scheduler evaluates registered jobs for due runs.
MaxCatchUpSlots	int	100	Maximum catch-up slots enqueued per tick. Prevents a flood after a long outage.

Runtime Registration and Removal

Jobs can be registered and unregistered at runtime without restarting the scheduler:

public class JobManagementService(TempoScheduler<ReportJob> scheduler)
{
    public Guid AddJob(ReportJob job, string cronExpression)
    {
        return scheduler.Register(
            job,
            new TempoSchedule.Cron(cronExpression),
            missedRunPolicy: MissedRunPolicy.Skip,
            overlapPolicy: OverlapPolicy.Skip);
    }

    public void RemoveJob(Guid scheduleId)
    {
        scheduler.Unregister(scheduleId);
    }
}

Scheduler Snapshot

var snapshot = scheduler.GetSnapshot();

foreach (var reg in snapshot.Registrations)
{
    Console.WriteLine($"{reg.ScheduleId}: next={reg.NextFireAt:u}, last={reg.LastFiredAt:u}");
}

---

ASP.NET Core Integration

Tempo registers itself as an IHostedService. The host manages the full queue lifecycle — no manual StartAsync or StopAsync calls are required.

var builder = WebApplication.CreateBuilder(args);

builder.Services
    .AddTempoQueue<NotificationWork, NotificationProcessor>(opts =>
    {
        opts.MessagesPerSecond = 50;
        opts.BurstCapacity = 100;
        opts.ProcessorTimeout = TimeSpan.FromSeconds(10);
        opts.CallbackTimeout = TimeSpan.FromSeconds(5);
    })
    .OnProcessed(...)
    .OnFailed(...)
    .OnQuotaExceeded(...);

builder.Services
    .AddHealthChecks()
    .AddTempoCheck<NotificationWork>(name: "tempo");

var app = builder.Build();
app.MapHealthChecks("/health");
app.Run();

---

Standalone Usage (No DI)

Tempo can be used directly without a dependency injection container:

var settings = new TempoQueueSettings
{
    MessagesPerSecond = 10,
    BurstCapacity = 20,
};

await using var queue = new TempoQueue<MyWork>(
    processor: new MyProcessor(),
    settings: settings,
    loggerFactory: LoggerFactory.Create(b => b.AddConsole()));

queue.OnProcessed = (id, work) =>
{
    Console.WriteLine($"Processed: {id}");
    return Task.CompletedTask;
};

await queue.StartAsync();

await queue.EnqueueFireAndForgetAsync(new MyWork(TenantId: tenantId, Priority: 0));

// Drain and stop cleanly.
await queue.StopAsync();

---

License

MIT — Copyright (c) 2025 Jattac Computing Company. See LICENSE for the full text.