Support consuming N messages from channel in one command by pradhankukiran · Pull Request #600 · indeedeng/iwf

pradhankukiran · 2026-03-17T20:48:38Z

Description

Implements server-side support for AtLeast/AtMost fields on InterStateChannelCommand (fields already exist in IDL but were unused). Enables three consumption modes:

Exact N: AtLeast=N, AtMost=N — wait for exactly N, consume exactly N
OneToAll: AtLeast=1 (no AtMost) — wait for at least 1, consume all available
ZeroToAll: AtLeast=0 (no AtMost) — don't wait, consume all available

Messages are consumed atomically. Gated behind global version 11 for backward compat with existing workflows.

Note: The Values field on InterStateChannelResult was added manually to the generated model. A corresponding update to iwf-idl/iwf.yaml is needed so make idl-code-gen produces the same result.

Checklist

Code compiles correctly
Tests for the changes have been added
All tests passing
This PR change is backwards-compatible
This PR CONTAINS a (planned) breaking change (it is not backwards compatible)

Related Issue

Closes #301

longquanzheng · 2026-03-18T19:01:40Z

thanks for contribution. Can you add test to make sure it works?

codecov · 2026-03-18T19:22:59Z

Codecov Report

❌ Patch coverage is 88.49558% with 13 lines in your changes missing coverage. Please review.
✅ Project coverage is 65.13%. Comparing base (0c659fb) to head (747f4d9).

Files with missing lines	Patch %	Lines
service/interpreter/workflowImpl.go	79.62%	5 Missing and 6 partials ⚠️
service/interpreter/deciderTriggerer.go	88.88%	1 Missing and 1 partial ⚠️

❌ Your patch check has failed because the patch coverage (88.49%) is below the target coverage (90.00%). You can increase the patch coverage or adjust the target coverage.

Additional details and impacted files

@@            Coverage Diff             @@
##             main     #600      +/-   ##
==========================================
+ Coverage   64.74%   65.13%   +0.39%     
==========================================
  Files          64       64              
  Lines        6595     6695     +100     
==========================================
+ Hits         4270     4361      +91     
- Misses       1997     2002       +5     
- Partials      328      332       +4

Files with missing lines	Coverage Δ
service/interfaces.go	`85.71% <ø> (ø)`
service/interpreter/InternalChannel.go	`81.35% <100.00%> (-12.93%)`	⬇️
service/interpreter/activityImpl.go	`65.20% <100.00%> (+0.78%)`	⬆️
service/interpreter/continueAsNewer.go	`87.16% <100.00%> (+0.08%)`	⬆️
service/interpreter/globalVersioner.go	`89.47% <100.00%> (+0.58%)`	⬆️
service/interpreter/deciderTriggerer.go	`92.30% <88.88%> (+5.82%)`	⬆️
service/interpreter/workflowImpl.go	`86.31% <79.62%> (-0.51%)`	⬇️

... and 1 file with indirect coverage changes

🚀 New features to boost your workflow:

❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.

pradhankukiran · 2026-03-28T12:18:18Z

@longquanzheng added integration tests covering all 4 consumption modes (ExactN, OneToAll, ZeroToAll, AtMostOnly) across both Temporal and Cadence backends, with and without ContinueAsNew.

Also fixed a bug where AtLeast=0, AtMost=1 would panic on an empty channel, Retrieve() doesn't handle empty channels, so the code now routes through the safe RetrieveUpToN path when atLeast == 0.

longquanzheng · 2026-05-28T05:18:01Z

 	commandRequest iwfidl.CommandRequest,
 	completedTimerCommands map[int]service.InternalTimerStatus,
 	completedSignalCommands, completedInterStateChannelCommands map[int]*iwfidl.EncodedObject,
+	completedInterStateChannelMultiCmds map[int][]*iwfidl.EncodedObject,


do you think we can just change completedInterStateChannelCommands to be map[int][]*iwfidl.EncodedObject?
Or that's a breaking change that will break the determinism?

Changing completedInterStateChannelCommands to map[int][]*EncodedObject would break continue-as-new compatibility. Existing snapshots have object values, not arrays, so old waiting workflows may fail unmarshal/replay. Separate CompletedInterStateChannelMultiCmds keeps old snapshot shape intact.

pradhankukiran · 2026-05-29T13:18:15Z

I am adding a follow-up commit. Plain is to keep CompletedInterStateChannelCommands as-is for continue-as-new compatibility, add InterStateChannelResult.values through IDL/codegen, validate invalid atLeast/atMost, and add focused unit coverage for the new channel/decider paths.
I’ll update generated Go from the IDL locally, but the canonical iwf-idl change may need a separate PR/update in the IDL repo.

longquanzheng · 2026-06-01T17:59:18Z

I am adding a follow-up commit. Plain is to keep CompletedInterStateChannelCommands as-is for continue-as-new compatibility, add InterStateChannelResult.values through IDL/codegen, validate invalid atLeast/atMost, and add focused unit coverage for the new channel/decider paths. I’ll update generated Go from the IDL locally, but the canonical iwf-idl change may need a separate PR/update in the IDL repo.

yes, please open a separate PR in idl repo and merge in that first.

longquanzheng · 2026-06-01T18:03:46Z

+		return fmt.Errorf("InterStateChannelCommand atLeast cannot be negative")
+	}
+	if cmd.HasAtMost() && cmd.GetAtMost() < 0 {
+		return fmt.Errorf("InterStateChannelCommand atMost cannot be negative")
+	}
+	if cmd.HasAtLeast() && cmd.HasAtMost() && cmd.GetAtMost() < cmd.GetAtLeast() {
+		return fmt.Errorf("InterStateChannelCommand atMost cannot be less than atLeast")


returning plain error here will cause workflow task running in a loop. we can return application erorr to fail the workflow directly.

longquanzheng · 2026-06-01T18:05:44Z

@@ -160,6 +160,9 @@ type (
 		CompletedTimerCommands             map[int]InternalTimerStatus   `json:"completedTimerCommands"`
 		CompletedSignalCommands            map[int]*iwfidl.EncodedObject `json:"completedSignalCommands"`
 		CompletedInterStateChannelCommands map[int]*iwfidl.EncodedObject `json:"completedInterStateChannelCommands"`


can you leave a comment here? are we deprecating that or it's still being used?

longquanzheng · 2026-06-01T18:10:57Z

 		stateExecutionLocal = startResponse.GetUpsertStateLocals()
 	}

+	if globalVersioner.IsAfterVersionOfChannelConsumeN() {


can you explain a bit more why this is necssary here?

I still don't understand why we have to do this. can you elaborate? maybe with an example?
thanks

longquanzheng · 2026-06-01T18:14:26Z


-				if received {
-					completedInterStateChannelCmds[idx] = interStateChannel.Retrieve(cmd.ChannelName)
+				for {


I don't think we need a for loop? or maybe I missed something(sorry btw I could be wrong, haven't touched the code for a while)

longquanzheng · 2026-06-01T18:20:53Z

+						if len(values) > 0 {
+							completedInterStateChannelCmds[idx] = values[0]
+						} else {
+							completedInterStateChannelMultiCmds[idx] = values


I feel like it's better not to always backfill the legacy field for single cmd? this means activity input will have duplicate data which cause the history size grow faster.

Since SDK need to change to support this anyway, we should change the SDK to only use the new field only.

For compatibility of old sdk, it will not send the mult-command, server will still need to check if it's single command, then fallback to the old code path and use the single command completed fields.

pradhankukiran · 2026-06-02T11:06:40Z

@longquanzheng Opened the IDL PR here: indeedeng/iwf-idl#91

longquanzheng · 2026-06-11T17:24:18Z

+				// Another command thread may consume from the same channel after this
+				// thread's Await condition is true. Re-check during retrieval; if data


may consume from the same channel after this thread's Await condition is true

I don't think that's possible in Cadence/Temporal's threading model(cooperative instead of preemtive).

Are you sure about this? Maybe you can write a test to confirm that behaviro for me.

Basically, once a thread(goroutine) got return from Await, it will continue to run until next blocking operation. There will not be any interruption between that.

pradhankukiran · 2026-06-11T20:03:17Z

Good point. Removed the retry loop and the comment. Since workflow goroutines are cooperative, once Await returns this thread runs the synchronous retrieval without another command thread interleaving first, so a single consume path is enough.

longquanzheng · 2026-06-12T22:09:01Z

we don't usually have test like this

longquanzheng · 2026-06-12T22:17:22Z

@longquanzheng Opened the IDL PR here: indeedeng/iwf-idl#91

oh , you also need to do it for iwf-sdk.yaml https://github.com/indeedeng/iwf-idl/blob/main/iwf-sdk.yaml

pradhankukiran force-pushed the feature/issue-301-channel-consume-n-messages branch from 5ee0e69 to 6a37310 Compare March 25, 2026 10:52

Support consuming N messages from channel in one command

1bd86db

pradhankukiran force-pushed the feature/issue-301-channel-consume-n-messages branch from 6a37310 to 1bd86db Compare March 28, 2026 12:16

longquanzheng requested review from ktrops and longquanzheng May 28, 2026 05:12

longquanzheng reviewed May 28, 2026

View reviewed changes

longquanzheng requested changes May 28, 2026

View reviewed changes

Merge remote-tracking branch 'upstream/main' into pr-600

4ff1dd3

Improve channel consume N support

2d0a83b

pradhankukiran requested a review from longquanzheng May 29, 2026 13:20

longquanzheng reviewed Jun 1, 2026

View reviewed changes

Address channel consume N review comments

db1b1b4

pradhankukiran requested a review from longquanzheng June 3, 2026 10:48

longquanzheng reviewed Jun 11, 2026

View reviewed changes

Remove redundant channel command retry loop

747f4d9

longquanzheng reviewed Jun 12, 2026

View reviewed changes

Comment thread service/interpreter/workflowImpl_test.go

longquanzheng Jun 12, 2026

Copy link
Copy Markdown

Contributor

Choose a reason for hiding this comment

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

we don't usually have test like this

pradhankukiran closed this Jun 13, 2026

pradhankukiran deleted the feature/issue-301-channel-consume-n-messages branch June 13, 2026 12:07

		// Another command thread may consume from the same channel after this
		// thread's Await condition is true. Re-check during retrieval; if data

Conversation

pradhankukiran commented Mar 17, 2026

Description

Checklist

Related Issue

Uh oh!

longquanzheng commented Mar 18, 2026

Uh oh!

codecov Bot commented Mar 18, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Codecov Report

Uh oh!

pradhankukiran commented Mar 28, 2026

Uh oh!

longquanzheng May 28, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

pradhankukiran commented May 29, 2026

Uh oh!

longquanzheng commented Jun 1, 2026

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!

pradhankukiran commented Jun 2, 2026

Uh oh!

Choose a reason for hiding this comment

Uh oh!

pradhankukiran commented Jun 11, 2026

Uh oh!

Choose a reason for hiding this comment

Uh oh!

longquanzheng commented Jun 12, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

codecov Bot commented Mar 18, 2026 •

edited

Loading

longquanzheng May 28, 2026 •

edited

Loading