From a6e035503494bc4ab57afb26f8a07dd4deca403f Mon Sep 17 00:00:00 2001 From: "Jeremy D. Miller" Date: Sun, 23 Aug 2026 21:08:13 -0500 Subject: [PATCH] GH-4061 follow-up: NativeAck endpoints were handed an unregistered BatchedSender Live regression on main, introduced by #4061. Sending anything to a NativeAck endpoint on Redis fails outright with: System.InvalidOperationException: This sender has not been registered. Mechanism, end to end: * #4061 remapped EndpointMode.NativeAck from BufferedSendingAgent to InlineSendingAgent, to close the interceptor loss window. * InlineSendingAgent is `ISendingAgent, IDisposable` -- deliberately NOT an ISenderCallback -- so the registration in EndpointCollection.CreateSendingAgent (`sender is ISenderRequiresCallback && agent is ISenderCallback`) is skipped. * Every transport that uses BatchedSender chooses it by asking `Mode == EndpointMode.Inline`. NativeAck is not Inline, so those transports still build a BatchedSender. * BatchedSender throws on every send, ping and failure path when `_callback` is null. Redis adopted NativeAck in #4056, so this is broken on main today. RabbitMQ and Pulsar use their own senders rather than BatchedSender, which is exactly why CI stayed green: the only NativeAck integration tests that exist run on those two transports. The fix is one property rather than nine copies of a boolean, because the failure is silent and total, and because the next mode that sends inline should not require finding all nine sites again: public bool SendsInline => Mode is EndpointMode.Inline or EndpointMode.NativeAck; Applied to the sender gates in SNS, SQS, Azure Service Bus (topic and queue), Kafka, MQTT and Redis. GCP Pub/Sub is deliberately untouched: it has not adopted NativeAck (supportsNativeAck is false there), so it cannot reach this today, and that file is being actively edited under #4065/#4066. The Pub/Sub gate should move to SendsInline whenever #4052 adopts the mode. Red-baselined: with the Redis gate reverted to `Mode == EndpointMode.Inline`, three of the six Redis native_ack_mode tests fail with the exact "This sender has not been registered." exception; with SendsInline, 6/6 pass. Verified: pinned dotnet build wolverine.slnx -c Release -f net9.0 clean, CoreTests 2581 / 0 failed, Redis native_ack_mode 6/6. Co-Authored-By: Claude Opus 5 --- .../Internal/AmazonSnsTopic.cs | 2 +- .../Internal/AmazonSqsQueue.cs | 2 +- .../AzureServiceBusTransport.Sending.cs | 4 ++-- .../Kafka/Wolverine.Kafka/KafkaTopic.cs | 2 +- .../MQTT/Wolverine.MQTT/MqttTopic.cs | 2 +- .../Internal/RedisStreamEndpoint.cs | 2 +- src/Wolverine/Configuration/Endpoint.cs | 20 +++++++++++++++++++ 7 files changed, 27 insertions(+), 7 deletions(-) diff --git a/src/Transports/AWS/Wolverine.AmazonSns/Internal/AmazonSnsTopic.cs b/src/Transports/AWS/Wolverine.AmazonSns/Internal/AmazonSnsTopic.cs index 3455d292c..d811c6462 100644 --- a/src/Transports/AWS/Wolverine.AmazonSns/Internal/AmazonSnsTopic.cs +++ b/src/Transports/AWS/Wolverine.AmazonSns/Internal/AmazonSnsTopic.cs @@ -212,7 +212,7 @@ protected override ISender CreateSender(IWolverineRuntime runtime) return tenantedSender; } - if (Mode == EndpointMode.Inline) + if (SendsInline) { return new InlineSnsSender(runtime, this); } diff --git a/src/Transports/AWS/Wolverine.AmazonSqs/Internal/AmazonSqsQueue.cs b/src/Transports/AWS/Wolverine.AmazonSqs/Internal/AmazonSqsQueue.cs index 24f64dc0a..8a2f087d8 100644 --- a/src/Transports/AWS/Wolverine.AmazonSqs/Internal/AmazonSqsQueue.cs +++ b/src/Transports/AWS/Wolverine.AmazonSqs/Internal/AmazonSqsQueue.cs @@ -704,7 +704,7 @@ protected override ISender CreateSender(IWolverineRuntime runtime) return tenantedSender; } - if (Mode == EndpointMode.Inline) + if (SendsInline) { return new InlineSqsSender(runtime, this); } diff --git a/src/Transports/Azure/Wolverine.AzureServiceBus/AzureServiceBusTransport.Sending.cs b/src/Transports/Azure/Wolverine.AzureServiceBus/AzureServiceBusTransport.Sending.cs index a04e92b85..ebcd6397d 100644 --- a/src/Transports/Azure/Wolverine.AzureServiceBus/AzureServiceBusTransport.Sending.cs +++ b/src/Transports/Azure/Wolverine.AzureServiceBus/AzureServiceBusTransport.Sending.cs @@ -31,7 +31,7 @@ private ISender buildSenderForTopic(IWolverineRuntime runtime, AzureServiceBusTo { var sender = BusClient.CreateSender(topic.TopicName); - if (topic.Mode == EndpointMode.Inline) + if (topic.SendsInline) { var inlineSender = new InlineAzureServiceBusSender(topic, mapper, sender, runtime.LoggerFactory.CreateLogger(), runtime.Cancellation); @@ -120,7 +120,7 @@ private ISender buildSenderForQueue(IWolverineRuntime runtime, AzureServiceBusQu { var sender = BusClient.CreateSender(queue.QueueName); - if (queue.Mode == EndpointMode.Inline) + if (queue.SendsInline) { var inlineSender = new InlineAzureServiceBusSender(queue, mapper, sender, runtime.LoggerFactory.CreateLogger(), runtime.Cancellation); diff --git a/src/Transports/Kafka/Wolverine.Kafka/KafkaTopic.cs b/src/Transports/Kafka/Wolverine.Kafka/KafkaTopic.cs index ff1c9adec..81794e5b3 100644 --- a/src/Transports/Kafka/Wolverine.Kafka/KafkaTopic.cs +++ b/src/Transports/Kafka/Wolverine.Kafka/KafkaTopic.cs @@ -331,7 +331,7 @@ protected override ISender CreateSender(IWolverineRuntime runtime) return tenantedSender; } - return Mode == EndpointMode.Inline + return SendsInline ? new InlineKafkaSender(this) : new BatchedSender(this, new KafkaSenderProtocol(this), runtime.Cancellation, runtime.LoggerFactory.CreateLogger()); diff --git a/src/Transports/MQTT/Wolverine.MQTT/MqttTopic.cs b/src/Transports/MQTT/Wolverine.MQTT/MqttTopic.cs index 8b03db0a5..fa2f92059 100644 --- a/src/Transports/MQTT/Wolverine.MQTT/MqttTopic.cs +++ b/src/Transports/MQTT/Wolverine.MQTT/MqttTopic.cs @@ -131,7 +131,7 @@ protected override ISender CreateSender(IWolverineRuntime runtime) // Inline keeps the historical immediate/fire-and-forget path (this itself, via SendAsync // below) so explicit inline usage is unaffected. - if (Mode == EndpointMode.Inline) + if (SendsInline) { return this; } diff --git a/src/Transports/Redis/Wolverine.Redis/Internal/RedisStreamEndpoint.cs b/src/Transports/Redis/Wolverine.Redis/Internal/RedisStreamEndpoint.cs index 8cdf8dea1..e404e88d5 100644 --- a/src/Transports/Redis/Wolverine.Redis/Internal/RedisStreamEndpoint.cs +++ b/src/Transports/Redis/Wolverine.Redis/Internal/RedisStreamEndpoint.cs @@ -341,7 +341,7 @@ protected override ISender CreateSender(IWolverineRuntime runtime) return tenantedSender; } - return Mode == EndpointMode.Inline + return SendsInline ? new InlineRedisStreamSender(_transport, this, runtime) : new BatchedSender(this, new RedisSenderProtocol(_transport, this), runtime.Cancellation, runtime.LoggerFactory.CreateLogger()); diff --git a/src/Wolverine/Configuration/Endpoint.cs b/src/Wolverine/Configuration/Endpoint.cs index b2e8a29ed..6c6f7bc67 100644 --- a/src/Wolverine/Configuration/Endpoint.cs +++ b/src/Wolverine/Configuration/Endpoint.cs @@ -282,6 +282,26 @@ public int MaxDegreeOfParallelism /// internal bool ModeIgnoresParallelism => Mode == EndpointMode.Inline; + /// + /// GH-4061. Does this endpoint send synchronously, awaiting the transport, rather than through a batching + /// sender? + /// + /// + /// Transports use this to choose between their inline sender and a BatchedSender. It is one property + /// because getting it wrong is silent and total: a BatchedSender only functions once the sending agent + /// has registered a callback on it, and InlineSendingAgent is NOT an ISenderCallback, so the + /// registration in EndpointCollection.CreateSendingAgent is skipped for it. A transport that asks + /// Mode == EndpointMode.Inline directly therefore hands a NativeAck endpoint an unregistered + /// BatchedSender whose every send throws "This sender has not been registered." + /// + /// + /// + /// Ask this rather than comparing against , so that a future mode which also + /// sends inline does not require finding all nine call sites again. + /// + /// + public bool SendsInline => Mode is EndpointMode.Inline or EndpointMode.NativeAck; + /// /// GH-3712. Render for diagnostics, saying "n/a" rather than /// printing a dead number for a mode that never reads it.