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`)

[bazarnov]

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 feat(source-pipedrive)!: 🚨🚨 migrate core streams to API v2, add primary keys and typed dates (3.0.0) 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 RLocks).
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):

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):

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 feat(source-pipedrive)!: 🚨🚨 migrate core streams to API v2, add primary keys and typed dates (3.0.0) 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 HttpClients 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.

---