perf(telemetry): batched off-pool writer for transactions + vertex_builds

.github/workflows/nightly_build.yml

@@ -283,11 +283,19 @@ jobs:
   #     python-versions: '["3.10", "3.11", "3.12", "3.13"]'
   #     ref: ${{ needs.create-nightly-tag.outputs.tag }}
 
+  stress-tests:
+    if: github.repository == 'langflow-ai/langflow' && !inputs.skip_backend_tests
+    name: Run Stress Tests
+    needs: [resolve-release-branch, create-nightly-tag]
+    uses: ./.github/workflows/stress-tests.yml
+    with:
+      ref: ${{ needs.resolve-release-branch.outputs.branch }}
+
   release-nightly-build:
     if: github.repository == 'langflow-ai/langflow' && always() && needs.create-nightly-tag.result == 'success' && needs.frontend-tests-linux.result != 'failure' && needs.backend-unit-tests.result != 'failure'
     name: Run Nightly Langflow Build
     needs:
-      [validate-inputs, create-nightly-tag, frontend-tests-linux, frontend-tests-windows, backend-unit-tests]
+      [validate-inputs, create-nightly-tag, frontend-tests-linux, frontend-tests-windows, backend-unit-tests, stress-tests]
     uses: ./.github/workflows/release_nightly.yml
     with:
       build_docker_base: true
@@ -316,12 +324,12 @@ jobs:
 
   slack-notification:
     name: Send Slack Notification
-    needs: [frontend-tests-linux, frontend-tests-windows, backend-unit-tests, release-nightly-build, db-migration-validation]
-    if: ${{ github.repository == 'langflow-ai/langflow' && !inputs.skip_slack && always() && (needs.release-nightly-build.result == 'failure' || needs.frontend-tests-linux.result == 'failure' || needs.frontend-tests-windows.result == 'failure' || needs.backend-unit-tests.result == 'failure' || needs.db-migration-validation.result == 'failure' || needs.release-nightly-build.result == 'success') }}
+    needs: [frontend-tests-linux, frontend-tests-windows, backend-unit-tests, stress-tests, release-nightly-build, db-migration-validation]
+    if: ${{ github.repository == 'langflow-ai/langflow' && !inputs.skip_slack && always() && (needs.release-nightly-build.result == 'failure' || needs.frontend-tests-linux.result == 'failure' || needs.frontend-tests-windows.result == 'failure' || needs.backend-unit-tests.result == 'failure' || needs.stress-tests.result == 'failure' || needs.db-migration-validation.result == 'failure' || needs.release-nightly-build.result == 'success') }}
     runs-on: ubuntu-latest
     steps:
       - name: Send failure notification to Slack
-        if: ${{ needs.release-nightly-build.result == 'failure' || needs.frontend-tests-linux.result == 'failure' || needs.frontend-tests-windows.result == 'failure' || needs.backend-unit-tests.result == 'failure' || needs.db-migration-validation.result == 'failure' }}
+        if: ${{ needs.release-nightly-build.result == 'failure' || needs.frontend-tests-linux.result == 'failure' || needs.frontend-tests-windows.result == 'failure' || needs.backend-unit-tests.result == 'failure' || needs.stress-tests.result == 'failure' || needs.db-migration-validation.result == 'failure' }}
         run: |
           # Determine which job failed
           FAILED_JOB="unknown"
@@ -333,6 +341,8 @@ jobs:
             FAILED_JOB="frontend-tests-windows (non-blocking)"
           elif [ "${{ needs.backend-unit-tests.result }}" == "failure" ]; then
             FAILED_JOB="backend-unit-tests"
+          elif [ "${{ needs.stress-tests.result }}" == "failure" ]; then
+            FAILED_JOB="stress-tests (non-blocking)"
           elif [ "${{ needs.db-migration-validation.result }}" == "failure" ]; then
             FAILED_JOB="db-migration-validation"
           fi

.github/workflows/stress-tests.yml

@@ -0,0 +1,56 @@
+name: Stress Tests
+
+on:
+  workflow_dispatch: {}
+  workflow_call:
+    inputs:
+      ref:
+        description: "Branch or tag to test"
+        required: false
+        type: string
+        default: ""
+
+jobs:
+  telemetry-writes:
+    name: Telemetry Writes (Postgres)
+    runs-on: ubuntu-latest
+
+    services:
+      postgres:
+        image: postgres:16
+        env:
+          POSTGRES_USER: langflow
+          POSTGRES_PASSWORD: langflow
+          POSTGRES_DB: langflow
+        ports:
+          - 5432:5432
+        options: >-
+          --health-cmd="pg_isready -U langflow"
+          --health-interval=10s
+          --health-timeout=5s
+          --health-retries=5
+
+    steps:
+      - name: Checkout code
+        uses: actions/checkout@v6
+        with:
+          ref: ${{ inputs.ref || github.sha }}
+
+      - name: Install uv
+        uses: astral-sh/setup-uv@v6
+        with:
+          enable-cache: true
+          cache-dependency-glob: "uv.lock"
+          python-version: "3.12"
+          prune-cache: false
+
+      - name: Install dependencies
+        run: uv sync --extra postgresql
+
+      - name: Run stress test
+        env:
+          DB_URL: "postgresql+psycopg://langflow:langflow@localhost:5432/langflow"  # pragma: allowlist secret
+          LANGFLOW_TELEMETRY_WRITER_ENABLED: "true"
+        run: |
+          uv run python src/backend/tests/stress/stress_telemetry_writes.py \
+            --concurrency 50 --seconds 15

.secrets.baseline

@@ -2352,6 +2352,29 @@
         "line_number": 774
       }
     ],
+    "src/backend/base/langflow/initial_setup/starter_projects/Pok\u00e9dex Agent.json": [
+      {
+        "type": "Hex High Entropy String",
+        "filename": "src/backend/base/langflow/initial_setup/starter_projects/Pok\u00e9dex Agent.json",
+        "hashed_secret": "54ed260e3bc31bc77ee06754dff850981d39a66c",
+        "is_verified": false,
+        "line_number": 115
+      },
+      {
+        "type": "Hex High Entropy String",
+        "filename": "src/backend/base/langflow/initial_setup/starter_projects/Pok\u00e9dex Agent.json",
+        "hashed_secret": "d6e6d7b4b115cd3b9d172623199f8c403055fecc",
+        "is_verified": false,
+        "line_number": 394
+      },
+      {
+        "type": "Hex High Entropy String",
+        "filename": "src/backend/base/langflow/initial_setup/starter_projects/Pok\u00e9dex Agent.json",
+        "hashed_secret": "ecdbe30d19e36761df3620b37270f23c8e3eaa4c",
+        "is_verified": false,
+        "line_number": 774
+      }
+    ],
     "src/backend/base/langflow/initial_setup/starter_projects/Portfolio Website Code Generator.json": [
       {
         "type": "Hex High Entropy String",

pyproject.toml

@@ -285,6 +285,13 @@ external = ["RUF027"]
 
 [tool.ruff.lint.per-file-ignores]
 "scripts/*" = ["D1", "INP", "T201"]
+"src/backend/tests/stress/*" = [
+    "D1",     # Docstrings: this is a manual CLI tool, not a test target
+    "INP001", # Stress tests live outside the unit test package
+    "S101",   # Asserts ok in stress harness
+    "S311",   # Standard PRNG is fine for stress payload jitter
+    "T201",   # Print statements: this is a CLI script
+]
 "scripts/gp/tests/*" = [
     "D1",      # Missing docstrings
     "PLR2004", # Magic value comparisons

src/backend/base/langflow/main.py

@@ -198,6 +198,23 @@ async def lifespan(_app: FastAPI):
             await initialize_services(fix_migration=fix_migration)
             await logger.adebug(f"Services initialized in {asyncio.get_event_loop().time() - start_time:.2f}s")
 
+            # Start the telemetry writer (no-op when telemetry_writer_enabled is False).
+            try:
+                from langflow.services.deps import get_telemetry_writer_service
+
+                telemetry_writer = get_telemetry_writer_service()
+                if telemetry_writer is not None and telemetry_writer.is_enabled():
+                    await telemetry_writer.start()
+            except Exception as exc:  # noqa: BLE001
+                # If the user explicitly opted in (telemetry_writer_enabled=True)
+                # but startup failed, this is an error not a warning — every
+                # subsequent write will silently fall back to the legacy direct-
+                # write path that this feature was built to replace.
+                await logger.aerror(
+                    f"Failed to start telemetry writer; transactions and vertex_build "
+                    f"writes will use the legacy direct-write path: {exc}"
+                )
+
             current_time = asyncio.get_event_loop().time()
             await logger.adebug("Setting up LLM caching")
             setup_llm_caching()

src/backend/base/langflow/services/deps.py

@@ -284,3 +284,10 @@ def get_memory_base_service():
     from langflow.services.memory_base.factory import MemoryBaseServiceFactory
 
     return get_service(ServiceType.MEMORY_BASE_SERVICE, MemoryBaseServiceFactory())
+
+
+def get_telemetry_writer_service():
+    """Return the TelemetryWriterService instance (always registered when langflow is installed)."""
+    from langflow.services.telemetry_writer.factory import TelemetryWriterServiceFactory
+
+    return get_service(ServiceType.TELEMETRY_WRITER_SERVICE, TelemetryWriterServiceFactory())

src/backend/base/langflow/services/schema.py

@@ -24,3 +24,4 @@ class ServiceType(str, Enum):
     JOB_SERVICE = "jobs_service"
     FLOW_EVENTS_SERVICE = "flow_events_service"
     MEMORY_BASE_SERVICE = "memory_base_service"
+    TELEMETRY_WRITER_SERVICE = "telemetry_writer_service"

src/backend/base/langflow/services/telemetry_writer/__init__.py

@@ -0,0 +1,12 @@
+"""Telemetry writer service.
+
+Async batched writer for transaction and vertex_build rows backed by a
+disk-persisted SQLite outbox and a dedicated database connection. Removes
+write-path contention from the request-handling connection pool so heavy load
+no longer triggers SQLite "database is locked" or PostgreSQL pool timeouts.
+"""
+
+from langflow.services.telemetry_writer.factory import TelemetryWriterServiceFactory
+from langflow.services.telemetry_writer.service import TelemetryWriterService
+
+__all__ = ["TelemetryWriterService", "TelemetryWriterServiceFactory"]

src/backend/base/langflow/services/telemetry_writer/factory.py

@@ -0,0 +1,20 @@
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+from typing_extensions import override
+
+from langflow.services.factory import ServiceFactory
+from langflow.services.telemetry_writer.service import TelemetryWriterService
+
+if TYPE_CHECKING:
+    from langflow.services.settings.service import SettingsService
+
+
+class TelemetryWriterServiceFactory(ServiceFactory):
+    def __init__(self) -> None:
+        super().__init__(TelemetryWriterService)
+
+    @override
+    def create(self, settings_service: SettingsService):
+        return TelemetryWriterService(settings_service)