test: Redis topic isolation integration test can hang until the 180s timeout

Start with api/tests/test_containers_integration_tests/libs/broadcast_channel/redis/test_channel.py::TestRedisBroadcastChannelIntegration::test_topic_isolation and run the existing API integration workflow. Inspect the producer, consumer, and executor lifecycle around lines 185-226, then verify that the test completes without the 180-second hang, preserves topic isolation, and terminates consumers on failure.

Self Checks

• I have read the Contributing Guide and Language Policy.
• This is a bug report, not a usage question.
• I have searched existing issues, including closed ones, for test_topic_isolation.
• I am submitting this report in English.

Dify version

Merge queue commit 7834f1b73ede1da92c107c1d42bc724eac682f0e, testing #42558 on September 20, 2026.

Cloud or Self Hosted

Self Hosted (Source) — GitHub Actions API integration tests, Linux, Python 3.12.3, four pytest-xdist workers, Redis 6-alpine via Testcontainers.

Steps to reproduce

Observed in merge queue run 35496444507, API Integration Tests shard 1/4.

The failing test is:

api/tests/test_containers_integration_tests/libs/broadcast_channel/redis/test_channel.py::TestRedisBroadcastChannelIntegration::test_topic_isolation

Run the API integration suite using its existing CI workflow. The failure is intermittent; a deterministic reproduction has not yet been established, and no local Docker integration run was performed for this report.

The test starts a producer before two consumers and relies on time.sleep(0.1) before publishing. There is no synchronization confirming that both Redis subscriptions are active.

✔️ Expected Behavior

Both subscriptions are ready before publishing. Each consumer receives only its own topic's message. If setup or receipt fails, subscriptions and worker threads terminate within a bounded time and the test reports the original failure.

❌ Actual Behavior

The first consumer's future.result(timeout=5.0) times out. While unwinding the ThreadPoolExecutor context, shutdown waits for a consumer that is still blocked iterating its subscription. The test ultimately hits the global 180-second timeout.

> received_by_topic1 = consumer1_future.result(timeout=5.0)
api/tests/test_containers_integration_tests/libs/broadcast_channel/redis/test_channel.py:219
...
> if lock.acquire(block, timeout):
E Failed: Timeout (>180.0s) from pytest-timeout.

FAILED .../redis/test_channel.py::TestRedisBroadcastChannelIntegration::test_topic_isolation
1 failed, 480 passed, 25 warnings in 256.35s

The other three API integration shards passed.

Suspected cause: if a consumer is scheduled late or subscription setup takes longer than 100 ms, the producer can publish before the subscription is active. Redis Pub/Sub does not replay that message, so the consumer keeps waiting. This race is present in the code and matches the observed stack; the logs do not directly establish the exact subscription/publish ordering.

Suggested fix: explicitly confirm both Redis subscriptions are active before publishing, replace the fixed sleep with synchronization, and ensure failure paths close subscriptions before executor shutdown so consumers cannot wait indefinitely. Merely increasing the sleep or global timeout would leave the race in place.

Relation to #42558: the failure was observed while that PR was in the merge queue. This test uses its own Redis container and does not depend on the PostgreSQL fixture changed by the PR. The logs provide no evidence of PostgreSQL tmpfs exhaustion; shared-runner load effects have not been ruled out. Track this test reliability issue separately from the tmpfs optimization.