From 7aaf3f9225b2af3089acf97e29230169e40d89c9 Mon Sep 17 00:00:00 2001 From: Tarcio Date: Wed, 2 Sep 2026 15:02:41 -0300 Subject: [PATCH 01/10] chore(deps): require starlette 1.5 for content-type-aware gzip The gzip middleware gained exclude_content_types and worker-thread offload in 1.5; the resolved pin was 1.3.1. fastapi 0.139.2 requires only starlette>=0.46.0, so nothing caps the bump. Locked with --upgrade-package so no other dependency moves in this change. --- src/backend/base/pyproject.toml | 1 + uv.lock | 8 +++++--- 2 files changed, 6 insertions(+), 3 deletions(-) diff --git a/src/backend/base/pyproject.toml b/src/backend/base/pyproject.toml index 82d0ff319897..09e7da1a9c6b 100644 --- a/src/backend/base/pyproject.toml +++ b/src/backend/base/pyproject.toml @@ -19,6 +19,7 @@ maintainers = [ dependencies = [ "lfx~=1.13.0", "fastapi>=0.139.0,<1.0.0", + "starlette>=1.5.0", "slowapi>=0.1.9,<1.0.0", "httpx[http2]>=0.27,<1.0.0", "h2>=4.4.1", diff --git a/uv.lock b/uv.lock index 7e8cf8dc3cfa..88165e3496b0 100644 --- a/uv.lock +++ b/uv.lock @@ -8193,6 +8193,7 @@ dependencies = [ { name = "spider-client" }, { name = "sqlalchemy", extra = ["aiosqlite"] }, { name = "sqlmodel" }, + { name = "starlette" }, { name = "structlog" }, { name = "traceloop-sdk" }, { name = "transformers" }, @@ -8793,6 +8794,7 @@ requires-dist = [ { name = "sqlalchemy", extras = ["postgresql-psycopg2binary"], marker = "extra == 'postgresql'", specifier = ">=2.0.38,<3.0.0" }, { name = "sqlmodel", specifier = "~=0.0.37" }, { name = "sseclient-py", marker = "extra == 'sseclient'", specifier = "==1.8.0" }, + { name = "starlette", specifier = ">=1.5.0" }, { name = "structlog", specifier = ">=25.4.0,<26.0.0" }, { name = "supabase", marker = "extra == 'supabase'", specifier = ">=2.6.0,<3.0.0" }, { name = "tiktoken", marker = "extra == 'docling-chunking'", specifier = ">=0.7.0" }, @@ -19040,15 +19042,15 @@ wheels = [ [[package]] name = "starlette" -version = "1.3.1" +version = "1.6.0" source = { registry = "https://pypi.org/simple" } dependencies = [ { name = "anyio" }, { name = "typing-extensions", marker = "python_full_version < '3.13'" }, ] -sdist = { url = "https://files.pythonhosted.org/packages/eb/e3/7c1dc7381d9f8ab7d854328ebfa884e62cb3f3d8549ddfd37c7814f42afa/starlette-1.3.1.tar.gz", hash = "sha256:05d0213193f2fbaae60e2ecb593b4add4262ad4e46536b54abe36f11a71724e0", size = 2703240, upload-time = "2026-06-12T09:23:11.602Z" } +sdist = { url = "https://files.pythonhosted.org/packages/b5/b4/205b0d5241d934e8add0c38aa924c4f9fb7330834ff11e5444db964ec3f9/starlette-1.6.0.tar.gz", hash = "sha256:d4e3ac5e546444960c710297a3c9fc3f7ebae1b7e963f3d36173b49da535be9b", size = 2716969, upload-time = "2026-08-08T18:27:57.512Z" } wheels = [ - { url = "https://files.pythonhosted.org/packages/ec/bb/2799cc2ede3ed41131f8975621e7213dfc7ef4acbbaadfa440f32500c370/starlette-1.3.1-py3-none-any.whl", hash = "sha256:c7372aae11c3c3f26a42df7bd626cec2f47d03483d261d369516a615a53714c6", size = 73632, upload-time = "2026-06-12T09:23:10.017Z" }, + { url = "https://files.pythonhosted.org/packages/c8/cb/6a6a47d5b464bd08695d254f3da6e7986cc70c9fa5d778eda57538edfe56/starlette-1.6.0-py3-none-any.whl", hash = "sha256:a86dd39d14bb45f85a3d18525215a9ef0cfd1f192ac793220e72598c90335f0c", size = 75969, upload-time = "2026-08-08T18:27:56.196Z" }, ] [[package]] From 786e13170d87e7cb9754d2866e729564f8b2156a Mon Sep 17 00:00:00 2001 From: Tarcio Date: Wed, 2 Sep 2026 15:06:54 -0300 Subject: [PATCH 02/10] perf(api): negotiate gzip for every API response GET /flows/{id} and the PATCH echo carried the whole graph uncompressed; on the 27 starter projects that is 4,514 KB of payload, 91.9% of it node templates. Level 6 takes it to 1,034 KB for 3.3 ms per flow, against 7.6 ms at the library default of 9 for the same 77%. Registered innermost, before ContentSizeLimitMiddleware: the BaseHTTPMiddleware layers above it turn every response into a stream, and a streamed response has no Content-Length for minimum_size to test, so a gzip registered outside them compresses 200-byte replies too. --- src/backend/base/langflow/main.py | 10 +++ .../unit/api/test_response_compression.py | 86 +++++++++++++++++++ 2 files changed, 96 insertions(+) create mode 100644 src/backend/tests/unit/api/test_response_compression.py diff --git a/src/backend/base/langflow/main.py b/src/backend/base/langflow/main.py index 00e00cf51361..7f80aae73f39 100644 --- a/src/backend/base/langflow/main.py +++ b/src/backend/base/langflow/main.py @@ -32,6 +32,7 @@ from pydantic import PydanticDeprecatedSince20 from pydantic_core import PydanticSerializationError from starlette.middleware.base import BaseHTTPMiddleware, RequestResponseEndpoint +from starlette.middleware.gzip import DEFAULT_EXCLUDED_CONTENT_TYPES, GZipMiddleware from langflow.api import log_router from langflow.api.health_check_router import health_check_router @@ -77,6 +78,9 @@ _tasks: list[asyncio.Task] = [] MAX_PORT = 65535 +GZIP_MINIMUM_SIZE = 1000 +GZIP_COMPRESS_LEVEL = 6 +GZIP_EXCLUDED_CONTENT_TYPES = (*DEFAULT_EXCLUDED_CONTENT_TYPES, "application/octet-stream") # Enterprise lifespan hook registry. Enterprise plugins append async callables # at app-construction time (plugin registration runs before the lifespan @@ -861,6 +865,12 @@ def create_app(): lifespan=lifespan, root_path=settings.root_path, ) + app.add_middleware( + GZipMiddleware, + minimum_size=GZIP_MINIMUM_SIZE, + compresslevel=GZIP_COMPRESS_LEVEL, + exclude_content_types=GZIP_EXCLUDED_CONTENT_TYPES, + ) app.add_middleware( ContentSizeLimitMiddleware, ) diff --git a/src/backend/tests/unit/api/test_response_compression.py b/src/backend/tests/unit/api/test_response_compression.py new file mode 100644 index 000000000000..9cff9ba9b4cb --- /dev/null +++ b/src/backend/tests/unit/api/test_response_compression.py @@ -0,0 +1,86 @@ +import json + +import pytest +from httpx import AsyncClient +from langflow.main import GZIP_EXCLUDED_CONTENT_TYPES, GZIP_MINIMUM_SIZE + +LARGE_NODE_COUNT = 40 + + +def _large_flow_payload() -> dict: + nodes = [ + { + "id": f"node-{index}", + "data": {"node": {"template": {"code": {"value": "from lfx.custom import Component\n" * 12}}}}, + } + for index in range(LARGE_NODE_COUNT) + ] + return { + "name": "compression fixture", + "description": "flow large enough to cross the compression threshold", + "data": {"nodes": nodes, "edges": []}, + "is_component": False, + "webhook": False, + } + + +async def _create_large_flow(client: AsyncClient, headers: dict) -> str: + payload = _large_flow_payload() + assert len(json.dumps(payload).encode()) > GZIP_MINIMUM_SIZE + response = await client.post("api/v1/flows/", json=payload, headers=headers) + assert response.status_code == 201 + return response.json()["id"] + + +@pytest.mark.asyncio +async def test_flow_read_is_compressed_when_the_client_accepts_gzip(client: AsyncClient, logged_in_headers): + flow_id = await _create_large_flow(client, logged_in_headers) + + response = await client.get(f"api/v1/flows/{flow_id}", headers={**logged_in_headers, "Accept-Encoding": "gzip"}) + + assert response.status_code == 200 + assert response.headers["content-encoding"] == "gzip" + assert "accept-encoding" in response.headers["vary"].lower() + assert response.json()["id"] == flow_id + + +@pytest.mark.asyncio +async def test_flow_read_is_untouched_when_the_client_does_not_accept_gzip(client: AsyncClient, logged_in_headers): + flow_id = await _create_large_flow(client, logged_in_headers) + + response = await client.get(f"api/v1/flows/{flow_id}", headers={**logged_in_headers, "Accept-Encoding": "identity"}) + + assert response.status_code == 200 + assert "content-encoding" not in response.headers + assert json.loads(response.content)["id"] == flow_id + + +@pytest.mark.asyncio +async def test_flow_update_echo_is_compressed(client: AsyncClient, logged_in_headers): + flow_id = await _create_large_flow(client, logged_in_headers) + payload = _large_flow_payload() + payload["name"] = "compression fixture renamed" + + response = await client.patch( + f"api/v1/flows/{flow_id}", + json=payload, + headers={**logged_in_headers, "Accept-Encoding": "gzip"}, + ) + + assert response.status_code == 200 + assert response.headers["content-encoding"] == "gzip" + assert response.json()["name"] == "compression fixture renamed" + + +@pytest.mark.asyncio +async def test_response_below_the_threshold_is_not_compressed(client: AsyncClient, logged_in_headers): + response = await client.get("api/v1/version", headers={**logged_in_headers, "Accept-Encoding": "gzip"}) + + assert response.status_code == 200 + assert len(response.content) < GZIP_MINIMUM_SIZE + assert "content-encoding" not in response.headers + + +def test_binary_and_streaming_content_types_are_excluded(): + for content_type in ("application/octet-stream", "application/zip", "text/event-stream", "image/png"): + assert content_type in GZIP_EXCLUDED_CONTENT_TYPES From f1237e5dd1eb7375866a0972cb1737420c84e5ac Mon Sep 17 00:00:00 2001 From: Tarcio Date: Wed, 2 Sep 2026 15:13:37 -0300 Subject: [PATCH 03/10] refactor(api): remove the unconditional compression helper compress_response gzipped every payload without reading Accept-Encoding, so a client that cannot decompress got a binary body on seven routes, GET /flows/ among them. The middleware now decides for the whole API and honours the header; these routes keep bypassing response_model validation through JSONResponse, as they did before. --- src/backend/base/langflow/api/v1/endpoints.py | 3 +- src/backend/base/langflow/api/v1/flows.py | 18 +- .../base/langflow/utils/compression.py | 19 -- .../tests/unit/utils/test_compression.py | 288 ------------------ 4 files changed, 9 insertions(+), 319 deletions(-) delete mode 100644 src/backend/base/langflow/utils/compression.py delete mode 100644 src/backend/tests/unit/utils/test_compression.py diff --git a/src/backend/base/langflow/api/v1/endpoints.py b/src/backend/base/langflow/api/v1/endpoints.py index c1c0b77aaf0a..2834328c92d1 100644 --- a/src/backend/base/langflow/api/v1/endpoints.py +++ b/src/backend/base/langflow/api/v1/endpoints.py @@ -116,7 +116,6 @@ ) from langflow.services.event_manager import create_webhook_event_manager, webhook_event_manager from langflow.services.telemetry.schema import RunPayload -from langflow.utils.compression import compress_response from langflow.utils.version import get_version_info if TYPE_CHECKING: @@ -275,7 +274,7 @@ async def get_all( all_types = translate_component_dict(visible_types_en, locale) if locale != "en" else visible_types_en component_display_names = build_component_display_names(visible_types_en) - return compress_response({**all_types, "component_display_names": component_display_names}) + return JSONResponse(content=jsonable_encoder({**all_types, "component_display_names": component_display_names})) except Exception as exc: raise HTTPException(status_code=500, detail=str(exc)) from exc diff --git a/src/backend/base/langflow/api/v1/flows.py b/src/backend/base/langflow/api/v1/flows.py index 3adca02cd85e..b3c4274fc1ea 100644 --- a/src/backend/base/langflow/api/v1/flows.py +++ b/src/backend/base/langflow/api/v1/flows.py @@ -11,6 +11,7 @@ import orjson from fastapi import APIRouter, Depends, File, HTTPException, Request, UploadFile, status from fastapi.encoders import jsonable_encoder +from fastapi.responses import JSONResponse from fastapi_pagination import Page, Params from fastapi_pagination.ext.sqlmodel import apaginate from lfx.log.logger import logger @@ -106,7 +107,6 @@ from langflow.services.database.models.user.model import User, UserRead from langflow.services.deps import get_catalog_policy_service, get_settings_service, get_storage_service from langflow.services.storage.service import StorageService -from langflow.utils.compression import compress_response from langflow.utils.i18n import translate_flow_notes, translate_starter_flows # Re-export helpers so existing ``from langflow.api.v1.flows import ...`` still works. @@ -447,13 +447,13 @@ async def read_flows( act=FlowAction.READ, ) if header_flows: - # Convert to FlowHeader objects and compress the response + # Convert to FlowHeader objects flow_headers = [FlowHeader.model_validate(flow, from_attributes=True) for flow in flows] - return compress_response(flow_headers) + return JSONResponse(content=jsonable_encoder(flow_headers)) # Convert to FlowRead while session is still active to avoid detached instance errors flow_reads = [FlowRead.model_validate(flow, from_attributes=True) for flow in flows] - return compress_response(flow_reads) + return JSONResponse(content=jsonable_encoder(flow_reads)) stmt = stmt.where(Flow.folder_id == folder_id) @@ -741,8 +741,6 @@ async def upsert_flow( Returns 201 for creation, 200 for update. Returns 404 if owned by another user. """ - from fastapi.responses import JSONResponse - # Read once, outside the retry loop: a rollback between attempts expires the ORM User # and a later attribute read would lazy-load outside the greenlet. writer_id = current_user.id @@ -1381,7 +1379,7 @@ async def read_basic_examples( blocked_template_keys=catalog_policy_snapshot.blocked_template_keys, ) ) - return compress_response(visible_flows) + return JSONResponse(content=jsonable_encoder(visible_flows)) async with _starter_flows_lock: # Double-check inside lock to prevent thundering herd @@ -1395,7 +1393,7 @@ async def read_basic_examples( blocked_template_keys=catalog_policy_snapshot.blocked_template_keys, ) ) - return compress_response(visible_flows) + return JSONResponse(content=jsonable_encoder(visible_flows)) # Ensure raw DB data is cached cached_flow_reads = _starter_flows_cache.get("starter_flows") @@ -1408,7 +1406,7 @@ async def read_basic_examples( ).first() if not starter_folder: - return compress_response([]) + return JSONResponse(content=jsonable_encoder([])) all_starter_folder_flows = ( await session.exec(select(Flow).where(Flow.folder_id == starter_folder.id)) @@ -1449,7 +1447,7 @@ async def read_basic_examples( blocked_template_keys=catalog_policy_snapshot.blocked_template_keys, ) ) - return compress_response(visible_flows) + return JSONResponse(content=jsonable_encoder(visible_flows)) @router.post("/expand/", status_code=200, dependencies=[Depends(get_current_active_user)], include_in_schema=False) diff --git a/src/backend/base/langflow/utils/compression.py b/src/backend/base/langflow/utils/compression.py deleted file mode 100644 index 30c93aa724d8..000000000000 --- a/src/backend/base/langflow/utils/compression.py +++ /dev/null @@ -1,19 +0,0 @@ -import gzip -import json -from typing import Any - -from fastapi import Response -from fastapi.encoders import jsonable_encoder - - -def compress_response(data: Any) -> Response: - """Compress data and return it as a FastAPI Response with appropriate headers.""" - json_data = json.dumps(jsonable_encoder(data)).encode("utf-8") - - compressed_data = gzip.compress(json_data, compresslevel=6) - - return Response( - content=compressed_data, - media_type="application/json", - headers={"Content-Encoding": "gzip", "Vary": "Accept-Encoding", "Content-Length": str(len(compressed_data))}, - ) diff --git a/src/backend/tests/unit/utils/test_compression.py b/src/backend/tests/unit/utils/test_compression.py deleted file mode 100644 index 37b4d8381d03..000000000000 --- a/src/backend/tests/unit/utils/test_compression.py +++ /dev/null @@ -1,288 +0,0 @@ -import gzip -import json -from datetime import date, datetime, timezone -from unittest.mock import patch - -from fastapi import Response -from langflow.utils.compression import compress_response - - -class TestCompressResponse: - """Test cases for compress_response function.""" - - def test_compress_response_simple_dict(self): - """Test compressing a simple dictionary.""" - data = {"message": "hello", "status": "success"} - - response = compress_response(data) - - assert isinstance(response, Response) - assert response.media_type == "application/json" - assert response.headers["Content-Encoding"] == "gzip" - assert response.headers["Vary"] == "Accept-Encoding" - assert "Content-Length" in response.headers - - # Decompress and verify content - decompressed = gzip.decompress(response.body) - parsed_data = json.loads(decompressed.decode("utf-8")) - assert parsed_data == data - - def test_compress_response_simple_list(self): - """Test compressing a simple list.""" - data = ["item1", "item2", "item3"] - - response = compress_response(data) - - assert isinstance(response, Response) - assert response.media_type == "application/json" - - # Decompress and verify content - decompressed = gzip.decompress(response.body) - parsed_data = json.loads(decompressed.decode("utf-8")) - assert parsed_data == data - - def test_compress_response_string(self): - """Test compressing a string.""" - data = "simple string message" - - response = compress_response(data) - - assert isinstance(response, Response) - - # Decompress and verify content - decompressed = gzip.decompress(response.body) - parsed_data = json.loads(decompressed.decode("utf-8")) - assert parsed_data == data - - def test_compress_response_number(self): - """Test compressing numeric data.""" - data = 42 - - response = compress_response(data) - - assert isinstance(response, Response) - - # Decompress and verify content - decompressed = gzip.decompress(response.body) - parsed_data = json.loads(decompressed.decode("utf-8")) - assert parsed_data == data - - def test_compress_response_boolean(self): - """Test compressing boolean data.""" - data = True - - response = compress_response(data) - - assert isinstance(response, Response) - - # Decompress and verify content - decompressed = gzip.decompress(response.body) - parsed_data = json.loads(decompressed.decode("utf-8")) - assert parsed_data == data - - def test_compress_response_none(self): - """Test compressing None value.""" - data = None - - response = compress_response(data) - - assert isinstance(response, Response) - - # Decompress and verify content - decompressed = gzip.decompress(response.body) - parsed_data = json.loads(decompressed.decode("utf-8")) - assert parsed_data is None - - def test_compress_response_nested_data(self): - """Test compressing nested data structures.""" - data = { - "users": [{"id": 1, "name": "Alice", "active": True}, {"id": 2, "name": "Bob", "active": False}], - "metadata": {"total": 2, "page": 1, "has_more": False}, - "settings": None, - } - - response = compress_response(data) - - assert isinstance(response, Response) - - # Decompress and verify content - decompressed = gzip.decompress(response.body) - parsed_data = json.loads(decompressed.decode("utf-8")) - assert parsed_data == data - - def test_compress_response_large_data(self): - """Test compressing large data to verify compression effectiveness.""" - # Create large data that should compress well (repeated patterns) - data = {"items": ["test_item"] * 1000, "metadata": {"repeated_value": "x" * 500}} - - response = compress_response(data) - - # Original JSON size - original_json = json.dumps(data).encode("utf-8") - original_size = len(original_json) - compressed_size = len(response.body) - - # Verify compression occurred (should be significantly smaller) - assert compressed_size < original_size - assert compressed_size < original_size * 0.5 # At least 50% compression - - # Verify content integrity - decompressed = gzip.decompress(response.body) - parsed_data = json.loads(decompressed.decode("utf-8")) - assert parsed_data == data - - def test_compress_response_unicode_data(self): - """Test compressing data with unicode characters.""" - data = { - "message": "Hello 世界! 🌍 Émojis and accénts", - "unicode_string": "テスト データ", - "special_chars": "àáâãäåæçèéêë", - } - - response = compress_response(data) - - assert isinstance(response, Response) - - # Decompress and verify content - decompressed = gzip.decompress(response.body) - parsed_data = json.loads(decompressed.decode("utf-8")) - assert parsed_data == data - - def test_compress_response_empty_dict(self): - """Test compressing empty dictionary.""" - data = {} - - response = compress_response(data) - - assert isinstance(response, Response) - - # Decompress and verify content - decompressed = gzip.decompress(response.body) - parsed_data = json.loads(decompressed.decode("utf-8")) - assert parsed_data == data - - def test_compress_response_empty_list(self): - """Test compressing empty list.""" - data = [] - - response = compress_response(data) - - assert isinstance(response, Response) - - # Decompress and verify content - decompressed = gzip.decompress(response.body) - parsed_data = json.loads(decompressed.decode("utf-8")) - assert parsed_data == data - - def test_compress_response_headers(self): - """Test that response has correct headers.""" - data = {"test": "data"} - - response = compress_response(data) - - # Check required headers - assert response.headers["Content-Encoding"] == "gzip" - assert response.headers["Vary"] == "Accept-Encoding" - assert response.headers["Content-Length"] == str(len(response.body)) - assert response.media_type == "application/json" - - def test_compress_response_content_length_accuracy(self): - """Test that Content-Length header matches actual body length.""" - data = {"message": "test", "numbers": [1, 2, 3, 4, 5]} - - response = compress_response(data) - - content_length = int(response.headers["Content-Length"]) - actual_length = len(response.body) - - assert content_length == actual_length - - @patch("langflow.utils.compression.jsonable_encoder") - def test_compress_response_jsonable_encoder_called(self, mock_encoder): - """Test that jsonable_encoder is called on the data.""" - data = {"test": "data"} - mock_encoder.return_value = data - - compress_response(data) - - mock_encoder.assert_called_once_with(data) - - @patch("langflow.utils.compression.gzip.compress") - def test_compress_response_gzip_compression_level(self, mock_compress): - """Test that gzip.compress is called with correct compression level.""" - data = {"test": "data"} - mock_compress.return_value = b"compressed_data" - - compress_response(data) - - # Verify gzip.compress was called with compresslevel=6 - mock_compress.assert_called_once() - call_args = mock_compress.call_args - assert call_args[1]["compresslevel"] == 6 - - def test_compress_response_with_custom_objects(self): - """Test compressing data with objects that need JSON encoding.""" - data = { - "timestamp": datetime(2023, 1, 1, 12, 0, 0, tzinfo=timezone.utc), - "date": date(2023, 1, 1), - "message": "test with custom objects", - } - - response = compress_response(data) - - assert isinstance(response, Response) - - # Decompress and verify content (datetime should be converted to string) - decompressed = gzip.decompress(response.body) - parsed_data = json.loads(decompressed.decode("utf-8")) - - # Check that custom objects were properly serialized - assert "timestamp" in parsed_data - assert "date" in parsed_data - assert parsed_data["message"] == "test with custom objects" - - def test_compress_response_compression_ratio(self): - """Test compression ratio with different types of data.""" - # Highly compressible data (lots of repetition) - repetitive_data = {"data": "a" * 1000} - - # Less compressible data (more random) - import random - import string - - random_data = {"data": ["".join(random.choices(string.ascii_letters, k=10)) for _ in range(100)]} # noqa: S311 - - rep_response = compress_response(repetitive_data) - rand_response = compress_response(random_data) - - rep_original = len(json.dumps(repetitive_data).encode("utf-8")) - rand_original = len(json.dumps(random_data).encode("utf-8")) - - rep_compressed = len(rep_response.body) - rand_compressed = len(rand_response.body) - - # Repetitive data should have better compression ratio - rep_ratio = rep_compressed / rep_original - rand_ratio = rand_compressed / rand_original - - assert rep_ratio < rand_ratio # Better compression for repetitive data - assert rep_ratio < 0.1 # Very good compression for repetitive data - - def test_compress_response_error_handling_invalid_json(self): - """Test error handling when data cannot be JSON serialized.""" - - # Create an object that cannot be JSON serialized - class NonSerializable: - def __init__(self): - self.func = lambda x: x - - data = {"object": NonSerializable()} - - # jsonable_encoder should handle this, but if it doesn't, test the behavior - try: - response = compress_response(data) - # If no exception, verify the response is still valid - assert isinstance(response, Response) - except (TypeError, ValueError): - # Expected behavior if jsonable_encoder can't handle the object - pass From 398635b5ab16cbbf2be3a7071c7098ede1257224 Mon Sep 17 00:00:00 2001 From: Tarcio Date: Wed, 2 Sep 2026 15:14:31 -0300 Subject: [PATCH 04/10] perf(db): add the version payload codec Round-trips a flow graph through gzip at level 6 and reuses FlowVersionSerializationError, which the API layer already translates to 422. Nothing calls it yet. --- .../models/flow_version/serialization.py | 30 ++++++++++ .../database/models/flow_version/__init__.py | 0 .../models/flow_version/test_serialization.py | 57 +++++++++++++++++++ 3 files changed, 87 insertions(+) create mode 100644 src/backend/base/langflow/services/database/models/flow_version/serialization.py create mode 100644 src/backend/tests/unit/services/database/models/flow_version/__init__.py create mode 100644 src/backend/tests/unit/services/database/models/flow_version/test_serialization.py diff --git a/src/backend/base/langflow/services/database/models/flow_version/serialization.py b/src/backend/base/langflow/services/database/models/flow_version/serialization.py new file mode 100644 index 000000000000..23155bd8c1a0 --- /dev/null +++ b/src/backend/base/langflow/services/database/models/flow_version/serialization.py @@ -0,0 +1,30 @@ +from __future__ import annotations + +import gzip +import json +from typing import Any + +from langflow.services.database.models.flow_version.exceptions import FlowVersionSerializationError + +COMPRESS_LEVEL = 6 + + +def pack(data: dict[str, Any] | None) -> bytes | None: + if data is None: + return None + try: + encoded = json.dumps(data, separators=(",", ":"), ensure_ascii=False).encode("utf-8") + except (TypeError, ValueError) as exc: + msg = "Flow version data could not be serialized." + raise FlowVersionSerializationError(msg) from exc + return gzip.compress(encoded, COMPRESS_LEVEL) + + +def unpack(blob: bytes | None) -> dict[str, Any] | None: + if blob is None: + return None + try: + return json.loads(gzip.decompress(blob)) + except (OSError, EOFError, ValueError, UnicodeDecodeError) as exc: + msg = "Flow version data could not be read." + raise FlowVersionSerializationError(msg) from exc diff --git a/src/backend/tests/unit/services/database/models/flow_version/__init__.py b/src/backend/tests/unit/services/database/models/flow_version/__init__.py new file mode 100644 index 000000000000..e69de29bb2d1 diff --git a/src/backend/tests/unit/services/database/models/flow_version/test_serialization.py b/src/backend/tests/unit/services/database/models/flow_version/test_serialization.py new file mode 100644 index 000000000000..e52ff00e7604 --- /dev/null +++ b/src/backend/tests/unit/services/database/models/flow_version/test_serialization.py @@ -0,0 +1,57 @@ +import gzip +import json + +import pytest +from langflow.services.database.models.flow_version.exceptions import FlowVersionSerializationError +from langflow.services.database.models.flow_version.serialization import pack, unpack + +GRAPH = { + "nodes": [{"id": "node-1", "data": {"node": {"template": {"code": {"value": "from lfx import x\n"}}}}}], + "edges": [], + "viewport": {"x": 0, "y": 0, "zoom": 1}, +} + + +def test_pack_then_unpack_returns_an_equal_document(): + assert unpack(pack(GRAPH)) == GRAPH + + +def test_pack_returns_gzip_bytes(): + packed = pack(GRAPH) + + assert isinstance(packed, bytes) + assert json.loads(gzip.decompress(packed)) == GRAPH + + +def test_pack_shrinks_a_repetitive_graph(): + graph = {"nodes": [GRAPH["nodes"][0] for _ in range(50)], "edges": []} + + packed = pack(graph) + + assert len(packed) < len(json.dumps(graph).encode()) / 4 + + +def test_none_passes_through_both_ways(): + assert pack(None) is None + assert unpack(None) is None + + +def test_non_ascii_survives_the_round_trip(): + graph = {"nodes": [], "edges": [], "name": "Análise de Sentimento — ação"} + + assert unpack(pack(graph)) == graph + + +def test_unpack_rejects_data_that_is_not_gzip(): + with pytest.raises(FlowVersionSerializationError): + unpack(b"not gzip at all") + + +def test_unpack_rejects_gzip_that_is_not_json(): + with pytest.raises(FlowVersionSerializationError): + unpack(gzip.compress(b"")) + + +def test_pack_rejects_a_document_json_cannot_encode(): + with pytest.raises(FlowVersionSerializationError): + pack({"nodes": {object()}}) From 301137068e3e5e1027220544ff3304aaa7f0869e Mon Sep 17 00:00:00 2001 From: Tarcio Date: Wed, 2 Sep 2026 15:17:33 -0300 Subject: [PATCH 05/10] perf(db): migrate flow_version.data to gzipped bytes Backfills in batches of 200 and verifies no row is left behind before dropping the JSON column, because a WHERE that silently matches nothing would drop the data instead of moving it. Column ids are left untyped so the update matches rows whatever spelling of UUID the engine stored. On PostgreSQL the new column takes STORAGE EXTERNAL: TOAST would otherwise spend write CPU compressing bytes that are already compressed. Verified on SQLite: 4 seeded versions, 37,650 bytes of JSON becoming 867 bytes, NULL preserved, and downgrade restoring the same 37,650 bytes. --- ...d3b7c1e05f84_compress_flow_version_data.py | 107 ++++++++++++++++++ 1 file changed, 107 insertions(+) create mode 100644 src/backend/base/langflow/alembic/versions/d3b7c1e05f84_compress_flow_version_data.py diff --git a/src/backend/base/langflow/alembic/versions/d3b7c1e05f84_compress_flow_version_data.py b/src/backend/base/langflow/alembic/versions/d3b7c1e05f84_compress_flow_version_data.py new file mode 100644 index 000000000000..a9dca6fc0b14 --- /dev/null +++ b/src/backend/base/langflow/alembic/versions/d3b7c1e05f84_compress_flow_version_data.py @@ -0,0 +1,107 @@ +"""Store flow version snapshots gzipped. + +Phase: MIGRATE +Revision ID: d3b7c1e05f84 +Revises: c9f2e5a7b1d4 +Create Date: 2026-09-02 +""" + +from __future__ import annotations + +import gzip +import json +from typing import TYPE_CHECKING + +import sqlalchemy as sa +from alembic import op +from langflow.utils import migration + +if TYPE_CHECKING: + from collections.abc import Sequence + +revision: str = "d3b7c1e05f84" # pragma: allowlist secret +down_revision: str | None = "c9f2e5a7b1d4" # pragma: allowlist secret +branch_labels: str | Sequence[str] | None = None +depends_on: str | Sequence[str] | None = None + +BATCH_SIZE = 200 +COMPRESS_LEVEL = 6 + +_flow_version = sa.table( + "flow_version", + sa.column("id"), + sa.column("data", sa.JSON()), + sa.column("data_gz", sa.LargeBinary()), +) + + +def _rows(conn, source: sa.Column): + offset = 0 + while True: + stmt = ( + sa.select(_flow_version.c.id, source) + .where(source.is_not(None)) + .order_by(_flow_version.c.id) + .limit(BATCH_SIZE) + .offset(offset) + ) + batch = conn.execute(stmt).fetchall() + if not batch: + return + yield batch + offset += BATCH_SIZE + + +def _set_external_storage(conn) -> None: + if conn.dialect.name == "postgresql": + op.execute("ALTER TABLE flow_version ALTER COLUMN data_gz SET STORAGE EXTERNAL") + + +def upgrade() -> None: + conn = op.get_bind() + if not migration.table_exists("flow_version", conn): + return + + if not migration.column_exists("flow_version", "data_gz", conn): + op.add_column("flow_version", sa.Column("data_gz", sa.LargeBinary(), nullable=True)) + _set_external_storage(conn) + + if migration.column_exists("flow_version", "data", conn): + migrated = 0 + for batch in _rows(conn, _flow_version.c.data): + for row in batch: + payload = gzip.compress( + json.dumps(row.data, separators=(",", ":"), ensure_ascii=False).encode("utf-8"), + COMPRESS_LEVEL, + ) + result = conn.execute( + _flow_version.update().where(_flow_version.c.id == row.id).values(data_gz=payload) + ) + migrated += result.rowcount + pending = conn.execute( + sa.select(sa.func.count()) + .select_from(_flow_version) + .where(_flow_version.c.data.is_not(None), _flow_version.c.data_gz.is_(None)) + ).scalar_one() + if pending: + msg = f"{pending} flow_version rows still hold uncompressed data after backfilling {migrated}" + raise RuntimeError(msg) + with op.batch_alter_table("flow_version") as batch_op: + batch_op.drop_column("data") + + +def downgrade() -> None: + conn = op.get_bind() + if not migration.table_exists("flow_version", conn): + return + + if not migration.column_exists("flow_version", "data", conn): + op.add_column("flow_version", sa.Column("data", sa.JSON(), nullable=True)) + + if migration.column_exists("flow_version", "data_gz", conn): + for batch in _rows(conn, _flow_version.c.data_gz): + for row in batch: + payload = json.loads(gzip.decompress(row.data_gz)) + conn.execute(_flow_version.update().where(_flow_version.c.id == row.id).values(data=payload)) + with op.batch_alter_table("flow_version") as batch_op: + batch_op.drop_column("data_gz") From 5705ec863504bc3404297e18db2bc987bd667cf6 Mon Sep 17 00:00:00 2001 From: Tarcio Date: Wed, 2 Sep 2026 15:23:57 -0300 Subject: [PATCH 06/10] perf(db): store version snapshots gzipped The compression sits in the column type, not in the call sites: FlowVersion.data still reads and writes a dict, so create_flow_version_entry, the activate path, the deployment mappers and variable.py are untouched and every existing test keeps constructing FlowVersion(data={...}). The attribute keeps its name while the column becomes data_gz, matching the migration. --- .../database/models/flow_version/model.py | 6 +++-- .../models/flow_version/serialization.py | 19 ++++++++++++++- .../models/flow_version/test_serialization.py | 23 +++++++++++++++++++ 3 files changed, 45 insertions(+), 3 deletions(-) diff --git a/src/backend/base/langflow/services/database/models/flow_version/model.py b/src/backend/base/langflow/services/database/models/flow_version/model.py index 01740411e41f..eebe19fc9d5b 100644 --- a/src/backend/base/langflow/services/database/models/flow_version/model.py +++ b/src/backend/base/langflow/services/database/models/flow_version/model.py @@ -6,7 +6,9 @@ from pydantic import BaseModel, computed_field, field_serializer from pydantic import Field as PydanticField from sqlalchemy import CheckConstraint, Column, DateTime, ForeignKey, UniqueConstraint, func -from sqlmodel import JSON, Field, SQLModel +from sqlmodel import Field, SQLModel + +from langflow.services.database.models.flow_version.serialization import GzippedJSON class FlowVersion(SQLModel, table=True): # type: ignore[call-arg] @@ -20,7 +22,7 @@ class FlowVersion(SQLModel, table=True): # type: ignore[call-arg] user_id: UUID | None = Field( sa_column=Column(ForeignKey("user.id", ondelete="SET NULL"), index=True, nullable=True), ) - data: dict | None = Field(default=None, sa_column=Column(JSON)) + data: dict | None = Field(default=None, sa_column=Column("data_gz", GzippedJSON)) version_number: int = Field(nullable=False, ge=1) description: str | None = Field(default=None, nullable=True, max_length=500) created_at: datetime = Field( diff --git a/src/backend/base/langflow/services/database/models/flow_version/serialization.py b/src/backend/base/langflow/services/database/models/flow_version/serialization.py index 23155bd8c1a0..2f10d295f4ec 100644 --- a/src/backend/base/langflow/services/database/models/flow_version/serialization.py +++ b/src/backend/base/langflow/services/database/models/flow_version/serialization.py @@ -2,10 +2,16 @@ import gzip import json -from typing import Any +from typing import TYPE_CHECKING, Any + +from sqlalchemy import LargeBinary +from sqlalchemy.types import TypeDecorator from langflow.services.database.models.flow_version.exceptions import FlowVersionSerializationError +if TYPE_CHECKING: + from sqlalchemy.engine.interfaces import Dialect + COMPRESS_LEVEL = 6 @@ -28,3 +34,14 @@ def unpack(blob: bytes | None) -> dict[str, Any] | None: except (OSError, EOFError, ValueError, UnicodeDecodeError) as exc: msg = "Flow version data could not be read." raise FlowVersionSerializationError(msg) from exc + + +class GzippedJSON(TypeDecorator): + impl = LargeBinary + cache_ok = True + + def process_bind_param(self, value: dict[str, Any] | None, dialect: Dialect) -> bytes | None: # noqa: ARG002 + return pack(value) + + def process_result_value(self, value: bytes | None, dialect: Dialect) -> dict[str, Any] | None: # noqa: ARG002 + return unpack(value) diff --git a/src/backend/tests/unit/services/database/models/flow_version/test_serialization.py b/src/backend/tests/unit/services/database/models/flow_version/test_serialization.py index e52ff00e7604..745a5f8fdf3a 100644 --- a/src/backend/tests/unit/services/database/models/flow_version/test_serialization.py +++ b/src/backend/tests/unit/services/database/models/flow_version/test_serialization.py @@ -4,6 +4,7 @@ import pytest from langflow.services.database.models.flow_version.exceptions import FlowVersionSerializationError from langflow.services.database.models.flow_version.serialization import pack, unpack +from sqlalchemy import Integer GRAPH = { "nodes": [{"id": "node-1", "data": {"node": {"template": {"code": {"value": "from lfx import x\n"}}}}}], @@ -55,3 +56,25 @@ def test_unpack_rejects_gzip_that_is_not_json(): def test_pack_rejects_a_document_json_cannot_encode(): with pytest.raises(FlowVersionSerializationError): pack({"nodes": {object()}}) + + +def test_the_column_stores_gzip_bytes_and_reads_back_a_dict(): + from langflow.services.database.models.flow_version.serialization import GzippedJSON + from sqlalchemy import Column, MetaData, Table, create_engine, select + + metadata = MetaData() + table = Table("sample", metadata, Column("id", Integer, primary_key=True), Column("payload", GzippedJSON)) + engine = create_engine("sqlite://") + metadata.create_all(engine) + + with engine.begin() as conn: + conn.execute(table.insert().values(id=1, payload=GRAPH)) + conn.execute(table.insert().values(id=2, payload=None)) + + with engine.connect() as conn: + assert conn.execute(select(table.c.payload).where(table.c.id == 1)).scalar_one() == GRAPH + assert conn.execute(select(table.c.payload).where(table.c.id == 2)).scalar_one() is None + raw = conn.exec_driver_sql("SELECT payload FROM sample WHERE id = 1").scalar_one() + + assert raw[:2] == b"\x1f\x8b" + assert b"lfx" not in raw From 32891a1c8d0675abc99aaa208ed186884038b37c Mon Sep 17 00:00:00 2001 From: Tarcio Date: Wed, 2 Sep 2026 17:59:40 -0300 Subject: [PATCH 07/10] test: prove the exclusion behaviour and the migration round-trip MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The excluded content types were asserted as tuple membership, which proves the configuration and not the behaviour: a response of each excluded type is now served through the same middleware configuration and checked for the absence of content-encoding, with a compressible type on the same app as the control. The migration had no automated coverage. It now runs upgrade and downgrade against a seeded table: every populated row round-trips, NULL survives both directions, the guard raises rather than dropping the column when a row was left behind, and a database without the table is a no-op. Batch size is patched to 2 so the paging is exercised. Also drops the redundant pytest.mark.asyncio decorators — asyncio_mode is auto in all three pyproject files. --- .../test_compress_flow_version_data.py | 108 ++++++++++++++++++ .../unit/api/test_response_compression.py | 56 +++++++-- 2 files changed, 156 insertions(+), 8 deletions(-) create mode 100644 src/backend/tests/unit/alembic/test_compress_flow_version_data.py diff --git a/src/backend/tests/unit/alembic/test_compress_flow_version_data.py b/src/backend/tests/unit/alembic/test_compress_flow_version_data.py new file mode 100644 index 000000000000..66d2128cc537 --- /dev/null +++ b/src/backend/tests/unit/alembic/test_compress_flow_version_data.py @@ -0,0 +1,108 @@ +"""Tests for the flow-version payload compression migration.""" + +from __future__ import annotations + +import gzip +import importlib +import json + +import pytest +import sqlalchemy as sa +from alembic.migration import MigrationContext +from alembic.operations import Operations + +_MIGRATION = importlib.import_module("langflow.alembic.versions.d3b7c1e05f84_compress_flow_version_data") + +GRAPH = { + "nodes": [{"id": f"node-{index}", "data": {"code": "from lfx import Component\n" * 8}} for index in range(6)], + "edges": [], + "name": "Análise de Sentimento — ação", +} +POPULATED_ROWS = 5 + + +def _engine_before_the_migration(): + engine = sa.create_engine("sqlite://") + metadata = sa.MetaData() + sa.Table( + "flow_version", + metadata, + sa.Column("id", sa.String(), primary_key=True), + sa.Column("data", sa.JSON(), nullable=True), + sa.Column("version_number", sa.Integer(), nullable=False), + ) + metadata.create_all(engine) + with engine.begin() as conn: + for index in range(POPULATED_ROWS): + conn.execute( + sa.text("INSERT INTO flow_version (id, data, version_number) VALUES (:id, :data, :n)"), + {"id": f"row-{index}", "data": json.dumps(GRAPH), "n": index + 1}, + ) + conn.execute( + sa.text("INSERT INTO flow_version (id, data, version_number) VALUES (:id, NULL, :n)"), + {"id": "row-null", "n": POPULATED_ROWS + 1}, + ) + return engine + + +def _run(engine, direction: str) -> None: + with engine.begin() as conn: + original = _MIGRATION.op + try: + _MIGRATION.op = Operations(MigrationContext.configure(conn)) + getattr(_MIGRATION, direction)() + finally: + _MIGRATION.op = original + + +def _columns(engine) -> set[str]: + return {column["name"] for column in sa.inspect(engine).get_columns("flow_version")} + + +def test_upgrade_compresses_every_populated_row_and_keeps_null(monkeypatch): + monkeypatch.setattr(_MIGRATION, "BATCH_SIZE", 2) + engine = _engine_before_the_migration() + + _run(engine, "upgrade") + + assert _columns(engine) == {"id", "data_gz", "version_number"} + with engine.connect() as conn: + rows = conn.execute(sa.text("SELECT id, data_gz FROM flow_version ORDER BY version_number")).fetchall() + stored = {row.id: row.data_gz for row in rows} + assert stored["row-null"] is None + for index in range(POPULATED_ROWS): + payload = stored[f"row-{index}"] + assert payload[:2] == b"\x1f\x8b" + assert json.loads(gzip.decompress(payload)) == GRAPH + + +def test_downgrade_restores_the_original_json(monkeypatch): + monkeypatch.setattr(_MIGRATION, "BATCH_SIZE", 2) + engine = _engine_before_the_migration() + + _run(engine, "upgrade") + _run(engine, "downgrade") + + assert _columns(engine) == {"id", "data", "version_number"} + with engine.connect() as conn: + rows = conn.execute(sa.text("SELECT id, data FROM flow_version ORDER BY version_number")).fetchall() + stored = {row.id: row.data for row in rows} + assert stored["row-null"] is None + for index in range(POPULATED_ROWS): + assert json.loads(stored[f"row-{index}"]) == GRAPH + + +def test_upgrade_refuses_to_drop_the_column_when_a_row_was_not_migrated(monkeypatch): + engine = _engine_before_the_migration() + monkeypatch.setattr(_MIGRATION, "_rows", lambda *_: iter(())) + + with pytest.raises(RuntimeError, match="still hold uncompressed data"): + _run(engine, "upgrade") + + +def test_upgrade_is_a_noop_without_the_table(): + engine = sa.create_engine("sqlite://") + + _run(engine, "upgrade") + + assert not sa.inspect(engine).has_table("flow_version") diff --git a/src/backend/tests/unit/api/test_response_compression.py b/src/backend/tests/unit/api/test_response_compression.py index 9cff9ba9b4cb..8c6fcbe1f150 100644 --- a/src/backend/tests/unit/api/test_response_compression.py +++ b/src/backend/tests/unit/api/test_response_compression.py @@ -1,8 +1,10 @@ import json import pytest -from httpx import AsyncClient -from langflow.main import GZIP_EXCLUDED_CONTENT_TYPES, GZIP_MINIMUM_SIZE +from fastapi import FastAPI, Response +from httpx import ASGITransport, AsyncClient +from langflow.main import GZIP_COMPRESS_LEVEL, GZIP_EXCLUDED_CONTENT_TYPES, GZIP_MINIMUM_SIZE +from starlette.middleware.gzip import GZipMiddleware LARGE_NODE_COUNT = 40 @@ -32,7 +34,6 @@ async def _create_large_flow(client: AsyncClient, headers: dict) -> str: return response.json()["id"] -@pytest.mark.asyncio async def test_flow_read_is_compressed_when_the_client_accepts_gzip(client: AsyncClient, logged_in_headers): flow_id = await _create_large_flow(client, logged_in_headers) @@ -44,7 +45,6 @@ async def test_flow_read_is_compressed_when_the_client_accepts_gzip(client: Asyn assert response.json()["id"] == flow_id -@pytest.mark.asyncio async def test_flow_read_is_untouched_when_the_client_does_not_accept_gzip(client: AsyncClient, logged_in_headers): flow_id = await _create_large_flow(client, logged_in_headers) @@ -55,7 +55,6 @@ async def test_flow_read_is_untouched_when_the_client_does_not_accept_gzip(clien assert json.loads(response.content)["id"] == flow_id -@pytest.mark.asyncio async def test_flow_update_echo_is_compressed(client: AsyncClient, logged_in_headers): flow_id = await _create_large_flow(client, logged_in_headers) payload = _large_flow_payload() @@ -72,7 +71,6 @@ async def test_flow_update_echo_is_compressed(client: AsyncClient, logged_in_hea assert response.json()["name"] == "compression fixture renamed" -@pytest.mark.asyncio async def test_response_below_the_threshold_is_not_compressed(client: AsyncClient, logged_in_headers): response = await client.get("api/v1/version", headers={**logged_in_headers, "Accept-Encoding": "gzip"}) @@ -81,6 +79,48 @@ async def test_response_below_the_threshold_is_not_compressed(client: AsyncClien assert "content-encoding" not in response.headers -def test_binary_and_streaming_content_types_are_excluded(): - for content_type in ("application/octet-stream", "application/zip", "text/event-stream", "image/png"): +EXCLUDED_UNDER_TEST = ("application/octet-stream", "application/zip", "text/event-stream", "image/png") + + +def _app_with_the_same_middleware() -> FastAPI: + app = FastAPI() + app.add_middleware( + GZipMiddleware, + minimum_size=GZIP_MINIMUM_SIZE, + compresslevel=GZIP_COMPRESS_LEVEL, + exclude_content_types=GZIP_EXCLUDED_CONTENT_TYPES, + ) + + @app.get("/payload") + async def payload(content_type: str) -> Response: + return Response(content=b"x" * (GZIP_MINIMUM_SIZE * 4), media_type=content_type) + + return app + + +def test_binary_and_streaming_content_types_are_configured_as_excluded(): + for content_type in EXCLUDED_UNDER_TEST: assert content_type in GZIP_EXCLUDED_CONTENT_TYPES + + +@pytest.mark.parametrize("content_type", EXCLUDED_UNDER_TEST) +async def test_excluded_content_types_are_not_compressed(content_type): + transport = ASGITransport(app=_app_with_the_same_middleware()) + async with AsyncClient(transport=transport, base_url="http://testserver") as client: + response = await client.get( + "/payload", params={"content_type": content_type}, headers={"Accept-Encoding": "gzip"} + ) + + assert response.status_code == 200 + assert "content-encoding" not in response.headers + assert len(response.content) == GZIP_MINIMUM_SIZE * 4 + + +async def test_a_compressible_type_on_the_same_app_is_compressed(): + transport = ASGITransport(app=_app_with_the_same_middleware()) + async with AsyncClient(transport=transport, base_url="http://testserver") as client: + response = await client.get( + "/payload", params={"content_type": "application/json"}, headers={"Accept-Encoding": "gzip"} + ) + + assert response.headers["content-encoding"] == "gzip" From 425be5e5a6c9fc92a3356874f52e903c09e66b81 Mon Sep 17 00:00:00 2001 From: Tarcio Date: Wed, 2 Sep 2026 19:03:46 -0300 Subject: [PATCH 08/10] fix(db): give the downgrade the same guard as the upgrade MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The upgrade counts what it wrote and refuses to drop the source column while a row is unmigrated; the downgrade dropped data_gz unconditionally. Same defect, opposite direction — a reverse backfill that silently matched nothing would delete the snapshots instead of restoring them. Both directions now share one backfill and one guard, so the asymmetry cannot come back. Also excludes PDF and the OOXML types from compression: docx, xlsx and pptx are ZIP containers and PDF carries compressed streams. Measured on a synthetic docx: 126 KB in, 126 KB out, 0.0% gain. Downloads are capped at max_file_size_upload, 1024 MB by default, which extrapolates to roughly 17 seconds of CPU per download for nothing. --- ...d3b7c1e05f84_compress_flow_version_data.py | 69 +++++++++++-------- src/backend/base/langflow/main.py | 9 ++- .../test_compress_flow_version_data.py | 15 +++- .../unit/api/test_response_compression.py | 11 ++- 4 files changed, 74 insertions(+), 30 deletions(-) diff --git a/src/backend/base/langflow/alembic/versions/d3b7c1e05f84_compress_flow_version_data.py b/src/backend/base/langflow/alembic/versions/d3b7c1e05f84_compress_flow_version_data.py index a9dca6fc0b14..b5887dcf52c5 100644 --- a/src/backend/base/langflow/alembic/versions/d3b7c1e05f84_compress_flow_version_data.py +++ b/src/backend/base/langflow/alembic/versions/d3b7c1e05f84_compress_flow_version_data.py @@ -57,6 +57,32 @@ def _set_external_storage(conn) -> None: op.execute("ALTER TABLE flow_version ALTER COLUMN data_gz SET STORAGE EXTERNAL") +def _backfill(conn, source: sa.Column, target: sa.Column, convert) -> int: + written = 0 + for batch in _rows(conn, source): + for row in batch: + value = convert(getattr(row, source.name)) + result = conn.execute( + _flow_version.update().where(_flow_version.c.id == row.id).values({target.name: value}) + ) + written += result.rowcount + return written + + +def _drop_when_nothing_is_pending(conn, source: sa.Column, target: sa.Column, written: int) -> None: + pending = conn.execute( + sa.select(sa.func.count()).select_from(_flow_version).where(source.is_not(None), target.is_(None)) + ).scalar_one() + if pending: + msg = ( + f"{pending} flow_version rows still hold data in {source.name} after backfilling " + f"{written} into {target.name}" + ) + raise RuntimeError(msg) + with op.batch_alter_table("flow_version") as batch_op: + batch_op.drop_column(source.name) + + def upgrade() -> None: conn = op.get_bind() if not migration.table_exists("flow_version", conn): @@ -67,27 +93,15 @@ def upgrade() -> None: _set_external_storage(conn) if migration.column_exists("flow_version", "data", conn): - migrated = 0 - for batch in _rows(conn, _flow_version.c.data): - for row in batch: - payload = gzip.compress( - json.dumps(row.data, separators=(",", ":"), ensure_ascii=False).encode("utf-8"), - COMPRESS_LEVEL, - ) - result = conn.execute( - _flow_version.update().where(_flow_version.c.id == row.id).values(data_gz=payload) - ) - migrated += result.rowcount - pending = conn.execute( - sa.select(sa.func.count()) - .select_from(_flow_version) - .where(_flow_version.c.data.is_not(None), _flow_version.c.data_gz.is_(None)) - ).scalar_one() - if pending: - msg = f"{pending} flow_version rows still hold uncompressed data after backfilling {migrated}" - raise RuntimeError(msg) - with op.batch_alter_table("flow_version") as batch_op: - batch_op.drop_column("data") + written = _backfill( + conn, + _flow_version.c.data, + _flow_version.c.data_gz, + lambda data: gzip.compress( + json.dumps(data, separators=(",", ":"), ensure_ascii=False).encode("utf-8"), COMPRESS_LEVEL + ), + ) + _drop_when_nothing_is_pending(conn, _flow_version.c.data, _flow_version.c.data_gz, written) def downgrade() -> None: @@ -99,9 +113,10 @@ def downgrade() -> None: op.add_column("flow_version", sa.Column("data", sa.JSON(), nullable=True)) if migration.column_exists("flow_version", "data_gz", conn): - for batch in _rows(conn, _flow_version.c.data_gz): - for row in batch: - payload = json.loads(gzip.decompress(row.data_gz)) - conn.execute(_flow_version.update().where(_flow_version.c.id == row.id).values(data=payload)) - with op.batch_alter_table("flow_version") as batch_op: - batch_op.drop_column("data_gz") + written = _backfill( + conn, + _flow_version.c.data_gz, + _flow_version.c.data, + lambda blob: json.loads(gzip.decompress(blob)), + ) + _drop_when_nothing_is_pending(conn, _flow_version.c.data_gz, _flow_version.c.data, written) diff --git a/src/backend/base/langflow/main.py b/src/backend/base/langflow/main.py index 7f80aae73f39..b83971e138b8 100644 --- a/src/backend/base/langflow/main.py +++ b/src/backend/base/langflow/main.py @@ -80,7 +80,14 @@ MAX_PORT = 65535 GZIP_MINIMUM_SIZE = 1000 GZIP_COMPRESS_LEVEL = 6 -GZIP_EXCLUDED_CONTENT_TYPES = (*DEFAULT_EXCLUDED_CONTENT_TYPES, "application/octet-stream") +GZIP_ALREADY_COMPRESSED_CONTENT_TYPES = ( + "application/octet-stream", + "application/pdf", + "application/vnd.openxmlformats-officedocument.wordprocessingml.document", + "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet", + "application/vnd.openxmlformats-officedocument.presentationml.presentation", +) +GZIP_EXCLUDED_CONTENT_TYPES = (*DEFAULT_EXCLUDED_CONTENT_TYPES, *GZIP_ALREADY_COMPRESSED_CONTENT_TYPES) # Enterprise lifespan hook registry. Enterprise plugins append async callables # at app-construction time (plugin registration runs before the lifespan diff --git a/src/backend/tests/unit/alembic/test_compress_flow_version_data.py b/src/backend/tests/unit/alembic/test_compress_flow_version_data.py index 66d2128cc537..a72f305bc8df 100644 --- a/src/backend/tests/unit/alembic/test_compress_flow_version_data.py +++ b/src/backend/tests/unit/alembic/test_compress_flow_version_data.py @@ -96,9 +96,22 @@ def test_upgrade_refuses_to_drop_the_column_when_a_row_was_not_migrated(monkeypa engine = _engine_before_the_migration() monkeypatch.setattr(_MIGRATION, "_rows", lambda *_: iter(())) - with pytest.raises(RuntimeError, match="still hold uncompressed data"): + with pytest.raises(RuntimeError, match="still hold data in data"): _run(engine, "upgrade") + assert "data" in _columns(engine) + + +def test_downgrade_refuses_to_drop_the_column_when_a_row_was_not_restored(monkeypatch): + engine = _engine_before_the_migration() + _run(engine, "upgrade") + monkeypatch.setattr(_MIGRATION, "_rows", lambda *_: iter(())) + + with pytest.raises(RuntimeError, match="still hold data in data_gz"): + _run(engine, "downgrade") + + assert "data_gz" in _columns(engine) + def test_upgrade_is_a_noop_without_the_table(): engine = sa.create_engine("sqlite://") diff --git a/src/backend/tests/unit/api/test_response_compression.py b/src/backend/tests/unit/api/test_response_compression.py index 8c6fcbe1f150..d2cfa7460ae2 100644 --- a/src/backend/tests/unit/api/test_response_compression.py +++ b/src/backend/tests/unit/api/test_response_compression.py @@ -79,7 +79,16 @@ async def test_response_below_the_threshold_is_not_compressed(client: AsyncClien assert "content-encoding" not in response.headers -EXCLUDED_UNDER_TEST = ("application/octet-stream", "application/zip", "text/event-stream", "image/png") +EXCLUDED_UNDER_TEST = ( + "application/octet-stream", + "application/zip", + "text/event-stream", + "image/png", + "application/pdf", + "application/vnd.openxmlformats-officedocument.wordprocessingml.document", + "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet", + "application/vnd.openxmlformats-officedocument.presentationml.presentation", +) def _app_with_the_same_middleware() -> FastAPI: From a139efb1a8080fec924d495325d6bc7b392dda82 Mon Sep 17 00:00:00 2001 From: Tarcio Date: Thu, 3 Sep 2026 14:30:13 -0300 Subject: [PATCH 09/10] revert(db): take the version-history compression out of this PR MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The at-rest half touches flow_version, which is where the multi-user editing work (#14903) is landing, and its Alembic revision hangs off the same head that work will branch from — two revisions on one parent leave the repository with divergent heads and force a merge migration on whoever lands second. Removed as one commit rather than three: the model, the codec and the migration are one unit, and splitting the removal would leave a commit where the model imports a module that no longer exists. What stays is the wire half, which shares no file with any of this. The removed work is preserved on perf/LE-2382-version-history-at-rest and comes back as its own PR under the same ticket once #14903 settles. --- ...d3b7c1e05f84_compress_flow_version_data.py | 122 ------------------ .../database/models/flow_version/model.py | 6 +- .../models/flow_version/serialization.py | 47 ------- .../test_compress_flow_version_data.py | 121 ----------------- .../database/models/flow_version/__init__.py | 0 .../models/flow_version/test_serialization.py | 80 ------------ 6 files changed, 2 insertions(+), 374 deletions(-) delete mode 100644 src/backend/base/langflow/alembic/versions/d3b7c1e05f84_compress_flow_version_data.py delete mode 100644 src/backend/base/langflow/services/database/models/flow_version/serialization.py delete mode 100644 src/backend/tests/unit/alembic/test_compress_flow_version_data.py delete mode 100644 src/backend/tests/unit/services/database/models/flow_version/__init__.py delete mode 100644 src/backend/tests/unit/services/database/models/flow_version/test_serialization.py diff --git a/src/backend/base/langflow/alembic/versions/d3b7c1e05f84_compress_flow_version_data.py b/src/backend/base/langflow/alembic/versions/d3b7c1e05f84_compress_flow_version_data.py deleted file mode 100644 index b5887dcf52c5..000000000000 --- a/src/backend/base/langflow/alembic/versions/d3b7c1e05f84_compress_flow_version_data.py +++ /dev/null @@ -1,122 +0,0 @@ -"""Store flow version snapshots gzipped. - -Phase: MIGRATE -Revision ID: d3b7c1e05f84 -Revises: c9f2e5a7b1d4 -Create Date: 2026-09-02 -""" - -from __future__ import annotations - -import gzip -import json -from typing import TYPE_CHECKING - -import sqlalchemy as sa -from alembic import op -from langflow.utils import migration - -if TYPE_CHECKING: - from collections.abc import Sequence - -revision: str = "d3b7c1e05f84" # pragma: allowlist secret -down_revision: str | None = "c9f2e5a7b1d4" # pragma: allowlist secret -branch_labels: str | Sequence[str] | None = None -depends_on: str | Sequence[str] | None = None - -BATCH_SIZE = 200 -COMPRESS_LEVEL = 6 - -_flow_version = sa.table( - "flow_version", - sa.column("id"), - sa.column("data", sa.JSON()), - sa.column("data_gz", sa.LargeBinary()), -) - - -def _rows(conn, source: sa.Column): - offset = 0 - while True: - stmt = ( - sa.select(_flow_version.c.id, source) - .where(source.is_not(None)) - .order_by(_flow_version.c.id) - .limit(BATCH_SIZE) - .offset(offset) - ) - batch = conn.execute(stmt).fetchall() - if not batch: - return - yield batch - offset += BATCH_SIZE - - -def _set_external_storage(conn) -> None: - if conn.dialect.name == "postgresql": - op.execute("ALTER TABLE flow_version ALTER COLUMN data_gz SET STORAGE EXTERNAL") - - -def _backfill(conn, source: sa.Column, target: sa.Column, convert) -> int: - written = 0 - for batch in _rows(conn, source): - for row in batch: - value = convert(getattr(row, source.name)) - result = conn.execute( - _flow_version.update().where(_flow_version.c.id == row.id).values({target.name: value}) - ) - written += result.rowcount - return written - - -def _drop_when_nothing_is_pending(conn, source: sa.Column, target: sa.Column, written: int) -> None: - pending = conn.execute( - sa.select(sa.func.count()).select_from(_flow_version).where(source.is_not(None), target.is_(None)) - ).scalar_one() - if pending: - msg = ( - f"{pending} flow_version rows still hold data in {source.name} after backfilling " - f"{written} into {target.name}" - ) - raise RuntimeError(msg) - with op.batch_alter_table("flow_version") as batch_op: - batch_op.drop_column(source.name) - - -def upgrade() -> None: - conn = op.get_bind() - if not migration.table_exists("flow_version", conn): - return - - if not migration.column_exists("flow_version", "data_gz", conn): - op.add_column("flow_version", sa.Column("data_gz", sa.LargeBinary(), nullable=True)) - _set_external_storage(conn) - - if migration.column_exists("flow_version", "data", conn): - written = _backfill( - conn, - _flow_version.c.data, - _flow_version.c.data_gz, - lambda data: gzip.compress( - json.dumps(data, separators=(",", ":"), ensure_ascii=False).encode("utf-8"), COMPRESS_LEVEL - ), - ) - _drop_when_nothing_is_pending(conn, _flow_version.c.data, _flow_version.c.data_gz, written) - - -def downgrade() -> None: - conn = op.get_bind() - if not migration.table_exists("flow_version", conn): - return - - if not migration.column_exists("flow_version", "data", conn): - op.add_column("flow_version", sa.Column("data", sa.JSON(), nullable=True)) - - if migration.column_exists("flow_version", "data_gz", conn): - written = _backfill( - conn, - _flow_version.c.data_gz, - _flow_version.c.data, - lambda blob: json.loads(gzip.decompress(blob)), - ) - _drop_when_nothing_is_pending(conn, _flow_version.c.data_gz, _flow_version.c.data, written) diff --git a/src/backend/base/langflow/services/database/models/flow_version/model.py b/src/backend/base/langflow/services/database/models/flow_version/model.py index eebe19fc9d5b..01740411e41f 100644 --- a/src/backend/base/langflow/services/database/models/flow_version/model.py +++ b/src/backend/base/langflow/services/database/models/flow_version/model.py @@ -6,9 +6,7 @@ from pydantic import BaseModel, computed_field, field_serializer from pydantic import Field as PydanticField from sqlalchemy import CheckConstraint, Column, DateTime, ForeignKey, UniqueConstraint, func -from sqlmodel import Field, SQLModel - -from langflow.services.database.models.flow_version.serialization import GzippedJSON +from sqlmodel import JSON, Field, SQLModel class FlowVersion(SQLModel, table=True): # type: ignore[call-arg] @@ -22,7 +20,7 @@ class FlowVersion(SQLModel, table=True): # type: ignore[call-arg] user_id: UUID | None = Field( sa_column=Column(ForeignKey("user.id", ondelete="SET NULL"), index=True, nullable=True), ) - data: dict | None = Field(default=None, sa_column=Column("data_gz", GzippedJSON)) + data: dict | None = Field(default=None, sa_column=Column(JSON)) version_number: int = Field(nullable=False, ge=1) description: str | None = Field(default=None, nullable=True, max_length=500) created_at: datetime = Field( diff --git a/src/backend/base/langflow/services/database/models/flow_version/serialization.py b/src/backend/base/langflow/services/database/models/flow_version/serialization.py deleted file mode 100644 index 2f10d295f4ec..000000000000 --- a/src/backend/base/langflow/services/database/models/flow_version/serialization.py +++ /dev/null @@ -1,47 +0,0 @@ -from __future__ import annotations - -import gzip -import json -from typing import TYPE_CHECKING, Any - -from sqlalchemy import LargeBinary -from sqlalchemy.types import TypeDecorator - -from langflow.services.database.models.flow_version.exceptions import FlowVersionSerializationError - -if TYPE_CHECKING: - from sqlalchemy.engine.interfaces import Dialect - -COMPRESS_LEVEL = 6 - - -def pack(data: dict[str, Any] | None) -> bytes | None: - if data is None: - return None - try: - encoded = json.dumps(data, separators=(",", ":"), ensure_ascii=False).encode("utf-8") - except (TypeError, ValueError) as exc: - msg = "Flow version data could not be serialized." - raise FlowVersionSerializationError(msg) from exc - return gzip.compress(encoded, COMPRESS_LEVEL) - - -def unpack(blob: bytes | None) -> dict[str, Any] | None: - if blob is None: - return None - try: - return json.loads(gzip.decompress(blob)) - except (OSError, EOFError, ValueError, UnicodeDecodeError) as exc: - msg = "Flow version data could not be read." - raise FlowVersionSerializationError(msg) from exc - - -class GzippedJSON(TypeDecorator): - impl = LargeBinary - cache_ok = True - - def process_bind_param(self, value: dict[str, Any] | None, dialect: Dialect) -> bytes | None: # noqa: ARG002 - return pack(value) - - def process_result_value(self, value: bytes | None, dialect: Dialect) -> dict[str, Any] | None: # noqa: ARG002 - return unpack(value) diff --git a/src/backend/tests/unit/alembic/test_compress_flow_version_data.py b/src/backend/tests/unit/alembic/test_compress_flow_version_data.py deleted file mode 100644 index a72f305bc8df..000000000000 --- a/src/backend/tests/unit/alembic/test_compress_flow_version_data.py +++ /dev/null @@ -1,121 +0,0 @@ -"""Tests for the flow-version payload compression migration.""" - -from __future__ import annotations - -import gzip -import importlib -import json - -import pytest -import sqlalchemy as sa -from alembic.migration import MigrationContext -from alembic.operations import Operations - -_MIGRATION = importlib.import_module("langflow.alembic.versions.d3b7c1e05f84_compress_flow_version_data") - -GRAPH = { - "nodes": [{"id": f"node-{index}", "data": {"code": "from lfx import Component\n" * 8}} for index in range(6)], - "edges": [], - "name": "Análise de Sentimento — ação", -} -POPULATED_ROWS = 5 - - -def _engine_before_the_migration(): - engine = sa.create_engine("sqlite://") - metadata = sa.MetaData() - sa.Table( - "flow_version", - metadata, - sa.Column("id", sa.String(), primary_key=True), - sa.Column("data", sa.JSON(), nullable=True), - sa.Column("version_number", sa.Integer(), nullable=False), - ) - metadata.create_all(engine) - with engine.begin() as conn: - for index in range(POPULATED_ROWS): - conn.execute( - sa.text("INSERT INTO flow_version (id, data, version_number) VALUES (:id, :data, :n)"), - {"id": f"row-{index}", "data": json.dumps(GRAPH), "n": index + 1}, - ) - conn.execute( - sa.text("INSERT INTO flow_version (id, data, version_number) VALUES (:id, NULL, :n)"), - {"id": "row-null", "n": POPULATED_ROWS + 1}, - ) - return engine - - -def _run(engine, direction: str) -> None: - with engine.begin() as conn: - original = _MIGRATION.op - try: - _MIGRATION.op = Operations(MigrationContext.configure(conn)) - getattr(_MIGRATION, direction)() - finally: - _MIGRATION.op = original - - -def _columns(engine) -> set[str]: - return {column["name"] for column in sa.inspect(engine).get_columns("flow_version")} - - -def test_upgrade_compresses_every_populated_row_and_keeps_null(monkeypatch): - monkeypatch.setattr(_MIGRATION, "BATCH_SIZE", 2) - engine = _engine_before_the_migration() - - _run(engine, "upgrade") - - assert _columns(engine) == {"id", "data_gz", "version_number"} - with engine.connect() as conn: - rows = conn.execute(sa.text("SELECT id, data_gz FROM flow_version ORDER BY version_number")).fetchall() - stored = {row.id: row.data_gz for row in rows} - assert stored["row-null"] is None - for index in range(POPULATED_ROWS): - payload = stored[f"row-{index}"] - assert payload[:2] == b"\x1f\x8b" - assert json.loads(gzip.decompress(payload)) == GRAPH - - -def test_downgrade_restores_the_original_json(monkeypatch): - monkeypatch.setattr(_MIGRATION, "BATCH_SIZE", 2) - engine = _engine_before_the_migration() - - _run(engine, "upgrade") - _run(engine, "downgrade") - - assert _columns(engine) == {"id", "data", "version_number"} - with engine.connect() as conn: - rows = conn.execute(sa.text("SELECT id, data FROM flow_version ORDER BY version_number")).fetchall() - stored = {row.id: row.data for row in rows} - assert stored["row-null"] is None - for index in range(POPULATED_ROWS): - assert json.loads(stored[f"row-{index}"]) == GRAPH - - -def test_upgrade_refuses_to_drop_the_column_when_a_row_was_not_migrated(monkeypatch): - engine = _engine_before_the_migration() - monkeypatch.setattr(_MIGRATION, "_rows", lambda *_: iter(())) - - with pytest.raises(RuntimeError, match="still hold data in data"): - _run(engine, "upgrade") - - assert "data" in _columns(engine) - - -def test_downgrade_refuses_to_drop_the_column_when_a_row_was_not_restored(monkeypatch): - engine = _engine_before_the_migration() - _run(engine, "upgrade") - monkeypatch.setattr(_MIGRATION, "_rows", lambda *_: iter(())) - - with pytest.raises(RuntimeError, match="still hold data in data_gz"): - _run(engine, "downgrade") - - assert "data_gz" in _columns(engine) - - -def test_upgrade_is_a_noop_without_the_table(): - engine = sa.create_engine("sqlite://") - - _run(engine, "upgrade") - - assert not sa.inspect(engine).has_table("flow_version") diff --git a/src/backend/tests/unit/services/database/models/flow_version/__init__.py b/src/backend/tests/unit/services/database/models/flow_version/__init__.py deleted file mode 100644 index e69de29bb2d1..000000000000 diff --git a/src/backend/tests/unit/services/database/models/flow_version/test_serialization.py b/src/backend/tests/unit/services/database/models/flow_version/test_serialization.py deleted file mode 100644 index 745a5f8fdf3a..000000000000 --- a/src/backend/tests/unit/services/database/models/flow_version/test_serialization.py +++ /dev/null @@ -1,80 +0,0 @@ -import gzip -import json - -import pytest -from langflow.services.database.models.flow_version.exceptions import FlowVersionSerializationError -from langflow.services.database.models.flow_version.serialization import pack, unpack -from sqlalchemy import Integer - -GRAPH = { - "nodes": [{"id": "node-1", "data": {"node": {"template": {"code": {"value": "from lfx import x\n"}}}}}], - "edges": [], - "viewport": {"x": 0, "y": 0, "zoom": 1}, -} - - -def test_pack_then_unpack_returns_an_equal_document(): - assert unpack(pack(GRAPH)) == GRAPH - - -def test_pack_returns_gzip_bytes(): - packed = pack(GRAPH) - - assert isinstance(packed, bytes) - assert json.loads(gzip.decompress(packed)) == GRAPH - - -def test_pack_shrinks_a_repetitive_graph(): - graph = {"nodes": [GRAPH["nodes"][0] for _ in range(50)], "edges": []} - - packed = pack(graph) - - assert len(packed) < len(json.dumps(graph).encode()) / 4 - - -def test_none_passes_through_both_ways(): - assert pack(None) is None - assert unpack(None) is None - - -def test_non_ascii_survives_the_round_trip(): - graph = {"nodes": [], "edges": [], "name": "Análise de Sentimento — ação"} - - assert unpack(pack(graph)) == graph - - -def test_unpack_rejects_data_that_is_not_gzip(): - with pytest.raises(FlowVersionSerializationError): - unpack(b"not gzip at all") - - -def test_unpack_rejects_gzip_that_is_not_json(): - with pytest.raises(FlowVersionSerializationError): - unpack(gzip.compress(b"")) - - -def test_pack_rejects_a_document_json_cannot_encode(): - with pytest.raises(FlowVersionSerializationError): - pack({"nodes": {object()}}) - - -def test_the_column_stores_gzip_bytes_and_reads_back_a_dict(): - from langflow.services.database.models.flow_version.serialization import GzippedJSON - from sqlalchemy import Column, MetaData, Table, create_engine, select - - metadata = MetaData() - table = Table("sample", metadata, Column("id", Integer, primary_key=True), Column("payload", GzippedJSON)) - engine = create_engine("sqlite://") - metadata.create_all(engine) - - with engine.begin() as conn: - conn.execute(table.insert().values(id=1, payload=GRAPH)) - conn.execute(table.insert().values(id=2, payload=None)) - - with engine.connect() as conn: - assert conn.execute(select(table.c.payload).where(table.c.id == 1)).scalar_one() == GRAPH - assert conn.execute(select(table.c.payload).where(table.c.id == 2)).scalar_one() is None - raw = conn.exec_driver_sql("SELECT payload FROM sample WHERE id = 1").scalar_one() - - assert raw[:2] == b"\x1f\x8b" - assert b"lfx" not in raw From 8b41f1992233f9fd7c5182df46d2a1d9bc1e06ad Mon Sep 17 00:00:00 2001 From: Tarcio Date: Fri, 4 Sep 2026 13:35:03 -0300 Subject: [PATCH 10/10] test(api): pin the streaming decision and the middleware position MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The description claimed build event streams were untouched because the library excludes text/event-stream. api/build.py streams application/x-ndjson, which is not excluded, so they are compressed — and minimum_size never applied to them either: gzip.py consults the floor only when more_body is false. Measured before deciding, per chunk with Z_SYNC_FLUSH as the middleware does: 300 small token events 17.6 KB -> 3.8 KB, 100 medium 47.7 KB -> 1.9 KB, 30 large 117.3 KB -> 1.0 KB, all under 0.7 ms. Only a stream carrying a single 36-byte event grows, 36 B -> 61 B. Keeping compression on ndjson is the decision; three tests pin it, including the excluded type staying uncompressed while streamed. The fourth test pins the registration order the design rests on. Verified it fails when the order inverts: moving the middleware above the BaseHTTPMiddleware layers turns two tests red, not zero. --- src/backend/base/langflow/main.py | 4 ++ .../unit/api/test_response_compression.py | 45 +++++++++++++++++++ 2 files changed, 49 insertions(+) diff --git a/src/backend/base/langflow/main.py b/src/backend/base/langflow/main.py index b83971e138b8..2205e499a2ea 100644 --- a/src/backend/base/langflow/main.py +++ b/src/backend/base/langflow/main.py @@ -80,6 +80,8 @@ MAX_PORT = 65535 GZIP_MINIMUM_SIZE = 1000 GZIP_COMPRESS_LEVEL = 6 +# application/x-ndjson is deliberately absent: build event streams compress 78-99% and, +# being streamed, bypass GZIP_MINIMUM_SIZE entirely. GZIP_ALREADY_COMPRESSED_CONTENT_TYPES = ( "application/octet-stream", "application/pdf", @@ -872,6 +874,8 @@ def create_app(): lifespan=lifespan, root_path=settings.root_path, ) + # Registered first so it sits innermost: the BaseHTTPMiddleware layers above turn every + # response into a stream, and a streamed response carries no Content-Length to test. app.add_middleware( GZipMiddleware, minimum_size=GZIP_MINIMUM_SIZE, diff --git a/src/backend/tests/unit/api/test_response_compression.py b/src/backend/tests/unit/api/test_response_compression.py index d2cfa7460ae2..ca6fe927a5c9 100644 --- a/src/backend/tests/unit/api/test_response_compression.py +++ b/src/backend/tests/unit/api/test_response_compression.py @@ -2,6 +2,7 @@ import pytest from fastapi import FastAPI, Response +from fastapi.responses import StreamingResponse from httpx import ASGITransport, AsyncClient from langflow.main import GZIP_COMPRESS_LEVEL, GZIP_EXCLUDED_CONTENT_TYPES, GZIP_MINIMUM_SIZE from starlette.middleware.gzip import GZipMiddleware @@ -104,6 +105,14 @@ def _app_with_the_same_middleware() -> FastAPI: async def payload(content_type: str) -> Response: return Response(content=b"x" * (GZIP_MINIMUM_SIZE * 4), media_type=content_type) + @app.get("/stream") + async def stream(content_type: str, events: int) -> StreamingResponse: + async def emit(): + for index in range(events): + yield (json.dumps({"event": "end_vertex", "id": index}) + "\n\n").encode() + + return StreamingResponse(emit(), media_type=content_type) + return app @@ -133,3 +142,39 @@ async def test_a_compressible_type_on_the_same_app_is_compressed(): ) assert response.headers["content-encoding"] == "gzip" + + +async def _stream_response(content_type: str, events: int): + transport = ASGITransport(app=_app_with_the_same_middleware()) + async with AsyncClient(transport=transport, base_url="http://testserver") as client: + return await client.get( + "/stream", + params={"content_type": content_type, "events": events}, + headers={"Accept-Encoding": "gzip"}, + ) + + +async def test_ndjson_event_streams_are_compressed(): + response = await _stream_response("application/x-ndjson", events=200) + + assert response.headers["content-encoding"] == "gzip" + assert "content-length" not in response.headers + + +async def test_a_stream_under_the_size_floor_is_compressed_like_any_other(): + response = await _stream_response("application/x-ndjson", events=1) + + assert len(response.content) < GZIP_MINIMUM_SIZE + assert response.headers["content-encoding"] == "gzip" + + +async def test_an_excluded_content_type_is_not_compressed_when_streamed(): + response = await _stream_response("text/event-stream", events=200) + + assert "content-encoding" not in response.headers + + +def test_gzip_is_registered_innermost(): + from langflow.main import create_app + + assert create_app().user_middleware[-1].cls is GZipMiddleware