Fix CloudWatch trigger stream lifecycle

[d-hervas]

Structured Triggerer logging reused one queued Watchtower handler across all CloudWatch stream paths. Watchtower retains queue, sequence-token, and worker state for every path until that handler closes, while Airflow keeps the remote I/O object alive for the Triggerer process.

This change owns one Watchtower handler per active structured-log path and closes/removes that handler when upload(path) signals completion. Other active paths remain live, a later reuse of the same path creates a fresh handler, and records racing with close cannot reopen the path during teardown. The legacy task-handler path keeps its current behavior.

Expected behavior

When upload(path) marks one trigger stream complete, all
queued records for that stream are delivered and its Watchtower queue, worker,
and sequence-token state are released. Other active streams must remain usable,
and later reuse of the same path must create fresh state.

Failure mode

A long-lived Triggerer can process many distinct trigger log paths, including sequentially. With Watchtower's default queued delivery, the first record for each path creates a queue, sequence-token entry, and worker thread. Airflow calls upload(path) when the trigger finishes, but the current implementation only flushes the shared handler. flush() delivers pending records without stopping the worker or removing that stream's state. Since the shared handler remains reachable for the lifetime of the Triggerer, retained state grows with the number of completed streams. Eventually the Triggerer's RSS or thread count can reach its container limit and the process is terminated with SIGKILL/OOM.

The retention can be reproduced without AWS credentials or API calls using Watchtower 3.4.0. This mirrors the current provider behavior of changing log_stream_name on one queued handler:

import logging
from itertools import count

import watchtower


class FakeLogsClient:
    def __init__(self):
        self.tokens = count(1)

    def put_log_events(self, **kwargs):
        return {"nextSequenceToken": str(next(self.tokens))}


handler = watchtower.CloudWatchLogHandler(
    log_group_name="reproducer",
    log_stream_name="unused",
    boto3_client=FakeLogsClient(),
    create_log_group=False,
    create_log_stream=False,
    use_queues=True,
    send_interval=60,
)

for job_id in range(300):
    handler.log_stream_name = (
        f"dag/task/attempt=1.log.trigger.{job_id}.log"
    )
    record = logging.LogRecord(
        "trigger",
        logging.INFO,
        "repro.py",
        1,
        "event",
        (),
        None,
    )
    handler.handle(record)

handler.flush()

print("queues:", len(handler.queues))
print("sequence tokens:", len(handler.sequence_tokens))
print("thread references:", len(handler.threads))
print("live workers:", sum(thread.is_alive() for thread in handler.threads))

handler.close()

Output:

queues: 300
sequence tokens: 300
thread references: 300
live workers: 300

Calling flush() therefore does not bound retained state as streams complete.
Calling close() stops all workers, but the current shared-handler design cannot do that for one completed stream without also closing other active streams.

This change gives each active path an independently closable handler.

Related work

#70635 adds read-side discovery of deferred-task trigger log streams so they remain visible after deferral. This PR addresses the separate write-side lifecycle of those streams and releases Watchtower's retained queues, sequence-token state, and worker threads when each stream completes.

Validation

All 39 CloudWatch task-handler tests pass on Python 3.10.21 and Python 3.14.7. New tests cover active-stream isolation, repeated completion, exact-path reuse, 100 sequential processor-driven paths with no retained handlers, construction and close/log races, and final shutdown. Ruff, the Airflow Python 3.10 CI image's pinned mypy for both changed files, the full pre-commit prek stage, and git diff --check also pass.

---

Was generative AI tooling used to co-author this PR?

• Yes: OpenAI Codex

[subhramit]

Some minor feedback

[o-nikolas]

I think this needs a few pairs of eyes, logging is critical code. Left one comment