From 8e301d4e1ef8b85f036d8f3cf076ff2a46e0267a Mon Sep 17 00:00:00 2001 From: Sipke Schoorstra Date: Mon, 25 May 2026 11:49:51 +0200 Subject: [PATCH] [codex] Scope console logs to workflow instances (#7535) * Avoid null endpoint DTO metadata in tests * Enforce console logs hub read permission * Remove unused console logs hub import * Support mapped endpoint metadata in auth tests * Reduce console log capture throughput impact * Address Copilot console logs review * Refactor task scheduling to support tenant-level background work and enhance logging functionality. * Introduce ConsoleStreamHook for stdout/stderr tee and enhance logging validation. Adjust test cases and startup warnings for distributed lock provider usage. * Refactor console logging pipeline with capture optimization and new ConsoleLogsHost; update tests accordingly. * Add Ansi SGR parser for console logs and associated unit tests * Remove ANSI color renderings and parsers; integrate ConsoleLogScopeAccessor for improved logging context with workflow instance ID support. * Address console logs code quality feedback * Address PR review feedback * Preserve console logs extension points * Stabilize console logs host lifecycle * Address final automated review comments * Tighten console log capture shutdown * Address console log review feedback * Address follow-up review feedback * Cover final review feedback * Avoid recursive console provider initialization * Guard console host lease shutdown * Preserve console log scope and provider lifetime * Correlate console log scope fallback * Tighten console scope correlation * Expose host services during provider construction * Redact ANSI-normalized console lines --- doc/wiki/workflow-runtime.md | 5 +- src/apps/Elsa.ModularServer.Web/Program.cs | 11 + .../Elsa.ModularServer.Web/appsettings.json | 5 + src/apps/Elsa.Server.Web/Program.cs | 2 +- .../Elsa.Common/Contracts/BackgroundTask.cs | 2 +- .../Options/DistributedLockingOptions.cs | 4 +- .../Features/MultitenancyFeature.cs | 5 +- .../Contracts/ITenantBackgroundWorkQueue.cs | 9 + .../TenantBackgroundWorkQueue.cs | 23 ++ .../TenantBackgroundWorkQueueWorker.cs | 37 ++ .../TenantTaskLifecycleCoordinator.cs | 51 ++- .../ShellFeatures/MultitenancyFeature.cs | 3 + .../Contracts/IConsoleLogCapture.cs | 3 +- .../HostServiceCollectionExtensions.cs | 39 ++ .../Extensions/ServiceCollectionExtensions.cs | 85 +++- .../Models/ConsoleLogFilter.cs | 1 + .../Models/ConsoleLogLine.cs | 1 + .../Options/ConsoleLogsOptions.cs | 8 +- .../Elsa.Diagnostics.ConsoleLogs/README.md | 29 +- .../Services/ConsoleAnsiEscapeSequences.cs | 10 + .../Services/ConsoleCaptureTee.cs | 300 +++++++------- .../Services/ConsoleLineBuffer.cs | 53 ++- .../Services/ConsoleLineFormatter.cs | 4 +- .../Services/ConsoleLogCaptureAdapter.cs | 22 + .../ConsoleLogCaptureHostedService.cs | 30 -- .../Services/ConsoleLogFilterEvaluator.cs | 3 + .../Services/ConsoleLogRedactor.cs | 16 +- .../Services/ConsoleLogScopeAccessor.cs | 217 ++++++++++ .../Services/ConsoleLogsHost.cs | 194 +++++++++ .../Services/ConsoleLogsHostedService.cs | 57 +++ .../ConsoleLogsProviderRegistration.cs | 85 ++++ .../Services/ConsoleStreamHook.cs | 207 ++++++++++ .../Services/ConsoleStreamTeeWriter.cs | 212 ++++++++++ .../HostedServices/AdminUserInitializer.cs | 2 +- .../Features/SchedulingFeature.cs | 5 +- .../Handlers/UpdateTenantSchedules.cs | 17 +- .../CreateSchedulesBackgroundTask.cs | 35 -- .../ShellFeatures/SchedulingFeature.cs | 6 +- .../CreateSchedulesStartupTask.cs | 50 +++ ...DistributedRuntimeLockProviderValidator.cs | 18 +- ...stributedRuntimeLockProviderStartupTask.cs | 2 +- .../TenantTaskLifecycleCoordinatorTests.cs | 110 ++++- .../Capture/ConsoleCaptureTeeTests.cs | 390 ++++++++++++------ .../Capture/ConsoleLineBufferTests.cs | 38 +- .../Capture/ConsoleLineFormatterTests.cs | 14 +- .../ConsoleHostStateCollection.cs | 15 + .../ConsoleLogsRegistrationTests.cs | 193 ++++++++- .../Filtering/ConsoleLogFilterTests.cs | 8 + .../ConsoleLogProviderRedactionTests.cs | 43 +- .../Redaction/ConsoleLogRedactorTests.cs | 20 + .../CreateSchedulesStartupTaskTests.cs | 78 ++++ ...ibutedRuntimeLockProviderValidatorTests.cs | 113 +++-- 52 files changed, 2410 insertions(+), 480 deletions(-) create mode 100644 src/modules/Elsa.Common/Multitenancy/Contracts/ITenantBackgroundWorkQueue.cs create mode 100644 src/modules/Elsa.Common/Multitenancy/Implementations/TenantBackgroundWorkQueue.cs create mode 100644 src/modules/Elsa.Common/Multitenancy/Implementations/TenantBackgroundWorkQueueWorker.cs create mode 100644 src/modules/Elsa.Diagnostics.ConsoleLogs/Extensions/HostServiceCollectionExtensions.cs create mode 100644 src/modules/Elsa.Diagnostics.ConsoleLogs/Services/ConsoleAnsiEscapeSequences.cs create mode 100644 src/modules/Elsa.Diagnostics.ConsoleLogs/Services/ConsoleLogCaptureAdapter.cs delete mode 100644 src/modules/Elsa.Diagnostics.ConsoleLogs/Services/ConsoleLogCaptureHostedService.cs create mode 100644 src/modules/Elsa.Diagnostics.ConsoleLogs/Services/ConsoleLogScopeAccessor.cs create mode 100644 src/modules/Elsa.Diagnostics.ConsoleLogs/Services/ConsoleLogsHost.cs create mode 100644 src/modules/Elsa.Diagnostics.ConsoleLogs/Services/ConsoleLogsHostedService.cs create mode 100644 src/modules/Elsa.Diagnostics.ConsoleLogs/Services/ConsoleLogsProviderRegistration.cs create mode 100644 src/modules/Elsa.Diagnostics.ConsoleLogs/Services/ConsoleStreamHook.cs create mode 100644 src/modules/Elsa.Diagnostics.ConsoleLogs/Services/ConsoleStreamTeeWriter.cs delete mode 100644 src/modules/Elsa.Scheduling/HostedServices/CreateSchedulesBackgroundTask.cs create mode 100644 src/modules/Elsa.Scheduling/StartupTasks/CreateSchedulesStartupTask.cs create mode 100644 test/unit/Elsa.Diagnostics.ConsoleLogs.UnitTests/ConsoleHostStateCollection.cs create mode 100644 test/unit/Elsa.Scheduling.UnitTests/StartupTasks/CreateSchedulesStartupTaskTests.cs diff --git a/doc/wiki/workflow-runtime.md b/doc/wiki/workflow-runtime.md index 889c8146b..33e7f4a6c 100644 --- a/doc/wiki/workflow-runtime.md +++ b/doc/wiki/workflow-runtime.md @@ -172,14 +172,15 @@ Distributed runtime support lives in [Elsa.Workflows.Runtime.Distributed](../../ ### Distributed Lock Provider Safety -The default workflow runtime lock provider is file-system based and writes under `App_Data/locks`. That provider is useful for single-host development and tests, but it is not safe for clustered deployments where nodes have separate file systems. When `UseDistributedRuntime()` is enabled, startup fails if Elsa detects the default file-system provider or the no-op provider unless the host explicitly opts in to local-only lock semantics: +The default workflow runtime lock provider is file-system based and writes under `App_Data/locks`. That provider is useful for single-host development and tests, but it is not safe for clustered deployments where nodes have separate file systems. When `UseDistributedRuntime()` is enabled, Elsa logs a startup warning if it detects the default file-system provider or the no-op provider unless the host explicitly acknowledges local-only lock semantics: ```csharp elsa.UseWorkflowRuntime(runtime => { runtime.UseDistributedRuntime(); - // Single-host development/test only. Do not use this for clustered production deployments. + // Single-host development/test only. Suppresses the startup warning. + // Do not use this for clustered production deployments. runtime.DistributedLockingOptions = options => options.AllowLocalLockProviderInDistributedRuntime = true; }); ``` diff --git a/src/apps/Elsa.ModularServer.Web/Program.cs b/src/apps/Elsa.ModularServer.Web/Program.cs index 4e0505c5a..b8a840d0e 100644 --- a/src/apps/Elsa.ModularServer.Web/Program.cs +++ b/src/apps/Elsa.ModularServer.Web/Program.cs @@ -1,6 +1,8 @@ using CShells.AspNetCore.Configuration; using CShells.AspNetCore.Extensions; using CShells.DependencyInjection; +using Elsa.Diagnostics.ConsoleLogs.Extensions; +using Elsa.Diagnostics.ConsoleLogs.Services; using Elsa.ModularServer.Web; using Elsa.ModularServer.Web.Catalog; using Elsa.ShellFeatures; @@ -13,10 +15,19 @@ using Nuplane; using Nuplane.Loading.Hosting.Builder; using Nuplane.Sources.Directory.Configuration; +// Install the console stream tee as early as possible — before WebApplication.CreateBuilder and +// any logger provider construction — so that downstream loggers (Microsoft.Extensions.Logging.Console, +// Kestrel/Hosting loggers, etc.) capture our tee writer instead of the raw Console.Out / Console.Error. +ConsoleStreamHook.Install(); + var builder = WebApplication.CreateBuilder(args); var services = builder.Services; var configuration = builder.Configuration; +// Console output is a single process-wide resource; the capture pipeline is owned by the root host so +// every shell's diagnostics endpoints (REST + SignalR) read from the same in-memory ring buffer. +services.AddConsoleLogsHost(); + var nuplaneConfiguration = configuration.GetSection("Nuplane"); services.AddNuplane(nuplaneConfiguration, nuplane => diff --git a/src/apps/Elsa.ModularServer.Web/appsettings.json b/src/apps/Elsa.ModularServer.Web/appsettings.json index ae813e68c..3b0d7ad3e 100644 --- a/src/apps/Elsa.ModularServer.Web/appsettings.json +++ b/src/apps/Elsa.ModularServer.Web/appsettings.json @@ -5,6 +5,11 @@ "Microsoft.AspNetCore": "Warning", "Microsoft.EntityFrameworkCore": "Warning", "Nuplane": "Warning" + }, + "Console": { + "FormatterOptions": { + "ColorBehavior": "Enabled" + } } }, "AllowedHosts": "*", diff --git a/src/apps/Elsa.Server.Web/Program.cs b/src/apps/Elsa.Server.Web/Program.cs index 6f3c615d5..b09694ddf 100644 --- a/src/apps/Elsa.Server.Web/Program.cs +++ b/src/apps/Elsa.Server.Web/Program.cs @@ -91,7 +91,7 @@ services runtime.UseEntityFrameworkCore(ef => ef.UseSqlite()); runtime.UseCache(); runtime.UseDistributedRuntime(); - // This sample host uses single-host file-system locks only for development or explicit single-host opt-in. + // This sample host acknowledges single-host file-system locks in development or explicit single-host opt-in. runtime.DistributedLockingOptions = options => options.AllowLocalLockProviderInDistributedRuntime = allowLocalDistributedRuntimeLockProvider; }) .UseWorkflowsApi() diff --git a/src/modules/Elsa.Common/Contracts/BackgroundTask.cs b/src/modules/Elsa.Common/Contracts/BackgroundTask.cs index 758d04516..1da99a40f 100644 --- a/src/modules/Elsa.Common/Contracts/BackgroundTask.cs +++ b/src/modules/Elsa.Common/Contracts/BackgroundTask.cs @@ -5,4 +5,4 @@ public abstract class BackgroundTask : IBackgroundTask public virtual Task ExecuteAsync(CancellationToken cancellationToken) => Task.CompletedTask; public virtual Task StartAsync(CancellationToken cancellationToken) => Task.CompletedTask; public virtual Task StopAsync(CancellationToken cancellationToken) => Task.CompletedTask; -} \ No newline at end of file +} diff --git a/src/modules/Elsa.Common/DistributedHosting/Options/DistributedLockingOptions.cs b/src/modules/Elsa.Common/DistributedHosting/Options/DistributedLockingOptions.cs index 2323304c6..27db12285 100644 --- a/src/modules/Elsa.Common/DistributedHosting/Options/DistributedLockingOptions.cs +++ b/src/modules/Elsa.Common/DistributedHosting/Options/DistributedLockingOptions.cs @@ -14,8 +14,8 @@ public class DistributedLockingOptions public TimeSpan LockAcquisitionTimeout { get; set; } = TimeSpan.FromMinutes(10); /// - /// Allows the distributed workflow runtime to use a lock provider that only coordinates within this node or local file system. - /// Enable this only for single-host development or tests. Clustered deployments should configure a cross-node provider such as Redis, SQL Server, or PostgreSQL. + /// Acknowledges that the distributed workflow runtime is using a lock provider that only coordinates within this node or local file system. + /// Enable this for single-host development or tests to suppress the startup warning. Clustered deployments should configure a cross-node provider such as Redis, SQL Server, or PostgreSQL. /// public bool AllowLocalLockProviderInDistributedRuntime { get; set; } } diff --git a/src/modules/Elsa.Common/Features/MultitenancyFeature.cs b/src/modules/Elsa.Common/Features/MultitenancyFeature.cs index b0d7a1689..d96aceee9 100644 --- a/src/modules/Elsa.Common/Features/MultitenancyFeature.cs +++ b/src/modules/Elsa.Common/Features/MultitenancyFeature.cs @@ -2,6 +2,7 @@ using Elsa.Common.Multitenancy; using Elsa.Common.Multitenancy.EventHandlers; using Elsa.Common.Multitenancy.HostedServices; using Elsa.Common.RecurringTasks; +using Elsa.Extensions; using Elsa.Features.Abstractions; using Elsa.Features.Services; using Microsoft.Extensions.DependencyInjection; @@ -50,8 +51,10 @@ public class MultitenancyFeature(IModule module) : FeatureBase(module) .AddSingleton() .AddScoped() .AddScoped() + .AddScoped() .AddScoped() .AddScoped() + .AddBackgroundTask() .AddScoped(_tenantsProviderFactory); } -} \ No newline at end of file +} diff --git a/src/modules/Elsa.Common/Multitenancy/Contracts/ITenantBackgroundWorkQueue.cs b/src/modules/Elsa.Common/Multitenancy/Contracts/ITenantBackgroundWorkQueue.cs new file mode 100644 index 000000000..b46b1e434 --- /dev/null +++ b/src/modules/Elsa.Common/Multitenancy/Contracts/ITenantBackgroundWorkQueue.cs @@ -0,0 +1,9 @@ +namespace Elsa.Common.Multitenancy; + +public delegate Task TenantBackgroundWorkItem(IServiceProvider serviceProvider, CancellationToken cancellationToken); + +public interface ITenantBackgroundWorkQueue +{ + ValueTask EnqueueAsync(TenantBackgroundWorkItem workItem, CancellationToken cancellationToken = default); + IAsyncEnumerable DequeueAllAsync(CancellationToken cancellationToken = default); +} diff --git a/src/modules/Elsa.Common/Multitenancy/Implementations/TenantBackgroundWorkQueue.cs b/src/modules/Elsa.Common/Multitenancy/Implementations/TenantBackgroundWorkQueue.cs new file mode 100644 index 000000000..6009c4d05 --- /dev/null +++ b/src/modules/Elsa.Common/Multitenancy/Implementations/TenantBackgroundWorkQueue.cs @@ -0,0 +1,23 @@ +using System.Threading.Channels; + +namespace Elsa.Common.Multitenancy; + +public class TenantBackgroundWorkQueue : ITenantBackgroundWorkQueue +{ + private readonly Channel _channel = Channel.CreateUnbounded(new UnboundedChannelOptions + { + SingleReader = true, + SingleWriter = false + }); + + public ValueTask EnqueueAsync(TenantBackgroundWorkItem workItem, CancellationToken cancellationToken = default) + { + ArgumentNullException.ThrowIfNull(workItem); + return _channel.Writer.WriteAsync(workItem, cancellationToken); + } + + public IAsyncEnumerable DequeueAllAsync(CancellationToken cancellationToken = default) + { + return _channel.Reader.ReadAllAsync(cancellationToken); + } +} diff --git a/src/modules/Elsa.Common/Multitenancy/Implementations/TenantBackgroundWorkQueueWorker.cs b/src/modules/Elsa.Common/Multitenancy/Implementations/TenantBackgroundWorkQueueWorker.cs new file mode 100644 index 000000000..2352739c7 --- /dev/null +++ b/src/modules/Elsa.Common/Multitenancy/Implementations/TenantBackgroundWorkQueueWorker.cs @@ -0,0 +1,37 @@ +using Elsa.Common.Helpers; +using Microsoft.Extensions.Logging; + +namespace Elsa.Common.Multitenancy; + +public class TenantBackgroundWorkQueueWorker( + ITenantBackgroundWorkQueue workQueue, + IServiceProvider serviceProvider, + ILogger logger) : BackgroundTask +{ + public override async Task ExecuteAsync(CancellationToken cancellationToken) + { + try + { + await using var enumerator = workQueue.DequeueAllAsync(cancellationToken).GetAsyncEnumerator(cancellationToken); + while (await enumerator.MoveNextAsync()) + { + try + { + await enumerator.Current(serviceProvider, cancellationToken); + } + catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested) + { + return; + } + catch (Exception e) when (!e.IsFatal()) + { + logger.LogError(e, "A tenant background work item failed."); + } + } + } + catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested) + { + logger.LogDebug("Tenant background work queue worker stopped because cancellation was requested."); + } + } +} diff --git a/src/modules/Elsa.Common/Multitenancy/Implementations/TenantTaskLifecycleCoordinator.cs b/src/modules/Elsa.Common/Multitenancy/Implementations/TenantTaskLifecycleCoordinator.cs index 3ec498510..3c3d249c5 100644 --- a/src/modules/Elsa.Common/Multitenancy/Implementations/TenantTaskLifecycleCoordinator.cs +++ b/src/modules/Elsa.Common/Multitenancy/Implementations/TenantTaskLifecycleCoordinator.cs @@ -92,27 +92,38 @@ public class TenantTaskLifecycleCoordinator(RecurringTaskScheduleManager schedul await taskExecutor.ExecuteTaskAsync(task, cancellationToken); } - private Task RunBackgroundTasksAsync(ITenantScope tenantScope, ITaskExecutor taskExecutor, TenantRuntimeState state, CancellationToken cancellationToken) + private async Task RunBackgroundTasksAsync(ITenantScope tenantScope, ITaskExecutor taskExecutor, TenantRuntimeState state, CancellationToken cancellationToken) { - var backgroundTasks = tenantScope.ServiceProvider.GetServices(); + var backgroundTasks = tenantScope.ServiceProvider.GetServices().ToList(); var backgroundTaskStarter = tenantScope.ServiceProvider.GetRequiredService(); var tenantCancellationToken = state.CancellationTokenSource?.Token ?? cancellationToken; foreach (var backgroundTask in backgroundTasks) { - var task = backgroundTaskStarter - .StartAsync(backgroundTask, tenantCancellationToken) - .ContinueWith(_ => taskExecutor.ExecuteTaskAsync(backgroundTask, tenantCancellationToken), - cancellationToken, - TaskContinuationOptions.RunContinuationsAsynchronously, - TaskScheduler.Default) - .Unwrap(); + await backgroundTaskStarter.StartAsync(backgroundTask, tenantCancellationToken); + state.BackgroundTasks.Add(new(backgroundTask, backgroundTaskStarter)); + + var task = RunBackgroundTaskAsync(backgroundTask, taskExecutor, tenantCancellationToken); if (!task.IsCompleted) state.RunningBackgroundTasks.Add(task); } + } - return Task.CompletedTask; + private async Task RunBackgroundTaskAsync(IBackgroundTask backgroundTask, ITaskExecutor taskExecutor, CancellationToken cancellationToken) + { + try + { + await taskExecutor.ExecuteTaskAsync(backgroundTask, cancellationToken); + } + catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested) + { + logger.LogInformation("Background task {TaskType} was cancelled", backgroundTask.GetType().Name); + } + catch (Exception e) when (!e.IsFatal()) + { + logger.LogError(e, "Background task {TaskType} failed with an error", backgroundTask.GetType().Name); + } } private async Task StartRecurringTasksAsync(ITenantScope tenantScope, ITaskExecutor taskExecutor, TenantRuntimeState state, CancellationToken cancellationToken) @@ -225,6 +236,23 @@ public class TenantTaskLifecycleCoordinator(RecurringTaskScheduleManager schedul } state.ScheduledTimers.Clear(); + foreach (var backgroundTask in state.BackgroundTasks) + { + try + { + await backgroundTask.Starter.StopAsync(backgroundTask.Task, cancellationToken); + } + catch (OperationCanceledException) + { + // Expected if caller requested cancellation during tenant deactivation. + } + catch (Exception e) when (!e.IsFatal()) + { + logger.LogError(e, "Failed to stop background task {TaskType}", backgroundTask.Task.GetType().Name); + } + } + state.BackgroundTasks.Clear(); + if (state.RunningBackgroundTasks.Count > 0) { try @@ -290,8 +318,11 @@ public class TenantTaskLifecycleCoordinator(RecurringTaskScheduleManager schedul // Since we exclusively use WaitAsync(), no kernel handle is ever created and disposal is a no-op. public SemaphoreSlim Gate { get; } = new(1, 1); public List RunningBackgroundTasks { get; } = []; + public List BackgroundTasks { get; } = []; public List ScheduledTimers { get; } = []; public List RecurringTasks { get; } = []; public CancellationTokenSource? CancellationTokenSource { get; set; } } + + private record BackgroundTaskRegistration(IBackgroundTask Task, IBackgroundTaskStarter Starter); } diff --git a/src/modules/Elsa.Common/ShellFeatures/MultitenancyFeature.cs b/src/modules/Elsa.Common/ShellFeatures/MultitenancyFeature.cs index da3d0136c..5fc8bc239 100644 --- a/src/modules/Elsa.Common/ShellFeatures/MultitenancyFeature.cs +++ b/src/modules/Elsa.Common/ShellFeatures/MultitenancyFeature.cs @@ -5,6 +5,7 @@ using Elsa.Common.Multitenancy.EventHandlers; using Elsa.Common.Multitenancy.HostedServices; using Elsa.Common.RecurringTasks; using Elsa.Common.ShellHandlers; +using Elsa.Extensions; using Microsoft.Extensions.DependencyInjection; namespace Elsa.Common.ShellFeatures; @@ -35,8 +36,10 @@ public class MultitenancyFeature : IShellFeature .AddSingleton() .AddScoped() .AddScoped() + .AddScoped() .AddScoped() .AddScoped() + .AddBackgroundTask() .AddScoped(_tenantsProviderFactory) // Transient per CShells 0.0.15 convention — the registry resolves IEnumerable and diff --git a/src/modules/Elsa.Diagnostics.ConsoleLogs/Contracts/IConsoleLogCapture.cs b/src/modules/Elsa.Diagnostics.ConsoleLogs/Contracts/IConsoleLogCapture.cs index 8d671d222..4138d9953 100644 --- a/src/modules/Elsa.Diagnostics.ConsoleLogs/Contracts/IConsoleLogCapture.cs +++ b/src/modules/Elsa.Diagnostics.ConsoleLogs/Contracts/IConsoleLogCapture.cs @@ -1,10 +1,9 @@ namespace Elsa.Diagnostics.ConsoleLogs.Contracts; +[Obsolete("Console log capture is host-owned. Use ConsoleLogsHost.Capture for host-level operations.")] public interface IConsoleLogCapture : IAsyncDisposable { ValueTask StartAsync(CancellationToken cancellationToken = default); - ValueTask StopAsync(CancellationToken cancellationToken = default); - ValueTask FlushIdleAsync(CancellationToken cancellationToken = default); } diff --git a/src/modules/Elsa.Diagnostics.ConsoleLogs/Extensions/HostServiceCollectionExtensions.cs b/src/modules/Elsa.Diagnostics.ConsoleLogs/Extensions/HostServiceCollectionExtensions.cs new file mode 100644 index 000000000..7391e8a6d --- /dev/null +++ b/src/modules/Elsa.Diagnostics.ConsoleLogs/Extensions/HostServiceCollectionExtensions.cs @@ -0,0 +1,39 @@ +using Elsa.Diagnostics.ConsoleLogs.Contracts; +using Elsa.Diagnostics.ConsoleLogs.Services; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.DependencyInjection.Extensions; +using Microsoft.Extensions.Logging; + +namespace Elsa.Diagnostics.ConsoleLogs.Extensions; + +public static class HostServiceCollectionExtensions +{ + /// + /// Registers the host-level console capture pipeline. Call this from Program.cs (ideally right + /// after ConsoleStreamHook.Install()) so that the capture singleton, the in-memory ring buffer, + /// the source registry, and the idle flush loop are owned by the root host rather than by any + /// individual shell. + /// + /// + /// Shell-level features that surface console logs (REST endpoints, SignalR hub) still call + /// AddConsoleLogsServices on their own service collection — those registrations resolve to the + /// same host-owned singletons exposed by . + /// + /// + public static IServiceCollection AddConsoleLogsHost(this IServiceCollection services, Action? configure = null) + { + ConsoleStreamHook.Install(); + + if (configure != null) + ConsoleLogsHost.Configure(configure); + + services.TryAddSingleton(_ => ConsoleLogsHost.ScopeAccessor); + services.AddSingleton(sp => sp.GetRequiredService()); +#pragma warning disable CS0618 + services.TryAddSingleton(); +#pragma warning restore CS0618 + services.TryAddEnumerable(ServiceDescriptor.Singleton()); + return services; + } +} + diff --git a/src/modules/Elsa.Diagnostics.ConsoleLogs/Extensions/ServiceCollectionExtensions.cs b/src/modules/Elsa.Diagnostics.ConsoleLogs/Extensions/ServiceCollectionExtensions.cs index 4f373975c..f14b947f5 100644 --- a/src/modules/Elsa.Diagnostics.ConsoleLogs/Extensions/ServiceCollectionExtensions.cs +++ b/src/modules/Elsa.Diagnostics.ConsoleLogs/Extensions/ServiceCollectionExtensions.cs @@ -1,27 +1,94 @@ -using Elsa.Diagnostics.ConsoleLogs.Providers.InMemory; +using Elsa.Diagnostics.ConsoleLogs.Contracts; using Elsa.Diagnostics.ConsoleLogs.RealTime; using Elsa.Diagnostics.ConsoleLogs.Services; using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.DependencyInjection.Extensions; +using Microsoft.Extensions.Hosting; +using Microsoft.Extensions.Logging; +using Microsoft.Extensions.Options; namespace Elsa.Diagnostics.ConsoleLogs.Extensions; public static class ServiceCollectionExtensions { + /// + /// Registers shell-level console-log consumers (SignalR, subscription manager) and exposes the + /// process-wide capture pipeline owned by through DI. Configuration + /// supplied here applies only if the host has not yet been initialized (first-wins). Prefer calling + /// AddConsoleLogsHost from Program.cs for deterministic host-level configuration. + /// public static IServiceCollection AddConsoleLogsServices(this IServiceCollection services, Action? configureOptions = null) { + ConsoleStreamHook.Install(); + var defaultProviderDescriptor = ServiceDescriptor.Singleton(_ => ConsoleLogsHost.Provider); + if (configureOptions != null) - services.Configure(configureOptions); + ConsoleLogsHost.Configure(configureOptions); + + services.TryAdd(defaultProviderDescriptor); + services.TryAddSingleton(new ConsoleLogsProviderRegistration(services, defaultProviderDescriptor)); + + ConsoleLogsProviderRegistration GetRegistration(IServiceProvider serviceProvider) => + serviceProvider.GetRequiredService(); + + void ConfigureCustomProvider(IServiceProvider serviceProvider) => + GetRegistration(serviceProvider).ConfigureProvider(serviceProvider); services.AddSignalR(); - services.AddOptions(); - services.TryAddSingleton(); - services.TryAddSingleton(); - services.TryAddSingleton(); - services.TryAddSingleton(); + + // Every shell resolves the same process-wide singletons — console output is a shared OS resource. + services.TryAddSingleton>(sp => + { + if (GetRegistration(sp).TryGetHostOptions(out var options)) + return options; + + ConfigureCustomProvider(sp); + return ConsoleLogsHost.Options; + }); + services.TryAddSingleton(sp => + { + if (GetRegistration(sp).TryGetHostSourceRegistry(out var sourceRegistry)) + return sourceRegistry; + + ConfigureCustomProvider(sp); + return ConsoleLogsHost.SourceRegistry; + }); + services.TryAddSingleton(sp => + { + if (GetRegistration(sp).TryGetHostRedactor(out var redactor)) + return redactor; + + ConfigureCustomProvider(sp); + return ConsoleLogsHost.Redactor; + }); + services.TryAddSingleton(sp => + { + if (GetRegistration(sp).TryGetHostFormatter(out var formatter)) + return formatter; + + ConfigureCustomProvider(sp); + return ConsoleLogsHost.Formatter; + }); + services.TryAddSingleton(sp => + { + if (GetRegistration(sp).TryGetHostScopeAccessor(out var scopeAccessor)) + return scopeAccessor; + + ConfigureCustomProvider(sp); + return ConsoleLogsHost.ScopeAccessor; + }); + services.AddSingleton(sp => + { + ConfigureCustomProvider(sp); + return ConsoleLogsHost.ScopeAccessor; + }); +#pragma warning disable CS0618 + services.TryAddSingleton(); +#pragma warning restore CS0618 + + // The subscription manager wraps a per-shell SignalR hub context, so it must be per-shell. services.TryAddSingleton(); - services.TryAddSingleton(); - services.AddHostedService(); + services.TryAddEnumerable(ServiceDescriptor.Singleton()); return services; } diff --git a/src/modules/Elsa.Diagnostics.ConsoleLogs/Models/ConsoleLogFilter.cs b/src/modules/Elsa.Diagnostics.ConsoleLogs/Models/ConsoleLogFilter.cs index 7415394f5..44f55afa4 100644 --- a/src/modules/Elsa.Diagnostics.ConsoleLogs/Models/ConsoleLogFilter.cs +++ b/src/modules/Elsa.Diagnostics.ConsoleLogs/Models/ConsoleLogFilter.cs @@ -5,6 +5,7 @@ public record ConsoleLogFilter public string? SourceId { get; init; } public ConsoleLogStream? Stream { get; init; } public string? Query { get; init; } + public string? WorkflowInstanceId { get; init; } public DateTimeOffset? From { get; init; } public DateTimeOffset? To { get; init; } public int? Limit { get; init; } diff --git a/src/modules/Elsa.Diagnostics.ConsoleLogs/Models/ConsoleLogLine.cs b/src/modules/Elsa.Diagnostics.ConsoleLogs/Models/ConsoleLogLine.cs index 836bef7a0..df1736442 100644 --- a/src/modules/Elsa.Diagnostics.ConsoleLogs/Models/ConsoleLogLine.cs +++ b/src/modules/Elsa.Diagnostics.ConsoleLogs/Models/ConsoleLogLine.cs @@ -9,6 +9,7 @@ public record ConsoleLogLine public ConsoleLogStream Stream { get; init; } public string Text { get; init; } = string.Empty; public ConsoleLogSource Source { get; init; } = default!; + public string? WorkflowInstanceId { get; init; } public bool Truncated { get; init; } public ConsoleLogDroppedSummary? Dropped { get; init; } } diff --git a/src/modules/Elsa.Diagnostics.ConsoleLogs/Options/ConsoleLogsOptions.cs b/src/modules/Elsa.Diagnostics.ConsoleLogs/Options/ConsoleLogsOptions.cs index d488b3843..46f74a1e4 100644 --- a/src/modules/Elsa.Diagnostics.ConsoleLogs/Options/ConsoleLogsOptions.cs +++ b/src/modules/Elsa.Diagnostics.ConsoleLogs/Options/ConsoleLogsOptions.cs @@ -8,7 +8,13 @@ public class ConsoleLogsOptions public int MaxRecentQuerySize { get; set; } = 1_000; public int MaxLineLength { get; set; } = 16_384; public TimeSpan IdleFlushTimeout { get; set; } = TimeSpan.FromSeconds(1); - public bool StripAnsiEscapeSequences { get; set; } = true; + + /// + /// When true, ANSI escape sequences (colors, cursor commands, etc.) are removed before the captured + /// line is published. The default is false so the wire payload preserves the original bytes the + /// process wrote; consumers (e.g., the Studio) decide whether to render or strip them client-side. + /// + public bool StripAnsiEscapeSequences { get; set; } = false; public TimeSpan SourceHeartbeatTimeout { get; set; } = TimeSpan.FromSeconds(30); public bool IncludeConsoleLogsInternalLogs { get; set; } public string RedactionReplacement { get; set; } = "[Redacted]"; diff --git a/src/modules/Elsa.Diagnostics.ConsoleLogs/README.md b/src/modules/Elsa.Diagnostics.ConsoleLogs/README.md index 37f99faaf..2fdb2ba8f 100644 --- a/src/modules/Elsa.Diagnostics.ConsoleLogs/README.md +++ b/src/modules/Elsa.Diagnostics.ConsoleLogs/README.md @@ -22,7 +22,7 @@ services.AddElsa(elsa => options.SubscriberChannelCapacity = 1_000; options.MaxRecentQuerySize = 1_000; options.MaxLineLength = 16_384; - options.StripAnsiEscapeSequences = true; + options.StripAnsiEscapeSequences = false; // pass colors through to consumers }); }); ``` @@ -33,6 +33,31 @@ Map the live hub: app.UseConsoleLogs(); ``` +## Capturing Colors + +The capture passes raw stdout/stderr bytes through to consumers. ANSI escape sequences (colors, cursor +moves) are preserved by default; consumers (e.g. the Studio's `Raw ANSI` toggle) decide whether to render +them or strip them on display. + +Whether colors actually arrive at the capture depends on whether the .NET console logger emits them. +`Microsoft.Extensions.Logging.Console`'s `SimpleConsoleFormatter` defaults to `LoggerColorBehavior.Default`, +which suppresses colors whenever `Console.IsOutputRedirected` is `true` — that includes most Docker/Kubernetes +deployments, IDE-hosted runs (JetBrains Rider, VS Code), and any piped stdout. To force colors regardless, +add the following to `appsettings.json` (no module dependency required): + +```json +"Logging": { + "Console": { + "FormatterOptions": { + "ColorBehavior": "Enabled" + } + } +} +``` + +With `ColorBehavior = Enabled`, the logger writes ANSI sequences such as `\x1b[32m` and `\x1b[0m` into the +tee, the capture publishes them verbatim, and the Studio renders them. + ## Contracts - Recent lines: `POST /diagnostics/console-logs/recent` @@ -45,7 +70,7 @@ Recent, source, and hub access all require the same permission. ## Safety Boundaries - Redaction runs before recent buffering, live streaming, endpoint responses, and provider storage. -- ANSI escape sequences are stripped by default. +- ANSI escape sequences are preserved by default; set `StripAnsiEscapeSequences = true` to strip them server-side. - Partial writes are buffered until newline, max line length, or idle flush. - Oversized lines are truncated to one event and marked as truncated. - Providers receive only redacted line text and redacted source metadata. diff --git a/src/modules/Elsa.Diagnostics.ConsoleLogs/Services/ConsoleAnsiEscapeSequences.cs b/src/modules/Elsa.Diagnostics.ConsoleLogs/Services/ConsoleAnsiEscapeSequences.cs new file mode 100644 index 000000000..8dea2fadc --- /dev/null +++ b/src/modules/Elsa.Diagnostics.ConsoleLogs/Services/ConsoleAnsiEscapeSequences.cs @@ -0,0 +1,10 @@ +using System.Text.RegularExpressions; + +namespace Elsa.Diagnostics.ConsoleLogs.Services; + +internal static class ConsoleAnsiEscapeSequences +{ + private static readonly Regex AnsiRegex = new(@"\x1B(?:[@-Z\\-_]|\[[0-?]*[ -/]*[@-~])", RegexOptions.Compiled | RegexOptions.CultureInvariant); + + public static string Strip(string text) => AnsiRegex.Replace(text, string.Empty); +} diff --git a/src/modules/Elsa.Diagnostics.ConsoleLogs/Services/ConsoleCaptureTee.cs b/src/modules/Elsa.Diagnostics.ConsoleLogs/Services/ConsoleCaptureTee.cs index 3bd64ac61..e04a08201 100644 --- a/src/modules/Elsa.Diagnostics.ConsoleLogs/Services/ConsoleCaptureTee.cs +++ b/src/modules/Elsa.Diagnostics.ConsoleLogs/Services/ConsoleCaptureTee.cs @@ -1,151 +1,179 @@ -using System.Text; using System.Threading.Channels; using Microsoft.Extensions.Options; namespace Elsa.Diagnostics.ConsoleLogs.Services; -public class ConsoleCaptureTee( - IConsoleLogProvider provider, - IConsoleLogSourceRegistry sourceRegistry, - IConsoleLogRedactor redactor, - ConsoleLineFormatter formatter, - IOptions options) : TextWriter, IConsoleLogCapture +/// +/// Process-wide adapter that turns raw stdout/stderr writes captured by +/// into fully formed records, then publishes them to the configured +/// on a single background task. +/// +/// +/// Owned by — there is one instance per process. Capture is wired up in the +/// constructor so that lines emitted during host startup — before any hosted service has run — are not +/// lost. The capture pipeline can never block the caller of Console.Write*: when the publish +/// channel is full, the current line is reported as dropped to the provider. +/// +/// +/// +/// To prevent the publish path itself from feeding new lines back into capture (e.g. when the in-memory +/// provider's SignalR fan-out logs internally), is set on +/// the publisher's async context for the lifetime of each +/// invocation. +/// +/// +public sealed class ConsoleCaptureTee : IAsyncDisposable, IDisposable { - private readonly object _lock = new(); - private readonly ConsoleLogsOptions _options = options.Value; - private readonly ConsoleLineBuffer _stdoutBuffer = new(options); - private readonly ConsoleLineBuffer _stderrBuffer = new(options); - private TextWriter? _originalOut; - private TextWriter? _originalError; - private Channel? _publishChannel; - private CancellationTokenSource? _publishCancellation; - private Task? _publishTask; - private int _startCount; + private readonly IConsoleLogProvider _provider; + private readonly IConsoleLogSourceRegistry _sourceRegistry; + private readonly IConsoleLogRedactor _redactor; + private readonly ConsoleLineFormatter _formatter; + private readonly ConsoleLogScopeAccessor _scopeAccessor; + private readonly ConsoleLineBuffer _stdoutBuffer; + private readonly ConsoleLineBuffer _stderrBuffer; + private readonly object _bufferLock = new(); + private readonly Channel _publishChannel; + private readonly CancellationTokenSource _shutdownCts = new(); + private readonly Task _publishTask; + private readonly IDisposable _subscription; private long _sequence; + private int _disposed; - public override Encoding Encoding => _originalOut?.Encoding ?? Encoding.UTF8; - - public ValueTask StartAsync(CancellationToken cancellationToken = default) + public ConsoleCaptureTee( + IConsoleLogProvider provider, + IConsoleLogSourceRegistry sourceRegistry, + IConsoleLogRedactor redactor, + ConsoleLineFormatter formatter, + ConsoleLogScopeAccessor scopeAccessor, + IOptions options) { - lock (_lock) + _provider = provider; + _sourceRegistry = sourceRegistry; + _redactor = redactor; + _formatter = formatter; + _scopeAccessor = scopeAccessor; + _stdoutBuffer = new(options); + _stderrBuffer = new(options); + + _publishChannel = Channel.CreateBounded(new BoundedChannelOptions(Math.Max(1, options.Value.CaptureChannelCapacity)) { - if (_startCount++ > 0) - return ValueTask.CompletedTask; + FullMode = BoundedChannelFullMode.Wait, + SingleReader = true, + SingleWriter = false + }); - _publishCancellation = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken); - _publishChannel = Channel.CreateBounded(new BoundedChannelOptions(Math.Max(1, _options.CaptureChannelCapacity)) - { - FullMode = BoundedChannelFullMode.Wait, - SingleReader = true, - SingleWriter = false - }); - _publishTask = PublishQueuedLinesAsync(_publishChannel.Reader, _publishCancellation.Token); - _originalOut = Console.Out; - _originalError = Console.Error; - Console.SetOut(new TeeWriter(_originalOut, this, ConsoleLogStream.Stdout)); - Console.SetError(new TeeWriter(_originalError, this, ConsoleLogStream.Stderr)); - } + _publishTask = Task.Run(() => PublishQueuedLinesAsync(_publishChannel.Reader, _shutdownCts.Token)); - return ValueTask.CompletedTask; - } - - public ValueTask StopAsync(CancellationToken cancellationToken = default) - { - Task? publishTask; - CancellationTokenSource? publishCancellation; - - lock (_lock) - { - if (_startCount == 0) - return ValueTask.CompletedTask; - - if (--_startCount > 0) - return ValueTask.CompletedTask; - - if (_originalOut != null) - Console.SetOut(_originalOut); - - if (_originalError != null) - Console.SetError(_originalError); - - FlushRemaining(ConsoleLogStream.Stdout); - FlushRemaining(ConsoleLogStream.Stderr); - _publishChannel?.Writer.TryComplete(); - publishTask = _publishTask; - publishCancellation = _publishCancellation; - _originalOut = null; - _originalError = null; - _publishChannel = null; - _publishCancellation = null; - _publishTask = null; - } - - return AwaitPublisherAsync(publishTask, publishCancellation, cancellationToken); - } - - public override void Write(char value) - { - Capture(ConsoleLogStream.Stdout, value.ToString()); - } - - public override ValueTask DisposeAsync() - { - return StopAsync(); + // Subscribe eagerly so that any output produced between this constructor and StartAsync is captured + // (the hosted service starts the flush loop, not the capture itself). + _subscription = ConsoleStreamHook.Subscribe(OnChunkCaptured); } public ValueTask FlushIdleAsync(CancellationToken cancellationToken = default) { - foreach (var stream in new[] { ConsoleLogStream.Stdout, ConsoleLogStream.Stderr }) - { - string? line; - lock (_lock) - line = GetBuffer(stream).FlushIfIdle(DateTimeOffset.UtcNow); + var now = DateTimeOffset.UtcNow; - if (line != null) - Publish(stream, line); - } + Publish(ConsoleLogStream.Stdout, FlushIfIdle(_stdoutBuffer, now), now); + Publish(ConsoleLogStream.Stderr, FlushIfIdle(_stderrBuffer, now), now); return ValueTask.CompletedTask; } - private void Capture(ConsoleLogStream stream, string value) + public async ValueTask DisposeAsync() { - IReadOnlyCollection lines; - lock (_lock) - lines = GetBuffer(stream).Append(value, DateTimeOffset.UtcNow); + await DisposeAsync(CancellationToken.None).ConfigureAwait(false); + } + + public async ValueTask DisposeAsync(CancellationToken cancellationToken) + { + if (Interlocked.Exchange(ref _disposed, 1) != 0) + return; + + _subscription.Dispose(); + + // Drain any pending buffered text before completing the channel. + FlushRemaining(ConsoleLogStream.Stdout); + FlushRemaining(ConsoleLogStream.Stderr); + + _publishChannel.Writer.TryComplete(); + + try + { + await _publishTask.WaitAsync(cancellationToken).ConfigureAwait(false); + } + catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested) + { + _shutdownCts.Cancel(); + } + catch (OperationCanceledException) + { + // Expected during shutdown. + } + + _shutdownCts.Dispose(); + } + + void IDisposable.Dispose() + { + // MS DI requires IDisposable on registered singletons even when the service exposes IAsyncDisposable. + // Bridge to DisposeAsync so the publish task is drained instead of orphaned. + DisposeAsync().AsTask().GetAwaiter().GetResult(); + } + + private void OnChunkCaptured(CapturedChunk chunk) + { + IReadOnlyCollection lines; + + var workflowInstanceId = _scopeAccessor.GetWorkflowInstanceId(); + + lock (_bufferLock) + lines = GetBuffer(chunk.Stream).Append(chunk.Text, chunk.TimestampUtc, workflowInstanceId); foreach (var line in lines) - Publish(stream, line); + Publish(chunk.Stream, line, chunk.TimestampUtc); + } + + private BufferedConsoleLine? FlushIfIdle(ConsoleLineBuffer buffer, DateTimeOffset now) + { + lock (_bufferLock) + return buffer.FlushIfIdle(now); } private void FlushRemaining(ConsoleLogStream stream) { - var line = GetBuffer(stream).Flush(); - if (line != null) - Publish(stream, line); + BufferedConsoleLine? line; + lock (_bufferLock) + line = GetBuffer(stream).Flush(); + + Publish(stream, line, DateTimeOffset.UtcNow); } - private ConsoleLineBuffer GetBuffer(ConsoleLogStream stream) => stream == ConsoleLogStream.Stdout ? _stdoutBuffer : _stderrBuffer; + private ConsoleLineBuffer GetBuffer(ConsoleLogStream stream) => + stream == ConsoleLogStream.Stdout ? _stdoutBuffer : _stderrBuffer; - private void Publish(ConsoleLogStream stream, string text) + private void Publish(ConsoleLogStream stream, BufferedConsoleLine? capturedLine, DateTimeOffset timestamp) { - var formatted = formatter.Format(text); - var now = DateTimeOffset.UtcNow; + if (capturedLine == null) + return; + + var text = capturedLine.Value.Text; + var workflowInstanceId = capturedLine.Value.WorkflowInstanceId ?? _scopeAccessor.TakeLoggedWorkflowInstanceId(text, timestamp); + var formatted = _formatter.Format(text); var line = new ConsoleLogLine { - Timestamp = now, - ReceivedAt = now, + Timestamp = timestamp, + ReceivedAt = timestamp, Sequence = Interlocked.Increment(ref _sequence), Stream = stream, Text = formatted.Text, - Source = sourceRegistry.Current, + Source = _sourceRegistry.Current, + WorkflowInstanceId = workflowInstanceId, Truncated = formatted.Truncated }; - if (_publishChannel?.Writer.TryWrite(line) != false) - return; - - if (provider is IConsoleLogDroppedLineReporter reporter) + // The capture path must not block console writers. When the bounded channel is full, drop this line + // and surface a summary so the UI can warn the operator. + if (!_publishChannel.Writer.TryWrite(line) && _provider is IConsoleLogDroppedLineReporter reporter) reporter.ReportDropped(new ConsoleLogDroppedSummary(line.Source.Id, stream, "CaptureChannelFull", 1)); } @@ -153,11 +181,14 @@ public class ConsoleCaptureTee( { try { - await foreach (var line in reader.ReadAllAsync(cancellationToken)) + await foreach (var line in reader.ReadAllAsync(cancellationToken).ConfigureAwait(false)) { + // Suppress capture on this async context so that any writes the provider performs + // (directly or via SignalR fan-out it invokes synchronously) do not loop back into us. + ConsoleStreamHook.SuppressCapture = true; try { - await provider.PublishAsync(redactor.Redact(line), cancellationToken); + await _provider.PublishAsync(_redactor.Redact(line), cancellationToken).ConfigureAwait(false); } catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested) { @@ -165,54 +196,17 @@ public class ConsoleCaptureTee( } catch (Exception) { - // Avoid logging from the console capture path; doing so would recurse through the same tee. + // Never log from the publish path — that would route right back through the same tee. + } + finally + { + ConsoleStreamHook.SuppressCapture = false; } } } catch (OperationCanceledException) { - // Normal shutdown completes or cancels the publish pump. - } - } - - private static async ValueTask AwaitPublisherAsync(Task? publishTask, CancellationTokenSource? publishCancellation, CancellationToken cancellationToken) - { - if (publishTask == null) - return; - - using var cancellation = publishCancellation; - - try - { - await publishTask.WaitAsync(cancellationToken); - } - catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested) - { - cancellation?.Cancel(); - throw; - } - } - - private sealed class TeeWriter(TextWriter original, ConsoleCaptureTee capture, ConsoleLogStream stream) : TextWriter - { - public override Encoding Encoding => original.Encoding; - - public override void Write(char value) - { - original.Write(value); - capture.Capture(stream, value.ToString()); - } - - public override void Write(string? value) - { - original.Write(value); - if (value != null) - capture.Capture(stream, value); - } - - public override void Flush() - { - original.Flush(); + // Normal shutdown. } } } diff --git a/src/modules/Elsa.Diagnostics.ConsoleLogs/Services/ConsoleLineBuffer.cs b/src/modules/Elsa.Diagnostics.ConsoleLogs/Services/ConsoleLineBuffer.cs index 98df9ed72..01ce458fa 100644 --- a/src/modules/Elsa.Diagnostics.ConsoleLogs/Services/ConsoleLineBuffer.cs +++ b/src/modules/Elsa.Diagnostics.ConsoleLogs/Services/ConsoleLineBuffer.cs @@ -3,37 +3,56 @@ using Microsoft.Extensions.Options; namespace Elsa.Diagnostics.ConsoleLogs.Services; +/// +/// Reassembles arbitrary stdout/stderr write chunks into complete lines. Lines are emitted whenever a +/// terminator (\n) is encountered, when the buffer reaches , +/// or when is called after the configured idle timeout has elapsed. +/// Empty lines (a bare \n with no preceding content on the current logical line) are preserved. +/// public class ConsoleLineBuffer(IOptions options) { private readonly StringBuilder _buffer = new(); private readonly ConsoleLogsOptions _options = options.Value; private DateTimeOffset? _lastWriteAt; + private bool _logicalLineHasContent; + private bool _workflowInstanceIdCaptured; + private string? _workflowInstanceId; - public IReadOnlyCollection Append(string value, DateTimeOffset now) + public IReadOnlyCollection Append(string value, DateTimeOffset now, string? workflowInstanceId = null) { - _lastWriteAt = now; - var lines = new List(); + var lines = new List(); - foreach (var ch in value.Where(x => x != '\r')) + foreach (var ch in value) { + if (ch == '\r') + continue; + if (ch == '\n') { if (_buffer.Length > 0) lines.Add(FlushBuffer()); + else if (!_logicalLineHasContent) + lines.Add(new(string.Empty, NormalizeWorkflowInstanceId(workflowInstanceId))); + _logicalLineHasContent = false; continue; } + CaptureScope(workflowInstanceId); _buffer.Append(ch); + _logicalLineHasContent = true; if (_buffer.Length >= _options.MaxLineLength) lines.Add(FlushBuffer()); } + // Track last activity only when characters remain pending so that FlushIfIdle has a deadline to act on. + _lastWriteAt = _buffer.Length > 0 ? now : null; + return lines; } - public string? FlushIfIdle(DateTimeOffset now) + public BufferedConsoleLine? FlushIfIdle(DateTimeOffset now) { if (_buffer.Length == 0 || _lastWriteAt == null) return null; @@ -41,16 +60,32 @@ public class ConsoleLineBuffer(IOptions options) return now - _lastWriteAt >= _options.IdleFlushTimeout ? FlushBuffer() : null; } - public string? Flush() + public BufferedConsoleLine? Flush() { return _buffer.Length == 0 ? null : FlushBuffer(); } - private string FlushBuffer() + private void CaptureScope(string? workflowInstanceId) + { + if (_workflowInstanceIdCaptured) + return; + + _workflowInstanceIdCaptured = true; + _workflowInstanceId = NormalizeWorkflowInstanceId(workflowInstanceId); + } + + private static string? NormalizeWorkflowInstanceId(string? workflowInstanceId) => + string.IsNullOrWhiteSpace(workflowInstanceId) ? null : workflowInstanceId; + + private BufferedConsoleLine FlushBuffer() { var line = _buffer.ToString(); + var workflowInstanceId = _workflowInstanceId; _buffer.Clear(); - _lastWriteAt = null; - return line; + _workflowInstanceIdCaptured = false; + _workflowInstanceId = null; + return new(line, workflowInstanceId); } } + +public readonly record struct BufferedConsoleLine(string Text, string? WorkflowInstanceId); diff --git a/src/modules/Elsa.Diagnostics.ConsoleLogs/Services/ConsoleLineFormatter.cs b/src/modules/Elsa.Diagnostics.ConsoleLogs/Services/ConsoleLineFormatter.cs index 674ca6165..b562901c4 100644 --- a/src/modules/Elsa.Diagnostics.ConsoleLogs/Services/ConsoleLineFormatter.cs +++ b/src/modules/Elsa.Diagnostics.ConsoleLogs/Services/ConsoleLineFormatter.cs @@ -1,17 +1,15 @@ -using System.Text.RegularExpressions; using Microsoft.Extensions.Options; namespace Elsa.Diagnostics.ConsoleLogs.Services; public class ConsoleLineFormatter(IOptions options) { - private static readonly Regex AnsiRegex = new(@"\x1B(?:[@-Z\\-_]|\[[0-?]*[ -/]*[@-~])", RegexOptions.Compiled | RegexOptions.CultureInvariant); private readonly ConsoleLogsOptions _options = options.Value; public FormattedConsoleLine Format(string text) { if (_options.StripAnsiEscapeSequences) - text = AnsiRegex.Replace(text, string.Empty); + text = ConsoleAnsiEscapeSequences.Strip(text); var truncated = text.Length > _options.MaxLineLength; if (truncated) diff --git a/src/modules/Elsa.Diagnostics.ConsoleLogs/Services/ConsoleLogCaptureAdapter.cs b/src/modules/Elsa.Diagnostics.ConsoleLogs/Services/ConsoleLogCaptureAdapter.cs new file mode 100644 index 000000000..40729a45d --- /dev/null +++ b/src/modules/Elsa.Diagnostics.ConsoleLogs/Services/ConsoleLogCaptureAdapter.cs @@ -0,0 +1,22 @@ +using Elsa.Diagnostics.ConsoleLogs.Contracts; + +namespace Elsa.Diagnostics.ConsoleLogs.Services; + +#pragma warning disable CS0618 +[Obsolete("Console log capture is host-owned. Use ConsoleLogsHost.Capture for host-level operations.")] +public sealed class ConsoleLogCaptureAdapter : IConsoleLogCapture +#pragma warning restore CS0618 +{ + public ValueTask StartAsync(CancellationToken cancellationToken = default) + { + ConsoleLogsHost.EnsureInitialized(); + return ValueTask.CompletedTask; + } + + public ValueTask StopAsync(CancellationToken cancellationToken = default) => ValueTask.CompletedTask; + + public ValueTask FlushIdleAsync(CancellationToken cancellationToken = default) => + ConsoleLogsHost.Capture.FlushIdleAsync(cancellationToken); + + public ValueTask DisposeAsync() => ValueTask.CompletedTask; +} diff --git a/src/modules/Elsa.Diagnostics.ConsoleLogs/Services/ConsoleLogCaptureHostedService.cs b/src/modules/Elsa.Diagnostics.ConsoleLogs/Services/ConsoleLogCaptureHostedService.cs deleted file mode 100644 index 5cf68d18b..000000000 --- a/src/modules/Elsa.Diagnostics.ConsoleLogs/Services/ConsoleLogCaptureHostedService.cs +++ /dev/null @@ -1,30 +0,0 @@ -using Microsoft.Extensions.Hosting; -using Microsoft.Extensions.Options; - -namespace Elsa.Diagnostics.ConsoleLogs.Services; - -public class ConsoleLogCaptureHostedService(IConsoleLogCapture capture, IOptions options) : BackgroundService -{ - public override async Task StartAsync(CancellationToken cancellationToken) - { - await capture.StartAsync(cancellationToken); - await base.StartAsync(cancellationToken); - } - - public override async Task StopAsync(CancellationToken cancellationToken) - { - await base.StopAsync(cancellationToken); - await capture.StopAsync(cancellationToken); - } - - protected override async Task ExecuteAsync(CancellationToken stoppingToken) - { - var interval = TimeSpan.FromMilliseconds(Math.Max(100, options.Value.IdleFlushTimeout.TotalMilliseconds / 2)); - - while (!stoppingToken.IsCancellationRequested) - { - await Task.Delay(interval, stoppingToken); - await capture.FlushIdleAsync(stoppingToken); - } - } -} diff --git a/src/modules/Elsa.Diagnostics.ConsoleLogs/Services/ConsoleLogFilterEvaluator.cs b/src/modules/Elsa.Diagnostics.ConsoleLogs/Services/ConsoleLogFilterEvaluator.cs index 04b9a7244..2e018ddfd 100644 --- a/src/modules/Elsa.Diagnostics.ConsoleLogs/Services/ConsoleLogFilterEvaluator.cs +++ b/src/modules/Elsa.Diagnostics.ConsoleLogs/Services/ConsoleLogFilterEvaluator.cs @@ -13,6 +13,9 @@ public static class ConsoleLogFilterEvaluator if (!ContainsText(line, filter.Query)) return false; + if (!EqualsFilter(line.WorkflowInstanceId, filter.WorkflowInstanceId)) + return false; + if (filter.From is { } from && line.ReceivedAt < from) return false; diff --git a/src/modules/Elsa.Diagnostics.ConsoleLogs/Services/ConsoleLogRedactor.cs b/src/modules/Elsa.Diagnostics.ConsoleLogs/Services/ConsoleLogRedactor.cs index 6f87d4a39..10b769839 100644 --- a/src/modules/Elsa.Diagnostics.ConsoleLogs/Services/ConsoleLogRedactor.cs +++ b/src/modules/Elsa.Diagnostics.ConsoleLogs/Services/ConsoleLogRedactor.cs @@ -49,8 +49,22 @@ public class ConsoleLogRedactor(IOptions options) : IConsole if (IsSensitiveName(name)) return _options.RedactionReplacement; - return _sensitiveTextPatterns.Aggregate(value, (current, pattern) => pattern.Replace(current, _options.RedactionReplacement)); + return RedactSensitiveText(value); } + private string RedactSensitiveText(string value) + { + var redacted = ApplySensitiveTextPatterns(value); + var normalized = ConsoleAnsiEscapeSequences.Strip(value); + if (normalized == value) + return redacted; + + var normalizedRedacted = ApplySensitiveTextPatterns(normalized); + return normalizedRedacted == normalized ? redacted : normalizedRedacted; + } + + private string ApplySensitiveTextPatterns(string value) => + _sensitiveTextPatterns.Aggregate(value, (current, pattern) => pattern.Replace(current, _options.RedactionReplacement)); + private bool IsSensitiveName(string name) => _sensitiveNames.Any(sensitiveName => name.Contains(sensitiveName, StringComparison.OrdinalIgnoreCase)); } diff --git a/src/modules/Elsa.Diagnostics.ConsoleLogs/Services/ConsoleLogScopeAccessor.cs b/src/modules/Elsa.Diagnostics.ConsoleLogs/Services/ConsoleLogScopeAccessor.cs new file mode 100644 index 000000000..7812df454 --- /dev/null +++ b/src/modules/Elsa.Diagnostics.ConsoleLogs/Services/ConsoleLogScopeAccessor.cs @@ -0,0 +1,217 @@ +using System.Collections; +using System.Linq; +using Microsoft.Extensions.Logging; + +namespace Elsa.Diagnostics.ConsoleLogs.Services; + +public sealed class ConsoleLogScopeAccessor : ILoggerProvider, ISupportExternalScope +{ + private const string WorkflowInstanceIdKey = "WorkflowInstanceId"; + private const int PendingWorkflowLogsCapacity = 1024; + private static readonly TimeSpan PendingWorkflowLogTtl = TimeSpan.FromSeconds(30); + private readonly IExternalScopeProvider _defaultScopeProvider = new LoggerExternalScopeProvider(); + private readonly object _lock = new(); + private readonly object _pendingWorkflowLogsLock = new(); + private readonly List _pendingWorkflowLogs = []; + private WeakReference[] _scopeProviders; + + public ConsoleLogScopeAccessor() + { + _scopeProviders = [new(_defaultScopeProvider)]; + } + + public ILogger CreateLogger(string categoryName) => new ScopeCapturingLogger(this); + public void Dispose() + { + } + + public void SetScopeProvider(IExternalScopeProvider scopeProvider) + { + lock (_lock) + { + var scopeProviders = GetLiveScopeProviders(); + if (scopeProviders.Any(x => ReferenceEquals(x, scopeProvider))) + return; + + _scopeProviders = [.. _scopeProviders, new WeakReference(scopeProvider)]; + } + } + + public string? GetWorkflowInstanceId() + { + var scopeProviders = GetLiveScopeProviders(); + string? value = null; + + foreach (var scopeProvider in scopeProviders) + { + var state = new ScopeSearchState(WorkflowInstanceIdKey); + scopeProvider.ForEachScope((scope, searchState) => searchState.Capture(scope), state); + if (state.Value != null) + value = state.Value; + } + + return value; + } + + internal string? TakeLoggedWorkflowInstanceId(string text, DateTimeOffset capturedAt) + { + if (string.IsNullOrWhiteSpace(text)) + return null; + + lock (_pendingWorkflowLogsLock) + { + PruneExpiredPendingWorkflowLogs(capturedAt); + + var index = _pendingWorkflowLogs.FindIndex(x => IsCorrelated(text, x.Message)); + if (index < 0) + return null; + + var workflowInstanceId = _pendingWorkflowLogs[index].WorkflowInstanceId; + _pendingWorkflowLogs.RemoveAt(index); + return workflowInstanceId; + } + } + + private IExternalScopeProvider[] GetLiveScopeProviders() + { + lock (_lock) + { + var liveScopeProviders = _scopeProviders + .Select(reference => (Reference: reference, Provider: reference.TryGetTarget(out var provider) ? provider : null)) + .Where(x => x.Provider != null) + .ToArray(); + + if (liveScopeProviders.Length != _scopeProviders.Length) + _scopeProviders = [.. liveScopeProviders.Select(x => x.Reference)]; + + return [.. liveScopeProviders.Select(x => x.Provider!)]; + } + } + + private sealed class ScopeSearchState(string key) + { + public string? Value { get; private set; } + + public void Capture(object? scope) + { + var value = TryGetScopeValue(scope, key); + if (value != null) + Value = value; + } + } + + private static string? TryGetScopeValue(object? scope, string key) + { + if (scope is not IEnumerable values) + return null; + + foreach (var value in values) + { + if (value == null) + continue; + + if (value is KeyValuePair kvp) + { + if (string.Equals(kvp.Key, key, StringComparison.OrdinalIgnoreCase)) + return kvp.Value?.ToString(); + + continue; + } + + var type = value.GetType(); + if (!type.IsGenericType || type.GetGenericTypeDefinition() != typeof(KeyValuePair<,>)) + continue; + + var itemKey = type.GetProperty(nameof(KeyValuePair.Key))?.GetValue(value) as string; + if (!string.Equals(itemKey, key, StringComparison.OrdinalIgnoreCase)) + continue; + + return type.GetProperty(nameof(KeyValuePair.Value))?.GetValue(value)?.ToString(); + } + + return null; + } + + private void CaptureLoggedWorkflowInstanceId( + TState state, + Exception? exception, + Func formatter) + { + var workflowInstanceId = GetWorkflowInstanceId(); + if (string.IsNullOrWhiteSpace(workflowInstanceId)) + return; + + string message; + try + { + message = formatter(state, exception); + } + catch (Exception) + { + return; + } + + if (string.IsNullOrWhiteSpace(message)) + return; + + var now = DateTimeOffset.UtcNow; + + lock (_pendingWorkflowLogsLock) + { + PruneExpiredPendingWorkflowLogs(now); + + while (_pendingWorkflowLogs.Count >= PendingWorkflowLogsCapacity) + _pendingWorkflowLogs.RemoveAt(0); + + _pendingWorkflowLogs.Add(new PendingWorkflowLog(message, workflowInstanceId, now)); + } + } + + private void PruneExpiredPendingWorkflowLogs(DateTimeOffset now) + { + for (var i = _pendingWorkflowLogs.Count - 1; i >= 0; i--) + { + if (now - _pendingWorkflowLogs[i].CapturedAt <= PendingWorkflowLogTtl) + continue; + + _pendingWorkflowLogs.RemoveAt(i); + } + } + + private static bool IsCorrelated(string capturedLine, string loggedMessage) + { + var trimmedLine = capturedLine.Trim(); + var trimmedMessage = loggedMessage.Trim(); + + return trimmedLine.Equals(trimmedMessage, StringComparison.Ordinal); + } + + private sealed class ScopeCapturingLogger(ConsoleLogScopeAccessor accessor) : ILogger + { + public IDisposable? BeginScope(TState state) where TState : notnull => NullScope.Instance; + + public bool IsEnabled(LogLevel logLevel) => logLevel != LogLevel.None; + + public void Log( + LogLevel logLevel, + EventId eventId, + TState state, + Exception? exception, + Func formatter) + { + if (IsEnabled(logLevel)) + accessor.CaptureLoggedWorkflowInstanceId(state, exception, formatter); + } + } + + private sealed record PendingWorkflowLog(string Message, string WorkflowInstanceId, DateTimeOffset CapturedAt); + + private sealed class NullScope : IDisposable + { + public static NullScope Instance { get; } = new(); + + public void Dispose() + { + } + } +} diff --git a/src/modules/Elsa.Diagnostics.ConsoleLogs/Services/ConsoleLogsHost.cs b/src/modules/Elsa.Diagnostics.ConsoleLogs/Services/ConsoleLogsHost.cs new file mode 100644 index 000000000..deb976cfd --- /dev/null +++ b/src/modules/Elsa.Diagnostics.ConsoleLogs/Services/ConsoleLogsHost.cs @@ -0,0 +1,194 @@ +using Elsa.Diagnostics.ConsoleLogs.Providers.InMemory; +using Microsoft.Extensions.Options; + +namespace Elsa.Diagnostics.ConsoleLogs.Services; + +/// +/// Process-wide owner of the console capture pipeline. Console output is, by nature, a single shared OS +/// resource — there is exactly one stdout and one stderr per process — so the capture, the in-memory ring +/// buffer, and the source registry are likewise single instances regardless of how many DI containers +/// (shells, tenants) want to consume them. This class is the seam that makes that fact explicit. +/// +/// +/// Configuration follows a first-wins model: calls made before the host is first +/// accessed are applied to a single instance. Subsequent calls are +/// ignored because the configuration would not apply to anything already captured. Call +/// AddConsoleLogsHost from Program.cs to set host-wide options deterministically. +/// +/// +/// +/// Initialization is lazy and thread-safe: the first access from any thread builds the pipeline. The host +/// also auto-installs , so simply touching , +/// , or is enough to start capturing. +/// +/// +public static class ConsoleLogsHost +{ + private static readonly object Lock = new(); + private static Action? _pendingConfigure; + private static Func? _providerFactory; + private static Lazy _state = CreateLazy(); + private static int _leases; + + /// + /// Applies a configuration delegate to the host options if the host has not yet been initialized. + /// Calls after initialization are ignored to preserve consistency between configuration and captured + /// data. Returns true when the delegate was accepted, false when the host had already + /// initialized. + /// + public static bool Configure(Action configure) + { + ArgumentNullException.ThrowIfNull(configure); + + lock (Lock) + { + if (_state.IsValueCreated) + return false; + + _pendingConfigure = (Action?)Delegate.Combine(_pendingConfigure, configure); + return true; + } + } + + /// + /// Configures the provider used by the process-wide capture pipeline. The factory must be registered + /// before the host is initialized. + /// + public static bool ConfigureProvider(Func, IConsoleLogSourceRegistry, IConsoleLogProvider> providerFactory) + { + ArgumentNullException.ThrowIfNull(providerFactory); + return ConfigureProvider(context => providerFactory(context.Options, context.SourceRegistry)); + } + + internal static bool ConfigureProvider(Func providerFactory) + { + ArgumentNullException.ThrowIfNull(providerFactory); + + lock (Lock) + { + if (_state.IsValueCreated || _providerFactory != null) + return false; + + _providerFactory = providerFactory; + return true; + } + } + + public static IOptions Options => _state.Value.Options; + public static IConsoleLogSourceRegistry SourceRegistry => _state.Value.SourceRegistry; + public static IConsoleLogRedactor Redactor => _state.Value.Redactor; + public static ConsoleLineFormatter Formatter => _state.Value.Formatter; + public static ConsoleLogScopeAccessor ScopeAccessor => _state.Value.ScopeAccessor; + public static IConsoleLogProvider Provider => _state.Value.Provider; + public static ConsoleCaptureTee Capture => _state.Value.Capture; + + /// + /// Triggers initialization without resolving any specific component. Returns true when this call + /// performed the initialization, false when the host was already initialized. + /// + public static bool EnsureInitialized() + { + var wasCreated = _state.IsValueCreated; + _ = _state.Value; + return !wasCreated; + } + + public static void AddReference() + { + lock (Lock) + _leases++; + } + + public static ValueTask ReleaseReferenceAsync(CancellationToken cancellationToken = default) + { + Lazy state; + + lock (Lock) + { + if (_leases <= 0) + return ValueTask.CompletedTask; + + _leases--; + if (_leases > 0) + return ValueTask.CompletedTask; + + state = ResetState(); + } + + return DisposeStateAsync(state, cancellationToken); + } + + /// + /// Tears down the host pipeline. Intended for tests and for orderly host shutdown. + /// + public static async ValueTask ShutdownAsync(CancellationToken cancellationToken = default) + { + Lazy state; + + lock (Lock) + { + state = ResetState(); + _leases = 0; + } + + await DisposeStateAsync(state, cancellationToken).ConfigureAwait(false); + } + + private static Lazy CreateLazy() => new(BuildState, LazyThreadSafetyMode.ExecutionAndPublication); + + private static Lazy ResetState() + { + var state = _state; + _state = CreateLazy(); + _pendingConfigure = null; + _providerFactory = null; + return state; + } + + private static async ValueTask DisposeStateAsync(Lazy state, CancellationToken cancellationToken) + { + if (state.IsValueCreated) + await state.Value.Capture.DisposeAsync(cancellationToken).ConfigureAwait(false); + } + + private static HostState BuildState() + { + var options = new ConsoleLogsOptions(); + + Func? providerFactory; + + lock (Lock) + { + _pendingConfigure?.Invoke(options); + providerFactory = _providerFactory; + } + + var wrappedOptions = Microsoft.Extensions.Options.Options.Create(options); + var sourceRegistry = new ConsoleLogSourceRegistry(wrappedOptions); + var redactor = new ConsoleLogRedactor(wrappedOptions); + var formatter = new ConsoleLineFormatter(wrappedOptions); + var scopeAccessor = new ConsoleLogScopeAccessor(); + var providerContext = new ConsoleLogsProviderContext(wrappedOptions, sourceRegistry, redactor, formatter, scopeAccessor); + var provider = providerFactory?.Invoke(providerContext) ?? new InMemoryConsoleLogProvider(wrappedOptions, sourceRegistry); + var capture = new ConsoleCaptureTee(provider, sourceRegistry, redactor, formatter, scopeAccessor, wrappedOptions); + + return new HostState(wrappedOptions, sourceRegistry, redactor, formatter, scopeAccessor, provider, capture); + } + + private sealed record HostState( + IOptions Options, + IConsoleLogSourceRegistry SourceRegistry, + IConsoleLogRedactor Redactor, + ConsoleLineFormatter Formatter, + ConsoleLogScopeAccessor ScopeAccessor, + IConsoleLogProvider Provider, + ConsoleCaptureTee Capture); +} + +internal sealed record ConsoleLogsProviderContext( + IOptions Options, + IConsoleLogSourceRegistry SourceRegistry, + IConsoleLogRedactor Redactor, + ConsoleLineFormatter Formatter, + ConsoleLogScopeAccessor ScopeAccessor); + diff --git a/src/modules/Elsa.Diagnostics.ConsoleLogs/Services/ConsoleLogsHostedService.cs b/src/modules/Elsa.Diagnostics.ConsoleLogs/Services/ConsoleLogsHostedService.cs new file mode 100644 index 000000000..f92395f80 --- /dev/null +++ b/src/modules/Elsa.Diagnostics.ConsoleLogs/Services/ConsoleLogsHostedService.cs @@ -0,0 +1,57 @@ +using Microsoft.Extensions.Hosting; + +namespace Elsa.Diagnostics.ConsoleLogs.Services; + +/// +/// Host-level hosted service that ensures the console capture is initialized at host build time, +/// periodically flushes idle partial lines, and drains pending writes during host shutdown. +/// +public sealed class ConsoleLogsHostedService : BackgroundService +{ + private readonly Action? _configureHost; + + public ConsoleLogsHostedService() : this((Action?)null) + { + } + + public ConsoleLogsHostedService(IServiceProvider serviceProvider) : this(() => + { + if (serviceProvider.GetService(typeof(ConsoleLogsProviderRegistration)) is ConsoleLogsProviderRegistration registration) + registration.ConfigureProvider(serviceProvider); + }) + { + } + + public ConsoleLogsHostedService(Action? configureHost) + { + _configureHost = configureHost; + } + + public override Task StartAsync(CancellationToken cancellationToken) + { + _configureHost?.Invoke(); + + // Touch the host so initialization happens before the first request, not lazily on first capture. + ConsoleLogsHost.AddReference(); + ConsoleLogsHost.EnsureInitialized(); + return base.StartAsync(cancellationToken); + } + + public override async Task StopAsync(CancellationToken cancellationToken) + { + await base.StopAsync(cancellationToken).ConfigureAwait(false); + await ConsoleLogsHost.ReleaseReferenceAsync(cancellationToken).ConfigureAwait(false); + } + + protected override async Task ExecuteAsync(CancellationToken stoppingToken) + { + var interval = TimeSpan.FromMilliseconds(Math.Max(100, ConsoleLogsHost.Options.Value.IdleFlushTimeout.TotalMilliseconds / 2)); + + while (!stoppingToken.IsCancellationRequested) + { + await Task.Delay(interval, stoppingToken).ConfigureAwait(false); + await ConsoleLogsHost.Capture.FlushIdleAsync(stoppingToken).ConfigureAwait(false); + } + } +} + diff --git a/src/modules/Elsa.Diagnostics.ConsoleLogs/Services/ConsoleLogsProviderRegistration.cs b/src/modules/Elsa.Diagnostics.ConsoleLogs/Services/ConsoleLogsProviderRegistration.cs new file mode 100644 index 000000000..6de9e9a79 --- /dev/null +++ b/src/modules/Elsa.Diagnostics.ConsoleLogs/Services/ConsoleLogsProviderRegistration.cs @@ -0,0 +1,85 @@ +using System.Diagnostics.CodeAnalysis; +using Elsa.Diagnostics.ConsoleLogs.Contracts; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Options; + +namespace Elsa.Diagnostics.ConsoleLogs.Services; + +internal sealed class ConsoleLogsProviderRegistration(IServiceCollection services, ServiceDescriptor defaultProviderDescriptor) +{ + private readonly object _lock = new(); + private ConsoleLogsProviderContext? _hostContext; + + public void ConfigureProvider(IServiceProvider serviceProvider) + { + var descriptor = GetCustomProviderDescriptor(); + if (descriptor == null) + return; + + ConsoleLogsHost.ConfigureProvider(context => ResolveProvider(serviceProvider, context)); + } + + public bool TryGetHostOptions([NotNullWhen(true)] out IOptions? options) + { + lock (_lock) + { + options = _hostContext?.Options; + return options != null; + } + } + + public bool TryGetHostSourceRegistry([NotNullWhen(true)] out IConsoleLogSourceRegistry? sourceRegistry) + { + lock (_lock) + { + sourceRegistry = _hostContext?.SourceRegistry; + return sourceRegistry != null; + } + } + + public bool TryGetHostRedactor([NotNullWhen(true)] out IConsoleLogRedactor? redactor) + { + lock (_lock) + { + redactor = _hostContext?.Redactor; + return redactor != null; + } + } + + public bool TryGetHostFormatter([NotNullWhen(true)] out ConsoleLineFormatter? formatter) + { + lock (_lock) + { + formatter = _hostContext?.Formatter; + return formatter != null; + } + } + + public bool TryGetHostScopeAccessor([NotNullWhen(true)] out ConsoleLogScopeAccessor? scopeAccessor) + { + lock (_lock) + { + scopeAccessor = _hostContext?.ScopeAccessor; + return scopeAccessor != null; + } + } + + private ServiceDescriptor? GetCustomProviderDescriptor() => + services.LastOrDefault(x => x.ServiceType == typeof(IConsoleLogProvider) && !ReferenceEquals(x, defaultProviderDescriptor)); + + private IConsoleLogProvider ResolveProvider(IServiceProvider serviceProvider, ConsoleLogsProviderContext context) + { + lock (_lock) + _hostContext = context; + + try + { + return serviceProvider.GetRequiredService(); + } + finally + { + lock (_lock) + _hostContext = null; + } + } +} diff --git a/src/modules/Elsa.Diagnostics.ConsoleLogs/Services/ConsoleStreamHook.cs b/src/modules/Elsa.Diagnostics.ConsoleLogs/Services/ConsoleStreamHook.cs new file mode 100644 index 000000000..1d28e3276 --- /dev/null +++ b/src/modules/Elsa.Diagnostics.ConsoleLogs/Services/ConsoleStreamHook.cs @@ -0,0 +1,207 @@ +namespace Elsa.Diagnostics.ConsoleLogs.Services; + +/// +/// Installs a process-wide tee on and so that every +/// captured write is fanned out to all currently registered subscribers, in addition to the original writers. +/// +/// +/// Downstream logger providers (e.g. Microsoft.Extensions.Logging.Console) capture Console.Out +/// once at construction. Therefore the tee must be installed BEFORE any such provider is constructed — +/// ideally as the very first line of Program.cs. Calling implicitly installs +/// the tee, so a host can also rely on subscribers attaching early in the pipeline. +/// +/// +/// +/// Writes captured before any subscriber attaches are kept in a bounded replay buffer and delivered to the +/// first subscriber (when replayPreviousWrites is true), preserving original timestamps. This bridges +/// the gap between and the first call, which typically happens +/// when the DI container is built and the capture singleton is resolved. +/// +/// +/// +/// Subscribers that themselves write to the console while handling a capture would cause infinite recursion. +/// Set to true on the publish path's context to +/// short-circuit dispatch for the duration of that asynchronous flow. +/// +/// +public static class ConsoleStreamHook +{ + private static readonly object Lock = new(); + private static readonly AsyncLocal SuppressFlag = new(); + private static volatile Subscription[] _subscriptions = []; + private static readonly ReplayBuffer Replay = new(capacity: 4096); + private static volatile bool _installed; + private static TextWriter? _originalOut; + private static TextWriter? _originalError; + + /// + /// While true on the current async context, captured writes will not be dispatched to subscribers. + /// Subscribers must set this to true while running their publish pipeline so that any logging triggered + /// by that pipeline does not feed back into capture. + /// + public static bool SuppressCapture + { + get => SuppressFlag.Value; + set => SuppressFlag.Value = value; + } + + /// + /// Installs the tee if not already installed. Safe to call multiple times. + /// + public static void Install() + { + if (_installed) + return; + + lock (Lock) + { + if (_installed) + return; + + _originalOut = Console.Out; + _originalError = Console.Error; + Console.SetOut(new ConsoleStreamTeeWriter(_originalOut, ConsoleLogStream.Stdout, Dispatch)); + Console.SetError(new ConsoleStreamTeeWriter(_originalError, ConsoleLogStream.Stderr, Dispatch)); + _installed = true; + } + } + + /// + /// Restores the original writers and clears all subscribers. Intended for tests. + /// + public static void Uninstall() + { + lock (Lock) + { + if (!_installed) + return; + + if (_originalOut != null) + Console.SetOut(_originalOut); + + if (_originalError != null) + Console.SetError(_originalError); + + _originalOut = null; + _originalError = null; + _subscriptions = []; + Replay.Clear(); + _installed = false; + } + } + + /// + /// Registers a subscriber that receives every captured write. Disposing the returned token removes the + /// subscription. Calling Subscribe auto-installs the tee. When + /// is true, the buffered writes captured since install are replayed to this subscriber synchronously. + /// + public static IDisposable Subscribe(Action handler, bool replayPreviousWrites = true) + { + ArgumentNullException.ThrowIfNull(handler); + + Install(); + var subscription = new Subscription(handler); + + lock (Lock) + { + _subscriptions = [.._subscriptions, subscription]; + + if (replayPreviousWrites) + { + foreach (var chunk in Replay.Snapshot()) + InvokeSafely(subscription, chunk); + + Replay.Clear(); + } + } + + return subscription; + } + + internal static void Remove(Subscription subscription) + { + lock (Lock) + { + _subscriptions = _subscriptions.Where(x => !ReferenceEquals(x, subscription)).ToArray(); + } + } + + private static void Dispatch(ConsoleLogStream stream, string value) + { + if (SuppressFlag.Value || value.Length == 0) + return; + + var chunk = new CapturedChunk(stream, value, DateTimeOffset.UtcNow); + var subscriptions = _subscriptions; + + if (subscriptions.Length == 0) + { + // No subscriber yet — retain for replay to the first one that attaches. + Replay.Add(chunk); + return; + } + + foreach (var subscription in subscriptions) + InvokeSafely(subscription, chunk); + } + + private static void InvokeSafely(Subscription subscription, CapturedChunk chunk) + { + try + { + subscription.Handler(chunk); + } + catch (Exception e) when (!IsCriticalException(e)) + { + // Never let a subscriber take down the writer. Subscribers are responsible for their own resilience. + } + } + + private static bool IsCriticalException(Exception exception) => + exception is OutOfMemoryException + or StackOverflowException + or AccessViolationException + or AppDomainUnloadedException + or BadImageFormatException + or CannotUnloadAppDomainException; + + internal sealed class Subscription(Action handler) : IDisposable + { + public Action Handler { get; } = handler; + public void Dispose() => Remove(this); + } + + private sealed class ReplayBuffer(int capacity) + { + private readonly object _lock = new(); + private readonly Queue _items = new(capacity); + + public void Add(CapturedChunk chunk) + { + lock (_lock) + { + if (_items.Count == capacity) + _items.Dequeue(); + + _items.Enqueue(chunk); + } + } + + public IReadOnlyList Snapshot() + { + lock (_lock) + return _items.ToArray(); + } + + public void Clear() + { + lock (_lock) + _items.Clear(); + } + } +} + +/// +/// A single chunk of text captured from stdout/stderr together with the moment of capture. +/// +public readonly record struct CapturedChunk(ConsoleLogStream Stream, string Text, DateTimeOffset TimestampUtc); diff --git a/src/modules/Elsa.Diagnostics.ConsoleLogs/Services/ConsoleStreamTeeWriter.cs b/src/modules/Elsa.Diagnostics.ConsoleLogs/Services/ConsoleStreamTeeWriter.cs new file mode 100644 index 000000000..577ca49cd --- /dev/null +++ b/src/modules/Elsa.Diagnostics.ConsoleLogs/Services/ConsoleStreamTeeWriter.cs @@ -0,0 +1,212 @@ +using System.Diagnostics.CodeAnalysis; +using System.Text; + +namespace Elsa.Diagnostics.ConsoleLogs.Services; + +/// +/// A that forwards every write to the original underlying writer and a configurable capture sink. +/// +internal sealed class ConsoleStreamTeeWriter(TextWriter original, ConsoleLogStream stream, Action capture) : TextWriter +{ + public TextWriter Original => original; + + public override Encoding Encoding => original.Encoding; + public override IFormatProvider FormatProvider => original.FormatProvider; + [AllowNull] + public override string NewLine { get => original.NewLine; set => original.NewLine = value; } + + // ── Write overloads ── + + public override void Write(char value) + { + original.Write(value); + capture(stream, value.ToString()); + } + + public override void Write(char[]? buffer) + { + original.Write(buffer); + if (buffer is { Length: > 0 }) + capture(stream, new string(buffer)); + } + + public override void Write(char[] buffer, int index, int count) + { + original.Write(buffer, index, count); + if (count > 0) + capture(stream, new string(buffer, index, count)); + } + + public override void Write(ReadOnlySpan buffer) + { + original.Write(buffer); + if (!buffer.IsEmpty) + capture(stream, buffer.ToString()); + } + + public override void Write(bool value) { original.Write(value); capture(stream, value ? "True" : "False"); } + public override void Write(int value) { original.Write(value); capture(stream, value.ToString(FormatProvider)); } + public override void Write(uint value) { original.Write(value); capture(stream, value.ToString(FormatProvider)); } + public override void Write(long value) { original.Write(value); capture(stream, value.ToString(FormatProvider)); } + public override void Write(ulong value) { original.Write(value); capture(stream, value.ToString(FormatProvider)); } + public override void Write(float value) { original.Write(value); capture(stream, value.ToString(FormatProvider)); } + public override void Write(double value) { original.Write(value); capture(stream, value.ToString(FormatProvider)); } + public override void Write(decimal value) { original.Write(value); capture(stream, value.ToString(FormatProvider)); } + + public override void Write(string? value) + { + original.Write(value); + if (value != null) + capture(stream, value); + } + + public override void Write(object? value) + { + original.Write(value); + var text = value?.ToString(); + if (text != null) + capture(stream, text); + } + + public override void Write(StringBuilder? value) + { + original.Write(value); + if (value is { Length: > 0 }) + capture(stream, value.ToString()); + } + + public override void Write(string format, object? arg0) { var text = string.Format(FormatProvider, format, arg0); original.Write(text); capture(stream, text); } + public override void Write(string format, object? arg0, object? arg1) { var text = string.Format(FormatProvider, format, arg0, arg1); original.Write(text); capture(stream, text); } + public override void Write(string format, object? arg0, object? arg1, object? arg2) { var text = string.Format(FormatProvider, format, arg0, arg1, arg2); original.Write(text); capture(stream, text); } + public override void Write(string format, params object?[] arg) { var text = string.Format(FormatProvider, format, arg); original.Write(text); capture(stream, text); } +#if NET9_0_OR_GREATER + public override void Write(string format, params ReadOnlySpan arg) { var text = string.Format(FormatProvider, format, arg); original.Write(text); capture(stream, text); } +#endif + + // ── WriteLine overloads ── + + public override void WriteLine() { original.WriteLine(); capture(stream, "\n"); } + public override void WriteLine(char value) { original.WriteLine(value); capture(stream, value + "\n"); } + + public override void WriteLine(char[]? buffer) + { + original.WriteLine(buffer); + capture(stream, (buffer is { Length: > 0 } ? new string(buffer) : "") + "\n"); + } + + public override void WriteLine(char[] buffer, int index, int count) + { + original.WriteLine(buffer, index, count); + capture(stream, new string(buffer, index, count) + "\n"); + } + + public override void WriteLine(ReadOnlySpan buffer) + { + original.WriteLine(buffer); + capture(stream, string.Concat(buffer, "\n")); + } + + public override void WriteLine(bool value) { original.WriteLine(value); capture(stream, (value ? "True" : "False") + "\n"); } + public override void WriteLine(int value) { original.WriteLine(value); capture(stream, value.ToString(FormatProvider) + "\n"); } + public override void WriteLine(uint value) { original.WriteLine(value); capture(stream, value.ToString(FormatProvider) + "\n"); } + public override void WriteLine(long value) { original.WriteLine(value); capture(stream, value.ToString(FormatProvider) + "\n"); } + public override void WriteLine(ulong value) { original.WriteLine(value); capture(stream, value.ToString(FormatProvider) + "\n"); } + public override void WriteLine(float value) { original.WriteLine(value); capture(stream, value.ToString(FormatProvider) + "\n"); } + public override void WriteLine(double value) { original.WriteLine(value); capture(stream, value.ToString(FormatProvider) + "\n"); } + public override void WriteLine(decimal value) { original.WriteLine(value); capture(stream, value.ToString(FormatProvider) + "\n"); } + + public override void WriteLine(string? value) + { + original.WriteLine(value); + capture(stream, (value ?? "") + "\n"); + } + + public override void WriteLine(object? value) + { + original.WriteLine(value); + capture(stream, (value?.ToString() ?? "") + "\n"); + } + + public override void WriteLine(StringBuilder? value) + { + original.WriteLine(value); + capture(stream, (value is { Length: > 0 } ? value.ToString() : "") + "\n"); + } + + public override void WriteLine(string format, object? arg0) { var text = string.Format(FormatProvider, format, arg0); original.WriteLine(text); capture(stream, text + "\n"); } + public override void WriteLine(string format, object? arg0, object? arg1) { var text = string.Format(FormatProvider, format, arg0, arg1); original.WriteLine(text); capture(stream, text + "\n"); } + public override void WriteLine(string format, object? arg0, object? arg1, object? arg2) { var text = string.Format(FormatProvider, format, arg0, arg1, arg2); original.WriteLine(text); capture(stream, text + "\n"); } + public override void WriteLine(string format, params object?[] arg) { var text = string.Format(FormatProvider, format, arg); original.WriteLine(text); capture(stream, text + "\n"); } +#if NET9_0_OR_GREATER + public override void WriteLine(string format, params ReadOnlySpan arg) { var text = string.Format(FormatProvider, format, arg); original.WriteLine(text); capture(stream, text + "\n"); } +#endif + + // ── Async Write overloads ── + + public override async Task WriteAsync(char value) { await original.WriteAsync(value); capture(stream, value.ToString()); } + + public override async Task WriteAsync(string? value) + { + await original.WriteAsync(value); + if (value != null) + capture(stream, value); + } + + public override async Task WriteAsync(char[] buffer, int index, int count) + { + await original.WriteAsync(buffer, index, count); + if (count > 0) + capture(stream, new string(buffer, index, count)); + } + + public override async Task WriteAsync(ReadOnlyMemory buffer, CancellationToken cancellationToken = default) + { + await original.WriteAsync(buffer, cancellationToken); + if (!buffer.IsEmpty) + capture(stream, buffer.ToString()); + } + + public override async Task WriteAsync(StringBuilder? value, CancellationToken cancellationToken = default) + { + await original.WriteAsync(value, cancellationToken); + if (value is { Length: > 0 }) + capture(stream, value.ToString()); + } + + // ── Async WriteLine overloads ── + + public override async Task WriteLineAsync() { await original.WriteLineAsync(); capture(stream, "\n"); } + public override async Task WriteLineAsync(char value) { await original.WriteLineAsync(value); capture(stream, value + "\n"); } + + public override async Task WriteLineAsync(string? value) + { + await original.WriteLineAsync(value); + capture(stream, (value ?? "") + "\n"); + } + + public override async Task WriteLineAsync(char[] buffer, int index, int count) + { + await original.WriteLineAsync(buffer, index, count); + capture(stream, new string(buffer, index, count) + "\n"); + } + + public override async Task WriteLineAsync(ReadOnlyMemory buffer, CancellationToken cancellationToken = default) + { + await original.WriteLineAsync(buffer, cancellationToken); + capture(stream, string.Concat(buffer.Span, "\n")); + } + + public override async Task WriteLineAsync(StringBuilder? value, CancellationToken cancellationToken = default) + { + await original.WriteLineAsync(value, cancellationToken); + capture(stream, (value is { Length: > 0 } ? value.ToString() : "") + "\n"); + } + + // ── Lifecycle ── + + public override void Flush() => original.Flush(); + public override Task FlushAsync() => original.FlushAsync(); + public override Task FlushAsync(CancellationToken cancellationToken) => original.FlushAsync(cancellationToken); + public override void Close() => original.Close(); +} + diff --git a/src/modules/Elsa.Identity/HostedServices/AdminUserInitializer.cs b/src/modules/Elsa.Identity/HostedServices/AdminUserInitializer.cs index 92c022e4e..e8759540b 100644 --- a/src/modules/Elsa.Identity/HostedServices/AdminUserInitializer.cs +++ b/src/modules/Elsa.Identity/HostedServices/AdminUserInitializer.cs @@ -20,7 +20,7 @@ public class AdminUserInitializer( ILogger logger) : BackgroundTask { - public override async Task StartAsync(CancellationToken cancellationToken) + public override async Task ExecuteAsync(CancellationToken cancellationToken) { var adminUserName = options.Value.AdminUserName; var adminPassword = options.Value.AdminPassword; diff --git a/src/modules/Elsa.Scheduling/Features/SchedulingFeature.cs b/src/modules/Elsa.Scheduling/Features/SchedulingFeature.cs index e29405991..7bac060de 100644 --- a/src/modules/Elsa.Scheduling/Features/SchedulingFeature.cs +++ b/src/modules/Elsa.Scheduling/Features/SchedulingFeature.cs @@ -7,8 +7,8 @@ using Elsa.Features.Attributes; using Elsa.Features.Services; using Elsa.Scheduling.Bookmarks; using Elsa.Scheduling.Handlers; -using Elsa.Scheduling.HostedServices; using Elsa.Scheduling.Services; +using Elsa.Scheduling.StartupTasks; using Elsa.Scheduling.TriggerPayloadValidators; using Elsa.Workflows.Management.Features; using Microsoft.Extensions.DependencyInjection; @@ -41,7 +41,6 @@ public class SchedulingFeature : FeatureBase { Services .AddSingleton() - .AddSingleton(sp => sp.GetRequiredService()) .AddSingleton(sp => sp.GetRequiredService()) .AddSingleton() .AddSingleton() @@ -50,7 +49,7 @@ public class SchedulingFeature : FeatureBase .AddScoped() .AddScoped() .AddScoped(WorkflowScheduler) - .AddBackgroundTask() + .AddStartupTask() .AddHandlersFrom() //Trigger payload validators. diff --git a/src/modules/Elsa.Scheduling/Handlers/UpdateTenantSchedules.cs b/src/modules/Elsa.Scheduling/Handlers/UpdateTenantSchedules.cs index 46456cb9b..64f289778 100644 --- a/src/modules/Elsa.Scheduling/Handlers/UpdateTenantSchedules.cs +++ b/src/modules/Elsa.Scheduling/Handlers/UpdateTenantSchedules.cs @@ -9,7 +9,7 @@ using Timer = Elsa.Scheduling.Activities.Timer; namespace Elsa.Scheduling.Handlers; -public class UpdateTenantSchedules : ITenantActivatedEvent, ITenantDeletedEvent +public class UpdateTenantSchedules : ITenantDeletedEvent { private static readonly string[] ActivityTypeNames = [ @@ -18,19 +18,6 @@ public class UpdateTenantSchedules : ITenantActivatedEvent, ITenantDeletedEvent ActivityTypeNameHelper.GenerateTypeName() ]; - public async Task TenantActivatedAsync(TenantActivatedEventArgs args) - { - var serviceProvider = args.TenantScope.ServiceProvider; - var cancellationToken = args.CancellationToken; - var triggerScheduler = args.TenantScope.ServiceProvider.GetRequiredService(); - var bookmarkScheduler = args.TenantScope.ServiceProvider.GetRequiredService(); - var triggers = (await GetTriggersAsync(serviceProvider, cancellationToken)).ToList(); - var bookmarks = (await GetBookmarksAsync(serviceProvider, cancellationToken)).ToList(); - - await triggerScheduler.ScheduleAsync(triggers, args.CancellationToken); - await bookmarkScheduler.ScheduleAsync(bookmarks, args.CancellationToken); - } - public async Task TenantDeletedAsync(TenantDeletedEventArgs args) { var serviceProvider = args.TenantScope.ServiceProvider; @@ -57,4 +44,4 @@ public class UpdateTenantSchedules : ITenantActivatedEvent, ITenantDeletedEvent var filter = BookmarkFilter.ByActivityTypeNames(ActivityTypeNames); return await bookmarkStore.FindManyAsync(filter, cancellationToken); } -} \ No newline at end of file +} diff --git a/src/modules/Elsa.Scheduling/HostedServices/CreateSchedulesBackgroundTask.cs b/src/modules/Elsa.Scheduling/HostedServices/CreateSchedulesBackgroundTask.cs deleted file mode 100644 index 510096cb6..000000000 --- a/src/modules/Elsa.Scheduling/HostedServices/CreateSchedulesBackgroundTask.cs +++ /dev/null @@ -1,35 +0,0 @@ -using Elsa.Common; -using Elsa.Workflows.Runtime; -using Elsa.Workflows.Runtime.Filters; -using JetBrains.Annotations; - -namespace Elsa.Scheduling.HostedServices; - -/// -/// Creates new schedules when using the default scheduler (which doesn't have its own persistence layer like Quartz or Hangfire). -/// -[UsedImplicitly] -public class CreateSchedulesBackgroundTask(ITriggerStore triggerStore, IBookmarkStore bookmarkStore, ITriggerScheduler triggerScheduler, IBookmarkScheduler bookmarkScheduler) : BackgroundTask -{ - /// - public override async Task ExecuteAsync(CancellationToken stoppingToken) - { - var stimulusNames = new[] - { - SchedulingStimulusNames.Cron, SchedulingStimulusNames.Timer, SchedulingStimulusNames.StartAt, SchedulingStimulusNames.Delay, - }; - var triggerFilter = new TriggerFilter - { - Names = stimulusNames - }; - var bookmarkFilter = new BookmarkFilter - { - Names = stimulusNames - }; - var triggers = (await triggerStore.FindManyAsync(triggerFilter, stoppingToken)).ToList(); - var bookmarks = (await bookmarkStore.FindManyAsync(bookmarkFilter, stoppingToken)).ToList(); - - await triggerScheduler.ScheduleAsync(triggers, stoppingToken); - await bookmarkScheduler.ScheduleAsync(bookmarks, stoppingToken); - } -} \ No newline at end of file diff --git a/src/modules/Elsa.Scheduling/ShellFeatures/SchedulingFeature.cs b/src/modules/Elsa.Scheduling/ShellFeatures/SchedulingFeature.cs index 149094b04..0c960b2cc 100644 --- a/src/modules/Elsa.Scheduling/ShellFeatures/SchedulingFeature.cs +++ b/src/modules/Elsa.Scheduling/ShellFeatures/SchedulingFeature.cs @@ -4,8 +4,8 @@ using Elsa.Common.ShellFeatures; using Elsa.Extensions; using Elsa.Scheduling.Bookmarks; using Elsa.Scheduling.Handlers; -using Elsa.Scheduling.HostedServices; using Elsa.Scheduling.Services; +using Elsa.Scheduling.StartupTasks; using Elsa.Scheduling.TriggerPayloadValidators; using Elsa.Workflows.Management.Extensions; using JetBrains.Annotations; @@ -37,7 +37,6 @@ public class SchedulingFeature : IShellFeature { services .AddSingleton() - .AddSingleton(sp => sp.GetRequiredService()) .AddSingleton(sp => sp.GetRequiredService()) .AddSingleton() .AddSingleton() @@ -46,10 +45,9 @@ public class SchedulingFeature : IShellFeature .AddScoped() .AddScoped() .AddScoped(WorkflowScheduler) - .AddBackgroundTask() + .AddStartupTask() .AddHandlersFrom() .AddTriggerPayloadValidator() .AddActivitiesFrom(); } } - diff --git a/src/modules/Elsa.Scheduling/StartupTasks/CreateSchedulesStartupTask.cs b/src/modules/Elsa.Scheduling/StartupTasks/CreateSchedulesStartupTask.cs new file mode 100644 index 000000000..4ab8971d6 --- /dev/null +++ b/src/modules/Elsa.Scheduling/StartupTasks/CreateSchedulesStartupTask.cs @@ -0,0 +1,50 @@ +using Elsa.Common; +using Elsa.Common.Multitenancy; +using Elsa.Workflows.Runtime; +using Elsa.Workflows.Runtime.Filters; +using Elsa.Workflows.Runtime.Tasks; +using Microsoft.Extensions.DependencyInjection; + +namespace Elsa.Scheduling.StartupTasks; + +/// +/// Enqueues schedule creation when using the default scheduler, which doesn't have its own persistence layer like Quartz or Hangfire. +/// +[TaskDependency(typeof(PopulateRegistriesStartupTask))] +public class CreateSchedulesStartupTask(IServiceProvider serviceProvider) : IStartupTask +{ + public async Task ExecuteAsync(CancellationToken cancellationToken) + { + var workQueue = serviceProvider.GetService(); + + if (workQueue != null) + await workQueue.EnqueueAsync(CreateSchedulesAsync, cancellationToken); + else + await CreateSchedulesAsync(serviceProvider, cancellationToken); + } + + private static async Task CreateSchedulesAsync(IServiceProvider serviceProvider, CancellationToken cancellationToken) + { + var triggerStore = serviceProvider.GetRequiredService(); + var bookmarkStore = serviceProvider.GetRequiredService(); + var triggerScheduler = serviceProvider.GetRequiredService(); + var bookmarkScheduler = serviceProvider.GetRequiredService(); + var stimulusNames = new[] + { + SchedulingStimulusNames.Cron, SchedulingStimulusNames.Timer, SchedulingStimulusNames.StartAt, SchedulingStimulusNames.Delay, + }; + var triggerFilter = new TriggerFilter + { + Names = stimulusNames + }; + var bookmarkFilter = new BookmarkFilter + { + Names = stimulusNames + }; + var triggers = (await triggerStore.FindManyAsync(triggerFilter, cancellationToken)).ToList(); + var bookmarks = (await bookmarkStore.FindManyAsync(bookmarkFilter, cancellationToken)).ToList(); + + await triggerScheduler.ScheduleAsync(triggers, cancellationToken); + await bookmarkScheduler.ScheduleAsync(bookmarks, cancellationToken); + } +} diff --git a/src/modules/Elsa.Workflows.Runtime.Distributed/Services/DistributedRuntimeLockProviderValidator.cs b/src/modules/Elsa.Workflows.Runtime.Distributed/Services/DistributedRuntimeLockProviderValidator.cs index c76cd0b02..143109473 100644 --- a/src/modules/Elsa.Workflows.Runtime.Distributed/Services/DistributedRuntimeLockProviderValidator.cs +++ b/src/modules/Elsa.Workflows.Runtime.Distributed/Services/DistributedRuntimeLockProviderValidator.cs @@ -10,7 +10,7 @@ using Microsoft.Extensions.Options; namespace Elsa.Workflows.Runtime.Distributed; /// -/// Validates that the distributed runtime is not accidentally backed by a local-only lock provider. +/// Logs when the distributed runtime is backed by a local-only lock provider. /// public class DistributedRuntimeLockProviderValidator( IDistributedLockProvider distributedLockProvider, @@ -18,7 +18,7 @@ public class DistributedRuntimeLockProviderValidator( ILogger logger) { /// - /// Validates the configured provider. + /// Checks the configured provider. /// public void Validate() { @@ -37,19 +37,17 @@ public class DistributedRuntimeLockProviderValidator( if (options.Value.AllowLocalLockProviderInDistributedRuntime) { - logger.LogWarning( - "Distributed workflow runtime is using local-only lock provider {LocalDistributedLockProviderType} through configured provider {DistributedLockProviderType}. This is safe only for single-host deployments.", + logger.LogDebug( + "Distributed workflow runtime is using acknowledged local-only lock provider {LocalDistributedLockProviderType} through configured provider {DistributedLockProviderType}.", localProviderTypeName, configuredProviderTypeName); return; } - var message = - $"The distributed workflow runtime is configured with local-only distributed lock provider '{localProviderTypeName}' through '{configuredProviderTypeName}'. " + - "This provider does not coordinate across application nodes and can allow concurrent workflow processing in clustered deployments. " + - "Configure a cross-node IDistributedLockProvider such as Redis, SQL Server, or PostgreSQL, or explicitly set DistributedLockingOptions.AllowLocalLockProviderInDistributedRuntime to true for single-host development/test deployments."; - - throw new InvalidOperationException(message); + logger.LogWarning( + "Distributed workflow runtime is using local-only lock provider {LocalDistributedLockProviderType} through configured provider {DistributedLockProviderType}. This provider does not coordinate across application nodes and can allow concurrent workflow processing in clustered deployments. Configure a cross-node IDistributedLockProvider such as Redis, SQL Server, or PostgreSQL for clustered deployments, or set DistributedLockingOptions.AllowLocalLockProviderInDistributedRuntime to true to acknowledge single-host development/test usage.", + localProviderTypeName, + configuredProviderTypeName); } private Type? FindLocalProviderType(IDistributedLockProvider provider, ISet? visited = null) diff --git a/src/modules/Elsa.Workflows.Runtime.Distributed/StartupTasks/ValidateDistributedRuntimeLockProviderStartupTask.cs b/src/modules/Elsa.Workflows.Runtime.Distributed/StartupTasks/ValidateDistributedRuntimeLockProviderStartupTask.cs index d59c2e17a..565fd9962 100644 --- a/src/modules/Elsa.Workflows.Runtime.Distributed/StartupTasks/ValidateDistributedRuntimeLockProviderStartupTask.cs +++ b/src/modules/Elsa.Workflows.Runtime.Distributed/StartupTasks/ValidateDistributedRuntimeLockProviderStartupTask.cs @@ -5,7 +5,7 @@ using JetBrains.Annotations; namespace Elsa.Workflows.Runtime.Distributed.StartupTasks; /// -/// Fails startup when distributed runtime uses a local-only lock provider without explicit opt-in. +/// Checks the distributed runtime lock provider during startup. /// [UsedImplicitly] [Order(-1000)] diff --git a/test/unit/Elsa.Common.UnitTests/Multitenancy/TenantTaskLifecycleCoordinatorTests.cs b/test/unit/Elsa.Common.UnitTests/Multitenancy/TenantTaskLifecycleCoordinatorTests.cs index 042d1fd10..bbed1435d 100644 --- a/test/unit/Elsa.Common.UnitTests/Multitenancy/TenantTaskLifecycleCoordinatorTests.cs +++ b/test/unit/Elsa.Common.UnitTests/Multitenancy/TenantTaskLifecycleCoordinatorTests.cs @@ -1,6 +1,7 @@ using Elsa.Common.Multitenancy; using Elsa.Common.RecurringTasks; using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Logging; using Microsoft.Extensions.Logging.Abstractions; using NSubstitute; @@ -18,6 +19,40 @@ public class TenantTaskLifecycleCoordinatorTests : IAsyncDisposable public ValueTask DisposeAsync() => _coordinator.DisposeAsync(); + [Fact] + public async Task ActivateAsync_WithQueuedWork_ReturnsBeforeWorkItemCompletes() + { + QueueBlockingWorkStartupTask? startupTask = null; + var serviceProvider = BuildTenantServiceProvider( + startupTasks: [sp => startupTask = ActivatorUtilities.CreateInstance(sp)], + backgroundTasks: [sp => ActivatorUtilities.CreateInstance(sp)]); + var activationTask = _coordinator.ActivateTenantAsync(new TenantActivatedEventArgs(_tenant, CreateTenantScope(_tenant, serviceProvider), CancellationToken.None)); + + var completedTask = await Task.WhenAny(activationTask, Task.Delay(TimeSpan.FromSeconds(1))); + + Assert.Same(activationTask, completedTask); + await activationTask; + await startupTask!.WorkStarted.Task.WaitAsync(TimeSpan.FromSeconds(1)); + Assert.False(startupTask.WorkCompleted.Task.IsCompleted); + + startupTask.ReleaseWork(); + await startupTask.WorkCompleted.Task.WaitAsync(TimeSpan.FromSeconds(1)); + } + + [Fact] + public async Task DeactivateAsync_WithRunningBackgroundTask_StopsTask() + { + var backgroundTask = new TrackingBackgroundTask(); + var serviceProvider = BuildTenantServiceProvider(backgroundTasks: [_ => backgroundTask]); + + await _coordinator.ActivateTenantAsync(new TenantActivatedEventArgs(_tenant, CreateTenantScope(_tenant, serviceProvider), CancellationToken.None)); + await backgroundTask.Started.Task.WaitAsync(TimeSpan.FromSeconds(1)); + await _coordinator.DeactivateTenantAsync(DeactivationArgs(serviceProvider)); + + Assert.True(backgroundTask.WasStopCalled); + Assert.True(backgroundTask.WasCancelled); + } + /// /// Regression test: before the fix, TryRemove was called before WaitAsync, /// so a cancelled WaitAsync would orphan the state (removed from the dictionary but @@ -80,6 +115,9 @@ public class TenantTaskLifecycleCoordinatorTests : IAsyncDisposable private TenantDeactivatedEventArgs DeactivationArgs(CancellationToken cancellationToken = default) => DeactivationArgs(_tenant, cancellationToken); + private TenantDeactivatedEventArgs DeactivationArgs(IServiceProvider serviceProvider, CancellationToken cancellationToken = default) => + new(_tenant, CreateTenantScope(_tenant, serviceProvider), cancellationToken); + private TenantDeactivatedEventArgs DeactivationArgs(Tenant tenant, CancellationToken cancellationToken = default) => new(tenant, CreateTenantScope(tenant, _serviceProvider), cancellationToken); @@ -91,13 +129,22 @@ public class TenantTaskLifecycleCoordinatorTests : IAsyncDisposable return new TenantTaskLifecycleCoordinator(scheduleManager, NullLogger.Instance); } - private static IServiceProvider BuildTenantServiceProvider(IEnumerable? recurringTasks = null) + private static IServiceProvider BuildTenantServiceProvider( + IEnumerable? recurringTasks = null, + IEnumerable>? startupTasks = null, + IEnumerable>? backgroundTasks = null) { var services = new ServiceCollection(); - services.AddSingleton(new InstantTaskExecutor()); - services.AddSingleton(Substitute.For()); + services.AddSingleton(new DirectTaskExecutor()); + services.AddSingleton(); + services.AddSingleton(); + services.AddSingleton>(NullLogger.Instance); + foreach (var taskFactory in startupTasks ?? []) + services.AddSingleton(taskFactory); foreach (var task in recurringTasks ?? []) services.AddSingleton(task); + foreach (var taskFactory in backgroundTasks ?? []) + services.AddSingleton(taskFactory); return services.BuildServiceProvider(); } @@ -126,8 +173,61 @@ public class TenantTaskLifecycleCoordinatorTests : IAsyncDisposable } } - private class InstantTaskExecutor : ITaskExecutor + private class QueueBlockingWorkStartupTask(ITenantBackgroundWorkQueue workQueue) : IStartupTask { - public Task ExecuteTaskAsync(ITask task, CancellationToken cancellationToken) => Task.CompletedTask; + private readonly TaskCompletionSource _releaseWork = new(TaskCreationOptions.RunContinuationsAsynchronously); + + public TaskCompletionSource WorkStarted { get; } = new(TaskCreationOptions.RunContinuationsAsynchronously); + public TaskCompletionSource WorkCompleted { get; } = new(TaskCreationOptions.RunContinuationsAsynchronously); + + public async Task ExecuteAsync(CancellationToken cancellationToken) + { + await workQueue.EnqueueAsync(async (_, ct) => + { + WorkStarted.SetResult(); + await _releaseWork.Task.WaitAsync(ct); + WorkCompleted.SetResult(); + }, cancellationToken); + } + + public void ReleaseWork() => _releaseWork.SetResult(); + } + + private class TrackingBackgroundTask : BackgroundTask + { + public TaskCompletionSource Started { get; } = new(TaskCreationOptions.RunContinuationsAsynchronously); + public bool WasCancelled { get; private set; } + public bool WasStopCalled { get; private set; } + + public override async Task ExecuteAsync(CancellationToken cancellationToken) + { + Started.SetResult(); + + try + { + await Task.Delay(Timeout.InfiniteTimeSpan, cancellationToken); + } + catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested) + { + WasCancelled = true; + } + } + + public override Task StopAsync(CancellationToken cancellationToken) + { + WasStopCalled = true; + return Task.CompletedTask; + } + } + + private class DirectTaskExecutor : ITaskExecutor + { + public Task ExecuteTaskAsync(ITask task, CancellationToken cancellationToken) => task.ExecuteAsync(cancellationToken); + } + + private class DirectBackgroundTaskStarter : IBackgroundTaskStarter + { + public Task StartAsync(IBackgroundTask task, CancellationToken cancellationToken) => task.StartAsync(cancellationToken); + public Task StopAsync(IBackgroundTask task, CancellationToken cancellationToken) => task.StopAsync(cancellationToken); } } diff --git a/test/unit/Elsa.Diagnostics.ConsoleLogs.UnitTests/Capture/ConsoleCaptureTeeTests.cs b/test/unit/Elsa.Diagnostics.ConsoleLogs.UnitTests/Capture/ConsoleCaptureTeeTests.cs index a5b0c9026..20ab44406 100644 --- a/test/unit/Elsa.Diagnostics.ConsoleLogs.UnitTests/Capture/ConsoleCaptureTeeTests.cs +++ b/test/unit/Elsa.Diagnostics.ConsoleLogs.UnitTests/Capture/ConsoleCaptureTeeTests.cs @@ -1,162 +1,292 @@ using Elsa.Diagnostics.ConsoleLogs.Contracts; using Elsa.Diagnostics.ConsoleLogs.Services; +using Microsoft.Extensions.Logging; namespace Elsa.Diagnostics.ConsoleLogs.UnitTests.Capture; -public class ConsoleCaptureTeeTests +[Collection(ConsoleHostStateCollection.Name)] +public class ConsoleCaptureTeeTests : IDisposable { - [Fact] - public async Task StartAsync_PreservesOriginalConsoleOutputAndPublishesLine() + private readonly TextWriter _originalOut = Console.Out; + private readonly TextWriter _originalError = Console.Error; + private readonly StringWriter _consoleOutput = new(); + private readonly CapturingProvider _provider = new(); + private readonly IOptions _options = Microsoft.Extensions.Options.Options.Create(new ConsoleLogsOptions()); + private readonly ConsoleLogScopeAccessor _scopeAccessor = new(); + + public ConsoleCaptureTeeTests() { - var originalOut = Console.Out; - var originalError = Console.Error; - using var consoleOutput = new StringWriter(); - var provider = new CapturingProvider(); - var registry = new ConsoleLogSourceRegistry(Microsoft.Extensions.Options.Options.Create(new ConsoleLogsOptions())); - var options = Microsoft.Extensions.Options.Options.Create(new ConsoleLogsOptions()); - var capture = new ConsoleCaptureTee(provider, registry, new ConsoleLogRedactor(options), new ConsoleLineFormatter(options), options); + // Tear down any process-wide host instance so its hook subscription doesn't double-publish. + ConsoleLogsHost.ShutdownAsync().AsTask().GetAwaiter().GetResult(); + ConsoleStreamHook.Uninstall(); + Console.SetOut(_consoleOutput); + } - try - { - Console.SetOut(consoleOutput); - await capture.StartAsync(); + public void Dispose() + { + ConsoleStreamHook.Uninstall(); + ConsoleLogsHost.ShutdownAsync().AsTask().GetAwaiter().GetResult(); + Console.SetOut(_originalOut); + Console.SetError(_originalError); + _consoleOutput.Dispose(); + } - Console.WriteLine("hello"); - await WaitForLineAsync(provider); + private ConsoleCaptureTee CreateCapture() => new( + _provider, + new ConsoleLogSourceRegistry(_options), + new ConsoleLogRedactor(_options), + new ConsoleLineFormatter(_options), + _scopeAccessor, + _options); - Assert.Equal($"hello{Environment.NewLine}", consoleOutput.ToString()); - Assert.Single(provider.Lines); - Assert.Equal("hello", provider.Lines[0].Text); - Assert.Equal(ConsoleLogStream.Stdout, provider.Lines[0].Stream); - } - finally - { - await capture.StopAsync(); - Console.SetOut(originalOut); - Console.SetError(originalError); - } + [Fact] + public async Task Capture_PreservesOriginalConsoleOutputAndPublishesLine() + { + await using var capture = CreateCapture(); + + Console.WriteLine("hello"); + await WaitForLinesAsync(_provider, 1); + + Assert.Equal($"hello{Environment.NewLine}", _consoleOutput.ToString()); + var line = Assert.Single(_provider.Lines); + Assert.Equal("hello", line.Text); + Assert.Equal(ConsoleLogStream.Stdout, line.Stream); } [Fact] - public async Task StartAsync_WhenAlreadyStarted_DoesNotWrapConsoleTwice() + public async Task Capture_RedactsBeforePublishingToProvider() { - var originalOut = Console.Out; - var originalError = Console.Error; - using var consoleOutput = new StringWriter(); - var provider = new CapturingProvider(); - var registry = new ConsoleLogSourceRegistry(Microsoft.Extensions.Options.Options.Create(new ConsoleLogsOptions())); - var options = Microsoft.Extensions.Options.Options.Create(new ConsoleLogsOptions()); - var capture = new ConsoleCaptureTee(provider, registry, new ConsoleLogRedactor(options), new ConsoleLineFormatter(options), options); + await using var capture = CreateCapture(); - try - { - Console.SetOut(consoleOutput); - await capture.StartAsync(); - await capture.StartAsync(); + Console.WriteLine("token=secret-value"); + await WaitForLinesAsync(_provider, 1); - Console.WriteLine("hello"); - await WaitForLineAsync(provider); - - Assert.Equal($"hello{Environment.NewLine}", consoleOutput.ToString()); - Assert.Single(provider.Lines); - } - finally - { - await capture.StopAsync(); - await capture.StopAsync(); - Console.SetOut(originalOut); - Console.SetError(originalError); - } + Assert.Equal("[Redacted]", _provider.Lines[0].Text); } [Fact] - public async Task StartAsync_RedactsBeforePublishingToProvider() + public async Task Capture_AfterDispose_StopsPublishing() { - var originalOut = Console.Out; - var originalError = Console.Error; - using var consoleOutput = new StringWriter(); - var provider = new CapturingProvider(); - var registry = new ConsoleLogSourceRegistry(Microsoft.Extensions.Options.Options.Create(new ConsoleLogsOptions())); - var options = Microsoft.Extensions.Options.Options.Create(new ConsoleLogsOptions()); - var capture = new ConsoleCaptureTee(provider, registry, new ConsoleLogRedactor(options), new ConsoleLineFormatter(options), options); + var capture = CreateCapture(); - try - { - Console.SetOut(consoleOutput); - await capture.StartAsync(); + Console.WriteLine("first"); + await WaitForLinesAsync(_provider, 1); - Console.WriteLine("token=secret-value"); - await WaitForLineAsync(provider); + await capture.DisposeAsync(); - Assert.Equal("[Redacted]", provider.Lines[0].Text); - } - finally - { - await capture.StopAsync(); - Console.SetOut(originalOut); - Console.SetError(originalError); - } + Console.WriteLine("second"); + await Task.Delay(50); + + Assert.Single(_provider.Lines); + Assert.Equal("first", _provider.Lines[0].Text); } [Fact] - public async Task StopAsync_WhenStartedTwice_KeepsCaptureActiveUntilSecondStop() + public async Task Capture_FansOutToMultipleConcurrentSubscribers() { - var originalOut = Console.Out; - var originalError = Console.Error; - using var consoleOutput = new StringWriter(); - var provider = new CapturingProvider(); - var registry = new ConsoleLogSourceRegistry(Microsoft.Extensions.Options.Options.Create(new ConsoleLogsOptions())); - var options = Microsoft.Extensions.Options.Options.Create(new ConsoleLogsOptions()); - var capture = new ConsoleCaptureTee(provider, registry, new ConsoleLogRedactor(options), new ConsoleLineFormatter(options), options); + var secondProvider = new CapturingProvider(); + await using var first = CreateCapture(); + await using var second = new ConsoleCaptureTee( + secondProvider, + new ConsoleLogSourceRegistry(_options), + new ConsoleLogRedactor(_options), + new ConsoleLineFormatter(_options), + _scopeAccessor, + _options); - try - { - Console.SetOut(consoleOutput); - await capture.StartAsync(); - await capture.StartAsync(); + Console.WriteLine("broadcast"); + await WaitForLinesAsync(_provider, 1); + await WaitForLinesAsync(secondProvider, 1); - await capture.StopAsync(); - Console.WriteLine("still captured"); - await WaitForLineAsync(provider); - - Assert.Single(provider.Lines); - Assert.Equal("still captured", provider.Lines[0].Text); - - await capture.StopAsync(); - Console.WriteLine("not captured"); - await Task.Delay(50); - - Assert.Single(provider.Lines); - } - finally - { - await capture.StopAsync(); - Console.SetOut(originalOut); - Console.SetError(originalError); - } + Assert.Equal("broadcast", _provider.Lines[^1].Text); + Assert.Equal("broadcast", secondProvider.Lines[^1].Text); } - private static async Task WaitForLineAsync(CapturingProvider provider) + [Fact] + public async Task Capture_ReplaysStartupWritesOnlyOnce() { - var timeout = DateTimeOffset.UtcNow.AddSeconds(2); - while (provider.Lines.Count == 0 && DateTimeOffset.UtcNow < timeout) + ConsoleStreamHook.Install(); + Console.WriteLine("startup"); + + var secondProvider = new CapturingProvider(); + await using var first = CreateCapture(); + await WaitForLinesAsync(_provider, 1); + + await using var second = new ConsoleCaptureTee( + secondProvider, + new ConsoleLogSourceRegistry(_options), + new ConsoleLogRedactor(_options), + new ConsoleLineFormatter(_options), + _scopeAccessor, + _options); + await Task.Delay(50); + + Assert.Equal("startup", _provider.Lines[0].Text); + Assert.Empty(secondProvider.Lines); + } + + [Fact] + public async Task Capture_SuppressesPublishPathReentrancy() + { + var loggingProvider = new ReentrantProvider(); + await using var capture = new ConsoleCaptureTee( + loggingProvider, + new ConsoleLogSourceRegistry(_options), + new ConsoleLogRedactor(_options), + new ConsoleLineFormatter(_options), + _scopeAccessor, + _options); + + Console.WriteLine("trigger"); + await WaitForLinesAsync(loggingProvider, 1); + await Task.Delay(50); // Give any reentrant publish a chance to misbehave. + + Assert.Single(loggingProvider.Lines); + Assert.Equal("trigger", loggingProvider.Lines[0].Text); + } + + [Fact] + public async Task Capture_ContinuesAfterProviderOperationCanceledException() + { + var provider = new OperationCanceledOnceProvider(); + await using var capture = new ConsoleCaptureTee( + provider, + new ConsoleLogSourceRegistry(_options), + new ConsoleLogRedactor(_options), + new ConsoleLineFormatter(_options), + _scopeAccessor, + _options); + + Console.WriteLine("first"); + Console.WriteLine("second"); + await WaitForLinesAsync(provider, 1); + + var line = Assert.Single(provider.Lines); + Assert.Equal("second", line.Text); + } + + [Fact] + public async Task Capture_AttachesCurrentWorkflowInstanceId() + { + using var loggerFactory = LoggerFactory.Create(builder => builder.AddProvider(_scopeAccessor)); + var logger = loggerFactory.CreateLogger("workflow"); + await using var capture = CreateCapture(); + using var scope = logger.BeginScope(new Dictionary + { + ["WorkflowInstanceId"] = "workflow-instance-a" + }); + + Console.WriteLine("scoped"); + await WaitForLinesAsync(_provider, 1); + + Assert.Equal("workflow-instance-a", _provider.Lines[0].WorkflowInstanceId); + } + + [Fact] + public async Task Capture_AttachesInnermostWorkflowInstanceId() + { + using var loggerFactory = LoggerFactory.Create(builder => builder.AddProvider(_scopeAccessor)); + var logger = loggerFactory.CreateLogger("workflow"); + await using var capture = CreateCapture(); + using var outerScope = logger.BeginScope(new Dictionary + { + ["WorkflowInstanceId"] = "workflow-instance-a" + }); + using var innerScope = logger.BeginScope(new Dictionary + { + ["WorkflowInstanceId"] = "workflow-instance-b" + }); + + Console.WriteLine("scoped"); + await WaitForLinesAsync(_provider, 1); + + Assert.Equal("workflow-instance-b", _provider.Lines[0].WorkflowInstanceId); + } + + [Fact] + public async Task Capture_PreservesScopeAcrossMultipleLoggerFactories() + { + using var firstLoggerFactory = LoggerFactory.Create(builder => builder.AddProvider(_scopeAccessor)); + using var secondLoggerFactory = LoggerFactory.Create(builder => builder.AddProvider(_scopeAccessor)); + var logger = firstLoggerFactory.CreateLogger("workflow"); + await using var capture = CreateCapture(); + using var scope = logger.BeginScope(new Dictionary + { + ["WorkflowInstanceId"] = "workflow-instance-a" + }); + + _ = secondLoggerFactory.CreateLogger("other"); + Console.WriteLine("scoped"); + await WaitForLinesAsync(_provider, 1); + + Assert.Equal("workflow-instance-a", _provider.Lines[0].WorkflowInstanceId); + } + + [Fact] + public async Task Capture_AttachesWorkflowInstanceIdCapturedBeforeAsyncConsoleWrite() + { + using var loggerFactory = LoggerFactory.Create(builder => builder.AddProvider(_scopeAccessor)); + var logger = loggerFactory.CreateLogger("workflow"); + await using var capture = CreateCapture(); + + using (logger.BeginScope(new Dictionary + { + ["WorkflowInstanceId"] = "workflow-instance-a" + })) + { + logger.LogInformation("scoped"); + } + + await Task.Run(() => Console.WriteLine("scoped")); + await WaitForLinesAsync(_provider, 1); + + Assert.Equal("workflow-instance-a", _provider.Lines[0].WorkflowInstanceId); + } + + [Fact] + public async Task Capture_DoesNotAttachCapturedWorkflowInstanceIdToUnrelatedConsoleWrite() + { + using var loggerFactory = LoggerFactory.Create(builder => builder.AddProvider(_scopeAccessor)); + var logger = loggerFactory.CreateLogger("workflow"); + await using var capture = CreateCapture(); + + using (logger.BeginScope(new Dictionary + { + ["WorkflowInstanceId"] = "workflow-instance-a" + })) + { + logger.LogInformation("scoped"); + } + + await Task.Run(() => Console.WriteLine("not scoped")); + await Task.Run(() => Console.WriteLine("scoped")); + await WaitForLinesAsync(_provider, 2); + + Assert.Null(_provider.Lines[0].WorkflowInstanceId); + Assert.Equal("workflow-instance-a", _provider.Lines[1].WorkflowInstanceId); + } + + private static async Task WaitForLinesAsync(CapturingProvider provider, int count) + { + var deadline = DateTimeOffset.UtcNow.AddSeconds(2); + while (provider.Lines.Count < count && DateTimeOffset.UtcNow < deadline) await Task.Delay(10); } - private sealed class CapturingProvider : IConsoleLogProvider + private class CapturingProvider : IConsoleLogProvider { public List Lines { get; } = []; - public ValueTask PublishAsync(ConsoleLogLine line, CancellationToken cancellationToken = default) + public virtual ValueTask PublishAsync(ConsoleLogLine line, CancellationToken cancellationToken = default) { - Lines.Add(line); + lock (Lines) + Lines.Add(line); + return ValueTask.CompletedTask; } - public ValueTask GetRecentAsync(ConsoleLogFilter filter, CancellationToken cancellationToken = default) - { - return ValueTask.FromResult(new RecentConsoleLogsResult(Lines)); - } + public ValueTask GetRecentAsync(ConsoleLogFilter filter, CancellationToken cancellationToken = default) => + ValueTask.FromResult(new RecentConsoleLogsResult(Lines)); public async IAsyncEnumerable SubscribeAsync(ConsoleLogFilter filter, [EnumeratorCancellation] CancellationToken cancellationToken = default) { @@ -165,9 +295,31 @@ public class ConsoleCaptureTeeTests yield return ConsoleLogStreamItem.FromLine(line); } - public ValueTask> ListSourcesAsync(CancellationToken cancellationToken = default) + public ValueTask> ListSourcesAsync(CancellationToken cancellationToken = default) => + ValueTask.FromResult>([]); + } + + private sealed class ReentrantProvider : CapturingProvider + { + public override ValueTask PublishAsync(ConsoleLogLine line, CancellationToken cancellationToken = default) { - return ValueTask.FromResult>([]); + // Simulate a provider whose internals write to the console (e.g., SignalR diagnostics). + // With SuppressCapture set on this async context, this MUST NOT recurse. + Console.WriteLine("[provider-internal-log]"); + return base.PublishAsync(line, cancellationToken); + } + } + + private sealed class OperationCanceledOnceProvider : CapturingProvider + { + private int _throwCount; + + public override ValueTask PublishAsync(ConsoleLogLine line, CancellationToken cancellationToken = default) + { + if (Interlocked.Increment(ref _throwCount) == 1) + throw new OperationCanceledException(); + + return base.PublishAsync(line, cancellationToken); } } } diff --git a/test/unit/Elsa.Diagnostics.ConsoleLogs.UnitTests/Capture/ConsoleLineBufferTests.cs b/test/unit/Elsa.Diagnostics.ConsoleLogs.UnitTests/Capture/ConsoleLineBufferTests.cs index df9cf532c..deac0f6a2 100644 --- a/test/unit/Elsa.Diagnostics.ConsoleLogs.UnitTests/Capture/ConsoleLineBufferTests.cs +++ b/test/unit/Elsa.Diagnostics.ConsoleLogs.UnitTests/Capture/ConsoleLineBufferTests.cs @@ -13,7 +13,7 @@ public class ConsoleLineBufferTests var lines = _buffer.Append("lo\n", DateTimeOffset.UtcNow); - Assert.Equal(["hello"], lines); + Assert.Equal(["hello"], lines.Select(x => x.Text)); } [Fact] @@ -21,8 +21,8 @@ public class ConsoleLineBufferTests { var lines = _buffer.Append("hello!", DateTimeOffset.UtcNow); - Assert.Equal(["hello"], lines); - Assert.Equal("!", _buffer.Flush()); + Assert.Equal(["hello"], lines.Select(x => x.Text)); + Assert.Equal("!", _buffer.Flush()?.Text); } [Fact] @@ -33,6 +33,36 @@ public class ConsoleLineBufferTests _buffer.Append("tail", now); Assert.Null(_buffer.FlushIfIdle(now.AddMilliseconds(500))); - Assert.Equal("tail", _buffer.FlushIfIdle(now.AddSeconds(2))); + Assert.Equal("tail", _buffer.FlushIfIdle(now.AddSeconds(2))?.Text); + } + + [Fact] + public void Append_PreservesEmptyLinesBetweenWrites() + { + var lines = _buffer.Append("a\n\nb\n", DateTimeOffset.UtcNow); + + Assert.Equal(["a", "", "b"], lines.Select(x => x.Text)); + } + + [Fact] + public void Append_PreservesWorkflowInstanceIdAcrossPartialWrites() + { + _buffer.Append("hel", DateTimeOffset.UtcNow, "workflow-instance-a"); + + var line = Assert.Single(_buffer.Append("lo\n", DateTimeOffset.UtcNow)); + + Assert.Equal("hello", line.Text); + Assert.Equal("workflow-instance-a", line.WorkflowInstanceId); + } + + [Fact] + public void Append_DoesNotAssignWorkflowInstanceIdAfterUnscopedPartialWrite() + { + _buffer.Append("hel", DateTimeOffset.UtcNow); + + var line = Assert.Single(_buffer.Append("lo\n", DateTimeOffset.UtcNow, "workflow-instance-a")); + + Assert.Equal("hello", line.Text); + Assert.Null(line.WorkflowInstanceId); } } diff --git a/test/unit/Elsa.Diagnostics.ConsoleLogs.UnitTests/Capture/ConsoleLineFormatterTests.cs b/test/unit/Elsa.Diagnostics.ConsoleLogs.UnitTests/Capture/ConsoleLineFormatterTests.cs index 4ce6ef5bd..ef5645295 100644 --- a/test/unit/Elsa.Diagnostics.ConsoleLogs.UnitTests/Capture/ConsoleLineFormatterTests.cs +++ b/test/unit/Elsa.Diagnostics.ConsoleLogs.UnitTests/Capture/ConsoleLineFormatterTests.cs @@ -5,16 +5,26 @@ namespace Elsa.Diagnostics.ConsoleLogs.UnitTests.Capture; public class ConsoleLineFormatterTests { [Fact] - public void Format_StripsAnsiByDefault() + public void Format_PreservesAnsiByDefault() { var formatter = new ConsoleLineFormatter(Microsoft.Extensions.Options.Options.Create(new ConsoleLogsOptions())); var result = formatter.Format("\u001b[31mred\u001b[0m"); - Assert.Equal("red", result.Text); + Assert.Equal("\u001b[31mred\u001b[0m", result.Text); Assert.False(result.Truncated); } + [Fact] + public void Format_StripsAnsiWhenConfigured() + { + var formatter = new ConsoleLineFormatter(Microsoft.Extensions.Options.Options.Create(new ConsoleLogsOptions { StripAnsiEscapeSequences = true })); + + var result = formatter.Format("\u001b[31mred\u001b[0m"); + + Assert.Equal("red", result.Text); + } + [Fact] public void Format_PreservesAnsiWhenConfigured() { diff --git a/test/unit/Elsa.Diagnostics.ConsoleLogs.UnitTests/ConsoleHostStateCollection.cs b/test/unit/Elsa.Diagnostics.ConsoleLogs.UnitTests/ConsoleHostStateCollection.cs new file mode 100644 index 000000000..221316362 --- /dev/null +++ b/test/unit/Elsa.Diagnostics.ConsoleLogs.UnitTests/ConsoleHostStateCollection.cs @@ -0,0 +1,15 @@ +using Elsa.Diagnostics.ConsoleLogs.Services; + +namespace Elsa.Diagnostics.ConsoleLogs.UnitTests; + +/// +/// xUnit collection used by every test class that mutates the process-wide +/// or . Members of this collection run sequentially, preventing parallel +/// tests from racing on the same static state. +/// +[CollectionDefinition(Name)] +public sealed class ConsoleHostStateCollection +{ + public const string Name = "ConsoleHostState"; +} + diff --git a/test/unit/Elsa.Diagnostics.ConsoleLogs.UnitTests/ConsoleLogsRegistrationTests.cs b/test/unit/Elsa.Diagnostics.ConsoleLogs.UnitTests/ConsoleLogsRegistrationTests.cs index 06b26c3aa..76af179cd 100644 --- a/test/unit/Elsa.Diagnostics.ConsoleLogs.UnitTests/ConsoleLogsRegistrationTests.cs +++ b/test/unit/Elsa.Diagnostics.ConsoleLogs.UnitTests/ConsoleLogsRegistrationTests.cs @@ -4,32 +4,209 @@ using Elsa.Diagnostics.ConsoleLogs.Providers.InMemory; using Elsa.Diagnostics.ConsoleLogs.Services; using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.Hosting; +using Microsoft.Extensions.Logging; +using Microsoft.Extensions.Options; namespace Elsa.Diagnostics.ConsoleLogs.UnitTests; -public class ConsoleLogsRegistrationTests +[Collection(ConsoleHostStateCollection.Name)] +public class ConsoleLogsRegistrationTests : IAsyncLifetime { + public async Task InitializeAsync() + { + await ConsoleLogsHost.ShutdownAsync(); + ConsoleStreamHook.Uninstall(); + } + + public async Task DisposeAsync() + { + await ConsoleLogsHost.ShutdownAsync(); + ConsoleStreamHook.Uninstall(); + } + [Fact] - public void AddConsoleLogsServices_WhenNoProviderIsConfigured_UsesInMemoryProvider() + public async Task AddConsoleLogsServices_ResolvesHostOwnedSingletons() { var services = new ServiceCollection(); - services.AddConsoleLogsServices(); - using var serviceProvider = services.BuildServiceProvider(); + await using var serviceProvider = services.BuildServiceProvider(); + Assert.IsType(serviceProvider.GetRequiredService()); Assert.IsType(serviceProvider.GetRequiredService()); Assert.IsType(serviceProvider.GetRequiredService()); + Assert.Same(ConsoleLogsHost.ScopeAccessor, serviceProvider.GetRequiredService()); + Assert.Contains(serviceProvider.GetServices(), x => ReferenceEquals(x, ConsoleLogsHost.ScopeAccessor)); + Assert.Contains(serviceProvider.GetServices(), x => x.GetType() == typeof(ConsoleLogsHostedService)); +#pragma warning disable CS0618 + Assert.IsType(serviceProvider.GetRequiredService()); +#pragma warning restore CS0618 + + // Same instance must be visible across separate shell containers — console output is process-wide. + var services2 = new ServiceCollection(); + services2.AddConsoleLogsServices(); + await using var serviceProvider2 = services2.BuildServiceProvider(); + + Assert.Same(serviceProvider.GetRequiredService(), serviceProvider2.GetRequiredService()); + Assert.Same(serviceProvider.GetRequiredService(), serviceProvider2.GetRequiredService()); } [Fact] - public void AddConsoleLogsServices_RegistersHostedCapture() + public async Task AddConsoleLogsServices_UsesCustomProviderForCapturePipeline() { + var provider = new CustomProvider(); var services = new ServiceCollection(); - + services.AddSingleton(provider); services.AddConsoleLogsServices(); - using var serviceProvider = services.BuildServiceProvider(); - Assert.Contains(serviceProvider.GetServices(), x => x.GetType() == typeof(ConsoleLogCaptureHostedService)); + await using var serviceProvider = services.BuildServiceProvider(); + + _ = serviceProvider.GetRequiredService(); + + foreach (var hostedService in serviceProvider.GetServices()) + await hostedService.StartAsync(CancellationToken.None); + + Assert.Same(provider, serviceProvider.GetRequiredService()); + Assert.Same(provider, ConsoleLogsHost.Provider); + + foreach (var hostedService in serviceProvider.GetServices()) + await hostedService.StopAsync(CancellationToken.None); + } + + [Fact] + public async Task AddConsoleLogsServices_UsesCustomProviderRegisteredAfterConsoleLogsServices() + { + var provider = new CustomProvider(); + var services = new ServiceCollection(); + services.AddConsoleLogsServices(); + services.AddSingleton(provider); + + await using var serviceProvider = services.BuildServiceProvider(); + + foreach (var hostedService in serviceProvider.GetServices()) + await hostedService.StartAsync(CancellationToken.None); + + Assert.Same(provider, serviceProvider.GetRequiredService()); + Assert.Same(provider, ConsoleLogsHost.Provider); + + foreach (var hostedService in serviceProvider.GetServices()) + await hostedService.StopAsync(CancellationToken.None); + } + + [Fact] + public async Task AddConsoleLogsServices_CreatesCustomProviderWithHostOwnedDependencies() + { + var services = new ServiceCollection(); + services.AddConsoleLogsServices(); + services.AddSingleton(); + + await using var serviceProvider = services.BuildServiceProvider(); + + foreach (var hostedService in serviceProvider.GetServices()) + await hostedService.StartAsync(CancellationToken.None); + + var provider = Assert.IsType(ConsoleLogsHost.Provider); + Assert.Same(serviceProvider.GetRequiredService(), provider); + Assert.Same(ConsoleLogsHost.Options, provider.Options); + Assert.Same(ConsoleLogsHost.SourceRegistry, provider.SourceRegistry); + Assert.Same(ConsoleLogsHost.Redactor, provider.Redactor); + Assert.Same(ConsoleLogsHost.Formatter, provider.Formatter); + Assert.Same(ConsoleLogsHost.ScopeAccessor, provider.ScopeAccessor); + + foreach (var hostedService in serviceProvider.GetServices()) + await hostedService.StopAsync(CancellationToken.None); + } + + [Fact] + public async Task ShutdownAsync_ResetsReferenceLeases() + { + ConsoleLogsHost.AddReference(); + ConsoleLogsHost.EnsureInitialized(); + + await ConsoleLogsHost.ShutdownAsync(); + + ConsoleLogsHost.AddReference(); + ConsoleLogsHost.EnsureInitialized(); + await ConsoleLogsHost.ReleaseReferenceAsync(); + + var provider = new CustomProvider(); + Assert.True(ConsoleLogsHost.ConfigureProvider((_, _) => provider)); + Assert.Same(provider, ConsoleLogsHost.Provider); + } + + [Fact] + public async Task AddConsoleLogsHost_RegistersHostedFlushService() + { + var services = new ServiceCollection(); + services.AddConsoleLogsHost(); + + await using var serviceProvider = services.BuildServiceProvider(); + + Assert.Contains(serviceProvider.GetServices(), x => x.GetType() == typeof(ConsoleLogsHostedService)); + Assert.Contains(serviceProvider.GetServices(), x => ReferenceEquals(x, ConsoleLogsHost.ScopeAccessor)); + } + + [Fact] + public void AddConsoleLogsHost_AppliesConfigurationWhenCalledBeforeFirstAccess() + { + var services = new ServiceCollection(); + services.AddConsoleLogsHost(options => options.RecentLogCapacity = 17); + + Assert.Equal(17, ConsoleLogsHost.Options.Value.RecentLogCapacity); + } + + [Fact] + public void AddConsoleLogsHost_DoesNotInitializeHostDuringRegistration() + { + var services = new ServiceCollection(); + services.AddConsoleLogsHost(); + + Assert.True(ConsoleLogsHost.Configure(options => options.RecentLogCapacity = 23)); + Assert.Equal(23, ConsoleLogsHost.Options.Value.RecentLogCapacity); + } + + private sealed class CustomProvider : IConsoleLogProvider + { + public ValueTask PublishAsync(ConsoleLogLine line, CancellationToken cancellationToken = default) => ValueTask.CompletedTask; + + public ValueTask GetRecentAsync(ConsoleLogFilter filter, CancellationToken cancellationToken = default) => + ValueTask.FromResult(new RecentConsoleLogsResult([])); + + public async IAsyncEnumerable SubscribeAsync(ConsoleLogFilter filter, [EnumeratorCancellation] CancellationToken cancellationToken = default) + { + await Task.CompletedTask; + yield break; + } + + public ValueTask> ListSourcesAsync(CancellationToken cancellationToken = default) => + ValueTask.FromResult>([]); + } + + private sealed class ProviderWithHostDependencies( + IOptions options, + IConsoleLogSourceRegistry sourceRegistry, + IConsoleLogRedactor redactor, + ConsoleLineFormatter formatter, + ConsoleLogScopeAccessor scopeAccessor) : IConsoleLogProvider + { + public IOptions Options { get; } = options; + public IConsoleLogSourceRegistry SourceRegistry { get; } = sourceRegistry; + public IConsoleLogRedactor Redactor { get; } = redactor; + public ConsoleLineFormatter Formatter { get; } = formatter; + public ConsoleLogScopeAccessor ScopeAccessor { get; } = scopeAccessor; + + public ValueTask PublishAsync(ConsoleLogLine line, CancellationToken cancellationToken = default) => ValueTask.CompletedTask; + + public ValueTask GetRecentAsync(ConsoleLogFilter filter, CancellationToken cancellationToken = default) => + ValueTask.FromResult(new RecentConsoleLogsResult([])); + + public async IAsyncEnumerable SubscribeAsync(ConsoleLogFilter filter, [EnumeratorCancellation] CancellationToken cancellationToken = default) + { + await Task.CompletedTask; + yield break; + } + + public ValueTask> ListSourcesAsync(CancellationToken cancellationToken = default) => + ValueTask.FromResult>([]); } } diff --git a/test/unit/Elsa.Diagnostics.ConsoleLogs.UnitTests/Filtering/ConsoleLogFilterTests.cs b/test/unit/Elsa.Diagnostics.ConsoleLogs.UnitTests/Filtering/ConsoleLogFilterTests.cs index 7c3a19ffb..2f44f05bf 100644 --- a/test/unit/Elsa.Diagnostics.ConsoleLogs.UnitTests/Filtering/ConsoleLogFilterTests.cs +++ b/test/unit/Elsa.Diagnostics.ConsoleLogs.UnitTests/Filtering/ConsoleLogFilterTests.cs @@ -9,6 +9,7 @@ public class ConsoleLogFilterTests Text = "workflow started", Stream = ConsoleLogStream.Stdout, Source = new ConsoleLogSource { Id = "source-a", DisplayName = "Source A", MachineName = "machine" }, + WorkflowInstanceId = "workflow-instance-a", ReceivedAt = DateTimeOffset.Parse("2026-05-18T10:00:00Z") }; @@ -33,4 +34,11 @@ public class ConsoleLogFilterTests Assert.False(ConsoleLogFilterEvaluator.Matches(_line, new ConsoleLogFilter { Query = "missing" })); Assert.False(ConsoleLogFilterEvaluator.Matches(_line, new ConsoleLogFilter { From = _line.ReceivedAt.AddSeconds(1) })); } + + [Fact] + public void Matches_FiltersByWorkflowInstanceId() + { + Assert.True(ConsoleLogFilterEvaluator.Matches(_line, new ConsoleLogFilter { WorkflowInstanceId = "workflow-instance-a" })); + Assert.False(ConsoleLogFilterEvaluator.Matches(_line, new ConsoleLogFilter { WorkflowInstanceId = "workflow-instance-b" })); + } } diff --git a/test/unit/Elsa.Diagnostics.ConsoleLogs.UnitTests/Redaction/ConsoleLogProviderRedactionTests.cs b/test/unit/Elsa.Diagnostics.ConsoleLogs.UnitTests/Redaction/ConsoleLogProviderRedactionTests.cs index 12ab23b6f..3e8721400 100644 --- a/test/unit/Elsa.Diagnostics.ConsoleLogs.UnitTests/Redaction/ConsoleLogProviderRedactionTests.cs +++ b/test/unit/Elsa.Diagnostics.ConsoleLogs.UnitTests/Redaction/ConsoleLogProviderRedactionTests.cs @@ -3,39 +3,68 @@ using Elsa.Diagnostics.ConsoleLogs.Services; namespace Elsa.Diagnostics.ConsoleLogs.UnitTests.Redaction; -public class ConsoleLogProviderRedactionTests +[Collection(ConsoleHostStateCollection.Name)] +public class ConsoleLogProviderRedactionTests : IAsyncLifetime { + public async Task InitializeAsync() + { + await ConsoleLogsHost.ShutdownAsync(); + ConsoleStreamHook.Uninstall(); + } + + public async Task DisposeAsync() + { + await ConsoleLogsHost.ShutdownAsync(); + ConsoleStreamHook.Uninstall(); + } + [Fact] public async Task CaptureTee_PublishesRedactedLinesToProvider() { var originalOut = Console.Out; var provider = new CapturingProvider(); - var registry = new ConsoleLogSourceRegistry(Microsoft.Extensions.Options.Options.Create(new ConsoleLogsOptions())); var options = Microsoft.Extensions.Options.Options.Create(new ConsoleLogsOptions()); - var capture = new ConsoleCaptureTee(provider, registry, new ConsoleLogRedactor(options), new ConsoleLineFormatter(options), options); try { + ConsoleStreamHook.Uninstall(); Console.SetOut(TextWriter.Null); - await capture.StartAsync(); + await using var capture = new ConsoleCaptureTee( + provider, + new ConsoleLogSourceRegistry(options), + new ConsoleLogRedactor(options), + new ConsoleLineFormatter(options), + new ConsoleLogScopeAccessor(), + options); + Console.WriteLine(string.Concat("pass", "word", "=", "sample-value")); - await capture.StopAsync(); + await WaitForLineAsync(provider); + Assert.Equal("[Redacted]", Assert.Single(provider.Lines).Text); } finally { - await capture.StopAsync(); + ConsoleStreamHook.Uninstall(); Console.SetOut(originalOut); } } + private static async Task WaitForLineAsync(CapturingProvider provider) + { + var deadline = DateTimeOffset.UtcNow.AddSeconds(2); + while (provider.Lines.Count == 0 && DateTimeOffset.UtcNow < deadline) + await Task.Delay(10); + } + private sealed class CapturingProvider : IConsoleLogProvider { public List Lines { get; } = []; public ValueTask PublishAsync(ConsoleLogLine line, CancellationToken cancellationToken = default) { - Lines.Add(line); + lock (Lines) + Lines.Add(line); + return ValueTask.CompletedTask; } diff --git a/test/unit/Elsa.Diagnostics.ConsoleLogs.UnitTests/Redaction/ConsoleLogRedactorTests.cs b/test/unit/Elsa.Diagnostics.ConsoleLogs.UnitTests/Redaction/ConsoleLogRedactorTests.cs index 6bc4e4773..083eb0677 100644 --- a/test/unit/Elsa.Diagnostics.ConsoleLogs.UnitTests/Redaction/ConsoleLogRedactorTests.cs +++ b/test/unit/Elsa.Diagnostics.ConsoleLogs.UnitTests/Redaction/ConsoleLogRedactorTests.cs @@ -16,6 +16,26 @@ public class ConsoleLogRedactorTests Assert.Equal("Authorization: [Redacted]", redacted.Text); } + [Fact] + public void Redact_MasksSensitiveLineTextSplitByAnsi() + { + var line = CreateLine("to\u001b[31mken=sample-token"); + + var redacted = _redactor.Redact(line); + + Assert.Equal("[Redacted]", redacted.Text); + } + + [Fact] + public void Redact_PreservesAnsiWhenLineTextIsNotSensitive() + { + var line = CreateLine("\u001b[31mhello\u001b[0m"); + + var redacted = _redactor.Redact(line); + + Assert.Equal("\u001b[31mhello\u001b[0m", redacted.Text); + } + [Fact] public void Redact_MasksSensitiveSourceMetadata() { diff --git a/test/unit/Elsa.Scheduling.UnitTests/StartupTasks/CreateSchedulesStartupTaskTests.cs b/test/unit/Elsa.Scheduling.UnitTests/StartupTasks/CreateSchedulesStartupTaskTests.cs new file mode 100644 index 000000000..d6c16a3a6 --- /dev/null +++ b/test/unit/Elsa.Scheduling.UnitTests/StartupTasks/CreateSchedulesStartupTaskTests.cs @@ -0,0 +1,78 @@ +using Elsa.Common; +using Elsa.Common.Multitenancy; +using Elsa.Scheduling.StartupTasks; +using Elsa.Workflows.Runtime; +using Elsa.Workflows.Runtime.Entities; +using Elsa.Workflows.Runtime.Filters; +using Elsa.Workflows.Runtime.Tasks; +using Microsoft.Extensions.DependencyInjection; +using NSubstitute; + +namespace Elsa.Scheduling.UnitTests.StartupTasks; + +public class CreateSchedulesStartupTaskTests +{ + private readonly StoredTrigger[] _triggers = [new() { WorkflowDefinitionId = "definition", WorkflowDefinitionVersionId = "version", ActivityId = "activity" }]; + private readonly StoredBookmark[] _bookmarks = [new() { Hash = "hash", WorkflowInstanceId = "instance" }]; + private readonly ITriggerStore _triggerStore = Substitute.For(); + private readonly IBookmarkStore _bookmarkStore = Substitute.For(); + private readonly ITriggerScheduler _triggerScheduler = Substitute.For(); + private readonly IBookmarkScheduler _bookmarkScheduler = Substitute.For(); + + public CreateSchedulesStartupTaskTests() + { + _triggerStore.FindManyAsync(Arg.Any(), Arg.Any()).Returns(_triggers); + _bookmarkStore.FindManyAsync(Arg.Any(), Arg.Any()).Returns(_bookmarks); + } + + [Fact] + public void Task_DependsOnPopulateRegistriesStartupTask() + { + var dependency = Assert.Single(typeof(CreateSchedulesStartupTask).GetCustomAttributes(typeof(TaskDependencyAttribute), false).Cast()); + + Assert.Equal(typeof(PopulateRegistriesStartupTask), dependency.DependencyTaskType); + } + + [Fact] + public async Task ExecuteAsync_WithoutTenantBackgroundQueue_SchedulesImmediately() + { + var task = new CreateSchedulesStartupTask(CreateServiceProvider()); + + await task.ExecuteAsync(CancellationToken.None); + + await _triggerScheduler.Received(1).ScheduleAsync(Arg.Is>(x => x.SequenceEqual(_triggers)), Arg.Any()); + await _bookmarkScheduler.Received(1).ScheduleAsync(Arg.Is>(x => x.SequenceEqual(_bookmarks)), Arg.Any()); + } + + [Fact] + public async Task ExecuteAsync_WithTenantBackgroundQueue_EnqueuesScheduleCreation() + { + TenantBackgroundWorkItem? workItem = null; + var workQueue = Substitute.For(); + workQueue.EnqueueAsync(Arg.Do(x => workItem = x), Arg.Any()).Returns(ValueTask.CompletedTask); + var serviceProvider = CreateServiceProvider(services => services.AddSingleton(workQueue)); + var task = new CreateSchedulesStartupTask(serviceProvider); + + await task.ExecuteAsync(CancellationToken.None); + + await workQueue.Received(1).EnqueueAsync(Arg.Any(), Arg.Any()); + await _triggerScheduler.DidNotReceive().ScheduleAsync(Arg.Any>(), Arg.Any()); + + Assert.NotNull(workItem); + await workItem(serviceProvider, CancellationToken.None); + + await _triggerScheduler.Received(1).ScheduleAsync(Arg.Is>(x => x.SequenceEqual(_triggers)), Arg.Any()); + await _bookmarkScheduler.Received(1).ScheduleAsync(Arg.Is>(x => x.SequenceEqual(_bookmarks)), Arg.Any()); + } + + private ServiceProvider CreateServiceProvider(Action? configureServices = null) + { + var services = new ServiceCollection(); + services.AddSingleton(_triggerStore); + services.AddSingleton(_bookmarkStore); + services.AddSingleton(_triggerScheduler); + services.AddSingleton(_bookmarkScheduler); + configureServices?.Invoke(services); + return services.BuildServiceProvider(); + } +} diff --git a/test/unit/Elsa.Workflows.Runtime.UnitTests/Distributed/DistributedRuntimeLockProviderValidatorTests.cs b/test/unit/Elsa.Workflows.Runtime.UnitTests/Distributed/DistributedRuntimeLockProviderValidatorTests.cs index 027dd95e2..0a31960bd 100644 --- a/test/unit/Elsa.Workflows.Runtime.UnitTests/Distributed/DistributedRuntimeLockProviderValidatorTests.cs +++ b/test/unit/Elsa.Workflows.Runtime.UnitTests/Distributed/DistributedRuntimeLockProviderValidatorTests.cs @@ -3,6 +3,7 @@ using Elsa.Common.DistributedHosting.DistributedLocks; using Elsa.Workflows.Runtime.Distributed; using Medallion.Threading; using Medallion.Threading.FileSystem; +using Microsoft.Extensions.Logging; using Microsoft.Extensions.Logging.Abstractions; namespace Elsa.Workflows.Runtime.UnitTests.Distributed; @@ -19,14 +20,24 @@ public class DistributedRuntimeLockProviderValidatorTests : IDisposable } [Fact] - public void Validate_Throws_WhenFileSystemProviderIsUsedWithoutOptIn() + public void Validate_DoesNotThrow_WhenFileSystemProviderIsUsedWithoutOptIn() { var validator = CreateValidator(_fileProvider); - var exception = Assert.Throws(validator.Validate); + validator.Validate(); + } - Assert.Contains(nameof(FileDistributedSynchronizationProvider), exception.Message); - Assert.Contains(nameof(DistributedLockingOptions.AllowLocalLockProviderInDistributedRuntime), exception.Message); + [Fact] + public void Validate_LogsWarning_WhenFileSystemProviderIsUsedWithoutOptIn() + { + var logger = new CapturingLogger(); + var validator = CreateValidator(_fileProvider, logger: logger); + + validator.Validate(); + + Assert.Contains(logger.Entries, entry => + entry.Level == LogLevel.Warning && + entry.Message.Contains("local-only lock provider", StringComparison.OrdinalIgnoreCase)); } [Fact] @@ -38,48 +49,49 @@ public class DistributedRuntimeLockProviderValidatorTests : IDisposable } [Fact] - public void Validate_Throws_WhenWrappedProviderUsesFileSystemProvider() + public void Validate_DoesNotThrow_WhenWrappedProviderUsesFileSystemProvider() { var validator = CreateValidator(new WrappedDistributedLockProvider(_fileProvider)); - var exception = Assert.Throws(validator.Validate); - - Assert.Contains(nameof(FileDistributedSynchronizationProvider), exception.Message); - Assert.Contains(nameof(WrappedDistributedLockProvider), exception.Message); + validator.Validate(); } [Fact] - public void Validate_Throws_WhenProviderUsesFileSystemProviderThroughCustomProperty() + public void Validate_LogsWarning_WhenWrappedProviderUsesFileSystemProvider() + { + var logger = new CapturingLogger(); + var validator = CreateValidator(new WrappedDistributedLockProvider(_fileProvider), logger: logger); + + validator.Validate(); + + Assert.Contains(logger.Entries, entry => + entry.Level == LogLevel.Warning && + entry.Message.Contains(nameof(WrappedDistributedLockProvider), StringComparison.Ordinal)); + } + + [Fact] + public void Validate_DoesNotThrow_WhenProviderUsesFileSystemProviderThroughCustomProperty() { var validator = CreateValidator(new CustomWrappedDistributedLockProvider(_fileProvider)); - var exception = Assert.Throws(validator.Validate); - - Assert.Contains(nameof(FileDistributedSynchronizationProvider), exception.Message); - Assert.Contains(nameof(CustomWrappedDistributedLockProvider), exception.Message); + validator.Validate(); } [Fact] - public void Validate_Throws_WhenProviderUsesFileSystemProviderThroughProviderCollection() + public void Validate_DoesNotThrow_WhenProviderUsesFileSystemProviderThroughProviderCollection() { var validator = CreateValidator(new CompositeDistributedLockProvider([new CustomDistributedLockProvider(), _fileProvider])); - var exception = Assert.Throws(validator.Validate); - - Assert.Contains(nameof(FileDistributedSynchronizationProvider), exception.Message); - Assert.Contains(nameof(CompositeDistributedLockProvider), exception.Message); + validator.Validate(); } [Fact] - public void Validate_Throws_WhenProviderUsesFileSystemProviderThroughNonGenericProviderCollection() + public void Validate_DoesNotThrow_WhenProviderUsesFileSystemProviderThroughNonGenericProviderCollection() { System.Collections.IEnumerable providers = new IDistributedLockProvider[] { new CustomDistributedLockProvider(), _fileProvider }; var validator = CreateValidator(new NonGenericCompositeDistributedLockProvider(providers)); - var exception = Assert.Throws(validator.Validate); - - Assert.Contains(nameof(FileDistributedSynchronizationProvider), exception.Message); - Assert.Contains(nameof(NonGenericCompositeDistributedLockProvider), exception.Message); + validator.Validate(); } [Fact] @@ -99,14 +111,11 @@ public class DistributedRuntimeLockProviderValidatorTests : IDisposable } [Fact] - public void Validate_Throws_WhenNullableProviderCollectionContainsFileSystemProvider() + public void Validate_DoesNotThrow_WhenNullableProviderCollectionContainsFileSystemProvider() { var validator = CreateValidator(new NullableCompositeDistributedLockProvider([new CustomDistributedLockProvider(), null, _fileProvider])); - var exception = Assert.Throws(validator.Validate); - - Assert.Contains(nameof(FileDistributedSynchronizationProvider), exception.Message); - Assert.Contains(nameof(NullableCompositeDistributedLockProvider), exception.Message); + validator.Validate(); } [Fact] @@ -141,20 +150,15 @@ public class DistributedRuntimeLockProviderValidatorTests : IDisposable new ValueEqualDistributedLockProvider("same", _fileProvider) ])); - var exception = Assert.Throws(validator.Validate); - - Assert.Contains(nameof(FileDistributedSynchronizationProvider), exception.Message); - Assert.Contains(nameof(CompositeDistributedLockProvider), exception.Message); + validator.Validate(); } [Fact] - public void Validate_Throws_WhenNoopProviderIsUsedWithoutOptIn() + public void Validate_DoesNotThrow_WhenNoopProviderIsUsedWithoutOptIn() { var validator = CreateValidator(new NoopDistributedSynchronizationProvider()); - var exception = Assert.Throws(validator.Validate); - - Assert.Contains(nameof(NoopDistributedSynchronizationProvider), exception.Message); + validator.Validate(); } [Fact] @@ -173,12 +177,15 @@ public class DistributedRuntimeLockProviderValidatorTests : IDisposable validator.Validate(); } - private static DistributedRuntimeLockProviderValidator CreateValidator(IDistributedLockProvider provider, Action? configure = null) + private static DistributedRuntimeLockProviderValidator CreateValidator( + IDistributedLockProvider provider, + Action? configure = null, + ILogger? logger = null) { var options = new DistributedLockingOptions(); configure?.Invoke(options); - return new(provider, Microsoft.Extensions.Options.Options.Create(options), NullLogger.Instance); + return new(provider, Microsoft.Extensions.Options.Options.Create(options), logger ?? NullLogger.Instance); } public void Dispose() @@ -298,4 +305,32 @@ public class DistributedRuntimeLockProviderValidatorTests : IDisposable public ValueTask TryAcquireAsync(TimeSpan timeout = default, CancellationToken cancellationToken = default) => throw new NotSupportedException(); } + + private sealed class CapturingLogger : ILogger + { + public List Entries { get; } = []; + + public IDisposable BeginScope(TState state) where TState : notnull => NullScope.Instance; + public bool IsEnabled(LogLevel logLevel) => true; + + public void Log( + LogLevel logLevel, + EventId eventId, + TState state, + Exception? exception, + Func formatter) + { + Entries.Add(new(logLevel, formatter(state, exception))); + } + } + + private sealed record LogEntry(LogLevel Level, string Message); + + private sealed class NullScope : IDisposable + { + public static readonly NullScope Instance = new(); + public void Dispose() + { + } + } }