Error in stage [RetryBidi] - "cannot push port twice"

[sebady]

This failure occurs intermittently in RetryBid on 0.9.X

Error in stage [RetryBidi]: requirement failed: Cannot push port (RetryBidi.out2) twice
java.lang.IllegalArgumentException: requirement failed: Cannot push port (RetryBidi.out2) twice
at scala.Predef$.require(Predef.scala:224)
at akka.stream.stage.GraphStageLogic.push(GraphStage.scala:463)
at org.squbs.streams.RetryBidi$$anon$1$$anon$4.onPush(RetryBidi.scala:286)
at akka.stream.impl.fusing.GraphInterpreter.processPush(GraphInterpreter.scala:747)
at akka.stream.impl.fusing.GraphInterpreter.execute(GraphInterpreter.scala:649)

[anilgursel]

Please see the code at https://github.com/paypal/squbs/blob/master/squbs-ext/src/main/scala/org/squbs/streams/RetryBidi.scala#L377

setHandler(in2, new InHandler {
      override def onPush(): Unit = {
        val (elem, context) = grab(in2)
        if (isFailure(elem)) {
          if (emitOrQueueFailure(context))
            push(out2, (elem, context))
          else if (isAvailable(out1)) pushIfReady()

          if (retryRegistry.nonEmpty && !hasBeenPulled(in2))
            pull(in2)
        } else {
          retryRegistry.get(uniqueId(context)) foreach(entry => {
            retryDelayQ.remove(entry._3)
            retryRegistry.remove(uniqueId(entry._2))
          })
          push(out2, (elem, context))
        }
      }

IMO, the problem happens when retries are exhausted. Because we push to downstream, but without getting a demand from downstream we immediately create demand on in2. For any element we actually send to downstream, we should not be creating a demand to in2 until we get a demand from out2. We should pull(in2) whenever we retry an element, but we should not do it for actual success and exhausted elements. Please see the success scenario, we do not do pull(in2), why would that scenario be different than retries exhausted scenario?

So, I think the demand creation should be inside else if:

if (emitOrQueueFailure(context)) push(out2, (elem, context))
else if (isAvailable(out1)) {
  pushIfReady()
  if (retryRegistry.nonEmpty && !hasBeenPulled(in2)) pull(in2)
}

And also, I am not sure if we need !hasBeenPulled(in2) check. Between the grab(in2) and !hasBeenPulled(in2) check, I do not see any pull(in2), so I think the check may be redundant.

[sebady]

It appear to me even in the case a retry is exhausted and element is emitted on out2.

if (emitOrQueueFailure(context)) push(out2, (elem, context))

We still need a pull(in2) to continue propagating demand on in2. Otherwise I'm not sure where that demand will come from as currently implemented.