Copilot commented on code in PR #4819: URL: https://github.com/apache/texera/pull/4819#discussion_r3177611402
########## amber/src/test/scala/org/apache/texera/amber/engine/architecture/messaginglayer/CongestionControlSpec.scala: ########## @@ -0,0 +1,153 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.texera.amber.engine.architecture.messaginglayer + +import org.apache.texera.amber.core.virtualidentity.{ActorVirtualIdentity, ChannelIdentity} +import org.apache.texera.amber.engine.architecture.common.WorkflowActor.NetworkMessage +import org.apache.texera.amber.engine.common.ambermessage.{ + WorkflowFIFOMessage, + WorkflowFIFOMessagePayload +} +import org.scalatest.flatspec.AnyFlatSpec +import org.scalatest.matchers.should.Matchers + +class CongestionControlSpec extends AnyFlatSpec with Matchers { + + private case object StubPayload extends WorkflowFIFOMessagePayload + private val channelId = ChannelIdentity( + ActorVirtualIdentity("from"), + ActorVirtualIdentity("to"), + isControl = false + ) + + private def netMsg(id: Long): NetworkMessage = + NetworkMessage(id, WorkflowFIFOMessage(channelId, sequenceNumber = id, StubPayload)) + + // ----- canSend / windowSize default ----- + + "CongestionControl" should "start with windowSize=1 (so canSend is true at zero in-flight)" in { + val cc = new CongestionControl + cc.canSend shouldBe true + } + + // ----- enqueue + send ----- + + "enqueueMessage + getBufferedMessagesToSend" should "release up to (windowSize - inTransit) messages" in { + val cc = new CongestionControl + cc.enqueueMessage(netMsg(0)) + cc.enqueueMessage(netMsg(1)) + cc.enqueueMessage(netMsg(2)) + val released = cc.getBufferedMessagesToSend.toList + // windowSize = 1, no in-flight yet → exactly one message released. + released should have length 1 + released.head.messageId shouldBe 0L + } + + it should "release nothing once inTransit reaches windowSize" in { + val cc = new CongestionControl + cc.markMessageInTransit(netMsg(0)) + // windowSize = 1, inTransit = 1 → already saturated. + cc.canSend shouldBe false + cc.enqueueMessage(netMsg(1)) + cc.getBufferedMessagesToSend.toList shouldBe empty + } + + // ----- markMessageInTransit ----- + + "markMessageInTransit" should "register the message in inTransit and stamp send time" in { + val cc = new CongestionControl + cc.markMessageInTransit(netMsg(7)) + cc.getInTransitMessages.map(_.messageId).toList shouldBe List(7L) + // No throw means sentTime got registered too — ack uses it next. + cc.ack(7L) + } + + // ----- ack: in-window growth ----- + + "ack within ackTimeLimit" should "double the window during slow start, then increment linearly past the threshold" in { + val cc = new CongestionControl + // ssThreshold defaults to 16 and windowSize to 1. Five quick acks should + // double 1→2→4→8→16, then increment to 17 on the next ack. + for (i <- 0 until 5) { + cc.markMessageInTransit(netMsg(i)) + cc.ack(i) + } + // After 5 acks: windowSize moves 1→2→4→8→16→17 (the fifth ack hits the + // linear branch because windowSize == ssThreshold == 16). + cc.getStatusReport should include("current window size = 17") + } + + it should "be a no-op for an unknown message id" in { + val cc = new CongestionControl + cc.markMessageInTransit(netMsg(0)) + val statusBefore = cc.getStatusReport + cc.ack(99) // not in inTransit + cc.getStatusReport shouldBe statusBefore + } + + // ----- ack: timeout shrinks the window ----- + + "ack outside ackTimeLimit" should "halve ssThreshold and reset windowSize to ssThreshold" in { + val cc = new CongestionControl + cc.markMessageInTransit(netMsg(0)) + // Forge a stale send time by sleeping past the 3s ack window. Skip in + // the regular path; instead use direct stamp via reflection-free path: + // we drain enough acks to grow windowSize, then artificially time out + // the next one using setSentTime on the underlying map. But the field + // is private — easiest is to drive a real timeout by sleeping. To keep + // the suite fast, this test acks exactly within the limit and pins the + // happy-path math. Out-of-window ack semantics are documented here as + // a comment; an integration test with controlled clock would be the + // right place to exercise the halving branch. + cc.ack(0) + cc.getStatusReport should include("current window size = 2") + } + + // ----- timed-out in-transit ----- + + "getTimedOutInTransitMessages" should "return nothing when nothing has been sitting past resendTimeLimit" in { + val cc = new CongestionControl + cc.markMessageInTransit(netMsg(0)) + cc.getTimedOutInTransitMessages.toList shouldBe empty + } + + // ----- getInTransitMessages / getAllMessages ----- + + "getAllMessages" should "return the union of inTransit and toBeSent buffers" in { + val cc = new CongestionControl + cc.markMessageInTransit(netMsg(0)) + cc.enqueueMessage(netMsg(1)) + cc.enqueueMessage(netMsg(2)) + val all = cc.getAllMessages.map(_.messageId).toSet + all shouldBe Set(0L, 1L, 2L) + } + + // ----- status report format ----- + + "getStatusReport" should "format the three core counters in the documented order" in { + val cc = new CongestionControl + cc.markMessageInTransit(netMsg(0)) + cc.enqueueMessage(netMsg(1)) + val status = cc.getStatusReport + status should include("current window size") + status should include("in transit = 1") + status should include("waiting = 1") Review Comment: The test name says it verifies the report format/order, but these assertions only check that three substrings appear somewhere in the string. A reorder of the fields or separator changes would still pass, so this doesn't actually guard the advertised formatting contract. ########## amber/src/test/scala/org/apache/texera/amber/engine/architecture/messaginglayer/CongestionControlSpec.scala: ########## @@ -0,0 +1,153 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.texera.amber.engine.architecture.messaginglayer + +import org.apache.texera.amber.core.virtualidentity.{ActorVirtualIdentity, ChannelIdentity} +import org.apache.texera.amber.engine.architecture.common.WorkflowActor.NetworkMessage +import org.apache.texera.amber.engine.common.ambermessage.{ + WorkflowFIFOMessage, + WorkflowFIFOMessagePayload +} +import org.scalatest.flatspec.AnyFlatSpec +import org.scalatest.matchers.should.Matchers + +class CongestionControlSpec extends AnyFlatSpec with Matchers { + + private case object StubPayload extends WorkflowFIFOMessagePayload + private val channelId = ChannelIdentity( + ActorVirtualIdentity("from"), + ActorVirtualIdentity("to"), + isControl = false + ) + + private def netMsg(id: Long): NetworkMessage = + NetworkMessage(id, WorkflowFIFOMessage(channelId, sequenceNumber = id, StubPayload)) + + // ----- canSend / windowSize default ----- + + "CongestionControl" should "start with windowSize=1 (so canSend is true at zero in-flight)" in { + val cc = new CongestionControl + cc.canSend shouldBe true + } + + // ----- enqueue + send ----- + + "enqueueMessage + getBufferedMessagesToSend" should "release up to (windowSize - inTransit) messages" in { + val cc = new CongestionControl + cc.enqueueMessage(netMsg(0)) + cc.enqueueMessage(netMsg(1)) + cc.enqueueMessage(netMsg(2)) + val released = cc.getBufferedMessagesToSend.toList + // windowSize = 1, no in-flight yet → exactly one message released. + released should have length 1 + released.head.messageId shouldBe 0L + } + + it should "release nothing once inTransit reaches windowSize" in { + val cc = new CongestionControl + cc.markMessageInTransit(netMsg(0)) + // windowSize = 1, inTransit = 1 → already saturated. + cc.canSend shouldBe false + cc.enqueueMessage(netMsg(1)) + cc.getBufferedMessagesToSend.toList shouldBe empty + } + + // ----- markMessageInTransit ----- + + "markMessageInTransit" should "register the message in inTransit and stamp send time" in { + val cc = new CongestionControl + cc.markMessageInTransit(netMsg(7)) + cc.getInTransitMessages.map(_.messageId).toList shouldBe List(7L) + // No throw means sentTime got registered too — ack uses it next. + cc.ack(7L) + } + + // ----- ack: in-window growth ----- + + "ack within ackTimeLimit" should "double the window during slow start, then increment linearly past the threshold" in { + val cc = new CongestionControl + // ssThreshold defaults to 16 and windowSize to 1. Five quick acks should + // double 1→2→4→8→16, then increment to 17 on the next ack. + for (i <- 0 until 5) { + cc.markMessageInTransit(netMsg(i)) + cc.ack(i) + } + // After 5 acks: windowSize moves 1→2→4→8→16→17 (the fifth ack hits the + // linear branch because windowSize == ssThreshold == 16). + cc.getStatusReport should include("current window size = 17") + } + + it should "be a no-op for an unknown message id" in { + val cc = new CongestionControl + cc.markMessageInTransit(netMsg(0)) + val statusBefore = cc.getStatusReport + cc.ack(99) // not in inTransit + cc.getStatusReport shouldBe statusBefore + } + + // ----- ack: timeout shrinks the window ----- + + "ack outside ackTimeLimit" should "halve ssThreshold and reset windowSize to ssThreshold" in { + val cc = new CongestionControl + cc.markMessageInTransit(netMsg(0)) + // Forge a stale send time by sleeping past the 3s ack window. Skip in + // the regular path; instead use direct stamp via reflection-free path: + // we drain enough acks to grow windowSize, then artificially time out + // the next one using setSentTime on the underlying map. But the field + // is private — easiest is to drive a real timeout by sleeping. To keep + // the suite fast, this test acks exactly within the limit and pins the + // happy-path math. Out-of-window ack semantics are documented here as + // a comment; an integration test with controlled clock would be the + // right place to exercise the halving branch. + cc.ack(0) + cc.getStatusReport should include("current window size = 2") Review Comment: This example never waits past `ackTimeLimit`; it immediately acks message `0` and then asserts the same `windowSize = 2` outcome as the happy-path test above. That means the timeout/backoff branch in `CongestionControl.ack` still has no coverage, even though this test name claims to exercise it. ########## amber/src/test/scala/org/apache/texera/amber/engine/architecture/messaginglayer/CongestionControlSpec.scala: ########## @@ -0,0 +1,153 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.texera.amber.engine.architecture.messaginglayer + +import org.apache.texera.amber.core.virtualidentity.{ActorVirtualIdentity, ChannelIdentity} +import org.apache.texera.amber.engine.architecture.common.WorkflowActor.NetworkMessage +import org.apache.texera.amber.engine.common.ambermessage.{ + WorkflowFIFOMessage, + WorkflowFIFOMessagePayload +} +import org.scalatest.flatspec.AnyFlatSpec +import org.scalatest.matchers.should.Matchers + +class CongestionControlSpec extends AnyFlatSpec with Matchers { + + private case object StubPayload extends WorkflowFIFOMessagePayload + private val channelId = ChannelIdentity( + ActorVirtualIdentity("from"), + ActorVirtualIdentity("to"), + isControl = false + ) + + private def netMsg(id: Long): NetworkMessage = + NetworkMessage(id, WorkflowFIFOMessage(channelId, sequenceNumber = id, StubPayload)) + + // ----- canSend / windowSize default ----- + + "CongestionControl" should "start with windowSize=1 (so canSend is true at zero in-flight)" in { + val cc = new CongestionControl + cc.canSend shouldBe true + } + + // ----- enqueue + send ----- + + "enqueueMessage + getBufferedMessagesToSend" should "release up to (windowSize - inTransit) messages" in { + val cc = new CongestionControl + cc.enqueueMessage(netMsg(0)) + cc.enqueueMessage(netMsg(1)) + cc.enqueueMessage(netMsg(2)) + val released = cc.getBufferedMessagesToSend.toList + // windowSize = 1, no in-flight yet → exactly one message released. + released should have length 1 + released.head.messageId shouldBe 0L + } + + it should "release nothing once inTransit reaches windowSize" in { + val cc = new CongestionControl + cc.markMessageInTransit(netMsg(0)) + // windowSize = 1, inTransit = 1 → already saturated. + cc.canSend shouldBe false + cc.enqueueMessage(netMsg(1)) + cc.getBufferedMessagesToSend.toList shouldBe empty + } + + // ----- markMessageInTransit ----- + + "markMessageInTransit" should "register the message in inTransit and stamp send time" in { + val cc = new CongestionControl + cc.markMessageInTransit(netMsg(7)) + cc.getInTransitMessages.map(_.messageId).toList shouldBe List(7L) + // No throw means sentTime got registered too — ack uses it next. + cc.ack(7L) + } + + // ----- ack: in-window growth ----- + + "ack within ackTimeLimit" should "double the window during slow start, then increment linearly past the threshold" in { + val cc = new CongestionControl + // ssThreshold defaults to 16 and windowSize to 1. Five quick acks should + // double 1→2→4→8→16, then increment to 17 on the next ack. + for (i <- 0 until 5) { + cc.markMessageInTransit(netMsg(i)) + cc.ack(i) + } + // After 5 acks: windowSize moves 1→2→4→8→16→17 (the fifth ack hits the + // linear branch because windowSize == ssThreshold == 16). + cc.getStatusReport should include("current window size = 17") + } + + it should "be a no-op for an unknown message id" in { + val cc = new CongestionControl + cc.markMessageInTransit(netMsg(0)) + val statusBefore = cc.getStatusReport + cc.ack(99) // not in inTransit + cc.getStatusReport shouldBe statusBefore + } + + // ----- ack: timeout shrinks the window ----- + + "ack outside ackTimeLimit" should "halve ssThreshold and reset windowSize to ssThreshold" in { + val cc = new CongestionControl + cc.markMessageInTransit(netMsg(0)) + // Forge a stale send time by sleeping past the 3s ack window. Skip in + // the regular path; instead use direct stamp via reflection-free path: + // we drain enough acks to grow windowSize, then artificially time out + // the next one using setSentTime on the underlying map. But the field + // is private — easiest is to drive a real timeout by sleeping. To keep + // the suite fast, this test acks exactly within the limit and pins the + // happy-path math. Out-of-window ack semantics are documented here as + // a comment; an integration test with controlled clock would be the + // right place to exercise the halving branch. + cc.ack(0) + cc.getStatusReport should include("current window size = 2") + } + + // ----- timed-out in-transit ----- + + "getTimedOutInTransitMessages" should "return nothing when nothing has been sitting past resendTimeLimit" in { + val cc = new CongestionControl + cc.markMessageInTransit(netMsg(0)) + cc.getTimedOutInTransitMessages.toList shouldBe empty + } + Review Comment: This only checks the fresh-message case. The branch that returns timed-out in-flight messages—the one `AkkaMessageTransferService.checkResend()` uses to trigger retransmission—remains untested, so regressions in resend bookkeeping or the timeout comparison would still slip through. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
