perf(telemetry): batched off-pool writer for transactions + vertex_builds - #13126
Merged
Conversation
…ilds Adds TelemetryWriterService that buffers transaction and vertex_build rows in memory and drains them in batched INSERTs via a dedicated AsyncEngine (pool_size=1 for SQLite, 2 for Postgres, max_overflow=0). Retention is amortized in a 60s sweeper instead of running on every insert. Producers (log_transaction, log_vertex_build) enqueue instead of opening a DB session, so telemetry traffic no longer competes with the request-handling pool. Falls back to the legacy direct-write path via LANGFLOW_TELEMETRY_WRITER_ENABLED=false. Durability: in-flight rows spill to a diskcache.Deque per PID on shutdown and are restored on next startup; orphan PID directories left by crashed workers are adopted.
Contributor
|
Important Review skippedAuto incremental reviews are disabled on this repository. Please check the settings in the CodeRabbit UI or the ⚙️ Run configurationConfiguration used: Path: .coderabbit.yaml Review profile: CHILL Plan: Pro Run ID: You can disable this status message by setting the Use the checkbox below for a quick retry:
Important Pre-merge checks failedPlease resolve all errors before merging. Addressing warnings is optional. ❌ Failed checks (1 error, 2 warnings)
✅ Passed checks (6 passed)
✨ Finishing Touches🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
Review feedback from pr-review-toolkit + silent-failure-hunter: - Retention sweep snapshots the dirty-flow sets before commit and only clears them after it lands; a crashed sweep no longer drops the flows on the floor, so per-flow caps cannot drift unboundedly. - Producer fall-through is no longer silent. transaction_service and log_vertex_build each log a one-shot WARNING when telemetry_writer_enabled is True but the writer is not running. - Lifespan startup failure now logs ERROR instead of WARNING — if the user opted into the writer and it didn't come up, that's an error. - Shutdown drain timeout no longer suppressed silently: logs a WARNING with pending row count + a hint to raise telemetry_writer_shutdown_drain_s. - Writer's flush loop catches asyncio.CancelledError separately and re-prepends the in-flight batch to the buffer so teardown's disk spill catches it. - After 6 consecutive batch failures the writer emits a loud ERROR with buffer depths so operators see sustained data-loss risk. Added tests: - test_sanitization_survives_writer_round_trip - test_retention_failure_preserves_dirty_flows - test_in_flight_batch_returned_on_cancel
github-actions
Bot
added
performance
Member
|
I was having a few nights' worth of worrying for this one. I added a bit of load to the DB pool exhaustion with the recent features. Thanks for this, I'll review it. nice work @ogabrielluiz ! 👍🏼 |
Contributor
Author
|
@dkaushik94 this still needs some work ahha running the locust tests now. |
Contributor
There was a problem hiding this comment.
Pull request overview
This PR introduces a new TelemetryWriterService to batch and offload transaction and vertex_build writes onto a dedicated database engine/pool, reducing contention with the request-handling DB pool under heavy load. It also adds disk-backed spill/replay for durability across graceful restarts and provides unit + stress tooling to validate behavior.
Changes:
- Add
TelemetryWriterService(batching writer task + periodic retention sweeper + disk outbox recovery) and register/start it in the backend service lifecycle. - Route transaction and vertex_build producers to enqueue into the writer when enabled, falling back to legacy direct DB writes when unavailable.
- Add unit tests for the writer service and a manual stress harness (plus ruff per-file ignores) to reproduce/benchmark high-concurrency write load.
Reviewed changes
Copilot reviewed 15 out of 17 changed files in this pull request and generated 5 comments.
Show a summary per file
| File | Description |
|---|---|
| src/lfx/src/lfx/services/settings/base.py | Adds telemetry-writer configuration knobs (enablement, batching, outbox, shutdown drain). |
| src/lfx/src/lfx/services/schema.py | Registers TELEMETRY_WRITER_SERVICE in the LFX service enum. |
| src/lfx/src/lfx/graph/utils.py | Routes log_vertex_build through the telemetry writer when enabled/running, with legacy fallback. |
| src/backend/base/langflow/services/transaction/service.py | Routes log_transaction through telemetry writer when enabled/running, with legacy fallback + one-shot warning. |
| src/backend/base/langflow/services/telemetry_writer/service.py | Implements the new batched writer + sweeper + disk outbox adoption/restore. |
| src/backend/base/langflow/services/telemetry_writer/factory.py | Adds service factory for telemetry writer. |
| src/backend/base/langflow/services/telemetry_writer/init.py | Exposes telemetry writer service/factory module exports. |
| src/backend/base/langflow/services/utils.py | Registers telemetry writer factory with the service manager. |
| src/backend/base/langflow/services/schema.py | Registers TELEMETRY_WRITER_SERVICE in the backend service enum. |
| src/backend/base/langflow/services/deps.py | Adds get_telemetry_writer_service() dependency accessor. |
| src/backend/base/langflow/main.py | Starts the telemetry writer during app lifespan after service initialization. |
| src/backend/tests/unit/services/telemetry_writer/test_service.py | Adds end-to-end unit tests against in-memory SQLite for batching/retention/spill/replay semantics. |
| src/backend/tests/unit/services/telemetry_writer/init.py | Declares the unit test package for telemetry writer tests. |
| src/backend/tests/stress/stress_telemetry_writes.py | Adds manual stress harness for high-concurrency telemetry writes. |
| src/backend/tests/stress/README.md | Documents how to run/interpret the stress harness. |
| src/backend/tests/stress/init.py | Declares the stress test package. |
| pyproject.toml | Adds ruff per-file ignores for the stress harness directory. |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
github-actions
Bot
added
performance
- _restore_from_disk + _adopt_orphan_outboxes now route through _enqueue so a large disk-spilled or orphan outbox can't bypass telemetry_writer_max_queue and OOM the process. Oldest rows are dropped and counted via the existing dropped_transactions / dropped_vertex_builds counters. - chmod 0o700 the outbox root + per-PID directory so sanitized-but-still- sensitive payloads aren't exposed cross-user on multi-tenant hosts. Suppressed on platforms where chmod is a no-op (Windows). - Added test_adopt_orphan_outboxes_honors_max_queue. The remaining two copilot notes (private API access to DatabaseService and diskcache.Cache.close) were also flagged by the in-tree review; tracking separately. The "diskcache not declared" note is a false positive — the dependency is at src/backend/base/pyproject.toml:76.
github-actions
Bot
added
performance
ogabrielluiz
marked this pull request as ready for review
May 14, 2026 20:04
github-actions
Bot
added
performance
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
Contributor
There was a problem hiding this comment.
Actionable comments posted: 3
🧹 Nitpick comments (5)
src/backend/tests/unit/services/telemetry_writer/test_service.py (1)
112-122: ⚡ Quick winAdd short docstrings to tests that currently lack one.
Several test functions here are missing docstrings; adding one-line purpose statements improves failure triage and keeps consistency with the backend test standard.
As per coding guidelines: "Each test should have a clear docstring explaining its purpose; complex test setups should be commented; mock usage should be documented; expected behaviors should be explicitly stated".
Also applies to: 125-133, 144-193, 219-241, 266-295, 323-424
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/backend/tests/unit/services/telemetry_writer/test_service.py` around lines 112 - 122, Several test functions (notably test_is_enabled_reads_settings and test_enqueue_when_not_running_returns_false and other tests in the file ranges indicated) lack one-line docstrings; add a concise single-line docstring at the top of each test function describing its purpose (e.g., "Verify writer reads enabled setting" for test_is_enabled_reads_settings, "Ensure enqueue returns False when writer not running" for test_enqueue_when_not_running_returns_false) to comply with the backend test standard and improve failure triage; apply the same pattern to the other test functions in the indicated ranges (125-133, 144-193, 219-241, 266-295, 323-424).src/backend/base/langflow/services/telemetry_writer/service.py (4)
359-376: 💤 Low valueWriter reads
batch_size/flush_intervalonly once at task start.Both are read at Lines 362–363 and never refreshed, while
_run_sweeperre-readstelemetry_writer_cleanup_interval_severy iteration (Line 434) and_enqueuere-readstelemetry_writer_max_queueevery call. As a result, runtime changes to these two settings only take effect on a process restart, which is surprising relative to neighboring code. Hoisting the reads inside the loop (or caching them on the service and refreshing on each iteration) would make the behavior uniform.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/backend/base/langflow/services/telemetry_writer/service.py` around lines 359 - 376, The loop in _run_writer currently reads telemetry_writer_batch_size and telemetry_writer_flush_interval_s only once at the top, so runtime changes aren't picked up; move the retrieval of batch_size and flush_interval into the while True loop (or read them from a cached attribute that is refreshed each iteration) so they are re-read each iteration from self.settings_service.settings (consistent with how _run_sweeper and _enqueue re-check settings). Update _run_writer to obtain batch_size = int(getattr(self.settings_service.settings, "telemetry_writer_batch_size", 200)) and flush_interval = float(getattr(self.settings_service.settings, "telemetry_writer_flush_interval_s", 0.5)) inside the loop before using them.
389-407: ⚡ Quick winFailure-escalation log fires exactly once and then stays silent.
if consecutive_failures == _FAILURE_ESCALATION_THRESHOLD:triggers at exactly 6. If the underlying DB stays broken (or the writer keeps failing for hours), every subsequent failure quietly incrementsfailed_batchesand re-applies backoff with no further ERROR signal — defeating the stated purpose of "operators see sustained data-loss risk." Consider either escalating on every Nth failure after the threshold, or upgrading the per-batch log to ERROR once past it:♻️ Proposed escalation tweak
- if consecutive_failures == _FAILURE_ESCALATION_THRESHOLD: - logger.error( - f"telemetry_writer: {consecutive_failures} consecutive batch failures, " - f"buffer depth tx={len(self._tx_buffer)} vb={len(self._vb_buffer)}" - ) + if ( + consecutive_failures == _FAILURE_ESCALATION_THRESHOLD + or ( + consecutive_failures > _FAILURE_ESCALATION_THRESHOLD + and consecutive_failures % _FAILURE_ESCALATION_THRESHOLD == 0 + ) + ): + logger.error( + f"telemetry_writer: {consecutive_failures} consecutive batch failures, " + f"buffer depth tx={len(self._tx_buffer)} vb={len(self._vb_buffer)}" + )🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/backend/base/langflow/services/telemetry_writer/service.py` around lines 389 - 407, The escalation ERROR log only fires once because the code checks if consecutive_failures == _FAILURE_ESCALATION_THRESHOLD; change this to either trigger for every failure beyond the threshold (use consecutive_failures >= _FAILURE_ESCALATION_THRESHOLD) or log at ERROR every Nth failure past the threshold (e.g., if consecutive_failures >= _FAILURE_ESCALATION_THRESHOLD and consecutive_failures % N == 0) and optionally promote the per-batch logger.exception call to logger.error when consecutive_failures is past the threshold; update the check near the exception handling block that references consecutive_failures, _FAILURE_ESCALATION_THRESHOLD, logger.exception/logger.error and failed_batches so operators continue to see ERRORs while failures persist.
173-192: 💤 Low valueRemove the redundant
self._writer_task.cancel()call in the timeout handler.When
asyncio.wait_for()times out, it automatically cancels the inner task before raisingTimeoutError. The explicitcancel()at line 187 is a no-op; the subsequentawait self._writer_task(withsuppress(asyncio.CancelledError)) completes the cancellation. Remove the explicitcancel()to align the timeout path with the explicit-cancel exception handler below it.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/backend/base/langflow/services/telemetry_writer/service.py` around lines 173 - 192, The timeout handler in the shutdown path redundantly calls self._writer_task.cancel() after asyncio.wait_for(self._writer_task, timeout=drain_timeout) raises asyncio.TimeoutError (wait_for has already cancelled the inner task); remove the explicit cancel() call in that except asyncio.TimeoutError block so the code mirrors the CancelledError handler and simply awaits self._writer_task inside the suppress(asyncio.CancelledError) block while keeping the warning log (refer to self._writer_task, drain_timeout, and the telemetry_writer_shutdown_drain_s setting in the message).
468-526: 🏗️ Heavy liftCollapse per-flow and per-vertex DELETEs into single windowed query to reduce transaction scope and contention on dedicated SQLite writer connection.
The retention pass issues multiple per-flow DELETEs for transactions and multiple per-(flow, vertex) DELETEs for vertex_builds, all within a single transaction and committed at the end. With
pool_size=1for SQLite, this extends the dedicated writer connection's hold, potentially blocking concurrent writes. In deployments with many flows/vertices, this can result in hundreds of database round-trips and multi-second transaction lock times.Suggested approach:
- For transactions, replace the per-flow loop with a single DELETE using a CTE +
ROW_NUMBER() OVER (PARTITION BY flow_id ORDER BY timestamp DESC), keeping only ranks ≤max_transactions.- Apply the same pattern for vertex_builds with
PARTITION BY flow_id, id.- Remove the inner
SELECT DISTINCT idloop once the per-(flow, vertex) DELETE is windowed.- Window functions are already used in Langflow migrations (e.g., recent migration 1b8b740a6fa3) and work across SQLite, PostgreSQL, and MySQL.
Alternatively, commit per-flow or in batches to reduce the duration of a single transaction.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/backend/base/langflow/services/telemetry_writer/service.py` around lines 468 - 526, The current retention logic holds a single writer transaction while issuing many per-flow and per-(flow,vertex) DELETEs (loops over tx_flows and vb_flows using TransactionTable and VertexBuildTable), which causes contention; replace those loops with single windowed DELETEs using a CTE with ROW_NUMBER() OVER (PARTITION BY ...) to keep only the top N rows instead of per-flow deletes: for transactions use PARTITION BY flow_id ORDER BY timestamp DESC and filter row_number > max_transactions; for vertex builds use PARTITION BY flow_id, id ORDER BY timestamp DESC, build_id DESC and filter row_number > max_per_vertex (also remove the select distinct id loop and the vertex_ids iteration), and for the global limit use a similar windowed CTE over build_id to enforce max_vertex_builds; perform these DELETEs in one short transaction via self._session_maker() (or alternatively flush/commit in batches) to reduce lock time and round-trips.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@src/backend/base/langflow/services/telemetry_writer/service.py`:
- Around line 415-427: The sweeper/flush race can remove newly-added flow_ids
from the shared sets because _flush adds to _dirty_tx_flows/_dirty_vb_flows
before await session.commit(), and _run_retention_pass snapshots then subtracts
them after commit; fix by changing _run_retention_pass to "capture-and-clear"
the live sets at the start (e.g., copy current _dirty_tx_flows/_dirty_vb_flows
into local variables and clear the original sets) and only restore any
unprocessed ids on failure, and ensure that after successful commit you do not
remove ids that were added after the snapshot; alternatively (recommended) also
move the additions in _flush (the calls that add flow_id when inserting into
TransactionTable/VertexBuildTable) to after await session.commit() so new dirty
marks occur only after a successful commit.
In `@src/backend/tests/stress/README.md`:
- Around line 28-30: The example uses an unset env variable
DB_URL="$LANGFLOW_DATABASE_URL"; replace that with a concrete Postgres DSN
matching the container started above (e.g. export
DB_URL="postgresql://user:password@localhost:5432/langflow_test") so the uv run
python src/backend/tests/stress/stress_telemetry_writes.py --concurrency 200
--seconds 15 command can run in a clean shell; update the README.md snippet to
show the explicit DSN instead of LANGFLOW_DATABASE_URL and keep the rest of the
example (the uv run invocation and flags) unchanged.
In `@src/backend/tests/unit/services/telemetry_writer/test_service.py`:
- Around line 244-250: The unbounded PID-probing loops using variable dead_pid
and function _pid_alive can hang CI; add a maximum probe count (e.g.,
MAX_PROBES) and increment a probe counter inside the loop that breaks and fails
the test with an explicit assertion or exception if the max is exceeded, then
proceed to set dead_dir = tmp_path / str(dead_pid); apply the same bounded-probe
pattern to the other loop referenced (the one around lines 276-277) so both
loops use _pid_alive with a capped number of attempts and a clear failure path.
---
Nitpick comments:
In `@src/backend/base/langflow/services/telemetry_writer/service.py`:
- Around line 359-376: The loop in _run_writer currently reads
telemetry_writer_batch_size and telemetry_writer_flush_interval_s only once at
the top, so runtime changes aren't picked up; move the retrieval of batch_size
and flush_interval into the while True loop (or read them from a cached
attribute that is refreshed each iteration) so they are re-read each iteration
from self.settings_service.settings (consistent with how _run_sweeper and
_enqueue re-check settings). Update _run_writer to obtain batch_size =
int(getattr(self.settings_service.settings, "telemetry_writer_batch_size", 200))
and flush_interval = float(getattr(self.settings_service.settings,
"telemetry_writer_flush_interval_s", 0.5)) inside the loop before using them.
- Around line 389-407: The escalation ERROR log only fires once because the code
checks if consecutive_failures == _FAILURE_ESCALATION_THRESHOLD; change this to
either trigger for every failure beyond the threshold (use consecutive_failures
>= _FAILURE_ESCALATION_THRESHOLD) or log at ERROR every Nth failure past the
threshold (e.g., if consecutive_failures >= _FAILURE_ESCALATION_THRESHOLD and
consecutive_failures % N == 0) and optionally promote the per-batch
logger.exception call to logger.error when consecutive_failures is past the
threshold; update the check near the exception handling block that references
consecutive_failures, _FAILURE_ESCALATION_THRESHOLD,
logger.exception/logger.error and failed_batches so operators continue to see
ERRORs while failures persist.
- Around line 173-192: The timeout handler in the shutdown path redundantly
calls self._writer_task.cancel() after asyncio.wait_for(self._writer_task,
timeout=drain_timeout) raises asyncio.TimeoutError (wait_for has already
cancelled the inner task); remove the explicit cancel() call in that except
asyncio.TimeoutError block so the code mirrors the CancelledError handler and
simply awaits self._writer_task inside the suppress(asyncio.CancelledError)
block while keeping the warning log (refer to self._writer_task, drain_timeout,
and the telemetry_writer_shutdown_drain_s setting in the message).
- Around line 468-526: The current retention logic holds a single writer
transaction while issuing many per-flow and per-(flow,vertex) DELETEs (loops
over tx_flows and vb_flows using TransactionTable and VertexBuildTable), which
causes contention; replace those loops with single windowed DELETEs using a CTE
with ROW_NUMBER() OVER (PARTITION BY ...) to keep only the top N rows instead of
per-flow deletes: for transactions use PARTITION BY flow_id ORDER BY timestamp
DESC and filter row_number > max_transactions; for vertex builds use PARTITION
BY flow_id, id ORDER BY timestamp DESC, build_id DESC and filter row_number >
max_per_vertex (also remove the select distinct id loop and the vertex_ids
iteration), and for the global limit use a similar windowed CTE over build_id to
enforce max_vertex_builds; perform these DELETEs in one short transaction via
self._session_maker() (or alternatively flush/commit in batches) to reduce lock
time and round-trips.
In `@src/backend/tests/unit/services/telemetry_writer/test_service.py`:
- Around line 112-122: Several test functions (notably
test_is_enabled_reads_settings and test_enqueue_when_not_running_returns_false
and other tests in the file ranges indicated) lack one-line docstrings; add a
concise single-line docstring at the top of each test function describing its
purpose (e.g., "Verify writer reads enabled setting" for
test_is_enabled_reads_settings, "Ensure enqueue returns False when writer not
running" for test_enqueue_when_not_running_returns_false) to comply with the
backend test standard and improve failure triage; apply the same pattern to the
other test functions in the indicated ranges (125-133, 144-193, 219-241,
266-295, 323-424).
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: CHILL
Plan: Pro
Run ID: 1f206269-2a31-4c75-847d-32ff1c40b645
📒 Files selected for processing (17)
pyproject.tomlsrc/backend/base/langflow/main.pysrc/backend/base/langflow/services/deps.pysrc/backend/base/langflow/services/schema.pysrc/backend/base/langflow/services/telemetry_writer/__init__.pysrc/backend/base/langflow/services/telemetry_writer/factory.pysrc/backend/base/langflow/services/telemetry_writer/service.pysrc/backend/base/langflow/services/transaction/service.pysrc/backend/base/langflow/services/utils.pysrc/backend/tests/stress/README.mdsrc/backend/tests/stress/__init__.pysrc/backend/tests/stress/stress_telemetry_writes.pysrc/backend/tests/unit/services/telemetry_writer/__init__.pysrc/backend/tests/unit/services/telemetry_writer/test_service.pysrc/lfx/src/lfx/graph/utils.pysrc/lfx/src/lfx/services/schema.pysrc/lfx/src/lfx/services/settings/base.py
Comment on lines
+415
to
+427
| if tx_batch: | ||
| await session.execute(TransactionTable.__table__.insert(), tx_batch) | ||
| for row in tx_batch: | ||
| flow_id = row.get("flow_id") | ||
| if flow_id is not None: | ||
| self._dirty_tx_flows.add(str(flow_id)) | ||
| if vb_batch: | ||
| await session.execute(VertexBuildTable.__table__.insert(), vb_batch) | ||
| for row in vb_batch: | ||
| flow_id = row.get("flow_id") | ||
| if flow_id is not None: | ||
| self._dirty_vb_flows.add(str(flow_id)) | ||
| await session.commit() |
Contributor
There was a problem hiding this comment.
Sweeper-vs-flush race can silently drop dirty flow_ids.
_run_retention_pass snapshots _dirty_*_flows at the start (Lines 463–466) and clears the snapshot from the live set via -= after commit (Lines 535–536). Meanwhile, _flush adds flow_ids to the same sets before its commit (Lines 420, 426). When a flush's dirty.add(...) interleaves between the snapshot and the post-commit -=, the freshly-flushed flow_id is removed from dirty even though its rows weren't accounted for by the just-finished retention pass. In single-threaded asyncio this window opens every time the sweeper yields on await session.commit() (Line 526) while a concurrent flush is in flight.
Effect: per-flow cap can overshoot by up to one flush's worth of rows for the affected flow, until that flow sees new activity (which re-marks it dirty). Bounded, self-healing — but a correctness wart worth removing.
A cleaner pattern is "capture-and-clear" at the start of the sweep, and restore-on-failure only:
♻️ Proposed dirty-set handover
- tx_flow_snapshot = set(self._dirty_tx_flows)
- vb_flow_snapshot = set(self._dirty_vb_flows)
+ # Hand off ownership of the current dirty set to this sweep. New
+ # flush activity during the sweep accumulates into a fresh set and
+ # is picked up on the next pass.
+ tx_flow_snapshot = set(self._dirty_tx_flows)
+ vb_flow_snapshot = set(self._dirty_vb_flows)
+ self._dirty_tx_flows -= tx_flow_snapshot
+ self._dirty_vb_flows -= vb_flow_snapshot
tx_flows = [_as_uuid(f) for f in tx_flow_snapshot]
vb_flows = [_as_uuid(f) for f in vb_flow_snapshot]
@@
await session.commit()
except Exception:
- # Sweep failed before commit — re-mark the snapshot as dirty so the
- # next sweep retries these flows. Without this the per-flow caps
- # could overshoot indefinitely until those flows see new writes.
+ # Sweep failed before commit — restore the handed-off snapshot
+ # so the next sweep retries these flows.
self._dirty_tx_flows |= tx_flow_snapshot
self._dirty_vb_flows |= vb_flow_snapshot
raise
- else:
- self._dirty_tx_flows -= tx_flow_snapshot
- self._dirty_vb_flows -= vb_flow_snapshotOptionally, moving _dirty_*_flows.add(...) in _flush to run after await session.commit() succeeds also reduces dirty noise when commits fail (the rows are re-flushed anyway, which will re-mark them).
Also applies to: 463-466, 535-536
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@src/backend/base/langflow/services/telemetry_writer/service.py` around lines
415 - 427, The sweeper/flush race can remove newly-added flow_ids from the
shared sets because _flush adds to _dirty_tx_flows/_dirty_vb_flows before await
session.commit(), and _run_retention_pass snapshots then subtracts them after
commit; fix by changing _run_retention_pass to "capture-and-clear" the live sets
at the start (e.g., copy current _dirty_tx_flows/_dirty_vb_flows into local
variables and clear the original sets) and only restore any unprocessed ids on
failure, and ensure that after successful commit you do not remove ids that were
added after the snapshot; alternatively (recommended) also move the additions in
_flush (the calls that add flow_id when inserting into
TransactionTable/VertexBuildTable) to after await session.commit() so new dirty
marks occur only after a successful commit.
Comment on lines
+28
to
+30
| export DB_URL="$LANGFLOW_DATABASE_URL" # any standard postgres DSN works | ||
| uv run python src/backend/tests/stress/stress_telemetry_writes.py \ | ||
| --concurrency 200 --seconds 15 |
Contributor
There was a problem hiding this comment.
Use a concrete Postgres DSN in the example.
Line 28 currently derives DB_URL from LANGFLOW_DATABASE_URL, which may be unset in a clean shell and makes this runbook step fail. Use an explicit DSN matching the container started above.
Suggested doc fix
-export DB_URL="$LANGFLOW_DATABASE_URL" # any standard postgres DSN works
+export DB_URL="postgresql+psycopg://langflow:langflow@localhost:55432/langflow"
+# (or any standard postgres DSN)
uv run python src/backend/tests/stress/stress_telemetry_writes.py \
--concurrency 200 --seconds 15📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
Suggested change
| export DB_URL="$LANGFLOW_DATABASE_URL" # any standard postgres DSN works | |
| uv run python src/backend/tests/stress/stress_telemetry_writes.py \ | |
| --concurrency 200 --seconds 15 | |
| export DB_URL="postgresql+psycopg://langflow:langflow@localhost:55432/langflow" | |
| # (or any standard postgres DSN) | |
| uv run python src/backend/tests/stress/stress_telemetry_writes.py \ | |
| --concurrency 200 --seconds 15 |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@src/backend/tests/stress/README.md` around lines 28 - 30, The example uses an
unset env variable DB_URL="$LANGFLOW_DATABASE_URL"; replace that with a concrete
Postgres DSN matching the container started above (e.g. export
DB_URL="postgresql://user:password@localhost:5432/langflow_test") so the uv run
python src/backend/tests/stress/stress_telemetry_writes.py --concurrency 200
--seconds 15 command can run in a clean shell; update the README.md snippet to
show the explicit DSN instead of LANGFLOW_DATABASE_URL and keep the rest of the
example (the uv run invocation and flags) unchanged.
Comment on lines
+244
to
+250
| while dead_pid > 0: | ||
| from langflow.services.telemetry_writer.service import _pid_alive | ||
|
|
||
| if not _pid_alive(dead_pid): | ||
| break | ||
| dead_pid += 1 | ||
| dead_dir = tmp_path / str(dead_pid) |
Contributor
There was a problem hiding this comment.
Bound dead-PID probing loops to prevent hung CI runs.
Both PID search loops are unbounded; if _pid_alive behaves unexpectedly on a runner, these tests can stall indefinitely. Add a bounded probe with a clear failure path.
Proposed patch
+def _find_dead_pid(start: int = 99_999, max_checks: int = 50_000) -> int:
+ from langflow.services.telemetry_writer.service import _pid_alive
+
+ pid = start
+ for _ in range(max_checks):
+ if not _pid_alive(pid):
+ return pid
+ pid += 1
+ pytest.fail("Unable to find an unused PID for orphan outbox setup")
+
def test_adopt_orphan_outboxes(tmp_path: Path) -> None:
# Simulate a dead worker that left rows in a sibling outbox.
- dead_pid = 99999
- while dead_pid > 0:
- from langflow.services.telemetry_writer.service import _pid_alive
-
- if not _pid_alive(dead_pid):
- break
- dead_pid += 1
+ dead_pid = _find_dead_pid()
@@
def test_adopt_orphan_outboxes_honors_max_queue(tmp_path: Path) -> None:
@@
- from langflow.services.telemetry_writer.service import _pid_alive
-
- dead_pid = 99999
- while _pid_alive(dead_pid):
- dead_pid += 1
+ dead_pid = _find_dead_pid()Also applies to: 276-277
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@src/backend/tests/unit/services/telemetry_writer/test_service.py` around
lines 244 - 250, The unbounded PID-probing loops using variable dead_pid and
function _pid_alive can hang CI; add a maximum probe count (e.g., MAX_PROBES)
and increment a probe counter inside the loop that breaks and fails the test
with an explicit assertion or exception if the max is exceeded, then proceed to
set dead_dir = tmp_path / str(dead_pid); apply the same bounded-probe pattern to
the other loop referenced (the one around lines 276-277) so both loops use
_pid_alive with a capped number of attempts and a clear failure path.
- Sweeper hands off dirty sets via capture-and-clear so concurrent flushes during a retention pass aren't wiped by the post-commit subtract; failure path restores the snapshot. - Stress README uses a concrete Postgres DSN matching the docker example instead of an unset env var. - Test PID-probe loops bounded via a shared helper with pytest.fail fallback.
github-actions
Bot
added
performance
Codecov Report❌ Patch coverage is Additional details and impacted files@@ Coverage Diff @@
## release-1.10.0 #13126 +/- ##
==================================================
+ Coverage 58.13% 58.31% +0.17%
==================================================
Files 2286 2288 +2
Lines 218405 219061 +656
Branches 32850 32922 +72
==================================================
+ Hits 126965 127735 +770
+ Misses 89985 89871 -114
Partials 1455 1455
Flags with carried forward coverage won't be shown. Click here to find out more.
🚀 New features to boost your workflow:
|
Contributor
Frontend Unit Test Coverage ReportCoverage Summary
Unit Test Results
|
Contributor
✅ Test Coverage AdvisorNo source changes detected without accompanying tests. Thanks for keeping coverage up! 🎉
|
Today the writer bounds memory by row count only, so a worker logging fat
vertex_build artifacts can hold tens of MB per row and silently dwarf the
configured max_queue. Add a size_strategy switch ('count' | 'bytes' |
'either', default 'count' for parity) with two byte thresholds:
- batch_size_bytes (256KB) caps per-flush INSERT size when bytes apply
- max_queue_bytes (200MB) drops oldest by bytes when bytes apply
Parallel sizes deques mirror the payload buffers so accounting stays
consistent across drain, spill, restore, and the cancel/retry rollback.
Single rows above the byte budget are still emitted (no row is refused);
operators get dropped_*_bytes counters alongside dropped_*.
github-actions
Bot
added
performance
…d paths Address gaps in the byte-aware strategy tests: - pin exact dropped counts and remaining buffer state for the drop-by-bytes case (was 'greater than zero') - compute exact drain count from encoded size (was a 1-4 range) - round-trip bytes through spill+restore to verify size deque is rebuilt - round-trip bytes through orphan adoption for the same reason - exercise the sweeper loop end-to-end to confirm heartbeat + prune are wired in (previously only unit-tested in isolation) Clarify in the settings docs that the byte caps measure encoded JSON size, not Python in-memory footprint.
github-actions
Bot
added
performance
# Conflicts: # .secrets.baseline # src/lfx/src/lfx/_assets/component_index.json
github-actions
Bot
added
performance
* fix(telemetry-writer): harden error handling and add missing test coverage Critical: - teardown() now re-raises CancelledError after awaiting the writer task so the asyncio cancellation chain propagates correctly on lifespan task kill - suppress(OSError) on owner file write replaced with explicit error log so operators know when disk-spilled rows will not be recoverable on restart High / important: - Escalation threshold check changed from == to >= so the error log fires on every failure past the threshold, not just the 6th - Dead `except OSError` on time.time() - time.monotonic() changed to `except Exception` since time functions cannot raise OSError - Removed redundant `from uuid import UUID` inside _run_retention_pass (already imported at module top) - Orphan directory cleanup: replaced blanket suppress(OSError) with per-child suppress so ENOTEMPTY on an individual rmdir doesn't abort the whole loop and leave the parent directory leaking silently; outer failure now logs at debug Tests (3 new): - test_retention_sweep_caps_vertex_builds_per_vertex: inserts 8 builds for a single vertex with max_per_vertex=3 so the per-vertex DELETE subquery actually executes (previous test used 8 distinct vertex IDs, bypassing it) - test_either_strategy_trips_on_bytes_first: verifies bytes can be the first trigger under 'either' strategy (previous test only covered the count-first path) - test_writer_retries_on_batch_failure: injects 2 flush failures then success; confirms failed_batches increments, rows are preserved in the buffer, and flushed_rows reflects the final successful write * ci: add stress-tests job to nightly build pipeline Wires the stress-tests workflow into nightly so telemetry write stress tests run automatically. Also gates release-nightly-build and slack-notification on stress-tests result so a stress regression blocks the nightly release and surfaces in Slack. * ci: run stress tests in nightly without blocking release Stress tests are informational for now — a failure is visible in the workflow run and Slack but does not gate the nightly release or build. * fix(telemetry-writer): address PR review on cancelled teardown and nightly stress tests - Add the stress-tests.yml reusable workflow (was untracked, so the nightly stress-tests job could never resolve) - Notify Slack on stress-tests failure (add to slack-notification gate and FAILED_JOB detection as non-blocking) - Run sweeper cancel, disk spill, and engine dispose in a finally so a cancelled teardown() still persists the in-memory buffer - Add tests for the cancelled-teardown spill path and the >= escalation threshold; drop the misleading wait-loop in the retry test
github-actions
Bot
added
performance
jordanrfrazier
left a comment
jordanrfrazier
left a comment
Collaborator
There was a problem hiding this comment.
lgtm. ran some tests, operation looks good. Didn't run any load tests.
jordanrfrazier
approved these changes
Jun 4, 2026
# Conflicts: # .github/workflows/nightly_build.yml # .secrets.baseline # src/lfx/src/lfx/services/schema.py
github-actions
Bot
added
performance
ogabrielluiz
enabled auto-merge
June 5, 2026 17:07
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Problem
Heavy load on the
transactionandvertex_buildtables saturates the request-handling DB pool and triggers row-level deadlocks during retention, causing:database is lockederrorsQueuePool ... timed outANDpsycopg.errors.DeadlockDetectedon the per-row retention DELETEToday the workaround is
LANGFLOW_TRANSACTIONS_STORAGE_ENABLED=falsein production, which loses all execution history.Even though the writes are dispatched off the request path via
BackgroundTasks, each background task still acquires a connection from the same pool the request handlers use, plus runs an INSERT + 1-2 retention DELETEs per row. The per-row DELETE-older query is what produces the Postgres deadlocks under concurrent load — many transactions racing for locks on the same older rows.Approach
New
TelemetryWriterService:log_transaction,log_vertex_build) enqueue rows into an in-memorycollections.dequeinstead of opening a DB session.AsyncEngine(pool_size=1 SQLite, 2 Postgres,max_overflow=0). That pool is structurally isolated — it cannot starve the request pool under any load.max_transactions_to_keep/max_vertex_builds_to_keep/max_vertex_builds_per_vertexcaps, just amortized). Retention runs once per cycle on one connection — deadlock is structurally impossible.outbox.sqlite, WAL mode, JSON payloads in a TEXT column,PRAGMA synchronous=FULLon the spill connection); on next startup that PID dir is restored and orphan PID dirs from crashed workers are adopted only if theirowner.jsonmatches the current host + boot identity.LANGFLOW_TELEMETRY_WRITER_ENABLED(legacy direct-write path stays as escape hatch).Why SQLite instead of
diskcacheThis branch originally used
diskcache.Dequefor the on-disk outbox. After review, two reasons to swap:diskcachedeserializes withpickleby default. An attacker with write access to the cache directory can plant a malicious payload that triggers RCE on read. Our outbox lives under/tmp/langflow_telemetry_outbox/<pid>and we deliberately adopt sibling-PID directories on startup — exactly the vulnerable pattern.diskcachewas not listed inpyproject.toml; it was only resolved transitively for Python 3.11–3.13 viaembedchain/unitxt. On Python 3.10 (a supported version) the import fails outright.Replacement is stdlib
sqlite3in WAL mode with JSON-in-TEXT payloads, encapsulated in a small_Outboxhelper:datetimeandUUIDacross the round-trip so SQLAlchemy's typed columns accept restored payloads.append_allencodes the whole buffer up front and only clears the deque after the transaction commits, so a mid-flight SQLite failure cannot half-drain the buffer.Hardening informed by industry practice
A survey of OpenTelemetry Collector, Fluent Bit, Vector, Prometheus remote_write, Datadog agent, Logstash, and Filebeat highlighted three gaps we closed:
owner.jsonat start (hostname + Linuxboot_id, ortime() - monotonic()as a portable fallback). On adoption we only proceed when host+boot match; cross-host or pre-owner dirs are logged and skipped. This prevents a recycled PID after a container restart from pulling in a stranger's spill data.PRAGMA synchronous=FULLon the spill connection.NORMALonly fsyncs on WAL checkpoint, which may never run if the process exits immediately after commit.FULLguarantees the shutdown commit hits the platter.telemetry_writer_max_queuewith drop-oldest. A backlogged buffer at shutdown can neither stall teardown nor fill disk; the spill cap matches the producer-side overflow policy.Why this is structurally safe
Baseline: direct stress harness (writer disabled)
200 concurrent worker coroutines, each calling
log_transaction+log_vertex_buildas fast as possible for 15s. Source:src/backend/tests/stress/stress_telemetry_writes.py.The "0 errors" only means the 30s pool/lock timeouts absorb the contention. The throughput collapse is the actual user-visible failure mode in production.
Results: direct stress harness (writer enabled)
Same harness, same scale, 30s drain grace.
All rows flushed to DB by end of run; retention caps still enforced by the sweeper.
Results: locust against running Langflow (Postgres 16, Basic Prompting flow)
End-to-end through HTTP
/api/v1/run/{flow_id}against a real Langflow backend with the defaultdb_connection_settings(pool_size=20, max_overflow=30). Source:src/backend/tests/locust/locustfile.py.Writer enabled
¹ Both failures were locust's own 30s client-side request timeout; not server-side.
Writer disabled (same 500 user / 90s load, head-to-head)
Same flow, same load, same DB. With the writer disabled, 258
psycopg.errors.DeadlockDetectederrors fire during the run — all on thetransactiontable's per-row retention DELETE. Throughput drops 36% and latency rises.Retention sweeper observed working in steady state: after a full 60s sweep cycle,
transactionrows trimmed from ~10k peak to exactly the configured 3000 cap,vertex_buildrows trimmed to exactly 200 (4 vertices × 50max_vertex_builds_per_vertex).Tradeoffs
sweep_interval × insert_raterows per flow above the cap, self-corrects within 60s.lfx/services/settings/base.py):telemetry_writer_enabled(defaultTrue),telemetry_writer_batch_size(200),telemetry_writer_flush_interval_s(0.5),telemetry_writer_cleanup_interval_s(60),telemetry_writer_max_queue(100k),telemetry_writer_outbox_dir(None → tmpdir),telemetry_writer_shutdown_drain_s(5.0).Files
src/backend/base/langflow/services/telemetry_writer/{__init__,service,factory}.pysrc/backend/tests/unit/services/telemetry_writer/test_service.py(20 end-to-end tests against in-memory SQLite, no mocking of persistence — including spill→restore→INSERT with realistic UUID/datetime payloads, owner-identity skip on cross-host and missing-owner dirs, spill cap drop-oldest, sanitization round-trip, retention failure preservation, and cancel-mid-flush recovery)src/backend/tests/stress/stress_telemetry_writes.py+ README (permanent regression test for this failure class)src/backend/base/langflow/services/transaction/service.py,src/lfx/src/lfx/graph/utils.pysrc/backend/base/langflow/main.pyServiceTypeenums,services/utils.py,services/deps.py,lfx/services/settings/base.pyObservability
When
telemetry_writer_enabled=True:telemetry_writer started (outbox=..., tx_pending=N, vb_pending=N).telemetry_writer_shutdown_drain_s.