fix(http): make the sqlite request cache thread-safe

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,

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"