airbytehq / airbytehq/airbyte

[source-slack] Threads stream re-syncs from start_date on every sync once thread count exceeds 10,000 (ThreadsStateMigration incompatible with global-cursor state)

Abierto
#85,049 0 comentarios 0 reacciones 0 asignados Ver en GitHub
autoteam community connectors/source/slack team/use
Lenguaje dominante
Python
Estrellas
22.1k
Forks
5.3k
Métricas de merge de PR
Métricas de PR pendientes

Descripción

## Connector Name

source-slack

## Connector Version

3.2.x (reproduced on current cloud version; the relevant code is unchanged on master as of 2026-08-25)

## What step the error happened?

During the sync

## Description

Once a workspace accumulates more than 10,000 thread partitions, the `threads` stream stops being incremental: every sync re-reads the entire window from `start_date - lookback_window`, re-issuing one `conversations.replies` call per thread parent. For a workspace with tens of thousands of threads this means every scheduled sync is a multi-hour full re-crawl that burns Slack API quota (sustained HTTP 429 backoffs) and re-emits hundreds of thousands of already-synced records. The `channel_messages` stream is unaffected and stays correctly incremental.

## Root cause analysis

Two mechanisms interact:

1. `ThreadsStateMigration` in `airbyte-integrations/connectors/source-slack/components.py` (lines ~163-203) has `should_migrate()` returning `True` unconditionally, so it runs at the start of every sync. `migrate()` computes `max(config start_date, cursors found in the state's top-level "states" list, top-level "float_ts")`, subtracts `lookback_window`, and overwrites `stream_state["parent_state"]` with that value.

2. The CDK's `ConcurrentPerPartitionCursor` (`airbyte-python-cdk/.../concurrent_partition_cursor.py`) switches a stream to a global cursor once it exceeds `SWITCH_TO_GLOBAL_LIMIT = 10_000` partitions. In global-cursor mode the serialized state contains the cursor under `"state"` and no longer contains the per-partition `"states"` key or a top-level `"float_ts"`.

Since the threads stream partitions per thread parent (`parent_key: ts`, `incremental_dependency: true` in manifest.yaml), any active workspace crosses 10,000 partitions quickly. After that, `migrate()` finds none of the keys it looks for and silently falls back to `start_date`, so the parent read-window is reset to `start_date - lookback_window` on every sync. The state itself is persisted correctly (the global cursor value is present and current in the connection state); it is just never read by the migration. Re-emitted records are not filtered against the cursor (the comment in `components.py` notes filtering was omitted for simplicity), so the full window lands in the destination each sync.

Observed state shape mid-sync (sanitized), showing the global cursor present but `parent_state` pinned at `start_date - 14 days`:

```json
{
"use_global_cursor": true,
"state": {"float_ts": 1787670300},
"parent_state": {"channel_messages": {"state": {"float_ts": 1754870400.0}}}
}
```

`1754870400` is exactly the configured `start_date` minus the 14-day `lookback_window`, not an observed cursor.

## Steps to reproduce

1. Configure source-slack against a workspace/channel set containing more than 10,000 threads (thread parents with replies), incremental sync mode for `channel_messages` and `threads`.
2. Run an initial sync; let it complete and checkpoint state.
3. Run a second sync with no new data.
4. Expected: `threads` reads only from its cursor (near-zero new records). Actual: `threads` re-reads from `start_date - lookback_window` (full history re-emitted), while `channel_messages` correctly reads only from its per-channel cursors.

## Suggested fix

`ThreadsStateMigration.migrate()` should also read the global-cursor state shape (`stream_state["state"]["float_ts"]` when `use_global_cursor` is set) when computing the parent window, or `should_migrate()` should return `False` when the state is already in the current format. Happy to open a PR implementing this.

## Impact

Above the 10k-partition threshold the threads stream is effectively full-refresh with incremental cost characteristics hidden from the user: sync duration and API usage scale with total history rather than new data, and rate limiting (429s on `conversations.replies`) throttles every sync.

Guía de contribución

Abrir la guía de contribución

Evaluación

Este issue todavía no se ha evaluado.

Recibe los nuevos issues en tu correo

Un resumen breve de issues de GitHub para principiantes.