[ESDB-169-3]: Migration of internal message dispatcher/scheduler to async

[sakno]

Changed: InMemoryBus becomes asynchronous dispatcher. Async handlers now can use IAsyncHandle<T> interface to enable async execution.
Removed: IBus interface is removed with no replacement to clearly distinguish roles of message dispatcher and message scheduler. Removed unused schedulers and related tests.

[github-actions]

Qodana for .NET

17 new problems were found

Inspection name	Severity	Problems
Redundant using directive	🔶 Warning	7
Namespace does not correspond to file location	🔶 Warning	2
Possible unassigned object created by 'new' expression	🔶 Warning	2
Type member is never used (private accessibility)	🔶 Warning	2
Private field can be converted into local variable	🔶 Warning	1
Redundant type arguments of method	🔶 Warning	1
Static field or auto-property in generic type	🔶 Warning	1
Suspicious type conversion or check	🔶 Warning	1

💡 Qodana analysis was run in the pull request mode: only the changed files were checked
☁️ View the detailed Qodana report

Detected 188 dependencies

Third-party software list

This page lists the third-party software dependencies used in EventStore

Dependency	Version	Licenses
BenchmarkDotNet.Annotations	0.13.10	MIT
BenchmarkDotNet	0.13.10	MIT
CommandLineParser	2.9.1	MIT
DotNext.IO	5.13.0	MIT
DotNext.Threading	5.13.0	MIT
DotNext.Unsafe	5.13.0	MIT
DotNext	5.13.0	MIT
Esprima	3.0.4	BSD-3-Clause
EventStore.Plugins	24.10.0	Apache-2.0 BSD-3-Clause OTHER-PERMISSIVE
Gee.External.Capstone	2.3.0	MIT
GitInfo	3.3.3	MIT
Google.Protobuf	3.27.2	BSD-3-Clause
Grpc.AspNetCore.Server.ClientFactory	2.64.0	Apache-2.0
Grpc.AspNetCore.Server	2.64.0	Apache-2.0
Grpc.AspNetCore	2.64.0	Apache-2.0
Grpc.Core.Api	2.64.0	Apache-2.0
Grpc.Net.Client	2.64.0	Apache-2.0
Grpc.Net.ClientFactory	2.64.0	Apache-2.0
Grpc.Net.Common	2.64.0	Apache-2.0
Grpc.Tools	2.65.0	Apache-2.0
Iced	1.17.0	MIT
JetBrains.Annotations	2023.3.0	MIT
Jint	3.0.0	BSD-2-Clause
Microsoft.Bcl.AsyncInterfaces	6.0.0	MIT
Microsoft.CSharp	4.7.0	MIT
Microsoft.CodeAnalysis.Analyzers	3.3.3	MIT
Microsoft.CodeAnalysis.Analyzers	3.3.4	MIT
Microsoft.CodeAnalysis.CSharp	4.1.0	MIT
Microsoft.CodeAnalysis.CSharp	4.7.0	MIT
Microsoft.CodeAnalysis.Common	4.1.0	MIT
Microsoft.CodeAnalysis.Common	4.7.0	MIT
Microsoft.Data.Sqlite.Core	8.0.0	MIT
Microsoft.Data.Sqlite	8.0.0	MIT
Microsoft.Diagnostics.NETCore.Client	0.2.452401	MIT
Microsoft.Diagnostics.Runtime	2.2.332302	MIT
Microsoft.Diagnostics.Tracing.TraceEvent	3.1.6	MIT
Microsoft.DotNet.PlatformAbstractions	3.1.6	MIT
Microsoft.Extensions.Configuration.Abstractions	8.0.0	MIT
Microsoft.Extensions.Configuration.Binder	8.0.0	MIT
Microsoft.Extensions.Configuration.EnvironmentVariables	3.1.0	Apache-2.0
Microsoft.Extensions.Configuration.FileExtensions	8.0.0	MIT
Microsoft.Extensions.Configuration.Json	8.0.0	MIT
Microsoft.Extensions.Configuration	8.0.0	MIT
Microsoft.Extensions.DependencyInjection.Abstractions	8.0.0	MIT
Microsoft.Extensions.DependencyInjection	8.0.0	MIT
Microsoft.Extensions.DependencyModel	8.0.1	MIT
Microsoft.Extensions.FileProviders.Abstractions	8.0.0	MIT
Microsoft.Extensions.FileProviders.Composite	8.0.0	MIT
Microsoft.Extensions.FileProviders.Embedded	8.0.0	MIT
Microsoft.Extensions.FileProviders.Physical	8.0.0	MIT
Microsoft.Extensions.FileSystemGlobbing	8.0.0	MIT
Microsoft.Extensions.Hosting.Abstractions	2.1.0	Apache-2.0
Microsoft.Extensions.Http	6.0.0	MIT
Microsoft.Extensions.Logging.Abstractions	8.0.0	MIT
Microsoft.Extensions.Logging.Configuration	3.1.0	Apache-2.0
Microsoft.Extensions.Logging	8.0.0	MIT
Microsoft.Extensions.ObjectPool	6.0.16	MIT
Microsoft.Extensions.Options.ConfigurationExtensions	3.1.0	Apache-2.0
Microsoft.Extensions.Options	8.0.0	MIT
Microsoft.Extensions.Primitives	8.0.0	MIT
Microsoft.FASTER.Core	1.9.16	MIT
Microsoft.IdentityModel.Abstractions	7.6.0	MIT
Microsoft.IdentityModel.JsonWebTokens	7.6.0	MIT
Microsoft.IdentityModel.Logging	7.6.0	MIT
Microsoft.IdentityModel.Tokens	7.6.0	MIT
Microsoft.NETCore.Platforms	1.1.0	MIT
Microsoft.NETCore.Platforms	1.1.1	MIT
Microsoft.NETCore.Platforms	2.0.0	MIT
Microsoft.NETCore.Platforms	5.0.0	MIT
Microsoft.NETCore.Targets	1.1.3	MIT
Microsoft.Net.Http.Headers	8.0.0	MIT
Microsoft.Win32.Registry	4.4.0	MIT
Microsoft.Win32.Registry	5.0.0	MIT
Mono.Posix.NETStandard	1.0.0	MIT
NETStandard.Library	2.0.3	MIT
Newtonsoft.Json	13.0.3	MIT
OpenTelemetry.Api	1.4.0-rc.1	Apache-2.0
OpenTelemetry.Exporter.Prometheus.AspNetCore	1.4.0-rc.1	Apache-2.0
OpenTelemetry.Extensions.DependencyInjection	1.4.0-rc.1	Apache-2.0
OpenTelemetry.Extensions.Hosting	1.4.0-rc.1	Apache-2.0
OpenTelemetry	1.4.0-rc.1	Apache-2.0
Perfolizer	0.2.1	MIT
Quickenshtein	1.5.1	MIT
SQLitePCLRaw.bundle_e_sqlite3	2.1.6	Apache-2.0
SQLitePCLRaw.core	2.1.6	Apache-2.0
SQLitePCLRaw.lib.e_sqlite3	2.1.6	Apache-2.0
SQLitePCLRaw.provider.e_sqlite3	2.1.6	Apache-2.0
Serilog.Enrichers.Process	3.0.0	Apache-2.0
Serilog.Enrichers.Thread	4.0.0	Apache-2.0
Serilog.Expressions	5.0.0	Apache-2.0
Serilog.Extensions.Logging	8.0.0	Apache-2.0
Serilog.Settings.Configuration	8.0.2	Apache-2.0
Serilog.Sinks.Async	2.0.0	Apache-2.0
Serilog.Sinks.Console	6.0.0	Apache-2.0
Serilog.Sinks.File	6.0.0	Apache-2.0
Serilog	4.0.1	Apache-2.0
SharpDotYaml.Extensions.Configuration	0.3.0	MIT
System.Buffers	4.5.1	MIT
System.CodeDom	5.0.0	MIT
System.Collections.Concurrent	4.3.0	MIT
System.Collections.Immutable	5.0.0	MIT
System.Collections.Immutable	7.0.0	MIT
System.Collections	4.3.0	MIT
System.ComponentModel.Composition	8.0.0	MIT
System.Configuration.ConfigurationManager	8.0.0	MIT
System.Diagnostics.Debug	4.3.0	MIT
System.Diagnostics.DiagnosticSource	8.0.1	MIT
System.Diagnostics.EventLog	8.0.0	MIT
System.Diagnostics.PerformanceCounter	8.0.0	MIT
System.Diagnostics.Tracing	4.3.0	MIT
System.Formats.Asn1	8.0.1	MIT
System.Globalization.Calendars	4.3.0	MIT
System.Globalization.Extensions	4.3.0	MIT
System.Globalization	4.3.0	MIT
System.IO.FileSystem.Primitives	4.3.0	MIT
System.IO.FileSystem	4.3.0	MIT
System.IO.Hashing	8.0.0	MIT
System.IO.Pipelines	8.0.0	MIT
System.IO	4.3.0	MIT
System.Interactive.Async	5.0.0	MIT
System.Linq.Async	6.0.1	MIT
System.Linq	4.3.0	MIT
System.Management	5.0.0	MIT
System.Memory	4.5.3	MIT
System.Memory	4.5.4	MIT
System.Memory	4.5.5	MIT
System.Net.Http	4.3.4	MIT
System.Net.Primitives	4.3.0	MIT
System.Numerics.Vectors	4.4.0	MIT
System.Reactive	6.0.1	MIT
System.Reflection.Metadata	5.0.0	MIT
System.Reflection.Metadata	7.0.0	MIT
System.Reflection.Primitives	4.3.0	MIT
System.Reflection	4.3.0	MIT
System.Resources.ResourceManager	4.3.0	MIT
System.Runtime.CompilerServices.Unsafe	5.0.0	MIT
System.Runtime.CompilerServices.Unsafe	6.0.0	MIT
System.Runtime.Extensions	4.3.0	MIT
System.Runtime.Handles	4.3.0	MIT
System.Runtime.InteropServices	4.3.0	MIT
System.Runtime.Numerics	4.3.0	MIT
System.Runtime	4.3.1	MIT
System.Security.AccessControl	4.4.0	MIT
System.Security.AccessControl	5.0.0	MIT
System.Security.Cryptography.Algorithms	4.3.0	MIT
System.Security.Cryptography.Cng	4.3.0	MIT
System.Security.Cryptography.Csp	4.3.0	MIT
System.Security.Cryptography.Encoding	4.3.0	MIT
System.Security.Cryptography.OpenSsl	4.3.0	MIT
System.Security.Cryptography.Pkcs	8.0.0	MIT
System.Security.Cryptography.Primitives	4.3.0	MIT
System.Security.Cryptography.ProtectedData	8.0.0	MIT
System.Security.Cryptography.X509Certificates	4.3.0	MIT
System.Security.Cryptography.Xml	8.0.0	MIT
System.Security.Principal.Windows	4.4.0	MIT
System.Security.Principal.Windows	5.0.0	MIT
System.ServiceModel.Http	6.2.0	MIT
System.ServiceModel.Primitives	6.2.0	MIT
System.Text.Encoding.CodePages	4.5.1	MIT
System.Text.Encoding.CodePages	7.0.0	MIT
System.Text.Encoding	4.3.0	MIT
System.Text.Encodings.Web	8.0.0	MIT
System.Text.Json	8.0.4	MIT
System.Text.RegularExpressions	4.3.1	MIT
System.Threading.Channels	8.0.0	MIT
System.Threading.Tasks.Extensions	4.5.4	MIT
System.Threading.Tasks	4.3.0	MIT
System.Threading	4.3.0	MIT
ThisAssembly.Constants	1.4.1	MIT
YamlDotNet	15.1.4	MIT
YamlDotnet	13.7.1	MIT
librdkafka.redist	2.5.0	Apache-2.0
runtime.debian.8-x64.runtime.native.System.Security.Cryptography.OpenSsl	4.3.2	MIT
runtime.fedora.23-x64.runtime.native.System.Security.Cryptography.OpenSsl	4.3.2	MIT
runtime.fedora.24-x64.runtime.native.System.Security.Cryptography.OpenSsl	4.3.2	MIT
runtime.native.System.Net.Http	4.3.0	MIT
runtime.native.System.Security.Cryptography.Apple	4.3.0	MIT
runtime.native.System.Security.Cryptography.OpenSsl	4.3.2	MIT
runtime.native.System	4.3.0	MIT
runtime.opensuse.13.2-x64.runtime.native.System.Security.Cryptography.OpenSsl	4.3.2	MIT
runtime.opensuse.42.1-x64.runtime.native.System.Security.Cryptography.OpenSsl	4.3.2	MIT
runtime.osx.10.10-x64.runtime.native.System.Security.Cryptography.Apple	4.3.0	MIT
runtime.osx.10.10-x64.runtime.native.System.Security.Cryptography.OpenSsl	4.3.2	MIT
runtime.rhel.7-x64.runtime.native.System.Security.Cryptography.OpenSsl	4.3.2	MIT
runtime.ubuntu.14.04-x64.runtime.native.System.Security.Cryptography.OpenSsl	4.3.2	MIT
runtime.ubuntu.16.04-x64.runtime.native.System.Security.Cryptography.OpenSsl	4.3.2	MIT
runtime.ubuntu.16.10-x64.runtime.native.System.Security.Cryptography.OpenSsl	4.3.2	MIT
system.uritemplate		MIT

Contact Qodana team

Contact us at qodana-support@jetbrains.com

• Or via our issue tracker: https://jb.gg/qodana-issue
• Or share your feedback: https://jb.gg/qodana-discussions

[sakno]

Depends on #4413 and #4412

[timothycoleman]

Buses and Messages

The ESDB server architecture is a message processing pipeline, with stages passing messages to each other via queues.

There are three main interfaces for message processing:

• IPublisher
• I(Async)Handle (previously: IHandle)
• IEnvelope

They are all similar in that they have a method that takes a message as a parameter. The primary difference is their intent.

• IPublisher is for sending messages.
  • If you want to send messages, get hold of an IPublisher implementation to send messages with.
• I(Async)Handle is for receiving messages.
  • If you want to receive messages, implement I(Async)Handle<T>
• IEnvelope is for sending messages in reply.
  • If you want a reply for the message you are sending, include an envelope with it for the receiver to call.

Interface non-guarantees

None of the interfaces (IPublisher, I(Async)Handle, IEnvelope) guarantee anything about the processing of the message. E.g. the interfaces do not guarantee:

• What processing is done with the message before control returns to the caller.
• What recipients the message will be passed to or how many (none, one, many).
• What order messages are processed in.
• Whether the messages are processed in sequence or concurrently.

Such guarantees can be obtained, but they depend on the concrete implementations, and often depend on the way in which the concrete implementations are arranged in the pipeline.

Best practices

• Ideally the sender of a message should care as little as possible about the way the message is handled by the pipeline. It just sends and that is it.
• Typically the sender of a message wanting a reply should include an envelope that does not burden the receiver. e.g. it should queue the reply rather than making the receiver do expensive work.

Introducing Async Handling

Everything above was true before async handling was introduced and remains true after it. We are not changing the message processing principles, or the intentions behind the interfaces.

We want to be able to handle messages asynchronously (e.g. so that reads can reach out to async APIs such as S3). So naturally we introduce IAsyncHandle, but we want to minimise its impact on the codebase.

On the calling side, places that used to call Handle will now need to call HandleAsync where there is the possibility of asynchronous handling. Luckily there are not many of these (InMemoryBus)

On the implementation side, end handlers implementing IHandle<T> remain as-is. Synchronous handling is still supported via async over sync, which minimises code impact and is fine.

When called non-concurrently (which is typical), the InMemoryBus provides strong guarantees (messages are processed one at a time, in order), so it must await the HandleAsync call to provide this guarantee. It can therefore no longer provide this guarantee through the IPublisher interface (because that returns void not Task). So InMemoryBus no longer implements IPublisher, which has the advantage of flagging, as compiler errors, every place that was relying on this guarantee when publishing (because InMemoryBus was the only IPublisher with such a guarantee). Those places can then be fixed by either:
1. (TESTS ONLY): Derive a special class SynchronousScheduler from InMemoryBus that implements IPublish and hackily gives the immediate guarantee by doing sync over async. The tests can later be gradually changed according to approaches 2 or 3
2. Convert the caller to async, have them await bus.Handle (e.g. we do this for MonitoringService when it collects stats)
3. Have the caller publish on to a queue if they didn't actually need the immediate processing of the bus.

There is now no implementation of IPublisher (outside of the tests) that guarantees messages are processed one at a time and in order. Later it would be possible to add an async version of IPublisher and create such an implementation, but this would be high impact in the code and ought to be rarely used (see best practices) (unless we enqueue messages asynchronously - as in Channels).

[timothycoleman]

Nice 👍