Skip to main content
Welcome. This site supports keyboard navigation and screen readers. Press ? at any time for keyboard shortcuts. Press [ to focus the sidebar, ] to focus the content. High-contrast themes are available via the toolbar.
serard@dev00:~/cv

The Data Model

"The saga state is not a cache — it's the source of truth. Model it like a domain."

Why Entity.Dsl?

The saga's state — which steps completed, which failed, what data each step produced, what conditions were evaluated — must survive process crashes, container restarts, and network partitions. This is not ephemeral progress tracking (that lives in Redis). This is the authoritative record of what happened.

Since we already have Entity.Dsl for attribute-driven EF Core, the distributed task's persistence model is itself an Entity.Dsl aggregate. The meta-DSL generates the meta-DSL's infrastructure. Turtles all the way down.

The Aggregate Root

[AggregateRoot("DistributedTaskInstance")]
[Table("DistributedTasks", Schema = "saga")]
[Timestampable]
[SoftDeletable]
public partial class DistributedTaskInstance
{
    [EntityId]
    public Guid Id { get; set; }

    [Property(Required = true, MaxLength = 200)]
    public string TaskName { get; set; } = "";

    [Property(Required = true, MaxLength = 200)]
    public string Queue { get; set; } = "";

    [Property]
    public DistributedTaskStatus Status { get; set; }

    [Property(MaxLength = 128)]
    public string? TraceId { get; set; }

    [Property]
    public string RequestPayload { get; set; } = "";  // serialized request

    [Property]
    public string? ResponsePayload { get; set; }       // serialized response

    [Property(MaxLength = 4000)]
    public string? ErrorMessage { get; set; }

    [Property]
    public int RetryCount { get; set; }

    [Property]
    public int MaxRetries { get; set; }

    [Property]
    public ListeningStrategy Listening { get; set; }

    [Property(MaxLength = 2000)]
    public string? WebhookUrl { get; set; }

    [Property]
    public DateTimeOffset? StartedAt { get; set; }

    [Property]
    public DateTimeOffset? CompletedAt { get; set; }

    [Property]
    public DateTimeOffset? CancelledAt { get; set; }

    [Composition]
    public List<SagaStepRecord> Steps { get; set; } = new();

    [Composition]
    public List<TaskAuditEntry> AuditLog { get; set; } = new();
}

public enum DistributedTaskStatus
{
    Submitted,
    Running,
    Completed,
    Failed,
    Cancelled,
    Compensating,
    CompensationFailed,
}

The [Composition] relationships mean:

  • Steps and audit entries are owned by the task (cascade delete)
  • They are accessed through the aggregate root, not via independent repositories
  • They form a single consistency boundary

Child Entities

SagaStepRecord — Per-Step State

[Entity("SagaStepRecord")]
[Table("SagaStepRecords", Schema = "saga")]
public partial class SagaStepRecord
{
    [EntityId]
    public Guid Id { get; set; }

    [Property(Required = true, MaxLength = 200)]
    public string StepName { get; set; } = "";

    [Property]
    public int Order { get; set; }

    [Property]
    public StepStatus Status { get; set; }

    [Property]
    public string? StepData { get; set; }  // serialized typed ValueObject

    [Property(MaxLength = 4000)]
    public string? ErrorMessage { get; set; }

    [Property(MaxLength = 4000)]
    public string? CompensationError { get; set; }

    [Property(MaxLength = 128)]
    public string? TraceId { get; set; }

    [Property]
    public int AttemptCount { get; set; }

    [Property]
    public DateTimeOffset? StartedAt { get; set; }

    [Property]
    public DateTimeOffset? CompletedAt { get; set; }

    [Composition]
    public List<StepConditionSnapshot> ConditionSnapshots { get; set; } = new();
}

public enum StepStatus
{
    Pending,
    Running,
    Completed,
    Failed,
    Skipped,
    Compensating,
    Compensated,
    CompensationFailed,
}

StepConditionSnapshot — Audit Trail for Conditions

[ValueObject("StepConditionSnapshot")]
public partial class StepConditionSnapshot
{
    [Property(Required = true, MaxLength = 200)]
    public string ConditionName { get; set; } = "";

    [Property(MaxLength = 1000)]
    public string Expression { get; set; } = "";

    [Property(MaxLength = 2000)]
    public string EvaluatedValue { get; set; } = "";

    [Property]
    public bool Passed { get; set; }

    [Property]
    public DateTimeOffset EvaluatedAt { get; set; }
}

As a [ValueObject], condition snapshots are mapped as EF Core owned types — stored inline in the step record's table or as a JSON column, depending on provider.

TaskAuditEntry — What Happened and When

[Entity("TaskAuditEntry")]
[Table("TaskAuditEntries", Schema = "saga")]
public partial class TaskAuditEntry
{
    [EntityId]
    public Guid Id { get; set; }

    [Property(Required = true, MaxLength = 100)]
    public string Action { get; set; } = "";

    [Property(MaxLength = 200)]
    public string? StepName { get; set; }

    [Property]
    public string? Details { get; set; }  // JSON with variable values

    [Property(MaxLength = 128)]
    public string? TraceId { get; set; }

    [Property]
    public DateTimeOffset Timestamp { get; set; }
}

Audit actions include: TaskSubmitted, StepStarted, StepCompleted, StepFailed, StepCompensating, StepCompensated, CompensationFailed, TaskCompleted, TaskFailed, TaskCancelled.

The Aggregate Boundary

Diagram

Everything inside the boundary is loaded and saved together. No independent repository for steps or audit entries — they are always accessed through the aggregate root.

Generated EF Core Code

Entity.Dsl generates the full persistence stack from the attributes above. Here's what comes out for the SagaStepRecord entity:

Configuration (Generation Gap)

// ── Generated: SagaStepRecordConfigurationBase.g.cs ── (Layer 1: always regenerated)
public abstract class SagaStepRecordConfigurationBase
{
    public virtual void ConfigureStepName(EntityTypeBuilder<SagaStepRecord> builder)
        => builder.Property(e => e.StepName).IsRequired().HasMaxLength(200);

    public virtual void ConfigureOrder(EntityTypeBuilder<SagaStepRecord> builder)
        => builder.Property(e => e.Order);

    public virtual void ConfigureStatus(EntityTypeBuilder<SagaStepRecord> builder)
        => builder.Property(e => e.Status).HasConversion<int>();

    public virtual void ConfigureStepData(EntityTypeBuilder<SagaStepRecord> builder)
        => builder.Property(e => e.StepData).HasColumnType("nvarchar(max)");

    public virtual void ConfigureErrorMessage(EntityTypeBuilder<SagaStepRecord> builder)
        => builder.Property(e => e.ErrorMessage).HasMaxLength(4000);

    public virtual void ConfigureCompensationError(EntityTypeBuilder<SagaStepRecord> builder)
        => builder.Property(e => e.CompensationError).HasMaxLength(4000);

    public virtual void ConfigureTraceId(EntityTypeBuilder<SagaStepRecord> builder)
        => builder.Property(e => e.TraceId).HasMaxLength(128);

    public virtual void ConfigureAttemptCount(EntityTypeBuilder<SagaStepRecord> builder)
        => builder.Property(e => e.AttemptCount);

    public virtual void ConfigureConditionSnapshots(EntityTypeBuilder<SagaStepRecord> builder)
        => builder.OwnsMany(e => e.ConditionSnapshots, cs =>
        {
            cs.Property(c => c.ConditionName).IsRequired().HasMaxLength(200);
            cs.Property(c => c.Expression).HasMaxLength(1000);
            cs.Property(c => c.EvaluatedValue).HasMaxLength(2000);
        });
}

// ── Generated: SagaStepRecordConfiguration.g.cs ── (Layer 2: partial stub)
public partial class SagaStepRecordConfiguration
    : SagaStepRecordConfigurationBase, IEntityTypeConfiguration<SagaStepRecord>
{
    public void Configure(EntityTypeBuilder<SagaStepRecord> builder)
    {
        builder.HasKey(e => e.Id);
        builder.Property(e => e.Id).ValueGeneratedOnAdd();
        ConfigureStepName(builder);
        ConfigureOrder(builder);
        ConfigureStatus(builder);
        ConfigureStepData(builder);
        ConfigureErrorMessage(builder);
        ConfigureCompensationError(builder);
        ConfigureTraceId(builder);
        ConfigureAttemptCount(builder);
        ConfigureConditionSnapshots(builder);
    }
}

The developer can override any Configure* method in a partial class (Layer 3) without touching generated code.

DbContext

// ── Generated: DistributedTaskDbContextBase.g.cs ──
public abstract class DistributedTaskDbContextBase : DbContext
{
    public DbSet<DistributedTaskInstance> Tasks => Set<DistributedTaskInstance>();

    protected override void OnModelCreating(ModelBuilder modelBuilder)
    {
        modelBuilder.Entity<DistributedTaskInstance>(entity =>
        {
            entity.HasMany(e => e.Steps)
                .WithOne()
                .OnDelete(DeleteBehavior.Cascade);  // Composition → Cascade
            entity.HasMany(e => e.AuditLog)
                .WithOne()
                .OnDelete(DeleteBehavior.Cascade);
        });

        modelBuilder.ApplyConfiguration(new DistributedTaskInstanceConfiguration());
        modelBuilder.ApplyConfiguration(new SagaStepRecordConfiguration());
        modelBuilder.ApplyConfiguration(new TaskAuditEntryConfiguration());
    }

    // Timestampable hook
    public override async Task<int> SaveChangesAsync(CancellationToken ct = default)
    {
        foreach (var entry in ChangeTracker.Entries<DistributedTaskInstance>())
        {
            if (entry.State == EntityState.Added)
                entry.Entity.CreatedAt = DateTimeOffset.UtcNow;
            if (entry.State == EntityState.Modified)
                entry.Entity.UpdatedAt = DateTimeOffset.UtcNow;
        }
        return await base.SaveChangesAsync(ct);
    }
}

Repository and Unit of Work

// ── Generated: IDistributedTaskInstanceRepository.g.cs ──
public interface IDistributedTaskInstanceRepository
    : IRepository<DistributedTaskInstance>
{
    Task<DistributedTaskInstance?> FindByIdAsync(Guid id);
}

// ── Generated: DistributedTaskInstanceRepository.g.cs ──
[Injectable(Scope = Scope.Scoped, As = typeof(IDistributedTaskInstanceRepository))]
public partial class DistributedTaskInstanceRepository
    : RepositoryBase<DistributedTaskInstance>, IDistributedTaskInstanceRepository
{
    public DistributedTaskInstanceRepository(DistributedTaskDbContext context)
        : base(context) { }

    public async Task<DistributedTaskInstance?> FindByIdAsync(Guid id)
        => await Query
            .Include(t => t.Steps)
                .ThenInclude(s => s.ConditionSnapshots)
            .Include(t => t.AuditLog)
            .FirstOrDefaultAsync(t => t.Id == id);
}

// ── Generated: IDistributedTaskUnitOfWork.g.cs ──
public interface IDistributedTaskUnitOfWork : IUnitOfWork
{
    IDistributedTaskInstanceRepository Tasks { get; }
    Task<int> SaveChangesAsync(CancellationToken ct = default);
}

// ── Generated: DistributedTaskUnitOfWork.g.cs ──
[Injectable(Scope = Scope.Scoped, As = typeof(IDistributedTaskUnitOfWork))]
public partial class DistributedTaskUnitOfWork : IDistributedTaskUnitOfWork
{
    private readonly DistributedTaskDbContext _context;
    private IDistributedTaskInstanceRepository? _tasks;

    public DistributedTaskUnitOfWork(DistributedTaskDbContext context)
        => _context = context;

    public IDistributedTaskInstanceRepository Tasks
        => _tasks ??= new DistributedTaskInstanceRepository(_context);

    public Task<int> SaveChangesAsync(CancellationToken ct = default)
        => _context.SaveChangesAsync(ct);
}

Typed Step Data — How It Persists

Each step's typed ValueObject (e.g., UploadSourceFilesStepData) is serialized into the SagaStepRecord.StepData column using the configured serializer (JSON by default). The mapping:

Developer writes:     stepData.UploadedKeys = keys;
Generator serializes: record.StepData = _serializer.Serialize(stepData);
DB stores:            {"UploadedKeys":["uploads/abc/file1.pdf","uploads/abc/file2.pdf"],"Bucket":"incoming-files"}
Next step reads:      var prev = context.GetStepData<UploadSourceFilesStepData>();
Generator deserializes: _serializer.Deserialize<UploadSourceFilesStepData>(record.StepData);

The developer API is fully typed. The persistence is transparent.

Automatic Archival

Completed and failed tasks accumulate. The ArchiveAfterDays property on [DistributedTask] generates a hosted service that periodically moves old records to an archive table:

// ── Generated: DistributedTaskArchiveService.g.cs ──
public class DistributedTaskArchiveService : BackgroundService
{
    private readonly IServiceScopeFactory _scopeFactory;
    private readonly ILogger<DistributedTaskArchiveService> _logger;
    private readonly TimeSpan _interval = TimeSpan.FromHours(6);

    protected override async Task ExecuteAsync(CancellationToken ct)
    {
        while (!ct.IsCancellationRequested)
        {
            await Task.Delay(_interval, ct);

            using var scope = _scopeFactory.CreateScope();
            var context = scope.ServiceProvider
                .GetRequiredService<DistributedTaskDbContext>();

            var cutoff = DateTimeOffset.UtcNow.AddDays(-30); // from ArchiveAfterDays

            // Move to archive table (same schema, no FK, no heavy indexes)
            var archived = await context.Database.ExecuteSqlRawAsync(@"
                INSERT INTO saga.DistributedTasks_Archive
                SELECT * FROM saga.DistributedTasks
                WHERE (Status = {0} OR Status = {1})
                  AND CompletedAt < {2};

                INSERT INTO saga.SagaStepRecords_Archive
                SELECT s.* FROM saga.SagaStepRecords s
                INNER JOIN saga.DistributedTasks t ON s.DistributedTaskInstanceId = t.Id
                WHERE (t.Status = {0} OR t.Status = {1})
                  AND t.CompletedAt < {2};

                INSERT INTO saga.TaskAuditEntries_Archive
                SELECT a.* FROM saga.TaskAuditEntries a
                INNER JOIN saga.DistributedTasks t ON a.DistributedTaskInstanceId = t.Id
                WHERE (t.Status = {0} OR t.Status = {1})
                  AND t.CompletedAt < {2};

                DELETE FROM saga.DistributedTasks
                WHERE (Status = {0} OR Status = {1})
                  AND CompletedAt < {2};",
                (int)DistributedTaskStatus.Completed,
                (int)DistributedTaskStatus.Failed,
                cutoff,
                ct);

            _logger.LogInformation("Archived {Count} old tasks", archived);
        }
    }
}

The archive tables are created via a migration generated alongside the main tables.

Database Schema

Diagram

Summary

What Lines Generated by
Aggregate root + 3 entities ~80 Developer (attributes)
EF Core configurations ~120 Entity.Dsl
DbContext ~40 Entity.Dsl
Repository + interface ~30 Entity.Dsl
Unit of Work + interface ~30 Entity.Dsl
Archive service ~40 DistributedTask.Dsl
Total generated ~260

The developer writes ~80 lines of attributed domain classes. Entity.Dsl generates ~260 lines of EF Core infrastructure. The archive service is generated by the DistributedTask source generator as an additional cross-cutting concern.

What's Next

Part VI catalogs every generated artifact — from the orchestrator kernel through the API controller to the DI registration — and maps out the complete Generation Gap: who writes what.