Skip to content

feat(reduce): track batch trigger reason on batch_time metric. #561

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
tryangul merged 5 commits into from
Aug 26, 2026
Merged

feat(reduce): track batch trigger reason on batch_time metric. #561

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
0757e36
feat(reduce): track batch trigger reason on batch_time metric.
tryangul Aug 25, 2026
b514ec2
Support 'force' trigger reason.
tryangul Aug 25, 2026
a69fef4
Rename trigger_reason -> flush_reason.
tryangul Aug 25, 2026
d42797c
Add python support through new BufferProtocol prop.
tryangul Aug 25, 2026
dbd44e6
lint
tryangul Aug 25, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
11 changes: 11 additions & 0 deletions arroyo/processing/strategies/buffer.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -42,6 +42,14 @@ def is_ready(self) -> bool:
"""
...

@property
def readiness_reason(self) -> str:
"""Returns why is_ready returned True.

Only meaningful when is_ready is True. Used for observability.
"""
...

@tryangul tryangul Aug 25, 2026 •
edited
Loading

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

Only the buffer knows why it's "ready", yet the metric is recorded so it must get propagated via an interface change. I went with an @Property method, but changing the return type of is_ready is another option.

The third option is an entirely new Protocol to propagate this.

It's not clear if this is entirely internal or a breaking API change. Greping in Sentry, I didn't find any implementors (hard to check since protocols are structural interfaces). Please let me know if we can proceed with this approach and how I should version it.

@tryangul tryangul Aug 25, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

@fpacifici

Comment on lines +45 to +51

@sentry sentry Bot Aug 25, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

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

Bug: Accessing the new readiness_reason property on custom buffer implementations without it will cause an AttributeError, which is a breaking API change.
Severity: MEDIUM

Suggested Fix

To avoid a breaking change, use getattr(self.__buffer, 'readiness_reason', None) to safely access the new property with a default value. This maintains backward compatibility for custom buffer implementations that have not been updated.

Prompt for AI Agent 
Review the code at the location below. A potential bug has been identified by an AI
agent. Verify if this is a real issue. If it is, propose a fix; if not, explain why it's
not valid.

Location: arroyo/processing/strategies/buffer.py#L45-L51

Potential issue: The `__flush` method now unconditionally accesses the
`readiness_reason` property on the buffer object. This introduces a breaking API change
for any custom buffer implementations that conform to the `BufferProtocol` but do not
include this new property. Accessing `self.__buffer.readiness_reason` on such an
implementation will raise an `AttributeError`, causing the processing strategy to crash.

Did we get this right? 👍 / 👎 to inform future reviews.


def append(self, message: BaseValue[TPayload]) -> None:
"""Accept a TPayload mutating the internal state of the batch builder."""
...
Expand Down Expand Up @@ -115,9 +123,12 @@ def __flush(self, force: bool) -> None:
)
)
self.__next_step.submit(buffer_msg)

flush_reason = self.__buffer.readiness_reason if self.__buffer.is_ready else "force"

@sentry sentry Bot Aug 25, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

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

Bug: A race condition in __flush can misreport a forced flush as time-based because the reason is checked after the submit() call.
Severity: LOW

Suggested Fix

Determine and store the flush reason before the self.__next_step.submit() call. If force=True, the reason is 'force'. Otherwise, check self.__buffer.is_ready to get the reason before submitting the data.

Prompt for AI Agent 
Review the code at the location below. A potential bug has been identified by an AI
agent. Verify if this is a real issue. If it is, propose a fix; if not, explain why it's
not valid.

Location: arroyo/processing/strategies/buffer.py#L127

Potential issue: A race condition exists in the `__flush` method. When a flush is
initiated with `force=True`, the flush reason is determined after the
`self.__next_step.submit()` call. If the `submit()` operation is slow, the time-based
flush condition (`time.time() >= self._buffer_until`) may become true during the call.
This causes the `flush_reason` to be incorrectly reported as 'time' instead of the
actual reason, 'force'.

Did we get this right? 👍 / 👎 to inform future reviews.

self.__metrics.timing(
"arroyo.strategies.reduce.batch_time",
time.time() - self.__init_time,
tags={"flush_reason": flush_reason},
)

# Reset to the empty state.
Expand Down
6 changes: 6 additions & 0 deletions arroyo/processing/strategies/reduce.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -48,6 +48,12 @@ def is_ready(self) -> bool:
or time.time() >= self._buffer_until
)

@property
def readiness_reason(self) -> str:
if self._buffer_size >= self.max_batch_size:
return "size"
return "time"

def append(self, message: BaseValue[TPayload]) -> None:
self._buffer = self.accumulator(self._buffer, message)
if self.compute_batch_size:
Expand Down
20 changes: 16 additions & 4 deletions rust-arroyo/src/processing/strategies/reduce.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -191,14 +191,26 @@ impl<T, TResult> Reduce<T, TResult> {
}

let batch_time = self.batch_state.batch_start_time.elapsed();
let batch_complete = self.batch_state.message_count >= self.max_batch_size
|| batch_time >= self.max_batch_time;
let size_trigger_complete = self.batch_state.message_count >= self.max_batch_size;
let time_trigger_complete = batch_time >= self.max_batch_time;

if !batch_complete && !force {
if !size_trigger_complete && !time_trigger_complete && !force {
return Ok(());
}

timer!("arroyo.strategies.reduce.batch_time.ms", batch_time);
let flush_reason = if size_trigger_complete {
"size"
} else if time_trigger_complete {
"time"
} else {
"force"
};

timer!(
"arroyo.strategies.reduce.batch_time.ms",
batch_time,
"flush_reason" => flush_reason
);
Comment thread
tryangul marked this conversation as resolved.

let batch_state = mem::replace(
&mut self.batch_state,
Expand Down
4 changes: 4 additions & 0 deletions tests/processing/strategies/test_buffer.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,10 @@ def is_empty(self) -> bool:
def is_ready(self) -> bool:
return len(self._buffer) >= 3

@property
def readiness_reason(self) -> str:
return "size"

def append(self, message: BaseValue[int]) -> None:
self._buffer.append(message.payload)

Expand Down
Loading