KeyNotFoundException in ASB LockRenewalBehavior

,
1

Hello and thanks in advance.
I’m following this example to try to renew the lock on a message using ASB transport.

When the behavior is invoked the line:

var (serviceBusConnection, path) = transportTransaction.Get<(ServiceBusConnection, string)>();

to retrive the connection string and the queue path throws this exception:

System.Collections.Generic.KeyNotFoundException: No item found in behavior context with key: System.ValueTuple`2[[Microsoft.Azure.ServiceBus.ServiceBusConnection, Microsoft.Azure.ServiceBus, Version=5.0.0.0, Culture=neutral, PublicKeyToken=7e34167dcc6d6d8c],[System.String, System.Private.CoreLib, Version=4.0.0.0, Culture=neutral, PublicKeyToken=7cec85d7bea7798e]]
       at NServiceBus.Extensibility.ContextBag.Get[T](String key) in /_/src/NServiceBus.Core/Extensibility/ContextBag.cs:line 74
       at NServiceBus.Extensibility.ContextBag.Get[T]() in /_/src/NServiceBus.Core/Extensibility/ContextBag.cs:line 25
       at Elfo.Alm.Accounts.Endpoint.Infrastructure.LockRenewalBehavior.Invoke(ITransportReceiveContext context, Func`1 next) in D:\BuildAgent_ADOS_1\_work\122\s\src\Backend\Endpoint\Infrastructure\NServiceBusExtensions.cs:line 51
       at NServiceBus.MainPipelineExecutor.Invoke(MessageContext messageContext) in /_/src/NServiceBus.Core/Pipeline/MainPipelineExecutor.cs:line 35
       at NServiceBus.Transport.AzureServiceBus.MessagePump.InnerProcessMessage(Task`1 receiveTask)

I register the behavior in this way:

public static void LockRenewal(this EndpointConfiguration endpointConfiguration, Action<LockRenewalOptions> action)
{
		var lockRenewalOptions = new LockRenewalOptions();
		action(lockRenewalOptions);

		var pipeline = endpointConfiguration.Pipeline;
		var renewLockTokenIn = lockRenewalOptions.LockDuration - lockRenewalOptions.ExecuteLockRenewalBefore;

		pipeline.Register(
				stepId: "LockRenewal",
				factoryMethod: builder => new LockRenewalBehavior(renewLockTokenIn),
				description: "Renew message lock token");
}

Could you please help me on what causes this error?
Is there another way to retrieve the ASB connection string and the queue path?

2

How do you configure your endpoint? Are you using SendsAtomicWithReceive mode (the default) or override it? This is required and explicitly mentioned in the sample.

And how do you host your endpoint?

3

Hi Sean, thank you.
this the method that configures the endpoint (with SendsAtomicWithReceive as default).

static void ConfigureEndpoint(EndpointConfiguration e, IConfiguration configuration)
{
    var transport = e.UseTransport<AzureServiceBusTransport>();
    transport.ConnectionString(configuration["NServiceBus:Transport"]);
    transport.SubscriptionRuleNamingConvention(ruleName => ruleName.FullName?.Split('.').Last().Replace("DomainEvent", string.Empty).Replace("Event", string.Empty));
    
    #region Override Lock Renewal
    
    e.LockRenewal(options =>
    {
        options.LockDuration = TimeSpan.FromMinutes(10);
        options.ExecuteLockRenewalBefore = TimeSpan.FromSeconds(10);
    });
    
    #endregion 
    
    var persistence = e.UsePersistence<SqlPersistence>();
    persistence.ConnectionBuilder(() => new SqlConnection(configuration.GetConnectionString("Db")));
    persistence.TablePrefix(string.Empty);
    var dialect = persistence.SqlDialect<SqlDialect.MsSqlServer>();
    dialect.Schema("nservicebus");
    var sagaSettings = persistence.SagaSettings();
    sagaSettings.JsonSettings(EventSerialization.Settings);

    var outboxSettings = e.EnableOutbox();
    outboxSettings.UseTransactionScope();
    outboxSettings.KeepDeduplicationDataFor(TimeSpan.FromDays(6));
    outboxSettings.RunDeduplicationDataCleanupEvery(TimeSpan.FromMinutes(15));

    e.Conventions().DefiningEventsAs(t => t.Namespace != null && t.Namespace.EndsWith("Events"))
                    .DefiningCommandsAs(t => t.Namespace != null && t.Namespace.EndsWith("Commands"));

    e.AuditProcessedMessagesTo(configuration["NServiceBus:AuditQueue"]);
    e.SendFailedMessagesTo("error");
    e.SendHeartbeatTo(serviceControlQueue: configuration["NServiceBus:ServiceControlInstance"],
        frequency: TimeSpan.FromSeconds(15),
        timeToLive: TimeSpan.FromSeconds(30));

    var endpointName = configuration["NServiceBus:EndpointName"];
    var machineName = $"{Dns.GetHostName()}.{IPGlobalProperties.GetIPGlobalProperties().DomainName}";
    var instanceIdentifier = $"{endpointName}@{machineName}";

    var metrics = e.EnableMetrics();
    metrics.SendMetricDataToServiceControl(serviceControlMetricsAddress: configuration["NServiceBus:ServiceControlMonitoringInstance"],
        interval: TimeSpan.FromSeconds(10),
        instanceId: instanceIdentifier);

    e.EnableInstallers();
}

We host the endpoint in a windows service.

4

Andrea looks like the outbox feature is interfering here. Outbox automatically downgrades the transport transaction mode and that’s what’s causing the behaviour to fail. You’d either need to remove the outbox or the behaviour.

I’ll update the sample to indicate that lock renewal is not compatible with the outbox feature.

5

oohhh…so this means that is practically impossible to try to renew the lock in case of outbox…
but this is true for this example that uses the behavior or is it impossibile in general with the ReceiveOnly transport level?

6

Sorry, my bad. The outbox feature sets the default. But you can override it by explicitly configuring your endpoint to use the desired transport transaction mode. In your endpoint configuration code, set the transport transaction mode to SendsAtomicWithReceive.

var transport = endpointConfiguration.UseTransport<MyTransport>();
transport.Transactions(TransportTransactionMode.SendsAtomicWithReceive);

That should allow you to use lock renewal behaviour along with the outbox.

7

Andrea,

The sample documentation has been updated. Also, the outbox documentation has already contained a note about this behaviour. Thank you for bringing it up.

8

Thank you Sean,
I try today.

1 Like