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..7a91da3af5 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(r"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"