fix: avoid StackOverflowError in PersistBarrier.awaitPersisted for large steps

[zhangxiangnan]

Problem

PersistBarrier.awaitPersisted folds a step's per-event barriers into a left-nested Completable.andThen(...) chain:

Completable result = Completable.complete();
for (Event event : events) {
  String eventId = event.id();
  if (eventId != null) {
    result = result.andThen(barrier(context, eventId));
  }
}
return result;

A single step can carry a large number of events — e.g. an agent transfer folds the sub-agent's events into the parent step. The nested andThen chain recurses once per element on both subscription and completion, so long sessions throw StackOverflowError:

java.lang.StackOverflowError
    at io.reactivex.rxjava3.internal.operators.completable.CompletableAndThenCompletable.subscribeActual(...)
    at io.reactivex.rxjava3.core.Completable.subscribe(...)
    ...

Fix

Await the per-event barriers with Completable.concat(Iterable) instead. concat drains its sources iteratively, so it stays stack-safe, while keeping the same await semantics (order is irrelevant — each CompletableSubject retains its terminal state).

Testing

Added a regression test largeStep_awaitsAllWithoutStackOverflow that awaits 50,000 events in a single step; it fails with StackOverflowError on the old implementation and passes with the fix.

mvn -pl core test -Dtest=PersistBarrierTest → Tests run: 10, Failures: 0, Errors: 0, Skipped: 0.

Made with Cursor

[google-cla]

Thanks for your pull request! It looks like this may be your first contribution to a Google open source project. Before we can look at your pull request, you'll need to sign a Contributor License Agreement (CLA).

View this failed invocation of the CLA check for more information.

For the most up to date status, view the checks section at the bottom of the pull request.

[hemasekhar-p]

Hi @zhangxiangnan, thank you for your contribution! We appreciate you taking the time to submit this pull request. As per contribution policy, please ensure your PR consists of a single commit. Could you please change your commits accordingly?

[zhangxiangnan]

Done — squashed into a single commit and rebased onto main.

[hemasekhar-p]

@zhangxiangnan, thank you for addressing the pre checks. Currently this PR is under review by our team, we will keep you posted if any additional information is required. thank you.