fix lowest pending request id, to clear lower

Author: somdoron

kvstore/src/test/scala/zio/kvstore/KVSessionSpec.scala

@@ -22,7 +22,7 @@ object KVSessionSpec extends ZIOSpecDefault:
         createdAt = now,
         sessionId = sessionId,
         requestId = RequestId(1),
-        lowestRequestId = RequestId(1),
+        lowestPendingRequestId = RequestId(1),
         command = KVCommand.Set("k", "v")
       )
       val (s2, _) = sm.apply(set).run(s1)
@@ -36,7 +36,7 @@ object KVSessionSpec extends ZIOSpecDefault:
         createdAt = now,
         sessionId = sessionId,
         requestId = RequestId(2),
-        lowestRequestId = RequestId(1),
+        lowestPendingRequestId = RequestId(1),
         command = KVCommand.Get("k")
       )
       val (_, result) = sm.apply(get).run(s3)

kvstore/src/test/scala/zio/kvstore/KVStateMachineSpec.scala

@@ -23,7 +23,7 @@ object KVStateMachineSpec extends ZIOSpecDefault:
         createdAt = now,
         sessionId = sessionId,
         requestId = RequestId(1),
-        lowestRequestId = RequestId(1),
+        lowestPendingRequestId = RequestId(1),
         command = KVCommand.Set("a", "v1")
       )
       val (s2, _) = sm.apply(set1).run(s1)
@@ -33,7 +33,7 @@ object KVStateMachineSpec extends ZIOSpecDefault:
         createdAt = now,
         sessionId = sessionId,
         requestId = RequestId(1),
-        lowestRequestId = RequestId(1),
+        lowestPendingRequestId = RequestId(1),
         command = KVCommand.Set("a", "v2")
       )
       val (s3, _) = sm.apply(set2).run(s2)
@@ -42,7 +42,7 @@ object KVStateMachineSpec extends ZIOSpecDefault:
         createdAt = now,
         sessionId = sessionId,
         requestId = RequestId(2),
-        lowestRequestId = RequestId(1),
+        lowestPendingRequestId = RequestId(1),
         command = KVCommand.Get("a")
       )
       val (_, result) = sm.apply(get).run(s3)

session-state-machine/src/main/scala/zio/raft/sessionstatemachine/SessionCommand.scala

@@ -51,7 +51,7 @@ object SessionCommand:
     createdAt: Instant,
     sessionId: SessionId,
     requestId: RequestId,
-    lowestRequestId: RequestId,
+    lowestPendingRequestId: RequestId,
     command: UC
   ) extends SessionCommand[UC, SR]:
     // Response type can be an error or the user command's response with server request envelopes

session-state-machine/src/main/scala/zio/raft/sessionstatemachine/SessionStateMachine.scala

@@ -28,8 +28,9 @@ import java.time.Instant
   *   - Response caching for duplicate requests
   *   - Server-initiated request management with cumulative acknowledgment
   *   - Session lifecycle coordination
-  *   - Cache cleanup driven by client-provided lowestRequestId (inclusive: removes requestId <= lowestRequestId)
-  *   - Evicted response detection: if requestId <= highestLowestRequestIdSeen and not in cache, return
+  *   - Cache cleanup driven by client-provided lowestPendingRequestId (exclusive: removes requestId <
+  *     lowestPendingRequestId)
+  *   - Evicted response detection: if requestId < highestLowestPendingRequestIdSeen and not in cache, return
   *     RequestError.ResponseEvicted
   *
   * ## Type Parameters
@@ -269,17 +270,17 @@ trait SessionStateMachine[UC <: Command, R, SR, UserSchema <: Tuple]
     *
     * Implementation of Raft dissertation Chapter 6.3 session management protocol:
     *   1. Check cache for (sessionId, requestId) 2. If cache hit, return cached response 3. If cache miss, check if
-    *      requestId <= highestLowestRequestIdSeen → response was evicted, return error 4. If cache miss + requestId >
-    *      highestLowestRequestIdSeen, execute command and update highestLowestRequestIdSeen
+    *      requestId < highestLowestPendingRequestIdSeen → response was evicted, return error 4. If cache miss +
+    *      requestId ≥ highestLowestPendingRequestIdSeen, execute command and update highestLowestPendingRequestIdSeen
     *
-    * This correctly handles out-of-order requests. The lowestRequestId from the client tells us which responses have
-    * been acknowledged (inclusive) and can be evicted. We only update highestLowestRequestIdSeen for requests we
-    * actually process.
+    * This correctly handles out-of-order requests. The lowestPendingRequestId from the client tells us the lowest
+    * sequence number without a response; we can evict responses with lower numbers. We only update
+    * highestLowestPendingRequestIdSeen for requests we actually process.
     */
   private def handleClientRequest(cmd: SessionCommand.ClientRequest[UC, SR])
     : State[HMap[Schema], Either[RequestError, (cmd.command.Response, List[ServerRequestEnvelope[SR]])]] =
     for
-      highestLowestSeen <- getHighestLowestRequestIdSeen(cmd.sessionId)
+      highestLowestSeen <- getHighestLowestPendingRequestIdSeen(cmd.sessionId)
       cachedOpt <- getCachedResponse((cmd.sessionId, cmd.requestId))
       result <- cachedOpt match
         case Some(cachedResponse) =>
@@ -288,17 +289,17 @@ trait SessionStateMachine[UC <: Command, R, SR, UserSchema <: Tuple]
 
         case None =>
           // Cache miss - check if response was evicted
-          // If requestId <= highestLowestRequestIdSeen, client has acknowledged receiving this response
-          if cmd.requestId.isLowerOrEqual(highestLowestSeen) then
-            // Client said "I have responses for all requestIds <= highestLowest", so this was evicted
+          // If requestId < highestLowestPendingRequestIdSeen, client has acknowledged receiving this response
+          if cmd.requestId.isLowerThan(highestLowestSeen) then
+            // Client said "I have responses for all requestIds < highestLowestPending", so this was evicted
             State.succeed(Left(RequestError.ResponseEvicted(cmd.sessionId, cmd.requestId)))
           else
-            // requestId >= highestLowestRequestIdSeen
+            // requestId >= highestLowestPendingRequestIdSeen
             // This is a valid request (not yet acknowledged), execute the command
             for
-              // Update highestLowestRequestIdSeen ONLY when actually executing a new request
-              _ <- updateHighestLowestRequestIdSeen(cmd.sessionId, cmd.lowestRequestId)
-              _ <- cleanupCache(cmd.sessionId, cmd.lowestRequestId)
+              // Update highestLowestPendingRequestIdSeen ONLY when actually executing a new request
+              _ <- updateHighestLowestPendingRequestIdSeen(cmd.sessionId, cmd.lowestPendingRequestId)
+              _ <- cleanupCache(cmd.sessionId, cmd.lowestPendingRequestId)
               (serverRequestsLog, response) <- applyCommand(cmd.command, cmd.createdAt).withLog
               assignedRequests <- addServerRequests(cmd.createdAt, serverRequestsLog)
               _ <- cacheResponse((cmd.sessionId, cmd.requestId), response)
@@ -318,31 +319,31 @@ trait SessionStateMachine[UC <: Command, R, SR, UserSchema <: Tuple]
   ): State[HMap[Schema], Unit] =
     State.update(_.updated["cache"](key, response))
 
-  /** Get the highest lowestRequestId seen from the client for a session.
+  /** Get the highest lowestPendingRequestId seen from the client for a session.
     *
-    * This tracks the highest value of lowestRequestId that the client has sent, indicating which responses the client
-    * has acknowledged receiving.
+    * This tracks the highest value of lowestPendingRequestId that the client has sent, indicating below which requestId
+    * responses have been acknowledged and can be discarded.
     *
-    * Returns RequestId.zero if no lowestRequestId has been seen yet (no requests have been acknowledged).
+    * Returns RequestId.zero if no lowestPendingRequestId has been seen yet (no requests have been acknowledged).
     */
-  private def getHighestLowestRequestIdSeen(sessionId: SessionId): State[HMap[Schema], RequestId] =
-    State.get.map(_.get["highestLowestRequestIdSeen"](sessionId).getOrElse(RequestId.zero))
+  private def getHighestLowestPendingRequestIdSeen(sessionId: SessionId): State[HMap[Schema], RequestId] =
+    State.get.map(_.get["highestLowestPendingRequestIdSeen"](sessionId).getOrElse(RequestId.zero))
 
-  /** Update the highest lowestRequestId seen from the client (only if lowestRequestId > current highest).
+  /** Update the highest lowestPendingRequestId seen from the client (only if it increased).
     *
-    * The lowestRequestId from the client indicates "I have received all responses for requestIds <= this value
-    * (inclusive)". We track the highest such value to detect evicted responses.
+    * The lowestPendingRequestId from the client indicates "I have not yet received a response for this requestId". We
+    * discard cached responses with lower requestIds and track the highest such value to detect evicted responses.
     */
-  private def updateHighestLowestRequestIdSeen(
+  private def updateHighestLowestPendingRequestIdSeen(
     sessionId: SessionId,
-    lowestRequestId: RequestId
+    lowestPendingRequestId: RequestId
   ): State[HMap[Schema], Unit] =
     State.update(state =>
-      state.get["highestLowestRequestIdSeen"](sessionId) match
-        case Some(current) if !lowestRequestId.isGreaterThan(current) =>
-          state // Don't update if lowestRequestId is not higher
+      state.get["highestLowestPendingRequestIdSeen"](sessionId) match
+        case Some(current) if !lowestPendingRequestId.isGreaterThan(current) =>
+          state // Don't update if not higher
         case _ =>
-          state.updated["highestLowestRequestIdSeen"](sessionId, lowestRequestId)
+          state.updated["highestLowestPendingRequestIdSeen"](sessionId, lowestPendingRequestId)
     )
 
   /** Handle ServerRequestAck with cumulative acknowledgment.
@@ -512,32 +513,33 @@ trait SessionStateMachine[UC <: Command, R, SR, UserSchema <: Tuple]
         .removedAll["serverRequests"](serverRequestKeysToRemove)
         .removed["metadata"](sessionId)
         .removed["lastServerRequestId"](sessionId)
-        .removed["highestLowestRequestIdSeen"](sessionId)
+        .removed["highestLowestPendingRequestIdSeen"](sessionId)
     }
 
   // ====================================================================================
   // HELPER METHODS - All return State[HMap[Schema], A] for composition
   // ====================================================================================
 
-  /** Clean up cached responses based on lowestRequestId (Lowest Sequence Number Protocol).
+  /** Clean up cached responses based on lowestPendingRequestId (Lowest Sequence Number Protocol).
     *
-    * Removes all cached responses for the session with requestId <= lowestRequestId. This allows the client to control
-    * cache cleanup by telling the server which responses it no longer needs (Chapter 6.3 of Raft dissertation).
+    * Removes all cached responses for the session with requestId < lowestPendingRequestId. This allows the client to
+    * control cache cleanup by telling the server the lowest requestId without a response (Chapter 6.3 of Raft
+    * dissertation).
     *
-    * The client sends lowestRequestId to indicate "I have received all responses up to and including this ID".
+    * The client sends lowestPendingRequestId to indicate "I have not yet received a response for this ID".
     *
     * Uses range queries to efficiently find and remove old cache entries.
     */
   private def cleanupCache(
     sessionId: SessionId,
-    lowestRequestId: RequestId
+    lowestPendingRequestId: RequestId
   ): State[HMap[Schema], Unit] =
     State.update { state =>
-      // Use range to find all cache entries for this session with requestId <= lowestRequestId
-      // Range is [from, until), so to include lowestRequestId, we use lowestRequestId.next as upper bound
+      // Use range to find all cache entries for this session with requestId < lowestPendingRequestId
+      // Range is [from, until), so use lowestPendingRequestId as the exclusive upper bound
       val keysToRemove = state.range["cache"](
         (sessionId, RequestId.zero),
-        (sessionId, lowestRequestId.next)
+        (sessionId, lowestPendingRequestId)
       ).map((key, _) => key)
 
       // Remove all old cache entries in one efficient operation

session-state-machine/src/main/scala/zio/raft/sessionstatemachine/package.scala

@@ -141,8 +141,8 @@ package object sessionstatemachine:
     /** Response was cached but has been evicted. Client must create a new session.
       *
       * This error occurs when:
-      *   1. Client retries a request with requestId <= highestLowestRequestIdSeen for the session 2. The response is
-      *      not in the cache (was evicted via cleanupCache)
+      *   1. Client retries a request with requestId < highestLowestPendingRequestIdSeen for the session 2. The response
+      *      is not in the cache (was evicted via cleanupCache)
       *
       * Per Raft dissertation Chapter 6.3, the client should create a new session and retry the operation.
       */
@@ -157,7 +157,7 @@ package object sessionstatemachine:
     *   - "serverRequests": ((SessionId, RequestId), PendingServerRequest[?]) - pending requests with composite key for
     *     efficiency
     *   - "lastServerRequestId": (SessionId, RequestId) - last assigned server request ID per session
-    *   - "highestLowestRequestIdSeen": (SessionId, RequestId) - highest lowestRequestId acknowledged by client (for
+    *   - "highestLowestPendingRequestIdSeen": (SessionId, RequestId) - highest lowestPendingRequestId observed (for
     *     eviction detection)
     *
     * Both cache and serverRequests use composite keys (SessionId, RequestId) for better performance:
@@ -168,10 +168,10 @@ package object sessionstatemachine:
     *   - Proper ordering: RequestId ordering is numeric (big-endian encoding), not lexicographic
     *   - No data duplication: sessionId and requestId are not stored in the value, only in the key
     *
-    * The highestLowestRequestIdSeen prefix enables detection of evicted responses:
-    *   - Client sends lowestRequestId indicating "I have received all responses <= this ID (inclusive)"
-    *   - We track the highest lowestRequestId value we've seen from the client
-    *   - When a ClientRequest arrives, we check if requestId <= highestLowestRequestIdSeen
+    * The highestLowestPendingRequestIdSeen prefix enables detection of evicted responses:
+    *   - Client sends lowestPendingRequestId indicating the lowest sequence number without a response
+    *   - We track the highest such value received from the client
+    *   - When a ClientRequest arrives, we check if requestId < highestLowestPendingRequestIdSeen
     *   - If yes and response is not in cache, we know it was evicted (client already acknowledged it)
     *   - This correctly handles out-of-order requests while preventing re-execution of acknowledged commands
     */
@@ -180,7 +180,7 @@ package object sessionstatemachine:
       ("cache", (SessionId, RequestId), R) *:
       ("serverRequests", (SessionId, RequestId), PendingServerRequest[SR]) *:
       ("lastServerRequestId", SessionId, RequestId) *:
-      ("highestLowestRequestIdSeen", SessionId, RequestId) *:
+      ("highestLowestPendingRequestIdSeen", SessionId, RequestId) *:
       EmptyTuple
 
   /** KeyLike instance for SessionId keys. Used by metadata, serverRequests, and lastServerRequestId prefixes.

session-state-machine/src/test/scala/zio/raft/sessionstatemachine/IdempotencySpec.scala

@@ -162,42 +162,41 @@ object IdempotencySpec extends ZIOSpecDefault:
           .asInstanceOf[SessionCommand[TestCommand, String]]
       val (state1, _) = sm.apply(createCmd).run(state0)
 
-      // First request (requestId=1, lowestRequestId=1)
+      // First request (requestId=1, lowestPendingRequestId=1)
       val cmd1: SessionCommand[TestCommand, String] =
         SessionCommand.ClientRequest(now, sessionId, RequestId(1), RequestId(1), TestCommand.Increment(10))
       val (state2, result1) = sm.apply(cmd1).run(state1)
       val Right((response1, _)) = (result1.asInstanceOf[Either[RequestError, (Int, List[Any])]]): @unchecked
 
-      // Second request (requestId=2, lowestRequestId=1)
+      // Second request (requestId=2, lowestPendingRequestId=1)
       val cmd2: SessionCommand[TestCommand, String] =
         SessionCommand.ClientRequest(now, sessionId, RequestId(2), RequestId(1), TestCommand.Increment(5))
       val (state3, result2) = sm.apply(cmd2).run(state2)
       val Right((response2, _)) = (result2.asInstanceOf[Either[RequestError, (Int, List[Any])]]): @unchecked
 
-      // Third request (requestId=3, lowestRequestId=2)
-      // Client says "I have received responses for requestIds <= 2 (inclusive)", so responses 1 and 2 can be evicted
-      // This triggers cache cleanup AND updates highestLowestRequestIdSeen to 2
+      // Third request (requestId=3, lowestPendingRequestId=2)
+      // Client says "lowestPendingRequestId=2", so responses with requestId < 2 can be evicted (only 1)
+      // This triggers cache cleanup AND updates highestLowestPendingRequestIdSeen to 2
       val cmd3: SessionCommand[TestCommand, String] =
         SessionCommand.ClientRequest(now, sessionId, RequestId(3), RequestId(2), TestCommand.Increment(1))
       val (state4, _) = sm.apply(cmd3).run(state3)
 
-      // Now retry request 2 - should fail with ResponseEvicted
-      // requestId(2) <= highestLowestRequestIdSeen(2) AND not in cache (was evicted)
+      // Now retry request 1 - should fail with ResponseEvicted (evicted)
       val cmd4: SessionCommand[TestCommand, String] =
-        SessionCommand.ClientRequest(now, sessionId, RequestId(2), RequestId(2), TestCommand.Increment(999))
+        SessionCommand.ClientRequest(now, sessionId, RequestId(1), RequestId(2), TestCommand.Increment(999))
       val (state5, result4) = sm.apply(cmd4).run(state4)
 
       (result4.asInstanceOf[Either[RequestError, (Int, List[Any])]]: @unchecked) match
         case Left(RequestError.ResponseEvicted(sid, rid)) =>
           assertTrue(
             sid == sessionId &&
-              rid == RequestId(2) &&
+              rid == RequestId(1) &&
               sm.callCount == 3 // Command was NOT executed again (only 3 commands processed)
           )
         case Right(_) =>
           assertTrue(false) // Should not succeed
     },
-    test("PC-3: Cache cleanup removes responses based on lowestRequestId") {
+    test("PC-3: Cache cleanup removes responses based on lowestPendingRequestId (exclusive)") {
       val sm = new TestStateMachine()
       val state0: HMap[CombinedSchema] = HMap.empty[CombinedSchema]
       val now = Instant.now()
@@ -228,16 +227,15 @@ object IdempotencySpec extends ZIOSpecDefault:
       assertTrue(cache4.get["cache"]((sessionId, RequestId(2))).isDefined) &&
       assertTrue(cache4.get["cache"]((sessionId, RequestId(3))).isDefined)
 
-      // Execute request 4 with lowestRequestId=2 (client says "I have received 1 and 2 (inclusive)")
+      // Execute request 4 with lowestPendingRequestId=2 (client says "lowest pending is 2")
       val cmd4: SessionCommand[TestCommand, String] =
         SessionCommand.ClientRequest(now, sessionId, RequestId(4), RequestId(2), TestCommand.Increment(1))
       val (state5, _) = sm.apply(cmd4).run(state4)
 
-      // Requests 1 and 2 should be cleaned up (<= lowestRequestId=2, inclusive)
-      // Request 3 and 4 should still be cached (> lowestRequestId=2)
+      // Requests with id < 2 should be cleaned up (only 1). 2, 3, 4 should still be cached.
       val cache5 = state5.asInstanceOf[HMap[CombinedSchema]]
       assertTrue(cache5.get["cache"]((sessionId, RequestId(1))).isEmpty) &&
-      assertTrue(cache5.get["cache"]((sessionId, RequestId(2))).isEmpty) &&
+      assertTrue(cache5.get["cache"]((sessionId, RequestId(2))).isDefined) &&
       assertTrue(cache5.get["cache"]((sessionId, RequestId(3))).isDefined) &&
       assertTrue(cache5.get["cache"]((sessionId, RequestId(4))).isDefined)
     }

specs/002-session-state-machine/ARCHITECTURE_FINAL.md

@@ -162,7 +162,7 @@ object SessionCommand:
   case class ClientRequest[UC <: Command](
     sessionId: SessionId,
     requestId: RequestId,
-    lowestRequestId: RequestId,  // For cache cleanup (Lowest Sequence Number Protocol)
+    lowestPendingRequestId: RequestId,  // For cache cleanup (Lowest Sequence Number Protocol, exclusive: evict < this)
     command: UC  // Already decoded! UC is a Command subtype
   ) extends SessionCommand[UC]:
     type Response = (command.Response, List[Any])  // (user response, server requests)