Skip to content

Take already-buffered messages without the per-take timeout in the paced engine. #23

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
DeviLab merged 1 commit into from
Aug 3, 2026
Merged

Take already-buffered messages without the per-take timeout in the paced engine. #23

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
1 change: 1 addition & 0 deletions jetstream/src/main/scala/com/evolution/natseffect/jetstream/impl/BufferedPullTransport.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -77,6 +77,7 @@ private[natseffect] object BufferedPullTransport {
// Per the ActiveSubscription contract, poll applies to the wait alone: a taken message is
// always classified, whatever cancellation is in flight
next = poll => poll(queue.take).flatMap(classify(inspect, lastPullSubject.get)),
tryNext = queue.tryTake.flatMap(_.traverse(classify(inspect, lastPullSubject.get))),
isActive = Async[F].delay(subscription.isActive)
)

Expand Down
96 changes: 59 additions & 37 deletions jetstream/src/main/scala/com/evolution/natseffect/jetstream/impl/PacedPullEngine.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -91,12 +91,14 @@ private[natseffect] object PacedPullEngine {
/** One established subscription, as the engine sees it: a step yielding the next classified [[Directive]], the pull handle, and the
* liveness check that tells a dead subscription apart from an idle one (the step just goes quiet either way). `next` must apply the
* given `Poll` to the message wait alone, so that waiting is the only cancelable point of the engine's masked take-to-handle step and a
* taken message is always classified. Torn down as a whole by the Resource that produced it.
* taken message is always classified. `tryNext` is the non-waiting form of the same step: `Some` exactly when a message was already
* buffered, classified like `next` would. Torn down as a whole by the Resource that produced it.
*/
final case class ActiveSubscription[F[_]](
consumerName: String,
pull: PullRequestOptions => F[Unit],
next: Poll[F] => F[Directive],
tryNext: F[Option[Directive]],
isActive: F[Boolean]
)

Expand Down Expand Up @@ -214,6 +216,16 @@ private[natseffect] object PacedPullEngine {
}
}

// The non-waiting form of the `awaitDirective`, for messages that are already buffered. There is
// nothing to cancel as there is no wait.
def pollDirective(active: ActiveSubscription[F]): F[Option[Directive]] =
F.uncancelable { _ =>
active.tryNext.flatTap {
case Some(Directive.Deliver(message)) => handleMessage(message)
case _ => F.unit
}
}

def probeAlive(state: Ref[F, SubscriptionState]): F[Boolean] =
state.get.flatMap { s =>
// Only a definite "consumer not found" counts as dead; transient lookup errors must not
Expand Down Expand Up @@ -328,44 +340,54 @@ private[natseffect] object PacedPullEngine {
def windowOver(early: Boolean): WindowOutcome =
if (deliveredAny) WindowOutcome.Handled else WindowOutcome.Idle(early)

def continueAfter(directive: Directive): F[WindowOutcome] =
directive match {
case Directive.Deliver(_) =>
// Already handled inside the masked step
if (remaining <= 1) F.pure[WindowOutcome](WindowOutcome.Handled)
else drainWindow(state, active, deadline, remaining - 1, deliveredAny = true)
case Directive.Skip => drainWindow(state, active, deadline, remaining, deliveredAny)
case Directive.PullOver =>
// The pull terminated server-side; ending well before the deadline is what
// pullLoop's hot-loop guard keys on
F.monotonic.map(end => windowOver(early = (deadline - end) > EarlyEmptyThreshold))
case Directive.Restart(reason) =>
F.pure[WindowOutcome](CycleEnd.Resubscribe(reason, Duration.Zero))
case Directive.Fail(e) =>
// The one recovery that is not a return value: raised so subscribeLoop's
// attempt routes it through the progressive-backoff path
F.raiseError[WindowOutcome](e)
}

unlessStopped[WindowOutcome](state)(CycleEnd.Stopped) {
F.monotonic.flatMap { now =>
val timeLeft = deadline - now
if (timeLeft <= Duration.Zero) F.pure(windowOver(early = false))
else
awaitDirective(active)
.timeout(timeLeft.max(1.milli))
.flatMap {
case Directive.Deliver(_) =>
// Already handled inside the masked step
if (remaining <= 1) F.pure[WindowOutcome](WindowOutcome.Handled)
else drainWindow(state, active, deadline, remaining - 1, deliveredAny = true)
case Directive.Skip => drainWindow(state, active, deadline, remaining, deliveredAny)
case Directive.PullOver =>
// The pull terminated server-side; ending well before the deadline is what
// pullLoop's hot-loop guard keys on
F.monotonic.map(end => windowOver(early = (deadline - end) > EarlyEmptyThreshold))
case Directive.Restart(reason) =>
F.pure[WindowOutcome](CycleEnd.Resubscribe(reason, Duration.Zero))
case Directive.Fail(e) =>
// The one recovery that is not a return value: raised so subscribeLoop's
// attempt routes it through the progressive-backoff path
F.raiseError[WindowOutcome](e)
}
.recoverWith {
case _: TimeoutException =>
// The window expired waiting (never "early") - or the deadline fired during a
// masked step, whose message was still fully processed and only this window's
// label was lost. A dead subscription goes quiet the same way, so distinguish
// via the liveness handle
active.isActive.flatMap {
case true => F.pure(windowOver(early = false))
case false =>
unlessStopped[WindowOutcome](state)(CycleEnd.Stopped)(
F.pure(CycleEnd.Resubscribe("subscription inactive", InitialRetryDelay))
)
// Hot path first: an already-buffered message is taken without the per-take deadline
// machinery below (a timer registration and racing fibers per take - measured to halve a
// fast handler's throughput).
pollDirective(active).flatMap {
case Some(directive) => continueAfter(directive)
case None =>
F.monotonic.flatMap { now =>
val timeLeft = deadline - now
if (timeLeft <= Duration.Zero) F.pure(windowOver(early = false))
else
awaitDirective(active)
.timeout(timeLeft.max(1.milli))
.flatMap(continueAfter)
.recoverWith {
case _: TimeoutException =>
// The window expired waiting (never "early") - or the deadline fired during a
// masked step, whose message was still fully processed and only this window's
// label was lost. A dead subscription goes quiet the same way, so distinguish
// via the liveness handle
active.isActive.flatMap {
case true => F.pure(windowOver(early = false))
case false =>
unlessStopped[WindowOutcome](state)(CycleEnd.Stopped)(
F.pure(CycleEnd.Resubscribe("subscription inactive", InitialRetryDelay))
)
}
}
}
}
}
}
}
Expand Down
41 changes: 38 additions & 3 deletions jetstream/src/test/scala/com/evolution/natseffect/jetstream/PacedPullEngineSpec.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -45,6 +45,7 @@ object PacedPullEngineSpec extends SimpleIOSuite {
consumerName = s"consumer-$n",
pull = _ => pulls.update(_ + 1),
next = poll => poll(directives.take),
tryNext = directives.tryTake,
isActive = IO.pure(true)
)
)
Expand Down Expand Up @@ -213,11 +214,19 @@ object PacedPullEngineSpec extends SimpleIOSuite {
for {
directives <- Queue.unbounded[IO, Directive]
_ <- directives.offer(Directive.Deliver(message(1)))
subscribes <- Ref.of[IO, Int](0)
pulls <- Ref.of[IO, Int](0)
completed <- Deferred[IO, Unit]

subscribe = fakeSubscribe(directives, subscribes, pulls)
// tryNext never answers, so the delivery goes through the *waiting* step - the one racing
// the window deadline - which is the path this test pins
subscribe = Resource.pure[IO, ActiveSubscription[IO]](
ActiveSubscription[IO](
consumerName = "consumer-1",
pull = _ => IO.unit,
next = poll => poll(directives.take),
tryNext = IO.pure(None),
isActive = IO.pure(true)
)
)
// The handler outlives the 300ms window; without the masked take-to-handle step the
// window timeout would cancel it mid-sleep and `completed` would never fire
handler = (_: JMessage) => IO.sleep(config.window + 200.millis) *> completed.complete(()).void
Expand All @@ -226,6 +235,32 @@ object PacedPullEngineSpec extends SimpleIOSuite {
} yield result
}

test("already-buffered messages are delivered without touching the waiting step") {
for {
directives <- Queue.unbounded[IO, Directive]
_ <- (1 to 4).toList.traverse_(n => directives.offer(Directive.Deliver(message(n))))
received <- Ref.of[IO, List[String]](Nil)
done <- Deferred[IO, Unit]

// The waiting step never yields anything: every delivery below can only have come through
// the tryNext fast path, which is what keeps the per-take deadline machinery off the hot path
subscribe = Resource.pure[IO, ActiveSubscription[IO]](
ActiveSubscription[IO](
consumerName = "consumer-1",
pull = _ => IO.unit,
next = poll => poll(IO.never),
tryNext = directives.tryTake,
isActive = IO.pure(true)
)
)
handler = (msg: JMessage) => received.updateAndGet(msg.getSubject :: _).flatMap(r => done.complete(()).void.whenA(r.size == 4))

order <- run(subscribe, handler).use { _ =>
done.get.timeout(5.seconds) *> received.get.map(_.reverse)
}
} yield expect.eql(order, List("test.1", "test.2", "test.3", "test.4"))
}

test("acquisition fails with the last error after the first-subscribe retry budget") {
for {
subscribes <- Ref.of[IO, Int](0)
Expand Down