RequestContextHTTPClient + Server implementation in local mode (#457)

RequestContextHTTPClient is a picklable RequestContext implementation
that provides
all RequestContext functionality in any Python thread and subprocess.

It exchanges small http requests with an HTTP Server which implements:
presigning of read/write blobs, commiting multipart uploads to blob
store,
adding request metrics, sending function progress updates. The BLOBs
are not sent to the Server. BLOB store client is directly invoked
by RequestContextHTTPClient.

User code can use RequestContextHTTPClient in a thread or subprocess by
passing the result of RequestContext.get()
as an argument when creating a Thread or Process. This usage scenario is
demonstrated
in the added tests.

When user passes RequestContextHTTPClient into a Thread the context is
passed by reference.
RequestContextHTTPClient is thread safe so this is okay. When user
passes RequestContextHTTPClient
into a Process, Python runtime pickles RequestContextHTTPClient and then
unpickles its clone
in the subprocess so the subprocess uses a clone of the original
RequestContextHTTPClient.
RequestContextHTTPClient has all the data required to reestablish http
connection to the
HTTP Server and tell the Server which allocation and request the http
requests are for.

This PR adds implementation of the HTTP Server in local mode.
Remote (FE) side HTTP Server implementation will be done in a
separate PR. New integration tests are covering all RequestContext
functionaility working correctly when accessed from a user function,
a thread and a subprocess.

Each LocalRunner (one per request) uses its own HTTP Server. Each
HTTP Server uses a unique ephemeral port where the LocalRunner
request's RequestContextHTTPClient connects to. So multiple local
runners
can exist at the same time without any problems.



<!-- CURSOR_SUMMARY -->
---

> [!NOTE]
> Introduces a picklable RequestContext HTTP client/server for local
mode with thread/process support, refactors BLOB store and internal
logging, and updates FE/local runner and tests accordingly.
> 
> - **Request Context (Local mode)**:
> - Add `RequestContextHTTPClient` (picklable, thread-safe) and
lightweight local HTTP server/router with handlers for `progress`,
`metrics`, and request state (`prepare_read`, `prepare_write`,
`commit_write`).
> - Local handlers print progress/metrics and serve file-backed BLOBs
for request state.
> - **BLOB Store**:
> - Introduce SDK `BLOB`/`BLOBChunk` pydantic models and `BLOBStore`
with local FS and S3 backends; parallel chunked `get/put` with
`InternalLogger`.
> - Remove old local blob store; add local `SerializedValueStore`
wrapper.
> - **Logging**:
> - Replace `FunctionExecutorLogger` with picklable `InternalLogger`;
unify usage across SDK/FE.
> - **Local Runner**:
> - Use per-request local HTTP server + `RequestContextHTTPClient`, new
value store, cleanup lifecycle; set multiprocessing start method via
`setup_multiprocessing`.
> - **Function Executor**:
> - Migrate to new `InternalLogger`/`BLOBStore`; add blob proto↔SDK
converters; validate progress attributes; minor proto doc tweak.
> - **Runtime hooks & multiprocessing**:
> - Add `_raise_multiprocessing_usage_error`; initialize spawn method;
improved SDK usage errors outside function threads.
> - **Cloud events**:
> - Move CloudEvents helpers under `applications` and update imports;
clarify JSON serialization errors.
> - **Tests**:
> - Add/adjust tests for blob store (local/S3), request context
(metrics/progress/state/request_id) in function, thread, and subprocess
scenarios; update benchmarks.
> - **Version**: bump to `0.2.89`.
> 
> <sup>Written by [Cursor
Bugbot](https://cursor.com/dashboard?tab=bugbot) for commit
e487a23. This will update automatically
on new commits. Configure
[here](https://cursor.com/dashboard?tab=bugbot).</sup>
<!-- /CURSOR_SUMMARY -->

Author: eabatalov

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "tensorlake"
-version = "0.2.88"
+version = "0.2.89"
 description = "Tensorlake SDK for Document Ingestion API and Serverless Applications"
 readme = "README.md"
 authors = ["Tensorlake Inc. <[email protected]>"]

src/tensorlake/applications/blob_store/__init__.py

@@ -0,0 +1,4 @@
+from .blob import BLOB, BLOBChunk
+from .blob_store import BLOBStore
+
+__all__ = ["BLOB", "BLOBChunk", "BLOBStore"]

src/tensorlake/applications/blob_store/blob.py

@@ -0,0 +1,22 @@
+from pydantic import BaseModel
+
+
+class BLOBChunk(BaseModel):
+    # URI of the chunk.
+    # S3 URI if the data is stored in S3.
+    # Starts with "file://"" prefix if the data is stored on a local file system.
+    uri: str
+    # Actual size of chunk data if the BLOB has data.
+    # Requested chunk size if the BLOB has no data yet (can be used for data upload).
+    size: int
+    # None if the BLOB has data.
+    # ETag of the chunk data (generated by BLOB store) if the BLOB chunk was used to upload data.
+    etag: str | None
+
+
+class BLOB(BaseModel):
+    # ID of the BLOB
+    id: str
+    # Ordered chunks of the BLOB if the BLOB has data.
+    # Ordered chunks of the BLOB that can be used to upload data if the BLOB has no data yet.
+    chunks: list[BLOBChunk]

…nction_executor/blob_store/blob_store.py …ke/applications/blob_store/blob_store.pysrc/tensorlake/function_executor/blob_store/blob_store.py renamed to src/tensorlake/applications/blob_store/blob_store.py src/tensorlake/function_executor/blob_store/blob_store.py renamed to src/tensorlake/applications/blob_store/blob_store.py

@@ -1,11 +1,11 @@
 from concurrent.futures import FIRST_EXCEPTION, Future, ThreadPoolExecutor, wait
 from dataclasses import dataclass
-from typing import List
+from typing import Any
 
-from tensorlake.applications import InternalError
+from tensorlake.applications.interface.exceptions import InternalError
+from tensorlake.applications.internal_logger import InternalLogger
 
-from ..logger import FunctionExecutorLogger
-from ..proto.function_executor_pb2 import BLOB, BLOBChunk
+from .blob import BLOB, BLOBChunk
 from .local_fs_blob_store import LocalFSBLOBStore
 from .s3_blob_store import S3BLOBStore
 
@@ -27,10 +27,13 @@ class _ChunkInfo:
 class BLOBStore:
     """Dispatches generic BLOB store calls to their real backends.
 
-    Implements chunking."""
+    Implements chunking. Thread-safe. Picklable.
+    """
 
-    def __init__(self, available_cpu_count: int, logger: FunctionExecutorLogger):
+    def __init__(self, available_cpu_count: int):
         """Creates a BLOB store that uses the supplied BLOB stores."""
+        self._available_cpu_count: int = available_cpu_count
+
         max_io_workers: int = min(
             available_cpu_count * _IO_WORKER_THREADS_PER_AVAILABLE_CPU,
             _MAX_WORKER_THREADS,
@@ -40,15 +43,20 @@ def __init__(self, available_cpu_count: int, logger: FunctionExecutorLogger):
         )
         self._local: LocalFSBLOBStore = LocalFSBLOBStore()
         self._s3: S3BLOBStore = S3BLOBStore(io_workers_count=max_io_workers)
-        logger.info(
-            "BLOBStore initialized",
-            available_cpu_count=available_cpu_count,
-            max_io_workers=max_io_workers,
+
+    def __getstate__(self):
+        """Get the state for pickling."""
+        return {
+            "available_cpu_count": self._available_cpu_count,
+        }
+
+    def __setstate__(self, state: dict[str, Any]):
+        """Set the state for unpickling."""
+        self.__init__(
+            available_cpu_count=state["available_cpu_count"],
         )
 
-    def get(
-        self, blob: BLOB, offset: int, size: int, logger: FunctionExecutorLogger
-    ) -> bytes:
+    def get(self, blob: BLOB, offset: int, size: int, logger: InternalLogger) -> bytes:
         """Returns binary data stored in BLOB with the supplied URI at the supplied offset.
 
         Raises InternalError on error.
@@ -60,7 +68,7 @@ def get(
 
         # Read data from BLOB chunks in parallel until all data is read.
         # Minimize data copying by not creating any intermediate bytes/bytearray objects.
-        read_chunk_futures: List[Future] = []
+        read_chunk_futures: list[Future] = []
         destination: bytearray = bytearray(size)
         destination_view: memoryview = memoryview(destination)
         read_offset: int = offset
@@ -110,7 +118,7 @@ def _read_into(
         blob_uri: str,
         blob_read_offset: int,
         destination: memoryview,
-        logger: FunctionExecutorLogger,
+        logger: InternalLogger,
     ) -> bytes:
         if _is_file_uri(blob_uri):
             self._local.get(
@@ -127,9 +135,7 @@ def _read_into(
                 logger=logger,
             )
 
-    def put(
-        self, blob: BLOB, data: List[bytes], logger: FunctionExecutorLogger
-    ) -> BLOB:
+    def put(self, blob: BLOB, data: list[bytes], logger: InternalLogger) -> BLOB:
         """Stores the supplied binary data into the supplied BLOB starting from its very beginning.
 
         Overwrites BLOB. Raises Exception on error.
@@ -150,14 +156,14 @@ def put(
         # Write data to BLOB chunks in parallel until all data is written.
         # Minimize data copying by not creating any intermediate bytes/bytearray objects.
         data_read_offset: int = 0
-        write_chunk_futures: List[Future] = []
-        uploaded_chunk_sizes: List[int] = []
+        write_chunk_futures: list[Future] = []
+        uploaded_chunk_sizes: list[int] = []
 
         data_ix: int = 0
         read_offset_inside_data: int = 0
         for chunk in blob.chunks:
             chunk: BLOBChunk
-            chunk_data: List[memoryview] = []
+            chunk_data: list[memoryview] = []
             chunk_data_size: int = 0
             chunk_offset: int = data_read_offset
             if data_ix == len(data):
@@ -198,6 +204,7 @@ def put(
         wait(write_chunk_futures, return_when=FIRST_EXCEPTION)
         uploaded_blob: BLOB = BLOB(
             id=blob.id,
+            chunks=[],
         )
         for ix, future in enumerate(write_chunk_futures):
             if future.exception() is not None:
@@ -206,8 +213,7 @@ def put(
                 ) from future.exception()
             # The futures list is ordered by the chunk index, so appending here preserves
             # the original chunks order.
-            uploaded_chunk: BLOBChunk = BLOBChunk()
-            uploaded_chunk.CopyFrom(blob.chunks[ix])
+            uploaded_chunk: BLOBChunk = blob.chunks[ix].model_copy()
             uploaded_chunk.size = uploaded_chunk_sizes[ix]
             uploaded_chunk.etag = future.result()
             uploaded_blob.chunks.append(uploaded_chunk)
@@ -218,8 +224,8 @@ def _write_chunk(
         self,
         chunk_uri: str,
         chunk_offset: int,
-        source: List[memoryview],
-        logger: FunctionExecutorLogger,
+        source: list[memoryview],
+        logger: InternalLogger,
     ) -> str:
         if _is_file_uri(chunk_uri):
             return self._local.put(

…ecutor/blob_store/local_fs_blob_store.py …ations/blob_store/local_fs_blob_store.pysrc/tensorlake/function_executor/blob_store/local_fs_blob_store.py renamed to src/tensorlake/applications/blob_store/local_fs_blob_store.py src/tensorlake/function_executor/blob_store/local_fs_blob_store.py renamed to src/tensorlake/applications/blob_store/local_fs_blob_store.py

@@ -3,9 +3,8 @@
 import os.path
 from typing import List
 
-from tensorlake.applications import InternalError
-
-from ..logger import FunctionExecutorLogger
+from tensorlake.applications.interface.exceptions import InternalError
+from tensorlake.applications.internal_logger import InternalLogger
 
 
 class LocalFSBLOBStore:
@@ -16,7 +15,7 @@ def get(
         uri: str,
         offset: int,
         destination: memoryview,
-        logger: FunctionExecutorLogger,
+        logger: InternalLogger,
     ) -> None:
         """Reads binary data stored in file at the supplied URI and offset into the destination memoryview.
 
@@ -40,7 +39,7 @@ def put(
         uri: str,
         offset: int,
         source: List[memoryview],
-        logger: FunctionExecutorLogger,
+        logger: InternalLogger,
     ) -> str:
         """Stores the supplied memoryviews of binary data in a file at the supplied URI and offset.
 

…ion_executor/blob_store/s3_blob_store.py …applications/blob_store/s3_blob_store.pysrc/tensorlake/function_executor/blob_store/s3_blob_store.py renamed to src/tensorlake/applications/blob_store/s3_blob_store.py src/tensorlake/function_executor/blob_store/s3_blob_store.py renamed to src/tensorlake/applications/blob_store/s3_blob_store.py

@@ -2,11 +2,10 @@
 
 import httpx
 
-from tensorlake.applications import InternalError
+from tensorlake.applications.interface.exceptions import InternalError
+from tensorlake.applications.internal_logger import InternalLogger
 from tensorlake.utils.retries import exponential_backoff
 
-from ..logger import FunctionExecutorLogger
-
 # Customers can upload and download large files, allow up to 1 hour per S3 operation.
 _S3_OPERATION_TIMEOUT_SEC = 1 * 60 * 60  # 1 hour
 # Keep established connections around for up to 1 hour to maximize download/upload throughput
@@ -35,7 +34,7 @@ def get(
         uri: str,
         offset: int,
         destination: memoryview,
-        logger: FunctionExecutorLogger,
+        logger: InternalLogger,
     ) -> bytes:
         """Reads binary data stored in S3 object at the supplied URI and offset into the destination memoryview.
 
@@ -57,10 +56,10 @@ def on_retry(
                     # because we're in streaming mode.
                     e.response.read()
                     response = e.response.text
-            except Exception as e:
+            except Exception as extract_response_exception:
                 logger.error(
                     "failed to extract status code and response from failed S3 get response",
-                    exc_info=e,
+                    exc_info=extract_response_exception,
                 )
 
             # The URI can be presigned, it should not be logged as it provides access to customer data.
@@ -105,10 +104,23 @@ def get_with_retries() -> bytes:
         except httpx.HTTPStatusError as e:
             # The URI can be presigned, it should not be logged as it provides access to customer data.
             # Response text doesn't contain the URI, so it can be logged.
+            status_code: str = "None"
+            response: str = "None"
+            try:
+                status_code = e.response.status_code
+                # .read() is required before accessing .text
+                # because we're in streaming mode.
+                e.response.read()
+                response = e.response.text
+            except Exception as extract_response_exception:
+                logger.error(
+                    "failed to extract status code and response from failed S3 get response",
+                    exc_info=extract_response_exception,
+                )
             logger.error(
                 "failed to get S3 object",
-                status_code=e.response.status_code,
-                response=e.response.text,
+                status_code=status_code,
+                response=response,
                 offset=offset,
                 size=len(destination),
             )
@@ -129,7 +141,7 @@ def put(
         self,
         uri: str,
         source: List[memoryview],
-        logger: FunctionExecutorLogger,
+        logger: InternalLogger,
     ) -> str:
         """Stores the supplied memoryviews in a S3 object at the supplied URI.
 
@@ -152,10 +164,10 @@ def on_retry(
                 if isinstance(e, httpx.HTTPStatusError):
                     status_code = e.response.status_code
                     response = e.response.text
-            except Exception as e:
+            except Exception as extract_response_exception:
                 logger.error(
                     "failed to extract status code and response from failed S3 put response",
-                    exc_info=e,
+                    exc_info=extract_response_exception,
                 )
 
             logger.error(

…orlake/function_executor/cloud_events.py …/tensorlake/applications/cloud_events.pysrc/tensorlake/function_executor/cloud_events.py renamed to src/tensorlake/applications/cloud_events.py src/tensorlake/function_executor/cloud_events.py renamed to src/tensorlake/applications/cloud_events.py

@@ -3,7 +3,7 @@
 import uuid
 from typing import Any
 
-from tensorlake.applications import SerializationError
+from .interface.exceptions import SerializationError
 
 
 def print_cloud_event(
@@ -14,6 +14,8 @@ def print_cloud_event(
 ) -> None:
     """
     Takes a dictionary representing an event produced by the executor, wraps it in a CloudEvent and prints it to stdout.
+
+    Raises SerializationError: If the event cannot be serialized to JSON.
     """
     print(_serialize_json(new_cloud_event(event, type, source, message)), flush=True)
 
@@ -56,7 +58,7 @@ def _serialize_json(obj: dict[str, Any]) -> str:
         obj: The dictionary to serialize
 
     Returns:
-        A JSON-serializable version of the object as a string
+        A version of the object serialized into a JSON string
 
     Raises:
         SerializationError: If the object cannot be serialized to JSON

src/tensorlake/applications/interface/request_context.py

@@ -32,14 +32,14 @@ def get(self, key: str, default: Any | None = None) -> Any | None:
 class RequestMetrics:
     """Abstract interface for reporting application request metrics."""
 
-    def timer(self, name: str, value: float):
+    def timer(self, name: str, value: float) -> None:
         """Records a duration metric with the supplied name and value.
 
         Raises TensorlakeError on error.
         """
         raise InternalError("RequestMetrics subclasses must implement timer method.")
 
-    def counter(self, name: str, value: int = 1):
+    def counter(self, name: str, value: int = 1) -> None:
         """Adds the supplied value to the counter with the supplied name.
 
         If the counter does not exist, it is created with the supplied value.
@@ -68,7 +68,8 @@ def update(
             attributes: A dictionary of key/value string pairs to pass to the progress update
 
         Raises:
-            SerializationError: If attributes cannot be serialized to JSON.
+            SDKUsageError: If attributes is not a dictionary of string key/value pairs or None.
+            TensorlakeError: On other errors.
         """
         raise InternalError("FunctionProgress subclasses must implement update method.")