You can implement FIFO (First-In-First-Out) job queues by using one orchestration instance per logical queue. Because an orchestration executes its logic sequentially, it naturally serializes jobs.
Key Implementation Details:
- Serialization: Only one job per queue is active at a time. Different queues run in parallel.
- Buffering: Incoming jobs are buffered using the External Events pattern.
- Persistence: Use
ContinueAsNew to carry unprocessed jobs (a backlog) forward to the next generation. This ensures that when history is reset, no jobs are lost. - Concurrency Control: To prevent history from growing indefinitely, limit the number of items processed in a single generation (e.g., using a
MaxUpdatesPerGeneration constant) and pass the remaining items in the Backlog via ContinueAsNew.
// One instance per resource ID applies each update to a primary store then a replica,
// strictly in FIFO order. Expected event: "Enqueue" (ResourceUpdate).
// The 4-type-parameter base declares ResourceUpdate as the event payload type, so the
// framework deserializes incoming events and passes OnEvent a typed value.
public class ResourceUpdateQueueOrchestration : TaskOrchestration<object, QueueState, ResourceUpdate, string>
{
const int MaxUpdatesPerGeneration = 100; // Max updates processed per generation
// Rebuilt deterministically on replay, since OnEvent replays in original enqueue order.
readonly Queue<ResourceUpdate> inbox = new Queue<ResourceUpdate>();
TaskCompletionSource<bool> newWork; // Wakes RunTask when an update arrives while parked.
public override async Task<object> RunTask(OrchestrationContext context, QueueState state)
{
// Re-seed with updates carried over from the previous generation (oldest first, so
// they stay ahead of any newly arriving events and FIFO order is preserved).
foreach (ResourceUpdate update in state?.Backlog ?? Enumerable.Empty<ResourceUpdate>())
{
this.inbox.Enqueue(update);
}
// Block until the first update arrives.
if (this.inbox.Count == 0)
{
this.newWork = new TaskCompletionSource<bool>();
await this.newWork.Task;
this.newWork = null;
}
// Drain in FIFO order. Limit the number processed per generation to bound history size; if there are more, we'll pick up the rest in the next generation.
for (int processed = 0; this.inbox.Count > 0 && processed < MaxUpdatesPerGeneration; processed++)
{
ResourceUpdate next = this.inbox.Dequeue();
await context.ScheduleTask<bool>(typeof(UpdatePrimaryStoreActivity), next);
await context.ScheduleTask<bool>(typeof(UpdateReplicaStoreActivity), next);
}
// Reset history, carrying any updates that arrived while we were busy.
context.ContinueAsNew(new QueueState { Backlog = this.inbox.ToArray() });
return null;
}
public override void OnEvent(OrchestrationContext context, string name, ResourceUpdate input)
{
if (name == "Enqueue")
{
this.inbox.Enqueue(input);
this.newWork?.TrySetResult(true);
}
}
}
public class QueueState
{
// Updates carried over from the previous generation, oldest first.
public ResourceUpdate[] Backlog { get; set; } = Array.Empty<ResourceUpdate>();
}
