Skip to content

feat: add overwrite rules to recurringSchedule.CreateOrUpdate #7

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
birdalicious merged 1 commit into from
Jun 16, 2026
Merged

feat: add overwrite rules to recurringSchedule.CreateOrUpdate #7

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion samples/Sheddueller.SampleHost/Program.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -179,7 +179,7 @@
"* * * * *",
(service, ct, progress) => service.RunRecurringAsync(progress, ct),
new RecurringScheduleOptions(Priority: 10, OverlapMode: RecurringOverlapMode.Skip),
cancellationToken).ConfigureAwait(false);
cancellationToken: cancellationToken).ConfigureAwait(false);
return RedirectWithMessage($"Recurring schedule 'demo:recurring' is {result}. The next occurrence will fire on the next minute boundary.");
});

Expand Down
20 changes: 14 additions & 6 deletions src/Sheddueller.Postgres/Internal/Operations/CreateOrUpdateRecurringScheduleOperation.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -10,36 +10,44 @@ public static async ValueTask<RecurringScheduleUpsertResult> ExecuteAsync(
UpsertRecurringScheduleRequest request,
CancellationToken cancellationToken)
{
var retry = PostgresRetryPolicies.Normalize(request.RetryPolicy);
await using var connection = await context.OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
await using var transaction = await connection.BeginTransactionAsync(cancellationToken).ConfigureAwait(false);
var existing = await PostgresSchedules.ReadScheduleDefinitionForUpdateAsync(context, connection, transaction, request.ScheduleKey, cancellationToken)
.ConfigureAwait(false);
var updateOptions = request.UpdateOptions ?? RecurringScheduleUpdateOptions.Default;

if (existing is null)
{
var insertRetry = PostgresRetryPolicies.Normalize(request.RetryPolicy);
var nextFireAtUtc = CronSchedule.GetNextOccurrenceAfter(
request.CronExpression,
await PostgresOperationContext.ReadTransactionTimestampAsync(connection, transaction, cancellationToken).ConfigureAwait(false));
await PostgresSchedules.InsertScheduleAsync(context, connection, transaction, request, retry, nextFireAtUtc, cancellationToken)
await PostgresSchedules.InsertScheduleAsync(context, connection, transaction, request, insertRetry, nextFireAtUtc, cancellationToken)
.ConfigureAwait(false);
await context.NotifyAsync(connection, transaction, cancellationToken).ConfigureAwait(false);
await transaction.CommitAsync(cancellationToken).ConfigureAwait(false);
return RecurringScheduleUpsertResult.Created;
}

if (existing.EqualsRequest(request))
var effectiveRequest = request with
{
CronExpression = updateOptions.OverwriteCronExpression ? request.CronExpression : existing.CronExpression,
};
var effectiveIsPaused = !updateOptions.OverwritePausedState && existing.IsPaused;

if (existing.EqualsRequest(effectiveRequest) && existing.IsPaused == effectiveIsPaused)
{
await transaction.CommitAsync(cancellationToken).ConfigureAwait(false);
return RecurringScheduleUpsertResult.Unchanged;
}

DateTimeOffset? updatedNextFireAtUtc = existing.IsPaused
var retry = PostgresRetryPolicies.Normalize(effectiveRequest.RetryPolicy);
DateTimeOffset? updatedNextFireAtUtc = effectiveIsPaused
? null
: CronSchedule.GetNextOccurrenceAfter(
request.CronExpression,
effectiveRequest.CronExpression,
await PostgresOperationContext.ReadTransactionTimestampAsync(connection, transaction, cancellationToken).ConfigureAwait(false));
await PostgresSchedules.UpdateScheduleAsync(context, connection, transaction, request, retry, updatedNextFireAtUtc, cancellationToken)
await PostgresSchedules.UpdateScheduleAsync(context, connection, transaction, effectiveRequest, retry, effectiveIsPaused, updatedNextFireAtUtc, cancellationToken)
.ConfigureAwait(false);
await context.NotifyAsync(connection, transaction, cancellationToken).ConfigureAwait(false);
await transaction.CommitAsync(cancellationToken).ConfigureAwait(false);
Expand Down
10 changes: 7 additions & 3 deletions src/Sheddueller.Postgres/Internal/Operations/PostgresSchedules.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -102,7 +102,7 @@ await PostgresOperationContext.ExecuteCountAsync(
values (
@schedule_key,
@cron_expression,
false,
@is_paused,
@overlap_mode,
@priority,
@service_type,
Expand All @@ -121,7 +121,7 @@ await PostgresOperationContext.ExecuteCountAsync(
transaction_timestamp(),
transaction_timestamp());
""",
command => AddScheduleParameters(command, request, retry, nextFireAtUtc),
command => AddScheduleParameters(command, request, retry, isPaused: false, nextFireAtUtc),
cancellationToken)
.ConfigureAwait(false);
await ReplaceScheduleGroupsAsync(context, connection, transaction, request.ScheduleKey, request.ConcurrencyGroupKeys, cancellationToken)
Expand All @@ -136,6 +136,7 @@ public static async ValueTask UpdateScheduleAsync(
NpgsqlTransaction transaction,
UpsertRecurringScheduleRequest request,
PostgresRetryPolicy retry,
bool isPaused,
DateTimeOffset? nextFireAtUtc,
CancellationToken cancellationToken)
{
Expand All @@ -145,6 +146,7 @@ await PostgresOperationContext.ExecuteCountAsync(
$"""
update {context.Names.RecurringSchedules}
set cron_expression = @cron_expression,
is_paused = @is_paused,
overlap_mode = @overlap_mode,
priority = @priority,
service_type = @service_type,
Expand All @@ -163,7 +165,7 @@ await PostgresOperationContext.ExecuteCountAsync(
updated_at_utc = transaction_timestamp()
where schedule_key = @schedule_key;
""",
command => AddScheduleParameters(command, request, retry, nextFireAtUtc),
command => AddScheduleParameters(command, request, retry, isPaused, nextFireAtUtc),
cancellationToken)
.ConfigureAwait(false);
await PostgresOperationContext.ExecuteCountAsync(
Expand Down Expand Up @@ -476,10 +478,12 @@ private static void AddScheduleParameters(
NpgsqlCommand command,
UpsertRecurringScheduleRequest request,
PostgresRetryPolicy retry,
bool isPaused,
DateTimeOffset? nextFireAtUtc)
{
command.Parameters.AddWithValue("schedule_key", request.ScheduleKey);
command.Parameters.AddWithValue("cron_expression", request.CronExpression);
command.Parameters.AddWithValue("is_paused", isPaused);
command.Parameters.AddWithValue("overlap_mode", PostgresConversion.ToText(request.OverlapMode));
command.Parameters.AddWithValue("priority", request.Priority);
command.Parameters.AddWithValue("service_type", request.ServiceType);
Expand Down
80 changes: 80 additions & 0 deletions src/Sheddueller.Testing/CapturingRecurringScheduleManager.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -37,7 +37,17 @@ public ValueTask<RecurringScheduleUpsertResult> CreateOrUpdateAsync(
string cronExpression,
Expression<Func<CancellationToken, Task>> work,
RecurringScheduleOptions? options = null,
RecurringScheduleUpdateOptions? updateOptions = null,
CancellationToken cancellationToken = default)
=> this.CurrentOrDiscardingFake().CreateOrUpdateAsync(scheduleKey, cronExpression, work, options, updateOptions, cancellationToken);

/// <inheritdoc />
public ValueTask<RecurringScheduleUpsertResult> CreateOrUpdateAsync(
string scheduleKey,
string cronExpression,
Expression<Func<CancellationToken, Task>> work,
RecurringScheduleOptions? options,
CancellationToken cancellationToken)
=> this.CurrentOrDiscardingFake().CreateOrUpdateAsync(scheduleKey, cronExpression, work, options, cancellationToken);

/// <inheritdoc />
Expand All @@ -46,7 +56,17 @@ public ValueTask<RecurringScheduleUpsertResult> CreateOrUpdateAsync(
string cronExpression,
Expression<Func<CancellationToken, ValueTask>> work,
RecurringScheduleOptions? options = null,
RecurringScheduleUpdateOptions? updateOptions = null,
CancellationToken cancellationToken = default)
=> this.CurrentOrDiscardingFake().CreateOrUpdateAsync(scheduleKey, cronExpression, work, options, updateOptions, cancellationToken);

/// <inheritdoc />
public ValueTask<RecurringScheduleUpsertResult> CreateOrUpdateAsync(
string scheduleKey,
string cronExpression,
Expression<Func<CancellationToken, ValueTask>> work,
RecurringScheduleOptions? options,
CancellationToken cancellationToken)
=> this.CurrentOrDiscardingFake().CreateOrUpdateAsync(scheduleKey, cronExpression, work, options, cancellationToken);

/// <inheritdoc />
Expand All @@ -55,7 +75,17 @@ public ValueTask<RecurringScheduleUpsertResult> CreateOrUpdateAsync(
string cronExpression,
Expression<Func<CancellationToken, IProgress<decimal>, Task>> work,
RecurringScheduleOptions? options = null,
RecurringScheduleUpdateOptions? updateOptions = null,
CancellationToken cancellationToken = default)
=> this.CurrentOrDiscardingFake().CreateOrUpdateAsync(scheduleKey, cronExpression, work, options, updateOptions, cancellationToken);

/// <inheritdoc />
public ValueTask<RecurringScheduleUpsertResult> CreateOrUpdateAsync(
string scheduleKey,
string cronExpression,
Expression<Func<CancellationToken, IProgress<decimal>, Task>> work,
RecurringScheduleOptions? options,
CancellationToken cancellationToken)
=> this.CurrentOrDiscardingFake().CreateOrUpdateAsync(scheduleKey, cronExpression, work, options, cancellationToken);

/// <inheritdoc />
Expand All @@ -64,7 +94,17 @@ public ValueTask<RecurringScheduleUpsertResult> CreateOrUpdateAsync(
string cronExpression,
Expression<Func<CancellationToken, IProgress<decimal>, ValueTask>> work,
RecurringScheduleOptions? options = null,
RecurringScheduleUpdateOptions? updateOptions = null,
CancellationToken cancellationToken = default)
=> this.CurrentOrDiscardingFake().CreateOrUpdateAsync(scheduleKey, cronExpression, work, options, updateOptions, cancellationToken);

/// <inheritdoc />
public ValueTask<RecurringScheduleUpsertResult> CreateOrUpdateAsync(
string scheduleKey,
string cronExpression,
Expression<Func<CancellationToken, IProgress<decimal>, ValueTask>> work,
RecurringScheduleOptions? options,
CancellationToken cancellationToken)
=> this.CurrentOrDiscardingFake().CreateOrUpdateAsync(scheduleKey, cronExpression, work, options, cancellationToken);

/// <inheritdoc />
Expand All @@ -73,7 +113,17 @@ public ValueTask<RecurringScheduleUpsertResult> CreateOrUpdateAsync<TService>(
string cronExpression,
Expression<Func<TService, CancellationToken, Task>> work,
RecurringScheduleOptions? options = null,
RecurringScheduleUpdateOptions? updateOptions = null,
CancellationToken cancellationToken = default)
=> this.CurrentOrDiscardingFake().CreateOrUpdateAsync(scheduleKey, cronExpression, work, options, updateOptions, cancellationToken);

/// <inheritdoc />
public ValueTask<RecurringScheduleUpsertResult> CreateOrUpdateAsync<TService>(
string scheduleKey,
string cronExpression,
Expression<Func<TService, CancellationToken, Task>> work,
RecurringScheduleOptions? options,
CancellationToken cancellationToken)
=> this.CurrentOrDiscardingFake().CreateOrUpdateAsync(scheduleKey, cronExpression, work, options, cancellationToken);

/// <inheritdoc />
Expand All @@ -82,7 +132,17 @@ public ValueTask<RecurringScheduleUpsertResult> CreateOrUpdateAsync<TService>(
string cronExpression,
Expression<Func<TService, CancellationToken, ValueTask>> work,
RecurringScheduleOptions? options = null,
RecurringScheduleUpdateOptions? updateOptions = null,
CancellationToken cancellationToken = default)
=> this.CurrentOrDiscardingFake().CreateOrUpdateAsync(scheduleKey, cronExpression, work, options, updateOptions, cancellationToken);

/// <inheritdoc />
public ValueTask<RecurringScheduleUpsertResult> CreateOrUpdateAsync<TService>(
string scheduleKey,
string cronExpression,
Expression<Func<TService, CancellationToken, ValueTask>> work,
RecurringScheduleOptions? options,
CancellationToken cancellationToken)
=> this.CurrentOrDiscardingFake().CreateOrUpdateAsync(scheduleKey, cronExpression, work, options, cancellationToken);

/// <inheritdoc />
Expand All @@ -91,7 +151,17 @@ public ValueTask<RecurringScheduleUpsertResult> CreateOrUpdateAsync<TService>(
string cronExpression,
Expression<Func<TService, CancellationToken, IProgress<decimal>, Task>> work,
RecurringScheduleOptions? options = null,
RecurringScheduleUpdateOptions? updateOptions = null,
CancellationToken cancellationToken = default)
=> this.CurrentOrDiscardingFake().CreateOrUpdateAsync(scheduleKey, cronExpression, work, options, updateOptions, cancellationToken);

/// <inheritdoc />
public ValueTask<RecurringScheduleUpsertResult> CreateOrUpdateAsync<TService>(
string scheduleKey,
string cronExpression,
Expression<Func<TService, CancellationToken, IProgress<decimal>, Task>> work,
RecurringScheduleOptions? options,
CancellationToken cancellationToken)
=> this.CurrentOrDiscardingFake().CreateOrUpdateAsync(scheduleKey, cronExpression, work, options, cancellationToken);

/// <inheritdoc />
Expand All @@ -100,7 +170,17 @@ public ValueTask<RecurringScheduleUpsertResult> CreateOrUpdateAsync<TService>(
string cronExpression,
Expression<Func<TService, CancellationToken, IProgress<decimal>, ValueTask>> work,
RecurringScheduleOptions? options = null,
RecurringScheduleUpdateOptions? updateOptions = null,
CancellationToken cancellationToken = default)
=> this.CurrentOrDiscardingFake().CreateOrUpdateAsync(scheduleKey, cronExpression, work, options, updateOptions, cancellationToken);

/// <inheritdoc />
public ValueTask<RecurringScheduleUpsertResult> CreateOrUpdateAsync<TService>(
string scheduleKey,
string cronExpression,
Expression<Func<TService, CancellationToken, IProgress<decimal>, ValueTask>> work,
RecurringScheduleOptions? options,
CancellationToken cancellationToken)
=> this.CurrentOrDiscardingFake().CreateOrUpdateAsync(scheduleKey, cronExpression, work, options, cancellationToken);

/// <inheritdoc />
Expand Down
Loading