This is an automated email from the ASF dual-hosted git repository. He-Pin pushed a commit to branch fix/cancel-receive-timeout-npe in repository https://gitbox.apache.org/repos/asf/pekko.git
commit c20332f172093ae9aa33c858b27661d3d8a057b2 Author: 虎鸣 <[email protected]> AuthorDate: Fri Jun 19 06:20:14 2026 +0800 fix(actor-typed): cancelReceiveTimeout 后丢弃 stale ReceiveTimeout 避免 NPE Motivation: cancelReceiveTimeout() 将 receiveTimeoutMsg 置 null, 但已通过 scheduleOnce 入队邮箱的 classic ReceiveTimeout 无法撤回; stale 消息以 null 传入 typed 行为栈, 导致 InterceptorImpl.receive 中 msg.getClass 抛出 NullPointerException。 Modification: ActorAdapter.aroundReceive 在转发 receiveTimeoutMsg 前做 null 检查, stale timeout 直接丢弃; 新增 CancelReceiveTimeoutSpec 回归测试。 Result: typed actor 在 cancelReceiveTimeout 与 stale ReceiveTimeout 竞态下 不再崩溃, 行为栈对调用方透明。 --- .../actor/typed/CancelReceiveTimeoutSpec.scala | 98 ++++++++++++++++++++++ .../typed/internal/adapter/ActorAdapter.scala | 6 +- 2 files changed, 103 insertions(+), 1 deletion(-) diff --git a/actor-typed-tests/src/test/scala/org/apache/pekko/actor/typed/CancelReceiveTimeoutSpec.scala b/actor-typed-tests/src/test/scala/org/apache/pekko/actor/typed/CancelReceiveTimeoutSpec.scala new file mode 100644 index 0000000000..a1547e1930 --- /dev/null +++ b/actor-typed-tests/src/test/scala/org/apache/pekko/actor/typed/CancelReceiveTimeoutSpec.scala @@ -0,0 +1,98 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * license agreements; and to You under the Apache License, version 2.0: + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * This file is part of the Apache Pekko project, which was derived from Akka. + */ + +package org.apache.pekko.actor.typed + +import scala.concurrent.duration._ +import scala.reflect.ClassTag + +import org.apache.pekko +import pekko.actor.testkit.typed.scaladsl.LogCapturing +import pekko.actor.testkit.typed.scaladsl.ScalaTestWithActorTestKit +import pekko.actor.testkit.typed.scaladsl.TestProbe +import pekko.actor.typed.scaladsl.Behaviors +import pekko.actor.typed.scaladsl.adapter._ + +import org.scalatest.wordspec.AnyWordSpecLike + +class CancelReceiveTimeoutSpec extends ScalaTestWithActorTestKit with AnyWordSpecLike with LogCapturing { + + sealed trait Command + case object SetAndCancelTimeout extends Command + case object Timeout extends Command + case object Ping extends Command + case object Pong extends Event + + sealed trait Event + case object Done extends Event + + // A no-op interceptor that deepens the behavior stack so the message + // passes through InterceptorImpl.receive — the exact code path where + // msg.getClass throws NPE on a null message. + private def noopInterceptor[T: ClassTag] = new BehaviorInterceptor[T, T] { + override def aroundReceive( + ctx: TypedActorContext[T], + msg: T, + target: BehaviorInterceptor.ReceiveTarget[T]): Behavior[T] = + target(ctx, msg) + } + + "A typed actor with receive timeout" must { + + // Regression test for #3084: cancelReceiveTimeout() sets receiveTimeoutMsg to + // null, but a classic ReceiveTimeout already enqueued in the mailbox cannot be + // retracted. Before the fix the stale ReceiveTimeout was forwarded to the typed + // behavior stack as a null message, causing NPE in InterceptorImpl.receive. + "silently discard a stale ReceiveTimeout after cancelReceiveTimeout" in { + val probe = TestProbe[Event]() + + def behavior: Behavior[Command] = + Behaviors.setup { context => + // Wrap in an interceptor so the message traverses InterceptorImpl.receive, + // the exact location where msg.getClass throws NPE on a null message. + Behaviors.intercept[Command, Command](() => noopInterceptor[Command]) { + Behaviors.receiveMessage { + case SetAndCancelTimeout => + // Set a very long timeout (won't actually fire), then immediately + // cancel it. This leaves receiveTimeoutMsg as null. + context.setReceiveTimeout(1.hour, Timeout) + context.cancelReceiveTimeout() + probe.ref ! Done + Behaviors.same + + case Timeout => + // Should never reach here: the timeout was cancelled and we send + // the classic ReceiveTimeout ourselves to simulate the stale case. + Behaviors.unhandled + + case Ping => + probe.ref ! Pong + Behaviors.same + } + } + } + + val ref = spawn(behavior) + + // Step 1: actor sets and cancels the receive timeout → receiveTimeoutMsg is now null + ref ! SetAndCancelTimeout + probe.expectMessage(Done) + + // Step 2: simulate a stale classic ReceiveTimeout arriving in the mailbox. + // This is what happens when the scheduler fires ReceiveTimeout before + // cancelReceiveTimeout() is processed but the actor dequeues them in the + // opposite order. + ref.toClassic ! pekko.actor.ReceiveTimeout + + // Step 3: verify the actor is still alive and responsive (no NPE crash). + ref ! Ping + probe.expectMessage(Pong) + } + } +} diff --git a/actor-typed/src/main/scala/org/apache/pekko/actor/typed/internal/adapter/ActorAdapter.scala b/actor-typed/src/main/scala/org/apache/pekko/actor/typed/internal/adapter/ActorAdapter.scala index 5b7b131b6c..7bee9f44f1 100644 --- a/actor-typed/src/main/scala/org/apache/pekko/actor/typed/internal/adapter/ActorAdapter.scala +++ b/actor-typed/src/main/scala/org/apache/pekko/actor/typed/internal/adapter/ActorAdapter.scala @@ -100,7 +100,11 @@ import pekko.util.OptionVal } else Terminated(ActorRefAdapter(ref)) handleSignal(msg) case classic.ReceiveTimeout => - handleMessage(ctx.receiveTimeoutMsg) + // cancelReceiveTimeout() sets receiveTimeoutMsg to null, but a classic ReceiveTimeout + // that was already enqueued in the mailbox before the cancel cannot be retracted. + // Discard the stale timeout to avoid passing null into the typed behavior stack (#3084). + val timeoutMsg = ctx.receiveTimeoutMsg + if (timeoutMsg != null) handleMessage(timeoutMsg) case wrapped: AdaptMessage[Any, T] @unchecked => withSafelyAdapted(() => wrapped.adapt()) { case AdaptWithRegisteredMessageAdapter(msg) => --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
