Skip stale JetStream pull statuses in paced consumer

CHANGELOG.md

@@ -5,6 +5,17 @@ All notable changes to this project are documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
+## [Unreleased]
+
+### Fixed
+
+- The paced consume engine (`KeyValue.watchPaced` / `watchAllPaced`, `StreamContext.createOrderedPacedConsumer` /
+  `getPacedConsumerContext`) mistook the server's routine expiry of an idle pull for an early pull terminus: the local window deadline
+  (armed at publish) always fired before the server's `expiresIn` clock (started at receipt), so the server's `408` landed at the head of
+  the next window and tripped the 500 ms early-terminus guard — a non-draining sleep and a redundant second pull per idle window.
+  Statuses are now matched to the pull they answer via jnats's per-pull reply subjects, and stale ones are skipped. No API or
+  configuration change; the callback engine is unaffected.
+
 ## [1.3.1] - 2026-07-28
 
 ### Added

docs/design/2026-07-13-paced-buffered-consumers.md

@@ -88,9 +88,13 @@ permanent configuration errors, while still absorbing transient ones; after the
 resubscribes retry indefinitely.
 
 `stop` lets the in-flight message finish and the loop exit at the next window boundary; `close`
-is prompt while waiting but also lets an in-flight message finish (the step is masked). A pull
-terminus arriving well before the window deadline is guarded with a 500ms sleep so terminus
-storms cannot re-pull in a hot loop; the guard self-disables for sub-second windows.
+is prompt while waiting but also lets an in-flight message finish (the step is masked). Statuses
+are matched to the pull they answer (jnats publishes every pull with a distinct reply subject);
+a stale status - typically the server's expiry `408` for a pull the engine already ended at its
+local deadline, which always fires first by one round trip - is skipped instead of ending the
+live window. A genuine pull terminus arriving well before the window deadline is guarded with a
+500ms sleep so terminus storms cannot re-pull in a hot loop; the guard self-disables for
+sub-second windows.
 
 ## Observability
 
@@ -106,11 +110,12 @@ the jnats connection `ErrorListener`, so existing log wiring works without a cus
   not interrupt; close waits for the in-flight message).
 - `PullStatusInterpreterSpec` — pins the status table so a jnats upgrade that shifts semantics
   fails loudly (the most upgrade-sensitive piece); `BufferedPullTransportSpec` pins the JetStream
-  verification in the classification.
+  verification and the stale-status gate in the classification.
 - `PacedOrderedConsumerContextSpec` / `PacedConsumerContextSpec` / `PacedKeyValueSpec` —
   server-backed behavior of the public API: warmup, recovery after consumer deletion, sequential
   continuation, stop, prompt idle release, handler failure, durable ack/redelivery, concurrent
-  consumes, KV watching.
+  consumes, KV watching, and the stale-terminus regression test (an idle consumer issues one
+  pull per window expiry, not two).
 - `PacedStressSpec` — the behavioral bar: 8 watchers x 1500 keys with a slow handler; zero drops,
   zero consumer recreations, strict revision order, pulls bounded by the batch budget.
 

jetstream/src/main/scala/com/evolution/natseffect/jetstream/impl/BufferedPullTransport.scala

@@ -1,10 +1,11 @@
 package com.evolution.natseffect.jetstream.impl
 
 import cats.effect.std.{Dispatcher, Queue}
-import cats.effect.{Async, Resource}
+import cats.effect.{Async, Ref, Resource}
 import cats.syntax.all.*
 import com.evolution.natseffect.impl.{CEMessageHandler, ConfiguringMessageHandler, JConnection, JMessage}
 import com.evolution.natseffect.jetstream.impl.PacedPullEngine.{ActiveSubscription, Directive}
+import io.nats.client.impl.JNatsPullSubscriptionImplOps
 import io.nats.client.{MessageHandler, PullSubscribeOptions}
 
 /** Transport for the paced engine: a JetStream pull subscription attached to a dispatcher whose exclusive default handler feeds a queue as
@@ -65,22 +66,45 @@ private[natseffect] object BufferedPullTransport {
           .delay(Option(subscription.getConsumerName))
           .flatMap(_.liftTo[F](new IllegalStateException("The pull subscription reports no consumer name")))
       )
+      // jnats publishes every pull request with its own reply subject, and the server addresses everything
+      // belonging to that pull, statuses included, to that subject. The subject is taken from the same internal
+      // jnats call that computes it (see `pullReturningSubject`), so it cannot drift from what is on the wire.
+      // Nothing can arrive before the first pull, so the initial value only needs to never equal a real subject.
+      lastPullSubject <- Resource.eval(Ref.of[F, String](""))
     } yield ActiveSubscription(
       consumerName = consumerName,
-      pull = options => Async[F].delay(subscription.pull(options)),
+      pull = options => subscription.pullReturningSubject(options).flatMap(lastPullSubject.set),
       // 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)),
+      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)
     )
 
   /** Classification of everything the subscription queue can carry, fused onto the take: statuses go to the interpreter, and data messages
     * are verified to be JetStream messages (anything else on the deliver subject is dropped as noise) before the consumer-type inspection
     * vets them. This is the single site that upholds what [[PacedPullEngine.Directive.Deliver]] promises - a genuine JetStream message.
+    *
+    * <p>A status is interpreted only when it answers the pull that is live now (statuses arrive on the reply subject of the pull they
+    * belong to, see `currentPullSubject` above); a status for an earlier pull is skipped as stale. The routine case: the engine's local
+    * window deadline is armed when the pull is published, while the server starts the same `expiresIn` clock only on receipt, so on an idle
+    * consumer the local deadline always fires first and the server's expiry `408` for the pull that just ended arrives at the head of the
+    * *next* window. Let through, it would end that window on the spot - indistinguishable from a dead consumer answering instantly - which
+    * cost a 500ms early-terminus guard sleep and a redundant second pull per window on a perfectly healthy consumer. Data messages carry no
+    * pull identity worth checking: a straggler delivered on the previous pull is still a wanted message and is handled normally.
     */
-  private[natseffect] def classify[F[_]: Async](inspect: JMessage => F[Directive.DataDirective])(message: JMessage): F[Directive] =
+  private[natseffect] def classify[F[_]: Async](
+    inspect: JMessage => F[Directive.DataDirective],
+    currentPullSubject: F[String]
+  )(message: JMessage): F[Directive] =
     Option(message.getStatus) match {
-      case Some(status)                 => (PullStatusInterpreter.interpret(status): Directive).pure[F]
+      case Some(status) =>
+        currentPullSubject
+          .map(pullSubject =>
+            if (message.getSubject == pullSubject) PullStatusInterpreter.interpret(status)
+            else Directive.Skip
+          )
+          .widen
       case None if !message.isJetStream => (Directive.Skip: Directive).pure[F]
       case None                         => inspect(message).widen
     }

jetstream/src/main/scala/com/evolution/natseffect/jetstream/impl/PacedPullEngine.scala

@@ -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]
   )
 
@@ -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
@@ -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))
+                          )
+                      }
                   }
-              }
+            }
         }
       }
     }

jetstream/src/main/scala/io/nats/client/impl/package.scala

@@ -5,6 +5,22 @@ import com.evolution.natseffect.jetstream.impl.JKeyValue
 
 package object impl {
 
+  /** We defer to the NatsJetStreamPullSubscription implementation for the per-pull reply subject, not to mirror its internal pull counter.
+    */
+  implicit class JNatsPullSubscriptionImplOps(a: JetStreamSubscription) {
+
+    /** Like `pull(PullRequestOptions)` (same `raiseStatusWarnings`, no observer), but returns the reply subject the pull request was
+      * published with - the subscription's wildcard inbox with `*` replaced by the internal pull counter. The server addresses everything
+      * belonging to the pull, statuses included, to this subject.
+      */
+    def pullReturningSubject[F[_]](options: PullRequestOptions)(implicit F: Sync[F]): F[String] = F.delay {
+      a match {
+        case impl: NatsJetStreamPullSubscription => impl._pull(options, true, null)
+        case _ => throw new IllegalArgumentException("Subscription is not a NatsJetStreamPullSubscription")
+      }
+    }
+  }
+
   /** We defer to NatsKeyValue implementation for stream and subject values calculation, not to reimplement the internal logic
     */
   implicit class JNatsKeyValueImplOps(a: JKeyValue) {

jetstream/src/test/scala/com/evolution/natseffect/jetstream/BufferedPullTransportSpec.scala

@@ -5,25 +5,34 @@ import com.evolution.natseffect.impl.JMessage
 import com.evolution.natseffect.jetstream.impl.BufferedPullTransport
 import com.evolution.natseffect.jetstream.impl.PacedPullEngine.Directive
 import io.nats.client.impl.NatsMessage
+import io.nats.client.support.Status
 import weaver.SimpleIOSuite
 
-/** The transport's classification fusion: non-JetStream messages are dropped before inspection, and genuine JetStream data messages reach
-  * the consumer-type inspection. This is the single site that upholds the `Directive.Deliver` promise of a verified JetStream message (the
-  * status branch of the fusion is pinned by `PullStatusInterpreterSpec` and exercised end-to-end by the server-backed specs).
+/** The transport's classification fusion: non-JetStream messages are dropped before inspection, genuine JetStream data messages reach the
+  * consumer-type inspection, and a status is interpreted only when it answers the current pull - a stale status is skipped. (The
+  * status-to-directive mapping itself is pinned by `PullStatusInterpreterSpec` and exercised end-to-end by the server-backed specs.)
   */
 object BufferedPullTransportSpec extends SimpleIOSuite {
 
+  private val CurrentPullSubject = "inbox.sub.2"
+
   private val inspect: JMessage => IO[Directive.DataDirective] =
     message => IO.pure(Directive.Deliver(message))
 
   private def classify(message: JMessage): IO[Directive] =
-    BufferedPullTransport.classify[IO](inspect)(message)
+    BufferedPullTransport.classify[IO](inspect, IO.pure(CurrentPullSubject))(message)
 
   private def jetStreamMessage(subject: String): JMessage =
     new NatsMessage(subject, null, Array.emptyByteArray) {
       override def isJetStream: Boolean = true
     }
 
+  private def statusMessage(subject: String): JMessage =
+    new NatsMessage(subject, null, Array.emptyByteArray) {
+      override def isStatusMessage: Boolean = true
+      override def getStatus: Status        = new Status(Status.REQUEST_TIMEOUT_CODE, "Request Timeout")
+    }
+
   test("non-JetStream messages are skipped before inspection") {
     val plain = NatsMessage.builder().subject("test").data("plain".getBytes).build()
     classify(plain).map(directive => expect.same(directive, Directive.Skip))
@@ -33,4 +42,12 @@ object BufferedPullTransportSpec extends SimpleIOSuite {
     val message = jetStreamMessage("test.1")
     classify(message).map(directive => expect.same(directive, Directive.Deliver(message)))
   }
+
+  test("a status answering the current pull is interpreted") {
+    classify(statusMessage(CurrentPullSubject)).map(directive => expect.same(directive, Directive.PullOver))
+  }
+
+  test("a status for a previous pull is stale and skipped") {
+    classify(statusMessage("inbox.sub.1")).map(directive => expect.same(directive, Directive.Skip))
+  }
 }