diff --git a/Orleans.slnx b/Orleans.slnx index c152e1ce77b..5f0aefd0046 100644 --- a/Orleans.slnx +++ b/Orleans.slnx @@ -34,7 +34,7 @@ - + @@ -77,7 +77,7 @@ - + diff --git a/src/Azure/Orleans.ScheduledJobs.AzureStorage/AzureStorageJobShard.Log.cs b/src/Azure/Orleans.DurableJobs.AzureStorage/AzureStorageJobShard.Log.cs similarity index 99% rename from src/Azure/Orleans.ScheduledJobs.AzureStorage/AzureStorageJobShard.Log.cs rename to src/Azure/Orleans.DurableJobs.AzureStorage/AzureStorageJobShard.Log.cs index a6f42ccf6bd..cd2ba74ec2a 100644 --- a/src/Azure/Orleans.ScheduledJobs.AzureStorage/AzureStorageJobShard.Log.cs +++ b/src/Azure/Orleans.DurableJobs.AzureStorage/AzureStorageJobShard.Log.cs @@ -1,7 +1,7 @@ using System; using Microsoft.Extensions.Logging; -namespace Orleans.ScheduledJobs.AzureStorage; +namespace Orleans.DurableJobs.AzureStorage; internal sealed partial class AzureStorageJobShard { diff --git a/src/Azure/Orleans.ScheduledJobs.AzureStorage/AzureStorageJobShard.cs b/src/Azure/Orleans.DurableJobs.AzureStorage/AzureStorageJobShard.cs similarity index 99% rename from src/Azure/Orleans.ScheduledJobs.AzureStorage/AzureStorageJobShard.cs rename to src/Azure/Orleans.DurableJobs.AzureStorage/AzureStorageJobShard.cs index 5fe10156f89..c73773882ea 100644 --- a/src/Azure/Orleans.ScheduledJobs.AzureStorage/AzureStorageJobShard.cs +++ b/src/Azure/Orleans.DurableJobs.AzureStorage/AzureStorageJobShard.cs @@ -18,7 +18,7 @@ using Orleans.Runtime; using Orleans.Serialization.Buffers.Adaptors; -namespace Orleans.ScheduledJobs.AzureStorage; +namespace Orleans.DurableJobs.AzureStorage; internal sealed partial class AzureStorageJobShard : JobShard { @@ -136,7 +136,7 @@ public async ValueTask InitializeAsync(CancellationToken cancellationToken) dueTime = retryEntries.newDueTime ?? dueTime; } - EnqueueJob(new ScheduledJob + EnqueueJob(new DurableJob { Id = op.Id, Name = op.Name!, diff --git a/src/Azure/Orleans.ScheduledJobs.AzureStorage/AzureStorageJobShardManager.cs b/src/Azure/Orleans.DurableJobs.AzureStorage/AzureStorageJobShardManager.cs similarity index 96% rename from src/Azure/Orleans.ScheduledJobs.AzureStorage/AzureStorageJobShardManager.cs rename to src/Azure/Orleans.DurableJobs.AzureStorage/AzureStorageJobShardManager.cs index 83b308dc3f4..4137af0419c 100644 --- a/src/Azure/Orleans.ScheduledJobs.AzureStorage/AzureStorageJobShardManager.cs +++ b/src/Azure/Orleans.DurableJobs.AzureStorage/AzureStorageJobShardManager.cs @@ -14,7 +14,7 @@ using Orleans.Hosting; using Orleans.Runtime; -namespace Orleans.ScheduledJobs.AzureStorage; +namespace Orleans.DurableJobs.AzureStorage; public sealed partial class AzureStorageJobShardManager : JobShardManager { @@ -57,12 +57,12 @@ public AzureStorageJobShardManager( { } - public override async Task> AssignJobShardsAsync(DateTimeOffset maxShardStartTime, CancellationToken cancellationToken) + public override async Task> AssignJobShardsAsync(DateTimeOffset maxShardStartTime, CancellationToken cancellationToken) { await InitializeIfNeeded(cancellationToken); LogAssigningShards(_logger, SiloAddress, maxShardStartTime, _containerName); - var result = new List(); + var result = new List(); await foreach (var blob in _client.GetBlobsAsync(traits: BlobTraits.Metadata, cancellationToken: cancellationToken, prefix: _blobPrefix)) { // Get the owner and creator of the shard @@ -174,7 +174,7 @@ async Task TryTakeOwnership(AzureStorageJobShard shard, IDictionary CreateShardAsync(DateTimeOffset minDueTime, DateTimeOffset maxDueTime, IDictionary metadata, CancellationToken cancellationToken) + public override async Task CreateShardAsync(DateTimeOffset minDueTime, DateTimeOffset maxDueTime, IDictionary metadata, CancellationToken cancellationToken) { await InitializeIfNeeded(cancellationToken); LogRegisteringShard(_logger, SiloAddress, minDueTime, maxDueTime, _containerName); @@ -217,7 +217,7 @@ public override async Task CreateShardAsync(DateTimeOffset minDueTime } } - public override async Task UnregisterShardAsync(IJobShard shard, CancellationToken cancellationToken) + public override async Task UnregisterShardAsync(Orleans.DurableJobs.IJobShard shard, CancellationToken cancellationToken) { var azureShard = shard as AzureStorageJobShard ?? throw new ArgumentException("Shard is not an AzureStorageJobShard", nameof(shard)); LogUnregisteringShard(_logger, shard.Id, SiloAddress); diff --git a/src/Azure/Orleans.ScheduledJobs.AzureStorage/Hosting/AzureStorageScheduledJobsExtensions.cs b/src/Azure/Orleans.DurableJobs.AzureStorage/Hosting/AzureStorageDurableJobsExtensions.cs similarity index 60% rename from src/Azure/Orleans.ScheduledJobs.AzureStorage/Hosting/AzureStorageScheduledJobsExtensions.cs rename to src/Azure/Orleans.DurableJobs.AzureStorage/Hosting/AzureStorageDurableJobsExtensions.cs index 983273821f5..ccb8d80fb4b 100644 --- a/src/Azure/Orleans.ScheduledJobs.AzureStorage/Hosting/AzureStorageScheduledJobsExtensions.cs +++ b/src/Azure/Orleans.DurableJobs.AzureStorage/Hosting/AzureStorageDurableJobsExtensions.cs @@ -3,36 +3,36 @@ using Microsoft.Extensions.Options; using Orleans.Configuration; using Orleans.Configuration.Internal; -using Orleans.ScheduledJobs; -using Orleans.ScheduledJobs.AzureStorage; +using Orleans.DurableJobs; +using Orleans.DurableJobs.AzureStorage; namespace Orleans.Hosting; /// -/// Extensions for configuring Azure Blob Storage scheduled jobs. +/// Extensions for configuring Azure Blob Storage durable jobs. /// -public static class AzureStorageScheduledJobsExtensions +public static class AzureStorageDurableJobsExtensions { /// - /// Adds scheduled jobs storage backed by Azure Blob Storage. + /// Adds durable jobs storage backed by Azure Blob Storage. /// /// /// The builder. /// /// - /// The delegate used to configure the scheduled jobs storage. + /// The delegate used to configure the durable jobs storage. /// /// /// The provided , for chaining. /// - public static ISiloBuilder UseAzureBlobScheduledJobs(this ISiloBuilder builder, Action configure) + public static ISiloBuilder UseAzureBlobDurableJobs(this ISiloBuilder builder, Action configure) { - builder.ConfigureServices(services => services.UseAzureBlobScheduledJobs(configure)); + builder.ConfigureServices(services => services.UseAzureBlobDurableJobs(configure)); return builder; } /// - /// Adds scheduled jobs storage backed by Azure Blob Storage. + /// Adds durable jobs storage backed by Azure Blob Storage. /// /// /// The builder. @@ -43,36 +43,36 @@ public static ISiloBuilder UseAzureBlobScheduledJobs(this ISiloBuilder builder, /// /// The provided , for chaining. /// - public static ISiloBuilder UseAzureBlobScheduledJobs(this ISiloBuilder builder, Action> configureOptions) + public static ISiloBuilder UseAzureBlobDurableJobs(this ISiloBuilder builder, Action> configureOptions) { - builder.ConfigureServices(services => services.UseAzureBlobScheduledJobs(configureOptions)); + builder.ConfigureServices(services => services.UseAzureBlobDurableJobs(configureOptions)); return builder; } /// - /// Adds scheduled jobs storage backed by Azure Blob Storage. + /// Adds durable jobs storage backed by Azure Blob Storage. /// /// /// The service collection. /// /// - /// The delegate used to configure the scheduled jobs storage. + /// The delegate used to configure the durable jobs storage. /// /// /// The provided , for chaining. /// - public static IServiceCollection UseAzureBlobScheduledJobs(this IServiceCollection services, Action configure) + public static IServiceCollection UseAzureBlobDurableJobs(this IServiceCollection services, Action configure) { - services.AddScheduledJobs(); + services.AddDurableJobs(); services.AddSingleton(); - services.AddFromExisting(); + services.AddFromExisting(); services.Configure(configure); services.ConfigureFormatter(); return services; } /// - /// Adds scheduled jobs storage backed by Azure Blob Storage. + /// Adds durable jobs storage backed by Azure Blob Storage. /// /// /// The service collection. @@ -83,11 +83,11 @@ public static IServiceCollection UseAzureBlobScheduledJobs(this IServiceCollecti /// /// The provided , for chaining. /// - public static IServiceCollection UseAzureBlobScheduledJobs(this IServiceCollection services, Action> configureOptions) + public static IServiceCollection UseAzureBlobDurableJobs(this IServiceCollection services, Action> configureOptions) { - services.AddScheduledJobs(); + services.AddDurableJobs(); services.AddSingleton(); - services.AddFromExisting(); + services.AddFromExisting(); configureOptions?.Invoke(services.AddOptions()); services.ConfigureFormatter(); services.AddTransient(sp => new AzureStorageJobShardOptionsValidator(sp.GetRequiredService>().Get(Options.DefaultName), Options.DefaultName)); diff --git a/src/Azure/Orleans.ScheduledJobs.AzureStorage/Hosting/AzureStorageJobShardOptions.cs b/src/Azure/Orleans.DurableJobs.AzureStorage/Hosting/AzureStorageJobShardOptions.cs similarity index 95% rename from src/Azure/Orleans.ScheduledJobs.AzureStorage/Hosting/AzureStorageJobShardOptions.cs rename to src/Azure/Orleans.DurableJobs.AzureStorage/Hosting/AzureStorageJobShardOptions.cs index f4f85e519ad..5139b1cd071 100644 --- a/src/Azure/Orleans.ScheduledJobs.AzureStorage/Hosting/AzureStorageJobShardOptions.cs +++ b/src/Azure/Orleans.DurableJobs.AzureStorage/Hosting/AzureStorageJobShardOptions.cs @@ -11,7 +11,7 @@ public class AzureStorageJobShardOptions public BlobServiceClient BlobServiceClient { get; set; } = null!; /// - /// Gets or sets the name of the container used to store scheduled jobs. + /// Gets or sets the name of the container used to store durable jobs. /// public string ContainerName { get; set; } = "jobs"; diff --git a/src/Azure/Orleans.ScheduledJobs.AzureStorage/Hosting/AzureStorageJobShardOptionsValidator.cs b/src/Azure/Orleans.DurableJobs.AzureStorage/Hosting/AzureStorageJobShardOptionsValidator.cs similarity index 100% rename from src/Azure/Orleans.ScheduledJobs.AzureStorage/Hosting/AzureStorageJobShardOptionsValidator.cs rename to src/Azure/Orleans.DurableJobs.AzureStorage/Hosting/AzureStorageJobShardOptionsValidator.cs diff --git a/src/Azure/Orleans.ScheduledJobs.AzureStorage/JobOperation.cs b/src/Azure/Orleans.DurableJobs.AzureStorage/JobOperation.cs similarity index 97% rename from src/Azure/Orleans.ScheduledJobs.AzureStorage/JobOperation.cs rename to src/Azure/Orleans.DurableJobs.AzureStorage/JobOperation.cs index 462c18f89f8..834e858ada3 100644 --- a/src/Azure/Orleans.ScheduledJobs.AzureStorage/JobOperation.cs +++ b/src/Azure/Orleans.DurableJobs.AzureStorage/JobOperation.cs @@ -4,10 +4,10 @@ using System.Text.Json.Serialization; using Orleans.Runtime; -namespace Orleans.ScheduledJobs.AzureStorage; +namespace Orleans.DurableJobs.AzureStorage; /// -/// Represents an operation to be performed on a scheduled job. +/// Represents an operation to be performed on a durable job. /// internal struct JobOperation { diff --git a/src/Azure/Orleans.ScheduledJobs.AzureStorage/NetstringJsonSerializer.cs b/src/Azure/Orleans.DurableJobs.AzureStorage/NetstringJsonSerializer.cs similarity index 99% rename from src/Azure/Orleans.ScheduledJobs.AzureStorage/NetstringJsonSerializer.cs rename to src/Azure/Orleans.DurableJobs.AzureStorage/NetstringJsonSerializer.cs index be38d023842..f47575c2162 100644 --- a/src/Azure/Orleans.ScheduledJobs.AzureStorage/NetstringJsonSerializer.cs +++ b/src/Azure/Orleans.DurableJobs.AzureStorage/NetstringJsonSerializer.cs @@ -10,7 +10,7 @@ using System.Threading.Tasks; using Orleans.Serialization.Buffers.Adaptors; -namespace Orleans.ScheduledJobs.AzureStorage; +namespace Orleans.DurableJobs.AzureStorage; /// /// Provides methods for serializing and deserializing JSON data using the netstring format. diff --git a/src/Azure/Orleans.ScheduledJobs.AzureStorage/Orleans.ScheduledJobs.AzureStorage.csproj b/src/Azure/Orleans.DurableJobs.AzureStorage/Orleans.DurableJobs.AzureStorage.csproj similarity index 64% rename from src/Azure/Orleans.ScheduledJobs.AzureStorage/Orleans.ScheduledJobs.AzureStorage.csproj rename to src/Azure/Orleans.DurableJobs.AzureStorage/Orleans.DurableJobs.AzureStorage.csproj index 150c3f67774..a6d8eb91a48 100644 --- a/src/Azure/Orleans.ScheduledJobs.AzureStorage/Orleans.ScheduledJobs.AzureStorage.csproj +++ b/src/Azure/Orleans.DurableJobs.AzureStorage/Orleans.DurableJobs.AzureStorage.csproj @@ -2,13 +2,13 @@ README.md - Microsoft.Orleans.ScheduledJobs.AzureStorage - Microsoft Orleans Azure Storage Scheduled Jobs Provider - Microsoft Orleans scheduled jobs provider backed by Azure Blob Storage + Microsoft.Orleans.DurableJobs.AzureStorage + Microsoft Orleans Azure Storage Durable Jobs Provider + Microsoft Orleans durable jobs provider backed by Azure Blob Storage $(PackageTags) Azure Storage $(DefaultTargetFrameworks) - Orleans.ScheduledJobs.AzureStorage - Orleans.ScheduledJobs.AzureStorage + Orleans.DurableJobs.AzureStorage + Orleans.DurableJobs.AzureStorage true $(DefineConstants) enable @@ -18,7 +18,7 @@ - + diff --git a/src/Azure/Orleans.ScheduledJobs.AzureStorage/Properties/AssemblyInfo.cs b/src/Azure/Orleans.DurableJobs.AzureStorage/Properties/AssemblyInfo.cs similarity index 100% rename from src/Azure/Orleans.ScheduledJobs.AzureStorage/Properties/AssemblyInfo.cs rename to src/Azure/Orleans.DurableJobs.AzureStorage/Properties/AssemblyInfo.cs diff --git a/src/Azure/Orleans.ScheduledJobs.AzureStorage/README.md b/src/Azure/Orleans.DurableJobs.AzureStorage/README.md similarity index 86% rename from src/Azure/Orleans.ScheduledJobs.AzureStorage/README.md rename to src/Azure/Orleans.DurableJobs.AzureStorage/README.md index 81d3599ffc5..8bc3def4cb0 100644 --- a/src/Azure/Orleans.ScheduledJobs.AzureStorage/README.md +++ b/src/Azure/Orleans.DurableJobs.AzureStorage/README.md @@ -1,7 +1,7 @@ -# Microsoft Orleans Scheduled Jobs for Azure Storage +# Microsoft Orleans Durable Jobs for Azure Storage ## Introduction -Microsoft Orleans Scheduled Jobs for Azure Storage provides persistent storage for Orleans scheduled jobs using Azure Blob Storage. This allows your Orleans applications to schedule jobs that survive silo restarts, grain deactivation, and cluster reconfigurations. Jobs are stored in append blobs, providing efficient storage and retrieval for time-based job scheduling. +Microsoft Orleans Durable Jobs for Azure Storage provides persistent storage for Orleans Durable Jobs using Azure Blob Storage. This allows your Orleans applications to schedule jobs that survive silo restarts, grain deactivation, and cluster reconfigurations. Jobs are stored in append blobs, providing efficient storage and retrieval for time-based job scheduling. ## Getting Started @@ -9,8 +9,8 @@ Microsoft Orleans Scheduled Jobs for Azure Storage provides persistent storage f To use this package, install it via NuGet along with the core package: ```shell -dotnet add package Microsoft.Orleans.ScheduledJobs -dotnet add package Microsoft.Orleans.ScheduledJobs.AzureStorage +dotnet add package Microsoft.Orleans.DurableJobs +dotnet add package Microsoft.Orleans.DurableJobs.AzureStorage ``` ### Configuration @@ -27,12 +27,12 @@ builder.UseOrleans(siloBuilder => { siloBuilder .UseAzureStorageClustering(options => options.ConfigureTableServiceClient("YOUR_STORAGE_ACCOUNT_URI")) - .UseAzureStorageScheduledJobs(options => + .UseAzureStorageDurableJobs(options => { options.Configure(o => { o.BlobServiceClient = new BlobServiceClient("YOUR_AZURE_STORAGE_CONNECTION_STRING"); - o.ContainerName = "scheduled-jobs"; + o.ContainerName = "durable-jobs"; }); }); }); @@ -53,7 +53,7 @@ builder.UseOrleans(siloBuilder => { siloBuilder .UseAzureStorageClustering(options => options.ConfigureTableServiceClient("YOUR_STORAGE_ACCOUNT_URI")) - .UseAzureStorageScheduledJobs(options => + .UseAzureStorageDurableJobs(options => { options.Configure(o => { @@ -61,7 +61,7 @@ builder.UseOrleans(siloBuilder => o.BlobServiceClient = new BlobServiceClient( new Uri("https://youraccount.blob.core.windows.net"), credential); - o.ContainerName = "scheduled-jobs"; + o.ContainerName = "durable-jobs"; }); }); }); @@ -78,18 +78,18 @@ builder.UseOrleans(siloBuilder => { siloBuilder .UseAzureStorageClustering(options => options.ConfigureTableServiceClient(connectionString)) - .UseAzureStorageScheduledJobs(options => + .UseAzureStorageDurableJobs(options => { options.Configure(o => { o.BlobServiceClient = new BlobServiceClient(connectionString); // Use different containers for different environments - o.ContainerName = $"scheduled-jobs-{Environment.GetEnvironmentVariable("ASPNETCORE_ENVIRONMENT")?.ToLowerInvariant()}"; + o.ContainerName = $"durable-jobs-{Environment.GetEnvironmentVariable("ASPNETCORE_ENVIRONMENT")?.ToLowerInvariant()}"; }); }) .ConfigureServices(services => { - services.Configure(options => + services.Configure(options => { // Shard duration: balance between latency and storage overhead options.ShardDuration = TimeSpan.FromMinutes(5); @@ -123,7 +123,7 @@ builder.UseOrleans(siloBuilder => ### Email Scheduling with Cancellation ```csharp using Orleans; -using Orleans.ScheduledJobs; +using Orleans.DurableJobs; public interface IEmailGrain : IGrainWithStringKey { @@ -131,15 +131,15 @@ public interface IEmailGrain : IGrainWithStringKey Task CancelScheduledEmail(); } -public class EmailGrain : Grain, IEmailGrain, IScheduledJobHandler +public class EmailGrain : Grain, IEmailGrain, IDurableJobHandler { - private readonly ILocalScheduledJobManager _jobManager; + private readonly ILocalDurableJobManager _jobManager; private readonly IEmailService _emailService; private readonly ILogger _logger; - private IScheduledJob? _scheduledEmailJob; + private IDurableJob? _durableEmailJob; public EmailGrain( - ILocalScheduledJobManager jobManager, + ILocalDurableJobManager jobManager, IEmailService emailService, ILogger logger) { @@ -157,7 +157,7 @@ public class EmailGrain : Grain, IEmailGrain, IScheduledJobHandler ["Body"] = body }; - _scheduledEmailJob = await _jobManager.ScheduleJobAsync( + _durableEmailJob = await _jobManager.ScheduleJobAsync( this.GetGrainId(), "SendEmail", sendTime, @@ -165,30 +165,30 @@ public class EmailGrain : Grain, IEmailGrain, IScheduledJobHandler _logger.LogInformation( "Scheduled email to {EmailAddress} for {SendTime} (JobId: {JobId})", - emailAddress, sendTime, _scheduledEmailJob.Id); + emailAddress, sendTime, _durableEmailJob.Id); } public async Task CancelScheduledEmail() { - if (_scheduledEmailJob is null) + if (_durableEmailJob is null) { _logger.LogWarning("No scheduled email to cancel"); return; } - var canceled = await _jobManager.TryCancelScheduledJobAsync(_scheduledEmailJob); + var canceled = await _jobManager.TryCancelDurableJobAsync(_durableEmailJob); if (canceled) { - _logger.LogInformation("Email job {JobId} canceled successfully", _scheduledEmailJob.Id); - _scheduledEmailJob = null; + _logger.LogInformation("Email job {JobId} canceled successfully", _durableEmailJob.Id); + _durableEmailJob = null; } else { - _logger.LogWarning("Failed to cancel email job {JobId} (may have already executed)", _scheduledEmailJob.Id); + _logger.LogWarning("Failed to cancel email job {JobId} (may have already executed)", _durableEmailJob.Id); } } - public async Task ExecuteJobAsync(IScheduledJobContext context, CancellationToken cancellationToken) + public async Task ExecuteJobAsync(IDurableJobContext context, CancellationToken cancellationToken) { var emailAddress = this.GetPrimaryKeyString(); var subject = context.Job.Metadata?["Subject"]; @@ -202,7 +202,7 @@ public class EmailGrain : Grain, IEmailGrain, IScheduledJobHandler { await _emailService.SendEmailAsync(emailAddress, subject, body, cancellationToken); _logger.LogInformation("Email sent successfully to {EmailAddress}", emailAddress); - _scheduledEmailJob = null; + _durableEmailJob = null; } catch (Exception ex) { @@ -221,16 +221,16 @@ public interface IOrderGrain : IGrainWithGuidKey Task CancelOrder(); } -public class OrderGrain : Grain, IOrderGrain, IScheduledJobHandler +public class OrderGrain : Grain, IOrderGrain, IDurableJobHandler { - private readonly ILocalScheduledJobManager _jobManager; + private readonly ILocalDurableJobManager _jobManager; private readonly IOrderService _orderService; private readonly IGrainFactory _grainFactory; private readonly ILogger _logger; private OrderDetails? _orderDetails; public OrderGrain( - ILocalScheduledJobManager jobManager, + ILocalDurableJobManager jobManager, IOrderService orderService, IGrainFactory grainFactory, ILogger logger) @@ -286,7 +286,7 @@ public class OrderGrain : Grain, IOrderGrain, IScheduledJobHandler _logger.LogInformation("Order {OrderId} canceled", orderId); } - public async Task ExecuteJobAsync(IScheduledJobContext context, CancellationToken cancellationToken) + public async Task ExecuteJobAsync(IDurableJobContext context, CancellationToken cancellationToken) { var step = context.Job.Metadata!["Step"]; var orderId = this.GetPrimaryKey(); @@ -311,7 +311,7 @@ public class OrderGrain : Grain, IOrderGrain, IScheduledJobHandler } } - private async Task HandlePaymentReminder(IScheduledJobContext context, CancellationToken ct) + private async Task HandlePaymentReminder(IDurableJobContext context, CancellationToken ct) { var orderId = this.GetPrimaryKey(); var order = await _orderService.GetOrderAsync(orderId, ct); @@ -430,7 +430,7 @@ public enum OrderStatus ### Concurrency Settings ```csharp -services.Configure(options => +services.Configure(options => { // Adjust based on your workload and Azure Storage limits options.MaxConcurrentJobsPerSilo = 50; @@ -449,8 +449,8 @@ services.Configure(options => ### Enable Logging ```csharp -builder.Logging.AddFilter("Orleans.ScheduledJobs", LogLevel.Information); -builder.Logging.AddFilter("Orleans.ScheduledJobs.AzureStorage", LogLevel.Information); +builder.Logging.AddFilter("Orleans.DurableJobs", LogLevel.Information); +builder.Logging.AddFilter("Orleans.DurableJobs.AzureStorage", LogLevel.Information); ``` ### Key Metrics to Monitor @@ -487,7 +487,7 @@ var blobServiceClient = new BlobServiceClient(storageAccountUri, credential); For more comprehensive documentation, please refer to: - [Microsoft Orleans Documentation](https://learn.microsoft.com/dotnet/orleans/) - [Azure Blob Storage Documentation](https://learn.microsoft.com/azure/storage/blobs/) -- [Orleans Scheduled Jobs Core Package](../../../Orleans.ScheduledJobs/README.md) +- [Orleans Durable Jobs Core Package](../../../Orleans.DurableJobs/README.md) ## Feedback & Contributing - If you have any issues or would like to provide feedback, please [open an issue on GitHub](https://github.com/dotnet/orleans/issues) diff --git a/src/Orleans.Core/Properties/AssemblyInfo.cs b/src/Orleans.Core/Properties/AssemblyInfo.cs index 552b4929821..75b425ac205 100644 --- a/src/Orleans.Core/Properties/AssemblyInfo.cs +++ b/src/Orleans.Core/Properties/AssemblyInfo.cs @@ -4,7 +4,7 @@ [assembly: InternalsVisibleTo("Orleans.CodeGeneration")] [assembly: InternalsVisibleTo("Orleans.CodeGeneration.Build")] [assembly: InternalsVisibleTo("Orleans.Runtime")] -[assembly: InternalsVisibleTo("Orleans.ScheduledJobs")] +[assembly: InternalsVisibleTo("Orleans.DurableJobs")] [assembly: InternalsVisibleTo("Orleans.Streaming")] [assembly: InternalsVisibleTo("Orleans.TestingHost")] diff --git a/src/Orleans.ScheduledJobs/ScheduledJob.cs b/src/Orleans.DurableJobs/DurableJob.cs similarity index 67% rename from src/Orleans.ScheduledJobs/ScheduledJob.cs rename to src/Orleans.DurableJobs/DurableJob.cs index 08e6175f18a..76632416b79 100644 --- a/src/Orleans.ScheduledJobs/ScheduledJob.cs +++ b/src/Orleans.DurableJobs/DurableJob.cs @@ -2,23 +2,23 @@ using System.Collections.Generic; using Orleans.Runtime; -namespace Orleans.ScheduledJobs; +namespace Orleans.DurableJobs; /// -/// Represents a scheduled job that will be executed at a specific time. +/// Represents a durable job that will be executed at a specific time. /// [GenerateSerializer] -[Alias("Orleans.ScheduledJobs.ScheduledJob")] -public sealed class ScheduledJob +[Alias("Orleans.DurableJobs.DurableJob")] +public sealed class DurableJob { /// - /// Gets the unique identifier for this scheduled job. + /// Gets the unique identifier for this durable job. /// [Id(0)] public required string Id { get; init; } /// - /// Gets the name of the scheduled job. + /// Gets the name of the durable job. /// [Id(1)] public required string Name { get; init; } @@ -36,13 +36,13 @@ public sealed class ScheduledJob public GrainId TargetGrainId { get; init; } /// - /// Gets the identifier of the shard that manages this scheduled job. + /// Gets the identifier of the shard that manages this durable job. /// [Id(4)] public required string ShardId { get; init; } /// - /// Gets optional metadata associated with this scheduled job. + /// Gets optional metadata associated with this durable job. /// [Id(5)] public IReadOnlyDictionary? Metadata { get; init; } diff --git a/src/Orleans.ScheduledJobs/Hosting/ScheduledJobsExtensions.cs b/src/Orleans.DurableJobs/Hosting/DurableJobsExtensions.cs similarity index 57% rename from src/Orleans.ScheduledJobs/Hosting/ScheduledJobsExtensions.cs rename to src/Orleans.DurableJobs/Hosting/DurableJobsExtensions.cs index 12d3297d06e..ea9319a4b42 100644 --- a/src/Orleans.ScheduledJobs/Hosting/ScheduledJobsExtensions.cs +++ b/src/Orleans.DurableJobs/Hosting/DurableJobsExtensions.cs @@ -3,70 +3,70 @@ using Microsoft.Extensions.Logging; using Orleans.Configuration.Internal; using Orleans.Runtime; -using Orleans.ScheduledJobs; +using Orleans.DurableJobs; namespace Orleans.Hosting; /// -/// Extensions to for configuring scheduled jobs. +/// Extensions to for configuring durable jobs. /// -public static class ScheduledJobsExtensions +public static class DurableJobsExtensions { /// - /// Adds support for scheduled jobs to this silo. + /// Adds support for durable jobs to this silo. /// /// The builder. /// The silo builder. - public static ISiloBuilder AddScheduledJobs(this ISiloBuilder builder) => builder.ConfigureServices(services => AddScheduledJobs(services)); + public static ISiloBuilder AddDurableJobs(this ISiloBuilder builder) => builder.ConfigureServices(services => AddDurableJobs(services)); /// - /// Adds support for scheduled jobs to this silo. + /// Adds support for durable jobs to this silo. /// /// The services. - public static void AddScheduledJobs(this IServiceCollection services) + public static void AddDurableJobs(this IServiceCollection services) { - if (services.Any(service => service.ServiceType.Equals(typeof(LocalScheduledJobManager)))) + if (services.Any(service => service.ServiceType.Equals(typeof(LocalDurableJobManager)))) { return; } - services.AddSingleton(); + services.AddSingleton(); services.AddSingleton(); - services.AddSingleton(); - services.AddFromExisting(); - services.AddFromExisting, LocalScheduledJobManager>(); - services.AddKeyedTransient(typeof(IScheduledJobReceiverExtension), (sp, _) => + services.AddSingleton(); + services.AddFromExisting(); + services.AddFromExisting, LocalDurableJobManager>(); + services.AddKeyedTransient(typeof(IDurableJobReceiverExtension), (sp, _) => { var grainContextAccessor = sp.GetRequiredService(); - return new ScheduledJobReceiverExtension(grainContextAccessor.GrainContext, sp.GetRequiredService>()); + return new DurableJobReceiverExtension(grainContextAccessor.GrainContext, sp.GetRequiredService>()); }); } /// - /// Configures scheduled jobs storage using an in-memory, non-persistent store. + /// Configures durable jobs storage using an in-memory, non-persistent store. /// /// /// Note that this is for development and testing scenarios only and should not be used in production. /// /// The silo host builder. /// The provided , for chaining. - public static ISiloBuilder UseInMemoryScheduledJobs(this ISiloBuilder builder) + public static ISiloBuilder UseInMemoryDurableJobs(this ISiloBuilder builder) { - builder.AddScheduledJobs(); + builder.AddDurableJobs(); - builder.ConfigureServices(services => services.UseInMemoryScheduledJobs()); + builder.ConfigureServices(services => services.UseInMemoryDurableJobs()); return builder; } /// - /// Configures scheduled jobs storage using an in-memory, non-persistent store. + /// Configures durable jobs storage using an in-memory, non-persistent store. /// /// /// Note that this is for development and testing scenarios only and should not be used in production. /// /// The service collection. /// The provided , for chaining. - internal static IServiceCollection UseInMemoryScheduledJobs(this IServiceCollection services) + internal static IServiceCollection UseInMemoryDurableJobs(this IServiceCollection services) { services.AddSingleton(sp => { diff --git a/src/Orleans.ScheduledJobs/Hosting/ScheduledJobsOptions.cs b/src/Orleans.DurableJobs/Hosting/DurableJobsOptions.cs similarity index 68% rename from src/Orleans.ScheduledJobs/Hosting/ScheduledJobsOptions.cs rename to src/Orleans.DurableJobs/Hosting/DurableJobsOptions.cs index 2418b53a59f..4298c77378a 100644 --- a/src/Orleans.ScheduledJobs/Hosting/ScheduledJobsOptions.cs +++ b/src/Orleans.DurableJobs/Hosting/DurableJobsOptions.cs @@ -2,14 +2,14 @@ using Microsoft.Extensions.Logging; using Microsoft.Extensions.Options; using Orleans.Runtime; -using Orleans.ScheduledJobs; +using Orleans.DurableJobs; namespace Orleans.Hosting; /// -/// Configuration options for the scheduled jobs feature. +/// Configuration options for the durable jobs feature. /// -public sealed class ScheduledJobsOptions +public sealed class DurableJobsOptions { /// /// Gets or sets the duration of each job shard. Smaller values reduce latency but increase overhead. @@ -38,9 +38,9 @@ public sealed class ScheduledJobsOptions /// the time when the job should be retried, or if the job should not be retried. /// Default: Retry up to 5 times with exponential backoff (2^n seconds). /// - public Func ShouldRetry { get; set; } = DefaultShouldRetry; + public Func ShouldRetry { get; set; } = DefaultShouldRetry; - private static DateTimeOffset? DefaultShouldRetry(IScheduledJobContext jobContext, Exception ex) + private static DateTimeOffset? DefaultShouldRetry(IDurableJobContext jobContext, Exception ex) { // Default retry logic: retry up to 5 times with exponential backoff if (jobContext.DequeueCount >= 5) @@ -52,12 +52,12 @@ public sealed class ScheduledJobsOptions } } -public sealed class ScheduledJobsOptionsValidator : IConfigurationValidator +public sealed class DurableJobsOptionsValidator : IConfigurationValidator { - private readonly ILogger _logger; - private readonly IOptions _options; + private readonly ILogger _logger; + private readonly IOptions _options; - public ScheduledJobsOptionsValidator(ILogger logger, IOptions options) + public DurableJobsOptionsValidator(ILogger logger, IOptions options) { _logger = logger; _options = options; @@ -68,12 +68,12 @@ public void ValidateConfiguration() var options = _options.Value; if (options.ShardDuration <= TimeSpan.Zero) { - throw new OrleansConfigurationException("ScheduledJobsOptions.ShardDuration must be greater than zero."); + throw new OrleansConfigurationException("DurableJobsOptions.ShardDuration must be greater than zero."); } if (options.ShouldRetry == null) { - throw new OrleansConfigurationException("ScheduledJobsOptions.ShouldRetry must not be null."); + throw new OrleansConfigurationException("DurableJobsOptions.ShouldRetry must not be null."); } - _logger.LogInformation("ScheduledJobsOptions validated: ShardDuration={ShardDuration}", options.ShardDuration); + _logger.LogInformation("DurableJobsOptions validated: ShardDuration={ShardDuration}", options.ShardDuration); } } diff --git a/src/Orleans.ScheduledJobs/IScheduledJobHandler.cs b/src/Orleans.DurableJobs/IDurableJobHandler.cs similarity index 60% rename from src/Orleans.ScheduledJobs/IScheduledJobHandler.cs rename to src/Orleans.DurableJobs/IDurableJobHandler.cs index 37d3b1bc710..5d29ac5104b 100644 --- a/src/Orleans.ScheduledJobs/IScheduledJobHandler.cs +++ b/src/Orleans.DurableJobs/IDurableJobHandler.cs @@ -1,17 +1,17 @@ using System.Threading; using System.Threading.Tasks; -namespace Orleans.ScheduledJobs; +namespace Orleans.DurableJobs; /// -/// Provides contextual information about a scheduled job execution. +/// Provides contextual information about a durable job execution. /// -public interface IScheduledJobContext +public interface IDurableJobContext { /// - /// Gets the scheduled job being executed. + /// Gets the durable job being executed. /// - ScheduledJob Job { get; } + DurableJob Job { get; } /// /// Gets the unique identifier for this execution run. @@ -25,16 +25,16 @@ public interface IScheduledJobContext } /// -/// Represents the execution context for a scheduled job. +/// Represents the execution context for a durable job. /// [GenerateSerializer] -internal class ScheduledJobContext : IScheduledJobContext +internal class DurableJobContext : IDurableJobContext { /// - /// Gets the scheduled job being executed. + /// Gets the durable job being executed. /// [Id(0)] - public ScheduledJob Job { get; } + public DurableJob Job { get; } /// /// Gets the unique identifier for this execution run. @@ -49,12 +49,12 @@ internal class ScheduledJobContext : IScheduledJobContext public int DequeueCount { get; } /// - /// Initializes a new instance of the class. + /// Initializes a new instance of the class. /// - /// The scheduled job to execute. + /// The durable job to execute. /// The unique identifier for this execution run. /// The number of times this job has been dequeued, including retries. - public ScheduledJobContext(ScheduledJob job, string runId, int retryCount) + public DurableJobContext(DurableJob job, string runId, int retryCount) { Job = job; RunId = runId; @@ -63,22 +63,22 @@ public ScheduledJobContext(ScheduledJob job, string runId, int retryCount) } /// -/// Defines the interface for handling scheduled job execution. -/// Grains implement this interface to receive and process scheduled jobs. +/// Defines the interface for handling durable job execution. +/// Grains implement this interface to receive and process durable jobs. /// /// /// -/// Grains that implement this interface can be targeted by scheduled jobs. +/// Grains that implement this interface can be targeted by durable jobs. /// The method is invoked when the job's due time is reached. /// /// -/// The following example demonstrates a grain that implements : +/// The following example demonstrates a grain that implements : /// -/// public class MyGrain : Grain, IScheduledJobHandler +/// public class MyGrain : Grain, IDurableJobHandler /// { -/// public Task ExecuteJobAsync(IScheduledJobContext context, CancellationToken cancellationToken) +/// public Task ExecuteJobAsync(IDurableJobContext context, CancellationToken cancellationToken) /// { -/// // Process the scheduled job +/// // Process the durable job /// var jobName = context.Job.Name; /// var dueTime = context.Job.DueTime; /// @@ -90,24 +90,24 @@ public ScheduledJobContext(ScheduledJob job, string runId, int retryCount) /// /// /// -public interface IScheduledJobHandler +public interface IDurableJobHandler { /// - /// Executes the scheduled job with the provided context. + /// Executes the durable job with the provided context. /// - /// The context containing information about the scheduled job execution. + /// The context containing information about the durable job execution. /// A token to monitor for cancellation requests. /// A task that represents the asynchronous job execution operation. /// /// - /// This method is invoked by the Orleans scheduled jobs infrastructure when a job's due time is reached. + /// This method is invoked by the Orleans durable jobs infrastructure when a job's due time is reached. /// Implementations should handle job execution logic and can use information from the /// to access job metadata, dequeue count for retry logic, and other execution details. /// /// /// If the method throws an exception and a retry policy is configured, the job may be retried. - /// The property can be used to determine if this is a retry attempt. + /// The property can be used to determine if this is a retry attempt. /// /// - Task ExecuteJobAsync(IScheduledJobContext context, CancellationToken cancellationToken); + Task ExecuteJobAsync(IDurableJobContext context, CancellationToken cancellationToken); } diff --git a/src/Orleans.DurableJobs/IDurableJobReceiverExtension.cs b/src/Orleans.DurableJobs/IDurableJobReceiverExtension.cs new file mode 100644 index 00000000000..58b55b2a344 --- /dev/null +++ b/src/Orleans.DurableJobs/IDurableJobReceiverExtension.cs @@ -0,0 +1,61 @@ +using System; +using System.Threading; +using System.Threading.Tasks; +using Microsoft.Extensions.Logging; +using Orleans.Runtime; + +namespace Orleans.DurableJobs; + +/// +/// Extension interface for grains that can receive durable job invocations. +/// +internal interface IDurableJobReceiverExtension : IGrainExtension +{ + /// + /// Delivers a durable job to the grain for execution. + /// + /// The context containing information about the durable job. + /// A token to monitor for cancellation requests. + /// A task that represents the asynchronous operation. + Task DeliverDurableJobAsync(IDurableJobContext context, CancellationToken cancellationToken); +} + +/// +internal sealed partial class DurableJobReceiverExtension : IDurableJobReceiverExtension +{ + private readonly IGrainContext _grain; + private readonly ILogger _logger; + + public DurableJobReceiverExtension(IGrainContext grain, ILogger logger) + { + _grain = grain; + _logger = logger; + } + + public async Task DeliverDurableJobAsync(IDurableJobContext context, CancellationToken cancellationToken) + { + if (_grain.GrainInstance is IDurableJobHandler handler) + { + try + { + await handler.ExecuteJobAsync(context, cancellationToken); + } + catch (Exception ex) + { + LogErrorExecutingDurableJob(ex, context.Job.Id, _grain.GrainId); + throw; + } + } + else + { + LogGrainDoesNotImplementHandler(_grain.GrainId); + throw new InvalidOperationException($"Grain {_grain.GrainId} does not implement IDurableJobHandler"); + } + } + + [LoggerMessage(Level = LogLevel.Error, Message = "Error executing durable job {JobId} on grain {GrainId}")] + private partial void LogErrorExecutingDurableJob(Exception exception, string jobId, GrainId grainId); + + [LoggerMessage(Level = LogLevel.Error, Message = "Grain {GrainId} does not implement IDurableJobHandler")] + private partial void LogGrainDoesNotImplementHandler(GrainId grainId); +} diff --git a/src/Orleans.ScheduledJobs/ILocalScheduledJobManager.cs b/src/Orleans.DurableJobs/ILocalDurableJobManager.cs similarity index 66% rename from src/Orleans.ScheduledJobs/ILocalScheduledJobManager.cs rename to src/Orleans.DurableJobs/ILocalDurableJobManager.cs index df138c78757..65870473f62 100644 --- a/src/Orleans.ScheduledJobs/ILocalScheduledJobManager.cs +++ b/src/Orleans.DurableJobs/ILocalDurableJobManager.cs @@ -4,29 +4,29 @@ using System.Threading.Tasks; using Orleans.Runtime; -namespace Orleans.ScheduledJobs; +namespace Orleans.DurableJobs; /// /// Provides functionality for scheduling and managing jobs on the local silo. /// -public interface ILocalScheduledJobManager +public interface ILocalDurableJobManager { /// /// Schedules a job to be executed at a specific time on the target grain. /// - /// The grain identifier of the target grain that will receive the scheduled job. + /// The grain identifier of the target grain that will receive the durable job. /// The name of the job for identification purposes. /// The date and time when the job should be executed. /// Optional metadata associated with the job. /// A cancellation token to cancel the operation. - /// A representing the asynchronous operation that returns the scheduled job. - Task ScheduleJobAsync(GrainId target, string jobName, DateTimeOffset dueTime, IReadOnlyDictionary? metadata, CancellationToken cancellationToken); + /// A representing the asynchronous operation that returns the durable job. + Task ScheduleJobAsync(GrainId target, string jobName, DateTimeOffset dueTime, IReadOnlyDictionary? metadata, CancellationToken cancellationToken); /// - /// Attempts to cancel a previously scheduled job. + /// Attempts to cancel a previously scheduled durable job. /// - /// The scheduled job to cancel. + /// The durable job to cancel. /// A cancellation token to cancel the operation. /// A representing the asynchronous operation that returns if the job was successfully canceled; otherwise, . - Task TryCancelScheduledJobAsync(ScheduledJob job, CancellationToken cancellationToken); + Task TryCancelDurableJobAsync(DurableJob job, CancellationToken cancellationToken); } diff --git a/src/Orleans.ScheduledJobs/InMemoryJobQueue.cs b/src/Orleans.DurableJobs/InMemoryJobQueue.cs similarity index 84% rename from src/Orleans.ScheduledJobs/InMemoryJobQueue.cs rename to src/Orleans.DurableJobs/InMemoryJobQueue.cs index a283315bba7..47d76945304 100644 --- a/src/Orleans.ScheduledJobs/InMemoryJobQueue.cs +++ b/src/Orleans.DurableJobs/InMemoryJobQueue.cs @@ -4,13 +4,13 @@ using System.Threading; using System.Threading.Tasks; -namespace Orleans.ScheduledJobs; +namespace Orleans.DurableJobs; /// -/// Provides an in-memory priority queue for managing scheduled jobs based on their due times. +/// Provides an in-memory priority queue for managing durable jobs based on their due times. /// Jobs are organized into time-based buckets and enumerated asynchronously as they become due. /// -internal sealed class InMemoryJobQueue : IAsyncEnumerable +internal sealed class InMemoryJobQueue : IAsyncEnumerable { private readonly PriorityQueue _queue = new(); private readonly Dictionary _jobsIdToBucket = new(); @@ -24,13 +24,13 @@ internal sealed class InMemoryJobQueue : IAsyncEnumerable public int Count => _jobsIdToBucket.Count; /// - /// Adds a scheduled job to the queue with the specified dequeue count. + /// Adds a durable job to the queue with the specified dequeue count. /// - /// The scheduled job to enqueue. + /// The durable job to enqueue. /// The number of times this job has been dequeued previously. /// Thrown when attempting to enqueue a job to a completed queue. /// Thrown when job is null. - public void Enqueue(ScheduledJob job, int dequeueCount) + public void Enqueue(DurableJob job, int dequeueCount) { ArgumentNullException.ThrowIfNull(job); @@ -58,7 +58,7 @@ public void MarkAsComplete() } /// - /// Cancels a scheduled job by removing it from the queue. + /// Cancels a durable job by removing it from the queue. /// /// The unique identifier of the job to cancel. /// True if the job was found and removed; false if the job was not found. @@ -91,10 +91,10 @@ public bool CancelJob(string jobId) /// The job is removed from its current bucket and added to a new bucket based on the specified due time. /// The dequeue count from the context is preserved. /// - public void RetryJobLater(IScheduledJobContext jobContext, DateTimeOffset newDueTime) + public void RetryJobLater(IDurableJobContext jobContext, DateTimeOffset newDueTime) { var jobId = jobContext.Job.Id; - var newJob = new ScheduledJob + var newJob = new DurableJob { Id = jobContext.Job.Id, Name = jobContext.Job.Name, @@ -118,14 +118,14 @@ public void RetryJobLater(IScheduledJobContext jobContext, DateTimeOffset newDue } /// - /// Returns an asynchronous enumerator that yields scheduled jobs as they become due. + /// Returns an asynchronous enumerator that yields durable jobs as they become due. /// /// A token to monitor for cancellation requests. /// - /// An async enumerator that returns instances for jobs that are due. + /// An async enumerator that returns instances for jobs that are due. /// The enumerator checks for due jobs every second and terminates when the queue is marked complete and empty. /// - public async IAsyncEnumerator GetAsyncEnumerator(CancellationToken cancellationToken = default) + public async IAsyncEnumerator GetAsyncEnumerator(CancellationToken cancellationToken = default) { using var timer = new PeriodicTimer(TimeSpan.FromSeconds(1)); while (true) @@ -169,7 +169,7 @@ public async IAsyncEnumerator GetAsyncEnumerator(Cancellat if (shouldYield) { - yield return new ScheduledJobContext(job, Guid.NewGuid().ToString(), dequeueCount + 1); + yield return new DurableJobContext(job, Guid.NewGuid().ToString(), dequeueCount + 1); } } @@ -204,20 +204,20 @@ private JobBucket GetJobBucket(DateTimeOffset dueTime) internal sealed class JobBucket { - private readonly Dictionary _jobs = new(); + private readonly Dictionary _jobs = new(); public int Count => _jobs.Count; public DateTimeOffset DueTime { get; private set; } - public IEnumerable<(ScheduledJob Job, int DequeueCount)> Jobs => _jobs.Values; + public IEnumerable<(DurableJob Job, int DequeueCount)> Jobs => _jobs.Values; public JobBucket(DateTimeOffset dueTime) { DueTime = dueTime; } - public void AddJob(ScheduledJob job, int dequeueCount) + public void AddJob(DurableJob job, int dequeueCount) { _jobs[job.Id] = (job, dequeueCount); } diff --git a/src/Orleans.ScheduledJobs/InMemoryJobShard.cs b/src/Orleans.DurableJobs/InMemoryJobShard.cs similarity index 97% rename from src/Orleans.ScheduledJobs/InMemoryJobShard.cs rename to src/Orleans.DurableJobs/InMemoryJobShard.cs index d063ab03748..148a14b0716 100644 --- a/src/Orleans.ScheduledJobs/InMemoryJobShard.cs +++ b/src/Orleans.DurableJobs/InMemoryJobShard.cs @@ -5,7 +5,7 @@ using System.Threading.Tasks; using Orleans.Runtime; -namespace Orleans.ScheduledJobs; +namespace Orleans.DurableJobs; [DebuggerDisplay("ShardId={Id}, StartTime={StartTime}, EndTime={EndTime}")] internal sealed class InMemoryJobShard : JobShard diff --git a/src/Orleans.ScheduledJobs/JobShard.cs b/src/Orleans.DurableJobs/JobShard.cs similarity index 85% rename from src/Orleans.ScheduledJobs/JobShard.cs rename to src/Orleans.DurableJobs/JobShard.cs index ca7c9a4b0a8..0e0eb6f3184 100644 --- a/src/Orleans.ScheduledJobs/JobShard.cs +++ b/src/Orleans.DurableJobs/JobShard.cs @@ -4,15 +4,15 @@ using System.Threading.Tasks; using Orleans.Runtime; -namespace Orleans.ScheduledJobs; +namespace Orleans.DurableJobs; /// -/// Represents a shard of scheduled jobs that manages a collection of jobs within a specific time range. -/// A job shard is responsible for storing, retrieving, and managing the lifecycle of scheduled jobs +/// Represents a shard of durable jobs that manages a collection of jobs within a specific time range. +/// A job shard is responsible for storing, retrieving, and managing the lifecycle of durable jobs /// that fall within its designated time window. /// /// -/// Job shards are used to partition scheduled jobs across time ranges to improve scalability +/// Job shards are used to partition durable jobs across time ranges to improve scalability /// and performance. Each shard has a defined start and end time that determines which jobs /// it manages. Shards can be marked as complete when all jobs within their time range /// have been processed. @@ -48,10 +48,10 @@ public interface IJobShard : IAsyncDisposable bool IsAddingCompleted { get; } /// - /// Consumes scheduled jobs from this shard in order of their due time. + /// Consumes durable jobs from this shard in order of their due time. /// - /// An asynchronous enumerable of scheduled job contexts. - IAsyncEnumerable ConsumeScheduledJobsAsync(); + /// An asynchronous enumerable of durable job contexts. + IAsyncEnumerable ConsumeDurableJobsAsync(); /// /// Gets the number of jobs currently scheduled in this shard. @@ -67,7 +67,7 @@ public interface IJobShard : IAsyncDisposable Task MarkAsCompleteAsync(CancellationToken cancellationToken); /// - /// Removes a scheduled job from this shard. + /// Removes a durable job from this shard. /// /// The unique identifier of the job to remove. /// A token to cancel the operation. @@ -81,7 +81,7 @@ public interface IJobShard : IAsyncDisposable /// The new due time for the job. /// A token to cancel the operation. /// A task that represents the asynchronous operation. - Task RetryJobLaterAsync(IScheduledJobContext jobContext, DateTimeOffset newDueTime, CancellationToken cancellationToken); + Task RetryJobLaterAsync(IDurableJobContext jobContext, DateTimeOffset newDueTime, CancellationToken cancellationToken); /// /// Attempts to schedule a new job on this shard. @@ -91,9 +91,9 @@ public interface IJobShard : IAsyncDisposable /// The time when the job should be executed. /// Optional metadata to associate with the job. /// A token to cancel the operation. - /// A task that represents the asynchronous operation. The task result contains the scheduled job if successful, or null if the job could not be scheduled (e.g., the shard was marked as complete). + /// A task that represents the asynchronous operation. The task result contains the durable job if successful, or null if the job could not be scheduled (e.g., the shard was marked as complete). /// Thrown when the due time is outside the shard's time range. - Task TryScheduleJobAsync(GrainId target, string jobName, DateTimeOffset dueTime, IReadOnlyDictionary? metadata, CancellationToken cancellationToken); + Task TryScheduleJobAsync(GrainId target, string jobName, DateTimeOffset dueTime, IReadOnlyDictionary? metadata, CancellationToken cancellationToken); } /// @@ -136,13 +136,13 @@ protected JobShard(string id, DateTimeOffset startTime, DateTimeOffset endTime) public ValueTask GetJobCountAsync() => ValueTask.FromResult(_jobQueue.Count); /// - public IAsyncEnumerable ConsumeScheduledJobsAsync() + public IAsyncEnumerable ConsumeDurableJobsAsync() { return _jobQueue; } /// - public async Task TryScheduleJobAsync(GrainId target, string jobName, DateTimeOffset dueTime, IReadOnlyDictionary? metadata, CancellationToken cancellationToken) + public async Task TryScheduleJobAsync(GrainId target, string jobName, DateTimeOffset dueTime, IReadOnlyDictionary? metadata, CancellationToken cancellationToken) { if (IsAddingCompleted) { @@ -155,7 +155,7 @@ public IAsyncEnumerable ConsumeScheduledJobsAsync() } var jobId = Guid.NewGuid().ToString(); - var job = new ScheduledJob + var job = new DurableJob { Id = jobId, TargetGrainId = target, @@ -186,7 +186,7 @@ public Task MarkAsCompleteAsync(CancellationToken cancellationToken) } /// - public async Task RetryJobLaterAsync(IScheduledJobContext jobContext, DateTimeOffset newDueTime, CancellationToken cancellationToken) + public async Task RetryJobLaterAsync(IDurableJobContext jobContext, DateTimeOffset newDueTime, CancellationToken cancellationToken) { await PersistRetryJobAsync(jobContext.Job.Id, newDueTime, cancellationToken); _jobQueue.RetryJobLater(jobContext, newDueTime); @@ -197,7 +197,7 @@ public async Task RetryJobLaterAsync(IScheduledJobContext jobContext, DateTimeOf /// /// The job to enqueue. /// The number of times this job has been dequeued. - protected void EnqueueJob(ScheduledJob job, int dequeueCount) + protected void EnqueueJob(DurableJob job, int dequeueCount) { _jobQueue.Enqueue(job, dequeueCount); } diff --git a/src/Orleans.ScheduledJobs/JobShardManager.cs b/src/Orleans.DurableJobs/JobShardManager.cs similarity index 99% rename from src/Orleans.ScheduledJobs/JobShardManager.cs rename to src/Orleans.DurableJobs/JobShardManager.cs index 1f3ff5f69b0..b7ab58063aa 100644 --- a/src/Orleans.ScheduledJobs/JobShardManager.cs +++ b/src/Orleans.DurableJobs/JobShardManager.cs @@ -5,7 +5,7 @@ using System.Threading.Tasks; using Orleans.Runtime; -namespace Orleans.ScheduledJobs; +namespace Orleans.DurableJobs; /// /// Manages the lifecycle of job shards for a specific silo. diff --git a/src/Orleans.ScheduledJobs/LocalScheduledJobManager.Log.cs b/src/Orleans.DurableJobs/LocalDurableJobManager.Log.cs similarity index 93% rename from src/Orleans.ScheduledJobs/LocalScheduledJobManager.Log.cs rename to src/Orleans.DurableJobs/LocalDurableJobManager.Log.cs index fefc40fd921..ee3ffc1a7fe 100644 --- a/src/Orleans.ScheduledJobs/LocalScheduledJobManager.Log.cs +++ b/src/Orleans.DurableJobs/LocalDurableJobManager.Log.cs @@ -2,9 +2,9 @@ using Microsoft.Extensions.Logging; using Orleans.Runtime; -namespace Orleans.ScheduledJobs; +namespace Orleans.DurableJobs; -internal partial class LocalScheduledJobManager +internal partial class LocalDurableJobManager { [LoggerMessage( Level = LogLevel.Debug, @@ -20,25 +20,25 @@ internal partial class LocalScheduledJobManager [LoggerMessage( Level = LogLevel.Information, - Message = "LocalScheduledJobManager starting" + Message = "LocalDurableJobManager starting" )] private static partial void LogStarting(ILogger logger); [LoggerMessage( Level = LogLevel.Information, - Message = "LocalScheduledJobManager started" + Message = "LocalDurableJobManager started" )] private static partial void LogStarted(ILogger logger); [LoggerMessage( Level = LogLevel.Information, - Message = "LocalScheduledJobManager stopping. Running shards: {RunningShardCount}" + Message = "LocalDurableJobManager stopping. Running shards: {RunningShardCount}" )] private static partial void LogStopping(ILogger logger, int runningShardCount); [LoggerMessage( Level = LogLevel.Information, - Message = "LocalScheduledJobManager stopped" + Message = "LocalDurableJobManager stopped" )] private static partial void LogStopped(ILogger logger); diff --git a/src/Orleans.ScheduledJobs/LocalScheduledJobManager.cs b/src/Orleans.DurableJobs/LocalDurableJobManager.cs similarity index 91% rename from src/Orleans.ScheduledJobs/LocalScheduledJobManager.cs rename to src/Orleans.DurableJobs/LocalDurableJobManager.cs index 2d1d3097364..433b1aefca0 100644 --- a/src/Orleans.ScheduledJobs/LocalScheduledJobManager.cs +++ b/src/Orleans.DurableJobs/LocalDurableJobManager.cs @@ -12,20 +12,20 @@ using Orleans.Runtime; using Orleans.Runtime.Internal; -namespace Orleans.ScheduledJobs; +namespace Orleans.DurableJobs; /// -internal partial class LocalScheduledJobManager : SystemTarget, ILocalScheduledJobManager, ILifecycleParticipant +internal partial class LocalDurableJobManager : SystemTarget, ILocalDurableJobManager, ILifecycleParticipant { private readonly JobShardManager _shardManager; private readonly ShardExecutor _shardExecutor; private readonly IAsyncEnumerable _clusterMembershipUpdates; - private readonly ILogger _logger; - private readonly ScheduledJobsOptions _options; + private readonly ILogger _logger; + private readonly DurableJobsOptions _options; private readonly CancellationTokenSource _cts = new(); private Task? _listenForClusterChangesTask; private Task? _periodicCheckTask; - + // Shard tracking state private readonly ConcurrentDictionary _shardCache = new(); private readonly ConcurrentDictionary _writeableShards = new(); @@ -35,13 +35,13 @@ internal partial class LocalScheduledJobManager : SystemTarget, ILocalScheduledJ private static readonly IDictionary EmptyMetadata = new Dictionary(); - public LocalScheduledJobManager( + public LocalDurableJobManager( JobShardManager shardManager, ShardExecutor shardExecutor, IClusterMembershipService clusterMembership, - IOptions options, + IOptions options, SystemTargetShared shared, - ILogger logger) + ILogger logger) : base(SystemTargetGrainId.CreateGrainType("job-manager"), shared) { _shardManager = shardManager; @@ -52,12 +52,12 @@ public LocalScheduledJobManager( } /// - public async Task ScheduleJobAsync(GrainId target, string jobName, DateTimeOffset dueTime, IReadOnlyDictionary? metadata, CancellationToken cancellationToken) + public async Task ScheduleJobAsync(GrainId target, string jobName, DateTimeOffset dueTime, IReadOnlyDictionary? metadata, CancellationToken cancellationToken) { LogSchedulingJob(_logger, jobName, target, dueTime); - + var shardKey = GetShardKey(dueTime); - + while (true) { // Fast path: shard already exists @@ -69,7 +69,7 @@ public async Task ScheduleJobAsync(GrainId target, string jobName, LogJobScheduled(_logger, jobName, job.Id, existingShard.Id, target); return job; } - + // Shard is full or no longer writable, remove from writable shards and try again _writeableShards.TryRemove(shardKey, out _); continue; @@ -89,7 +89,7 @@ public async Task ScheduleJobAsync(GrainId target, string jobName, using var linkedCts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken, _cts.Token); var endTime = shardKey.Add(_options.ShardDuration); var newShard = await _shardManager.CreateShardAsync(shardKey, endTime, EmptyMetadata, linkedCts.Token); - + LogCreatingNewShard(_logger, shardKey); _writeableShards[shardKey] = newShard; _shardCache.TryAdd(newShard.Id, newShard); @@ -105,7 +105,7 @@ public async Task ScheduleJobAsync(GrainId target, string jobName, public void Participate(ISiloLifecycle lifecycle) { lifecycle.Subscribe( - nameof(LocalScheduledJobManager), + nameof(LocalDurableJobManager), ServiceLifecycleStage.Active, ct => Start(ct), ct => Stop(ct)); @@ -118,7 +118,7 @@ private Task Start(CancellationToken ct) using (var _ = new ExecutionContextSuppressor()) { _listenForClusterChangesTask = Task.Factory.StartNew( - state => ((LocalScheduledJobManager)state!).ProcessMembershipUpdates(), + state => ((LocalDurableJobManager)state!).ProcessMembershipUpdates(), this, CancellationToken.None, TaskCreationOptions.None, @@ -126,7 +126,7 @@ private Task Start(CancellationToken ct) _listenForClusterChangesTask.Ignore(); _periodicCheckTask = Task.Factory.StartNew( - state => ((LocalScheduledJobManager)state!).PeriodicShardCheck(), + state => ((LocalDurableJobManager)state!).PeriodicShardCheck(), this, CancellationToken.None, TaskCreationOptions.None, @@ -141,9 +141,9 @@ private Task Start(CancellationToken ct) private async Task Stop(CancellationToken ct) { LogStopping(_logger, _runningShards.Count); - + _cts.Cancel(); - + if (_listenForClusterChangesTask is not null) { await _listenForClusterChangesTask.SuppressThrowing(); @@ -153,17 +153,17 @@ private async Task Stop(CancellationToken ct) { await _periodicCheckTask.SuppressThrowing(); } - + await Task.WhenAll(_runningShards.Values.ToArray()); - + LogStopped(_logger); } /// - public async Task TryCancelScheduledJobAsync(ScheduledJob job, CancellationToken cancellationToken) + public async Task TryCancelDurableJobAsync(DurableJob job, CancellationToken cancellationToken) { LogCancellingJob(_logger, job.Id, job.Name, job.ShardId); - + if (!_shardCache.TryGetValue(job.ShardId, out var shard)) { LogJobCancellationFailed(_logger, job.Id, job.Name, job.ShardId); @@ -235,7 +235,7 @@ private async Task PeriodicShardCheck() await Task.WhenAny(timerTask, signalTask); LogCheckingPendingShards(_logger); - + // Clean up old writable shards that have passed their time window var now = DateTimeOffset.UtcNow; foreach (var key in _writeableShards.Keys.ToArray()) @@ -246,7 +246,7 @@ private async Task PeriodicShardCheck() _writeableShards.TryRemove(key, out _); } } - + // Query ShardManager for assigned shards (source of truth) var shards = await _shardManager.AssignJobShardsAsync(DateTime.UtcNow.AddHours(1), _cts.Token); if (shards.Count > 0) @@ -255,7 +255,7 @@ private async Task PeriodicShardCheck() foreach (var shard in shards) { _shardCache.TryAdd(shard.Id, shard); - + if (!_runningShards.ContainsKey(shard.Id)) { TryActivateShard(shard); @@ -306,7 +306,7 @@ private async Task RunShardWithCleanupAsync(IJobShard shard) try { await _shardExecutor.RunShardAsync(shard, _cts.Token); - + // Unregister the shard from the manager try { @@ -323,7 +323,7 @@ private async Task RunShardWithCleanupAsync(IJobShard shard) // Clean up tracking and dispose the shard _shardCache.TryRemove(shard.Id, out _); _runningShards.TryRemove(shard.Id, out _); - + try { await shard.DisposeAsync(); diff --git a/src/Orleans.ScheduledJobs/Orleans.ScheduledJobs.csproj b/src/Orleans.DurableJobs/Orleans.DurableJobs.csproj similarity index 82% rename from src/Orleans.ScheduledJobs/Orleans.ScheduledJobs.csproj rename to src/Orleans.DurableJobs/Orleans.DurableJobs.csproj index 6b86863d93f..fa1505157f3 100644 --- a/src/Orleans.ScheduledJobs/Orleans.ScheduledJobs.csproj +++ b/src/Orleans.DurableJobs/Orleans.DurableJobs.csproj @@ -1,8 +1,8 @@ - Microsoft.Orleans.ScheduledJobs - Microsoft Orleans Scheduled Jobs Library - Scheduled Jobs library for Microsoft Orleans used on the server. + Microsoft.Orleans.DurableJobs + Microsoft Orleans Durable Jobs Library + Durable Jobs library for Microsoft Orleans used on the server. README.md $(DefaultTargetFrameworks) true diff --git a/src/Orleans.ScheduledJobs/Properties/AssemblyInfo.cs b/src/Orleans.DurableJobs/Properties/AssemblyInfo.cs similarity index 100% rename from src/Orleans.ScheduledJobs/Properties/AssemblyInfo.cs rename to src/Orleans.DurableJobs/Properties/AssemblyInfo.cs diff --git a/src/Orleans.ScheduledJobs/README.md b/src/Orleans.DurableJobs/README.md similarity index 81% rename from src/Orleans.ScheduledJobs/README.md rename to src/Orleans.DurableJobs/README.md index 927754127ce..ace184938c9 100644 --- a/src/Orleans.ScheduledJobs/README.md +++ b/src/Orleans.DurableJobs/README.md @@ -1,7 +1,7 @@ -# Microsoft Orleans Scheduled Jobs +# Microsoft Orleans Durable Jobs ## Introduction -Microsoft Orleans Scheduled Jobs provides a distributed, scalable system for scheduling one-time jobs that execute at a specific time. Unlike Orleans Reminders which are designed for recurring tasks, Scheduled Jobs are ideal for one-time future events such as appointment notifications, delayed processing, scheduled workflow steps, and time-based triggers. +Microsoft Orleans Durable Jobs provides a distributed, scalable system for scheduling one-time jobs that execute at a specific time. Unlike Orleans Reminders which are designed for recurring tasks, Durable Jobs are ideal for one-time future events such as appointment notifications, delayed processing, scheduled workflow steps, and time-based triggers. **Key Features:** - **At Least One-time Execution**: Jobs are scheduled to run at least once @@ -17,13 +17,13 @@ Microsoft Orleans Scheduled Jobs provides a distributed, scalable system for sch To use this package, install it via NuGet: ```shell -dotnet add package Microsoft.Orleans.ScheduledJobs +dotnet add package Microsoft.Orleans.DurableJobs ``` For production scenarios with persistence, also install a storage provider: ```shell -dotnet add package Microsoft.Orleans.ScheduledJobs.AzureStorage +dotnet add package Microsoft.Orleans.DurableJobs.AzureStorage ``` ### Configuration @@ -39,8 +39,8 @@ builder.UseOrleans(siloBuilder => { siloBuilder .UseLocalhostClustering() - // Configure in-memory scheduled jobs (no persistence) - .UseInMemoryScheduledJobs(); + // Configure in-memory Durable Jobs (no persistence) + .UseInMemoryDurableJobs(); }); await builder.Build().RunAsync(); @@ -57,13 +57,13 @@ builder.UseOrleans(siloBuilder => { siloBuilder .UseLocalhostClustering() - // Configure Azure Storage scheduled jobs - .UseAzureStorageScheduledJobs(options => + // Configure Azure Storage Durable Jobs + .UseAzureStorageDurableJobs(options => { options.Configure(o => { o.BlobServiceClient = new Azure.Storage.Blobs.BlobServiceClient("YOUR_CONNECTION_STRING"); - o.ContainerName = "scheduled-jobs"; + o.ContainerName = "durable-jobs"; }); }); }); @@ -77,10 +77,10 @@ builder.UseOrleans(siloBuilder => { siloBuilder .UseLocalhostClustering() - .UseInMemoryScheduledJobs() + .UseInMemoryDurableJobs() .ConfigureServices(services => { - services.Configure(options => + services.Configure(options => { // Duration of each job shard (jobs are partitioned by time) options.ShardDuration = TimeSpan.FromMinutes(5); @@ -108,10 +108,10 @@ builder.UseOrleans(siloBuilder => ### Basic Job Scheduling -#### 1. Implement the IScheduledJobHandler Interface +#### 1. Implement the IDurableJobHandler Interface ```csharp using Orleans; -using Orleans.ScheduledJobs; +using Orleans.DurableJobs; public interface INotificationGrain : IGrainWithStringKey { @@ -119,14 +119,14 @@ public interface INotificationGrain : IGrainWithStringKey Task CancelScheduledNotification(); } -public class NotificationGrain : Grain, INotificationGrain, IScheduledJobHandler +public class NotificationGrain : Grain, INotificationGrain, IDurableJobHandler { - private readonly ILocalScheduledJobManager _jobManager; + private readonly ILocalDurableJobManager _jobManager; private readonly ILogger _logger; - private IScheduledJob? _scheduledJob; + private IDurableJob? _durableJob; public NotificationGrain( - ILocalScheduledJobManager jobManager, + ILocalDurableJobManager jobManager, ILogger logger) { _jobManager = jobManager; @@ -141,7 +141,7 @@ public class NotificationGrain : Grain, INotificationGrain, IScheduledJobHandler ["Message"] = message }; - _scheduledJob = await _jobManager.ScheduleJobAsync( + _durableJob = await _jobManager.ScheduleJobAsync( this.GetGrainId(), "SendNotification", sendTime, @@ -149,28 +149,28 @@ public class NotificationGrain : Grain, INotificationGrain, IScheduledJobHandler _logger.LogInformation( "Scheduled notification for user {UserId} at {SendTime} (JobId: {JobId})", - userId, sendTime, _scheduledJob.Id); + userId, sendTime, _durableJob.Id); } public async Task CancelScheduledNotification() { - if (_scheduledJob is null) + if (_durableJob is null) { _logger.LogWarning("No scheduled notification to cancel"); return; } - var canceled = await _jobManager.TryCancelScheduledJobAsync(_scheduledJob); - _logger.LogInformation("Notification {JobId} canceled: {Canceled}", _scheduledJob.Id, canceled); + var canceled = await _jobManager.TryCancelDurableJobAsync(_durableJob); + _logger.LogInformation("Notification {JobId} canceled: {Canceled}", _durableJob.Id, canceled); if (canceled) { - _scheduledJob = null; + _durableJob = null; } } - // This method is called when the scheduled job executes - public Task ExecuteJobAsync(IScheduledJobContext context, CancellationToken cancellationToken) + // This method is called when the durable job executes + public Task ExecuteJobAsync(IDurableJobContext context, CancellationToken cancellationToken) { var userId = this.GetPrimaryKeyString(); var message = context.Job.Metadata?["Message"]; @@ -182,7 +182,7 @@ public class NotificationGrain : Grain, INotificationGrain, IScheduledJobHandler // Send the notification here // If this throws an exception, the job can be retried based on your retry policy - _scheduledJob = null; + _durableJob = null; return Task.CompletedTask; } } @@ -196,15 +196,15 @@ public interface IOrderGrain : IGrainWithGuidKey Task CancelOrder(); } -public class OrderGrain : Grain, IOrderGrain, IScheduledJobHandler +public class OrderGrain : Grain, IOrderGrain, IDurableJobHandler { - private readonly ILocalScheduledJobManager _jobManager; + private readonly ILocalDurableJobManager _jobManager; private readonly IOrderService _orderService; private readonly IGrainFactory _grainFactory; private readonly ILogger _logger; public OrderGrain( - ILocalScheduledJobManager jobManager, + ILocalDurableJobManager jobManager, IOrderService orderService, IGrainFactory grainFactory, ILogger logger) @@ -253,7 +253,7 @@ public class OrderGrain : Grain, IOrderGrain, IScheduledJobHandler await _orderService.CancelOrderAsync(orderId); } - public async Task ExecuteJobAsync(IScheduledJobContext context, CancellationToken cancellationToken) + public async Task ExecuteJobAsync(IDurableJobContext context, CancellationToken cancellationToken) { var step = context.Job.Metadata!["Step"]; var orderId = this.GetPrimaryKey(); @@ -270,7 +270,7 @@ public class OrderGrain : Grain, IOrderGrain, IScheduledJobHandler } } - private async Task HandleDeliveryReminder(IScheduledJobContext context, CancellationToken ct) + private async Task HandleDeliveryReminder(IDurableJobContext context, CancellationToken ct) { var customerId = context.Job.Metadata!["CustomerId"]; var orderNumber = context.Job.Metadata["OrderNumber"]; @@ -299,12 +299,12 @@ public class OrderGrain : Grain, IOrderGrain, IScheduledJobHandler #### Job with Retry Logic ```csharp -public class PaymentProcessorGrain : Grain, IScheduledJobHandler +public class PaymentProcessorGrain : Grain, IDurableJobHandler { private readonly IPaymentService _paymentService; private readonly ILogger _logger; - public Task ExecuteJobAsync(IScheduledJobContext context, CancellationToken cancellationToken) + public Task ExecuteJobAsync(IDurableJobContext context, CancellationToken cancellationToken) { var paymentId = context.Job.Metadata?["PaymentId"]; @@ -333,11 +333,11 @@ public class PaymentProcessorGrain : Grain, IScheduledJobHandler #### Tracking Job Completion ```csharp -public class WorkflowGrain : Grain, IScheduledJobHandler +public class WorkflowGrain : Grain, IDurableJobHandler { private readonly Dictionary _pendingJobs = new(); - public async Task ScheduleWorkflowStep(string stepName, DateTimeOffset executeAt) + public async Task ScheduleWorkflowStep(string stepName, DateTimeOffset executeAt) { var job = await _jobManager.ScheduleJobAsync( this.GetGrainId(), @@ -357,7 +357,7 @@ public class WorkflowGrain : Grain, IScheduledJobHandler } } - public Task ExecuteJobAsync(IScheduledJobContext context, CancellationToken cancellationToken) + public Task ExecuteJobAsync(IDurableJobContext context, CancellationToken cancellationToken) { // Execute the workflow step... @@ -406,13 +406,13 @@ public class WorkflowGrain : Grain, IScheduledJobHandler ## Configuration Reference -### ScheduledJobsOptions +### DurableJobsOptions | Property | Type | Default | Description | |----------|------|---------|-------------| | `ShardDuration` | `TimeSpan` | 1 minute | Duration of each job shard. Smaller values reduce latency but increase overhead. | | `MaxConcurrentJobsPerSilo` | `int` | 100 | Maximum number of jobs that can execute simultaneously on a silo. | -| `ShouldRetry` | `Func` | 3 retries with exp. backoff | Determines if a failed job should be retried. Return the new due time or `null` to not retry. | +| `ShouldRetry` | `Func` | 3 retries with exp. backoff | Determines if a failed job should be retried. Return the new due time or `null` to not retry. | ## Best Practices @@ -423,7 +423,7 @@ public class WorkflowGrain : Grain, IScheduledJobHandler 2. **Implement Idempotent Job Handlers**: Jobs may be retried, ensure handlers are idempotent ```csharp - public async Task ExecuteJobAsync(IScheduledJobContext context, CancellationToken ct) + public async Task ExecuteJobAsync(IDurableJobContext context, CancellationToken ct) { var jobId = context.Job.Id; // Check if already processed @@ -446,7 +446,7 @@ public class WorkflowGrain : Grain, IScheduledJobHandler 4. **Handle Cancellation**: Respect the cancellation token ```csharp - public async Task ExecuteJobAsync(IScheduledJobContext context, CancellationToken ct) + public async Task ExecuteJobAsync(IDurableJobContext context, CancellationToken ct) { await SomeLongRunningOperation(ct); } diff --git a/src/Orleans.ScheduledJobs/ShardExecutor.Log.cs b/src/Orleans.DurableJobs/ShardExecutor.Log.cs similarity index 98% rename from src/Orleans.ScheduledJobs/ShardExecutor.Log.cs rename to src/Orleans.DurableJobs/ShardExecutor.Log.cs index 6ef04c98393..f038dea1f66 100644 --- a/src/Orleans.ScheduledJobs/ShardExecutor.Log.cs +++ b/src/Orleans.DurableJobs/ShardExecutor.Log.cs @@ -2,7 +2,7 @@ using Microsoft.Extensions.Logging; using Orleans.Runtime; -namespace Orleans.ScheduledJobs; +namespace Orleans.DurableJobs; internal sealed partial class ShardExecutor { diff --git a/src/Orleans.ScheduledJobs/ShardExecutor.cs b/src/Orleans.DurableJobs/ShardExecutor.cs similarity index 88% rename from src/Orleans.ScheduledJobs/ShardExecutor.cs rename to src/Orleans.DurableJobs/ShardExecutor.cs index 5d9bcb6b67f..25187657caf 100644 --- a/src/Orleans.ScheduledJobs/ShardExecutor.cs +++ b/src/Orleans.DurableJobs/ShardExecutor.cs @@ -8,27 +8,27 @@ using Orleans.Hosting; using Orleans.Runtime; -namespace Orleans.ScheduledJobs; +namespace Orleans.DurableJobs; /// -/// Handles the execution of job shards and individual scheduled jobs. +/// Handles the execution of job shards and individual durable jobs. /// internal sealed partial class ShardExecutor { private readonly IInternalGrainFactory _grainFactory; private readonly ILogger _logger; - private readonly ScheduledJobsOptions _options; + private readonly DurableJobsOptions _options; private readonly SemaphoreSlim _jobConcurrencyLimiter; /// /// Initializes a new instance of the class. /// /// The grain factory for creating grain references. - /// The scheduled jobs configuration options. + /// The durable jobs configuration options. /// The logger instance. public ShardExecutor( IInternalGrainFactory grainFactory, - IOptions options, + IOptions options, ILogger logger) { _grainFactory = grainFactory; @@ -61,7 +61,7 @@ public async Task RunShardAsync(IJobShard shard, CancellationToken cancellationT LogBeginProcessingShard(_logger, shard.Id); // Process all jobs in the shard - await foreach (var jobContext in shard.ConsumeScheduledJobsAsync().WithCancellation(cancellationToken)) + await foreach (var jobContext in shard.ConsumeDurableJobsAsync().WithCancellation(cancellationToken)) { // Wait for concurrency slot await _jobConcurrencyLimiter.WaitAsync(cancellationToken); @@ -84,7 +84,7 @@ public async Task RunShardAsync(IJobShard shard, CancellationToken cancellationT } private async Task RunJobAsync( - IScheduledJobContext jobContext, + IDurableJobContext jobContext, IJobShard shard, ConcurrentDictionary runningTasks, CancellationToken cancellationToken) @@ -97,9 +97,9 @@ private async Task RunJobAsync( var target = _grainFactory .GetGrain(jobContext.Job.TargetGrainId) - .AsReference(); + .AsReference(); - await target.DeliverScheduledJobAsync(jobContext, cancellationToken); + await target.DeliverDurableJobAsync(jobContext, cancellationToken); await shard.RemoveJobAsync(jobContext.Job.Id, cancellationToken); LogJobExecutedSuccessfully(_logger, jobContext.Job.Id, jobContext.Job.Name); diff --git a/src/Orleans.Runtime/Properties/AssemblyInfo.cs b/src/Orleans.Runtime/Properties/AssemblyInfo.cs index 73a98bd2f18..0e27d969e74 100644 --- a/src/Orleans.Runtime/Properties/AssemblyInfo.cs +++ b/src/Orleans.Runtime/Properties/AssemblyInfo.cs @@ -2,7 +2,7 @@ [assembly: InternalsVisibleTo("Orleans.Streaming")] [assembly: InternalsVisibleTo("Orleans.Reminders")] -[assembly: InternalsVisibleTo("Orleans.ScheduledJobs")] +[assembly: InternalsVisibleTo("Orleans.DurableJobs")] [assembly: InternalsVisibleTo("Orleans.Journaling")] [assembly: InternalsVisibleTo("Orleans.TestingHost")] diff --git a/src/Orleans.ScheduledJobs/IScheduledJobReceiverExtension.cs b/src/Orleans.ScheduledJobs/IScheduledJobReceiverExtension.cs deleted file mode 100644 index fa217c34d0c..00000000000 --- a/src/Orleans.ScheduledJobs/IScheduledJobReceiverExtension.cs +++ /dev/null @@ -1,61 +0,0 @@ -using System; -using System.Threading; -using System.Threading.Tasks; -using Microsoft.Extensions.Logging; -using Orleans.Runtime; - -namespace Orleans.ScheduledJobs; - -/// -/// Extension interface for grains that can receive scheduled job invocations. -/// -internal interface IScheduledJobReceiverExtension : IGrainExtension -{ - /// - /// Delivers a scheduled job to the grain for execution. - /// - /// The context containing information about the scheduled job. - /// A token to monitor for cancellation requests. - /// A task that represents the asynchronous operation. - Task DeliverScheduledJobAsync(IScheduledJobContext context, CancellationToken cancellationToken); -} - -/// -internal sealed partial class ScheduledJobReceiverExtension : IScheduledJobReceiverExtension -{ - private readonly IGrainContext _grain; - private readonly ILogger _logger; - - public ScheduledJobReceiverExtension(IGrainContext grain, ILogger logger) - { - _grain = grain; - _logger = logger; - } - - public async Task DeliverScheduledJobAsync(IScheduledJobContext context, CancellationToken cancellationToken) - { - if (_grain.GrainInstance is IScheduledJobHandler handler) - { - try - { - await handler.ExecuteJobAsync(context, cancellationToken); - } - catch (Exception ex) - { - LogErrorExecutingScheduledJob(ex, context.Job.Id, _grain.GrainId); - throw; - } - } - else - { - LogGrainDoesNotImplementHandler(_grain.GrainId); - throw new InvalidOperationException($"Grain {_grain.GrainId} does not implement IScheduledJobHandler"); - } - } - - [LoggerMessage(Level = LogLevel.Error, Message = "Error executing scheduled job {JobId} on grain {GrainId}")] - private partial void LogErrorExecutingScheduledJob(Exception exception, string jobId, GrainId grainId); - - [LoggerMessage(Level = LogLevel.Error, Message = "Grain {GrainId} does not implement IScheduledJobHandler")] - private partial void LogGrainDoesNotImplementHandler(GrainId grainId); -} diff --git a/src/Orleans.TestingHost/Orleans.TestingHost.csproj b/src/Orleans.TestingHost/Orleans.TestingHost.csproj index 56fff77461d..dd2e935309d 100644 --- a/src/Orleans.TestingHost/Orleans.TestingHost.csproj +++ b/src/Orleans.TestingHost/Orleans.TestingHost.csproj @@ -12,7 +12,7 @@ - + diff --git a/test/DefaultCluster.Tests/InMemoryDurableJobsTests.cs b/test/DefaultCluster.Tests/InMemoryDurableJobsTests.cs new file mode 100644 index 00000000000..d4c036f414c --- /dev/null +++ b/test/DefaultCluster.Tests/InMemoryDurableJobsTests.cs @@ -0,0 +1,68 @@ +using System.Threading.Tasks; +using Tester.DurableJobs; +using TestExtensions; +using Xunit; + +namespace DefaultCluster.Tests; + +public class InMemoryDurableJobsTests : HostedTestClusterEnsureDefaultStarted +{ + private readonly DurableJobTestsRunner _runner; + + public InMemoryDurableJobsTests(DefaultClusterFixture fixture) : base(fixture) + { + _runner = new DurableJobTestsRunner(this.GrainFactory); + } + + [Fact, TestCategory("BVT"), TestCategory("DurableJobs")] + public Task DurableJobGrain() + => _runner.DurableJobGrain(); + + [Fact, TestCategory("BVT"), TestCategory("DurableJobs")] + public Task JobExecutionOrder() + => _runner.JobExecutionOrder(); + + [Fact, TestCategory("BVT"), TestCategory("DurableJobs")] + public Task PastDueTime() + => _runner.PastDueTime(); + + [Fact, TestCategory("BVT"), TestCategory("DurableJobs")] + public Task JobWithMetadata() + => _runner.JobWithMetadata(); + + [Fact, TestCategory("BVT"), TestCategory("DurableJobs")] + public Task MultipleGrains() + => _runner.MultipleGrains(); + + [Fact, TestCategory("BVT"), TestCategory("DurableJobs")] + public Task DuplicateJobNames() + => _runner.DuplicateJobNames(); + + [Fact, TestCategory("BVT"), TestCategory("DurableJobs")] + public Task CancelNonExistentJob() + => _runner.CancelNonExistentJob(); + + [Fact, TestCategory("BVT"), TestCategory("DurableJobs")] + public Task CancelAlreadyExecutedJob() + => _runner.CancelAlreadyExecutedJob(); + + [Fact, TestCategory("BVT"), TestCategory("DurableJobs")] + public Task ConcurrentScheduling() + => _runner.ConcurrentScheduling(); + + [Fact, TestCategory("BVT"), TestCategory("DurableJobs")] + public Task JobPropertiesVerification() + => _runner.JobPropertiesVerification(); + + [Fact, TestCategory("BVT"), TestCategory("DurableJobs")] + public Task DequeueCount() + => _runner.DequeueCount(); + + [Fact, TestCategory("BVT"), TestCategory("DurableJobs")] + public Task ScheduleJobOnAnotherGrain() + => _runner.ScheduleJobOnAnotherGrain(); + + [Fact, TestCategory("BVT"), TestCategory("DurableJobs")] + public Task JobRetry() + => _runner.JobRetry(); +} diff --git a/test/DefaultCluster.Tests/InMemoryScheduledJobTests.cs b/test/DefaultCluster.Tests/InMemoryScheduledJobTests.cs deleted file mode 100644 index a11022460f3..00000000000 --- a/test/DefaultCluster.Tests/InMemoryScheduledJobTests.cs +++ /dev/null @@ -1,68 +0,0 @@ -using System.Threading.Tasks; -using Tester.ScheduledJobs; -using TestExtensions; -using Xunit; - -namespace DefaultCluster.Tests; - -public class InMemoryScheduledJobsTests : HostedTestClusterEnsureDefaultStarted -{ - private readonly ScheduledJobTestsRunner _runner; - - public InMemoryScheduledJobsTests(DefaultClusterFixture fixture) : base(fixture) - { - _runner = new ScheduledJobTestsRunner(this.GrainFactory); - } - - [Fact, TestCategory("BVT"), TestCategory("ScheduledJobs")] - public Task ScheduledJobGrain() - => _runner.ScheduledJobGrain(); - - [Fact, TestCategory("BVT"), TestCategory("ScheduledJobs")] - public Task JobExecutionOrder() - => _runner.JobExecutionOrder(); - - [Fact, TestCategory("BVT"), TestCategory("ScheduledJobs")] - public Task PastDueTime() - => _runner.PastDueTime(); - - [Fact, TestCategory("BVT"), TestCategory("ScheduledJobs")] - public Task JobWithMetadata() - => _runner.JobWithMetadata(); - - [Fact, TestCategory("BVT"), TestCategory("ScheduledJobs")] - public Task MultipleGrains() - => _runner.MultipleGrains(); - - [Fact, TestCategory("BVT"), TestCategory("ScheduledJobs")] - public Task DuplicateJobNames() - => _runner.DuplicateJobNames(); - - [Fact, TestCategory("BVT"), TestCategory("ScheduledJobs")] - public Task CancelNonExistentJob() - => _runner.CancelNonExistentJob(); - - [Fact, TestCategory("BVT"), TestCategory("ScheduledJobs")] - public Task CancelAlreadyExecutedJob() - => _runner.CancelAlreadyExecutedJob(); - - [Fact, TestCategory("BVT"), TestCategory("ScheduledJobs")] - public Task ConcurrentScheduling() - => _runner.ConcurrentScheduling(); - - [Fact, TestCategory("BVT"), TestCategory("ScheduledJobs")] - public Task JobPropertiesVerification() - => _runner.JobPropertiesVerification(); - - [Fact, TestCategory("BVT"), TestCategory("ScheduledJobs")] - public Task DequeueCount() - => _runner.DequeueCount(); - - [Fact, TestCategory("BVT"), TestCategory("ScheduledJobs")] - public Task ScheduleJobOnAnotherGrain() - => _runner.ScheduleJobOnAnotherGrain(); - - [Fact, TestCategory("BVT"), TestCategory("ScheduledJobs")] - public Task JobRetry() - => _runner.JobRetry(); -} diff --git a/test/Extensions/TesterAzureUtils/ScheduledJobs/AzureStorageBlobScheduledJobsTests.cs b/test/Extensions/TesterAzureUtils/DurableJobs/AzureStorageBlobDurableJobsTests.cs similarity index 55% rename from test/Extensions/TesterAzureUtils/ScheduledJobs/AzureStorageBlobScheduledJobsTests.cs rename to test/Extensions/TesterAzureUtils/DurableJobs/AzureStorageBlobDurableJobsTests.cs index 129de88638a..e371ac1901f 100644 --- a/test/Extensions/TesterAzureUtils/ScheduledJobs/AzureStorageBlobScheduledJobsTests.cs +++ b/test/Extensions/TesterAzureUtils/DurableJobs/AzureStorageBlobDurableJobsTests.cs @@ -4,22 +4,22 @@ using Orleans.Configuration; using Orleans.TestingHost; using Tester; -using Tester.ScheduledJobs; +using Tester.DurableJobs; using TestExtensions; using Xunit; -namespace Tester.AzureUtils.ScheduledJobs; +namespace Tester.AzureUtils.DurableJobs; -public class AzureStorageBlobScheduledJobsTests : TestClusterPerTest +public class AzureStorageBlobDurableJobsTests : TestClusterPerTest { - private ScheduledJobTestsRunner _runner; + private DurableJobTestsRunner _runner; protected override void CheckPreconditionsOrThrow() => TestUtils.CheckForAzureStorage(); public override async Task InitializeAsync() { await base.InitializeAsync(); - _runner = new ScheduledJobTestsRunner(this.GrainFactory); + _runner = new DurableJobTestsRunner(this.GrainFactory); } protected override void ConfigureTestCluster(TestClusterBuilder builder) @@ -32,60 +32,60 @@ public class SiloHostConfigurator : ISiloConfigurator public void Configure(ISiloBuilder hostBuilder) { hostBuilder - .UseAzureBlobScheduledJobs(options => options.ConfigureTestDefaults()) + .UseAzureBlobDurableJobs(options => options.ConfigureTestDefaults()) .AddMemoryGrainStorageAsDefault(); } } - [SkippableFact, TestCategory("Azure"), TestCategory("ScheduledJobs")] - public Task ScheduledJobGrain() - => _runner.ScheduledJobGrain(); + [SkippableFact, TestCategory("Azure"), TestCategory("DurableJobs")] + public Task DurableJobGrain() + => _runner.DurableJobGrain(); - [SkippableFact, TestCategory("Azure"), TestCategory("ScheduledJobs")] + [SkippableFact, TestCategory("Azure"), TestCategory("DurableJobs")] public Task JobExecutionOrder() => _runner.JobExecutionOrder(); - [SkippableFact, TestCategory("Azure"), TestCategory("ScheduledJobs")] + [SkippableFact, TestCategory("Azure"), TestCategory("DurableJobs")] public Task PastDueTime() => _runner.PastDueTime(); - [SkippableFact, TestCategory("Azure"), TestCategory("ScheduledJobs")] + [SkippableFact, TestCategory("Azure"), TestCategory("DurableJobs")] public Task JobWithMetadata() => _runner.JobWithMetadata(); - [SkippableFact, TestCategory("Azure"), TestCategory("ScheduledJobs")] + [SkippableFact, TestCategory("Azure"), TestCategory("DurableJobs")] public Task MultipleGrains() => _runner.MultipleGrains(); - [SkippableFact, TestCategory("Azure"), TestCategory("ScheduledJobs")] + [SkippableFact, TestCategory("Azure"), TestCategory("DurableJobs")] public Task DuplicateJobNames() => _runner.DuplicateJobNames(); - [SkippableFact, TestCategory("Azure"), TestCategory("ScheduledJobs")] + [SkippableFact, TestCategory("Azure"), TestCategory("DurableJobs")] public Task CancelNonExistentJob() => _runner.CancelNonExistentJob(); - [SkippableFact, TestCategory("Azure"), TestCategory("ScheduledJobs")] + [SkippableFact, TestCategory("Azure"), TestCategory("DurableJobs")] public Task CancelAlreadyExecutedJob() => _runner.CancelAlreadyExecutedJob(); - [SkippableFact, TestCategory("Azure"), TestCategory("ScheduledJobs")] + [SkippableFact, TestCategory("Azure"), TestCategory("DurableJobs")] public Task ConcurrentScheduling() => _runner.ConcurrentScheduling(); - [SkippableFact, TestCategory("Azure"), TestCategory("ScheduledJobs")] + [SkippableFact, TestCategory("Azure"), TestCategory("DurableJobs")] public Task JobPropertiesVerification() => _runner.JobPropertiesVerification(); - [SkippableFact, TestCategory("Azure"), TestCategory("ScheduledJobs")] + [SkippableFact, TestCategory("Azure"), TestCategory("DurableJobs")] public Task DequeueCount() => _runner.DequeueCount(); - [SkippableFact, TestCategory("Azure"), TestCategory("ScheduledJobs")] + [SkippableFact, TestCategory("Azure"), TestCategory("DurableJobs")] public Task ScheduleJobOnAnotherGrain() => _runner.ScheduleJobOnAnotherGrain(); - [SkippableFact, TestCategory("Azure"), TestCategory("ScheduledJobs")] + [SkippableFact, TestCategory("Azure"), TestCategory("DurableJobs")] public Task JobRetry() => _runner.JobRetry(); } diff --git a/test/Extensions/TesterAzureUtils/ScheduledJobs/AzureStorageJobShardBatchingTests.cs b/test/Extensions/TesterAzureUtils/DurableJobs/AzureStorageJobShardBatchingTests.cs similarity index 96% rename from test/Extensions/TesterAzureUtils/ScheduledJobs/AzureStorageJobShardBatchingTests.cs rename to test/Extensions/TesterAzureUtils/DurableJobs/AzureStorageJobShardBatchingTests.cs index 9590b04673a..b3e2846efe7 100644 --- a/test/Extensions/TesterAzureUtils/ScheduledJobs/AzureStorageJobShardBatchingTests.cs +++ b/test/Extensions/TesterAzureUtils/DurableJobs/AzureStorageJobShardBatchingTests.cs @@ -9,18 +9,18 @@ using Microsoft.Extensions.Options; using Orleans.Hosting; using Orleans.Runtime; -using Orleans.ScheduledJobs; -using Orleans.ScheduledJobs.AzureStorage; +using Orleans.DurableJobs; +using Orleans.DurableJobs.AzureStorage; using Tester.AzureUtils; using Xunit; -namespace Tester.AzureUtils.ScheduledJobs; +namespace Tester.AzureUtils.DurableJobs; /// /// Azure Storage-specific tests for job shard batching functionality. /// These tests verify Azure-specific batching behaviors that don't apply to all providers. /// -[TestCategory("ScheduledJobs")] +[TestCategory("DurableJobs")] public class AzureStorageJobShardBatchingTests : AzureStorageBasicTests, IAsyncDisposable { private readonly IDictionary _metadata = new Dictionary @@ -120,7 +120,7 @@ public async Task AzureStorageJobShard_MultipleOperationsBatched() var consumedJobs = new List(); var cts = new CancellationTokenSource(TimeSpan.FromSeconds(20)); - await foreach (var jobCtx in shards[0].ConsumeScheduledJobsAsync().WithCancellation(cts.Token)) + await foreach (var jobCtx in shards[0].ConsumeDurableJobsAsync().WithCancellation(cts.Token)) { consumedJobs.Add(jobCtx.Job.Name); await shards[0].RemoveJobAsync(jobCtx.Job.Id, CancellationToken.None); @@ -168,7 +168,7 @@ public async Task AzureStorageJobShard_PartialBatchFlushesOnTimeout() var consumedJobs = new List(); var cts = new CancellationTokenSource(TimeSpan.FromSeconds(20)); - await foreach (var jobCtx in shards[0].ConsumeScheduledJobsAsync().WithCancellation(cts.Token)) + await foreach (var jobCtx in shards[0].ConsumeDurableJobsAsync().WithCancellation(cts.Token)) { consumedJobs.Add(jobCtx.Job.Name); await shards[0].RemoveJobAsync(jobCtx.Job.Id, CancellationToken.None); @@ -221,7 +221,7 @@ public async Task AzureStorageJobShard_MaxBatchSizeEnforced() var consumedJobs = new List(); var cts = new CancellationTokenSource(TimeSpan.FromSeconds(30)); - await foreach (var jobCtx in shards[0].ConsumeScheduledJobsAsync().WithCancellation(cts.Token)) + await foreach (var jobCtx in shards[0].ConsumeDurableJobsAsync().WithCancellation(cts.Token)) { consumedJobs.Add(jobCtx.Job.Name); await shards[0].RemoveJobAsync(jobCtx.Job.Id, CancellationToken.None); @@ -288,7 +288,7 @@ public async Task AzureStorageJobShard_MetadataOperationsBreakBatches() var consumedJobs = new List(); var cts = new CancellationTokenSource(TimeSpan.FromSeconds(20)); - await foreach (var jobCtx in shards[0].ConsumeScheduledJobsAsync().WithCancellation(cts.Token)) + await foreach (var jobCtx in shards[0].ConsumeDurableJobsAsync().WithCancellation(cts.Token)) { consumedJobs.Add(jobCtx.Job.Name); await shards[0].RemoveJobAsync(jobCtx.Job.Id, CancellationToken.None); diff --git a/test/Extensions/TesterAzureUtils/ScheduledJobs/AzureStorageJobShardManagerTestFixture.cs b/test/Extensions/TesterAzureUtils/DurableJobs/AzureStorageJobShardManagerTestFixture.cs similarity index 90% rename from test/Extensions/TesterAzureUtils/ScheduledJobs/AzureStorageJobShardManagerTestFixture.cs rename to test/Extensions/TesterAzureUtils/DurableJobs/AzureStorageJobShardManagerTestFixture.cs index 3126bdfe0e8..c73aee7238e 100644 --- a/test/Extensions/TesterAzureUtils/ScheduledJobs/AzureStorageJobShardManagerTestFixture.cs +++ b/test/Extensions/TesterAzureUtils/DurableJobs/AzureStorageJobShardManagerTestFixture.cs @@ -5,12 +5,12 @@ using Microsoft.Extensions.Options; using Orleans.Hosting; using Orleans.Runtime; -using Orleans.ScheduledJobs; -using Orleans.ScheduledJobs.AzureStorage; +using Orleans.DurableJobs; +using Orleans.DurableJobs.AzureStorage; using Tester.AzureUtils; -using Tester.ScheduledJobs; +using Tester.DurableJobs; -namespace Orleans.Tests.ScheduledJobs.AzureStorage; +namespace Orleans.Tests.DurableJobs.AzureStorage; /// /// Azure Storage implementation of . diff --git a/test/Extensions/TesterAzureUtils/ScheduledJobs/AzureStorageJobShardManagerTests.cs b/test/Extensions/TesterAzureUtils/DurableJobs/AzureStorageJobShardManagerTests.cs similarity index 96% rename from test/Extensions/TesterAzureUtils/ScheduledJobs/AzureStorageJobShardManagerTests.cs rename to test/Extensions/TesterAzureUtils/DurableJobs/AzureStorageJobShardManagerTests.cs index 3ba5b944414..7dd17df53a7 100644 --- a/test/Extensions/TesterAzureUtils/ScheduledJobs/AzureStorageJobShardManagerTests.cs +++ b/test/Extensions/TesterAzureUtils/DurableJobs/AzureStorageJobShardManagerTests.cs @@ -11,21 +11,21 @@ using Microsoft.Extensions.Logging.Abstractions; using Microsoft.Extensions.Options; using Orleans.Internal; -using Orleans.ScheduledJobs; -using Orleans.ScheduledJobs.AzureStorage; -using Orleans.Tests.ScheduledJobs.AzureStorage; -using Tester.ScheduledJobs; +using Orleans.DurableJobs; +using Orleans.DurableJobs.AzureStorage; +using Orleans.Tests.DurableJobs.AzureStorage; +using Tester.DurableJobs; using Xunit; using Xunit.Sdk; -namespace Tester.AzureUtils.ScheduledJobs; +namespace Tester.AzureUtils.DurableJobs; /// /// Azure Storage-specific tests for job shard manager functionality. /// Common tests are delegated to for reusability across providers. /// Provider-specific tests (e.g., batching) remain here. /// -[TestCategory("ScheduledJobs")] +[TestCategory("DurableJobs")] public class AzureStorageJobShardManagerTests : AzureStorageBasicTests, IAsyncDisposable { private readonly AzureStorageJobShardManagerTestFixture _fixture; @@ -146,4 +146,4 @@ public Task AzureStorageJobShardManager_ShardRegistrationRetry_IdCollisions() [SkippableFact, TestCategory("Azure"), TestCategory("Functional")] public Task AzureStorageJobShardManager_UnregisterShard_WithJobsRemaining() => _runner.UnregisterShard_WithJobsRemaining(); -} \ No newline at end of file +} diff --git a/test/Extensions/TesterAzureUtils/ScheduledJobs/NetstringJsonSerializerTests.cs b/test/Extensions/TesterAzureUtils/DurableJobs/NetstringJsonSerializerTests.cs similarity index 99% rename from test/Extensions/TesterAzureUtils/ScheduledJobs/NetstringJsonSerializerTests.cs rename to test/Extensions/TesterAzureUtils/DurableJobs/NetstringJsonSerializerTests.cs index 349028428a1..44ec9df426f 100644 --- a/test/Extensions/TesterAzureUtils/ScheduledJobs/NetstringJsonSerializerTests.cs +++ b/test/Extensions/TesterAzureUtils/DurableJobs/NetstringJsonSerializerTests.cs @@ -8,12 +8,12 @@ using System.Threading.Tasks; using FluentAssertions; using Orleans.Runtime; -using Orleans.ScheduledJobs.AzureStorage; +using Orleans.DurableJobs.AzureStorage; using Xunit; -namespace Tester.AzureUtils.ScheduledJobs; +namespace Tester.AzureUtils.DurableJobs; -[TestCategory("ScheduledJobs"), TestCategory("BVT")] +[TestCategory("DurableJobs"), TestCategory("BVT")] public class NetstringJsonSerializerTests { private static byte[] EncodeToBytes(JobOperation operation) diff --git a/test/Extensions/TesterAzureUtils/Tester.AzureUtils.csproj b/test/Extensions/TesterAzureUtils/Tester.AzureUtils.csproj index 8246ee7899d..b71f1ab0918 100644 --- a/test/Extensions/TesterAzureUtils/Tester.AzureUtils.csproj +++ b/test/Extensions/TesterAzureUtils/Tester.AzureUtils.csproj @@ -20,7 +20,7 @@ - + diff --git a/test/Grains/TestGrainInterfaces/IScheduledJobGrain.cs b/test/Grains/TestGrainInterfaces/IDurableJobGrain.cs similarity index 55% rename from test/Grains/TestGrainInterfaces/IScheduledJobGrain.cs rename to test/Grains/TestGrainInterfaces/IDurableJobGrain.cs index 35afe873bbf..0dabddd3a97 100644 --- a/test/Grains/TestGrainInterfaces/IScheduledJobGrain.cs +++ b/test/Grains/TestGrainInterfaces/IDurableJobGrain.cs @@ -4,15 +4,15 @@ using System.Text; using System.Threading.Tasks; using Orleans.Concurrency; -using Orleans.ScheduledJobs; +using Orleans.DurableJobs; namespace UnitTests.GrainInterfaces; -public interface IScheduledJobGrain : IGrainWithStringKey +public interface IDurableJobGrain : IGrainWithStringKey { - Task ScheduleJobAsync(string jobName, DateTimeOffset scheduledTime, IReadOnlyDictionary metadata = null); + Task ScheduleJobAsync(string jobName, DateTimeOffset scheduledTime, IReadOnlyDictionary metadata = null); - Task TryCancelJobAsync(ScheduledJob job); + Task TryCancelJobAsync(DurableJob job); Task HasJobRan(string jobId); @@ -21,7 +21,7 @@ public interface IScheduledJobGrain : IGrainWithStringKey Task GetJobExecutionTime(string jobId); - Task GetJobContext(string jobId); + Task GetJobContext(string jobId); Task WasCancellationTokenCancelled(string jobId); } diff --git a/test/Grains/TestGrainInterfaces/IRetryTestGrain.cs b/test/Grains/TestGrainInterfaces/IRetryTestGrain.cs index ebc1762b706..8d25b63f8af 100644 --- a/test/Grains/TestGrainInterfaces/IRetryTestGrain.cs +++ b/test/Grains/TestGrainInterfaces/IRetryTestGrain.cs @@ -2,13 +2,13 @@ using System.Collections.Generic; using System.Threading.Tasks; using Orleans.Concurrency; -using Orleans.ScheduledJobs; +using Orleans.DurableJobs; namespace UnitTests.GrainInterfaces; public interface IRetryTestGrain : IGrainWithStringKey { - Task ScheduleJobAsync(string jobName, DateTimeOffset scheduledTime, IReadOnlyDictionary metadata = null); + Task ScheduleJobAsync(string jobName, DateTimeOffset scheduledTime, IReadOnlyDictionary metadata = null); Task HasJobSucceeded(string jobId); @@ -19,5 +19,5 @@ public interface IRetryTestGrain : IGrainWithStringKey Task> GetJobDequeueCountHistory(string jobId); - Task GetFinalJobContext(string jobId); + Task GetFinalJobContext(string jobId); } diff --git a/test/Grains/TestGrainInterfaces/ISchedulerGrain.cs b/test/Grains/TestGrainInterfaces/ISchedulerGrain.cs index f18e5ec9f45..0876f00d138 100644 --- a/test/Grains/TestGrainInterfaces/ISchedulerGrain.cs +++ b/test/Grains/TestGrainInterfaces/ISchedulerGrain.cs @@ -1,10 +1,10 @@ using System; using System.Threading.Tasks; -using Orleans.ScheduledJobs; +using Orleans.DurableJobs; namespace UnitTests.GrainInterfaces; public interface ISchedulerGrain : IGrainWithStringKey { - Task ScheduleJobOnAnotherGrainAsync(string targetGrainKey, string jobName, DateTimeOffset scheduledTime); + Task ScheduleJobOnAnotherGrainAsync(string targetGrainKey, string jobName, DateTimeOffset scheduledTime); } diff --git a/test/Grains/TestGrainInterfaces/TestGrainInterfaces.csproj b/test/Grains/TestGrainInterfaces/TestGrainInterfaces.csproj index 58838b8f1fe..5b12aadec69 100644 --- a/test/Grains/TestGrainInterfaces/TestGrainInterfaces.csproj +++ b/test/Grains/TestGrainInterfaces/TestGrainInterfaces.csproj @@ -11,7 +11,7 @@ - + diff --git a/test/Grains/TestGrains/ScheduledJobGrain.cs b/test/Grains/TestGrains/DurableJobGrain.cs similarity index 67% rename from test/Grains/TestGrains/ScheduledJobGrain.cs rename to test/Grains/TestGrains/DurableJobGrain.cs index 634033b7ba0..2e5b05cb044 100644 --- a/test/Grains/TestGrains/ScheduledJobGrain.cs +++ b/test/Grains/TestGrains/DurableJobGrain.cs @@ -5,23 +5,23 @@ using System.Threading; using System.Threading.Tasks; using Microsoft.Extensions.Logging; -using Orleans.ScheduledJobs; +using Orleans.DurableJobs; using UnitTests.GrainInterfaces; namespace UnitTests.Grains; -public class ScheduledJobGrain : Grain, IScheduledJobGrain, IScheduledJobHandler +public class DurableJobGrain : Grain, IDurableJobGrain, IDurableJobHandler { private Dictionary jobRunStatus = new(); private Dictionary jobExecutionTimes = new(); - private Dictionary jobContexts = new(); + private Dictionary jobContexts = new(); private Dictionary cancellationTokenStatus = new(); - private readonly ILocalScheduledJobManager _localScheduledJobManager; - private readonly ILogger _logger; + private readonly ILocalDurableJobManager _localDurableJobManager; + private readonly ILogger _logger; - public ScheduledJobGrain(ILocalScheduledJobManager localScheduledJobManager, ILogger logger) + public DurableJobGrain(ILocalDurableJobManager localDurableJobManager, ILogger logger) { - _localScheduledJobManager = localScheduledJobManager; + _localDurableJobManager = localDurableJobManager; _logger = logger; } @@ -30,7 +30,7 @@ public Task HasJobRan(string jobId) return Task.FromResult(jobRunStatus.TryGetValue(jobId, out var taskResult) && taskResult.Task.IsCompleted); } - public Task ExecuteJobAsync(IScheduledJobContext ctx, CancellationToken cancellationToken) + public Task ExecuteJobAsync(IDurableJobContext ctx, CancellationToken cancellationToken) { _logger.LogInformation("Job {JobId} received at {ReceivedTime}", ctx.Job.Id, DateTime.UtcNow); jobExecutionTimes[ctx.Job.Id] = DateTimeOffset.UtcNow; @@ -40,9 +40,9 @@ public Task ExecuteJobAsync(IScheduledJobContext ctx, CancellationToken cancella return Task.CompletedTask; } - public async Task ScheduleJobAsync(string jobName, DateTimeOffset scheduledTime, IReadOnlyDictionary metadata = null) + public async Task ScheduleJobAsync(string jobName, DateTimeOffset scheduledTime, IReadOnlyDictionary metadata = null) { - var job = await _localScheduledJobManager.ScheduleJobAsync(this.GetGrainId(), jobName, scheduledTime, metadata, CancellationToken.None); + var job = await _localDurableJobManager.ScheduleJobAsync(this.GetGrainId(), jobName, scheduledTime, metadata, CancellationToken.None); jobRunStatus[job.Id] = new TaskCompletionSource(); return job; } @@ -59,9 +59,9 @@ public async Task WaitForJobToRun(string jobId) await taskResult.Task; } - public async Task TryCancelJobAsync(ScheduledJob job) + public async Task TryCancelJobAsync(DurableJob job) { - return await _localScheduledJobManager.TryCancelScheduledJobAsync(job, CancellationToken.None); + return await _localDurableJobManager.TryCancelDurableJobAsync(job, CancellationToken.None); } public Task GetJobExecutionTime(string jobId) @@ -74,7 +74,7 @@ public Task GetJobExecutionTime(string jobId) return Task.FromResult(time); } - public Task GetJobContext(string jobId) + public Task GetJobContext(string jobId) { if (!jobContexts.TryGetValue(jobId, out var ctx)) { diff --git a/test/Grains/TestGrains/RetryTestGrain.cs b/test/Grains/TestGrains/RetryTestGrain.cs index 506b4b2fee6..7cb57b04afc 100644 --- a/test/Grains/TestGrains/RetryTestGrain.cs +++ b/test/Grains/TestGrains/RetryTestGrain.cs @@ -3,23 +3,23 @@ using System.Threading; using System.Threading.Tasks; using Microsoft.Extensions.Logging; -using Orleans.ScheduledJobs; +using Orleans.DurableJobs; using UnitTests.GrainInterfaces; namespace UnitTests.Grains; -public class RetryTestGrain : Grain, IRetryTestGrain, IScheduledJobHandler +public class RetryTestGrain : Grain, IRetryTestGrain, IDurableJobHandler { private readonly Dictionary _jobSuccessStatus = new(); private readonly Dictionary _jobExecutionAttempts = new(); private readonly Dictionary> _jobDequeueCountHistory = new(); - private readonly Dictionary _finalJobContexts = new(); - private readonly ILocalScheduledJobManager _localScheduledJobManager; + private readonly Dictionary _finalJobContexts = new(); + private readonly ILocalDurableJobManager _localDurableJobManager; private readonly ILogger _logger; - public RetryTestGrain(ILocalScheduledJobManager localScheduledJobManager, ILogger logger) + public RetryTestGrain(ILocalDurableJobManager localDurableJobManager, ILogger logger) { - _localScheduledJobManager = localScheduledJobManager; + _localDurableJobManager = localDurableJobManager; _logger = logger; } @@ -28,7 +28,7 @@ public Task HasJobSucceeded(string jobId) return Task.FromResult(_jobSuccessStatus.TryGetValue(jobId, out var tcs) && tcs.Task.IsCompleted); } - public Task ExecuteJobAsync(IScheduledJobContext ctx, CancellationToken cancellationToken) + public Task ExecuteJobAsync(IDurableJobContext ctx, CancellationToken cancellationToken) { var jobId = ctx.Job.Id; @@ -76,9 +76,9 @@ public Task ExecuteJobAsync(IScheduledJobContext ctx, CancellationToken cancella return Task.CompletedTask; } - public async Task ScheduleJobAsync(string jobName, DateTimeOffset scheduledTime, IReadOnlyDictionary metadata = null) + public async Task ScheduleJobAsync(string jobName, DateTimeOffset scheduledTime, IReadOnlyDictionary metadata = null) { - var job = await _localScheduledJobManager.ScheduleJobAsync( + var job = await _localDurableJobManager.ScheduleJobAsync( this.GetGrainId(), jobName, scheduledTime, @@ -122,7 +122,7 @@ public Task> GetJobDequeueCountHistory(string jobId) return Task.FromResult(history); } - public Task GetFinalJobContext(string jobId) + public Task GetFinalJobContext(string jobId) { if (!_finalJobContexts.TryGetValue(jobId, out var ctx)) { diff --git a/test/Grains/TestGrains/SchedulerGrain.cs b/test/Grains/TestGrains/SchedulerGrain.cs index f4e657bc742..91bf9c823d0 100644 --- a/test/Grains/TestGrains/SchedulerGrain.cs +++ b/test/Grains/TestGrains/SchedulerGrain.cs @@ -2,30 +2,30 @@ using System.Threading; using System.Threading.Tasks; using Microsoft.Extensions.Logging; -using Orleans.ScheduledJobs; +using Orleans.DurableJobs; using UnitTests.GrainInterfaces; namespace UnitTests.Grains; public class SchedulerGrain : Grain, ISchedulerGrain { - private readonly ILocalScheduledJobManager _localScheduledJobManager; + private readonly ILocalDurableJobManager _localDurableJobManager; private readonly IGrainFactory _grainFactory; private readonly ILogger _logger; public SchedulerGrain( - ILocalScheduledJobManager localScheduledJobManager, + ILocalDurableJobManager localDurableJobManager, IGrainFactory grainFactory, ILogger logger) { - _localScheduledJobManager = localScheduledJobManager; + _localDurableJobManager = localDurableJobManager; _grainFactory = grainFactory; _logger = logger; } - public async Task ScheduleJobOnAnotherGrainAsync(string targetGrainKey, string jobName, DateTimeOffset scheduledTime) + public async Task ScheduleJobOnAnotherGrainAsync(string targetGrainKey, string jobName, DateTimeOffset scheduledTime) { - var targetGrain = _grainFactory.GetGrain(targetGrainKey); + var targetGrain = _grainFactory.GetGrain(targetGrainKey); var targetGrainId = targetGrain.GetGrainId(); _logger.LogInformation( @@ -34,7 +34,7 @@ public async Task ScheduleJobOnAnotherGrainAsync(string targetGrai targetGrainKey, this.GetPrimaryKeyString()); - var job = await _localScheduledJobManager.ScheduleJobAsync( + var job = await _localDurableJobManager.ScheduleJobAsync( targetGrainId, jobName, scheduledTime, diff --git a/test/Grains/TestGrains/TestGrains.csproj b/test/Grains/TestGrains/TestGrains.csproj index 52f316aa37b..11223facc73 100644 --- a/test/Grains/TestGrains/TestGrains.csproj +++ b/test/Grains/TestGrains/TestGrains.csproj @@ -12,6 +12,6 @@ - + diff --git a/test/NonSilo.Tests/ScheduledJobs/InMemoryJobQueueTests.cs b/test/NonSilo.Tests/ScheduledJobs/InMemoryJobQueueTests.cs index 43bd6ccfbf1..85b99b57beb 100644 --- a/test/NonSilo.Tests/ScheduledJobs/InMemoryJobQueueTests.cs +++ b/test/NonSilo.Tests/ScheduledJobs/InMemoryJobQueueTests.cs @@ -3,14 +3,14 @@ using System.Linq; using System.Threading; using System.Threading.Tasks; -using Orleans.ScheduledJobs; +using Orleans.DurableJobs; using Orleans.Runtime; using NSubstitute; using Xunit; -namespace NonSilo.Tests.ScheduledJobs; +namespace NonSilo.Tests.DurableJobs; -[TestCategory("ScheduledJobs")] +[TestCategory("DurableJobs")] public class InMemoryJobQueueTests { [Fact] @@ -62,7 +62,7 @@ public async Task GetAsyncEnumerator_ReturnsJobsInDueTimeOrder() queue.Enqueue(job2, 0); queue.MarkAsComplete(); - var results = new List(); + var results = new List(); await foreach (var context in queue.WithCancellation(CancellationToken.None)) { results.Add(context); @@ -252,7 +252,7 @@ public async Task GetAsyncEnumerator_RespectsEmptyBuckets() queue.CancelJob("job2"); queue.MarkAsComplete(); - var results = new List(); + var results = new List(); await foreach (var context in queue.WithCancellation(CancellationToken.None)) { results.Add(context); @@ -325,9 +325,9 @@ await Assert.ThrowsAnyAsync(async () => }); } - private static ScheduledJob CreateJob(string id, DateTimeOffset dueTime) + private static DurableJob CreateJob(string id, DateTimeOffset dueTime) { - return new ScheduledJob + return new DurableJob { Id = id, Name = id, @@ -338,9 +338,9 @@ private static ScheduledJob CreateJob(string id, DateTimeOffset dueTime) }; } - private static IScheduledJobContext CreateJobContext(ScheduledJob job, string runId, int dequeueCount) + private static IDurableJobContext CreateJobContext(DurableJob job, string runId, int dequeueCount) { - var context = Substitute.For(); + var context = Substitute.For(); context.Job.Returns(job); context.RunId.Returns(runId); context.DequeueCount.Returns(dequeueCount); diff --git a/test/TestInfrastructure/TestExtensions/DefaultClusterFixture.cs b/test/TestInfrastructure/TestExtensions/DefaultClusterFixture.cs index 000483df482..6b0f6ab32a4 100644 --- a/test/TestInfrastructure/TestExtensions/DefaultClusterFixture.cs +++ b/test/TestInfrastructure/TestExtensions/DefaultClusterFixture.cs @@ -58,7 +58,7 @@ public void Configure(ISiloBuilder hostBuilder) hostBuilder .Configure(o => o.ClientGatewayShutdownNotificationTimeout = default) .UseInMemoryReminderService() - .UseInMemoryScheduledJobs() + .UseInMemoryDurableJobs() .AddMemoryGrainStorageAsDefault() .AddMemoryGrainStorage("MemoryStore"); } diff --git a/test/Tester/ScheduledJobs/ScheduledJobTestsRunner.cs b/test/Tester/DurableJobs/DurableJobTestsRunner.cs similarity index 83% rename from test/Tester/ScheduledJobs/ScheduledJobTestsRunner.cs rename to test/Tester/DurableJobs/DurableJobTestsRunner.cs index 530a90f6262..0417d0ca669 100644 --- a/test/Tester/ScheduledJobs/ScheduledJobTestsRunner.cs +++ b/test/Tester/DurableJobs/DurableJobTestsRunner.cs @@ -4,27 +4,28 @@ using System.Threading.Tasks; using Orleans; using Orleans.Internal; -using Orleans.ScheduledJobs; +using Orleans.DurableJobs; using Xunit; +using UnitTests.GrainInterfaces; -namespace Tester.ScheduledJobs; +namespace Tester.DurableJobs; /// -/// Contains the test logic for scheduled jobs that can be run against different providers. +/// Contains the test logic for durable jobs that can be run against different providers. /// This class is provider-agnostic and can be reused by test classes for InMemory, Azure, and other providers. /// -public class ScheduledJobTestsRunner +public class DurableJobTestsRunner { private readonly IGrainFactory _grainFactory; - public ScheduledJobTestsRunner(IGrainFactory grainFactory) + public DurableJobTestsRunner(IGrainFactory grainFactory) { _grainFactory = grainFactory; } - public async Task ScheduledJobGrain() + public async Task DurableJobGrain() { - var grain = _grainFactory.GetGrain("test-job-grain"); + var grain = _grainFactory.GetGrain("test-job-grain"); var dueTime = DateTimeOffset.UtcNow.AddSeconds(5); var job1 = await grain.ScheduleJobAsync("TestJob", dueTime); Assert.NotNull(job1); @@ -45,7 +46,7 @@ public async Task ScheduledJobGrain() } catch (TimeoutException) { - Assert.Fail($"The scheduled job {job.Name} did not run within the expected time."); + Assert.Fail($"The durable job {job.Name} did not run within the expected time."); } } // Verify the canceled job did not run @@ -54,7 +55,7 @@ public async Task ScheduledJobGrain() public async Task JobExecutionOrder() { - var grain = _grainFactory.GetGrain("test-execution-order"); + var grain = _grainFactory.GetGrain("test-execution-order"); var baseTime = DateTimeOffset.UtcNow.AddSeconds(2); var job1 = await grain.ScheduleJobAsync("FirstJob", baseTime); @@ -75,7 +76,7 @@ public async Task JobExecutionOrder() public async Task PastDueTime() { - var grain = _grainFactory.GetGrain("test-past-due"); + var grain = _grainFactory.GetGrain("test-past-due"); var pastTime = DateTimeOffset.UtcNow.AddSeconds(-5); var job = await grain.ScheduleJobAsync("PastDueJob", pastTime); @@ -87,7 +88,7 @@ public async Task PastDueTime() public async Task JobWithMetadata() { - var grain = _grainFactory.GetGrain("test-metadata"); + var grain = _grainFactory.GetGrain("test-metadata"); var dueTime = DateTimeOffset.UtcNow.AddSeconds(3); var metadata = new Dictionary { @@ -114,9 +115,9 @@ public async Task JobWithMetadata() public async Task MultipleGrains() { - var grain1 = _grainFactory.GetGrain("test-grain-1"); - var grain2 = _grainFactory.GetGrain("test-grain-2"); - var grain3 = _grainFactory.GetGrain("test-grain-3"); + var grain1 = _grainFactory.GetGrain("test-grain-1"); + var grain2 = _grainFactory.GetGrain("test-grain-2"); + var grain3 = _grainFactory.GetGrain("test-grain-3"); var dueTime = DateTimeOffset.UtcNow.AddSeconds(3); var job1 = await grain1.ScheduleJobAsync("Job1", dueTime); @@ -138,7 +139,7 @@ public async Task MultipleGrains() public async Task DuplicateJobNames() { - var grain = _grainFactory.GetGrain("test-duplicate-names"); + var grain = _grainFactory.GetGrain("test-duplicate-names"); var dueTime = DateTimeOffset.UtcNow.AddSeconds(3); var job1 = await grain.ScheduleJobAsync("SameName", dueTime); @@ -164,12 +165,12 @@ public async Task DuplicateJobNames() public async Task CancelNonExistentJob() { - var grain = _grainFactory.GetGrain("test-cancel-nonexistent"); + var grain = _grainFactory.GetGrain("test-cancel-nonexistent"); var dueTime = DateTimeOffset.UtcNow.AddSeconds(10); var job = await grain.ScheduleJobAsync("RealJob", dueTime); - var fakeJob = new ScheduledJob + var fakeJob = new DurableJob { Id = "non-existent-id", Name = "FakeJob", @@ -187,7 +188,7 @@ public async Task CancelNonExistentJob() public async Task CancelAlreadyExecutedJob() { - var grain = _grainFactory.GetGrain("test-cancel-executed"); + var grain = _grainFactory.GetGrain("test-cancel-executed"); var dueTime = DateTimeOffset.UtcNow.AddSeconds(2); var job = await grain.ScheduleJobAsync("QuickJob", dueTime); @@ -201,11 +202,11 @@ public async Task CancelAlreadyExecutedJob() public async Task ConcurrentScheduling() { - var grain = _grainFactory.GetGrain("test-concurrent"); + var grain = _grainFactory.GetGrain("test-concurrent"); var baseTime = DateTimeOffset.UtcNow.AddSeconds(5); var jobCount = 20; - var scheduleTasks = new List>(); + var scheduleTasks = new List>(); for (int i = 0; i < jobCount; i++) { scheduleTasks.Add(grain.ScheduleJobAsync($"ConcurrentJob{i}", baseTime.AddMilliseconds(i * 100))); @@ -227,7 +228,7 @@ public async Task ConcurrentScheduling() public async Task JobPropertiesVerification() { - var grain = _grainFactory.GetGrain("test-properties"); + var grain = _grainFactory.GetGrain("test-properties"); var dueTime = DateTimeOffset.UtcNow.AddSeconds(3); var metadata = new Dictionary { ["Key"] = "Value" }; @@ -254,7 +255,7 @@ public async Task JobPropertiesVerification() public async Task DequeueCount() { - var grain = _grainFactory.GetGrain("test-dequeue-count"); + var grain = _grainFactory.GetGrain("test-dequeue-count"); var dueTime = DateTimeOffset.UtcNow.AddSeconds(3); var job = await grain.ScheduleJobAsync("DequeueTestJob", dueTime); @@ -268,8 +269,8 @@ public async Task DequeueCount() public async Task ScheduleJobOnAnotherGrain() { - var schedulerGrain = _grainFactory.GetGrain("scheduler-grain"); - var targetGrain = _grainFactory.GetGrain("target-grain"); + var schedulerGrain = _grainFactory.GetGrain("scheduler-grain"); + var targetGrain = _grainFactory.GetGrain("target-grain"); var dueTime = DateTimeOffset.UtcNow.AddSeconds(3); var job = await schedulerGrain.ScheduleJobOnAnotherGrainAsync("target-grain", "CrossGrainJob", dueTime); @@ -290,7 +291,7 @@ public async Task ScheduleJobOnAnotherGrain() public async Task JobRetry() { - var grain = _grainFactory.GetGrain("retry-test-grain"); + var grain = _grainFactory.GetGrain("retry-test-grain"); var dueTime = DateTimeOffset.UtcNow.AddSeconds(2); var metadata = new Dictionary { diff --git a/test/Tester/ScheduledJobs/IJobShardManagerTestFixture.cs b/test/Tester/DurableJobs/IJobShardManagerTestFixture.cs similarity index 94% rename from test/Tester/ScheduledJobs/IJobShardManagerTestFixture.cs rename to test/Tester/DurableJobs/IJobShardManagerTestFixture.cs index f77d65f32ef..acd09191e18 100644 --- a/test/Tester/ScheduledJobs/IJobShardManagerTestFixture.cs +++ b/test/Tester/DurableJobs/IJobShardManagerTestFixture.cs @@ -2,9 +2,9 @@ using System.Collections.Generic; using System.Threading.Tasks; using Orleans.Runtime; -using Orleans.ScheduledJobs; +using Orleans.DurableJobs; -namespace Tester.ScheduledJobs; +namespace Tester.DurableJobs; /// /// Defines the contract for provider-specific test fixtures used by . diff --git a/test/Tester/ScheduledJobs/InMemoryJobShardManagerTestFixture.cs b/test/Tester/DurableJobs/InMemoryJobShardManagerTestFixture.cs similarity index 93% rename from test/Tester/ScheduledJobs/InMemoryJobShardManagerTestFixture.cs rename to test/Tester/DurableJobs/InMemoryJobShardManagerTestFixture.cs index 39509fe8d65..e1115183215 100644 --- a/test/Tester/ScheduledJobs/InMemoryJobShardManagerTestFixture.cs +++ b/test/Tester/DurableJobs/InMemoryJobShardManagerTestFixture.cs @@ -1,8 +1,8 @@ using System.Threading.Tasks; using Orleans.Runtime; -using Orleans.ScheduledJobs; +using Orleans.DurableJobs; -namespace Tester.ScheduledJobs; +namespace Tester.DurableJobs; /// /// InMemory implementation of . diff --git a/test/Tester/ScheduledJobs/InMemoryJobShardManagerTests.cs b/test/Tester/DurableJobs/InMemoryJobShardManagerTests.cs similarity index 96% rename from test/Tester/ScheduledJobs/InMemoryJobShardManagerTests.cs rename to test/Tester/DurableJobs/InMemoryJobShardManagerTests.cs index beb104ee8d8..de8a3f10227 100644 --- a/test/Tester/ScheduledJobs/InMemoryJobShardManagerTests.cs +++ b/test/Tester/DurableJobs/InMemoryJobShardManagerTests.cs @@ -1,13 +1,13 @@ using System.Threading.Tasks; using Xunit; -namespace Tester.ScheduledJobs; +namespace Tester.DurableJobs; /// /// Tests for using the . /// These tests verify shard lifecycle management, ownership, and failover semantics for the InMemory provider. /// -[TestCategory("BVT"), TestCategory("ScheduledJobs")] +[TestCategory("BVT"), TestCategory("DurableJobs")] public class InMemoryJobShardManagerTests : IAsyncLifetime { private readonly InMemoryJobShardManagerTestFixture _fixture; diff --git a/test/Tester/ScheduledJobs/JobShardManagerTestsRunner.cs b/test/Tester/DurableJobs/JobShardManagerTestsRunner.cs similarity index 95% rename from test/Tester/ScheduledJobs/JobShardManagerTestsRunner.cs rename to test/Tester/DurableJobs/JobShardManagerTestsRunner.cs index 4040727dd01..9ad0ff31534 100644 --- a/test/Tester/ScheduledJobs/JobShardManagerTestsRunner.cs +++ b/test/Tester/DurableJobs/JobShardManagerTestsRunner.cs @@ -6,14 +6,14 @@ using System.Threading; using System.Threading.Tasks; using Orleans.Runtime; -using Orleans.ScheduledJobs; +using Orleans.DurableJobs; using Xunit; -namespace Tester.ScheduledJobs; +namespace Tester.DurableJobs; /// /// Contains provider-agnostic test logic for job shard managers that can be run against different providers. -/// This class is similar to but operates at the infrastructure layer, +/// This class is similar to but operates at the infrastructure layer, /// testing shard lifecycle management, ownership, and failover semantics. /// public class JobShardManagerTestsRunner @@ -135,7 +135,7 @@ public async Task ReadFrozenShard() var counter = 1; var cts = new CancellationTokenSource(TimeSpan.FromSeconds(20)); - await foreach (var jobCtx in shard1.ConsumeScheduledJobsAsync().WithCancellation(cts.Token)) + await foreach (var jobCtx in shard1.ConsumeDurableJobsAsync().WithCancellation(cts.Token)) { Assert.Equal($"job{counter}", jobCtx.Job.Name); await shard1.RemoveJobAsync(jobCtx.Job.Id, cts.Token); @@ -173,7 +173,7 @@ public async Task LiveShard() var cts = new CancellationTokenSource(TimeSpan.FromSeconds(10)); await shard1.MarkAsCompleteAsync(CancellationToken.None); await shard1.RemoveJobAsync(jobToCancel.Id, CancellationToken.None); - await foreach (var jobCtx in shard1.ConsumeScheduledJobsAsync().WithCancellation(cts.Token)) + await foreach (var jobCtx in shard1.ConsumeDurableJobsAsync().WithCancellation(cts.Token)) { Assert.Equal($"job{counter}", jobCtx.Job.Name); await shard1.RemoveJobAsync(jobCtx.Job.Id, CancellationToken.None); @@ -221,7 +221,7 @@ public async Task JobMetadata() var job2 = await shard.TryScheduleJobAsync(GrainId.Create("type", "target2"), "job2", DateTime.UtcNow.AddSeconds(2), jobMetadata2, CancellationToken.None); var job3 = await shard.TryScheduleJobAsync(GrainId.Create("type", "target3"), "job3", DateTime.UtcNow.AddSeconds(3), null, CancellationToken.None); - // Verify metadata is set on the scheduled jobs + // Verify metadata is set on the durable jobs Assert.Equal(jobMetadata1, job1.Metadata); Assert.Equal(jobMetadata2, job2.Metadata); Assert.Null(job3.Metadata); @@ -235,9 +235,9 @@ public async Task JobMetadata() shard = shards[0]; // Consume jobs and verify metadata is preserved - var consumedJobs = new List(); + var consumedJobs = new List(); var cts = new CancellationTokenSource(TimeSpan.FromSeconds(30)); - await foreach (var jobCtx in shard.ConsumeScheduledJobsAsync().WithCancellation(cts.Token)) + await foreach (var jobCtx in shard.ConsumeDurableJobsAsync().WithCancellation(cts.Token)) { consumedJobs.Add(jobCtx.Job); await shard.RemoveJobAsync(jobCtx.Job.Id, CancellationToken.None); @@ -370,7 +370,7 @@ public async Task StopProcessingShard() var counter = 1; var cts = new CancellationTokenSource(TimeSpan.FromSeconds(40)); - await foreach (var jobCtx in shard1.ConsumeScheduledJobsAsync().WithCancellation(cts.Token)) + await foreach (var jobCtx in shard1.ConsumeDurableJobsAsync().WithCancellation(cts.Token)) { Assert.Equal($"job{counter}", jobCtx.Job.Name); if (counter == 2) @@ -400,7 +400,7 @@ public async Task RetryJobLater() // Schedule a job var job = await shard1.TryScheduleJobAsync(GrainId.Create("type", "target1"), "job1", DateTime.UtcNow.AddSeconds(1), null, CancellationToken.None); var cts = new CancellationTokenSource(TimeSpan.FromSeconds(40)); - await foreach (var jobCtx in shard1.ConsumeScheduledJobsAsync().WithCancellation(cts.Token)) + await foreach (var jobCtx in shard1.ConsumeDurableJobsAsync().WithCancellation(cts.Token)) { Assert.Equal("job1", jobCtx.Job.Name); var newDueTime = DateTimeOffset.UtcNow.AddSeconds(1); @@ -409,7 +409,7 @@ public async Task RetryJobLater() } // Consume again - await foreach (var jobCtx in shard1.ConsumeScheduledJobsAsync().WithCancellation(cts.Token)) + await foreach (var jobCtx in shard1.ConsumeDurableJobsAsync().WithCancellation(cts.Token)) { Assert.Equal("job1", jobCtx.Job.Name); Assert.NotEqual(job.DueTime, jobCtx.Job.DueTime); @@ -450,7 +450,7 @@ public async Task JobCancellation() var consumedJobs = new List(); var cts = new CancellationTokenSource(TimeSpan.FromSeconds(30)); - await foreach (var jobCtx in shard.ConsumeScheduledJobsAsync().WithCancellation(cts.Token)) + await foreach (var jobCtx in shard.ConsumeDurableJobsAsync().WithCancellation(cts.Token)) { consumedJobs.Add(jobCtx.Job.Name); @@ -484,7 +484,7 @@ public async Task JobCancellation() var hasJobs = false; cts = new CancellationTokenSource(TimeSpan.FromSeconds(5)); - await foreach (var jobCtx in shard.ConsumeScheduledJobsAsync().WithCancellation(cts.Token)) + await foreach (var jobCtx in shard.ConsumeDurableJobsAsync().WithCancellation(cts.Token)) { hasJobs = true; break; @@ -546,7 +546,7 @@ public async Task UnregisterShard_WithJobsRemaining() var consumedJobs = new List(); var cts = new CancellationTokenSource(TimeSpan.FromSeconds(20)); - await foreach (var jobCtx in shards[0].ConsumeScheduledJobsAsync().WithCancellation(cts.Token)) + await foreach (var jobCtx in shards[0].ConsumeDurableJobsAsync().WithCancellation(cts.Token)) { consumedJobs.Add(jobCtx.Job.Name); await shards[0].RemoveJobAsync(jobCtx.Job.Id, CancellationToken.None);