fix(api): prevent dropped workflow_started events in Redis Streams by zl86790 · Pull Request #40964 · langgenius/dify

zl86790 · 2026-08-19T07:11:18Z

Summary

When using Redis Streams as the event transport (EVENT_BUS_REDIS_CHANNEL_TYPE=streams), the workflow_started event (and sometimes the first few node events) can silently disappear on concurrent streaming runs. The events are actually in Redis — the SSE subscriber just never sees them.

The root cause: the workflow execution task gets dispatched to Celery right away, before the SSE response has even started streaming. The Redis Streams subscription only gets set up later, and it starts reading from $ (only new messages). If the worker publishes workflow_started before that subscription exists, it's gone for good. This shows up much more often under concurrent load, which matches what's reported in the issue (55% loss with 8 concurrent requests).

Reverting $ back to 0-0 isn't an option — that was changed in #34030 specifically to stop replaying stale workflow_paused events on reconnect (#34040), and reverting it would bring that bug straight back.

Fix: grab a checkpoint (the current tail of the stream) synchronously, before dispatching the task. The subscriber then resumes from that exact point once it actually connects, no matter how late that happens. If reading the checkpoint itself fails, we let it fail loudly instead of quietly falling back to something unsafe.

Pub/sub and sharded transports aren't affected — they already gate task start on a real subscription with a fallback timer, so this failure mode doesn't apply to them.

Fixes #40948

Tests

Added unit tests for the new checkpoint() / subscribe_from() methods on the streams channel: reading the last entry id, empty-stream case, checkpoint errors propagating instead of being swallowed, and that a subscriber resuming from a checkpoint picks up events published before it actually starts listening (while still not replaying anything from before the checkpoint).
Added a regression test that dispatches the task — letting it publish both workflow_started and workflow_finished — before the subscriber even calls retrieve_events(), matching the real ordering. Also fixed the fake Redis Streams test double, which didn't model $ semantics at all and had been masking this bug.
Added a test covering the case where the checkpoint read fails: task must not start, error propagates.
ruff check — clean.
Full relevant test suite (streams channel, streaming utils, app generate service) — 80 passed.
Haven't run make type-check locally, worth double-checking in CI.

Checklist

This change requires a documentation update, included: Dify Document
I understand that this PR may be closed in case there was no previous discussion or issues. (This doesn't apply to typos!)
I've added a test for each change that was introduced, and I tried as much as possible to make a single atomic change.
I've updated the documentation accordingly.
I ran make lint && make type-check (backend) and cd web && pnpm exec vp staged (frontend) to appease the lint gods

github-actions · 2026-08-19T09:03:19Z

Pyrefly Type Coverage

Metric	Base	PR	Delta
Type coverage	59.71%	59.71%	+0.00%
Strict coverage	59.29%	59.29%	+0.00%
Typed symbols	39,742	39,757	+15
Untyped symbols	27,010	27,019	+9
Modules	3171	3171	0

github-actions · 2026-08-19T09:03:26Z

Pyrefly Diff

base → PR

--- /tmp/pyrefly_base.txt	2026-08-19 09:02:40.754971977 +0000
+++ /tmp/pyrefly_pr.txt	2026-08-19 09:02:27.721953129 +0000
@@ -8087,35 +8087,39 @@
 ERROR Argument `FakeRedisClient` is not assignable to parameter `client` with type `Redis[Unknown] | RedisCluster[Unknown]` in function `libs.broadcast_channel.redis._subscription.RedisSubscriptionBase.__init__` [bad-argument-type]
     --> tests/unit_tests/libs/broadcast_channel/redis/test_channel_unit_tests.py:1127:24
 ERROR Class member `FailExpireRedis.expire` overrides a member in a parent class but is missing an `@override` decorator [missing-override-decorator]
-  --> tests/unit_tests/libs/broadcast_channel/redis/test_streams_channel_unit_tests.py:75:9
+  --> tests/unit_tests/libs/broadcast_channel/redis/test_streams_channel_unit_tests.py:81:9
+ERROR Class member `FailXrevrangeRedis.xrevrange` overrides a member in a parent class but is missing an `@override` decorator [missing-override-decorator]
+  --> tests/unit_tests/libs/broadcast_channel/redis/test_streams_channel_unit_tests.py:86:9
 ERROR Argument `FakeStreamsRedis` is not assignable to parameter `redis_client` with type `Redis[Unknown] | RedisCluster[Unknown]` in function `libs.broadcast_channel.redis.streams_channel.StreamsBroadcastChannel.__init__` [bad-argument-type]
-   --> tests/unit_tests/libs/broadcast_channel/redis/test_streams_channel_unit_tests.py:147:36
+   --> tests/unit_tests/libs/broadcast_channel/redis/test_streams_channel_unit_tests.py:158:36
 ERROR Argument `FakeStreamsRedis` is not assignable to parameter `redis_client` with type `Redis[Unknown] | RedisCluster[Unknown]` in function `libs.broadcast_channel.redis.streams_channel.StreamsBroadcastChannel.__init__` [bad-argument-type]
-   --> tests/unit_tests/libs/broadcast_channel/redis/test_streams_channel_unit_tests.py:173:41
+   --> tests/unit_tests/libs/broadcast_channel/redis/test_streams_channel_unit_tests.py:184:41
 ERROR Argument `FakeStreamsRedis` is not assignable to parameter `redis_client` with type `Redis[Unknown] | RedisCluster[Unknown]` in function `libs.broadcast_channel.redis.streams_channel.StreamsBroadcastChannel.__init__` [bad-argument-type]
-   --> tests/unit_tests/libs/broadcast_channel/redis/test_streams_channel_unit_tests.py:180:41
+   --> tests/unit_tests/libs/broadcast_channel/redis/test_streams_channel_unit_tests.py:191:41
 ERROR Argument `FailExpireRedis` is not assignable to parameter `redis_client` with type `Redis[Unknown] | RedisCluster[Unknown]` in function `libs.broadcast_channel.redis.streams_channel.StreamsBroadcastChannel.__init__` [bad-argument-type]
-   --> tests/unit_tests/libs/broadcast_channel/redis/test_streams_channel_unit_tests.py:193:43
+   --> tests/unit_tests/libs/broadcast_channel/redis/test_streams_channel_unit_tests.py:204:43
+ERROR Argument `FailXrevrangeRedis` is not assignable to parameter `redis_client` with type `Redis[Unknown] | RedisCluster[Unknown]` in function `libs.broadcast_channel.redis.streams_channel.StreamsBroadcastChannel.__init__` [bad-argument-type]
+   --> tests/unit_tests/libs/broadcast_channel/redis/test_streams_channel_unit_tests.py:259:43
 ERROR Argument `FakeStreamsRedis` is not assignable to parameter `redis_client` with type `Redis[Unknown] | RedisCluster[Unknown]` in function `libs.broadcast_channel.redis.streams_channel.StreamsBroadcastChannel.__init__` [bad-argument-type]
-   --> tests/unit_tests/libs/broadcast_channel/redis/test_streams_channel_unit_tests.py:244:43
+   --> tests/unit_tests/libs/broadcast_channel/redis/test_streams_channel_unit_tests.py:339:43
 ERROR Argument `TestStreamsSubscription.test_listener_normalizes_supported_payloads_and_ignores_unsupported_shapes.OneShotRedis` is not assignable to parameter `client` with type `Redis[Unknown] | RedisCluster[Unknown]` in function `libs.broadcast_channel.redis.streams_channel._StreamsSubscription.__init__` [bad-argument-type]
-   --> tests/unit_tests/libs/broadcast_channel/redis/test_streams_channel_unit_tests.py:269:45
+   --> tests/unit_tests/libs/broadcast_channel/redis/test_streams_channel_unit_tests.py:364:45
 ERROR Argument `object` is not assignable to parameter `o` with type `Buffer | Iterable[SupportsIndex] | SupportsBytes | SupportsIndex` in function `bytes.__new__` [bad-argument-type]
-   --> tests/unit_tests/libs/broadcast_channel/redis/test_streams_channel_unit_tests.py:277:35
+   --> tests/unit_tests/libs/broadcast_channel/redis/test_streams_channel_unit_tests.py:372:35
 ERROR Argument `TestStreamsSubscription.test_listener_ignores_close_signal_from_another_subscription.OneShotRedis` is not assignable to parameter `client` with type `Redis[Unknown] | RedisCluster[Unknown]` in function `libs.broadcast_channel.redis.streams_channel._StreamsSubscription.__init__` [bad-argument-type]
-   --> tests/unit_tests/libs/broadcast_channel/redis/test_streams_channel_unit_tests.py:302:45
+   --> tests/unit_tests/libs/broadcast_channel/redis/test_streams_channel_unit_tests.py:397:45
 ERROR Argument `FakeStreamsRedis` is not assignable to parameter `client` with type `Redis[Unknown] | RedisCluster[Unknown]` in function `libs.broadcast_channel.redis.streams_channel._StreamsSubscription.__init__` [bad-argument-type]
-   --> tests/unit_tests/libs/broadcast_channel/redis/test_streams_channel_unit_tests.py:346:45
+   --> tests/unit_tests/libs/broadcast_channel/redis/test_streams_channel_unit_tests.py:441:45
 ERROR Argument `FakeStreamsRedis` is not assignable to parameter `client` with type `Redis[Unknown] | RedisCluster[Unknown]` in function `libs.broadcast_channel.redis.streams_channel._StreamsSubscription.__init__` [bad-argument-type]
-   --> tests/unit_tests/libs/broadcast_channel/redis/test_streams_channel_unit_tests.py:354:45
+   --> tests/unit_tests/libs/broadcast_channel/redis/test_streams_channel_unit_tests.py:449:45
 ERROR Argument `FakeStreamsRedis` is not assignable to parameter `client` with type `Redis[Unknown] | RedisCluster[Unknown]` in function `libs.broadcast_channel.redis.streams_channel._StreamsSubscription.__init__` [bad-argument-type]
-   --> tests/unit_tests/libs/broadcast_channel/redis/test_streams_channel_unit_tests.py:363:45
+   --> tests/unit_tests/libs/broadcast_channel/redis/test_streams_channel_unit_tests.py:458:45
 ERROR Argument `FakeStreamsRedis` is not assignable to parameter `client` with type `Redis[Unknown] | RedisCluster[Unknown]` in function `libs.broadcast_channel.redis.streams_channel._StreamsSubscription.__init__` [bad-argument-type]
-   --> tests/unit_tests/libs/broadcast_channel/redis/test_streams_channel_unit_tests.py:371:45
+   --> tests/unit_tests/libs/broadcast_channel/redis/test_streams_channel_unit_tests.py:466:45
 ERROR Argument `BlockingRedis` is not assignable to parameter `client` with type `Redis[Unknown] | RedisCluster[Unknown]` in function `libs.broadcast_channel.redis.streams_channel._StreamsSubscription.__init__` [bad-argument-type]
-   --> tests/unit_tests/libs/broadcast_channel/redis/test_streams_channel_unit_tests.py:391:45
+   --> tests/unit_tests/libs/broadcast_channel/redis/test_streams_channel_unit_tests.py:486:45
 ERROR Argument `FakeStreamsRedis` is not assignable to parameter `client` with type `Redis[Unknown] | RedisCluster[Unknown]` in function `libs.broadcast_channel.redis.streams_channel._StreamsSubscription.__init__` [bad-argument-type]
-   --> tests/unit_tests/libs/broadcast_channel/redis/test_streams_channel_unit_tests.py:409:45
+   --> tests/unit_tests/libs/broadcast_channel/redis/test_streams_channel_unit_tests.py:504:45
 ERROR Class member `TestCronCompatibility.setUp` overrides a member in a parent class but is missing an `@override` decorator [missing-override-decorator]
   --> tests/unit_tests/libs/test_cron_compatibility.py:21:9
 ERROR Class member `TestTimezoneCompatibility.setUp` overrides a member in a parent class but is missing an `@override` decorator [missing-override-decorator]
@@ -8561,15 +8565,15 @@
 ERROR Argument `Literal['app']` is not assignable to parameter `value` with type `ApiTokenType | SQLCoreOperations[ApiTokenType]` in function `sqlalchemy.orm.base.Mapped.__set__` [bad-argument-type]
   --> tests/unit_tests/services/test_api_token_service.py:44:26
 ERROR Class member `_TrackingRateLimit.exit` overrides a member in a parent class but is missing an `@override` decorator [missing-override-decorator]
-   --> tests/unit_tests/services/test_app_generate_service.py:574:17
+   --> tests/unit_tests/services/test_app_generate_service.py:614:17
 ERROR Class member `_TrackingRateLimit.exit` overrides a member in a parent class but is missing an `@override` decorator [missing-override-decorator]
-   --> tests/unit_tests/services/test_app_generate_service.py:608:17
+   --> tests/unit_tests/services/test_app_generate_service.py:648:17
 ERROR Class member `_TrackingRateLimit.exit` overrides a member in a parent class but is missing an `@override` decorator [missing-override-decorator]
-   --> tests/unit_tests/services/test_app_generate_service.py:640:17
+   --> tests/unit_tests/services/test_app_generate_service.py:680:17
 ERROR Argument `_FakeStreams` is not assignable to parameter `redis_client` with type `Redis[Unknown] | RedisCluster[Unknown]` in function `libs.broadcast_channel.redis.streams_channel.StreamsBroadcastChannel.__init__` [bad-argument-type]
-  --> tests/unit_tests/services/test_app_generate_service_streaming_integration.py:96:36
+   --> tests/unit_tests/services/test_app_generate_service_streaming_integration.py:108:36
 ERROR Argument `_FakeRedisClient` is not assignable to parameter `redis_client` with type `Redis[Unknown] | RedisCluster[Unknown]` in function `libs.broadcast_channel.redis.pubsub_channel.BroadcastChannel.__init__` [bad-argument-type]
-   --> tests/unit_tests/services/test_app_generate_service_streaming_integration.py:116:34
+   --> tests/unit_tests/services/test_app_generate_service_streaming_integration.py:129:34
 ERROR Argument `str | None` is not assignable to parameter `tenant_id` with type `str` in function `services.app_service.AppService.create_app` [bad-argument-type]
    --> tests/unit_tests/services/test_app_service.py:118:17
 ERROR Argument `str | None` is not assignable to parameter `tenant_id` with type `str` in function `services.app_service.AppService.create_app` [bad-argument-type]

wylswz · 2026-08-19T09:08:41Z

        channel_type = dify_config.PUBSUB_REDIS_CHANNEL_TYPE
        if channel_type == "streams":
-            # With Redis Streams, we can safely start right away; consumers can read past events.
+            checkpoint = getattr(topic, "checkpoint", None)


we could check the type of topic instead of using getattr

Updated to check StreamsTopic explicitly instead of using getattr. Thanks :P

Please also refactor this line, thank you.

Copilot

Pull request overview

This PR fixes a race in the Redis Streams event transport where workflow_started (and early node events) can be published by a fast Celery worker before the SSE listener issues its first XREAD, causing the SSE client to miss those events when the subscription starts from $. The fix introduces a synchronous stream checkpoint captured before dispatch, then resumes consumption from that checkpoint once the SSE subscriber actually begins reading.

Changes:

Add StreamsTopic.checkpoint() and StreamsTopic.subscribe_from() so subscribers can resume from an explicitly captured stream position rather than lazily resolving $.
Capture a streams checkpoint in AppGenerateService._build_streaming_task_on_subscribe() before dispatching the background task, and plumb start_id into retrieve_events() / stream_topic_events().
Expand and correct unit/integration tests (including improving the fake Redis Streams $ semantics) to reproduce and prevent the dispatch-before-subscribe ordering regression.

Reviewed changes

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

Show a summary per file

File	Description
api/services/app_generate_service.py	Captures a streams checkpoint before dispatch and passes `start_id` through streaming generators.
api/libs/broadcast_channel/redis/streams_channel.py	Adds `checkpoint()` / `subscribe_from()` and supports explicit `start_id` in the subscription listener.
api/core/app/apps/streaming_utils.py	Adds `start_id` parameter and uses `subscribe_from(start_id)` when provided.
api/core/app/apps/message_based_app_generator.py	Threads `start_id` into `stream_topic_events()` for message-based streaming.
api/tests/unit_tests/libs/broadcast_channel/redis/test_streams_channel_unit_tests.py	Adds unit tests for checkpointing + subscribe-from behavior and error propagation.
api/tests/unit_tests/services/test_app_generate_service.py	Updates unit tests to new `(on_subscribe, start_id)` return and adds checkpoint ordering/error tests.
api/tests/unit_tests/services/test_app_generate_service_streaming_integration.py	Adds regression coverage for dispatch-before-subscribe ordering and fixes fake Redis `$` behavior.

💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.

wylswz · 2026-08-19T10:05:52Z

@QuantumGhost Could you review this stability enhancement on Redis message streaming?

QuantumGhost · 2026-08-20T07:50:24Z

AI disclosure: This comment was drafted with Codex using GPT-5.6-sol. I have reviewed it and am responsible for the content.

I agree with capturing the stream position before dispatching the Celery task, but I think the Redis Streams cursor/checkpoint details should remain inside the broadcast-channel abstraction.

After some thought, I would propose capturing the start_id inside the method _StreamsSubscription._start_if_needed.

The suggested fix is like the following:

class _StreamsSubscription(Subscription):
    def __init__(self, client: Redis | RedisCluster, key: str):
        self._client = client
        self._key = key
        self._start_id: bytes | str | None = None
        ...

    def _resolve_start_id(self) -> bytes | str:
        entries = self._client.xrevrange(self._key, count=1)
        if not entries:
            return "0-0"

        entry_id, _fields = entries[0]
        return entry_id

    def _start_if_needed(self) -> None:
        """This method must be called with `_lock` held."""
        if self._listener is not None or self._closed:
            return

        # Fix the logical read boundary synchronously before `__enter__`
        # returns and before the producer can be started.
        self._start_id = self._resolve_start_id()

        self._listener = threading.Thread(
            target=self._listen,
            name=f"redis-streams-sub-{self._key}",
            daemon=True,
        )
        self._listener.start()

    def _listen(self) -> None:
        start_id = self._start_id
        assert start_id is not None

        last_id = start_id
        ...

This approach minimizes change scope, avoids leaking unnecessary concepts into application services.

The race currently looks like this:

API thread                         Listener thread              Celery worker
──────────                         ───────────────              ─────────────
subscription.__enter__()
  start listener thread
return from __enter__()

on_subscribe()
  task.delay()
                                                                publish event
                                   XREAD stream $
                                   "$" resolves after event
                                   event is skipped

zl86790 · 2026-08-20T09:12:47Z

hi @QuantumGhost,
Thanks for the detailed suggestion, I've updated the code to follow your approach — mind taking another look when you get a chance?

fix(api): prevent dropped workflow_started events in Redis Streams

0909c16

zl86790 requested review from QuantumGhost and laipz8200 as code owners August 19, 2026 07:11

dosubot Bot added the size:M This PR changes 30-99 lines, ignoring generated files. label Aug 19, 2026

zl86790 added 2 commits August 19, 2026 15:15

Merge branch 'main' into fix/streams-channel-workflow-started-race

6e0e5b8

Merge branch 'main' into fix/streams-channel-workflow-started-race

fc432d5

[autofix.ci] apply automated fixes

9d4503c

wylswz reviewed Aug 19, 2026

View reviewed changes

fix(api): use isinstance check instead of getattr for streams checkpoint

dfe01c9

wylswz requested a lite review from Copilot August 19, 2026 09:51

Copilot started reviewing on behalf of wylswz August 19, 2026 09:52 View session

Copilot AI reviewed Aug 19, 2026

View reviewed changes

Comment thread api/services/app_generate_service.py Outdated

Comment thread api/core/app/apps/streaming_utils.py Outdated

zl86790 added 3 commits August 19, 2026 18:42

Merge branch 'main' into fix/streams-channel-workflow-started-race

9679a77

fix(api): fail loudly when streams checkpoint/resume path is unsupported

392bdff

Merge branch 'main' into fix/streams-channel-workflow-started-race

38c9379

QuantumGhost removed the request for review from laipz8200 August 20, 2026 06:22

snaka mentioned this pull request Aug 20, 2026

[Bug] workflow_started is dropped for concurrent streaming runs with the streams event transport (regression of #32518 via #34030) #40948

Open

6 tasks

refactor(api): resolve streams start id inside _StreamsSubscription

0c90306

Merge branch 'main' into fix/streams-channel-workflow-started-race

a3a3d61

Navigation Menu

Uh oh!

fix(api): prevent dropped workflow_started events in Redis Streams - #40964

fix(api): prevent dropped workflow_started events in Redis Streams#40964
zl86790 wants to merge 10 commits into
langgenius:mainfrom
zl86790:fix/streams-channel-workflow-started-race

zl86790 commented Aug 19, 2026 •

edited

Loading

Uh oh!

github-actions Bot commented Aug 19, 2026

Uh oh!

github-actions Bot commented Aug 19, 2026

Uh oh!

wylswz Aug 19, 2026

Uh oh!

zl86790 Aug 19, 2026

Uh oh!

wylswz Aug 20, 2026

Uh oh!

Copilot AI left a comment

Uh oh!

Uh oh!

Uh oh!

wylswz commented Aug 19, 2026

Uh oh!

QuantumGhost commented Aug 20, 2026 •

edited

Loading

Uh oh!

zl86790 commented Aug 20, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

Uh oh!

Conversation

zl86790 commented Aug 19, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Summary

Tests

Checklist

Uh oh!

github-actions Bot commented Aug 19, 2026

Pyrefly Type Coverage

Uh oh!

github-actions Bot commented Aug 19, 2026

Pyrefly Diff

Uh oh!

wylswz Aug 19, 2026

Choose a reason for hiding this comment

Uh oh!

zl86790 Aug 19, 2026

Choose a reason for hiding this comment

Uh oh!

wylswz Aug 20, 2026

Choose a reason for hiding this comment

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull request overview

Reviewed changes

Uh oh!

Uh oh!

Uh oh!

wylswz commented Aug 19, 2026

Uh oh!

QuantumGhost commented Aug 20, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

zl86790 commented Aug 20, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

zl86790 commented Aug 19, 2026 •

edited

Loading

QuantumGhost commented Aug 20, 2026 •

edited

Loading