refactor(pubsub): move worker logic into OutstandingPublishes #4051

PhongChuong · 2025-12-12T00:15:15Z

Refactor by:

moving batch size logic from Worker to OutstandingPublishes
simplify Batch by passing initial message size instead of the topic string
simplify Batch by returning a JoinHandle instead of appending to inflight FuturesUnordered data structure

Towards: #4012

codecov · 2025-12-12T00:26:01Z

Codecov Report

✅ All modified and coverable lines are covered by tests.
✅ Project coverage is 95.28%. Comparing base (07c6f2c) to head (2b79566).
⚠️ Report is 27 commits behind head on main.

Additional details and impacted files

@@           Coverage Diff           @@
##             main    #4051   +/-   ##
=======================================
  Coverage   95.27%   95.28%           
=======================================
  Files         173      173           
  Lines        6624     6635   +11     
=======================================
+ Hits         6311     6322   +11     
  Misses        313      313

☔ View full report in Codecov by Sentry.
📢 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.

suzmue

Can you also update the commit message description to describe a bit about the change that you are making (what has changed / why). Thanks!

src/pubsub/src/publisher/worker.rs

suzmue · 2025-12-12T18:22:24Z

src/pubsub/src/publisher/worker.rs

-                                // all of those resolve as well.
-                                let mut flushing = std::mem::take(&mut inflight);
-                                while flushing.next().await.is_some() {}
+                                outstanding.flush(self.client.clone(), self.topic_name.clone());


This seems to me like it has changed the behavior so that it is no longer waiting on all of the batches to have returned, which is a regression. Flush just sends the existing batch, but it doesn't wait on any of those to resolve.

Notably this doesn't appear to be caught in the tests so if we can add a test for this too that would be great.

I think there was a misunderstanding on flush on my part. In my head, flush is to signal to the publisher to send out all the pending batches/publish but not await until it finishes.

If the contract of flush is to await, then are we waiting for all pending publishes? Or just the pending batch?

suzmue · 2025-12-12T18:23:06Z

src/pubsub/src/publisher/worker.rs

                        },
                        Some(ToWorker::Flush(tx)) => {
+                            // TODO(#4012): To guarantee ordering, we should wait for the
+                            // inflight batch to complete so that messages are publish in order.


This comment feels like it belongs within OutstandingPublish NOT here, since this is no longer responsible for deciding when to send the message.

See comment regarding moving inflight back into Worker.

In that case, I think having the comment here makes more sense.

I still don't think this main loop is handling keeping track of which batches have outstanding publishes. But since this is a TODO, I am fine leaving the comment here. But it seems likely this sort of decision should be handled by the OutstandingPublish.

It does not handle outstanding publish yet as this is simply a refactoring.

Regarding where to handle awaiting the inflight batch, I actually think it is better handled by the Worker as it has it avoids spawning additional tasks in OutstandingPublishes to push pending batches along.

Lets discuss offline outside of this PR.

suzmue · 2025-12-12T18:23:27Z

src/pubsub/src/publisher/worker.rs

+    message_limit: u32,
+    byte_threshold: u32,
+    pending_batch: Batch,
+    inflight: Option<tokio::task::JoinHandle<()>>,


inflight is unused?

I moved inflight back up to Worker. I think having the task control logic purely in Worker is easier to follow then having it distributed across Worker and OutstandingPublishes.
PTAL.

src/pubsub/src/publisher/worker.rs

Co-authored-by: Suzy Mueller <[email protected]>

PhongChuong · 2025-12-15T18:07:00Z

@suzmue , Thanks for the review.
I've added inflight back to worker as having all the task control logic within Worker turns out to be better.
I think we need to discuss the Flush contract in a bit more detail in regards to what it would await on (pending batch vs pending publishes) or if it should await at all.

PTAL.

src/pubsub/src/publisher/batch.rs

suzmue · 2025-12-15T19:26:39Z

src/pubsub/src/publisher/worker.rs

                        },
                        Some(ToWorker::Flush(tx)) => {
+                            // TODO(#4012): To guarantee ordering, we should wait for the
+                            // inflight batch to complete so that messages are publish in order.


I still don't think this main loop is handling keeping track of which batches have outstanding publishes. But since this is a TODO, I am fine leaving the comment here. But it seems likely this sort of decision should be handled by the OutstandingPublish.

suzmue · 2025-12-15T19:29:42Z

src/pubsub/src/publisher/worker.rs

+        msg: BundledMessage,
+        inflight: &mut FuturesUnordered<tokio::task::JoinHandle<()>>,
+    ) {
+        self.pending_batch.push(msg);


This can be in a future PR, but what if adding the msg increases the size > the limit of the size of a single message (not just the configured max size but the actual max size we are allowed to send). I think we should:

flush the existing batch

start a new batch with the message.

Added a TODO.

src/pubsub/src/publisher/worker.rs

PhongChuong added 5 commits December 11, 2025 23:34

add todo

ceba0e9

fmt

f9c8d3b

refactor(pubsub): move worker logic into OutstandingPublishes

eaf2bb2

fix worker run comment

e774622

merge and fmt

73b1c9a

product-auto-label bot added the api: pubsub Issues related to the Pub/Sub API. label Dec 12, 2025

PhongChuong marked this pull request as ready for review December 12, 2025 00:33

PhongChuong requested a review from a team as a code owner December 12, 2025 00:33

PhongChuong requested a review from suzmue December 12, 2025 00:34

suzmue reviewed Dec 12, 2025

View reviewed changes

PhongChuong and others added 2 commits December 15, 2025 11:57

Apply suggestion from @suzmue

788bbd2

Co-authored-by: Suzy Mueller <[email protected]>

address review comments

310d6c0

suzmue reviewed Dec 15, 2025

View reviewed changes

PhongChuong added 2 commits December 15, 2025 21:05

address comment

cacc776

add back missing comment for main worker loop

2b79566

suzmue approved these changes Dec 16, 2025

View reviewed changes

PhongChuong merged commit 317b8d2 into googleapis:main Dec 17, 2025
29 checks passed

PhongChuong deleted the orderingRefactor branch December 17, 2025 16:51

refactor(pubsub): move worker logic into OutstandingPublishes #4051

refactor(pubsub): move worker logic into OutstandingPublishes #4051

Uh oh!

Conversation

PhongChuong commented Dec 12, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

codecov bot commented Dec 12, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Codecov Report

Uh oh!

suzmue left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

PhongChuong commented Dec 15, 2025

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

PhongChuong commented Dec 12, 2025 •

edited

Loading

codecov bot commented Dec 12, 2025 •

edited

Loading