From d1f784cb98eb64c756dadbe1e6ad2734364fe252 Mon Sep 17 00:00:00 2001 From: Tom Date: Tue, 23 Jun 2026 11:08:31 +0200 Subject: [PATCH 01/33] feat(core): register a shared DirectoriesConfig in RegisterCoreServices --- .../Extensions/RegisterDefaultServicesExtensions.cs | 2 ++ .../Core/RegisterDefaultServicesExtensionsTests.cs | 12 ++++++++++++ 2 files changed, 14 insertions(+) diff --git a/src/SquidStd.Services.Core/Extensions/RegisterDefaultServicesExtensions.cs b/src/SquidStd.Services.Core/Extensions/RegisterDefaultServicesExtensions.cs index 1ce76469..148ff3a6 100644 --- a/src/SquidStd.Services.Core/Extensions/RegisterDefaultServicesExtensions.cs +++ b/src/SquidStd.Services.Core/Extensions/RegisterDefaultServicesExtensions.cs @@ -4,6 +4,7 @@ using SquidStd.Abstractions.Extensions.Container; using SquidStd.Abstractions.Extensions.Services; using SquidStd.Core.Data.Bootstrap; +using SquidStd.Core.Directories; using SquidStd.Core.Data.Jobs; using SquidStd.Core.Data.Metrics; using SquidStd.Core.Data.Storage; @@ -79,6 +80,7 @@ public IContainer RegisterCoreServices() /// The same container for chaining. public IContainer RegisterCoreServices(string configName, string configDirectory) { + container.RegisterInstance(new DirectoriesConfig(configDirectory, [])); container.RegisterDataSerializer(); container.RegisterDefaultCoreConfigSections(); container.RegisterConfigManagerService(configName, configDirectory); diff --git a/tests/SquidStd.Tests/Services/Core/RegisterDefaultServicesExtensionsTests.cs b/tests/SquidStd.Tests/Services/Core/RegisterDefaultServicesExtensionsTests.cs index 74e259a7..e8d2ee14 100644 --- a/tests/SquidStd.Tests/Services/Core/RegisterDefaultServicesExtensionsTests.cs +++ b/tests/SquidStd.Tests/Services/Core/RegisterDefaultServicesExtensionsTests.cs @@ -2,6 +2,7 @@ using SquidStd.Abstractions.Data.Internal.Config; using SquidStd.Abstractions.Data.Internal.Services; using SquidStd.Core.Data.Bootstrap; +using SquidStd.Core.Directories; using SquidStd.Core.Data.Jobs; using SquidStd.Core.Data.Metrics; using SquidStd.Core.Data.Storage; @@ -180,6 +181,17 @@ public void RegisterCoreServices_RegistersSecretServices() Assert.True(container.IsRegistered()); } + [Fact] + public void RegisterCoreServices_RegistersDirectoriesConfig() + { + using var temp = new TempDirectory(); + using var container = new Container(); + + container.RegisterCoreServices("app", temp.Path); + + Assert.True(container.IsRegistered()); + } + [Fact] public void RegisterSecretServices_RegistersSecretProtectorAndStore() { From 1a7323267b000b3df7bf3554ac904a3e9cc11b28 Mon Sep 17 00:00:00 2001 From: Tom Date: Tue, 23 Jun 2026 11:11:41 +0200 Subject: [PATCH 02/33] feat(templating): add SquidStd.Templating with Scriban renderer and named registry --- SquidStd.slnx | 1 + .../Interfaces/ITemplateRenderer.cs | 16 +++ .../Services/ScribanTemplateRenderer.cs | 103 ++++++++++++++++++ .../SquidStd.Templating.csproj | 20 ++++ src/SquidStd.Templating/TemplateException.cs | 17 +++ tests/SquidStd.Tests/SquidStd.Tests.csproj | 1 + .../ScribanTemplateRendererTests.cs | 71 ++++++++++++ 7 files changed, 229 insertions(+) create mode 100644 src/SquidStd.Templating/Interfaces/ITemplateRenderer.cs create mode 100644 src/SquidStd.Templating/Services/ScribanTemplateRenderer.cs create mode 100644 src/SquidStd.Templating/SquidStd.Templating.csproj create mode 100644 src/SquidStd.Templating/TemplateException.cs create mode 100644 tests/SquidStd.Tests/Templating/ScribanTemplateRendererTests.cs diff --git a/SquidStd.slnx b/SquidStd.slnx index 12d68a83..5a6da28f 100644 --- a/SquidStd.slnx +++ b/SquidStd.slnx @@ -18,6 +18,7 @@ + diff --git a/src/SquidStd.Templating/Interfaces/ITemplateRenderer.cs b/src/SquidStd.Templating/Interfaces/ITemplateRenderer.cs new file mode 100644 index 00000000..8fa4ee8c --- /dev/null +++ b/src/SquidStd.Templating/Interfaces/ITemplateRenderer.cs @@ -0,0 +1,16 @@ +namespace SquidStd.Templating.Interfaces; + +/// +/// Renders Scriban templates from strings or registered named templates. +/// +public interface ITemplateRenderer +{ + /// Renders an ad-hoc template string against a model. + ValueTask RenderAsync(string template, object? model, CancellationToken cancellationToken = default); + + /// Compiles and registers a named template (replacing any existing one with the same name). + void Register(string name, string template); + + /// Renders a previously registered template by name. + ValueTask RenderByNameAsync(string name, object? model, CancellationToken cancellationToken = default); +} diff --git a/src/SquidStd.Templating/Services/ScribanTemplateRenderer.cs b/src/SquidStd.Templating/Services/ScribanTemplateRenderer.cs new file mode 100644 index 00000000..c3b703b6 --- /dev/null +++ b/src/SquidStd.Templating/Services/ScribanTemplateRenderer.cs @@ -0,0 +1,103 @@ +using System.Collections.Concurrent; +using Scriban; +using Scriban.Syntax; +using SquidStd.Abstractions.Interfaces.Services; +using SquidStd.Core.Directories; +using SquidStd.Templating.Interfaces; + +namespace SquidStd.Templating.Services; + +/// +/// Scriban-backed . Named templates are compiled and cached; ad-hoc +/// strings are parsed per call. On start it auto-loads templates/**/*.tmpl. +/// +public sealed class ScribanTemplateRenderer : ITemplateRenderer, ISquidStdService +{ + private const string TemplateExtension = ".tmpl"; + + private readonly DirectoriesConfig _directories; + private readonly ConcurrentDictionary _named = new(StringComparer.Ordinal); + + public ScribanTemplateRenderer(DirectoriesConfig directories) + { + _directories = directories; + } + + /// + public async ValueTask RenderAsync(string template, object? model, CancellationToken cancellationToken = default) + { + ArgumentException.ThrowIfNullOrWhiteSpace(template); + + return await RenderCompiledAsync(Parse(template, "(inline)"), model, "(inline)"); + } + + /// + public void Register(string name, string template) + { + ArgumentException.ThrowIfNullOrWhiteSpace(name); + ArgumentException.ThrowIfNullOrWhiteSpace(template); + + _named[name] = Parse(template, name); + } + + /// + public async ValueTask RenderByNameAsync(string name, object? model, CancellationToken cancellationToken = default) + { + ArgumentException.ThrowIfNullOrWhiteSpace(name); + + if (!_named.TryGetValue(name, out var compiled)) + { + throw new InvalidOperationException($"No template registered with name '{name}'."); + } + + return await RenderCompiledAsync(compiled, model, name); + } + + /// + public async ValueTask StartAsync(CancellationToken cancellationToken = default) + { + var directory = _directories["templates"]; + + if (!Directory.Exists(directory)) + { + return; + } + + foreach (var file in Directory.EnumerateFiles(directory, "*" + TemplateExtension, SearchOption.AllDirectories)) + { + var relative = Path.GetRelativePath(directory, file).Replace('\\', '/'); + var name = relative[..^TemplateExtension.Length]; + var content = await File.ReadAllTextAsync(file, cancellationToken); + + Register(name, content); + } + } + + /// + public ValueTask StopAsync(CancellationToken cancellationToken = default) + => ValueTask.CompletedTask; + + private static Template Parse(string text, string name) + { + var template = Template.Parse(text); + + if (template.HasErrors) + { + throw new TemplateException($"Template '{name}' has parse errors: {string.Join("; ", template.Messages)}"); + } + + return template; + } + + private static async ValueTask RenderCompiledAsync(Template template, object? model, string name) + { + try + { + return await template.RenderAsync(model); + } + catch (ScriptRuntimeException ex) + { + throw new TemplateException($"Template '{name}' failed to render: {ex.Message}", ex); + } + } +} diff --git a/src/SquidStd.Templating/SquidStd.Templating.csproj b/src/SquidStd.Templating/SquidStd.Templating.csproj new file mode 100644 index 00000000..c088a852 --- /dev/null +++ b/src/SquidStd.Templating/SquidStd.Templating.csproj @@ -0,0 +1,20 @@ + + + + net10.0 + enable + enable + true + + + + + + + + + + + + + diff --git a/src/SquidStd.Templating/TemplateException.cs b/src/SquidStd.Templating/TemplateException.cs new file mode 100644 index 00000000..73f2bd57 --- /dev/null +++ b/src/SquidStd.Templating/TemplateException.cs @@ -0,0 +1,17 @@ +namespace SquidStd.Templating; + +/// +/// Raised when a template fails to parse or render. +/// +public sealed class TemplateException : Exception +{ + public TemplateException(string message) + : base(message) + { + } + + public TemplateException(string message, Exception innerException) + : base(message, innerException) + { + } +} diff --git a/tests/SquidStd.Tests/SquidStd.Tests.csproj b/tests/SquidStd.Tests/SquidStd.Tests.csproj index 59802f79..ad45b91e 100644 --- a/tests/SquidStd.Tests/SquidStd.Tests.csproj +++ b/tests/SquidStd.Tests/SquidStd.Tests.csproj @@ -48,6 +48,7 @@ + diff --git a/tests/SquidStd.Tests/Templating/ScribanTemplateRendererTests.cs b/tests/SquidStd.Tests/Templating/ScribanTemplateRendererTests.cs new file mode 100644 index 00000000..1b5348db --- /dev/null +++ b/tests/SquidStd.Tests/Templating/ScribanTemplateRendererTests.cs @@ -0,0 +1,71 @@ +using SquidStd.Core.Directories; +using SquidStd.Templating; +using SquidStd.Templating.Services; +using SquidStd.Tests.Support; + +namespace SquidStd.Tests.Templating; + +public class ScribanTemplateRendererTests +{ + private static ScribanTemplateRenderer NewRenderer(string root) + => new(new DirectoriesConfig(root, [])); + + [Fact] + public async Task RenderAsync_RendersModel() + { + using var temp = new TempDirectory(); + var renderer = NewRenderer(temp.Path); + + var result = await renderer.RenderAsync("Hi {{ user.name }}", new { User = new { Name = "squid" } }); + + Assert.Equal("Hi squid", result); + } + + [Fact] + public async Task RegisterThenRenderByName_Works() + { + using var temp = new TempDirectory(); + var renderer = NewRenderer(temp.Path); + renderer.Register("greet", "Hi {{ user.name }}"); + + var result = await renderer.RenderByNameAsync("greet", new { User = new { Name = "squid" } }); + + Assert.Equal("Hi squid", result); + } + + [Fact] + public async Task RenderByNameAsync_UnknownName_Throws() + { + using var temp = new TempDirectory(); + var renderer = NewRenderer(temp.Path); + + await Assert.ThrowsAsync(async () => await renderer.RenderByNameAsync("nope", null)); + } + + [Fact] + public async Task RenderAsync_MalformedTemplate_Throws() + { + using var temp = new TempDirectory(); + var renderer = NewRenderer(temp.Path); + + await Assert.ThrowsAsync(async () => await renderer.RenderAsync("{{ for x in }}", null)); + } + + [Fact] + public void Register_MalformedTemplate_Throws() + { + using var temp = new TempDirectory(); + var renderer = NewRenderer(temp.Path); + + Assert.Throws(() => renderer.Register("bad", "{{ for x in }}")); + } + + [Fact] + public async Task RenderAsync_EmptyTemplate_Throws() + { + using var temp = new TempDirectory(); + var renderer = NewRenderer(temp.Path); + + await Assert.ThrowsAsync(async () => await renderer.RenderAsync(string.Empty, null)); + } +} From 2cdda180fea82a4145d66817a424b973e2743ea7 Mon Sep 17 00:00:00 2001 From: Tom Date: Tue, 23 Jun 2026 11:12:48 +0200 Subject: [PATCH 03/33] test(templating): cover startup auto-load of templates/*.tmpl --- .../Templating/ScribanTemplateRendererTests.cs | 16 ++++++++++++++++ 1 file changed, 16 insertions(+) diff --git a/tests/SquidStd.Tests/Templating/ScribanTemplateRendererTests.cs b/tests/SquidStd.Tests/Templating/ScribanTemplateRendererTests.cs index 1b5348db..414f63ba 100644 --- a/tests/SquidStd.Tests/Templating/ScribanTemplateRendererTests.cs +++ b/tests/SquidStd.Tests/Templating/ScribanTemplateRendererTests.cs @@ -68,4 +68,20 @@ public async Task RenderAsync_EmptyTemplate_Throws() await Assert.ThrowsAsync(async () => await renderer.RenderAsync(string.Empty, null)); } + + [Fact] + public async Task StartAsync_AutoLoadsTemplatesFromDirectory() + { + using var temp = new TempDirectory(); + var emails = Path.Combine(temp.Path, "templates", "emails"); + Directory.CreateDirectory(emails); + await File.WriteAllTextAsync(Path.Combine(emails, "welcome.tmpl"), "Welcome {{ user.name }}"); + + var renderer = NewRenderer(temp.Path); + await renderer.StartAsync(); + + var result = await renderer.RenderByNameAsync("emails/welcome", new { User = new { Name = "squid" } }); + + Assert.Equal("Welcome squid", result); + } } From 96dfc739ef5bfd4f33c6a5981f89522a84223d36 Mon Sep 17 00:00:00 2001 From: Tom Date: Tue, 23 Jun 2026 11:13:50 +0200 Subject: [PATCH 04/33] feat(templating): add AddTemplating DI registration --- .../TemplatingRegistrationExtensions.cs | 23 +++++++++++++++++ .../Templating/TemplatingRegistrationTests.cs | 25 +++++++++++++++++++ 2 files changed, 48 insertions(+) create mode 100644 src/SquidStd.Templating/Extensions/TemplatingRegistrationExtensions.cs create mode 100644 tests/SquidStd.Tests/Templating/TemplatingRegistrationTests.cs diff --git a/src/SquidStd.Templating/Extensions/TemplatingRegistrationExtensions.cs b/src/SquidStd.Templating/Extensions/TemplatingRegistrationExtensions.cs new file mode 100644 index 00000000..5ea4f59c --- /dev/null +++ b/src/SquidStd.Templating/Extensions/TemplatingRegistrationExtensions.cs @@ -0,0 +1,23 @@ +using DryIoc; +using SquidStd.Abstractions.Extensions.Services; +using SquidStd.Templating.Interfaces; +using SquidStd.Templating.Services; + +namespace SquidStd.Templating.Extensions; + +/// +/// DryIoc registration helpers for the templating module. +/// +public static class TemplatingRegistrationExtensions +{ + /// + /// Registers the Scriban template renderer as a singleton SquidStd service (so its startup + /// auto-load of templates/*.tmpl runs with the host). Requires a registered DirectoriesConfig. + /// + public static IContainer AddTemplating(this IContainer container) + { + ArgumentNullException.ThrowIfNull(container); + + return container.RegisterStdService(-1); + } +} diff --git a/tests/SquidStd.Tests/Templating/TemplatingRegistrationTests.cs b/tests/SquidStd.Tests/Templating/TemplatingRegistrationTests.cs new file mode 100644 index 00000000..7046a789 --- /dev/null +++ b/tests/SquidStd.Tests/Templating/TemplatingRegistrationTests.cs @@ -0,0 +1,25 @@ +using DryIoc; +using SquidStd.Core.Directories; +using SquidStd.Templating.Extensions; +using SquidStd.Templating.Interfaces; +using SquidStd.Tests.Support; + +namespace SquidStd.Tests.Templating; + +public class TemplatingRegistrationTests +{ + [Fact] + public async Task AddTemplating_ResolvesRendererAndRenders() + { + using var temp = new TempDirectory(); + var container = new Container(); + container.RegisterInstance(new DirectoriesConfig(temp.Path, [])); + + container.AddTemplating(); + + var renderer = container.Resolve(); + var result = await renderer.RenderAsync("Hi {{ user.name }}", new { User = new { Name = "squid" } }); + + Assert.Equal("Hi squid", result); + } +} From 58fef582233d238dd7e5f5395266f8c8d025bbe1 Mon Sep 17 00:00:00 2001 From: Tom Date: Tue, 23 Jun 2026 11:29:20 +0200 Subject: [PATCH 05/33] docs(templating): add package README, DocFX article, toc and root index entries --- README.md | 1 + docs/articles/templating.md | 1 + docs/articles/toc.yml | 2 + src/SquidStd.Templating/README.md | 62 +++++++++++++++++++++++++++++++ 4 files changed, 66 insertions(+) create mode 100644 docs/articles/templating.md create mode 100644 src/SquidStd.Templating/README.md diff --git a/README.md b/README.md index a879589d..306d9ab4 100644 --- a/README.md +++ b/README.md @@ -86,6 +86,7 @@ block until cancellation for long-running hosts. | `SquidStd.Storage` | Local file storage backend (`AddFileStorage`). | [![readme](https://img.shields.io/badge/readme-1390A3.svg)](src/SquidStd.Storage/README.md) · [![NuGet](https://img.shields.io/nuget/v/SquidStd.Storage.svg)](https://www.nuget.org/packages/SquidStd.Storage/) | | `SquidStd.Storage.S3` | S3/MinIO storage backend (`AddS3Storage`). | [![readme](https://img.shields.io/badge/readme-1390A3.svg)](src/SquidStd.Storage.S3/README.md) · [![NuGet](https://img.shields.io/nuget/v/SquidStd.Storage.S3.svg)](https://www.nuget.org/packages/SquidStd.Storage.S3/) | | `SquidStd.Scripting.Lua` | Lua scripting engine with attribute-based modules and event bridging. | [![readme](https://img.shields.io/badge/readme-1390A3.svg)](src/SquidStd.Scripting.Lua/README.md) · [![NuGet](https://img.shields.io/nuget/v/SquidStd.Scripting.Lua.svg)](https://www.nuget.org/packages/SquidStd.Scripting.Lua/) | +| `SquidStd.Templating` | Scriban templating with a named-template registry and `templates/*.tmpl` auto-load (`AddTemplating`). | [![readme](https://img.shields.io/badge/readme-1390A3.svg)](src/SquidStd.Templating/README.md) · [![NuGet](https://img.shields.io/nuget/v/SquidStd.Templating.svg)](https://www.nuget.org/packages/SquidStd.Templating/) | ## Architecture diff --git a/docs/articles/templating.md b/docs/articles/templating.md new file mode 100644 index 00000000..3bcd926d --- /dev/null +++ b/docs/articles/templating.md @@ -0,0 +1 @@ +[!include[](../../src/SquidStd.Templating/README.md)] diff --git a/docs/articles/toc.yml b/docs/articles/toc.yml index 229340cd..9dbf4625 100644 --- a/docs/articles/toc.yml +++ b/docs/articles/toc.yml @@ -36,6 +36,8 @@ href: storage-s3.md - name: SquidStd.Scripting.Lua href: scripting-lua.md +- name: SquidStd.Templating + href: templating.md - name: Serialization href: serialization.md - name: Scheduler diff --git a/src/SquidStd.Templating/README.md b/src/SquidStd.Templating/README.md new file mode 100644 index 00000000..6cc99300 --- /dev/null +++ b/src/SquidStd.Templating/README.md @@ -0,0 +1,62 @@ +

+ SquidStd +

+ +

SquidStd.Templating

+ +

+ NuGet + Downloads + docs + license +

+ +Templating for SquidStd, backed by [Scriban](https://github.com/scriban/scriban). `ITemplateRenderer` +renders ad-hoc template strings and named templates (compiled and cached), and auto-loads +`*.tmpl` files from the `templates/` directory at startup. + +## Install + +```bash +dotnet add package SquidStd.Templating +``` + +## Features + +- `ITemplateRenderer.RenderAsync(template, model)` — render an ad-hoc template string. +- `Register(name, template)` + `RenderByNameAsync(name, model)` — compiled, cached named templates. +- Startup auto-load: every `templates/**/*.tmpl` is registered by relative path without extension (`emails/welcome.tmpl` → `emails/welcome`). +- Scriban default member naming (`snake_case`): a `UserName` property is `{{ user.user_name }}`. +- Parse/render failures surface as `TemplateException`; unknown names as `InvalidOperationException`. + +## Usage + +```csharp +using DryIoc; +using SquidStd.Templating.Extensions; +using SquidStd.Templating.Interfaces; + +container.AddTemplating(); // after RegisterCoreServices (a shared DirectoriesConfig is registered there) + +var renderer = container.Resolve(); + +// ad-hoc +var hi = await renderer.RenderAsync("Hi {{ user.name }}!", new { User = new { Name = "squid" } }); + +// named (registered manually or auto-loaded from templates/) +renderer.Register("welcome", "Welcome {{ user.name }}"); +var welcome = await renderer.RenderByNameAsync("welcome", new { User = new { Name = "squid" } }); +``` + +## Key types + +| Type | Purpose | +|------|---------| +| `ITemplateRenderer` | Render ad-hoc and named templates. | +| `ScribanTemplateRenderer` | Scriban implementation; compiles/caches named templates and auto-loads `templates/*.tmpl`. | +| `TemplateException` | Raised on parse/render failures. | +| `TemplatingRegistrationExtensions` | `AddTemplating(...)` registration. | + +## License + +MIT — part of [SquidStd](https://github.com/tgiachi/squid-std). From 5ed2778fa8a5652395ab7f3bf5f5ba4fcdfcc248 Mon Sep 17 00:00:00 2001 From: Tom Date: Tue, 23 Jun 2026 12:00:19 +0200 Subject: [PATCH 06/33] feat(messaging): add topic pub/sub contracts, facade and event-bus bridge --- .../Data/Events/TopicMessageEvent.cs | 9 +++++ .../Interfaces/IMessageTopic.cs | 13 +++++++ .../Interfaces/ITopicEventBridge.cs | 11 ++++++ .../Interfaces/ITopicProvider.cs | 16 +++++++++ .../Services/MessageTopic.cs | 34 +++++++++++++++++++ .../Services/TopicEventBridge.cs | 25 ++++++++++++++ 6 files changed, 108 insertions(+) create mode 100644 src/SquidStd.Messaging.Abstractions/Data/Events/TopicMessageEvent.cs create mode 100644 src/SquidStd.Messaging.Abstractions/Interfaces/IMessageTopic.cs create mode 100644 src/SquidStd.Messaging.Abstractions/Interfaces/ITopicEventBridge.cs create mode 100644 src/SquidStd.Messaging.Abstractions/Interfaces/ITopicProvider.cs create mode 100644 src/SquidStd.Messaging.Abstractions/Services/MessageTopic.cs create mode 100644 src/SquidStd.Messaging.Abstractions/Services/TopicEventBridge.cs diff --git a/src/SquidStd.Messaging.Abstractions/Data/Events/TopicMessageEvent.cs b/src/SquidStd.Messaging.Abstractions/Data/Events/TopicMessageEvent.cs new file mode 100644 index 00000000..b3a43554 --- /dev/null +++ b/src/SquidStd.Messaging.Abstractions/Data/Events/TopicMessageEvent.cs @@ -0,0 +1,9 @@ +using SquidStd.Core.Interfaces.Events; + +namespace SquidStd.Messaging.Abstractions.Data.Events; + +/// +/// Event published on the in-process event bus when a topic message is bridged. is the +/// deserialized message. +/// +public sealed record TopicMessageEvent(string Topic, object Data) : IEvent; diff --git a/src/SquidStd.Messaging.Abstractions/Interfaces/IMessageTopic.cs b/src/SquidStd.Messaging.Abstractions/Interfaces/IMessageTopic.cs new file mode 100644 index 00000000..e1b1c2be --- /dev/null +++ b/src/SquidStd.Messaging.Abstractions/Interfaces/IMessageTopic.cs @@ -0,0 +1,13 @@ +namespace SquidStd.Messaging.Abstractions.Interfaces; + +/// +/// Typed facade for publishing to and subscribing to topics (fan-out). +/// +public interface IMessageTopic +{ + /// Publishes a typed message to a topic. + Task PublishAsync(string topic, TMessage message, CancellationToken cancellationToken = default); + + /// Subscribes a typed handler to a topic. Dispose to unsubscribe. + IDisposable Subscribe(string topic, Func handler); +} diff --git a/src/SquidStd.Messaging.Abstractions/Interfaces/ITopicEventBridge.cs b/src/SquidStd.Messaging.Abstractions/Interfaces/ITopicEventBridge.cs new file mode 100644 index 00000000..2ba8483c --- /dev/null +++ b/src/SquidStd.Messaging.Abstractions/Interfaces/ITopicEventBridge.cs @@ -0,0 +1,11 @@ +namespace SquidStd.Messaging.Abstractions.Interfaces; + +/// +/// Bridges a topic into the in-process event bus: each message of type T on the topic is republished +/// as a TopicMessageEvent. +/// +public interface ITopicEventBridge +{ + /// Starts bridging the given topic; dispose the result to stop. + IDisposable Bridge(string topic); +} diff --git a/src/SquidStd.Messaging.Abstractions/Interfaces/ITopicProvider.cs b/src/SquidStd.Messaging.Abstractions/Interfaces/ITopicProvider.cs new file mode 100644 index 00000000..c5882dc4 --- /dev/null +++ b/src/SquidStd.Messaging.Abstractions/Interfaces/ITopicProvider.cs @@ -0,0 +1,16 @@ +using SquidStd.Abstractions.Interfaces.Services; + +namespace SquidStd.Messaging.Abstractions.Interfaces; + +/// +/// Byte-level publish/subscribe transport: every current subscriber of a topic receives every message +/// (transient, at-most-once fan-out). +/// +public interface ITopicProvider : ISquidStdService, IAsyncDisposable +{ + /// Publishes a raw payload to a topic (fan-out to all current subscribers). + Task PublishAsync(string topic, ReadOnlyMemory payload, CancellationToken cancellationToken = default); + + /// Subscribes a raw handler to a topic. Dispose to unsubscribe. + IDisposable Subscribe(string topic, Func, CancellationToken, Task> handler); +} diff --git a/src/SquidStd.Messaging.Abstractions/Services/MessageTopic.cs b/src/SquidStd.Messaging.Abstractions/Services/MessageTopic.cs new file mode 100644 index 00000000..f6c4a6cc --- /dev/null +++ b/src/SquidStd.Messaging.Abstractions/Services/MessageTopic.cs @@ -0,0 +1,34 @@ +using SquidStd.Core.Interfaces.Serialization; +using SquidStd.Messaging.Abstractions.Interfaces; + +namespace SquidStd.Messaging.Abstractions.Services; + +/// +/// Typed facade over an : serializes outgoing messages and deserializes +/// incoming payloads. +/// +public sealed class MessageTopic : IMessageTopic +{ + private readonly ITopicProvider _provider; + private readonly IDataSerializer _serializer; + private readonly IDataDeserializer _deserializer; + + public MessageTopic(ITopicProvider provider, IDataSerializer serializer, IDataDeserializer deserializer) + { + _provider = provider; + _serializer = serializer; + _deserializer = deserializer; + } + + /// + public Task PublishAsync(string topic, TMessage message, CancellationToken cancellationToken = default) + => _provider.PublishAsync(topic, _serializer.Serialize(message), cancellationToken); + + /// + public IDisposable Subscribe(string topic, Func handler) + { + ArgumentNullException.ThrowIfNull(handler); + + return _provider.Subscribe(topic, (payload, cancellationToken) => handler(_deserializer.Deserialize(payload), cancellationToken)); + } +} diff --git a/src/SquidStd.Messaging.Abstractions/Services/TopicEventBridge.cs b/src/SquidStd.Messaging.Abstractions/Services/TopicEventBridge.cs new file mode 100644 index 00000000..9cebd1f8 --- /dev/null +++ b/src/SquidStd.Messaging.Abstractions/Services/TopicEventBridge.cs @@ -0,0 +1,25 @@ +using SquidStd.Core.Interfaces.Events; +using SquidStd.Messaging.Abstractions.Data.Events; +using SquidStd.Messaging.Abstractions.Interfaces; + +namespace SquidStd.Messaging.Abstractions.Services; + +/// +/// One-way Topic → EventBus bridge. Subscribes a topic and republishes each message as a +/// on the . +/// +public sealed class TopicEventBridge : ITopicEventBridge +{ + private readonly IMessageTopic _topic; + private readonly IEventBus _eventBus; + + public TopicEventBridge(IMessageTopic topic, IEventBus eventBus) + { + _topic = topic; + _eventBus = eventBus; + } + + /// + public IDisposable Bridge(string topic) + => _topic.Subscribe(topic, (data, cancellationToken) => _eventBus.PublishAsync(new TopicMessageEvent(topic, data!), cancellationToken)); +} From 06cee7025a0f04a3a2417ff5ac96061839c96522 Mon Sep 17 00:00:00 2001 From: Tom Date: Tue, 23 Jun 2026 12:05:26 +0200 Subject: [PATCH 07/33] feat(messaging): add in-memory topic provider, facade/bridge registration and tests --- .../MessagingRegistrationExtensions.cs | 4 + .../Services/InMemoryTopicProvider.cs | 96 +++++++++++++++++++ .../Messaging/InMemoryTopicProviderTests.cs | 58 +++++++++++ .../Messaging/MessageTopicTests.cs | 29 ++++++ .../MessagingRegistrationExtensionsTests.cs | 15 +++ .../Messaging/TopicEventBridgeTests.cs | 49 ++++++++++ 6 files changed, 251 insertions(+) create mode 100644 src/SquidStd.Messaging/Services/InMemoryTopicProvider.cs create mode 100644 tests/SquidStd.Tests/Messaging/InMemoryTopicProviderTests.cs create mode 100644 tests/SquidStd.Tests/Messaging/MessageTopicTests.cs create mode 100644 tests/SquidStd.Tests/Messaging/TopicEventBridgeTests.cs diff --git a/src/SquidStd.Messaging/Extensions/MessagingRegistrationExtensions.cs b/src/SquidStd.Messaging/Extensions/MessagingRegistrationExtensions.cs index 92a3299a..a2ca9fee 100644 --- a/src/SquidStd.Messaging/Extensions/MessagingRegistrationExtensions.cs +++ b/src/SquidStd.Messaging/Extensions/MessagingRegistrationExtensions.cs @@ -37,6 +37,10 @@ public static IContainer AddInMemoryMessaging(this IContainer container, Messagi container.Register(Reuse.Singleton); container.Register(Reuse.Singleton); + container.Register(Reuse.Singleton); + container.Register(Reuse.Singleton); + container.Register(Reuse.Singleton); + return container; } diff --git a/src/SquidStd.Messaging/Services/InMemoryTopicProvider.cs b/src/SquidStd.Messaging/Services/InMemoryTopicProvider.cs new file mode 100644 index 00000000..3a950261 --- /dev/null +++ b/src/SquidStd.Messaging/Services/InMemoryTopicProvider.cs @@ -0,0 +1,96 @@ +using System.Collections.Concurrent; +using Serilog; +using SquidStd.Messaging.Abstractions.Interfaces; + +namespace SquidStd.Messaging.Services; + +/// +/// In-memory : fan-out delivery to all current subscribers of a topic. +/// Transient and at-most-once; exceptions in one subscriber are isolated. +/// +public sealed class InMemoryTopicProvider : ITopicProvider +{ + private readonly ILogger _logger = Log.ForContext(); + private readonly ConcurrentDictionary, CancellationToken, Task>>> _topics = new(StringComparer.Ordinal); + private int _disposed; + + /// + public ValueTask StartAsync(CancellationToken cancellationToken = default) + => ValueTask.CompletedTask; + + /// + public ValueTask StopAsync(CancellationToken cancellationToken = default) + => DisposeAsync(); + + /// + public async Task PublishAsync(string topic, ReadOnlyMemory payload, CancellationToken cancellationToken = default) + { + ArgumentException.ThrowIfNullOrWhiteSpace(topic); + + if (!_topics.TryGetValue(topic, out var handlers)) + { + return; + } + + foreach (var handler in handlers.Values) + { + try + { + await handler(payload, cancellationToken); + } + catch (Exception ex) + { + _logger.Error(ex, "Topic '{Topic}' subscriber failed", topic); + } + } + } + + /// + public IDisposable Subscribe(string topic, Func, CancellationToken, Task> handler) + { + ArgumentException.ThrowIfNullOrWhiteSpace(topic); + ArgumentNullException.ThrowIfNull(handler); + + var handlers = _topics.GetOrAdd(topic, static _ => new()); + var id = Guid.NewGuid(); + handlers[id] = handler; + + return new Subscription(handlers, id); + } + + /// + public ValueTask DisposeAsync() + { + if (Interlocked.Exchange(ref _disposed, 1) != 0) + { + return ValueTask.CompletedTask; + } + + _topics.Clear(); + + return ValueTask.CompletedTask; + } + + private sealed class Subscription : IDisposable + { + private readonly ConcurrentDictionary, CancellationToken, Task>> _handlers; + private readonly Guid _id; + private int _disposed; + + public Subscription(ConcurrentDictionary, CancellationToken, Task>> handlers, Guid id) + { + _handlers = handlers; + _id = id; + } + + public void Dispose() + { + if (Interlocked.Exchange(ref _disposed, 1) != 0) + { + return; + } + + _handlers.TryRemove(_id, out _); + } + } +} diff --git a/tests/SquidStd.Tests/Messaging/InMemoryTopicProviderTests.cs b/tests/SquidStd.Tests/Messaging/InMemoryTopicProviderTests.cs new file mode 100644 index 00000000..7757d639 --- /dev/null +++ b/tests/SquidStd.Tests/Messaging/InMemoryTopicProviderTests.cs @@ -0,0 +1,58 @@ +using System.Text; +using SquidStd.Messaging.Services; + +namespace SquidStd.Tests.Messaging; + +public class InMemoryTopicProviderTests +{ + private static ReadOnlyMemory Bytes(string s) => Encoding.UTF8.GetBytes(s); + + [Fact] + public async Task Publish_FansOutToAllSubscribers() + { + await using var provider = new InMemoryTopicProvider(); + var a = 0; + var b = 0; + provider.Subscribe("t", (_, _) => { Interlocked.Increment(ref a); return Task.CompletedTask; }); + provider.Subscribe("t", (_, _) => { Interlocked.Increment(ref b); return Task.CompletedTask; }); + + await provider.PublishAsync("t", Bytes("x")); + + Assert.Equal(1, a); + Assert.Equal(1, b); + } + + [Fact] + public async Task Dispose_StopsDelivery() + { + await using var provider = new InMemoryTopicProvider(); + var count = 0; + var subscription = provider.Subscribe("t", (_, _) => { Interlocked.Increment(ref count); return Task.CompletedTask; }); + + subscription.Dispose(); + await provider.PublishAsync("t", Bytes("x")); + + Assert.Equal(0, count); + } + + [Fact] + public async Task Publish_NoSubscribers_IsNoOp() + { + await using var provider = new InMemoryTopicProvider(); + + await provider.PublishAsync("t", Bytes("x")); + } + + [Fact] + public async Task FailingSubscriber_IsIsolated() + { + await using var provider = new InMemoryTopicProvider(); + var received = 0; + provider.Subscribe("t", (_, _) => throw new InvalidOperationException("boom")); + provider.Subscribe("t", (_, _) => { Interlocked.Increment(ref received); return Task.CompletedTask; }); + + await provider.PublishAsync("t", Bytes("x")); + + Assert.Equal(1, received); + } +} diff --git a/tests/SquidStd.Tests/Messaging/MessageTopicTests.cs b/tests/SquidStd.Tests/Messaging/MessageTopicTests.cs new file mode 100644 index 00000000..dcbcafa2 --- /dev/null +++ b/tests/SquidStd.Tests/Messaging/MessageTopicTests.cs @@ -0,0 +1,29 @@ +using SquidStd.Core.Json; +using SquidStd.Messaging.Abstractions.Interfaces; +using SquidStd.Messaging.Abstractions.Services; +using SquidStd.Messaging.Services; + +namespace SquidStd.Tests.Messaging; + +public class MessageTopicTests +{ + private sealed class Ping + { + public string Source { get; set; } = ""; + } + + [Fact] + public async Task PublishSubscribe_RoundTripsTypedMessage() + { + await using var provider = new InMemoryTopicProvider(); + var serializer = new JsonDataSerializer(); + IMessageTopic topic = new MessageTopic(provider, serializer, serializer); + var received = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + topic.Subscribe("pings", (ping, _) => { received.TrySetResult(ping); return Task.CompletedTask; }); + + await topic.PublishAsync("pings", new Ping { Source = "w1" }); + + var got = await received.Task.WaitAsync(TimeSpan.FromSeconds(5)); + Assert.Equal("w1", got.Source); + } +} diff --git a/tests/SquidStd.Tests/Messaging/MessagingRegistrationExtensionsTests.cs b/tests/SquidStd.Tests/Messaging/MessagingRegistrationExtensionsTests.cs index 140608b0..e06330f0 100644 --- a/tests/SquidStd.Tests/Messaging/MessagingRegistrationExtensionsTests.cs +++ b/tests/SquidStd.Tests/Messaging/MessagingRegistrationExtensionsTests.cs @@ -1,9 +1,11 @@ using DryIoc; +using SquidStd.Core.Interfaces.Events; using SquidStd.Core.Interfaces.Metrics; using SquidStd.Core.Interfaces.Serialization; using SquidStd.Messaging.Extensions; using SquidStd.Messaging.Services; using SquidStd.Messaging.Abstractions.Interfaces; +using SquidStd.Services.Core.Services; namespace SquidStd.Tests.Messaging; @@ -34,4 +36,17 @@ public void AddInMemoryMessaging_MetricsAndProviderShareInstance() Assert.Same(container.Resolve(), container.Resolve()); } + + [Fact] + public void AddInMemoryMessaging_RegistersTopicServices() + { + using var container = new Container(); + container.RegisterInstance(new EventBusService()); + + container.AddInMemoryMessaging(); + + Assert.NotNull(container.Resolve()); + Assert.NotNull(container.Resolve()); + Assert.NotNull(container.Resolve()); + } } diff --git a/tests/SquidStd.Tests/Messaging/TopicEventBridgeTests.cs b/tests/SquidStd.Tests/Messaging/TopicEventBridgeTests.cs new file mode 100644 index 00000000..d68276d7 --- /dev/null +++ b/tests/SquidStd.Tests/Messaging/TopicEventBridgeTests.cs @@ -0,0 +1,49 @@ +using SquidStd.Core.Interfaces.Events; +using SquidStd.Core.Json; +using SquidStd.Messaging.Abstractions.Data.Events; +using SquidStd.Messaging.Abstractions.Interfaces; +using SquidStd.Messaging.Abstractions.Services; +using SquidStd.Messaging.Services; +using SquidStd.Services.Core.Services; + +namespace SquidStd.Tests.Messaging; + +public class TopicEventBridgeTests +{ + private sealed class Beat + { + public string Worker { get; set; } = ""; + } + + private sealed class CapturingListener : IAsyncEventListener + { + public TaskCompletionSource Received { get; } = new(TaskCreationOptions.RunContinuationsAsynchronously); + + public Task HandleAsync(TopicMessageEvent eventData, CancellationToken cancellationToken) + { + Received.TrySetResult(eventData); + + return Task.CompletedTask; + } + } + + [Fact] + public async Task Bridge_RepublishesTopicMessageOnEventBus() + { + await using var provider = new InMemoryTopicProvider(); + var serializer = new JsonDataSerializer(); + IMessageTopic topic = new MessageTopic(provider, serializer, serializer); + var eventBus = new EventBusService(); + ITopicEventBridge bridge = new TopicEventBridge(topic, eventBus); + + var listener = new CapturingListener(); + eventBus.RegisterAsyncListener(listener); + using var subscription = bridge.Bridge("workers.heartbeat"); + + await topic.PublishAsync("workers.heartbeat", new Beat { Worker = "w1" }); + + var evt = await listener.Received.Task.WaitAsync(TimeSpan.FromSeconds(5)); + Assert.Equal("workers.heartbeat", evt.Topic); + Assert.Equal("w1", Assert.IsType(evt.Data).Worker); + } +} From 83407ec1ad08ab9c870738fcc41abe4a5b9e283c Mon Sep 17 00:00:00 2001 From: Tom Date: Tue, 23 Jun 2026 12:07:41 +0200 Subject: [PATCH 08/33] feat(messaging): add RabbitMQ fanout topic provider and registration --- ...RabbitMqMessagingRegistrationExtensions.cs | 4 + .../Services/RabbitMqTopicProvider.cs | 218 ++++++++++++++++++ .../RabbitMq/RabbitMqTopicProviderTests.cs | 59 +++++ 3 files changed, 281 insertions(+) create mode 100644 src/SquidStd.Messaging.RabbitMq/Services/RabbitMqTopicProvider.cs create mode 100644 tests/SquidStd.Tests/Messaging/RabbitMq/RabbitMqTopicProviderTests.cs diff --git a/src/SquidStd.Messaging.RabbitMq/Extensions/RabbitMqMessagingRegistrationExtensions.cs b/src/SquidStd.Messaging.RabbitMq/Extensions/RabbitMqMessagingRegistrationExtensions.cs index b56cd98c..839e8ccf 100644 --- a/src/SquidStd.Messaging.RabbitMq/Extensions/RabbitMqMessagingRegistrationExtensions.cs +++ b/src/SquidStd.Messaging.RabbitMq/Extensions/RabbitMqMessagingRegistrationExtensions.cs @@ -49,6 +49,10 @@ public static IContainer AddRabbitMqMessaging( ); container.Register(Reuse.Singleton); + container.Register(Reuse.Singleton); + container.Register(Reuse.Singleton); + container.Register(Reuse.Singleton); + return container; } diff --git a/src/SquidStd.Messaging.RabbitMq/Services/RabbitMqTopicProvider.cs b/src/SquidStd.Messaging.RabbitMq/Services/RabbitMqTopicProvider.cs new file mode 100644 index 00000000..ccb60449 --- /dev/null +++ b/src/SquidStd.Messaging.RabbitMq/Services/RabbitMqTopicProvider.cs @@ -0,0 +1,218 @@ +using RabbitMQ.Client; +using RabbitMQ.Client.Events; +using Serilog; +using SquidStd.Messaging.Abstractions.Interfaces; +using SquidStd.Messaging.RabbitMq.Data.Config; + +namespace SquidStd.Messaging.RabbitMq.Services; + +/// +/// RabbitMQ : topics map to fanout exchanges; each subscriber binds an +/// exclusive auto-delete queue and consumes with auto-ack (transient, at-most-once fan-out). +/// +public sealed class RabbitMqTopicProvider : ITopicProvider +{ + private readonly ILogger _logger = Log.ForContext(); + private readonly RabbitMqOptions _options; + private readonly SemaphoreSlim _publishLock = new(1, 1); + private readonly Lock _exchangeSync = new(); + private readonly HashSet _declared = new(StringComparer.Ordinal); + private IConnection? _connection; + private IChannel? _publishChannel; + private int _disposed; + + public RabbitMqTopicProvider(RabbitMqOptions options) + { + _options = options; + } + + /// + public async ValueTask StartAsync(CancellationToken cancellationToken = default) + { + var factory = new ConnectionFactory { AutomaticRecoveryEnabled = true }; + + if (_options.Uri is not null) + { + factory.Uri = _options.Uri; + } + else + { + factory.HostName = _options.HostName; + factory.Port = _options.Port; + factory.VirtualHost = _options.VirtualHost; + factory.UserName = _options.UserName; + factory.Password = _options.Password; + } + + _connection = await factory.CreateConnectionAsync(cancellationToken); + _publishChannel = await _connection.CreateChannelAsync(cancellationToken: cancellationToken); + } + + /// + public ValueTask StopAsync(CancellationToken cancellationToken = default) + => DisposeAsync(); + + /// + public async Task PublishAsync(string topic, ReadOnlyMemory payload, CancellationToken cancellationToken = default) + { + ArgumentException.ThrowIfNullOrWhiteSpace(topic); + var channel = _publishChannel ?? throw new InvalidOperationException("Provider not started."); + + await EnsureExchangeAsync(channel, topic, cancellationToken); + + var properties = new BasicProperties(); + + await _publishLock.WaitAsync(cancellationToken); + + try + { + await channel.BasicPublishAsync( + exchange: topic, + routingKey: string.Empty, + mandatory: false, + basicProperties: properties, + body: payload, + cancellationToken: cancellationToken + ); + } + finally + { + _publishLock.Release(); + } + } + + /// + public IDisposable Subscribe(string topic, Func, CancellationToken, Task> handler) + { + ArgumentException.ThrowIfNullOrWhiteSpace(topic); + ArgumentNullException.ThrowIfNull(handler); + + var connection = _connection ?? throw new InvalidOperationException("Provider not started."); + var subscription = new Subscription(this, connection, topic, handler); + subscription.Start(); + + return subscription; + } + + private async Task EnsureExchangeAsync(IChannel channel, string topic, CancellationToken cancellationToken) + { + lock (_exchangeSync) + { + if (!_declared.Add(topic)) + { + return; + } + } + + await channel.ExchangeDeclareAsync( + exchange: topic, + type: ExchangeType.Fanout, + durable: false, + autoDelete: false, + cancellationToken: cancellationToken + ); + } + + /// + public async ValueTask DisposeAsync() + { + if (Interlocked.Exchange(ref _disposed, 1) != 0) + { + return; + } + + if (_publishChannel is not null) + { + await _publishChannel.CloseAsync(); + await _publishChannel.DisposeAsync(); + } + + if (_connection is not null) + { + await _connection.CloseAsync(); + await _connection.DisposeAsync(); + } + + _publishLock.Dispose(); + } + + private sealed class Subscription : IDisposable + { + private readonly RabbitMqTopicProvider _provider; + private readonly IConnection _connection; + private readonly string _topic; + private readonly Func, CancellationToken, Task> _handler; + private IChannel? _channel; + private string? _consumerTag; + private int _disposed; + + public Subscription( + RabbitMqTopicProvider provider, + IConnection connection, + string topic, + Func, CancellationToken, Task> handler + ) + { + _provider = provider; + _connection = connection; + _topic = topic; + _handler = handler; + } + + public void Start() + => StartAsync().GetAwaiter().GetResult(); + + private async Task StartAsync() + { + _channel = await _connection.CreateChannelAsync(); + + await _channel.ExchangeDeclareAsync(_topic, ExchangeType.Fanout, durable: false, autoDelete: false); + + var queue = await _channel.QueueDeclareAsync(queue: string.Empty, durable: false, exclusive: true, autoDelete: true); + await _channel.QueueBindAsync(queue: queue.QueueName, exchange: _topic, routingKey: string.Empty); + + var consumer = new AsyncEventingBasicConsumer(_channel); + consumer.ReceivedAsync += OnReceivedAsync; + + _consumerTag = await _channel.BasicConsumeAsync(queue.QueueName, autoAck: true, consumer: consumer); + } + + private async Task OnReceivedAsync(object sender, BasicDeliverEventArgs args) + { + try + { + await _handler(args.Body, CancellationToken.None); + } + catch (Exception ex) + { + _provider._logger.Warning(ex, "RabbitMq topic '{Topic}' handler failed", _topic); + } + } + + public void Dispose() + { + if (Interlocked.Exchange(ref _disposed, 1) != 0) + { + return; + } + + if (_channel is not null) + { + try + { + if (_consumerTag is not null) + { + _channel.BasicCancelAsync(_consumerTag).GetAwaiter().GetResult(); + } + + _channel.CloseAsync().GetAwaiter().GetResult(); + _channel.DisposeAsync().AsTask().GetAwaiter().GetResult(); + } + catch + { + // Best-effort teardown. + } + } + } + } +} diff --git a/tests/SquidStd.Tests/Messaging/RabbitMq/RabbitMqTopicProviderTests.cs b/tests/SquidStd.Tests/Messaging/RabbitMq/RabbitMqTopicProviderTests.cs new file mode 100644 index 00000000..28775802 --- /dev/null +++ b/tests/SquidStd.Tests/Messaging/RabbitMq/RabbitMqTopicProviderTests.cs @@ -0,0 +1,59 @@ +using System.Text; +using SquidStd.Messaging.RabbitMq.Data.Config; +using SquidStd.Messaging.RabbitMq.Services; + +namespace SquidStd.Tests.Messaging.RabbitMq; + +[Collection(RabbitMqCollection.Name)] +public class RabbitMqTopicProviderTests +{ + private static readonly TimeSpan Timeout = TimeSpan.FromSeconds(30); + + private readonly RabbitMqContainerFixture _fixture; + + public RabbitMqTopicProviderTests(RabbitMqContainerFixture fixture) + { + _fixture = fixture; + } + + private RabbitMqTopicProvider NewProvider() + => new(new RabbitMqOptions { Uri = new Uri(_fixture.AmqpUri) }); + + private static ReadOnlyMemory Bytes(string s) => Encoding.UTF8.GetBytes(s); + private static string Topic() => "topic-" + Guid.NewGuid().ToString("N"); + + [Fact] + public async Task Publish_FansOutToAllSubscribers() + { + await using var provider = NewProvider(); + await provider.StartAsync(); + var topic = Topic(); + var a = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var b = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + provider.Subscribe(topic, (payload, _) => { a.TrySetResult(Encoding.UTF8.GetString(payload.Span)); return Task.CompletedTask; }); + provider.Subscribe(topic, (payload, _) => { b.TrySetResult(Encoding.UTF8.GetString(payload.Span)); return Task.CompletedTask; }); + + // Allow the broker to register the bindings before publishing. + await Task.Delay(300); + await provider.PublishAsync(topic, Bytes("hello")); + + Assert.Equal("hello", await a.Task.WaitAsync(Timeout)); + Assert.Equal("hello", await b.Task.WaitAsync(Timeout)); + } + + [Fact] + public async Task Publish_BeforeAnySubscriber_IsNotReceived() + { + await using var provider = NewProvider(); + await provider.StartAsync(); + var topic = Topic(); + + await provider.PublishAsync(topic, Bytes("lost")); + + var received = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + provider.Subscribe(topic, (payload, _) => { received.TrySetResult(Encoding.UTF8.GetString(payload.Span)); return Task.CompletedTask; }); + await Task.Delay(300); + + await Assert.ThrowsAsync(async () => await received.Task.WaitAsync(TimeSpan.FromSeconds(1))); + } +} From 00d98789ba4af63c429e2115a3bcb00fe82fb57b Mon Sep 17 00:00:00 2001 From: Tom Date: Tue, 23 Jun 2026 12:19:45 +0200 Subject: [PATCH 09/33] build(workers): scaffold SquidStd.Workers.Abstractions project --- SquidStd.slnx | 1 + .../SquidStd.Workers.Abstractions.csproj | 10 ++++++++++ 2 files changed, 11 insertions(+) create mode 100644 src/SquidStd.Workers.Abstractions/SquidStd.Workers.Abstractions.csproj diff --git a/SquidStd.slnx b/SquidStd.slnx index 5a6da28f..dea09079 100644 --- a/SquidStd.slnx +++ b/SquidStd.slnx @@ -19,6 +19,7 @@ + diff --git a/src/SquidStd.Workers.Abstractions/SquidStd.Workers.Abstractions.csproj b/src/SquidStd.Workers.Abstractions/SquidStd.Workers.Abstractions.csproj new file mode 100644 index 00000000..2f0e9b5b --- /dev/null +++ b/src/SquidStd.Workers.Abstractions/SquidStd.Workers.Abstractions.csproj @@ -0,0 +1,10 @@ + + + + net10.0 + enable + enable + true + + + From c8a9fbd2e504b0e3a2a772791dfd828882712721 Mon Sep 17 00:00:00 2001 From: Tom Date: Tue, 23 Jun 2026 12:20:32 +0200 Subject: [PATCH 10/33] feat(workers): add WorkerStatusType enum --- .../Types/WorkerStatusType.cs | 17 +++++++++++++++++ 1 file changed, 17 insertions(+) create mode 100644 src/SquidStd.Workers.Abstractions/Types/WorkerStatusType.cs diff --git a/src/SquidStd.Workers.Abstractions/Types/WorkerStatusType.cs b/src/SquidStd.Workers.Abstractions/Types/WorkerStatusType.cs new file mode 100644 index 00000000..d3aa947c --- /dev/null +++ b/src/SquidStd.Workers.Abstractions/Types/WorkerStatusType.cs @@ -0,0 +1,17 @@ +namespace SquidStd.Workers.Abstractions.Types; + +/// +/// Lifecycle status of a worker. Workers report or ; +/// the manager assigns when a worker's heartbeats stop arriving. +/// +public enum WorkerStatusType +{ + /// The worker is running and has no job in progress. + Idle, + + /// The worker is currently processing a job. + Busy, + + /// The manager has not seen a heartbeat within the expected window. + Offline +} From 2a74d589bd930ebb01e19420a665cde3ec03afd6 Mon Sep 17 00:00:00 2001 From: Tom Date: Tue, 23 Jun 2026 12:20:32 +0200 Subject: [PATCH 11/33] feat(workers): add JobRequest, WorkerHeartbeat and WorkerInfo contracts --- .../Data/JobRequest.cs | 8 ++++++++ .../Data/WorkerHeartbeat.cs | 16 ++++++++++++++++ .../Data/WorkerInfo.cs | 18 ++++++++++++++++++ 3 files changed, 42 insertions(+) create mode 100644 src/SquidStd.Workers.Abstractions/Data/JobRequest.cs create mode 100644 src/SquidStd.Workers.Abstractions/Data/WorkerHeartbeat.cs create mode 100644 src/SquidStd.Workers.Abstractions/Data/WorkerInfo.cs diff --git a/src/SquidStd.Workers.Abstractions/Data/JobRequest.cs b/src/SquidStd.Workers.Abstractions/Data/JobRequest.cs new file mode 100644 index 00000000..0effc8c4 --- /dev/null +++ b/src/SquidStd.Workers.Abstractions/Data/JobRequest.cs @@ -0,0 +1,8 @@ +namespace SquidStd.Workers.Abstractions.Data; + +/// +/// A unit of work the manager enqueues on the jobs queue for a worker to execute. +/// +/// Logical name of the job, used by the worker to pick a handler. +/// Opaque string key/value arguments for the job. +public sealed record JobRequest(string JobName, IReadOnlyDictionary Parameters); diff --git a/src/SquidStd.Workers.Abstractions/Data/WorkerHeartbeat.cs b/src/SquidStd.Workers.Abstractions/Data/WorkerHeartbeat.cs new file mode 100644 index 00000000..46799817 --- /dev/null +++ b/src/SquidStd.Workers.Abstractions/Data/WorkerHeartbeat.cs @@ -0,0 +1,16 @@ +using SquidStd.Workers.Abstractions.Types; + +namespace SquidStd.Workers.Abstractions.Data; + +/// +/// A liveness signal a worker publishes on the heartbeat topic at a fixed interval. +/// +/// Stable identity of the reporting worker. +/// UTC time the heartbeat was produced. +/// Self-reported status ( or ). +/// Name of the job in progress, or null when idle. +public sealed record WorkerHeartbeat( + string WorkerId, + DateTime TimestampUtc, + WorkerStatusType Status, + string? CurrentJob); diff --git a/src/SquidStd.Workers.Abstractions/Data/WorkerInfo.cs b/src/SquidStd.Workers.Abstractions/Data/WorkerInfo.cs new file mode 100644 index 00000000..4510a105 --- /dev/null +++ b/src/SquidStd.Workers.Abstractions/Data/WorkerInfo.cs @@ -0,0 +1,18 @@ +using SquidStd.Workers.Abstractions.Types; + +namespace SquidStd.Workers.Abstractions.Data; + +/// +/// The manager-side view of a worker, folded from incoming heartbeats. +/// +/// Stable identity of the worker. +/// Last known status; the manager sets on missed heartbeats. +/// UTC time the manager first saw this worker. +/// UTC time of the most recent heartbeat. +/// Name of the job last reported in progress, or null. +public sealed record WorkerInfo( + string WorkerId, + WorkerStatusType Status, + DateTime FirstSeenUtc, + DateTime LastSeenUtc, + string? CurrentJob); From 94f44c0352499d65655d8b363aa3f19ea85392bd Mon Sep 17 00:00:00 2001 From: Tom Date: Tue, 23 Jun 2026 12:20:32 +0200 Subject: [PATCH 12/33] feat(workers): add WorkerChannels conventional channel names --- .../WorkerChannels.cs | 14 ++++++++++++++ 1 file changed, 14 insertions(+) create mode 100644 src/SquidStd.Workers.Abstractions/WorkerChannels.cs diff --git a/src/SquidStd.Workers.Abstractions/WorkerChannels.cs b/src/SquidStd.Workers.Abstractions/WorkerChannels.cs new file mode 100644 index 00000000..6514a147 --- /dev/null +++ b/src/SquidStd.Workers.Abstractions/WorkerChannels.cs @@ -0,0 +1,14 @@ +namespace SquidStd.Workers.Abstractions; + +/// +/// Default names of the messaging channels the workers system uses. Shared so the manager +/// and workers agree out of the box; either side may override them via its own configuration. +/// +public static class WorkerChannels +{ + /// Competing-consumers queue the manager publishes JobRequests onto. + public const string JobQueue = "squidstd.workers.jobs"; + + /// Fan-out topic workers publish WorkerHeartbeats onto. + public const string HeartbeatTopic = "squidstd.workers.heartbeat"; +} From 3122cdfea71796ac1e700869f218b821880a4391 Mon Sep 17 00:00:00 2001 From: Tom Date: Tue, 23 Jun 2026 12:21:34 +0200 Subject: [PATCH 13/33] test(workers): reference SquidStd.Workers.Abstractions from test project --- tests/SquidStd.Tests/SquidStd.Tests.csproj | 1 + 1 file changed, 1 insertion(+) diff --git a/tests/SquidStd.Tests/SquidStd.Tests.csproj b/tests/SquidStd.Tests/SquidStd.Tests.csproj index ad45b91e..71f02f75 100644 --- a/tests/SquidStd.Tests/SquidStd.Tests.csproj +++ b/tests/SquidStd.Tests/SquidStd.Tests.csproj @@ -49,6 +49,7 @@ + From db448515e76472cd7eb45ffc0edf9aae3c68978c Mon Sep 17 00:00:00 2001 From: Tom Date: Tue, 23 Jun 2026 12:21:35 +0200 Subject: [PATCH 14/33] test(workers): add contract serialization round-trip tests --- .../Workers/WorkerContractsTests.cs | 95 +++++++++++++++++++ 1 file changed, 95 insertions(+) create mode 100644 tests/SquidStd.Tests/Workers/WorkerContractsTests.cs diff --git a/tests/SquidStd.Tests/Workers/WorkerContractsTests.cs b/tests/SquidStd.Tests/Workers/WorkerContractsTests.cs new file mode 100644 index 00000000..e6e2b003 --- /dev/null +++ b/tests/SquidStd.Tests/Workers/WorkerContractsTests.cs @@ -0,0 +1,95 @@ +using SquidStd.Core.Json; +using SquidStd.Workers.Abstractions; +using SquidStd.Workers.Abstractions.Data; +using SquidStd.Workers.Abstractions.Types; + +namespace SquidStd.Tests.Workers; + +public class WorkerContractsTests +{ + private static readonly JsonDataSerializer Serializer = new(); + + [Fact] + public void JobRequest_RoundTrips_PreservingNameAndParameters() + { + var original = new JobRequest( + "resize-image", + new Dictionary { ["width"] = "800", ["height"] = "600" }); + + var bytes = Serializer.Serialize(original); + var restored = Serializer.Deserialize(bytes); + + Assert.Equal("resize-image", restored.JobName); + Assert.Equal("800", restored.Parameters["width"]); + Assert.Equal("600", restored.Parameters["height"]); + } + + [Fact] + public void WorkerHeartbeat_RoundTrips_WithCurrentJob() + { + var original = new WorkerHeartbeat( + "worker-1", + new DateTime(2026, 6, 23, 10, 0, 0, DateTimeKind.Utc), + WorkerStatusType.Busy, + "resize-image"); + + var restored = Serializer.Deserialize(Serializer.Serialize(original)); + + Assert.Equal(original, restored); + } + + [Fact] + public void WorkerHeartbeat_RoundTrips_WhenIdleWithNullJob() + { + var original = new WorkerHeartbeat( + "worker-2", + new DateTime(2026, 6, 23, 10, 5, 0, DateTimeKind.Utc), + WorkerStatusType.Idle, + null); + + var restored = Serializer.Deserialize(Serializer.Serialize(original)); + + Assert.Equal(original, restored); + Assert.Null(restored.CurrentJob); + } + + [Theory] + [InlineData(WorkerStatusType.Idle)] + [InlineData(WorkerStatusType.Busy)] + [InlineData(WorkerStatusType.Offline)] + public void WorkerHeartbeat_RoundTrips_EveryStatus(WorkerStatusType status) + { + var original = new WorkerHeartbeat( + "worker-3", + new DateTime(2026, 6, 23, 11, 0, 0, DateTimeKind.Utc), + status, + null); + + var restored = Serializer.Deserialize(Serializer.Serialize(original)); + + Assert.Equal(status, restored.Status); + } + + [Fact] + public void WorkerInfo_RoundTrips_PreservingAllFields() + { + var original = new WorkerInfo( + "worker-4", + WorkerStatusType.Offline, + new DateTime(2026, 6, 23, 9, 0, 0, DateTimeKind.Utc), + new DateTime(2026, 6, 23, 9, 30, 0, DateTimeKind.Utc), + null); + + var restored = Serializer.Deserialize(Serializer.Serialize(original)); + + Assert.Equal(original, restored); + } + + [Fact] + public void WorkerChannels_NamesAreNonEmptyAndDistinct() + { + Assert.False(string.IsNullOrWhiteSpace(WorkerChannels.JobQueue)); + Assert.False(string.IsNullOrWhiteSpace(WorkerChannels.HeartbeatTopic)); + Assert.NotEqual(WorkerChannels.JobQueue, WorkerChannels.HeartbeatTopic); + } +} From f0e42ff03203e2d7cc52a9d4f9aa4702b95a69a8 Mon Sep 17 00:00:00 2001 From: Tom Date: Tue, 23 Jun 2026 12:41:30 +0200 Subject: [PATCH 15/33] build(workers): scaffold SquidStd.Workers project --- SquidStd.slnx | 1 + src/SquidStd.Workers/SquidStd.Workers.csproj | 15 +++++++++++++++ tests/SquidStd.Tests/SquidStd.Tests.csproj | 1 + 3 files changed, 17 insertions(+) create mode 100644 src/SquidStd.Workers/SquidStd.Workers.csproj diff --git a/SquidStd.slnx b/SquidStd.slnx index dea09079..77846e19 100644 --- a/SquidStd.slnx +++ b/SquidStd.slnx @@ -20,6 +20,7 @@ + diff --git a/src/SquidStd.Workers/SquidStd.Workers.csproj b/src/SquidStd.Workers/SquidStd.Workers.csproj new file mode 100644 index 00000000..ff5daeb4 --- /dev/null +++ b/src/SquidStd.Workers/SquidStd.Workers.csproj @@ -0,0 +1,15 @@ + + + + net10.0 + enable + enable + true + + + + + + + + diff --git a/tests/SquidStd.Tests/SquidStd.Tests.csproj b/tests/SquidStd.Tests/SquidStd.Tests.csproj index 71f02f75..73dd9959 100644 --- a/tests/SquidStd.Tests/SquidStd.Tests.csproj +++ b/tests/SquidStd.Tests/SquidStd.Tests.csproj @@ -50,6 +50,7 @@ + From 4fa86748d8f338f176df86ecc8223d2a23f10819 Mon Sep 17 00:00:00 2001 From: Tom Date: Tue, 23 Jun 2026 12:42:50 +0200 Subject: [PATCH 16/33] feat(workers): carry ActiveJobs and MaxConcurrency in heartbeat and worker info --- .../Data/WorkerHeartbeat.cs | 8 ++++-- .../Data/WorkerInfo.cs | 8 ++++-- .../Workers/WorkerContractsTests.cs | 28 ++++++------------- 3 files changed, 18 insertions(+), 26 deletions(-) diff --git a/src/SquidStd.Workers.Abstractions/Data/WorkerHeartbeat.cs b/src/SquidStd.Workers.Abstractions/Data/WorkerHeartbeat.cs index 46799817..1213b09a 100644 --- a/src/SquidStd.Workers.Abstractions/Data/WorkerHeartbeat.cs +++ b/src/SquidStd.Workers.Abstractions/Data/WorkerHeartbeat.cs @@ -7,10 +7,12 @@ namespace SquidStd.Workers.Abstractions.Data; /// /// Stable identity of the reporting worker. /// UTC time the heartbeat was produced. -/// Self-reported status ( or ). -/// Name of the job in progress, or null when idle. +/// Self-reported status ( when no job is running, else ). +/// Number of jobs currently in progress on this worker. +/// Maximum number of jobs the worker runs at once. public sealed record WorkerHeartbeat( string WorkerId, DateTime TimestampUtc, WorkerStatusType Status, - string? CurrentJob); + int ActiveJobs, + int MaxConcurrency); diff --git a/src/SquidStd.Workers.Abstractions/Data/WorkerInfo.cs b/src/SquidStd.Workers.Abstractions/Data/WorkerInfo.cs index 4510a105..22eac1ab 100644 --- a/src/SquidStd.Workers.Abstractions/Data/WorkerInfo.cs +++ b/src/SquidStd.Workers.Abstractions/Data/WorkerInfo.cs @@ -7,12 +7,14 @@ namespace SquidStd.Workers.Abstractions.Data; /// /// Stable identity of the worker. /// Last known status; the manager sets on missed heartbeats. +/// Jobs in progress as of the most recent heartbeat. +/// Maximum number of jobs the worker runs at once. /// UTC time the manager first saw this worker. /// UTC time of the most recent heartbeat. -/// Name of the job last reported in progress, or null. public sealed record WorkerInfo( string WorkerId, WorkerStatusType Status, + int ActiveJobs, + int MaxConcurrency, DateTime FirstSeenUtc, - DateTime LastSeenUtc, - string? CurrentJob); + DateTime LastSeenUtc); diff --git a/tests/SquidStd.Tests/Workers/WorkerContractsTests.cs b/tests/SquidStd.Tests/Workers/WorkerContractsTests.cs index e6e2b003..8e9df30b 100644 --- a/tests/SquidStd.Tests/Workers/WorkerContractsTests.cs +++ b/tests/SquidStd.Tests/Workers/WorkerContractsTests.cs @@ -25,34 +25,20 @@ public void JobRequest_RoundTrips_PreservingNameAndParameters() } [Fact] - public void WorkerHeartbeat_RoundTrips_WithCurrentJob() + public void WorkerHeartbeat_RoundTrips_PreservingAllFields() { var original = new WorkerHeartbeat( "worker-1", new DateTime(2026, 6, 23, 10, 0, 0, DateTimeKind.Utc), WorkerStatusType.Busy, - "resize-image"); + 3, + 8); var restored = Serializer.Deserialize(Serializer.Serialize(original)); Assert.Equal(original, restored); } - [Fact] - public void WorkerHeartbeat_RoundTrips_WhenIdleWithNullJob() - { - var original = new WorkerHeartbeat( - "worker-2", - new DateTime(2026, 6, 23, 10, 5, 0, DateTimeKind.Utc), - WorkerStatusType.Idle, - null); - - var restored = Serializer.Deserialize(Serializer.Serialize(original)); - - Assert.Equal(original, restored); - Assert.Null(restored.CurrentJob); - } - [Theory] [InlineData(WorkerStatusType.Idle)] [InlineData(WorkerStatusType.Busy)] @@ -63,7 +49,8 @@ public void WorkerHeartbeat_RoundTrips_EveryStatus(WorkerStatusType status) "worker-3", new DateTime(2026, 6, 23, 11, 0, 0, DateTimeKind.Utc), status, - null); + 0, + 4); var restored = Serializer.Deserialize(Serializer.Serialize(original)); @@ -76,9 +63,10 @@ public void WorkerInfo_RoundTrips_PreservingAllFields() var original = new WorkerInfo( "worker-4", WorkerStatusType.Offline, + 2, + 8, new DateTime(2026, 6, 23, 9, 0, 0, DateTimeKind.Utc), - new DateTime(2026, 6, 23, 9, 30, 0, DateTimeKind.Utc), - null); + new DateTime(2026, 6, 23, 9, 30, 0, DateTimeKind.Utc)); var restored = Serializer.Deserialize(Serializer.Serialize(original)); From 1b82a796f37ce30d31bc2e748d8e68d6e3805993 Mon Sep 17 00:00:00 2001 From: Tom Date: Tue, 23 Jun 2026 12:43:21 +0200 Subject: [PATCH 17/33] feat(workers): add WorkersConfig config section --- .../Data/Config/WorkersConfig.cs | 27 +++++++++++++++++++ 1 file changed, 27 insertions(+) create mode 100644 src/SquidStd.Workers/Data/Config/WorkersConfig.cs diff --git a/src/SquidStd.Workers/Data/Config/WorkersConfig.cs b/src/SquidStd.Workers/Data/Config/WorkersConfig.cs new file mode 100644 index 00000000..d9bf2b4f --- /dev/null +++ b/src/SquidStd.Workers/Data/Config/WorkersConfig.cs @@ -0,0 +1,27 @@ +using SquidStd.Workers.Abstractions; + +namespace SquidStd.Workers.Data.Config; + +/// +/// Configuration for the worker runtime (config section "workers"). +/// +public sealed class WorkersConfig +{ + /// + /// Stable worker identity. When blank, the runtime falls back to the machine name + /// (in Docker, the container hostname). + /// + public string WorkerId { get; set; } = string.Empty; + + /// Seconds between heartbeats. Falls back to 10 when not positive. + public int HeartbeatIntervalSeconds { get; set; } = 10; + + /// Maximum jobs processed in parallel. Falls back to the processor count when not positive. + public int MaxConcurrency { get; set; } + + /// Queue the worker consumes jobs from. + public string JobQueue { get; set; } = WorkerChannels.JobQueue; + + /// Topic the worker publishes heartbeats to. + public string HeartbeatTopic { get; set; } = WorkerChannels.HeartbeatTopic; +} From 544b5e4b7e579f59c19b9737b659101e37361a7f Mon Sep 17 00:00:00 2001 From: Tom Date: Tue, 23 Jun 2026 12:43:21 +0200 Subject: [PATCH 18/33] feat(workers): add IJobHandler and JobHandlerNotFoundException --- .../Exceptions/JobHandlerNotFoundException.cs | 17 +++++++++++++++++ src/SquidStd.Workers/Interfaces/IJobHandler.cs | 15 +++++++++++++++ 2 files changed, 32 insertions(+) create mode 100644 src/SquidStd.Workers/Exceptions/JobHandlerNotFoundException.cs create mode 100644 src/SquidStd.Workers/Interfaces/IJobHandler.cs diff --git a/src/SquidStd.Workers/Exceptions/JobHandlerNotFoundException.cs b/src/SquidStd.Workers/Exceptions/JobHandlerNotFoundException.cs new file mode 100644 index 00000000..61a4d92d --- /dev/null +++ b/src/SquidStd.Workers/Exceptions/JobHandlerNotFoundException.cs @@ -0,0 +1,17 @@ +namespace SquidStd.Workers.Exceptions; + +/// +/// Thrown when a job arrives whose name has no registered . +/// +public sealed class JobHandlerNotFoundException : Exception +{ + /// The job name that had no handler. + public string JobName { get; } + + /// Initializes the exception for the given job name. + public JobHandlerNotFoundException(string jobName) + : base($"No job handler is registered for job '{jobName}'.") + { + JobName = jobName; + } +} diff --git a/src/SquidStd.Workers/Interfaces/IJobHandler.cs b/src/SquidStd.Workers/Interfaces/IJobHandler.cs new file mode 100644 index 00000000..d64a706e --- /dev/null +++ b/src/SquidStd.Workers/Interfaces/IJobHandler.cs @@ -0,0 +1,15 @@ +using SquidStd.Workers.Abstractions.Data; + +namespace SquidStd.Workers.Interfaces; + +/// +/// Handles jobs of a single named kind. Implemented by consumers of the worker library. +/// +public interface IJobHandler +{ + /// The this handler processes. + string JobName { get; } + + /// Executes the job. + Task HandleAsync(JobRequest job, CancellationToken cancellationToken); +} From 2e9d1053415e6724c5d7c5a7f2ed9c1decbb90aa Mon Sep 17 00:00:00 2001 From: Tom Date: Tue, 23 Jun 2026 12:44:17 +0200 Subject: [PATCH 19/33] feat(workers): add JobDispatcher routing jobs to handlers by name --- .../Interfaces/IJobDispatcher.cs | 15 +++++ .../Services/JobDispatcher.cs | 41 ++++++++++++++ .../Workers/JobDispatcherTests.cs | 46 ++++++++++++++++ .../Workers/Support/RecordingJobHandler.cs | 55 +++++++++++++++++++ 4 files changed, 157 insertions(+) create mode 100644 src/SquidStd.Workers/Interfaces/IJobDispatcher.cs create mode 100644 src/SquidStd.Workers/Services/JobDispatcher.cs create mode 100644 tests/SquidStd.Tests/Workers/JobDispatcherTests.cs create mode 100644 tests/SquidStd.Tests/Workers/Support/RecordingJobHandler.cs diff --git a/src/SquidStd.Workers/Interfaces/IJobDispatcher.cs b/src/SquidStd.Workers/Interfaces/IJobDispatcher.cs new file mode 100644 index 00000000..fdae86b5 --- /dev/null +++ b/src/SquidStd.Workers/Interfaces/IJobDispatcher.cs @@ -0,0 +1,15 @@ +using SquidStd.Workers.Abstractions.Data; + +namespace SquidStd.Workers.Interfaces; + +/// +/// Routes a to the registered for its job name. +/// +public interface IJobDispatcher +{ + /// + /// Dispatches the job to its handler. + /// + /// No handler matches the job name. + Task DispatchAsync(JobRequest job, CancellationToken cancellationToken); +} diff --git a/src/SquidStd.Workers/Services/JobDispatcher.cs b/src/SquidStd.Workers/Services/JobDispatcher.cs new file mode 100644 index 00000000..9d481073 --- /dev/null +++ b/src/SquidStd.Workers/Services/JobDispatcher.cs @@ -0,0 +1,41 @@ +using Serilog; +using SquidStd.Workers.Abstractions.Data; +using SquidStd.Workers.Exceptions; +using SquidStd.Workers.Interfaces; + +namespace SquidStd.Workers.Services; + +/// +/// Default : indexes the registered handlers by job name. +/// +public sealed class JobDispatcher : IJobDispatcher +{ + private readonly ILogger _logger = Log.ForContext(); + private readonly Dictionary _handlers = new(StringComparer.Ordinal); + + public JobDispatcher(IEnumerable handlers) + { + foreach (var handler in handlers) + { + if (_handlers.ContainsKey(handler.JobName)) + { + _logger.Warning( + "Duplicate job handler for '{JobName}'; the last registration wins.", + handler.JobName); + } + + _handlers[handler.JobName] = handler; + } + } + + /// + public Task DispatchAsync(JobRequest job, CancellationToken cancellationToken) + { + if (!_handlers.TryGetValue(job.JobName, out var handler)) + { + throw new JobHandlerNotFoundException(job.JobName); + } + + return handler.HandleAsync(job, cancellationToken); + } +} diff --git a/tests/SquidStd.Tests/Workers/JobDispatcherTests.cs b/tests/SquidStd.Tests/Workers/JobDispatcherTests.cs new file mode 100644 index 00000000..6114d4dc --- /dev/null +++ b/tests/SquidStd.Tests/Workers/JobDispatcherTests.cs @@ -0,0 +1,46 @@ +using SquidStd.Tests.Workers.Support; +using SquidStd.Workers.Abstractions.Data; +using SquidStd.Workers.Exceptions; +using SquidStd.Workers.Services; + +namespace SquidStd.Tests.Workers; + +public class JobDispatcherTests +{ + private static JobRequest Job(string name) + => new(name, new Dictionary()); + + [Fact] + public async Task DispatchAsync_InvokesHandlerMatchingJobName() + { + var resize = new RecordingJobHandler("resize"); + var encode = new RecordingJobHandler("encode"); + var dispatcher = new JobDispatcher([resize, encode]); + + await dispatcher.DispatchAsync(Job("encode"), CancellationToken.None); + + Assert.Empty(resize.Received); + Assert.Single(encode.Received); + } + + [Fact] + public async Task DispatchAsync_ThrowsWhenNoHandlerMatches() + { + var dispatcher = new JobDispatcher([new RecordingJobHandler("resize")]); + + var ex = await Assert.ThrowsAsync( + () => dispatcher.DispatchAsync(Job("unknown"), CancellationToken.None)); + + Assert.Equal("unknown", ex.JobName); + } + + [Fact] + public async Task DispatchAsync_PropagatesHandlerException() + { + var boom = new RecordingJobHandler("resize") { ThrowOnHandle = new InvalidOperationException("boom") }; + var dispatcher = new JobDispatcher([boom]); + + await Assert.ThrowsAsync( + () => dispatcher.DispatchAsync(Job("resize"), CancellationToken.None)); + } +} diff --git a/tests/SquidStd.Tests/Workers/Support/RecordingJobHandler.cs b/tests/SquidStd.Tests/Workers/Support/RecordingJobHandler.cs new file mode 100644 index 00000000..75fef96a --- /dev/null +++ b/tests/SquidStd.Tests/Workers/Support/RecordingJobHandler.cs @@ -0,0 +1,55 @@ +using SquidStd.Workers.Abstractions.Data; +using SquidStd.Workers.Interfaces; + +namespace SquidStd.Tests.Workers.Support; + +/// +/// Configurable for tests: records the jobs it received and can be told +/// to throw, or to block until released, so concurrency and error paths can be exercised. +/// +public sealed class RecordingJobHandler : IJobHandler +{ + private readonly List _received = []; + private readonly Lock _sync = new(); + + public RecordingJobHandler(string jobName) + { + JobName = jobName; + } + + public string JobName { get; } + + public Exception? ThrowOnHandle { get; set; } + + /// When set, awaits this before returning. + public TaskCompletionSource? Gate { get; set; } + + public IReadOnlyList Received + { + get + { + lock (_sync) + { + return _received.ToArray(); + } + } + } + + public async Task HandleAsync(JobRequest job, CancellationToken cancellationToken) + { + lock (_sync) + { + _received.Add(job); + } + + if (ThrowOnHandle is not null) + { + throw ThrowOnHandle; + } + + if (Gate is not null) + { + await Gate.Task.WaitAsync(cancellationToken); + } + } +} From a3eff821fd88dfa1079863ac7108fc15f891d619 Mon Sep 17 00:00:00 2001 From: Tom Date: Tue, 23 Jun 2026 12:44:57 +0200 Subject: [PATCH 20/33] feat(workers): add WorkerState shared runtime state --- .../Interfaces/IWorkerState.cs | 27 ++++++++++ src/SquidStd.Workers/Services/WorkerState.cs | 40 +++++++++++++++ .../Workers/WorkerStateTests.cs | 51 +++++++++++++++++++ 3 files changed, 118 insertions(+) create mode 100644 src/SquidStd.Workers/Interfaces/IWorkerState.cs create mode 100644 src/SquidStd.Workers/Services/WorkerState.cs create mode 100644 tests/SquidStd.Tests/Workers/WorkerStateTests.cs diff --git a/src/SquidStd.Workers/Interfaces/IWorkerState.cs b/src/SquidStd.Workers/Interfaces/IWorkerState.cs new file mode 100644 index 00000000..685e510c --- /dev/null +++ b/src/SquidStd.Workers/Interfaces/IWorkerState.cs @@ -0,0 +1,27 @@ +using SquidStd.Workers.Abstractions.Types; + +namespace SquidStd.Workers.Interfaces; + +/// +/// Shared runtime state of a worker, read by the heartbeat service and mutated by the consumer. +/// +public interface IWorkerState +{ + /// Stable worker identity. + string WorkerId { get; } + + /// Maximum jobs processed in parallel. + int MaxConcurrency { get; } + + /// Jobs currently in progress. + int ActiveJobs { get; } + + /// while any job is active, otherwise . + WorkerStatusType Status { get; } + + /// Records that a job started (increments ). + void JobStarted(); + + /// Records that a job finished (decrements ). + void JobFinished(); +} diff --git a/src/SquidStd.Workers/Services/WorkerState.cs b/src/SquidStd.Workers/Services/WorkerState.cs new file mode 100644 index 00000000..8b96046b --- /dev/null +++ b/src/SquidStd.Workers/Services/WorkerState.cs @@ -0,0 +1,40 @@ +using SquidStd.Workers.Abstractions.Types; +using SquidStd.Workers.Data.Config; +using SquidStd.Workers.Interfaces; + +namespace SquidStd.Workers.Services; + +/// +/// Default backed by an interlocked counter; resolves identity and +/// concurrency from at construction. +/// +public sealed class WorkerState : IWorkerState +{ + private int _activeJobs; + + /// + public string WorkerId { get; } + + /// + public int MaxConcurrency { get; } + + public WorkerState(WorkersConfig config) + { + WorkerId = string.IsNullOrWhiteSpace(config.WorkerId) ? Environment.MachineName : config.WorkerId; + MaxConcurrency = config.MaxConcurrency > 0 ? config.MaxConcurrency : Environment.ProcessorCount; + } + + /// + public int ActiveJobs => Volatile.Read(ref _activeJobs); + + /// + public WorkerStatusType Status => ActiveJobs == 0 ? WorkerStatusType.Idle : WorkerStatusType.Busy; + + /// + public void JobStarted() + => Interlocked.Increment(ref _activeJobs); + + /// + public void JobFinished() + => Interlocked.Decrement(ref _activeJobs); +} diff --git a/tests/SquidStd.Tests/Workers/WorkerStateTests.cs b/tests/SquidStd.Tests/Workers/WorkerStateTests.cs new file mode 100644 index 00000000..8e8b7942 --- /dev/null +++ b/tests/SquidStd.Tests/Workers/WorkerStateTests.cs @@ -0,0 +1,51 @@ +using SquidStd.Workers.Data.Config; +using SquidStd.Workers.Services; +using SquidStd.Workers.Abstractions.Types; + +namespace SquidStd.Tests.Workers; + +public class WorkerStateTests +{ + [Fact] + public void Status_IsIdleWhenNoActiveJobs() + { + var state = new WorkerState(new WorkersConfig { WorkerId = "w1", MaxConcurrency = 4 }); + + Assert.Equal(WorkerStatusType.Idle, state.Status); + Assert.Equal(0, state.ActiveJobs); + } + + [Fact] + public void Status_IsBusyWhileJobsActive() + { + var state = new WorkerState(new WorkersConfig { WorkerId = "w1", MaxConcurrency = 4 }); + + state.JobStarted(); + state.JobStarted(); + + Assert.Equal(WorkerStatusType.Busy, state.Status); + Assert.Equal(2, state.ActiveJobs); + + state.JobFinished(); + state.JobFinished(); + + Assert.Equal(WorkerStatusType.Idle, state.Status); + Assert.Equal(0, state.ActiveJobs); + } + + [Fact] + public void WorkerId_FallsBackToMachineNameWhenBlank() + { + var state = new WorkerState(new WorkersConfig { WorkerId = " ", MaxConcurrency = 4 }); + + Assert.Equal(Environment.MachineName, state.WorkerId); + } + + [Fact] + public void MaxConcurrency_FallsBackToProcessorCountWhenNotPositive() + { + var state = new WorkerState(new WorkersConfig { WorkerId = "w1", MaxConcurrency = 0 }); + + Assert.Equal(Environment.ProcessorCount, state.MaxConcurrency); + } +} From 420f685422e129b5c3aaad36388fdbfea9ab8699 Mon Sep 17 00:00:00 2001 From: Tom Date: Tue, 23 Jun 2026 12:45:48 +0200 Subject: [PATCH 21/33] feat(workers): add WorkerConsumerService consuming and dispatching jobs --- .../Services/WorkerConsumerService.cs | 86 +++++++++++++++++++ .../Workers/Support/FakeMessageQueue.cs | 23 +++++ .../Workers/WorkerConsumerServiceTests.cs | 76 ++++++++++++++++ 3 files changed, 185 insertions(+) create mode 100644 src/SquidStd.Workers/Services/WorkerConsumerService.cs create mode 100644 tests/SquidStd.Tests/Workers/Support/FakeMessageQueue.cs create mode 100644 tests/SquidStd.Tests/Workers/WorkerConsumerServiceTests.cs diff --git a/src/SquidStd.Workers/Services/WorkerConsumerService.cs b/src/SquidStd.Workers/Services/WorkerConsumerService.cs new file mode 100644 index 00000000..e598cde5 --- /dev/null +++ b/src/SquidStd.Workers/Services/WorkerConsumerService.cs @@ -0,0 +1,86 @@ +using Serilog; +using SquidStd.Abstractions.Interfaces.Services; +using SquidStd.Messaging.Abstractions.Interfaces; +using SquidStd.Workers.Abstractions; +using SquidStd.Workers.Abstractions.Data; +using SquidStd.Workers.Data.Config; +using SquidStd.Workers.Exceptions; +using SquidStd.Workers.Interfaces; + +namespace SquidStd.Workers.Services; + +/// +/// Subscribes to the jobs queue and dispatches each to its handler, +/// bounded by . +/// +public sealed class WorkerConsumerService : ISquidStdService, IQueueMessageListenerAsync +{ + private readonly ILogger _logger = Log.ForContext(); + private readonly IMessageQueue _queue; + private readonly IJobDispatcher _dispatcher; + private readonly IWorkerState _state; + private readonly SemaphoreSlim _semaphore; + private readonly string _queueName; + private IDisposable? _subscription; + + /// Constructs the consumer wired to the messaging queue and worker state. + public WorkerConsumerService(IMessageQueue queue, IJobDispatcher dispatcher, IWorkerState state, WorkersConfig config) + { + _queue = queue; + _dispatcher = dispatcher; + _state = state; + _semaphore = new SemaphoreSlim(state.MaxConcurrency); + _queueName = string.IsNullOrWhiteSpace(config.JobQueue) ? WorkerChannels.JobQueue : config.JobQueue; + } + + /// + public ValueTask StartAsync(CancellationToken cancellationToken = default) + { + _subscription = _queue.Subscribe(_queueName, this); + _logger.Information("Worker consuming jobs from queue '{Queue}'.", _queueName); + + return ValueTask.CompletedTask; + } + + /// + public ValueTask StopAsync(CancellationToken cancellationToken = default) + { + _subscription?.Dispose(); + _subscription = null; + + return ValueTask.CompletedTask; + } + + /// + public async Task HandleAsync(JobRequest message, CancellationToken cancellationToken) + { + await _semaphore.WaitAsync(cancellationToken); + _state.JobStarted(); + + try + { + await _dispatcher.DispatchAsync(message, cancellationToken); + } + catch (JobHandlerNotFoundException) + { + _logger.Warning("No handler for job '{JobName}'; dropping.", message.JobName); + } + catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested) + { + _logger.Debug("Job '{JobName}' cancelled during shutdown; will be redelivered.", message.JobName); + + throw; + } + catch (Exception ex) + { + _logger.Error(ex, "Job '{JobName}' failed; it will be retried.", message.JobName); + + throw; + } + finally + { + _state.JobFinished(); + _semaphore.Release(); + } + } +} diff --git a/tests/SquidStd.Tests/Workers/Support/FakeMessageQueue.cs b/tests/SquidStd.Tests/Workers/Support/FakeMessageQueue.cs new file mode 100644 index 00000000..1a1812d2 --- /dev/null +++ b/tests/SquidStd.Tests/Workers/Support/FakeMessageQueue.cs @@ -0,0 +1,23 @@ +using SquidStd.Messaging.Abstractions.Interfaces; + +namespace SquidStd.Tests.Workers.Support; + +/// Inert for unit tests that drive the consumer's HandleAsync directly. +public sealed class FakeMessageQueue : IMessageQueue +{ + public Task PublishAsync(string queueName, TMessage message, CancellationToken cancellationToken = default) + => Task.CompletedTask; + + public IDisposable Subscribe(string queueName, IQueueMessageListener listener) + => new Subscription(); + + public IDisposable Subscribe(string queueName, IQueueMessageListenerAsync listener) + => new Subscription(); + + private sealed class Subscription : IDisposable + { + public void Dispose() + { + } + } +} diff --git a/tests/SquidStd.Tests/Workers/WorkerConsumerServiceTests.cs b/tests/SquidStd.Tests/Workers/WorkerConsumerServiceTests.cs new file mode 100644 index 00000000..7f04c1e3 --- /dev/null +++ b/tests/SquidStd.Tests/Workers/WorkerConsumerServiceTests.cs @@ -0,0 +1,76 @@ +using SquidStd.Tests.Workers.Support; +using SquidStd.Workers.Abstractions.Data; +using SquidStd.Workers.Data.Config; +using SquidStd.Workers.Services; + +namespace SquidStd.Tests.Workers; + +public class WorkerConsumerServiceTests +{ + private static JobRequest Job(string name) + => new(name, new Dictionary()); + + private static WorkerConsumerService Build(WorkerState state, params RecordingJobHandler[] handlers) + => new(new FakeMessageQueue(), new JobDispatcher(handlers), state, new WorkersConfig()); + + [Fact] + public async Task HandleAsync_RunsMatchingHandler() + { + var handler = new RecordingJobHandler("resize"); + var state = new WorkerState(new() { WorkerId = "w1", MaxConcurrency = 2 }); + var consumer = Build(state, handler); + + await consumer.HandleAsync(Job("resize"), CancellationToken.None); + + Assert.Single(handler.Received); + Assert.Equal(0, state.ActiveJobs); + } + + [Fact] + public async Task HandleAsync_DropsUnknownJobWithoutThrowing() + { + var state = new WorkerState(new() { WorkerId = "w1", MaxConcurrency = 2 }); + var consumer = Build(state, new RecordingJobHandler("resize")); + + var ex = await Record.ExceptionAsync(() => consumer.HandleAsync(Job("unknown"), CancellationToken.None)); + + Assert.Null(ex); + Assert.Equal(0, state.ActiveJobs); + } + + [Fact] + public async Task HandleAsync_RethrowsHandlerExceptionForRequeue() + { + var handler = new RecordingJobHandler("resize") { ThrowOnHandle = new InvalidOperationException("boom") }; + var state = new WorkerState(new() { WorkerId = "w1", MaxConcurrency = 2 }); + var consumer = Build(state, handler); + + await Assert.ThrowsAsync( + () => consumer.HandleAsync(Job("resize"), CancellationToken.None)); + + Assert.Equal(0, state.ActiveJobs); + } + + [Fact] + public async Task HandleAsync_NeverExceedsMaxConcurrency() + { + var handler = new RecordingJobHandler("resize") { Gate = new TaskCompletionSource() }; + var state = new WorkerState(new() { WorkerId = "w1", MaxConcurrency = 2 }); + var consumer = Build(state, handler); + + // Launch 5 concurrent dispatches against a handler that blocks on its gate. + var inFlight = Enumerable.Range(0, 5) + .Select(_ => consumer.HandleAsync(Job("resize"), CancellationToken.None)) + .ToArray(); + + // Give the semaphore time to admit as many as it will, then assert the cap held. + await Task.Delay(200); + Assert.True(state.ActiveJobs <= 2, $"ActiveJobs was {state.ActiveJobs}, expected <= 2"); + + // Release the gate so everything drains. + handler.Gate!.SetResult(); + await Task.WhenAll(inFlight); + Assert.Equal(0, state.ActiveJobs); + Assert.Equal(5, handler.Received.Count); + } +} From d47ec8597208e866f3b90e7a573761ab704ca764 Mon Sep 17 00:00:00 2001 From: Tom Date: Tue, 23 Jun 2026 12:46:35 +0200 Subject: [PATCH 22/33] feat(workers): add WorkerHeartbeatService publishing periodic heartbeats --- .../Services/WorkerHeartbeatService.cs | 117 ++++++++++++++++++ .../Workers/WorkerHeartbeatServiceTests.cs | 69 +++++++++++ 2 files changed, 186 insertions(+) create mode 100644 src/SquidStd.Workers/Services/WorkerHeartbeatService.cs create mode 100644 tests/SquidStd.Tests/Workers/WorkerHeartbeatServiceTests.cs diff --git a/src/SquidStd.Workers/Services/WorkerHeartbeatService.cs b/src/SquidStd.Workers/Services/WorkerHeartbeatService.cs new file mode 100644 index 00000000..c5fa548a --- /dev/null +++ b/src/SquidStd.Workers/Services/WorkerHeartbeatService.cs @@ -0,0 +1,117 @@ +using Serilog; +using SquidStd.Abstractions.Interfaces.Services; +using SquidStd.Messaging.Abstractions.Interfaces; +using SquidStd.Workers.Abstractions; +using SquidStd.Workers.Abstractions.Data; +using SquidStd.Workers.Data.Config; +using SquidStd.Workers.Interfaces; + +namespace SquidStd.Workers.Services; + +/// +/// Publishes a on the heartbeat topic immediately on start and then once +/// per configured interval. +/// +public sealed class WorkerHeartbeatService : ISquidStdService +{ + private const int DefaultIntervalSeconds = 10; + + private readonly ILogger _logger = Log.ForContext(); + private readonly IMessageTopic _topic; + private readonly IWorkerState _state; + private readonly TimeSpan _interval; + private readonly string _topicName; + private CancellationTokenSource? _loopCts; + private Task? _loopTask; + + public WorkerHeartbeatService(IMessageTopic topic, IWorkerState state, WorkersConfig config) + { + _topic = topic; + _state = state; + + var seconds = config.HeartbeatIntervalSeconds > 0 ? config.HeartbeatIntervalSeconds : DefaultIntervalSeconds; + if (config.HeartbeatIntervalSeconds <= 0) + { + _logger.Warning( + "HeartbeatIntervalSeconds was {Value}; falling back to {Default}s.", + config.HeartbeatIntervalSeconds, + DefaultIntervalSeconds); + } + + _interval = TimeSpan.FromSeconds(seconds); + _topicName = string.IsNullOrWhiteSpace(config.HeartbeatTopic) ? WorkerChannels.HeartbeatTopic : config.HeartbeatTopic; + } + + /// + public ValueTask StartAsync(CancellationToken cancellationToken = default) + { + _loopCts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken); + _loopTask = Task.Run(() => RunLoopAsync(_loopCts.Token), CancellationToken.None); + + return ValueTask.CompletedTask; + } + + /// + public async ValueTask StopAsync(CancellationToken cancellationToken = default) + { + if (_loopCts is null) + { + return; + } + + await _loopCts.CancelAsync(); + + try + { + if (_loopTask is not null) + { + await _loopTask; + } + } + catch (OperationCanceledException) + { + // Expected on shutdown. + } + finally + { + _loopCts.Dispose(); + _loopCts = null; + _loopTask = null; + } + } + + private async Task RunLoopAsync(CancellationToken cancellationToken) + { + using var timer = new PeriodicTimer(_interval); + + await PublishAsync(cancellationToken); + + while (await timer.WaitForNextTickAsync(cancellationToken)) + { + await PublishAsync(cancellationToken); + } + } + + private async Task PublishAsync(CancellationToken cancellationToken) + { + try + { + var heartbeat = new WorkerHeartbeat( + _state.WorkerId, + DateTime.UtcNow, + _state.Status, + _state.ActiveJobs, + _state.MaxConcurrency); + + await _topic.PublishAsync(_topicName, heartbeat, cancellationToken); + } + catch (OperationCanceledException) + { + throw; + } + catch (Exception ex) + { + _logger.Error(ex, "Failed to publish heartbeat; will retry on the next tick."); + } + } +} diff --git a/tests/SquidStd.Tests/Workers/WorkerHeartbeatServiceTests.cs b/tests/SquidStd.Tests/Workers/WorkerHeartbeatServiceTests.cs new file mode 100644 index 00000000..6efcf659 --- /dev/null +++ b/tests/SquidStd.Tests/Workers/WorkerHeartbeatServiceTests.cs @@ -0,0 +1,69 @@ +using DryIoc; +using SquidStd.Core.Interfaces.Events; +using SquidStd.Messaging.Abstractions.Interfaces; +using SquidStd.Messaging.Extensions; +using SquidStd.Services.Core.Services; +using SquidStd.Workers.Abstractions; +using SquidStd.Workers.Abstractions.Data; +using SquidStd.Workers.Abstractions.Types; +using SquidStd.Workers.Data.Config; +using SquidStd.Workers.Services; + +namespace SquidStd.Tests.Workers; + +public class WorkerHeartbeatServiceTests +{ + private static (IMessageTopic topic, WorkerState state) NewMessaging(WorkersConfig config) + { + var container = new Container(); + container.RegisterInstance(new EventBusService()); + container.AddInMemoryMessaging(); + + return (container.Resolve(), new WorkerState(config)); + } + + [Fact] + public async Task StartAsync_PublishesHeartbeatImmediately() + { + var config = new WorkersConfig { WorkerId = "w1", MaxConcurrency = 8, HeartbeatIntervalSeconds = 60 }; + var (topic, state) = NewMessaging(config); + + var received = new TaskCompletionSource(); + using var _ = topic.Subscribe( + WorkerChannels.HeartbeatTopic, + (hb, _) => { received.TrySetResult(hb); return Task.CompletedTask; }); + + var service = new WorkerHeartbeatService(topic, state, config); + await service.StartAsync(); + + var heartbeat = await received.Task.WaitAsync(TimeSpan.FromSeconds(5)); + await service.StopAsync(); + + Assert.Equal("w1", heartbeat.WorkerId); + Assert.Equal(8, heartbeat.MaxConcurrency); + Assert.Equal(0, heartbeat.ActiveJobs); + Assert.Equal(WorkerStatusType.Idle, heartbeat.Status); + } + + [Fact] + public async Task Heartbeat_ReflectsBusyStateWithActiveJobs() + { + var config = new WorkersConfig { WorkerId = "w1", MaxConcurrency = 8, HeartbeatIntervalSeconds = 60 }; + var (topic, state) = NewMessaging(config); + state.JobStarted(); + + var received = new TaskCompletionSource(); + using var _ = topic.Subscribe( + WorkerChannels.HeartbeatTopic, + (hb, _) => { received.TrySetResult(hb); return Task.CompletedTask; }); + + var service = new WorkerHeartbeatService(topic, state, config); + await service.StartAsync(); + + var heartbeat = await received.Task.WaitAsync(TimeSpan.FromSeconds(5)); + await service.StopAsync(); + + Assert.Equal(1, heartbeat.ActiveJobs); + Assert.Equal(WorkerStatusType.Busy, heartbeat.Status); + } +} From fbb3033e482529b6fb9e19832c7bfdfa3ef4e939 Mon Sep 17 00:00:00 2001 From: Tom Date: Tue, 23 Jun 2026 12:47:14 +0200 Subject: [PATCH 23/33] feat(workers): add AddWorkers and AddJobHandler registration extensions --- .../WorkersRegistrationExtensions.cs | 46 +++++++++++++ .../WorkersRegistrationExtensionsTests.cs | 65 +++++++++++++++++++ 2 files changed, 111 insertions(+) create mode 100644 src/SquidStd.Workers/Extensions/WorkersRegistrationExtensions.cs create mode 100644 tests/SquidStd.Tests/Workers/WorkersRegistrationExtensionsTests.cs diff --git a/src/SquidStd.Workers/Extensions/WorkersRegistrationExtensions.cs b/src/SquidStd.Workers/Extensions/WorkersRegistrationExtensions.cs new file mode 100644 index 00000000..18eb37ba --- /dev/null +++ b/src/SquidStd.Workers/Extensions/WorkersRegistrationExtensions.cs @@ -0,0 +1,46 @@ +using DryIoc; +using SquidStd.Abstractions.Extensions.Config; +using SquidStd.Abstractions.Extensions.Services; +using SquidStd.Workers.Data.Config; +using SquidStd.Workers.Interfaces; +using SquidStd.Workers.Services; + +namespace SquidStd.Workers.Extensions; + +/// +/// DryIoc registration helpers for the worker runtime. +/// +public static class WorkersRegistrationExtensions +{ + /// + /// Registers the worker runtime: the "workers" config section, shared state, job dispatcher, and the + /// consumer + heartbeat lifecycle services. + /// + public static IContainer AddWorkers(this IContainer container) + { + ArgumentNullException.ThrowIfNull(container); + + container.RegisterConfigSection("workers", static () => new WorkersConfig(), -50); + + container.Register(Reuse.Singleton); + container.Register(Reuse.Singleton); + + container.RegisterStdService(100); + container.RegisterStdService(110); + + return container; + } + + /// + /// Registers a job handler so the dispatcher can route jobs to it by name. + /// + public static IContainer AddJobHandler(this IContainer container) + where THandler : class, IJobHandler + { + ArgumentNullException.ThrowIfNull(container); + + container.Register(Reuse.Singleton); + + return container; + } +} diff --git a/tests/SquidStd.Tests/Workers/WorkersRegistrationExtensionsTests.cs b/tests/SquidStd.Tests/Workers/WorkersRegistrationExtensionsTests.cs new file mode 100644 index 00000000..5d4b73b4 --- /dev/null +++ b/tests/SquidStd.Tests/Workers/WorkersRegistrationExtensionsTests.cs @@ -0,0 +1,65 @@ +using DryIoc; +using SquidStd.Core.Interfaces.Events; +using SquidStd.Messaging.Extensions; +using SquidStd.Services.Core.Services; +using SquidStd.Workers.Data.Config; +using SquidStd.Workers.Extensions; +using SquidStd.Workers.Interfaces; +using SquidStd.Workers.Services; + +namespace SquidStd.Tests.Workers; + +public class WorkersRegistrationExtensionsTests +{ + private static Container NewContainer() + { + var container = new Container(); + container.RegisterInstance(new EventBusService()); + container.AddInMemoryMessaging(); + // ConfigManager normally registers config instances; register one directly for the test. + container.RegisterInstance(new WorkersConfig { WorkerId = "w1", MaxConcurrency = 4 }); + + return container; + } + + [Fact] + public void AddWorkers_RegistersResolvableServices() + { + using var container = NewContainer(); + + container.AddWorkers(); + + Assert.NotNull(container.Resolve()); + Assert.NotNull(container.Resolve()); + Assert.NotNull(container.Resolve()); + Assert.NotNull(container.Resolve()); + } + + [Fact] + public async Task AddJobHandler_MakesHandlerReachableFromDispatcher() + { + using var container = NewContainer(); + + container.AddWorkers(); + container.AddJobHandler(); + + var dispatcher = container.Resolve(); + await dispatcher.DispatchAsync(new("echo", new Dictionary()), CancellationToken.None); + + Assert.Equal(1, EchoJobHandler.Calls); + } + + private sealed class EchoJobHandler : IJobHandler + { + public static int Calls; + + public string JobName => "echo"; + + public Task HandleAsync(SquidStd.Workers.Abstractions.Data.JobRequest job, CancellationToken cancellationToken) + { + Interlocked.Increment(ref Calls); + + return Task.CompletedTask; + } + } +} From 1b6d1f7fd4b3353fd80ac706f093eb87f8e8278d Mon Sep 17 00:00:00 2001 From: Tom Date: Tue, 23 Jun 2026 14:24:17 +0200 Subject: [PATCH 24/33] build(workers-manager): scaffold SquidStd.Workers.Manager project --- SquidStd.slnx | 1 + .../SquidStd.Workers.Manager.csproj | 20 +++++++++++++++++++ tests/SquidStd.Tests/SquidStd.Tests.csproj | 1 + 3 files changed, 22 insertions(+) create mode 100644 src/SquidStd.Workers.Manager/SquidStd.Workers.Manager.csproj diff --git a/SquidStd.slnx b/SquidStd.slnx index 77846e19..6d73a395 100644 --- a/SquidStd.slnx +++ b/SquidStd.slnx @@ -21,6 +21,7 @@ + diff --git a/src/SquidStd.Workers.Manager/SquidStd.Workers.Manager.csproj b/src/SquidStd.Workers.Manager/SquidStd.Workers.Manager.csproj new file mode 100644 index 00000000..6b1b3dcc --- /dev/null +++ b/src/SquidStd.Workers.Manager/SquidStd.Workers.Manager.csproj @@ -0,0 +1,20 @@ + + + + net10.0 + enable + enable + true + + + + + + + + + + + + + diff --git a/tests/SquidStd.Tests/SquidStd.Tests.csproj b/tests/SquidStd.Tests/SquidStd.Tests.csproj index 73dd9959..7e8df98e 100644 --- a/tests/SquidStd.Tests/SquidStd.Tests.csproj +++ b/tests/SquidStd.Tests/SquidStd.Tests.csproj @@ -51,6 +51,7 @@ + From 12874cde4069dd4c8465a9dc6a05b75a7138c3a1 Mon Sep 17 00:00:00 2001 From: Tom Date: Tue, 23 Jun 2026 14:24:54 +0200 Subject: [PATCH 25/33] feat(workers-manager): add config, status-change event and enqueue request types --- .../Data/Config/WorkerManagerConfig.cs | 21 +++++++++++++++++++ .../Data/EnqueueJobRequest.cs | 8 +++++++ .../Data/Events/WorkerStatusChangedEvent.cs | 13 ++++++++++++ 3 files changed, 42 insertions(+) create mode 100644 src/SquidStd.Workers.Manager/Data/Config/WorkerManagerConfig.cs create mode 100644 src/SquidStd.Workers.Manager/Data/EnqueueJobRequest.cs create mode 100644 src/SquidStd.Workers.Manager/Data/Events/WorkerStatusChangedEvent.cs diff --git a/src/SquidStd.Workers.Manager/Data/Config/WorkerManagerConfig.cs b/src/SquidStd.Workers.Manager/Data/Config/WorkerManagerConfig.cs new file mode 100644 index 00000000..beef18cd --- /dev/null +++ b/src/SquidStd.Workers.Manager/Data/Config/WorkerManagerConfig.cs @@ -0,0 +1,21 @@ +using SquidStd.Workers.Abstractions; + +namespace SquidStd.Workers.Manager.Data.Config; + +/// +/// Configuration for the worker manager (config section "workerManager"). +/// +public sealed class WorkerManagerConfig +{ + /// Seconds without a heartbeat before a worker is considered Offline. Falls back to 30 when not positive. + public int OfflineTimeoutSeconds { get; set; } = 30; + + /// Seconds between offline sweeps. Falls back to 10 when not positive. + public int SweepIntervalSeconds { get; set; } = 10; + + /// Queue jobs are enqueued onto. + public string JobQueue { get; set; } = WorkerChannels.JobQueue; + + /// Topic heartbeats are collected from. + public string HeartbeatTopic { get; set; } = WorkerChannels.HeartbeatTopic; +} diff --git a/src/SquidStd.Workers.Manager/Data/EnqueueJobRequest.cs b/src/SquidStd.Workers.Manager/Data/EnqueueJobRequest.cs new file mode 100644 index 00000000..8869f5f7 --- /dev/null +++ b/src/SquidStd.Workers.Manager/Data/EnqueueJobRequest.cs @@ -0,0 +1,8 @@ +namespace SquidStd.Workers.Manager.Data; + +/// +/// Request body for enqueuing a job via the manager's HTTP surface. +/// +/// Logical name of the job. +/// Optional string key/value arguments; treated as empty when null. +public sealed record EnqueueJobRequest(string JobName, IReadOnlyDictionary? Parameters); diff --git a/src/SquidStd.Workers.Manager/Data/Events/WorkerStatusChangedEvent.cs b/src/SquidStd.Workers.Manager/Data/Events/WorkerStatusChangedEvent.cs new file mode 100644 index 00000000..9b15b767 --- /dev/null +++ b/src/SquidStd.Workers.Manager/Data/Events/WorkerStatusChangedEvent.cs @@ -0,0 +1,13 @@ +using SquidStd.Core.Interfaces.Events; +using SquidStd.Workers.Abstractions.Types; + +namespace SquidStd.Workers.Manager.Data.Events; + +/// +/// Published on the event bus when a worker's status changes: discovered ( null), +/// gone Offline (via the sweep), or returned (Offline → Idle/Busy). +/// +/// The worker whose status changed. +/// Previous status, or null when the worker was just discovered. +/// The new status. +public sealed record WorkerStatusChangedEvent(string WorkerId, WorkerStatusType? OldStatus, WorkerStatusType NewStatus) : IEvent; From 7d73cb944c3620ea99634d538d29d4531831e35f Mon Sep 17 00:00:00 2001 From: Tom Date: Tue, 23 Jun 2026 14:25:57 +0200 Subject: [PATCH 26/33] feat(workers-manager): add WorkerRegistry folding heartbeats into worker state --- .../Interfaces/IWorkerRegistry.cs | 15 +++ .../Services/WorkerRegistry.cs | 105 ++++++++++++++++++ .../Manager/WorkerRegistryTests.cs | 97 ++++++++++++++++ 3 files changed, 217 insertions(+) create mode 100644 src/SquidStd.Workers.Manager/Interfaces/IWorkerRegistry.cs create mode 100644 src/SquidStd.Workers.Manager/Services/WorkerRegistry.cs create mode 100644 tests/SquidStd.Tests/Manager/WorkerRegistryTests.cs diff --git a/src/SquidStd.Workers.Manager/Interfaces/IWorkerRegistry.cs b/src/SquidStd.Workers.Manager/Interfaces/IWorkerRegistry.cs new file mode 100644 index 00000000..1859d6ca --- /dev/null +++ b/src/SquidStd.Workers.Manager/Interfaces/IWorkerRegistry.cs @@ -0,0 +1,15 @@ +using SquidStd.Workers.Abstractions.Data; + +namespace SquidStd.Workers.Manager.Interfaces; + +/// +/// Read access to the manager's in-memory view of known workers. +/// +public interface IWorkerRegistry +{ + /// Returns a snapshot of all known workers. + IReadOnlyCollection GetAll(); + + /// Returns the worker with the given id, or null when unknown. + WorkerInfo? Get(string workerId); +} diff --git a/src/SquidStd.Workers.Manager/Services/WorkerRegistry.cs b/src/SquidStd.Workers.Manager/Services/WorkerRegistry.cs new file mode 100644 index 00000000..d7804ae8 --- /dev/null +++ b/src/SquidStd.Workers.Manager/Services/WorkerRegistry.cs @@ -0,0 +1,105 @@ +using SquidStd.Workers.Abstractions.Data; +using SquidStd.Workers.Abstractions.Types; +using SquidStd.Workers.Manager.Data.Config; +using SquidStd.Workers.Manager.Data.Events; +using SquidStd.Workers.Manager.Interfaces; + +namespace SquidStd.Workers.Manager.Services; + +/// +/// In-memory registry of workers, folded from heartbeats. Pure: it returns status transitions for the +/// caller to publish, and never touches the event bus or a real clock (the sweep takes the time as a parameter). +/// +public sealed class WorkerRegistry : IWorkerRegistry +{ + private readonly Lock _sync = new(); + private readonly Dictionary _workers = new(StringComparer.Ordinal); + private readonly TimeSpan _offlineTimeout; + + public WorkerRegistry(WorkerManagerConfig config) + { + var seconds = config.OfflineTimeoutSeconds > 0 ? config.OfflineTimeoutSeconds : 30; + _offlineTimeout = TimeSpan.FromSeconds(seconds); + } + + /// + public IReadOnlyCollection GetAll() + { + lock (_sync) + { + return _workers.Values.ToArray(); + } + } + + /// + public WorkerInfo? Get(string workerId) + { + lock (_sync) + { + return _workers.GetValueOrDefault(workerId); + } + } + + /// + /// Folds a heartbeat into the registry. Returns a transition only when the worker is newly discovered + /// or has returned from ; otherwise null. + /// + public WorkerStatusChangedEvent? Record(WorkerHeartbeat heartbeat) + { + var now = DateTime.UtcNow; + + lock (_sync) + { + if (!_workers.TryGetValue(heartbeat.WorkerId, out var existing)) + { + _workers[heartbeat.WorkerId] = new WorkerInfo( + heartbeat.WorkerId, + heartbeat.Status, + heartbeat.ActiveJobs, + heartbeat.MaxConcurrency, + now, + now); + + return new WorkerStatusChangedEvent(heartbeat.WorkerId, null, heartbeat.Status); + } + + var old = existing.Status; + _workers[heartbeat.WorkerId] = existing with + { + Status = heartbeat.Status, + ActiveJobs = heartbeat.ActiveJobs, + MaxConcurrency = heartbeat.MaxConcurrency, + LastSeenUtc = now + }; + + return old == WorkerStatusType.Offline && heartbeat.Status != WorkerStatusType.Offline + ? new WorkerStatusChangedEvent(heartbeat.WorkerId, old, heartbeat.Status) + : null; + } + } + + /// + /// Marks workers Offline whose last heartbeat is older than the configured timeout relative to + /// . Returns the resulting transitions. + /// + public IReadOnlyList Sweep(DateTime nowUtc) + { + var changes = new List(); + + lock (_sync) + { + foreach (var (id, info) in _workers.ToArray()) + { + if (info.Status == WorkerStatusType.Offline || nowUtc - info.LastSeenUtc <= _offlineTimeout) + { + continue; + } + + _workers[id] = info with { Status = WorkerStatusType.Offline }; + changes.Add(new WorkerStatusChangedEvent(id, info.Status, WorkerStatusType.Offline)); + } + } + + return changes; + } +} diff --git a/tests/SquidStd.Tests/Manager/WorkerRegistryTests.cs b/tests/SquidStd.Tests/Manager/WorkerRegistryTests.cs new file mode 100644 index 00000000..1cf24a7d --- /dev/null +++ b/tests/SquidStd.Tests/Manager/WorkerRegistryTests.cs @@ -0,0 +1,97 @@ +using SquidStd.Workers.Abstractions.Data; +using SquidStd.Workers.Abstractions.Types; +using SquidStd.Workers.Manager.Data.Config; +using SquidStd.Workers.Manager.Services; + +namespace SquidStd.Tests.Manager; + +public class WorkerRegistryTests +{ + private static WorkerRegistry NewRegistry(int offlineTimeoutSeconds = 30) + => new(new WorkerManagerConfig { OfflineTimeoutSeconds = offlineTimeoutSeconds }); + + private static WorkerHeartbeat Heartbeat(string id, WorkerStatusType status, int activeJobs = 0) + => new(id, DateTime.UtcNow, status, activeJobs, 8); + + [Fact] + public void Record_NewWorker_ReturnsDiscoveredTransitionAndStores() + { + var registry = NewRegistry(); + + var change = registry.Record(Heartbeat("w1", WorkerStatusType.Idle)); + + Assert.NotNull(change); + Assert.Equal("w1", change!.WorkerId); + Assert.Null(change.OldStatus); + Assert.Equal(WorkerStatusType.Idle, change.NewStatus); + + var info = registry.Get("w1"); + Assert.NotNull(info); + Assert.Equal(WorkerStatusType.Idle, info!.Status); + Assert.Equal(8, info.MaxConcurrency); + } + + [Fact] + public void Record_ExistingWorker_IdleToBusy_NoTransitionButStateUpdated() + { + var registry = NewRegistry(); + registry.Record(Heartbeat("w1", WorkerStatusType.Idle)); + + var change = registry.Record(Heartbeat("w1", WorkerStatusType.Busy, activeJobs: 2)); + + Assert.Null(change); + Assert.Equal(WorkerStatusType.Busy, registry.Get("w1")!.Status); + Assert.Equal(2, registry.Get("w1")!.ActiveJobs); + } + + [Fact] + public void Sweep_MarksOverdueWorkerOffline_AndReturnsTransition() + { + var registry = NewRegistry(offlineTimeoutSeconds: 30); + registry.Record(Heartbeat("w1", WorkerStatusType.Idle)); + + var changes = registry.Sweep(DateTime.UtcNow.AddSeconds(31)); + + var change = Assert.Single(changes); + Assert.Equal("w1", change.WorkerId); + Assert.Equal(WorkerStatusType.Offline, change.NewStatus); + Assert.Equal(WorkerStatusType.Offline, registry.Get("w1")!.Status); + } + + [Fact] + public void Sweep_DoesNotReEmitAlreadyOfflineWorker() + { + var registry = NewRegistry(offlineTimeoutSeconds: 30); + registry.Record(Heartbeat("w1", WorkerStatusType.Idle)); + registry.Sweep(DateTime.UtcNow.AddSeconds(31)); + + var changes = registry.Sweep(DateTime.UtcNow.AddSeconds(62)); + + Assert.Empty(changes); + } + + [Fact] + public void Record_AfterOffline_ReturnsReturnTransition() + { + var registry = NewRegistry(offlineTimeoutSeconds: 30); + registry.Record(Heartbeat("w1", WorkerStatusType.Idle)); + registry.Sweep(DateTime.UtcNow.AddSeconds(31)); + + var change = registry.Record(Heartbeat("w1", WorkerStatusType.Busy)); + + Assert.NotNull(change); + Assert.Equal(WorkerStatusType.Offline, change!.OldStatus); + Assert.Equal(WorkerStatusType.Busy, change.NewStatus); + } + + [Fact] + public void GetAll_ReturnsAllKnownWorkers() + { + var registry = NewRegistry(); + registry.Record(Heartbeat("w1", WorkerStatusType.Idle)); + registry.Record(Heartbeat("w2", WorkerStatusType.Busy)); + + Assert.Equal(2, registry.GetAll().Count); + Assert.Null(registry.Get("missing")); + } +} From 4bc4ce90516c6e9d3a927a2fa95ad57c510e4efe Mon Sep 17 00:00:00 2001 From: Tom Date: Tue, 23 Jun 2026 14:26:46 +0200 Subject: [PATCH 27/33] feat(workers-manager): add JobScheduler publishing jobs to the queue --- .../Interfaces/IJobScheduler.cs | 13 +++++ .../Services/JobScheduler.cs | 30 +++++++++++ .../Manager/JobSchedulerTests.cs | 52 +++++++++++++++++++ 3 files changed, 95 insertions(+) create mode 100644 src/SquidStd.Workers.Manager/Interfaces/IJobScheduler.cs create mode 100644 src/SquidStd.Workers.Manager/Services/JobScheduler.cs create mode 100644 tests/SquidStd.Tests/Manager/JobSchedulerTests.cs diff --git a/src/SquidStd.Workers.Manager/Interfaces/IJobScheduler.cs b/src/SquidStd.Workers.Manager/Interfaces/IJobScheduler.cs new file mode 100644 index 00000000..25efa737 --- /dev/null +++ b/src/SquidStd.Workers.Manager/Interfaces/IJobScheduler.cs @@ -0,0 +1,13 @@ +namespace SquidStd.Workers.Manager.Interfaces; + +/// +/// Enqueues jobs for workers to consume. +/// +public interface IJobScheduler +{ + /// Enqueues a job with parameters. + Task EnqueueAsync(string jobName, IReadOnlyDictionary parameters, CancellationToken cancellationToken = default); + + /// Enqueues a job with no parameters. + Task EnqueueAsync(string jobName, CancellationToken cancellationToken = default); +} diff --git a/src/SquidStd.Workers.Manager/Services/JobScheduler.cs b/src/SquidStd.Workers.Manager/Services/JobScheduler.cs new file mode 100644 index 00000000..82336a01 --- /dev/null +++ b/src/SquidStd.Workers.Manager/Services/JobScheduler.cs @@ -0,0 +1,30 @@ +using SquidStd.Messaging.Abstractions.Interfaces; +using SquidStd.Workers.Abstractions; +using SquidStd.Workers.Abstractions.Data; +using SquidStd.Workers.Manager.Data.Config; +using SquidStd.Workers.Manager.Interfaces; + +namespace SquidStd.Workers.Manager.Services; + +/// +/// Default : publishes s onto the configured jobs queue. +/// +public sealed class JobScheduler : IJobScheduler +{ + private readonly IMessageQueue _queue; + private readonly string _queueName; + + public JobScheduler(IMessageQueue queue, WorkerManagerConfig config) + { + _queue = queue; + _queueName = string.IsNullOrWhiteSpace(config.JobQueue) ? WorkerChannels.JobQueue : config.JobQueue; + } + + /// + public Task EnqueueAsync(string jobName, IReadOnlyDictionary parameters, CancellationToken cancellationToken = default) + => _queue.PublishAsync(_queueName, new JobRequest(jobName, parameters), cancellationToken); + + /// + public Task EnqueueAsync(string jobName, CancellationToken cancellationToken = default) + => EnqueueAsync(jobName, new Dictionary(), cancellationToken); +} diff --git a/tests/SquidStd.Tests/Manager/JobSchedulerTests.cs b/tests/SquidStd.Tests/Manager/JobSchedulerTests.cs new file mode 100644 index 00000000..7da319c6 --- /dev/null +++ b/tests/SquidStd.Tests/Manager/JobSchedulerTests.cs @@ -0,0 +1,52 @@ +using DryIoc; +using SquidStd.Core.Interfaces.Events; +using SquidStd.Messaging.Abstractions.Interfaces; +using SquidStd.Messaging.Extensions; +using SquidStd.Services.Core.Services; +using SquidStd.Workers.Abstractions; +using SquidStd.Workers.Abstractions.Data; +using SquidStd.Workers.Manager.Data.Config; +using SquidStd.Workers.Manager.Services; + +namespace SquidStd.Tests.Manager; + +public class JobSchedulerTests +{ + [Fact] + public async Task EnqueueAsync_PublishesJobRequestToConfiguredQueue() + { + var container = new Container(); + container.RegisterInstance(new EventBusService()); + container.AddInMemoryMessaging(); + var queue = container.Resolve(); + + var received = new TaskCompletionSource(); + using var _ = queue.Subscribe( + WorkerChannels.JobQueue, + new DelegateListener(job => received.TrySetResult(job))); + + var scheduler = new JobScheduler(queue, new WorkerManagerConfig()); + await scheduler.EnqueueAsync("resize", new Dictionary { ["w"] = "100" }); + + var job = await received.Task.WaitAsync(TimeSpan.FromSeconds(5)); + Assert.Equal("resize", job.JobName); + Assert.Equal("100", job.Parameters["w"]); + } + + private sealed class DelegateListener : IQueueMessageListenerAsync + { + private readonly Action _onMessage; + + public DelegateListener(Action onMessage) + { + _onMessage = onMessage; + } + + public Task HandleAsync(JobRequest message, CancellationToken cancellationToken) + { + _onMessage(message); + + return Task.CompletedTask; + } + } +} From fcd6bee5c84b17ec7ca430ab2e7589ba515f735c Mon Sep 17 00:00:00 2001 From: Tom Date: Tue, 23 Jun 2026 14:27:38 +0200 Subject: [PATCH 28/33] feat(workers-manager): add HeartbeatCollectorService folding heartbeats into the registry --- .../Services/HeartbeatCollectorService.cs | 65 +++++++++++++++++++ .../Manager/HeartbeatCollectorServiceTests.cs | 62 ++++++++++++++++++ 2 files changed, 127 insertions(+) create mode 100644 src/SquidStd.Workers.Manager/Services/HeartbeatCollectorService.cs create mode 100644 tests/SquidStd.Tests/Manager/HeartbeatCollectorServiceTests.cs diff --git a/src/SquidStd.Workers.Manager/Services/HeartbeatCollectorService.cs b/src/SquidStd.Workers.Manager/Services/HeartbeatCollectorService.cs new file mode 100644 index 00000000..91317e0a --- /dev/null +++ b/src/SquidStd.Workers.Manager/Services/HeartbeatCollectorService.cs @@ -0,0 +1,65 @@ +using Serilog; +using SquidStd.Abstractions.Interfaces.Services; +using SquidStd.Core.Interfaces.Events; +using SquidStd.Messaging.Abstractions.Interfaces; +using SquidStd.Workers.Abstractions; +using SquidStd.Workers.Abstractions.Data; +using SquidStd.Workers.Manager.Data.Config; + +namespace SquidStd.Workers.Manager.Services; + +/// +/// Subscribes to the heartbeat topic and folds each into the registry, +/// publishing any resulting status transition on the event bus. +/// +public sealed class HeartbeatCollectorService : ISquidStdService +{ + private readonly ILogger _logger = Log.ForContext(); + private readonly IMessageTopic _topic; + private readonly WorkerRegistry _registry; + private readonly IEventBus _eventBus; + private readonly string _topicName; + private IDisposable? _subscription; + + public HeartbeatCollectorService(IMessageTopic topic, WorkerRegistry registry, IEventBus eventBus, WorkerManagerConfig config) + { + _topic = topic; + _registry = registry; + _eventBus = eventBus; + _topicName = string.IsNullOrWhiteSpace(config.HeartbeatTopic) ? WorkerChannels.HeartbeatTopic : config.HeartbeatTopic; + } + + /// + public ValueTask StartAsync(CancellationToken cancellationToken = default) + { + _subscription = _topic.Subscribe(_topicName, OnHeartbeatAsync); + _logger.Information("Worker manager collecting heartbeats from topic '{Topic}'.", _topicName); + + return ValueTask.CompletedTask; + } + + /// + public ValueTask StopAsync(CancellationToken cancellationToken = default) + { + _subscription?.Dispose(); + _subscription = null; + + return ValueTask.CompletedTask; + } + + private async Task OnHeartbeatAsync(WorkerHeartbeat heartbeat, CancellationToken cancellationToken) + { + try + { + var change = _registry.Record(heartbeat); + if (change is not null) + { + await _eventBus.PublishAsync(change, cancellationToken); + } + } + catch (Exception ex) + { + _logger.Error(ex, "Failed to process heartbeat from '{WorkerId}'.", heartbeat.WorkerId); + } + } +} diff --git a/tests/SquidStd.Tests/Manager/HeartbeatCollectorServiceTests.cs b/tests/SquidStd.Tests/Manager/HeartbeatCollectorServiceTests.cs new file mode 100644 index 00000000..736e52c1 --- /dev/null +++ b/tests/SquidStd.Tests/Manager/HeartbeatCollectorServiceTests.cs @@ -0,0 +1,62 @@ +using DryIoc; +using SquidStd.Core.Interfaces.Events; +using SquidStd.Messaging.Abstractions.Interfaces; +using SquidStd.Messaging.Extensions; +using SquidStd.Services.Core.Services; +using SquidStd.Workers.Abstractions; +using SquidStd.Workers.Abstractions.Data; +using SquidStd.Workers.Abstractions.Types; +using SquidStd.Workers.Manager.Data.Config; +using SquidStd.Workers.Manager.Data.Events; +using SquidStd.Workers.Manager.Services; + +namespace SquidStd.Tests.Manager; + +public class HeartbeatCollectorServiceTests +{ + [Fact] + public async Task Collector_RecordsHeartbeat_AndPublishesDiscoveredEvent() + { + var container = new Container(); + var eventBus = new EventBusService(); + container.RegisterInstance(eventBus); + container.AddInMemoryMessaging(); + + var topic = container.Resolve(); + var registry = new WorkerRegistry(new WorkerManagerConfig()); + + var discovered = new TaskCompletionSource(); + eventBus.RegisterAsyncListener(new DelegateListener(e => discovered.TrySetResult(e))); + + var service = new HeartbeatCollectorService(topic, registry, eventBus, new WorkerManagerConfig()); + await service.StartAsync(); + + await topic.PublishAsync( + WorkerChannels.HeartbeatTopic, + new WorkerHeartbeat("w1", DateTime.UtcNow, WorkerStatusType.Idle, 0, 8)); + + var change = await discovered.Task.WaitAsync(TimeSpan.FromSeconds(5)); + await service.StopAsync(); + + Assert.Equal("w1", change.WorkerId); + Assert.Null(change.OldStatus); + Assert.NotNull(registry.Get("w1")); + } + + private sealed class DelegateListener : IAsyncEventListener + { + private readonly Action _onEvent; + + public DelegateListener(Action onEvent) + { + _onEvent = onEvent; + } + + public Task HandleAsync(WorkerStatusChangedEvent eventData, CancellationToken cancellationToken) + { + _onEvent(eventData); + + return Task.CompletedTask; + } + } +} From 23efd3718995ef72c84d51db39777e31d7011b03 Mon Sep 17 00:00:00 2001 From: Tom Date: Tue, 23 Jun 2026 14:28:44 +0200 Subject: [PATCH 29/33] feat(workers-manager): add WorkerOfflineSweepService marking stale workers offline --- .../Services/WorkerOfflineSweepService.cs | 80 +++++++++++++++++++ .../Manager/Support/FakeTimerService.cs | 41 ++++++++++ .../Manager/WorkerOfflineSweepServiceTests.cs | 67 ++++++++++++++++ 3 files changed, 188 insertions(+) create mode 100644 src/SquidStd.Workers.Manager/Services/WorkerOfflineSweepService.cs create mode 100644 tests/SquidStd.Tests/Manager/Support/FakeTimerService.cs create mode 100644 tests/SquidStd.Tests/Manager/WorkerOfflineSweepServiceTests.cs diff --git a/src/SquidStd.Workers.Manager/Services/WorkerOfflineSweepService.cs b/src/SquidStd.Workers.Manager/Services/WorkerOfflineSweepService.cs new file mode 100644 index 00000000..34d63a8d --- /dev/null +++ b/src/SquidStd.Workers.Manager/Services/WorkerOfflineSweepService.cs @@ -0,0 +1,80 @@ +using Serilog; +using SquidStd.Abstractions.Interfaces.Services; +using SquidStd.Core.Interfaces.Events; +using SquidStd.Core.Interfaces.Timing; +using SquidStd.Workers.Manager.Data.Config; + +namespace SquidStd.Workers.Manager.Services; + +/// +/// Periodically marks workers Offline whose heartbeats have stopped, using the timer wheel +/// (), and publishes the resulting transitions. +/// +public sealed class WorkerOfflineSweepService : ISquidStdService +{ + private const int DefaultIntervalSeconds = 10; + + private readonly ILogger _logger = Log.ForContext(); + private readonly ITimerService _timer; + private readonly WorkerRegistry _registry; + private readonly IEventBus _eventBus; + private readonly TimeSpan _interval; + private string? _timerId; + + public WorkerOfflineSweepService(ITimerService timer, WorkerRegistry registry, IEventBus eventBus, WorkerManagerConfig config) + { + _timer = timer; + _registry = registry; + _eventBus = eventBus; + + var seconds = config.SweepIntervalSeconds > 0 ? config.SweepIntervalSeconds : DefaultIntervalSeconds; + if (config.SweepIntervalSeconds <= 0) + { + _logger.Warning( + "SweepIntervalSeconds was {Value}; falling back to {Default}s.", + config.SweepIntervalSeconds, + DefaultIntervalSeconds); + } + + _interval = TimeSpan.FromSeconds(seconds); + } + + /// + public ValueTask StartAsync(CancellationToken cancellationToken = default) + { + _timerId = _timer.RegisterTimer("worker-offline-sweep", _interval, OnTick, delay: _interval, repeat: true); + + return ValueTask.CompletedTask; + } + + /// + public ValueTask StopAsync(CancellationToken cancellationToken = default) + { + if (_timerId is not null) + { + _timer.UnregisterTimer(_timerId); + _timerId = null; + } + + return ValueTask.CompletedTask; + } + + /// Runs one sweep and publishes the transitions. Public so it can be driven directly in tests. + public async Task RunSweepAsync() + { + try + { + foreach (var change in _registry.Sweep(DateTime.UtcNow)) + { + await _eventBus.PublishAsync(change, CancellationToken.None); + } + } + catch (Exception ex) + { + _logger.Error(ex, "Worker offline sweep failed."); + } + } + + private void OnTick() + => _ = RunSweepAsync(); +} diff --git a/tests/SquidStd.Tests/Manager/Support/FakeTimerService.cs b/tests/SquidStd.Tests/Manager/Support/FakeTimerService.cs new file mode 100644 index 00000000..03857243 --- /dev/null +++ b/tests/SquidStd.Tests/Manager/Support/FakeTimerService.cs @@ -0,0 +1,41 @@ +using SquidStd.Core.Interfaces.Timing; + +namespace SquidStd.Tests.Manager.Support; + +/// Captures timer registration for tests without running a real timer wheel. +public sealed class FakeTimerService : ITimerService +{ + public string? RegisteredName { get; private set; } + + public TimeSpan RegisteredInterval { get; private set; } + + public bool RegisteredRepeat { get; private set; } + + public string? UnregisteredId { get; private set; } + + public string RegisterTimer(string name, TimeSpan interval, Action callback, TimeSpan? delay = null, bool repeat = false) + { + RegisteredName = name; + RegisteredInterval = interval; + RegisteredRepeat = repeat; + + return "timer-1"; + } + + public void UnregisterAllTimers() + { + } + + public bool UnregisterTimer(string timerId) + { + UnregisteredId = timerId; + + return true; + } + + public int UnregisterTimersByName(string name) + => 0; + + public int UpdateTicksDelta(long timestampMilliseconds) + => 0; +} diff --git a/tests/SquidStd.Tests/Manager/WorkerOfflineSweepServiceTests.cs b/tests/SquidStd.Tests/Manager/WorkerOfflineSweepServiceTests.cs new file mode 100644 index 00000000..d1a0dfec --- /dev/null +++ b/tests/SquidStd.Tests/Manager/WorkerOfflineSweepServiceTests.cs @@ -0,0 +1,67 @@ +using SquidStd.Services.Core.Services; +using SquidStd.Tests.Manager.Support; +using SquidStd.Workers.Abstractions.Data; +using SquidStd.Workers.Abstractions.Types; +using SquidStd.Workers.Manager.Data.Config; +using SquidStd.Workers.Manager.Data.Events; +using SquidStd.Workers.Manager.Services; +using SquidStd.Core.Interfaces.Events; + +namespace SquidStd.Tests.Manager; + +public class WorkerOfflineSweepServiceTests +{ + [Fact] + public async Task StartAsync_RegistersRepeatingTimer_StopAsyncUnregisters() + { + var timer = new FakeTimerService(); + var registry = new WorkerRegistry(new WorkerManagerConfig()); + var service = new WorkerOfflineSweepService(timer, registry, new EventBusService(), new WorkerManagerConfig { SweepIntervalSeconds = 5 }); + + await service.StartAsync(); + + Assert.Equal("worker-offline-sweep", timer.RegisteredName); + Assert.Equal(TimeSpan.FromSeconds(5), timer.RegisteredInterval); + Assert.True(timer.RegisteredRepeat); + + await service.StopAsync(); + Assert.Equal("timer-1", timer.UnregisteredId); + } + + [Fact] + public async Task RunSweepAsync_MarksOverdueWorkerOffline_AndPublishesTransition() + { + var registry = new WorkerRegistry(new WorkerManagerConfig { OfflineTimeoutSeconds = 1 }); + registry.Record(new WorkerHeartbeat("w1", DateTime.UtcNow, WorkerStatusType.Idle, 0, 8)); + await Task.Delay(1100); + + var eventBus = new EventBusService(); + var offline = new TaskCompletionSource(); + eventBus.RegisterAsyncListener(new DelegateListener(e => offline.TrySetResult(e))); + + var service = new WorkerOfflineSweepService(new FakeTimerService(), registry, eventBus, new WorkerManagerConfig()); + await service.RunSweepAsync(); + + var change = await offline.Task.WaitAsync(TimeSpan.FromSeconds(5)); + Assert.Equal("w1", change.WorkerId); + Assert.Equal(WorkerStatusType.Offline, change.NewStatus); + Assert.Equal(WorkerStatusType.Offline, registry.Get("w1")!.Status); + } + + private sealed class DelegateListener : IAsyncEventListener + { + private readonly Action _onEvent; + + public DelegateListener(Action onEvent) + { + _onEvent = onEvent; + } + + public Task HandleAsync(WorkerStatusChangedEvent eventData, CancellationToken cancellationToken) + { + _onEvent(eventData); + + return Task.CompletedTask; + } + } +} From b92c6b75b82dd9d83ed1fe5e68b68884dcba590d Mon Sep 17 00:00:00 2001 From: Tom Date: Tue, 23 Jun 2026 14:29:40 +0200 Subject: [PATCH 30/33] feat(workers-manager): add HTTP endpoints for querying workers and enqueuing jobs --- .../Endpoints/WorkerManagerEndpoints.cs | 48 +++++++++++ .../WorkerManagerEndpointsExtensions.cs | 25 ++++++ .../Manager/WorkerManagerEndpointsTests.cs | 83 +++++++++++++++++++ 3 files changed, 156 insertions(+) create mode 100644 src/SquidStd.Workers.Manager/Endpoints/WorkerManagerEndpoints.cs create mode 100644 src/SquidStd.Workers.Manager/Extensions/WorkerManagerEndpointsExtensions.cs create mode 100644 tests/SquidStd.Tests/Manager/WorkerManagerEndpointsTests.cs diff --git a/src/SquidStd.Workers.Manager/Endpoints/WorkerManagerEndpoints.cs b/src/SquidStd.Workers.Manager/Endpoints/WorkerManagerEndpoints.cs new file mode 100644 index 00000000..95765ec7 --- /dev/null +++ b/src/SquidStd.Workers.Manager/Endpoints/WorkerManagerEndpoints.cs @@ -0,0 +1,48 @@ +using Microsoft.AspNetCore.Http; +using Microsoft.AspNetCore.Http.HttpResults; +using SquidStd.Workers.Abstractions.Data; +using SquidStd.Workers.Manager.Data; +using SquidStd.Workers.Manager.Interfaces; + +namespace SquidStd.Workers.Manager.Endpoints; + +/// +/// Typed minimal-API handlers for the worker manager surface. Static so they can be unit-tested directly. +/// +public static class WorkerManagerEndpoints +{ + /// Lists all known workers. + public static Ok> GetWorkers(IWorkerRegistry registry) + => TypedResults.Ok(registry.GetAll()); + + /// Returns a single worker, or 404 when unknown. + public static Results, NotFound> GetWorker(string id, IWorkerRegistry registry) + { + var info = registry.Get(id); + + return info is null ? TypedResults.NotFound() : TypedResults.Ok(info); + } + + /// Enqueues a job; 400 on a blank name, 503 when the queue is unavailable. + public static async Task, ProblemHttpResult>> EnqueueJob( + EnqueueJobRequest request, + IJobScheduler scheduler, + CancellationToken cancellationToken) + { + if (string.IsNullOrWhiteSpace(request.JobName)) + { + return TypedResults.BadRequest("JobName is required."); + } + + try + { + await scheduler.EnqueueAsync(request.JobName, request.Parameters ?? new Dictionary(), cancellationToken); + + return TypedResults.Accepted((string?)null); + } + catch (Exception) + { + return TypedResults.Problem(statusCode: StatusCodes.Status503ServiceUnavailable, detail: "Job queue unavailable."); + } + } +} diff --git a/src/SquidStd.Workers.Manager/Extensions/WorkerManagerEndpointsExtensions.cs b/src/SquidStd.Workers.Manager/Extensions/WorkerManagerEndpointsExtensions.cs new file mode 100644 index 00000000..6290d917 --- /dev/null +++ b/src/SquidStd.Workers.Manager/Extensions/WorkerManagerEndpointsExtensions.cs @@ -0,0 +1,25 @@ +using Microsoft.AspNetCore.Builder; +using Microsoft.AspNetCore.Routing; +using SquidStd.Workers.Manager.Endpoints; + +namespace SquidStd.Workers.Manager.Extensions; + +/// +/// Maps the opt-in worker manager HTTP endpoints. +/// +public static class WorkerManagerEndpointsExtensions +{ + /// + /// Maps GET /workers, GET /workers/{id}, and POST /jobs. + /// + public static IEndpointRouteBuilder MapWorkerManagerEndpoints(this IEndpointRouteBuilder endpoints) + { + ArgumentNullException.ThrowIfNull(endpoints); + + endpoints.MapGet("/workers", WorkerManagerEndpoints.GetWorkers); + endpoints.MapGet("/workers/{id}", WorkerManagerEndpoints.GetWorker); + endpoints.MapPost("/jobs", WorkerManagerEndpoints.EnqueueJob); + + return endpoints; + } +} diff --git a/tests/SquidStd.Tests/Manager/WorkerManagerEndpointsTests.cs b/tests/SquidStd.Tests/Manager/WorkerManagerEndpointsTests.cs new file mode 100644 index 00000000..589e2324 --- /dev/null +++ b/tests/SquidStd.Tests/Manager/WorkerManagerEndpointsTests.cs @@ -0,0 +1,83 @@ +using DryIoc; +using Microsoft.AspNetCore.Http.HttpResults; +using SquidStd.Core.Interfaces.Events; +using SquidStd.Messaging.Abstractions.Interfaces; +using SquidStd.Messaging.Extensions; +using SquidStd.Services.Core.Services; +using SquidStd.Workers.Abstractions.Data; +using SquidStd.Workers.Abstractions.Types; +using SquidStd.Workers.Manager.Data; +using SquidStd.Workers.Manager.Data.Config; +using SquidStd.Workers.Manager.Endpoints; +using SquidStd.Workers.Manager.Services; + +namespace SquidStd.Tests.Manager; + +public class WorkerManagerEndpointsTests +{ + private static WorkerRegistry RegistryWith(params string[] workerIds) + { + var registry = new WorkerRegistry(new WorkerManagerConfig()); + foreach (var id in workerIds) + { + registry.Record(new WorkerHeartbeat(id, DateTime.UtcNow, WorkerStatusType.Idle, 0, 8)); + } + + return registry; + } + + private static JobScheduler NewScheduler() + { + var container = new Container(); + container.RegisterInstance(new EventBusService()); + container.AddInMemoryMessaging(); + + return new JobScheduler(container.Resolve(), new WorkerManagerConfig()); + } + + [Fact] + public void GetWorkers_ReturnsOkWithAll() + { + var result = WorkerManagerEndpoints.GetWorkers(RegistryWith("w1", "w2")); + + Assert.Equal(2, result.Value!.Count); + } + + [Fact] + public void GetWorker_ReturnsOk_WhenPresent() + { + var result = WorkerManagerEndpoints.GetWorker("w1", RegistryWith("w1")); + + Assert.IsType>(result.Result); + } + + [Fact] + public void GetWorker_ReturnsNotFound_WhenAbsent() + { + var result = WorkerManagerEndpoints.GetWorker("missing", RegistryWith("w1")); + + Assert.IsType(result.Result); + } + + [Fact] + public async Task EnqueueJob_ReturnsAccepted_AndSchedulesJob() + { + var result = await WorkerManagerEndpoints.EnqueueJob( + new EnqueueJobRequest("resize", new Dictionary()), + NewScheduler(), + CancellationToken.None); + + Assert.IsType(result.Result); + } + + [Fact] + public async Task EnqueueJob_ReturnsBadRequest_WhenJobNameBlank() + { + var result = await WorkerManagerEndpoints.EnqueueJob( + new EnqueueJobRequest(" ", null), + NewScheduler(), + CancellationToken.None); + + Assert.IsType>(result.Result); + } +} From 38c990e8c519f854d427de2fdca25335c074f930 Mon Sep 17 00:00:00 2001 From: Tom Date: Tue, 23 Jun 2026 14:30:42 +0200 Subject: [PATCH 31/33] feat(workers-manager): add AddWorkerManager registration extension --- .../WorkerManagerRegistrationExtensions.cs | 42 +++++++++++++ ...orkerManagerRegistrationExtensionsTests.cs | 62 +++++++++++++++++++ 2 files changed, 104 insertions(+) create mode 100644 src/SquidStd.Workers.Manager/Extensions/WorkerManagerRegistrationExtensions.cs create mode 100644 tests/SquidStd.Tests/Manager/WorkerManagerRegistrationExtensionsTests.cs diff --git a/src/SquidStd.Workers.Manager/Extensions/WorkerManagerRegistrationExtensions.cs b/src/SquidStd.Workers.Manager/Extensions/WorkerManagerRegistrationExtensions.cs new file mode 100644 index 00000000..4b825bab --- /dev/null +++ b/src/SquidStd.Workers.Manager/Extensions/WorkerManagerRegistrationExtensions.cs @@ -0,0 +1,42 @@ +using DryIoc; +using SquidStd.Abstractions.Extensions.Config; +using SquidStd.Abstractions.Extensions.Services; +using SquidStd.Core.Data.Timing; +using SquidStd.Services.Core.Services.Scheduling; +using SquidStd.Workers.Manager.Data.Config; +using SquidStd.Workers.Manager.Interfaces; +using SquidStd.Workers.Manager.Services; + +namespace SquidStd.Workers.Manager.Extensions; + +/// +/// DryIoc registration helpers for the worker manager. +/// +public static class WorkerManagerRegistrationExtensions +{ + /// + /// Registers the worker manager: config section, registry, job scheduler, the collector and sweep + /// lifecycle services, and the timer-wheel pump (only if it is not already registered). + /// + public static IContainer AddWorkerManager(this IContainer container) + { + ArgumentNullException.ThrowIfNull(container); + + container.RegisterConfigSection("workerManager", static () => new WorkerManagerConfig(), -50); + + container.Register(Reuse.Singleton); + container.RegisterMapping(); + container.Register(Reuse.Singleton); + + container.RegisterStdService(100); + container.RegisterStdService(110); + + if (!container.IsRegistered()) + { + container.RegisterConfigSection("timerWheelPump", static () => new TimerWheelPumpConfig(), -90); + container.RegisterStdService(-1); + } + + return container; + } +} diff --git a/tests/SquidStd.Tests/Manager/WorkerManagerRegistrationExtensionsTests.cs b/tests/SquidStd.Tests/Manager/WorkerManagerRegistrationExtensionsTests.cs new file mode 100644 index 00000000..56c6c7c2 --- /dev/null +++ b/tests/SquidStd.Tests/Manager/WorkerManagerRegistrationExtensionsTests.cs @@ -0,0 +1,62 @@ +using DryIoc; +using SquidStd.Core.Interfaces.Events; +using SquidStd.Core.Interfaces.Timing; +using SquidStd.Messaging.Extensions; +using SquidStd.Services.Core.Services; +using SquidStd.Services.Core.Services.Scheduling; +using SquidStd.Tests.Manager.Support; +using SquidStd.Workers.Manager.Data.Config; +using SquidStd.Workers.Manager.Extensions; +using SquidStd.Workers.Manager.Interfaces; +using SquidStd.Workers.Manager.Services; + +namespace SquidStd.Tests.Manager; + +public class WorkerManagerRegistrationExtensionsTests +{ + private static Container NewContainer() + { + var container = new Container(); + container.RegisterInstance(new EventBusService()); + container.AddInMemoryMessaging(); + // RegisterCoreServices normally provides ITimerService; supply a fake so the sweep service resolves. + container.RegisterInstance(new FakeTimerService()); + // ConfigManager normally registers config instances; register one directly for the test. + container.RegisterInstance(new WorkerManagerConfig()); + + return container; + } + + [Fact] + public void AddWorkerManager_RegistersResolvableServices() + { + using var container = NewContainer(); + + container.AddWorkerManager(); + + Assert.NotNull(container.Resolve()); + Assert.NotNull(container.Resolve()); + Assert.NotNull(container.Resolve()); + Assert.NotNull(container.Resolve()); + } + + [Fact] + public void AddWorkerManager_RegistersTimerWheelPump_WhenAbsent() + { + using var container = NewContainer(); + + container.AddWorkerManager(); + + Assert.True(container.IsRegistered()); + } + + [Fact] + public void AddWorkerManager_SharesRegistryInstanceAcrossInterfaceAndConcrete() + { + using var container = NewContainer(); + + container.AddWorkerManager(); + + Assert.Same(container.Resolve(), container.Resolve()); + } +} From b9639102e2c40877e8ae91f0aabacfc21c96b395 Mon Sep 17 00:00:00 2001 From: Tom Date: Tue, 23 Jun 2026 14:32:04 +0200 Subject: [PATCH 32/33] docs(workers): add READMEs and DocFX articles for the workers system --- docs/articles/toc.yml | 6 +++ docs/articles/workers-abstractions.md | 1 + docs/articles/workers-manager.md | 1 + docs/articles/workers.md | 1 + src/SquidStd.Workers.Abstractions/README.md | 35 +++++++++++++ src/SquidStd.Workers.Manager/README.md | 50 ++++++++++++++++++ src/SquidStd.Workers/README.md | 58 +++++++++++++++++++++ 7 files changed, 152 insertions(+) create mode 100644 docs/articles/workers-abstractions.md create mode 100644 docs/articles/workers-manager.md create mode 100644 docs/articles/workers.md create mode 100644 src/SquidStd.Workers.Abstractions/README.md create mode 100644 src/SquidStd.Workers.Manager/README.md create mode 100644 src/SquidStd.Workers/README.md diff --git a/docs/articles/toc.yml b/docs/articles/toc.yml index 9dbf4625..280a0eb9 100644 --- a/docs/articles/toc.yml +++ b/docs/articles/toc.yml @@ -38,6 +38,12 @@ href: scripting-lua.md - name: SquidStd.Templating href: templating.md +- name: SquidStd.Workers.Abstractions + href: workers-abstractions.md +- name: SquidStd.Workers + href: workers.md +- name: SquidStd.Workers.Manager + href: workers-manager.md - name: Serialization href: serialization.md - name: Scheduler diff --git a/docs/articles/workers-abstractions.md b/docs/articles/workers-abstractions.md new file mode 100644 index 00000000..274c4be9 --- /dev/null +++ b/docs/articles/workers-abstractions.md @@ -0,0 +1 @@ +[!include[](../../src/SquidStd.Workers.Abstractions/README.md)] diff --git a/docs/articles/workers-manager.md b/docs/articles/workers-manager.md new file mode 100644 index 00000000..2d372ba9 --- /dev/null +++ b/docs/articles/workers-manager.md @@ -0,0 +1 @@ +[!include[](../../src/SquidStd.Workers.Manager/README.md)] diff --git a/docs/articles/workers.md b/docs/articles/workers.md new file mode 100644 index 00000000..22565bdb --- /dev/null +++ b/docs/articles/workers.md @@ -0,0 +1 @@ +[!include[](../../src/SquidStd.Workers/README.md)] diff --git a/src/SquidStd.Workers.Abstractions/README.md b/src/SquidStd.Workers.Abstractions/README.md new file mode 100644 index 00000000..32063e9f --- /dev/null +++ b/src/SquidStd.Workers.Abstractions/README.md @@ -0,0 +1,35 @@ +

+ SquidStd +

+ +

SquidStd.Workers.Abstractions

+ +

+ NuGet + Downloads + docs + license +

+ +Shared, dependency-free contracts between the SquidStd worker runtime and manager: the job descriptor, the +worker heartbeat, the manager-side worker view, the worker status enum, and the conventional channel names. + +## Install + +```bash +dotnet add package SquidStd.Workers.Abstractions +``` + +## Key types + +| Type | Purpose | +|------|---------| +| `JobRequest` | A job (name + string parameters) enqueued by the manager and consumed by a worker. | +| `WorkerHeartbeat` | Liveness signal a worker publishes (status, active jobs, max concurrency). | +| `WorkerInfo` | The manager-side view of a worker, folded from heartbeats. | +| `WorkerStatusType` | `Idle` / `Busy` / `Offline`. | +| `WorkerChannels` | Default jobs-queue and heartbeat-topic names. | + +## License + +MIT — part of [SquidStd](https://github.com/tgiachi/squid-std). diff --git a/src/SquidStd.Workers.Manager/README.md b/src/SquidStd.Workers.Manager/README.md new file mode 100644 index 00000000..a1e33a74 --- /dev/null +++ b/src/SquidStd.Workers.Manager/README.md @@ -0,0 +1,50 @@ +

+ SquidStd +

+ +

SquidStd.Workers.Manager

+ +

+ NuGet + Downloads + docs + license +

+ +The SquidStd worker manager. Enqueues jobs, collects worker heartbeats into an in-memory registry, marks +workers `Offline` on a periodic sweep, publishes status-transition events, and exposes an opt-in ASP.NET +minimal-API surface. + +## Install + +```bash +dotnet add package SquidStd.Workers.Manager +``` + +## Usage + +```csharp +using DryIoc; +using SquidStd.Workers.Manager.Extensions; + +container.AddInMemoryMessaging(); // or AddRabbitMqMessaging(...) +container.AddWorkerManager(); + +// In an ASP.NET app, map the opt-in HTTP surface: +app.MapWorkerManagerEndpoints(); // GET /workers, GET /workers/{id}, POST /jobs +``` + +## Key types + +| Type | Purpose | +|------|---------| +| `IJobScheduler` | `EnqueueAsync(jobName, parameters)` onto the jobs queue. | +| `IWorkerRegistry` | `GetAll()` / `Get(id)` over the live worker view. | +| `WorkerStatusChangedEvent` | Published on the event bus on discover / offline / return. | +| `WorkerManagerConfig` | `OfflineTimeoutSeconds`, `SweepIntervalSeconds`, queue/topic names. | +| `WorkerManagerRegistrationExtensions` | `AddWorkerManager()`. | +| `WorkerManagerEndpointsExtensions` | `MapWorkerManagerEndpoints()`. | + +## License + +MIT — part of [SquidStd](https://github.com/tgiachi/squid-std). diff --git a/src/SquidStd.Workers/README.md b/src/SquidStd.Workers/README.md new file mode 100644 index 00000000..753a2a65 --- /dev/null +++ b/src/SquidStd.Workers/README.md @@ -0,0 +1,58 @@ +

+ SquidStd +

+ +

SquidStd.Workers

+ +

+ NuGet + Downloads + docs + license +

+ +The SquidStd worker runtime. Consumes `JobRequest`s from the jobs queue, dispatches each to a named +`IJobHandler` (up to `MaxConcurrency` in parallel), and publishes a `WorkerHeartbeat` on the heartbeat topic +every few seconds. + +## Install + +```bash +dotnet add package SquidStd.Workers +``` + +## Usage + +```csharp +using DryIoc; +using SquidStd.Workers.Extensions; + +container.AddInMemoryMessaging(); // or AddRabbitMqMessaging(...) +container.AddWorkers(); +container.AddJobHandler(); +``` + +```csharp +public sealed class ResizeImageHandler : IJobHandler +{ + public string JobName => "resize-image"; + + public Task HandleAsync(JobRequest job, CancellationToken cancellationToken) + { + // job.Parameters["width"], ... + return Task.CompletedTask; + } +} +``` + +## Key types + +| Type | Purpose | +|------|---------| +| `IJobHandler` | Handles jobs of one named kind. | +| `WorkersConfig` | `WorkerId`, `HeartbeatIntervalSeconds`, `MaxConcurrency`, queue/topic names. | +| `WorkersRegistrationExtensions` | `AddWorkers()` and `AddJobHandler()`. | + +## License + +MIT — part of [SquidStd](https://github.com/tgiachi/squid-std). From 3ed8b15c08d21ee8e3f9b38060f112ecf1314abb Mon Sep 17 00:00:00 2001 From: Tom Date: Tue, 23 Jun 2026 14:37:14 +0200 Subject: [PATCH 33/33] test(workers): add end-to-end RabbitMQ integration test for worker and manager --- .../Workers/WorkerSystemIntegrationTests.cs | 151 ++++++++++++++++++ 1 file changed, 151 insertions(+) create mode 100644 tests/SquidStd.Tests/Integration/Workers/WorkerSystemIntegrationTests.cs diff --git a/tests/SquidStd.Tests/Integration/Workers/WorkerSystemIntegrationTests.cs b/tests/SquidStd.Tests/Integration/Workers/WorkerSystemIntegrationTests.cs new file mode 100644 index 00000000..d02ed74e --- /dev/null +++ b/tests/SquidStd.Tests/Integration/Workers/WorkerSystemIntegrationTests.cs @@ -0,0 +1,151 @@ +using DryIoc; +using SquidStd.Core.Interfaces.Events; +using SquidStd.Messaging.Abstractions.Interfaces; +using SquidStd.Messaging.RabbitMq.Data.Config; +using SquidStd.Messaging.RabbitMq.Extensions; +using SquidStd.Services.Core.Services; +using SquidStd.Tests.Messaging.RabbitMq; +using SquidStd.Workers.Abstractions.Data; +using SquidStd.Workers.Data.Config; +using SquidStd.Workers.Interfaces; +using SquidStd.Workers.Services; +using SquidStd.Workers.Manager.Data.Config; +using SquidStd.Workers.Manager.Services; + +namespace SquidStd.Tests.Integration.Workers; + +/// +/// End-to-end test of the workers system over a real RabbitMQ broker (Testcontainers): the manager +/// enqueues a job that the worker consumes and runs (real queue), and the worker's heartbeat reaches the +/// manager's registry (real fan-out topic). +/// +[Collection(RabbitMqCollection.Name)] +public class WorkerSystemIntegrationTests +{ + private static readonly TimeSpan Timeout = TimeSpan.FromSeconds(30); + + private readonly RabbitMqContainerFixture _fixture; + + public WorkerSystemIntegrationTests(RabbitMqContainerFixture fixture) + { + _fixture = fixture; + } + + [Fact] + public async Task Manager_EnqueuesJob_WorkerRunsIt_AndHeartbeatReachesRegistry() + { + using var container = new Container(); + container.RegisterInstance(new EventBusService()); + container.AddRabbitMqMessaging(new RabbitMqOptions { Uri = new Uri(_fixture.AmqpUri) }); + + // Unique channel names per run so parallel/other tests on the shared broker do not interfere. + var suffix = Guid.NewGuid().ToString("N"); + var jobQueue = $"it.jobs.{suffix}"; + var heartbeatTopic = $"it.hb.{suffix}"; + + var workersConfig = new WorkersConfig + { + WorkerId = "it-worker", + MaxConcurrency = 2, + HeartbeatIntervalSeconds = 1, + JobQueue = jobQueue, + HeartbeatTopic = heartbeatTopic + }; + var managerConfig = new WorkerManagerConfig + { + JobQueue = jobQueue, + HeartbeatTopic = heartbeatTopic + }; + + // Open the RabbitMQ connections. + var queueProvider = container.Resolve(); + var topicProvider = container.Resolve(); + await queueProvider.StartAsync(); + await topicProvider.StartAsync(); + + var queue = container.Resolve(); + var topic = container.Resolve(); + var eventBus = container.Resolve(); + + // Worker side. + var jobRan = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var handler = new CapturingJobHandler("greet", jobRan); + var workerState = new WorkerState(workersConfig); + var dispatcher = new JobDispatcher([handler]); + var consumer = new WorkerConsumerService(queue, dispatcher, workerState, workersConfig); + var heartbeatService = new WorkerHeartbeatService(topic, workerState, workersConfig); + + // Manager side. + var registry = new WorkerRegistry(managerConfig); + var collector = new HeartbeatCollectorService(topic, registry, eventBus, managerConfig); + var scheduler = new JobScheduler(queue, managerConfig); + + try + { + await collector.StartAsync(); + await consumer.StartAsync(); + await heartbeatService.StartAsync(); + + // Let the broker register the queue consumer and the topic binding before publishing. + await Task.Delay(500); + + await scheduler.EnqueueAsync("greet", new Dictionary { ["name"] = "squid" }); + + // The worker consumed the job from the real queue and ran the handler. + var ranJob = await jobRan.Task.WaitAsync(Timeout); + Assert.Equal("greet", ranJob.JobName); + Assert.Equal("squid", ranJob.Parameters["name"]); + + // The worker's heartbeat reached the manager's registry over the real topic. + var seen = await PollForWorker(registry, "it-worker"); + Assert.NotNull(seen); + Assert.Equal("it-worker", seen!.WorkerId); + Assert.Equal(2, seen.MaxConcurrency); + } + finally + { + await heartbeatService.StopAsync(); + await consumer.StopAsync(); + await collector.StopAsync(); + await topicProvider.StopAsync(); + await queueProvider.StopAsync(); + } + } + + private static async Task PollForWorker(WorkerRegistry registry, string workerId) + { + var deadline = DateTime.UtcNow + Timeout; + while (DateTime.UtcNow < deadline) + { + var info = registry.Get(workerId); + if (info is not null) + { + return info; + } + + await Task.Delay(200); + } + + return null; + } + + private sealed class CapturingJobHandler : IJobHandler + { + private readonly TaskCompletionSource _completion; + + public CapturingJobHandler(string jobName, TaskCompletionSource completion) + { + JobName = jobName; + _completion = completion; + } + + public string JobName { get; } + + public Task HandleAsync(JobRequest job, CancellationToken cancellationToken) + { + _completion.TrySetResult(job); + + return Task.CompletedTask; + } + } +}