Added abstraction for Secondary Indexes building (#5035)

Added needed abstractions for handling secondary indexes and plugged them to plugin. This is the first step before plugging the real DuckDB implementation. 

1. `SecondaryIndexSubscription` - responsible for listening to log and indexing events that were stored there.
2. Indexing and committing happens in the implementation of `ISecondaryIndexProcessor`.
3. `SecondaryIndexCheckpointTracker` is responsible for triggering index commits. Commits happen when the maximum size of pending batches is reached or when the maximum time has elapsed since the previous commit.
4. Commit settings can be configured through options: CheckpointCommitBatchSize and CheckpointCommitDelayMs exposed through database settings.
5. All of that is handled by the `SecondaryIndexBuilder` hosted service, responsible for running the subscription.

You can see how it will work for DuckDB in simplified fake in-memory implementations.

Besides that plugged integration tests based on the Connectors plugins. Expanded them to make it possible to run them in parallel.

**_Note:_ I also made `_clusterNodeMutex` NOT be acquired for the in-memory database in `ClusterVNodeHostedService`  (so just like the db block). I also released it on dispose it accordingly.** This enables having multiple instances of `ClusterVNodeHostedService` running in integration tests. Previous tests in connectors assumed that only one `ClusterVNodeHostedService` would be running in the assembly fixture, but that doesn't allow for testing different configuration settings (e.g. if the plugin was enabled or disabled).

Author: oskardudycz

src/Connectors/KurrentDB.Connectors.Tests/ClusterVNodeApp.cs

@@ -14,6 +14,7 @@
 using KurrentDB.Core.Messages;
 // using KurrentDB.Surge.Testing.Extensions;
 using Microsoft.AspNetCore.Builder;
+using Microsoft.AspNetCore.Hosting;
 using Microsoft.Extensions.Configuration;
 using Microsoft.Extensions.DependencyInjection;
 using Microsoft.Extensions.Hosting;
@@ -38,7 +39,7 @@ public class ClusterVNodeApp : IAsyncDisposable {
 
     public async Task<(ClusterVNodeOptions Options, IServiceProvider Services)> Start(TimeSpan? readinessTimeout = null, Dictionary<string, string?>? overrides = null, Action<IServiceCollection>? configureServices = null) {
         var settings = overrides is not null
-            ? DefaultSettings.With(x => overrides.ForEach((key, value) => x[key] = value))
+            ? DefaultSettings.ToDictionary().With(x => overrides.ForEach((key, value) => x[key] = value))
             : DefaultSettings;
 
         var options = GetClusterVNodeOptions(settings);
@@ -54,6 +55,8 @@ public class ClusterVNodeApp : IAsyncDisposable {
             .With(x => x.Services.AddSingleton<IHostedService>(esdb))
             .With(x => configureServices?.Invoke(x.Services));
 
+        builder.WebHost.ConfigureKestrel(serverOptions => serverOptions.ListenAnyIP(0));
+
         App = builder.Build().With(x => esdb.Node.Startup.Configure(x));
 
         await App.StartAsync();

src/Connectors/KurrentDB.Connectors.Tests/ClusterVNodeFixture.cs

@@ -37,9 +37,10 @@ protected ClusterVNodeFixture() {
     public ILoggerFactory LoggerFactory { get; }
     public Faker          Faker         { get; }
 
-    public Action<IServiceCollection> ConfigureServices { get; init; } = _ => { };
-    public Func<Task>                 OnSetup           { get; init; } = () => Task.CompletedTask;
-    public Func<Task>                 OnTearDown        { get; init; } = () => Task.CompletedTask;
+    public Action<IServiceCollection>   ConfigureServices { get; init; } = _ => { };
+    public Dictionary<string, string?>? Configuration     { get; init; }
+    public Func<Task>                   OnSetup           { get; init; } = () => Task.CompletedTask;
+    public Func<Task>                   OnTearDown        { get; init; } = () => Task.CompletedTask;
 
     public ClusterVNodeOptions NodeOptions  { get; private set; } = null!;
     public IServiceProvider    NodeServices { get; private set; } = null!;
@@ -48,7 +49,10 @@ protected ClusterVNodeFixture() {
     public ISubscriber Subscriber => NodeServices.GetRequiredService<ISubscriber>();
 
     public async Task InitializeAsync() {
-        var (options, services) = await ClusterVNodeApp.Start(configureServices: ConfigureServices);
+        var (options, services) = await ClusterVNodeApp.Start(
+	        configureServices: ConfigureServices,
+	        overrides: Configuration
+	    );
 
         NodeServices = services;
         NodeOptions  = options;
@@ -59,6 +63,7 @@ public async Task InitializeAsync() {
     public async Task DisposeAsync() {
         try {
             await OnTearDown();
+            await ClusterVNodeApp.DisposeAsync();
         }
         catch {
             // ignored

src/KurrentDB.Core.Tests/Helpers/MiniClusterNode.cs

@@ -72,8 +72,7 @@ public MiniClusterNode(string pathname, int debugIndex, IPEndPoint internalTcp,
 	IPEndPoint httpEndPoint, EndPoint[] gossipSeeds, ISubsystem[] subsystems = null,
 	bool enableTrustedAuth = false, int memTableSize = 1000, bool inMemDb = true,
 	bool disableFlushToDisk = false, bool readOnlyReplica = false, int nodePriority = 0,
-	string intHostAdvertiseAs = null, IExpiryStrategy expiryStrategy = null,
-	IEnumerable<IVirtualStreamReader> virtualStreamReaders = null) {
+	string intHostAdvertiseAs = null, IExpiryStrategy expiryStrategy = null) {
 
 		if (RuntimeInformation.IsOSX) {
 			AppContext.SetSwitch("System.Net.Http.SocketsHttpHandler.Http2UnencryptedSupport",
@@ -194,7 +193,7 @@ public MiniClusterNode(string pathname, int debugIndex, IPEndPoint internalTcp,
 						options.Application.AllowAnonymousEndpointAccess,
 						options.Application.AllowAnonymousStreamAccess,
 						options.Application.OverrideAnonymousEndpointAccessForGossip).Create(components.MainQueue)]))),
-			virtualStreamReaders ?? [],
+			virtualStreamReader: null,
 			Array.Empty<IPersistentSubscriptionConsumerStrategyFactory>(),
 			new OptionsCertificateProvider(),
 			configuration: inMemConf,

src/KurrentDB.Core.Tests/Helpers/MiniNode.cs

@@ -216,10 +216,6 @@ public MiniNode(string pathname,
 				HighHasher = hash32bit ? new ConstantHasher(0) : options.HighHasher,
 			});
 
-		var virtualStreamReaders = subsystems
-			.OfType<ISecondaryIndexingPlugin>()
-			.SelectMany(indexingPlugin => indexingPlugin.IndicesVirtualStreamReaders);
-
 		Node = new ClusterVNode<TStreamId>(options, logFormatFactory,
 			new AuthenticationProviderFactory(
 				c => authenticationProviderFactory ?? new InternalAuthenticationProviderFactory(
@@ -231,7 +227,6 @@ public MiniNode(string pathname,
 						options.Application.AllowAnonymousEndpointAccess,
 						options.Application.AllowAnonymousStreamAccess,
 						options.Application.OverrideAnonymousEndpointAccessForGossip).Create(c.MainQueue)]))),
-			virtualStreamReaders: virtualStreamReaders,
 			expiryStrategy: expiryStrategy,
 			certificateProvider: new OptionsCertificateProvider(),
 			configuration: inMemConf,

src/KurrentDB.Core/ClusterVNode.cs

@@ -103,7 +103,7 @@ public static ClusterVNode<TStreamId> Create<TStreamId>(
 		ILogFormatAbstractorFactory<TStreamId> logFormatAbstractorFactory,
 		AuthenticationProviderFactory authenticationProviderFactory = null,
 		AuthorizationProviderFactory authorizationProviderFactory = null,
-		IEnumerable<IVirtualStreamReader> virtualStreamReaders = null,
+		VirtualStreamReader virtualStreamReader = null,
 		IReadOnlyList<IPersistentSubscriptionConsumerStrategyFactory> factories = null,
 		CertificateProvider certificateProvider = null,
 		IConfiguration configuration = null,
@@ -115,7 +115,7 @@ public static ClusterVNode<TStreamId> Create<TStreamId>(
 			logFormatAbstractorFactory,
 			authenticationProviderFactory,
 			authorizationProviderFactory,
-			virtualStreamReaders,
+			virtualStreamReader,
 			factories,
 			certificateProvider,
 			configuration,
@@ -235,7 +235,7 @@ public ClusterVNode(ClusterVNodeOptions options,
 		ILogFormatAbstractorFactory<TStreamId> logFormatAbstractorFactory,
 		AuthenticationProviderFactory authenticationProviderFactory = null,
 		AuthorizationProviderFactory authorizationProviderFactory = null,
-		IEnumerable<IVirtualStreamReader> virtualStreamReaders = null,
+		VirtualStreamReader virtualStreamReader = null,
 		IReadOnlyList<IPersistentSubscriptionConsumerStrategyFactory>
 			additionalPersistentSubscriptionConsumerStrategyFactories = null,
 		CertificateProvider certificateProvider = null,
@@ -782,16 +782,16 @@ void StartSubsystems() {
 		var nodeStatusListener = new NodeStateListenerService(_mainQueue, memLog);
 		_mainBus.Subscribe<SystemMessage.StateChangeMessage>(nodeStatusListener);
 
-		var inMemReader = new VirtualStreamReader([
+		virtualStreamReader ??= new VirtualStreamReader();
+		virtualStreamReader.Register(
 			gossipListener.Stream,
-			nodeStatusListener.Stream,
-			..virtualStreamReaders ?? []
-		]);
+			nodeStatusListener.Stream
+		);
 
 		// Storage Reader
 		var storageReader = new StorageReaderService<TStreamId>(_mainQueue, _mainBus, readIndex,
 			logFormat.SystemStreams,
-			readerThreadsCount, Db.Config.WriterCheckpoint.AsReadOnly(), inMemReader, _queueStatsManager,
+			readerThreadsCount, Db.Config.WriterCheckpoint.AsReadOnly(), virtualStreamReader, _queueStatsManager,
 			trackers.QueueTrackers);
 
 		_mainBus.Subscribe<SystemMessage.SystemInit>(storageReader);
@@ -1157,7 +1157,7 @@ GossipAdvertiseInfo GetGossipAdvertiseInfo() {
 		_mainBus.Subscribe<StorageMessage.EventCommitted>(subscrQueue);
 		_mainBus.Subscribe<StorageMessage.InMemoryEventCommitted>(subscrQueue);
 
-		var subscription = new SubscriptionsService<TStreamId>(_mainQueue, subscrQueue, _authorizationProvider, readIndex, inMemReader);
+		var subscription = new SubscriptionsService<TStreamId>(_mainQueue, subscrQueue, _authorizationProvider, readIndex, virtualStreamReader);
 		subscrBus.Subscribe<SystemMessage.SystemStart>(subscription);
 		subscrBus.Subscribe<SystemMessage.BecomeShuttingDown>(subscription);
 		subscrBus.Subscribe<TcpMessage.ConnectionClosed>(subscription);

src/KurrentDB.Core/Configuration/ClusterVNodeOptionsExtensions.cs

@@ -7,6 +7,7 @@
 using System.Net;
 using System.Security.Cryptography.X509Certificates;
 using EventStore.Plugins;
+using EventStore.Plugins.Subsystems;
 using KurrentDB.Common.Exceptions;
 using KurrentDB.Common.Utils;
 using KurrentDB.Core.Certificates;
@@ -20,6 +21,9 @@ public static ClusterVNodeOptions Reload(this ClusterVNodeOptions options) =>
 			? options
 			: ClusterVNodeOptions.FromConfiguration(options.ConfigurationRoot);
 
+	public static ClusterVNodeOptions WithPlugableComponents(this ClusterVNodeOptions options, ISubsystemsPlugin subsystemsPlugin) =>
+		options with { PlugableComponents = [.. options.PlugableComponents, .. subsystemsPlugin.GetSubsystems()] };
+
 	public static ClusterVNodeOptions WithPlugableComponent(this ClusterVNodeOptions options, IPlugableComponent plugableComponent) =>
 		options with { PlugableComponents = [.. options.PlugableComponents, plugableComponent] };
 

src/KurrentDB.Core/Services/Storage/InMemory/VirtualStreamReader.cs

@@ -9,12 +9,15 @@
 namespace KurrentDB.Core.Services.Storage.InMemory;
 
 public class VirtualStreamReader : IVirtualStreamReader {
-	private readonly IVirtualStreamReader[] _readers;
+	private IVirtualStreamReader[] _readers;
 
-	public VirtualStreamReader(IVirtualStreamReader[] readers) {
-		_readers = readers;
+	public VirtualStreamReader(IVirtualStreamReader[] readers = null) {
+		_readers = readers ?? [];
 	}
 
+	public void Register(params IVirtualStreamReader[] readers) =>
+		_readers = [.._readers, ..readers];
+
 	public ValueTask<ReadStreamEventsForwardCompleted> ReadForwards(ReadStreamEventsForward msg, CancellationToken token) {
 		if (TryGetReader(msg.EventStreamId, out var reader))
 			return reader.ReadForwards(msg, token);

src/KurrentDB.SecondaryIndexing.Tests/Indices/FakeSecondaryIndex.cs

@@ -0,0 +1,31 @@
+// 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 KurrentDB.Core.Data;
+using KurrentDB.Core.Services.Storage.InMemory;
+using KurrentDB.SecondaryIndexing.Indices;
+
+namespace KurrentDB.SecondaryIndexing.Tests.Indices;
+
+public class FakeSecondaryIndex : ISecondaryIndex {
+	public FakeSecondaryIndex(string streamName) {
+		Committed = [];
+		Processor = new FakeSecondaryIndexProcessor(Committed, Pending);
+		Readers = [new FakeVirtualStreamReader(streamName, Committed.AsReadOnly())];
+	}
+
+	public IList<ResolvedEvent> Committed { get; }
+	public IList<ResolvedEvent> Pending { get; } = new List<ResolvedEvent>();
+
+	public ISecondaryIndexProcessor Processor { get; }
+	public IReadOnlyList<IVirtualStreamReader> Readers { get; }
+	public ValueTask Init(CancellationToken ct) => ValueTask.CompletedTask;
+
+	public ValueTask<ulong?> GetLastPosition(CancellationToken ct) =>
+		ValueTask.FromResult(Committed.Select(@event => (ulong?)@event.Event.LogPosition).FirstOrDefault());
+
+	public ValueTask<ulong?> GetLastSequence(CancellationToken ct) =>
+		ValueTask.FromResult(Committed.Select(@event => (ulong?)@event.Event.EventNumber).FirstOrDefault());
+
+	public void Dispose() { }
+}

src/KurrentDB.SecondaryIndexing.Tests/Indices/FakeSecondaryIndexProcessor.cs

@@ -0,0 +1,29 @@
+// 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 FluentStorage.Utils.Extensions;
+using KurrentDB.Core.Data;
+using KurrentDB.SecondaryIndexing.Indices;
+
+namespace KurrentDB.SecondaryIndexing.Tests.Indices;
+
+public class FakeSecondaryIndexProcessor(IList<ResolvedEvent> committed, IList<ResolvedEvent>? pending = null): ISecondaryIndexProcessor {
+	private readonly object _lock = new();
+	private readonly IList<ResolvedEvent> _pending = pending ?? [];
+
+	public ValueTask Index(ResolvedEvent resolvedEvent, CancellationToken token = default) {
+		lock (_lock) {
+			_pending.Add(resolvedEvent);
+		}
+
+		return ValueTask.CompletedTask;
+	}
+
+	public ValueTask Commit(CancellationToken token = default) {
+		lock (_lock) {
+			committed.AddRange(_pending);
+			_pending.Clear();
+		}
+		return ValueTask.CompletedTask;
+	}
+}

src/KurrentDB.SecondaryIndexing.Tests/FakeVirtualStreamReader.cs renamed to src/KurrentDB.SecondaryIndexing.Tests/Indices/FakeVirtualStreamReader.cs

@@ -5,18 +5,19 @@
 using KurrentDB.Core.Messages;
 using KurrentDB.Core.Services.Storage.InMemory;
 
-namespace KurrentDB.SecondaryIndexing.Tests;
+namespace KurrentDB.SecondaryIndexing.Tests.Indices;
 
-public class FakeVirtualStreamReader(string streamName, ResolvedEvent[] events) : IVirtualStreamReader {
+internal class FakeVirtualStreamReader(string streamName, IReadOnlyList<ResolvedEvent> events) : IVirtualStreamReader {
 	public ValueTask<ClientMessage.ReadStreamEventsForwardCompleted> ReadForwards(
 		ClientMessage.ReadStreamEventsForward msg,
 		CancellationToken token) {
 
 		ReadStreamResult result;
-		long nextEventNumber, lastEventNumber;
+		long nextEventNumber, lastEventNumber, lastPosition;
+		bool isEndOfStream;
 
 		var readEvents = events
-			.Where(e =>  e.Event.EventNumber >= msg.FromEventNumber)
+			.Skip((int)msg.FromEventNumber)
 			.ToArray();
 
 		ResolvedEvent? lastEvent = readEvents.Length > 0 ? readEvents.Last(): null;
@@ -25,13 +26,16 @@ public class FakeVirtualStreamReader(string streamName, ResolvedEvent[] events)
 			result = ReadStreamResult.NoStream;
 			nextEventNumber = -1;
 			lastEventNumber = ExpectedVersion.NoStream;
+			lastPosition = -1;
+			isEndOfStream = true;
 		} else {
 			result = ReadStreamResult.Success;
-			lastEventNumber = lastEvent.Value.Event.EventNumber;
+			lastEventNumber = Array.IndexOf(readEvents, lastEvent);
 			nextEventNumber =  lastEventNumber + 1;
+			lastPosition = lastEvent.Value.Event.TransactionPosition;
+			isEndOfStream = nextEventNumber == events.Count;
 		}
 
-
 		return ValueTask.FromResult(new ClientMessage.ReadStreamEventsForwardCompleted(
 			msg.CorrelationId,
 			msg.EventStreamId,
@@ -44,19 +48,20 @@ public class FakeVirtualStreamReader(string streamName, ResolvedEvent[] events)
 			error: string.Empty,
 			nextEventNumber: nextEventNumber,
 			lastEventNumber: lastEventNumber,
-			isEndOfStream: true,
-			tfLastCommitPosition: 0L));
+			isEndOfStream: isEndOfStream,
+			tfLastCommitPosition: lastPosition));
 	}
 
 	public ValueTask<ClientMessage.ReadStreamEventsBackwardCompleted> ReadBackwards(
 		ClientMessage.ReadStreamEventsBackward msg,
 		CancellationToken token) {
 
 		ReadStreamResult result;
-		long nextEventNumber, lastEventNumber;
+		long nextEventNumber, lastEventNumber, lastPosition;
 
 		var readEvents = events
-			.Where(e =>  e.Event.EventNumber <= msg.FromEventNumber)
+			.Reverse()
+			.Skip((int)msg.FromEventNumber)
 			.ToArray();
 
 		ResolvedEvent? lastEvent = readEvents.Length > 0 ? readEvents.First(): null;
@@ -65,10 +70,12 @@ public class FakeVirtualStreamReader(string streamName, ResolvedEvent[] events)
 			result = ReadStreamResult.NoStream;
 			nextEventNumber = -1;
 			lastEventNumber = ExpectedVersion.NoStream;
+			lastPosition = -1;
 		} else {
 			result = ReadStreamResult.Success;
-			lastEventNumber = lastEvent.Value.Event.EventNumber;
+			lastEventNumber = Array.IndexOf(readEvents, lastEvent);
 			nextEventNumber =  lastEventNumber - 1;
+			lastPosition = lastEvent.Value.Event.LogPosition;
 		}
 
 		return ValueTask.FromResult(new ClientMessage.ReadStreamEventsBackwardCompleted(
@@ -77,21 +84,21 @@ public class FakeVirtualStreamReader(string streamName, ResolvedEvent[] events)
 			msg.FromEventNumber,
 			msg.MaxCount,
 			result,
-			readEvents,
+			readEvents.Reverse().ToArray(),
 			StreamMetadata.Empty,
 			isCachePublic: false,
 			error: string.Empty,
 			nextEventNumber: nextEventNumber,
 			lastEventNumber: lastEventNumber,
-			isEndOfStream: true,
-			tfLastCommitPosition: 0L));
+			isEndOfStream: lastEventNumber == 0,
+			tfLastCommitPosition: lastPosition));
 	}
 
 	public long GetLastEventNumber(string streamId) =>
-		events.Length > 0 ? events.Last().Event.EventNumber : -1;
+		events.Count > 0 ? events.Last().Event.EventNumber : -1;
 
 	public long GetLastIndexedPosition(string streamId) =>
-		events.Length > 0 ? events.Last().Event.LogPosition : -1;
+		events.Count > 0 ? events.Last().Event.LogPosition : -1;
 
 	public bool CanReadStream(string streamId) =>
 		streamId == streamName;