Conversation
Test Results 48 files ± 0 48 suites ±0 16m 5s ⏱️ +50s Results for commit e38e910. ± Comparison against base commit ed6cfd1. This pull request removes 9 and adds 59 tests. Note that renamed tests count towards both.♻️ This comment has been updated with latest results. |
51f5c55 to
327fd91
Compare
327fd91 to
7207869
Compare
PR Summary by QodoFix subscription consume-path bugs and reduce hot-path allocations
AI Description
Diagram
High-Level Assessment
Files changed (57)
|
Code Review by Qodo
1.
|
…tions - MurmurHash3 hashes the partition key through a span byte view instead of reading an unpinned string; AllowUnsafeBlocks is no longer needed - MessageConsumeContextConverter caches are thread-safe - TracingFilter disposes only the activities it created, so trace-flag changes made after handling take effect, and it keeps the error status a failed handler set instead of overwriting it with OK - ConsumePipe validates the second filter's context type when the pipe is composed, not on every message - DefaultConsumer logging scope is a KeyValuePair array - MessageConsumeContext.Items is allocated on first use and published atomically, so racing first accesses share one bag - Checkpointed runs build their ack and nack delegates once per run Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
…easures - TracedEventHandler and TracedCommandService use a shared static DiagnosticListener, so metrics come from every instance, not only the most recently created one - Handlers, command services and the traced event writer skip measures when nobody listens; Measure uses Stopwatch - Activity names are cached per message type - ActivityStatus.Ok() is a shared instance; GetParentTag reads the tag without enumerating - The subscription duration metric test runs for every store Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
- ThrowingCommandService returns the inner result on success instead of always throwing - Resolver null-check messages are built only on failure - ApplicationEventSource builds error event text only when enabled Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
DefaultEventSerializer and DefaultStaticEventSerializer allocated a new FailedToDeserialize per unknown type, content-type mismatch or empty payload, which is every unregistered event on $all. Records are immutable, so share one static instance per error kind. Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
- Postgres, SQL Server, SQLite and Redis subscriptions deserialize metadata with the configured IMetadataSerializer through DeserializeMeta, so malformed metadata is logged (or throws DeserializationException under ThrowOnError) instead of faulting the poll loop - Metadata is skipped for events whose payload didn't deserialize - Redis $all resolves each link with a single-entry XRANGE - Postgres builds its query text once per schema instead of per access Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
…DB and brokers - KurrentDB (all, stream, persistent), RabbitMQ, Service Bus and Pub/Sub subscriptions don't deserialize or build metadata when the payload didn't deserialize; such events are acknowledged without entering the pipe - Pub/Sub deserializes from the message memory without copying it - The Cloud Run receive log moves to Debug and logs the message id only Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
7207869 to
e38e910
Compare
Summary
This PR contains low-risk bug fixes and allocation cuts on the subscription consume path, plus a few from diagnostics, application and persistence. It has no redesigns and no public signature changes. Each commit covers one area and builds on its own.
fix(subscriptions)TracingFilterdisposes only the activities it created and keeps a failed handler's error status.ConsumePipevalidates filter types when the pipe is composed. Items dictionary is allocated lazily; logging scope is an array; ack/nack delegates are built once per run.fix(diagnostics)fix(application)ThrowingCommandServicereturns the result on success. Failure-only strings are built only on failure.perf(domain)Aggregate.Changesdoesn't allocate a new view on each access.perf(serialization)fix(sql)$alllink resolution reads exactly one entry. Postgres builds its query text once per schema instead of on every access.perf(subscriptions)Observable behaviour changes
Subscription spans on the synchronous path. This covers Service Bus, Pub/Sub, Cloud Run and KurrentDB persistent subscriptions.
TracingFilterused to dispose the subscription activity it reused, which stopped the span beforeEventSubscription.Handlerhad finished with it. Now the handler ends the span:Recordedbefore it stops, so failed spans that a sampler had dropped are now exported.TracingFilterno longer overwrites the error status a failed handler set (viaNack) with OK. This also corrects failed spans that were already exported, which used to show OK.Ignored messages are unchanged: their span was already marked not-recorded before it stopped. Checkpointed subscriptions (the async path) are unaffected, because there the filter always starts and disposes its own span.
Metrics from every instance. Duration and error metrics now come from every traced handler and command service. Before, only the most recently created one was observed.
Stopwatch, so they are monotonic.DiagnosticListenerobserver that subscribes with a predicate rejecting the measure event no longer receives it. The built-in metrics subscribe without a predicate.SQL and Redis metadata. Postgres, SQL Server and SQLite subscriptions now use the metadata serializer they are given, including one registered in DI. Before, it was injected but ignored. The concrete Redis subscriptions don't take one and keep the default. A
RedisSubscriptionBasesubclass that passes one now has it honoured.Before, malformed metadata threw out of the poll loop, whatever
ThrowOnErrorwas set to. The subscription dropped and resubscribed from the same checkpoint indefinitely. Now:ThrowOnError, the error is logged, the event is delivered withMetadata == null, and the checkpoint moves on. Handlers that dereference metadata without a null check now run for such events.ThrowOnError, the loop behaves as before. The exception is now aDeserializationException, and an error is logged on each attempt.No metadata on payload-less events. This applies when the payload didn't deserialize: it was empty, its type isn't registered, or deserialization failed and the failure was caught. On every transport the context then has
Metadata == null. Nothing reads it, because such contexts are ignored and acknowledged without entering the pipe or starting an activity. The visible effects:ThrowOnErroron every transport, and on SQL and Redis in either mode.ConsumePipechecks filter types when it is built. An incompatible second filter throwsInvalidContextTypeExceptionwhile the pipe is composed. Before, aConsumeFilter<,>-derived first filter already threwArgumentExceptionon every message for the same mismatch. The only composition that used to work and is now rejected: a first filter that implementsIConsumeFilter<,>directly (skipping that per-message check), declares aTOutthe second filter can't consume, yet passes it compatible contexts at runtime.ThrowingCommandServicereturns the result on success. It used to always throwApplicationException, a regression from d0499d4 (2024). This restores the earlier behaviour.Redis
$allresolves each link with a single-entryXRANGE, which picks the same entry as before whenever it exists. A linked entry can only go missing through an externalXDELorXTRIM.Cloud Run Pub/Sub receive log moved from Info to Debug. It logs the message id only, and no longer the payload and attributes.
Shared instances.
ActivityStatus.Ok()andFailedToDeserializeresults are now shared instances, andAggregate.Changesreturns the same read-only view on every access. This is only visible through reference equality.DefaultConsumerlogging scope state is aKeyValuePairarray instead of a dictionary, with the same keys and values.Public API
BaseTracer.StartMeasurewent fromprivatetoprivate protected.Testing
On net10.0, locally, these suites pass:
Not verified yet:
src/Benchmarksbefore marking this ready.🤖 Generated with Claude Code