Activities take a long time to start

[well0549]

I implemented the following pattern.

• Get a message from a Queue
• Perfom some activities (usually 3 or 4)
• Put message on another queue

I have the same setup in a .net6 (non isolated) and that works great and fast.
I started with mainly the same codebase in .net 8.0 isolated.

Now i see the following behaviour:

• Sometimes all the activities are executed within a few seconds
• Sometimes it wil take several minutes before an activity is executed. (always the first, GetServiceUrl)

Once that first activity is executes (after a while), the remaining activities of that orchestration will run as soon as the first activity is completed.

Below is a sample of an orchestration,

     [Function(nameof(RunCleanup))]
       public static async Task RunCleanup([OrchestrationTrigger] TaskOrchestrationContext context)
       {
           TaskOptions taskOptions = new TaskOptions(new RetryPolicy(maxNumberOfAttempts: 5, firstRetryInterval: TimeSpan.FromSeconds(10)));
           FunctionGuard request = context.GetInput<FunctionGuard>();
           FunctionGuard cleanGuard = request.CloneShallow();
           try
           {

               FunctionGuard advanceGuardResult = await context.CallActivityAsync<FunctionGuard>(nameof(GetServiceUrl), request,taskOptions);
               cleanGuard = advanceGuardResult.CloneShallow();
               if (!advanceGuardResult.Success)
                   throw new Exception($"Error setting DurableGuard to Running. : {advanceGuardResult.ExtraFunctionOutput}");

               FunctionGuard attachResult = await context.CallActivityAsync<FunctionGuard>(nameof(CleanAttachments), advanceGuardResult,taskOptions);
               cleanGuard = attachResult.CloneShallow();
               if (!attachResult.Success)
                   throw new Exception($"Error deleting attachments. : {attachResult.ExtraFunctionOutput}");

               FunctionGuard payloadResult = await context.CallActivityAsync<FunctionGuard>(nameof(CleanPayload), attachResult,taskOptions);
               cleanGuard = payloadResult.CloneShallow();
               if (!payloadResult.Success)
                   throw new Exception($"Error deleting payload. : {payloadResult.ExtraFunctionOutput}");

               FunctionGuard claimResult = await context.CallActivityAsync<FunctionGuard>(nameof(CleanClaim),  payloadResult,taskOptions);
               cleanGuard = claimResult.CloneShallow();
               if (!claimResult.Success)
                   throw new Exception($"Error deleting Claim. : {claimResult.ExtraFunctionOutput}");

               FunctionGuard processGuardResult = await context.CallActivityAsync<FunctionGuard>(nameof(CleanProcessGuard),  claimResult,taskOptions);
               cleanGuard = processGuardResult.CloneShallow();
               if (!processGuardResult.Success)
                   throw new Exception($"Error deleting ProcessGuard. : {processGuardResult.ExtraFunctionOutput}");

           }
           catch (Exception ex)
           {
               await ClaimPointer.ClaimException(StringCompression.Decompress(request.FunctionInput), nameof(DurableCleanup), ex, "DurableException");
           }
           finally
           {
               // Now delete the function guards
               await context.CallActivityAsync(nameof(DeleteGuards), cleanGuard,taskOptions);
               context.SetCustomStatus("Guards deleted.");
               // Put the orchestrator to sleep for 10 minutes
               // In this way we ensure a duplicate cannot be created. All the service bus timeouts should be less than that. 
               context.SetCustomStatus("Processing completed, waiting for messages with same id.");
               DateTime dueTime = context.CurrentUtcDateTime.AddMinutes(20);
               await context.CreateTimer(dueTime, CancellationToken.None);
               context.SetCustomStatus("Timer completed");
           }
       }

A typical activity will look like this:

        [Function(nameof(CleanClaim))]
        public static async Task<FunctionGuard> CleanClaim([ActivityTrigger] FunctionGuard functionInput)
        {
            ActivityStartResult activityStartResult = await DurableHelper.ActivityStarter(functionInput, nameof(CleanClaim), true, false);
            // see if we have to run tis activity or not
            if (!activityStartResult.ContinueWithActivity)
            {
                return activityStartResult.FunctionGuardResult;
            }

            // Restore the original claim form the ActivityStarter
            ClaimPointer originalClaim = activityStartResult.originalClaimPointer;
            functionInput = activityStartResult.FunctionGuardResult;

            try
            {
                await originalClaim.Claim.DeleteBlobAsync();
                BlobGuard guard = new(originalClaim.Claim.PayloadId);
                await BlobGuard.DeleteBlobGuardAsync(guard);
                if (SNFCConfiguration.Config.Log.CurrentLogLevelValue > 4)
                    await SNFCConfiguration.Config.Log.LogMessageAsync(SNFCLogLevel.Debug, SNFCLogCategory.Function, functionInput.FunctionName, functionInput.ActivityName, $"Deleted Claim {originalClaim.Claim.PayloadName}, {originalClaim.Claim.PayloadId}, {originalClaim.Claim.Qualifier}, {originalClaim.Claim.Version}, {originalClaim.Claim.Bytes}", originalClaim.ConversationId);
            }
            catch (Exception clex)
            {
                await SNFCConfiguration.Config.Log.LogMessageAsync(SNFCLogLevel.Error, SNFCLogCategory.Function, functionInput.FunctionName, functionInput.ActivityName, $"Unexpected error deleting claim : {clex}", originalClaim.ConversationId);
            }
            functionInput.Success = true;
            functionInput.Status = "Completed";
            functionInput.FunctionOutput = functionInput.FunctionInput;
            await FunctionGuard.UpsertFunctionGuardAsync(functionInput);
            // The original input and output is no longer needed (persisted in guard) so clear to reduce orchestration persistence
            functionInput.MinimizeSerialization();

            return functionInput;
        }

I fiddled a lot with the Host.json but nothing seems to work.

  "extensions": {
    "serviceBus": {
      "clientRetryOptions": {
        "mode": "exponential",
        "tryTimeout": "00:05:00",
        "delay": "00:00:01.00",
        "maxDelay": "00:01:00",
        "maxRetries": 3
      },
      "prefetchCount": 20,
      "transportType": "amqpWebSockets",
      "autoCompleteMessages": true,
      "maxAutoLockRenewalDuration": "00:05:00",
      "maxConcurrentSessions": 4,
      "maxMessageBatchSize": 20,
      "sessionIdleTimeout": "00:00:10",
      "enableCrossEntityTransactions": false
    },
    "durableTask": {
      "useGracefulShutdown": true,
      "maxConcurrentActivityFunctions": 6,
      "maxConcurrentOrchestratorFunctions": 3,
      "hubName": "durablecleanup",
      "overridableExistingInstanceStates": 1,
      "storageProvider": {
        "partitionCount": 4
      },
      "tracing": {
        "traceInputsAndOutputs": false,
        "traceReplayEvents": false
      }
    }
  }

The functionGuard class is actually a table entity where I keep some metadata about the message.

If i look at the Instances table storage i can see 1 record with the status pending
There are is no records in the History for the partitionkey of the instance.

		<PackageReference Include="Microsoft.Azure.Functions.Worker" Version="1.22.0" />
		<PackageReference Include="Microsoft.Azure.Functions.Worker.Extensions.ServiceBus" Version="5.18.0" />
		<PackageReference Include="Microsoft.Azure.Functions.Worker.Extensions.Timer" Version="4.3.0" />
		<PackageReference Include="Microsoft.Azure.Functions.Worker.Sdk" Version="1.17.3-preview1" />
		<PackageReference Include="Microsoft.Azure.Functions.Worker.Extensions.Http.AspNetCore" Version="1.2.1" />

I need the preview because i want to clean up the function history

[pavan530]

@well0549 Were you able to figure out the issue for the above?