Azure / Azure/azure-functions-durable-js

`context.df.Task.any` launches tasks even if the task was already launched and sometimes return result from a different `Task`

Open
#574 6 comments 1 reaction 0 assignees View on GitHub
Needs: Investigation :mag: P2 question
Dominant language
TypeScript
Stars
142
Forks
66
Avg merge
3d 19h
Merged PRs (30d)
4

Description

**Describe the bug**

First I would like to explain the idea I am trying to achieve with Durable Functions. I have two activities:
- `activity-get-next-products-batch` - resolves a batch of products from data source and places into a blob storage file
- `activity-process-products` - processes a batch of products produced by previous activity. In real world this can run really long

I want to achieve a result where each batch processing starts as soon as possible, but at the same time we keep producing next batch.

So when we resolve the batch, we do two actions
- Start next `activity-get-next-products-batch`
- Start `activity-process-products` for processing of the batch, this is pushed in the array of processing tasks to be awaited.

Then we do `Task.any` to wait for any of the jobs to finish (keep in mind that there can be multiple `activity-process-products` running ideally).

In this set-up `Task.any` can return two tasks:
- `activity-process-products` - then we filter finished job from the list of tasks
- `activity-get-next-products-batch` - then we schedule both jobs again, and keep waiting for previuos unfinished jobs with new call to `Task.any`

My expectations that if a job is already launched it won't be launched again we will keep waiting for it to finish, but what happnes in reallity that `activity-get-next-products-batch` invocation runs longer than `activity-process-products`, and when the code just want to wait for the batch to arrive the runtime schedules another job, rather than waiting for already scheduled. See gant chart obtained from Durable Functions Monitor.

image

And from the logs here we starting the job as `TaskEventId 2` it runs longer that another job started for processing of products.

![image](https://github.com/Azure/azure-functions-durable-js/assets/44115170/abaf2730-d485-41a3-b966-bbb3734f5d11)

Then when processing is finished we just wait for the next batch to arrive, but the runtime schedules the task again, rather than waiting for already launched task... See the `input` is the same, but `TaskEventId` is `3` this time. The code will be attached below.

![image](https://github.com/Azure/azure-functions-durable-js/assets/44115170/cec71bfe-a7b2-4e4e-af6d-762741d0aac7)

---

Another issue that I see that when in the code that when both tasks for `any` finish around the same time, the `results` are being returned from a wrong Task, though the code checks that the Task being returned `===` the task that was scheduled to gext batch.

Here on the screenshot below you can see that both tasks finished but have different output. Then we see logs that `Get Products Task Finished`, this code is
```ts
const finishedTask: Task = yield context.df.Task.any(tasksToWait);

if (getProductsTask === finishedTask) {
```
But what happens is that the `Task.any` actually returned result from a complete different task.

![image](https://github.com/Azure/azure-functions-durable-js/assets/44115170/98c7680f-9196-4a26-b729-efc1e97509cb)

**Investigative information**

- Durable Functions extension version: 2.1.12
- durable-functions npm module version: 2.1.3
- Language (JavaScript/TypeScript) and version: TypeScript 4.9.3
- Node.js version: ~18.12.1 (Azure Function Apps ~18)

***If deployed to Azure App Service***

- Timeframe issue observed: 2024-01-24T11:12:47.3924557Z - 2024-01-24T11:13:08.2962719Z
- Function App name: fa-cms-dev-ne-002
- Function name(s): bloomreach-full-feed-orchestrator
- Region: NE
- Orchestration instance ID(s): en-2024-01-24-asypevhui4i6dyu5njd3c

**To Reproduce**
Steps to reproduce the behavior:

1. Go to '...'
2. Click on '....'
3. Scroll down to '....'
4. See error

The code of orchestrator.

```tsimport * as df from 'durable-functions';
import { Logger } from '../../common/logger';
import { Task } from 'durable-functions/lib/src/task';
import { BloomreachFullFeedOrchestratorParams, GetNextProductsBatchResult, ProcessProductsResult } from './types';
import { RetryOptions } from 'durable-functions';

export default df.orchestrator(function* (context) {
const logger = new Logger(context.log, {
correlationId: context.df.newGuid(context.df.instanceId),
});
logger.setContext('bloomreach-full-feed-orchestrator');
if (!context.df.isReplaying) {
logger.info(`isReplaying=` + String(context.df.isReplaying));
}

if (!context.df.isReplaying) {
logger.info(`Input payload:`, {
input: JSON.stringify(context.df.getInput()),
});
}

const params: BloomreachFullFeedOrchestratorParams = context.df.getInput();

let lastCommerceId = null;
let batchNumber = 1;

if (!context.df.isReplaying) {
logger.info('Scheduled first task to get products.');
}

let getProductsTask: Task | null = context.df.callActivityWithRetry(
'activity-get-next-products-batch',
new RetryOptions(1000, 1),
{
lastCommerceId,
batchNumber,
orchestratorId: context.bindingData.instanceId,
}
);
let processingTasks: Task[] = [];
const processingResults: ProcessProductsResult[] = [];

while (processingTasks.length > 0 || getProductsTask !== null) {
const tasksToWait = [...processingTasks];

if (getProductsTask !== null) {
tasksToWait.push(getProductsTask);
}

if (!context.df.isReplaying) {
logger.info(`Waiting for any of the tasks to finish.`);
}

const finishedTask: Task = yield context.df.Task.any(tasksToWait);

if (getProductsTask === finishedTask) {
if (!context.df.isReplaying) {
logger.info('Get Products Task Finished.');
}

if (!finishedTask.isCompleted) {
if (!context.df.isReplaying) {
logger.warn(`Get Products Task Failed...`);
}
}

const result = finishedTask.result as GetNextProductsBatchResult;

if (!context.df.isReplaying) {
logger.info(`Result=${JSON.stringify(result)}`);
}

lastCommerceId = result.lastCommerceId;

if (!context.df.isReplaying) {
logger.info('Starting new processing job activity.');
}

processingTasks.push(
context.df.callActivityWithRetry('activity-process-products', new RetryOptions(1000, 1), {
locale: params.locale,
inputFilePath: result.filePath,
})
);

if (result.hasMoreProducts) {
if (!context.df.isReplaying) {
logger.info('Starting getting new batch activity.');
}

batchNumber++;
getProductsTask = context.df.callActivity('activity-get-next-products-batch', {
lastCommerceId,
batchNumber,
orchestratorId: context.bindingData.instanceId,
});
} else {
getProductsTask = null;
if (!context.df.isReplaying) {
logger.info('Finished products downloading from the Commercetools.');
}
}
} else {
if (!context.df.isReplaying) {
logger.info(
`Batch processing activity finished. Removing from the active list. Length=${processingTasks.length}`
);
}

processingTasks = processingTasks.filter((task) => finishedTask !== task);

if (!context.df.isReplaying) {
logger.info(`LengthAfterFilter=${processingTasks.length}`);
}

processingResults.push(finishedTask.result as ProcessProductsResult);
}
}

if (!context.df.isReplaying) {
logger.info(`Finished processing :)`);
}

// then use processing results to merge the files together and do
});

```

**Expected behavior**

a) The `Task.any` doesn't schedule a new task for already launched task.
b) The `Task.any` return the valid result for the task rather than a result from different task.

**Actual behavior**

a) The `Task.any` schedule a new task for already launched task.
b) The `Task.any` return the invalid result for the task

**Screenshots**

See above in the bug description

**Known workarounds**
MAY BE this could be workaround-ed if I treat each activity as a sub-orchestrator in "singleton" mode, where each of them has an `id`, and than parent somehow will do the similar logic with different subochestrators...

Contributor guide

Open the contributing guide

Research direction

Start with the Task.any entry point in durable-functions/lib/src/task and trace how it handles the tasks supplied by the orchestrator code. Reproduce the scenario using the listed durable-functions and Durable Functions extension versions, focusing on overlapping get-products and process-products tasks. Done means already-launched tasks are not scheduled again and Task.any returns the corresponding task and result.

Written by the indexing model from the issue text.

Assessment

Tech stack
typescript
Domain
backend, distributed-systems
Issue type
Bug
Difficulty
4/5
Estimated time
3-5 days
Activity status
Stale
Clarity
Mostly clear
Newbie friendliness
35/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.