This post explores the Dapr Jobs building block through a small weather operations app. From the UI, a user can schedule a one-time refresh, a finite recurring refresh, or a cron-based refresh that also evaluates weather alerts. Dapr Scheduler owns the schedule and calls the API when work is due; the API does not need a background loop that keeps checking the clock.
Aspire brings the local pieces together: the Blazor web app, the Minimal API, the API's Dapr sidecar, and RedisInsight. The sample also uses Redis-backed Dapr state, pub/sub, and lock components so scheduled work remains safe across restarts and browser pages update automatically when a refresh finishes.
Here's a glimpse of the demo app, built up incrementally throughout this post.
Prerequisites
- Docker Desktop - Docker Desktop provides the local container runtime used by the backing services and management UI.
- .NET Aspire - Aspire composes, runs, and observes the local distributed application.
- Dapr CLI - the CLI installs and initializes the local Dapr runtime, including Dapr Scheduler and Redis.
Demo App
The demo app is based on the Aspire starter template, which includes a frontend (ASP.NET Core Blazor App), a backend (ASP.NET Core Minimal API), ServiceDefaults, and an AppHost project.
dotnet new aspire-starter --output dapr-jobs-aspireThe solution structure should look similar to the one below.
NOTE: A new project, dapr-jobs-aspire-Contracts, has been added to hold shared contracts consumed by the API and web client.
With the initial structure in place, let's move on to the AppHost.
App Host
Aspire's AppHost is where the distributed application is declared, code-first. In this demo, it starts the API service, its Dapr sidecar, the Blazor frontend, and RedisInsight. It attaches statestore.yaml, pubsub.yaml, and lockstore.yaml to the API sidecar.
Start by adding the CommunityToolkit.Aspire.Hosting.Dapr NuGet package to the AppHost project. It provides extension methods and resource definitions for an Aspire AppHost to configure Dapr resources.
dotnet add package CommunityToolkit.Aspire.Hosting.DaprCreate the folder structure shown below to keep the Dapr component files organized.
components/statestore.yaml
Configures the statestore Redis state store. It durably holds the weather dashboard, active mission, alerts, and execution IDs; ETags protect those records from conflicting writes.
apiVersion: dapr.io/v1alpha1
kind: Component
metadata:
name: statestore
spec:
type: state.redis
version: v1
metadata:
- name: redisHost
value: localhost:6379
- name: redisPassword
value: ""
- name: actorStateStore
value: "true"components/pubsub.yaml
Configures the pubsub Redis pub/sub component. In this single-replica demo, it carries a weather-update message from the scheduled refresh back to the API's SSE stream, which causes connected browsers to update automatically.
apiVersion: dapr.io/v1alpha1
kind: Component
metadata:
name: pubsub
spec:
type: pubsub.redis
version: v1
metadata:
- name: redisHost
value: localhost:6379
- name: redisPassword
value: ""components/lockstore.yaml
Configures the lockstore Redis lock component. It grants a short-lived lease for schedule creation, replacement, and cancellation so only one such operation proceeds at a time.
apiVersion: dapr.io/v1alpha1
kind: Component
metadata:
name: lockstore
spec:
type: lock.redis
version: v1
metadata:
- name: redisHost
value: localhost:6379
- name: redisPassword
value: ""AppHost.cs
using CommunityToolkit.Aspire.Hosting.Dapr;
var builder = DistributedApplication.CreateBuilder(args);
builder.AddDapr();
var daprComponentsPath = Path.Combine(builder.Environment.ContentRootPath, "components");
// RedisInsight UI
builder.AddContainer("redisinsight", "redis/redisinsight")
.WithHttpEndpoint(port: 5540, targetPort: 5540, name: "http")
.WithEnvironment("RI_REDIS_HOST", "host.docker.internal")
.WithEnvironment("RI_REDIS_PORT", "6379")
.WithEnvironment("RI_REDIS_ALIAS", "Dapr state store")
.WithHttpHealthCheck("/api/health/")
.WithUrlForEndpoint("http", url =>
{
url.Url = "/";
url.DisplayText = "Redis Insight";
});;
var apiService = builder.AddProject<Projects.dapr_jobs_aspire_ApiService>("apiservice")
.WithDaprSidecar(sidecar =>
{
sidecar.WithOptions(new DaprSidecarOptions
{
AppId = "weather-operations-api",
ResourcesPaths = [daprComponentsPath]
});
})
.WithHttpHealthCheck("/health")
.WithUrlForEndpoint("http", url =>
{
url.Url = "/";
url.DisplayText = "API Service";
})
.WithUrlForEndpoint("https", url =>
{
url.Url = "/";
url.DisplayText = "API Service";
});;
builder.AddProject<Projects.dapr_jobs_aspire_Web>("webfrontend")
.WithExternalHttpEndpoints()
.WithHttpHealthCheck("/health")
.WithReference(apiService)
.WaitFor(apiService)
.WithUrlForEndpoint("http", url =>
{
url.Url = "/";
url.DisplayText = "Web App";
})
.WithUrlForEndpoint("https", url =>
{
url.Url = "/";
url.DisplayText = "Web App";
});;
builder.Build().Run();The following diagram illustrates the overall application topology.
CAS - Compare-And-Set
NOTE: Redis plays three roles here: state store, pub/sub broker, and lock store. Notice that AppHost isn't responsible for managing it, the demo app instead points to the Redis instance already set up by dapr init, reusing it rather than creating a new one.
API Service
The API is the Dapr-enabled backend powering the weather-operations demo. It exposes minimal API endpoints to view forecasts, schedule one-time, recurring, or alert-driven weather refreshes, check the active Dapr Job, and cancel it when needed. When a scheduled callback fires, it refreshes the in-memory weather state and publishes forecast-updated events through Redis pub/sub and server-sent events, letting the web UI update automatically. A Redis-backed distributed lock guards schedule changes so only one managed weather job runs at a time.
The ApiService project needs three NuGet packages: Dapr.AspNetCore, Dapr.Client, and Dapr.Jobs. Dapr.AspNetCore contains the reference assemblies for developing Dapr services with ASP.NET Core, Dapr.Client contains the reference assemblies for developing Dapr services in general, and Dapr.Jobs provides the SDK for scheduling and handling jobs with Dapr.
dotnet add package Dapr.AspNetCore
dotnet add package Dapr.Client
dotnet add package Dapr.JobsProgram.cs
using System.Text.Json;
using Dapr;
using Dapr.Jobs.Extensions;
using dapr_jobs_aspire.ApiService;
using dapr_jobs_aspire.Contracts;
using Scalar.AspNetCore;
var builder = WebApplication.CreateBuilder(args);
builder.AddServiceDefaults();
builder.Services.AddProblemDetails();
builder.Services.AddSingleton<WeatherUpdateStream>();
builder.Services.AddSingleton<WeatherStation>();
builder.Services.AddSingleton<WeatherRefreshCoordinator>();
builder.Services.AddSingleton<WeatherUpdatePublisher>();
builder.Services.AddDaprClient();
builder.Services.AddDaprJobsClient();
builder.Services.AddOpenApi();
var app = builder.Build();
app.UseExceptionHandler();
if (app.Environment.IsDevelopment())
{
app.MapOpenApi();
app.MapScalarApiReference();
}
app.MapGet("/", () => "Weather Operations API is running. Navigate to /weatherforecast or /weather-operations.");
app.MapGet("/weatherforecast", async Task<WeatherForecast[]> (WeatherStation station, CancellationToken cancellationToken) =>
(await station.GetDashboardAsync(cancellationToken)).Snapshot.Forecasts)
.WithName("GetWeatherForecast");
app.MapDaprScheduledJobHandler(async (string jobName, ReadOnlyMemory<byte> jobPayload, WeatherStation weatherStation, WeatherUpdatePublisher updatePublisher, ILogger? logger, CancellationToken cancellationToken) =>
{
var payload = JsonSerializer.Deserialize<WeatherRefreshPayload>(jobPayload.Span, new JsonSerializerOptions(JsonSerializerDefaults.Web))
?? throw new InvalidOperationException("The Dapr Job payload was empty or invalid.");
var execution = await weatherStation.RefreshFromJobAsync(jobName, payload, cancellationToken);
if (execution.ShouldNotify)
{
await updatePublisher.PublishAsync(execution.Dashboard.Snapshot, cancellationToken);
}
logger?.LogInformation("Dapr triggered weather job {JobName}; forecast version {Version} is {Outcome}.", jobName, execution.Dashboard.Snapshot.Version, execution.WasApplied ? "ready" : "unchanged");
});
app.MapSubscribeHandler();
app.MapPost("/weather-events", [Topic(WeatherUpdatePublisher.PubSubName, WeatherUpdatePublisher.TopicName)] (WeatherUpdate update, WeatherUpdateStream updates) =>
{
updates.Publish(update);
return TypedResults.Ok();
});
app.MapWeatherOperations();
app.MapDefaultEndpoints();
app.Run();
// Allows the API host to be exercised through WebApplicationFactory in the test project.
public partial class Program;WeatherOperationsEndpoints.cs
It defines the /weather-operations minimal API routes for viewing dashboard state, streaming forecast updates, scheduling or cancelling the single managed weather refresh job, and retrieving job details from Dapr Scheduler.
using Dapr.Jobs;
using Dapr.Jobs.Models;
using Microsoft.AspNetCore.Http.HttpResults;
using dapr_jobs_aspire.Contracts;
namespace dapr_jobs_aspire.ApiService;
public static class WeatherOperationsEndpoints
{
public const string WeatherRefreshJobName = "weather-forecast-refresh";
private static readonly DaprJobSchedule ForecastCronSchedule = DaprJobSchedule.FromCronExpression(
new CronExpressionBuilder()
.On(OnCronPeriod.Second, 0)
.Every(EveryCronPeriod.Minute, 1));
public static IEndpointRouteBuilder MapWeatherOperations(this IEndpointRouteBuilder app)
{
var operations = app.MapGroup("/weather-operations")
.WithTags("Weather Operations");
operations.MapGet("/", async Task<Ok<WeatherOperationsDashboard>> (WeatherStation station, CancellationToken cancellationToken) =>
TypedResults.Ok(await station.GetDashboardAsync(cancellationToken)))
.WithName("GetWeatherOperations");
operations.MapGet("/updates", (WeatherUpdateStream updates, CancellationToken cancellationToken) =>
TypedResults.ServerSentEvents(updates.Subscribe(cancellationToken), eventType: "forecast-updated"))
.WithName("StreamWeatherUpdates");
operations.MapPost("/refreshes/once", async Task<Results<Ok<WeatherOperationsDashboard>, ProblemHttpResult>> (
WeatherOneTimeRefreshRequest request,
WeatherRefreshCoordinator coordinator,
ILogger<WeatherStation> logger,
CancellationToken cancellationToken) =>
{
if (request.DelaySeconds is < 5 or > 3600)
{
return TypedResults.Problem("Choose a delay between 5 and 3,600 seconds.", statusCode: StatusCodes.Status400BadRequest);
}
var dueAt = DateTimeOffset.UtcNow.AddSeconds(request.DelaySeconds);
const string displayName = "Forecast refresh";
var payload = WeatherRefreshPayload.Create(WeatherRefreshMode.Once, displayName, dueAt, intervalSeconds: null);
try
{
return TypedResults.Ok(await coordinator.ScheduleAsync(payload, DaprJobSchedule.FromDateTime(dueAt), $"Once, after {request.DelaySeconds} seconds", 1, dueAt, cancellationToken));
}
catch (Exception exception)
{
logger.LogError(exception, "Could not schedule the one-time weather refresh.");
return TypedResults.Problem("The Dapr sidecar or Scheduler is unavailable.", statusCode: StatusCodes.Status503ServiceUnavailable);
}
})
.WithName("ScheduleOneTimeWeatherRefresh");
operations.MapPost("/refreshes/recurring", async Task<Results<Ok<WeatherOperationsDashboard>, ProblemHttpResult>> (
WeatherRepeatingRefreshRequest request,
WeatherRefreshCoordinator coordinator,
ILogger<WeatherStation> logger,
CancellationToken cancellationToken) =>
{
if (request.IntervalSeconds is < 5 or > 3600 || request.Repeats is < 1 or > 20)
{
return TypedResults.Problem("Choose an interval between 5 and 3,600 seconds and 1 to 20 repeats.", statusCode: StatusCodes.Status400BadRequest);
}
const string displayName = "Recurring forecast refresh";
var firstDueAt = DateTimeOffset.UtcNow.AddSeconds(request.IntervalSeconds);
var payload = WeatherRefreshPayload.Create(WeatherRefreshMode.Recurring, displayName, firstDueAt, request.IntervalSeconds);
try
{
return TypedResults.Ok(await coordinator.ScheduleAsync(payload, DaprJobSchedule.FromDuration(TimeSpan.FromSeconds(request.IntervalSeconds)), $"Every {request.IntervalSeconds} seconds, {request.Repeats} times", request.Repeats, firstDueAt, cancellationToken));
}
catch (Exception exception)
{
logger.LogError(exception, "Could not schedule the recurring weather refresh.");
return TypedResults.Problem("The Dapr sidecar or Scheduler is unavailable.", statusCode: StatusCodes.Status503ServiceUnavailable);
}
})
.WithName("ScheduleRecurringWeatherRefresh");
operations.MapPost("/refreshes/alerts", async Task<Results<Ok<WeatherOperationsDashboard>, ProblemHttpResult>> (
WeatherRefreshCoordinator coordinator,
ILogger<WeatherStation> logger,
CancellationToken cancellationToken) =>
{
const string displayName = "Refresh & alerts";
var payload = WeatherRefreshPayload.Create(WeatherRefreshMode.Alerts, displayName, DateTimeOffset.UtcNow, intervalSeconds: null, evaluateWeatherRules: true);
try
{
return TypedResults.Ok(await coordinator.ScheduleAsync(payload, ForecastCronSchedule, "Every minute · weather rules", null, null, cancellationToken));
}
catch (Exception exception)
{
logger.LogError(exception, "Could not schedule refresh and alerts.");
return TypedResults.Problem("The Dapr sidecar or Scheduler is unavailable.", statusCode: StatusCodes.Status503ServiceUnavailable);
}
})
.WithName("ScheduleRefreshAndAlerts");
operations.MapDelete("/refreshes/current", async Task<Results<Ok<WeatherOperationsDashboard>, ProblemHttpResult>> (
WeatherRefreshCoordinator coordinator,
ILogger<WeatherStation> logger,
CancellationToken cancellationToken) =>
{
try
{
return TypedResults.Ok(await coordinator.CancelAsync(cancellationToken));
}
catch (Exception exception)
{
logger.LogError(exception, "Could not cancel the active weather refresh.");
return TypedResults.Problem("The Dapr sidecar or Scheduler is unavailable.", statusCode: StatusCodes.Status503ServiceUnavailable);
}
})
.WithName("CancelWeatherRefresh");
operations.MapGet("/jobs/{jobName}", async Task<IResult> (string jobName, DaprJobsClient jobs, ILogger<WeatherStation> logger, CancellationToken cancellationToken) =>
{
try
{
var job = await jobs.GetJobAsync(jobName, cancellationToken);
return TypedResults.Ok(new WeatherJobDetails(
job.Name ?? jobName,
job.Schedule.ExpressionValue,
job.RepeatCount));
}
catch (Exception exception)
{
logger.LogError(exception, "Could not retrieve weather job {JobName}.", jobName);
return TypedResults.Problem("The requested Dapr job could not be retrieved.", statusCode: StatusCodes.Status503ServiceUnavailable);
}
})
.WithName("GetWeatherJob");
return app;
}
}WeatherRefreshCoordinator.cs
It safely schedules and cancels that job through Dapr Jobs. It uses the Redis-backed lockstore distributed lock to serialize schedule mutations and keeps the local mission state consistent if Scheduler operations fail.
using Dapr.Client;
using Dapr.Jobs;
using Dapr.Jobs.Extensions;
using Dapr.Jobs.Models;
using dapr_jobs_aspire.Contracts;
namespace dapr_jobs_aspire.ApiService;
#pragma warning disable DAPR_DISTRIBUTEDLOCK // Dapr's distributed-lock API is still marked experimental.
public sealed class WeatherRefreshCoordinator(DaprClient dapr, DaprJobsClient jobs, WeatherStation station)
{
private const string LockStoreName = "lockstore";
private const string LockResource = "weather-forecast-refresh";
private const int LockExpirySeconds = 60;
public async Task<WeatherOperationsDashboard> ScheduleAsync(
WeatherRefreshPayload payload,
DaprJobSchedule schedule,
string scheduleDescription,
int? repeats,
DateTimeOffset? nextRunAt,
CancellationToken cancellationToken)
{
await using var lease = await AcquireLeaseAsync(cancellationToken);
try
{
var dashboard = await station.ScheduleMissionAsync(WeatherOperationsEndpoints.WeatherRefreshJobName, payload, scheduleDescription, repeats, nextRunAt, cancellationToken);
try
{
await jobs.ScheduleJobWithPayloadAsync(WeatherOperationsEndpoints.WeatherRefreshJobName, schedule, payload, repeats: repeats, overwrite: true, cancellationToken: cancellationToken);
return dashboard;
}
catch
{
await station.ClearMissionIfCurrentAsync(payload.MissionId, "Scheduling failed", "The Dapr Scheduler did not accept this refresh.", cancellationToken);
throw;
}
}
finally
{
await ReleaseLeaseAsync(lease, cancellationToken);
}
}
public async Task<WeatherOperationsDashboard> CancelAsync(CancellationToken cancellationToken)
{
await using var lease = await AcquireLeaseAsync(cancellationToken);
try
{
var dashboard = await station.GetDashboardAsync(cancellationToken);
var mission = dashboard.Mission;
if (mission is null) return dashboard;
var cancelled = await station.CancelMissionAsync(WeatherOperationsEndpoints.WeatherRefreshJobName, cancellationToken);
try
{
await jobs.DeleteJobAsync(WeatherOperationsEndpoints.WeatherRefreshJobName, cancellationToken);
return cancelled;
}
catch
{
await station.RestoreMissionIfCurrentAsync(mission, cancellationToken);
throw;
}
}
finally
{
await ReleaseLeaseAsync(lease, cancellationToken);
}
}
private async Task<TryLockResponse> AcquireLeaseAsync(CancellationToken cancellationToken)
{
var owner = Guid.NewGuid().ToString("N");
var lease = await dapr.Lock(LockStoreName, LockResource, owner, LockExpirySeconds, cancellationToken);
if (!lease.Success)
{
await lease.DisposeAsync();
throw new InvalidOperationException("Another weather schedule operation is in progress.");
}
return lease;
}
private async Task ReleaseLeaseAsync(TryLockResponse lease, CancellationToken cancellationToken)
{
var response = await dapr.Unlock(LockStoreName, LockResource, lease.LockOwner, cancellationToken);
if (response.status is not LockStatus.Success and not LockStatus.LockDoesNotExist)
{
throw new InvalidOperationException($"Could not release the weather schedule lock: {response.status}.");
}
}
}
#pragma warning restore DAPR_DISTRIBUTEDLOCKWeatherStation.cs
It owns the in-memory weather forecast and mission state. It applies scheduled refreshes, tracks versions and outcomes, evaluates refresh payloads, and produces the dashboard data returned to clients
using Dapr.Client;
using dapr_jobs_aspire.Contracts;
namespace dapr_jobs_aspire.ApiService;
public sealed class WeatherStation(DaprClient dapr)
{
// Registered by the AppHost from its Redis-backed Dapr component manifest.
private const string StateStoreName = "statestore";
private const string DashboardStateKey = "weather-dashboard";
private const int ProcessedExecutionLimit = 32;
private static readonly StateOptions FirstWrite = new() { Concurrency = ConcurrencyMode.FirstWrite };
private static readonly string[] Summaries = ["Clear skies", "Light cloud", "Cool breeze", "Warm front", "Bright afternoon", "Rain watch", "Storm cells", "Windy conditions", "Heat advisory", "Calm evening"];
public async Task<WeatherOperationsDashboard> GetDashboardAsync(CancellationToken cancellationToken) => (await ReadStateAsync(cancellationToken)).ToDashboard();
public async Task<WeatherOperationsDashboard> ScheduleMissionAsync(string jobName, WeatherRefreshPayload payload, string scheduleDescription, int? repeats, DateTimeOffset? nextRunAt, CancellationToken cancellationToken)
{
var state = (await UpdateStateAsync(current => current with
{
Mission = new(jobName, payload.Mode, payload.DisplayName, payload.MissionId, scheduleDescription, repeats, nextRunAt, DateTimeOffset.UtcNow),
Alert = null,
CompletedMissionRuns = 0,
ProcessedExecutionIds = [],
Events = AddEvent(current.Events, "Mission scheduled", $"{payload.DisplayName} is ready for Dapr to trigger.", jobName)
}, cancellationToken)).State;
return state.ToDashboard();
}
public async Task<WeatherOperationsDashboard> CancelMissionAsync(string jobName, CancellationToken cancellationToken)
{
var state = (await UpdateStateAsync(current => current.Mission?.JobName == jobName
? current with { Mission = null, Alert = null, CompletedMissionRuns = 0, ProcessedExecutionIds = [], Events = AddEvent(current.Events, "Mission cancelled", $"{current.Mission.DisplayName} was removed from the Dapr scheduler.", jobName) }
: current, cancellationToken)).State;
return state.ToDashboard();
}
public async Task ClearMissionIfCurrentAsync(string missionId, string title, string detail, CancellationToken cancellationToken)
{
await UpdateStateAsync(current => current.Mission?.MissionId == missionId
? current with { Mission = null, Alert = null, CompletedMissionRuns = 0, ProcessedExecutionIds = [], Events = AddEvent(current.Events, title, detail, WeatherOperationsEndpoints.WeatherRefreshJobName) }
: current, cancellationToken);
}
public async Task RestoreMissionIfCurrentAsync(WeatherMission mission, CancellationToken cancellationToken)
{
await UpdateStateAsync(current => current.Mission is null
? current with { Mission = mission, Events = AddEvent(current.Events, "Cancellation failed", $"{mission.DisplayName} is still scheduled in Dapr.", mission.JobName) }
: current, cancellationToken);
}
public async Task<WeatherJobExecutionResult> RefreshFromJobAsync(string jobName, WeatherRefreshPayload payload, CancellationToken cancellationToken)
{
if (jobName != WeatherOperationsEndpoints.WeatherRefreshJobName) throw new InvalidOperationException($"Unsupported Dapr Job '{jobName}'.");
if (string.IsNullOrWhiteSpace(payload.MissionId)) throw new InvalidOperationException("The Dapr Job payload does not identify its mission.");
var executionId = payload.GetExecutionId(jobName, DateTimeOffset.UtcNow);
var update = await UpdateStateAsync(current =>
{
if (current.Mission?.MissionId != payload.MissionId || current.ProcessedExecutionIds.Contains(executionId, StringComparer.Ordinal)) return current;
var snapshot = CreateSnapshot(payload.DisplayName, current.Snapshot.Version + 1);
var alert = payload.EvaluateWeatherRules ? EvaluateWeatherAlert(snapshot) : null;
var events = AddEvent(current.Events, "Weather refresh completed", $"Forecast version {snapshot.Version} arrived from Dapr.", jobName);
if (alert is not null) events = AddEvent(events, "Weather alert", alert.Detail, jobName);
var completedRuns = current.CompletedMissionRuns;
var mission = current.Mission;
if (mission.Repeats is { } repeats && ++completedRuns >= repeats)
{
events = AddEvent(events, "Mission complete", $"{mission.DisplayName} finished its scheduled refreshes.", jobName);
mission = null;
completedRuns = 0;
}
return current with
{
Snapshot = snapshot,
Mission = mission,
Alert = alert,
CompletedMissionRuns = completedRuns,
ProcessedExecutionIds = [.. current.ProcessedExecutionIds.Append(executionId).TakeLast(ProcessedExecutionLimit)],
Events = events
};
}, cancellationToken);
var shouldNotify = update.WasChanged ||
update.State.ProcessedExecutionIds.Contains(executionId, StringComparer.Ordinal);
return new(update.State.ToDashboard(), update.WasChanged, shouldNotify);
}
private async Task<WeatherStationState> ReadStateAsync(CancellationToken cancellationToken)
{
var entry = await dapr.GetStateEntryAsync<WeatherStationState>(StateStoreName, DashboardStateKey, cancellationToken: cancellationToken);
return entry.Value ?? CreateInitialState();
}
private async Task<WeatherStateUpdate> UpdateStateAsync(Func<WeatherStationState, WeatherStationState> update, CancellationToken cancellationToken)
{
for (var attempt = 0; attempt < 8; attempt++)
{
var entry = await dapr.GetStateEntryAsync<WeatherStationState>(StateStoreName, DashboardStateKey, cancellationToken: cancellationToken);
var current = entry.Value ?? CreateInitialState();
var next = update(current);
if (ReferenceEquals(current, next)) return new(current, false);
if (await dapr.TrySaveStateAsync(StateStoreName, DashboardStateKey, next, entry.ETag, FirstWrite, cancellationToken: cancellationToken)) return new(next, true);
}
throw new InvalidOperationException("Weather state changed concurrently too often. Dapr will retry this job invocation.");
}
private static WeatherStationState CreateInitialState() => new(CreateSnapshot("Initial station reading", 1), null, null, 0, [], []);
private static WeatherSnapshot CreateSnapshot(string refreshReason, int version) => new(version, DateTimeOffset.UtcNow, refreshReason, Enumerable.Range(1, 5).Select(index => new WeatherForecast(DateOnly.FromDateTime(DateTime.UtcNow.AddDays(index)), Random.Shared.Next(-20, 55), Summaries[Random.Shared.Next(Summaries.Length)])).ToArray());
private static WeatherAlert? EvaluateWeatherAlert(WeatherSnapshot forecast)
{
var rain = forecast.Forecasts.FirstOrDefault(day => day.Summary is "Rain watch" or "Storm cells");
if (rain is not null) return new("Rain alert", $"{rain.Summary} expected {rain.Date:ddd}.", forecast.Version);
var freeze = forecast.Forecasts.FirstOrDefault(day => day.TemperatureC <= 0);
if (freeze is not null) return new("Freeze warning", $"{freeze.TemperatureC}°C expected {freeze.Date:ddd}.", forecast.Version);
var heat = forecast.Forecasts.FirstOrDefault(day => day.TemperatureC >= 35 || day.Summary == "Heat advisory");
return heat is null ? null : new("Heat alert", $"{heat.TemperatureC}°C expected {heat.Date:ddd}.", forecast.Version);
}
private static WeatherRefreshEvent[] AddEvent(IEnumerable<WeatherRefreshEvent> events, string title, string detail, string jobName) => [new(DateTimeOffset.UtcNow, title, detail, jobName), .. events.Take(11)];
}
public sealed record WeatherJobExecutionResult(WeatherOperationsDashboard Dashboard, bool WasApplied, bool ShouldNotify);
public sealed record WeatherRefreshPayload(WeatherRefreshMode Mode, string DisplayName, string MissionId, DateTimeOffset FirstDueAt, int? IntervalSeconds, bool EvaluateWeatherRules = false)
{
public static WeatherRefreshPayload Create(WeatherRefreshMode mode, string displayName, DateTimeOffset firstDueAt, int? intervalSeconds, bool evaluateWeatherRules = false) => new(mode, displayName, Guid.NewGuid().ToString("N"), firstDueAt, intervalSeconds, evaluateWeatherRules);
public string GetExecutionId(string jobName, DateTimeOffset observedAt)
{
var occurrence = Mode switch
{
WeatherRefreshMode.Once => "once",
WeatherRefreshMode.Recurring when IntervalSeconds is > 0 => Math.Max(0, (long)Math.Floor((observedAt - FirstDueAt).TotalSeconds / IntervalSeconds.Value)).ToString(),
WeatherRefreshMode.Alerts => (observedAt.ToUnixTimeSeconds() / 60L).ToString(),
_ => throw new InvalidOperationException($"Unsupported refresh mode '{Mode}'.")
};
return $"{jobName}:{MissionId}:{occurrence}";
}
}
internal sealed record WeatherStationState(WeatherSnapshot Snapshot, WeatherMission? Mission, WeatherAlert? Alert, int CompletedMissionRuns, string[] ProcessedExecutionIds, WeatherRefreshEvent[] Events)
{
public WeatherOperationsDashboard ToDashboard() => new(Snapshot, Mission, Alert, Events);
}
internal sealed record WeatherStateUpdate(WeatherStationState State, bool WasChanged);WeatherUpdatePublisher.cs
It broadcasts refreshed forecast snapshots as forecast-updated messages through the configured Dapr Redis pub/sub component, so updates can move between API instances.
using Dapr.Client;
using dapr_jobs_aspire.Contracts;
namespace dapr_jobs_aspire.ApiService;
public sealed class WeatherUpdatePublisher(DaprClient dapr)
{
public const string PubSubName = "pubsub";
public const string TopicName = "weather-updates";
public Task PublishAsync(WeatherSnapshot snapshot, CancellationToken cancellationToken) =>
dapr.PublishEventAsync(PubSubName, TopicName, new WeatherUpdate(snapshot.Version, snapshot.RefreshedAt), cancellationToken);
}WeatherUpdateStream.cs
It maintains the server-sent-events subscriber stream. It receives published update messages and delivers them to connected browser clients through the same-origin /weather/updates endpoint.
using System.Collections.Concurrent;
using System.Runtime.CompilerServices;
using System.Threading.Channels;
using dapr_jobs_aspire.Contracts;
namespace dapr_jobs_aspire.ApiService;
public sealed class WeatherUpdateStream
{
private readonly ConcurrentDictionary<Guid, Channel<WeatherUpdate>> subscribers = new();
public void Publish(WeatherSnapshot snapshot)
{
Publish(new WeatherUpdate(snapshot.Version, snapshot.RefreshedAt));
}
public void Publish(WeatherUpdate update)
{
foreach (var subscriber in subscribers.Values)
{
subscriber.Writer.TryWrite(update);
}
}
public async IAsyncEnumerable<WeatherUpdate> Subscribe(
[EnumeratorCancellation] CancellationToken cancellationToken)
{
var id = Guid.NewGuid();
var subscriber = Channel.CreateBounded<WeatherUpdate>(new BoundedChannelOptions(1)
{
FullMode = BoundedChannelFullMode.DropOldest,
SingleReader = true
});
subscribers.TryAdd(id, subscriber);
try
{
await foreach (var update in subscriber.Reader.ReadAllAsync(cancellationToken))
{
yield return update;
}
}
finally
{
subscribers.TryRemove(id, out _);
}
}
}Service Defaults
There's no change in the ServiceDefaults project. It remains as it was generated by the template.
Web App
The Web app is an interactive Blazor Server application that provides the weather-operations dashboard. It calls the API service through service discovery to display forecast and job state, start or cancel refresh schedules, and proxy the API's SSE stream through /weather/updates so the browser updates automatically. The UI code has intentionally been left out here to keep the focus on the Dapr Jobs; if you'd like access to it, feel free to get in touch.
Running the App
With almost everything in place, let's run the application.
aspire runOpen the Aspire dashboard, select the webfrontend resource, and navigate to Weather. Choose a one-time or recurring refresh to see the forecast version change when Dapr Scheduler invokes the API. Choose Refresh & alerts to keep refreshing every minute and display an alert when a generated forecast meets one of the weather rules.
The Aspire dashboard lets you inspect the web app, API, and API sidecar together. RedisInsight can inspect the Redis data behind statestore, while the logs make the lifecycle easy to trace: schedule created, sidecar callback received, forecast persisted, pub/sub update sent, and browser refreshed.
Scalar provides an interactive OpenAPI reference for the API. In development, MapScalarApiReference() exposes a modern UI at /scalar where you can browse all endpoints, along with their requests and responses, generated from the API's OpenAPI document.
Conclusion
This demo highlights how Dapr Jobs provides a consistent way to schedule and reliably run future work, whether a one-time refresh, a recurring interval, or a cron-based schedule. Instead of building and maintaining separate timer logic for each mode, ApiService hands scheduling and callback delivery off to Dapr's jobs API through a single, consistent contract, backed by the Scheduler service's at least once execution guarantee.
State persistence, a distributed lock for serializing schedule change, and pub/sub for propagating updates keep the application focused on the weather-refresh workflow instead of scheduling mechanics.
Combined with Aspire's orchestration, the path from a scheduled job to an automatically refreshed UI remains easy to run, inspect, and understand.