WATCHER: buffer producer dbt events and write _dbt_event once per node

[pankajastro]

Problem

In ExecutionMode.WATCHER, the producer pushed the structured _dbt_event to XCom on every allowlisted dbt event (~4–5 per node — NodeStart/NodeCompiling/NodeExecuting/NodeFinished), each through the global xcom_set_lock. This is largely write-amplification: the consumer reads _dbt_event only once at terminal (since #2805), so the intermediate pushes are wasted and add lock contention to the multi-threaded dbt run. The per-event overwrite also let a trailing NodeFinished clobber a captured error message.

Part of the #2736 watcher performance regression; delivers the producer-write reduction tracked in #2777.

Change (local watcher producer)

• Buffer in memory, flush once at terminal. Accumulate each node's events in an in-process per-node buffer (node_event_buffer) and write _dbt_event to XCom only when the node reaches a terminal state — ~1 write per success node, ~2 per failure — instead of one per event.
• Reliable error message. An error message wins and is never overwritten by a later (often empty) message; subprocess mode's RunResultError — which arrives after the terminal event — triggers a second flush; the literal "None" status is ignored; the authoritative terminal status is stamped on flush.
• Thread-safe. A lock guards the buffer, since dbt-runner callbacks fire from multiple threads.
• Backward compatible. Callers that don't pass a buffer (e.g. the Kubernetes watcher) keep the original per-event push via _process_dbt_log_event.

No consumer-side change: the consumer already logs _dbt_event once at terminal (execute_complete/poke).

Scope

Final of the three split PRs from this work:

1. duplicate consumer logs — merged (Fix duplicate consumer logs in WATCHER mode #2805)
2. dbt-runner error capture on the per-event path — merged (Capture dbt-runner error message in WATCHER consumer logs #2806)
3. this PR — per-node event buffering (the architectural change + reduced XCom writes)

Testing

• Unit tests for the buffered path: terminal flush, error-message-wins, subprocess RunResultError re-flush, "None"-status guard, and the allowlist / missing-unique_id guards.
• Full tests/operators/_watcher/ suite passes (309); pre-commit (ruff/black/mypy) passes.

References

• [Bug] Watcher mode 1.14.x >3x slower than 1.13.1 on large DAGs (800+ models) - DAG times out after forced Airflow 3.x migration #2736 — watcher performance regression
• Watcher producer: push _dbt_event only for a curated subset (allowlist) of dbt event types #2777 — reduce producer _dbt_event writes (delivered here via buffering)
• Include non-error logs in consumer task output #2456

🤖 Generated with Claude Code

[Copilot]

Pull request overview

This PR improves WATCHER-mode producer→consumer dbt log shipping by buffering per-node structured events in memory and flushing to XCom only at terminal states, ensuring reliable error-message capture across dbt invocation modes and preventing duplicate consumer logs caused by repeated polling.

Changes:

• Add an in-memory per-node dbt event buffer in the local watcher producer and flush structured events to XCom at terminal status (with an additional re-flush when late error text arrives).
• Stop the trigger from logging the per-node structured event on every poll; log the structured event once in the consumer at completion.
• Add unit tests covering runner vs subprocess error-message capture, non-overwrite behavior, and buffer-flush guards.

Reviewed changes

Copilot reviewed 5 out of 5 changed files in this pull request and generated 1 comment.

Show a summary per file

File	Description
cosmos/operators/_watcher/base.py	Implements buffered structured-event accumulation/flush, integrates it into log parsing, and adjusts consumer logging to be “once per node”.
cosmos/operators/_watcher/triggerer.py	Removes per-poll structured-event XCom fetch/log to avoid duplicate logs from trigger polling.
cosmos/operators/watcher.py	Adds and wires a per-node event buffer into the local watcher producer operator lifecycle.
tests/operators/_watcher/test_watcher_base.py	Adds unit tests for buffered event accumulation/flush and error-message capture semantics.
tests/operators/_watcher/test_triggerer.py	Updates trigger tests to reflect removal of _log_dbt_event polling behavior.

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[codecov]

Codecov Report

✅ All modified and coverable lines are covered by tests.
✅ Project coverage is 98.39%. Comparing base (0f4182a) to head (a6d326b).
⚠️ Report is 4 commits behind head on main.

Additional details and impacted files

@@            Coverage Diff             @@
##             main    #2792      +/-   ##
==========================================
+ Coverage   98.37%   98.39%   +0.01%     
==========================================
  Files         107      107              
  Lines        8021     8080      +59     
==========================================
+ Hits         7891     7950      +59     
  Misses        130      130              

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.

[Copilot]

Pull request overview

Copilot reviewed 5 out of 5 changed files in this pull request and generated 2 comments.

[Copilot]

Pull request overview

Copilot reviewed 5 out of 5 changed files in this pull request and generated 2 comments.

[Copilot]

Pull request overview

Copilot reviewed 6 out of 6 changed files in this pull request and generated no new comments.

[tatiana]

@pankajastro, thanks for the improvements! I feel we tried to do too much in a single PR:

1. Removing duplicate consumer logs
2. Improving tackling the error message across dbt Runner/subprocess
3. Changing local watcher _dbt_event persistence from per-event XCom writes to a per-event to an in-memory buffer, flushed at terminal state

The first two seem lower-risk correctness files and can be merged independently - via individual smaller PRs.

The buffering piece is the bigger architectural change because it changes crash behaviour if the producer is OOM-killed, for instance, and overlaps with the #2777 proposal to reduce XCom _dbt_event writes.

Please, split this into smaller PRs.

Also, for each PR, illustrate via the Airflow UI/command line how it looked before/after the fix, detailing which DAG was used to reproduce that use case.

Some concerns about the biggest architectural change - which we could discuss in the follow-up PR:

0. Evaluate and review options

This approach is an alternative to ticket #2777, which was also assigned to you during this cycle. It is an alternative to reduce _dbt_event writes. In a dedicated PR, we should review which approach makes the most sense.

1. Accuracy during producer hard-killing events

(relates to 0 as well)

One concern with the local watcher buffering approach is that we now keep the structured dbt event in memory and only flush it to XCom at the terminal state, with a possible re-flush when a later error event carries the message. That reduces XCom writes, but it also changes the failure mode if the producer process is hard-killed, for example, by OOMKilled/SIGKILL. In that case, Python will not run the terminal flush, so any in-flight node event that only exists in node_event_buffer is lost.

That may be an acceptable tradeoff, especially if this PR is intentionally optimising local watcher XCom write volume, but I think we should call it out explicitly. Is the intended behaviour that, when a hard producer death occurs, in-flight structured events/error text are lost, and consumers fall back to the producer-failed path? If yes, can we document that in the documentation/PR/comments/tests? If not, we may need some form of periodic checkpointing instead of only terminal flushes.

2. Support ExecutionMode.WATCHER_KUBERNETES

(relates to 0 as well)

Could you confirm if keeping ExecutionMode.WATCHER_KUBERNETES on the unbuffered path is intentional? With #2543 merged, the Kubernetes watcher now has a per-execution state path via callback_extra_kwargs, so it looks like it could also pass a node_event_buffer into store_dbt_resource_status_from_log.

If this PR is intentionally scoped to localExecutionMode.WATCHER only, I think we should make that explicit in the PR description and make sure we create a follow-up PR - to be included in the Cosmos 1.15.0 release - that makes the approach in ExecutionMode.WATCHER_KUBERNETES` consistent. Otherwise, the K8s implementation still keeps the original per-event XCom writes while local WATCHER gets the buffered path - we don't want to have these two patterns in the code base.

[pankajastro]

Thanks @tatiana, agree it was too much in one PR. I've split it:

• Fix duplicate consumer logs in WATCHER mode #2805 — duplicate consumer logs
• Capture dbt-runner error message in WATCHER consumer logs #2806 — error-message capture (dbt-runner mode)
• this PR → rebased down to just the buffering change once the two above land

A few clarifications:

Why #2806 is dbt-runner only. The two modes put the error in different places:

• dbt-runner: in NodeFinished.run_result.message — i.e. on the terminal event itself, so a single per-event push captures it. Clean and stateless.
• subprocess: in a RunResultError that arrives after the terminal event, with node_status="None". On the per-event path the later event would overwrite the terminal status, and a trailing empty message could wipe the error.

So subprocess capture needs retained per-node state to merge "status from the terminal event" + "message from the later error" — which is the buffer. That's why robust subprocess capture rides with the buffering PR (the re-flush + "don't overwrite the error"/"ignore None status" guards), not #2806.

Kubernetes is intentionally scoped out. Local WATCHER gets the buffered path; WATCHER_KUBERNETES stays per-event. Bringing K8s in line is already tracked by #2457, so I'll do it there rather than open a new follow-up — agreed we shouldn't keep two patterns long-term.

On in-memory loss during a hard producer kill: not a regression. The buffer flushes per node at that node's terminal state, so completed nodes are already in XCom. Only in-flight nodes are lost — but those have no terminal status under the old per-event path either, and the consumer fails them via the producer-failure fallback. Same outcome; we just don't keep partial logs (still in the producer log). I'll call this out in the buffering PR, along with the #2777 comparison.