Redis durablejobs storage

Orleans.slnx

@@ -98,6 +98,7 @@
   </Folder>
   <Folder Name="/src/Extensions/Redis/">
     <Project Path="src/Redis/Orleans.Clustering.Redis/Orleans.Clustering.Redis.csproj" />
+    <Project Path="src/Redis/Orleans.DurableJobs.Redis/Orleans.DurableJobs.Redis.csproj" />
     <Project Path="src/Redis/Orleans.GrainDirectory.Redis/Orleans.GrainDirectory.Redis.csproj" />
     <Project Path="src/Redis/Orleans.Persistence.Redis/Orleans.Persistence.Redis.csproj" />
     <Project Path="src/Redis/Orleans.Reminders.Redis/Orleans.Reminders.Redis.csproj" />

src/Redis/Orleans.DurableJobs.Redis/Hosting/RedisDurableJobsExtensions.cs

@@ -0,0 +1,71 @@
+using Microsoft.Extensions.DependencyInjection;
+using Microsoft.Extensions.Options;
+using Orleans.Configuration;
+using Orleans.Configuration.Internal;
+using Orleans.DurableJobs;
+using Orleans.DurableJobs.Redis;
+
+namespace Orleans.Hosting;
+
+/// <summary>
+/// Extensions for configuring Redis durable jobs.
+/// </summary>
+public static class RedisDurableJobsExtensions
+{
+    /// <summary>
+    /// Adds durable jobs storage backed by Redis.
+    /// </summary>
+    /// <param name="builder">The builder.</param>
+    /// <param name="configure">The delegate used to configure the durable jobs storage.</param>
+    /// <returns>The provided <see cref="ISiloBuilder"/>, for chaining.</returns>
+    public static ISiloBuilder UseRedisDurableJobs(this ISiloBuilder builder, Action<RedisJobShardOptions> configure)
+    {
+        builder.ConfigureServices(services => services.UseRedisDurableJobs(configure));
+        return builder;
+    }
+
+    /// <summary>
+    /// Adds durable jobs storage backed by Redis.
+    /// </summary>
+    /// <param name="builder">The builder.</param>
+    /// <param name="configureOptions">The configuration delegate.</param>
+    /// <returns>The provided <see cref="ISiloBuilder"/>, for chaining.</returns>
+    public static ISiloBuilder UseRedisDurableJobs(this ISiloBuilder builder, Action<OptionsBuilder<RedisJobShardOptions>> configureOptions)
+    {
+        builder.ConfigureServices(services => services.UseRedisDurableJobs(configureOptions));
+        return builder;
+    }
+
+    /// <summary>
+    /// Adds durable jobs storage backed by Redis.
+    /// </summary>
+    /// <param name="services">The service collection.</param>
+    /// <param name="configure">The delegate used to configure the durable jobs storage.</param>
+    /// <returns>The provided <see cref="IServiceCollection"/>, for chaining.</returns>
+    public static IServiceCollection UseRedisDurableJobs(this IServiceCollection services, Action<RedisJobShardOptions> configure)
+    {
+        services.AddDurableJobs();
+        services.AddSingleton<RedisJobShardManager>();
+        services.AddFromExisting<JobShardManager, RedisJobShardManager>();
+        services.Configure(configure);
+        services.ConfigureFormatter<RedisJobShardOptions>();
+        return services;
+    }
+
+    /// <summary>
+    /// Adds durable jobs storage backed by Redis.
+    /// </summary>
+    /// <param name="services">The service collection.</param>
+    /// <param name="configureOptions">The configuration delegate.</param>
+    /// <returns>The provided <see cref="IServiceCollection"/>, for chaining.</returns>
+    public static IServiceCollection UseRedisDurableJobs(this IServiceCollection services, Action<OptionsBuilder<RedisJobShardOptions>>? configureOptions)
+    {
+        services.AddDurableJobs();
+        services.AddSingleton<RedisJobShardManager>();
+        services.AddFromExisting<JobShardManager, RedisJobShardManager>();
+        configureOptions?.Invoke(services.AddOptions<RedisJobShardOptions>());
+        services.ConfigureFormatter<RedisJobShardOptions>();
+        services.AddTransient<IConfigurationValidator>(sp => new RedisJobShardOptionsValidator(sp.GetRequiredService<IOptionsMonitor<RedisJobShardOptions>>().Get(Options.DefaultName), Options.DefaultName));
+        return services;
+    }
+}

src/Redis/Orleans.DurableJobs.Redis/Hosting/RedisJobShardOptions.cs

@@ -0,0 +1,66 @@
+using StackExchange.Redis;
+
+namespace Orleans.Hosting;
+
+/// <summary>
+/// Options for configuring the Redis durable jobs provider.
+/// </summary>
+public class RedisJobShardOptions
+{
+    /// <summary>
+    /// Gets or sets the Redis client configuration.
+    /// </summary>
+    [RedactRedisConfigurationOptions]
+    public ConfigurationOptions? ConfigurationOptions { get; set; }
+
+    /// <summary>
+    /// Gets or sets a delegate to create the Redis connection multiplexer.
+    /// </summary>
+    /// <remarks>
+    /// This delegate is called once during initialization to create the connection.
+    /// </remarks>
+    public Func<RedisJobShardOptions, Task<IConnectionMultiplexer>> CreateMultiplexer { get; set; } = DefaultCreateMultiplexer;
+
+    /// <summary>
+    /// Gets or sets the prefix for shard identifiers.
+    /// </summary>
+    /// <remarks>
+    /// This prefix is used to namespace shards in Redis, allowing multiple applications to share the same Redis instance.
+    /// </remarks>
+    public string ShardPrefix { get; set; } = "shard";
+
+    /// <summary>
+    /// Gets or sets the maximum number of retries when creating a shard in case of ID collisions.
+    /// </summary>
+    public int MaxShardCreationRetries { get; set; } = 5;
+
+    /// <summary>
+    /// Gets or sets the maximum number of job operations to batch together in a single write.
+    /// Default is 128 operations.
+    /// </summary>
+    public int MaxBatchSize { get; set; } = 128;
+
+    /// <summary>
+    /// Gets or sets the minimum number of job operations to batch together before flushing.
+    /// Default is 1 operation (immediate flush, optimized for latency).
+    /// </summary>
+    public int MinBatchSize { get; set; } = 1;
+
+    /// <summary>
+    /// Gets or sets the maximum time to wait for additional operations if the minimum batch size isn't reached
+    /// before flushing a batch.
+    /// Default is 100 milliseconds.
+    /// </summary>
+    public TimeSpan BatchFlushInterval { get; set; } = TimeSpan.FromMilliseconds(100);
+
+    /// <summary>
+    /// The default multiplexer creation delegate.
+    /// </summary>
+    public static async Task<IConnectionMultiplexer> DefaultCreateMultiplexer(RedisJobShardOptions options)
+        => await ConnectionMultiplexer.ConnectAsync(options.ConfigurationOptions!);
+}
+
+internal class RedactRedisConfigurationOptions : RedactAttribute
+{
+    public override string Redact(object value) => value is ConfigurationOptions cfg ? cfg.ToString(includePassword: false) : base.Redact(value);
+}

src/Redis/Orleans.DurableJobs.Redis/Hosting/RedisJobShardOptionsValidator.cs

@@ -0,0 +1,64 @@
+namespace Orleans.Hosting;
+
+/// <summary>
+/// Validates <see cref="RedisJobShardOptions"/>.
+/// </summary>
+public class RedisJobShardOptionsValidator : IConfigurationValidator
+{
+    private readonly RedisJobShardOptions _options;
+    private readonly string _name;
+
+    /// <summary>
+    /// Initializes a new instance of the <see cref="RedisJobShardOptionsValidator"/> class.
+    /// </summary>
+    /// <param name="options">The options.</param>
+    /// <param name="name">The name.</param>
+    public RedisJobShardOptionsValidator(RedisJobShardOptions options, string name)
+    {
+        _options = options;
+        _name = name;
+    }
+
+    /// <inheritdoc/>
+    public void ValidateConfiguration()
+    {
+        if (_options.ConfigurationOptions is null)
+        {
+            throw new OrleansConfigurationException($"Invalid configuration for {nameof(RedisJobShardOptions)} with name '{_name}'. {nameof(_options.ConfigurationOptions)} is required.");
+        }
+
+        if (_options.CreateMultiplexer is null)
+        {
+            throw new OrleansConfigurationException($"Invalid configuration for {nameof(RedisJobShardOptions)} with name '{_name}'. {nameof(_options.CreateMultiplexer)} is required.");
+        }
+        if (string.IsNullOrWhiteSpace(_options.ShardPrefix))
+        {
+            throw new OrleansConfigurationException($"Invalid configuration for {nameof(RedisJobShardOptions)} with name '{_name}'. {nameof(_options.ShardPrefix)} is required.");
+        }
+
+        if (_options.MaxShardCreationRetries < 1)
+        {
+            throw new OrleansConfigurationException($"Invalid configuration for {nameof(RedisJobShardOptions)} with name '{_name}'. {nameof(_options.MaxShardCreationRetries)} must be at least 1.");
+        }
+
+        if (_options.MaxBatchSize < 1)
+        {
+            throw new OrleansConfigurationException($"Invalid configuration for {nameof(RedisJobShardOptions)} with name '{_name}'. {nameof(_options.MaxBatchSize)} must be at least 1.");
+        }
+
+        if (_options.MinBatchSize < 1)
+        {
+            throw new OrleansConfigurationException($"Invalid configuration for {nameof(RedisJobShardOptions)} with name '{_name}'. {nameof(_options.MinBatchSize)} must be at least 1.");
+        }
+
+        if (_options.MinBatchSize > _options.MaxBatchSize)
+        {
+            throw new OrleansConfigurationException($"Invalid configuration for {nameof(RedisJobShardOptions)} with name '{_name}'. {nameof(_options.MinBatchSize)} must not exceed {nameof(_options.MaxBatchSize)}.");
+        }
+
+        if (_options.BatchFlushInterval < TimeSpan.Zero)
+        {
+            throw new OrleansConfigurationException($"Invalid configuration for {nameof(RedisJobShardOptions)} with name '{_name}'. {nameof(_options.BatchFlushInterval)} must not be negative.");
+        }
+    }
+}

src/Redis/Orleans.DurableJobs.Redis/JobOperation.cs

@@ -0,0 +1,106 @@
+using System.Text.Json.Serialization;
+
+namespace Orleans.DurableJobs.Redis;
+
+/// <summary>
+/// Represents an operation to be performed on a durable job.
+/// </summary>
+internal readonly struct JobOperation
+{
+    /// <summary>
+    /// The type of operation to perform.
+    /// </summary>
+    public enum OperationType
+    {
+        Add,
+        Remove,
+        Retry,
+    }
+
+    /// <summary>
+    /// Gets or sets the type of operation.
+    /// </summary>
+    public OperationType Type { get; init; }
+
+    /// <summary>
+    /// Gets or sets the job identifier.
+    /// </summary>
+    public string Id { get; init; }
+
+    /// <summary>
+    /// Gets or sets the job name (only used for Add operations).
+    /// </summary>
+    public string? Name { get; init; }
+
+    /// <summary>
+    /// Gets or sets the due time (used for Add and Retry operations).
+    /// </summary>
+    public DateTimeOffset? DueTime { get; init; }
+
+    /// <summary>
+    /// Gets or sets the target grain ID (only used for Add operations).
+    /// </summary>
+    public GrainId? TargetGrainId { get; init; }
+
+    /// <summary>
+    /// Gets or sets the job metadata (only used for Add operations).
+    /// </summary>
+    public IReadOnlyDictionary<string, string>? Metadata { get; init; }
+
+    /// <summary>
+    /// Creates an Add operation for scheduling a new job.
+    /// </summary>
+    /// <param name="id">The job identifier.</param>
+    /// <param name="name">The job name.</param>
+    /// <param name="dueTime">The job due time.</param>
+    /// <param name="targetGrainId">The target grain ID.</param>
+    /// <param name="metadata">The job metadata.</param>
+    /// <returns>A new JobOperation for adding a job.</returns>
+    /// <exception cref="ArgumentException">Thrown when <paramref name="id"/> or <paramref name="name"/> is null or empty.</exception>
+    public static JobOperation CreateAddOperation(string id, string name, DateTimeOffset dueTime, GrainId targetGrainId, IReadOnlyDictionary<string, string>? metadata)
+    {
+        ArgumentException.ThrowIfNullOrEmpty(id);
+        ArgumentException.ThrowIfNullOrEmpty(name);
+
+        return new() { Type = OperationType.Add, Id = id, Name = name, DueTime = dueTime, TargetGrainId = targetGrainId, Metadata = metadata };
+    }
+
+    /// <summary>
+    /// Creates a Remove operation for canceling a job.
+    /// </summary>
+    /// <param name="id">The job identifier.</param>
+    /// <returns>A new JobOperation for removing a job.</returns>
+    /// <exception cref="ArgumentException">Thrown when <paramref name="id"/> is null or empty.</exception>
+    public static JobOperation CreateRemoveOperation(string id)
+    {
+        ArgumentException.ThrowIfNullOrEmpty(id);
+
+        return new() { Type = OperationType.Remove, Id = id };
+    }
+
+    /// <summary>
+    /// Creates a Retry operation for rescheduling a job.
+    /// </summary>
+    /// <param name="id">The job identifier.</param>
+    /// <param name="dueTime">The new due time.</param>
+    /// <returns>A new JobOperation for retrying a job.</returns>
+    /// <exception cref="ArgumentException">Thrown when <paramref name="id"/> is null or empty.</exception>
+    public static JobOperation CreateRetryOperation(string id, DateTimeOffset dueTime)
+    {
+        ArgumentException.ThrowIfNullOrEmpty(id);
+
+        return new() { Type = OperationType.Retry, Id = id, DueTime = dueTime };
+    }
+}
+
+/// <summary>
+/// JSON serialization context for JobOperation with compile-time source generation.
+/// </summary>
+[JsonSerializable(typeof(JobOperation))]
+[JsonSourceGenerationOptions(
+    DefaultIgnoreCondition = JsonIgnoreCondition.WhenWritingDefault,
+    PropertyNamingPolicy = JsonKnownNamingPolicy.CamelCase,
+    WriteIndented = false)]
+internal partial class JobOperationJsonContext : JsonSerializerContext
+{
+}

src/Redis/Orleans.DurableJobs.Redis/Orleans.DurableJobs.Redis.csproj

@@ -0,0 +1,29 @@
+<Project Sdk="Microsoft.NET.Sdk">
+
+  <PropertyGroup>
+    <PackageReadmeFile>README.md</PackageReadmeFile>
+    <PackageId>Microsoft.Orleans.DurableJobs.Redis</PackageId>
+    <Title>Microsoft Orleans Redis Durable Jobs Provider</Title>
+    <Description>Microsoft Orleans durable jobs provider backed by Redis</Description>
+    <PackageTags>$(PackageTags) Redis</PackageTags>
+    <TargetFrameworks>$(DefaultTargetFrameworks)</TargetFrameworks>
+    <AssemblyName>Orleans.DurableJobs.Redis</AssemblyName>
+    <RootNamespace>Orleans.DurableJobs.Redis</RootNamespace>
+    <OrleansBuildTimeCodeGen>true</OrleansBuildTimeCodeGen>
+    <DefineConstants>$(DefineConstants)</DefineConstants>
+    <Nullable>enable</Nullable>
+    <VersionSuffix Condition="$(VersionSuffix) != ''">$(VersionSuffix).alpha.1</VersionSuffix>
+    <VersionSuffix Condition="$(VersionSuffix) == ''">alpha.1</VersionSuffix>
+  </PropertyGroup>
+
+  <ItemGroup>
+    <ProjectReference Include="$(SourceRoot)src\Orleans.Runtime\Orleans.Runtime.csproj" />
+    <ProjectReference Include="$(SourceRoot)src\Orleans.DurableJobs\Orleans.DurableJobs.csproj" />
+    <PackageReference Include="StackExchange.Redis" />
+  </ItemGroup>
+
+  <ItemGroup>
+    <None Include="README.md" Pack="true" PackagePath="\" />
+  </ItemGroup>
+
+</Project>

src/Redis/Orleans.DurableJobs.Redis/Properties/AssemblyInfo.cs

@@ -0,0 +1,3 @@
+using System.Runtime.CompilerServices;
+
+[assembly: InternalsVisibleTo("Tester.Redis")]