Skip to content

fix(topic): add read-ahead buffer flow control to TopicListener#2234

Open
kprokopenko wants to merge 5 commits into
masterfrom
fix/topic-listener-buffer-flow-control
Open

fix(topic): add read-ahead buffer flow control to TopicListener#2234
kprokopenko wants to merge 5 commits into
masterfrom
fix/topic-listener-buffer-flow-control

Conversation

@kprokopenko

Copy link
Copy Markdown
Collaborator

Summary

  • Add centralized read-ahead buffer flow control to TopicListener, matching the topic reader model: one shared BufferSize limit across all partitions via freeBytes and sendMessagesLoop.
  • Partition workers release buffer credit through FreeBufferFromBatch (including on handler errors and queued batches on shutdown) instead of sending per-partition ReadRequest messages.
  • Add public topicoptions.WithListenerBufferSizeBytes option (default 1 MiB, same as topic reader).

Test plan

  • go test -race ./internal/topic/topiclistenerinternal/... ./internal/xsync/...
  • golangci-lint run ./...
  • go test -race ./...

Made with Cursor

Centralize BufferSize accounting in streamListener (like topic reader) so
partition workers release credits instead of sending per-partition ReadRequests.

Co-authored-by: Cursor <cursoragent@cursor.com>
@kprokopenko kprokopenko added topic issue for topic service no slo labels Jul 9, 2026
@github-actions

github-actions Bot commented Jul 9, 2026

Copy link
Copy Markdown

github.com/ydb-platform/ydb-go-sdk/v3/topic/topicoptions

compatible changes

WithListenerBufferSizeBytes: added

summary

Inferred base version: v3.143.0
Suggested version: v3.144.0

@codecov

codecov Bot commented Jul 9, 2026

Copy link
Copy Markdown

Codecov Report

❌ Patch coverage is 97.80220% with 2 lines in your changes missing coverage. Please review.
✅ Project coverage is 75.48%. Comparing base (9efa606) to head (8121543).

Files with missing lines Patch % Lines
...nal/topic/topiclistenerinternal/stream_listener.go 97.05% 1 Missing and 1 partial ⚠️
Additional details and impacted files
@@            Coverage Diff             @@
##           master    #2234      +/-   ##
==========================================
+ Coverage   75.43%   75.48%   +0.05%     
==========================================
  Files         445      445              
  Lines       45996    46043      +47     
==========================================
+ Hits        34698    34757      +59     
+ Misses       9343     9332      -11     
+ Partials     1955     1954       -1     
Flag Coverage Δ
experiment 75.37% <96.70%> (+0.06%) ⬆️
go-1.21.x 72.29% <93.40%> (-0.07%) ⬇️
go-1.26.x 75.47% <97.80%> (+0.07%) ⬆️
integration 56.91% <71.42%> (-0.23%) ⬇️
macOS 45.54% <94.50%> (+0.30%) ⬆️
ubuntu 75.48% <97.80%> (+0.05%) ⬆️
unit 45.56% <94.50%> (+0.31%) ⬆️
windows 45.56% <94.50%> (+0.31%) ⬆️
ydb-24.4 56.45% <60.43%> (-0.18%) ⬇️
ydb-edge 56.74% <71.42%> (-0.30%) ⬇️
ydb-latest 56.64% <60.43%> (-0.21%) ⬇️
ydb-nightly 75.37% <96.70%> (+0.06%) ⬆️

Flags with carried forward coverage won't be shown. Click here to find out more.

☔ 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.

kprokopenko and others added 2 commits July 9, 2026 20:52
Cover queued batch drain on handler error, metadata validation failures,
freeBytes coalescing, end-to-end credit return, and send error handling.

Co-authored-by: Cursor <cursoragent@cursor.com>
Add tests for newStreamListener seed, StopPartition routing, shutdown
free-buffer branches, and WithListenerBufferSizeBytes option.

Co-authored-by: Cursor <cursoragent@cursor.com>

@robot-vibe-db robot-vibe-db Bot left a comment

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

AI Review Summary

Verdict: ✅ No critical issues found

Critical issues

No critical issues found.

Other findings

  • Minor | High: Missing zero-size guard in freeBufferFromBatch — can send pointless ReadRequest{BytesSize: 0} to the server — internal/topic/topiclistenerinternal/stream_listener.go:554
  • Minor | Medium: sendDataRequest in sendMessagesLoop freeBytes case goes through sendMessage, causing a spurious hasNewMessagesToSend signal and redundant empty flush on the next select iteration — internal/topic/topiclistenerinternal/stream_listener.go:283
  • Nit | Low: WithListenerBufferSizeBytes accepts zero/negative values silently; validation happens later in Validate() but the godoc doesn't mention the constraint — topic/topicoptions/topicoptions_listener.go:19

This review was generated automatically. Critical issues require attention; other findings are advisory.
If this comment was useful, please give it a 👍 — it helps us improve the review bot.

Comment thread internal/topic/topiclistenerinternal/stream_listener.go
Comment thread internal/topic/topiclistenerinternal/stream_listener.go
Comment thread topic/topicoptions/topicoptions_listener.go
@robot-vibe-db

robot-vibe-db Bot commented Jul 10, 2026

Copy link
Copy Markdown

Full analysis log

Analysis performed by claude, claude-opus-4-6.

- Skip zero-size ReadRequest in freeBufferFromBatch when nothing was charged.
- Enqueue the coalesced ReadRequest without signaling hasNewMessagesToSend in
  the freeBytes case, avoiding a stale signal and a redundant empty flush.
- Document that WithListenerBufferSizeBytes requires a positive value.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
@kprokopenko kprokopenko requested a review from rekby July 10, 2026 15:48
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

ai_reviewed no slo topic issue for topic service

Projects

None yet

Development

Successfully merging this pull request may close these issues.

1 participant