csharp

public interface IQueue<T> : IQueue where T : class
{
    AsyncEvent<EnqueuingEventArgs<T>> Enqueuing { get; }
    AsyncEvent<EnqueuedEventArgs<T>> Enqueued { get; }
    AsyncEvent<DequeuedEventArgs<T>> Dequeued { get; }
    AsyncEvent<LockRenewedEventArgs<T>> LockRenewed { get; }
    AsyncEvent<CompletedEventArgs<T>> Completed { get; }
    AsyncEvent<AbandonedEventArgs<T>> Abandoned { get; }
    AsyncEvent<QueueDeletedEventArgs<T>> QueueDeleted { get; }

    void AttachBehavior(IQueueBehavior<T> behavior);
    Task<string> EnqueueAsync(T data, QueueEntryOptions options = null);
    Task<IQueueEntry<T>> DequeueAsync(CancellationToken cancellationToken);
    Task<IQueueEntry<T>> DequeueAsync(TimeSpan? timeout = null);
    Task RenewLockAsync(IQueueEntry<T> queueEntry);
    Task CompleteAsync(IQueueEntry<T> queueEntry);
    Task AbandonAsync(IQueueEntry<T> queueEntry);
    Task<IEnumerable<T>> GetDeadletterItemsAsync(CancellationToken cancellationToken = default);
    Task StartWorkingAsync(Func<IQueueEntry<T>, CancellationToken, Task> handler,
                           bool autoComplete = false,
                           CancellationToken cancellationToken = default);
}

public interface IQueue : IHaveSerializer, IDisposable
{
    Task<QueueStats> GetQueueStatsAsync();
    Task DeleteQueueAsync();
    string QueueId { get; }
}

csharp

using Foundatio.Queues;

var queue = new InMemoryQueue<WorkItem>();

// Enqueue work
await queue.EnqueueAsync(new WorkItem { Id = 1, Data = "Hello" });

// Dequeue and process
var entry = await queue.DequeueAsync();
Console.WriteLine($"Processing: {entry.Value.Data}");
await entry.CompleteAsync();

csharp

// dotnet add package Foundatio.Redis

using Foundatio.Redis.Queues;

var queue = new RedisQueue<WorkItem>(o => {
    o.ConnectionMultiplexer = redis;
    o.Name = "work-items";
    o.WorkItemTimeout = TimeSpan.FromMinutes(5);
});

csharp

// dotnet add package Foundatio.AzureServiceBus

using Foundatio.AzureServiceBus.Queues;

var queue = new AzureServiceBusQueue<WorkItem>(o => {
    o.ConnectionString = "...";
    o.Name = "work-items";
});

csharp

// dotnet add package Foundatio.AzureStorage

using Foundatio.AzureStorage.Queues;

var queue = new AzureStorageQueue<WorkItem>(o => {
    o.ConnectionString = "...";
    o.Name = "work-items";
});

csharp

// dotnet add package Foundatio.AWS

using Foundatio.AWS.Queues;

var queue = new SQSQueue<WorkItem>(o => {
    o.Region = RegionEndpoint.USEast1;
    o.QueueName = "work-items";
});

txt

                    ┌─────────┐
                    │ Queued  │
                    └────┬────┘
                         │
                         ▼
              ┌──────────────────┐
              │ Dequeued/Working │
              └────┬─────────────┘
                   │
                   ▼
            ┌──────────────┐
            │  Processing  │
            └──┬────────┬──┘
               │        │
        Success│        │Failure
               │        │
               ▼        ▼
          ┌────────┐  ┌───────────┐
          │Complete│  │ Abandoned │
          └────────┘  └─────┬─────┘
                            │
                            ▼
                       ┌─────────┐
                       │ Retry?  │
                       └──┬───┬──┘
                   Yes    │   │    No
                          │   │
                          ▼   ▼
                    ┌─────────┐  ┌──────────────┐
                    │ Queued  │  │ Dead Letter  │
                    └─────────┘  └──────────────┘

csharp

var entry = await queue.DequeueAsync();
try
{
    await ProcessAsync(entry.Value);
    await entry.CompleteAsync();
}
catch
{
    await entry.AbandonAsync();
    throw;
}

csharp

var entry = await queue.DequeueAsync();
if (!CanProcess(entry.Value))
{
    // Return to queue for later processing
    await entry.AbandonAsync();
    return;
}

Recommended Approach

Use QueueJobBase<T> for queue processing (see Jobs - Queue Processor Jobs). For manual processing, call RenewLockAsync() periodically within your processing logic.

csharp

public class VideoProcessorJob : QueueJobBase<VideoWorkItem>
{
    private readonly IVideoService _videoService;

    public VideoProcessorJob(IQueue<VideoWorkItem> queue, IVideoService videoService)
        : base(queue) => _videoService = videoService;

    protected override async Task<JobResult> ProcessQueueEntryAsync(
        QueueEntryContext<VideoWorkItem> context)
    {
        var workItem = context.QueueEntry.Value;
        var startTime = DateTime.UtcNow;

        try
        {
            // Start processing
            await _videoService.StartProcessingAsync(workItem.VideoId);

            // Renew lock if processing is taking longer than expected
            if (DateTime.UtcNow - startTime > TimeSpan.FromMinutes(3))
            {
                await context.QueueEntry.RenewLockAsync();
            }

            await _videoService.CompleteProcessingAsync(workItem.VideoId);
            return JobResult.Success;
        }
        catch (Exception ex)
        {
            return JobResult.FromException(ex);
        }
    }
}

Manual Lock Renewal

Most processing should complete within the WorkItemTimeout. If you regularly need lock renewal, increase the WorkItemTimeout instead. Manual renewal should only be used for truly variable-duration workloads where you cannot predict processing time accurately.

csharp

public class OrderProcessorJob : QueueJobBase<OrderWorkItem>
{
    private readonly ILockProvider _lockProvider;
    private readonly IOrderService _orderService;

    public OrderProcessorJob(
        IQueue<OrderWorkItem> queue,
        ILockProvider lockProvider,
        IOrderService orderService) : base(queue)
    {
        _lockProvider = lockProvider;
        _orderService = orderService;
    }

    // Override to lock on the order ID instead of the queue entry ID
    protected override Task<ILock> GetQueueEntryLockAsync(
        IQueueEntry<OrderWorkItem> queueEntry,
        CancellationToken cancellationToken = default)
    {
        // Lock on the business key (order ID) to prevent concurrent processing
        // of the same order across all queue entries
        string lockKey = $"order:{queueEntry.Value.OrderId}";
        return _lockProvider.AcquireAsync(lockKey, TimeSpan.FromMinutes(5), cancellationToken);
    }

    protected override async Task<JobResult> ProcessQueueEntryAsync(
        QueueEntryContext<OrderWorkItem> context)
    {
        // This will only execute if we successfully acquired the lock
        // Multiple queue entries for the same order will be serialized
        var orderId = context.QueueEntry.Value.OrderId;

        await _orderService.ProcessAsync(orderId, context.CancellationToken);
        return JobResult.Success;
    }
}

Lock Provider Selection

Use a distributed lock provider (e.g., CacheLockProvider with Redis) in production to coordinate across multiple instances. For single-instance scenarios, CacheLockProvider with InMemoryCacheClient is sufficient.

Recommended Approach

For production applications, use QueueJobBase<T> with Foundatio.Extensions.Hosting for reliable, automatic background processing. See Jobs - Queue Processor Jobs for details. The patterns below are for advanced scenarios or custom integrations.

csharp

while (!cancellationToken.IsCancellationRequested)
{
    var entry = await queue.DequeueAsync(cancellationToken);
    if (entry == null)
        continue;

    try
    {
        await ProcessAsync(entry.Value);
        await entry.CompleteAsync();
    }
    catch (Exception ex)
    {
        _logger.LogError(ex, "Failed to process {Id}", entry.Value.Id);
        await entry.AbandonAsync();
    }
}

csharp

// Start processing in background
await queue.StartWorkingAsync(
    async (entry, ct) =>
    {
        await ProcessAsync(entry.Value);
    },
    autoComplete: true,  // Automatically complete on success
    cancellationToken
);

csharp

await queue.EnqueueAsync(new WorkItem { Id = 1 }, new QueueEntryOptions
{
    UniqueId = "unique-id",           // Dedupe by ID
    CorrelationId = "request-123",    // For tracing
    DeliveryDelay = TimeSpan.FromMinutes(5),  // Delayed delivery
    Properties = new Dictionary<string, string>
    {
        ["priority"] = "high"
    }
});

csharp

var queue = new InMemoryQueue<WorkItem>();

queue.Enqueuing.AddHandler((sender, args) =>
{
    _logger.LogInformation("Enqueuing: {Data}", args.Data);
    return Task.CompletedTask;
});

queue.Enqueued.AddHandler((sender, args) =>
{
    _logger.LogInformation("Enqueued: {Id}", args.Entry.Id);
    return Task.CompletedTask;
});

queue.Dequeued.AddHandler((sender, args) =>
{
    _logger.LogInformation("Dequeued: {Id}", args.Entry.Id);
    return Task.CompletedTask;
});

queue.Completed.AddHandler((sender, args) =>
{
    _logger.LogInformation("Completed: {Id}", args.Entry.Id);
    return Task.CompletedTask;
});

queue.Abandoned.AddHandler((sender, args) =>
{
    _logger.LogWarning("Abandoned: {Id}", args.Entry.Id);
    return Task.CompletedTask;
});

queue.QueueDeleted.AddHandler((sender, args) =>
{
    _logger.LogInformation("Queue deleted");
    return Task.CompletedTask;
});

csharp

public class LoggingQueueBehavior<T> : QueueBehaviorBase<T> where T : class
{
    private readonly ILogger _logger;

    public LoggingQueueBehavior(ILogger logger) => _logger = logger;

    protected override Task OnEnqueued(object sender, EnqueuedEventArgs<T> args)
    {
        _logger.LogInformation("Enqueued {Id}", args.Entry.Id);
        return Task.CompletedTask;
    }

    protected override Task OnDequeued(object sender, DequeuedEventArgs<T> args)
    {
        _logger.LogInformation("Dequeued {Id}", args.Entry.Id);
        return Task.CompletedTask;
    }

    protected override Task OnCompleted(object sender, CompletedEventArgs<T> args)
    {
        _logger.LogInformation("Completed {Id} in {Duration}ms",
            args.Entry.Id, args.Entry.ProcessingTime.TotalMilliseconds);
        return Task.CompletedTask;
    }

    protected override Task OnAbandoned(object sender, AbandonedEventArgs<T> args)
    {
        _logger.LogWarning("Abandoned {Id}, attempt {Attempt}",
            args.Entry.Id, args.Entry.Attempts);
        return Task.CompletedTask;
    }

    protected override Task OnQueueDeleted(object sender, QueueDeletedEventArgs<T> args)
    {
        _logger.LogInformation("Queue deleted");
        return Task.CompletedTask;
    }
}

// Attach to queue
queue.AttachBehavior(new LoggingQueueBehavior<WorkItem>(logger));

csharp

// Your message must implement IHaveUniqueIdentifier
public class OrderWorkItem : IHaveUniqueIdentifier
{
    public int OrderId { get; set; }
    public string UniqueIdentifier => $"order:{OrderId}";
}

// Attach the behavior
var cache = new InMemoryCacheClient();
queue.AttachBehavior(new DuplicateDetectionQueueBehavior<OrderWorkItem>(
    cache,
    loggerFactory,
    detectionWindow: TimeSpan.FromMinutes(10)  // How long to remember seen IDs
));

// Duplicates are automatically discarded
await queue.EnqueueAsync(new OrderWorkItem { OrderId = 123 }); // ✅ Enqueued
await queue.EnqueueAsync(new OrderWorkItem { OrderId = 123 }); // ❌ Discarded (duplicate)
await queue.EnqueueAsync(new OrderWorkItem { OrderId = 456 }); // ✅ Enqueued

csharp

// ✅ Correct: separate instances for each queue
queue1.AttachBehavior(new LoggingQueueBehavior<WorkItem>(logger));
queue2.AttachBehavior(new LoggingQueueBehavior<WorkItem>(logger));

// ❌ Throws QueueException: same instance attached twice
var behavior = new LoggingQueueBehavior<WorkItem>(logger);
queue1.AttachBehavior(behavior);
queue2.AttachBehavior(behavior); // throws QueueException

csharp

var queue = new InMemoryQueue<WorkItem>(o => o
    .Behaviors(
        new LoggingQueueBehavior<WorkItem>(logger),
        new DuplicateDetectionQueueBehavior<WorkItem>(cache, loggerFactory),
        new MetricsQueueBehavior<WorkItem>(metrics)
    ));

csharp

var stats = await queue.GetQueueStatsAsync();

Console.WriteLine($"Queued: {stats.Queued}");
Console.WriteLine($"Working: {stats.Working}");
Console.WriteLine($"Dead Letter: {stats.Deadletter}");
Console.WriteLine($"Enqueued: {stats.Enqueued}");
Console.WriteLine($"Dequeued: {stats.Dequeued}");
Console.WriteLine($"Completed: {stats.Completed}");
Console.WriteLine($"Abandoned: {stats.Abandoned}");
Console.WriteLine($"Errors: {stats.Errors}");
Console.WriteLine($"Timeouts: {stats.Timeouts}");

csharp

// Get dead letter items
var deadLetters = await queue.GetDeadletterItemsAsync();

foreach (var item in deadLetters)
{
    _logger.LogWarning("Dead letter: {Id}", item.Id);

    // Optionally re-queue for retry
    await queue.EnqueueAsync(item);
}

csharp

var stats = await queue.GetQueueStatsAsync();
if (stats.Deadletter > 0)
{
    _logger.LogWarning("Dead letter queue has {Count} items", stats.Deadletter);
    // Alert operations team, trigger investigation
}

text

Dequeue → Deserialize fails → Abandon (attempt incremented)
  → Still has retries? → Re-queued with backoff delay
  → Retries exhausted? → Moved to dead letter queue

csharp

var queue = new RedisQueue<WorkItem>(o =>
{
    // For predictable workloads: typical duration + padding
    // Example: If processing takes 2 minutes, set to 4-5 minutes
    o.WorkItemTimeout = TimeSpan.FromMinutes(5);

    // For variable workloads: maximum expected duration + buffer
    // Example: If processing can take up to 10 minutes, set to 15 minutes
    o.WorkItemTimeout = TimeSpan.FromMinutes(15);
});

Timeout Too Short

If WorkItemTimeout is too short, entries will be re-queued before processing completes, leading to duplicate processing attempts and wasted resources.

Monitoring and Adjustment

Monitor your queue processing times and adjust WorkItemTimeout based on actual metrics. Use Application Insights, logging, or custom telemetry to track processing duration over time.

csharp

var queue = new InMemoryQueue<WorkItem>(o =>
{
    o.Retries = 3;                              // Max retry attempts
    o.RetryDelay = TimeSpan.FromMinutes(1);     // Base delay between retries
    o.RetryMultipliers = new[] { 1, 3, 5, 10 }; // Exponential backoff multipliers
});

csharp

// ✅ Good: Small message with reference
public record ProcessImageWorkItem
{
    public required string ImageBlobPath { get; init; }  // Reference to storage
    public required string OutputPath { get; init; }
    public required ImageProcessingOptions Options { get; init; }
}

// ❌ Bad: Large payload in message
public record ProcessImageWorkItem
{
    public required byte[] ImageData { get; init; }  // Could be megabytes!
    public required ImageProcessingOptions Options { get; init; }
}

csharp

// Store large data in blob storage
var blobPath = $"work-items/{Guid.NewGuid()}.json";
await fileStorage.SaveObjectAsync(blobPath, largePayload);

// Enqueue reference only
await queue.EnqueueAsync(new WorkItem
{
    PayloadPath = blobPath,
    PayloadSize = largePayload.Length
});

// In worker: retrieve the payload
var entry = await queue.DequeueAsync();
var payload = await fileStorage.GetObjectAsync<LargePayload>(entry.Value.PayloadPath);
await ProcessAsync(payload);
await entry.CompleteAsync();

// Clean up blob after processing
await fileStorage.DeleteFileAsync(entry.Value.PayloadPath);

csharp

// In-memory (development)
services.AddSingleton<IQueue<WorkItem>>(sp =>
    new InMemoryQueue<WorkItem>());

// Redis (production)
services.AddSingleton<IQueue<WorkItem>>(sp =>
    new RedisQueue<WorkItem>(o => {
        o.ConnectionMultiplexer = sp.GetRequiredService<IConnectionMultiplexer>();
        o.Name = "work-items";
    }));

Automatic Queue Processing

For automatic background processing of queue items, use QueueJobBase<T> with Foundatio.Extensions.Hosting. See Jobs - Queue Processor Jobs for details.

// Register queue and processor job
services.AddSingleton<IQueue<OrderWorkItem>>(sp => new InMemoryQueue<OrderWorkItem>());
services.AddJob<OrderProcessorJob>();  // Automatically processes queue items

Multiple Queues ​

services.AddSingleton<IQueue<OrderWorkItem>>(sp =>
    new InMemoryQueue<OrderWorkItem>(o => o.Name = "orders"));

services.AddSingleton<IQueue<EmailWorkItem>>(sp =>
    new InMemoryQueue<EmailWorkItem>(o => o.Name = "emails"));

Queue Exceptions ​

Queue operations throw QueueException for queue-specific error conditions. This provides a consistent, predictable exception type across all queue implementations (in-memory, Redis, Azure, AWS, etc.).

using Foundatio.Queues;

try
{
    // Attempting to reuse a behavior instance throws QueueException
    var behavior = new LoggingQueueBehavior<WorkItem>(logger);
    queue1.AttachBehavior(behavior);
    queue2.AttachBehavior(behavior); // throws QueueException
}
catch (QueueException ex)
{
    logger.LogError(ex, "Queue operation failed: {Message}", ex.Message);
}

Cancellation Token Behavior ​

Understanding how cancellation tokens are handled internally is important for building reliable queue consumers.

Resource Creation Uses Disposal Token ​

When you call EnqueueAsync, DequeueAsync, or GetDeadletterItemsAsync, the queue may need to create infrastructure (e.g., SQS queues, Azure Service Bus queues, Redis streams). These setup operations use an internal disposal token — not the caller's cancellation token. This means:

• Queue creation only aborts when the queue is disposed, never because a single caller cancelled their operation.
• A cancelled DequeueAsync call (e.g., from a zero timeout) will not prevent queue creation from completing.
• Multiple concurrent callers cannot interfere with each other's setup.

Linked Cancellation for Operations ​

The caller's cancellation token is combined with the disposal token into a linked token for the actual operation (dequeue, deadletter retrieval, etc.). This means:

• Operations cancel when either the caller cancels or the queue is disposed.
• Graceful shutdown via Dispose() cancels all in-flight operations promptly.

// This will never prevent queue creation, even though it times out immediately
var entry = await queue.DequeueAsync(TimeSpan.Zero);

// The cancellation token only affects the dequeue wait, not infrastructure setup
using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(5));
var entry = await queue.DequeueAsync(cts.Token);

If you are writing a custom IQueue<T> implementation by extending QueueBase<T>:

• EnsureQueueCreatedAsync always receives DisposedCancellationToken. Use it for all setup operations (lock acquisition, API calls, etc.).
• DequeueImplAsync receives a linked token (caller + disposal). Respect it for the wait/poll operation.
• EnqueueImplAsync does not receive a cancellation token — keep enqueue fast and non-blocking.

Best Practices ​

1. Proper Resource Disposal ​

Queues implement IDisposable and should be properly disposed:

// ✅ Good: Using statement for short-lived queues
await using var queue = new InMemoryQueue<WorkItem>();
await queue.EnqueueAsync(new WorkItem { Id = 1 });

// ✅ Good: DI container manages lifetime
services.AddSingleton<IQueue<WorkItem>>(sp =>
    new InMemoryQueue<WorkItem>());

// ❌ Bad: Not disposing
var queue = new InMemoryQueue<WorkItem>();
// ... use queue
// Queue is never disposed, resources leak

2. Use Typed Messages ​

// ✅ Good: Typed, versioned messages
public record OrderWorkItem
{
    public int Version { get; init; } = 1;
    public required int OrderId { get; init; }
    public required DateTime CreatedAt { get; init; }
}

// ❌ Bad: Generic, untyped
public class WorkItem
{
    public object Data { get; set; }
}

2. Handle Idempotency ​

var entry = await queue.DequeueAsync();

// Check if already processed
if (await _processedIds.ContainsAsync(entry.Value.Id))
{
    await entry.CompleteAsync();
    return;
}

// Process
await ProcessAsync(entry.Value);

// Mark as processed
await _processedIds.AddAsync(entry.Value.Id);
await entry.CompleteAsync();

3. Set Appropriate Timeouts ​

var queue = new RedisQueue<WorkItem>(o => {
    o.WorkItemTimeout = TimeSpan.FromMinutes(5);  // How long to process
    o.RetryDelay = TimeSpan.FromSeconds(30);      // Delay before retry
    o.Retries = 3;                                 // Max retries
});

4. Monitor Queue Depth ​

var stats = await queue.GetQueueStatsAsync();
if (stats.Queued > 1000)
{
    _logger.LogWarning("Queue depth is high: {Depth}", stats.Queued);
    // Consider scaling workers
}

5. Use Delayed Delivery for Scheduling ​

// Schedule for later
await queue.EnqueueAsync(reminder, new QueueEntryOptions
{
    DeliveryDelay = TimeSpan.FromHours(24)
});

Queue Name vs Queue ID ​

Every queue has two distinct identifiers that serve different purposes:

Name is what controls which data is read and written. All distributed queue implementations (Redis, SQS, Azure Service Bus, Azure Storage) route messages based on Name, not QueueId. Two processes or application restarts using the same Name will naturally share queue data and continue from where the other left off.

QueueId exists purely so that multiple queue instances within the same process (e.g., priority queues or keyed queues) produce distinguishable log output. It has no effect on the backing store.

// Both processes use the same Name → they share the same queue in Redis
var queue = new RedisQueue<WorkItem>(o => {
    o.ConnectionMultiplexer = redis;
    o.Name = "work-items";  // This is the stable backing-store identifier
});

InMemoryQueue

InMemoryQueue is an in-process implementation only. It cannot share data across processes regardless of Name or QueueId. Use a distributed implementation (Redis, SQS, Azure) for cross-process sharing.