Archiver Observability

docs/adr/0021-move-archive-from-esb.md

@@ -0,0 +1,32 @@
+# 20. Move Archive Methods from External Service Bus to Outbox Archiver to Reduce Complexity
+
+Date: 2024-10-28
+
+## Status
+
+Adopted
+
+## Context
+
+The `ExternalServiceBus` class is a mediator between producer and outbox. It suffers from complexity, see ADR 0020.
+
+The `ExternalServiceBus` class has a number of methods that are related to archiving messages. These methods are:
+ - Archive
+ - ArchiveAsync
+
+The OutboxArchiver and the TimedOuboxArchiver are the only callers of these ExternalBus Archive methods. The TimedOutboxArchiver provides both a timer that fires the archiver at a given periodic interval, and uses the global distributed lock to ensure only one archiver runs.
+
+## Decision
+
+Whilst the `Archive` methods were not called out by CodeScene analysis, they do add to the overall set of responsibilities of `ExternalServiceBus`. As they have different reasons to change to `ExternalServiceBus` they should be moved to a separate class.
+
+We will move the implementation from `ExternalServiceBus` into `OutboxArchiver`. We will call `OutboxArchiver` from  `TimedOutboxArchiver`.
+
+## Consequences
+
+One, we have moved these functions, it makes sense to rename the `ExternalServiceBus` class to 'OutboxProducerMediator' as this better describes its role within our codebase.
+
+
+
+
+

src/Paramore.Brighter.Extensions.DependencyInjection/ServiceCollectionExtensions.cs

@@ -30,7 +30,6 @@ THE SOFTWARE. */
 using Paramore.Brighter.FeatureSwitch;
 using Paramore.Brighter.Logging;
 using System.Text.Json;
-using System.Transactions;
 using Paramore.Brighter.DynamoDb;
 using Paramore.Brighter.Observability;
 using Polly.Registry;
@@ -94,7 +93,7 @@ public static IBrighterBuilder BrighterHandlerBuilder(IServiceCollection service
             var mapperRegistry = new ServiceCollectionMessageMapperRegistry(services, options.MapperLifetime);
             services.TryAddSingleton(mapperRegistry);
 
-            services.TryAddSingleton<IAmARequestContextFactory>(options.RequestContextFactory);
+            services.TryAddSingleton(options.RequestContextFactory);
 
             if (options.FeatureSwitchRegistry != null)
                 services.TryAddSingleton(options.FeatureSwitchRegistry);
@@ -136,9 +135,6 @@ public static IBrighterBuilder BrighterHandlerBuilder(IServiceCollection service
         /// </summary>
         /// <param name="brighterBuilder">The Brighter builder to add this option to</param>
         /// <param name="configure">A callback that allows you to configure <see cref="ExternalBusConfiguration"/> options</param>
-        /// <param name="transactionProvider">The transaction provider for the outbox, can be null for in-memory default
-        /// of <see cref="CommittableTransactionProvider"/> which you must set the generic type to <see cref="CommittableTransaction"/> for
-        /// </param>
         /// <param name="serviceLifetime">The lifetime of the transaction provider</param>
         /// <returns>The Brighter builder to allow chaining of requests</returns>
         public static IBrighterBuilder UseExternalBus(
@@ -222,8 +218,8 @@ public static IBrighterBuilder UseExternalBus(
 
             brighterBuilder.Services.TryAddSingleton<IAmExternalBusConfiguration>(busConfiguration);
 
-            brighterBuilder.Services.TryAdd(new ServiceDescriptor(typeof(IAmAnExternalBusService),
-               (serviceProvider) => BuildExternalBus(
+            brighterBuilder.Services.TryAdd(new ServiceDescriptor(typeof(IAmAnOutboxProducerMediator),
+               (serviceProvider) => BuildOutBoxProducerMediator(
                    serviceProvider, transactionType, busConfiguration, brighterBuilder.PolicyRegistry, outbox
                    ),
                ServiceLifetime.Singleton));
@@ -237,7 +233,7 @@ private static INeedInstrumentation AddEventBus(
             INeedMessaging messagingBuilder,
             IUseRpc useRequestResponse)
         {
-            var eventBus = provider.GetService<IAmAnExternalBusService>();
+            var eventBus = provider.GetService<IAmAnOutboxProducerMediator>();
             var eventBusConfiguration = provider.GetService<IAmExternalBusConfiguration>();
             var serviceActivatorOptions = provider.GetService<IServiceActivatorOptions>();
 
@@ -325,17 +321,17 @@ private static object BuildCommandProcessor(IServiceProvider provider)
             return commandProcessor;
         }
 
-        private static IAmAnExternalBusService BuildExternalBus(IServiceProvider serviceProvider,
+        private static IAmAnOutboxProducerMediator BuildOutBoxProducerMediator(IServiceProvider serviceProvider,
             Type transactionType,
             ExternalBusConfiguration busConfiguration,
             IPolicyRegistry<string> policyRegistry,
             IAmAnOutbox outbox) 
         {
             //Because the bus has specialized types as members, we need to create the bus type dynamically
             //again to prevent someone configuring Brighter from having to pass generic types
-            var busType = typeof(ExternalBusService<,>).MakeGenericType(typeof(Message), transactionType);
+            var busType = typeof(OutboxProducerMediator<,>).MakeGenericType(typeof(Message), transactionType);
 
-            return (IAmAnExternalBusService)Activator.CreateInstance(
+            return (IAmAnOutboxProducerMediator)Activator.CreateInstance(
                 busType,
                 busConfiguration.ProducerRegistry,
                 policyRegistry,
@@ -344,13 +340,11 @@ private static IAmAnExternalBusService BuildExternalBus(IServiceProvider service
                 TransformFactoryAsync(serviceProvider),
                 Tracer(serviceProvider),
                 outbox,
-                busConfiguration.ArchiveProvider,
                 RequestContextFactory(serviceProvider),
                 busConfiguration.OutboxTimeout,
                 busConfiguration.MaxOutStandingMessages,
                 busConfiguration.MaxOutStandingCheckInterval,
                 busConfiguration.OutBoxBag,
-                busConfiguration.ArchiveBatchSize,
                 TimeProvider.System,
                 busConfiguration.InstrumentationOptions);
         }

src/Paramore.Brighter.Extensions.Hosting/HostedServiceCollectionExtensions.cs

@@ -57,7 +57,7 @@ public static IBrighterBuilder UseOutboxArchiver<TTransaction>(this IBrighterBui
             brighterBuilder.Services.TryAddSingleton<TimedOutboxArchiverOptions>(options);
             brighterBuilder.Services.AddSingleton<IAmAnArchiveProvider>(archiveProvider);
 
-            brighterBuilder.Services.AddHostedService<TimedOutboxArchiver>();
+            brighterBuilder.Services.AddHostedService<TimedOutboxArchiver<Message, TTransaction>>();
 
             return brighterBuilder;
         }

src/Paramore.Brighter.Extensions.Hosting/TimedOutboxArchiver.cs

@@ -25,29 +25,43 @@ THE SOFTWARE. */
 using System;
 using System.Threading;
 using System.Threading.Tasks;
-using Microsoft.Extensions.DependencyInjection;
 using Microsoft.Extensions.Hosting;
 using Microsoft.Extensions.Logging;
 using Paramore.Brighter.Logging;
+// ReSharper disable StaticMemberInGenericType
 
 namespace Paramore.Brighter.Extensions.Hosting
 {
-
     /// <summary>
     /// The archiver will find messages in the outbox that are older than a certain age and archive them
     /// </summary>
-    /// <param name="serviceScopeFactory">Needed to create a scope within which to create a <see cref="CommandProcessor"/></param>
-    /// <param name="distributedLock">Used to ensure that only one instance of the <see cref="TimedOutboxSweeper"/> is running</param>
-    /// <param name="options">The <see cref="TimedOutboxArchiverOptions"/> that control how the archiver runs, such as interval</param>
-    public class TimedOutboxArchiver(
-        IServiceScopeFactory serviceScopeFactory,
-        IDistributedLock distributedLock,
-        TimedOutboxArchiverOptions options
-        )
-        : IHostedService, IDisposable
+    public class TimedOutboxArchiver<TMessage, TTransaction> : IHostedService, IDisposable where TMessage : Message
     {
         private static readonly ILogger s_logger = ApplicationLogging.CreateLogger<TimedOutboxSweeper>();
         private Timer _timer;
+        private readonly OutboxArchiver<TMessage, TTransaction> _archiver;
+        private readonly TimeSpan _dispatchedSince;
+        private readonly IDistributedLock _distributedLock;
+        private readonly TimedOutboxArchiverOptions _options;
+
+        /// <summary>
+        /// The archiver will find messages in the outbox that are older than a certain age and archive them
+        /// </summary>
+        /// <param name="archiver">The archiver to use</param>
+        /// <param name="dispatchedSince">How old should a message be, in order to archive it?</param>
+        /// <param name="distributedLock">Used to ensure that only one instance of the <see cref="TimedOutboxSweeper"/> is running</param>
+        /// <param name="options">The <see cref="TimedOutboxArchiverOptions"/> that control how the archiver runs, such as interval</param>
+        public TimedOutboxArchiver(
+            OutboxArchiver<TMessage, TTransaction> archiver,
+            TimeSpan dispatchedSince,
+            IDistributedLock distributedLock,
+            TimedOutboxArchiverOptions options)
+        {
+            _archiver = archiver;
+            _dispatchedSince = dispatchedSince;
+            _distributedLock = distributedLock;
+            _options = options;
+        }
 
         private const string LockingResourceName = "Archiver";
 
@@ -60,8 +74,8 @@ public Task StartAsync(CancellationToken cancellationToken)
         {
             s_logger.LogInformation("Outbox Archiver Service is starting");
 
-            _timer = new Timer(async (e) => await Archive(e, cancellationToken), null, TimeSpan.Zero,
-                TimeSpan.FromSeconds(options.TimerInterval));
+            _timer = new Timer(async e => await Archive(e, cancellationToken), null, TimeSpan.Zero,
+                TimeSpan.FromSeconds(_options.TimerInterval));
 
             return Task.CompletedTask;
         }
@@ -90,24 +104,21 @@ public void Dispose()
 
         private async Task Archive(object state, CancellationToken cancellationToken)
         {
-            var lockId = await distributedLock.ObtainLockAsync(LockingResourceName, cancellationToken); 
+            var lockId = await _distributedLock.ObtainLockAsync(LockingResourceName, cancellationToken); 
             if (lockId != null)
             {
-                var scope = serviceScopeFactory.CreateScope();
                 s_logger.LogInformation("Outbox Archiver looking for messages to Archive");
                 try
                 {
-                    IAmAnExternalBusService externalBusService = scope.ServiceProvider.GetService<IAmAnExternalBusService>();
-
-                    await externalBusService.ArchiveAsync(options.MinimumAge,  new RequestContext(), cancellationToken);
+                    await _archiver.ArchiveAsync(_dispatchedSince, new RequestContext(), cancellationToken);
                 }
                 catch (Exception e)
                 {
                     s_logger.LogError(e, "Error while sweeping the outbox");
                 }
                 finally
                 {
-                    await distributedLock.ReleaseLockAsync(LockingResourceName, lockId, cancellationToken);
+                    await _distributedLock.ReleaseLockAsync(LockingResourceName, lockId, cancellationToken);
                 }
 
                 s_logger.LogInformation("Outbox Sweeper sleeping");

src/Paramore.Brighter.ServiceActivator/ControlBus/ControlBusReceiverBuilder.cs

@@ -158,7 +158,7 @@ public Dispatcher Build(string hostName)
 
             var outbox = new SinkOutboxSync();
 
-            var externalBus = new ExternalBusService<Message, CommittableTransaction>(
+            var mediator = new OutboxProducerMediator<Message, CommittableTransaction>(
                 producerRegistry: producerRegistry,
                 policyRegistry: new DefaultPolicy(),
                 mapperRegistry: outgoingMessageMapperRegistry,
@@ -175,7 +175,7 @@ public Dispatcher Build(string hostName)
             commandProcessor = CommandProcessorBuilder.StartNew()
                 .Handlers(new HandlerConfiguration(subscriberRegistry, new ControlBusHandlerFactorySync(_dispatcher, () => commandProcessor)))
                 .Policies(policyRegistry)
-                .ExternalBus(ExternalBusType.FireAndForget, externalBus)
+                .ExternalBus(ExternalBusType.FireAndForget, mediator)
                 .ConfigureInstrumentation(null, InstrumentationOptions.None)
                 .RequestContextFactory(new InMemoryRequestContextFactory())
                 .Build();