[source-slack] Threads stream re-syncs from start_date on every sync once thread count exceeds 10,000 (ThreadsStateMigration incompatible with global-cursor state)
- 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
Evaluación
Este issue todavía no se ha evaluado.