From 392b4f0efc5932153fd98998d152ab2735f23e45 Mon Sep 17 00:00:00 2001 From: Baz <22987674+bazarnov@users.noreply.github.com> Date: Mon, 5 Oct 2026 21:23:25 +0300 Subject: [PATCH 1/2] fix(http): make the sqlite request cache thread-safe requests_cache shares one sqlite3 connection between threads and locks only writes. All partitions of a stream share one HttpClient, so cache reads raced other threads on that connection and failed with EOFError or "no more rows available", or crashed the interpreter on SQLite builds without serialized mode. Without REQUEST_CACHE_PATH every cache also shares one in-memory database, where concurrent connections fail with "database table is locked". - Hold the backend lock while reading in the existing SQLiteDict.__getitem__ monkeypatch and deserialize after releasing it. - Give every cache backend HttpClient creates one process-wide RLock. Fixes #1150 --- .../sources/streams/http/http_client.py | 28 +++++--- .../sources/streams/http/test_http_client.py | 68 +++++++++++++++++++ 2 files changed, 88 insertions(+), 8 deletions(-) diff --git a/airbyte_cdk/sources/streams/http/http_client.py b/airbyte_cdk/sources/streams/http/http_client.py index 12f792c47a..50d03039cc 100644 --- a/airbyte_cdk/sources/streams/http/http_client.py +++ b/airbyte_cdk/sources/streams/http/http_client.py @@ -4,6 +4,7 @@ import logging import os +import threading import urllib from pathlib import Path from typing import Any, Callable, Dict, List, Mapping, Optional, Tuple, Union @@ -82,24 +83,32 @@ def monkey_patched_get_item(self, key): # type: ignore # this interface is a copy/paste from the requests_cache lib """ - con.execute can lead to `sqlite3.InterfaceError: bad parameter or other API misuse`. There was a fix implemented - [here](https://github.com/requests-cache/requests-cache/commit/5ca6b9cdcb2797dd2fed485872110ccd72aee55d#diff-f43db4a5edf931647c32dec28ea7557aae4cae8444af4b26c8ecbe88d8c925aaL330-R332) - but there is still no official releases of requests_cache that this is part of. Hence, we will monkeypatch it for now. + requests_cache shares one sqlite3 connection between threads but only takes `self._lock` for + writes. A read that races another thread on that connection can return an empty value + (`EOFError: Ran out of input`), raise that thread's result code (`sqlite3.DatabaseError: no more + rows available`) or, when SQLite is not built in serialized mode, crash the interpreter. Reads + take the lock too, until requests_cache locks them itself. """ - with self.connection() as con: + with self._lock, self.connection() as con: # Using placeholders here with python 3.12+ and concurrency results in the error: # sqlite3.InterfaceError: bad parameter or other API misuse cur = con.execute(f"SELECT value FROM {self.table_name} WHERE key='{key}'") row = cur.fetchone() cur.close() - if not row: - raise KeyError(key) + if not row: + raise KeyError(key) - return self.deserialize(key, row[0]) + return self.deserialize(key, row[0]) requests_cache.SQLiteDict.__getitem__ = monkey_patched_get_item # type: ignore # see the method doc for more information +# Every cache backend HttpClient creates uses this lock. Without REQUEST_CACHE_PATH they all share +# one in-memory database, where a connection using a table another one is writing fails with +# `database table is locked` instead of waiting. +# ponytail: one lock per process, use one lock per sqlite path if contention ever shows +_SQLITE_CACHE_LOCK = threading.RLock() + def _as_auxiliary_request_log( log_message: Any, title: Optional[str] = None, description: Optional[str] = None @@ -213,7 +222,10 @@ def _request_session(self) -> requests.Session: # * https://github.com/requests-cache/requests-cache/commit/7fa89ffda300331c37d8fad7f773348a3b5b0236#diff-f43db4a5edf931647c32dec28ea7557aae4cae8444af4b26c8ecbe88d8c925aaR238 # * https://github.com/requests-cache/requests-cache/commit/7fa89ffda300331c37d8fad7f773348a3b5b0236#diff-2e7f95b7d7be270ff1a8118f817ea3e6663cdad273592e536a116c24e6d23c18R164-R168 # * `If the application running SQLite crashes, the data will be safe, but the database [might become corrupted](https://www.sqlite.org/howtocorrupt.html#cfgerr) if the operating system crashes or the computer loses power before that data has been written to the disk surface.` in [this description](https://www.sqlite.org/pragma.html#pragma_synchronous). - backend = requests_cache.SQLiteCache(sqlite_path, fast_save=True, wal=True) + # The backend creates its tables when built, so build it under the lock too + with _SQLITE_CACHE_LOCK: + backend = requests_cache.SQLiteCache(sqlite_path, fast_save=True, wal=True) + backend.responses._lock = backend.redirects._lock = _SQLITE_CACHE_LOCK return CachedLimiterSession( cache_name=sqlite_path, backend=backend, diff --git a/unit_tests/sources/streams/http/test_http_client.py b/unit_tests/sources/streams/http/test_http_client.py index ad3ee5acd1..518cf5638b 100644 --- a/unit_tests/sources/streams/http/test_http_client.py +++ b/unit_tests/sources/streams/http/test_http_client.py @@ -3,6 +3,8 @@ import json import logging import os +import re +import threading import time from datetime import timedelta from unittest.mock import MagicMock, patch @@ -41,6 +43,7 @@ Oauth2Authenticator, TokenAuthenticator, ) +from airbyte_cdk.utils.constants import ENV_REQUEST_CACHE_PATH from airbyte_cdk.utils.datetime_helpers import ab_datetime_now from airbyte_cdk.utils.traced_exception import AirbyteTracedException @@ -763,6 +766,71 @@ def test_given_different_headers_then_response_is_not_cached(requests_mock): assert second_response.json()["test"] == "second response" +def test_cache_read_holds_the_lock_shared_by_every_cache(tmp_path, monkeypatch): + monkeypatch.setenv(ENV_REQUEST_CACHE_PATH, str(tmp_path)) + first, second = ( + HttpClient(name=name, logger=MagicMock(), use_cache=True)._session.cache + for name in ("first", "second") + ) + assert first.responses._lock is second.responses._lock is second.redirects._lock + + responses = second.responses + connection = responses._connection + lock_owned_during_read = [] + + class SpyCursor: + def __init__(self, cursor): + self.cursor = cursor + + def fetchone(self): + lock_owned_during_read.append(responses._lock._is_owned()) + return self.cursor.fetchone() + + def close(self): + lock_owned_during_read.append(responses._lock._is_owned()) + self.cursor.close() + + class SpyConnection: + def execute(self, *args): + lock_owned_during_read.append(responses._lock._is_owned()) + return SpyCursor(connection.execute(*args)) + + with patch.object(responses, "_connection", SpyConnection()): + assert responses.get("missing_key") is None + + assert lock_owned_during_read == [True, True, True] + + +@pytest.mark.parametrize("in_memory", [False, True], ids=["file", "in_memory"]) +def test_concurrent_cached_sends_do_not_fail(requests_mock, tmp_path, monkeypatch, in_memory): + if in_memory: + monkeypatch.delenv(ENV_REQUEST_CACHE_PATH, raising=False) + else: + monkeypatch.setenv(ENV_REQUEST_CACHE_PATH, str(tmp_path)) + requests_mock.get(re.compile("https://api.example.com/.*"), json={"data": [1]}) + clients = [HttpClient(name="same", logger=MagicMock(), use_cache=True) for _ in range(2)] + errors = [] + + def send(thread_id: int) -> None: + for i in range(50): + # even requests cache a url of their own, odd ones read urls every thread caches + path = str(i % 5) if i % 2 else f"{thread_id}/{i}" + try: + clients[thread_id % 2].send_request( + "GET", f"https://api.example.com/{path}", request_kwargs={} + ) + except Exception as exception: + errors.append(exception) + + threads = [threading.Thread(target=send, args=(thread_id,)) for thread_id in range(8)] + for thread in threads: + thread.start() + for thread in threads: + thread.join() + + assert errors == [] + + def test_given_noproxy_for_another_url_when_send_request_then_do_not_break(requests_mock): http_client = HttpClient(name="test", logger=MagicMock(), use_cache=True) os.environ["no_proxy"] = "another.com" From 2e052267ac9826b607990c4038a27b50d5c5bb36 Mon Sep 17 00:00:00 2001 From: Baz <22987674+bazarnov@users.noreply.github.com> Date: Tue, 6 Oct 2026 11:02:35 +0300 Subject: [PATCH 2/2] test: escape the dots of the mocked host regex --- unit_tests/sources/streams/http/test_http_client.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/unit_tests/sources/streams/http/test_http_client.py b/unit_tests/sources/streams/http/test_http_client.py index 518cf5638b..7a91da3af5 100644 --- a/unit_tests/sources/streams/http/test_http_client.py +++ b/unit_tests/sources/streams/http/test_http_client.py @@ -807,7 +807,7 @@ def test_concurrent_cached_sends_do_not_fail(requests_mock, tmp_path, monkeypatc monkeypatch.delenv(ENV_REQUEST_CACHE_PATH, raising=False) else: monkeypatch.setenv(ENV_REQUEST_CACHE_PATH, str(tmp_path)) - requests_mock.get(re.compile("https://api.example.com/.*"), json={"data": [1]}) + requests_mock.get(re.compile(r"https://api\.example\.com/.*"), json={"data": [1]}) clients = [HttpClient(name="same", logger=MagicMock(), use_cache=True) for _ in range(2)] errors = []