From 4468ee21c100016a810bc7a1a90ff3223687abfa Mon Sep 17 00:00:00 2001 From: George Date: Fri, 24 Jul 2026 19:09:23 +0200 Subject: [PATCH 01/22] Preserve core options when adding worker --- ...EventsWorkerServiceCollectionExtensions.cs | 12 ------ .../TinyEventsServiceCollectionExtensions.cs | 19 ++++++++- .../TinyEventsWorkerTests.cs | 40 ++++++++++++++++++- 3 files changed, 56 insertions(+), 15 deletions(-) diff --git a/src/TinyEvents.Worker/DependencyInjection/TinyEventsWorkerServiceCollectionExtensions.cs b/src/TinyEvents.Worker/DependencyInjection/TinyEventsWorkerServiceCollectionExtensions.cs index eb5d3d0..d54f47f 100644 --- a/src/TinyEvents.Worker/DependencyInjection/TinyEventsWorkerServiceCollectionExtensions.cs +++ b/src/TinyEvents.Worker/DependencyInjection/TinyEventsWorkerServiceCollectionExtensions.cs @@ -40,17 +40,5 @@ private static void ConfigureTinyEventsForWorker( options.BatchSize = workerOptions.BatchSize; options.ClaimTimeout = workerOptions.ClaimTimeout; }); - - services.Replace(ServiceDescriptor.Singleton(CreateCoreOptions(workerOptions))); - } - - private static TinyEventsOptions CreateCoreOptions(TinyEventsWorkerOptions workerOptions) - { - return new TinyEventsOptions - { - WorkerId = workerOptions.WorkerId, - BatchSize = workerOptions.BatchSize, - ClaimTimeout = workerOptions.ClaimTimeout - }; } } diff --git a/src/TinyEvents/DependencyInjection/TinyEventsServiceCollectionExtensions.cs b/src/TinyEvents/DependencyInjection/TinyEventsServiceCollectionExtensions.cs index 6b3bb9c..60288bb 100644 --- a/src/TinyEvents/DependencyInjection/TinyEventsServiceCollectionExtensions.cs +++ b/src/TinyEvents/DependencyInjection/TinyEventsServiceCollectionExtensions.cs @@ -24,12 +24,29 @@ private static void RegisterCoreServices( Action? configure) { services.TryAddSingleton(TimeProvider.System); - services.TryAddSingleton(CreateOptions(configure)); + ConfigureOptions(services, configure); services.TryAddSingleton(); services.TryAddScoped(); services.TryAddScoped(); } + private static void ConfigureOptions( + IServiceCollection services, + Action? configure) + { + var existingOptions = services + .LastOrDefault(descriptor => descriptor.ServiceType == typeof(TinyEventsOptions)) + ?.ImplementationInstance as TinyEventsOptions; + + if (existingOptions is not null) + { + configure?.Invoke(existingOptions); + return; + } + + services.TryAddSingleton(CreateOptions(configure)); + } + private static TinyEventsOptions CreateOptions(Action? configure) { var options = new TinyEventsOptions(); diff --git a/tests/TinyEvents.Worker.Tests/TinyEventsWorkerTests.cs b/tests/TinyEvents.Worker.Tests/TinyEventsWorkerTests.cs index c66777c..b811619 100644 --- a/tests/TinyEvents.Worker.Tests/TinyEventsWorkerTests.cs +++ b/tests/TinyEvents.Worker.Tests/TinyEventsWorkerTests.cs @@ -61,7 +61,11 @@ public void Worker_options_apply_when_worker_registered_before_core() options.BatchSize = 3; options.ClaimTimeout = TimeSpan.FromSeconds(11); }); - services.UseTinyEvents(); + services.UseTinyEvents(options => + { + options.MaxAttempts = 9; + options.RetryDelay = TimeSpan.FromSeconds(13); + }); using var provider = services.BuildServiceProvider(); var coreOptions = provider.GetRequiredService(); @@ -69,6 +73,8 @@ public void Worker_options_apply_when_worker_registered_before_core() Assert.Equal("worker-before", coreOptions.WorkerId); Assert.Equal(3, coreOptions.BatchSize); Assert.Equal(TimeSpan.FromSeconds(11), coreOptions.ClaimTimeout); + Assert.Equal(9, coreOptions.MaxAttempts); + Assert.Equal(TimeSpan.FromSeconds(13), coreOptions.RetryDelay); } [Fact] @@ -76,7 +82,11 @@ public void Worker_options_apply_when_worker_registered_after_core() { var services = new ServiceCollection(); - services.UseTinyEvents(); + services.UseTinyEvents(options => + { + options.MaxAttempts = 9; + options.RetryDelay = TimeSpan.FromSeconds(13); + }); services.AddTinyEventsWorker(options => { options.WorkerId = "worker-after"; @@ -90,6 +100,32 @@ public void Worker_options_apply_when_worker_registered_after_core() Assert.Equal("worker-after", coreOptions.WorkerId); Assert.Equal(4, coreOptions.BatchSize); Assert.Equal(TimeSpan.FromSeconds(12), coreOptions.ClaimTimeout); + Assert.Equal(9, coreOptions.MaxAttempts); + Assert.Equal(TimeSpan.FromSeconds(13), coreOptions.RetryDelay); + } + + [Fact] + public void Add_tiny_events_worker_preserves_core_retry_configuration() + { + var services = new ServiceCollection(); + + services.UseTinyEvents(options => + { + options.MaxAttempts = 11; + options.RetryDelay = TimeSpan.FromSeconds(17); + }); + services.AddTinyEventsWorker(options => + { + options.WorkerId = "worker-1"; + options.BatchSize = 12; + options.ClaimTimeout = TimeSpan.FromSeconds(45); + }); + + using var provider = services.BuildServiceProvider(); + var coreOptions = provider.GetRequiredService(); + + Assert.Equal(11, coreOptions.MaxAttempts); + Assert.Equal(TimeSpan.FromSeconds(17), coreOptions.RetryDelay); } [Fact] From eaa992c4d08e4fc90e5bf9fd748a4e2e4877766e Mon Sep 17 00:00:00 2001 From: George Date: Fri, 24 Jul 2026 19:14:05 +0200 Subject: [PATCH 02/22] Validate TinyEvents option values --- .../TinyEventsWorkerOptions.cs | 57 +++++++++++++- src/TinyEvents/Options/TinyEventsOptions.cs | 76 ++++++++++++++++++- .../Processing/TinyOutboxProcessorTests.cs | 41 ++++++++++ .../TinyEventsWorkerTests.cs | 44 +++++++++++ 4 files changed, 211 insertions(+), 7 deletions(-) diff --git a/src/TinyEvents.Worker/TinyEventsWorkerOptions.cs b/src/TinyEvents.Worker/TinyEventsWorkerOptions.cs index 854c8ee..eff5910 100644 --- a/src/TinyEvents.Worker/TinyEventsWorkerOptions.cs +++ b/src/TinyEvents.Worker/TinyEventsWorkerOptions.cs @@ -2,6 +2,9 @@ namespace TinyEvents.Worker; public sealed class TinyEventsWorkerOptions { + private int batchSize = 50; + private TimeSpan pollingInterval = TimeSpan.FromSeconds(5); + private TimeSpan claimTimeout = TimeSpan.FromMinutes(5); private string? workerId; public string? WorkerId @@ -22,9 +25,57 @@ public string? WorkerId } } - public int BatchSize { get; set; } = 50; + public int BatchSize + { + get + { + return batchSize; + } + + set + { + if (value <= 0) + { + throw new ArgumentOutOfRangeException(nameof(value), "Batch size must be greater than zero."); + } + + batchSize = value; + } + } + + public TimeSpan PollingInterval + { + get + { + return pollingInterval; + } + + set + { + if (value < TimeSpan.Zero) + { + throw new ArgumentOutOfRangeException(nameof(value), "Polling interval cannot be negative."); + } + + pollingInterval = value; + } + } + + public TimeSpan ClaimTimeout + { + get + { + return claimTimeout; + } - public TimeSpan PollingInterval { get; set; } = TimeSpan.FromSeconds(5); + set + { + if (value <= TimeSpan.Zero) + { + throw new ArgumentOutOfRangeException(nameof(value), "Claim timeout must be greater than zero."); + } - public TimeSpan ClaimTimeout { get; set; } = TimeSpan.FromMinutes(5); + claimTimeout = value; + } + } } diff --git a/src/TinyEvents/Options/TinyEventsOptions.cs b/src/TinyEvents/Options/TinyEventsOptions.cs index 2c647ca..2384d0e 100644 --- a/src/TinyEvents/Options/TinyEventsOptions.cs +++ b/src/TinyEvents/Options/TinyEventsOptions.cs @@ -2,15 +2,83 @@ namespace TinyEvents; public sealed class TinyEventsOptions { + private int batchSize = 50; + private int maxAttempts = 5; + private TimeSpan retryDelay = TimeSpan.FromSeconds(30); + private TimeSpan claimTimeout = TimeSpan.FromMinutes(5); private string? workerId; - public int BatchSize { get; set; } = 50; + public int BatchSize + { + get + { + return batchSize; + } + + set + { + if (value <= 0) + { + throw new ArgumentOutOfRangeException(nameof(value), "Batch size must be greater than zero."); + } + + batchSize = value; + } + } + + public int MaxAttempts + { + get + { + return maxAttempts; + } + + set + { + if (value <= 0) + { + throw new ArgumentOutOfRangeException(nameof(value), "Maximum attempts must be greater than zero."); + } + + maxAttempts = value; + } + } + + public TimeSpan RetryDelay + { + get + { + return retryDelay; + } + + set + { + if (value < TimeSpan.Zero) + { + throw new ArgumentOutOfRangeException(nameof(value), "Retry delay cannot be negative."); + } + + retryDelay = value; + } + } - public int MaxAttempts { get; set; } = 5; + public TimeSpan ClaimTimeout + { + get + { + return claimTimeout; + } - public TimeSpan RetryDelay { get; set; } = TimeSpan.FromSeconds(30); + set + { + if (value <= TimeSpan.Zero) + { + throw new ArgumentOutOfRangeException(nameof(value), "Claim timeout must be greater than zero."); + } - public TimeSpan ClaimTimeout { get; set; } = TimeSpan.FromMinutes(5); + claimTimeout = value; + } + } public string? WorkerId { diff --git a/tests/TinyEvents.Tests/Processing/TinyOutboxProcessorTests.cs b/tests/TinyEvents.Tests/Processing/TinyOutboxProcessorTests.cs index 2af4849..0bcadbd 100644 --- a/tests/TinyEvents.Tests/Processing/TinyOutboxProcessorTests.cs +++ b/tests/TinyEvents.Tests/Processing/TinyOutboxProcessorTests.cs @@ -80,6 +80,47 @@ public void Options_reject_empty_configured_worker_id() Assert.Throws(() => new TinyEventsOptions { WorkerId = " " }); } + [Theory] + [InlineData(0)] + [InlineData(-1)] + public void Options_reject_non_positive_batch_size(int batchSize) + { + Assert.Throws(() => new TinyEventsOptions { BatchSize = batchSize }); + } + + [Theory] + [InlineData(0)] + [InlineData(-1)] + public void Options_reject_non_positive_max_attempts(int maxAttempts) + { + Assert.Throws(() => new TinyEventsOptions { MaxAttempts = maxAttempts }); + } + + [Fact] + public void Options_reject_negative_retry_delay() + { + Assert.Throws(() => new TinyEventsOptions { RetryDelay = TimeSpan.FromMilliseconds(-1) }); + } + + [Theory] + [InlineData(0)] + [InlineData(-1)] + public void Options_reject_non_positive_claim_timeout(int milliseconds) + { + Assert.Throws(() => new TinyEventsOptions { ClaimTimeout = TimeSpan.FromMilliseconds(milliseconds) }); + } + + [Fact] + public void Options_allow_zero_retry_delay() + { + var options = new TinyEventsOptions + { + RetryDelay = TimeSpan.Zero + }; + + Assert.Equal(TimeSpan.Zero, options.RetryDelay); + } + [Fact] public async Task Process_pending_async_invokes_matching_consumer() { diff --git a/tests/TinyEvents.Worker.Tests/TinyEventsWorkerTests.cs b/tests/TinyEvents.Worker.Tests/TinyEventsWorkerTests.cs index b811619..5a93dac 100644 --- a/tests/TinyEvents.Worker.Tests/TinyEventsWorkerTests.cs +++ b/tests/TinyEvents.Worker.Tests/TinyEventsWorkerTests.cs @@ -137,6 +137,50 @@ public void Add_tiny_events_worker_rejects_empty_configured_worker_id() () => services.AddTinyEventsWorker(options => options.WorkerId = " ")); } + [Theory] + [InlineData(0)] + [InlineData(-1)] + public void Add_tiny_events_worker_rejects_non_positive_batch_size(int batchSize) + { + var services = new ServiceCollection(); + + Assert.Throws( + () => services.AddTinyEventsWorker(options => options.BatchSize = batchSize)); + } + + [Fact] + public void Add_tiny_events_worker_rejects_negative_polling_interval() + { + var services = new ServiceCollection(); + + Assert.Throws( + () => services.AddTinyEventsWorker(options => options.PollingInterval = TimeSpan.FromMilliseconds(-1))); + } + + [Theory] + [InlineData(0)] + [InlineData(-1)] + public void Add_tiny_events_worker_rejects_non_positive_claim_timeout(int milliseconds) + { + var services = new ServiceCollection(); + + Assert.Throws( + () => services.AddTinyEventsWorker(options => options.ClaimTimeout = TimeSpan.FromMilliseconds(milliseconds))); + } + + [Fact] + public void Add_tiny_events_worker_allows_zero_polling_interval() + { + var services = new ServiceCollection(); + + services.AddTinyEventsWorker(options => options.PollingInterval = TimeSpan.Zero); + + using var provider = services.BuildServiceProvider(); + var workerOptions = provider.GetRequiredService(); + + Assert.Equal(TimeSpan.Zero, workerOptions.PollingInterval); + } + [Fact] public void Add_tiny_events_worker_leaves_worker_id_unset_when_not_configured() { From d3d5c2dd9c0f4c2e34d7608b9a7281d76975182e Mon Sep 17 00:00:00 2001 From: George Date: Fri, 24 Jul 2026 19:18:45 +0200 Subject: [PATCH 03/22] Stop processing when canceled after claim --- .../Processing/TinyOutboxProcessor.cs | 1 + .../Processing/TinyOutboxProcessorTests.cs | 72 +++++++++++++++++++ 2 files changed, 73 insertions(+) diff --git a/src/TinyEvents/Processing/TinyOutboxProcessor.cs b/src/TinyEvents/Processing/TinyOutboxProcessor.cs index 37c7f71..ce6ffd2 100644 --- a/src/TinyEvents/Processing/TinyOutboxProcessor.cs +++ b/src/TinyEvents/Processing/TinyOutboxProcessor.cs @@ -78,6 +78,7 @@ public async ValueTask ProcessPendingAsync(CancellationToken cancellationToken = foreach (var message in messages) { + cancellationToken.ThrowIfCancellationRequested(); await ProcessMessageAsync(message, workerId, cancellationToken); } } diff --git a/tests/TinyEvents.Tests/Processing/TinyOutboxProcessorTests.cs b/tests/TinyEvents.Tests/Processing/TinyOutboxProcessorTests.cs index 0bcadbd..7d8831e 100644 --- a/tests/TinyEvents.Tests/Processing/TinyOutboxProcessorTests.cs +++ b/tests/TinyEvents.Tests/Processing/TinyOutboxProcessorTests.cs @@ -279,6 +279,23 @@ public async Task Process_pending_async_passes_cancellation_token_to_consumer() Assert.Equal(cancellation.Token, CancellationRecordingConsumer.CancellationToken); } + [Fact] + public async Task Process_pending_async_stops_before_processing_when_canceled_after_claim() + { + RecordingConsumer.Consumed.Clear(); + using var cancellation = new CancellationTokenSource(); + var store = new CancelAfterClaimStore( + cancellation, + NewProcessingMessage(new UserCreated(Guid.NewGuid(), "user@example.com"))); + var processor = BuildProcessor(store); + + await Assert.ThrowsAsync( + async () => await processor.ProcessPendingAsync(cancellation.Token)); + + Assert.Empty(RecordingConsumer.Consumed); + Assert.Equal(0, store.MarkFailedCount); + } + private static ITinyOutboxProcessor BuildProcessor( ITinyOutboxStore store, bool includeSecondConsumer = false, @@ -503,4 +520,59 @@ public ValueTask MarkFailedAsync( return ValueTask.CompletedTask; } } + + private sealed class CancelAfterClaimStore : ITinyOutboxStore + { + private readonly CancellationTokenSource cancellation; + private readonly IReadOnlyList claimedMessages; + + public CancelAfterClaimStore( + CancellationTokenSource cancellation, + params TinyOutboxMessage[] claimedMessages) + { + this.cancellation = cancellation; + this.claimedMessages = claimedMessages; + } + + public int MarkFailedCount { get; private set; } + + public ValueTask AddAsync( + TinyOutboxMessage message, + CancellationToken cancellationToken) + { + throw new NotSupportedException(); + } + + public ValueTask> ClaimPendingAsync( + int maxCount, + string workerId, + DateTimeOffset now, + TimeSpan claimTimeout, + CancellationToken cancellationToken) + { + cancellation.Cancel(); + return ValueTask.FromResult(claimedMessages); + } + + public ValueTask MarkProcessedAsync( + Guid messageId, + string workerId, + DateTimeOffset processedAtUtc, + CancellationToken cancellationToken) + { + throw new InvalidOperationException("Canceled processing should not mark messages as processed."); + } + + public ValueTask MarkFailedAsync( + Guid messageId, + string workerId, + string error, + int attemptCount, + DateTimeOffset? nextAttemptAtUtc, + CancellationToken cancellationToken) + { + MarkFailedCount++; + return ValueTask.CompletedTask; + } + } } From 33593234635041cd7185cf2bd4940f9d9837a7d2 Mon Sep 17 00:00:00 2001 From: George Date: Fri, 24 Jul 2026 19:33:09 +0200 Subject: [PATCH 04/22] Add generated event dispatcher contract --- .../Registry/ITinyEventDispatcher.cs | 13 ++ .../Registry/TinyEventDispatcher.cs | 39 ++++++ .../Registry/TinyEventDispatcherTests.cs | 113 ++++++++++++++++++ 3 files changed, 165 insertions(+) create mode 100644 src/TinyEvents/Registry/ITinyEventDispatcher.cs create mode 100644 src/TinyEvents/Registry/TinyEventDispatcher.cs create mode 100644 tests/TinyEvents.Tests/Registry/TinyEventDispatcherTests.cs diff --git a/src/TinyEvents/Registry/ITinyEventDispatcher.cs b/src/TinyEvents/Registry/ITinyEventDispatcher.cs new file mode 100644 index 0000000..50d1fe2 --- /dev/null +++ b/src/TinyEvents/Registry/ITinyEventDispatcher.cs @@ -0,0 +1,13 @@ +namespace TinyEvents; + +public interface ITinyEventDispatcher +{ + string EventTypeName { get; } + + Type EventType { get; } + + ValueTask DispatchAsync( + IServiceProvider serviceProvider, + object eventInstance, + CancellationToken cancellationToken); +} diff --git a/src/TinyEvents/Registry/TinyEventDispatcher.cs b/src/TinyEvents/Registry/TinyEventDispatcher.cs new file mode 100644 index 0000000..9c59e61 --- /dev/null +++ b/src/TinyEvents/Registry/TinyEventDispatcher.cs @@ -0,0 +1,39 @@ +using Microsoft.Extensions.DependencyInjection; + +namespace TinyEvents; + +public sealed class TinyEventDispatcher : ITinyEventDispatcher +{ + public TinyEventDispatcher(string eventTypeName) + { + if (string.IsNullOrWhiteSpace(eventTypeName)) + { + throw new ArgumentException("Event type name is required.", nameof(eventTypeName)); + } + + EventTypeName = eventTypeName; + } + + public string EventTypeName { get; } + + public Type EventType => typeof(TEvent); + + public async ValueTask DispatchAsync( + IServiceProvider serviceProvider, + object eventInstance, + CancellationToken cancellationToken) + { + if (serviceProvider is null) + { + throw new ArgumentNullException(nameof(serviceProvider)); + } + + var typedEvent = (TEvent)eventInstance; + var consumers = serviceProvider.GetServices>(); + + foreach (var consumer in consumers) + { + await consumer.ConsumeAsync(typedEvent, cancellationToken); + } + } +} diff --git a/tests/TinyEvents.Tests/Registry/TinyEventDispatcherTests.cs b/tests/TinyEvents.Tests/Registry/TinyEventDispatcherTests.cs new file mode 100644 index 0000000..fd4788a --- /dev/null +++ b/tests/TinyEvents.Tests/Registry/TinyEventDispatcherTests.cs @@ -0,0 +1,113 @@ +using Microsoft.Extensions.DependencyInjection; +using Xunit; + +namespace TinyEvents.Tests; + +public sealed class TinyEventDispatcherTests +{ + [Fact] + public void Constructor_rejects_empty_event_type_name() + { + Assert.Throws(() => new TinyEventDispatcher(" ")); + } + + [Fact] + public void Event_type_returns_generic_event_type() + { + var dispatcher = new TinyEventDispatcher("test-event"); + + Assert.Equal(typeof(UserCreated), dispatcher.EventType); + } + + [Fact] + public async Task Dispatch_async_invokes_matching_consumer() + { + RecordingConsumer.Consumed.Clear(); + var services = new ServiceCollection(); + services.AddSingleton, RecordingConsumer>(); + using var provider = services.BuildServiceProvider(); + var dispatcher = new TinyEventDispatcher("test-event"); + var eventInstance = new UserCreated(Guid.NewGuid()); + + await dispatcher.DispatchAsync(provider, eventInstance, CancellationToken.None); + + Assert.Equal(eventInstance, Assert.Single(RecordingConsumer.Consumed)); + } + + [Fact] + public async Task Dispatch_async_invokes_multiple_matching_consumers() + { + RecordingConsumer.Consumed.Clear(); + SecondRecordingConsumer.Consumed.Clear(); + var services = new ServiceCollection(); + services.AddSingleton, RecordingConsumer>(); + services.AddSingleton, SecondRecordingConsumer>(); + using var provider = services.BuildServiceProvider(); + var dispatcher = new TinyEventDispatcher("test-event"); + var eventInstance = new UserCreated(Guid.NewGuid()); + + await dispatcher.DispatchAsync(provider, eventInstance, CancellationToken.None); + + Assert.Equal(eventInstance, Assert.Single(RecordingConsumer.Consumed)); + Assert.Equal(eventInstance, Assert.Single(SecondRecordingConsumer.Consumed)); + } + + [Fact] + public async Task Dispatch_async_passes_cancellation_token_to_consumers() + { + CancellationRecordingConsumer.CancellationToken = default; + var services = new ServiceCollection(); + services.AddSingleton, CancellationRecordingConsumer>(); + using var provider = services.BuildServiceProvider(); + var dispatcher = new TinyEventDispatcher("test-event"); + using var cancellation = new CancellationTokenSource(); + + await dispatcher.DispatchAsync(provider, new UserCreated(Guid.NewGuid()), cancellation.Token); + + Assert.Equal(cancellation.Token, CancellationRecordingConsumer.CancellationToken); + } + + [Fact] + public async Task Dispatch_async_rejects_null_service_provider() + { + var dispatcher = new TinyEventDispatcher("test-event"); + + await Assert.ThrowsAsync( + async () => await dispatcher.DispatchAsync(null!, new UserCreated(Guid.NewGuid()), CancellationToken.None)); + } + + private sealed record UserCreated(Guid UserId); + + private sealed class RecordingConsumer : IEventConsumer + { + public static List Consumed { get; } = new List(); + + public ValueTask ConsumeAsync(UserCreated @event, CancellationToken cancellationToken) + { + Consumed.Add(@event); + return ValueTask.CompletedTask; + } + } + + private sealed class SecondRecordingConsumer : IEventConsumer + { + public static List Consumed { get; } = new List(); + + public ValueTask ConsumeAsync(UserCreated @event, CancellationToken cancellationToken) + { + Consumed.Add(@event); + return ValueTask.CompletedTask; + } + } + + private sealed class CancellationRecordingConsumer : IEventConsumer + { + public static CancellationToken CancellationToken { get; set; } + + public ValueTask ConsumeAsync(UserCreated @event, CancellationToken cancellationToken) + { + CancellationToken = cancellationToken; + return ValueTask.CompletedTask; + } + } +} From 87f5b03213fd4c1d583e1fa5967e064298d1f509 Mon Sep 17 00:00:00 2001 From: George Date: Fri, 24 Jul 2026 19:35:21 +0200 Subject: [PATCH 05/22] Emit generated event dispatchers --- .../Emission/EventTypeDescriptorEmitter.cs | 5 +++++ .../TinyEventsSourceGeneratorTests.cs | 3 +++ 2 files changed, 8 insertions(+) diff --git a/src/TinyEvents.SourceGen/Emission/EventTypeDescriptorEmitter.cs b/src/TinyEvents.SourceGen/Emission/EventTypeDescriptorEmitter.cs index ae22fa1..31038d4 100644 --- a/src/TinyEvents.SourceGen/Emission/EventTypeDescriptorEmitter.cs +++ b/src/TinyEvents.SourceGen/Emission/EventTypeDescriptorEmitter.cs @@ -14,5 +14,10 @@ public static void Emit( writer.Write(", typeof("); writer.Write(descriptor.EventTypeName); writer.WriteLine(")));"); + writer.Write("global::Microsoft.Extensions.DependencyInjection.ServiceCollectionServiceExtensions.AddSingleton(services, new global::TinyEvents.TinyEventDispatcher<"); + writer.Write(descriptor.EventTypeName); + writer.Write(">("); + writer.Write(StringLiteral.From(descriptor.EventTypeDisplayName)); + writer.WriteLine("));"); } } diff --git a/tests/TinyEvents.SourceGen.Tests/TinyEventsSourceGeneratorTests.cs b/tests/TinyEvents.SourceGen.Tests/TinyEventsSourceGeneratorTests.cs index 6c81971..dfcb68b 100644 --- a/tests/TinyEvents.SourceGen.Tests/TinyEventsSourceGeneratorTests.cs +++ b/tests/TinyEvents.SourceGen.Tests/TinyEventsSourceGeneratorTests.cs @@ -188,6 +188,8 @@ public ValueTask ConsumeAsync(UserCreated @event, CancellationToken cancellation Assert.Contains("IEventConsumer", source); Assert.Contains("global::MyApp.SendWelcomeEmail", source); Assert.Contains("TinyEventTypeDescriptor", source); + Assert.Contains("ITinyEventDispatcher", source); + Assert.Contains("TinyEventDispatcher", source); Assert.Contains("\"MyApp.UserCreated\"", source); Assert.Contains("TinyEventsBootstrap.AddContribution", source); } @@ -227,6 +229,7 @@ public ValueTask ConsumeAsync(UserCreated @event, CancellationToken cancellation Assert.Contains("global::MyApp.SendWelcomeEmail", source); Assert.Contains("global::MyApp.UpdateProjection", source); Assert.Equal(1, source.Split("TinyEventTypeDescriptor").Length - 1); + Assert.Equal(1, source.Split("TinyEventDispatcher<").Length - 1); } [Fact] From 2c88144c47f02f63b5b4ed8592bc0c251ae35172 Mon Sep 17 00:00:00 2001 From: George Date: Fri, 24 Jul 2026 19:38:10 +0200 Subject: [PATCH 06/22] Dispatch events without reflection --- .../Processing/TinyOutboxProcessor.cs | 76 ++++++------------- .../AdoNetPostgreSqlEndToEndRuntimeTests.cs | 4 +- .../EfCorePostgreSqlStoreRuntimeTests.cs | 4 +- .../Processing/TinyOutboxProcessorTests.cs | 4 +- 4 files changed, 29 insertions(+), 59 deletions(-) diff --git a/src/TinyEvents/Processing/TinyOutboxProcessor.cs b/src/TinyEvents/Processing/TinyOutboxProcessor.cs index ce6ffd2..e77d143 100644 --- a/src/TinyEvents/Processing/TinyOutboxProcessor.cs +++ b/src/TinyEvents/Processing/TinyOutboxProcessor.cs @@ -1,20 +1,11 @@ -using System.Reflection; -using Microsoft.Extensions.DependencyInjection; - namespace TinyEvents; public sealed class TinyOutboxProcessor : ITinyOutboxProcessor { - private static readonly MethodInfo ProcessTypedMessageMethod = - typeof(TinyOutboxProcessor).GetMethod( - nameof(ProcessTypedMessageAsync), - BindingFlags.Instance | BindingFlags.NonPublic) - ?? throw new InvalidOperationException("Typed message processor method was not found."); - private readonly IServiceProvider serviceProvider; private readonly ITinyOutboxStore store; private readonly ITinyEventSerializer serializer; - private readonly Dictionary eventTypes; + private readonly Dictionary dispatchers; private readonly TinyEventsOptions options; private readonly TimeProvider timeProvider; @@ -22,7 +13,7 @@ public TinyOutboxProcessor( IServiceProvider serviceProvider, ITinyOutboxStore store, ITinyEventSerializer serializer, - IEnumerable eventTypes, + IEnumerable dispatchers, TinyEventsOptions options, TimeProvider timeProvider) { @@ -41,9 +32,9 @@ public TinyOutboxProcessor( throw new ArgumentNullException(nameof(serializer)); } - if (eventTypes is null) + if (dispatchers is null) { - throw new ArgumentNullException(nameof(eventTypes)); + throw new ArgumentNullException(nameof(dispatchers)); } if (options is null) @@ -59,7 +50,7 @@ public TinyOutboxProcessor( this.serviceProvider = serviceProvider; this.store = store; this.serializer = serializer; - this.eventTypes = BuildEventTypeMap(eventTypes); + this.dispatchers = BuildDispatcherMap(dispatchers); this.options = options; this.timeProvider = timeProvider; } @@ -107,42 +98,21 @@ private async ValueTask InvokeConsumersAsync( TinyOutboxMessage message, CancellationToken cancellationToken) { - var eventType = ResolveEventType(message); - var eventInstance = serializer.Deserialize(message.Payload, eventType); - var method = ProcessTypedMessageMethod.MakeGenericMethod(eventType); - - var result = method.Invoke(this, new[] { eventInstance, cancellationToken }); - - if (result is null) - { - throw new InvalidOperationException("Typed event processing returned no task."); - } - - await (ValueTask)result; + var dispatcher = ResolveDispatcher(message); + var eventInstance = serializer.Deserialize(message.Payload, dispatcher.EventType); + await dispatcher.DispatchAsync(serviceProvider, eventInstance, cancellationToken); } - private Type ResolveEventType(TinyOutboxMessage message) + private ITinyEventDispatcher ResolveDispatcher(TinyOutboxMessage message) { - if (eventTypes.TryGetValue(message.EventType, out var eventType)) + if (dispatchers.TryGetValue(message.EventType, out var dispatcher)) { - return eventType; + return dispatcher; } throw new InvalidOperationException($"Event type '{message.EventType}' is not registered."); } - private async ValueTask ProcessTypedMessageAsync( - TEvent @event, - CancellationToken cancellationToken) - { - var consumers = serviceProvider.GetServices>(); - - foreach (var consumer in consumers) - { - await consumer.ConsumeAsync(@event, cancellationToken); - } - } - private async ValueTask MarkProcessedAsync( TinyOutboxMessage message, string workerId, @@ -183,33 +153,33 @@ await store.MarkFailedAsync( return timeProvider.GetUtcNow().Add(options.RetryDelay); } - private static Dictionary BuildEventTypeMap(IEnumerable descriptors) + private static Dictionary BuildDispatcherMap(IEnumerable dispatchers) { - var eventTypes = new Dictionary(StringComparer.Ordinal); + var dispatcherMap = new Dictionary(StringComparer.Ordinal); - foreach (var descriptor in descriptors) + foreach (var dispatcher in dispatchers) { - AddEventType(eventTypes, descriptor); + AddDispatcher(dispatcherMap, dispatcher); } - return eventTypes; + return dispatcherMap; } - private static void AddEventType( - Dictionary eventTypes, - TinyEventTypeDescriptor descriptor) + private static void AddDispatcher( + Dictionary dispatchers, + ITinyEventDispatcher dispatcher) { - if (eventTypes.TryGetValue(descriptor.EventTypeName, out var existingType)) + if (dispatchers.TryGetValue(dispatcher.EventTypeName, out var existingDispatcher)) { - if (existingType == descriptor.EventType) + if (existingDispatcher.EventType == dispatcher.EventType) { return; } throw new InvalidOperationException( - $"Event type name '{descriptor.EventTypeName}' is registered for both '{existingType.FullName}' and '{descriptor.EventType.FullName}'."); + $"Event type name '{dispatcher.EventTypeName}' is registered for both '{existingDispatcher.EventType.FullName}' and '{dispatcher.EventType.FullName}'."); } - eventTypes.Add(descriptor.EventTypeName, descriptor.EventType); + dispatchers.Add(dispatcher.EventTypeName, dispatcher); } } diff --git a/tests/TinyEvents.PostgreSql.Tests/AdoNetPostgreSqlEndToEndRuntimeTests.cs b/tests/TinyEvents.PostgreSql.Tests/AdoNetPostgreSqlEndToEndRuntimeTests.cs index dde4255..6ee78c9 100644 --- a/tests/TinyEvents.PostgreSql.Tests/AdoNetPostgreSqlEndToEndRuntimeTests.cs +++ b/tests/TinyEvents.PostgreSql.Tests/AdoNetPostgreSqlEndToEndRuntimeTests.cs @@ -62,8 +62,8 @@ private ServiceProvider BuildServices() return connection; }); }); - services.AddSingleton( - new TinyEventTypeDescriptor(typeof(UserCreated).FullName!, typeof(UserCreated))); + services.AddSingleton( + new TinyEventDispatcher(typeof(UserCreated).FullName!)); services.AddScoped, RecordingConsumer>(); return services.BuildServiceProvider(); diff --git a/tests/TinyEvents.PostgreSql.Tests/EfCorePostgreSqlStoreRuntimeTests.cs b/tests/TinyEvents.PostgreSql.Tests/EfCorePostgreSqlStoreRuntimeTests.cs index 0e7ddf7..7c86a82 100644 --- a/tests/TinyEvents.PostgreSql.Tests/EfCorePostgreSqlStoreRuntimeTests.cs +++ b/tests/TinyEvents.PostgreSql.Tests/EfCorePostgreSqlStoreRuntimeTests.cs @@ -158,8 +158,8 @@ private ServiceProvider BuildServices() { options.TableName = "TinyOutbox"; }); - services.AddSingleton( - new TinyEventTypeDescriptor(typeof(UserCreated).FullName!, typeof(UserCreated))); + services.AddSingleton( + new TinyEventDispatcher(typeof(UserCreated).FullName!)); services.AddScoped, RecordingConsumer>(); return services.BuildServiceProvider(); diff --git a/tests/TinyEvents.Tests/Processing/TinyOutboxProcessorTests.cs b/tests/TinyEvents.Tests/Processing/TinyOutboxProcessorTests.cs index 7d8831e..986ba2c 100644 --- a/tests/TinyEvents.Tests/Processing/TinyOutboxProcessorTests.cs +++ b/tests/TinyEvents.Tests/Processing/TinyOutboxProcessorTests.cs @@ -319,8 +319,8 @@ private static ITinyOutboxProcessor BuildProcessor( ClaimTimeout = claimTimeout ?? TimeSpan.FromMinutes(5) }); services.AddSingleton(); - services.AddSingleton( - new TinyEventTypeDescriptor(typeof(UserCreated).FullName!, typeof(UserCreated))); + services.AddSingleton( + new TinyEventDispatcher(typeof(UserCreated).FullName!)); services.AddSingleton(); services.AddSingleton, RecordingConsumer>(); From 7bafa2bec74e4b9ac0ff173e8c68454d9507cd53 Mon Sep 17 00:00:00 2001 From: George Date: Fri, 24 Jul 2026 19:42:51 +0200 Subject: [PATCH 07/22] Remove event type descriptor registrations --- docs/getting-started.md | 2 +- docs/source-generator.md | 15 ++++++----- .../Emission/ConsumerRegistrationEmitter.cs | 4 +-- .../Emission/EventDispatcherEmitter.cs | 18 +++++++++++++ .../Emission/EventTypeDescriptorEmitter.cs | 23 ----------------- ...scriptorPlan.cs => EventDispatcherPlan.cs} | 4 +-- .../Planning/TinyEventsGenerationPlan.cs | 8 +++--- .../Planning/TinyEventsGenerationPlanner.cs | 12 ++++----- .../Registry/TinyEventTypeDescriptor.cs | 25 ------------------- .../TinyEventsSourceGeneratorTests.cs | 2 -- 10 files changed, 40 insertions(+), 73 deletions(-) create mode 100644 src/TinyEvents.SourceGen/Emission/EventDispatcherEmitter.cs delete mode 100644 src/TinyEvents.SourceGen/Emission/EventTypeDescriptorEmitter.cs rename src/TinyEvents.SourceGen/Planning/{EventTypeDescriptorPlan.cs => EventDispatcherPlan.cs} (78%) delete mode 100644 src/TinyEvents/Registry/TinyEventTypeDescriptor.cs diff --git a/docs/getting-started.md b/docs/getting-started.md index 12533d9..852baff 100644 --- a/docs/getting-started.md +++ b/docs/getting-started.md @@ -137,7 +137,7 @@ var processor = provider.GetRequiredService(); await processor.ProcessPendingAsync(ct); ``` -Processing resolves generated `TinyEventTypeDescriptor` services to deserialize the payload, then resolves `IEnumerable>` from dependency injection. +Processing resolves a generated `ITinyEventDispatcher` for the stored event type, deserializes the payload, and invokes matching `IEventConsumer` services through dependency injection. For hosted processing: diff --git a/docs/source-generator.md b/docs/source-generator.md index 39cb52d..59f279c 100644 --- a/docs/source-generator.md +++ b/docs/source-generator.md @@ -23,7 +23,7 @@ public sealed class SendWelcomeEmail : IEventConsumer For each discovered consumer, the generator emits: - DI registration for `IEventConsumer` -- event type descriptor registration for deserialization +- event dispatcher registration for deserialization and consumer invocation - a generated contribution - a module initializer that adds the contribution to TinyEvents bootstrap @@ -31,9 +31,8 @@ The generated registration is equivalent to: ```csharp services.AddScoped, SendWelcomeEmail>(); -services.AddSingleton(new TinyEventTypeDescriptor( - "MyApp.UserCreated", - typeof(UserCreated))); +services.AddSingleton( + new TinyEventDispatcher("MyApp.UserCreated")); ``` The generated code is packaged as an `ITinyEventsContribution`. A module initializer calls: @@ -50,11 +49,11 @@ TinyEvents does not scan assemblies at runtime to find consumers. Runtime processing uses: -1. Generated `TinyEventTypeDescriptor` services to resolve the stored event type name. +1. Generated `ITinyEventDispatcher` services to resolve the stored event type name. 2. `ITinyEventSerializer` to deserialize the payload. -3. Dependency injection to resolve `IEnumerable>`. +3. Dependency injection to resolve and invoke `IEventConsumer` services. -Dependency injection is the consumer registry. `TinyEventTypeDescriptor` is only the event-name-to-CLR-type map needed for deserialization. +Dependency injection is the consumer registry. `ITinyEventDispatcher` is the event-name-to-typed-dispatch map needed for deserialization and consumer invocation. ## Contribution Bootstrap @@ -88,7 +87,7 @@ Analysis reads Roslyn syntax and symbols. Model stores TinyEvents-owned facts. -Planning creates consumer registration and event descriptor plans. +Planning creates consumer registration and event dispatcher plans. Emission writes generated C#. diff --git a/src/TinyEvents.SourceGen/Emission/ConsumerRegistrationEmitter.cs b/src/TinyEvents.SourceGen/Emission/ConsumerRegistrationEmitter.cs index 3e45ce0..a72fc03 100644 --- a/src/TinyEvents.SourceGen/Emission/ConsumerRegistrationEmitter.cs +++ b/src/TinyEvents.SourceGen/Emission/ConsumerRegistrationEmitter.cs @@ -21,9 +21,9 @@ public static void Emit( WriteConsumerRegistration(writer, registration); } - foreach (var descriptor in plan.EventTypeDescriptors) + foreach (var dispatcher in plan.EventDispatchers) { - EventTypeDescriptorEmitter.Emit(writer, descriptor); + EventDispatcherEmitter.Emit(writer, dispatcher); } writer.Unindent(); diff --git a/src/TinyEvents.SourceGen/Emission/EventDispatcherEmitter.cs b/src/TinyEvents.SourceGen/Emission/EventDispatcherEmitter.cs new file mode 100644 index 0000000..b645db8 --- /dev/null +++ b/src/TinyEvents.SourceGen/Emission/EventDispatcherEmitter.cs @@ -0,0 +1,18 @@ +using TinyEvents.SourceGen.Emission.Writing; +using TinyEvents.SourceGen.Planning; + +namespace TinyEvents.SourceGen.Emission; + +internal static class EventDispatcherEmitter +{ + public static void Emit( + SourceWriter writer, + EventDispatcherPlan dispatcher) + { + writer.Write("global::Microsoft.Extensions.DependencyInjection.ServiceCollectionServiceExtensions.AddSingleton(services, new global::TinyEvents.TinyEventDispatcher<"); + writer.Write(dispatcher.EventTypeName); + writer.Write(">("); + writer.Write(StringLiteral.From(dispatcher.EventTypeDisplayName)); + writer.WriteLine("));"); + } +} diff --git a/src/TinyEvents.SourceGen/Emission/EventTypeDescriptorEmitter.cs b/src/TinyEvents.SourceGen/Emission/EventTypeDescriptorEmitter.cs deleted file mode 100644 index 31038d4..0000000 --- a/src/TinyEvents.SourceGen/Emission/EventTypeDescriptorEmitter.cs +++ /dev/null @@ -1,23 +0,0 @@ -using TinyEvents.SourceGen.Emission.Writing; -using TinyEvents.SourceGen.Planning; - -namespace TinyEvents.SourceGen.Emission; - -internal static class EventTypeDescriptorEmitter -{ - public static void Emit( - SourceWriter writer, - EventTypeDescriptorPlan descriptor) - { - writer.Write("global::Microsoft.Extensions.DependencyInjection.ServiceCollectionServiceExtensions.AddSingleton(services, new global::TinyEvents.TinyEventTypeDescriptor("); - writer.Write(StringLiteral.From(descriptor.EventTypeDisplayName)); - writer.Write(", typeof("); - writer.Write(descriptor.EventTypeName); - writer.WriteLine(")));"); - writer.Write("global::Microsoft.Extensions.DependencyInjection.ServiceCollectionServiceExtensions.AddSingleton(services, new global::TinyEvents.TinyEventDispatcher<"); - writer.Write(descriptor.EventTypeName); - writer.Write(">("); - writer.Write(StringLiteral.From(descriptor.EventTypeDisplayName)); - writer.WriteLine("));"); - } -} diff --git a/src/TinyEvents.SourceGen/Planning/EventTypeDescriptorPlan.cs b/src/TinyEvents.SourceGen/Planning/EventDispatcherPlan.cs similarity index 78% rename from src/TinyEvents.SourceGen/Planning/EventTypeDescriptorPlan.cs rename to src/TinyEvents.SourceGen/Planning/EventDispatcherPlan.cs index df80ead..1674784 100644 --- a/src/TinyEvents.SourceGen/Planning/EventTypeDescriptorPlan.cs +++ b/src/TinyEvents.SourceGen/Planning/EventDispatcherPlan.cs @@ -1,8 +1,8 @@ namespace TinyEvents.SourceGen.Planning; -internal sealed class EventTypeDescriptorPlan +internal sealed class EventDispatcherPlan { - public EventTypeDescriptorPlan( + public EventDispatcherPlan( string eventTypeName, string eventTypeDisplayName) { diff --git a/src/TinyEvents.SourceGen/Planning/TinyEventsGenerationPlan.cs b/src/TinyEvents.SourceGen/Planning/TinyEventsGenerationPlan.cs index b51b6d4..216e4bd 100644 --- a/src/TinyEvents.SourceGen/Planning/TinyEventsGenerationPlan.cs +++ b/src/TinyEvents.SourceGen/Planning/TinyEventsGenerationPlan.cs @@ -6,15 +6,15 @@ internal sealed class TinyEventsGenerationPlan { public TinyEventsGenerationPlan( IReadOnlyList consumerRegistrations, - IReadOnlyList eventTypeDescriptors) + IReadOnlyList eventDispatchers) { ConsumerRegistrations = consumerRegistrations; - EventTypeDescriptors = eventTypeDescriptors; + EventDispatchers = eventDispatchers; } public IReadOnlyList ConsumerRegistrations { get; } - public IReadOnlyList EventTypeDescriptors { get; } + public IReadOnlyList EventDispatchers { get; } - public bool HasContent => ConsumerRegistrations.Count > 0 || EventTypeDescriptors.Count > 0; + public bool HasContent => ConsumerRegistrations.Count > 0 || EventDispatchers.Count > 0; } diff --git a/src/TinyEvents.SourceGen/Planning/TinyEventsGenerationPlanner.cs b/src/TinyEvents.SourceGen/Planning/TinyEventsGenerationPlanner.cs index 81f941a..ace9cd6 100644 --- a/src/TinyEvents.SourceGen/Planning/TinyEventsGenerationPlanner.cs +++ b/src/TinyEvents.SourceGen/Planning/TinyEventsGenerationPlanner.cs @@ -14,13 +14,13 @@ public static TinyEventsGenerationPlan Plan(DiscoveryResult discovery) .ThenBy(registration => registration.ImplementationTypeName, StringComparer.Ordinal) .ToArray(); - var descriptors = discovery.Consumers + var dispatchers = discovery.Consumers .GroupBy(consumer => consumer.EventTypeDisplayName, StringComparer.Ordinal) - .Select(group => CreateEventTypeDescriptor(group.First())) - .OrderBy(descriptor => descriptor.EventTypeDisplayName, StringComparer.Ordinal) + .Select(group => CreateEventDispatcher(group.First())) + .OrderBy(dispatcher => dispatcher.EventTypeDisplayName, StringComparer.Ordinal) .ToArray(); - return new TinyEventsGenerationPlan(registrations, descriptors); + return new TinyEventsGenerationPlan(registrations, dispatchers); } private static ConsumerRegistrationPlan CreateConsumerRegistration(DiscoveredConsumer consumer) @@ -30,9 +30,9 @@ private static ConsumerRegistrationPlan CreateConsumerRegistration(DiscoveredCon consumer.EventTypeName); } - private static EventTypeDescriptorPlan CreateEventTypeDescriptor(DiscoveredConsumer consumer) + private static EventDispatcherPlan CreateEventDispatcher(DiscoveredConsumer consumer) { - return new EventTypeDescriptorPlan( + return new EventDispatcherPlan( consumer.EventTypeName, consumer.EventTypeDisplayName); } diff --git a/src/TinyEvents/Registry/TinyEventTypeDescriptor.cs b/src/TinyEvents/Registry/TinyEventTypeDescriptor.cs deleted file mode 100644 index da646e9..0000000 --- a/src/TinyEvents/Registry/TinyEventTypeDescriptor.cs +++ /dev/null @@ -1,25 +0,0 @@ -namespace TinyEvents; - -public sealed class TinyEventTypeDescriptor -{ - public TinyEventTypeDescriptor(string eventTypeName, Type eventType) - { - if (string.IsNullOrWhiteSpace(eventTypeName)) - { - throw new ArgumentException("Event type name is required.", nameof(eventTypeName)); - } - - if (eventType is null) - { - throw new ArgumentNullException(nameof(eventType)); - } - - EventTypeName = eventTypeName; - EventType = eventType; - } - - public string EventTypeName { get; } - - public Type EventType { get; } -} - diff --git a/tests/TinyEvents.SourceGen.Tests/TinyEventsSourceGeneratorTests.cs b/tests/TinyEvents.SourceGen.Tests/TinyEventsSourceGeneratorTests.cs index dfcb68b..0dfba0c 100644 --- a/tests/TinyEvents.SourceGen.Tests/TinyEventsSourceGeneratorTests.cs +++ b/tests/TinyEvents.SourceGen.Tests/TinyEventsSourceGeneratorTests.cs @@ -187,7 +187,6 @@ public ValueTask ConsumeAsync(UserCreated @event, CancellationToken cancellation Assert.Contains("IEventConsumer", source); Assert.Contains("global::MyApp.SendWelcomeEmail", source); - Assert.Contains("TinyEventTypeDescriptor", source); Assert.Contains("ITinyEventDispatcher", source); Assert.Contains("TinyEventDispatcher", source); Assert.Contains("\"MyApp.UserCreated\"", source); @@ -228,7 +227,6 @@ public ValueTask ConsumeAsync(UserCreated @event, CancellationToken cancellation Assert.Contains("global::MyApp.SendWelcomeEmail", source); Assert.Contains("global::MyApp.UpdateProjection", source); - Assert.Equal(1, source.Split("TinyEventTypeDescriptor").Length - 1); Assert.Equal(1, source.Split("TinyEventDispatcher<").Length - 1); } From 5be52799e886cb764af3a6162ff37e40604cc260 Mon Sep 17 00:00:00 2001 From: George Date: Fri, 24 Jul 2026 19:50:28 +0200 Subject: [PATCH 08/22] Cover loaded consumer assembly contributions --- ...EventsSourceGeneratorMultiAssemblyTests.cs | 229 ++++++++++++++++++ 1 file changed, 229 insertions(+) create mode 100644 tests/TinyEvents.SourceGen.Tests/TinyEventsSourceGeneratorMultiAssemblyTests.cs diff --git a/tests/TinyEvents.SourceGen.Tests/TinyEventsSourceGeneratorMultiAssemblyTests.cs b/tests/TinyEvents.SourceGen.Tests/TinyEventsSourceGeneratorMultiAssemblyTests.cs new file mode 100644 index 0000000..9a3696f --- /dev/null +++ b/tests/TinyEvents.SourceGen.Tests/TinyEventsSourceGeneratorMultiAssemblyTests.cs @@ -0,0 +1,229 @@ +using System.Reflection; +using Microsoft.Extensions.DependencyInjection; +using Xunit; + +namespace TinyEvents.SourceGen.Tests; + +public sealed class TinyEventsSourceGeneratorMultiAssemblyTests +{ + [Fact] + public async Task Loaded_consumer_assembly_contribution_processes_event_in_host_services() + { + var consumerAssembly = CompileExternalConsumerAssembly(); + var userId = Guid.NewGuid(); + var store = new ProbeOutboxStore(); + await AddExternalUserCreatedMessageAsync(store, consumerAssembly, userId); + using var provider = BuildHostProvider(store); + var processor = provider.GetRequiredService(); + + await processor.ProcessPendingAsync(); + + Assert.True(ExternalRuntimeProbeContains(consumerAssembly, userId)); + Assert.Equal(TinyOutboxMessageStatus.Processed, Assert.Single(store.Snapshot()).Status); + } + + private static Assembly CompileExternalConsumerAssembly() + { + return SourceGeneratorTestHost.CompileAndLoad( + """ + using System; + using System.Collections.Generic; + using System.Threading; + using System.Threading.Tasks; + using TinyEvents; + + namespace ExternalConsumers; + + public sealed record UserCreated(Guid UserId); + + public sealed class RecordUserCreated : IEventConsumer + { + public ValueTask ConsumeAsync(UserCreated @event, CancellationToken cancellationToken) + { + RuntimeProbe.Consumed.Add(@event.UserId); + return ValueTask.CompletedTask; + } + } + + public static class RuntimeProbe + { + public static readonly List Consumed = new List(); + + public static bool Contains(Guid userId) + { + return Consumed.Contains(userId); + } + } + """); + } + + private static async ValueTask AddExternalUserCreatedMessageAsync( + ProbeOutboxStore store, + Assembly consumerAssembly, + Guid userId) + { + var eventType = consumerAssembly.GetType("ExternalConsumers.UserCreated")!; + var eventInstance = Activator.CreateInstance(eventType, userId)!; + var serializer = new SystemTextJsonTinyEventSerializer(); + + await store.AddAsync( + new TinyOutboxMessage + { + Id = Guid.NewGuid(), + EventType = eventType.FullName!, + Payload = serializer.Serialize(eventInstance, eventType), + Status = TinyOutboxMessageStatus.Pending, + CreatedAtUtc = DateTimeOffset.UtcNow + }, + CancellationToken.None); + } + + private static ServiceProvider BuildHostProvider(ProbeOutboxStore store) + { + var services = new ServiceCollection(); + services.UseTinyEvents(options => + { + options.WorkerId = "host-worker"; + }); + services.AddSingleton(store); + services.AddSingleton(store); + + return services.BuildServiceProvider(); + } + + private static bool ExternalRuntimeProbeContains( + Assembly consumerAssembly, + Guid userId) + { + var runtimeProbe = consumerAssembly.GetType("ExternalConsumers.RuntimeProbe")!; + var contains = runtimeProbe.GetMethod("Contains")!; + return (bool)contains.Invoke(null, new object[] { userId })!; + } + + private sealed class ProbeOutboxStore : ITinyOutboxStore, ITinyOutboxWriter + { + private readonly List messages = new List(); + + public ValueTask AddAsync( + TinyOutboxMessage message, + CancellationToken cancellationToken) + { + messages.Add(message); + return ValueTask.CompletedTask; + } + + public ValueTask> ClaimPendingAsync( + int maxCount, + string workerId, + DateTimeOffset now, + TimeSpan claimTimeout, + CancellationToken cancellationToken) + { + var claimedMessages = new List(); + + for (var index = 0; index < messages.Count && claimedMessages.Count < maxCount; index++) + { + var message = messages[index]; + + if (message.Status != TinyOutboxMessageStatus.Pending) + { + continue; + } + + var claimedMessage = Claim(message, workerId, now, claimTimeout); + messages[index] = claimedMessage; + claimedMessages.Add(claimedMessage); + } + + return ValueTask.FromResult>(claimedMessages); + } + + public ValueTask MarkProcessedAsync( + Guid messageId, + string workerId, + DateTimeOffset processedAtUtc, + CancellationToken cancellationToken) + { + Replace( + messageId, + message => new TinyOutboxMessage + { + Id = message.Id, + EventType = message.EventType, + Payload = message.Payload, + Status = TinyOutboxMessageStatus.Processed, + AttemptCount = message.AttemptCount, + ClaimedBy = message.ClaimedBy, + ClaimedAtUtc = message.ClaimedAtUtc, + ClaimExpiresAtUtc = message.ClaimExpiresAtUtc, + CreatedAtUtc = message.CreatedAtUtc, + ProcessedAtUtc = processedAtUtc + }); + + return ValueTask.CompletedTask; + } + + public ValueTask MarkFailedAsync( + Guid messageId, + string workerId, + string error, + int attemptCount, + DateTimeOffset? nextAttemptAtUtc, + CancellationToken cancellationToken) + { + Replace( + messageId, + message => new TinyOutboxMessage + { + Id = message.Id, + EventType = message.EventType, + Payload = message.Payload, + Status = TinyOutboxMessageStatus.Pending, + AttemptCount = attemptCount, + ClaimedBy = message.ClaimedBy, + ClaimedAtUtc = message.ClaimedAtUtc, + ClaimExpiresAtUtc = message.ClaimExpiresAtUtc, + CreatedAtUtc = message.CreatedAtUtc, + NextAttemptAtUtc = nextAttemptAtUtc, + LastError = error + }); + + return ValueTask.CompletedTask; + } + + public IReadOnlyList Snapshot() + { + return messages.ToArray(); + } + + private void Replace( + Guid messageId, + Func replace) + { + var index = messages.FindIndex(message => message.Id == messageId); + messages[index] = replace(messages[index]); + } + + private TinyOutboxMessage Claim( + TinyOutboxMessage message, + string workerId, + DateTimeOffset now, + TimeSpan claimTimeout) + { + return new TinyOutboxMessage + { + Id = message.Id, + EventType = message.EventType, + Payload = message.Payload, + Status = TinyOutboxMessageStatus.Processing, + AttemptCount = message.AttemptCount, + ClaimedBy = workerId, + ClaimedAtUtc = now, + ClaimExpiresAtUtc = now.Add(claimTimeout), + CreatedAtUtc = message.CreatedAtUtc, + NextAttemptAtUtc = message.NextAttemptAtUtc, + LastError = message.LastError + }; + } + } +} From f4fddb04a666adb60074547e63da73832387d773 Mon Sep 17 00:00:00 2001 From: George Date: Fri, 24 Jul 2026 19:53:34 +0200 Subject: [PATCH 09/22] Document generated dispatcher bootstrap behavior --- README.md | 6 +++--- docs/architecture.md | 12 +++++++----- docs/getting-started.md | 2 +- docs/source-generator.md | 4 +++- 4 files changed, 14 insertions(+), 10 deletions(-) diff --git a/README.md b/README.md index 505b611..31d87b5 100644 --- a/README.md +++ b/README.md @@ -35,7 +35,7 @@ TinyEvents is deliberately different: - `PublishAsync` stores an outbox message. - Consumers run later through the processor or hosted worker. -- Incremental source generation registers consumers and event type descriptors automatically. +- Incremental source generation registers consumers and event dispatchers automatically. - Runtime dispatch resolves consumers from Microsoft dependency injection. - Worker claiming is database-backed and lease-based. - Delivery is at-least-once, not exactly-once. @@ -139,7 +139,7 @@ public sealed class SendWelcomeEmail : IEventConsumer The source generator discovers concrete closed consumers and emits: - `IEventConsumer` DI registrations -- event type descriptors for deserialization +- event dispatchers for deserialization and consumer invocation - a module-initialized contribution to TinyEvents bootstrap No runtime assembly scanning is required, and normal consumers do not need manual DI registration. @@ -265,7 +265,7 @@ TinyEvents is intentionally small. - Domain/application event handling with outbox reliability. - Plain event objects, no marker interface. - No runtime scanning. -- Source generation for consumer registration and event type descriptors. +- Source generation for consumer registration and event dispatchers. - Contribution-based bootstrap for generated registrations. - Provider isolation. - Database-backed lease claiming. diff --git a/docs/architecture.md b/docs/architecture.md index 0ebd6f1..626ca09 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -24,7 +24,7 @@ TinyEvents is designed to stay small enough to reason about. - Providers are isolated in separate class libraries. - No runtime assembly scanning. - Source generation removes registration and event-map boilerplate. -- Generated contributions register consumers with dependency injection. +- Generated contributions register consumers and event dispatchers with dependency injection. - Delivery is at-least-once, not exactly-once. - Database-backed claim leases support multiple workers. - Consumers should be idempotent. @@ -119,10 +119,10 @@ ADO.NET publishing inserts the message through the current application transacti 1. Resolve the current worker id. 2. Claim pending or expired processing messages. -3. Resolve event type through generated event descriptors. +3. Resolve the event dispatcher for the stored event type. 4. Deserialize the payload. 5. Resolve all `IEventConsumer` instances from DI. -6. Invoke consumers. +6. Invoke consumers through the generated dispatcher. 7. Mark processed if all consumers succeed. 8. Mark failed or scheduled for retry when a consumer fails. @@ -178,6 +178,8 @@ The contribution system is the bridge between compile-time discovery and runtime 1. The generator emits an `ITinyEventsContribution`. 2. A module initializer adds it to `TinyEventsBootstrap`. 3. `UseTinyEvents` or a provider registration method applies contributions. -4. Consumers and event type descriptors become normal DI services. +4. Consumers and event dispatchers become normal DI services. -Runtime processing does not use a custom consumer registry. It resolves consumers directly from `IServiceProvider`. +TinyEvents does not scan assemblies or load consumer assemblies at runtime. Contributions are available only after the assembly that contains them has been loaded and its module initializer has run. Call TinyEvents registration after consumer assemblies are loaded. + +Runtime processing does not use a custom consumer registry. It resolves consumers directly from `IServiceProvider` through generated dispatchers. diff --git a/docs/getting-started.md b/docs/getting-started.md index 852baff..af6e619 100644 --- a/docs/getting-started.md +++ b/docs/getting-started.md @@ -50,7 +50,7 @@ services.UseSqlServerEntityFrameworkCoreOutbox(); `UseSqlServerEntityFrameworkCoreOutbox` registers TinyEvents core services, the EF Core outbox writer, and the EF Core outbox store. -It also applies generated TinyEvents contributions for the assemblies loaded in the process. Those contributions contain the consumer registrations and event type descriptors emitted by the source generator. +It also applies generated TinyEvents contributions for assemblies already loaded in the process. Those contributions contain the consumer and event dispatcher registrations emitted by the source generator. For PostgreSQL EF Core, use the PostgreSQL provider package and registration method: diff --git a/docs/source-generator.md b/docs/source-generator.md index 59f279c..e7f9ebb 100644 --- a/docs/source-generator.md +++ b/docs/source-generator.md @@ -59,7 +59,9 @@ Dependency injection is the consumer registry. `ITinyEventDispatcher` is the eve Generated contributions make multi-assembly projects work without runtime scanning. -Each assembly that contains consumers can contribute registrations. When the host calls a TinyEvents registration method, bootstrap applies the collected contributions once per service collection. +Each assembly that contains consumers generates its own contribution. When that assembly is loaded, its module initializer adds the contribution to TinyEvents bootstrap. When the host calls a TinyEvents registration method, bootstrap applies the collected contributions once per service collection. + +TinyEvents does not scan application assemblies or force-load referenced assemblies. If a consumer assembly has not been loaded before TinyEvents registration runs, that assembly's consumers and event dispatchers will not be registered in that service collection. For example: From 8fdb37be089e037e6ef49c83871bb059aa76571d Mon Sep 17 00:00:00 2001 From: George Date: Fri, 24 Jul 2026 20:03:49 +0200 Subject: [PATCH 10/22] Handle lost outbox leases explicitly --- .../TinyPostgreSqlAdoNetOutboxStore.cs | 18 ++- .../TinyPostgreSqlEfCoreOutboxStore.cs | 18 ++- .../TinySqlServerAdoNetOutboxStore.cs | 18 ++- .../TinySqlServerEfCoreOutboxStore.cs | 18 ++- .../Outbox/TinyOutboxLeaseLostException.cs | 12 ++ .../Processing/TinyOutboxProcessor.cs | 11 +- src/TinyEvents/Properties/AssemblyInfo.cs | 7 + .../TinyPostgreSqlAdoNetWriterTests.cs | 38 +++++- .../TinySqlServerAdoNetProviderTests.cs | 38 +++++- .../Outbox/InMemoryTinyOutboxStore.cs | 5 + .../Outbox/InMemoryTinyOutboxStoreTests.cs | 74 ++++++---- .../Processing/TinyOutboxProcessorTests.cs | 126 ++++++++++++++++++ 12 files changed, 340 insertions(+), 43 deletions(-) create mode 100644 src/TinyEvents/Outbox/TinyOutboxLeaseLostException.cs create mode 100644 src/TinyEvents/Properties/AssemblyInfo.cs diff --git a/src/TinyEvents.PostgreSql.AdoNet/TinyPostgreSqlAdoNetOutboxStore.cs b/src/TinyEvents.PostgreSql.AdoNet/TinyPostgreSqlAdoNetOutboxStore.cs index 902d034..61f5c2f 100644 --- a/src/TinyEvents.PostgreSql.AdoNet/TinyPostgreSqlAdoNetOutboxStore.cs +++ b/src/TinyEvents.PostgreSql.AdoNet/TinyPostgreSqlAdoNetOutboxStore.cs @@ -73,7 +73,8 @@ public async ValueTask MarkProcessedAsync( TinyPostgreSqlAdoNetCommandParameters.Add(command, "@ProcessedStatus", (int)TinyOutboxMessageStatus.Processed); TinyPostgreSqlAdoNetCommandParameters.Add(command, "@ProcessingStatus", (int)TinyOutboxMessageStatus.Processing); - await command.ExecuteNonQueryAsync(cancellationToken); + var affectedRows = await command.ExecuteNonQueryAsync(cancellationToken); + ThrowIfLeaseWasLost(affectedRows, messageId, workerId, "processed"); } public async ValueTask MarkFailedAsync( @@ -106,7 +107,8 @@ public async ValueTask MarkFailedAsync( TinyPostgreSqlAdoNetCommandParameters.Add(command, "@LastError", error); TinyPostgreSqlAdoNetCommandParameters.Add(command, "@ProcessingStatus", (int)TinyOutboxMessageStatus.Processing); - await command.ExecuteNonQueryAsync(cancellationToken); + var affectedRows = await command.ExecuteNonQueryAsync(cancellationToken); + ThrowIfLeaseWasLost(affectedRows, messageId, workerId, "failed"); } private async ValueTask CreateOpenConnectionAsync(CancellationToken cancellationToken) @@ -176,4 +178,16 @@ private static TinyOutboxMessageStatus GetFailedStatus(DateTimeOffset? nextAttem return TinyOutboxMessageStatus.Pending; } + + private static void ThrowIfLeaseWasLost( + int affectedRows, + Guid messageId, + string workerId, + string operation) + { + if (affectedRows == 0) + { + throw new TinyOutboxLeaseLostException(messageId, workerId, operation); + } + } } diff --git a/src/TinyEvents.PostgreSql.EntityFrameworkCore/TinyPostgreSqlEfCoreOutboxStore.cs b/src/TinyEvents.PostgreSql.EntityFrameworkCore/TinyPostgreSqlEfCoreOutboxStore.cs index c3342dd..acab25b 100644 --- a/src/TinyEvents.PostgreSql.EntityFrameworkCore/TinyPostgreSqlEfCoreOutboxStore.cs +++ b/src/TinyEvents.PostgreSql.EntityFrameworkCore/TinyPostgreSqlEfCoreOutboxStore.cs @@ -75,7 +75,8 @@ public async ValueTask MarkProcessedAsync( AddParameter(command, "@ProcessedStatus", (int)TinyOutboxMessageStatus.Processed); AddParameter(command, "@ProcessingStatus", (int)TinyOutboxMessageStatus.Processing); - await command.ExecuteNonQueryAsync(cancellationToken); + var affectedRows = await command.ExecuteNonQueryAsync(cancellationToken); + ThrowIfLeaseWasLost(affectedRows, messageId, workerId, "processed"); } public async ValueTask MarkFailedAsync( @@ -107,7 +108,8 @@ public async ValueTask MarkFailedAsync( AddParameter(command, "@LastError", error); AddParameter(command, "@ProcessingStatus", (int)TinyOutboxMessageStatus.Processing); - await command.ExecuteNonQueryAsync(cancellationToken); + var affectedRows = await command.ExecuteNonQueryAsync(cancellationToken); + ThrowIfLeaseWasLost(affectedRows, messageId, workerId, "failed"); } private async ValueTask CreateCommandAsync(CancellationToken cancellationToken) @@ -200,4 +202,16 @@ private static TinyOutboxMessageStatus GetFailedStatus(DateTimeOffset? nextAttem return TinyOutboxMessageStatus.Pending; } + + private static void ThrowIfLeaseWasLost( + int affectedRows, + Guid messageId, + string workerId, + string operation) + { + if (affectedRows == 0) + { + throw new TinyOutboxLeaseLostException(messageId, workerId, operation); + } + } } diff --git a/src/TinyEvents.SqlServer.AdoNet/TinySqlServerAdoNetOutboxStore.cs b/src/TinyEvents.SqlServer.AdoNet/TinySqlServerAdoNetOutboxStore.cs index 994ba1f..579d20b 100644 --- a/src/TinyEvents.SqlServer.AdoNet/TinySqlServerAdoNetOutboxStore.cs +++ b/src/TinyEvents.SqlServer.AdoNet/TinySqlServerAdoNetOutboxStore.cs @@ -76,7 +76,8 @@ public async ValueTask MarkProcessedAsync( TinySqlServerAdoNetCommandParameters.Add(command, "@ProcessedStatus", (int)TinyOutboxMessageStatus.Processed); TinySqlServerAdoNetCommandParameters.Add(command, "@ProcessingStatus", (int)TinyOutboxMessageStatus.Processing); - await command.ExecuteNonQueryAsync(cancellationToken); + var affectedRows = await command.ExecuteNonQueryAsync(cancellationToken); + ThrowIfLeaseWasLost(affectedRows, messageId, workerId, "processed"); } public async ValueTask MarkFailedAsync( @@ -109,7 +110,8 @@ public async ValueTask MarkFailedAsync( TinySqlServerAdoNetCommandParameters.Add(command, "@LastError", error); TinySqlServerAdoNetCommandParameters.Add(command, "@ProcessingStatus", (int)TinyOutboxMessageStatus.Processing); - await command.ExecuteNonQueryAsync(cancellationToken); + var affectedRows = await command.ExecuteNonQueryAsync(cancellationToken); + ThrowIfLeaseWasLost(affectedRows, messageId, workerId, "failed"); } private async ValueTask CreateOpenConnectionAsync(CancellationToken cancellationToken) @@ -179,4 +181,16 @@ private static TinyOutboxMessageStatus GetFailedStatus(DateTimeOffset? nextAttem return TinyOutboxMessageStatus.Pending; } + + private static void ThrowIfLeaseWasLost( + int affectedRows, + Guid messageId, + string workerId, + string operation) + { + if (affectedRows == 0) + { + throw new TinyOutboxLeaseLostException(messageId, workerId, operation); + } + } } diff --git a/src/TinyEvents.SqlServer.EntityFrameworkCore/TinySqlServerEfCoreOutboxStore.cs b/src/TinyEvents.SqlServer.EntityFrameworkCore/TinySqlServerEfCoreOutboxStore.cs index 4bf78f0..899d3f1 100644 --- a/src/TinyEvents.SqlServer.EntityFrameworkCore/TinySqlServerEfCoreOutboxStore.cs +++ b/src/TinyEvents.SqlServer.EntityFrameworkCore/TinySqlServerEfCoreOutboxStore.cs @@ -75,7 +75,8 @@ public async ValueTask MarkProcessedAsync( AddParameter(command, "@ProcessedStatus", (int)TinyOutboxMessageStatus.Processed); AddParameter(command, "@ProcessingStatus", (int)TinyOutboxMessageStatus.Processing); - await command.ExecuteNonQueryAsync(cancellationToken); + var affectedRows = await command.ExecuteNonQueryAsync(cancellationToken); + ThrowIfLeaseWasLost(affectedRows, messageId, workerId, "processed"); } public async ValueTask MarkFailedAsync( @@ -107,7 +108,8 @@ public async ValueTask MarkFailedAsync( AddParameter(command, "@LastError", error); AddParameter(command, "@ProcessingStatus", (int)TinyOutboxMessageStatus.Processing); - await command.ExecuteNonQueryAsync(cancellationToken); + var affectedRows = await command.ExecuteNonQueryAsync(cancellationToken); + ThrowIfLeaseWasLost(affectedRows, messageId, workerId, "failed"); } private async ValueTask CreateCommandAsync(CancellationToken cancellationToken) @@ -200,4 +202,16 @@ private static TinyOutboxMessageStatus GetFailedStatus(DateTimeOffset? nextAttem return TinyOutboxMessageStatus.Pending; } + + private static void ThrowIfLeaseWasLost( + int affectedRows, + Guid messageId, + string workerId, + string operation) + { + if (affectedRows == 0) + { + throw new TinyOutboxLeaseLostException(messageId, workerId, operation); + } + } } diff --git a/src/TinyEvents/Outbox/TinyOutboxLeaseLostException.cs b/src/TinyEvents/Outbox/TinyOutboxLeaseLostException.cs new file mode 100644 index 0000000..6a37318 --- /dev/null +++ b/src/TinyEvents/Outbox/TinyOutboxLeaseLostException.cs @@ -0,0 +1,12 @@ +namespace TinyEvents; + +internal sealed class TinyOutboxLeaseLostException : InvalidOperationException +{ + public TinyOutboxLeaseLostException( + Guid messageId, + string workerId, + string operation) + : base($"Outbox message '{messageId}' could not be marked as {operation} because worker '{workerId}' no longer owns a processing lease.") + { + } +} diff --git a/src/TinyEvents/Processing/TinyOutboxProcessor.cs b/src/TinyEvents/Processing/TinyOutboxProcessor.cs index e77d143..5b39645 100644 --- a/src/TinyEvents/Processing/TinyOutboxProcessor.cs +++ b/src/TinyEvents/Processing/TinyOutboxProcessor.cs @@ -88,9 +88,18 @@ private async ValueTask ProcessMessageAsync( { throw; } + catch (TinyOutboxLeaseLostException) + { + } catch (Exception exception) { - await MarkFailedAsync(message, workerId, exception, cancellationToken); + try + { + await MarkFailedAsync(message, workerId, exception, cancellationToken); + } + catch (TinyOutboxLeaseLostException) + { + } } } diff --git a/src/TinyEvents/Properties/AssemblyInfo.cs b/src/TinyEvents/Properties/AssemblyInfo.cs new file mode 100644 index 0000000..feca562 --- /dev/null +++ b/src/TinyEvents/Properties/AssemblyInfo.cs @@ -0,0 +1,7 @@ +using System.Runtime.CompilerServices; + +[assembly: InternalsVisibleTo("TinyEvents.SqlServer.AdoNet")] +[assembly: InternalsVisibleTo("TinyEvents.SqlServer.EntityFrameworkCore")] +[assembly: InternalsVisibleTo("TinyEvents.PostgreSql.AdoNet")] +[assembly: InternalsVisibleTo("TinyEvents.PostgreSql.EntityFrameworkCore")] +[assembly: InternalsVisibleTo("TinyEvents.Tests")] diff --git a/tests/TinyEvents.PostgreSql.AdoNet.Tests/TinyPostgreSqlAdoNetWriterTests.cs b/tests/TinyEvents.PostgreSql.AdoNet.Tests/TinyPostgreSqlAdoNetWriterTests.cs index 622e7e4..184b254 100644 --- a/tests/TinyEvents.PostgreSql.AdoNet.Tests/TinyPostgreSqlAdoNetWriterTests.cs +++ b/tests/TinyEvents.PostgreSql.AdoNet.Tests/TinyPostgreSqlAdoNetWriterTests.cs @@ -335,6 +335,17 @@ public async Task Mark_processed_uses_worker_connection_factory() Assert.Equal(1, factory.CallCount); } + [Fact] + public async Task Mark_processed_throws_when_no_rows_are_updated() + { + var store = NewStore(new RecordingWorkerConnectionFactory(new RecordingConnection(affectedRows: 0))); + + var exception = await Assert.ThrowsAnyAsync( + async () => await store.MarkProcessedAsync(Guid.NewGuid(), "worker", DateTimeOffset.UtcNow, CancellationToken.None)); + + Assert.Contains("no longer owns a processing lease", exception.Message); + } + [Fact] public async Task Mark_failed_uses_worker_connection_factory() { @@ -346,6 +357,17 @@ public async Task Mark_failed_uses_worker_connection_factory() Assert.Equal(1, factory.CallCount); } + [Fact] + public async Task Mark_failed_throws_when_no_rows_are_updated() + { + var store = NewStore(new RecordingWorkerConnectionFactory(new RecordingConnection(affectedRows: 0))); + + var exception = await Assert.ThrowsAnyAsync( + async () => await store.MarkFailedAsync(Guid.NewGuid(), "worker", "boom", 1, null, CancellationToken.None)); + + Assert.Contains("no longer owns a processing lease", exception.Message); + } + [Fact] public async Task Worker_connection_is_disposed_after_worker_operation_if_owned_by_factory() { @@ -461,8 +483,14 @@ public ValueTask CreateOpenConnectionAsync(CancellationToken cance private sealed class RecordingConnection : DbConnection { private readonly RecordingParameterCollection parameters = new RecordingParameterCollection(); + private readonly int affectedRows; private ConnectionState state = ConnectionState.Open; + public RecordingConnection(int affectedRows = 1) + { + this.affectedRows = affectedRows; + } + public RecordingCommand LastCommand { get; private set; } public bool WasDisposed { get; private set; } @@ -504,7 +532,7 @@ protected override DbTransaction BeginDbTransaction(IsolationLevel isolationLeve protected override DbCommand CreateDbCommand() { - LastCommand = new RecordingCommand(this); + LastCommand = new RecordingCommand(this, affectedRows); return LastCommand; } @@ -555,10 +583,12 @@ private sealed class RecordingCommand : DbCommand { private readonly RecordingParameterCollection parameters = new RecordingParameterCollection(); private readonly DbConnection connection; + private readonly int affectedRows; - public RecordingCommand(DbConnection connection) + public RecordingCommand(DbConnection connection, int affectedRows) { this.connection = connection; + this.affectedRows = affectedRows; } public override string CommandText { get; set; } = string.Empty; @@ -595,7 +625,7 @@ public override void Cancel() public override int ExecuteNonQuery() { - return 1; + return affectedRows; } public override object ExecuteScalar() @@ -626,7 +656,7 @@ protected override Task ExecuteDbDataReaderAsync( public override Task ExecuteNonQueryAsync(CancellationToken cancellationToken) { - return Task.FromResult(1); + return Task.FromResult(affectedRows); } } diff --git a/tests/TinyEvents.SqlServer.AdoNet.Tests/TinySqlServerAdoNetProviderTests.cs b/tests/TinyEvents.SqlServer.AdoNet.Tests/TinySqlServerAdoNetProviderTests.cs index c203a9c..b85dc6f 100644 --- a/tests/TinyEvents.SqlServer.AdoNet.Tests/TinySqlServerAdoNetProviderTests.cs +++ b/tests/TinyEvents.SqlServer.AdoNet.Tests/TinySqlServerAdoNetProviderTests.cs @@ -210,6 +210,17 @@ public async Task MarkProcessedAsync_uses_worker_connection_factory() Assert.Equal(1, factory.CallCount); } + [Fact] + public async Task MarkProcessedAsync_throws_when_no_rows_are_updated() + { + var store = NewStore(new RecordingWorkerConnectionFactory(new RecordingConnection(affectedRows: 0))); + + var exception = await Assert.ThrowsAnyAsync( + async () => await store.MarkProcessedAsync(Guid.NewGuid(), "worker", DateTimeOffset.UtcNow, CancellationToken.None)); + + Assert.Contains("no longer owns a processing lease", exception.Message); + } + [Fact] public async Task MarkFailedAsync_uses_worker_connection_factory() { @@ -221,6 +232,17 @@ public async Task MarkFailedAsync_uses_worker_connection_factory() Assert.Equal(1, factory.CallCount); } + [Fact] + public async Task MarkFailedAsync_throws_when_no_rows_are_updated() + { + var store = NewStore(new RecordingWorkerConnectionFactory(new RecordingConnection(affectedRows: 0))); + + var exception = await Assert.ThrowsAnyAsync( + async () => await store.MarkFailedAsync(Guid.NewGuid(), "worker", "boom", 1, null, CancellationToken.None)); + + Assert.Contains("no longer owns a processing lease", exception.Message); + } + [Fact] public void AdoNet_store_does_not_implement_writer() { @@ -405,8 +427,14 @@ public ValueTask CreateOpenConnectionAsync(CancellationToken cance private sealed class RecordingConnection : DbConnection { + private readonly int affectedRows; private ConnectionState state = ConnectionState.Open; + public RecordingConnection(int affectedRows = 1) + { + this.affectedRows = affectedRows; + } + public RecordingCommand LastCommand { get; private set; } public bool WasDisposed { get; private set; } @@ -453,7 +481,7 @@ protected override DbTransaction BeginDbTransaction(IsolationLevel isolationLeve protected override DbCommand CreateDbCommand() { - LastCommand = new RecordingCommand(this); + LastCommand = new RecordingCommand(this, affectedRows); return LastCommand; } @@ -504,10 +532,12 @@ private sealed class RecordingCommand : DbCommand { private readonly RecordingParameterCollection parameters = new RecordingParameterCollection(); private readonly DbConnection connection; + private readonly int affectedRows; - public RecordingCommand(DbConnection connection) + public RecordingCommand(DbConnection connection, int affectedRows) { this.connection = connection; + this.affectedRows = affectedRows; } public override string CommandText { get; set; } = string.Empty; @@ -544,7 +574,7 @@ public override void Cancel() public override int ExecuteNonQuery() { - return 1; + return affectedRows; } public override object ExecuteScalar() @@ -568,7 +598,7 @@ protected override DbDataReader ExecuteDbDataReader(CommandBehavior behavior) public override Task ExecuteNonQueryAsync(CancellationToken cancellationToken) { - return Task.FromResult(1); + return Task.FromResult(affectedRows); } } diff --git a/tests/TinyEvents.Tests/Outbox/InMemoryTinyOutboxStore.cs b/tests/TinyEvents.Tests/Outbox/InMemoryTinyOutboxStore.cs index 47914e5..f07eb2e 100644 --- a/tests/TinyEvents.Tests/Outbox/InMemoryTinyOutboxStore.cs +++ b/tests/TinyEvents.Tests/Outbox/InMemoryTinyOutboxStore.cs @@ -65,6 +65,7 @@ public ValueTask MarkProcessedAsync( ReplaceClaimedMessage( messageId, workerId, + operation: "processed", message => MarkProcessed(message, processedAtUtc)); } @@ -96,6 +97,7 @@ public ValueTask MarkFailedAsync( ReplaceClaimedMessage( messageId, workerId, + operation: "failed", message => MarkFailed(message, error, attemptCount, nextAttemptAtUtc)); } @@ -182,6 +184,7 @@ private static TinyOutboxMessage Claim( private void ReplaceClaimedMessage( Guid messageId, string workerId, + string operation, Func replace) { for (var index = 0; index < messages.Count; index++) @@ -198,6 +201,8 @@ private void ReplaceClaimedMessage( messages[index] = replace(message); return; } + + throw new TinyOutboxLeaseLostException(messageId, workerId, operation); } private static TinyOutboxMessage Copy(TinyOutboxMessage message) diff --git a/tests/TinyEvents.Tests/Outbox/InMemoryTinyOutboxStoreTests.cs b/tests/TinyEvents.Tests/Outbox/InMemoryTinyOutboxStoreTests.cs index 296d4b0..dead743 100644 --- a/tests/TinyEvents.Tests/Outbox/InMemoryTinyOutboxStoreTests.cs +++ b/tests/TinyEvents.Tests/Outbox/InMemoryTinyOutboxStoreTests.cs @@ -116,20 +116,30 @@ await store.AddAsync( } [Fact] - public async Task Mark_processed_async_updates_only_message_claimed_by_worker() + public async Task Mark_processed_async_throws_when_message_is_not_claimed_by_worker() { var store = new InMemoryTinyOutboxStore(); var messageId = Guid.NewGuid(); var processedAt = DateTimeOffset.UtcNow; await store.AddAsync(NewProcessingMessage(id: messageId, workerId: "worker-1"), CancellationToken.None); - await store.MarkProcessedAsync( - messageId, - workerId: "worker-2", - processedAtUtc: processedAt, - cancellationToken: CancellationToken.None); + await Assert.ThrowsAsync( + async () => await store.MarkProcessedAsync( + messageId, + workerId: "worker-2", + processedAtUtc: processedAt, + cancellationToken: CancellationToken.None)); Assert.Equal(TinyOutboxMessageStatus.Processing, Assert.Single(store.Snapshot()).Status); + } + + [Fact] + public async Task Mark_processed_async_updates_message_claimed_by_worker() + { + var store = new InMemoryTinyOutboxStore(); + var messageId = Guid.NewGuid(); + var processedAt = DateTimeOffset.UtcNow; + await store.AddAsync(NewProcessingMessage(id: messageId, workerId: "worker-1"), CancellationToken.None); await store.MarkProcessedAsync( messageId, @@ -151,32 +161,43 @@ await store.AddAsync( NewPendingMessage(id: messageId, claimedBy: "worker-1"), CancellationToken.None); - await store.MarkProcessedAsync( - messageId, - workerId: "worker-1", - processedAtUtc: DateTimeOffset.UtcNow, - cancellationToken: CancellationToken.None); + await Assert.ThrowsAsync( + async () => await store.MarkProcessedAsync( + messageId, + workerId: "worker-1", + processedAtUtc: DateTimeOffset.UtcNow, + cancellationToken: CancellationToken.None)); Assert.Equal(TinyOutboxMessageStatus.Pending, Assert.Single(store.Snapshot()).Status); } [Fact] - public async Task Mark_failed_async_updates_only_message_claimed_by_worker() + public async Task Mark_failed_async_throws_when_message_is_not_claimed_by_worker() { var store = new InMemoryTinyOutboxStore(); var messageId = Guid.NewGuid(); var nextAttemptAt = DateTimeOffset.UtcNow.AddSeconds(30); await store.AddAsync(NewProcessingMessage(id: messageId, workerId: "worker-1"), CancellationToken.None); - await store.MarkFailedAsync( - messageId, - workerId: "worker-2", - error: "nope", - attemptCount: 1, - nextAttemptAtUtc: nextAttemptAt, - cancellationToken: CancellationToken.None); + await Assert.ThrowsAsync( + async () => await store.MarkFailedAsync( + messageId, + workerId: "worker-2", + error: "nope", + attemptCount: 1, + nextAttemptAtUtc: nextAttemptAt, + cancellationToken: CancellationToken.None)); Assert.Equal(TinyOutboxMessageStatus.Processing, Assert.Single(store.Snapshot()).Status); + } + + [Fact] + public async Task Mark_failed_async_updates_message_claimed_by_worker() + { + var store = new InMemoryTinyOutboxStore(); + var messageId = Guid.NewGuid(); + var nextAttemptAt = DateTimeOffset.UtcNow.AddSeconds(30); + await store.AddAsync(NewProcessingMessage(id: messageId, workerId: "worker-1"), CancellationToken.None); await store.MarkFailedAsync( messageId, @@ -202,13 +223,14 @@ await store.AddAsync( NewPendingMessage(id: messageId, claimedBy: "worker-1"), CancellationToken.None); - await store.MarkFailedAsync( - messageId, - workerId: "worker-1", - error: "boom", - attemptCount: 1, - nextAttemptAtUtc: null, - cancellationToken: CancellationToken.None); + await Assert.ThrowsAsync( + async () => await store.MarkFailedAsync( + messageId, + workerId: "worker-1", + error: "boom", + attemptCount: 1, + nextAttemptAtUtc: null, + cancellationToken: CancellationToken.None)); var message = Assert.Single(store.Snapshot()); Assert.Equal(TinyOutboxMessageStatus.Pending, message.Status); diff --git a/tests/TinyEvents.Tests/Processing/TinyOutboxProcessorTests.cs b/tests/TinyEvents.Tests/Processing/TinyOutboxProcessorTests.cs index 986ba2c..063774f 100644 --- a/tests/TinyEvents.Tests/Processing/TinyOutboxProcessorTests.cs +++ b/tests/TinyEvents.Tests/Processing/TinyOutboxProcessorTests.cs @@ -296,6 +296,36 @@ await Assert.ThrowsAsync( Assert.Equal(0, store.MarkFailedCount); } + [Fact] + public async Task Process_pending_async_continues_when_mark_processed_loses_lease() + { + RecordingConsumer.Consumed.Clear(); + var firstMessage = NewProcessingMessage(new UserCreated(Guid.NewGuid(), "first@example.com")); + var secondMessage = NewProcessingMessage(new UserCreated(Guid.NewGuid(), "second@example.com")); + var store = new LeaseLostOnFirstProcessedStore(firstMessage, secondMessage); + var processor = BuildProcessor(store); + + await processor.ProcessPendingAsync(); + + Assert.Equal(2, RecordingConsumer.Consumed.Count); + Assert.Equal(secondMessage.Id, Assert.Single(store.ProcessedMessageIds)); + Assert.Equal(0, store.MarkFailedCount); + } + + [Fact] + public async Task Process_pending_async_continues_when_mark_failed_loses_lease() + { + ThrowingConsumer.Throw = true; + var message = NewProcessingMessage(new UserCreated(Guid.NewGuid(), "user@example.com")); + var store = new LeaseLostOnFailedStore(message); + var processor = BuildProcessor(store, includeThrowingConsumer: true); + + await processor.ProcessPendingAsync(); + + Assert.Equal(1, store.MarkFailedCount); + ThrowingConsumer.Throw = false; + } + private static ITinyOutboxProcessor BuildProcessor( ITinyOutboxStore store, bool includeSecondConsumer = false, @@ -575,4 +605,100 @@ public ValueTask MarkFailedAsync( return ValueTask.CompletedTask; } } + + private sealed class LeaseLostOnFirstProcessedStore : ITinyOutboxStore + { + private readonly IReadOnlyList claimedMessages; + private bool hasLostLease; + + public LeaseLostOnFirstProcessedStore(params TinyOutboxMessage[] claimedMessages) + { + this.claimedMessages = claimedMessages; + } + + public List ProcessedMessageIds { get; } = new List(); + + public int MarkFailedCount { get; private set; } + + public ValueTask> ClaimPendingAsync( + int maxCount, + string workerId, + DateTimeOffset now, + TimeSpan claimTimeout, + CancellationToken cancellationToken) + { + return ValueTask.FromResult(claimedMessages); + } + + public ValueTask MarkProcessedAsync( + Guid messageId, + string workerId, + DateTimeOffset processedAtUtc, + CancellationToken cancellationToken) + { + if (!hasLostLease) + { + hasLostLease = true; + throw new TinyOutboxLeaseLostException(messageId, workerId, "processed"); + } + + ProcessedMessageIds.Add(messageId); + return ValueTask.CompletedTask; + } + + public ValueTask MarkFailedAsync( + Guid messageId, + string workerId, + string error, + int attemptCount, + DateTimeOffset? nextAttemptAtUtc, + CancellationToken cancellationToken) + { + MarkFailedCount++; + return ValueTask.CompletedTask; + } + } + + private sealed class LeaseLostOnFailedStore : ITinyOutboxStore + { + private readonly IReadOnlyList claimedMessages; + + public LeaseLostOnFailedStore(params TinyOutboxMessage[] claimedMessages) + { + this.claimedMessages = claimedMessages; + } + + public int MarkFailedCount { get; private set; } + + public ValueTask> ClaimPendingAsync( + int maxCount, + string workerId, + DateTimeOffset now, + TimeSpan claimTimeout, + CancellationToken cancellationToken) + { + return ValueTask.FromResult(claimedMessages); + } + + public ValueTask MarkProcessedAsync( + Guid messageId, + string workerId, + DateTimeOffset processedAtUtc, + CancellationToken cancellationToken) + { + return ValueTask.CompletedTask; + } + + public ValueTask MarkFailedAsync( + Guid messageId, + string workerId, + string error, + int attemptCount, + DateTimeOffset? nextAttemptAtUtc, + CancellationToken cancellationToken) + { + MarkFailedCount++; + throw new TinyOutboxLeaseLostException(messageId, workerId, "failed"); + } + } } From 184a7d9833fd4bd2674bbd6807addc3ac3a787bc Mon Sep 17 00:00:00 2001 From: George Date: Fri, 24 Jul 2026 20:10:05 +0200 Subject: [PATCH 11/22] Keep worker running after iteration failures --- .../TinyEvents.Worker.csproj | 1 + .../TinyEventsBackgroundService.cs | 41 ++++++++++++-- .../TinyEventsWorkerTests.cs | 53 +++++++++++++++++++ 3 files changed, 92 insertions(+), 3 deletions(-) diff --git a/src/TinyEvents.Worker/TinyEvents.Worker.csproj b/src/TinyEvents.Worker/TinyEvents.Worker.csproj index 81ddc08..af22c73 100644 --- a/src/TinyEvents.Worker/TinyEvents.Worker.csproj +++ b/src/TinyEvents.Worker/TinyEvents.Worker.csproj @@ -15,6 +15,7 @@ + diff --git a/src/TinyEvents.Worker/TinyEventsBackgroundService.cs b/src/TinyEvents.Worker/TinyEventsBackgroundService.cs index 18c9f3a..b1a7d53 100644 --- a/src/TinyEvents.Worker/TinyEventsBackgroundService.cs +++ b/src/TinyEvents.Worker/TinyEventsBackgroundService.cs @@ -1,5 +1,7 @@ using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.Hosting; +using Microsoft.Extensions.Logging; +using Microsoft.Extensions.Logging.Abstractions; namespace TinyEvents.Worker; @@ -7,10 +9,19 @@ public sealed class TinyEventsBackgroundService : BackgroundService { private readonly IServiceScopeFactory scopeFactory; private readonly TinyEventsWorkerOptions options; + private readonly ILogger logger; public TinyEventsBackgroundService( IServiceScopeFactory scopeFactory, TinyEventsWorkerOptions options) + : this(scopeFactory, options, NullLogger.Instance) + { + } + + public TinyEventsBackgroundService( + IServiceScopeFactory scopeFactory, + TinyEventsWorkerOptions options, + ILogger logger) { if (scopeFactory is null) { @@ -22,8 +33,14 @@ public TinyEventsBackgroundService( throw new ArgumentNullException(nameof(options)); } + if (logger is null) + { + throw new ArgumentNullException(nameof(logger)); + } + this.scopeFactory = scopeFactory; this.options = options; + this.logger = logger; } public async ValueTask ProcessOnceAsync(CancellationToken cancellationToken = default) @@ -38,9 +55,27 @@ protected override async Task ExecuteAsync(CancellationToken stoppingToken) { while (!stoppingToken.IsCancellationRequested) { - await ProcessOnceAsync(stoppingToken); - await Task.Delay(options.PollingInterval, stoppingToken); + try + { + await ProcessOnceAsync(stoppingToken); + } + catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested) + { + return; + } + catch (Exception exception) + { + logger.LogError(exception, "TinyEvents worker processing iteration failed."); + } + + try + { + await Task.Delay(options.PollingInterval, stoppingToken); + } + catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested) + { + return; + } } } } - diff --git a/tests/TinyEvents.Worker.Tests/TinyEventsWorkerTests.cs b/tests/TinyEvents.Worker.Tests/TinyEventsWorkerTests.cs index 5a93dac..7419a17 100644 --- a/tests/TinyEvents.Worker.Tests/TinyEventsWorkerTests.cs +++ b/tests/TinyEvents.Worker.Tests/TinyEventsWorkerTests.cs @@ -228,6 +228,28 @@ public async Task Background_service_creates_scope_per_processing_iteration() Assert.NotEqual(ScopedProcessor.InstanceIds[0], ScopedProcessor.InstanceIds[1]); } + [Fact] + public async Task Background_service_continues_after_processing_iteration_fails() + { + FailingThenRecordingProcessor.Reset(); + var services = new ServiceCollection(); + services.AddSingleton(); + services.AddSingleton(new TinyEventsWorkerOptions + { + PollingInterval = TimeSpan.FromMilliseconds(1) + }); + services.AddSingleton(); + using var provider = services.BuildServiceProvider(); + var worker = provider.GetRequiredService(); + using var cancellation = new CancellationTokenSource(TimeSpan.FromSeconds(5)); + + await worker.StartAsync(cancellation.Token); + await FailingThenRecordingProcessor.SecondCall.Task.WaitAsync(cancellation.Token); + await worker.StopAsync(CancellationToken.None).WaitAsync(cancellation.Token); + + Assert.True(FailingThenRecordingProcessor.CallCount >= 2); + } + private sealed class RecordingProcessor : ITinyOutboxProcessor { public static int CallCount { get; set; } @@ -251,4 +273,35 @@ public ValueTask ProcessPendingAsync(CancellationToken cancellationToken = defau return ValueTask.CompletedTask; } } + + private sealed class FailingThenRecordingProcessor : ITinyOutboxProcessor + { + public static int CallCount { get; private set; } + + public static TaskCompletionSource SecondCall { get; private set; } = NewTaskCompletionSource(); + + public static void Reset() + { + CallCount = 0; + SecondCall = NewTaskCompletionSource(); + } + + public ValueTask ProcessPendingAsync(CancellationToken cancellationToken = default) + { + CallCount++; + + if (CallCount == 1) + { + throw new InvalidOperationException("database failed"); + } + + SecondCall.TrySetResult(); + return ValueTask.CompletedTask; + } + + private static TaskCompletionSource NewTaskCompletionSource() + { + return new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + } + } } From f35be65e9843cfade18c96cab642ea8b12ce08eb Mon Sep 17 00:00:00 2001 From: George Date: Fri, 24 Jul 2026 20:11:57 +0200 Subject: [PATCH 12/22] Document worker lease loss behavior --- docs/architecture.md | 4 ++++ docs/workers.md | 6 ++++++ 2 files changed, 10 insertions(+) diff --git a/docs/architecture.md b/docs/architecture.md index 626ca09..c19411a 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -128,6 +128,8 @@ ADO.NET publishing inserts the message through the current application transacti One outbox message represents one event, not one message per consumer. If one consumer fails, the event message is retried. +If marking a message as processed or failed no longer updates a row because the worker lost ownership of the processing lease, the processor leaves the message alone and continues. Lease loss is not recorded as a consumer failure. + ## Claiming Contract ```csharp @@ -165,6 +167,8 @@ public interface ITinyOutboxStore Provider claiming must be atomic. Query-then-update claiming is not acceptable for DB providers. +Provider completion and failure updates must validate affected row counts. A mark operation that updates no rows means the worker no longer owns a processing lease for that message. + New database providers must implement atomic claiming safely for their database engine. Query-then-update is not acceptable for multi-worker processing. ## Bootstrap diff --git a/docs/workers.md b/docs/workers.md index f13c80b..a1e0b90 100644 --- a/docs/workers.md +++ b/docs/workers.md @@ -77,6 +77,8 @@ The hosted worker: - registers an `IHostedService` - creates a scope per processing iteration - calls `ITinyOutboxProcessor.ProcessPendingAsync` +- logs processing-iteration failures +- continues polling after non-cancellation processing failures - waits `PollingInterval` - stops claiming new work when cancellation is requested @@ -84,6 +86,8 @@ The hosted worker: On shutdown, TinyEvents does not scan and release claims. If processing does not complete, claims expire naturally. +A hosted worker can remain running while processing iterations repeatedly fail, for example during a database outage. Treat worker logs and host-level health checks as part of production operations. + ## Marking Processed Or Failed Providers mark messages only when: @@ -94,6 +98,8 @@ Providers mark messages only when: This prevents worker A from marking worker B's work. +If the mark operation affects no rows, TinyEvents treats that as a lost lease. The processor does not record a failed attempt for that message and continues with the next claimed message. + ## Claim Timeout TinyEvents v1 does not implement claim renewal or heartbeat. From 72b8f1e0ad5544963eeaf781f643399868ee964d Mon Sep 17 00:00:00 2001 From: George Date: Sat, 25 Jul 2026 10:57:08 +0200 Subject: [PATCH 13/22] Document retry attempt semantics --- docs/architecture.md | 4 ++++ docs/workers.md | 24 ++++++++++++++++++++++++ 2 files changed, 28 insertions(+) diff --git a/docs/architecture.md b/docs/architecture.md index c19411a..5e308cd 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -130,6 +130,10 @@ One outbox message represents one event, not one message per consumer. If one co If marking a message as processed or failed no longer updates a row because the worker lost ownership of the processing lease, the processor leaves the message alone and continues. Lease loss is not recorded as a consumer failure. +Failures that are recorded through `MarkFailedAsync` advance the message attempt count. Before `MaxAttempts` is reached, the message is made pending again and delayed until `NextAttemptAtUtc`. When `MaxAttempts` is reached, the message is marked failed and no next attempt is scheduled. + +Unknown event types are processing failures because the processor cannot resolve a generated dispatcher for the stored event type. + ## Claiming Contract ```csharp diff --git a/docs/workers.md b/docs/workers.md index a1e0b90..092e49e 100644 --- a/docs/workers.md +++ b/docs/workers.md @@ -100,6 +100,30 @@ This prevents worker A from marking worker B's work. If the mark operation affects no rows, TinyEvents treats that as a lost lease. The processor does not record a failed attempt for that message and continues with the next claimed message. +## Retries And Attempts + +TinyEvents increments `AttemptCount` only when processing fails and the processor records that failure through the outbox store. + +When processing fails before `MaxAttempts` is reached: + +- `AttemptCount` is incremented +- `LastError` stores the failure message +- `Status` returns to `Pending` +- `NextAttemptAtUtc` is set to the current time plus `RetryDelay` + +When processing fails and the next attempt would reach `MaxAttempts`: + +- `AttemptCount` is incremented +- `LastError` stores the failure message +- `Status` becomes `Failed` +- `NextAttemptAtUtc` is cleared + +Cancellation requested through the worker cancellation token is not recorded as a failed attempt. + +Lost leases are not recorded as failed attempts. Another worker may already own the message or may reclaim it after the current lease expires. + +An unknown event type is treated as a processing failure. It follows the same attempt and retry rules as a consumer failure. + ## Claim Timeout TinyEvents v1 does not implement claim renewal or heartbeat. From c36bb0ecb3c00fd719c141f95992fb26ee05f658 Mon Sep 17 00:00:00 2001 From: George Date: Sat, 25 Jul 2026 11:00:06 +0200 Subject: [PATCH 14/22] Document unknown event type handling --- docs/source-generator.md | 2 ++ docs/workers.md | 15 +++++++++++++++ 2 files changed, 17 insertions(+) diff --git a/docs/source-generator.md b/docs/source-generator.md index e7f9ebb..8867d3b 100644 --- a/docs/source-generator.md +++ b/docs/source-generator.md @@ -63,6 +63,8 @@ Each assembly that contains consumers generates its own contribution. When that TinyEvents does not scan application assemblies or force-load referenced assemblies. If a consumer assembly has not been loaded before TinyEvents registration runs, that assembly's consumers and event dispatchers will not be registered in that service collection. +An outbox message whose stored event type has no registered dispatcher is treated as a processing failure by the worker. It follows the normal retry and max-attempt rules. Load consumer assemblies before TinyEvents registration so their generated contributions are available. + For example: ```csharp diff --git a/docs/workers.md b/docs/workers.md index 092e49e..7eec204 100644 --- a/docs/workers.md +++ b/docs/workers.md @@ -124,6 +124,21 @@ Lost leases are not recorded as failed attempts. Another worker may already own An unknown event type is treated as a processing failure. It follows the same attempt and retry rules as a consumer failure. +## Unknown Event Types + +TinyEvents stores the event type name in each outbox row. At processing time, that name must match a generated `ITinyEventDispatcher` registration in the current service provider. + +If no dispatcher is registered for the stored event type, TinyEvents records the message as a processing failure. The message follows the normal retry and max-attempt rules. + +Common causes are: + +- the assembly containing the consumer was not loaded before TinyEvents registration +- TinyEvents registration ran before generated contributions were available +- an old outbox row references an event type that the application no longer handles +- a row was inserted manually with an invalid event type name + +If the missing dispatcher is caused by startup or registration order, the message can succeed on a later attempt after the application is corrected. If the application will never handle that event type, the message eventually reaches `Failed`. + ## Claim Timeout TinyEvents v1 does not implement claim renewal or heartbeat. From 4e0c3934566f2a502c2530cec431f9e9418b898b Mon Sep 17 00:00:00 2001 From: George Date: Sat, 25 Jul 2026 11:03:33 +0200 Subject: [PATCH 15/22] Align SQL Server EF outbox string lengths --- .../TinyEventsModelBuilderExtensions.cs | 3 +- .../TinySqlServerEfCoreProviderTests.cs | 30 +++++++++++++++++++ 2 files changed, 32 insertions(+), 1 deletion(-) diff --git a/src/TinyEvents.SqlServer.EntityFrameworkCore/TinyEventsModelBuilderExtensions.cs b/src/TinyEvents.SqlServer.EntityFrameworkCore/TinyEventsModelBuilderExtensions.cs index 43d1974..c553c53 100644 --- a/src/TinyEvents.SqlServer.EntityFrameworkCore/TinyEventsModelBuilderExtensions.cs +++ b/src/TinyEvents.SqlServer.EntityFrameworkCore/TinyEventsModelBuilderExtensions.cs @@ -19,9 +19,10 @@ public static ModelBuilder UseTinyEventsOutbox( { entity.ToTable(parsedTableName.Table, parsedTableName.Schema); entity.HasKey(message => message.Id); - entity.Property(message => message.EventType).IsRequired(); + entity.Property(message => message.EventType).IsRequired().HasMaxLength(512); entity.Property(message => message.Payload).IsRequired(); entity.Property(message => message.Status).IsRequired(); + entity.Property(message => message.ClaimedBy).HasMaxLength(256); entity.Property(message => message.CreatedAtUtc).IsRequired(); entity.HasIndex(message => new diff --git a/tests/TinyEvents.SqlServer.EntityFrameworkCore.Tests/TinySqlServerEfCoreProviderTests.cs b/tests/TinyEvents.SqlServer.EntityFrameworkCore.Tests/TinySqlServerEfCoreProviderTests.cs index 15458cc..2ac497e 100644 --- a/tests/TinyEvents.SqlServer.EntityFrameworkCore.Tests/TinySqlServerEfCoreProviderTests.cs +++ b/tests/TinyEvents.SqlServer.EntityFrameworkCore.Tests/TinySqlServerEfCoreProviderTests.cs @@ -90,6 +90,26 @@ public void EF_model_builder_adds_claim_owner_lookup_index() index => HasProperties(index, nameof(TinyOutboxMessage.ClaimedBy), nameof(TinyOutboxMessage.Status))); } + [Fact] + public void EF_model_builder_limits_event_type_length() + { + using var dbContext = NewTestDbContext(); + + var property = GetOutboxProperty(dbContext, nameof(TinyOutboxMessage.EventType)); + + Assert.Equal(512, property.GetMaxLength()); + } + + [Fact] + public void EF_model_builder_limits_claimed_by_length() + { + using var dbContext = NewTestDbContext(); + + var property = GetOutboxProperty(dbContext, nameof(TinyOutboxMessage.ClaimedBy)); + + Assert.Equal(256, property.GetMaxLength()); + } + [Fact] public void Use_sql_server_entity_framework_core_outbox_registers_writer_and_store() { @@ -263,6 +283,16 @@ private static TestDbContext NewTestDbContext() return new TestDbContext(options); } + private static Microsoft.EntityFrameworkCore.Metadata.IProperty GetOutboxProperty( + DbContext dbContext, + string propertyName) + { + var entity = dbContext.Model.FindEntityType(typeof(TinyOutboxMessage)); + + Assert.NotNull(entity); + return entity.FindProperty(propertyName)!; + } + private static bool HasProperties( Microsoft.EntityFrameworkCore.Metadata.IReadOnlyIndex index, params string[] propertyNames) From 2313b8f15d5eacefca21697f124313d049690ce3 Mon Sep 17 00:00:00 2001 From: George Date: Sat, 25 Jul 2026 11:08:14 +0200 Subject: [PATCH 16/22] Document EF Core worker transaction behavior --- docs/architecture.md | 2 ++ docs/ef-core.md | 6 ++++++ docs/postgresql/ef-core.md | 4 ++++ docs/sql-server/ef-core.md | 4 ++++ 4 files changed, 16 insertions(+) diff --git a/docs/architecture.md b/docs/architecture.md index 5e308cd..5f5bf9e 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -115,6 +115,8 @@ EF Core publishing adds the message to the current `DbContext`. ADO.NET publishing inserts the message through the current application transaction. +EF Core worker claim and mark commands use the scoped `DbContext` relational connection and attach to `DbContext.Database.CurrentTransaction` when one exists. TinyEvents does not create or complete EF Core transactions for worker operations. + ## Processing Flow 1. Resolve the current worker id. diff --git a/docs/ef-core.md b/docs/ef-core.md index eb30cae..1d6cb5e 100644 --- a/docs/ef-core.md +++ b/docs/ef-core.md @@ -21,3 +21,9 @@ Worker claiming is database-specific: - SQL Server uses SQL Server locking hints and atomic update/output SQL. - PostgreSQL uses `FOR UPDATE SKIP LOCKED` inside an atomic update/returning SQL statement. + +Worker claim and mark operations use the relational connection from the scoped `DbContext`. + +If `DbContext.Database.CurrentTransaction` exists, TinyEvents attaches the claim or mark command to that transaction. TinyEvents does not begin, commit, or roll back an EF Core transaction for worker operations. + +The hosted worker creates a fresh dependency-injection scope for each processing iteration. In normal hosted-worker usage, claim and mark commands run without an application transaction unless your application explicitly creates one in that worker scope. diff --git a/docs/postgresql/ef-core.md b/docs/postgresql/ef-core.md index c6a1606..5844478 100644 --- a/docs/postgresql/ef-core.md +++ b/docs/postgresql/ef-core.md @@ -61,6 +61,10 @@ The PostgreSQL EF Core store opens the underlying relational connection when nee Claiming is atomic and lease-based. PostgreSQL uses `FOR UPDATE SKIP LOCKED` inside an update/returning statement. +If the scoped `DbContext` has a current EF Core transaction, TinyEvents attaches claim and mark commands to that transaction. TinyEvents does not start, commit, or roll back that transaction. + +The hosted worker creates a fresh scope for each processing iteration, so claim and mark commands normally run outside an application-owned transaction. + ## Migrations Use normal EF Core migrations: diff --git a/docs/sql-server/ef-core.md b/docs/sql-server/ef-core.md index ddd7918..e5232c8 100644 --- a/docs/sql-server/ef-core.md +++ b/docs/sql-server/ef-core.md @@ -61,6 +61,10 @@ The SQL Server EF Core store opens the underlying relational connection when nee Claiming is atomic and lease-based. SQL Server uses locking hints and update/output SQL. +If the scoped `DbContext` has a current EF Core transaction, TinyEvents attaches claim and mark commands to that transaction. TinyEvents does not start, commit, or roll back that transaction. + +The hosted worker creates a fresh scope for each processing iteration, so claim and mark commands normally run outside an application-owned transaction. + ## Migrations Use normal EF Core migrations: From d541b530526cb0029bf238f0673208861d7c593f Mon Sep 17 00:00:00 2001 From: George Date: Sat, 25 Jul 2026 11:12:35 +0200 Subject: [PATCH 17/22] Document provider schema parity --- docs/postgresql/ado-net.md | 2 ++ docs/postgresql/ef-core.md | 2 ++ docs/schema-and-migrations.md | 5 +++++ docs/sql-server/ado-net.md | 2 ++ docs/sql-server/ef-core.md | 2 ++ 5 files changed, 13 insertions(+) diff --git a/docs/postgresql/ado-net.md b/docs/postgresql/ado-net.md index 65e76b9..b644484 100644 --- a/docs/postgresql/ado-net.md +++ b/docs/postgresql/ado-net.md @@ -86,3 +86,5 @@ The package also includes the default PostgreSQL script: ```text schema/postgresql/001_CreateTinyOutbox.sql ``` + +The default PostgreSQL schema uses `text` for `EventType`, `Payload`, `ClaimedBy`, and `LastError`. diff --git a/docs/postgresql/ef-core.md b/docs/postgresql/ef-core.md index 5844478..eb02b16 100644 --- a/docs/postgresql/ef-core.md +++ b/docs/postgresql/ef-core.md @@ -55,6 +55,8 @@ protected override void OnModelCreating(ModelBuilder modelBuilder) The provider option controls SQL claiming and marking. The model builder extension controls EF mapping and migrations. +The PostgreSQL mapping uses `text` for `EventType`, `Payload`, `ClaimedBy`, and `LastError`. + ## Worker Claiming The PostgreSQL EF Core store opens the underlying relational connection when needed and executes PostgreSQL claim/mark statements. diff --git a/docs/schema-and-migrations.md b/docs/schema-and-migrations.md index 5eca842..535c29e 100644 --- a/docs/schema-and-migrations.md +++ b/docs/schema-and-migrations.md @@ -33,6 +33,11 @@ The v1 statuses are: - `Processed` - `Failed` +Provider schemas keep the same logical columns and indexes, but database types follow each provider: + +- SQL Server maps `EventType` to `NVARCHAR(512)`, `Payload` to `NVARCHAR(MAX)`, `ClaimedBy` to `NVARCHAR(256)`, and `LastError` to `NVARCHAR(MAX)`. +- PostgreSQL maps `EventType`, `Payload`, `ClaimedBy`, and `LastError` to `text`. + ## EF Core EF Core applications create the schema through normal EF Core migrations. diff --git a/docs/sql-server/ado-net.md b/docs/sql-server/ado-net.md index b2dd7d7..de27ac3 100644 --- a/docs/sql-server/ado-net.md +++ b/docs/sql-server/ado-net.md @@ -86,3 +86,5 @@ The package also includes the default SQL Server script: ```text schema/sqlserver/001_CreateTinyOutbox.sql ``` + +The default SQL Server schema uses `NVARCHAR(512)` for `EventType`, `NVARCHAR(MAX)` for `Payload`, `NVARCHAR(256)` for `ClaimedBy`, and `NVARCHAR(MAX)` for `LastError`. diff --git a/docs/sql-server/ef-core.md b/docs/sql-server/ef-core.md index e5232c8..044879e 100644 --- a/docs/sql-server/ef-core.md +++ b/docs/sql-server/ef-core.md @@ -55,6 +55,8 @@ protected override void OnModelCreating(ModelBuilder modelBuilder) The provider option controls SQL claiming and marking. The model builder extension controls EF mapping and migrations. +The SQL Server mapping uses `NVARCHAR(512)` for `EventType`, `NVARCHAR(MAX)` for `Payload`, `NVARCHAR(256)` for `ClaimedBy`, and `NVARCHAR(MAX)` for `LastError`. + ## Worker Claiming The SQL Server EF Core store opens the underlying relational connection when needed and executes SQL Server claim/mark statements. From 158f2d8765c94b62a661ee1e7200f66cde5bfbb2 Mon Sep 17 00:00:00 2001 From: George Date: Sat, 25 Jul 2026 11:18:12 +0200 Subject: [PATCH 18/22] Log outbox processing outcomes --- .../Processing/TinyOutboxProcessor.cs | 77 ++++++++++++++++++- src/TinyEvents/TinyEvents.csproj | 1 + .../Processing/TinyOutboxProcessorTests.cs | 75 +++++++++++++++++- 3 files changed, 146 insertions(+), 7 deletions(-) diff --git a/src/TinyEvents/Processing/TinyOutboxProcessor.cs b/src/TinyEvents/Processing/TinyOutboxProcessor.cs index 5b39645..cbbfded 100644 --- a/src/TinyEvents/Processing/TinyOutboxProcessor.cs +++ b/src/TinyEvents/Processing/TinyOutboxProcessor.cs @@ -1,3 +1,6 @@ +using Microsoft.Extensions.Logging; +using Microsoft.Extensions.Logging.Abstractions; + namespace TinyEvents; public sealed class TinyOutboxProcessor : ITinyOutboxProcessor @@ -8,6 +11,7 @@ public sealed class TinyOutboxProcessor : ITinyOutboxProcessor private readonly Dictionary dispatchers; private readonly TinyEventsOptions options; private readonly TimeProvider timeProvider; + private readonly ILogger logger; public TinyOutboxProcessor( IServiceProvider serviceProvider, @@ -16,6 +20,25 @@ public TinyOutboxProcessor( IEnumerable dispatchers, TinyEventsOptions options, TimeProvider timeProvider) + : this( + serviceProvider, + store, + serializer, + dispatchers, + options, + timeProvider, + NullLogger.Instance) + { + } + + public TinyOutboxProcessor( + IServiceProvider serviceProvider, + ITinyOutboxStore store, + ITinyEventSerializer serializer, + IEnumerable dispatchers, + TinyEventsOptions options, + TimeProvider timeProvider, + ILogger logger) { if (serviceProvider is null) { @@ -47,12 +70,18 @@ public TinyOutboxProcessor( throw new ArgumentNullException(nameof(timeProvider)); } + if (logger is null) + { + throw new ArgumentNullException(nameof(logger)); + } + this.serviceProvider = serviceProvider; this.store = store; this.serializer = serializer; this.dispatchers = BuildDispatcherMap(dispatchers); this.options = options; this.timeProvider = timeProvider; + this.logger = logger; } public async ValueTask ProcessPendingAsync(CancellationToken cancellationToken = default) @@ -88,17 +117,20 @@ private async ValueTask ProcessMessageAsync( { throw; } - catch (TinyOutboxLeaseLostException) + catch (TinyOutboxLeaseLostException exception) { + LogLeaseLost(message, workerId, exception, "marked as processed"); } catch (Exception exception) { try { - await MarkFailedAsync(message, workerId, exception, cancellationToken); + var failure = await MarkFailedAsync(message, workerId, exception, cancellationToken); + LogProcessingFailure(message, workerId, exception, failure); } - catch (TinyOutboxLeaseLostException) + catch (TinyOutboxLeaseLostException leaseLostException) { + LogLeaseLost(message, workerId, leaseLostException, "recording a processing failure"); } } } @@ -134,7 +166,7 @@ await store.MarkProcessedAsync( cancellationToken); } - private async ValueTask MarkFailedAsync( + private async ValueTask MarkFailedAsync( TinyOutboxMessage message, string workerId, Exception exception, @@ -150,6 +182,8 @@ await store.MarkFailedAsync( attemptCount, nextAttemptAtUtc, cancellationToken); + + return new RecordedFailure(attemptCount, nextAttemptAtUtc); } private DateTimeOffset? GetNextAttemptAtUtc(int attemptCount) @@ -191,4 +225,39 @@ private static void AddDispatcher( dispatchers.Add(dispatcher.EventTypeName, dispatcher); } + + private void LogProcessingFailure( + TinyOutboxMessage message, + string workerId, + Exception exception, + RecordedFailure failure) + { + logger.LogWarning( + exception, + "TinyEvents outbox message {MessageId} for event type {EventType} failed processing on worker {WorkerId} at attempt {AttemptCount}. Next attempt at {NextAttemptAtUtc}.", + message.Id, + message.EventType, + workerId, + failure.AttemptCount, + failure.NextAttemptAtUtc); + } + + private void LogLeaseLost( + TinyOutboxMessage message, + string workerId, + TinyOutboxLeaseLostException exception, + string operation) + { + logger.LogWarning( + exception, + "TinyEvents outbox message {MessageId} for event type {EventType} lost its processing lease while {Operation} on worker {WorkerId}.", + message.Id, + message.EventType, + operation, + workerId); + } + + private readonly record struct RecordedFailure( + int AttemptCount, + DateTimeOffset? NextAttemptAtUtc); } diff --git a/src/TinyEvents/TinyEvents.csproj b/src/TinyEvents/TinyEvents.csproj index ee338ad..cd105b3 100644 --- a/src/TinyEvents/TinyEvents.csproj +++ b/src/TinyEvents/TinyEvents.csproj @@ -18,6 +18,7 @@ + diff --git a/tests/TinyEvents.Tests/Processing/TinyOutboxProcessorTests.cs b/tests/TinyEvents.Tests/Processing/TinyOutboxProcessorTests.cs index 063774f..1221867 100644 --- a/tests/TinyEvents.Tests/Processing/TinyOutboxProcessorTests.cs +++ b/tests/TinyEvents.Tests/Processing/TinyOutboxProcessorTests.cs @@ -1,4 +1,5 @@ using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Logging; using Xunit; namespace TinyEvents.Tests; @@ -174,6 +175,27 @@ public async Task Process_pending_async_marks_failed_and_schedules_retry_when_co ThrowingConsumer.Throw = false; } + [Fact] + public async Task Process_pending_async_logs_recorded_processing_failure() + { + ThrowingConsumer.Throw = true; + var logger = new RecordingLogger(); + var store = new InMemoryTinyOutboxStore(); + await store.AddAsync(NewPendingMessage(new UserCreated(Guid.NewGuid(), "user@example.com")), CancellationToken.None); + var processor = BuildProcessor( + store, + includeThrowingConsumer: true, + logger: logger); + + await processor.ProcessPendingAsync(); + + var entry = Assert.Single(logger.Entries); + Assert.Equal(LogLevel.Warning, entry.LogLevel); + Assert.Contains("failed processing", entry.Message, StringComparison.Ordinal); + Assert.IsType(entry.Exception); + ThrowingConsumer.Throw = false; + } + [Fact] public async Task Process_pending_async_marks_processed_with_current_worker_id() { @@ -300,29 +322,39 @@ await Assert.ThrowsAsync( public async Task Process_pending_async_continues_when_mark_processed_loses_lease() { RecordingConsumer.Consumed.Clear(); + var logger = new RecordingLogger(); var firstMessage = NewProcessingMessage(new UserCreated(Guid.NewGuid(), "first@example.com")); var secondMessage = NewProcessingMessage(new UserCreated(Guid.NewGuid(), "second@example.com")); var store = new LeaseLostOnFirstProcessedStore(firstMessage, secondMessage); - var processor = BuildProcessor(store); + var processor = BuildProcessor(store, logger: logger); await processor.ProcessPendingAsync(); Assert.Equal(2, RecordingConsumer.Consumed.Count); Assert.Equal(secondMessage.Id, Assert.Single(store.ProcessedMessageIds)); Assert.Equal(0, store.MarkFailedCount); + Assert.Contains(logger.Entries, entry => + entry.LogLevel == LogLevel.Warning + && entry.Message.Contains("lost its processing lease", StringComparison.Ordinal) + && entry.Exception is TinyOutboxLeaseLostException); } [Fact] public async Task Process_pending_async_continues_when_mark_failed_loses_lease() { ThrowingConsumer.Throw = true; + var logger = new RecordingLogger(); var message = NewProcessingMessage(new UserCreated(Guid.NewGuid(), "user@example.com")); var store = new LeaseLostOnFailedStore(message); - var processor = BuildProcessor(store, includeThrowingConsumer: true); + var processor = BuildProcessor(store, includeThrowingConsumer: true, logger: logger); await processor.ProcessPendingAsync(); Assert.Equal(1, store.MarkFailedCount); + Assert.Contains(logger.Entries, entry => + entry.LogLevel == LogLevel.Warning + && entry.Message.Contains("lost its processing lease", StringComparison.Ordinal) + && entry.Exception is TinyOutboxLeaseLostException); ThrowingConsumer.Throw = false; } @@ -334,7 +366,8 @@ private static ITinyOutboxProcessor BuildProcessor( TimeProvider? timeProvider = null, string? workerId = "worker-1", int batchSize = 10, - TimeSpan? claimTimeout = null) + TimeSpan? claimTimeout = null, + ILogger? logger = null) { var services = new ServiceCollection(); @@ -354,6 +387,11 @@ private static ITinyOutboxProcessor BuildProcessor( services.AddSingleton(); services.AddSingleton, RecordingConsumer>(); + if (logger is not null) + { + services.AddSingleton(logger); + } + if (includeSecondConsumer) { services.AddSingleton, SecondRecordingConsumer>(); @@ -701,4 +739,35 @@ public ValueTask MarkFailedAsync( throw new TinyOutboxLeaseLostException(messageId, workerId, "failed"); } } + + private sealed class RecordingLogger : ILogger + { + public List Entries { get; } = new List(); + + public IDisposable? BeginScope(TState state) + where TState : notnull + { + return null; + } + + public bool IsEnabled(LogLevel logLevel) + { + return true; + } + + public void Log( + LogLevel logLevel, + EventId eventId, + TState state, + Exception? exception, + Func formatter) + { + Entries.Add(new LogEntry(logLevel, formatter(state, exception), exception)); + } + } + + private sealed record LogEntry( + LogLevel LogLevel, + string Message, + Exception? Exception); } From 51d6408b503a07d20bc310f68ebfbb1d4f77462b Mon Sep 17 00:00:00 2001 From: George Date: Sat, 25 Jul 2026 11:28:53 +0200 Subject: [PATCH 19/22] Hide provider implementation details --- .../ITinyPostgreSqlAdoNetWorkerConnectionFactory.cs | 2 +- .../TinyPostgreSqlAdoNetWorkerConnectionFactory.cs | 2 +- src/TinyEvents.PostgreSql.AdoNet/Properties/AssemblyInfo.cs | 4 ++++ .../Sql/TinyPostgreSqlAdoNetSql.cs | 2 +- .../Sql/TinyPostgreSqlAdoNetTableName.cs | 2 +- .../TinyPostgreSqlAdoNetOutboxStore.cs | 2 +- .../TinyPostgreSqlAdoNetOutboxWriter.cs | 2 +- .../Properties/AssemblyInfo.cs | 4 ++++ .../Sql/TinyPostgreSqlEfCoreSql.cs | 2 +- .../Sql/TinyPostgreSqlEfCoreTableName.cs | 2 +- .../TinyPostgreSqlEfCoreOutboxStore.cs | 2 +- .../TinyPostgreSqlEfCoreOutboxWriter.cs | 2 +- .../ITinySqlServerAdoNetWorkerConnectionFactory.cs | 2 +- .../Connections/TinySqlServerAdoNetWorkerConnectionFactory.cs | 2 +- src/TinyEvents.SqlServer.AdoNet/Properties/AssemblyInfo.cs | 4 ++++ src/TinyEvents.SqlServer.AdoNet/Sql/TinySqlServerAdoNetSql.cs | 3 +-- .../Sql/TinySqlServerAdoNetTableName.cs | 2 +- .../TinySqlServerAdoNetOutboxStore.cs | 2 +- .../TinySqlServerAdoNetOutboxWriter.cs | 2 +- .../Properties/AssemblyInfo.cs | 4 ++++ .../Sql/TinySqlServerEfCoreSql.cs | 3 +-- .../Sql/TinySqlServerEfCoreTableName.cs | 2 +- .../TinySqlServerEfCoreOutboxStore.cs | 2 +- .../TinySqlServerEfCoreOutboxWriter.cs | 3 +-- 24 files changed, 36 insertions(+), 23 deletions(-) create mode 100644 src/TinyEvents.PostgreSql.AdoNet/Properties/AssemblyInfo.cs create mode 100644 src/TinyEvents.PostgreSql.EntityFrameworkCore/Properties/AssemblyInfo.cs create mode 100644 src/TinyEvents.SqlServer.AdoNet/Properties/AssemblyInfo.cs create mode 100644 src/TinyEvents.SqlServer.EntityFrameworkCore/Properties/AssemblyInfo.cs diff --git a/src/TinyEvents.PostgreSql.AdoNet/Connections/ITinyPostgreSqlAdoNetWorkerConnectionFactory.cs b/src/TinyEvents.PostgreSql.AdoNet/Connections/ITinyPostgreSqlAdoNetWorkerConnectionFactory.cs index e973141..615cb08 100644 --- a/src/TinyEvents.PostgreSql.AdoNet/Connections/ITinyPostgreSqlAdoNetWorkerConnectionFactory.cs +++ b/src/TinyEvents.PostgreSql.AdoNet/Connections/ITinyPostgreSqlAdoNetWorkerConnectionFactory.cs @@ -2,7 +2,7 @@ namespace TinyEvents.PostgreSql.AdoNet; -public interface ITinyPostgreSqlAdoNetWorkerConnectionFactory +internal interface ITinyPostgreSqlAdoNetWorkerConnectionFactory { ValueTask CreateOpenConnectionAsync(CancellationToken cancellationToken); } diff --git a/src/TinyEvents.PostgreSql.AdoNet/Connections/TinyPostgreSqlAdoNetWorkerConnectionFactory.cs b/src/TinyEvents.PostgreSql.AdoNet/Connections/TinyPostgreSqlAdoNetWorkerConnectionFactory.cs index 521caf6..ce4d73e 100644 --- a/src/TinyEvents.PostgreSql.AdoNet/Connections/TinyPostgreSqlAdoNetWorkerConnectionFactory.cs +++ b/src/TinyEvents.PostgreSql.AdoNet/Connections/TinyPostgreSqlAdoNetWorkerConnectionFactory.cs @@ -3,7 +3,7 @@ namespace TinyEvents.PostgreSql.AdoNet; -public sealed class TinyPostgreSqlAdoNetWorkerConnectionFactory : ITinyPostgreSqlAdoNetWorkerConnectionFactory +internal sealed class TinyPostgreSqlAdoNetWorkerConnectionFactory : ITinyPostgreSqlAdoNetWorkerConnectionFactory { private readonly TinyEventsPostgreSqlAdoNetOptions options; private readonly IServiceProvider serviceProvider; diff --git a/src/TinyEvents.PostgreSql.AdoNet/Properties/AssemblyInfo.cs b/src/TinyEvents.PostgreSql.AdoNet/Properties/AssemblyInfo.cs new file mode 100644 index 0000000..e51eb43 --- /dev/null +++ b/src/TinyEvents.PostgreSql.AdoNet/Properties/AssemblyInfo.cs @@ -0,0 +1,4 @@ +using System.Runtime.CompilerServices; + +[assembly: InternalsVisibleTo("TinyEvents.PostgreSql.AdoNet.Tests")] +[assembly: InternalsVisibleTo("TinyEvents.PostgreSql.Tests")] diff --git a/src/TinyEvents.PostgreSql.AdoNet/Sql/TinyPostgreSqlAdoNetSql.cs b/src/TinyEvents.PostgreSql.AdoNet/Sql/TinyPostgreSqlAdoNetSql.cs index 1211514..4352716 100644 --- a/src/TinyEvents.PostgreSql.AdoNet/Sql/TinyPostgreSqlAdoNetSql.cs +++ b/src/TinyEvents.PostgreSql.AdoNet/Sql/TinyPostgreSqlAdoNetSql.cs @@ -1,6 +1,6 @@ namespace TinyEvents.PostgreSql.AdoNet; -public static class TinyPostgreSqlAdoNetSql +internal static class TinyPostgreSqlAdoNetSql { public static string Insert(TinyPostgreSqlAdoNetTableName tableName) { diff --git a/src/TinyEvents.PostgreSql.AdoNet/Sql/TinyPostgreSqlAdoNetTableName.cs b/src/TinyEvents.PostgreSql.AdoNet/Sql/TinyPostgreSqlAdoNetTableName.cs index fc2e7c1..fbaa101 100644 --- a/src/TinyEvents.PostgreSql.AdoNet/Sql/TinyPostgreSqlAdoNetTableName.cs +++ b/src/TinyEvents.PostgreSql.AdoNet/Sql/TinyPostgreSqlAdoNetTableName.cs @@ -1,6 +1,6 @@ namespace TinyEvents.PostgreSql.AdoNet; -public sealed class TinyPostgreSqlAdoNetTableName +internal sealed class TinyPostgreSqlAdoNetTableName { private readonly string[] parts; diff --git a/src/TinyEvents.PostgreSql.AdoNet/TinyPostgreSqlAdoNetOutboxStore.cs b/src/TinyEvents.PostgreSql.AdoNet/TinyPostgreSqlAdoNetOutboxStore.cs index 61f5c2f..859d981 100644 --- a/src/TinyEvents.PostgreSql.AdoNet/TinyPostgreSqlAdoNetOutboxStore.cs +++ b/src/TinyEvents.PostgreSql.AdoNet/TinyPostgreSqlAdoNetOutboxStore.cs @@ -2,7 +2,7 @@ namespace TinyEvents.PostgreSql.AdoNet; -public sealed class TinyPostgreSqlAdoNetOutboxStore : ITinyOutboxStore +internal sealed class TinyPostgreSqlAdoNetOutboxStore : ITinyOutboxStore { private readonly ITinyPostgreSqlAdoNetWorkerConnectionFactory connectionFactory; private readonly TinyPostgreSqlAdoNetTableName tableName; diff --git a/src/TinyEvents.PostgreSql.AdoNet/TinyPostgreSqlAdoNetOutboxWriter.cs b/src/TinyEvents.PostgreSql.AdoNet/TinyPostgreSqlAdoNetOutboxWriter.cs index 87c1284..f3d591b 100644 --- a/src/TinyEvents.PostgreSql.AdoNet/TinyPostgreSqlAdoNetOutboxWriter.cs +++ b/src/TinyEvents.PostgreSql.AdoNet/TinyPostgreSqlAdoNetOutboxWriter.cs @@ -1,6 +1,6 @@ namespace TinyEvents.PostgreSql.AdoNet; -public sealed class TinyPostgreSqlAdoNetOutboxWriter : ITinyOutboxWriter +internal sealed class TinyPostgreSqlAdoNetOutboxWriter : ITinyOutboxWriter { private readonly TinyEventsPostgreSqlAdoNetOptions options; private readonly IServiceProvider serviceProvider; diff --git a/src/TinyEvents.PostgreSql.EntityFrameworkCore/Properties/AssemblyInfo.cs b/src/TinyEvents.PostgreSql.EntityFrameworkCore/Properties/AssemblyInfo.cs new file mode 100644 index 0000000..e20b439 --- /dev/null +++ b/src/TinyEvents.PostgreSql.EntityFrameworkCore/Properties/AssemblyInfo.cs @@ -0,0 +1,4 @@ +using System.Runtime.CompilerServices; + +[assembly: InternalsVisibleTo("TinyEvents.PostgreSql.EntityFrameworkCore.Tests")] +[assembly: InternalsVisibleTo("TinyEvents.PostgreSql.Tests")] diff --git a/src/TinyEvents.PostgreSql.EntityFrameworkCore/Sql/TinyPostgreSqlEfCoreSql.cs b/src/TinyEvents.PostgreSql.EntityFrameworkCore/Sql/TinyPostgreSqlEfCoreSql.cs index 85306eb..48c0ab3 100644 --- a/src/TinyEvents.PostgreSql.EntityFrameworkCore/Sql/TinyPostgreSqlEfCoreSql.cs +++ b/src/TinyEvents.PostgreSql.EntityFrameworkCore/Sql/TinyPostgreSqlEfCoreSql.cs @@ -1,6 +1,6 @@ namespace TinyEvents.PostgreSql.EntityFrameworkCore; -public static class TinyPostgreSqlEfCoreSql +internal static class TinyPostgreSqlEfCoreSql { public static string ClaimPending(TinyPostgreSqlEfCoreTableName tableName) { diff --git a/src/TinyEvents.PostgreSql.EntityFrameworkCore/Sql/TinyPostgreSqlEfCoreTableName.cs b/src/TinyEvents.PostgreSql.EntityFrameworkCore/Sql/TinyPostgreSqlEfCoreTableName.cs index c484966..28fbcef 100644 --- a/src/TinyEvents.PostgreSql.EntityFrameworkCore/Sql/TinyPostgreSqlEfCoreTableName.cs +++ b/src/TinyEvents.PostgreSql.EntityFrameworkCore/Sql/TinyPostgreSqlEfCoreTableName.cs @@ -1,6 +1,6 @@ namespace TinyEvents.PostgreSql.EntityFrameworkCore; -public sealed class TinyPostgreSqlEfCoreTableName +internal sealed class TinyPostgreSqlEfCoreTableName { private readonly string[] parts; diff --git a/src/TinyEvents.PostgreSql.EntityFrameworkCore/TinyPostgreSqlEfCoreOutboxStore.cs b/src/TinyEvents.PostgreSql.EntityFrameworkCore/TinyPostgreSqlEfCoreOutboxStore.cs index acab25b..982b2d4 100644 --- a/src/TinyEvents.PostgreSql.EntityFrameworkCore/TinyPostgreSqlEfCoreOutboxStore.cs +++ b/src/TinyEvents.PostgreSql.EntityFrameworkCore/TinyPostgreSqlEfCoreOutboxStore.cs @@ -5,7 +5,7 @@ namespace TinyEvents.PostgreSql.EntityFrameworkCore; -public sealed class TinyPostgreSqlEfCoreOutboxStore : ITinyOutboxStore +internal sealed class TinyPostgreSqlEfCoreOutboxStore : ITinyOutboxStore where TDbContext : DbContext { private readonly TDbContext dbContext; diff --git a/src/TinyEvents.PostgreSql.EntityFrameworkCore/TinyPostgreSqlEfCoreOutboxWriter.cs b/src/TinyEvents.PostgreSql.EntityFrameworkCore/TinyPostgreSqlEfCoreOutboxWriter.cs index 5ab591c..d6ef78f 100644 --- a/src/TinyEvents.PostgreSql.EntityFrameworkCore/TinyPostgreSqlEfCoreOutboxWriter.cs +++ b/src/TinyEvents.PostgreSql.EntityFrameworkCore/TinyPostgreSqlEfCoreOutboxWriter.cs @@ -2,7 +2,7 @@ namespace TinyEvents.PostgreSql.EntityFrameworkCore; -public sealed class TinyPostgreSqlEfCoreOutboxWriter : ITinyOutboxWriter +internal sealed class TinyPostgreSqlEfCoreOutboxWriter : ITinyOutboxWriter where TDbContext : DbContext { private readonly TDbContext dbContext; diff --git a/src/TinyEvents.SqlServer.AdoNet/Connections/ITinySqlServerAdoNetWorkerConnectionFactory.cs b/src/TinyEvents.SqlServer.AdoNet/Connections/ITinySqlServerAdoNetWorkerConnectionFactory.cs index 15ab303..db63b29 100644 --- a/src/TinyEvents.SqlServer.AdoNet/Connections/ITinySqlServerAdoNetWorkerConnectionFactory.cs +++ b/src/TinyEvents.SqlServer.AdoNet/Connections/ITinySqlServerAdoNetWorkerConnectionFactory.cs @@ -2,7 +2,7 @@ namespace TinyEvents.SqlServer.AdoNet; -public interface ITinySqlServerAdoNetWorkerConnectionFactory +internal interface ITinySqlServerAdoNetWorkerConnectionFactory { ValueTask CreateOpenConnectionAsync(CancellationToken cancellationToken); } diff --git a/src/TinyEvents.SqlServer.AdoNet/Connections/TinySqlServerAdoNetWorkerConnectionFactory.cs b/src/TinyEvents.SqlServer.AdoNet/Connections/TinySqlServerAdoNetWorkerConnectionFactory.cs index ba662ee..cdc615b 100644 --- a/src/TinyEvents.SqlServer.AdoNet/Connections/TinySqlServerAdoNetWorkerConnectionFactory.cs +++ b/src/TinyEvents.SqlServer.AdoNet/Connections/TinySqlServerAdoNetWorkerConnectionFactory.cs @@ -3,7 +3,7 @@ namespace TinyEvents.SqlServer.AdoNet; -public sealed class TinySqlServerAdoNetWorkerConnectionFactory : ITinySqlServerAdoNetWorkerConnectionFactory +internal sealed class TinySqlServerAdoNetWorkerConnectionFactory : ITinySqlServerAdoNetWorkerConnectionFactory { private readonly TinyEventsSqlServerAdoNetOptions options; private readonly IServiceProvider serviceProvider; diff --git a/src/TinyEvents.SqlServer.AdoNet/Properties/AssemblyInfo.cs b/src/TinyEvents.SqlServer.AdoNet/Properties/AssemblyInfo.cs new file mode 100644 index 0000000..7d5a29f --- /dev/null +++ b/src/TinyEvents.SqlServer.AdoNet/Properties/AssemblyInfo.cs @@ -0,0 +1,4 @@ +using System.Runtime.CompilerServices; + +[assembly: InternalsVisibleTo("TinyEvents.SqlServer.AdoNet.Tests")] +[assembly: InternalsVisibleTo("TinyEvents.SqlServer.Tests")] diff --git a/src/TinyEvents.SqlServer.AdoNet/Sql/TinySqlServerAdoNetSql.cs b/src/TinyEvents.SqlServer.AdoNet/Sql/TinySqlServerAdoNetSql.cs index 81a3491..9548a60 100644 --- a/src/TinyEvents.SqlServer.AdoNet/Sql/TinySqlServerAdoNetSql.cs +++ b/src/TinyEvents.SqlServer.AdoNet/Sql/TinySqlServerAdoNetSql.cs @@ -1,6 +1,6 @@ namespace TinyEvents.SqlServer.AdoNet; -public static class TinySqlServerAdoNetSql +internal static class TinySqlServerAdoNetSql { public static string Insert(TinySqlServerAdoNetTableName tableName) { @@ -129,4 +129,3 @@ public static string MarkFailed(TinySqlServerAdoNetTableName tableName) """; } } - diff --git a/src/TinyEvents.SqlServer.AdoNet/Sql/TinySqlServerAdoNetTableName.cs b/src/TinyEvents.SqlServer.AdoNet/Sql/TinySqlServerAdoNetTableName.cs index e0fb915..78959a6 100644 --- a/src/TinyEvents.SqlServer.AdoNet/Sql/TinySqlServerAdoNetTableName.cs +++ b/src/TinyEvents.SqlServer.AdoNet/Sql/TinySqlServerAdoNetTableName.cs @@ -1,6 +1,6 @@ namespace TinyEvents.SqlServer.AdoNet; -public sealed class TinySqlServerAdoNetTableName +internal sealed class TinySqlServerAdoNetTableName { private readonly string[] parts; diff --git a/src/TinyEvents.SqlServer.AdoNet/TinySqlServerAdoNetOutboxStore.cs b/src/TinyEvents.SqlServer.AdoNet/TinySqlServerAdoNetOutboxStore.cs index 579d20b..7841437 100644 --- a/src/TinyEvents.SqlServer.AdoNet/TinySqlServerAdoNetOutboxStore.cs +++ b/src/TinyEvents.SqlServer.AdoNet/TinySqlServerAdoNetOutboxStore.cs @@ -3,7 +3,7 @@ namespace TinyEvents.SqlServer.AdoNet; -public sealed class TinySqlServerAdoNetOutboxStore : ITinyOutboxStore +internal sealed class TinySqlServerAdoNetOutboxStore : ITinyOutboxStore { private readonly TinyEventsSqlServerAdoNetOptions options; private readonly ITinySqlServerAdoNetWorkerConnectionFactory connectionFactory; diff --git a/src/TinyEvents.SqlServer.AdoNet/TinySqlServerAdoNetOutboxWriter.cs b/src/TinyEvents.SqlServer.AdoNet/TinySqlServerAdoNetOutboxWriter.cs index 69cde2c..0ca881e 100644 --- a/src/TinyEvents.SqlServer.AdoNet/TinySqlServerAdoNetOutboxWriter.cs +++ b/src/TinyEvents.SqlServer.AdoNet/TinySqlServerAdoNetOutboxWriter.cs @@ -1,6 +1,6 @@ namespace TinyEvents.SqlServer.AdoNet; -public sealed class TinySqlServerAdoNetOutboxWriter : ITinyOutboxWriter +internal sealed class TinySqlServerAdoNetOutboxWriter : ITinyOutboxWriter { private readonly TinyEventsSqlServerAdoNetOptions options; private readonly IServiceProvider serviceProvider; diff --git a/src/TinyEvents.SqlServer.EntityFrameworkCore/Properties/AssemblyInfo.cs b/src/TinyEvents.SqlServer.EntityFrameworkCore/Properties/AssemblyInfo.cs new file mode 100644 index 0000000..ebb3dfd --- /dev/null +++ b/src/TinyEvents.SqlServer.EntityFrameworkCore/Properties/AssemblyInfo.cs @@ -0,0 +1,4 @@ +using System.Runtime.CompilerServices; + +[assembly: InternalsVisibleTo("TinyEvents.SqlServer.EntityFrameworkCore.Tests")] +[assembly: InternalsVisibleTo("TinyEvents.SqlServer.Tests")] diff --git a/src/TinyEvents.SqlServer.EntityFrameworkCore/Sql/TinySqlServerEfCoreSql.cs b/src/TinyEvents.SqlServer.EntityFrameworkCore/Sql/TinySqlServerEfCoreSql.cs index df29ab0..662553f 100644 --- a/src/TinyEvents.SqlServer.EntityFrameworkCore/Sql/TinySqlServerEfCoreSql.cs +++ b/src/TinyEvents.SqlServer.EntityFrameworkCore/Sql/TinySqlServerEfCoreSql.cs @@ -1,6 +1,6 @@ namespace TinyEvents.SqlServer.EntityFrameworkCore; -public static class TinySqlServerEfCoreSql +internal static class TinySqlServerEfCoreSql { public static string ClaimPending(TinySqlServerEfCoreTableName tableName) { @@ -88,4 +88,3 @@ public static string MarkFailed(TinySqlServerEfCoreTableName tableName) """; } } - diff --git a/src/TinyEvents.SqlServer.EntityFrameworkCore/Sql/TinySqlServerEfCoreTableName.cs b/src/TinyEvents.SqlServer.EntityFrameworkCore/Sql/TinySqlServerEfCoreTableName.cs index 2b8b0b8..168bcd8 100644 --- a/src/TinyEvents.SqlServer.EntityFrameworkCore/Sql/TinySqlServerEfCoreTableName.cs +++ b/src/TinyEvents.SqlServer.EntityFrameworkCore/Sql/TinySqlServerEfCoreTableName.cs @@ -1,6 +1,6 @@ namespace TinyEvents.SqlServer.EntityFrameworkCore; -public sealed class TinySqlServerEfCoreTableName +internal sealed class TinySqlServerEfCoreTableName { private readonly string[] parts; diff --git a/src/TinyEvents.SqlServer.EntityFrameworkCore/TinySqlServerEfCoreOutboxStore.cs b/src/TinyEvents.SqlServer.EntityFrameworkCore/TinySqlServerEfCoreOutboxStore.cs index 899d3f1..77ef5c7 100644 --- a/src/TinyEvents.SqlServer.EntityFrameworkCore/TinySqlServerEfCoreOutboxStore.cs +++ b/src/TinyEvents.SqlServer.EntityFrameworkCore/TinySqlServerEfCoreOutboxStore.cs @@ -5,7 +5,7 @@ namespace TinyEvents.SqlServer.EntityFrameworkCore; -public sealed class TinySqlServerEfCoreOutboxStore : ITinyOutboxStore +internal sealed class TinySqlServerEfCoreOutboxStore : ITinyOutboxStore where TDbContext : DbContext { private readonly TDbContext dbContext; diff --git a/src/TinyEvents.SqlServer.EntityFrameworkCore/TinySqlServerEfCoreOutboxWriter.cs b/src/TinyEvents.SqlServer.EntityFrameworkCore/TinySqlServerEfCoreOutboxWriter.cs index de6b754..2904e0a 100644 --- a/src/TinyEvents.SqlServer.EntityFrameworkCore/TinySqlServerEfCoreOutboxWriter.cs +++ b/src/TinyEvents.SqlServer.EntityFrameworkCore/TinySqlServerEfCoreOutboxWriter.cs @@ -2,7 +2,7 @@ namespace TinyEvents.SqlServer.EntityFrameworkCore; -public sealed class TinySqlServerEfCoreOutboxWriter : ITinyOutboxWriter +internal sealed class TinySqlServerEfCoreOutboxWriter : ITinyOutboxWriter where TDbContext : DbContext { private readonly TDbContext dbContext; @@ -31,4 +31,3 @@ public ValueTask AddAsync( return ValueTask.CompletedTask; } } - From 5af2324aef4704e963df690f923d27f9bbdf5c53 Mon Sep 17 00:00:00 2001 From: George Date: Sat, 25 Jul 2026 11:36:20 +0200 Subject: [PATCH 20/22] Use one package release train --- .github/workflows/release-package.yml | 110 ------------------ Directory.Build.props | 11 +- docs/releasing.md | 41 +------ .../TinyEvents.PostgreSql.AdoNet.csproj | 8 -- ...ents.PostgreSql.EntityFrameworkCore.csproj | 8 -- .../TinyEvents.SqlServer.AdoNet.csproj | 8 -- ...vents.SqlServer.EntityFrameworkCore.csproj | 8 -- .../TinyEvents.Worker.csproj | 8 -- src/TinyEvents/TinyEvents.csproj | 8 -- 9 files changed, 14 insertions(+), 196 deletions(-) delete mode 100644 .github/workflows/release-package.yml diff --git a/.github/workflows/release-package.yml b/.github/workflows/release-package.yml deleted file mode 100644 index 45dafb4..0000000 --- a/.github/workflows/release-package.yml +++ /dev/null @@ -1,110 +0,0 @@ -name: Release Package - -on: - workflow_dispatch: - inputs: - package: - description: "Package to publish" - required: true - type: choice - options: - - TinyEvents - - TinyEvents.Worker - - TinyEvents.SqlServer.AdoNet - - TinyEvents.SqlServer.EntityFrameworkCore - - TinyEvents.PostgreSql.AdoNet - - TinyEvents.PostgreSql.EntityFrameworkCore - version: - description: "Package version to publish" - required: true - type: string - -permissions: - contents: read - -jobs: - publish: - runs-on: ubuntu-latest - environment: nuget-prod - - env: - DOTNET_NOLOGO: true - PACKAGE_ID: ${{ inputs.package }} - PACKAGE_VERSION: ${{ inputs.version }} - - steps: - - name: Checkout - uses: actions/checkout@v4 - with: - fetch-depth: 0 - - - name: Setup .NET - uses: actions/setup-dotnet@v4 - with: - dotnet-version: | - 8.0.x - 10.0.x - - - name: Resolve package project - shell: bash - run: | - case "$PACKAGE_ID" in - TinyEvents) - PROJECT="src/TinyEvents/TinyEvents.csproj" - ;; - TinyEvents.Worker) - PROJECT="src/TinyEvents.Worker/TinyEvents.Worker.csproj" - ;; - TinyEvents.SqlServer.AdoNet) - PROJECT="src/TinyEvents.SqlServer.AdoNet/TinyEvents.SqlServer.AdoNet.csproj" - ;; - TinyEvents.SqlServer.EntityFrameworkCore) - PROJECT="src/TinyEvents.SqlServer.EntityFrameworkCore/TinyEvents.SqlServer.EntityFrameworkCore.csproj" - ;; - TinyEvents.PostgreSql.AdoNet) - PROJECT="src/TinyEvents.PostgreSql.AdoNet/TinyEvents.PostgreSql.AdoNet.csproj" - ;; - TinyEvents.PostgreSql.EntityFrameworkCore) - PROJECT="src/TinyEvents.PostgreSql.EntityFrameworkCore/TinyEvents.PostgreSql.EntityFrameworkCore.csproj" - ;; - *) - echo "Unsupported package: $PACKAGE_ID" - exit 1 - ;; - esac - - echo "PROJECT=$PROJECT" >> "$GITHUB_ENV" - - - name: Restore - run: dotnet restore - - - name: Build (Release) - run: dotnet build -c Release --no-restore - - - name: Test (Release) - run: dotnet test -c Release --no-build --no-restore - env: - TINYEVENTS_RUN_SQLSERVER_TESTS: true - TINYEVENTS_RUN_POSTGRESQL_TESTS: true - - - name: Pack selected package - run: dotnet pack "$PROJECT" -c Release --no-build -o ./artifacts /p:PackageVersion="$PACKAGE_VERSION" /p:Version="$PACKAGE_VERSION" - - - name: Verify selected package - shell: bash - run: | - PACKAGE="./artifacts/${PACKAGE_ID}.${PACKAGE_VERSION}.nupkg" - - if [ ! -f "$PACKAGE" ]; then - echo "ERROR: Expected $PACKAGE but it was not found." - exit 1 - fi - - - name: Publish selected package to NuGet - run: dotnet nuget push "./artifacts/${PACKAGE_ID}.${PACKAGE_VERSION}.nupkg" --api-key "${{ secrets.NUGET_API_KEY }}" --source "https://api.nuget.org/v3/index.json" --skip-duplicate - - - name: Upload selected package - uses: actions/upload-artifact@v4 - with: - name: nupkg - path: ./artifacts/*.nupkg diff --git a/Directory.Build.props b/Directory.Build.props index 88d664b..220f677 100644 --- a/Directory.Build.props +++ b/Directory.Build.props @@ -5,5 +5,14 @@ enable true + + 0.1.0-alpha.2 + Jorge Durban Antunano + MIT + https://github.com/george2006/TinyEvents + https://github.com/george2006/TinyEvents + git + README.md + false + - diff --git a/docs/releasing.md b/docs/releasing.md index 062831d..2cd5f75 100644 --- a/docs/releasing.md +++ b/docs/releasing.md @@ -1,8 +1,8 @@ # Releasing -TinyEvents has two release paths. +TinyEvents uses one release train. -Use the full release workflow when every package should move together. Use the package release workflow when one package needs its own version bump. +All published TinyEvents packages move together with the same version. The shared development version lives in `Directory.Build.props`. Release workflows override that version from the git tag when packages are packed. ## Full Release Train @@ -23,7 +23,7 @@ This workflow: - verifies every expected package exists - publishes every package to NuGet -Use this when the suite should share one version: +The release train publishes: - `TinyEvents` - `TinyEvents.Worker` @@ -32,40 +32,7 @@ Use this when the suite should share one version: - `TinyEvents.PostgreSql.AdoNet` - `TinyEvents.PostgreSql.EntityFrameworkCore` -## Single Package Release - -The `Release Package` workflow is manual. - -Use it when one package needs a dedicated version without forcing every package to publish the same version. - -Inputs: - -- `package`: the package id to publish -- `version`: the exact NuGet version to publish - -Example: - -```text -package: TinyEvents.PostgreSql.AdoNet -version: 0.1.0-alpha.3 -``` - -This workflow still restores, builds, and tests the whole solution before publishing. It only packs and pushes the selected package. - -Use this for: - -- provider-specific alpha fixes -- package metadata fixes -- documentation or schema-content fixes that affect one package -- small package-specific corrections that do not require a release train - -## Before A Single Package Release - -Check the selected package dependency story before publishing. - -Provider packages reference `TinyEvents` in the repository through project references. During packing, NuGet dependencies are generated from those project references. Make sure the dependency version is the one you intend consumers to use. - -If the provider package requires a newer core package, publish the core package first or use the full release train. +Do not publish a provider package independently during the alpha hardening phase. Provider packages reference `TinyEvents`, and keeping one version across the train avoids accidental dependency mismatches. ## Local Checks diff --git a/src/TinyEvents.PostgreSql.AdoNet/TinyEvents.PostgreSql.AdoNet.csproj b/src/TinyEvents.PostgreSql.AdoNet/TinyEvents.PostgreSql.AdoNet.csproj index 8d66e94..6034802 100644 --- a/src/TinyEvents.PostgreSql.AdoNet/TinyEvents.PostgreSql.AdoNet.csproj +++ b/src/TinyEvents.PostgreSql.AdoNet/TinyEvents.PostgreSql.AdoNet.csproj @@ -2,16 +2,8 @@ net8.0 TinyEvents.PostgreSql.AdoNet - 0.1.0-alpha.2 - Jorge Durban Antunano PostgreSQL ADO.NET provider for TinyEvents outbox storage and worker claiming. - MIT - https://github.com/george2006/TinyEvents - https://github.com/george2006/TinyEvents - git tinyevents;outbox;postgresql;postgres;ado.net;dotnet;source-generator - README.md - false diff --git a/src/TinyEvents.PostgreSql.EntityFrameworkCore/TinyEvents.PostgreSql.EntityFrameworkCore.csproj b/src/TinyEvents.PostgreSql.EntityFrameworkCore/TinyEvents.PostgreSql.EntityFrameworkCore.csproj index 2bb01dc..f1d46d0 100644 --- a/src/TinyEvents.PostgreSql.EntityFrameworkCore/TinyEvents.PostgreSql.EntityFrameworkCore.csproj +++ b/src/TinyEvents.PostgreSql.EntityFrameworkCore/TinyEvents.PostgreSql.EntityFrameworkCore.csproj @@ -2,16 +2,8 @@ net8.0 TinyEvents.PostgreSql.EntityFrameworkCore - 0.1.0-alpha.2 - Jorge Durban Antunano PostgreSQL Entity Framework Core provider for TinyEvents outbox storage and worker claiming. - MIT - https://github.com/george2006/TinyEvents - https://github.com/george2006/TinyEvents - git tinyevents;outbox;postgresql;postgres;efcore;dotnet;source-generator - README.md - false diff --git a/src/TinyEvents.SqlServer.AdoNet/TinyEvents.SqlServer.AdoNet.csproj b/src/TinyEvents.SqlServer.AdoNet/TinyEvents.SqlServer.AdoNet.csproj index 4c9bf0d..d658941 100644 --- a/src/TinyEvents.SqlServer.AdoNet/TinyEvents.SqlServer.AdoNet.csproj +++ b/src/TinyEvents.SqlServer.AdoNet/TinyEvents.SqlServer.AdoNet.csproj @@ -2,16 +2,8 @@ net8.0 TinyEvents.SqlServer.AdoNet - 0.1.0-alpha.2 - Jorge Durban Antunano SQL Server ADO.NET provider for TinyEvents outbox storage and worker claiming. - MIT - https://github.com/george2006/TinyEvents - https://github.com/george2006/TinyEvents - git tinyevents;outbox;sqlserver;ado.net;dotnet;source-generator - README.md - false diff --git a/src/TinyEvents.SqlServer.EntityFrameworkCore/TinyEvents.SqlServer.EntityFrameworkCore.csproj b/src/TinyEvents.SqlServer.EntityFrameworkCore/TinyEvents.SqlServer.EntityFrameworkCore.csproj index eb814eb..43b9f15 100644 --- a/src/TinyEvents.SqlServer.EntityFrameworkCore/TinyEvents.SqlServer.EntityFrameworkCore.csproj +++ b/src/TinyEvents.SqlServer.EntityFrameworkCore/TinyEvents.SqlServer.EntityFrameworkCore.csproj @@ -2,16 +2,8 @@ net8.0 TinyEvents.SqlServer.EntityFrameworkCore - 0.1.0-alpha.2 - Jorge Durban Antunano SQL Server Entity Framework Core provider for TinyEvents outbox storage and worker claiming. - MIT - https://github.com/george2006/TinyEvents - https://github.com/george2006/TinyEvents - git tinyevents;outbox;sqlserver;efcore;dotnet;source-generator - README.md - false diff --git a/src/TinyEvents.Worker/TinyEvents.Worker.csproj b/src/TinyEvents.Worker/TinyEvents.Worker.csproj index af22c73..b22b05a 100644 --- a/src/TinyEvents.Worker/TinyEvents.Worker.csproj +++ b/src/TinyEvents.Worker/TinyEvents.Worker.csproj @@ -2,16 +2,8 @@ net8.0 TinyEvents.Worker - 0.1.0-alpha.2 - Jorge Durban Antunano Hosted worker integration for TinyEvents outbox processing. - MIT - https://github.com/george2006/TinyEvents - https://github.com/george2006/TinyEvents - git events;outbox;worker;hosting;dotnet;tinyevents - README.md - false diff --git a/src/TinyEvents/TinyEvents.csproj b/src/TinyEvents/TinyEvents.csproj index cd105b3..ce2bf26 100644 --- a/src/TinyEvents/TinyEvents.csproj +++ b/src/TinyEvents/TinyEvents.csproj @@ -2,16 +2,8 @@ net8.0 TinyEvents - 0.1.0-alpha.2 - Jorge Durban Antunano Provider-agnostic outbox-first application event library for durable same-process event processing. - MIT - https://github.com/george2006/TinyEvents - https://github.com/george2006/TinyEvents - git events;outbox;source-generator;dotnet;dependency-injection;tinyevents - README.md - false $(MSBuildThisFileDirectory)..\TinyEvents.SourceGen\bin\$(Configuration)\netstandard2.0\TinyEvents.SourceGen.dll From 8d0ab20d7f9bbc99f7e0fb93d4dd110d97f86ff6 Mon Sep 17 00:00:00 2001 From: George Date: Sat, 25 Jul 2026 11:44:10 +0200 Subject: [PATCH 21/22] Add local package smoke validation --- samples/TinyEvents.PackageSmoke/README.md | 32 ++++- .../TinyEvents.PackageSmoke.csproj | 13 +- scripts/Test-PackageSmoke.ps1 | 112 ++++++++++++++++++ 3 files changed, 149 insertions(+), 8 deletions(-) create mode 100644 scripts/Test-PackageSmoke.ps1 diff --git a/samples/TinyEvents.PackageSmoke/README.md b/samples/TinyEvents.PackageSmoke/README.md index f5e6685..142b411 100644 --- a/samples/TinyEvents.PackageSmoke/README.md +++ b/samples/TinyEvents.PackageSmoke/README.md @@ -1,6 +1,32 @@ # TinyEvents Package Smoke -Run this sample after publishing packages to NuGet: +## Local Package Build Smoke + +Run this sample against locally packed packages without touching a database: + +```powershell +.\scripts\Test-PackageSmoke.ps1 +``` + +That command builds the solution, packs the complete TinyEvents release train with a unique local version, restores this sample from those local packages, and builds the sample. + +## Local Package Runtime Smoke + +To start the sample SQL Server and PostgreSQL containers and run the database smoke paths: + +```powershell +.\scripts\Test-PackageSmoke.ps1 -StartDatabases -Run +``` + +To run against already-started databases, omit `-StartDatabases`: + +```powershell +.\scripts\Test-PackageSmoke.ps1 -Run +``` + +`-Run` uses the default SQL Server port `14334` and PostgreSQL port `54324` unless the environment variables below override them. + +To run this sample after publishing packages to NuGet: ```bash docker compose -f samples/TinyEvents.PackageSmoke/docker-compose.yml up -d @@ -22,4 +48,6 @@ To use a different PostgreSQL instance: set TINYEVENTS_PACKAGE_SMOKE_POSTGRESQL=Host=localhost;Port=5432;Database=TinyEventsPackageSmoke;Username=postgres;Password=your-password; ``` -The sample references the public `0.1.0-alpha.2` packages instead of local project references. It verifies that the core package, SQL Server providers, PostgreSQL providers, worker package, dependency injection extensions, source-generator consumer registration, publishing, claiming, and processing can be consumed from NuGet. +The sample references package versions through `TinyEventsPackageVersion`, which defaults to the shared version from `Directory.Build.props`. The local smoke script overrides that property with the temporary locally packed package version. + +It verifies that the core package, SQL Server providers, PostgreSQL providers, worker package, dependency injection extensions, source-generator consumer registration, publishing, claiming, and processing can be consumed from NuGet packages instead of project references. diff --git a/samples/TinyEvents.PackageSmoke/TinyEvents.PackageSmoke.csproj b/samples/TinyEvents.PackageSmoke/TinyEvents.PackageSmoke.csproj index 8128955..4667e6b 100644 --- a/samples/TinyEvents.PackageSmoke/TinyEvents.PackageSmoke.csproj +++ b/samples/TinyEvents.PackageSmoke/TinyEvents.PackageSmoke.csproj @@ -4,17 +4,18 @@ net8.0 false enable + $(Version) - - - - - - + + + + + + diff --git a/scripts/Test-PackageSmoke.ps1 b/scripts/Test-PackageSmoke.ps1 new file mode 100644 index 0000000..c999d45 --- /dev/null +++ b/scripts/Test-PackageSmoke.ps1 @@ -0,0 +1,112 @@ +param( + [string]$PackageVersion = "", + [switch]$Run, + [switch]$StartDatabases +) + +$ErrorActionPreference = "Stop" + +function Invoke-Native { + param( + [string]$FilePath, + [string[]]$Arguments + ) + + & $FilePath @Arguments + + if ($LASTEXITCODE -ne 0) { + throw "Command failed with exit code ${LASTEXITCODE}: $FilePath $($Arguments -join ' ')" + } +} + +$repoRoot = Resolve-Path (Join-Path $PSScriptRoot "..") +$solution = Join-Path $repoRoot "TinyEvents.sln" +$sampleProject = Join-Path $repoRoot "samples\TinyEvents.PackageSmoke\TinyEvents.PackageSmoke.csproj" +$composeFile = Join-Path $repoRoot "samples\TinyEvents.PackageSmoke\docker-compose.yml" + +if ([string]::IsNullOrWhiteSpace($PackageVersion)) { + $PackageVersion = "0.1.0-local.$(Get-Date -Format 'yyyyMMddHHmmss')" +} + +$artifactRoot = Join-Path $repoRoot "artifacts\package-smoke\$PackageVersion" +$packagesDirectory = Join-Path $artifactRoot "packages" +$nugetConfig = Join-Path $artifactRoot "NuGet.config" + +New-Item -ItemType Directory -Force -Path $packagesDirectory | Out-Null + +$projects = @( + "src\TinyEvents\TinyEvents.csproj", + "src\TinyEvents.Worker\TinyEvents.Worker.csproj", + "src\TinyEvents.SqlServer.AdoNet\TinyEvents.SqlServer.AdoNet.csproj", + "src\TinyEvents.SqlServer.EntityFrameworkCore\TinyEvents.SqlServer.EntityFrameworkCore.csproj", + "src\TinyEvents.PostgreSql.AdoNet\TinyEvents.PostgreSql.AdoNet.csproj", + "src\TinyEvents.PostgreSql.EntityFrameworkCore\TinyEvents.PostgreSql.EntityFrameworkCore.csproj" +) + +Write-Host "Building TinyEvents release train..." +Invoke-Native "dotnet" @("restore", $solution) +Invoke-Native "dotnet" @("build", $solution, "-c", "Release", "--no-restore") + +Write-Host "Packing TinyEvents release train as $PackageVersion..." +foreach ($project in $projects) { + $projectPath = Join-Path $repoRoot $project + Invoke-Native "dotnet" @( + "pack", + $projectPath, + "-c", + "Release", + "--no-build", + "-o", + $packagesDirectory, + "/p:PackageVersion=$PackageVersion", + "/p:Version=$PackageVersion") +} + +@" + + + + + + + + +"@ | Set-Content -Path $nugetConfig -Encoding UTF8 + +Write-Host "Restoring package smoke sample from local TinyEvents packages..." +Invoke-Native "dotnet" @( + "restore", + $sampleProject, + "--configfile", + $nugetConfig, + "--no-cache", + "--force", + "/p:TinyEventsPackageVersion=$PackageVersion") + +Write-Host "Building package smoke sample..." +Invoke-Native "dotnet" @( + "build", + $sampleProject, + "-c", + "Release", + "--no-restore", + "/p:TinyEventsPackageVersion=$PackageVersion") + +if ($StartDatabases) { + Write-Host "Starting package smoke databases..." + Invoke-Native "docker" @("compose", "-f", $composeFile, "up", "-d", "--wait") +} + +if ($Run) { + Write-Host "Running package smoke sample..." + Invoke-Native "dotnet" @( + "run", + "--project", + $sampleProject, + "-c", + "Release", + "--no-build", + "--no-restore") +} + +Write-Host "TinyEvents package smoke validation passed for $PackageVersion." From 28e3958c8a6c9e50643b76aac48a13141b96a2cb Mon Sep 17 00:00:00 2001 From: George Date: Sat, 25 Jul 2026 11:47:03 +0200 Subject: [PATCH 22/22] Align PostgreSQL lease-loss runtime tests --- .../AdoNetPostgreSqlStoreRuntimeTests.cs | 5 +++-- .../EfCorePostgreSqlStoreRuntimeTests.cs | 5 +++-- 2 files changed, 6 insertions(+), 4 deletions(-) diff --git a/tests/TinyEvents.PostgreSql.Tests/AdoNetPostgreSqlStoreRuntimeTests.cs b/tests/TinyEvents.PostgreSql.Tests/AdoNetPostgreSqlStoreRuntimeTests.cs index 126b8f3..d5c7502 100644 --- a/tests/TinyEvents.PostgreSql.Tests/AdoNetPostgreSqlStoreRuntimeTests.cs +++ b/tests/TinyEvents.PostgreSql.Tests/AdoNetPostgreSqlStoreRuntimeTests.cs @@ -100,7 +100,7 @@ public async Task Competing_workers_claim_message_only_once() } [PostgreSqlIntegrationFact] - public async Task Mark_processed_updates_only_message_owned_by_worker() + public async Task Mark_processed_throws_when_message_is_owned_by_another_worker() { await fixture.ResetSchemaAsync(); var ownedId = Guid.NewGuid(); @@ -110,7 +110,8 @@ public async Task Mark_processed_updates_only_message_owned_by_worker() var store = NewStore(); await store.MarkProcessedAsync(ownedId, "worker-1", DateTimeOffset.UtcNow, CancellationToken.None); - await store.MarkProcessedAsync(otherId, "worker-1", DateTimeOffset.UtcNow, CancellationToken.None); + await Assert.ThrowsAnyAsync( + async () => await store.MarkProcessedAsync(otherId, "worker-1", DateTimeOffset.UtcNow, CancellationToken.None)); Assert.Equal(TinyOutboxMessageStatus.Processed, await ReadStatusAsync(ownedId)); Assert.Equal(TinyOutboxMessageStatus.Processing, await ReadStatusAsync(otherId)); diff --git a/tests/TinyEvents.PostgreSql.Tests/EfCorePostgreSqlStoreRuntimeTests.cs b/tests/TinyEvents.PostgreSql.Tests/EfCorePostgreSqlStoreRuntimeTests.cs index 7c86a82..2210bef 100644 --- a/tests/TinyEvents.PostgreSql.Tests/EfCorePostgreSqlStoreRuntimeTests.cs +++ b/tests/TinyEvents.PostgreSql.Tests/EfCorePostgreSqlStoreRuntimeTests.cs @@ -110,7 +110,7 @@ await InsertOutboxMessageAsync( } [PostgreSqlIntegrationFact] - public async Task Mark_processed_updates_only_message_owned_by_worker() + public async Task Mark_processed_throws_when_message_is_owned_by_another_worker() { await fixture.ResetSchemaAsync(); var ownedId = Guid.NewGuid(); @@ -121,7 +121,8 @@ public async Task Mark_processed_updates_only_message_owned_by_worker() var store = NewStore(dbContext); await store.MarkProcessedAsync(ownedId, "worker-1", DateTimeOffset.UtcNow, CancellationToken.None); - await store.MarkProcessedAsync(otherId, "worker-1", DateTimeOffset.UtcNow, CancellationToken.None); + await Assert.ThrowsAnyAsync( + async () => await store.MarkProcessedAsync(otherId, "worker-1", DateTimeOffset.UtcNow, CancellationToken.None)); Assert.Equal(TinyOutboxMessageStatus.Processed, await ReadStatusAsync(ownedId)); Assert.Equal(TinyOutboxMessageStatus.Processing, await ReadStatusAsync(otherId));