more tests

Author: somdoron

client-server-client/src/test/scala/zio/raft/client/PendingQueriesSpec.scala

@@ -39,6 +39,39 @@ object PendingQueriesSpec extends ZIOSpecDefault {
           r <- p.await
         } yield assertTrue(r == ByteVector(7))
       }
+
+      test("resendExpired resends only timed-out queries and updates lastSentAt") {
+        for {
+          sentRef <- Ref.make(0)
+          transport = new ClientTransport {
+            def connect(address: String) = ZIO.unit
+            def disconnect() = ZIO.unit
+            def sendMessage(message: ClientMessage) = sentRef.update(_ + 1).unit
+            def incomingMessages = ZStream.empty
+          }
+
+          now <- Clock.instant
+          currentTime = now.plusSeconds(60)
+          timeout = 10.seconds
+
+          p1 <- Promise.make[Nothing, ByteVector]
+          p2 <- Promise.make[Nothing, ByteVector]
+          cid1 = CorrelationId.fromString("exp-1")
+          cid2 = CorrelationId.fromString("ok-2")
+
+          // cid1 last sent long ago (expired); cid2 recently (not expired)
+          pq0 = PendingQueries.empty
+            .add(cid1, ByteVector(1), p1, currentTime.minusSeconds(60))
+            .add(cid2, ByteVector(2), p2, currentTime.minusSeconds(5))
+
+          pq1 <- pq0.resendExpired(transport, currentTime, timeout)
+          sent <- sentRef.get
+          d1 = pq1.queries(cid1)
+          d2 = pq1.queries(cid2)
+        } yield assertTrue(sent == 1) &&
+          assertTrue(d1.lastSentAt == currentTime) &&
+          assertTrue(d2.lastSentAt == currentTime.minusSeconds(5))
+      }
     }
 }
 

client-server-client/src/test/scala/zio/raft/client/RaftClientSpec.scala

@@ -178,6 +178,45 @@ object RaftClientSpec extends ZIOSpec[TestEnvironment & ZContext] {
       }
     }
 
+    // ==================================================================
+    // Query Submission Tests
+    // ==================================================================
+
+    suiteAll("Query Submission") {
+      test("should send Query and complete promise when QueryResponse received") {
+        val port = findOpenPort
+        for {
+          mockServer <- ZSocket.server
+          _ <- mockServer.bind(s"tcp://0.0.0.0:$port")
+
+          client <- RaftClient.make(
+            clusterMembers = Map(MemberId.fromString("node1") -> s"tcp://127.0.0.1:$port"),
+            capabilities = Map("test" -> "v1")
+          )
+          _ <- client.run().forkScoped
+          _ <- client.connect()
+
+          (routingId, createMsg) <- expectMessage[CreateSession](mockServer)
+
+          sessionId <- SessionId.generate()
+          _ <- sendServerMessage(mockServer, routingId, SessionCreated(sessionId, createMsg.nonce))
+
+          // Issue a query from the client
+          payload = ByteVector.fromValidHex("c0ffee")
+          queryFiber <- client.query(payload).fork
+
+          // Wait for Query message
+          (_, queryMsg) <- waitForMessage[Query](mockServer)
+
+          // Respond with matching correlationId
+          result = ByteVector.fromValidHex("beaded")
+          _ <- sendServerMessage(mockServer, routingId, QueryResponse(queryMsg.correlationId, result))
+
+          response <- queryFiber.join
+        } yield assertTrue(response == result)
+      }
+    }
+
     // ==========================================================================
     // Request Submission Tests
     // ==========================================================================

client-server-server/src/test/scala/zio/raft/server/QueryServerSpec.scala

@@ -45,33 +45,17 @@ object QueryServerSpec extends ZIOSpec[TestEnvironment & ZContext]:
 
   override def spec = suiteAll("Server Query Handling") {
 
-    test("Follower rejects Query with NotLeaderAnymore SessionClosed") {
-      for
-        server <- RaftServer.make(s"tcp://0.0.0.0:$testPort")
-        _ <- ZIO.sleep(200.millis)
-        client <- ZSocket.client
-        _ <- client.connect(serverAddress)
-        // Send Query without a session (server derives routing but is follower)
-        now <- Clock.instant
-        _ <- sendClientMessage(client, Query(CorrelationId.fromString("q1"), ByteVector(1,2,3), now))
-        msg <- waitForMessage[SessionClosed](client)
-      yield assertTrue(msg.reason == SessionCloseReason.NotLeaderAnymore)
-    }
-
     test("Leader forwards Query to Raft and send QueryResponse back") {
       for
         server <- RaftServer.make(s"tcp://0.0.0.0:$testPort")
-        _ <- ZIO.sleep(200.millis)
         _ <- server.stepUp(Map.empty)
-        _ <- ZIO.sleep(100.millis)
 
         client <- ZSocket.client
         _ <- client.connect(serverAddress)
 
         // Create session first
         nonce <- Nonce.generate()
         _ <- sendClientMessage(client, CreateSession(Map("kv" -> "v1"), nonce))
-        _ <- ZIO.sleep(100.millis)
         action <- server.raftActions.take(1).runCollect.map(_.head)
         sessionId = action.asInstanceOf[RaftAction.CreateSession].sessionId
         _ <- server.confirmSessionCreation(sessionId)
@@ -84,7 +68,6 @@ object QueryServerSpec extends ZIOSpec[TestEnvironment & ZContext]:
         _ <- sendClientMessage(client, Query(corr, payload, now))
 
         // Verify RaftAction.Query queued
-        _ <- ZIO.sleep(100.millis)
         qAction <- server.raftActions.take(1).runCollect.map(_.head)
         verified = qAction.isInstanceOf[RaftAction.Query]
         _ <- ZIO.fail(new RuntimeException("Expected RaftAction.Query")).unless(verified).ignore