Sparkdo.Messaging.EntityFrameworkCore 3.0.0-preview.1

Sparkdo.Messaging.EntityFrameworkCore

把「幂等标记」「待发消息」「Saga 状态」落进宿主自己的 DbContext，于是 「业务工作单元 + Inbox + Outbox 同一次提交」这件事真的成立——一次 SaveChanges 就是一次原子落地。

<PackageReference Include="Sparkdo.Messaging.EntityFrameworkCore" Version="1.0.0" />
<PackageReference Include="Npgsql.EntityFrameworkCore.PostgreSQL" Version="..." />

1. 接线（一条链，可整段照抄）

// ① 宿主自己的上下文：业务实体与消息表必须共用一个 DbContext
public sealed class AppDbContext(DbContextOptions<AppDbContext> options) : DbContext(options)
{
    public DbSet<Order> Orders => Set<Order>();

    protected override void OnModelCreating(ModelBuilder modelBuilder)
    {
        base.OnModelCreating(modelBuilder);
        modelBuilder.ConfigureSparkdoMessaging();   // 三张消息表加进这个模型
    }
}

// ② 四行接线
builder.Services.AddDbContext<AppDbContext>(b => b.UseNpgsql(connectionString));

builder.Services
    .AddSparkdoMessaging(serializers, bus => bus
        .AddTransport(new RabbitMQTransport(transportOptions), TransportDefault.Default)
        // ③ 外层先开工作单元，内层再判幂等：提交成功＝标记落库
        .Use(WorkUnitMiddleware.Create(sp => sp.GetRequiredService<IWorkUnit>()))
        .Use(InboxMiddleware.Create(sp => sp.GetRequiredService<IMessageInbox>(), tenant: "orders", handlerId: "sparkdo.billing.order-placed-handler")))
    .AddMessagingEntityFrameworkCore<AppDbContext>()   // 只登记四个端口
    .AddMessagingOutboxDispatcher()                    // 显式起中继：只入队不投递的角色（Web 端）不调
    .AddMessagingPersistenceCleanup()                  // 显式起清理：保留期是部署决定，不调就不清
    .AddSparkdoMessagingHosting();

builder.Services.AddHealthChecks().AddMessagingOutboxHealthCheck();   // 可选：/health 里报中继在不在推进

宿主注册了多个上下文时用工厂重载挑一个：

builder.Services.AddMessagingEntityFrameworkCore(provider => provider.GetRequiredService<AppDbContext>());

处理器的登记必须是作用域形态，否则它拿不到本条投递的工作单元与上下文：

builder.AddScopedHandler<SettleOrderHandler, Settled>();      // 不是 AddHandler

三条顺序与重复调用的坑

2. 同事务的处理器长什么样

public sealed class SettleOrderHandler(
    AppDbContext db,
    WorkUnitAccessor workUnits,
    IMessageSerializerRegistry serializers,
    IMessageOutbox outbox) : IEventHandler<Settled>
{
    public async ValueTask HandleAsync(Settled message, CancellationToken ct)
    {
        var unit = workUnits.Current
            ?? throw new InvalidOperationException("这条投递没有工作单元：管道上没挂 WorkUnitMiddleware。");

        db.Orders.Add(new Order(message.OrderId, message.Amount));      // 业务写

        await outbox.EnqueueAsync(unit, BuildEnvelope(message), BuildBytes(message, serializers), ct);
    }
}

Envelope 怎么拼（原样发布与发件箱同一份形状）：

var envelope = new Envelope
{
    MessageId = MessageId.New(),
    MessageType = new MessageIdentity("urn:message:sparkdo.billing:settlement-notice"),
    ContentType = MessageContentType.Json,
    Destination = EndpointAddress.Parse("rabbitmq://topic/urn:message:sparkdo.billing:settlement-notice"),
    Source = EndpointAddress.Parse("rabbitmq://in/urn:message:sparkdo.billing:settled/sparkdo.billing.settle-order-handler"),
};

var buffer = new ArrayBufferWriter<byte>();
serializer.Write(buffer, notice);                    // 用登记过的那个读写器实例

outbox.EnqueueAsync(unit, envelope, buffer.WrittenMemory.ToArray(), ct);

3. 中继（发件箱 → 传输）

中继默认不跑：AddMessagingOutboxDispatcher() 才把它接进宿主生命周期（按实现类型去重，调几次都只起一台）。 接上之后宿主启动时它开始一轮轮取投，停机时收回去。要自己掌控节奏就解析 IOutboxDispatcher：

var dispatcher = provider.GetRequiredService<IOutboxDispatcher>();

var sent = await dispatcher.RunOnceAsync(ct);   // 跑一轮，返回发出几条
await dispatcher.RunAsync(ct);                  // 常驻循环：DispatcherInterval 一轮

观察面与限流

中继把「在不在推进」摊成三份可读的东西：端口上的快照、核心 Meter 上的指标、可选的健康检查。

健康检查判定只看快照：连续失败 ≥ DispatcherUnhealthyFailureThreshold（默认 5）→ 红；失败过但没到阈值 → 黄； 还没跑完第一轮 → 黄；其余 → 绿。data 里带积压数（现查 SentAtTicks 与 DeadAt 都为空的行数，2 秒超时）； 数不出来只记 backlog = "count-failed"，不把检查拖红——观测失败不是服务故障。

失败后的等待：第一轮失败等 DispatcherInterval，之后每轮翻倍，封顶 DispatcherMaxFailureBackoff（默认 30 秒）， 再按 DispatcherFailureJitterRatio（默认 0.25）抖动；成功一轮立即回到 DispatcherInterval。抖动是为了不让多台中继 在同一刻重试、把恢复中的目的地再压垮一次。

限速可选：DispatcherMaxMessagesPerSecond 默认 0 表示不限。设了之后单轮内按最小间隔（1/速率）整流，跨轮不累计； 它治的是「一批 64 条一口气砸向目的地」的尖峰，不是令牌桶。每目标并发不做：发送天然串行，要并发就得改认领归属，属于另一个档位。

死信处置与重驱

标死不是销毁：行留在发件箱表里，SentAt 仍为空、DeadAt 非空，payload 原样在库。

重驱是人工动作，不接后台循环；端口随 AddMessagingEntityFrameworkCore 一起登记，不调就一行都不动：

var redrive = provider.GetRequiredService<IMessagingDeadLetterRedrive>();

// 只放回这一条目的地的死信；一次最多 100 条（默认），按 Id 从旧到新。
var redriven = await redrive.RedriveAsync(new DeadLetterRedriveFilter
{
    Destination = "memory://in/urn:message:tests:order-placed",
});

// 想确认还剩多少，用上面的「查」再数一次。

DeadLetterRedriveFilter 三项取交集：Destination 与 Identity 留空即不限，MaxCount 默认 100、硬顶 1000。 每次重驱记一条 Info 日志，并在核心 Meter 上记一笔 messaging.outbox.redrive.count。

4. 清理与保留

三张表只进不出会一直长大，所以本包给「已经没用」的行一个出口。清理默认不跑： AddMessagingPersistenceCleanup() 才把它接进宿主生命周期（与中继同一体例，按实现类型去重）。

builder.Services
    .AddMessagingEntityFrameworkCore<AppDbContext>()
    .AddMessagingOutboxDispatcher()
    .AddMessagingPersistenceCleanup(options =>
    {
        options.InboxCompletedRetention = TimeSpan.FromDays(30);   // 大于本宿主的最大重投/重放窗口
        options.OutboxSentRetention = TimeSpan.FromDays(30);
    });

四类删除谓词：

不自动清的：

清理是幂等的、可以多实例同跑（重复删除只影响计数）；每批删除是原子的，取消时已删批次不回滚。 删除只记计数日志，不归档：需要审计的宿主在清理之前自己导出。

过期租约的完成行要人工收尾：把租约清掉，下一轮清理就会认它（示例是 Postgres；表名按宿主改）：

update sparkdo_messaging_saga
set "LeaseOwner" = null, "LeaseExpiresAt" = null
where "CompletedAtTicks" is not null and "LeaseExpiresAt" < now();

别直接删：清租约之后仍走标准谓词，保留期判断一条不少。SQLite 把 DateTimeOffset 存成文本，比较口径按实际列类型写。

升级与回填

StartedAtTicks / CompletedAtTicks / SentAtTicks 都是新增可空列：宿主的迁移生成并执行后，旧行为 null。 Saga 的 CompletedAtTicks 列与 (CompletedAtTicks) 索引也在同一次迁移里；它没有可回填的来源（记的是判据命中的那一刻）， 升级前写入的行一直为 null、永不自动清，等后续事件写回时按当时的终态补上。 想连旧行一起清就回填（示例是 Postgres 的 timestamptz 口径；621355968000000000 是 Unix 历元的 ticks 起点）：

update sparkdo_messaging_inbox
set "StartedAtTicks" = (621355968000000000 + extract(epoch from "StartedAt") * 10000000)::bigint,
    "CompletedAtTicks" = case when "CompletedAt" is null then null
        else (621355968000000000 + extract(epoch from "CompletedAt") * 10000000)::bigint end
where "StartedAtTicks" is null;

发件箱同形（把表与列换成 sparkdo_messaging_outbox / SentAt / SentAtTicks）。回填用应用侧脚本 （DateTimeOffset.UtcTicks）最稳——SQL 里的历元换算要按 provider 的时区口径对齐，写错不如不回填。

保留期必须大于宿主的最大重投/重放窗口：删掉完成标记之后，一条老消息的再次投递会被当成新消费重新处理。 至少一次语义允许重复，但业务未必受得了第二次副作用。

5. 三张表

默认表名就是 MessagingTables.Inbox/Outbox/Saga 三个常量。要换表名或 schema 不用覆盖 OnModelCreating， 在 ConfigureSparkdoMessaging 的委托里改（只认字母、数字、下划线），例如：

modelBuilder.ConfigureSparkdoMessaging(options =>
{
    options.Schema = "messaging";
    options.InboxTableName = "order_inbox";
    options.OutboxTableName = "order_outbox";
    options.SagaTableName = "order_saga";
});

schema 只在支持它的 provider 上出证（本仓是 Postgres 档）；SQLite 上只改表名，不要设 schema——Saga 的原生 插入会带上 schema 前缀，SQLite 不认。

逐实体还能再定制（ConfigureInboxEntity/ConfigureOutboxEntity/ConfigureSagaEntity，在默认映射之后跑， 例如补列宽、加宿主自己的索引），整体再补用 ConfigureModel。内置列不能删、不能改类型——Saga 的原生插入 从模型元数据取表名与列名，删了列会在中继启动后第一次建行时抛错（列只能追加，且须可空或有默认值）。

迁移怎么给

模型在宿主的 OnModelCreating 里（ConfigureSparkdoMessaging 把三张表加进去），所以标准 dotnet ef migrations add 会连这三张表一起生成——不用给本包单独建迁移项目。用例走的是 Database.EnsureCreated()：它只建默认 schema、不产生迁移历史，所以别拿它当交付路径； 测试里每次开一个新库而不是新 schema，正是这个原因。

6. 语义与边界（逐条实测）

7. provider 档位与出证

本包不绑 provider：三张表的映射里没有一句 provider 专有的东西，换 provider 只是把 UseSqlite(...) 换成 UseNpgsql(connectionString)。

出证方式：同一套语义用例换 provider 再跑一遍，开关是环境变量 SPARKDO_MESSAGING_POSTGRES （形如 Host=localhost;Port=5432;Database=sparkdo_messaging;Username=sparkdo;Password=sparkdo）； 没设就跳过，跳过文案里带着这条形参。每条用例一个临时库（用完 drop database ... with (force)）， 并行跑也不会互相看见——与 SQLite 档位一个临时文件同义。

# WSL 里起一台 Postgres（示例：系统包，端口让开 Windows 侧已占用的 5432）
sudo apt-get install -y postgresql
sudo sed -i 's/^port = 5432/port = 55432/' /etc/postgresql/*/main/postgresql.conf
sudo service postgresql restart
sudo -u postgres psql -c "create role sparkdo login password 'sparkdo' superuser createdb"

# Windows 侧跑用例
$env:SPARKDO_MESSAGING_POSTGRES='Host=localhost;Port=55432;Database=sparkdo_messaging;Username=sparkdo;Password=sparkdo'
dotnet test src/messaging/test/Sparkdo.Messaging.EntityFrameworkCore.Tests

读数：2026-10-19 在同一台库上复跑（WSL 内跑 Windows 产物）41/41、0 跳过。此前 2026-10-15 在 Postgres 18.6 上整工程 29/29（其中 4 条要真库）。端到端那几条把整摞一起跑： 总线 + Transport.Postgres + 本包——发布 → 序列化 → 库 → 占位取走 → 反序列化 → 工作单元与幂等中间件 → 处理器 → 结算，以及请求/响应（依赖临时订阅与回投地址，是这条链上最绕的一段）。

「传输过了契约」与「总线在这条传输上跑得通」是两件事：地址由核心拼、订阅按身份建、 幂等与工作单元按跳开作用域——这些都在传输边界之外，只有整摞跑过才算数。

从 Windows 连 WSL 里的库不稳；同一份产物在 WSL 里直接跑的跑法见 Sparkdo.Messaging.Transport.Postgres 的 README，把这个工程名换进去即可。 PostgresEvidenceTests 会显式断言 ProviderName == "postgres"——环境变量没生效却静默退回 SQLite 的话， 那套用例会全绿而证据是假的。