2 lock each message for update

generate_data.sql

@@ -0,0 +1,9 @@
+INSERT INTO outbox_messages
+(topic, "key", "type", payload, headers)
+SELECT
+    'topic-' || n.v%2, 
+	n.v, 
+	'type',
+	'{"byte250":"0123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789"}', 
+	'{"traceparent": "00-9588ce39e007adce6d63a55779ea22e1-500119597384256a-00"}'
+FROM generate_series(1,1000000) AS n(v);

src/Outbox.WebApi/BackgroundServices/OutboxBackgroundService.cs

@@ -1,10 +1,10 @@
 using System.Text;
 using Confluent.Kafka;
-using Medallion.Threading;
 using Microsoft.EntityFrameworkCore;
 using Microsoft.Extensions.Options;
 using Outbox.Configurations;
 using Outbox.Entities;
+using Outbox.Extensions;
 
 namespace Outbox.WebApi.BackgroundServices;
 
@@ -13,20 +13,17 @@ public class OutboxBackgroundService : BackgroundService, IOutboxMessagesProcess
     private readonly IServiceProvider _serviceProvider;
     private readonly IOptions<OutboxConfiguration> _outboxOptions;
     private readonly ILogger<OutboxBackgroundService> _logger;
-    private readonly IDistributedLockProvider _distributedLockProvider;
 
     private readonly AutoResetEvent _autoResetEvent = new(false);
 
     public OutboxBackgroundService(
         IServiceProvider serviceProvider, 
         IOptions<OutboxConfiguration> outboxOptions,
-        ILogger<OutboxBackgroundService> logger,
-        IDistributedLockProvider distributedLockProvider)
+        ILogger<OutboxBackgroundService> logger)
     {
         _serviceProvider = serviceProvider;
         _outboxOptions = outboxOptions;
         _logger = logger;
-        _distributedLockProvider = distributedLockProvider;
     }
 
     protected override async Task ExecuteAsync(CancellationToken stoppingToken)
@@ -52,25 +49,25 @@ protected override async Task ExecuteAsync(CancellationToken stoppingToken)
 
     private async Task<int> ProcessMessagesAsync(AppDbContext dbContext, CancellationToken cancellationToken)
     {
-        //pg_bouncer must be in session mode
-        await using var advisoryLock = await _distributedLockProvider.AcquireLockAsync("Outbox", cancellationToken: cancellationToken);
-
-        var offset = await dbContext.OutboxOffsets
-            .FirstAsync(cancellationToken);
+        var transaction = await dbContext.Database.BeginTransactionAsync(cancellationToken);
 
         var outboxMessages = await dbContext.OutboxMessages
             .AsNoTracking()
-            .Where(x => x.Id > offset.LastProcessedId)
             .OrderBy(x => x.Id)
             .Take(_outboxOptions.Value.BatchSize)
+            .ForUpdateSkipLocked()
             .ToArrayAsync(cancellationToken);
 
         if (!outboxMessages.Any()) return 0;
 
         await ProcessOutboxMessagesAsync(outboxMessages, cancellationToken);
 
-        offset.LastProcessedId = outboxMessages.Last().Id;
-        await dbContext.SaveChangesAsync(cancellationToken);
+        var messageIds = outboxMessages.Select(x => x.Id).ToArray();
+        await dbContext.OutboxMessages
+            .Where(x => messageIds.Contains(x.Id))
+            .ExecuteDeleteAsync(cancellationToken);
+
+        await transaction.CommitAsync(cancellationToken);
 
         return outboxMessages.Length;
     }

src/Outbox.WebApi/Outbox.WebApi.csproj

@@ -9,7 +9,6 @@
 
     <ItemGroup>
         <PackageReference Include="Confluent.Kafka.DependencyInjection" Version="3.2.0" />
-        <PackageReference Include="DistributedLock.Postgres" Version="1.3.0" />
         <PackageReference Include="EFCore.MigrationExtensions.PostgreSQL" Version="9.0.1" />
         <PackageReference Include="Microsoft.AspNetCore.OpenApi" Version="9.0.2"/>
         <PackageReference Include="Microsoft.EntityFrameworkCore.Design" Version="9.0.3">

src/Outbox.WebApi/Program.cs

@@ -1,8 +1,6 @@
 using System.Diagnostics;
 using Confluent.Kafka;
 using EFCore.MigrationExtensions.PostgreSQL;
-using Medallion.Threading;
-using Medallion.Threading.Postgres;
 using Microsoft.EntityFrameworkCore;
 using Npgsql;
 using Outbox;
@@ -30,9 +28,6 @@
 
     optionsBuilder.UseSqlObjects();
 });
-builder.Services.AddSingleton<IDistributedLockProvider>(_ => 
-    new PostgresDistributedSynchronizationProvider(builder.Configuration.GetConnectionString("Outbox")!));
-
 
 builder.Services.AddKafkaClient();
 
@@ -44,10 +39,7 @@
 builder.Services.AddSingleton<IOutboxMessagesProcessor>(sp => sp.GetRequiredService<OutboxBackgroundService>());
 
 builder.Services.AddOpenTelemetry()
-    .WithTracing(bld =>
-    {
-        //bld.AddSource(ActivitySources.OutboxSource);
-    });
+    .WithTracing();
 
 var app = builder.Build();