Kanject.Core.Recurring.Provider.CacheDb 1.5.0

Kanject.Core.Recurring.Provider.CacheDb

An IRecurringDataProvider implementation (from Kanject.Core.Recurring.Abstractions) that keeps recurring-job leases and checkpoints in any ICacheDb (from Kanject.Core.CacheDb.Abstractions). Use it when a Kanject.Core [RecurringHosted] job sets LeaseKey or CheckpointKey and you want a CacheDb provider as the backing store.

Installation

dotnet add package Kanject.Core                               # [Recurring] / [RecurringHosted]
dotnet add package Kanject.Core.Recurring.Provider.CacheDb
dotnet add package Kanject.Core.CacheDb.Provider.InMemory     # local-development store

Targets .NET 8, .NET 9 and .NET 10. Kanject.Core.Recurring.Abstractions comes in transitively. This package doesn't include a CacheDb provider. You have to install and register one.

Quick start

You need three pieces: a job marked with [RecurringHosted], a CacheDb provider, and this package bound to that provider's lock extensions.

// InvoiceSweeper.cs
using Kanject.Core.Annotations.Attributes.Recurring;
using Kanject.Core.Annotations.Attributes.Recurring.Enums;

namespace Billing.Jobs;

public sealed class InvoiceSweeper(IInvoiceStore invoices)
{
    [Recurring(1, RecurringRateUnit.Minutes)]
    [RecurringHosted(LeaseKey = "invoice-sweep", LeaseDurationSeconds = 120, CheckpointKey = "invoice-sweep")]
    public Task SweepAsync(CancellationToken cancellationToken) =>
        invoices.CloseExpiredAsync(cancellationToken);
}

public interface IInvoiceStore
{
    Task CloseExpiredAsync(CancellationToken cancellationToken);
}

// Program.cs
using Billing.Jobs;
using Kanject.Core.CacheDb.Provider.InMemory.Extensions;
using Kanject.Core.Recurring.Provider.CacheDb.Extensions;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Hosting;

var builder = Host.CreateApplicationBuilder(args);

builder.Services.AddSingleton<IInvoiceStore, SqlInvoiceStore>(); // your implementation

// 1. A CacheDb provider. In-memory works within one process only, so use it for local development.
builder.Services.AddInMemoryCacheDb("billing");

// 2. Leases and checkpoints over the registered ICacheDb. The delegates must come from
//    the SAME provider package that registered ICacheDb.
builder.Services.AddCacheDbRecurringDataProvider(options =>
{
    options.AcquireLock = DbLockExtensions.AcquireLockAsync;
    options.ReleaseLock = DbLockExtensions.ReleaseLockAsync;
});

// 3. Generated by [RecurringHosted] as Add{Type}{Method}Recurring, in the job's namespace.
builder.Services.AddInvoiceSweeperSweepAsyncRecurring();

await builder.Build().RunAsync();

On each tick, the generated service does three things:

1. It tries to take the invoice-sweep lease for 120 seconds. If another holder has it, the tick is skipped.
2. It runs SweepAsync, then reads the stored checkpoint and saves the next one (iteration + 1, the stored state kept).
3. It releases the lease, unless the 120 seconds have already run out.

Going to production

With the in-memory provider, leases coordinate only inside one process, and checkpoints are lost on restart. For more than one instance, register a durable provider instead: Kanject.Core.CacheDb.Provider.DynamoDb or Kanject.Core.CacheDb.Provider.S3Express. Bind AcquireLock / ReleaseLock to that provider's DbLockExtensions.AcquireLockAsync / ReleaseLockAsync, which have the same signatures and live in the provider's own …Extensions namespace. The job code and the AddCacheDbRecurringDataProvider call don't change.

The lock extensions only work with their own provider's ICacheDb. If you mix packages, for example in-memory delegates with a DynamoDB ICacheDb, every lease call fails with NotSupportedException.

Registration and options

AddCacheDbRecurringDataProvider(Action<CacheDbRecurringDataProviderOptions> configure) works like this:

• It runs configure immediately, then checks that both delegates are set. If either is missing, it throws InvalidOperationException right there, at registration.
• It registers the options instance and CacheDbRecurringDataProvider as the singleton IRecurringDataProvider, both with TryAdd. If you call it twice, the first registration wins.
• It returns the IServiceCollection so you can chain calls.
• It needs an ICacheDb registered in the container.

The constructor runs the same delegate check, which covers the direct-construction path. ValidateOnBuild can't catch a missing delegate: it checks that dependencies resolve and never runs constructors. That's why the registration method checks eagerly.

AcquireCacheDbLockAsync and ReleaseCacheDbLockAsync are ordinary delegates. You can also point them at your own lambdas, for example to wrap the provider call with logging. To create the provider without DI, call new CacheDbRecurringDataProvider(cacheDb, options).

How data is stored

• Keys.
  • Lease keys are built with cacheDb.FormatCacheKey(KeyPrefix, "lease", leaseKey).
  • Checkpoint keys are built with cacheDb.FormatCacheKey(KeyPrefix, "checkpoint", checkpointKey).
  • This keeps both kinds of record separate from your other cache entries, so a job can use the same name for LeaseKey and CheckpointKey.
  • The final format depends on the provider. With the in-memory provider registered as "billing", the lease above is stored as billing_kanject:recurring:lease:invoice-sweep.
• Acquiring a lease.
  • It calls AcquireLock(cacheDb, lockId, DateTime.UtcNow + duration, comment: null, data: holderData, overrideData: false).
  • On success, it returns Acquired = true with the formatted lock id as LeaseId.
  • If the provider throws DbLockConflictException, it returns Acquired = false with the conflict's LockId, Data (as CurrentHolderData) and LockDuration (as RemainingDuration).
  • Any other exception propagates.
  • overrideData: false stops a caller that loses the race from overwriting the current holder's data.
• Releasing a lease.
  • It calls ReleaseLock(cacheDb, lockId).
  • The CacheDb providers delete the lock by id without checking who holds it. If an iteration runs longer than the lease, another instance can take the lease, and a release would delete that instance's lease.
  • The generated hosted service guards against this: it skips the release once LeaseDurationSeconds has elapsed since it acquired the lease. Code that calls ReleaseLeaseAsync directly, such as a Lambda handler, needs the same check.
  • An overrunning iteration can still overlap the next holder's iteration, so size LeaseDurationSeconds above your worst-case iteration time.
• Checkpoints.
  • They are serialized with System.Text.Json using web defaults (camelCase).
  • They are written as a string through ICacheDb.CacheDataAsync, with an expiry of DateTime.UtcNow + (retention ?? CheckpointRetention).
  • A stored payload that can't be deserialized is read as null, and the next save overwrites it.
• Cancellation. Every method checks the token when it starts. The token isn't passed on to the cache or lock calls.

Public surface at a glance

Related packages

License

Licensed under the Kanject Code Libraries License Agreement (KCLLA); the full text ships in this package as LICENSE.md. Organizations whose trailing-twelve-month gross revenue and total funding raised are each below US$250,000 may use it at no cost under the Free Tier. At or above either threshold a commercial license is required — contact commercial@kanjectbusiness.com.