[release/v26.0] [DEV-1850] Fix connector accounting when connectors stop on their own

src/Connectors/KurrentDB.Connectors.Tests/Infrastructure/SystemConnectorsFactoryTests.cs

@@ -0,0 +1,62 @@
+// Copyright (c) Kurrent, Inc and/or licensed to Kurrent, Inc under one or more agreements.
+// Kurrent, Inc licenses this file to you under the Kurrent License v1 (see LICENSE.md).
+
+using System.Diagnostics.Metrics;
+using Kurrent.Surge.Connectors;
+using KurrentDB.Connectors.Infrastructure.Connect.Components.Connectors;
+using Microsoft.Extensions.Configuration;
+using Microsoft.Extensions.DependencyInjection;
+
+namespace KurrentDB.Connectors.Tests.Infrastructure;
+
+public class SystemConnectorsFactoryTests(ITestOutputHelper output, ConnectorsAssemblyFixture fixture) : ConnectorsIntegrationTests(output, fixture) {
+    [Fact]
+    public Task counts_only_disposed_connector_as_closed() => Fixture.TestWithTimeout(async cts => {
+        // Arrange
+        var factory = Fixture.NodeServices.GetRequiredService<ISystemConnectorFactory>();
+
+        var firstId  = ConnectorId.From(Fixture.NewConnectorId());
+        var secondId = ConnectorId.From(Fixture.NewConnectorId());
+
+        var closed = new List<string>();
+
+        using var listener = new MeterListener {
+            InstrumentPublished = (instrument, meterListener) => {
+                if (instrument.Meter.Name == "Kurrent.Connectors" && instrument.Name == "kurrent_connector_active_total")
+                    meterListener.EnableMeasurementEvents(instrument);
+            }
+        };
+
+        listener.SetMeasurementEventCallback<int>((_, measurement, tags, _) => {
+            if (measurement >= 0)
+                return;
+
+            var connectorId = tags.ToArray().FirstOrDefault(tag => tag.Key == "connector_id").Value?.ToString();
+            if (connectorId is not null)
+                closed.Add(connectorId);
+        });
+
+        listener.Start();
+
+        var first  = factory.CreateConnector(firstId, SerilogSinkSettings());
+        var second = factory.CreateConnector(secondId, SerilogSinkSettings());
+
+        await first.Connect(cts.Token);
+        await second.Connect(cts.Token);
+
+        // Act
+        await first.DisposeAsync();
+
+        // Assert
+        closed.Should().Equal(firstId.ToString());
+
+        await second.DisposeAsync();
+
+        closed.Should().Equal(firstId.ToString(), secondId.ToString());
+    });
+
+    static IConfiguration SerilogSinkSettings() =>
+        new ConfigurationBuilder()
+            .AddInMemoryCollection([new("InstanceTypeName", "serilog-sink")])
+            .Build();
+}

src/Connectors/KurrentDB.Connectors.Tests/Planes/Control/ConnectorsActivatorTests.cs

@@ -8,115 +8,120 @@ namespace KurrentDB.Connectors.Tests.Planes.Control;
 
 [Trait("Category", "ControlPlane")]
 public class ConnectorsActivatorTests {
-	[Fact]
-	public async Task connector_activates() {
-		// Arrange
-		var connectorId = ConnectorId.From(Guid.NewGuid());
-		var settings = new Dictionary<string, string?>();
-		var revision = 1;
+	const int Revision = 1;
+
+	static (ConnectorsActivator Sut, ConnectorId ConnectorId) CreateSut(TestConnector connector) =>
+		(new ConnectorsActivator((_, _) => connector), ConnectorId.From(Guid.NewGuid()));
+
+	static ValueTask<ActivateResult> Activate(ConnectorsActivator sut, ConnectorId connectorId) =>
+		sut.Activate(connectorId, NoSettings, Revision);
 
-		var testConnector = new TestConnector(failOnConnect: false);
+	static readonly Dictionary<string, string?> NoSettings = [];
 
-		var sut = new ConnectorsActivator(CreateConnector);
+	[Fact]
+	public async Task connector_activates() {
+		var connector = new TestConnector();
+		var (sut, connectorId) = CreateSut(connector);
 
-		// Act
-		var result = await sut.Activate(connectorId, settings, revision);
+		var result = await Activate(sut, connectorId);
 
-		// Assert
 		result.Success.Should().BeTrue();
 		result.Type.Should().Be(ActivateResultType.Activated);
-		testConnector.IsDisposed.Should().BeFalse();
-		testConnector.ConnectionAttempt.Should().Be(1);
-		return;
-
-		IConnector CreateConnector(ConnectorId connectorId1, IDictionary<string, string?> dictionary) => testConnector;
+		connector.DisposeCount.Should().Be(0);
+		connector.ConnectionAttempt.Should().Be(1);
 	}
 
 	[Fact]
 	public async Task connector_disposed_when_connect_throws_exception() {
-		// Arrange
-		var connectorId = ConnectorId.From(Guid.NewGuid());
-		var settings = new Dictionary<string, string?>();
-		var revision = 1;
 		var exception = new InvalidOperationException("Connection failed");
+		var connector = new TestConnector(failOnConnect: true, exception);
+		var (sut, connectorId) = CreateSut(connector);
 
-		var testConnector = new TestConnector(failOnConnect: true, exception);
-
-		var sut = new ConnectorsActivator(CreateConnector);
+		var result = await Activate(sut, connectorId);
 
-		// Act
-		var result = await sut.Activate(connectorId, settings, revision);
-
-		// Assert
 		result.Failure.Should().BeTrue();
 		result.Type.Should().Be(ActivateResultType.Unknown);
 		result.Error.Should().Be(exception);
-		testConnector.IsDisposed.Should().BeTrue();
-		testConnector.ConnectionAttempt.Should().Be(1);
-		return;
-
-		IConnector CreateConnector(ConnectorId connectorId1, IDictionary<string, string?> dictionary) => testConnector;
+		connector.DisposeCount.Should().Be(1);
+		connector.ConnectionAttempt.Should().Be(1);
+		connector.Stopped.Status.Should().Be(TaskStatus.RanToCompletion);
 	}
 
 	[Fact]
 	public async Task connector_disposed_when_connect_throws_validation_exception() {
-		// Arrange
-		var connectorId = ConnectorId.From(Guid.NewGuid());
-		var settings = new Dictionary<string, string?>();
-		var revision = 1;
 		var validationException = new FluentValidation.ValidationException("Invalid configuration");
+		var connector = new TestConnector(failOnConnect: true, validationException);
+		var (sut, connectorId) = CreateSut(connector);
 
-		var testConnector = new TestConnector(failOnConnect: true, validationException);
-
-		var sut = new ConnectorsActivator(CreateConnector);
-
-		// Act
-		var result = await sut.Activate(connectorId, settings, revision);
+		var result = await Activate(sut, connectorId);
 
-		// Assert
 		result.Failure.Should().BeTrue();
 		result.Type.Should().Be(ActivateResultType.InvalidConfiguration);
 		result.Error.Should().Be(validationException);
-		testConnector.IsDisposed.Should().BeTrue();
-		testConnector.ConnectionAttempt.Should().Be(1);
-		return;
+		connector.DisposeCount.Should().Be(1);
+		connector.ConnectionAttempt.Should().Be(1);
+	}
+
+	[Fact]
+	public async Task deactivates_once() {
+		var connector = new TestConnector();
+		var (sut, connectorId) = CreateSut(connector);
 
-		IConnector CreateConnector(ConnectorId connectorId1, IDictionary<string, string?> dictionary) => testConnector;
+		await Activate(sut, connectorId);
+
+		var result = await sut.Deactivate(connectorId);
+
+		result.Type.Should().Be(DeactivateResultType.Deactivated);
+		connector.DisposeCount.Should().Be(1);
+
+		var repeated = await sut.Deactivate(connectorId);
+
+		repeated.Type.Should().Be(DeactivateResultType.ConnectorNotFound);
+		connector.DisposeCount.Should().Be(1);
 	}
 
 	[Fact]
-	public async Task connector_stopped_task_completes_on_connect_failure() {
-		// Arrange
-		var connectorId = ConnectorId.From(Guid.NewGuid());
-		var settings = new Dictionary<string, string?>();
-		var revision = 1;
-		var exception = new InvalidOperationException("Connection failed");
+	public async Task deactivates_self_stopped_connector() {
+		var connector = new TestConnector();
+		var (sut, connectorId) = CreateSut(connector);
 
-		var testConnector = new TestConnector(failOnConnect: true, exception);
+		await Activate(sut, connectorId);
 
-		var sut = new ConnectorsActivator(CreateConnector);
+		// a sink failing against an unreachable broker
+		connector.SimulateSelfTermination(new InvalidOperationException("simulated connector crash"));
 
-		// Act
-		var result = await sut.Activate(connectorId, settings, revision);
+		var result = await sut.Deactivate(connectorId);
 
-		// Assert
-		result.Failure.Should().BeTrue();
-		testConnector.Stopped.IsCompleted.Should().BeTrue();
-		testConnector.Stopped.Status.Should().Be(TaskStatus.RanToCompletion);
-		return;
+		result.Type.Should().Be(DeactivateResultType.Deactivated);
+		connector.DisposeCount.Should().Be(1);
+	}
 
-		IConnector CreateConnector(ConnectorId connectorId1, IDictionary<string, string?> dictionary) => testConnector;
+	[Theory]
+	[InlineData(false)]
+	[InlineData(true)]
+	public async Task waits_for_deactivation(bool faulted) {
+		var connector = new TestConnector();
+		var (sut, connectorId) = CreateSut(connector);
+
+		await Activate(sut, connectorId);
+
+		var waiting = sut.WaitForDeactivation(connectorId);
+		connector.SimulateSelfTermination(faulted ? new InvalidOperationException("simulated connector crash") : null);
+		var result = await waiting;
+
+		result.Type.Should().Be(DeactivateResultType.Deactivated);
+		connector.DisposeCount.Should().Be(1);
 	}
 }
 
 internal class TestConnector(bool failOnConnect = false, Exception? exception = null) : IConnector {
-	readonly TaskCompletionSource _stoppedTcs = new();
+	readonly TaskCompletionSource _stoppedTcs = new(TaskCreationOptions.RunContinuationsAsynchronously);
 
 	public ConnectorId ConnectorId => ConnectorId.From(Guid.NewGuid());
 	public ConnectorState State { get; private set; } = ConnectorState.Unspecified;
 	public Task Stopped => _stoppedTcs.Task;
 
-	public bool IsDisposed { get; private set; }
+	public int DisposeCount      { get; private set; }
 	public int ConnectionAttempt { get; private set; }
 
 	public Task Connect(CancellationToken stoppingToken) {
@@ -131,8 +136,17 @@ public Task Connect(CancellationToken stoppingToken) {
 		return Task.CompletedTask;
 	}
 
+	public void SimulateSelfTermination(Exception? error = null) {
+		State = ConnectorState.Stopped;
+
+		if (error is null)
+			_stoppedTcs.TrySetResult();
+		else
+			_stoppedTcs.TrySetException(error);
+	}
+
 	public ValueTask DisposeAsync() {
-		IsDisposed = true;
+		DisposeCount++;
 		State = ConnectorState.Stopped;
 
 		_stoppedTcs.TrySetResult();

src/Connectors/KurrentDB.Connectors/Infrastructure/Connect/Components/Connectors/SystemConnectorsFactory.cs

@@ -45,8 +45,6 @@ public class SystemConnectorsFactory(SystemConnectorsFactoryOptions options, ISe
     SystemConnectorsFactoryOptions Options  { get; } = options;
     IServiceProvider               Services { get; } = services;
 
-    static DisposeCallback? OnDisposeCallback;
-
     public IConnector CreateConnector(ConnectorId connectorId, IConfiguration configuration) {
         var options       = configuration.GetRequiredOptions<ConnectorOptions>();
         var validator     = Services.GetRequiredService<IConnectorValidator>();
@@ -86,29 +84,37 @@ SinkConnector CreateSinkConnector() {
                 connector = new SqlReducerSink(connector, reducer);
 	        }
 
-	        ConnectorMetrics.TrackSinkConnectorCreated(connector.GetType(), connectorId);
+	        Type instanceType = connector.GetType();
 
-	        OnDisposeCallback = () => ConnectorMetrics.TrackSinkConnectorClosed(connector.GetType(), connectorId);
+	        ConnectorMetrics.TrackSinkConnectorCreated(instanceType, connectorId);
 
 	        var sinkProxy = new SinkProxy(connectorId, connector, config, Services);
 
 	        var processor = ConfigureSinkProcessor(connectorId, Options.Interceptors, sinkOptions, sinkProxy);
 
-	        return new SinkConnector(processor, sinkProxy);
+	        return new SinkConnector(
+		        processor,
+		        sinkProxy,
+		        () => ConnectorMetrics.TrackSinkConnectorClosed(instanceType, connectorId)
+	        );
         }
 
         SourceConnector CreateSourceConnector() {
 	        var sourceOptions = configuration.GetRequiredOptions<SourceOptions>();
 
-	        ConnectorMetrics.TrackSourceConnectorCreated(connector.GetType(), connectorId);
+	        Type instanceType = connector.GetType();
 
-	        OnDisposeCallback = () => ConnectorMetrics.TrackSourceConnectorClosed(connector.GetType(), connectorId);
+	        ConnectorMetrics.TrackSourceConnectorCreated(instanceType, connectorId);
 
 	        var sourceProxy = new SourceProxy(connectorId, connector, configuration, Services);
 
 	        var processor = ConfigureSourceProcessor(connectorId, Options.Interceptors, sourceOptions, sourceProxy);
 
-	        return new SourceConnector(connectorId, processor);
+	        return new SourceConnector(
+		        connectorId,
+		        processor,
+		        () => ConnectorMetrics.TrackSourceConnectorClosed(instanceType, connectorId)
+	        );
         }
 
         dynamic CreateConnectorInstance(string connectorTypeName) {
@@ -215,7 +221,7 @@ IProcessor ConfigureSourceProcessor(ConnectorId connectorId, LinkedList<Intercep
         return new SourceProcessor(connectorId, interceptors, producer, sourceProxy, loggingOptions);
     }
 
-    sealed class SinkConnector(IProcessor processor, SinkProxy sinkProxy) : IConnector {
+    sealed class SinkConnector(IProcessor processor, SinkProxy sinkProxy, DisposeCallback disposeCallback) : IConnector {
         public ConnectorId    ConnectorId { get; } = ConnectorId.From(processor.ProcessorId);
         public ConnectorState State       { get; } = (ConnectorState)processor.State;
 
@@ -227,28 +233,47 @@ public async Task Connect(CancellationToken stoppingToken) {
         }
 
         public async ValueTask DisposeAsync() {
-            await sinkProxy.DisposeAsync();
-            await processor.DisposeAsync();
-            OnDisposeCallback?.Invoke();
+            // the processor is disposed even when the sink fails to close, and the
+            // connector stops counting as active either way, otherwise a sink that
+            // cannot be closed leaves work running and is counted forever
+            try {
+                await sinkProxy.DisposeAsync();
+            }
+            finally {
+                try {
+                    await processor.DisposeAsync();
+                }
+                finally {
+                    disposeCallback.Invoke();
+                }
+            }
         }
     }
 
-    sealed class SourceConnector(ConnectorId connectorId, IProcessor SourceProcessor) : BackgroundService, IConnector {
+    sealed class SourceConnector(ConnectorId connectorId, IProcessor sourceProcessor, DisposeCallback disposeCallback) : BackgroundService, IConnector {
         public ConnectorId    ConnectorId { get; } = ConnectorId.From(connectorId);
-        public ConnectorState State       => (ConnectorState)SourceProcessor.State;
+        public ConnectorState State       => (ConnectorState)sourceProcessor.State;
 
-        public Task Stopped => SourceProcessor.Stopped;
+        public Task Stopped => sourceProcessor.Stopped;
 
         protected override async Task ExecuteAsync(CancellationToken stoppingToken) =>
-            await SourceProcessor.Activate(stoppingToken).ConfigureAwait(false);
+            await sourceProcessor.Activate(stoppingToken).ConfigureAwait(false);
 
         public async Task Connect(CancellationToken stoppingToken) =>
             await StartAsync(stoppingToken).ConfigureAwait(false);
 
         public async ValueTask DisposeAsync() {
-            await StopAsync(CancellationToken.None).ConfigureAwait(false);
-            await SourceProcessor.DisposeAsync().ConfigureAwait(false);
-            OnDisposeCallback?.Invoke();
+            try {
+                await StopAsync(CancellationToken.None).ConfigureAwait(false);
+            }
+            finally {
+                try {
+                    await sourceProcessor.DisposeAsync().ConfigureAwait(false);
+                }
+                finally {
+                    disposeCallback.Invoke();
+                }
+            }
         }
     }
 }