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/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..5f5bf9e 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. @@ -115,19 +115,27 @@ 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. 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. 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. + +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 @@ -165,6 +173,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 @@ -178,6 +188,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. + +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`. +Runtime processing does not use a custom consumer registry. It resolves consumers directly from `IServiceProvider` through generated dispatchers. 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/getting-started.md b/docs/getting-started.md index 12533d9..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: @@ -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/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 c6a1606..eb02b16 100644 --- a/docs/postgresql/ef-core.md +++ b/docs/postgresql/ef-core.md @@ -55,12 +55,18 @@ 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. 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/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/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/source-generator.md b/docs/source-generator.md index 39cb52d..8867d3b 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,17 +49,21 @@ 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 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. + +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: @@ -88,7 +91,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/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 ddd7918..044879e 100644 --- a/docs/sql-server/ef-core.md +++ b/docs/sql-server/ef-core.md @@ -55,12 +55,18 @@ 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. 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: diff --git a/docs/workers.md b/docs/workers.md index f13c80b..7eec204 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,47 @@ 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. + +## 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. + +## 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. 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." 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/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.AdoNet/TinyPostgreSqlAdoNetOutboxStore.cs b/src/TinyEvents.PostgreSql.AdoNet/TinyPostgreSqlAdoNetOutboxStore.cs index 902d034..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; @@ -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.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/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.PostgreSql.EntityFrameworkCore/TinyPostgreSqlEfCoreOutboxStore.cs b/src/TinyEvents.PostgreSql.EntityFrameworkCore/TinyPostgreSqlEfCoreOutboxStore.cs index c3342dd..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; @@ -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.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.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 ae22fa1..0000000 --- a/src/TinyEvents.SourceGen/Emission/EventTypeDescriptorEmitter.cs +++ /dev/null @@ -1,18 +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(")));"); - } -} 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.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/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.AdoNet/TinySqlServerAdoNetOutboxStore.cs b/src/TinyEvents.SqlServer.AdoNet/TinySqlServerAdoNetOutboxStore.cs index 994ba1f..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; @@ -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.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/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.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/src/TinyEvents.SqlServer.EntityFrameworkCore/TinySqlServerEfCoreOutboxStore.cs b/src/TinyEvents.SqlServer.EntityFrameworkCore/TinySqlServerEfCoreOutboxStore.cs index 4bf78f0..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; @@ -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.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; } } - 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.Worker/TinyEvents.Worker.csproj b/src/TinyEvents.Worker/TinyEvents.Worker.csproj index 81ddc08..b22b05a 100644 --- a/src/TinyEvents.Worker/TinyEvents.Worker.csproj +++ b/src/TinyEvents.Worker/TinyEvents.Worker.csproj @@ -2,19 +2,12 @@ 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.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/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/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/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/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 37c7f71..cbbfded 100644 --- a/src/TinyEvents/Processing/TinyOutboxProcessor.cs +++ b/src/TinyEvents/Processing/TinyOutboxProcessor.cs @@ -1,30 +1,44 @@ -using System.Reflection; -using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Logging; +using Microsoft.Extensions.Logging.Abstractions; 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; + private readonly ILogger logger; public TinyOutboxProcessor( IServiceProvider serviceProvider, ITinyOutboxStore store, ITinyEventSerializer serializer, - IEnumerable eventTypes, + 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) { @@ -41,9 +55,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) @@ -56,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.eventTypes = BuildEventTypeMap(eventTypes); + this.dispatchers = BuildDispatcherMap(dispatchers); this.options = options; this.timeProvider = timeProvider; + this.logger = logger; } public async ValueTask ProcessPendingAsync(CancellationToken cancellationToken = default) @@ -78,6 +98,7 @@ public async ValueTask ProcessPendingAsync(CancellationToken cancellationToken = foreach (var message in messages) { + cancellationToken.ThrowIfCancellationRequested(); await ProcessMessageAsync(message, workerId, cancellationToken); } } @@ -96,9 +117,21 @@ private async ValueTask ProcessMessageAsync( { throw; } + catch (TinyOutboxLeaseLostException exception) + { + LogLeaseLost(message, workerId, exception, "marked as processed"); + } catch (Exception exception) { - await MarkFailedAsync(message, workerId, exception, cancellationToken); + try + { + var failure = await MarkFailedAsync(message, workerId, exception, cancellationToken); + LogProcessingFailure(message, workerId, exception, failure); + } + catch (TinyOutboxLeaseLostException leaseLostException) + { + LogLeaseLost(message, workerId, leaseLostException, "recording a processing failure"); + } } } @@ -106,42 +139,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, @@ -154,7 +166,7 @@ await store.MarkProcessedAsync( cancellationToken); } - private async ValueTask MarkFailedAsync( + private async ValueTask MarkFailedAsync( TinyOutboxMessage message, string workerId, Exception exception, @@ -170,6 +182,8 @@ await store.MarkFailedAsync( attemptCount, nextAttemptAtUtc, cancellationToken); + + return new RecordedFailure(attemptCount, nextAttemptAtUtc); } private DateTimeOffset? GetNextAttemptAtUtc(int attemptCount) @@ -182,33 +196,68 @@ 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); } + + 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/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/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/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/src/TinyEvents/TinyEvents.csproj b/src/TinyEvents/TinyEvents.csproj index ee338ad..ce2bf26 100644 --- a/src/TinyEvents/TinyEvents.csproj +++ b/src/TinyEvents/TinyEvents.csproj @@ -2,22 +2,15 @@ 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 + 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.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/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 0e7ddf7..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)); @@ -158,8 +159,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.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 + }; + } + } +} diff --git a/tests/TinyEvents.SourceGen.Tests/TinyEventsSourceGeneratorTests.cs b/tests/TinyEvents.SourceGen.Tests/TinyEventsSourceGeneratorTests.cs index 6c81971..0dfba0c 100644 --- a/tests/TinyEvents.SourceGen.Tests/TinyEventsSourceGeneratorTests.cs +++ b/tests/TinyEvents.SourceGen.Tests/TinyEventsSourceGeneratorTests.cs @@ -187,7 +187,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); } @@ -226,7 +227,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] 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.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) 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 2af4849..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; @@ -80,6 +81,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() { @@ -133,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() { @@ -238,6 +301,63 @@ 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); + } + + [Fact] + 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, 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, 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; + } + private static ITinyOutboxProcessor BuildProcessor( ITinyOutboxStore store, bool includeSecondConsumer = false, @@ -246,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(); @@ -261,11 +382,16 @@ 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>(); + if (logger is not null) + { + services.AddSingleton(logger); + } + if (includeSecondConsumer) { services.AddSingleton, SecondRecordingConsumer>(); @@ -462,4 +588,186 @@ 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; + } + } + + 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"); + } + } + + 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); } 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; + } + } +} diff --git a/tests/TinyEvents.Worker.Tests/TinyEventsWorkerTests.cs b/tests/TinyEvents.Worker.Tests/TinyEventsWorkerTests.cs index c66777c..7419a17 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] @@ -101,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() { @@ -148,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; } @@ -171,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); + } + } }