Conversation
Recover per plugin
Flush MarkPublished once per batch instead of once per message, cache ChangeSubscriber type resolution, and compile PostgreSQL column setters instead of using live reflection per row.
…mer channels - LZ4 decompress buffer was sized via a 255x worst-case multiplier; now prefixed with the real uncompressed length so allocation is exact. - RayTreeMeter.ChangeTag() called ChangeType.ToString() on every metric emission; the 3 known values are now precomputed statics. - Kafka/RabbitMQ consumer buffers were unbounded channels that could grow without limit if a subscriber fell behind; both are now bounded (Kafka to a fixed capacity, RabbitMQ reusing the existing PrefetchCount option) with blocking writes applying backpressure to the poll/receive loop.
Updates Microsoft.Extensions.*, EF Core, Npgsql, StackExchange.Redis, Confluent.Kafka, MessagePack, OpenTelemetry, and Testcontainers.* to their latest patch/minor versions; adds a DependencyInjection.Abstractions reference to RayTree.Core.
Ranked findings from a perf/concurrency/data-access review: reflection hot paths, connection-per-call patterns, an unawaited-task shutdown race, and asymmetric publisher/subscriber init parallelism.
CreateChange did MakeGenericType + Activator.CreateInstance + reflective PropertyInfo.SetValue on every changed entity, every SaveChanges call — the hottest reflection cost in the interceptor path. Now compiles one factory delegate per entity type via Expression.Lambda and caches it in a ConcurrentDictionary, so steady-state cost is a delegate invocation.
OnNotification fires an untracked Task.Run per delivery. StopAsync only
awaited the LISTEN and fallback-polling loops, so Dispose() could free
_notificationSemaphore while a notification handler was still running —
its `finally { Release() }` would then throw ObjectDisposedException
into an unobserved task. Now every notification task is tracked in a
ConcurrentDictionary and StopAsync waits for them (bounded by the same
5s timeout used for the other loops) before returning.
PostgreSqlOutbox<TEntity> and PostgreSqlRepository<TEntity> opened a new NpgsqlConnection + OpenAsync on every write/read call (outbox writes, polls, claims, cleanup, repository CRUD). Each now builds one NpgsqlDataSource in its constructor and issues commands through it (_dataSource.CreateCommand / OpenConnectionAsync for the batch-delete loop that reuses one connection across iterations), avoiding repeated connection-string parsing and enabling Npgsql's per-data-source prepared-statement reuse. Schema-migration code (InitializeAsync) is unchanged — startup-only, not a hot path. Verified against a real Postgres container: 116/116 tests pass.
… types Repositories, outboxes, and publishers were each initialized with a sequential foreach + await, and OutboxPublisherService.StartAsync was called one entity type at a time. One slow init (e.g. a locking Postgres ALTER TABLE for one entity type) delayed every other entity type's startup. ChangeSubscriber.InitializeAsync already parallelizes consumer init for the same reason; ChangePublisher now does too, via Task.WhenAll per phase.
CaptureChanges called _trackedEntityTypes.Contains(entityType) for every ChangeTracker entry on every SaveChanges; the field was IEnumerable<Type>, making that a linear scan. Now a HashSet<Type> (built once in the constructor if the caller didn't already pass one), so the lookup is O(1).
WriteOutboxAsync awaited WriteTypedAsync one change at a time; a SaveChanges touching N entities did N sequential DB round trips. Now collects the write tasks and awaits them with Task.WhenAll. Safe for concurrent writes to the same outbox instance — PostgreSqlOutbox.WriteAsync uses per-call commands/connections and InMemoryOutbox uses Interlocked/ConcurrentDictionary.
Both ProcessMessageAsync and ProcessIsolatedMessageAsync did handlers.Where(h => h.ChangeType == envelope.ChangeType).ToList() on every message — an iterator plus a List<> allocated per dispatch. Now a HasMatchingHandler helper does a plain for-loop existence check (no closure, no list), and the dispatch loop filters inline while iterating the original registration list directly.
…y access AddKeyParameters/MapEntity used raw PropertyInfo.GetValue/SetValue instead of the compiled-delegate cache EntityColumnMapper already provided (and that PostgreSqlOutbox.ReadEntityChange already used for setters). Added a symmetric compiled-getter cache to EntityColumnMapper and routed both methods through it.
Enum.Parse is reflection-based; KafkaConsumer.ParseEnvelope, RabbitMqConsumer.ParseEnvelope, and PostgreSqlOutbox.ReadEntityChange each called it once per message/row. Added a local ParseChangeType switch to each (duplicated per plugin by existing convention — see ComputeBackoffDelay for the same pattern), falling back to Enum.Parse for any value outside the 3 known ones so behavior is unchanged for unrecognized input. Verified: RabbitMQ 24/24, Kafka 17/17 (then an unrelated native librdkafka crash in the test host — reproduced identically on the pre-change commit, so it's a pre-existing environment flake, not a regression), Postgres 115/116 (the 1 failure is the already-known timing-sensitive FallbackPolling_DoesNotRedeliver_AlreadyPublishedChange test, confirmed flaky under full-suite load in earlier commits too).
CompressAsync/DecompressAsync copied the buffered payload a second time via ToArray() before handing it to LZ4Codec. GetBuffer() returns the internal array directly (no copy); every use is now bounded by ms.Length since the returned array can be longer than the written data. Left as-is: full elimination of the source->MemoryStream buffering step would need the streaming LZ4 Frame API (K4os.Compression.LZ4.Streams), a new package dependency and a wire-format change — a bigger call than this task's scope, flagging separately rather than pulling it in unasked.
Typeless mode embeds and reflectively resolves a runtime type name on every serialize/deserialize call, but TEntity is already known statically at both call sites — no need to carry or resolve it from the payload. ContractlessStandardResolver still handles plain POCOs without requiring [MessagePackObject] attributes on user entity types. BREAKING WIRE FORMAT CHANGE: messages already serialized with Typeless (e.g. sitting unpublished in an outbox, or in-flight in a broker) will not deserialize with this resolver. Safe for a fresh deploy or an environment with no queued MessagePack-encoded messages; coordinate a drain/flush before upgrading a live system that uses this serializer. Verified: 8/8 MessagePack serializer tests pass (in-process round trip, not a persisted-fixture wire-compatibility test).
RoutingKeySelector's default delegate ran envelope.ChangeType.ToString() then .ToLower() on every publish — two allocations, and ToLower() is culture-sensitive with no StringComparison. Replaced with a switch over the 3 known ChangeType values, falling back to ToLowerInvariant() for any future value.
Documents this session's work: the outbox/EF-Core/PostgreSQL/Kafka/ RabbitMQ/LZ4/MessagePack performance and concurrency fixes, the NotificationBasedPublisher shutdown race fix, and the dependency bump. Bumps VersionPrefix to match. Removes tasks.md now that all scheduled items are done.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
No description provided.