# Pipeline Task Lifecycle And Thread Model

Scope:
- `pipeline/internal/otel_collector/extension/taskqueue/`
- `pipeline/internal/otel_collector/receiver/metadb/`
- `pipeline/internal/otel_collector/processor/thanos/`
- `pipeline/internal/otel_collector/exporter/otlphttpexporter/`

This document describes how one task window moves through the pipeline, which goroutines are involved, and how `ack`, `nack`, `timeout`, `requeue`, and `bypass` interact.

## 1) Core task fields

File: `pipeline/internal/otel_collector/extension/taskqueue/task_model.go`

Important fields:
- `Timestamp`: task start timestamp
- `EndTimestamp`: task end timestamp
- `Step`: task resolution and truncation unit
- `AckTimestamp`: downstream progress chosen for ack
- `retriedCount`: number of retries already performed
- `firstFailedAt`: time when the task first failed or first timed out
- `retryStrategy`: current retry mode

## 2) Main goroutines

### G1: taskqueue dispatcher

File: `pipeline/internal/otel_collector/extension/taskqueue/runners.go`

Responsibilities:
- dequeue one task from the internal queue
- stamp `createAt`
- send the task into the consumer channel
- wait for one of:
  - `ack`
  - `nack`
  - timeout
  - shutdown

This is the goroutine that decides whether to:
- requeue the same task
- advance progress with `AckTimestamp`
- generate the next task window

### G2: metadb receiver worker

File: `pipeline/internal/otel_collector/receiver/metadb/fetcher_by_taskqueue.go`

Responsibilities:
- receive a task from `taskqueue.Subscribe(...)`
- wait until the task end is not in the future
- call `fetchData(...)`
- inject task into context
- call `nextConsumer.ConsumeMetrics(...)`

### G3: processor call stack

File: `pipeline/internal/otel_collector/processor/thanos/thanos.go`

Responsibilities:
- query metrics per cluster and per metric group
- decide whether one task should fail fast
- compute `lastExistTimestamp`
- set `task.AckTimestamp` when any datapoint exists

This work runs on the receiver call stack rather than on a separate taskqueue-specific goroutine.

### G4: exporter worker

File: `pipeline/internal/otel_collector/exporter/otlphttpexporter/factory.go`

Responsibilities:
- send metrics to downstream endpoint
- call `Ack(task)` on success
- call `NackWithRequeue(task)` on failure

The worker itself is managed by OpenTelemetry exporter helper queueing.

### G5: ack/nack sender helper goroutines

File: `pipeline/internal/otel_collector/extension/taskqueue/consumer.go`

Responsibilities:
- asynchronously send one ack or nack event into `ackChan` or `nackChan`

## 3) Thread model overview

```mermaid
flowchart LR
    Q[TaskQueue internal queue]
    G1[G1 runDispatchConsumer]
    G2[G2 metadb fetchTaskFromTaskQueueAndRun]
    G3[G3 thanos doQuery\nsame call stack as G2]
    G4[G4 exporter worker]
    ACK[ackChan or nackChan]
    P[(persist downstream_ts)]

    Q --> G1
    G1 -->|send task| G2
    G2 -->|fetchData ok| G3
    G2 -->|fetchData err -> Nack| ACK

    G3 -->|core metric err -> Nack| ACK
    G3 -->|soft metric err -> log only| G3
    G3 -->|set AckTimestamp| G4
    G3 -->|no datapoint -> Nack| ACK

    G4 -->|send ok -> Ack| ACK
    G4 -->|send err -> Nack| ACK

    ACK --> G1
    G1 -->|Ack: persist and generate next| P
    P --> Q
    G1 -->|Nack before 5m: requeue same task| Q
    G1 -->|Timeout before 5m: requeue same task| Q
    G1 -->|Nack or timeout after 5m: ack endTs and continue| P
```

## 4) State machine for one task

```mermaid
stateDiagram-v2
    [*] --> Queued
    Queued --> Dispatched: dequeue and send to consumer
    Dispatched --> Running: receiver + processor + exporter path

    Running --> Acked: exporter send succeeds
    Running --> Failed: processor nack or exporter nack
    Running --> Failed: receiver fetchData error
    Running --> TimedOut: no ack or nack before timeout

    Failed --> Requeued: retry elapsed < 5m since firstFailedAt
    TimedOut --> Requeued: retry elapsed < 5m since firstFailedAt
    Requeued --> Queued

    Failed --> Bypassed: retry elapsed >= 5m since firstFailedAt
    TimedOut --> Bypassed: retry elapsed >= 5m since firstFailedAt

    Acked --> NextTaskGenerated
    Bypassed --> NextTaskGenerated
    NextTaskGenerated --> Queued
```

## 5) Worked example with timestamps and data

Assumptions:
- `step = 15s`
- task window is `[10:00:00, 10:01:00]`
- one cluster `12345`
- exporter is `otlphttp`

### Example A: normal success path

Metrics found in the window:
- `tidb_cloud.db_total_connection` last datapoint at `10:00:45`
- `tidb_cloud.db_query_per_second` last datapoint at `10:01:00`
- `tidb_cloud.resource_manager_resource_unit_read_request_unit` last datapoint at `10:00:30`

Processor result:
- `lastExistTimestamp = 10:01:00`
- `task.AckTimestamp = 10:01:00`

End-to-end sequence:

```mermaid
sequenceDiagram
    participant G1 as G1 taskqueue
    participant G2 as G2 receiver
    participant G3 as G3 thanos
    participant G4 as G4 exporter

    G1->>G2: Task [10:00:00,10:01:00]
    G2->>G3: ConsumeMetrics(taskCtx)
    G3->>G3: total_connection -> 10:00:45
    G3->>G3: qps -> 10:01:00
    G3->>G3: RU read -> 10:00:30
    G3->>G3: set AckTimestamp = 10:01:00
    G3->>G4: export metrics
    G4-->>G1: Ack(task AckTimestamp=10:01:00)
    G1->>G1: persist downstream_ts=10:01:00
    G1->>G1: generate next task
```

### Example B: empty window retries for 5 minutes, then bypasses

Task window:
- `Timestamp = 10:00:00`
- `EndTimestamp = 10:01:00`

Observed behavior:
- no metric returns any datapoint
- processor `nack`s the task

Timeline:
- first failure at `10:01:05` sets `firstFailedAt = 10:01:05`
- retries continue at `10:02`, `10:03`, `10:04`, `10:05`
- retry at or after `10:06:05` crosses the `5m` bypass duration

Bypass result:
- `AckTimestamp = EndTimestamp.Truncate(Step) = 10:01:00`
- taskqueue persists `downstream_ts = 10:01:00`
- next task is generated

```mermaid
sequenceDiagram
    participant G1 as G1 taskqueue
    participant G2 as G2 receiver
    participant G3 as G3 thanos

    G1->>G2: Task [10:00:00,10:01:00]
    G2->>G3: ConsumeMetrics(taskCtx)
    G3->>G3: no datapoints found
    G3-->>G1: Nack(task)

    G1->>G1: set firstFailedAt = 10:01:05
    G1->>G1: retry elapsed < 5m
    G1->>G1: requeue same task

    G1->>G2: same Task again
    G2->>G3: ConsumeMetrics(taskCtx)
    G3-->>G1: Nack(task)

    G1->>G1: retry elapsed >= 5m
    G1->>G1: AckTimestamp = 10:01:00
    G1->>G1: persist downstream_ts=10:01:00
    G1->>G1: generate next task
```

### Example C: one core metric failure retries the entire window

Task window:
- `Timestamp = 10:00:00`
- `EndTimestamp = 10:01:00`

Observed data:
- `tidb_cloud.db_total_connection` succeeds with last datapoint `10:00:30`
- `tidb_cloud.db_active_connections` succeeds with last datapoint `10:00:45`
- `tidb_cloud.db_kv_request_duration` fails

Outcome:
- processor immediately `nack`s the task
- taskqueue retries the same task window
- already queried partial data in that attempt is not acked
- after 5 minutes since `firstFailedAt`, taskqueue bypasses the window if the failure persists

## 6) Practical reading order

To understand the runtime in code order:

1. `pipeline/internal/otel_collector/extension/taskqueue/runners.go`
2. `pipeline/internal/otel_collector/receiver/metadb/fetcher_by_taskqueue.go`
3. `pipeline/internal/otel_collector/processor/thanos/thanos.go`
4. `pipeline/internal/otel_collector/exporter/otlphttpexporter/factory.go`
5. `pipeline/internal/otel_collector/extension/taskqueue/task_generator.go`
6. `pipeline/internal/otel_collector/extension/taskqueue/timestamp_manage.go`