Cached parent stream with more than one partition corrupts the shared sqlite request cache under concurrent reads (`no such table: responses`, `file is not a database`, `no more rows available`)

Found while certifying `source-pipedrive` (airbytehq/airbyte#85812). Verified against CDK 7.28.3 (the `source-declarative-manifest:7.28.3` base image) and 7.23.6; line numbers are for 7.28.3 unless marked.

## Summary

`ConcurrentDeclarativeSource` forces `use_cache: true` on every parent stream requester. `HttpClient` then backs the requester with a `requests_cache` sqlite database whose file name is derived from the stream name only. When that parent stream has more than one partition (a `ListPartitionRouter`, or several children each holding their own copy of the parent), the worker threads hit the same sqlite database through several `SQLiteCache` instances that do not share a lock, and through one `sqlite3` connection per instance that is used from several threads with `check_same_thread=False`. The result is intermittent sqlite corruption errors that fail the parent stream after it emitted part of its records.

Observed errors:

- `sqlite3.DatabaseError: no such table: responses` and `sqlite3.DatabaseError: file is not a database` (source-pipedrive unit tests on PR airbytehq/airbyte#85812, deterministic once the cached `deals` parent received a two-value `ListPartitionRouter`; master's `mailThreads` parent with four folders flakes the same way).
- `sqlite3.DatabaseError: no more rows available` raised from `requests_cache/backends/sqlite.py:303` (`BEGIN IMMEDIATE` in `_acquire_sqlite_lock`) in the minimal reproduction below, against CDK 7.28.3 and requests-cache 1.3.3.

## Where it comes from (7.28.3)

1. `airbyte_cdk/sources/declarative/concurrent_declarative_source.py:521-536` (`_initialize_cache_for_parent_streams` >> `_set_cache_if_not_disabled` sets `requester["use_cache"] = True` for every parent unless the manifest says `use_cache: false`). 7.23.6: lines 493-508.
2. `airbyte_cdk/sources/streams/http/http_client.py:156-161` `cache_filename` returns `f"{self._name}.sqlite"`, so every `HttpClient` created for a stream called `deals` writes to the same file. 7.23.6: lines 143-148.
3. `airbyte_cdk/sources/streams/http/http_client.py:163-190` `_request_session` builds `requests_cache.SQLiteCache(sqlite_path, fast_save=True, wal=True)` (`PRAGMA synchronous=OFF` + WAL) and wraps it in `CachedLimiterSession`. 7.23.6: lines 150-179.
4. `airbyte_cdk/entrypoint.py:172-179` sets `REQUEST_CACHE_PATH` to one `TemporaryDirectory` for the whole `read`, so in production and in `entrypoint_wrapper.read` based unit tests the cache is a real file, not `file::memory:`.
5. `airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py:3151-3154` `create_parent_stream_config` instantiates the parent stream again for every child (`self._create_component_from_model(model.stream, ...)`). With a parent read as a top-level stream plus three children, four `SQLiteCache` instances open the same `parent.sqlite` (measured by patching `SQLiteCache.__init__`: `{'.../parent.sqlite': 4}`, four distinct `RLock`s).
6. `airbyte_cdk/sources/declarative/stream_slicers/declarative_partition_generator.py:44-50` the `DeclarativePartitionFactory` docstring says one retriever per thread, but `self._retriever = retriever` is a single retriever, so all partitions of one stream instance share one `HttpClient` and one `CachedLimiterSession` across the `num_workers` threads (`concurrent_source.py:63`, default `_LOWEST_SAFE_CONCURRENCY_LEVEL = 2` at `concurrent_declarative_source.py:146`).
7. `requests_cache/backends/sqlite.py:212-218,248-267` (1.3.3): each `SQLiteDict` keeps one `sqlite3.Connection` with `check_same_thread=False`; reads run on that shared connection without the lock ("Read operations can be run in parallel"), writes take `BEGIN IMMEDIATE`. The lock is per instance, so instances 1-4 above never serialize against each other. `http_client.py:77-95` already monkeypatches `SQLiteDict.__getitem__` because of `sqlite3.InterfaceError: bad parameter or other API misuse` from this same sharing.

## Minimal reproduction

Manifest (cached parent with a `ListPartitionRouter`, three children with a `SubstreamPartitionRouter` on it; `use_cache` is not set anywhere, the CDK turns it on):

```yaml
version: 6.48.15
type: DeclarativeSource
check: { type: CheckStream, stream_names: [parent] }
spec: { type: Spec, connection_specification: { type: object, properties: {} } }
concurrency_level: { type: ConcurrencyLevel, default_concurrency: 4, max_concurrency: 4 }
definitions:
parent:
type: DeclarativeStream
name: parent
primary_key: [id]
retriever:
type: SimpleRetriever
requester: { type: HttpRequester, url_base: https://api.example.com/, path: parent, http_method: GET }
record_selector: { type: RecordSelector, extractor: { type: DpathExtractor, field_path: [data] } }
paginator:
type: DefaultPaginator
page_token_option: { type: RequestOption, inject_into: request_parameter, field_name: cursor }
pagination_strategy:
type: CursorPagination
cursor_value: "{{ response.get('next_cursor') }}"
stop_condition: "{{ not response.get('next_cursor') }}"
partition_router:
type: ListPartitionRouter
values: [a, b]
cursor_field: folder
request_option: { type: RequestOption, inject_into: request_parameter, field_name: folder }
incremental_sync:
type: DatetimeBasedCursor
cursor_field: update_time
datetime_format: "%Y-%m-%dT%H:%M:%SZ"
start_datetime: { type: MinMaxDatetime, datetime: "{{ config['start_date'] }}", datetime_format: "%Y-%m-%dT%H:%M:%SZ" }
start_time_option: { type: RequestOption, inject_into: request_parameter, field_name: updated_since }
schema_loader: { type: InlineSchemaLoader, schema: { type: object, properties: { id: { type: integer }, update_time: { type: string } } } }
child: # repeated as child0, child1, child2 with paths parent/{{ stream_partition.pid }}/c0, c1, c2
type: DeclarativeStream
name: child0
primary_key: [id]
retriever:
type: SimpleRetriever
requester: { type: HttpRequester, url_base: https://api.example.com/, path: "parent/{{ stream_partition.pid }}/c0", http_method: GET }
record_selector: { type: RecordSelector, extractor: { type: DpathExtractor, field_path: [data] } }
partition_router:
type: SubstreamPartitionRouter
parent_stream_configs:
- type: ParentStreamConfig
parent_key: id
partition_field: pid
incremental_dependency: true
stream: "#/definitions/parent"
schema_loader: { type: InlineSchemaLoader, schema: { type: object, properties: { id: { type: integer } } } }
streams: [ "#/definitions/parent", "#/definitions/child0", "#/definitions/child1", "#/definitions/child2" ]
```

HttpMocker test outline (`airbyte_cdk.test.mock_http`, `airbyte_cdk.test.entrypoint_wrapper.read`):

```python
def _page(ids, next_cursor=None):
body = {"data": [{"id": i, "update_time": "2024-01-02T00:00:00Z"} for i in ids]}
if next_cursor:
body["next_cursor"] = next_cursor
return HttpResponse(json.dumps(body), 200)

@pytest.mark.parametrize("attempt", range(20))
def test_cached_parent_with_partitions_survives_concurrent_read(attempt):
catalog = (CatalogBuilder().with_stream("parent", SyncMode.incremental)
.with_stream("child0", SyncMode.full_refresh)
.with_stream("child1", SyncMode.full_refresh)
.with_stream("child2", SyncMode.full_refresh).build())
with HttpMocker() as m:
q = {"updated_since": "2024-01-01T00:00:00Z"}
m.get(HttpRequest(BASE + "parent", {**q, "folder": "a"}), _page([1, 2], "p2a"))
m.get(HttpRequest(BASE + "parent", {**q, "folder": "a", "cursor": "p2a"}), _page([3, 4]))
m.get(HttpRequest(BASE + "parent", {**q, "folder": "b"}), _page([5, 6], "p2b"))
m.get(HttpRequest(BASE + "parent", {**q, "folder": "b", "cursor": "p2b"}), _page([7, 8]))
for pid in range(1, 9):
for c in range(3):
m.get(HttpRequest(BASE + f"parent/{pid}/c{c}"), _page([pid * 10 + c]))
source = ConcurrentDeclarativeSource(source_config=MANIFEST, config=CONFIG, catalog=catalog, state=None)
output = read(source, CONFIG, catalog)
assert not output.errors
assert len(output.records_for_stream("parent")) == 8
```

Running the reproduction above (same manifest, 8 reads per process, CDK 7.28.3, requests-cache 1.3.3, Python 3.11.12, macOS) gives failures such as:

```
run 0: {'parent': 6, 'child0': 8, 'child1': 8, 'child2': 8} errors=2
An unexpected error occurred in stream parent: DatabaseError | no more rows available |
File ".../requests_cache/backends/sqlite.py", line 303, in _acquire_sqlite_lock
self._connection.execute('BEGIN IMMEDIATE')
sqlite3.DatabaseError: no more rows available
During the sync, the following streams did not sync successfully: parent: DatabaseError('no more rows available')
```

Frequency in this minimal setup is low (2 failing reads out of about 60 across runs, higher with `max_concurrency: 8` and six partition values). source-pipedrive reproduces it on every run of its unit suite as soon as the cached `deals` parent gets a two-value `ListPartitionRouter`, and `mailThreads` (four folders, parent of `mail`) flakes on master.

## Expected

A parent stream with N partitions read by M worker threads returns all its records and all child streams complete, exactly like the same manifest with `use_cache: false` on the parent requester (which passes every time).

## Actual

The parent stream fails with a sqlite `DatabaseError` after emitting a subset of its records (6 of 8 in the run above); the child streams, which read the parent through their own copies, still complete, so the sync ends with a partially loaded parent table and a failed-stream trace.

## Impact

- Every production sync of a manifest connector runs with `num_workers >= 2` (`_LOWEST_SAFE_CONCURRENCY_LEVEL = 2`, most connectors set `concurrency_level` higher), so any cached parent that yields more than one partition (list router, datetime slices, several children) is exposed. The failure is intermittent, looks like an infrastructure error, and leaves partial parent data.
- Connector authors currently work around it structurally: PR airbytehq/airbyte#85812 had to add `deals_archived` as a separate stream instead of a second partition of `deals`, and documents "a cached parent stream must not get a multi-value partition router" in AGENTS.md.

## Suggested fixes

1. Share one cache backend per sqlite path inside the process. In `HttpClient._request_session`, keep a module-level `dict[str, SQLiteCache]` keyed by `sqlite_path` and reuse the instance; all `HttpClient`s for the same stream name then share the backend's `RLock` and connection, which is what `requests_cache` expects. Smallest change, keeps the cross-child cache hits that motivated caching.
2. Serialize cache access: wrap `CachedLimiterSession.send` (or the backend's `SQLiteDict.connection`) in one lock per sqlite path so reads never interleave with `BEGIN IMMEDIATE` writes on the shared connection. Slower, but the parent request rate is bounded by the API anyway.
3. Per-partition cache files: include a hash of the stream slice (or the `HttpClient` instance id) in `cache_filename`. Avoids the race but also defeats the cache for the child copies, and leaks one sqlite file per partition.
4. Do not enable caching when the parent has a partition router or `incremental_sync` slices (`_initialize_cache_for_parent_streams`): simplest, loses the optimisation only for the cases that break today.

Whatever the fix, please add a regression test along the lines of the outline above (`unit_tests/sources/declarative/`), or a lower-level one in `unit_tests/sources/streams/http/test_http_client.py`: two `HttpClient(name="same", use_cache=True)` instances with `REQUEST_CACHE_PATH` set, eight threads calling `send()` against `requests_mock`, assert no `sqlite3.DatabaseError`.

---

Start by reading airbyte_cdk/sources/streams/http/http_client.py, concurrent_declarative_source.py, and the referenced requests-cache SQLite backend behavior. Run the reproduction or add it under unit_tests/sources/declarative/, then inspect unit_tests/sources/streams/http/test_http_client.py for the lower-level option. Done means concurrent cached parent reads complete without sqlite errors, return all records, and pass the regression test.

Vom Indexierungsmodell aus dem Issue-Text verfasst.