InfluxDB v2.3.0: Stuck task runs

Nobody has claimed this yet.

__Steps to reproduce:__

1. Set up a task
2. Wait for the task's execution
3. The task is executed but sometimes it is stuck

__Expected behaviour:__
All task are executed and finished in finite time.

__Actual behaviour:__
The task is sometimes stuck without any status progress.

__Environment info:__

Basic installation using package manager.
`apt install influxdb2` [using version 2.3.0](https://github.com/influxdata/influxdb/tree/090f681737)
OS: Linux 4.19.0-17-amd64 Debian 4.19.194-3 x86_64
Disk info: SSD

__Config:__
```
bolt-path = "/mnt/ssd2/influxdb2/influxd.bolt"
engine-path = "/mnt/ssd2/influxdb2/engine"
http-read-timeout = "30m"
http-write-timeout = "30m"
http-bind-address = ":8086"
query-concurrency = 20
query-queue-size = 200
storage-max-concurrent-compactions = 1
storage-compact-throughput-burst = 1485760
storage-compact-throughput = 1485760
storage-tsm-use-madv-willneed = true
max-series-per-database = 0
```

__Logs:__
No logs were found.
It might be included later.

__Task Setup:__
```
option task = {name: "top_content", every: 1d, offset: 1h}

data =
from(bucket: "mcollector_raw_data")
|> range(start: -task.every)
|> filter(
fn: (r) => r["_measurement"] == "app_watch" and r["_field"] == "channelEvent" or r["_field"] == "user",
)
|> pivot(rowKey: ["_time"], columnKey: ["_field"], valueColumn: "_value")
|> group(columns: ["user"])
|> filter(fn: (r) => r["channelEvent"] != "")
|> unique(column: "channelEvent")
|> group(columns: ["channelEvent"])
|> aggregateWindow(every: 4d, fn: count, column: "user")
|> set(key: "_field", value: "user_count")
|> yield(name: "_field")
|> rename(columns: {user: "_value"})
|> set(key: "_measurement", value: "top_content")
|> to(bucket: "mcollector_aggregations", tagColumns: ["channelEvent"])
```

__Task Run:__
(refer to the following picture of the InfluxDB UI)
From oldest to newest:
* first task was stuck
* the two following tasks got finished
* and the remaining tasks got stuck again


API response:
```
{
"runs": [
{
"id": "0c0b8568a0562000",
"taskID": "0b1da7efccf7d000",
"status": "started",
"scheduledFor": "2023-10-30T00:00:00Z",
"startedAt": "2023-10-30T01:00:00.003568993Z",
"log": [
{
"runID": "0c0b8568a0562000",
"time": "2023-10-30T01:00:00.002924719Z",
"message": "Started task from script: \"option task = {name: \\\"top_content\\\", every: 1d, offset: 1h}\\n\\ndata =\\n from(bucket: \\\"mcollector_raw_data\\\")\\n |> range(start: -task.every)\\n |> filter(\\n fn: (r) => r[\\\"_measurement\\\"] == \\\"app_watch\\\" and r[\\\"_field\\\"] == \\\"channelEvent\\\" or r[\\\"_field\\\"] == \\\"user\\\",\\n )\\n |> pivot(rowKey: [\\\"_time\\\"], columnKey: [\\\"_field\\\"], valueColumn: \\\"_value\\\")\\n |> group(columns: [\\\"user\\\"])\\n |> filter(fn: (r) => r[\\\"channelEvent\\\"] != \\\"\\\")\\n |> unique(column: \\\"channelEvent\\\")\\n |> group(columns: [\\\"channelEvent\\\"])\\n |> aggregateWindow(every: 4d, fn: count, column: \\\"user\\\")\\n |> set(key: \\\"_field\\\", value: \\\"user_count\\\")\\n |> yield(name: \\\"_field\\\")\\n |> rename(columns: {user: \\\"_value\\\"})\\n |> set(key: \\\"_measurement\\\", value: \\\"top_content\\\")\\n |> to(bucket: \\\"mcollector_aggregations\\\", tagColumns: [\\\"channelEvent\\\"])\""
}
]
},
{
"id": "0c148889a0962000",
"taskID": "0b1da7efccf7d000",
"status": "started",
"scheduledFor": "2023-11-06T00:00:00Z",
"startedAt": "2023-11-06T01:00:00.008447752Z",
"log": [
{
"runID": "0c148889a0962000",
"time": "2023-11-06T01:00:00.007330033Z",
"message": "Started task from script: \"option task = {name: \\\"top_content\\\", every: 1d, offset: 1h}\\n\\ndata =\\n from(bucket: \\\"mcollector_raw_data\\\")\\n |> range(start: -task.every)\\n |> filter(\\n fn: (r) => r[\\\"_measurement\\\"] == \\\"app_watch\\\" and r[\\\"_field\\\"] == \\\"channelEvent\\\" or r[\\\"_field\\\"] == \\\"user\\\",\\n )\\n |> pivot(rowKey: [\\\"_time\\\"], columnKey: [\\\"_field\\\"], valueColumn: \\\"_value\\\")\\n |> group(columns: [\\\"user\\\"])\\n |> filter(fn: (r) => r[\\\"channelEvent\\\"] != \\\"\\\")\\n |> unique(column: \\\"channelEvent\\\")\\n |> group(columns: [\\\"channelEvent\\\"])\\n |> aggregateWindow(every: 4d, fn: count, column: \\\"user\\\")\\n |> set(key: \\\"_field\\\", value: \\\"user_count\\\")\\n |> yield(name: \\\"_field\\\")\\n |> rename(columns: {user: \\\"_value\\\"})\\n |> set(key: \\\"_measurement\\\", value: \\\"top_content\\\")\\n |> to(bucket: \\\"mcollector_aggregations\\\", tagColumns: [\\\"channelEvent\\\"])\""
}
]
},
{
"id": "0c15d220a0962000",
"taskID": "0b1da7efccf7d000",
"status": "started",
"scheduledFor": "2023-11-07T00:00:00Z",
"startedAt": "2023-11-07T01:00:00.006719187Z",
"log": [
{
"runID": "0c15d220a0962000",
"time": "2023-11-07T01:00:00.005677347Z",
"message": "Started task from script: \"option task = {name: \\\"top_content\\\", every: 1d, offset: 1h}\\n\\ndata =\\n from(bucket: \\\"mcollector_raw_data\\\")\\n |> range(start: -task.every)\\n |> filter(\\n fn: (r) => r[\\\"_measurement\\\"] == \\\"app_watch\\\" and r[\\\"_field\\\"] == \\\"channelEvent\\\" or r[\\\"_field\\\"] == \\\"user\\\",\\n )\\n |> pivot(rowKey: [\\\"_time\\\"], columnKey: [\\\"_field\\\"], valueColumn: \\\"_value\\\")\\n |> group(columns: [\\\"user\\\"])\\n |> filter(fn: (r) => r[\\\"channelEvent\\\"] != \\\"\\\")\\n |> unique(column: \\\"channelEvent\\\")\\n |> group(columns: [\\\"channelEvent\\\"])\\n |> aggregateWindow(every: 4d, fn: count, column: \\\"user\\\")\\n |> set(key: \\\"_field\\\", value: \\\"user_count\\\")\\n |> yield(name: \\\"_field\\\")\\n |> rename(columns: {user: \\\"_value\\\"})\\n |> set(key: \\\"_measurement\\\", value: \\\"top_content\\\")\\n |> to(bucket: \\\"mcollector_aggregations\\\", tagColumns: [\\\"channelEvent\\\"])\""
}
]
},
{
"id": "0c133ef2a0962000",
"taskID": "0b1da7efccf7d000",
"status": "success",
"scheduledFor": "2023-11-05T00:00:00Z",
"startedAt": "2023-11-05T01:00:00.007061808Z",
"finishedAt": "2023-11-05T01:09:19.997668468Z",
"log": [
{
"runID": "0c133ef2a0962000",
"time": "2023-11-05T01:00:00.006444839Z",
"message": "Started task from script: \"option task = {name: \\\"top_content\\\", every: 1d, offset: 1h}\\n\\ndata =\\n from(bucket: \\\"mcollector_raw_data\\\")\\n |> range(start: -task.every)\\n |> filter(\\n fn: (r) => r[\\\"_measurement\\\"] == \\\"app_watch\\\" and r[\\\"_field\\\"] == \\\"channelEvent\\\" or r[\\\"_field\\\"] == \\\"user\\\",\\n )\\n |> pivot(rowKey: [\\\"_time\\\"], columnKey: [\\\"_field\\\"], valueColumn: \\\"_value\\\")\\n |> group(columns: [\\\"user\\\"])\\n |> filter(fn: (r) => r[\\\"channelEvent\\\"] != \\\"\\\")\\n |> unique(column: \\\"channelEvent\\\")\\n |> group(columns: [\\\"channelEvent\\\"])\\n |> aggregateWindow(every: 4d, fn: count, column: \\\"user\\\")\\n |> set(key: \\\"_field\\\", value: \\\"user_count\\\")\\n |> yield(name: \\\"_field\\\")\\n |> rename(columns: {user: \\\"_value\\\"})\\n |> set(key: \\\"_measurement\\\", value: \\\"top_content\\\")\\n |> to(bucket: \\\"mcollector_aggregations\\\", tagColumns: [\\\"channelEvent\\\"])\""
},
{
"runID": "0c133ef2a0962000",
"time": "2023-11-05T01:09:19.997114148Z",
"message": "Completed(success)"
}
]
},
{
"id": "0c11f55ba0562000",
"taskID": "0b1da7efccf7d000",
"status": "success",
"scheduledFor": "2023-11-04T00:00:00Z",
"startedAt": "2023-11-04T01:00:00.004565228Z",
"finishedAt": "2023-11-04T01:08:15.168789709Z",
"log": [
{
"runID": "0c11f55ba0562000",
"time": "2023-11-04T01:00:00.003842208Z",
"message": "Started task from script: \"option task = {name: \\\"top_content\\\", every: 1d, offset: 1h}\\n\\ndata =\\n from(bucket: \\\"mcollector_raw_data\\\")\\n |> range(start: -task.every)\\n |> filter(\\n fn: (r) => r[\\\"_measurement\\\"] == \\\"app_watch\\\" and r[\\\"_field\\\"] == \\\"channelEvent\\\" or r[\\\"_field\\\"] == \\\"user\\\",\\n )\\n |> pivot(rowKey: [\\\"_time\\\"], columnKey: [\\\"_field\\\"], valueColumn: \\\"_value\\\")\\n |> group(columns: [\\\"user\\\"])\\n |> filter(fn: (r) => r[\\\"channelEvent\\\"] != \\\"\\\")\\n |> unique(column: \\\"channelEvent\\\")\\n |> group(columns: [\\\"channelEvent\\\"])\\n |> aggregateWindow(every: 4d, fn: count, column: \\\"user\\\")\\n |> set(key: \\\"_field\\\", value: \\\"user_count\\\")\\n |> yield(name: \\\"_field\\\")\\n |> rename(columns: {user: \\\"_value\\\"})\\n |> set(key: \\\"_measurement\\\", value: \\\"top_content\\\")\\n |> to(bucket: \\\"mcollector_aggregations\\\", tagColumns: [\\\"channelEvent\\\"])\""
},
{
"runID": "0c11f55ba0562000",
"time": "2023-11-04T01:08:15.168081479Z",
"message": "Completed(success)"
}
]
}
]
}
```

__Metrics:__
**The metrics that are included are from different runs. The past runs were not available anymore. If it would be helpful, we are able to show data from any given time window.**
Based on the metrics the task starting on the timeline finished without problems.
When the tasks got stuck the memory has increased without it being freed.


The same time period but different metric active tasks.
From left to right the tasks were finished without problems (there is an increase and decrease that depicts the tasks being finished)
So we suspect that there is a connection between tasks started and stuck with memory increase.
There are more tasks than one being executed. (that is why the number shown is about 8)


Start by reproducing the task with the provided Flux script on InfluxDB v2.3.0 and inspect task-run status, logs, and memory behavior over the reported time window. Done means scheduled task runs consistently finish with status progress and do not leave memory growing; the issue does not identify source files or tests.

Written by the indexing model from the issue text.